# Architecture (Carbon 2.0)

> How Pipeline::run moves an update from datasource to processor, with the exact ordering, concurrency, error, metrics and shutdown rules.

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`](https://github.com/sevenlabs-hq/carbon/blob/v2.0.0/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()`](https://carbon.sevenlabs.org/reference/pipeline-builder.md). The builder collects:

- **Datasources**, each paired with a [`DatasourceId`](https://carbon.sevenlabs.org/concepts/datasources.md#datasource-ids). `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](https://carbon.sevenlabs.org/concepts/processors.md) and its [filters](https://carbon.sevenlabs.org/concepts/filters.md).
- **Metrics exporters** added with `metrics()`.
- **Settings**: the [shutdown strategy](https://carbon.sevenlabs.org/concepts/shutdown.md) (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](https://carbon.sevenlabs.org/concepts/metrics.md) and returns the `Pipeline`. In 2.0.0 it doesn't validate anything and always returns `Ok`.

![What the pipeline does with each update](https://carbon.sevenlabs.org/diagrams/pipeline-loop.svg)

**Diagram: What the pipeline does with each update.** Each datasource runs in its own Tokio task and sends (Update, DatasourceId) pairs into one bounded channel. The pipeline loop takes one update at a time and routes it by type. Update::Account goes to the account pipes, which filter, decode and call the processor. Update::Transaction goes first to the instruction pipes, which filter, decode and process each instruction, then to the transaction pipes, which filter, decode every instruction with a decoder collection and call the processor once. Update::AccountDeletion and Update::BlockDetails go to the deletion and block-details pipes, which filter and call the processor without a decoder.

## 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`](https://carbon.sevenlabs.org/concepts/metrics.md#log-metrics) 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](https://carbon.sevenlabs.org/concepts/shutdown.md#ctrl-c) 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`](https://carbon.sevenlabs.org/concepts/processors.md#account-pipes) 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`](https://carbon.sevenlabs.org/concepts/instructions.md), 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`](https://carbon.sevenlabs.org/concepts/processors.md#instruction-pipes). 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`](https://carbon.sevenlabs.org/concepts/transactions.md) 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](https://carbon.sevenlabs.org/concepts/filters.md) 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](https://carbon.sevenlabs.org/concepts/processors.md#share-state-with-the-rest-of-your-program). 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](https://carbon.sevenlabs.org/concepts/error-handling.md) 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](https://carbon.sevenlabs.org/concepts/metrics.md).

## Example

This program feeds three hand-built account updates through the [stream message datasource](https://carbon.sevenlabs.org/datasources/stream-message.md), 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.

```sh
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
```

`src/main.rs`:

```rust
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
}
```

```sh
cargo run
```

Output:

```text
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](https://carbon.sevenlabs.org/concepts/updates.md#account-deletions)), so only the deletion pipe sees it.

## Next steps

- [Updates](https://carbon.sevenlabs.org/concepts/updates.md): The four update types a datasource can send.
- [Processors](https://carbon.sevenlabs.org/concepts/processors.md): What each pipe passes to your code, and how to keep state.
- [Shutdown](https://carbon.sevenlabs.org/concepts/shutdown.md): How the pipeline stops, and what happens to queued updates.

---

Carbon 2.0.0 documentation · Web: https://carbon.sevenlabs.org/concepts/architecture · Last updated 2026-11-09 · Index of all pages: https://carbon.sevenlabs.org/llms.txt