RPC blockSubscribe

RpcBlockSubscribe opens a blockSubscribe WebSocket subscription and emits a block-details update for every block, followed by a transaction update for each successful transaction in it. Use it for live transactions from an RPC node that has block subscriptions enabled, when you don't have a Geyser gRPC endpoint.

Quick example#

This program subscribes to confirmed blocks that mention the Orca Whirlpool program, prints each block and each swapV2 instruction. It compiles as shown; to run it, set RPC_WS_URL to an endpoint that supports blockSubscribe.

cargo new block-subscribe && cd block-subscribe
cargo add carbon-core@2.0.0 carbon-log-metrics@2.0.0 carbon-orca-whirlpool-decoder@2.0.0 carbon-rpc-block-subscribe-datasource@2.0.0
cargo add solana-client@=4.2.2
cargo add tokio@1 --features full
cargo add env_logger@0.11 log@0.4

use {
    carbon_core::{
        datasource::BlockDetails, error::CarbonResult,
        instruction::InstructionProcessorInputType, pipeline::Pipeline, processor::Processor,
    },
    carbon_log_metrics::LogMetrics,
    carbon_orca_whirlpool_decoder::{
        instructions::OrcaWhirlpoolInstruction, OrcaWhirlpoolDecoder, PROGRAM_ID,
    },
    carbon_rpc_block_subscribe_datasource::{Filters, RpcBlockSubscribe},
    solana_client::rpc_config::{
        CommitmentConfig, RpcBlockSubscribeConfig, RpcBlockSubscribeFilter, TransactionDetails,
        UiTransactionEncoding,
    },
    std::{env, sync::Arc},
};

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

    let config = RpcBlockSubscribeConfig {
        commitment: Some(CommitmentConfig::confirmed()),
        encoding: Some(UiTransactionEncoding::Base64), // required: Carbon can't decode JSON payloads
        transaction_details: Some(TransactionDetails::Full),
        show_rewards: Some(false),
        max_supported_transaction_version: Some(1),
    };

    let datasource = RpcBlockSubscribe::new(
        env::var("RPC_WS_URL").expect("RPC_WS_URL must be set"),
        Filters::new(
            RpcBlockSubscribeFilter::MentionsAccountOrProgram(PROGRAM_ID.to_string()),
            Some(config),
        ),
    );

    Pipeline::builder()
        .datasource(datasource)
        .metrics(Arc::new(LogMetrics::new()))
        .instruction(OrcaWhirlpoolDecoder, SwapLogger)
        .block_details(BlockLogger)
        .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(())
    }
}

struct BlockLogger;

impl Processor<BlockDetails> for BlockLogger {
    async fn process(&mut self, block: &BlockDetails) -> CarbonResult<()> {
        println!("block {} time={:?} hash={:?}", block.slot, block.block_time, block.block_hash);
        Ok(())
    }
}

RPC_WS_URL=wss://your-node.example.com RUST_LOG=info cargo run

Against the public endpoint the subscription is refused and retried every few seconds:

[ERROR carbon_rpc_block_subscribe_datasource] Failed to subscribe to block updates: SubscribeFailed { reason: "Method not found (-32601)", message: "{\"jsonrpc\":\"2.0\",\"error\":{\"code\":-32601,\"message\":\"Method not found\"},\"id\":1}" }

Configuration#

Constructors#

pub fn new(rpc_ws_url: String, filters: Filters) -> Self
pub fn with_disconnect_notifier(
    rpc_ws_url: String,
    filters: Filters,
    disconnect_notifier: tokio::sync::mpsc::Sender<DatasourceDisconnection>,
) -> Self
pub fn with_stream_timeout(self, timeout: Duration) -> Self

Filters#

pub const fn new(
    block_filter: RpcBlockSubscribeFilter,
    block_subscribe_config: Option<RpcBlockSubscribeConfig>,
) -> Self

RpcBlockSubscribeFilter (from solana_client::rpc_config) is All or MentionsAccountOrProgram(String), the latter taking a base58 address. RpcBlockSubscribeConfig:

DatasourceDisconnection#

Sent with try_send on the first block after a reconnect: source ("rpc-websocket"), disconnect_time, last_slot_before_disconnect, first_slot_after_reconnect, missed_slots. Use it to trigger a backfill of the gap with RpcBlockCrawler.

Behaviour#

• Updates per block. One BlockDetails (slot, block hash, previous block hash, rewards, reward partitions, block time, block height), then one Transaction per transaction whose status is Ok. Transactions carry the block's block_time and block_hash, is_vote: false, and index set to the position in the notification's transaction list. With MentionsAccountOrProgram that list contains only matching transactions, so index is not the position in the full block.
• update_types(). Declares only Transaction, although BlockDetails is emitted too. The pipeline does not check this list, so .block_details(…) processors receive the blocks.
• Reconnects. On stream close or stream_timeout, the datasource waits 3 seconds and reconnects. Connection failures are retried every 3 seconds up to 10 times, subscription failures up to 11 times, then consume returns an error. The counter resets after a successful subscription. Missed blocks are not replayed.
• Backpressure. Updates use try_send. If the BlockDetails send fails, the subscription is dropped and reopened. If a transaction send fails, the rest of that block's transactions are dropped. Both log Error sending ….
• Finishing. Never finishes on its own.
• Metrics. block_subscribe_blocks_received_total, block_subscribe_transactions_processed_total (counters), block_subscribe_block_process_time_nanoseconds, block_subscribe_transaction_process_time_nanoseconds (histograms).

Troubleshooting#

Failed to subscribe to block updates: … Method not found (-32601)#

The endpoint doesn't support blockSubscribe. The public mainnet endpoint never does. Use a node with --rpc-pubsub-enable-block-subscription, a provider that exposes it, or switch to Yellowstone gRPC or the block crawler.

Every transaction logs Failed to decode transaction#

Set encoding: Some(UiTransactionEncoding::Base64) in RpcBlockSubscribeConfig.

Block stream timeout - no messages for 30s#

The node stopped sending blocks, often because the filter matched nothing for 30 seconds. With a narrow MentionsAccountOrProgram filter on a quiet program, raise the timeout with .with_stream_timeout(Duration::from_secs(300)).

Blocks are missing after a reconnect#

Expected: nothing is replayed. Pass a disconnect_notifier and backfill last_slot_before_disconnect..first_slot_after_reconnect with the block crawler.

See also#

• Stream with RPC
• RPC block crawler
• Processor inputs for BlockDetails
• Datasources concept and update types