# Stream with RPC websockets (Carbon 2.0)

> Stream account changes with programSubscribe on any Solana RPC, including the public one, and whole blocks with blockSubscribe on providers that enable it.

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](https://carbon.sevenlabs.org/guides/stream-with-yellowstone.md) 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.

### Step 1: Create a project

```sh
cargo new dlmm-accounts
cd dlmm-accounts
```

### Step 2: Add the dependencies

`Cargo.toml`:

```toml
[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"] }
```

`solana-client` provides the subscription config types. Pin it to `=4.2.2`, the version Carbon 2.0.0 uses, so one copy of the Solana types is in your build.

### Step 3: Write the pipeline

`src/main.rs`:

```rust
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(())
    }
}
```

The filters run on the RPC node, so only `LbPair` accounts cross the network. `encoding: Some(UiAccountEncoding::Base64)` is required: the datasource can't rebuild accounts from the node's default encoding.

### Step 4: Run it

```sh
cargo run
```

A pool prints each time it changes, with its active bin:

Output:

```text
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
…
```

Press `Ctrl+C` to stop.

### 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`](https://carbon.sevenlabs.org/guides/load-account-state.md) and combine the two as in [Multiple datasources](https://carbon.sevenlabs.org/guides/multiple-datasources.md#load-state-then-follow-changes).
- **No transaction signature.** `input.metadata.transaction_signature` is always `None`, so a [`DeduplicationFilter`](https://carbon.sevenlabs.org/concepts/filters.md#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](https://carbon.sevenlabs.org/concepts/updates.md#account-deletions), 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:

`Cargo.toml`:

```toml
[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"] }
```

`src/main.rs`:

```rust
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(())
    }
}
```

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

Each block prints first, followed by its swaps:

Output:

```text
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
…
```

> **Warning: Base64, full details and version 1 are required.** Set `encoding: Some(UiTransactionEncoding::Base64)`, `transaction_details: Some(TransactionDetails::Full)` and `max_supported_transaction_version: Some(1)`. Carbon can't rebuild a transaction from `Json` or `JsonParsed` payloads, and without version `1` the node refuses blocks that contain versioned transactions.

### 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`](https://carbon.sevenlabs.org/concepts/filters.md#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](https://carbon.sevenlabs.org/guides/backfill-transactions.md#crawl-a-slot-range).
- **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](https://carbon.sevenlabs.org/guides/stream-with-yellowstone.md). 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

- [Load account state](https://carbon.sevenlabs.org/guides/load-account-state.md): Fetch every matching account before streaming changes.
- [Multiple datasources](https://carbon.sevenlabs.org/guides/multiple-datasources.md): Load state with gPA and follow it with programSubscribe.
- [Stream with Yellowstone gRPC](https://carbon.sevenlabs.org/guides/stream-with-yellowstone.md): Transactions and accounts in real time, with filters on both.

---

Carbon 2.0.0 documentation · Web: https://carbon.sevenlabs.org/guides/stream-with-rpc · Last updated 2026-11-09 · Index of all pages: https://carbon.sevenlabs.org/llms.txt