Jetstreamer (legacy)

JetstreamerDatasource runs the Jetstreamer firehose over a slot range or an epoch, reading blocks from the public Old Faithful archive over HTTP with several threads in parallel, and emits transactions (and optionally block details). Use it for fast historical backfills over long ranges without an RPC provider.

Quick example#

This Carbon 1.0 program replays 100 mainnet slots, keeps successful non-vote transactions that touch the Orca Whirlpool program, and prints each swapV2. It needs no API key. Building it pulls in RocksDB, so install a C++ compiler and libclang first (sudo apt install build-essential clang libclang-dev).

cargo new jetstreamer && cd jetstreamer
cargo add carbon-core@1.0.0 carbon-log-metrics@1.0.0 carbon-orca-whirlpool-decoder@1.0.0 carbon-jetstreamer-datasource@1.0.0
cargo add tokio@1 --features full
cargo add env_logger@0.11 log@0.4
# jetstreamer-firehose 0.2.0 breaks with rseek 0.4 (reqwest 0.13); pin the older release
cargo update -p rseek --precise 0.3.0

use {
    carbon_core::{
        error::CarbonResult, instruction::InstructionProcessorInputType, pipeline::Pipeline,
        processor::Processor,
    },
    carbon_jetstreamer_datasource::{
        filter::{JetstreamerFilter, TransactionFilter},
        range::JetstreamerRange,
        JetstreamerDatasource,
    },
    carbon_log_metrics::LogMetrics,
    carbon_orca_whirlpool_decoder::{
        instructions::OrcaWhirlpoolInstruction, OrcaWhirlpoolDecoder, PROGRAM_ID,
    },
    std::{collections::HashSet, sync::Arc},
};

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

    let filter = JetstreamerFilter {
        include_transactions: true,
        include_blocks: false,
        transaction_filters: vec![TransactionFilter::new(
            Some(false),                    // vote
            Some(false),                    // failed
            HashSet::from([PROGRAM_ID]),    // account_include
            HashSet::new(),                 // account_exclude
            HashSet::new(),                 // account_required
        )],
    };

    let datasource = JetstreamerDatasource::new_with_old_faithful_mainnet(
        JetstreamerRange::new_with_slots(376_000_000, 376_000_100), // end slot is exclusive
        filter,
        4,    // firehose threads
        None, // stats tracking interval (slots)
    );

    Pipeline::builder()
        .datasource(datasource)
        .metrics(Arc::new(LogMetrics::new()))
        .instruction(OrcaWhirlpoolDecoder, SwapLogger)
        .build()?
        .run()
        .await
}

struct SwapLogger;

impl Processor<InstructionProcessorInputType<'_, OrcaWhirlpoolInstruction>> for SwapLogger {
    async fn process(
        &mut self,
        input: &InstructionProcessorInputType<'_, OrcaWhirlpoolInstruction>,
    ) -> CarbonResult<()> {
        let tx = &input.metadata.transaction_metadata;
        if let OrcaWhirlpoolInstruction::SwapV2 { data, accounts, .. } = input.decoded_instruction {
            println!("slot {} swapV2 pool={} amount={}", tx.slot, accounts.whirlpool, data.amount);
        }
        Ok(())
    }
}

RUST_LOG=info cargo run

The 100 slots took about 7 seconds. Four threads each replay 25 slots, so output from different threads interleaves:

slot 376000050 swapV2 pool=2nJjE2ba3iGtefN4UNM1KN5FdYCwssU4Bzhc8URceqh6 amount=169535616
slot 376000001 swapV2 pool=44W73kGYQgXCTNkGxUmHv8DDBPCxojBcX49uuKmbFc9U amount=355597021341
slot 376000001 swapV2 pool=VPAZrfafQQerQjx4ZFHmKTXDXh4SStCdoFs4J3GfYjT amount=1617157367
slot 376000075 swapV2 pool=44W73kGYQgXCTNkGxUmHv8DDBPCxojBcX49uuKmbFc9U amount=364530054
...
[INFO  jetstreamer::firehose] 🚒 firehose finished successfully.
[INFO  carbon_core::pipeline] update_receiver closed, shutting down.
[INFO  carbon_log_metrics]   jetstreamer_transactions_sent_total: 227
[INFO  carbon_log_metrics]   jetstreamer_transactions_filtered_out_total: 118827

Configuration#

JetstreamerDatasource#

pub fn new(
    range: JetstreamerRange,
    filter: JetstreamerFilter,
    threads: u64,
    tracking_interval_slots: Option<u64>,
    archive_url: Option<String>,
    network: Option<String>,
) -> Self

pub fn new_with_old_faithful_mainnet(
    range: JetstreamerRange,
    filter: JetstreamerFilter,
    threads: u64,
    tracking_interval_slots: Option<u64>,
) -> Self // archive_url and network = None

JetstreamerRange (carbon_jetstreamer_datasource::range)#

JetstreamerFilter and TransactionFilter (carbon_jetstreamer_datasource::filter)#

TransactionFilter::new(vote, failed, account_include, account_exclude, account_required):

Accounts are matched against static keys and lookup-table addresses.

Behaviour#

• Fetching. The firehose reads the Old Faithful compact indexes for the epoch, then streams blocks from the archive, one thread per sub-range.
• Ordering. Sequential within a thread, interleaved across threads. Use threads: 1 for slot order.
• Updates. Transactions carry full metadata from the archive, is_vote and index (position in the slot); block_time and block_hash are None. Failed transactions are emitted unless a filter excludes them. BlockDetails carry hash, parent hash, rewards, partitions, block time and height.
• Backpressure. Updates are sent with send().await; nothing is dropped.
• Finishing. consume returns when the firehose finishes the range, and a pipeline with only this datasource shuts down. Firehose errors become FailedToConsumeDatasource.
• Cancellation. The datasource ignores the cancellation token. With the default ShutdownStrategy::ProcessPending, Ctrl-C does not stop the pipeline until the range is done; set .shutdown_strategy(ShutdownStrategy::Immediate) to exit at once.
• Metrics. Counters jetstreamer_blocks_sent_total, jetstreamer_transactions_sent_total, jetstreamer_transactions_filtered_in_total, jetstreamer_transactions_filtered_out_total; gauges jetstreamer_internal_slots_processed, jetstreamer_internal_blocks_processed, jetstreamer_internal_transactions_processed (filled only with tracking_interval_slots).

Troubleshooting#

expected F to return RequestBuilder, but it returns RequestBuilder in jetstreamer-firehose#

rseek 0.4 moved to reqwest 0.13 while jetstreamer-firehose 0.2.0 uses 0.12. Run cargo update -p rseek --precise 0.3.0.

Trait or type mismatch errors between carbon_core versions#

You mixed this crate with carbon-core 2.0.0 or 2.0.0 decoders. Use carbon-core, carbon-log-metrics and decoders at 1.0.0 in the same crate.

Build fails in clang-sys#

Install libclang (sudo apt install clang libclang-dev) or set LIBCLANG_PATH.

No transactions are emitted#

include_transactions defaults to false. Set it to true.

Ctrl-C doesn't stop the pipeline#

See Cancellation above: use ShutdownStrategy::Immediate.

See also#

• Backfill transactions
• RPC block crawler: the Carbon 2.0 alternative for slot ranges.
• Migrate to v2
• Datasources concept