From 38bed87f35adea24005efdf18ac5199401478528 Mon Sep 17 00:00:00 2001 From: Ruediger Klaehn Date: Mon, 8 Jun 2026 12:04:58 +0300 Subject: [PATCH 1/7] Rename and extend --- zero-rtt-blob-service/.DS_Store | Bin 0 -> 8196 bytes zero-rtt-blob-service/Cargo.lock | 4416 +++++++++++++++++ zero-rtt-blob-service/Cargo.toml | 24 + ...7164900b19ac2a4eab7d8e564644a8-client.qlog | 218 + ...16fd34a6439bc903a65c403f61a3b5-client.qlog | 120 + ...5dc538bebe9b0dc528494bacb1499e-client.qlog | 219 + ...9e35a31a47a2f3b016b9bbcef3ed16-client.qlog | 182 + zero-rtt-blob-service/src/main.rs | 450 ++ 8 files changed, 5629 insertions(+) create mode 100644 zero-rtt-blob-service/.DS_Store create mode 100644 zero-rtt-blob-service/Cargo.lock create mode 100644 zero-rtt-blob-service/Cargo.toml create mode 100644 zero-rtt-blob-service/client/zero-rtt-blob-service-client-1780660628454-8b8114180e7164900b19ac2a4eab7d8e564644a8-client.qlog create mode 100644 zero-rtt-blob-service/client/zero-rtt-blob-service-client-1780660628467-f4319b3d1716fd34a6439bc903a65c403f61a3b5-client.qlog create mode 100644 zero-rtt-blob-service/client/zero-rtt-blob-service-client-1780661128119-27e23aabcf5dc538bebe9b0dc528494bacb1499e-client.qlog create mode 100644 zero-rtt-blob-service/client/zero-rtt-blob-service-client-1780661128134-d60456d9879e35a31a47a2f3b016b9bbcef3ed16-client.qlog create mode 100644 zero-rtt-blob-service/src/main.rs diff --git a/zero-rtt-blob-service/.DS_Store b/zero-rtt-blob-service/.DS_Store new file mode 100644 index 0000000000000000000000000000000000000000..a773755bf51da873344f11d10d4567e4c796cf39 GIT binary patch literal 8196 zcmeHMU2GIp6u#fI(3yeK+tQY@8(>2tKc$uOSIUoZ`-fB@knWb20&P|C_*o;n~DC$-pSdj0Y-TaDDu4eVH)UI%-`#x`N=@yYxpiutI!`^EI_8#B ze%3E~{n^4n(Mr3novYkubNel`oZ-y_uH$Dd$1DU6mSK{~p)t!b+;YELbPO->O-Y5S zsal4&oj$#3eRpSKLvQby&cx|0;(p_Xo-=1OwWWRSwvo(nd(v^w$!Q?m3Rr#Bw4SP- zlkJrT9|#OHoEJTF#3VbbWDb(GH*V2z z!mPRK#i3oC-?C)+iZ$JvwmvwxbN9ZP`UMSKZ_=B`N|tBkY;&}18UC>YSubxohUFZc zAoLvHI%1gxtr4GTX(u~r&hXYnZHr^^c(rw9l<{KDI#IOzQ@mYOXZV7K`lZX_`XPf} zZjWn-(G32css)<%W%2kyp&2NcR2!kg%2jba?a>@~HZ<()()ILlT3+sH(skW>Jw9q@ zbEeG=O+Ke*vmPHSNo@J9uyfmXJwBdyCnsqtUL!ig?qpv){)l+;?yTqUoiZJIt1IE- zhe$W!9m^KXy=7up*gH(U!HQA8a2hf!-wl)m$R3)=0Wn;kVFc4?YhA?SBk{D#zf@_J zET?1_W>{Ipy zyT-m{-?8u6kL(wAo&AAY)T03$jfi0-R-*%Jun|f0p&z@j4coP?K2_NG!KEdbs3Rm$pe!zA7hTrk0Qm3ejrZg!Fl@_H{ zS*>&^89}Vq!ZU(uePgh`Baj#7JY)!seEx z?W?-h_iRhv;F2U?g>V04aX{T!gb%12taKIUW@IqwODfS*9aW>mdGn+yh|BAwbwXdR z66g6G*}P9*sS?}yT-of1b*dC!TnU=`+SodcqKMbXW=Cw3rgF$eufA2Ic;dA)ix;)E zZht`6ITLZN$&vbFgDS-YSN}a(-)0}O%j^o9A+r9&erCV2zn~IDTamyDBB`Kg7kaQ6 zTd)<|F@zo1iD9B^8lxD)I3|!Mx*ow%SSY}Qj}jioG*R{}o+QFPi|6n>Ucd#sj92km zNZohvZWS3Ho=rwvswQJ8Rd8(AIZFK%0?&4dT6cs%jr3|0`Tx-EzyIIS*pX9&K!m{m z0|C?yrUu1NfZOe^BG*pP^(bAU2)k)10~e} blake3(data) (32 bytes) +//! GET (0x01) || hash_32_bytes -> data_bytes (or empty) +//! +//! Storage: in-memory `HashMap`. Wiped on server restart. + +use std::{ + env, + str::FromStr, + sync::Arc, + time::{Duration, Instant}, +}; + +use blake3::Hash; +use bytes::Bytes; +use clap::{Parser, Subcommand}; +use data_encoding::HEXLOWER; +use iroh::{ + EndpointAddr, RelayMode, SecretKey, TransportAddr, + endpoint::{ + Connection, Endpoint, Incoming, PathId, QuicTransportConfig, RecvStream, SendStream, + ServerConfig, ValidationTokenConfig, ZeroRttStatus, presets, + }, +}; +use iroh_tickets::endpoint::EndpointTicket; +use n0_error::{Result, StdResultExt}; +use noq_proto::{BloomTokenLog, NoneTokenLog}; +use papaya::HashMap; +use tokio::io::{AsyncReadExt, AsyncWriteExt}; +use tracing::{info, trace}; + +const ALPN: &[u8] = b"zero-rtt-blob-service/v1"; +const MAX_BLOB: usize = 10 * 1024; +const TAG_PUT: u8 = 0x00; +const TAG_GET: u8 = 0x01; +const TOKENS_PER_VALIDATION: u32 = 2; +const TOKEN_LIFETIME: Duration = Duration::from_secs(14 * 24 * 60 * 60); // 2 weeks + +// --------------------------------------------------------------------------- +// CLI +// --------------------------------------------------------------------------- + +#[derive(Parser)] +#[command(name = "zero-rtt-blob-service")] +struct Args { + #[command(subcommand)] + mode: Mode, +} + +#[derive(Subcommand)] +enum Mode { + /// Run a server that stores blobs in memory. + Server(ServerArgs), + /// Run a client that uploads or downloads a blob. + Client(ClientArgs), +} + +#[derive(Parser)] +struct ServerArgs { + /// Disable issuance of NEW_TOKEN validation tokens. + #[clap(long)] + no_tokens: bool, +} + +#[derive(Parser)] +struct ClientArgs { + /// Ticket of the server to connect to, in `@,,...` form. + /// The ticket must contain at least one IP address — relay-only tickets are + /// refused because 0-RTT via a relay is not interesting for this demo. + ticket: String, + + #[command(subcommand)] + op: ClientOp, + + /// Discard the in-memory token store between rounds, so no stored + /// token is presented to the server. + #[clap(long)] + no_tokens: bool, + + /// Number of requests to run. + #[clap(long = "n", default_value = "2")] + n: u32, +} + +#[derive(Subcommand)] +enum ClientOp { + /// Upload up to 10 KiB of data read from stdin. Prints the BLAKE3 hash. + Put, + /// Download the blob with the given hex-encoded BLAKE3 hash to stdout. + Get { hash: String }, +} + +// --------------------------------------------------------------------------- +// Server +// --------------------------------------------------------------------------- + +async fn run_server(args: ServerArgs) -> Result<()> { + // Leave relays and address discovery (QAD) enabled on the server: that's how it + // learns its own public address (including behind 1:1 NAT) so the ticket carries + // a reachable direct address. It doesn't affect the measurement — the ticket is + // built from IP addresses only (see `wait_for_ip_addr`), and the client runs with + // relays disabled, so the client always dials the server directly. + let endpoint = Endpoint::builder(presets::N0) + .alpns(vec![ALPN.to_vec()]) + .transport_config(qlog_transport_config("zero-rtt-blob-service-server")) + .secret_key(get_or_generate_secret_key()?) + .bind() + .await?; + let ip_addr = wait_for_ip_addr(&endpoint).await?; + println!("ticket: {}", EndpointTicket::from(ip_addr)); + + let server_config = Arc::new(make_server_config(&endpoint, !args.no_tokens)); + let store: Arc> = Arc::new(HashMap::new()); + + let accept = async move { + while let Some(incoming) = endpoint.accept().await { + let store = store.clone(); + let cfg = server_config.clone(); + tokio::spawn(async move { + if let Err(e) = serve_connection(incoming, cfg, store).await { + trace!("connection ended: {e:?}"); + } + }); + } + }; + tokio::select! { + _ = accept => info!("accept finished"), + _ = tokio::signal::ctrl_c() => info!("Ctrl-C received"), + } + Ok(()) +} + +async fn wait_for_ip_addr(endpoint: &Endpoint) -> Result { + let deadline = Instant::now() + Duration::from_secs(10); + loop { + let addr = endpoint.addr(); + let ip_addr = EndpointAddr::from_parts( + addr.id, + addr.ip_addrs().copied().map(TransportAddr::Ip), + ); + if !ip_addr.is_empty() { + return Ok(ip_addr); + } + if Instant::now() >= deadline { + n0_error::bail_any!("timed out waiting for local IP addresses"); + } + tokio::time::sleep(Duration::from_millis(200)).await; + } +} + +/// Build the server config. +/// +/// With `tokens` enabled, the server issues `NEW_TOKEN` address-validation tokens +/// and validates the ones clients present (via [`BloomTokenLog`]), which lifts the +/// 3x anti-amplification limit on 0-RTT connections so large 0.5-RTT responses +/// flow immediately. +/// +/// With `tokens` disabled (`--no-tokens`) the server neither issues (`sent` = 0) +/// nor validates ([`NoneTokenLog`] ignores any token a client presents), so every +/// 0-RTT connection's address stays unvalidated until the handshake completes and +/// a large 0.5-RTT response stalls on the 3x budget. Note that the default config +/// (`ValidationTokenConfig::default()`, used by `Incoming::accept`) issues and +/// validates by default, so we must set this explicitly rather than rely on it. +fn make_server_config(endpoint: &Endpoint, tokens: bool) -> ServerConfig { + let mut validation_token = ValidationTokenConfig::default(); + if tokens { + validation_token + .log(Arc::new(BloomTokenLog::default())) + .sent(TOKENS_PER_VALIDATION) + .lifetime(TOKEN_LIFETIME); + } else { + validation_token.log(Arc::new(NoneTokenLog)).sent(0); + } + endpoint + .create_server_config_builder(vec![ALPN.to_vec()]) + .set_validation_token_config(validation_token) + .build() +} + +async fn serve_connection( + incoming: Incoming, + server_config: Arc, + store: Arc>, +) -> Result<()> { + let accepting = incoming.accept_with(server_config).anyerr()?; + let conn = accepting.into_0rtt(); + while let Ok((send, recv)) = conn.accept_bi().await { + let store = store.clone(); + tokio::spawn(async move { + if let Err(e) = handle_request(send, recv, &store).await { + trace!("request error: {e:?}"); + } + }); + } + conn.closed().await; + Ok(()) +} + +async fn handle_request( + mut send: SendStream, + mut recv: RecvStream, + store: &HashMap, +) -> Result<()> { + let req = recv.read_to_end(1 + MAX_BLOB).await.anyerr()?; + let Some((&tag, body)) = req.split_first() else { + return Ok(()); + }; + match tag { + TAG_PUT => { + let hash = blake3::hash(body); + store.pin().insert(hash, Bytes::copy_from_slice(body)); + send.write_all(hash.as_bytes()).await.anyerr()?; + } + TAG_GET => { + let Ok(bytes): std::result::Result<[u8; 32], _> = body.try_into() else { + return Ok(()); + }; + let data = store.pin().get(&Hash::from_bytes(bytes)).cloned(); + if let Some(data) = data { + send.write_all(&data).await.anyerr()?; + } + } + _ => {} + } + send.finish().anyerr() +} + +// --------------------------------------------------------------------------- +// Client +// --------------------------------------------------------------------------- + +async fn run_client(args: ClientArgs) -> Result<()> { + let ticket = EndpointTicket::from_str(&args.ticket).std_context("invalid ticket")?; + let remote: EndpointAddr = ticket.into(); + if remote.ip_addrs().next().is_none() { + n0_error::bail_any!("ticket has no IP addresses — this demo requires direct IP paths"); + } + + let shared = if args.no_tokens { + None + } else { + Some(client_endpoint().await?) + }; + + match &args.op { + ClientOp::Put => { + let endpoint = match &shared { + Some(ep) => ep.clone(), + None => client_endpoint().await?, + }; + let t0 = Instant::now(); + let request = build_put_request().await?; + let hash_bytes = do_round(&endpoint, remote.clone(), request).await?; + print_hash(&hash_bytes)?; + eprintln!("put: {} us", t0.elapsed().as_micros()); + + if args.n > 1 { + let Ok(bytes): std::result::Result<[u8; 32], _> = hash_bytes.as_slice().try_into() else { + n0_error::bail_any!("expected 32-byte hash, got {} bytes", hash_bytes.len()); + }; + let hash = Hash::from_bytes(bytes); + let request = build_get_request(&hash); + for round in 1..args.n { + let t0 = Instant::now(); + let resp = do_round(&endpoint, remote.clone(), request.clone()).await?; + // Don't dump the (potentially large) blob — just echo its hash so + // the round-trip is verifiable without flooding stdout. + println!("{}", blake3::hash(&resp).to_hex()); + eprintln!("get {round}: {} us", t0.elapsed().as_micros()); + } + } + + if args.no_tokens { + endpoint.close().await; + } + } + ClientOp::Get { hash } => { + let endpoint = match &shared { + Some(ep) => ep.clone(), + None => client_endpoint().await?, + }; + let request = build_get_request(&parse_hash(hash)?); + for round in 0..args.n { + let t0 = Instant::now(); + let resp = do_round(&endpoint, remote.clone(), request.clone()).await?; + tokio::io::stdout().write_all(&resp).await.anyerr()?; + eprintln!("get {round}: {} us", t0.elapsed().as_micros()); + } + if args.no_tokens { + endpoint.close().await; + } + } + } + if let Some(ep) = shared { + ep.close().await; + } + Ok(()) +} + +async fn client_endpoint() -> Result { + let ep = Endpoint::builder(presets::N0) + .transport_config(qlog_transport_config("zero-rtt-blob-service-client")) + .relay_mode(RelayMode::Disabled) + .clear_address_lookup() + .bind() + .await?; + Ok(ep) +} + +fn qlog_transport_config(prefix: &str) -> QuicTransportConfig { + QuicTransportConfig::builder() + .qlog_from_env(prefix) + .build() +} + +async fn do_round(endpoint: &Endpoint, remote: EndpointAddr, request: Bytes) -> Result> { + let connecting = endpoint + .connect_with_opts(remote, ALPN, Default::default()) + .await?; + let (conn, resp) = match connecting.into_0rtt() { + Ok(zrtt) => { + let (send, mut recv) = zrtt.open_bi().await.anyerr()?; + // Queue the request as 0-RTT early data *before* driving the handshake + // to completion. The earlier version spawned this write and raced it + // against `handshake_completed()` — the handshake almost always won, so + // the bytes were flushed only once 1-RTT keys existed and went out in a + // 1-RTT packet. The server still reports "0-RTT accepted" (resumption + // was accepted), but no early data ever actually rides on the wire. + // Writing inline and awaiting it guarantees the bytes are buffered as + // 0-RTT data first. This matches iroh's own 0-RTT test. + write_request(send, request.clone()).await?; + match zrtt.handshake_completed().await? { + ZeroRttStatus::Accepted(conn) => { + println!("0-RTT accepted"); + (conn, recv.read_to_end(MAX_BLOB).await.anyerr()?) + } + ZeroRttStatus::Rejected(conn) => { + println!("0-RTT rejected; retrying on a fresh stream"); + let resp = roundtrip(&conn, &request).await?; + (conn, resp) + } + } + } + Err(connecting) => { + println!("0-RTT not possible from our side"); + let conn = connecting.await.anyerr()?; + let resp = roundtrip(&conn, &request).await?; + (conn, resp) + } + }; + // Don't close the connection right away. Right after the handshake the server + // sends a couple of `NewSessionTicket` messages, and those resumption tickets + // are exactly what makes the *next* connection eligible for 0-RTT. The ticket + // typically arrives in the same flight as the response, so closing here would + // race the connection driver and tear things down before the ticket is stored + // in the session cache — leaving every subsequent round stuck at 1-RTT. + // + // Instead, hand the connection off to a background task that lingers for a + // couple of RTTs so the driver can ingest the ticket, then closes cleanly. The + // request itself has already completed, so this doesn't add to the round's + // latency. + let linger = conn + .rtt(PathId::ZERO) + .map(|rtt| rtt * 2) + .unwrap_or_else(|| Duration::from_millis(100)); + tokio::spawn(async move { + tokio::time::sleep(linger).await; + conn.close(0u8.into(), b""); + }); + Ok(resp) +} + +async fn roundtrip(conn: &Connection, request: &[u8]) -> Result> { + let (mut send, mut recv) = conn.open_bi().await.anyerr()?; + send.write_all(request).await.anyerr()?; + send.finish().anyerr()?; + recv.read_to_end(MAX_BLOB).await.anyerr() +} + +async fn write_request(mut send: SendStream, request: Bytes) -> Result<()> { + send.write_all(&request).await.anyerr()?; + send.finish().anyerr() +} + +async fn build_put_request() -> Result { + let mut buf = vec![TAG_PUT]; + tokio::io::stdin().read_to_end(&mut buf).await.anyerr()?; + if buf.len() - 1 > MAX_BLOB { + n0_error::bail_any!("blob exceeds {MAX_BLOB} bytes"); + } + Ok(buf.into()) +} + +fn build_get_request(hash: &Hash) -> Bytes { + let mut buf = vec![TAG_GET]; + buf.extend_from_slice(hash.as_bytes()); + buf.into() +} + +fn print_hash(resp: &[u8]) -> Result<()> { + let Ok(bytes): std::result::Result<[u8; 32], _> = resp.try_into() else { + n0_error::bail_any!("expected 32-byte hash, got {} bytes", resp.len()); + }; + println!("{}", Hash::from_bytes(bytes).to_hex()); + Ok(()) +} + +fn parse_hash(s: &str) -> Result { + let bytes = HEXLOWER + .decode(s.as_bytes()) + .std_context("invalid hex")?; + let array: [u8; 32] = bytes + .as_slice() + .try_into() + .std_context("hash must be 32 bytes")?; + Ok(Hash::from_bytes(array)) +} + +// --------------------------------------------------------------------------- +// Misc +// --------------------------------------------------------------------------- + +fn get_or_generate_secret_key() -> Result { + if let Ok(secret) = env::var("IROH_SECRET") { + SecretKey::from_str(&secret).std_context("Invalid IROH_SECRET") + } else { + let key = SecretKey::generate(); + eprintln!( + "Generated new secret key: {}", + HEXLOWER.encode(&key.to_bytes()) + ); + eprintln!("Set IROH_SECRET to reuse this key."); + Ok(key) + } +} + +#[tokio::main] +async fn main() -> Result<()> { + tracing_subscriber::fmt::init(); + match Args::parse().mode { + Mode::Server(args) => run_server(args).await, + Mode::Client(args) => run_client(args).await, + } +} From 17c5446924115fb879e14751ff63cb42ec798603 Mon Sep 17 00:00:00 2001 From: Ruediger Klaehn Date: Mon, 8 Jun 2026 12:34:04 +0300 Subject: [PATCH 2/7] Use github patch --- zero-rtt-blob-service/Cargo.lock | 4 ++++ zero-rtt-blob-service/Cargo.toml | 10 ++++++---- 2 files changed, 10 insertions(+), 4 deletions(-) diff --git a/zero-rtt-blob-service/Cargo.lock b/zero-rtt-blob-service/Cargo.lock index cec5187..de3973d 100644 --- a/zero-rtt-blob-service/Cargo.lock +++ b/zero-rtt-blob-service/Cargo.lock @@ -1584,6 +1584,7 @@ dependencies = [ [[package]] name = "iroh" version = "1.0.0-rc.1" +source = "git+https://github.com/n0-computer/iroh?branch=persistent-token-store#fc1044a0ec6b501f3fa4fe36261a5f8c074b9195" dependencies = [ "backon", "blake3", @@ -1635,6 +1636,7 @@ dependencies = [ [[package]] name = "iroh-base" version = "1.0.0-rc.1" +source = "git+https://github.com/n0-computer/iroh?branch=persistent-token-store#fc1044a0ec6b501f3fa4fe36261a5f8c074b9195" dependencies = [ "curve25519-dalek", "data-encoding", @@ -1655,6 +1657,7 @@ dependencies = [ [[package]] name = "iroh-dns" version = "1.0.0-rc.1" +source = "git+https://github.com/n0-computer/iroh?branch=persistent-token-store#fc1044a0ec6b501f3fa4fe36261a5f8c074b9195" dependencies = [ "arc-swap", "cfg_aliases", @@ -1704,6 +1707,7 @@ dependencies = [ [[package]] name = "iroh-relay" version = "1.0.0-rc.1" +source = "git+https://github.com/n0-computer/iroh?branch=persistent-token-store#fc1044a0ec6b501f3fa4fe36261a5f8c074b9195" dependencies = [ "blake3", "bytes", diff --git a/zero-rtt-blob-service/Cargo.toml b/zero-rtt-blob-service/Cargo.toml index 6d58499..c7d2e99 100644 --- a/zero-rtt-blob-service/Cargo.toml +++ b/zero-rtt-blob-service/Cargo.toml @@ -17,8 +17,10 @@ tokio = { version = "1", features = ["full"] } tracing = "0.1" tracing-subscriber = "0.3" -# Local iroh with the shared client-side QUIC token store fix (NEW_TOKEN tokens -# now survive across connections). See branch `persistent-token-store`. +# iroh branch with the shared client-side QUIC token store fix: NEW_TOKEN +# address-validation tokens now survive across connections (the default minted a +# fresh, empty store per connection). iroh-base is patched too so iroh-tickets +# (still from crates.io) resolves the same iroh-base as the patched iroh. [patch.crates-io] -iroh = { path = "/Users/rklaehn/projects_git/iroh/iroh" } -iroh-base = { path = "/Users/rklaehn/projects_git/iroh/iroh-base" } +iroh = { git = "https://github.com/n0-computer/iroh", branch = "persistent-token-store" } +iroh-base = { git = "https://github.com/n0-computer/iroh", branch = "persistent-token-store" } From 41977ec805398cfb77db2ceaeff3e35113ab86e6 Mon Sep 17 00:00:00 2001 From: Ruediger Klaehn Date: Mon, 8 Jun 2026 12:54:29 +0300 Subject: [PATCH 3/7] print direct addrs. --- zero-rtt-blob-service/src/main.rs | 8 ++++++++ 1 file changed, 8 insertions(+) diff --git a/zero-rtt-blob-service/src/main.rs b/zero-rtt-blob-service/src/main.rs index 0656905..44d2ce7 100644 --- a/zero-rtt-blob-service/src/main.rs +++ b/zero-rtt-blob-service/src/main.rs @@ -113,6 +113,10 @@ async fn run_server(args: ServerArgs) -> Result<()> { .bind() .await?; let ip_addr = wait_for_ip_addr(&endpoint).await?; + eprintln!( + "direct addresses: {:?}", + ip_addr.ip_addrs().copied().collect::>() + ); println!("ticket: {}", EndpointTicket::from(ip_addr)); let server_config = Arc::new(make_server_config(&endpoint, !args.no_tokens)); @@ -241,6 +245,10 @@ async fn run_client(args: ClientArgs) -> Result<()> { if remote.ip_addrs().next().is_none() { n0_error::bail_any!("ticket has no IP addresses — this demo requires direct IP paths"); } + eprintln!( + "dialing direct addresses: {:?}", + remote.ip_addrs().copied().collect::>() + ); let shared = if args.no_tokens { None From b22024373756a315eb6135ee7bed4212fb778eea Mon Sep 17 00:00:00 2001 From: Ruediger Klaehn Date: Mon, 8 Jun 2026 13:52:38 +0300 Subject: [PATCH 4/7] Reconfigure pacer, add per packet logging --- zero-rtt-blob-service/src/main.rs | 137 +++++++++++++++++++----------- 1 file changed, 88 insertions(+), 49 deletions(-) diff --git a/zero-rtt-blob-service/src/main.rs b/zero-rtt-blob-service/src/main.rs index 44d2ce7..e4107b9 100644 --- a/zero-rtt-blob-service/src/main.rs +++ b/zero-rtt-blob-service/src/main.rs @@ -70,30 +70,38 @@ struct ServerArgs { #[derive(Parser)] struct ClientArgs { - /// Ticket of the server to connect to, in `@,,...` form. - /// The ticket must contain at least one IP address — relay-only tickets are - /// refused because 0-RTT via a relay is not interesting for this demo. - ticket: String, - #[command(subcommand)] op: ClientOp, - - /// Discard the in-memory token store between rounds, so no stored - /// token is presented to the server. - #[clap(long)] - no_tokens: bool, - - /// Number of requests to run. - #[clap(long = "n", default_value = "2")] - n: u32, } #[derive(Subcommand)] enum ClientOp { - /// Upload up to 10 KiB of data read from stdin. Prints the BLAKE3 hash. - Put, + /// Upload up to 10 KiB of data from stdin, then re-GET it for the remaining + /// rounds. Prints the BLAKE3 hash. + Put { + #[command(flatten)] + common: CommonArgs, + }, /// Download the blob with the given hex-encoded BLAKE3 hash to stdout. - Get { hash: String }, + Get { + #[command(flatten)] + common: CommonArgs, + /// Hex-encoded BLAKE3 hash to download. + hash: String, + }, +} + +/// Arguments shared by every client operation, flattened into each subcommand so +/// the ticket and flags come *after* `put`/`get` (e.g. `client put -n 10`). +#[derive(clap::Args)] +struct CommonArgs { + /// Ticket of the server to connect to. Must contain at least one IP address — + /// relay-only tickets are refused because 0-RTT via a relay isn't interesting here. + ticket: String, + + /// Number of requests to run: round 0 (cold full handshake) plus the 0-RTT rounds. + #[clap(short = 'n', long = "rounds", default_value = "2")] + n: u32, } // --------------------------------------------------------------------------- @@ -108,7 +116,7 @@ async fn run_server(args: ServerArgs) -> Result<()> { // relays disabled, so the client always dials the server directly. let endpoint = Endpoint::builder(presets::N0) .alpns(vec![ALPN.to_vec()]) - .transport_config(qlog_transport_config("zero-rtt-blob-service-server")) + .transport_config(server_transport_config("zero-rtt-blob-service-server")) .secret_key(get_or_generate_secret_key()?) .bind() .await?; @@ -240,7 +248,10 @@ async fn handle_request( // --------------------------------------------------------------------------- async fn run_client(args: ClientArgs) -> Result<()> { - let ticket = EndpointTicket::from_str(&args.ticket).std_context("invalid ticket")?; + let (ticket, n) = match &args.op { + ClientOp::Put { common } | ClientOp::Get { common, .. } => (&common.ticket, common.n), + }; + let ticket = EndpointTicket::from_str(ticket).std_context("invalid ticket")?; let remote: EndpointAddr = ticket.into(); if remote.ip_addrs().next().is_none() { n0_error::bail_any!("ticket has no IP addresses — this demo requires direct IP paths"); @@ -250,31 +261,23 @@ async fn run_client(args: ClientArgs) -> Result<()> { remote.ip_addrs().copied().collect::>() ); - let shared = if args.no_tokens { - None - } else { - Some(client_endpoint().await?) - }; + let endpoint = client_endpoint().await?; match &args.op { - ClientOp::Put => { - let endpoint = match &shared { - Some(ep) => ep.clone(), - None => client_endpoint().await?, - }; + ClientOp::Put { .. } => { let t0 = Instant::now(); let request = build_put_request().await?; let hash_bytes = do_round(&endpoint, remote.clone(), request).await?; print_hash(&hash_bytes)?; eprintln!("put: {} us", t0.elapsed().as_micros()); - if args.n > 1 { + if n > 1 { let Ok(bytes): std::result::Result<[u8; 32], _> = hash_bytes.as_slice().try_into() else { n0_error::bail_any!("expected 32-byte hash, got {} bytes", hash_bytes.len()); }; let hash = Hash::from_bytes(bytes); let request = build_get_request(&hash); - for round in 1..args.n { + for round in 1..n { let t0 = Instant::now(); let resp = do_round(&endpoint, remote.clone(), request.clone()).await?; // Don't dump the (potentially large) blob — just echo its hash so @@ -283,31 +286,18 @@ async fn run_client(args: ClientArgs) -> Result<()> { eprintln!("get {round}: {} us", t0.elapsed().as_micros()); } } - - if args.no_tokens { - endpoint.close().await; - } } - ClientOp::Get { hash } => { - let endpoint = match &shared { - Some(ep) => ep.clone(), - None => client_endpoint().await?, - }; + ClientOp::Get { hash, .. } => { let request = build_get_request(&parse_hash(hash)?); - for round in 0..args.n { + for round in 0..n { let t0 = Instant::now(); let resp = do_round(&endpoint, remote.clone(), request.clone()).await?; tokio::io::stdout().write_all(&resp).await.anyerr()?; eprintln!("get {round}: {} us", t0.elapsed().as_micros()); } - if args.no_tokens { - endpoint.close().await; - } } } - if let Some(ep) = shared { - ep.close().await; - } + endpoint.close().await; Ok(()) } @@ -327,6 +317,30 @@ fn qlog_transport_config(prefix: &str) -> QuicTransportConfig { .build() } +/// Server transport config: qlog plus a deliberately huge initial congestion +/// window. +/// +/// QUIC's pacer spreads roughly one congestion window over one RTT, and its +/// per-burst capacity is `window * 2ms / RTT`. At a ~250 ms RTT the default +/// ~14 KiB window dribbles a multi-packet response out one packet every ~20 ms, +/// which buries the 0-RTT / token win under congestion-control pacing. Our +/// responses are bounded at `MAX_BLOB` (16 KiB), so we set a window large enough +/// (`16 KiB * RTT / 2ms` ≈ 2 MiB; we use a round 8 MiB for margin) that the pacer +/// bursts the whole response in a single flight. Only ever 16 KiB actually flies. +/// +/// This is a DEMO setting on a known-good path. A real server facing arbitrary +/// clients must not do this — see the anti-amplification and congestion-fairness +/// caveats. It does not affect the token A/B: an unvalidated (no-token) connection +/// is still capped at the 3x amplification budget regardless of the window. +fn server_transport_config(prefix: &str) -> QuicTransportConfig { + let mut congestion = noq_proto::congestion::NewRenoConfig::default(); + congestion.initial_window(8 * 1024 * 1024); + QuicTransportConfig::builder() + .qlog_from_env(prefix) + .congestion_controller_factory(Arc::new(congestion)) + .build() +} + async fn do_round(endpoint: &Endpoint, remote: EndpointAddr, request: Bytes) -> Result> { let connecting = endpoint .connect_with_opts(remote, ALPN, Default::default()) @@ -343,10 +357,11 @@ async fn do_round(endpoint: &Endpoint, remote: EndpointAddr, request: Bytes) -> // Writing inline and awaiting it guarantees the bytes are buffered as // 0-RTT data first. This matches iroh's own 0-RTT test. write_request(send, request.clone()).await?; + let recv_t0 = Instant::now(); match zrtt.handshake_completed().await? { ZeroRttStatus::Accepted(conn) => { println!("0-RTT accepted"); - (conn, recv.read_to_end(MAX_BLOB).await.anyerr()?) + (conn, read_response(&mut recv, recv_t0).await?) } ZeroRttStatus::Rejected(conn) => { println!("0-RTT rejected; retrying on a fresh stream"); @@ -388,7 +403,31 @@ async fn roundtrip(conn: &Connection, request: &[u8]) -> Result> { let (mut send, mut recv) = conn.open_bi().await.anyerr()?; send.write_all(request).await.anyerr()?; send.finish().anyerr()?; - recv.read_to_end(MAX_BLOB).await.anyerr() + let recv_t0 = Instant::now(); + read_response(&mut recv, recv_t0).await +} + +/// Read a response to its end, logging each chunk's arrival time relative to `t0` +/// (when the request was sent). A stall in the response stream — e.g. the +/// anti-amplification stall when no validation token is presented — shows up as a +/// gap between consecutive lines, while smooth pacing shows up as evenly spaced +/// lines a few ms apart. +async fn read_response(recv: &mut RecvStream, t0: Instant) -> Result> { + let mut data = Vec::new(); + let mut buf = [0u8; 4096]; + while let Some(n) = recv.read(&mut buf).await.anyerr()? { + data.extend_from_slice(&buf[..n]); + eprintln!( + " recv +{:>6} us | +{:>5} B | {:>6} B total", + t0.elapsed().as_micros(), + n, + data.len() + ); + if data.len() > MAX_BLOB { + n0_error::bail_any!("response exceeds {MAX_BLOB} bytes"); + } + } + Ok(data) } async fn write_request(mut send: SendStream, request: Bytes) -> Result<()> { From 9255695fbb7caec4d53902635e3fde9f01311b27 Mon Sep 17 00:00:00 2001 From: Ruediger Klaehn Date: Mon, 8 Jun 2026 14:46:14 +0300 Subject: [PATCH 5/7] Allow fixed port --- zero-rtt-blob-service/src/main.rs | 35 ++++++++++++++++++++++++------- 1 file changed, 28 insertions(+), 7 deletions(-) diff --git a/zero-rtt-blob-service/src/main.rs b/zero-rtt-blob-service/src/main.rs index e4107b9..7b2a9aa 100644 --- a/zero-rtt-blob-service/src/main.rs +++ b/zero-rtt-blob-service/src/main.rs @@ -66,6 +66,13 @@ struct ServerArgs { /// Disable issuance of NEW_TOKEN validation tokens. #[clap(long)] no_tokens: bool, + + /// Bind a fixed UDP port (IPv4 only) instead of a random one. With a fixed + /// `IROH_SECRET` this keeps the ticket stable across restarts. Binding IPv4 + /// only also stops advertising an IPv6 address that IPv4-only clients would + /// otherwise waste a dial attempt on. + #[clap(long)] + port: Option, } #[derive(Parser)] @@ -114,12 +121,20 @@ async fn run_server(args: ServerArgs) -> Result<()> { // a reachable direct address. It doesn't affect the measurement — the ticket is // built from IP addresses only (see `wait_for_ip_addr`), and the client runs with // relays disabled, so the client always dials the server directly. - let endpoint = Endpoint::builder(presets::N0) + let mut builder = Endpoint::builder(presets::N0) .alpns(vec![ALPN.to_vec()]) .transport_config(server_transport_config("zero-rtt-blob-service-server")) - .secret_key(get_or_generate_secret_key()?) - .bind() - .await?; + .secret_key(get_or_generate_secret_key()?); + if let Some(port) = args.port { + // Replace the default dual-stack ephemeral binds with a single fixed + // IPv4 socket, so the ticket is stable across restarts and carries no + // (unreachable, for IPv4-only clients) IPv6 address. + builder = builder + .clear_ip_transports() + .bind_addr(format!("0.0.0.0:{port}").as_str()) + .anyerr()?; + } + let endpoint = builder.bind().await?; let ip_addr = wait_for_ip_addr(&endpoint).await?; eprintln!( "direct addresses: {:?}", @@ -414,15 +429,21 @@ async fn roundtrip(conn: &Connection, request: &[u8]) -> Result> { /// lines a few ms apart. async fn read_response(recv: &mut RecvStream, t0: Instant) -> Result> { let mut data = Vec::new(); - let mut buf = [0u8; 4096]; + let mut buf = [0u8; MAX_BLOB]; + let mut prev = Duration::ZERO; while let Some(n) = recv.read(&mut buf).await.anyerr()? { data.extend_from_slice(&buf[..n]); + let elapsed = t0.elapsed(); + // `Δ` is the gap since the previous chunk: tiny for a paced/bursted + // response, but ~1 RTT on the chunk right after an amplification stall. eprintln!( - " recv +{:>6} us | +{:>5} B | {:>6} B total", - t0.elapsed().as_micros(), + " recv +{:>6} us Δ{:>+7} us | +{:>5} B | {:>6} B total", + elapsed.as_micros(), + elapsed.saturating_sub(prev).as_micros(), n, data.len() ); + prev = elapsed; if data.len() > MAX_BLOB { n0_error::bail_any!("response exceeds {MAX_BLOB} bytes"); } From 337724f6a127e8042a8f834dc7cb6c2cf7c463cb Mon Sep 17 00:00:00 2001 From: Ruediger Klaehn Date: Mon, 8 Jun 2026 15:07:49 +0300 Subject: [PATCH 6/7] Don't dump the data even on get --- zero-rtt-blob-service/src/main.rs | 7 ++++--- 1 file changed, 4 insertions(+), 3 deletions(-) diff --git a/zero-rtt-blob-service/src/main.rs b/zero-rtt-blob-service/src/main.rs index 7b2a9aa..eb9d1c0 100644 --- a/zero-rtt-blob-service/src/main.rs +++ b/zero-rtt-blob-service/src/main.rs @@ -32,7 +32,7 @@ use iroh_tickets::endpoint::EndpointTicket; use n0_error::{Result, StdResultExt}; use noq_proto::{BloomTokenLog, NoneTokenLog}; use papaya::HashMap; -use tokio::io::{AsyncReadExt, AsyncWriteExt}; +use tokio::io::AsyncReadExt; use tracing::{info, trace}; const ALPN: &[u8] = b"zero-rtt-blob-service/v1"; @@ -89,7 +89,8 @@ enum ClientOp { #[command(flatten)] common: CommonArgs, }, - /// Download the blob with the given hex-encoded BLAKE3 hash to stdout. + /// Download the blob with the given hex-encoded BLAKE3 hash. Prints the + /// BLAKE3 hash of what was received (for verification), not the data itself. Get { #[command(flatten)] common: CommonArgs, @@ -307,7 +308,7 @@ async fn run_client(args: ClientArgs) -> Result<()> { for round in 0..n { let t0 = Instant::now(); let resp = do_round(&endpoint, remote.clone(), request.clone()).await?; - tokio::io::stdout().write_all(&resp).await.anyerr()?; + println!("{}", blake3::hash(&resp).to_hex()); eprintln!("get {round}: {} us", t0.elapsed().as_micros()); } } From 27208b8986aaf08eb18b12fd7c38637614276eb7 Mon Sep 17 00:00:00 2001 From: Ruediger Klaehn Date: Mon, 8 Jun 2026 15:16:38 +0300 Subject: [PATCH 7/7] Flip no-token to token By default tokens are off now! --- zero-rtt-blob-service/src/main.rs | 41 +++++++++++++++++++------------ 1 file changed, 25 insertions(+), 16 deletions(-) diff --git a/zero-rtt-blob-service/src/main.rs b/zero-rtt-blob-service/src/main.rs index eb9d1c0..80bfc37 100644 --- a/zero-rtt-blob-service/src/main.rs +++ b/zero-rtt-blob-service/src/main.rs @@ -63,9 +63,13 @@ enum Mode { #[derive(Parser)] struct ServerArgs { - /// Disable issuance of NEW_TOKEN validation tokens. + /// Issue and validate NEW_TOKEN address-validation tokens. Off by default + /// (matching iroh's default), so a 0-RTT response larger than the 3x + /// amplification budget stalls until the handshake validates the client's + /// address. With this flag the client's replayed token validates the address + /// up front and the stall disappears. #[clap(long)] - no_tokens: bool, + tokens: bool, /// Bind a fixed UDP port (IPv4 only) instead of a random one. With a fixed /// `IROH_SECRET` this keeps the ticket stable across restarts. Binding IPv4 @@ -143,7 +147,7 @@ async fn run_server(args: ServerArgs) -> Result<()> { ); println!("ticket: {}", EndpointTicket::from(ip_addr)); - let server_config = Arc::new(make_server_config(&endpoint, !args.no_tokens)); + let server_config = Arc::new(make_server_config(&endpoint, args.tokens)); let store: Arc> = Arc::new(HashMap::new()); let accept = async move { @@ -189,12 +193,13 @@ async fn wait_for_ip_addr(endpoint: &Endpoint) -> Result { /// 3x anti-amplification limit on 0-RTT connections so large 0.5-RTT responses /// flow immediately. /// -/// With `tokens` disabled (`--no-tokens`) the server neither issues (`sent` = 0) -/// nor validates ([`NoneTokenLog`] ignores any token a client presents), so every -/// 0-RTT connection's address stays unvalidated until the handshake completes and -/// a large 0.5-RTT response stalls on the 3x budget. Note that the default config -/// (`ValidationTokenConfig::default()`, used by `Incoming::accept`) issues and -/// validates by default, so we must set this explicitly rather than rely on it. +/// With `tokens` disabled (the default — pass `--tokens` to enable) the server +/// neither issues (`sent` = 0) nor validates ([`NoneTokenLog`] ignores any token a +/// client presents), so every 0-RTT connection's address stays unvalidated until +/// the handshake completes and a large 0.5-RTT response stalls on the 3x budget. +/// Note that `ValidationTokenConfig::default()` (used by `Incoming::accept`) +/// actually issues and validates, so the disabled case must be set explicitly +/// rather than relying on the default. fn make_server_config(endpoint: &Endpoint, tokens: bool) -> ServerConfig { let mut validation_token = ValidationTokenConfig::default(); if tokens { @@ -358,6 +363,12 @@ fn server_transport_config(prefix: &str) -> QuicTransportConfig { } async fn do_round(endpoint: &Endpoint, remote: EndpointAddr, request: Bytes) -> Result> { + // Start the receive clock before connecting, so the per-chunk `recv` timing + // reflects the *full* request latency from this client's point of view: + // ~1 RTT to the first byte for an accepted 0-RTT round, but ~2 RTT for a cold + // round (where the handshake completes inside `connecting.await` before the + // request can even be sent). + let t0 = Instant::now(); let connecting = endpoint .connect_with_opts(remote, ALPN, Default::default()) .await?; @@ -373,15 +384,14 @@ async fn do_round(endpoint: &Endpoint, remote: EndpointAddr, request: Bytes) -> // Writing inline and awaiting it guarantees the bytes are buffered as // 0-RTT data first. This matches iroh's own 0-RTT test. write_request(send, request.clone()).await?; - let recv_t0 = Instant::now(); match zrtt.handshake_completed().await? { ZeroRttStatus::Accepted(conn) => { println!("0-RTT accepted"); - (conn, read_response(&mut recv, recv_t0).await?) + (conn, read_response(&mut recv, t0).await?) } ZeroRttStatus::Rejected(conn) => { println!("0-RTT rejected; retrying on a fresh stream"); - let resp = roundtrip(&conn, &request).await?; + let resp = roundtrip(&conn, &request, t0).await?; (conn, resp) } } @@ -389,7 +399,7 @@ async fn do_round(endpoint: &Endpoint, remote: EndpointAddr, request: Bytes) -> Err(connecting) => { println!("0-RTT not possible from our side"); let conn = connecting.await.anyerr()?; - let resp = roundtrip(&conn, &request).await?; + let resp = roundtrip(&conn, &request, t0).await?; (conn, resp) } }; @@ -415,12 +425,11 @@ async fn do_round(endpoint: &Endpoint, remote: EndpointAddr, request: Bytes) -> Ok(resp) } -async fn roundtrip(conn: &Connection, request: &[u8]) -> Result> { +async fn roundtrip(conn: &Connection, request: &[u8], t0: Instant) -> Result> { let (mut send, mut recv) = conn.open_bi().await.anyerr()?; send.write_all(request).await.anyerr()?; send.finish().anyerr()?; - let recv_t0 = Instant::now(); - read_response(&mut recv, recv_t0).await + read_response(&mut recv, t0).await } /// Read a response to its end, logging each chunk's arrival time relative to `t0`