Stream with Helius LaserStream

Helius LaserStream is a hosted, Yellowstone-compatible gRPC stream that can replay recent slots. Carbon's LaserStream datasource uses that to start from a past slot and to resume where it left off after a disconnect. This guide streams Meteora DLMM swaps from LaserStream and shows how from_slot and replay behave.

You need:

• Rust 1.96.1 or newer.
• A Helius account on a plan that includes LaserStream, and its API key.
• The LaserStream endpoint for your region, such as https://laserstream-mainnet-ewr.helius-rpc.com. Helius's dashboard and docs list the regional endpoints.

If you already use Yellowstone gRPC, the filters are identical; only the datasource type and its config differ.

cargo new dlmm-laserstream
cd dlmm-laserstream

[package]
name = "dlmm-laserstream"
version = "0.1.0"
edition = "2024"

[dependencies]
carbon-core = "2.0.0"
carbon-helius-laserstream-datasource = "2.0.0"
carbon-log-metrics = "2.0.0"
carbon-meteora-dlmm-decoder = "2.0.0"
env_logger = "0.11.11"
log = "0.4.35"
rustls = { version = "0.23", default-features = false, features = ["std", "ring"] }
tokio = { version = "1.54.1", features = ["full"] }
yellowstone-grpc-proto = { version = "=12.7.0", default-features = false }

use {
    carbon_core::{
        error::CarbonResult, instruction::InstructionProcessorInputType, pipeline::Pipeline,
        processor::Processor,
    },
    carbon_helius_laserstream_datasource::{
        BlockFilters, LaserStreamClientConfig, LaserStreamGeyserClient,
    },
    carbon_log_metrics::LogMetrics,
    carbon_meteora_dlmm_decoder::{
        instructions::MeteoraDlmmInstruction, MeteoraDlmmDecoder, PROGRAM_ID,
    },
    std::{collections::HashMap, env, sync::Arc},
    yellowstone_grpc_proto::geyser::{CommitmentLevel, SubscribeRequestFilterTransactions},
};

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

    // The gRPC client uses rustls; pick its crypto provider before connecting.
    rustls::crypto::ring::default_provider()
        .install_default()
        .expect("failed to install the rustls crypto provider");

    // Successful, non-vote transactions that invoke Meteora DLMM.
    let mut transaction_filters = HashMap::new();
    transaction_filters.insert(
        "meteora_dlmm".to_string(),
        SubscribeRequestFilterTransactions {
            vote: Some(false),
            failed: Some(false),
            account_required: vec![PROGRAM_ID.to_string()],
            ..Default::default()
        },
    );

    // Start from a past slot if FROM_SLOT is set; otherwise start at the tip.
    let from_slot = env::var("FROM_SLOT")
        .ok()
        .map(|slot| slot.parse().expect("FROM_SLOT must be a slot number"));

    let config = LaserStreamClientConfig {
        from_slot,
        ..LaserStreamClientConfig::default() // replay_enabled: true
    };

    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(),
        config,
    );

    Pipeline::builder()
        .datasource(datasource)
        .metrics(Arc::new(LogMetrics::new()))
        .instruction(MeteoraDlmmDecoder, SwapLogger)
        .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(())
    }
}

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

slot 467110307 swap  ue82HW7rk3DcWiCTzRrnMzMeKDYDLWUiyeMBtMi4tLtqiKaRGgbr7EYigU9RbpftsFsgEC7P6kgxkNWJgZhvKZs: amount_in=10298583
slot 467110307 swap  48kbDBkFMweb1bVYk9o3uGSR9CQS5pkgm8YCGB9Qg9ymui5dXmwjjoiSAHfF4eQ8mt6j688pkrG8n33GnZX4cffd: amount_in=1880866
slot 467110307 swap2 2d6GNQzSzxgV85mnzD4ed6JyxdLYdwBtyCH8LujdYaxySSKfD8x96yJLtxmuVYt6XGRjy8m3AutxyxqiumscEb4W: amount_in=381756
…

Verify it's streaming#

Run with RUST_LOG=info. LogMetrics logs counters every five seconds; laserstream_transaction_updates_received_total should grow steadily. To check replay, note a slot from the output, stop the program, and start it again with FROM_SLOT set a few hundred slots earlier: the first lines printed should have slots at or after FROM_SLOT, and then catch up to the tip.

FROM_SLOT=467110000 LASERSTREAM_URL="https://laserstream-mainnet-ewr.helius-rpc.com" HELIUS_API_KEY="your-helius-api-key" cargo run

Start from a past slot#

from_slot asks LaserStream to replay from that slot before streaming live data. The slot must still be in Helius's replay window; check Helius's LaserStream docs for how far back your plan goes. Use it to:

• Resume after a restart. Store the last slot your processor handled (in your database, next to the data) and pass it as from_slot on startup. You'll see that slot's updates again, so make writes idempotent or add a DeduplicationFilter.
• Cover a deploy gap. Start the new version from a slot before the old one stopped.

For history beyond the replay window, use a backfill instead.

How replay handles disconnects#

LaserStreamClientConfig::default() has replay_enabled: true. The datasource then adds an internal slot subscription to the request and records the latest slot it reports. When the stream ends or errors, it waits 3 seconds and subscribes again with from_slot set to:

Updates from the resumed slots arrive a second time. Instruction and account pipes can drop them with a DeduplicationFilter; key your storage on signature and instruction path (or pubkey and slot) so a repeat is harmless.

After 10 consecutive failed subscribe attempts, the datasource stops. If it's your pipeline's only datasource, the pipeline then exits. Run a second stream for redundancy (Multiple datasources) or restart the process from a supervisor with from_slot.

Unlike the Yellowstone datasource, the LaserStream datasource has no stream timeout and no disconnect notifier. A connection that stays open but stops sending data isn't detected; watch laserstream_transaction_updates_received_total with Prometheus and alert when it stops growing.

Filters and settings#

The account, transaction and block filters are the same Yellowstone types as in Stream with Yellowstone gRPC, and the datasource turns them into the same updates. LaserStreamClientConfig has these public fields:

LaserStreamClientConfig::new takes the same values positionally, ending with replay_enabled and from_slot.

Like the Yellowstone datasource, it never waits for the pipeline: when the pipeline's channel is full, updates are dropped with Failed to send transaction update …. A replay from an old from_slot delivers a burst, so raise .channel_buffer_size(…) on the builder (default 1,000) when you replay.

Common problems#

Panic: Could not automatically determine the process-level CryptoProvider from Rustls crate features#

No rustls crypto provider is selected. Add the rustls dependency from the steps and call rustls::crypto::ring::default_provider().install_default() first thing in main.

Pipeline exits immediately with error consuming datasource: FailedToConsumeDatasource("gRPC transport error: …")#

The client couldn't connect. Check that LASERSTREAM_URL is a LaserStream gRPC endpoint with https://, not an RPC URL.

Geyser stream error, will reconnect: Status { … } every 3 seconds#

The server accepted the connection but rejected the stream: Unauthenticated or PermissionDenied for a wrong API key or a plan without LaserStream, or 405 Method Not Allowed when the URL points at an HTTP RPC endpoint. These retries don't count toward the reconnection limit, so the datasource keeps retrying until you fix the cause.

Failed to subscribe, will retry after 5s delay repeated, then Max reconnection attempts reached: 10#

The subscribe request itself failed ten times in a row, for example because the endpoint is unreachable or from_slot is outside the replay window. The retries are 3 seconds apart, despite the message. After the tenth, the datasource stops.

The stream starts at the tip although FROM_SLOT is set#

replay_enabled is false. Leave it at the default true.

Every swap appears twice after a reconnect#

That's the replay resuming from the last slot. Add a DeduplicationFilter to the pipe, or make your writes idempotent.

Next steps#