RPC programSubscribe

RpcProgramSubscribe opens a programSubscribe WebSocket subscription and emits an account update each time an account owned by the program changes. Accounts that drop to zero lamports arrive as account deletions. Use it for live account state when you don't have a Geyser provider; it works on the public endpoint.

Quick example#

This program streams changes to Orca Whirlpool pool accounts from the public endpoint and prints each pool's current tick and liquidity.

cargo new program-subscribe && cd program-subscribe
cargo add carbon-core@2.0.0 carbon-log-metrics@2.0.0 carbon-orca-whirlpool-decoder@2.0.0 carbon-rpc-program-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::{
        account::AccountProcessorInputType, error::CarbonResult, pipeline::Pipeline,
        processor::Processor,
    },
    carbon_log_metrics::LogMetrics,
    carbon_orca_whirlpool_decoder::{
        accounts::OrcaWhirlpoolAccount, OrcaWhirlpoolDecoder, PROGRAM_ID,
    },
    carbon_rpc_program_subscribe_datasource::{Filters, RpcProgramSubscribe},
    solana_client::{
        rpc_config::{RpcAccountInfoConfig, RpcProgramAccountsConfig, UiAccountEncoding},
        rpc_filter::RpcFilterType,
    },
    std::sync::Arc,
};

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

    let config = RpcProgramAccountsConfig {
        // Only `Whirlpool` pool accounts (653 bytes).
        filters: Some(vec![RpcFilterType::DataSize(653)]),
        account_config: RpcAccountInfoConfig {
            encoding: Some(UiAccountEncoding::Base64),
            ..Default::default()
        },
        ..Default::default()
    };

    let datasource = RpcProgramSubscribe::new(
        "wss://api.mainnet-beta.solana.com".to_string(),
        Filters::new(PROGRAM_ID, Some(config)),
    );

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

struct PoolLogger;

impl Processor<AccountProcessorInputType<'_, OrcaWhirlpoolAccount>> for PoolLogger {
    async fn process(
        &mut self,
        input: &AccountProcessorInputType<'_, OrcaWhirlpoolAccount>,
    ) -> CarbonResult<()> {
        if let OrcaWhirlpoolAccount::Whirlpool(pool) = &input.decoded_account.data {
            println!(
                "slot {} pool {} tick={} liquidity={}",
                input.metadata.slot, input.metadata.pubkey, pool.tick_current_index, pool.liquidity
            );
        }
        Ok(())
    }
}

cargo run

Updates start within a second. On mainnet this subscription produced about 2,300 updates in 25 seconds:

slot 467112567 pool VY1ZQXjqBwvuWgVfTfhqanJe96GGoQrX7xZZDrWPGiT tick=-44384 liquidity=2559193909761
slot 467112567 pool HxA6SKW5qA4o12fjVgTpXdq2YnZ5Zv1s7SB4FFomsyLM tick=66176 liquidity=1192468585558
slot 467112567 pool 6R4r93V5fcMzc13CL2enEepDSYcr4Qx3ptZBDwudTXCo tick=5929 liquidity=225897662074
slot 467112567 pool 7qbRF6YsyGuLUVs6Y1q64bdVrfe4ZcUUz1JRdoVNUJnm tick=-24426 liquidity=565651282086

Press Ctrl-C to stop.

Configuration#

RpcProgramSubscribe::new#

pub const fn new(rpc_ws_url: String, filters: Filters) -> Self

Filters#

pub const fn new(pubkey: Pubkey, program_subscribe_config: Option<RpcProgramAccountsConfig>) -> Self

The fields of RpcProgramAccountsConfig that matter here:

Behaviour#

• Updates. Each notification becomes an AccountUpdate with the notification's context slot and transaction_signature: None. An account with zero lamports becomes an AccountDeletion instead.
• No snapshot. Only accounts that change after the subscription starts are emitted. Load the initial state with GpaDatasource in the same pipeline; see Load account state.
• Ordering. In the order the node sends notifications.
• Reconnects. When the stream closes, the datasource waits 3 seconds, opens a new client and subscribes again. Failing to connect is retried every 3 seconds, up to 10 attempts; failing to subscribe, up to 11. After that consume returns an error. The attempt counter resets after each successful subscription. Changes during the gap are not replayed.
• Backpressure. Updates are sent with try_send. If the pipeline channel is full, the datasource logs Error sending account update, drops the subscription and reconnects after 3 seconds. Raise the pipeline's channel_buffer_size for busy programs.
• Finishing. Never finishes on its own; stops on cancellation or Ctrl-C.
• Metrics. program_subscribe_accounts_processed_total (counter), program_subscribe_account_process_time_nanoseconds (histogram).

Troubleshooting#

Every event logs Error decoding account event#

The subscription uses the server's default encoding, which Carbon cannot decode. Set account_config.encoding: Some(UiAccountEncoding::Base64) as in the example.

Failed to subscribe to program updates#

Logged every 3 seconds. The endpoint rejected programSubscribe: check the URL scheme (wss://, not https://), the program ID, and your filters (a Memcmp filter with invalid bytes is rejected). The error text after Failed to subscribe to program updates: is the server's reason.

Updates stop and Error sending account update appears#

The pipeline is slower than the stream. The datasource reconnects, but changes during the gap are lost. Speed up the processor, raise channel_buffer_size, or filter more narrowly.

No updates for an account that exists#

The account hasn't changed since you subscribed. Load current state first with GpaDatasource.

See also#

• Stream with RPC
• Load account state
• Multiple datasources
• Datasources concept and update types