Stream with RPC websockets

Solana RPC nodes push updates over websockets. Carbon has two websocket datasources: rpc-program-subscribe streams every change to a program's accounts and works on any RPC, the public endpoint included; rpc-block-subscribe streams confirmed blocks with their transactions, but only on RPC providers that enable blockSubscribe. Use them when you don't have a Geyser gRPC endpoint.

You need: Rust 1.96.1 or newer. For account updates, nothing else: this guide runs on wss://api.mainnet-beta.solana.com. For block updates, a websocket URL from an RPC provider that supports blockSubscribe.

Stream account changes#

This program streams every Meteora DLMM pool (LbPair account) as swaps and liquidity changes write to it.

cargo new dlmm-accounts
cd dlmm-accounts

[package]
name = "dlmm-accounts"
version = "0.1.0"
edition = "2024"

[dependencies]
carbon-core = "2.0.0"
carbon-meteora-dlmm-decoder = "2.0.0"
carbon-rpc-program-subscribe-datasource = "2.0.0"
env_logger = "0.11.11"
log = "0.4.35"
solana-client = "=4.2.2"
tokio = { version = "1.54.1", features = ["full"] }

use {
    carbon_core::{
        account::AccountProcessorInputType, error::CarbonResult, pipeline::Pipeline,
        processor::Processor,
    },
    carbon_meteora_dlmm_decoder::{accounts::MeteoraDlmmAccount, MeteoraDlmmDecoder, PROGRAM_ID},
    carbon_rpc_program_subscribe_datasource::{Filters, RpcProgramSubscribe},
    solana_client::{
        rpc_config::{
            CommitmentConfig, RpcAccountInfoConfig, RpcProgramAccountsConfig, UiAccountEncoding,
        },
        rpc_filter::{Memcmp, RpcFilterType},
    },
};

// First 8 bytes of every Meteora DLMM `LbPair` account.
const LB_PAIR_DISCRIMINATOR: [u8; 8] = [33, 11, 49, 98, 181, 101, 177, 13];

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

    let config = RpcProgramAccountsConfig {
        filters: Some(vec![RpcFilterType::Memcmp(Memcmp::new_raw_bytes(
            0,
            LB_PAIR_DISCRIMINATOR.to_vec(),
        ))]),
        account_config: RpcAccountInfoConfig {
            encoding: Some(UiAccountEncoding::Base64),
            commitment: Some(CommitmentConfig::confirmed()),
            ..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)
        .account(MeteoraDlmmDecoder, PairLogger)
        .build()?
        .run()
        .await
}

struct PairLogger;

impl Processor<AccountProcessorInputType<'_, MeteoraDlmmAccount>> for PairLogger {
    async fn process(
        &mut self,
        input: &AccountProcessorInputType<'_, MeteoraDlmmAccount>,
    ) -> CarbonResult<()> {
        if let MeteoraDlmmAccount::LbPair(pair) = &input.decoded_account.data {
            println!(
                "slot {} pair {}: active_id={} bin_step={}",
                input.metadata.slot, input.metadata.pubkey, pair.active_id, pair.bin_step
            );
        }
        Ok(())
    }
}

cargo run

slot 467109944 pair 8obGpjiUu7QTJHK58YHCoz5HxobmrVP2x5zpMZu3c4BT: active_id=-78 bin_step=80
slot 467109944 pair EXXfbVutbuexDZtPw8pkSvWxLFoUMBHd544CAGkmtPZm: active_id=-2912 bin_step=10
slot 467109944 pair 54Vp27uLaw4wNLo5n7r4fcC6zLamoQc28xBARjss4EUJ: active_id=-2625 bin_step=16
slot 467109945 pair 2tFYCTqYJ2pyrb7CeA824o2Ybq3jq7agvr3Qpv2uVDmL: active_id=-543 bin_step=80
slot 467109945 pair GMgh4NtWrGaUf1RR2kcXD7LY1jou1qFAuSsQeKp5ow4a: active_id=4903 bin_step=5
slot 467109946 pair GMgh4NtWrGaUf1RR2kcXD7LY1jou1qFAuSsQeKp5ow4a: active_id=4904 bin_step=5
…

How the account stream behaves#

• Only changes. The subscription sends an account when it's written, not its current state. Load the starting state with getProgramAccounts and combine the two as in Multiple datasources.
• No transaction signature. input.metadata.transaction_signature is always None, so a DeduplicationFilter lets every update through. input.metadata.slot is the slot of the notification.
• Closed accounts. An update with zero lamports arrives as an account deletion, not an account update: add .account_deletions(processor) to handle it.
• Reconnects. When the websocket closes, the datasource waits 3 seconds and subscribes again. After 10 failed connection attempts in a row it stops with an error. Changes made while disconnected aren't replayed.
• Backpressure. If the pipeline's channel is full (1,000 updates by default), the datasource logs Error sending account update, drops the update and reconnects. Raise .channel_buffer_size(…) on the builder if your processor is slow.

The public endpoint limits subscriptions per IP address, and busy programs send a lot of data. Narrow the filters (by discriminator, DataSize, or a Memcmp on a field such as a mint) and use a provider's websocket for production.

Stream blocks#

rpc-block-subscribe sends each block's transactions as transaction updates and the block itself as a BlockDetails update. It needs an RPC node started with --rpc-pubsub-enable-block-subscription. Most providers enable it on paid plans; the public endpoint rejects it with Method not found.

Add the datasource to a project set up like the one above:

[dependencies]
carbon-core = "2.0.0"
carbon-meteora-dlmm-decoder = "2.0.0"
carbon-rpc-block-subscribe-datasource = "2.0.0"
env_logger = "0.11.11"
log = "0.4.35"
solana-client = "=4.2.2"
tokio = { version = "1.54.1", features = ["full"] }

use {
    carbon_core::{
        datasource::BlockDetails, error::CarbonResult,
        instruction::InstructionProcessorInputType, pipeline::Pipeline, processor::Processor,
    },
    carbon_meteora_dlmm_decoder::{
        instructions::MeteoraDlmmInstruction, MeteoraDlmmDecoder, PROGRAM_ID,
    },
    carbon_rpc_block_subscribe_datasource::{Filters, RpcBlockSubscribe},
    solana_client::rpc_config::{
        CommitmentConfig, RpcBlockSubscribeConfig, RpcBlockSubscribeFilter, TransactionDetails,
        UiTransactionEncoding,
    },
    std::env,
};

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

    let filters = Filters::new(
        // Only transactions that mention Meteora DLMM.
        RpcBlockSubscribeFilter::MentionsAccountOrProgram(PROGRAM_ID.to_string()),
        Some(RpcBlockSubscribeConfig {
            commitment: Some(CommitmentConfig::confirmed()),
            encoding: Some(UiTransactionEncoding::Base64),
            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,
    );

    Pipeline::builder()
        .datasource(datasource)
        .instruction(MeteoraDlmmDecoder, SwapLogger)
        .block_details(BlockLogger)
        .build()?
        .run()
        .await
}

struct SwapLogger;

impl Processor<InstructionProcessorInputType<'_, MeteoraDlmmInstruction>> for SwapLogger {
    async fn process(
        &mut self,
        input: &InstructionProcessorInputType<'_, MeteoraDlmmInstruction>,
    ) -> CarbonResult<()> {
        let meta = &input.metadata.transaction_metadata;
        match input.decoded_instruction {
            MeteoraDlmmInstruction::Swap { data, .. } => {
                println!("slot {} swap  {}: amount_in={}", meta.slot, meta.signature, data.amount_in)
            }
            MeteoraDlmmInstruction::Swap2 { data, .. } => {
                println!("slot {} swap2 {}: amount_in={}", meta.slot, meta.signature, data.amount_in)
            }
            _ => {}
        }
        Ok(())
    }
}

struct BlockLogger;

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

RPC_WS_URL="wss://your-provider.example/?api-key=…" cargo run

Each block prints first, followed by its swaps:

block 467110307 time=Some(1794182812)
slot 467110307 swap  ue82HW7rk3DcWiCTzRrnMzMeKDYDLWUiyeMBtMi4tLtqiKaRGgbr7EYigU9RbpftsFsgEC7P6kgxkNWJgZhvKZs: amount_in=10298583
slot 467110307 swap  48kbDBkFMweb1bVYk9o3uGSR9CQS5pkgm8YCGB9Qg9ymui5dXmwjjoiSAHfF4eQ8mt6j688pkrG8n33GnZX4cffd: amount_in=1880866
slot 467110307 swap2 2d6GNQzSzxgV85mnzD4ed6JyxdLYdwBtyCH8LujdYaxySSKfD8x96yJLtxmuVYt6XGRjy8m3AutxyxqiumscEb4W: amount_in=381756
block 467110308 time=Some(1794182812)
slot 467110308 swap2 2miT4n9yFXZMdUTVF4My1RVvHrbbZMwD64vYyppzdkxNg5qVtYJgsm4dL8MPYtBRrJm2V3EzP4ZfYJKjEc61Wzr1: amount_in=28328938705735
…

How the block stream behaves#

• Filter. MentionsAccountOrProgram takes one address. To follow several programs, use RpcBlockSubscribeFilter::All (every transaction on the network, which is a lot of data) or one datasource per program with a DeduplicationFilter.
• Failed transactions are skipped. Only successful ones reach your pipes.
• Metadata. Each transaction gets the block's slot, time and hash. transaction_metadata.index is the transaction's position in the list the node sent; with MentionsAccountOrProgram that list holds only the matching transactions, so it isn't the position in the full block.
• Block details. The BlockDetails update carries slot, block_hash, previous_block_hash, block_time, block_height and, when you request them, rewards.
• Stale connections. If no block arrives for 30 seconds, the datasource reconnects. Change it with RpcBlockSubscribe::new(url, filters).with_stream_timeout(Duration::from_secs(60)).
• Gaps. Blocks produced while disconnected aren't replayed. Construct the datasource with RpcBlockSubscribe::with_disconnect_notifier(url, filters, sender) to receive a DatasourceDisconnection with the missed slot range after each reconnect, then fill it with the block crawler.
• Reconnects. After a disconnect it waits 3 seconds and subscribes again. It gives up after 10 failed connection attempts, or 11 rejected subscriptions, in a row.

Common problems#

Failed to subscribe to block updates: SubscribeFailed { reason: "Method not found (-32601)", … }#

The node doesn't allow blockSubscribe. The public endpoint never does. Use a provider whose websocket enables it, or stream with Yellowstone gRPC. After 11 rejections the datasource stops and the pipeline exits.

Error decoding account event for every update#

The program subscription uses the node's default encoding. Set encoding: Some(UiAccountEncoding::Base64) in RpcAccountInfoConfig, as in the example. Passing None as the config has the same problem.

Failed to decode transaction for every transaction#

The block subscription uses Json or JsonParsed encoding. Set encoding: Some(UiTransactionEncoding::Base64).

Failed to create RPC subscribe client#

The websocket URL is wrong or unreachable. It must start with wss:// (or ws://), not https://. Providers usually give a separate websocket URL.

Nothing prints#

The filters match nothing, or the decoder doesn't match the program you subscribed to. Remove the Memcmp filter to test, and check that the account discriminator bytes are those of the account type you want.

Next steps#