# RPC programSubscribe (Carbon 2.0) > Stream live changes to every account a program owns over a Solana RPC WebSocket, including the free public endpoint. `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. | | | | -------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------ | | Crate | `carbon-rpc-program-subscribe-datasource` 2.0.0 | | Install | `cargo add carbon-rpc-program-subscribe-datasource@2.0.0` | | Emits | `AccountUpdate`, `AccountDeletion` | | Mode | Live. Only changes after the subscription starts; no initial snapshot. Runs until cancelled. | | Provider | Any Solana RPC WebSocket endpoint (`wss://…`). The public `wss://api.mainnet-beta.solana.com` works. | | API docs | [docs.rs](https://docs.rs/carbon-rpc-program-subscribe-datasource/2.0.0/carbon_rpc_program_subscribe_datasource/) | | Source | [`datasources/rpc-program-subscribe-datasource`](https://github.com/sevenlabs-hq/carbon/blob/v2.0.0/datasources/rpc-program-subscribe-datasource/src/lib.rs) | ## Quick example This program streams changes to Orca Whirlpool pool accounts from the public endpoint and prints each pool's current tick and liquidity. ```sh 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 ``` `src/main.rs`: ```rust 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> 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(()) } } ``` ```sh cargo run ``` Updates start within a second. On mainnet this subscription produced about 2,300 updates in 25 seconds: Output: ```text 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` ```rust pub const fn new(rpc_ws_url: String, filters: Filters) -> Self ``` | Parameter | Type | Description | | ------------ | --------- | ---------------------------------------- | | `rpc_ws_url` | `String` | WebSocket endpoint, `ws://` or `wss://`. | | `filters` | `Filters` | Program and subscription config. | ### `Filters` ```rust pub const fn new(pubkey: Pubkey, program_subscribe_config: Option) -> Self ``` | Field | Type | Default | Meaning | | -------------------------- | ---------------------------------- | ------- | --------------------------------------------------------------------- | | `pubkey` | `Pubkey` | — | Program ID to subscribe to. | | `program_subscribe_config` | `Option` | `None` | Passed to `programSubscribe` unchanged (`solana_client::rpc_config`). | The fields of `RpcProgramAccountsConfig` that matter here: | Field | Meaning | | --------------------------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------- | | `filters` | `DataSize` and `Memcmp` filters, as for [`getProgramAccounts`](https://carbon.sevenlabs.org/datasources/rpc-gpa.md#rpcprogramaccountsconfig). | | `account_config.encoding` | **Set `Some(UiAccountEncoding::Base64)`.** The datasource does not set it, and with the server default every event fails with `Error decoding account event`. | | `account_config.commitment` | Commitment of notifications. `None` uses the server default (`finalized` on Agave). | | `account_config.data_slice` | Return part of the data. Decoders usually fail on sliced data. | ## 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`](https://carbon.sevenlabs.org/datasources/rpc-gpa.md) in the same pipeline; see [Load account state](https://carbon.sevenlabs.org/guides/load-account-state.md). - **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`](https://carbon.sevenlabs.org/reference/pipeline-builder.md) 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`](https://carbon.sevenlabs.org/datasources/rpc-gpa.md). ## See also - [Stream with RPC](https://carbon.sevenlabs.org/guides/stream-with-rpc.md) - [Load account state](https://carbon.sevenlabs.org/guides/load-account-state.md) - [Multiple datasources](https://carbon.sevenlabs.org/guides/multiple-datasources.md) - [Datasources concept](https://carbon.sevenlabs.org/concepts/datasources.md) and [update types](https://carbon.sevenlabs.org/concepts/updates.md) --- Carbon 2.0.0 documentation · Web: https://carbon.sevenlabs.org/datasources/rpc-program-subscribe · Last updated 2026-11-09 · Index of all pages: https://carbon.sevenlabs.org/llms.txt