# Stream with Helius LaserStream (Carbon 2.0) > Stream transactions and accounts from Helius LaserStream, start from a past slot with from_slot, and resume after disconnects with slot replay. 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](https://carbon.sevenlabs.org/guides/stream-with-yellowstone.md), the filters are identical; only the datasource type and its config differ. ### Step 1: Create a project ```sh cargo new dlmm-laserstream cd dlmm-laserstream ``` ### Step 2: Add the dependencies `Cargo.toml`: ```toml [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 } ``` The filters are `yellowstone-grpc-proto` types, so pin it to `=12.7.0`, the version the datasource uses. `rustls` lets you choose the TLS crypto provider the gRPC client needs. ### Step 3: Write the pipeline `src/main.rs`: ```rust 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> 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(()) } } ``` The API key goes in the second argument; the datasource sends it as the `x-token` header. ### Step 4: Run it ```sh LASERSTREAM_URL="https://laserstream-mainnet-ewr.helius-rpc.com" HELIUS_API_KEY="your-helius-api-key" cargo run ``` Each decoded swap prints one line: Output: ```text slot 467110307 swap ue82HW7rk3DcWiCTzRrnMzMeKDYDLWUiyeMBtMi4tLtqiKaRGgbr7EYigU9RbpftsFsgEC7P6kgxkNWJgZhvKZs: amount_in=10298583 slot 467110307 swap 48kbDBkFMweb1bVYk9o3uGSR9CQS5pkgm8YCGB9Qg9ymui5dXmwjjoiSAHfF4eQ8mt6j688pkrG8n33GnZX4cffd: amount_in=1880866 slot 467110307 swap2 2d6GNQzSzxgV85mnzD4ed6JyxdLYdwBtyCH8LujdYaxySSKfD8x96yJLtxmuVYt6XGRjy8m3AutxyxqiumscEb4W: amount_in=381756 … ``` Press `Ctrl+C` to stop. ## 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. ```sh 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`](https://carbon.sevenlabs.org/concepts/filters.md#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](https://carbon.sevenlabs.org/guides/backfill-transactions.md) instead. > **Warning: from_slot needs replay_enabled.** With `replay_enabled: false`, the datasource clears `from_slot` before every subscribe, including the first, so the stream always starts at the tip. ## 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: | Commitment | Resumes from | | ------------------------ | ----------------------------------------------------------------------- | | `Confirmed`, `Finalized` | The last slot seen | | `Processed`, or `None` | The last slot seen minus 31, since processed slots can still be dropped | 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](https://carbon.sevenlabs.org/guides/multiple-datasources.md)) 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](https://carbon.sevenlabs.org/guides/prometheus.md) 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](https://carbon.sevenlabs.org/guides/stream-with-yellowstone.md#filter-transactions), and the datasource turns them into the same updates. `LaserStreamClientConfig` has these public fields: | Field | Default | Description | | ---------------------------- | ----------------------- | --------------------------------------------------------------------------------------- | | `replay_enabled` | `true` | Tracks slots and resumes from the last one after a reconnect. Required for `from_slot`. | | `from_slot` | `None` | Slot to start the first subscription from. | | `connect_timeout`, `timeout` | 15 s | Connection and request timeouts. | | `compression` | `None` | `Some(CompressionEncoding::Gzip)` or `Zstd`. | | `max_decoding_message_size` | tonic's default (4 MiB) | Raise it for large blocks or accounts. | | `tls_config` | system roots | A custom `ClientTlsConfig`. | | `tcp_nodelay` | `None` | Sets `TCP_NODELAY`. | `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 - [Stream with Yellowstone gRPC](https://carbon.sevenlabs.org/guides/stream-with-yellowstone.md): Filters, commitment and reconnect options in detail. - [Multiple datasources](https://carbon.sevenlabs.org/guides/multiple-datasources.md): Combine LaserStream with a second stream or a backfill. - [Helius LaserStream datasource](https://carbon.sevenlabs.org/datasources/helius-laserstream.md): The datasource's full reference. --- Carbon 2.0.0 documentation · Web: https://carbon.sevenlabs.org/guides/stream-with-laserstream · Last updated 2026-11-09 · Index of all pages: https://carbon.sevenlabs.org/llms.txt