Helius enhanced WebSockets

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.

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.

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

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<InstructionProcessorInputType<'_, OrcaWhirlpoolInstruction>> 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(())
    }
}

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:

[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#

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

For example:

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#

pub fn new(accounts: Vec<Pubkey>, transactions: Option<RpcTransactionsConfig>) -> CarbonResult<Self>

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:

options: TransactionSubscribeOptions. Its Default is not usable with Carbon, because it asks for JsonParsed:

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 or RpcProgramSubscribe 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 or speed up processors.

See also#

• Helius LaserStream: gRPC on Helius, with replay.
• Stream with RPC
• Datasources concept and update types