# Pipeline builder (Carbon 2.0) > Every PipelineBuilder method, Pipeline::run, ShutdownStrategy and the pipeline defaults in Carbon 2.0.0. `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](https://carbon.sevenlabs.org/concepts/architecture.md). Source: [`crates/core/src/pipeline.rs`](https://github.com/sevenlabs-hq/carbon/blob/v2.0.0/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. `Cargo.toml`: ```toml [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" ``` `src/main.rs`: ```rust 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> 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> = 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 } ``` ```sh 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: Output: ```text [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 | Setting | Default | Change with | | ---------------------------------------------------------- | ---------------------------------------------------------------------------------------------------- | ----------------------------------------------------------------- | | Channel capacity between datasources and the pipeline loop | `1_000` updates (`DEFAULT_CHANNEL_BUFFER_SIZE`) | [`channel_buffer_size`](#channel_buffer_size) | | Shutdown strategy | `ShutdownStrategy::ProcessPending` | [`shutdown_strategy`](#shutdown_strategy) | | Cancellation token | None. `run` creates a private token, so only Ctrl-C or the end of all datasources stops the pipeline | [`datasource_cancellation_token`](#datasource_cancellation_token) | | Datasource ID | A random UUID v4 per datasource (`DatasourceId::new_unique()`) | [`datasource_with_id`](#datasource_with_id) | | Filters | None. Every update reaches every pipe of the matching kind | the `*_with_filters` methods | | Metrics exporters | None | [`metrics`](#metrics) | ## `Pipeline::builder` Returns an empty `PipelineBuilder` with the [defaults](#defaults). ```rust 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.** ```rust let builder = Pipeline::builder(); ``` **See also.** [`build`](#build), [docs.rs](https://docs.rs/carbon-core/2.0.0/carbon_core/pipeline/struct.Pipeline.html#method.builder). ## `PipelineBuilder::new` Same as [`Pipeline::builder`](#pipelinebuilder). ```rust 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.** ```rust let builder = PipelineBuilder::new(); ``` **See also.** [docs.rs](https://docs.rs/carbon-core/2.0.0/carbon_core/pipeline/struct.PipelineBuilder.html). ## `datasource` Adds a datasource with a random ID. ```rust 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`](#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.** ```rust Pipeline::builder().datasource(datasource) ``` **See also.** [`datasource_with_id`](#datasource_with_id), [`Datasource`](https://carbon.sevenlabs.org/reference/traits.md#datasource), [Datasources](https://carbon.sevenlabs.org/concepts/datasources.md), [docs.rs](https://docs.rs/carbon-core/2.0.0/carbon_core/pipeline/struct.PipelineBuilder.html#method.datasource). ## `datasource_with_id` Adds a datasource with an ID you choose. ```rust 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.** ```rust let crawler_id = DatasourceId::new_named("dlmm-crawler"); Pipeline::builder().datasource_with_id(datasource, crawler_id.clone()) ``` **See also.** [`DatasourceId`](https://carbon.sevenlabs.org/reference/updates.md#datasourceid), [`DatasourceFilter`](https://carbon.sevenlabs.org/reference/filters.md#datasourcefilter), [Multiple datasources](https://carbon.sevenlabs.org/guides/multiple-datasources.md). ## `account` Adds an account pipe: a decoder and the processor that receives what it decodes. ```rust pub fn account( self, decoder: impl for<'a> AccountDecoder<'a, AccountType = T> + Send + Sync + 'static, processor: P, ) -> Self where T: Send + Sync + 'static, P: for<'a> Processor> + Send + Sync + 'static, ``` **Details.** For every `Update::Account`, the pipe calls `decoder.decode_account(&account)`. When it returns `Some`, the processor receives an [`AccountProcessorInputType`](https://carbon.sevenlabs.org/reference/processor-inputs.md#accountprocessorinputtype). 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`](#account_deletions) for them. **Example.** ```rust Pipeline::builder() .datasource(GpaDatasource::new(rpc_url, PROGRAM_ID)) .account(MeteoraDlmmDecoder, PoolLogger) ``` **See also.** [`account_with_filters`](#account_with_filters), [`AccountDecoder`](https://carbon.sevenlabs.org/reference/traits.md#accountdecoder), [Load account state](https://carbon.sevenlabs.org/guides/load-account-state.md). ## `account_with_filters` Adds an account pipe that only runs for updates every filter accepts. ```rust pub fn account_with_filters( self, decoder: impl for<'a> AccountDecoder<'a, AccountType = T> + Send + Sync + 'static, processor: P, filters: Vec>, ) -> Self ``` **Details.** Same bounds as [`account`](#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.** ```rust Pipeline::builder().account_with_filters( MeteoraDlmmDecoder, PoolLogger, vec![Box::new(SlotRangeFilter::from(400_000_000, None))], ) ``` **See also.** [Filters reference](https://carbon.sevenlabs.org/reference/filters.md), [Filters](https://carbon.sevenlabs.org/concepts/filters.md). ## `account_deletions` Adds a processor for account closures. ```rust pub fn account_deletions

(self, processor: P) -> Self where P: Processor + 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.** ```rust Pipeline::builder().account_deletions(DeletionLogger) ``` **See also.** [`AccountDeletion`](https://carbon.sevenlabs.org/reference/updates.md#accountdeletion), [`account_deletions_with_filters`](#account_deletions_with_filters). ## `account_deletions_with_filters` Adds a deletion processor behind filters. ```rust pub fn account_deletions_with_filters

( self, processor: P, filters: Vec>, ) -> Self where P: Processor + Send + Sync + 'static, ``` **Details.** Calls `Filter::filter_account_deletion`. `DeduplicationFilter` does not implement this hook, so it accepts every deletion. **Example.** ```rust Pipeline::builder().account_deletions_with_filters( DeletionLogger, vec![Box::new(DatasourceFilter::new(DatasourceId::new_named("geyser")))], ) ``` **See also.** [Filters reference](https://carbon.sevenlabs.org/reference/filters.md#filter). ## `block_details` Adds a processor for block-level metadata. ```rust pub fn block_details

(self, processor: P) -> Self where P: Processor + 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.** ```rust Pipeline::builder().block_details(BlockLogger) ``` **See also.** [`BlockDetails`](https://carbon.sevenlabs.org/reference/updates.md#blockdetails), [`block_details_with_filters`](#block_details_with_filters). ## `block_details_with_filters` Adds a block-details processor behind filters. ```rust pub fn block_details_with_filters

( self, processor: P, filters: Vec>, ) -> Self where P: Processor + Send + Sync + 'static, ``` **Details.** Calls `Filter::filter_block_details`. **Example.** ```rust Pipeline::builder().block_details_with_filters( BlockLogger, vec![Box::new(SlotRangeFilter::between(400_000_000, None, 400_100_000, None))], ) ``` **See also.** [Filters reference](https://carbon.sevenlabs.org/reference/filters.md#slotrangefilter). ## `instruction` Adds an instruction pipe: a decoder and the processor that receives each instruction it decodes. ```rust pub fn instruction( self, decoder: impl for<'a> InstructionDecoder<'a, InstructionType = T> + Send + Sync + 'static, processor: P, ) -> Self where T: Send + Sync + 'static, P: for<'a> Processor> + 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`](https://carbon.sevenlabs.org/reference/processor-inputs.md#instructionprocessorinputtype), including the instruction's own children in `nested_instructions`. **Example.** ```rust Pipeline::builder() .datasource(datasource) .instruction(MeteoraDlmmDecoder, SwapLogger) ``` **See also.** [`instruction_with_filters`](#instruction_with_filters), [Instructions](https://carbon.sevenlabs.org/concepts/instructions.md), [`InstructionDecoder`](https://carbon.sevenlabs.org/reference/traits.md#instructiondecoder). ## `instruction_with_filters` Adds an instruction pipe behind filters. ```rust pub fn instruction_with_filters( self, decoder: impl for<'a> InstructionDecoder<'a, InstructionType = T> + Send + Sync + 'static, processor: P, filters: Vec>, ) -> Self ``` **Details.** Same bounds as [`instruction`](#instruction). Calls `Filter::filter_instruction` once per instruction, before decoding, so filters also see instructions of other programs. **Example.** ```rust let filters: Vec> = 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`](https://carbon.sevenlabs.org/reference/filters.md#deduplicationfilter). ## `transaction` Adds a transaction pipe: one processor call per transaction with every instruction the collection `T` recognizes. ```rust pub fn transaction(self, processor: P) -> Self where T: InstructionDecoderCollection + 'static, P: for<'a> Processor> + 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.** ```rust Pipeline::builder() .datasource(datasource) .transaction::(TxLogger) ``` **See also.** [`transaction_with_filters`](#transaction_with_filters), [`InstructionDecoderCollection`](https://carbon.sevenlabs.org/reference/traits.md#instructiondecodercollection), [Transactions](https://carbon.sevenlabs.org/concepts/transactions.md). ## `transaction_with_filters` Adds a transaction pipe behind filters. ```rust pub fn transaction_with_filters( self, processor: P, filters: Vec>, ) -> Self where T: InstructionDecoderCollection + 'static, P: for<'a> Processor> + Send + Sync + 'static, ``` **Details.** Calls `Filter::filter_transaction`. `DeduplicationFilter` does not implement this hook, so it does not deduplicate transaction pipes. **Example.** ```rust Pipeline::builder().transaction_with_filters::( TxLogger, vec![Box::new(SlotRangeFilter::from(400_000_000, Some(0)))], ) ``` **See also.** [Filters reference](https://carbon.sevenlabs.org/reference/filters.md). ## `metrics` Adds a metrics exporter. ```rust pub fn metrics(self, exporter: Arc) -> 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.** ```rust Pipeline::builder().metrics(Arc::new(LogMetrics::new())) ``` **See also.** [Metrics reference](https://carbon.sevenlabs.org/reference/metrics.md), [`MetricsExporter`](https://carbon.sevenlabs.org/reference/traits.md#metricsexporter). ## `datasource_cancellation_token` Gives the pipeline a `CancellationToken` you control. ```rust 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`](#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](#ctrl-c-is-ignored-while-an-update-is-processing). **Example.** ```rust 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](https://carbon.sevenlabs.org/concepts/shutdown.md), [`run`](#run). ## `shutdown_strategy` Sets what happens when the pipeline receives Ctrl-C. ```rust 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`](#shutdownstrategy) for the exact behaviour of each variant. **Example.** ```rust Pipeline::builder().shutdown_strategy(ShutdownStrategy::Immediate) ``` **See also.** [`ShutdownStrategy`](#shutdownstrategy), [Shutdown](https://carbon.sevenlabs.org/concepts/shutdown.md). ## `channel_buffer_size` Sets the capacity of the channel between datasources and the pipeline loop. ```rust 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.** ```rust Pipeline::builder().channel_buffer_size(10_000) ``` **See also.** [`DEFAULT_CHANNEL_BUFFER_SIZE`](#default_channel_buffer_size), [Production](https://carbon.sevenlabs.org/guides/production.md). ## `build` Finishes the builder and returns the `Pipeline`. ```rust pub fn build(self) -> CarbonResult ``` **Details.** Registers the [pipeline metrics](https://carbon.sevenlabs.org/reference/metrics.md) 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.** ```rust let mut pipeline = Pipeline::builder() .datasource(datasource) .instruction(MeteoraDlmmDecoder, SwapLogger) .build()?; ``` **See also.** [`run`](#run), [docs.rs](https://docs.rs/carbon-core/2.0.0/carbon_core/pipeline/struct.PipelineBuilder.html#method.build). ## `run` Runs the pipeline until it stops. ```rust 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](#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](https://carbon.sevenlabs.org/reference/errors.md). **Example.** ```rust Pipeline::builder() .datasource(datasource) .instruction(MeteoraDlmmDecoder, SwapLogger) .build()? .run() .await?; ``` **See also.** [Architecture](https://carbon.sevenlabs.org/concepts/architecture.md), [Error handling](https://carbon.sevenlabs.org/concepts/error-handling.md). ### Stop conditions | Event | What `run` does | Updates still in the channel | | ------------------------------------------------------------------------------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------------- | ---------------------------- | | Every datasource task returns (finite sources such as GPA or a bounded crawler) | Processes everything left in the channel, logs `update_receiver closed, shutting down.`, exports, returns `Ok(())` | Processed | | Your cancellation token is cancelled | Finishes the current update, exports, returns `Ok(())` | Dropped | | Ctrl-C with `ShutdownStrategy::Immediate` | Cancels the token, logs `shutting down the pipeline immediately.`, exports, returns `Ok(())` | Dropped | | Ctrl-C with `ShutdownStrategy::ProcessPending` | Cancels the token, logs `shutting down the pipeline after processing pending updates.`, then stops at the next loop iteration where the cancelled token is selected | Mostly dropped (see below) | ### 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](#example), so the signal is never missed. ## `ShutdownStrategy` What `run` does when it receives Ctrl-C. ```rust #[derive(Default, PartialEq, Debug)] pub enum ShutdownStrategy { Immediate, #[default] ProcessPending, } ``` **Details.** | Variant | Behaviour in 2.0.0 | | -------------------------- | ----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | | `Immediate` | Cancels the datasources' token, exports metrics, shuts down exporters and returns. Updates already in the channel are dropped. | | `ProcessPending` (default) | Cancels the token and keeps looping. On the next iteration the cancelled token and the channel are both ready, and `tokio::select!` picks between them at random, so the pipeline usually exits after zero or a few more updates. It does **not** reliably drain the channel. | 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.** ```rust Pipeline::builder().shutdown_strategy(ShutdownStrategy::Immediate) ``` **See also.** [Shutdown](https://carbon.sevenlabs.org/concepts/shutdown.md), [docs.rs](https://docs.rs/carbon-core/2.0.0/carbon_core/pipeline/enum.ShutdownStrategy.html). ## `DEFAULT_CHANNEL_BUFFER_SIZE` The default channel capacity. ```rust pub const DEFAULT_CHANNEL_BUFFER_SIZE: usize = 1_000; ``` **Details.** Used by `PipelineBuilder::default()`. Override it with [`channel_buffer_size`](#channel_buffer_size). **Example.** ```rust let size = DEFAULT_CHANNEL_BUFFER_SIZE * 10; Pipeline::builder().channel_buffer_size(size) ``` **See also.** [docs.rs](https://docs.rs/carbon-core/2.0.0/carbon_core/pipeline/constant.DEFAULT_CHANNEL_BUFFER_SIZE.html). ## `Pipeline` fields The built pipeline. Its fields are public. ```rust pub struct Pipeline { pub datasources: Vec<(DatasourceId, Arc)>, pub account_pipes: Vec>, pub account_deletion_pipes: Vec>, pub block_details_pipes: Vec>, pub instruction_pipes: Vec InstructionPipes<'a>>>, pub transaction_pipes: Vec TransactionPipes<'a>>>, pub exporters: Vec>, pub datasource_cancellation_token: Option, 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.** ```rust let mut pipeline = Pipeline::builder().datasource(datasource).build()?; println!("{} datasource(s)", pipeline.datasources.len()); pipeline.run().await?; ``` **See also.** [docs.rs](https://docs.rs/carbon-core/2.0.0/carbon_core/pipeline/struct.Pipeline.html). --- Carbon 2.0.0 documentation · Web: https://carbon.sevenlabs.org/reference/pipeline-builder · Last updated 2026-11-09 · Index of all pages: https://carbon.sevenlabs.org/llms.txt