Helius LaserStream

LaserStreamGeyserClient subscribes to Helius LaserStream, a hosted gRPC service that speaks the Yellowstone protocol, and emits account and transaction updates. Unlike the Yellowstone datasource, it tracks the last slot it saw and asks LaserStream to replay from that slot after a reconnect. Use it for real-time indexing on Helius when gaps after a disconnect are not acceptable.

Quick example#

This program streams confirmed, successful Orca Whirlpool transactions and prints each swapV2. It compiles as shown; to run it, set LASERSTREAM_URL and HELIUS_API_KEY.

cargo new laserstream && cd laserstream
cargo add carbon-core@2.0.0 carbon-log-metrics@2.0.0 carbon-orca-whirlpool-decoder@2.0.0 carbon-helius-laserstream-datasource@2.0.0
cargo add yellowstone-grpc-proto@=12.7.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_laserstream_datasource::{
        BlockFilters, LaserStreamClientConfig, LaserStreamGeyserClient,
    },
    carbon_log_metrics::LogMetrics,
    carbon_orca_whirlpool_decoder::{
        instructions::OrcaWhirlpoolInstruction, OrcaWhirlpoolDecoder, PROGRAM_ID,
    },
    std::{collections::HashMap, env, sync::Arc},
    yellowstone_grpc_proto::geyser::{CommitmentLevel, SubscribeRequestFilterTransactions},
};

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

    let mut transaction_filters = HashMap::new();
    transaction_filters.insert(
        "orca_whirlpool".to_string(),
        SubscribeRequestFilterTransactions {
            vote: Some(false),
            failed: Some(false),
            account_include: vec![PROGRAM_ID.to_string()],
            ..Default::default()
        },
    );

    let datasource = LaserStreamGeyserClient::new(
        env::var("LASERSTREAM_URL").expect("LASERSTREAM_URL must be set"),
        Some(env::var("HELIUS_API_KEY").expect("HELIUS_API_KEY must be set")),
        Some(CommitmentLevel::Confirmed),
        HashMap::new(), // account filters
        transaction_filters,
        BlockFilters::default(),
        LaserStreamClientConfig::default(), // replay_enabled: true
    );

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

LASERSTREAM_URL=https://laserstream-mainnet-ewr.helius-rpc.com HELIUS_API_KEY=your-helius-api-key RUST_LOG=info cargo run

Each matching swap prints one slot … swapV2 pool=… amount=… line. The Stream with LaserStream guide covers endpoints and keys.

Configuration#

LaserStreamGeyserClient::new#

pub const fn new(
    endpoint: String,
    x_token: Option<String>,
    commitment: Option<CommitmentLevel>,
    account_filters: HashMap<String, SubscribeRequestFilterAccounts>,
    transaction_filters: HashMap<String, SubscribeRequestFilterTransactions>,
    block_filters: BlockFilters,
    geyser_config: LaserStreamClientConfig,
) -> Self

There is no disconnect notifier or stream timeout on this datasource.

LaserStreamClientConfig#

pub const fn new(
    compression: Option<CompressionEncoding>,
    connect_timeout: Option<Duration>,
    timeout: Option<Duration>,
    max_decoding_message_size: Option<usize>,
    tls_config: Option<ClientTlsConfig>,
    tcp_nodelay: Option<bool>,
    replay_enabled: bool,
    from_slot: Option<u64>,
) -> Self

To start from a recent past slot:

let geyser_config = LaserStreamClientConfig {
    from_slot: Some(467_000_000),
    ..LaserStreamClientConfig::default()
};

Behaviour#

• Connecting. consume connects first and returns FailedToConsumeDatasource if that fails, which stops a pipeline that has no other datasource.
• Replay. With replay_enabled, the datasource adds an internal slot subscription and remembers the last slot it saw. On reconnect it sets from_slot to that slot for Confirmed and Finalized, or to 31 slots earlier for Processed (and when commitment is None). Messages from the resumed slot can arrive twice; add a deduplication filter if your processor isn't idempotent.
• Updates. Same mapping as Yellowstone: account messages become AccountUpdate or AccountDeletion, transaction messages become Transaction with is_vote and index from Geyser, and block messages expand into transactions and accounts. Failed and vote transactions pass unless your filter excludes them. No BlockDetails are emitted.
• Reconnects. After the stream ends, errors or fails to subscribe, the datasource waits 3 seconds and subscribes again. After 10 consecutive failed attempts it logs Max reconnection attempts reached: 10 and stops, which shuts down a pipeline that has no other datasource. A successful subscription resets the count. There is no inactivity timeout: a connection that stays open but silent is not detected.
• Backpressure. Updates use try_send and are dropped with Failed to send … when the pipeline channel is full.
• Metrics. Counters laserstream_account_updates_received_total, laserstream_account_deletions_received_total, laserstream_transaction_updates_received_total; histograms laserstream_account_process_time_nanoseconds, laserstream_account_deletion_process_time_nanoseconds, laserstream_transaction_process_time_nanoseconds.

Troubleshooting#

The pipeline exits with FailedToConsumeDatasource("gRPC transport error: …")#

The first connection failed. Check the regional endpoint URL, including https://, and your network.

Failed to subscribe, will retry after 5s delay#

The subscription was rejected, usually because the API key is wrong or the plan doesn't include LaserStream. The actual delay is 3 seconds; after 10 failures the datasource stops.

from_slot has no effect#

You set replay_enabled: false. The datasource only sends from_slot when replay is enabled.

Duplicate transactions after a reconnect#

Expected with replay. Deduplicate by signature, or make writes idempotent (for example, upserts keyed by signature).

The stream stops without an error#

There is no inactivity timeout. Monitor laserstream_transaction_updates_received_total (see Prometheus) and restart the process if it stops growing.

See also#

• Stream with LaserStream
• Yellowstone gRPC for filter types and the full update mapping
• Filters for deduplication
• Datasources concept and update types