Jito Shredstream gRPC (legacy)

JitoShredstreamGrpcClient subscribes to the SubscribeEntries stream of a Jito shredstream-proxy, deserializes the entries the leader is producing, and emits every transaction in them. Transactions arrive before they are executed, so they have no status, logs, balances or inner instructions. Use it only for latency-critical signals (such as reacting to a swap before the block lands), never for indexing state.

Quick example#

This Carbon 1.0 program prints every Orca Whirlpool swapV2 the proxy sees. It compiles as shown; to run it, set SHREDSTREAM_URL to the proxy's gRPC address.

cargo new shredstream && cd shredstream
cargo add carbon-core@1.0.0 carbon-log-metrics@1.0.0 carbon-orca-whirlpool-decoder@1.0.0 carbon-jito-shredstream-grpc-datasource@1.0.0
cargo add tokio@1 --features full
cargo add env_logger@0.11 log@0.4

use {
    carbon_core::{
        error::CarbonResult, instruction::InstructionProcessorInputType, pipeline::Pipeline,
        processor::Processor,
    },
    carbon_jito_shredstream_grpc_datasource::JitoShredstreamGrpcClient,
    carbon_log_metrics::LogMetrics,
    carbon_orca_whirlpool_decoder::{instructions::OrcaWhirlpoolInstruction, OrcaWhirlpoolDecoder},
    std::{env, sync::Arc},
};

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

    // The gRPC address of your shredstream-proxy, e.g. http://127.0.0.1:9999
    let datasource = JitoShredstreamGrpcClient::new(
        env::var("SHREDSTREAM_URL").expect("SHREDSTREAM_URL must be set"),
    );

    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;
        // No execution metadata: the transaction may still fail on chain.
        if let OrcaWhirlpoolInstruction::SwapV2 { data, accounts, .. } = input.decoded_instruction {
            println!("slot {} swapV2 (unconfirmed) pool={} amount={}", tx.slot, accounts.whirlpool, data.amount);
        }
        Ok(())
    }
}

SHREDSTREAM_URL=http://127.0.0.1:9999 RUST_LOG=info cargo run

With no proxy listening, the pipeline stops at once:

[ERROR carbon_core::pipeline] error consuming datasource: FailedToConsumeDatasource("transport error")
[INFO  carbon_core::pipeline] update_receiver closed, shutting down.

Configuration#

JitoShredstreamGrpcClient::new#

pub fn new(endpoint: String) -> Self

endpoint is the proxy's gRPC URL, for example http://127.0.0.1:9999. There are no other options.

Behaviour#

• Connecting. consume connects first and returns FailedToConsumeDatasource("transport error") if the proxy isn't reachable.
• Entries. Each message holds a slot and bincode-encoded entries. Entries already seen (by entry hash, in a cache of about 4,096) are skipped and counted as duplicates. Messages are processed concurrently, so updates are not strictly ordered.
• Updates. One Transaction per transaction in each entry, with slot from the message and meta set to TransactionStatusMeta { status: Ok(()), ..Default::default() }: no fee, balances, logs, inner instructions or loaded addresses. is_vote is always false and votes are included. index and block_hash are None. block_time is the local receive time in milliseconds since the Unix epoch, not seconds.
• What decodes. Top-level instructions decode normally. CPIs, log events and anything that reads token balances or loaded addresses don't, because the metadata is empty. With no loaded addresses, accounts that come from address lookup tables are missing from the instruction, so decoders usually return nothing for those instructions.
• Backpressure. Updates use try_send. When the pipeline channel is full, the rest of that message's transactions are dropped with Failed to send transaction update ….
• Reconnects. None. If the subscription fails or the stream ends, the task logs Failed to subscribe or Grpc stream error and stops; a pipeline with only this datasource then shuts down. Run it under a supervisor that restarts the process.
• Metrics. Counters jito_shredstream_grpc_entry_updates_received_total, jito_shredstream_grpc_duplicate_entries_total; histogram jito_shredstream_grpc_entry_process_time_nanoseconds.

Troubleshooting#

FailedToConsumeDatasource("transport error")#

The proxy isn't running, or its gRPC service is disabled or on another port. Start shredstream-proxy with its gRPC service enabled and point SHREDSTREAM_URL at that port.

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.

Decoded swaps never appear in the chain#

Expected for some: shredstream shows transactions before execution, and some fail. Confirm with a Yellowstone or RPC datasource if you need final results.

block_time values look 1000 times too large#

They are milliseconds of local receive time. Divide by 1000 for Unix seconds.

See also#

• Yellowstone gRPC for executed transactions with metadata
• Migrate to v2
• Datasources concept