Pipeline builder

Pipeline connects datasources to processors. You create one with Pipeline::builder(), register datasources, pipes and metrics exporters on the returned PipelineBuilder, call build(), and then run().await. This page lists every method in carbon_core::pipeline for Carbon 2.0.0. For the model behind it, see Architecture.

Source: crates/core/src/pipeline.rs.

Example#

This complete program crawls recent Meteora DLMM transactions on the public RPC endpoint. It uses a named datasource, an instruction pipe with two filters, a metrics exporter, a cancellation token, an explicit shutdown strategy and a larger channel. The fragments in the entries below come from this program.

[dependencies]
carbon-core = "2.0.0"
carbon-log-metrics = "2.0.0"
carbon-meteora-dlmm-decoder = "2.0.0"
carbon-rpc-transaction-crawler-datasource = "2.0.0"
env_logger = "0.11"
tokio = { version = "1", features = ["full"] }
tokio-util = "0.7"

use {
    carbon_core::{
        datasource::DatasourceId,
        error::CarbonResult,
        filter::{DatasourceFilter, DeduplicationFilter, Filter},
        instruction::InstructionProcessorInputType,
        pipeline::{Pipeline, ShutdownStrategy},
        processor::Processor,
    },
    carbon_log_metrics::LogMetrics,
    carbon_meteora_dlmm_decoder::{
        instructions::MeteoraDlmmInstruction, MeteoraDlmmDecoder, PROGRAM_ID,
    },
    carbon_rpc_transaction_crawler_datasource::{ConnectionConfig, Filters, RpcTransactionCrawler},
    std::{sync::Arc, time::Duration},
    tokio_util::sync::CancellationToken,
};

struct SwapLogger;

impl Processor<InstructionProcessorInputType<'_, MeteoraDlmmInstruction>> for SwapLogger {
    async fn process(
        &mut self,
        input: &InstructionProcessorInputType<'_, MeteoraDlmmInstruction>,
    ) -> CarbonResult<()> {
        if let MeteoraDlmmInstruction::Swap { data, .. } = input.decoded_instruction {
            println!(
                "swap {} slot={} amount_in={}",
                input.metadata.transaction_metadata.signature,
                input.metadata.transaction_metadata.slot,
                data.amount_in
            );
        }
        Ok(())
    }
}

#[tokio::main]
async fn main() -> CarbonResult<()> {
    env_logger::init();

    let crawler_id = DatasourceId::new_named("dlmm-crawler");
    let datasource = RpcTransactionCrawler::new(
        "https://api.mainnet-beta.solana.com".to_string(),
        PROGRAM_ID,
        ConnectionConfig::default(),
        Filters::new(None, None, None),
        None,
    );

    // Stop on Ctrl-C, or after 60 seconds.
    let token = CancellationToken::new();
    let stop = token.clone();
    tokio::spawn(async move {
        tokio::select! {
            _ = tokio::signal::ctrl_c() => {}
            _ = tokio::time::sleep(Duration::from_secs(60)) => {}
        }
        stop.cancel();
    });

    let filters: Vec<Box<dyn Filter>> = vec![
        Box::new(DatasourceFilter::new(crawler_id.clone())),
        Box::new(DeduplicationFilter::new(Duration::from_secs(300))),
    ];

    Pipeline::builder()
        .datasource_with_id(datasource, crawler_id)
        .instruction_with_filters(MeteoraDlmmDecoder, SwapLogger, filters)
        .metrics(Arc::new(LogMetrics::new()))
        .datasource_cancellation_token(token)
        .shutdown_strategy(ShutdownStrategy::Immediate)
        .channel_buffer_size(10_000)
        .build()?
        .run()
        .await
}

RUST_LOG=info cargo run

The public endpoint rate-limits the crawler, so swaps arrive slowly. After 60 seconds the token fires and the pipeline exports metrics one last time and exits with status 0:

[2026-11-09T00:39:21Z INFO  carbon_core::pipeline] starting pipeline. num_datasources: 1, num_exporters: 1, num_account_pipes: 0, num_account_deletion_pipes: 0, num_instruction_pipes: 1, num_transaction_pipes: 0
swap 21qMkWAt3sGx1cN3E1gfyLqMPmXtd2Npnou5rREVBxnDfdW1wS7NSgyZx3Y4Y9uEfzXdyh83HSybpNBosV7BWaeR slot=467119379 amount_in=4313584534
[2026-11-09T00:40:21Z INFO  carbon_log_metrics] 00:01:00 (+376.772µs) | 34 processed (100%), 34 successful, 0 failed (0%), 0 in queue
[2026-11-09T00:40:21Z INFO  carbon_core::pipeline] update_receiver closed, shutting down.
[2026-11-09T00:40:21Z INFO  carbon_core::pipeline] pipeline shutdown complete.

Defaults#

Pipeline::builder#

Returns an empty PipelineBuilder with the defaults.

pub fn builder() -> PipelineBuilder

Details. Equivalent to PipelineBuilder::new() and PipelineBuilder::default(). A pipeline with no datasources builds and runs, but run returns as soon as it starts because the update channel closes immediately.

Example.

let builder = Pipeline::builder();

See also. build, docs.rs.

PipelineBuilder::new#

Same as Pipeline::builder.

pub fn new() -> Self

Details. PipelineBuilder also implements Default. All of its fields are public (datasources, account_pipes, account_deletion_pipes, block_details_pipes, instruction_pipes, transaction_pipes, exporters, datasource_cancellation_token, shutdown_strategy, channel_buffer_size), but the pipe fields hold internal *Pipe types. Use the methods instead.

Example.

let builder = PipelineBuilder::new();

See also. docs.rs.

datasource#

Adds a datasource with a random ID.

pub fn datasource(self, datasource: impl Datasource + 'static) -> Self

Details. The ID is DatasourceId::new_unique(), a UUID v4 string you cannot read back. Use datasource_with_id when a filter needs to match it. run spawns one Tokio task per datasource. You can add any number of datasources; their updates share one channel and are processed in arrival order.

Example.

Pipeline::builder().datasource(datasource)

See also. datasource_with_id, Datasource, Datasources, docs.rs.

datasource_with_id#

Adds a datasource with an ID you choose.

pub fn datasource_with_id(self, datasource: impl Datasource + 'static, id: DatasourceId) -> Self

Details. Create the ID with DatasourceId::new_named("name"). Every update from this datasource carries the ID, and filters read it from FilterContext::datasource_id. The pipeline does not check that IDs are unique: two datasources with the same ID are indistinguishable to filters.

Example.

let crawler_id = DatasourceId::new_named("dlmm-crawler");
Pipeline::builder().datasource_with_id(datasource, crawler_id.clone())

See also. DatasourceId, DatasourceFilter, Multiple datasources.

account#

Adds an account pipe: a decoder and the processor that receives what it decodes.

pub fn account<T, P>(
    self,
    decoder: impl for<'a> AccountDecoder<'a, AccountType = T> + Send + Sync + 'static,
    processor: P,
) -> Self
where
    T: Send + Sync + 'static,
    P: for<'a> Processor<AccountProcessorInputType<'a, T>> + Send + Sync + 'static,

Details. For every Update::Account, the pipe calls decoder.decode_account(&account). When it returns Some, the processor receives an AccountProcessorInputType<T>. When it returns None (for example, the owner is a different program), the pipe skips the update. Zero-lamport accounts arrive as Update::AccountDeletion instead and never reach account pipes; register account_deletions for them.

Example.

Pipeline::builder()
    .datasource(GpaDatasource::new(rpc_url, PROGRAM_ID))
    .account(MeteoraDlmmDecoder, PoolLogger)

See also. account_with_filters, AccountDecoder, Load account state.

account_with_filters#

Adds an account pipe that only runs for updates every filter accepts.

pub fn account_with_filters<T, P>(
    self,
    decoder: impl for<'a> AccountDecoder<'a, AccountType = T> + Send + Sync + 'static,
    processor: P,
    filters: Vec<Box<dyn Filter + 'static>>,
) -> Self

Details. Same bounds as account. The pipeline calls Filter::filter_account on each filter in order and stops at the first Reject, so put stateful filters such as DeduplicationFilter last. Filters run before the decoder.

Example.

Pipeline::builder().account_with_filters(
    MeteoraDlmmDecoder,
    PoolLogger,
    vec![Box::new(SlotRangeFilter::from(400_000_000, None))],
)

See also. Filters reference, Filters.

account_deletions#

Adds a processor for account closures.

pub fn account_deletions<P>(self, processor: P) -> Self
where
    P: Processor<AccountDeletion> + Send + Sync + 'static,

Details. The processor receives &AccountDeletion for every Update::AccountDeletion, with no decoding step. In 2.0.0 the datasources that emit deletions are Yellowstone gRPC, Helius LaserStream, Helius Atlas WebSocket, RPC program subscribe and stream message: any account update with lamports == 0 becomes a deletion. The deletion carries the account's final state in account.

Example.

Pipeline::builder().account_deletions(DeletionLogger)

See also. AccountDeletion, account_deletions_with_filters.

account_deletions_with_filters#

Adds a deletion processor behind filters.

pub fn account_deletions_with_filters<P>(
    self,
    processor: P,
    filters: Vec<Box<dyn Filter + 'static>>,
) -> Self
where
    P: Processor<AccountDeletion> + Send + Sync + 'static,

Details. Calls Filter::filter_account_deletion. DeduplicationFilter does not implement this hook, so it accepts every deletion.

Example.

Pipeline::builder().account_deletions_with_filters(
    DeletionLogger,
    vec![Box::new(DatasourceFilter::new(DatasourceId::new_named("geyser")))],
)

See also. Filters reference.

block_details#

Adds a processor for block-level metadata.

pub fn block_details<P>(self, processor: P) -> Self
where
    P: Processor<BlockDetails> + Send + Sync + 'static,

Details. The processor receives &BlockDetails for every Update::BlockDetails. In 2.0.0 only the RPC block subscribe datasource emits block details; with any other datasource this processor never runs.

Example.

Pipeline::builder().block_details(BlockLogger)

See also. BlockDetails, block_details_with_filters.

block_details_with_filters#

Adds a block-details processor behind filters.

pub fn block_details_with_filters<P>(
    self,
    processor: P,
    filters: Vec<Box<dyn Filter + 'static>>,
) -> Self
where
    P: Processor<BlockDetails> + Send + Sync + 'static,

Details. Calls Filter::filter_block_details.

Example.

Pipeline::builder().block_details_with_filters(
    BlockLogger,
    vec![Box::new(SlotRangeFilter::between(400_000_000, None, 400_100_000, None))],
)

See also. Filters reference.

instruction#

Adds an instruction pipe: a decoder and the processor that receives each instruction it decodes.

pub fn instruction<T, P>(
    self,
    decoder: impl for<'a> InstructionDecoder<'a, InstructionType = T> + Send + Sync + 'static,
    processor: P,
) -> Self
where
    T: Send + Sync + 'static,
    P: for<'a> Processor<InstructionProcessorInputType<'a, T>> + Send + Sync + 'static,

Details. For every Update::Transaction, the pipeline rebuilds the instruction tree and flattens it depth-first: each top-level instruction, then its CPIs. The pipe tries the decoder on every instruction, including inner ones, so a program reached by CPI is processed like a top-level call. When the decoder returns Some, the processor receives an InstructionProcessorInputType<T>, including the instruction's own children in nested_instructions.

Example.

Pipeline::builder()
    .datasource(datasource)
    .instruction(MeteoraDlmmDecoder, SwapLogger)

See also. instruction_with_filters, Instructions, InstructionDecoder.

instruction_with_filters#

Adds an instruction pipe behind filters.

pub fn instruction_with_filters<T, P>(
    self,
    decoder: impl for<'a> InstructionDecoder<'a, InstructionType = T> + Send + Sync + 'static,
    processor: P,
    filters: Vec<Box<dyn Filter + 'static>>,
) -> Self

Details. Same bounds as instruction. Calls Filter::filter_instruction once per instruction, before decoding, so filters also see instructions of other programs.

Example.

let filters: Vec<Box<dyn Filter>> = vec![
    Box::new(DatasourceFilter::new(crawler_id.clone())),
    Box::new(DeduplicationFilter::new(Duration::from_secs(300))),
];
Pipeline::builder().instruction_with_filters(MeteoraDlmmDecoder, SwapLogger, filters)

See also. DeduplicationFilter.

transaction#

Adds a transaction pipe: one processor call per transaction with every instruction the collection T recognizes.

pub fn transaction<T, P>(self, processor: P) -> Self
where
    T: InstructionDecoderCollection + 'static,
    P: for<'a> Processor<TransactionProcessorInputType<'a, T>> + Send + Sync + 'static,

Details. There is no decoder argument: T::parse_instruction runs on each instruction of the flattened list (top-level and inner), and the ones it parses are passed in order in instructions. The processor runs for every transaction, even when instructions is empty, so return early when it is. You usually have to name T with a turbofish or let the processor's type infer it.

Example.

Pipeline::builder()
    .datasource(datasource)
    .transaction::<AllInstructions, _>(TxLogger)

See also. transaction_with_filters, InstructionDecoderCollection, Transactions.

transaction_with_filters#

Adds a transaction pipe behind filters.

pub fn transaction_with_filters<T, P>(
    self,
    processor: P,
    filters: Vec<Box<dyn Filter + 'static>>,
) -> Self
where
    T: InstructionDecoderCollection + 'static,
    P: for<'a> Processor<TransactionProcessorInputType<'a, T>> + Send + Sync + 'static,

Details. Calls Filter::filter_transaction. DeduplicationFilter does not implement this hook, so it does not deduplicate transaction pipes.

Example.

Pipeline::builder().transaction_with_filters::<AllInstructions, _>(
    TxLogger,
    vec![Box::new(SlotRangeFilter::from(400_000_000, Some(0)))],
)

See also. Filters reference.

metrics#

Adds a metrics exporter.

pub fn metrics(self, exporter: Arc<dyn MetricsExporter>) -> Self

Details. You can add several exporters. run calls each exporter's initialize once at startup, then export and shutdown once when the pipeline stops. It does not call export while running: exporters that report periodically, such as LogMetrics, start their own timer in initialize. An error from any of these calls makes run return that error.

Example.

Pipeline::builder().metrics(Arc::new(LogMetrics::new()))

See also. Metrics reference, MetricsExporter.

datasource_cancellation_token#

Gives the pipeline a CancellationToken you control.

pub fn datasource_cancellation_token(self, cancellation_token: CancellationToken) -> Self

Details. run passes a clone of the token to every datasource's consume. When you cancel it, the pipeline finishes the update it is processing, exports metrics, shuts down exporters and returns Ok(()). It does not drain updates still in the channel, whatever the shutdown_strategy. Without this method, run creates a private token that only Ctrl-C cancels. Use this method in production: see Ctrl-C is ignored while processing.

Example.

let token = CancellationToken::new();
let stop = token.clone();
tokio::spawn(async move {
    tokio::signal::ctrl_c().await.ok();
    stop.cancel();
});
Pipeline::builder().datasource_cancellation_token(token)

See also. Shutdown, run.

shutdown_strategy#

Sets what happens when the pipeline receives Ctrl-C.

pub fn shutdown_strategy(self, shutdown_strategy: ShutdownStrategy) -> Self

Details. Applies only to Ctrl-C handled by run itself. Cancelling your own token always behaves like Immediate. See ShutdownStrategy for the exact behaviour of each variant.

Example.

Pipeline::builder().shutdown_strategy(ShutdownStrategy::Immediate)

See also. ShutdownStrategy, Shutdown.

channel_buffer_size#

Sets the capacity of the channel between datasources and the pipeline loop.

pub fn channel_buffer_size(self, size: usize) -> Self

Details. Default 1_000 updates. All datasources share the one channel. When it is full, datasources that use send().await wait (backpressure), and datasources that use try_send drop the update and log an error. Most streaming datasources in 2.0.0 use try_send, and so does the RPC transaction crawler unless ConnectionConfig::blocking_send is true. Raise the size when processors are slower than the source. A size of 0 panics in run (Tokio channels need a capacity of at least 1). The carbon_updates_queued gauge shows the queue length.

Example.

Pipeline::builder().channel_buffer_size(10_000)

See also. DEFAULT_CHANNEL_BUFFER_SIZE, Production.

build#

Finishes the builder and returns the Pipeline.

pub fn build(self) -> CarbonResult<Pipeline>

Details. Registers the pipeline metrics in the global MetricsRegistry (and the Postgres metrics when carbon-core's postgres feature is on), then moves every field into the Pipeline. It never returns Err in 2.0.0, and it does not validate anything: no check for zero datasources, duplicate IDs, or datasources whose update_types() don't match the registered pipes. Registration is not deduplicated, so building two pipelines in one process lists each pipeline metric twice in exported snapshots.

Example.

let mut pipeline = Pipeline::builder()
    .datasource(datasource)
    .instruction(MeteoraDlmmDecoder, SwapLogger)
    .build()?;

See also. run, docs.rs.

run#

Runs the pipeline until it stops.

pub async fn run(&mut self) -> CarbonResult<()>

Details. In order, run:

1. Logs starting pipeline. num_datasources: … at info level.
2. Calls initialize on each exporter.
3. Creates the update channel (capacity channel_buffer_size) and spawns one Tokio task per datasource, each calling consume(id, sender, token).
4. Processes updates one at a time, in arrival order. For each update, it runs every matching pipe in registration order: account pipes for Account, deletion pipes for AccountDeletion, block pipes for BlockDetails. For Transaction, it runs all instruction pipes over the flattened instruction list, then all transaction pipes.
5. Stops when one of the stop conditions happens, then calls export with a final snapshot and shutdown on each exporter.

Processing is sequential: a slow processor delays every other pipe and fills the channel. Use tokio::spawn inside a processor to move slow work off the loop.

Errors. When a processor or decoder step returns Err, the pipeline logs error processing update (…): … at error level, increments carbon_updates_failed_total and moves on to the next update. The remaining pipes for that same update are skipped. When a datasource's consume returns Err, the pipeline logs error consuming datasource: … and keeps running with the other datasources. run itself returns Err only when an exporter's initialize, export or shutdown fails. A pipeline whose only datasource failed still returns Ok(()), so your process exits with status 0. See Errors.

Example.

Pipeline::builder()
    .datasource(datasource)
    .instruction(MeteoraDlmmDecoder, SwapLogger)
    .build()?
    .run()
    .await?;

See also. Architecture, Error handling.

Stop conditions#

Ctrl-C is ignored while an update is processing#

run listens for Ctrl-C with a fresh tokio::signal::ctrl_c() future on each loop iteration, and it installs Tokio's signal handler, which replaces the default "terminate on SIGINT" behaviour. A signal that arrives while a processor is running is lost. So a pipeline that is always busy can ignore Ctrl-C entirely, until you send SIGTERM or SIGKILL. Listen for Ctrl-C yourself and cancel a token, as in the example, so the signal is never missed.

ShutdownStrategy#

What run does when it receives Ctrl-C.

#[derive(Default, PartialEq, Debug)]
pub enum ShutdownStrategy {
    Immediate,
    #[default]
    ProcessPending,
}

Details.

Neither variant applies when you cancel your own token: that path always stops without draining. To process everything that was fetched, let the datasources finish on their own (the channel then drains completely), or stop your datasource from producing and wait for the carbon_updates_queued gauge to reach 0 before you cancel.

Example.

Pipeline::builder().shutdown_strategy(ShutdownStrategy::Immediate)

See also. Shutdown, docs.rs.

DEFAULT_CHANNEL_BUFFER_SIZE#

The default channel capacity.

pub const DEFAULT_CHANNEL_BUFFER_SIZE: usize = 1_000;

Details. Used by PipelineBuilder::default(). Override it with channel_buffer_size.

Example.

let size = DEFAULT_CHANNEL_BUFFER_SIZE * 10;
Pipeline::builder().channel_buffer_size(size)

See also. docs.rs.

Pipeline fields#

The built pipeline. Its fields are public.

pub struct Pipeline {
    pub datasources: Vec<(DatasourceId, Arc<dyn Datasource>)>,
    pub account_pipes: Vec<Box<dyn AccountPipes>>,
    pub account_deletion_pipes: Vec<Box<dyn AccountDeletionPipes>>,
    pub block_details_pipes: Vec<Box<dyn BlockDetailsPipes>>,
    pub instruction_pipes: Vec<Box<dyn for<'a> InstructionPipes<'a>>>,
    pub transaction_pipes: Vec<Box<dyn for<'a> TransactionPipes<'a>>>,
    pub exporters: Vec<Arc<dyn MetricsExporter>>,
    pub datasource_cancellation_token: Option<CancellationToken>,
    pub shutdown_strategy: ShutdownStrategy,
    pub channel_buffer_size: usize,
}

Details. The fields are the builder's values, unchanged. You can read them (for example pipeline.datasources.len()) or adjust channel_buffer_size and shutdown_strategy before calling run. The *Pipes traits are internal plumbing implemented by AccountPipe, InstructionPipe and the others; you don't need to implement them.

Example.

let mut pipeline = Pipeline::builder().datasource(datasource).build()?;
println!("{} datasource(s)", pipeline.datasources.len());
pipeline.run().await?;

See also. docs.rs.