Ingesting the Solana Firehose: High-Throughput Decoding with Rust

In Part 1: Economics & System Design, we established the blueprint. We proved that by optimizing Helius RPC subscriptions and decoupling our ingestion layer from our database using NATS, we can track all Solana swap volume without burning our runway on infrastructure costs.

Now, we have to actually build the ingestion engine.

When you are dealing with thousands of transactions per second across Raydium, Orca, and Jupiter, traditional REST polling is obsolete. We need a persistent, real-time WebSocket connection. But keeping a socket alive at this scale, decoding the payload without causing CPU spikes, and forwarding it to our broker requires strict memory management.

Here is how we architect the Rust ingestion service to survive the firehose on a highly cost-effective 4-CPU pod.

Connection Resilience & The Reconnection Loop

Public and private WebSocket connections drop. Network blips happen. Helius might briefly cycle a node. If your Rust service uses a naive TCP stream that panics on a closed socket, your pipeline is dead in the water and your database starts missing swaps.

We wrap our WebSocket connection in a tokio asynchronous reconnection loop equipped with exponential backoff and heartbeat monitoring. Instead of crashing, the service detects a missed ping/pong heartbeat or a dropped stream, pauses, and safely reconnects.

async fn stream_with_retry(ws_url: &str, tx: mpsc::Sender<RawEvent>) {
    let mut backoff = Duration::from_secs(1);

    loop {
        match connect_async(ws_url).await {
            Ok((mut socket, _)) => {
                tracing::info!("Connected to Helius Enhanced WebSocket");
                backoff = Duration::from_secs(1); // Reset on success

                while let Some(msg) = socket.next().await {
                    process_message(msg, &tx).await;
                }
            }
            Err(e) => {
                tracing::error!("Disconnected: {}. Retrying in {:?}...", e, backoff);
                tokio::time::sleep(backoff).await;
                // Exponential backoff, capped at 30 s
                backoff = std::cmp::min(backoff * 2, Duration::from_secs(30));
            }
        }
    }
}

The backoff starts at 1 second and doubles on each consecutive failure, capped at 30 seconds. On a successful reconnect, it resets to 1 second. This pattern ensures the service does not hammer Helius during an outage, while recovering instantly under normal network conditions.

By handling socket death gracefully, this loop ensures that our container never has to be manually restarted by Kubernetes or Docker during a network hiccup. The entire reconnection lifecycle is invisible to the downstream NATS publisher.

Zero-Copy Decoding with the Carbon Crate

The fastest way to max out a 4-CPU server is attempting to fully deserialize every single transaction payload on Solana into a massive JSON object. The vast majority of the firehose is noise — NFT mints, token transfers, and governance votes that our swap indexer doesn't care about. If we allocate memory for those, we lose.

Filtering by Program Discriminators

To solve this, we leverage the Carbon framework. Carbon is an indexing framework built for Solana that allows us to pipeline our data: Datasource → Decoder → Processor.

Instead of deserializing the entire block, we configure our Helius datasource to apply a RpcBlockSubscribeFilter. This drops non-swap transactions at the RPC layer in nanoseconds — before they ever enter our process memory. For the transactions that do come through, Carbon's generated decoders turn the raw byte arrays directly into typed Rust structures without unnecessary allocations.

use carbon_core::{
    pipeline::Pipeline,
    processor::Processor,
};
use carbon_rpc_block_subscribe_datasource::{
    Filters, RpcBlockSubscribe,
};

// 1. Filter noise at the RPC layer — nanosecond cost
let datasource = RpcBlockSubscribe::new(
    helius_ws_url.to_string(),
    Filters::new(
        RpcBlockSubscribeFilter::MentionsAccountOrProgram(
            RAYDIUM_PROGRAM_ID.to_string(),
        ),
        None,
    ),
);

// 2. Carbon pipeline: Datasource → Decoder → Processor
Pipeline::builder()
    .datasource(datasource)
    .instruction(RaydiumDecoder, SwapProcessor::new(nats_tx))
    .build()?
    .run()
    .await;

By strictly filtering by Program Discriminators (like Raydium's program ID) and using Carbon to map byte arrays directly into structs, our CPU utilization stays incredibly low, easily fitting within our compute constraints.

Managing Backpressure & NATS Publishing

We are now successfully isolating and decoding swap events. But what happens when Solana processes a massive burst of volume in a single slot, and our internal network to NATS briefly lags for 10 milliseconds?

If our pipeline is perfectly synchronous, the WebSocket reader will block while waiting for the NATS publish to succeed. The incoming TCP buffer will instantly fill up, and Helius will aggressively disconnect our socket for being too slow.

Decoupling I/O with Bounded Channels

To prevent this, we completely decouple our I/O tasks using tokio::sync::mpsc (Multi-Producer, Single-Consumer) bounded channels.

// Bounded channel — acts as an in-memory ring buffer
let (tx, mut rx) = mpsc::channel::<DecodedSwap>(10_000);

// Task 1: Producer — WebSocket reader / Carbon pipeline
tokio::spawn(async move {
    run_carbon_pipeline(tx).await;
});

// Task 2: Consumer — NATS publisher, drains independently
tokio::spawn(async move {
    let nc = async_nats::connect("nats://localhost:4222").await.unwrap();

    while let Some(swap) = rx.recv().await {
        let payload = bincode::serialize(&swap).unwrap();

        if let Err(e) = nc.publish(
            "solana.swaps.raydium",
            payload.into(),
        ).await {
            tracing::error!("NATS publish failed: {}", e);
        }
    }
});

The bounded channel with a capacity of 10,000 acts as a shock absorber. The WebSocket reader (Producer) can instantly dump decoded swaps into the channel and immediately go back to reading the socket.

This two-task architecture is the key to operating safely on a cost-effective pod. The WebSocket reader and the NATS publisher run as independent tokio tasks on the same thread pool, coordinating through the channel without any locks, mutexes, or shared mutable state.

What's Next?

Our Rust service is now resilient, highly efficient, and pumping structured data into our message broker without dropping packets. To summarise what we have built:

• Exponential backoff reconnection loop — the pipeline self-heals from network blips with no human intervention and no data gap.
• Carbon framework with program discriminator filtering — non-swap transactions are dropped at the RPC layer in nanoseconds, keeping CPU and memory usage minimal.
• Bounded mpsc channels — the WebSocket reader and NATS publisher run as fully decoupled async tasks, absorbing traffic spikes without blocking the incoming socket.