# Jetstreamer (legacy) (Carbon 2.0) > Replay historical slots or whole epochs at high speed from the Old Faithful archive with Jetstreamer. Carbon 1.x only. > **Danger: Carbon 1.x only.** `carbon-jetstreamer-datasource` is not part of Carbon 2.0. The latest release is **1.0.0**, built on `carbon-core` **1.0.0** and the Solana 3.x crates, because Jetstreamer doesn't support the Solana 4 stack yet. It does not work in a Carbon 2.0 pipeline: use it in a Carbon 1.0 application, with 1.0.0 decoders and `carbon-log-metrics` 1.0.0. The source in the `v2.0.0` tag is identical to the published 1.0.0 crate but is excluded from the 2.0 workspace. `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. | | | | -------- | ---------------------------------------------------------------------------------------------------------------------------------------- | | Crate | `carbon-jetstreamer-datasource` 1.0.0 (**Carbon 1.x**: `carbon-core` 1.0.0) | | Install | `cargo add carbon-jetstreamer-datasource@1.0.0` | | Emits | `Transaction`, and `BlockDetails` when `include_blocks` is set. `update_types()` declares only `Transaction`. | | Mode | Historical. Finishes at the end of the range. | | Provider | None: Old Faithful's public archive (`https://files.old-faithful.net/`) by default. | | API docs | [docs.rs](https://docs.rs/carbon-jetstreamer-datasource/1.0.0/carbon_jetstreamer_datasource/) | | Source | [`datasources/jetstreamer-datasource`](https://github.com/sevenlabs-hq/carbon/blob/v2.0.0/datasources/jetstreamer-datasource/src/lib.rs) | ## 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`). ```sh 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 ``` `src/main.rs`: ```rust 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> 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(()) } } ``` ```sh RUST_LOG=info cargo run ``` The 100 slots took about 7 seconds. Four threads each replay 25 slots, so output from different threads interleaves: Output: ```text 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` ```rust pub fn new( range: JetstreamerRange, filter: JetstreamerFilter, threads: u64, tracking_interval_slots: Option, archive_url: Option, network: Option, ) -> Self pub fn new_with_old_faithful_mainnet( range: JetstreamerRange, filter: JetstreamerFilter, threads: u64, tracking_interval_slots: Option, ) -> Self // archive_url and network = None ``` | Parameter | Type | Meaning | | ------------------------- | ------------------- | --------------------------------------------------------------------------------------------------------- | | `range` | `JetstreamerRange` | Slots or epoch to replay. | | `filter` | `JetstreamerFilter` | Which updates to emit. | | `threads` | `u64` | Firehose threads; the range is split evenly between them. Must be at least 1. | | `tracking_interval_slots` | `Option` | When set, the firehose reports progress every this many slots into the `jetstreamer_internal_*` gauges. | | `archive_url` | `Option` | Sets `JETSTREAMER_COMPACT_INDEX_BASE_URL` for the process. `None` uses `https://files.old-faithful.net/`. | | `network` | `Option` | Sets `JETSTREAMER_NETWORK` for the process (default `mainnet`). | ### `JetstreamerRange` (`carbon_jetstreamer_datasource::range`) | Constructor | Meaning | | ---------------------------------------------- | ---------------------------------------- | | `JetstreamerRange::new_with_slots(start, end)` | `start..end`: the end slot is exclusive. | | `JetstreamerRange::new_with_epoch(epoch)` | Every slot of the epoch. | ### `JetstreamerFilter` and `TransactionFilter` (`carbon_jetstreamer_datasource::filter`) | Field | Type | Default | Meaning | | ---------------------- | ------------------------ | ------- | ------------------------------------------------------------------------------- | | `include_transactions` | `bool` | `false` | Emit `Transaction` updates. Set it, or you receive no transactions. | | `include_blocks` | `bool` | `false` | Emit a `BlockDetails` update per block. | | `transaction_filters` | `Vec` | empty | A transaction passes if it matches any filter. Empty: every transaction passes. | `TransactionFilter::new(vote, failed, account_include, account_exclude, account_required)`: | Field | Type | Meaning | | ------------------ | ----------------- | ------------------------------------------------------------------ | | `vote` | `Option` | Match only votes (`Some(true)`) or only non-votes (`Some(false)`). | | `failed` | `Option` | Match only failed or only successful transactions. | | `account_include` | `HashSet` | At least one of these accounts (when non-empty). | | `account_exclude` | `HashSet` | None of these accounts. | | `account_required` | `HashSet` | All of these accounts. | 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](https://carbon.sevenlabs.org/guides/backfill-transactions.md) - [RPC block crawler](https://carbon.sevenlabs.org/datasources/rpc-block-crawler.md): the Carbon 2.0 alternative for slot ranges. - [Migrate to v2](https://carbon.sevenlabs.org/guides/migrate-to-v2.md) - [Datasources concept](https://carbon.sevenlabs.org/concepts/datasources.md) --- Carbon 2.0.0 documentation · Web: https://carbon.sevenlabs.org/datasources/jetstreamer · Last updated 2026-11-09 · Index of all pages: https://carbon.sevenlabs.org/llms.txt