Trait-based framework for building structured observers on top of Jetstreamer's firehose.

Overview

Plugins let you react to every block, transaction, reward, entry, and stats update emitted by jetstreamer_firehose. Combined with the JetstreamerRunner, they provide a high-throughput analytics pipeline capable of exceeding 2.7 million transactions per second on the right hardware. All events originate from Old Faithful's CAR archive and are streamed over the network into your local runner.

The framework offers:

A [Plugin] trait with async hook points for each data type.
[PluginRunner] for coordinating multiple plugins with shared ClickHouse connections (used internally by JetstreamerRunner).
Built-in plugins under [plugins] that demonstrate common batching strategies and metrics.
See JetstreamerRunner in the jetstreamer crate for the easiest way to run plugins.

ClickHouse Integration

Jetstreamer plugins are typically paired with ClickHouse for persistence. Runner instances honor the following environment variables:

JETSTREAMER_CLICKHOUSE_DSN (default http://localhost:8123): HTTP(S) DSN handed to every plugin that requests a database handle.
JETSTREAMER_CLICKHOUSE_MODE (default auto): toggles the bundled ClickHouse helper. Set to remote to opt out of spawning the helper while still writing to a cluster, local to always spawn, or off to disable ClickHouse entirely.

When the mode is auto, Jetstreamer inspects the DSN at runtime and only launches the embedded helper for local endpoints, enabling native clustering workflows out of the box.

Write Durability

Writes issued by the runner and the bundled plugins are never silently dropped. Inserts use async_insert with wait_for_async_insert=1 (an acknowledgment means durably flushed), failures are retried with exponential backoff for up to 10 minutes, in-flight write tasks are tracked and drained at shutdown (so runtime teardown never cancels a batch mid-delivery), and a write that is still failing after the horizon aborts the run with a message that includes the exact command to resume from the lowest unprocessed slot. Retries provide at-least-once delivery: every bundled table is a ReplacingMergeTree keyed on its logical identity, so replayed batches deduplicate on merge — query with FINAL (or tolerate transient duplicates) when reading while ingestion is active.

Batching ClickHouse Writes

ClickHouse (and any sinks you invoke inside hook handlers) can apply backpressure on large numbers of tiny inserts. Plugins should buffer work locally and flush in batches on a cadence that matches their workload. The default [PluginRunner] configuration triggers stats pulses every 100 slots, which offers a reasonable heartbeat without thrashing the database. The bundled [plugins::program_tracking::ProgramTrackingPlugin] mirrors this approach by accumulating ProgramEvent rows per worker thread and issuing a single batch insert every 1,000 slots. Adopting a similar strategy keeps long-running replays responsive even under peak throughput.

Ordering Guarantees

Also note that because Jetstreamer spawns parallel threads that process different subranges of the overall slot range at the same time, while each thread sees a purely sequential view of transactions, downstream services such as databases that consume this data will see writes in a fairly arbitrary order, so you should design your database tables and shared data structures accordingly.

Examples

Defining a Plugin

use std::sync::Arc;
use clickhouse::Client;
use futures_util::FutureExt;
use jetstreamer_firehose::firehose::TransactionData;
use jetstreamer_plugin::{Plugin, PluginFuture};

struct CountingPlugin;

impl Plugin for CountingPlugin {
    fn name(&self) -> &'static str { "counting" }

    fn on_transaction<'a>(
        &'a self,
        _thread_id: usize,
        _db: Option<Arc<Client>>,
        transaction: &'a TransactionData,
    ) -> PluginFuture<'a> {
        async move {
            println!("saw tx {} in slot {}", transaction.signature, transaction.slot);
            Ok(())
        }
        .boxed()
    }
}
# let _plugin = CountingPlugin;

Running Plugins with `PluginRunner`

use std::sync::Arc;
use jetstreamer_firehose::epochs;
use jetstreamer_plugin::{Plugin, PluginRunner};

struct LoggingPlugin;

impl Plugin for LoggingPlugin {
    fn name(&self) -> &'static str { "logging" }
}

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let mut runner = PluginRunner::new("http://localhost:8123", 1, false, false, None);
    runner.register(Box::new(LoggingPlugin));
    let runner = Arc::new(runner);

    let (start, _) = epochs::epoch_to_slot_range(800);
    let (_, end_inclusive) = epochs::epoch_to_slot_range(805);
    runner
        .clone()
        .run(start..(end_inclusive + 1), false)
        .await?;
    Ok(())
}

jetstreamer-plugin 0.7.0