Architecture

This page describes exactly what Pipeline::run does in Carbon 2.0.0, from the moment you call it until it returns. Read it when you need to know in which order processors run, what happens when one is slow or fails, and when the pipeline stops. The behavior described here comes from crates/core/src/pipeline.rs.

In short#

Each datasource runs in its own task and sends updates into one shared channel. The pipeline takes updates off that channel one at a time and hands each one to every pipe registered for its type; a pipe filters it, decodes it and awaits your processor before the next update starts. A processor error is logged and counted, and processing carries on. The pipeline stops when you cancel its token, when you press Ctrl-C, or when every datasource has finished.

The rest of this page is the exact behaviour behind each of those sentences.

The parts of a pipeline#

You build a pipeline with Pipeline::builder(). The builder collects:

• Datasources, each paired with a DatasourceId. datasource() assigns a random UUID; datasource_with_id() uses the ID you give it.
• Pipes, one per account, instruction, transaction, account_deletions or block_details call. A pipe bundles a decoder (where the update type needs one), your processor and its filters.
• Metrics exporters added with metrics().
• Settings: the shutdown strategy (default ProcessPending), an optional cancellation token, and channel_buffer_size (default DEFAULT_CHANNEL_BUFFER_SIZE, 1,000).

build() registers the pipeline's built-in metrics in the global metrics registry and returns the Pipeline. In 2.0.0 it doesn't validate anything and always returns Ok.

What run does#

1. Logs a start line with the number of datasources, exporters and pipes, at info level.
2. Initializes exporters. It calls MetricsExporter::initialize on each exporter in order. LogMetrics starts its five-second flush task here, and PrometheusMetrics::with_server starts its HTTP server. If an exporter returns an error, run returns that error before any datasource starts.
3. Creates the channel, a bounded Tokio MPSC channel of (Update, DatasourceId) with room for channel_buffer_size updates.
4. Picks the cancellation token: the one you passed to datasource_cancellation_token(), or a new one.
5. Starts every datasource in its own Tokio task with tokio::spawn, calling consume(id, sender, token) with a clone of the channel's sender and of the token. If consume returns an error, the task logs error consuming datasource: … and ends. The pipeline isn't told and doesn't retry; the other datasources keep running.
6. Drops its own sender, so the channel closes once every datasource has dropped its copy.
7. Runs the loop. Each iteration waits for whichever comes first:
   • The token is cancelled: exports metrics once, shuts the exporters down and returns Ok(()).
   • Ctrl-C: cancels the token. With ShutdownStrategy::Immediate it exports metrics, shuts the exporters down and returns. With ProcessPending it logs a message and continues the loop. The loop notices Ctrl-C only while it's waiting for an update; see Shutdown for what that means in practice.
   • An update arrives: processes it as described below, then starts the next iteration.
   • The channel is closed and empty (every datasource finished): logs update_receiver closed, shutting down., exports metrics, shuts the exporters down and returns Ok(()).

How each update is processed#

For every update it receives, the loop increments carbon_updates_received_total, processes the update through every matching pipe, records the time it took in two histograms, and increments either carbon_updates_successful_total or carbon_updates_failed_total, then carbon_updates_processed_total. It then sets carbon_updates_queued to the number of updates still waiting in the channel.

The update's variant decides which pipes see it. For each pipe, the pipeline runs the pipe's filters in order and stops at the first Reject. A pipe runs only if every filter returns Accept.

Account updates#

The pipeline builds an AccountMetadata (slot, pubkey, transaction signature) and visits each account pipe in the order you registered them. A pipe whose filters accept calls its decoder's decode_account. If the decoder returns None, the account isn't for this pipe and the processor isn't called. Otherwise the processor receives an AccountProcessorInputType with the metadata, the decoded account and the raw account.

Transactions#

1. The pipeline converts the TransactionUpdate into a TransactionMetadata, shared through an Arc. The fee payer is the first static account key; a transaction without one fails with MissingFeePayer before any pipe runs.
2. It builds the instruction list: each top-level instruction, followed by its inner instructions in execution order, with account keys resolved from the message and, for v0 transactions, the loaded address lookup tables. Legacy, v0 and V1 messages are supported. An inner instruction without a valid stack height (2 to 5) is dropped, together with the rest of its group, and a warning is logged.
3. It rebuilds that list into a tree of NestedInstructions, and flattens the tree depth-first.
4. Instruction pipes run first. For each instruction pipe, in registration order, the pipeline walks every instruction of the transaction in depth-first order: top-level instruction 0, its inner instructions, top-level instruction 1, and so on. Each instruction that passes the pipe's filters goes to the pipe's decoder; when the decoder returns Some, the processor is called with an InstructionProcessorInputType. The first pipe sees all instructions before the second pipe sees any.
5. Transaction pipes run next, in registration order. Their filters receive the transaction metadata and the instruction tree. A pipe that accepts decodes every instruction with its InstructionDecoderCollection and calls the processor once with the decoded list, even if the list is empty.

Account deletions and block details#

AccountDeletion and BlockDetails updates go to the account_deletions and block_details pipes, in registration order, after filtering. There is no decoder: the processor receives the update itself.

Ordering and concurrency guarantees#

• One update at a time. The loop processes an update completely, through every matching pipe, before it takes the next one off the channel. Processors never run concurrently with each other, and a processor's process calls never overlap, which is why process takes &mut self.
• Registration order within an update. Pipes of the same kind run in the order you added them. For a transaction, every instruction pipe runs before any transaction pipe.
• Execution order within a transaction. Instruction pipes see instructions in depth-first order, which is the order in which they executed.
• Channel order across updates. Updates are processed in the order they entered the channel. With one datasource, that's the order it sends them in. With several, their updates interleave in whatever order they arrive. Carbon doesn't sort by slot or remove duplicates unless you add a filter for it.
• No parallelism inside the pipeline. To use more cores, make process cheap and hand heavy work to tasks or channels you own. See Sharing state. Once you do, ordering across that work is up to you.

When processing falls behind#

A slow processor slows the whole loop, and updates accumulate in the channel up to channel_buffer_size. What happens when the channel is full depends on the datasource:

• Datasources that send with send().await wait until there's room. Back-pressure reaches the source and nothing is lost. The RPC and Helius getProgramAccounts datasources, the Helius GTFA datasource, the validator snapshot datasource and the stream message datasource behave this way, as does the RPC transaction crawler with blocking_send: true.
• Datasources that send with try_send drop the update and log an error. Yellowstone gRPC, Helius LaserStream, Helius Atlas websockets, RPC block subscribe, RPC program subscribe, the RPC block crawler, and the transaction crawler with its default blocking_send: false behave this way.

Watch carbon_updates_queued. If it stays near channel_buffer_size, raise the buffer with .channel_buffer_size(…) to absorb bursts, and make the processors faster for sustained load.

Errors#

• A processor returns Err. The pipe's ? stops processing of that update: pipes after it don't run for this update, and for a transaction the remaining instructions and the transaction pipes are skipped too. The loop logs error processing update ({update:?}): {error:?} at error level, counts the update in carbon_updates_failed_total and moves on to the next update. Nothing is retried.
• A datasource's consume returns Err. The error is logged and that datasource stops. The pipeline continues with the others; if none are left, the channel closes and run returns Ok(()).
• A processor panics. The pipeline doesn't catch panics. The panic propagates out of run, so with #[tokio::main] the program exits.

Error handling covers patterns for these cases.

Metrics flushing#

The pipeline updates its counters, gauge and histograms on every update, but it calls MetricsExporter::export only once, when it shuts down, followed by shutdown on each exporter. Periodic export is the exporter's job: LogMetrics logs a snapshot every five seconds from its own task, and Prometheus reads the registry each time it scrapes /metrics. See Metrics.

Example#

This program feeds three hand-built account updates through the stream message datasource, registers two account pipes and a deletion pipe, and prints what each processor sees. The source is finite, so the pipeline returns once it has processed all three.

cargo add carbon-core@2.0.0 carbon-stream-message-datasource@2.0.0
cargo add solana-account@4.3.2 solana-pubkey@4.2.1
cargo add tokio --features full

use {
    carbon_core::{
        account::{AccountDecoder, AccountProcessorInputType, DecodedAccount},
        datasource::{AccountDeletion, AccountUpdate},
        error::CarbonResult,
        pipeline::Pipeline,
        processor::Processor,
    },
    carbon_stream_message_datasource::{StreamMessageClient, UnifiedMessage},
    solana_account::Account,
    solana_pubkey::Pubkey,
};

/// Accepts every account and exposes the length of its data.
struct AnyAccount;

impl<'a> AccountDecoder<'a> for AnyAccount {
    type AccountType = usize;

    fn decode_account(&self, account: &'a Account) -> Option<DecodedAccount<usize>> {
        Some(DecodedAccount {
            lamports: account.lamports,
            data: account.data.len(),
            owner: account.owner,
            executable: account.executable,
            rent_epoch: account.rent_epoch,
        })
    }
}

struct Print(&'static str);

impl Processor<AccountProcessorInputType<'_, usize>> for Print {
    async fn process(&mut self, input: &AccountProcessorInputType<'_, usize>) -> CarbonResult<()> {
        println!(
            "{}: slot {} lamports {}",
            self.0, input.metadata.slot, input.decoded_account.lamports
        );
        Ok(())
    }
}

impl Processor<AccountDeletion> for Print {
    async fn process(&mut self, deletion: &AccountDeletion) -> CarbonResult<()> {
        println!("{}: slot {} account closed", self.0, deletion.slot);
        Ok(())
    }
}

#[tokio::main]
async fn main() -> CarbonResult<()> {
    // Hand-built updates stand in for a real datasource.
    let (tx, rx) = tokio::sync::mpsc::channel(16);
    for (slot, lamports) in [(100, 2_039_280), (101, 1_461_600), (102, 0)] {
        let update = AccountUpdate {
            pubkey: Pubkey::new_unique(),
            account: Account { lamports, ..Account::default() },
            slot,
            transaction_signature: None,
        };
        tx.send(UnifiedMessage::Account(update)).await.unwrap();
    }
    drop(tx); // The source is finite: the pipeline returns once it drains.

    Pipeline::builder()
        .datasource(StreamMessageClient::new(rx))
        .account(AnyAccount, Print("pipe A"))
        .account(AnyAccount, Print("pipe B"))
        .account_deletions(Print("deletions"))
        .build()?
        .run()
        .await
}

cargo run

pipe A: slot 100 lamports 2039280
pipe B: slot 100 lamports 2039280
pipe A: slot 101 lamports 1461600
pipe B: slot 101 lamports 1461600
deletions: slot 102 account closed

Both account pipes see each update, in registration order, before the next update starts. The zero-lamport account arrives as an AccountDeletion (see Updates), so only the deletion pipe sees it.

Next steps#