# Shutdown (Carbon 2.0)

> How a pipeline stops, ShutdownStrategy, datasource_cancellation_token, Ctrl-C handling, in-flight and queued updates, and finite datasources.

A pipeline stops in one of three ways: its datasources finish, its cancellation token is cancelled, or the process receives Ctrl-C. This page explains each, what happens to updates that are in flight or queued, and how to stop a pipeline reliably.

## How a pipeline stops

| Trigger                                          | What `run` does                                                                                        | Queued updates       |
| ------------------------------------------------ | ------------------------------------------------------------------------------------------------------ | -------------------- |
| Every datasource finished and dropped its sender | Processes everything left in the channel, exports metrics, shuts the exporters down, returns `Ok(())`. | All processed.       |
| The cancellation token is cancelled              | Finishes the update it's processing, exports metrics, shuts the exporters down, returns `Ok(())`.      | Dropped.             |
| Ctrl-C                                           | Cancels the token, then behaves according to the `ShutdownStrategy`.                                   | Dropped (see below). |

In every case the update being processed when the stop arrives runs to completion: the pipeline never interrupts a processor. When updates are waiting in the channel, the loop may process one or two more before it notices the cancellation, because it picks at random among the events that are ready. `run` doesn't wait for datasource tasks to finish. They see the cancelled token and stop on their own, or are dropped with the Tokio runtime when `main` returns.

## `ShutdownStrategy`

```rust
pub enum ShutdownStrategy {
    Immediate,
    #[default]
    ProcessPending,
}
```

Set it with `.shutdown_strategy(ShutdownStrategy::Immediate)`. The strategy only affects Ctrl-C:

- **`Immediate`**: on Ctrl-C the pipeline cancels the token, exports metrics and returns.
- **`ProcessPending`** (default): on Ctrl-C the pipeline cancels the token, logs `shutting down the pipeline after processing pending updates.` and keeps looping. The intent is to drain the channel, but in 2.0.0 the loop also watches the token it has cancelled and exits as soon as it sees it, so at most a few more updates are processed. Don't rely on `ProcessPending` to drain the queue.

Neither strategy affects a cancellation that comes from your own token: the pipeline returns after the current update.

## Stop with a cancellation token

Pass a `tokio_util::sync::CancellationToken` to the builder and cancel it when you want to stop:

```rust
let shutdown = CancellationToken::new();

let pipeline = Pipeline::builder()
    .datasource(datasource)
    .instruction(decoder, processor)
    .datasource_cancellation_token(shutdown.clone())
    .build()?;
```

The pipeline hands the same token to every datasource, so cancelling it stops the sources and the processing loop together. Cancel it from a signal handler, a timer, an admin endpoint or another part of your program. A cancelled token stays cancelled, so the pipeline sees it at its next iteration whenever you cancel it.

To run for a fixed time, for example in a test:

```rust
let stop = shutdown.clone();
tokio::spawn(async move {
    tokio::time::sleep(Duration::from_secs(30)).await;
    stop.cancel();
});
```

## Ctrl-C

The pipeline listens for Ctrl-C itself, but only while it waits for the next update. In 2.0.0 a Ctrl-C that arrives while a processor is running is lost, and because Tokio has replaced the default signal handler, the process doesn't exit either. A busy pipeline is almost always running a processor, so Ctrl-C can appear to do nothing.

Handle Ctrl-C in a task of your own and cancel the token from there, as the example below does. Your task waits for the signal the whole time, so it never misses it.

## Finite datasources

Some datasources end on their own: `getProgramAccounts` after one fetch, the validator snapshot after its last account, Helius GTFA after the last page, the block crawler at its `end_slot`, and custom datasources that return. When the last finite datasource drops its sender, the pipeline processes every remaining update and returns `Ok(())`. This is the only way to drain the channel completely.

Others never end: Geyser streams, websocket subscriptions, the block crawler without an `end_slot`, and the transaction crawler, which keeps polling for new transactions after it reaches the end of history. A pipeline with any of these runs until you cancel it.

A datasource whose `consume` returns early but leaves spawned tasks holding the sender (as many published ones do) still counts as running until those tasks end.

## Example

The `Ticker` datasource below sends a `BlockDetails` update every 400 ms until it's cancelled. `main` handles Ctrl-C in its own task and cancels the pipeline's token.

```sh
cargo add carbon-core@2.0.0 async-trait tokio-util
cargo add tokio --features full
```

`src/main.rs`:

```rust
use {
    async_trait::async_trait,
    carbon_core::{
        datasource::{BlockDetails, Datasource, DatasourceId, Update, UpdateType},
        error::CarbonResult,
        pipeline::Pipeline,
        processor::Processor,
    },
    std::time::Duration,
    tokio::sync::mpsc::Sender,
    tokio_util::sync::CancellationToken,
};

/// Emits one `BlockDetails` per 400 ms until cancelled.
struct Ticker;

#[async_trait]
impl Datasource for Ticker {
    async fn consume(
        &self,
        id: DatasourceId,
        sender: Sender<(Update, DatasourceId)>,
        cancellation_token: CancellationToken,
    ) -> CarbonResult<()> {
        let mut interval = tokio::time::interval(Duration::from_millis(400));
        for slot in 0.. {
            tokio::select! {
                _ = cancellation_token.cancelled() => break,
                _ = interval.tick() => {
                    let update = Update::BlockDetails(BlockDetails {
                        slot,
                        block_hash: None,
                        previous_block_hash: None,
                        rewards: None,
                        num_reward_partitions: None,
                        block_time: None,
                        block_height: None,
                    });
                    if sender.send((update, id.clone())).await.is_err() {
                        break;
                    }
                }
            }
        }
        Ok(())
    }

    fn update_types(&self) -> Vec<UpdateType> {
        vec![UpdateType::BlockDetails]
    }
}

struct PrintSlot;

impl Processor<BlockDetails> for PrintSlot {
    async fn process(&mut self, block: &BlockDetails) -> CarbonResult<()> {
        println!("slot {}", block.slot);
        Ok(())
    }
}

#[tokio::main]
async fn main() -> CarbonResult<()> {
    let shutdown = CancellationToken::new();

    // Listen for Ctrl-C in a task of its own so the signal is never missed.
    let on_ctrl_c = shutdown.clone();
    tokio::spawn(async move {
        tokio::signal::ctrl_c().await.ok();
        println!("Ctrl-C received, stopping");
        on_ctrl_c.cancel();
    });

    Pipeline::builder()
        .datasource(Ticker)
        .block_details(PrintSlot)
        .datasource_cancellation_token(shutdown)
        .build()?
        .run()
        .await?;

    println!("pipeline stopped");
    Ok(())
}
```

Run it and press Ctrl-C after a second or two:

```sh
cargo run
```

Output:

```text
slot 0
slot 1
slot 2
slot 3
Ctrl-C received, stopping
pipeline stopped
```

## Next steps

- [Error handling](https://carbon.sevenlabs.org/concepts/error-handling.md): What the pipeline does when processors and datasources fail.
- [Architecture](https://carbon.sevenlabs.org/concepts/architecture.md): The full processing loop, step by step.
- [Run in production](https://carbon.sevenlabs.org/guides/production.md): Deploy, monitor and restart pipelines safely.

---

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