Shutdown

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#

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#

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:

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:

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.

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

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:

cargo run

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

Next steps#