JetStream

JetStream is an RPC framework designed to be a high performance, low latency, secure, and reliable RPC framework.

Transport Backends

JetStream supports multiple transport backends:

• quinn - QUIC transport with TLS/mTLS support
• iroh - P2P transport with built-in NAT traversal
• webtransport - WebTransport transport for browser and server environments

Features

• Bidirectional streaming
• 0-RTT
• mTLS
• Binary encoding
• Cross-platform (Linux, macOS, Windows, WebAssembly)
• Sessions and lanes — many independent streams per peer, or one multiplexed by tag
• Cross-language — TypeScript and Swift clients with wire-compatible codegen

For detailed API documentation, see the rustdoc documentation.

Examples

• echo - Basic QUIC-based echo service example
• iroh_echo - Echo service using iroh transport
• wasm_example - WebAssembly example
• wasm_example_bindings - WebAssembly bindings example
• session_lanes - Sessions and lanes in process, with no network under them
• quic_session - The same model over an mTLS QUIC connection

Sessions and Lanes

A session is the association with a peer. A lane is one ordered, reliable, flow-controlled sequence of frames obtained from it.

That split is the whole model. A lane is exactly what ClientTransport<P> already describes from one end and ServiceTransport<P> from the other — sessions do not change those bounds, they add a way to obtain a lane without naming the concrete transport. A session orders nothing; ordering is a property of a lane and only of a lane.

The normative rules live in the sessions and lanes specification. This page is the guide to using them.

Why

ClientTransport<P> describes one stream. Until now nothing described the thing that can open another one, and the result was uneven:

Servers accepted many streams; clients opened one and multiplexed by tag. The iroh case was the sharpest: a transport whose whole point is many independent streams plus key-based addressing was reachable only as a flat duplex.

Both strategies are legitimate and both are still supported. They differ in what they guarantee:

• Tag multiplexing — many calls over one lane. Ordered with respect to each other, and subject to head-of-line blocking: one slow response delays the bytes behind it.
• Lane multiplexing — one lane per call, or per group of calls. Unordered with respect to each other, independent, no head-of-line blocking between them.

Pick per call site. Mux has not gone anywhere, and a client that holds a single transport and multiplexes by tag keeps working unchanged.

The trait

#![allow(unused)]
fn main() {
#[async_trait::async_trait]
pub trait Session<P: Protocol>: Send + Sync {
    /// The end of a lane this peer opened, driven as the RPC caller.
    type ClientLane: ClientTransport<P>;
    /// The end of a lane the peer opened, served as the RPC callee.
    type ServiceLane: ServiceTransport<P>;

    fn capabilities(&self) -> Capabilities;
    fn context(&self) -> Context;

    async fn open_lane(&self) -> Result<Self::ClientLane, SessionError>;
    async fn accept_lane(&self) -> Result<Self::ServiceLane, SessionError>;
    async fn close(&self);
}
}

Both directions take &self, so a session is usable from several tasks at once and opening a lane never needs exclusive access.

The two lane types differ because the model is symmetric: where the transport allows it either peer may open a lane, and the opener is the RPC caller on that lane. open_lane yields the caller’s end; accept_lane yields the callee’s. A session is not fixed to a role.

In process, end to end

LocalSession implements the model with no network under it, which makes it the shortest way to see the whole thing. A lane goes straight into jetstream_rpc::server::run on one side and into a generated channel on the other — there is no adapter in between, because a lane already is a transport.

//! Sessions and lanes, without a network.
//!
//! A **session** is the association with a peer: it carries identity,
//! reports capabilities, and opens and accepts lanes. A **lane** is one
//! ordered, reliable sequence of frames — exactly what a
//! `ClientTransport` already is from one end and a `ServiceTransport`
//! from the other. A session orders nothing; ordering belongs to a lane
//! and only to a lane.
//!
//! This example uses `LocalSession`, the in-process binding, so it needs
//! no sockets and no certificates. The model is the same one the QUIC,
//! iroh and WebTransport bindings implement — see `quic_session.rs` for
//! the same shape over a real connection.
//!
//! ```console
//! cargo run --example session_lanes
//! ```

use std::time::Duration;

use jetstream::prelude::*;
use jetstream_macros::service;
use jetstream_rpc::session::{
    Capabilities, Capability, LaneSupport, LocalSession, Session, SessionError,
};
use tokio::time::sleep;

use crate::sleepy_protocol::{SleepyChannel, SleepyService};

#[service]
pub trait Sleepy {
    /// Nap for `ms` milliseconds, then say so.
    async fn nap(&self, ctx: Context, ms: u32) -> Result<String>;
}

#[derive(Debug, Clone)]
struct SleepyServer;

impl Sleepy for SleepyServer {
    async fn nap(&self, _ctx: Context, ms: u32) -> Result<String> {
        sleep(Duration::from_millis(ms as u64)).await;
        Ok(format!("slept {ms}ms"))
    }
}

#[tokio::main]
async fn main() {
    let pair = LocalSession::<SleepyChannel>::pair();

    // A session says what it can do rather than leaving callers to
    // infer it from the concrete transport type. Every binding reports
    // its own row: this one has many lanes but no datagrams, no
    // transport-level identity, and no migration.
    let caps = Session::<SleepyChannel>::capabilities(&pair.client);
    assert_eq!(caps, Capabilities::in_process());
    println!(
        "in-process session: lanes={} datagrams={} identity={}",
        caps.lanes, caps.datagrams, caps.identity
    );

    // Asking for something the session does not have fails loudly. The
    // alternative — emulating a datagram channel over a lane — would be
    // silently wrong about ordering and delivery, so it is not offered.
    match caps.require(Capability::Datagrams) {
        Err(SessionError::Unsupported(what)) => {
            println!("this session has no {what}, and says so")
        }
        other => panic!("expected the capability to be refused: {other:?}"),
    }

    // Serve every lane the peer opens. One task per lane: a lane is a
    // `ServiceTransport`, so it goes straight into the RPC loop with no
    // adapter in between.
    let server = pair.server.clone();
    let serving = tokio::spawn(async move {
        while let Ok(lane) =
            Session::<SleepyChannel>::accept_lane(&server).await
        {
            tokio::spawn(async move {
                let mut service = SleepyService {
                    inner: SleepyServer,
                };
                let _ = jetstream_rpc::server::run(&mut service, lane).await;
            });
        }
    });

    // Three lanes on the one session. `LaneSupport::Many` is what makes
    // this legal; a byte-stream session would refuse the second open
    // with `SessionError::LaneLimitReached` rather than queue behind the
    // first.
    assert_eq!(caps.lanes, LaneSupport::Many);
    let mut calls = Vec::new();
    for (lane_no, nap_ms) in [200u32, 20, 20].into_iter().enumerate() {
        let lane = Session::<SleepyChannel>::open_lane(&pair.client)
            .await
            .expect("the session is open");
        let channel = SleepyChannel::new(4, Box::new(lane));
        calls.push(tokio::spawn(async move {
            let started = std::time::Instant::now();
            let answer = channel.nap(Context::default(), nap_ms).await.unwrap();
            (lane_no, answer, started.elapsed())
        }));
    }

    // The lanes are independent sequences, so the long nap on the first
    // one does not hold up the other two.
    for call in calls {
        let (lane_no, answer, took) = call.await.unwrap();
        println!("lane {lane_no}: {answer} (round trip {took:?})");
    }

    // Closing the session ends every lane on it. Work in flight fails
    // rather than hanging, and a later open is refused outright.
    Session::<SleepyChannel>::close(&pair.client).await;
    match Session::<SleepyChannel>::open_lane(&pair.client).await {
        Err(SessionError::Closed) => {
            println!("the closed session opens no more lanes")
        }
        other => panic!("expected the session to be closed: {other:?}"),
    }

    serving.abort();
}

cargo run --example session_lanes

Over a real connection

The same code against an mTLS QUIC connection: capabilities read from the live connection, peer identity from the handshake, several independent lanes on one connection, and a datagram alongside them.

//! The same session model over a real QUIC connection.
//!
//! `session_lanes.rs` shows the model with no network under it. This one
//! stands up an mTLS QUIC connection on loopback with the certificates in
//! `certs/` and does the same things over it: reads the session's
//! capabilities, learns who the peer is from the handshake, opens several
//! independent lanes on the one connection, and sends a datagram
//! alongside them.
//!
//! The iroh binding (`jetstream_iroh::IrohSession`) is the same shape;
//! the only differences are in the row it reports — identity by public
//! key rather than by certificate.
//!
//! ```console
//! cargo run --example quic_session --features quic
//! ```

use std::{net::SocketAddr, path::Path, sync::Arc, time::Duration};

use jetstream::prelude::*;
use jetstream_macros::service;
use jetstream_quic::QuicSession;
use jetstream_rpc::{
    context::Peer,
    session::{decode_datagram, Capabilities, Capability, Datagrams, Session},
};
use rustls::pki_types::{CertificateDer, PrivateKeyDer};
use tokio::time::{sleep, timeout};

use crate::sleepy_protocol::{SleepyChannel, SleepyService, Tmessage, Tnap};

#[service]
pub trait Sleepy {
    /// Nap for `ms` milliseconds, then say so.
    async fn nap(&self, ctx: Context, ms: u32) -> Result<String>;
}

#[derive(Debug, Clone)]
struct SleepyServer;

impl Sleepy for SleepyServer {
    async fn nap(&self, _ctx: Context, ms: u32) -> Result<String> {
        sleep(Duration::from_millis(ms as u64)).await;
        Ok(format!("slept {ms}ms"))
    }
}

static CA_CERT_PEM: &str = concat!(env!("CARGO_MANIFEST_DIR"), "/certs/ca.pem");
static CLIENT_CERT_PEM: &str =
    concat!(env!("CARGO_MANIFEST_DIR"), "/certs/client.pem");
static CLIENT_KEY_PEM: &str =
    concat!(env!("CARGO_MANIFEST_DIR"), "/certs/client.key");
static SERVER_CERT_PEM: &str =
    concat!(env!("CARGO_MANIFEST_DIR"), "/certs/server.pem");
static SERVER_KEY_PEM: &str =
    concat!(env!("CARGO_MANIFEST_DIR"), "/certs/server.key");

const ALPN: &[u8] = b"jetstream-session-example";

fn load_certs(path: &str) -> Vec<CertificateDer<'static>> {
    let data = std::fs::read(Path::new(path)).expect("failed to read cert");
    rustls_pemfile::certs(&mut &*data)
        .filter_map(|cert| cert.ok())
        .collect()
}

fn load_key(path: &str) -> PrivateKeyDer<'static> {
    let data = std::fs::read(Path::new(path)).expect("failed to read key");
    rustls_pemfile::private_key(&mut &*data)
        .expect("failed to parse key")
        .expect("no key in file")
}

/// A server endpoint that requires a client certificate.
fn server_endpoint() -> (quinn::Endpoint, SocketAddr) {
    let mut roots = rustls::RootCertStore::empty();
    roots.add(load_certs(CA_CERT_PEM).pop().unwrap()).unwrap();
    let verifier =
        rustls::server::WebPkiClientVerifier::builder(Arc::new(roots))
            .build()
            .unwrap();

    let mut tls = rustls::ServerConfig::builder()
        .with_client_cert_verifier(verifier)
        .with_single_cert(load_certs(SERVER_CERT_PEM), load_key(SERVER_KEY_PEM))
        .unwrap();
    tls.alpn_protocols = vec![ALPN.to_vec()];

    let config = quinn::ServerConfig::with_crypto(Arc::new(
        quinn::crypto::rustls::QuicServerConfig::try_from(tls).unwrap(),
    ));
    let endpoint =
        quinn::Endpoint::server(config, "127.0.0.1:0".parse().unwrap())
            .unwrap();
    let addr = endpoint.local_addr().unwrap();
    (endpoint, addr)
}

/// A client endpoint that presents the client certificate.
fn client_endpoint() -> quinn::Endpoint {
    let mut roots = rustls::RootCertStore::empty();
    roots.add(load_certs(CA_CERT_PEM).pop().unwrap()).unwrap();
    let mut tls = rustls::ClientConfig::builder()
        .with_root_certificates(roots)
        .with_client_auth_cert(
            load_certs(CLIENT_CERT_PEM),
            load_key(CLIENT_KEY_PEM),
        )
        .unwrap();
    tls.alpn_protocols = vec![ALPN.to_vec()];

    let mut endpoint =
        quinn::Endpoint::client("127.0.0.1:0".parse().unwrap()).unwrap();
    endpoint.set_default_client_config(quinn::ClientConfig::new(Arc::new(
        quinn::crypto::rustls::QuicClientConfig::try_from(tls).unwrap(),
    )));
    endpoint
}

#[tokio::main]
async fn main() {
    rustls::crypto::ring::default_provider()
        .install_default()
        .ok();

    let (server_endpoint, addr) = server_endpoint();
    let client_endpoint = client_endpoint();

    let accepting = tokio::spawn(async move {
        let connection = server_endpoint.accept().await.unwrap().await.unwrap();
        // `new_owned` hands the endpoint to the session: dropping the
        // last handle on a quinn `Endpoint` closes every connection
        // opened from it, so the session is the right owner.
        QuicSession::<SleepyChannel>::new_owned(connection, server_endpoint)
    });

    let connection = client_endpoint
        .connect(addr, "localhost")
        .unwrap()
        .await
        .unwrap();
    let client =
        QuicSession::<SleepyChannel>::new_owned(connection, client_endpoint);
    let server = accepting.await.unwrap();

    // QUIC's row: many lanes, datagrams, a certificate for identity, and
    // survival across a change of network path. Datagram support is read
    // from the live connection rather than from the row, so this is what
    // the two peers actually negotiated.
    let caps = Session::<SleepyChannel>::capabilities(&client);
    assert_eq!(caps, Capabilities::quic());
    println!(
        "quic session: lanes={} datagrams={} identity={} migration={}",
        caps.lanes, caps.datagrams, caps.identity, caps.migration
    );

    // A certificate says who answered, not where to find them, so a
    // caller still has to carry the address it dialled.
    assert!(caps.identity.requires_address());
    match Session::<SleepyChannel>::context(&server).peer() {
        Some(Peer::Tls(tls)) => {
            let leaf = tls.leaf().expect("the client presented a chain");
            println!("the server authenticated {}", leaf.fingerprint);
        }
        other => panic!("expected a TLS peer, got {other:?}"),
    }

    // Serve every lane the peer opens, one task each.
    let serving = tokio::spawn({
        let server = server.clone();
        async move {
            while let Ok(lane) =
                Session::<SleepyChannel>::accept_lane(&server).await
            {
                tokio::spawn(async move {
                    let mut service = SleepyService {
                        inner: SleepyServer,
                    };
                    let _ =
                        jetstream_rpc::server::run(&mut service, lane).await;
                });
            }
        }
    });

    // Each lane is its own QUIC bidirectional stream, so the slow call on
    // the first does not hold up the others. Before sessions the iroh and
    // QUIC clients opened one stream at connect and recovered concurrency
    // by tag, which shares one ordered sequence between every call.
    let mut calls = Vec::new();
    for (lane_no, nap_ms) in [200u32, 20, 20].into_iter().enumerate() {
        let lane = Session::<SleepyChannel>::open_lane(&client).await.unwrap();
        let channel = SleepyChannel::new(4, Box::new(lane));
        calls.push(tokio::spawn(async move {
            let started = std::time::Instant::now();
            let answer = channel.nap(Context::default(), nap_ms).await.unwrap();
            (lane_no, answer, started.elapsed())
        }));
    }
    for call in calls {
        let (lane_no, answer, took) = call.await.unwrap();
        println!("lane {lane_no}: {answer} (round trip {took:?})");
    }

    // The datagram channel belongs to the session, not to any lane. It
    // is unordered and unreliable by construction, so it is deliberately
    // not a lane and none of the ordering guarantees apply to it.
    caps.require(Capability::Datagrams).unwrap();
    let limit = Datagrams::<SleepyChannel>::max_datagram_size(&client)
        .expect("the peer advertised datagram support");
    println!("datagrams up to {limit} bytes");

    // Sending names the direction: a session is not fixed to a role, so
    // the end acting as caller sends requests and the end acting as
    // callee sends responses.
    client
        .send_request_datagram(Frame {
            tag: 1,
            msg: Tmessage::Nap(Tnap { ms: 0 }),
        })
        .await
        .unwrap();

    let bytes = timeout(Duration::from_secs(5), server.recv_datagram_bytes())
        .await
        .expect("the datagram should have arrived")
        .unwrap();
    let got: Frame<Tmessage> = decode_datagram(bytes).unwrap();
    println!("datagram arrived out of band: tag {}", got.tag);

    // Closing the session ends every lane on it; work in flight fails
    // rather than hanging.
    Session::<SleepyChannel>::close(&client).await;
    serving.abort();
}

cargo run --example quic_session --features quic

jetstream_iroh::IrohSession is the same shape. The difference is the row it reports: identity by public key rather than by certificate.

Capabilities

Capabilities are reported, not inferred — not from the concrete transport type, and not reduced to a lowest common denominator. A caller picks a multiplexing strategy without knowing which transport it holds.

¹ HTTP/3 carries datagrams, but this binding has no Datagrams<P> impl, so it reports the capability absent rather than promising something a caller cannot then use.

² Unless the caller supplies one — see byte streams below.

³ The type works, and a caller running its own h3 accept loop can use it. But H3Service owns the HTTP/3 connection and routes through a private handler that calls accept_bi directly, so the session is not yet reachable from the shipped server path. Reconciling the two wants an untyped lane RpcRouter can consume, since the router negotiates a version per stream and dispatches to whichever protocol answers, while Session<P> is typed to one protocol.

Ask before you rely on something:

#![allow(unused)]
fn main() {
use jetstream_rpc::session::{Capability, Session};

// Branch on it...
if session.capabilities().supports(Capability::ManyLanes) {
    // a lane per call
} else {
    // one lane, multiplexed by tag
}

// ...or refuse to continue without it.
session.capabilities().require(Capability::Datagrams)?;
}

require fails with SessionError::Unsupported(capability). That is deliberate: the alternative — emulating a datagram channel over a lane — would be silently wrong about ordering and delivery, so it is not the easy path.

Two capabilities are read from the live connection rather than from the row. A QUIC or iroh peer that never advertised DATAGRAM support is detected on its own, and without_datagrams() lets a caller that disabled them locally say so:

#![allow(unused)]
fn main() {
let session = QuicSession::<MyChannel>::new(connection).without_datagrams();
assert!(!Session::<MyChannel>::capabilities(&session).datagrams);
}

Identity

Session::context() returns the peer in the form the transport established it, and every lane the session hands out carries the same context — a handler reading only its lane sees the same peer the session does.

IdentityKind is the model, which is what a caller has to branch on:

• IdentityKind::None — no transport-level authentication.
• IdentityKind::Certificate — a TLS peer certificate. QUIC, WebTransport.
• IdentityKind::Key — a public key the transport itself established. iroh.

IdentityKind::requires_address() is the question worth asking. Under Key the identity is the address, so a caller must not be made to carry placement information alongside it; under the other two it must.

Datagrams

The datagram channel belongs to the session, not to any lane. It is unordered and unreliable by construction, so it is deliberately not a ClientTransport — none of the lane ordering guarantees apply to it, and the type system says so.

A transport implements three raw methods — max_datagram_size, send_datagram_bytes, recv_datagram_bytes — and gets the framed ones free:

#![allow(unused)]
fn main() {
// Sending names the direction, because a session is not fixed to a role.
session.send_request_datagram(frame).await?;   // as the caller
session.send_response_datagram(frame).await?;  // as the callee

// Receiving cannot: one queue takes one reader, so the caller decodes.
let frame: Frame<MyRequest> =
    decode_datagram(session.recv_datagram_bytes().await?)?;
}

A session that reports no datagram channel refuses traffic on it rather than leaving the check to the caller: both send_datagram_bytes and recv_datagram_bytes fail with SessionError::Unsupported. Receiving is the reason — awaiting a datagram on an endpoint whose receive buffer is switched off never yields, and parking forever is the one failure mode the degradation rule exists to prevent.

A datagram carries a complete frame or it is discarded — there is nothing to reassemble it from, and trailing bytes after the frame mean it is not a complete frame either. check_datagram_size rejects a frame larger than the path allows at the sender rather than fragmenting it.

Lifetime

Closing a session ends every lane on it: calls in flight fail rather than hang, and a later open_lane is refused with SessionError::Closed. Closing a lane does not close its session.

SessionError distinguishes the cases a caller has to tell apart — Closed for a deliberate close, LaneClosed for one lane ending under a session that is still up, LaneLimitReached for a second open on a one-lane transport, Unsupported for a capability the session does not have, and Transport for an actual fault.

Byte streams

A byte-stream transport needs no new type. A TCP client is already a Framed value satisfying ClientTransport, so SingleLaneSession just wraps it and reports LaneSupport::One: the first open succeeds and every later one fails with SessionError::LaneLimitReached, rather than blocking or silently sharing the lane that is already open.

#![allow(unused)]
fn main() {
use jetstream_rpc::session::SingleLaneSession;

// A stream that knows its own peer — TCP, a unix socket.
let session = SingleLaneSession::<MyChannel, _, _>::service_io(stream);

// One that does not — TLS, stdio, anything in memory. The caller states
// the identity the framed bytes never see.
let session = SingleLaneSession::<MyChannel, _, _>::service_io_with_context(
    stream, peer_context,
);
}

Version negotiation

Version negotiation is scoped to a lane. A Tversion on one lane must not disturb its siblings — see r[jetstream.version.negotiation.reset] and the version negotiation spec.

Subscriptions

A subscription is a streaming response: one request, many responses sharing its tag, terminated explicitly.

That is the whole shape. It adds no new role — the subscriber is still the caller, the producer is still the service — and no new transport requirement: a subscription is realisable on any session JetStream supports, including one with a single lane.

The normative rules live in the subscriptions specification. This page is the guide to using them.

Why not a reverse call

The obvious way to let a service push is to make it the caller — open a lane the other way and let the service issue requests. JetStream does not, and the reason is not that it cannot: it is that the subscriber then has no correlation key to cancel by, no place to put backpressure that the producer can see, and nothing that ends. A subscription keeps all three by staying a call: the tag that identifies the request identifies the subscription for as long as it lives, and the terminator that frees the tag is the thing that ended.

The cost is that a live subscription holds a tag for its whole life. That is real, and the specification counts it.

Three things the shape has to get right

The end is a value. A subscription that reports on an operation — an upload, a build, a room closing — has a result, and a sequence that ends with the absence of an item has nowhere to put one. So the terminator is an Item::Done(_) in the sequence rather than the sequence stopping:

while let Some(item) = events.next().await {
    match item? {
        Item::Next(event) => render(event),
        Item::Done(closed) => return Ok(closed.last_seq),
    }
}

This is also what makes fan-in work. select_all yields items and drops a finished stream silently, so merging subscriptions whose end was None shows every item and no ending at all. subscription::merge labels each one, so a merged consumer can still say which ended and with what.

Cancelling reaches the work. Dropping a subscription is how a Rust caller cancels, and it must stop the producer, not merely stop reading it. Dropping sends a cancellation naming the subscription’s tag; the dispatcher cancels the token the producer holds; the producer’s next send fails even if it watches nothing else. A producer that does watch can stop between items:

tokio::select! {
    _ = producer.cancelled() => break,
    next = expensive_inference() => producer.send(next).await?,
}

A subscription does not take its lane. A room that stays open is the normal case, and a dispatcher that serves a subscription by consuming it never reads that lane again — no second request served, and no cancellation able to arrive. server::run serves the subscriptions in flight and the requests still arriving, together.

Declaring one

A subscription is a method on an ordinary #[service] trait, marked #[subscription] and returning Subscription<Item, Done>:

#[service]
pub trait Room {
    /// Unary. Unchanged by any of this.
    async fn post(&self, ctx: Context, who: String, body: String)
        -> Result<u64>;

    /// Streaming: many `Event`s share the request's tag, and the end
    /// carries a `Closed`.
    #[subscription]
    fn events(&self, ctx: Context, from: u64) -> Subscription<Event, Closed>;
}

The declaration is the protocol’s, not the call site’s: a dispatcher has to route on it before it decodes the payload, and a caller must not be able to turn a unary method into a subscription by asking differently.

A streaming method costs no per-method message id. The terminator and the cancellation take global ids below MESSAGE_ID_START, so 102 + 2 * index is untouched and no existing protocol is renumbered. That has one consequence worth knowing: because RDONE is a single id, and a decoder is handed the type byte without the tag, a terminator that carries a typed payload names its method in the payload. One byte, and a protocol can have as many subscription methods as it likes.

The method is not async, and the subscription opens when it is first read — which is what lets the signature be the plain fn it should be while acquiring a tag and sending a request stay asynchronous. It also means nothing is on the wire until something reads, so “subscribe, then act, then read” acts before the subscription exists.

establish() says “now”:

let mut events = room.events(Context::default(), from);
events.establish().await;   // the request is on the wire

That is necessary and, across lanes, not sufficient. A subscription on its own lane is unordered against a call on another — jetstream.lane.no-cross-lane-order — and nothing at the RPC layer can change that. Which is what the cursor is for: from is not decoration, it is what makes the sequence well-defined regardless of when the request arrived, and a producer that ignores it and only forwards live events will drop messages. Reliably, as it turns out: the first version of the room in this guide did exactly that, and deadlocked in process.

A room

The example is a chat room: one room, many subscribers, no transport in the protocol. It runs over a LocalSession; the same code runs over QUIC, iroh or WebTransport, because what it needs from a session is lanes and nothing else.

cargo run --example chat_room

joined room-42
posted #1
  heard: ada says is anyone there?
  heard: ada says is anyone there?
  heard: ada says is anyone there?
producers alive: 3
producers alive after one left: 2
  grace heard: grace says here
  grace's subscription closed at #2
  ada heard: grace says here
  ada's subscription closed at #2
producers alive at the end: 0

Three lines of that are the three rules above. posted #1 is answered while all three subscriptions are open. producers alive after one left: 2 is a dropped subscription stopping the work behind it, not just the delivery. And closed at #2 is a result carried out through a merge and still attributable to the subscriber that received it. (Which of the two subscribers the merge reports first is not fixed — there is no ordering between distinct subscriptions, and the specification says so.)

//! A chat room, the way a Durable Object would host one.
//!
//! One room, many subscribers, and no transport anywhere in the
//! protocol. The room is served over a `LocalSession` here; the same
//! code serves over QUIC, iroh or WebTransport, because what it needs
//! from a session is lanes and nothing else.
//!
//! A **subscription** is one request and many responses sharing its tag,
//! terminated explicitly. The three things that makes hard, and that
//! this example is really about:
//!
//! * the end is a *value* — `Item::Done(Closed { last_seq })` — so a
//!   subscription can report a result, and so the end survives a merge;
//! * cancelling reaches the *producer*, not just the delivery, so
//!   dropping a subscription stops the work behind it;
//! * a subscription does not take the lane it is served on, so `post`
//!   still works while every subscriber is listening.
//!
//! ```console
//! cargo run --example chat_room
//! ```

use std::sync::{
    atomic::{AtomicU64, AtomicUsize, Ordering::SeqCst},
    Arc,
};

use futures::StreamExt;
use jetstream::prelude::*;
use jetstream_macros::service;
use jetstream_rpc::session::{LocalSession, Session};
use tokio::sync::broadcast;

use crate::room_protocol::{RoomChannel, RoomService};

#[derive(Debug, Clone, JetStreamWireFormat)]
pub struct Event {
    pub seq: u64,
    pub who: String,
    pub body: String,
}

/// What the room says when a subscription ends normally. This is the
/// value a plain `Stream` has nowhere to put.
#[derive(Debug, Clone, JetStreamWireFormat)]
pub struct Closed {
    pub last_seq: u64,
}

#[service(uses(super::{Closed, Event}))]
pub trait Room {
    /// Unary, and unchanged by any of this.
    async fn post(
        &self,
        ctx: Context,
        who: String,
        body: String,
    ) -> Result<u64>;

    /// Streaming: every event from `from` onwards, until the subscriber
    /// leaves or the room closes.
    #[subscription]
    fn events(&self, ctx: Context, from: u64) -> Subscription<Event, Closed>;
}

#[derive(Clone)]
struct ChatRoom {
    events: broadcast::Sender<Event>,
    /// Everything said so far.
    ///
    /// r[impl jetstream.subscription.surface.producer]
    /// A producer driven by a *cursor*, not only by a live sender. This
    /// is not decoration: a subscription opens on its own lane, and
    /// `jetstream.lane.no-cross-lane-order` means it is unordered
    /// against a `post` on another one. Without a log to replay from,
    /// "subscribe, then post, then read" is a race that the in-process
    /// session loses reliably — which is how this example found it.
    log: Arc<std::sync::Mutex<Vec<Event>>>,
    seq: Arc<AtomicU64>,
    /// Cancelled when the room itself closes, which is the *other* way a
    /// subscription ends — and the one that carries a result.
    closing: subscription::CancellationToken,
    /// How many producers are alive. The example prints it to show that
    /// cancelling a subscription stops the work, rather than leaving it
    /// running with nowhere to deliver.
    producers: Arc<AtomicUsize>,
}

impl Room for ChatRoom {
    async fn post(
        &self,
        _ctx: Context,
        who: String,
        body: String,
    ) -> Result<u64> {
        let seq = self.seq.fetch_add(1, SeqCst) + 1;
        let event = Event { seq, who, body };
        self.log.lock().unwrap().push(event.clone());
        // Nobody listening is not an error: the room does not know or
        // care who is.
        let _ = self.events.send(event);
        Ok(seq)
    }

    fn events(&self, _ctx: Context, from: u64) -> Subscription<Event, Closed> {
        // Subscribe to the live feed *before* reading the log, so
        // nothing said in between is missed. Anything the replay
        // already covered is skipped by sequence below.
        let mut feed = self.events.subscribe();
        let backlog: Vec<Event> = self
            .log
            .lock()
            .unwrap()
            .iter()
            .filter(|e| e.seq >= from)
            .cloned()
            .collect();
        let seq = self.seq.clone();
        let closing = self.closing.clone();
        let producers = self.producers.clone();

        // The producer can send *and* learn that its subscriber has
        // gone — the half a bare `Sender` cannot do.
        Subscription::producing(64, move |producer| async move {
            producers.fetch_add(1, SeqCst);
            let mut sent_through = from.saturating_sub(1);
            for event in backlog {
                sent_through = event.seq;
                if producer.send(event).await.is_err() {
                    producers.fetch_sub(1, SeqCst);
                    return;
                }
            }
            let ending = loop {
                tokio::select! {
                    // Biased, and in this order, because
                    // r[jetstream.subscription.cancel] puts the
                    // terminator *after* every item already emitted.
                    // Left to chance, a subscriber can miss the last
                    // message because the room closed in the same
                    // instant — which is what happened before this line
                    // was here.
                    biased;

                    // The subscriber left. Stop the work; there is no
                    // result to report to someone who is not there, and
                    // nothing to drain for them either.
                    _ = producer.cancelled() => break None,
                    got = feed.recv() => match got {
                        Ok(event) => {
                            // Skip what the replay already covered.
                            if event.seq > sent_through {
                                sent_through = event.seq;
                                if producer.send(event).await.is_err() {
                                    break None;
                                }
                            }
                        }
                        Err(broadcast::error::RecvError::Lagged(_)) => continue,
                        Err(_) => {
                            break Some(Closed { last_seq: seq.load(SeqCst) })
                        }
                    },
                    // The room closed. That *is* a result — and it is
                    // reached only once the feed has nothing left.
                    _ = closing.cancelled() => {
                        break Some(Closed { last_seq: seq.load(SeqCst) });
                    }
                }
            };
            if let Some(closed) = ending {
                producer.finish(closed).await;
            }
            producers.fetch_sub(1, SeqCst);
        })
    }
}

/// Built over the **session**, not over a lane: a channel handed one
/// transport can only tag-multiplex, which would make the caller's
/// construction the realisation choice.
///
/// On this session — `LaneSupport::Many` — every subscription gets a
/// lane of its own, for independent flow control. On a session with one
/// lane the same code tag-multiplexes, and the caller's types do not
/// change. That is `jetstream.subscription.realisation.opaque`.
struct RoomOn {
    session: LocalSession<RoomChannel>,
    endpoint: subscription::Endpoint,
}

impl RoomOn {
    async fn lane(&self) -> Result<RoomChannel> {
        let lane = Session::<RoomChannel>::open_lane(&self.session)
            .await
            .map_err(|e| Error::new(e.to_string()))?;
        Ok(RoomChannel::new(64, Box::new(lane)))
    }
}

#[tokio::main]
async fn main() -> Result<()> {
    let pair = LocalSession::<RoomChannel>::pair();

    let room = ChatRoom {
        events: broadcast::channel(256).0,
        log: Arc::new(std::sync::Mutex::new(Vec::new())),
        seq: Arc::new(AtomicU64::new(0)),
        closing: subscription::CancellationToken::new(),
        producers: Arc::new(AtomicUsize::new(0)),
    };
    let producers = room.producers.clone();
    let closing = room.closing.clone();

    // One task per lane. A lane is a `ServiceTransport`, so it goes
    // straight into the RPC loop with no adapter in between.
    let serving = {
        let server = pair.server.clone();
        tokio::spawn(async move {
            while let Ok(lane) =
                Session::<RoomChannel>::accept_lane(&server).await
            {
                let inner = room.clone();
                tokio::spawn(async move {
                    let mut service = RoomService { inner };
                    let _ =
                        jetstream_rpc::server::run(&mut service, lane).await;
                });
            }
        })
    };

    let on = RoomOn {
        session: pair.client.clone(),
        endpoint: subscription::Endpoint::from("room-42"),
    };
    println!("joined {}", String::from_utf8_lossy(on.endpoint.as_bytes()));

    let poster = on.lane().await?;
    // Ada subscribes on the **same lane** the posts below go out on.
    // With every subscription on a lane of its own, `posted #1` would
    // print even with the dispatcher serving a subscription to
    // exhaustion — the subscription lanes would hang and the unary lane
    // would answer regardless. That is what made an earlier version of
    // this example decorative on its central claim.
    let mut ada = poster.events(Context::default(), 0);
    // Grace and the one who leaves get lanes of their own, which is the
    // other realisation and equally invisible in the types.
    let mut grace = on.lane().await?.events(Context::default(), 0);
    let mut leaving = on.lane().await?.events(Context::default(), 0);

    // A subscription opens when it is first read, so say "now" instead.
    // Necessary, and *not* sufficient: each of these is on its own lane
    // and the post below is on another, and there is no ordering
    // between lanes. That is what `from` is for, and why the room keeps
    // a log — the two together are what make this deterministic.
    for who in [&mut ada, &mut grace, &mut leaving] {
        who.establish().await;
    }

    let seq = poster
        .post(Context::default(), "ada".into(), "is anyone there?".into())
        .await?;
    println!("posted #{seq}");

    for who in [&mut ada, &mut grace, &mut leaving] {
        match who.next().await.expect("an event")? {
            Item::Next(event) => {
                println!("  heard: {} says {}", event.who, event.body)
            }
            Item::Done(_) => panic!("not yet"),
        }
    }
    settle().await;
    println!("producers alive: {}", producers.load(SeqCst));

    // Dropping a subscription cancels it, and cancelling reaches the
    // work: the producer behind it stops, rather than carrying on with
    // nowhere to deliver.
    drop(leaving);
    settle().await;
    println!("producers alive after one left: {}", producers.load(SeqCst));

    // A subscription is one call among the lane's others, so posting
    // works while every subscriber is listening.
    poster
        .post(Context::default(), "grace".into(), "here".into())
        .await?;
    // Closing the room ends every subscription with a *result*.
    closing.cancel();

    // Fan-in: merged, and still able to say which subscription ended and
    // with what. `select_all` over bare item streams would show every
    // event and no ending at all.
    let mut merged = subscription::merge([("ada", ada), ("grace", grace)]);
    while let Some(next) = merged.next().await {
        match next {
            (who, Ok(Item::Next(event))) => {
                println!("  {who} heard: {} says {}", event.who, event.body)
            }
            (who, Ok(Item::Done(closed))) => {
                println!(
                    "  {who}'s subscription closed at #{}",
                    closed.last_seq
                )
            }
            // The key survives a failure too, so a fan-in can say which
            // room went away rather than only that one did.
            (who, Err(e)) => println!("  {who} failed: {e}"),
        }
    }

    settle().await;
    println!("producers alive at the end: {}", producers.load(SeqCst));
    serving.abort();
    Ok(())
}

/// Let the spawned halves catch up. Everything here is in-process, so
/// this is a scheduling nudge rather than a timeout.
async fn settle() {
    tokio::time::sleep(std::time::Duration::from_millis(50)).await;
}

Realisation

Nothing in the caller’s type says whether a subscription got a lane of its own or a tag on a lane it shares. On a session reporting LaneSupport::Many the example opens one lane per subscription, for independent flow control; on a session with one lane the same code tag-multiplexes. That choice belongs to the session, and a caller that had to branch on it would not have had the transport abstracted, only reported.

Which is also why a channel is built over a session rather than over a lane. A channel handed one transport can only ever tag-multiplex, so the caller’s construction would have made the realisation choice.

JetStream Iroh

Iroh is a peer-to-peer networking library that provides NAT traversal, hole punching, and relay fallback. JetStream integrates with Iroh to enable peer-to-peer RPC communication.

Features

• NAT Traversal: Automatic hole punching for direct peer-to-peer connections
• Relay Fallback: Falls back to relay servers when direct connection isn’t possible
• Discovery: Uses JetStream’s discovery service at discovery.jetstream.rs
• ALPN-based Protocol Negotiation: Each service defines its own protocol version

Example

Here’s a complete example of an echo service using Iroh transport:

#![cfg(feature = "iroh")]
use crate::square_protocol::{SquareChannel, SquareService};
use futures::stream::FuturesUnordered;
use futures::StreamExt;
use jetstream::prelude::*;
use jetstream_macros::service;

#[service(tracing)]
pub trait Square {
    async fn square(&self, ctx: Context, i: u32) -> Result<String>;
}

#[derive(Debug, Clone)]
struct SquareServer;

impl Square for SquareServer {
    async fn square(&self, _ctx: Context, i: u32) -> Result<String> {
        Ok((i * i).to_string())
    }
}

#[tokio::main]
async fn main() {
    // Initialize tracing subscriber
    tracing_subscriber::fmt()
        .with_max_level(tracing::Level::DEBUG)
        .with_thread_ids(true)
        .with_span_events(
            tracing_subscriber::fmt::format::FmtSpan::ENTER
                | tracing_subscriber::fmt::format::FmtSpan::EXIT,
        )
        .init();
    // Build the server router with the echo service
    let router = jetstream_iroh::server_builder(SquareService {
        inner: SquareServer {},
    })
    .await
    .unwrap();

    // Wait for the server endpoint to be reachable via the relay before
    // publishing its address, otherwise clients can dial before the
    // relay connection is established and time out.
    router.endpoint().online().await;

    // get our own address. At this point we have a running router
    // that's ready to accept connections.
    let addr = router.endpoint().addr();

    // Build client transport and connect
    let transport = jetstream_iroh::client_builder::<SquareChannel>(addr)
        .await
        .unwrap();

    let ec = SquareChannel::new(10, Box::new(transport));
    let mut futures = FuturesUnordered::new();
    for i in 0..1000 {
        futures.push(ec.square(Context::default(), i));
    }

    while let Some(result) = futures.next().await {
        let response = result.unwrap();
        println!("Response: {}", response);
    }
    router.shutdown().await.unwrap();
}

Server Setup

To create an Iroh server, use the server_builder function:

#![allow(unused)]
fn main() {
use jetstream_iroh::server_builder;

let router = server_builder(EchoService { inner: EchoServer {} })
    .await
    .unwrap();

// Get the node address to share with clients
let addr = router.endpoint().node_addr();
}

Client Setup

To connect to an Iroh server, use the client_builder function:

#![allow(unused)]
fn main() {
use jetstream_iroh::client_builder;

let mut transport = client_builder::<EchoChannel>(addr)
    .await
    .unwrap();
}

Discovery

JetStream uses a custom discovery service for Iroh nodes. The discovery URL is:

https://discovery.jetstream.rs

This allows nodes to find each other using their public keys without needing to exchange IP addresses directly.

Sessions and Lanes

IrohSession binds an iroh connection to the session model, which is what lets a client open more than one stream: client_builder opens a single bidi stream at connect and recovers concurrency by tag, whereas a session opens a lane per call.

#![allow(unused)]
fn main() {
use jetstream_iroh::IrohSession;
use jetstream_rpc::session::{IdentityKind, Session};

let session = IrohSession::<SquareChannel>::new(connection);
let lane = Session::<SquareChannel>::open_lane(&session).await?;
}

iroh is the one transport reporting IdentityKind::Key: the peer’s public key is its address, so requires_address() is false and a caller need not carry placement information alongside the identity.

Feature Flag

To use Iroh transport, enable the iroh feature in your Cargo.toml:

[dependencies]
jetstream = { version = "10", features = ["iroh"] }

For more details, see the jetstream_iroh API documentation.

JetStream QUIC

QUIC is a modern transport protocol that provides multiplexed connections over UDP with built-in TLS 1.3 encryption. JetStream QUIC provides the transport layer using quinn.

Features

• ALPN-based Routing: Route connections to different handlers based on ALPN protocol negotiation
• TLS 1.3: Built-in secure transport with rustls
• 0-RTT: Support for zero round-trip connection resumption
• mTLS Support: Mutual TLS authentication for client certificate verification
• Peer Identity: Extract client certificate information (CN, fingerprint, SANs) in request handlers

Architecture

The crate is organized around these core components:

• Server: The main QUIC server that accepts incoming connections
• Router: Routes connections to protocol handlers based on ALPN
• ProtocolHandler: Trait for implementing custom protocol handlers
• Client: QUIC client for connecting to servers

For HTTP/3 support, see jetstream_http.

Example

Here’s a complete echo service example:

use std::{net::SocketAddr, path::Path, sync::Arc};

use echo_protocol::EchoChannel;
use jetstream::prelude::*;
use jetstream_macros::service;
use jetstream_quic::{
    Client, QuicRouter, QuicRouterHandler, QuicTransport, Server,
};

use rustls::pki_types::{CertificateDer, PrivateKeyDer};

#[service]
pub trait Echo {
    async fn ping(&self) -> Result<()>;
}

#[derive(Clone)]
struct EchoImpl {}

impl Echo for EchoImpl {
    async fn ping(&self) -> Result<()> {
        eprintln!("Ping received");
        eprintln!("Pong sent");
        Ok(())
    }
}

pub static CA_CERT_PEM: &str =
    concat!(env!("CARGO_MANIFEST_DIR"), "/certs/ca.pem");
pub static CLIENT_CERT_PEM: &str =
    concat!(env!("CARGO_MANIFEST_DIR"), "/certs/client.pem");
pub static CLIENT_KEY_PEM: &str =
    concat!(env!("CARGO_MANIFEST_DIR"), "/certs/client.key");
pub static SERVER_CERT_PEM: &str =
    concat!(env!("CARGO_MANIFEST_DIR"), "/certs/server.pem");
pub static SERVER_KEY_PEM: &str =
    concat!(env!("CARGO_MANIFEST_DIR"), "/certs/server.key");

fn load_certs(path: &str) -> Vec<CertificateDer<'static>> {
    let data = std::fs::read(Path::new(path)).expect("Failed to read cert");
    rustls_pemfile::certs(&mut &*data)
        .filter_map(|r| r.ok())
        .collect()
}

fn load_key(path: &str) -> PrivateKeyDer<'static> {
    let data = std::fs::read(Path::new(path)).expect("Failed to read key");
    rustls_pemfile::private_key(&mut &*data)
        .expect("Failed to parse key")
        .expect("No key found")
}

async fn server(
    addr: SocketAddr,
) -> std::result::Result<(), Box<dyn std::error::Error + Send + Sync>> {
    let server_cert = load_certs(SERVER_CERT_PEM).pop().unwrap();
    let server_key = load_key(SERVER_KEY_PEM);
    let ca_cert = load_certs(CA_CERT_PEM).pop().unwrap();

    let mut root_store = rustls::RootCertStore::empty();
    root_store.add(ca_cert).expect("Failed to add CA cert");
    let client_verifier =
        rustls::server::WebPkiClientVerifier::builder(Arc::new(root_store))
            .allow_unauthenticated()
            .build()
            .expect("Failed to build client verifier");

    // Register the EchoService via the per-stream RPC router
    let echo_service = echo_protocol::EchoService { inner: EchoImpl {} };

    let rpc_router = Arc::new(
        jetstream_rpc::Router::new()
            .with_handler(echo_protocol::PROTOCOL_NAME, echo_service),
    );
    let quic_handler = QuicRouterHandler::new(rpc_router);

    let mut router = QuicRouter::new();
    router.register(Arc::new(quic_handler));

    let server = Server::new_with_mtls(
        vec![server_cert],
        server_key,
        client_verifier,
        addr,
        router,
    );

    eprintln!("Server listening on {}", addr);
    server.run().await;

    Ok(())
}

async fn client(
    addr: SocketAddr,
) -> std::result::Result<(), Box<dyn std::error::Error + Send + Sync>> {
    // Wait for server to start
    tokio::time::sleep(tokio::time::Duration::from_millis(100)).await;

    let ca_cert = load_certs(CA_CERT_PEM).pop().unwrap();
    let client_cert = load_certs(CLIENT_CERT_PEM).pop().unwrap();
    let client_key = load_key(CLIENT_KEY_PEM);

    let alpn = vec![b"jetstream".to_vec()];
    let bind_addr: SocketAddr = "0.0.0.0:0".parse().unwrap();
    let client = Client::new_with_mtls(
        ca_cert,
        client_cert,
        client_key,
        alpn,
        bind_addr,
    )?;

    let connection = client.connect(addr, "localhost").await?;

    // Open a bidirectional stream and wrap it in QuicTransport
    let (send, recv) = connection.open_bi().await?;
    let transport: QuicTransport<EchoChannel> = (send, recv).into();
    let mut chan = EchoChannel::new(10, Box::new(transport));
    chan.negotiate_version(u32::MAX).await?;

    eprintln!("Ping sent");
    chan.ping().await?;
    eprintln!("Pong received");

    Ok(())
}

#[tokio::main]
async fn main() {
    // Install the ring crypto provider for rustls
    rustls::crypto::ring::default_provider()
        .install_default()
        .ok();

    let addr: SocketAddr = "127.0.0.1:4433".parse().unwrap();
    tokio::select! {
      _ = server(addr) => {},
      _ = client(addr) => {},
    }
}

Defining a Service

Use the #[service] macro to define an RPC service:

#![allow(unused)]
fn main() {
use jetstream::prelude::*;
use jetstream_macros::service;

#[service]
pub trait Echo {
    async fn ping(&mut self) -> Result<()>;
    async fn echo(&mut self, message: String) -> Result<String>;
}
}

The macro generates:

• EchoChannel - Client-side channel for calling methods
• EchoService - Server-side wrapper for your implementation

Implementing the Service

#![allow(unused)]
fn main() {
#[derive(Clone)]
struct EchoImpl;

impl Echo for EchoImpl {
    async fn ping(&mut self) -> Result<()> {
        Ok(())
    }

    async fn echo(&mut self, message: String) -> Result<String> {
        Ok(message)
    }
}
}

Server Setup

#![allow(unused)]
fn main() {
use std::sync::Arc;
use jetstream_quic::{Server, Router};

// Create the service
let echo_service = echo_protocol::EchoService { inner: EchoImpl {} };

// Register with router
let mut router = Router::new();
router.register(Arc::new(echo_service));

// Create and run server
let server = Server::new_with_addr(cert, key, addr, router);
server.run().await;
}

Client Setup

#![allow(unused)]
fn main() {
use jetstream_quic::{Client, QuicTransport};
use jetstream_rpc::Protocol;

// Create client
let alpn = vec![EchoChannel::VERSION.as_bytes().to_vec()];
let client = Client::new_with_mtls(ca_cert, client_cert, client_key, alpn)?;

// Connect
let connection = client.connect(addr, "localhost").await?;

// Open stream and create channel
let (send, recv) = connection.open_bi().await?;
let transport: QuicTransport<EchoChannel> = (send, recv).into();
let mut chan = EchoChannel::new(10, Box::new(transport));

// Call methods
chan.ping().await?;
let response = chan.echo("Hello".to_string()).await?;
}

TLS Certificates

Generate development certificates:

cd certs
./generate_certs.sh

This generates:

• ca.pem / ca.key - Certificate Authority
• server.pem / server.key - Server certificate
• client.pem / client.key - Client certificate
• client.p12 - PKCS12 bundle for browser import

For production, use certificates from a trusted CA or Let’s Encrypt.

Mutual TLS (mTLS)

JetStream QUIC supports mutual TLS authentication:

#![allow(unused)]
fn main() {
// Build a client certificate verifier from a CA cert
let mut root_store = rustls::RootCertStore::empty();
root_store.add(ca_cert).expect("Failed to add CA cert");
let client_verifier =
    rustls::server::WebPkiClientVerifier::builder(Arc::new(root_store))
        .allow_unauthenticated()
        .build()
        .expect("Failed to build client verifier");

let server = Server::new_with_mtls(
    server_cert,
    server_key,
    client_verifier,  // any Arc<dyn ClientCertVerifier>
    addr,
    router,
);
}

Accessing Peer Certificate Info

#![allow(unused)]
fn main() {
use jetstream_rpc::context::{Context, Peer};

// In your service implementation or handler
if let Some(Peer::Tls(tls_peer)) = ctx.peer() {
    if let Some(leaf) = tls_peer.leaf() {
        println!("Client CN: {:?}", leaf.common_name);
        println!("Fingerprint: {}", leaf.fingerprint);
        println!("DNS SANs: {:?}", leaf.dns_names);
    }
}
}

Sessions and Lanes

QuicSession binds a QUIC connection to the session model: one lane per bidirectional stream, the connection’s datagram channel, and the peer’s certificate for identity.

#![allow(unused)]
fn main() {
use jetstream_quic::QuicSession;
use jetstream_rpc::session::Session;

let session = QuicSession::<EchoChannel>::new(connection);

// A lane per call instead of one stream multiplexed by tag.
let lane = Session::<EchoChannel>::open_lane(&session).await?;
let channel = EchoChannel::new(4, Box::new(lane));
}

Use QuicSession::new_owned(connection, endpoint) when nothing else holds the endpoint: dropping the last handle on a quinn Endpoint closes every connection opened from it, so the session is the right owner.

A worked example is in examples/quic_session.rs.

Dependencies

Add to your Cargo.toml:

[dependencies]
jetstream = "13"
jetstream_quic = "13"
jetstream_macros = "13"
tokio = { version = "1", features = ["full"] }

For more details, see the jetstream_quic API documentation.

JetStream HTTP

JetStream HTTP provides HTTP/2 and HTTP/3 server support with a unified Axum router interface. It enables serving both protocols on the same port using TCP for HTTP/2 and UDP (QUIC) for HTTP/3.

Features

• HTTP/2 + HTTP/3: Serve both protocols simultaneously on the same port
• Axum Integration: Use familiar Axum routers and handlers
• Alt-Svc Header: Automatically advertise HTTP/3 availability to HTTP/2 clients
• Context Extractor: Access connection metadata (peer certificates, remote address) in handlers
• mTLS Support: Mutual TLS authentication for both protocols

Architecture

The crate provides these components:

• H3Service: HTTP/3 protocol handler for use with jetstream_quic
• AltSvcLayer: Tower layer that adds Alt-Svc header to advertise HTTP/3
• JetStreamContext: Axum extractor for accessing connection context

Example

Here’s a complete example serving HTTP/2 and HTTP/3:

use std::fs;
use std::net::SocketAddr;
use std::path::Path;
use std::sync::Arc;

use axum::Router;
use hyper_util::rt::{TokioExecutor, TokioIo};
use hyper_util::server::conn::auto::Builder as HttpBuilder;
use hyper_util::service::TowerToHyperService;
use jetstream::prelude::*;
use jetstream_http::{AltSvcLayer, H3Service};
use jetstream_macros::service;
use tokio::net::TcpListener;
use tokio_rustls::TlsAcceptor;
use tower_http::services::ServeDir;
use tracing::{error, info};

use rustls::pki_types::{CertificateDer, PrivateKeyDer};

// r[impl jetstream.webtransport.http-example]
#[service]
pub trait EchoHttp {
    async fn ping(&self, message: String) -> Result<String>;
    async fn add(&self, a: i32, b: i32) -> Result<i32>;
}

#[derive(Clone)]
struct EchoHttpImpl;

impl EchoHttp for EchoHttpImpl {
    async fn ping(&self, message: String) -> Result<String> {
        Ok(message)
    }
    async fn add(&self, a: i32, b: i32) -> Result<i32> {
        Ok(a + b)
    }
}

pub static CA_PEM: &str = concat!(env!("CARGO_MANIFEST_DIR"), "/certs/ca.pem");
pub static SERVER_PEM: &str =
    concat!(env!("CARGO_MANIFEST_DIR"), "/certs/server.pem");
pub static SERVER_KEY: &str =
    concat!(env!("CARGO_MANIFEST_DIR"), "/certs/server.key");

fn load_certs(path: &str) -> Vec<CertificateDer<'static>> {
    let data = fs::read(Path::new(path)).expect("Failed to read cert");
    rustls_pemfile::certs(&mut &*data)
        .filter_map(|r| r.ok())
        .collect()
}

fn load_key(path: &str) -> PrivateKeyDer<'static> {
    let data = fs::read(Path::new(path)).expect("Failed to read key");
    rustls_pemfile::private_key(&mut &*data)
        .expect("Failed to parse key")
        .expect("No key found")
}

pub static APP_DIST: &str =
    concat!(env!("CARGO_MANIFEST_DIR"), "/examples/app/dist");

/// Run HTTP/2 server with TLS
async fn run_http2_server(
    addr: SocketAddr,
    router: Router,
    tls_acceptor: TlsAcceptor,
) -> std::result::Result<(), Box<dyn std::error::Error + Send + Sync>> {
    let listener = TcpListener::bind(addr).await?;

    loop {
        let (stream, _remote_addr) = listener.accept().await?;
        let tls_acceptor = tls_acceptor.clone();
        let router = router.clone();

        tokio::spawn(async move {
            let tls_stream = match tls_acceptor.accept(stream).await {
                Ok(s) => s,
                Err(e) => {
                    error!("TLS accept error: {}", e);
                    return;
                }
            };

            let io = TokioIo::new(tls_stream);
            let svc = TowerToHyperService::new(router);

            if let Err(e) = HttpBuilder::new(TokioExecutor::new())
                .serve_connection(io, svc)
                .await
            {
                error!("HTTP/2 connection error: {}", e);
            }
        });
    }
}

/// Run HTTP/3 server with QUIC + WebTransport
async fn run_http3_server(
    addr: SocketAddr,
    router: Router,
    server_cert: CertificateDer<'static>,
    server_key: PrivateKeyDer<'static>,
    ca_cert: Option<CertificateDer<'static>>,
) {
    // Register the EchoHttp service as a WebTransport handler
    let echo = echohttp_protocol::EchoHttpService {
        inner: EchoHttpImpl,
    };
    let rpc_router = Arc::new(
        jetstream_rpc::Router::new()
            .with_handler(echohttp_protocol::PROTOCOL_NAME, echo),
    );

    let mut quic_router = jetstream_quic::QuicRouter::new();

    let server = if let Some(ca) = ca_cert {
        let mut root_store = rustls::RootCertStore::empty();
        root_store.add(ca).expect("Failed to add CA cert");
        let client_verifier =
            rustls::server::WebPkiClientVerifier::builder(Arc::new(root_store))
                .allow_unauthenticated()
                .build()
                .expect("Failed to build client verifier");

        let h3_service = Arc::new(H3Service::new_with_cert_verifier(
            router,
            rpc_router,
            client_verifier.clone(),
        ));
        quic_router.register(h3_service);

        jetstream_quic::Server::new_with_mtls(
            vec![server_cert],
            server_key,
            client_verifier,
            addr,
            quic_router,
        )
    } else {
        let h3_service = Arc::new(H3Service::new(router, rpc_router));
        quic_router.register(h3_service);

        jetstream_quic::Server::new_with_addr(
            vec![server_cert],
            server_key,
            addr,
            quic_router,
        )
    };

    server.run().await;
}

#[tokio::main]
async fn main() -> std::result::Result<(), Box<dyn std::error::Error>> {
    // Initialize tracing
    tracing_subscriber::fmt::init();

    // Install the ring crypto provider for rustls
    rustls::crypto::ring::default_provider()
        .install_default()
        .ok();

    // Check for --mtls flag
    let mtls_enabled = std::env::args().any(|arg| arg == "--mtls");

    let server_certs = load_certs(SERVER_PEM);
    let server_key = load_key(SERVER_KEY);
    let ca_cert = if mtls_enabled {
        Some(load_certs(CA_PEM).pop().unwrap())
    } else {
        None
    };

    // Create shared Axum router serving React app build + Alt-Svc header
    let router = Router::new()
        .fallback_service(ServeDir::new(APP_DIST))
        .layer(AltSvcLayer::new(4433));

    // Setup TLS config for HTTP/2
    let mut tls_config = rustls::ServerConfig::builder()
        .with_no_client_auth()
        .with_single_cert(server_certs.clone(), server_key.clone_key())?;
    tls_config.alpn_protocols = vec![b"h2".to_vec(), b"http/1.1".to_vec()];
    let tls_acceptor = TlsAcceptor::from(Arc::new(tls_config));

    let addr: SocketAddr = "127.0.0.1:4433".parse()?;

    info!("=== JetStream HTTP + WebTransport Server ===");
    info!("Listening on https://{}", addr);
    info!("  - HTTP/2 over TLS (TCP)");
    info!("  - HTTP/3 + WebTransport over QUIC (UDP)");
    info!(
        "  - WebTransport protocol: {}",
        echohttp_protocol::PROTOCOL_VERSION
    );
    if mtls_enabled {
        info!("mTLS enabled - client certificates required");
    } else {
        info!("No client auth (use --mtls to enable)");
    }

    // Run both servers concurrently on the same port (TCP for HTTP/2, UDP for HTTP/3)
    tokio::select! {
        result = run_http2_server(addr, router.clone(), tls_acceptor) => {
            if let Err(e) = result {
                error!("HTTP/2 server error: {}", e);
            }
        }
        _ = run_http3_server(
            addr,
            router,
            server_certs.into_iter().next().unwrap(),
            server_key,
            ca_cert,
        ) => {}
    }

    Ok(())
}

Quick Start

1. Create an Axum Router

#![allow(unused)]
fn main() {
use axum::{routing::get, Router};
use jetstream_http::{AltSvcLayer, JetStreamContext};

async fn handler(ctx: JetStreamContext) -> &'static str {
    println!("Request from: {:?}", ctx.remote());
    "Hello, World!"
}

let router = Router::new()
    .route("/", get(handler))
    .layer(AltSvcLayer::new(4433));
}

2. Set Up HTTP/2 Server (TCP + TLS)

#![allow(unused)]
fn main() {
use hyper_util::rt::{TokioExecutor, TokioIo};
use hyper_util::server::conn::auto::Builder as HttpBuilder;
use hyper_util::service::TowerToHyperService;
use tokio::net::TcpListener;
use tokio_rustls::TlsAcceptor;

let listener = TcpListener::bind("127.0.0.1:4433").await?;
let tls_acceptor = TlsAcceptor::from(Arc::new(tls_config));

loop {
    let (stream, _) = listener.accept().await?;
    let tls_stream = tls_acceptor.accept(stream).await?;
    let io = TokioIo::new(tls_stream);
    let svc = TowerToHyperService::new(router.clone());
    
    HttpBuilder::new(TokioExecutor::new())
        .serve_connection(io, svc)
        .await?;
}
}

3. Set Up HTTP/3 Server (QUIC)

#![allow(unused)]
fn main() {
use jetstream_http::H3Service;
use jetstream_quic::{Router as QuicRouter, Server};

let h3_service = Arc::new(H3Service::new(router));

let mut quic_router = QuicRouter::new();
quic_router.register(h3_service);

let server = Server::new_with_addr(cert, key, addr, quic_router);
server.run().await;
}

Alt-Svc Layer

The AltSvcLayer adds the Alt-Svc header to HTTP/2 responses, telling browsers that HTTP/3 is available:

#![allow(unused)]
fn main() {
use jetstream_http::AltSvcLayer;

// Advertise HTTP/3 on port 4433 with 24-hour max-age
let layer = AltSvcLayer::new(4433);
// Adds: Alt-Svc: h3=":4433"; ma=86400

// Or with custom value
let layer = AltSvcLayer::with_value("h3=\":443\"; ma=3600, h3-29=\":443\"");
}

When browsers receive this header over HTTP/2, they will attempt to upgrade to HTTP/3 on subsequent requests.

Context Extractor

Use JetStreamContext to access connection metadata in your handlers:

#![allow(unused)]
fn main() {
use jetstream_http::JetStreamContext;
use jetstream_rpc::context::{Peer, RemoteAddr};

async fn handler(ctx: JetStreamContext) -> String {
    // Get remote address
    let addr = match ctx.remote() {
        Some(RemoteAddr::IpAddr(ip)) => ip.to_string(),
        _ => "unknown".to_string(),
    };

    // Get peer certificate info (for mTLS)
    if let Some(Peer::Tls(tls_peer)) = ctx.peer() {
        if let Some(leaf) = tls_peer.leaf() {
            return format!(
                "Hello {}! (CN: {:?})",
                addr,
                leaf.common_name
            );
        }
    }

    format!("Hello {}!", addr)
}
}

TLS Configuration

HTTP/2 TLS Config

#![allow(unused)]
fn main() {
let mut tls_config = rustls::ServerConfig::builder()
    .with_no_client_auth()
    .with_single_cert(certs, key)?;

// Important: set ALPN for HTTP/2
tls_config.alpn_protocols = vec![b"h2".to_vec(), b"http/1.1".to_vec()];
}

HTTP/3 (QUIC) TLS

QUIC/HTTP/3 TLS is handled automatically by jetstream_quic::Server.

Certificates

Generate development certificates:

cd certs
./generate_certs.sh

Files created:

• server.pem / server.key - Server certificate
• ca.pem - CA certificate (for client trust)
• client.pem / client.key - Client certificate (for mTLS)
• client.p12 - PKCS12 bundle for browser import
• server.crt - DER format for Chrome

Testing with Chrome

Chrome supports HTTP/3. To test:

# Start the server
cargo run --example http --features quic

# Launch Chrome with QUIC enabled
./examples/launch_chrome.sh

The launch_chrome.sh script:

1. Extracts the server’s SPKI hash
2. Launches Chrome with --origin-to-force-quic-on flag
3. Provides instructions for importing certificates

Sessions and Lanes

WebTransportSession binds a WebTransport session to the session model — a lane per bidirectional stream, with the client certificate for identity where one was presented.

It reports no datagram capability. HTTP/3 carries datagrams, but this binding has no Datagrams<P> impl yet, and reporting a capability a caller then cannot use is worse than reporting its absence.

The type is also not yet reachable from H3Service, which owns the HTTP/3 connection and routes through its own handler; a caller running its own h3 accept loop can use it today.

Dependencies

Add to your Cargo.toml:

[dependencies]
jetstream_http = "0.1"
jetstream_quic = "13"
axum = "0.8"
hyper-util = { version = "0.1", features = ["full"] }
tokio-rustls = "0.26"
tower = "0.5"

For more details, see the jetstream_http API documentation.

TypeScript

JetStream provides TypeScript packages for wire format encoding and RPC communication, published to the GitHub Package Registry under @sevki.

Installation

# Configure npm to use GitHub Package Registry for @sevki scope
echo "@sevki:registry=https://npm.pkg.github.com" >> .npmrc

pnpm add @sevki/jetstream-wireformat @sevki/jetstream-rpc

WireFormat Codecs

The @sevki/jetstream-wireformat package provides binary codecs for all JetStream primitive and composite types. Every codec implements the WireFormat<T> interface:

interface WireFormat<T> {
  byteSize(value: T): number;
  encode(value: T, writer: BinaryWriter): void;
  decode(reader: BinaryReader): T;
}

Primitives

import { BinaryReader, BinaryWriter, u32Codec, stringCodec, boolCodec } from '@sevki/jetstream-wireformat';

// Encode a u32
const writer = new BinaryWriter();
u32Codec.encode(42, writer);
const bytes = writer.toUint8Array();

// Decode it back
const reader = new BinaryReader(bytes);
const value = u32Codec.decode(reader); // 42

Available primitive codecs: u8Codec, u16Codec, u32Codec, u64Codec, i16Codec, i32Codec, i64Codec, f32Codec, f64Codec, boolCodec, stringCodec.

Composite Types

Build codecs for collections and optional types:

import { vecCodec, optionCodec, mapCodec, stringCodec, u32Codec } from '@sevki/jetstream-wireformat';

// Vec<String>
const tagsCodec = vecCodec(stringCodec);

// Option<u32>
const maybeIdCodec = optionCodec(u32Codec);

// HashMap<String, u32>
const scoresCodec = mapCodec(stringCodec, u32Codec);

Structs and Enums

Use the structCodec and enumCodec helpers, or generate codecs from Rust types using jetstream_codegen:

import { BinaryReader, BinaryWriter, u32Codec } from '@sevki/jetstream-wireformat';
import type { WireFormat } from '@sevki/jetstream-wireformat';

// A manually defined codec for a Point struct
interface Point {
  x: number;
  y: number;
}

const pointCodec: WireFormat<Point> = {
  byteSize(value: Point): number {
    return u32Codec.byteSize(value.x) + u32Codec.byteSize(value.y);
  },
  encode(value: Point, writer: BinaryWriter): void {
    u32Codec.encode(value.x, writer);
    u32Codec.encode(value.y, writer);
  },
  decode(reader: BinaryReader): Point {
    const x = u32Codec.decode(reader);
    const y = u32Codec.decode(reader);
    return { x, y };
  },
};

Code Generation

Instead of writing codecs by hand, use jetstream_codegen to generate TypeScript types and codecs from Rust source files:

cargo run -p jetstream_codegen -- \
  --input src/types.rs \
  --ts-out generated/

Given a Rust file:

#![allow(unused)]
fn main() {
#[derive(JetStreamWireFormat)]
pub struct Point {
    pub x: u32,
    pub y: u32,
}

#[derive(JetStreamWireFormat)]
pub enum Shape {
    Circle(u32),
    Rectangle { width: u32, height: u32 },
}
}

The codegen produces TypeScript interfaces and WireFormat<T> codec objects that are wire-compatible with the Rust implementations.

RPC

The @sevki/jetstream-rpc package provides the RPC runtime for multiplexed request/response communication.

Generated Code

The codegen generates RPC client and handler types from #[service] trait definitions:

#![allow(unused)]
fn main() {
// Rust service definition
#[service]
pub trait EchoHttp {
    async fn ping(&mut self, message: String) -> Result<String>;
    async fn add(&mut self, a: i32, b: i32) -> Result<i32>;
}
}

cargo run -p jetstream_codegen -- \
  --input examples/http.rs \
  --ts-out generated/

This generates:

• Request/response types: TPing, RPing, TAdd, RAdd with codecs
• Frame unions: Tmessage, Rmessage discriminated unions with FramerCodec implementations
• Framer wrappers: TmessageFramer, RmessageFramer classes implementing the Framer interface
• rmessageDecode: A decoder function for use with Mux and WebTransportTransport
• Protocol constants: PROTOCOL_NAME (e.g., 'rs.jetstream.proto/echohttp') and PROTOCOL_VERSION (e.g., 'rs.jetstream.proto/echohttp/15.0.0+bfd7d20e')
• EchoHttpClient: A typed client class with async methods and version negotiation
• EchoHttpHandler: A handler interface for implementing server-side dispatch
• dispatchEchoHttp: A dispatch function that routes Tmessage frames to handler methods

Version Negotiation

Before making RPC calls, clients must perform a Tversion/Rversion handshake to negotiate the protocol version and maximum message size. The generated client provides a static negotiate method:

import { EchoHttpClient, rmessageDecode, PROTOCOL_NAME } from './generated/echohttp_rpc.js';

// Open a WebTransport session and bidi stream
const session = new WebTransport(`https://api.example.com:4433/${PROTOCOL_NAME}`);
await session.ready;
const stream = await session.createBidirectionalStream();

// Negotiate version on the raw stream before creating the Mux
const negotiated = await EchoHttpClient.negotiate(stream.readable, stream.writable);
console.log(`Negotiated: ${negotiated.version}, msize: ${negotiated.msize}`);

You can also call negotiateVersion directly from @sevki/jetstream-rpc:

import { negotiateVersion } from '@sevki/jetstream-rpc';
import { PROTOCOL_VERSION } from './generated/echohttp_rpc.js';

const negotiated = await negotiateVersion(stream.readable, stream.writable, PROTOCOL_VERSION);

After negotiation, the stream is ready for Mux framing.

Client Usage

import { Mux } from '@sevki/jetstream-rpc';
import { EchoHttpClient, rmessageDecode, PROTOCOL_NAME } from './generated/echohttp_rpc.js';

// 1. Connect
const session = new WebTransport(`https://api.example.com:4433/${PROTOCOL_NAME}`);
await session.ready;
const stream = await session.createBidirectionalStream();

// 2. Negotiate version
await EchoHttpClient.negotiate(stream.readable, stream.writable);

// 3. Create transport and mux
const transport = new WebTransportTransport(stream, rmessageDecode);
const mux = new Mux(transport);
await mux.start();

// 4. Create client and make RPC calls
const client = new EchoHttpClient(mux);
const reply = await client.ping("hello");  // "hello"
const sum = await client.add(2, 3);        // 5

// 5. Cleanup
await mux.close();
session.close();

Handler (Server-Side)

Implement the generated handler interface to serve RPCs:

import { EchoHttpHandler, dispatchEchoHttp } from './generated/echohttp_rpc.js';

const handler: EchoHttpHandler = {
  async ping(ctx, message) {
    return message; // echo back
  },
  async add(ctx, a, b) {
    return a + b;
  },
};

Frame Wire Format

RPC frames follow the format [size:u32 LE][type:u8][tag:u16 LE][payload] where size includes itself (minimum 7 bytes). The tag field enables multiplexing concurrent requests over a single connection.

Special message types:

• TVERSION (100) / RVERSION (101) — version negotiation
• MESSAGE_ID_START (102) — first service method ID
• RJETSTREAMERROR (5) — error response frames

Mux

The Mux class handles tag allocation, request/response matching, and concurrent RPC dispatch:

import { Mux } from '@sevki/jetstream-rpc';

const mux = new Mux(transport);
await mux.start();

// Each rpc() call acquires a tag, sends a frame, waits for the matching response, and releases the tag
const response = await mux.rpc(request);

await mux.close();

Protocol Interface

Every generated service implements the Protocol interface:

interface Protocol<TReq extends Framer, TRes extends Framer> {
  readonly VERSION: string;
  readonly NAME: string;
}

• NAME is the protocol name used for routing (e.g., URI path, ALPN)
• VERSION is the full version string used during Tversion/Rversion negotiation

React

@sevki/jetstream-react provides React hooks for bidirectional RPC over WebTransport. Downstream (browser) can call upstream services, and upstream can push RPCs downstream.

Installation

pnpm add @sevki/jetstream-react @sevki/jetstream-rpc @sevki/jetstream-wireformat

react (>=18) is a peer dependency.

Quick Start

import { JetStreamProvider, useJetStream, useJetStreamStatus, useRPC } from '@sevki/jetstream-react';
import { EchoHttpClient, rmessageDecode, PROTOCOL_VERSION, PROTOCOL_NAME } from './generated/echohttp_rpc.js';

function App() {
  return (
    <JetStreamProvider url={`https://api.example.com:4433/${PROTOCOL_NAME}`}>
      <EchoDemo />
    </JetStreamProvider>
  );
}

function EchoDemo() {
  const status = useJetStreamStatus();
  const echo = useJetStream(EchoHttpClient, rmessageDecode, PROTOCOL_VERSION);

  const { data, error, isLoading } = useRPC(
    () => (echo ? echo.ping('hello') : Promise.resolve('')),
    [echo],
  );

  return (
    <div>
      <p>Connection: {status}</p>
      {isLoading && <p>Loading...</p>}
      {error && <p>Error: {error.message}</p>}
      {data && <p>Echo: {data}</p>}
    </div>
  );
}

Provider

JetStreamProvider manages the WebTransport session lifecycle. It establishes the session on mount and closes it on unmount.

<JetStreamProvider url="https://api.example.com:4433/rs.jetstream.proto/echohttp">
  <App />
</JetStreamProvider>

Props:

• url (required) — the upstream WebTransport URL, typically https://host:port/{PROTOCOL_NAME}
• children — React children

The provider exposes connection state and protocol version through context:

• session — the WebTransport instance (or null before connected)
• state — 'connecting' | 'connected' | 'disconnected' | 'error'
• protocolVersion — the negotiated protocol version string (or null before negotiation)

useJetStreamStatus

Read the current connection state:

function StatusIndicator() {
  const status = useJetStreamStatus();
  return <span className={status}>{status}</span>;
}

useJetStream

Creates a memoized RPC client for calling upstream services. Handles stream creation, version negotiation, and Mux setup automatically.

const client = useJetStream(ClientClass, responseDecode, protocolVersion);

Parameters:

• ClientClass — the generated client constructor (e.g., EchoHttpClient)
• responseDecode — the generated response decoder function (e.g., rmessageDecode)
• protocolVersion — the protocol version string for Tversion negotiation (e.g., PROTOCOL_VERSION)

Returns: the client instance (or null while connecting).

The hook:

1. Opens a bidirectional stream on the WebTransport session
2. Performs Tversion/Rversion negotiation on the raw stream
3. Stores the negotiated version in the provider context
4. Creates a WebTransportTransport and Mux over the stream
5. Constructs and returns the client

The returned client is stable across re-renders as long as the session and constructor remain the same.

import { EchoHttpClient, rmessageDecode, PROTOCOL_VERSION } from './generated/echohttp_rpc.js';

function MyComponent() {
  const echo = useJetStream(EchoHttpClient, rmessageDecode, PROTOCOL_VERSION);

  if (!echo) return <p>Connecting...</p>;

  return <button onClick={() => echo.ping('hi')}>Ping</button>;
}

useRPC

Reactive wrapper around a single RPC call. Re-executes when dependencies change.

const { data, error, isLoading, refetch } = useRPC(fn, deps);

Parameters:

• fn — a function returning a Promise<T> (the RPC call)
• deps — dependency array (like useEffect); the call re-runs when dependencies change

Returns:

• data: T | undefined — the resolved value
• error: Error | undefined — the rejection reason
• isLoading: boolean — whether a request is in flight
• refetch: () => void — manually re-execute the call

When dependencies change, the previous in-flight request is discarded (stale results are never applied).

function SearchResults() {
  const [query, setQuery] = useState('');
  const search = useJetStream(SearchClient, rmessageDecode, PROTOCOL_VERSION);

  const { data, error, isLoading } = useRPC(
    () => (search ? search.find(query) : Promise.resolve([])),
    [search, query],
  );

  return (
    <div>
      <input value={query} onChange={e => setQuery(e.target.value)} />
      {isLoading && <p>Loading...</p>}
      {error && <p>Error: {error.message}</p>}
      {data?.map(item => <div key={item.id}>{item.name}</div>)}
    </div>
  );
}

useHandler

Registers a handler for incoming upstream-initiated RPCs (push notifications, cache invalidation, etc.).

import { NotificationHandler } from './generated/notification_rpc.js';

function NotificationBadge() {
  const { events, error } = useHandler(NotificationHandler, {
    async notify(ctx, title, body) {
      return { ack: true };
    },
  });

  const unread = events.filter(e => e.method === 'notify').length;
  return <span>{unread > 0 && `${unread} new`}</span>;
}

The handler is registered on mount and unregistered on unmount. Incoming bidirectional streams from upstream are dispatched to the matching handler.

Imperative Calls and Mutations

For imperative operations (mutations, fire-and-forget), call the client directly. The generated client methods return plain Promises:

function AddButton() {
  const echo = useJetStream(EchoHttpClient, rmessageDecode, PROTOCOL_VERSION);
  const [sum, setSum] = useState<number>();

  return (
    <button
      disabled={!echo}
      onClick={async () => {
        if (echo) setSum(await echo.add(2, 3));
      }}
    >
      Add 2 + 3 {sum !== undefined && `= ${sum}`}
    </button>
  );
}

For richer mutation state (loading, error, retries, cache invalidation), use TanStack Query:

import { useMutation } from '@tanstack/react-query';

function AddForm() {
  const echo = useJetStream(EchoHttpClient, rmessageDecode, PROTOCOL_VERSION);

  const { mutate, data, isPending } = useMutation({
    mutationFn: ({ a, b }: { a: number; b: number }) => echo!.add(a, b),
  });

  return (
    <div>
      <button onClick={() => mutate({ a: 2, b: 3 })} disabled={isPending || !echo}>
        Add 2 + 3
      </button>
      {data !== undefined && <p>Sum: {data}</p>}
    </div>
  );
}

Full Example

import { useState } from 'react';
import {
  JetStreamProvider,
  useJetStream,
  useJetStreamStatus,
  useRPC,
} from '@sevki/jetstream-react';
import {
  EchoHttpClient,
  rmessageDecode,
  PROTOCOL_VERSION,
  PROTOCOL_NAME,
} from './generated/echohttp_rpc.js';

const SERVER_URL = `https://127.0.0.1:4433/${PROTOCOL_NAME}`;

function EchoDemo() {
  const status = useJetStreamStatus();
  const echo = useJetStream(EchoHttpClient, rmessageDecode, PROTOCOL_VERSION);
  const [message, setMessage] = useState('hello');
  const [sum, setSum] = useState<number | null>(null);

  const { data, error, isLoading } = useRPC(
    () => (echo ? echo.ping(message) : Promise.resolve('')),
    [echo, message],
  );

  return (
    <div>
      <h1>JetStream Echo Demo</h1>
      <p>Connection: {status}</p>
      <p>Protocol: {PROTOCOL_VERSION}</p>

      <h2>Ping</h2>
      <input
        value={message}
        onChange={e => setMessage(e.target.value)}
        placeholder="Type a message..."
      />
      {isLoading && <p>Loading...</p>}
      {error && <p>Error: {error.message}</p>}
      {data && <p>Echo: {data}</p>}

      <h2>Add</h2>
      <button
        disabled={!echo}
        onClick={async () => {
          if (echo) setSum(await echo.add(2, 3));
        }}
      >
        Add 2 + 3
      </button>
      {sum !== null && <p>Sum: {sum}</p>}
    </div>
  );
}

export default function App() {
  return (
    <JetStreamProvider url={SERVER_URL}>
      <EchoDemo />
    </JetStreamProvider>
  );
}

Package Structure

Swift

JetStream provides Swift packages for wire format encoding and RPC communication via Swift Package Manager.

Installation

Add the JetStream package to your Package.swift:

dependencies: [
    .package(url: "https://github.com/sevki/jetstream.git", from: "15.0.0"),
],
targets: [
    .target(
        name: "MyApp",
        dependencies: [
            .product(name: "JetStreamWireFormat", package: "jetstream"),
            .product(name: "JetStreamRpc", package: "jetstream"),
        ]
    ),
]

WireFormat Protocol

The JetStreamWireFormat module defines the WireFormat protocol that all serializable types conform to:

public protocol WireFormat: Sized {
    func byteSize() -> UInt32
    func encode(writer: inout BinaryWriter) throws
    static func decode(reader: inout BinaryReader) throws -> Self
}

Primitives

All Swift standard types (UInt8, UInt16, UInt32, UInt64, Int16, Int32, Int64, Float, Double, Bool, String) conform to WireFormat:

import JetStreamWireFormat

// Encode a UInt32
var writer = BinaryWriter()
try UInt32(42).encode(writer: &writer)
let bytes = writer.bytes

// Decode it back
var reader = BinaryReader(data: bytes)
let value = try UInt32.decode(reader: &reader) // 42

Collections and Optionals

Arrays, Dictionaries, Sets, and Optionals conform to WireFormat when their elements do:

// Array<String>
let tags: [String] = ["hello", "world"]
try tags.encode(writer: &writer)

// Optional<UInt32>
let maybeId: UInt32? = 42
try maybeId.encode(writer: &writer)

// Dictionary<String, UInt32>
let scores: [String: UInt32] = ["alice": 100, "bob": 200]
try scores.encode(writer: &writer)

Structs and Enums

Define custom types conforming to WireFormat, or generate them from Rust types using jetstream_codegen:

import JetStreamWireFormat

public struct Point: WireFormat {
    public var x: UInt32
    public var y: UInt32

    public func byteSize() -> UInt32 {
        return x.byteSize() + y.byteSize()
    }

    public func encode(writer: inout BinaryWriter) throws {
        try x.encode(writer: &writer)
        try y.encode(writer: &writer)
    }

    public static func decode(reader: inout BinaryReader) throws -> Point {
        let x = try UInt32.decode(reader: &reader)
        let y = try UInt32.decode(reader: &reader)
        return Point(x: x, y: y)
    }
}

Code Generation

Instead of writing conformances by hand, use jetstream_codegen to generate Swift types from Rust source files:

cargo run -p jetstream_codegen -- \
  --input src/types.rs \
  --swift-out generated/

Given a Rust file:

#![allow(unused)]
fn main() {
#[derive(JetStreamWireFormat)]
pub struct Point {
    pub x: u32,
    pub y: u32,
}

#[derive(JetStreamWireFormat)]
pub enum Shape {
    Circle(u32),
    Rectangle { width: u32, height: u32 },
}
}

The codegen produces Swift structs and enums conforming to WireFormat that are wire-compatible with the Rust implementations.

RPC

The JetStreamRpc module provides the RPC runtime for multiplexed request/response communication.

Generated Client

The codegen generates RPC client classes and handler protocols from #[service] trait definitions:

#![allow(unused)]
fn main() {
// Rust service definition
#[service]
pub trait Echo {
    async fn ping(&mut self, message: String) -> Result<String>;
    async fn add(&mut self, a: u32, b: u32) -> Result<u32>;
}
}

cargo run -p jetstream_codegen -- \
  --input examples/echo.rs \
  --swift-out generated/

This generates a typed EchoClient class and an EchoHandler protocol:

// Generated client usage
let client = EchoClient(mux: mux)
let reply = try await client.ping(message: "hello")
let sum = try await client.add(a: 2, b: 3)

// Generated handler protocol — implement this for your server
public protocol EchoHandler {
    func ping(message: String) async throws -> String
    func add(a: UInt32, b: UInt32) async throws -> UInt32
}

Framer Protocol

The Framer protocol handles message type dispatch for RPC envelopes:

public protocol Framer {
    func messageType() -> UInt8
    func byteSize() -> UInt32
    func encode(writer: inout BinaryWriter) throws
    static func decode(reader: inout BinaryReader, type: UInt8) throws -> Self
}

Generated Tmessage and Rmessage enums conform to Framer and dispatch encoding/decoding based on the message type byte.

Frame Wire Format

RPC frames follow the format [size:u32 LE][type:u8][tag:u16 LE][payload] where size includes itself (minimum 7 bytes). The tag field enables multiplexing concurrent requests over a single connection.

Mux

The Mux class handles tag allocation, request/response matching, and concurrent RPC dispatch:

let mux = Mux(transport: transport)

// Each rpc() call acquires a tag, sends a frame, waits for the matching response, and releases the tag
let response = try await mux.rpc(request)

🦀 Crates

For detailed API documentation of all crates in this repository, please see the rustdoc documentation.

The main crates include:

• jetstream - The main crate with the jetstream::prelude module
• jetstream_9p - 9P protocol implementation
• jetstream_error - Error types and handling
• jetstream_iroh - Iroh peer-to-peer transport
• jetstream_libc - libc bindings
• jetstream_macros - Procedural macros
• jetstream_quic - QUIC/HTTP3 transport
• jetstream_rpc - RPC framework core
• jetstream_ufs - UFS (Unix File System) implementation
• jetstream_websocket - WebSocket transport
• jetstream_wireformat - Encoding and decoding logic

📦 jetstream

📦 jetstream_9p

📦 jetstream_error

📦 jetstream_http

📦 jetstream_iroh

📦 jetstream_libc

📦 jetstream_macros

📦 jetstream_quic

📦 jetstream_rpc

📦 jetstream_ufs

📦 jetstream_wireformat

Changelog

16.3.1 (2026-09-08)

Bug Fixes

• ufs: declare that jetstream_ufs is Linux-only (31a5166)
• ufs: declare that jetstream_ufs is Linux-only (58da1ce)

16.3.0 (2026-09-02)

Features

• macros: #[subscription], and a chat room the macro writes (e45b5db)
• rpc: route streaming responses, and stop panicking on stray tags (2626f8f)
• rpc: say “now” — Subscription::establish (e70c302)
• rpc: serve subscriptions, with cancellation reaching the producer (360ca97)
• rpc: the typed surface, and a chat room that runs on it (53e9182)
• rpc: the wire pieces subscriptions need (26d066d)

Bug Fixes

• example: the room’s terminator takes the method it ends (95b6947)
• macros: a failure ends the subscription, and a served one keeps its span (492e808)
• macros: three from codex on #699 (047dd37)
• rpc: a merged failure keeps its key, and the example earns its output (2ba5407)
• rpc: a subscription must not take the lane it is served on (ae611c5)
• rpc: a subscription that fails says so, and then stops, from codex on #698 (50e3706)
• rpc: acknowledge every cancellation, not only the last, from codex on #697 (2b51c89)
• rpc: drop a mut the unfold made redundant (7c8260e)
• rpc: four more ways the client lost a subscription, from codex on #696 (be75f6f)
• rpc: four ways the client failed a subscription, from codex on #696 (08f7838)
• rpc: four ways the dispatcher failed, from codex on #697 (a2d7a91)
• rpc: opening a subscription keeps the means to cancel it, from codex on #699 (7bbd271)
• rpc: serving an opened subscription must not cancel what it serves (4e536de)
• rpc: the default build has no select! (4bc6d52)
• rpc: the supplied terminator has to know which method it ends (3cd5334)
• rpc: zero is a reserved binding, not a plausible one (f078ad5)

16.2.0 (2026-08-31)

Features

• rpc: draft the session and lane model (55806d1)
• rpc: draft the session and lane model (a78d4da)
• transports: bind QUIC, WebTransport and byte streams to sessions (e5da2dc)
• transports: bind QUIC, WebTransport and byte streams to sessions (a5c6eac)

Bug Fixes

• http: stop advertising WebTransport datagrams that do not exist (2c59228)
• iroh: let a caller report datagrams it disabled locally (cab9499)
• iroh: let a caller report datagrams it disabled locally (da97af5)
• iroh: map a closed datagram channel to Closed as well (a34a958)
• iroh: map a closed datagram channel to Closed as well (3c43232)
• iroh: refuse datagram traffic on a session that reports none (a061d3a)
• iroh: report a deliberate close as Closed, not as a transport fault (64527ff)
• iroh: report a deliberate close as Closed, not as a transport fault (0874ec1)
• iroh: report the datagram capability the connection negotiated (aaf5dea)
• iroh: report the datagram capability the connection negotiated (278e7d4)
• iroh: un-flake the oversize test, share the datagram override (d0bd2a3)
• iroh: un-flake the oversize test, share the datagram override (aabf455)
• quic: let a caller report datagrams it disabled locally (404dc31)
• quic: map a closed datagram channel to Closed as well (469b57d)
• quic: refuse datagram traffic on a session that reports none (9c63e56)
• quic: report a deliberate close as Closed, not as a transport fault (4a7b4ec)
• quic: report the datagram capability the connection negotiated (96f6cfd)
• quic: set service_io’s session context, own the endpoint, close on drop (72708c4)
• quic: un-flake the oversize test, share the datagram override (8e98599)
• rpc: close a session from either end and at every exit (91e9497)
• rpc: close a session from either end and at every exit (6a872a5)
• rpc: close the gaps the last round’s fixes left open (a274695)
• rpc: close the gaps the last round’s fixes left open (1de60cb)
• rpc: decode datagrams by role, and close iroh on the last drop (012b4f7)
• rpc: decode datagrams by role, and close iroh on the last drop (b48ae8a)
• rpc: do not report the end of a lane over a queued frame (11d9caa)
• rpc: do not report the end of a lane over a queued frame (3ccd224)
• rpc: end a closed lane for its peer, and name a lost open correctly (ebebd05)
• rpc: end a closed lane for its peer, and name a lost open correctly (9939181)
• rpc: end a lane for its peer, not for its refcount (23b6603)
• rpc: end a lane for its peer, not for its refcount (7b85c85)
• rpc: give lanes a cancellation token their session owns (4a8446f)
• rpc: give lanes a cancellation token their session owns (592d8a5)
• rpc: make datagram receive singular, since the queue is (a0c5b2f)
• rpc: make datagram receive singular, since the queue is (76eea26)
• rpc: make session close reach lanes it handed out (343ffe2)
• rpc: make session close reach lanes it handed out (ed6aac0)
• rpc: reject an ordered ticket issued by another lane (57b9382)
• rpc: reject an ordered ticket issued by another lane (a4c1c6d)
• rpc: stamp a code on transport errors that carry none (2f90b23)
• rpc: stamp a code on transport errors that carry none (c71a187)
• rpc: tell a lane-only close apart from a session close (5d3b9f3)
• rpc: tell a lane-only close apart from a session close (43f17c1)
• transports: address both review rounds on the transport bindings (d361c42)

16.1.3 (2026-08-24)

Bug Fixes

• bazel: add crate annotation to link serde dep for chrono (7e1c939)
• bazel: upgrade Rust toolchain to 1.89.0, fix SVG compile_data, fix test file staging (ae13cb5)
• bazel: use deps instead of extra_deps in crate.annotation for chrono (834e73b)
• bazel: wire serde dep for chrono via crate annotation (7d1c06e)

16.1.2 (2026-07-17)

Bug Fixes

• build: silence trailing-semicolon-in-macro lint from cfg_aliases! (a979f0a)
• build: silence trailing-semicolon-in-macro lint from cfg_aliases! (6f6047f)
• deps: bump cfg_aliases to 0.2.2 to fix nightly build (c07c7b9)

16.1.1 (2026-07-10)

Bug Fixes

• add test-specific iroh builders that don’t require external relay (3e19eb3)
• correct pnpm allowBuilds config to actually approve esbuild’s postinstall script (67c4d96)
• keep iroh Connection/Endpoint alive in client_builder to stop examples hanging (303e955)
• resolve compile errors from iroh 1.0.2 and axum-test 20.0.0 API changes, fix pnpm build approval config (b6a8aef)
• wait for iroh endpoint to come online in benchmark, same as example (e195701)

16.1.0 (2026-05-02)

Features

• wireformat tuple support with 12 items (cd442bb)
• wireformat tuple support with 12 items (cf59a41)

Bug Fixes

• CI macos sed bug (2fffa4a)
• ci: resolve benchmark workflow checkout failure on Cargo.lock conflict (7c26430)
• gen is reserved in rust 2024 (075e3e2)
• jetstream: add hashbrown support to wireformat, update rcgen to 0.14, misc fixes (f5659cb)
• single item tuple (f727793)
• updated rcgen api (016ada8)

16.0.0 (2026-02-22)

⚠ BREAKING CHANGES

• jetstream: WebTransportHandler trait removed in favor of Router

Features

• jetstream: add router-based dispatch and WebTransport cert auth (5bb5ff8)
• jetstream: add router-based dispatch and WebTransport cert auth (b6ca4f9)

15.2.0 (2026-02-12)

Features

• jetstream_react package and webtransport implmentation (3309a0c)
• jetstream_wireformat for swift and typescript (e0f73fc)
• move webtransport to it’s own package (2ddbdf2)

15.1.0 (2026-02-10)

Features

• implement backtrace for error (ac9d388)
• implement backtrace for error (bc85295)

15.0.0 (2026-02-07)

⚠ BREAKING CHANGES

• mtls now accepts a client verifier

Features

• mtls now accepts a client verifier (3b8fe8c)

14.1.0 (2026-02-06)

Features

• version is devived in crate (0a2a1ea)
• version is fixed, service now has a uses attribute (3c9e03d)

14.0.0 (2026-01-29)

⚠ BREAKING CHANGES

• fully implenent jetstream_quic, introduced related traits

Features

• fully implenent jetstream_quic, introduced related traits (2b1e4a6)
• introduce jetstream_http, supports http1.1,http2 and h3 (10c203b)

13.0.0 (2026-01-25)

⚠ BREAKING CHANGES

• add multiplexing ability to client.
• make error types more isomorphic
• move quic and webocket to their own crates.
• use more futures
• fix proken publish
• splits up packages
• move modules to sensible parents
• protocol -> coding
• merge all the creates

Features

• add cleanup code serverside. add cloudflare docs to mdbook (91da99c)
• add context (4771c84)
• add i16,i32,i64,i128 types (ba741bd)
• add i16,i32,i64,i128 types (3f0751a)
• add iroh docs (6b7392f)
• add multiple tag pool backends (c85aafb)
• add multiplexing ability to client. (91a238e)
• add prost_wireformat macro (beeb947)
• Add support for bool in WireFormat (640c6ca)
• Add tests for jetstream modules (18462d9)
• add tracing feature (5cb2907)
• add websocket transport (3f4054c)
• add wireformat for SystemTime (169f74a)
• added async_trait support service macro (9a86185)
• autopub (73a0844)
• better error handling (faae10a)
• channels can now be split (ef646a5)
• distributes jetsream (d1477d5)
• dt features, remove rustdoc_to_md_book (0398159)
• enum support (c4552d8)
• fix proken publish (a7272c0)
• hide filesystem behind a feautre-flag (9aa880d)
• introduce iroh examples (86e001f)
• introduce jetstream cloudflare module. (6e8cc01)
• Ip primitives (1c263b5)
• jetstream cluster (80b5727)
• jetstream distributed features (52b5e89)
• jetstream_distributed: placement (8a93788)
• jetstream_libc (aba123b)
• jetstream_rpc supports wasm (e97e6ca)
• macros: service macro to remove boilerplate code (e0a9295)
• modularize components (7262a66)
• move quic and webocket to their own crates. (7d0ba9f)
• prost_wireformat: make derives optional (12dfacb)
• publish npm package to gh (9b332b9)
• release please (5fe2180)
• release please (7d7bedd)
• revamp service (#147) (6d96be8)
• rust-clippy code scanning (3dfb39f)
• rustdoc_to_mdbook (32deadf)
• start quic server with config (#159) (3433981)
• update mdbook (f213c71)
• use more futures (467b6f5)
• use sccache (#142) (89f96ab)
• use serde_bytes::ByteBuf instead of Bytes (a1101d9)
• use tag pool in mux (cd5dd5d)
• virtio support (ce13217)
• wasm-support (#260) (5cbff0d)
• wireformat: add u128 (c76f6c4)

Bug Fixes

• add a script to update the workspace versions (9fc4ca2)
• add comment about cf exec model (c979856)
• add support for libc in windows and mac (1fb3b5b)
• add toasty types fix skip (0fe936a)
• auto-release (964036c)
• auto-release feature (6505b0f)
• benchmarks (af695e2)
• benchmarks (94767dd)
• benchmarks and add iroh_benchmarks (56dbc12)
• benchmarks code (c0d0f59)
• bothced update (b3b7003)
• broken lock file (6148cf7)
• broken okid lockfile (63bae5e)
• broken release-please (089bb22)
• broken use statement for async_trait (cce4df6)
• bump deps (0dbd81b)
• bump okid (7ed2940)
• bump zero copy (#140) (4bb933f)
• cargo lock resolver (327d8bd)
• cargo publish workspace (8db9367)
• cargo toml version issues (3481015)
• Change git user configuration in workflow (c3d92c1)
• change service shape (194252d)
• ci (e7418e5)
• ci release-please (de391e5)
• ci workflows (4c12f04)
• ci-builds, drop 1.76 (5664e57)
• ci/cd (#157) (3f7ff1e)
• cleanup code and remove unused mermaid (c8975d5)
• collections can only be u16::MAX (6ada71a)
• correct release workflow step ordering (0c7bbcd)
• criterion benchmark test, framed implementation (25b1611)
• Delete CHANGELOG.md (abcc79d)
• delete release-plz (a7e3199)
• dependency order (9a0e1f9)
• dependency order and cycle (6a6c997)
• dependency sudoku (7df5f9b)
• disable running echo on windows (828d6be)
• disable running echo on windows (7f44956)
• docs (995da2f)
• elide lifetimes in ufs (b778ff9)
• extern async_trait and trait_variant and lazy_static (71e93e2)
• extern_more (1bd58d8)
• extract git config to separate step (679d1a7)
• failing docs (232ddee)
• failing serde tests (81db0de)
• filesystem under feature flag, rm newline (de4cf79)
• formatting (049e584)
• formatting (04aace4)
• fuzz target (a494e15)
• hide documentation for iroh (47b104d)
• ignore e2e tests (e066dde)
• iroh example needs -F iroh (5246fb2)
• keep reciver types as is in generated code (7d95671)
• lint errors (4f50d0b)
• lint errors and rm unnecessary compiler_error (27aed1e)
• macros: protocol macro fix to spread fn args (b261a28)
• make data io::Read (12a864e)
• make data io::Read (910c75a)
• make data io::Read (77b3680)
• make error types more isomorphic (aee2fa2)
• make qid eq,hash (522de0d)
• make TagPool use a channel (beb6209)
• make the proto mod same vis as trait (a43c0a2)
• make websocket pub (6096ff9)
• make websocket pub again (21acf4b)
• mdbook changelog (5f03b6b)
• mdbook-changelog version issue (7043b8e)
• more tests (9a5071f)
• move term-transcript to dev, gate s2n windows (5fa7f4c)
• npm scoping issue (0d06884)
• only build docs in release (02b02bc)
• option<T> support (2e224ca)
• pin toasty (03b6b18)
• publish script (e8685cc)
• readme (dc77722)
• recursive self call (c51d455)
• reenable sccache (757bb7e)
• reexport tokio_util from rpc (cf28c40)
• release again (3a6e65e)
• release order (0973046)
• release please (53a698b)
• release please (92eeee5)
• release please (3a2f4df)
• release workflow (4abeb24)
• release-please-config.json (885527f)
• remove distributed (3aba1f6)
• remove expects (8a58cf5)
• remove jj (02fefd9)
• remove prost (5a5ff9b)
• remove redundant io imports and use fully qualified paths (7de4b9e)
• remove redundant push from version update step (923d696)
• remove tracing from sink and stream (97ac0a5)
• remove unsafe code (a286d93)
• revert serde_bytes to bytes with serde (2e02460)
• rollback toasty (ae0790b)
• rust.yml workflow (9dc7bc0)
• rustdoc to mdbook (2098ad1)
• semaphor should use permit.forget() (340323c)
• service macro works correctly with client calls (eb1fd0f)
• simpler workflows for github actions (17185dd)
• snapshots for macros (fab686d)
• some tests (c65ded1)
• switch to release-plz (a07aaec)
• target os cfg attrs (16e5c84)
• trait_variant use (#155) (3cd5665)
• typo (901ffbd)
• unused git (3fe908e)
• update 9p to use trait_variant (96db410)
• Update CHANGELOG.md (6ab562f)
• Update client_tests.rs (4c50132)
• update doc links (80a5e1e)
• update docs (74e2867)
• Update lib.rs for extern crates (ffb6777)
• update release flow (#144) (36dd4af)
• Update release-please-config.json (e592cf3)
• Update release.yml (b058b38)
• Update release.yml (#122) (566fe1f)
• update snapshot tests (b9fde4c)
• update to v2 upload sarif (e38bacb)
• use cargo publish workspace (e28a0f7)
• version (ca861a7)
• version (822bf0e)
• versioning issues (4a3111c)
• versions in workspace (8b90b5a)
• warnings (62d8013)
• wasm-pack build error. rm rustdoc2mdbook (68076cc)
• wasm32 feature gates (755f9c1)
• wireformat: from_bytes doesn’t require a mutable buf (437c35c)

Code Refactoring

• merge all the creates (faa0a1a)
• move modules to sensible parents (4eba5fb)
• protocol -> coding (5f86bc7)

12.1.0 (2026-01-25)

Features

• publish npm package to gh (9b332b9)

Bug Fixes

• Change git user configuration in workflow (c3d92c1)
• npm scoping issue (0d06884)

12.0.0 (2026-01-24)

⚠ BREAKING CHANGES

• add multiplexing ability to client.

Features

• add multiple tag pool backends (c85aafb)
• add multiplexing ability to client. (91a238e)
• use tag pool in mux (cd5dd5d)

Bug Fixes

• benchmarks code (c0d0f59)
• delete release-plz (a7e3199)
• lint errors and rm unnecessary compiler_error (27aed1e)
• make TagPool use a channel (beb6209)
• semaphor should use permit.forget() (340323c)

11.1.0 (2026-01-21)

Features

• add prost_wireformat macro (beeb947)
• add wireformat for SystemTime (169f74a)
• prost_wireformat: make derives optional (12dfacb)

11.0.0 (2026-01-20)

⚠ BREAKING CHANGES

• make error types more isomorphic
• move quic and webocket to their own crates.
• use more futures
• fix proken publish
• splits up packages
• move modules to sensible parents
• protocol -> coding
• merge all the creates

Features

• add cleanup code serverside. add cloudflare docs to mdbook (91da99c)
• add context (4771c84)
• add i16,i32,i64,i128 types (ba741bd)
• add i16,i32,i64,i128 types (3f0751a)
• add iroh docs (6b7392f)
• Add support for bool in WireFormat (640c6ca)
• Add tests for jetstream modules (18462d9)
• add tracing feature (5cb2907)
• add websocket transport (3f4054c)
• added async_trait support service macro (9a86185)
• autopub (73a0844)
• better error handling (faae10a)
• channels can now be split (ef646a5)
• distributes jetsream (d1477d5)
• dt features, remove rustdoc_to_md_book (0398159)
• enum support (c4552d8)
• fix proken publish (a7272c0)
• hide filesystem behind a feautre-flag (9aa880d)
• introduce iroh examples (86e001f)
• introduce jetstream cloudflare module. (6e8cc01)
• Ip primitives (1c263b5)
• jetstream cluster (80b5727)
• jetstream distributed features (52b5e89)
• jetstream_distributed: placement (8a93788)
• jetstream_libc (aba123b)
• jetstream_rpc supports wasm (e97e6ca)
• macros: service macro to remove boilerplate code (e0a9295)
• modularize components (7262a66)
• move quic and webocket to their own crates. (7d0ba9f)
• release please (5fe2180)
• release please (7d7bedd)
• release please (044cceb)
• revamp service (#147) (6d96be8)
• rust-clippy code scanning (3dfb39f)
• rustdoc_to_mdbook (32deadf)
• start quic server with config (#159) (3433981)
• update mdbook (f213c71)
• use more futures (467b6f5)
• use sccache (#142) (89f96ab)
• use serde_bytes::ByteBuf instead of Bytes (a1101d9)
• virtio support (ce13217)
• wasm-support (#260) (5cbff0d)
• wireformat: add u128 (c76f6c4)

Bug Fixes

• add a script to update the workspace versions (9fc4ca2)
• add comment about cf exec model (c979856)
• add support for libc in windows and mac (1fb3b5b)
• add toasty types fix skip (0fe936a)
• auto-release (964036c)
• auto-release feature (6505b0f)
• benchmarks (af695e2)
• benchmarks (94767dd)
• benchmarks and add iroh_benchmarks (56dbc12)
• bothced update (b3b7003)
• broken lock file (6148cf7)
• broken okid lockfile (63bae5e)
• broken release-please (089bb22)
• broken use statement for async_trait (cce4df6)
• bump deps (0dbd81b)
• bump okid (7ed2940)
• bump zero copy (#140) (4bb933f)
• cargo lock resolver (327d8bd)
• cargo publish workspace (8db9367)
• cargo toml version issues (3481015)
• change service shape (194252d)
• ci (e7418e5)
• ci release-please (de391e5)
• ci workflows (4c12f04)
• ci-builds, drop 1.76 (5664e57)
• ci/cd (#157) (3f7ff1e)
• cleanup code and remove unused mermaid (c8975d5)
• collections can only be u16::MAX (6ada71a)
• criterion benchmark test, framed implementation (25b1611)
• Delete CHANGELOG.md (abcc79d)
• dependency order (9a0e1f9)
• dependency order and cycle (6a6c997)
• dependency sudoku (7df5f9b)
• disable running echo on windows (828d6be)
• disable running echo on windows (7f44956)
• docs (995da2f)
• elide lifetimes in ufs (b778ff9)
• extern async_trait and trait_variant and lazy_static (71e93e2)
• extern_more (1bd58d8)
• failing docs (232ddee)
• failing serde tests (81db0de)
• filesystem under feature flag, rm newline (de4cf79)
• formatting (049e584)
• formatting (04aace4)
• fuzz target (a494e15)
• hide documentation for iroh (47b104d)
• ignore e2e tests (e066dde)
• iroh example needs -F iroh (5246fb2)
• keep reciver types as is in generated code (7d95671)
• lint errors (4f50d0b)
• macros: protocol macro fix to spread fn args (b261a28)
• make data io::Read (12a864e)
• make data io::Read (910c75a)
• make data io::Read (77b3680)
• make error types more isomorphic (aee2fa2)
• make qid eq,hash (522de0d)
• make the proto mod same vis as trait (a43c0a2)
• make websocket pub (6096ff9)
• make websocket pub again (21acf4b)
• mdbook changelog (5f03b6b)
• mdbook-changelog version issue (7043b8e)
• more tests (9a5071f)
• move term-transcript to dev, gate s2n windows (5fa7f4c)
• only build docs in release (02b02bc)
• option<T> support (2e224ca)
• pin toasty (03b6b18)
• publish script (e8685cc)
• readme (dc77722)
• recursive self call (c51d455)
• reenable sccache (757bb7e)
• reexport tokio_util from rpc (cf28c40)
• release again (3a6e65e)
• release please (53a698b)
• release please (92eeee5)
• release please (3a2f4df)
• release workflow (4abeb24)
• release-please-config.json (885527f)
• remove distributed (3aba1f6)
• remove expects (8a58cf5)
• remove jj (02fefd9)
• remove prost (5a5ff9b)
• remove redundant io imports and use fully qualified paths (7de4b9e)
• remove tracing from sink and stream (97ac0a5)
• remove unsafe code (a286d93)
• revert serde_bytes to bytes with serde (2e02460)
• rollback toasty (ae0790b)
• rust.yml workflow (9dc7bc0)
• rustdoc to mdbook (2098ad1)
• service macro works correctly with client calls (eb1fd0f)
• simpler workflows for github actions (17185dd)
• snapshots for macros (fab686d)
• some tests (c65ded1)
• target os cfg attrs (16e5c84)
• trait_variant use (#155) (3cd5665)
• typo (901ffbd)
• unused git (3fe908e)
• update 9p to use trait_variant (96db410)
• Update CHANGELOG.md (6ab562f)
• Update client_tests.rs (4c50132)
• update doc links (80a5e1e)
• update docs (74e2867)
• Update lib.rs for extern crates (ffb6777)
• update release flow (#144) (36dd4af)
• Update release-please-config.json (e592cf3)
• Update release.yml (b058b38)
• Update release.yml (#122) (566fe1f)
• update to v2 upload sarif (e38bacb)
• use cargo publish workspace (e28a0f7)
• version (ca861a7)
• version (822bf0e)
• versions in workspace (8b90b5a)
• warnings (62d8013)
• wasm-pack build error. rm rustdoc2mdbook (68076cc)
• wasm32 feature gates (755f9c1)
• wireformat: from_bytes doesn’t require a mutable buf (437c35c)

Code Refactoring

• merge all the creates (faa0a1a)
• move modules to sensible parents (4eba5fb)
• protocol -> coding (5f86bc7)

10.1.0 (2026-01-20)

Features

• better error handling (faae10a)
• update mdbook (f213c71)

Bug Fixes

• benchmarks (af695e2)
• cargo publish workspace (8db9367)
• cleanup code and remove unused mermaid (c8975d5)
• mdbook-changelog version issue (7043b8e)
• move term-transcript to dev, gate s2n windows (5fa7f4c)
• publish script (e8685cc)

10.0.0 (2025-11-19)

⚠ BREAKING CHANGES

• make error types more isomorphic

Bug Fixes

• make error types more isomorphic (aee2fa2)

9.5.0 (2025-10-30)

Features

• add cleanup code serverside. add cloudflare docs to mdbook (91da99c)
• introduce jetstream cloudflare module. (6e8cc01)

Bug Fixes

• add comment about cf exec model (c979856)
• remove tracing from sink and stream (97ac0a5)
• remove unsafe code (a286d93)

9.4.2 (2025-10-28)

Bug Fixes

• dependency sudoku (7df5f9b)

9.4.1 (2025-10-17)

Bug Fixes

• remove redundant io imports and use fully qualified paths (7de4b9e)
• target os cfg attrs (16e5c84)

9.4.0 (2025-10-15)

Features

• add context (4771c84)

Bug Fixes

• cargo toml version issues (3481015)
• collections can only be u16::MAX (6ada71a)
• recursive self call (c51d455)
• wasm32 feature gates (755f9c1)

9.3.0 (2025-10-14)

Features

• add tracing feature (5cb2907)

9.2.0 (2025-10-05)

Features

• dt features, remove rustdoc_to_md_book (0398159)

Bug Fixes

• iroh example needs -F iroh (5246fb2)
• wasm-pack build error. rm rustdoc2mdbook (68076cc)

9.1.2 (2025-10-05)

Bug Fixes

• failing docs (232ddee)

9.1.1 (2025-10-05)

Bug Fixes

• benchmarks and add iroh_benchmarks (56dbc12)
• hide documentation for iroh (47b104d)

9.1.0 (2025-10-02)

Features

• introduce iroh examples (86e001f)

Bug Fixes

• add support for libc in windows and mac (1fb3b5b)
• disable running echo on windows (828d6be)
• disable running echo on windows (7f44956)

9.0.0 (2025-09-29)

⚠ BREAKING CHANGES

• move quic and webocket to their own crates.

Features

• move quic and webocket to their own crates. (7d0ba9f)

Bug Fixes

• simpler workflows for github actions (17185dd)

8.3.1 (2025-06-06)

Bug Fixes

• update docs (74e2867)

8.3.0 (2025-05-31)

Features

• jetstream_libc (aba123b)

8.2.1 (2025-05-29)

Bug Fixes

• rustdoc to mdbook (2098ad1)

8.2.0 (2025-05-29)

Features

• jetstream distributed features (52b5e89)

Bug Fixes

• Update lib.rs for extern crates (ffb6777)

8.1.5 (2025-04-05)

Bug Fixes

• rollback toasty (ae0790b)
• version (ca861a7)

8.1.4 (2025-04-04)

Bug Fixes

• add toasty types fix skip (0fe936a)
• ci-builds, drop 1.76 (5664e57)
• pin toasty (03b6b18)

8.1.3 (2025-04-04)

Bug Fixes

• extern async_trait and trait_variant and lazy_static (71e93e2)
• extern_more (1bd58d8)

8.1.2 (2025-02-15)

Bug Fixes

• make websocket pub again (21acf4b)

8.1.1 (2025-02-15)

Bug Fixes

• make websocket pub (6096ff9)

8.1.0 (2025-02-15)

Features

• add websocket transport (3f4054c)
• channels can now be split (ef646a5)

8.0.10 (2025-02-14)

Bug Fixes

• broken lock file (6148cf7)

8.0.9 (2025-02-14)

Bug Fixes

• remove prost (5a5ff9b)

8.0.8 (2025-02-01)

Bug Fixes

• remove expects (8a58cf5)

8.0.7 (2025-01-30)

Bug Fixes

• release please (53a698b)

8.0.6 (2025-01-30)

Bug Fixes

• release please (92eeee5)
• release please (3a2f4df)

8.0.5 (2025-01-30)

Bug Fixes

• use cargo publish workspace (e28a0f7)

8.0.4 (2025-01-30)

Bug Fixes

• dependency order and cycle (6a6c997)
• formatting (049e584)
• reexport tokio_util from rpc (cf28c40)
• rust.yml workflow (9dc7bc0)

8.0.3 (2025-01-30)

Bug Fixes

• cargo lock resolver (327d8bd)
• dependency order (9a0e1f9)

8.0.2 (2025-01-30)

Bug Fixes

• criterion benchmark test, framed implementation (25b1611)
• failing serde tests (81db0de)
• reenable sccache (757bb7e)
• service macro works correctly with client calls (eb1fd0f)
• snapshots for macros (fab686d)

8.0.1 (2025-01-28)

Bug Fixes

• remove distributed (3aba1f6)

8.0.0 (2025-01-27)

⚠ BREAKING CHANGES

• use more futures

Features

• use more futures (467b6f5)

Bug Fixes

• formatting (04aace4)

7.4.0 (2025-01-18)

Features

• jetstream_rpc supports wasm (e97e6ca)

7.3.0 (2025-01-17)

Features

• wasm-support (#260) (5cbff0d)

7.2.1 (2024-12-10)

Bug Fixes

• keep reciver types as is in generated code (7d95671)

7.2.0 (2024-12-09)

Features

• Ip primitives (1c263b5)

7.1.2 (2024-12-06)

Bug Fixes

• elide lifetimes in ufs (b778ff9)

7.1.1 (2024-11-25)

Bug Fixes

• broken use statement for async_trait (cce4df6)

7.1.0 (2024-11-25)

Features

• added async_trait support service macro (9a86185)

7.0.4 (2024-11-23)

Bug Fixes

• fuzz target (a494e15)

7.0.3 (2024-11-21)

Bug Fixes

• Delete CHANGELOG.md (abcc79d)
• mdbook changelog (5f03b6b)

7.0.1 (2024-11-21)

Bug Fixes

• ci (e7418e5)
• docs (995da2f)

7.0.0 (2024-11-21)

⚠ BREAKING CHANGES

• fix proken publish

Features

• fix proken publish (a7272c0)

6.6.2 (2024-11-21)

Bug Fixes

• bump okid (7ed2940)

6.6.1 (2024-11-20)

Bug Fixes

• broken okid lockfile (63bae5e)

6.6.0 (2024-11-20)

Features

• distributes jetsream (d1477d5)
• jetstream_distributed: placement (8a93788)

Bug Fixes

• change service shape (194252d)
• typo (901ffbd)

6.5.0 (2024-11-20)

Features

• jetstream cluster (80b5727)
• rustdoc_to_mdbook (32deadf)

Bug Fixes

• only build docs in release (02b02bc)
• update doc links (80a5e1e)

6.4.2 (2024-11-18)

Bug Fixes

• bump deps (0dbd81b)

6.4.1 (2024-11-11)

Bug Fixes

• make the proto mod same vis as trait (a43c0a2)

6.4.0 (2024-11-10)

Features

• Add support for bool in WireFormat (640c6ca)
• Add tests for jetstream modules (18462d9)

Bug Fixes

• benchmarks (94767dd)
• more tests (9a5071f)
• some tests (c65ded1)

6.3.4 (2024-11-10)

Bug Fixes

• make qid eq,hash (522de0d)

6.3.3 (2024-11-10)

Bug Fixes

• update 9p to use trait_variant (96db410)

6.3.2 (2024-11-10)

Bug Fixes

• option<T> support (2e224ca)

6.3.1 (2024-11-09)

Bug Fixes

• ci workflows (4c12f04)

6.3.0 (2024-11-09)

Features

• add i16,i32,i64,i128 types (ba741bd)
• enum support (c4552d8)

Bug Fixes

• remove jj (02fefd9)

6.2.0 (2024-11-08)

Features

• add i16,i32,i64,i128 types (3f0751a)

6.1.0 (2024-11-08)

Features

• start quic server with config (#159) (3433981)

6.0.2 (2024-11-08)

Bug Fixes

• ci/cd (#157) (3f7ff1e)

6.0.1 (2024-11-08)

Bug Fixes

• trait_variant use (#155) (3cd5665)

6.0.0 (2024-11-07)

⚠ BREAKING CHANGES

• splits up packages
• move modules to sensible parents
• protocol -> coding
• merge all the creates

Features

• autopub (73a0844)
• hide filesystem behind a feautre-flag (9aa880d)
• macros: service macro to remove boilerplate code (e0a9295)
• modularize components (7262a66)
• release please (7d7bedd)
• release please (044cceb)
• revamp service (#147) (6d96be8)
• rust-clippy code scanning (3dfb39f)
• use sccache (#142) (89f96ab)
• use serde_bytes::ByteBuf instead of Bytes (a1101d9)
• virtio support (ce13217)
• wireformat: add u128 (c76f6c4)

Bug Fixes

• auto-release (964036c)
• auto-release feature (6505b0f)
• bothced update (b3b7003)
• broken release-please (089bb22)
• bump zero copy (#140) (4bb933f)
• ci release-please (de391e5)
• filesystem under feature flag, rm newline (de4cf79)
• ignore e2e tests (e066dde)
• lint errors (4f50d0b)
• macros: protocol macro fix to spread fn args (b261a28)
• make data io::Read (12a864e)
• make data io::Read (910c75a)
• make data io::Read (77b3680)
• readme (dc77722)
• release again (3a6e65e)
• release workflow (4abeb24)
• revert serde_bytes to bytes with serde (2e02460)
• unused git (3fe908e)
• Update client_tests.rs (4c50132)
• update release flow (#144) (36dd4af)
• Update release.yml (b058b38)
• Update release.yml (#122) (566fe1f)
• update to v2 upload sarif (e38bacb)
• version (822bf0e)
• warnings (62d8013)
• wireformat: from_bytes doesn’t require a mutable buf (437c35c)

Code Refactoring

• merge all the creates (faa0a1a)
• move modules to sensible parents (4eba5fb)
• protocol -> coding (5f86bc7)

5.4.2 (2024-11-07)

Bug Fixes

• auto-release (964036c)

5.4.2 (2024-11-07)

Bug Fixes

• auto-release (964036c)

5.4.1 (2024-11-07)

Bug Fixes

• readme (dc77722)

5.4.0 (2024-11-07)

Features

• revamp service (#147) (6d96be8)

5.3.0 (2024-10-23)

Features

• use sccache (#142) (89f96ab)

Bug Fixes

• update release flow (#144) (36dd4af)

5.2.3 (2024-10-10)

Bug Fixes

• bump zero copy (#140) (4bb933f)
• unused git (3fe908e)

5.2.2 (2024-10-07)

Bug Fixes

• revert serde_bytes to bytes with serde (2e02460)

5.2.1 (2024-10-06)

Bug Fixes

• wireformat: from_bytes doesn’t require a mutable buf (437c35c)

5.2.0 (2024-10-06)

Features

• use serde_bytes::ByteBuf instead of Bytes (a1101d9)

Bug Fixes

• lint errors (4f50d0b)

5.1.4 (2024-10-03)

Bug Fixes

• lint errors (4f50d0b)

5.1.3 (2024-10-03)

Bug Fixes

• Update release.yml (b058b38)

5.1.2 (2024-10-03)

Bug Fixes

• Update release.yml (#122) (566fe1f)

5.1.1 (2024-10-03)

Bug Fixes

• release again (3a6e65e)
• warnings (62d8013)

5.1.0 (2024-10-03)

Features

• wireformat: add u128 (c76f6c4)

Bug Fixes

• release workflow (4abeb24)

5.0.0 (2024-10-03)

⚠ BREAKING CHANGES

• splits up packages

Features

• modularize components (7262a66)

Bug Fixes

• version (822bf0e)

4.0.0 (2024-10-01)

⚠ BREAKING CHANGES

• move modules to sensible parents

Code Refactoring

• move modules to sensible parents (4eba5fb)

3.0.0 (2024-03-30)

⚠ BREAKING CHANGES

• protocol -> coding
• merge all the creates

Features

• autopub (73a0844)
• hide filesystem behind a feautre-flag (9aa880d)
• macros: service macro to remove boilerplate code (e0a9295)
• release please (7d7bedd)
• release please (044cceb)
• rust-clippy code scanning (3dfb39f)
• virtio support (ce13217)

Bug Fixes

• auto-release feature (6505b0f)
• bothced update (b3b7003)
• broken release-please (089bb22)
• ci release-please (de391e5)
• filesystem under feature flag, rm newline (de4cf79)
• ignore e2e tests (e066dde)
• macros: protocol macro fix to spread fn args (b261a28)
• make data io::Read (12a864e)
• make data io::Read (910c75a)
• make data io::Read (77b3680)
• Update client_tests.rs (4c50132)
• update to v2 upload sarif (e38bacb)

Code Refactoring

• merge all the creates (faa0a1a)
• protocol -> coding (5f86bc7)

2.0.2 (2024-03-30)

Bug Fixes

• make data io::Read (12a864e)
• make data io::Read (910c75a)
• make data io::Read (77b3680)

2.0.1 (2024-03-29)

Bug Fixes

• macros: protocol macro fix to spread fn args (b261a28)

2.0.0 (2024-03-29)

⚠ BREAKING CHANGES

• protocol -> coding

Code Refactoring

• protocol -> coding (5f86bc7)

1.1.1 (2024-03-29)

Bug Fixes

• ignore e2e tests (e066dde)

1.1.0 (2024-03-29)

Features

• macros: service macro to remove boilerplate code (e0a9295)

1.0.0 (2024-03-25)

⚠ BREAKING CHANGES

• merge all the creates

Code Refactoring

• merge all the creates (faa0a1a)

0.6.0 (2024-03-21)

Features

• virtio support (ce13217)

0.5.1 (2024-03-15)

Bug Fixes

• filesystem under feature flag, rm newline (de4cf79)
• Update client_tests.rs (4c50132)

0.5.0 (2024-03-15)

Features

• rust-clippy code scanning (3dfb39f)

Bug Fixes

• update to v2 upload sarif (e38bacb)

0.4.0 (2024-03-14)

Features

• autopub (73a0844)
• hide filesystem behind a feautre-flag (9aa880d)

0.3.2 (2024-03-14)

Bug Fixes

• auto-release feature (6505b0f)

0.3.1 (2024-03-14)

Bug Fixes

• broken release-please (089bb22)
• ci release-please (de391e5)

0.3.0 (2024-03-14)

Features

• release please (7d7bedd)
• release please (044cceb)

0.2.0 (2024-03-14)

Features

• release please (044cceb)

JetStream Cloudflare (deprecated)

Cloudflare Workers support was removed, along with the jetstream_radar example crate that this page walked through.

The page is kept because it was published under this path. Its three code listings used to be {{#include}} directives pointing into components/jetstream_radar/, which is no longer in the repository — mdbook logged an error for each one and rendered the page with three empty code blocks, so what stood here was not an example anyone could follow.

For a current, working example of defining a service and connecting to it, see the QUIC and iroh examples.