# Helius enhanced WebSockets (Carbon 2.0) > Stream transactions and account changes from Helius enhanced (Atlas) WebSockets with transactionSubscribe and accountSubscribe. `HeliusWebsocket` connects to Helius enhanced WebSockets (`wss://atlas-mainnet.helius-rpc.com`), runs one `transactionSubscribe` with your filter and one `accountSubscribe` per account you list, and emits transaction and account updates. It watches the Clock sysvar to detect a stalled connection and reconnects everything when it does. Use it for live transactions on Helius when you don't need gRPC. | | | | -------- | ------------------------------------------------------------------------------------------------------------------------------------------------ | | Crate | `carbon-helius-atlas-ws-datasource` 2.0.0 | | Install | `cargo add carbon-helius-atlas-ws-datasource@2.0.0` | | Emits | `Transaction`, `AccountUpdate`, `AccountDeletion` | | Mode | Live. Runs until cancelled or until reconnects keep failing. | | Provider | A Helius API key on a plan that includes enhanced WebSockets. You also need the `helius` crate (2.0.0) for the filter types. | | API docs | [docs.rs](https://docs.rs/carbon-helius-atlas-ws-datasource/2.0.0/carbon_helius_atlas_ws_datasource/) | | Source | [`datasources/helius-atlas-ws-datasource`](https://github.com/sevenlabs-hq/carbon/blob/v2.0.0/datasources/helius-atlas-ws-datasource/src/lib.rs) | ## Quick example This program subscribes to successful Orca Whirlpool transactions and prints each `swapV2`. It compiles as shown; to run it, set `HELIUS_API_KEY`. ```sh cargo new helius-ws && cd helius-ws cargo add carbon-core@2.0.0 carbon-log-metrics@2.0.0 carbon-orca-whirlpool-decoder@2.0.0 carbon-helius-atlas-ws-datasource@2.0.0 cargo add helius@2.0.0 cargo add tokio@1 --features full cargo add env_logger@0.11 log@0.4 ``` `src/main.rs`: ```rust use { carbon_core::{ error::CarbonResult, instruction::InstructionProcessorInputType, pipeline::Pipeline, processor::Processor, }, carbon_helius_atlas_ws_datasource::{Filters, HeliusWebsocket}, carbon_log_metrics::LogMetrics, carbon_orca_whirlpool_decoder::{ instructions::OrcaWhirlpoolInstruction, OrcaWhirlpoolDecoder, PROGRAM_ID, }, helius::types::{ enhanced_websocket::TransactionDetails, Cluster, RpcTransactionsConfig, TransactionCommitment, TransactionSubscribeFilter, TransactionSubscribeOptions, UiEnhancedTransactionEncoding, }, std::{env, sync::Arc}, }; #[tokio::main] async fn main() -> CarbonResult<()> { env_logger::init(); let transactions = RpcTransactionsConfig { filter: TransactionSubscribeFilter { vote: Some(false), failed: Some(false), account_include: Some(vec![PROGRAM_ID.to_string()]), ..Default::default() }, options: TransactionSubscribeOptions { commitment: Some(TransactionCommitment::Confirmed), // Required: the helius default is JsonParsed, which Carbon can't decode. encoding: Some(UiEnhancedTransactionEncoding::Base64), transaction_details: Some(TransactionDetails::Full), show_rewards: Some(false), max_supported_transaction_version: Some(1), }, }; let datasource = HeliusWebsocket::new( env::var("HELIUS_API_KEY").expect("HELIUS_API_KEY must be set"), Filters::new(vec![], Some(transactions))?, Cluster::MainnetBeta, ); Pipeline::builder() .datasource(datasource) .metrics(Arc::new(LogMetrics::new())) .instruction(OrcaWhirlpoolDecoder, SwapLogger) .build()? .run() .await } struct SwapLogger; impl Processor> 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(()) } } ``` ```sh HELIUS_API_KEY=your-helius-api-key RUST_LOG=info cargo run ``` Each matching swap prints one `slot … swapV2 pool=… amount=…` line. With an invalid key the connection is refused and retried every 3 seconds: Output: ```text [ERROR carbon_helius_atlas_ws_datasource] Failed to create Helius client: Unable to connect to server: HTTP error: 401 Unauthorized ``` `TransactionDetails` must be imported from `helius::types::enhanced_websocket`: `helius::types::TransactionDetails` is a different type used by Helius' HTTP API. ## Configuration ### `HeliusWebsocket` ```rust pub const fn new(api_key: String, filters: Filters, cluster: Cluster) -> Self pub fn with_ping_interval_secs(self, ping_interval_secs: u64) -> Self pub fn with_pong_timeout_secs(self, pong_timeout_secs: u64) -> Self pub fn with_transaction_idle_timeout_secs(self, transaction_idle_timeout_secs: u64) -> Self ``` | Field | Type | Default | Meaning | | ------------------------------- | ------------------------ | ----------------------- | ---------------------------------------------------------------------------------------------------------------- | | `api_key` | `String` | — | Helius API key. | | `filters` | `Filters` | — | What to subscribe to. | | `cluster` | `helius::types::Cluster` | — | `MainnetBeta` or `Devnet`. `StakedMainnetBeta` has no enhanced WebSocket URL and fails to connect. | | `ping_interval_secs` | `Option` | `None` (10 s) | Interval between WebSocket pings. | | `pong_timeout_secs` | `Option` | `None` (3 missed pongs) | Close the connection after this long without a pong, rounded up to whole ping intervals. `0` disables the check. | | `transaction_idle_timeout_secs` | `Option` | `None` | Reconnect when no transaction arrives for this long. Leave unset for quiet filters. | For example: ```rust let datasource = HeliusWebsocket::new(api_key, filters, Cluster::MainnetBeta) .with_ping_interval_secs(10) .with_pong_timeout_secs(30) .with_transaction_idle_timeout_secs(60); ``` ### `Filters` ```rust pub fn new(accounts: Vec, transactions: Option) -> CarbonResult ``` | Field | Type | Meaning | | -------------- | ------------------------------- | ------------------------------------------------------------------------------------------------------------------ | | `accounts` | `Vec` | One `accountSubscribe` per address. Opened without options, so the server's default encoding and commitment apply. | | `transactions` | `Option` | One `transactionSubscribe`. | `new` returns `Err(Custom("Error creating Filters for the Helius WebSocket: accounts and transactions can't be both empty"))` when both are empty. ### `RpcTransactionsConfig` (from `helius::types`) `filter: TransactionSubscribeFilter`: | Field | Type | Meaning | | ------------------ | --------------------- | -------------------------------------------------------------------------------- | | `vote` | `Option` | Include or exclude vote transactions. | | `failed` | `Option` | Include or exclude failed transactions. Carbon drops failed transactions anyway. | | `signature` | `Option` | A single signature. | | `account_include` | `Option>` | Any of these accounts (OR). | | `account_exclude` | `Option>` | None of these accounts. | | `account_required` | `Option>` | All of these accounts (AND). | `options: TransactionSubscribeOptions`. Its `Default` is not usable with Carbon, because it asks for `JsonParsed`: | Field | Type | Set to | Meaning | | ----------------------------------- | --------------------------------------- | ----------- | ------------------------------------------------------------------------------------------------- | | `commitment` | `Option` | `Confirmed` | `Processed`, `Confirmed` or `Finalized`. | | `encoding` | `Option` | `Base64` | **Required.** Any JSON encoding makes every transaction fail with `Failed to decode transaction`. | | `transaction_details` | `Option` | `Full` | Other levels are skipped with `Skipping non-full Helius transaction notification`. | | `show_rewards` | `Option` | as needed | Include rewards in the metadata. | | `max_supported_transaction_version` | `Option` | `Some(1)` | Highest transaction version to receive. | ## Behaviour - **Subscriptions.** Each connection runs a Clock sysvar subscription, one task per account, and one transaction task. - **Stall detection.** If no Clock update arrives for 5 seconds, or the Clock jumps more than 10 slots, all subscriptions are closed and reopened. With `transaction_idle_timeout_secs`, a silent transaction stream does the same. - **Reconnects.** Between connections the datasource waits 3 seconds. Building the client is retried up to 30 times in a row, then `consume` returns an error. Missed data is not replayed. - **Partial failures.** If one `accountSubscribe` or the `transactionSubscribe` call fails, or its stream closes, only that task stops (with an error in the log); the others keep running until the next reconnect. - **Transaction updates.** Failed transactions are dropped. `slot` comes from the notification; `index`, `block_time` and `block_hash` are `None`. `is_vote` is copied from your filter's `vote` value, not from the transaction. Return data in `meta.return_data` holds the base64 text as bytes rather than the decoded bytes. - **Account updates.** `AccountUpdate` (or `AccountDeletion` at zero lamports) with the notification slot and `transaction_signature: None`. - **Backpressure.** Updates use `try_send`. When the pipeline channel is full, the update is dropped and that subscription's task stops until the next reconnect. - **Metrics.** Counters `helius_atlas_ws_transaction_updates_received_total`, `helius_atlas_ws_account_updates_received_total`, `helius_atlas_ws_account_deletions_received_total`; histograms `helius_atlas_ws_transaction_process_time_nanoseconds`, `helius_atlas_ws_account_process_time_nanoseconds`, `helius_atlas_ws_account_deletion_process_time_nanoseconds`, `helius_atlas_ws_clock_process_time_nanoseconds`. ## Troubleshooting ### `Failed to create Helius client: Unable to connect to server: HTTP error: 401 Unauthorized` The API key is wrong, or your plan doesn't include enhanced WebSockets. After 30 attempts the datasource gives up. ### Every transaction logs `Failed to decode transaction` `options.encoding` is `JsonParsed` (the helius default) or another JSON encoding. Set `Some(UiEnhancedTransactionEncoding::Base64)`. ### `Error decoding Helius WS Account event` Account subscriptions use the server's default encoding, which Carbon can't always decode (for example for larger accounts). For program-owned accounts, use [Yellowstone](https://carbon.sevenlabs.org/datasources/yellowstone-grpc.md) or [`RpcProgramSubscribe`](https://carbon.sevenlabs.org/datasources/rpc-program-subscribe.md) with Base64 instead. ### `No Clock updates received for 5 seconds, triggering reconnection` The connection stalled; the datasource reconnects on its own. Frequent occurrences point to network problems or an overloaded process. ### Transactions stop after `Error sending transaction update` The pipeline channel was full, and the transaction task stopped. It resumes at the next reconnect. Raise the pipeline's [`channel_buffer_size`](https://carbon.sevenlabs.org/reference/pipeline-builder.md) or speed up processors. ## See also - [Helius LaserStream](https://carbon.sevenlabs.org/datasources/helius-laserstream.md): gRPC on Helius, with replay. - [Stream with RPC](https://carbon.sevenlabs.org/guides/stream-with-rpc.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/helius-atlas-ws · Last updated 2026-11-09 · Index of all pages: https://carbon.sevenlabs.org/llms.txt