tx-receiver SDK

tx-receiver is the Rust receiver for what Slipstream pushes over UDP: shreds and decoded transactions. It is a thin wrapper over UdpSocket with lock-free counters and Solana-MTU filtering, with a blocking API and a Tokio one. The source and examples are on GitHub at reflow-research/txReciever.

Install#

The crate is published on GitHub, not crates.io. Pull it as a git dependency:

[dependencies]
tx-receiver = { git = "https://github.com/reflow-research/txReciever.git" }

With the async API:

[dependencies]
tx-receiver = { git = "https://github.com/reflow-research/txReciever.git", features = ["tokio"] }

In production, pin a commit or tag with rev = "..." or tag = "...".

Sync API#

A blocking loop with no runtime, best when the receiver is the only work on its thread.

use tx_receiver::TxReceiver;

fn main() -> Result<(), Box<dyn std::error::Error>> {
    let mut rx = TxReceiver::bind("0.0.0.0:8003")?;

    for tx in rx.iter() {
        let tx = tx?;
        process(tx.payload()); // &[u8]
    }
    Ok(())
}

fn process(_: &[u8]) {}

The alternatives to iter():

// One datagram at a time; Ok(None) means it was oversized and skipped.
let maybe_tx = rx.recv()?;

// Zero-alloc: write into your own buffer.
let mut buf = [0u8; 1232];
if let Some((len, source)) = rx.recv_into(&mut buf)? {
    process(&buf[..len]);
}

// Bounded blocking.
rx.set_read_timeout(Some(std::time::Duration::from_millis(500)))?;

Async API#

The Tokio variant takes ownership of a non-blocking socket, with the same buffer model and stats. recv_into(&mut buf) is there too for a zero-alloc receive.

use tx_receiver::AsyncTxReceiver;

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let mut rx = AsyncTxReceiver::bind("0.0.0.0:8003").await?;

    loop {
        match rx.recv().await? {
            Some(tx) => process(tx.payload()),
            None => continue, // oversized, dropped
        }
    }
}

fn process(_: &[u8]) {}

Configuration#

Build a Config instead of calling bind() to tune the buffers or fan out.

use tx_receiver::{Config, TxReceiver};

let cfg = Config::new("0.0.0.0:8003".parse()?)
    .with_recv_buffer(32 * 1024 * 1024) // SO_RCVBUF; clamped by rmem_max
    .with_reuse_port(true)              // fan-out across multiple bound sockets
    .with_max_payload(1232);            // Solana MTU (default)

let mut rx = TxReceiver::with_config(cfg)?;

Stats#

rx.stats() returns an Arc<Stats> you can clone into a metrics thread; every accessor is atomic.

let stats = rx.stats(); // Arc<Stats>, clone freely

std::thread::spawn(move || loop {
    std::thread::sleep(std::time::Duration::from_secs(1));
    let s = stats.snapshot();
    println!(
        "received={} bytes={} oversized={} errors={}",
        s.received, s.bytes, s.oversized, s.errors,
    );
});

Errors#

Run the examples#

From the repository's root:

cargo run --release --example sync_drain -- 0.0.0.0:8003
cargo run --release --features tokio --example async_drain -- 0.0.0.0:8003

Both print packets and bytes per second, oversized and errors once a second.

OS tuning for high tx rates#

If netstat -su shows packet receive errors climbing while oversized and errors stay flat, the kernel is dropping datagrams before your loop reads them. Raise the socket buffer ceiling:

sudo sysctl -w net.core.rmem_max=134217728
sudo sysctl -w net.core.rmem_default=134217728

Then claim it with Config::with_recv_buffer(128 * 1024 * 1024). To fan out, start several receivers with reuse_port(true) bound to the same address; the kernel hashes flows across them.