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.