joule-code/CHANGELOG.md

@@ -4,6 +4,21 @@ All notable changes to the Joule Code standard are recorded here. The format fol
 
 ## [Unreleased]
 
+### Changed — `joule-moq-relay`: one object held once, and a socket that keeps what arrives
+
+- **`limits.quic_socket_buffer_bytes`** (4 MiB by default, and for the links a relay dials too): the
+  size asked of the kernel for the UDP socket's buffers. Each fan-out's acknowledgements arrive
+  together, and past the buffer the kernel drops them, which reads as loss. The system caps the
+  size (`net.core.rmem_max` on Linux), and the relay logs a warning when it gets less.
+- **An object is held once**, from when it is read until the last subscriber's QUIC stream has it
+  acknowledged: the cache, each session's queue and each QUIC send buffer share one allocation
+  (`SendStream::write_parts`, `SubgroupCodec::encode_object_head`). The fan-out no longer copies a
+  payload per subscriber while it holds the relay's lock. A WebTransport or WebSocket session, and
+  a session bridged to an earlier draft, still has its own copy in its transport.
+- **`examples/fanout_load.rs`**: one publisher, N subscribers of one track and a bystander on its
+  own, reporting the relay's memory and CPU, what arrived and how late, and the packets the kernel
+  dropped at each end.
+
 ### Added — `joule-moq-relay`: relay-to-relay cascading
 
 - **`[[upstream]]`**: relays this one dials, over native QUIC (`moqt://`), WebTransport

joule-code/joule-code-rs/crates/joule-moq-relay/Cargo.toml

@@ -32,7 +32,7 @@ tokio-rustls       = { version = "0.26", default-features = false, features = ["
 # The platform's roots, for an upstream whose front holds a public certificate.
 rustls-native-certs = "0.8"
 tokio              = { version = "1", features = ["rt-multi-thread", "macros", "sync", "time", "net", "io-util", "signal"] }
-bytes              = "1"
+bytes              = "1.9"
 futures            = "0.3"
 async-trait        = "0.1"
 tracing            = "0.1"

joule-code/joule-code-rs/crates/joule-moq-relay/README.md

@@ -326,6 +326,18 @@ not verify.
   unaltered and nothing is re-signed. A refused object ends its downstream streams rather
   than leaving a hole in them.
 
+## Load
+
+`examples/fanout_load.rs` measures the fan-out: one publisher, N subscribers of one track over
+native QUIC, and a bystander on a track of its own, whose lateness is what the audience costs
+everyone else on the relay. It reports the relay's resident memory and CPU, how many objects
+arrived, how late, and the packets the kernel dropped at each end. A deployment sets
+`net.core.rmem_max` to at least `limits.quic_socket_buffer_bytes` (4 MiB by default): the relay
+asks for that, Linux quietly gives no more than its limit, and the relay logs a warning. An object
+is held once from when it is read until the last subscriber's QUIC stream has it acknowledged: the
+cache, each session's queue and each QUIC send buffer share one allocation. A WebTransport or
+WebSocket session, and a session bridged to an earlier draft, has its own copy in its transport.
+
 ## Tokens and room grants
 
 `[[auth.grants]]` rules name a token by the SHA-256 of its value (Token Type 0), a

joule-code/joule-code-rs/crates/joule-moq-relay/examples/fanout_load.rs

@@ -0,0 +1,316 @@
+//! A load test of the relay's fan-out: one publisher and N subscribers of one track, over native
+//! QUIC on loopback, the relay a process of its own, so what it spends is its own.
+//!
+//! ```text
+//! fanout_load --subscribers N [--payload B] [--rate R] [--seconds S] [--relay-threads T] [--relay-cpus 0-7]
+//!             [--keyframe-every K --keyframe-bytes B] [--per-endpoint E]
+//! ```
+//!
+//! It prints one JSON object: the relay's resident memory before the subscribers, with them idle
+//! and at its peak; its CPU over the run; how many objects arrived of how many sent; and how late.
+//! A subscriber's lateness is from the publisher's clock, which is the subscriber's: one host. Run
+//! the driver on other CPUs than the relay's (`taskset -c 8-31 fanout_load ...`), so that the clients
+//! never take the relay's time. Every wait is bounded, and the relay exits when the driver's end of
+//! its standard input closes, so a killed run leaves nothing behind. Linux, for `/proc`.
+//!
+//! A **bystander** runs beside the audience: a track of its own with one subscriber, at the same
+//! rate and with small objects. It shares nothing with the audience but the relay, so its lateness
+//! is what the audience's fan-out costs everyone else on the relay.
+
+use joule_moq::draft22::data::{SubgroupHeader, SubgroupId, SubgroupObject};
+use joule_moq::draft22::{Parameters, UniEvent};
+use joule_moq::{TrackName, TrackNamespace};
+use joule_moq_relay::client::{Client, ClientOptions};
+use joule_moq_relay::transport::quic;
+use joule_moq_relay::{RelayConfig, RelayServer};
+use std::io::{BufRead, Read, Write};
+use std::process::{Child, Command, Stdio};
+use std::sync::atomic::{AtomicU64, Ordering};
+use std::sync::Arc;
+use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH};
+use tokio::time::timeout;
+
+/// Latency buckets of 50 µs, up to 10 s; the last holds everything later.
+const BUCKET_US: u64 = 50;
+const BUCKETS: usize = 200_000;
+const STEP: Duration = Duration::from_secs(30);
+
+fn arg<T: std::str::FromStr>(args: &[String], flag: &str, default: T) -> T {
+    args.iter().position(|a| a == flag).and_then(|i| args.get(i + 1)).and_then(|v| v.parse().ok()).unwrap_or(default)
+}
+
+fn relay_main(args: &[String]) {
+    let threads: usize = arg(args, "--relay-threads", 8);
+    let rt = tokio::runtime::Builder::new_multi_thread().worker_threads(threads).enable_all().build().expect("a runtime");
+    rt.block_on(async {
+        let mut c = RelayConfig::loopback();
+        c.websocket_listen = None;
+        c.limits.max_sessions = 1 << 20;
+        let server = RelayServer::start(c).await.expect("the relay starts");
+        let cert: String = server.identity.chain[0].as_ref().iter().map(|b| format!("{b:02x}")).collect();
+        println!("{} {cert}", server.quic_addr.expect("a UDP endpoint"));
+        std::io::stdout().flush().expect("stdout");
+        // Until the driver goes.
+        tokio::task::spawn_blocking(|| {
+            let mut sink = Vec::new();
+            let _ = std::io::stdin().read_to_end(&mut sink);
+        })
+        .await
+        .expect("the wait");
+        std::process::exit(0);
+    });
+}
+
+fn proc_kb(pid: u32, field: &str) -> u64 {
+    std::fs::read_to_string(format!("/proc/{pid}/status")).ok().and_then(|s| s.lines().find(|l| l.starts_with(field)).and_then(|l| l.split_whitespace().nth(1)?.parse().ok())).unwrap_or(0)
+}
+
+/// The process's user and system CPU, in seconds.
+fn proc_cpu(pid: u32) -> f64 {
+    let Ok(s) = std::fs::read_to_string(format!("/proc/{pid}/stat")) else { return 0.0 };
+    let after = s.rsplit(')').next().unwrap_or("");
+    let f: Vec<&str> = after.split_whitespace().collect();
+    let t = |i: usize| f.get(i).and_then(|v| v.parse::<u64>().ok()).unwrap_or(0);
+    (t(11) + t(12)) as f64 / 100.0
+}
+
+/// The packets the kernel dropped on UDP sockets on loopback: the relay's, at `port`, and the rest.
+fn udp_drops(port: u16) -> (u64, u64) {
+    let Ok(s) = std::fs::read_to_string("/proc/net/udp") else { return (0, 0) };
+    let (mut relay, mut others) = (0, 0);
+    for l in s.lines().skip(1) {
+        let f: Vec<&str> = l.split_whitespace().collect();
+        let (Some(local), Some(drops)) = (f.get(1), f.last().and_then(|d| d.parse::<u64>().ok())) else { continue };
+        let Some((host, p)) = local.split_once(':') else { continue };
+        if host != "0100007F" {
+            continue;
+        }
+        if u16::from_str_radix(p, 16) == Ok(port) {
+            relay += drops;
+        } else {
+            others += drops;
+        }
+    }
+    (relay, others)
+}
+
+fn now_us() -> u64 {
+    SystemTime::now().duration_since(UNIX_EPOCH).expect("after 1970").as_micros() as u64
+}
+
+fn spawn_relay(args: &[String]) -> (Child, std::net::SocketAddr, Vec<u8>) {
+    let exe = std::env::current_exe().expect("this binary");
+    let threads: usize = arg(args, "--relay-threads", 8);
+    let cpus: String = arg(args, "--relay-cpus", "0-7".to_string());
+    let mut child = Command::new("taskset")
+        .args(["-c", &cpus])
+        .arg(&exe)
+        .args(["relay", "--relay-threads", &threads.to_string()])
+        .stdin(Stdio::piped())
+        .stdout(Stdio::piped())
+        .stderr(Stdio::null())
+        .spawn()
+        .expect("taskset runs");
+    let mut line = String::new();
+    std::io::BufReader::new(child.stdout.take().expect("its stdout")).read_line(&mut line).expect("its address");
+    let (addr, cert) = line.trim().split_once(' ').expect("an address and a certificate");
+    let cert = (0..cert.len() / 2).map(|i| u8::from_str_radix(&cert[2 * i..2 * i + 2], 16).expect("hex")).collect();
+    (child, addr.parse().expect("an address"), cert)
+}
+
+struct Seen {
+    objects: AtomicU64,
+    buckets: Vec<AtomicU64>,
+}
+
+impl Seen {
+    fn note(&self, late_us: u64) {
+        self.objects.fetch_add(1, Ordering::Relaxed);
+        self.buckets[((late_us / BUCKET_US) as usize).min(BUCKETS - 1)].fetch_add(1, Ordering::Relaxed);
+    }
+
+    /// The latency, in milliseconds, at each fraction.
+    fn at(&self, fractions: &[f64]) -> Vec<f64> {
+        let counts: Vec<u64> = self.buckets.iter().map(|b| b.load(Ordering::Relaxed)).collect();
+        let total: u64 = counts.iter().sum();
+        fractions
+            .iter()
+            .map(|f| {
+                let want = ((total as f64) * f).ceil() as u64;
+                let mut sum = 0;
+                for (i, c) in counts.iter().enumerate() {
+                    sum += c;
+                    if sum >= want.max(1) {
+                        return (i as u64 * BUCKET_US) as f64 / 1000.0;
+                    }
+                }
+                0.0
+            })
+            .collect()
+    }
+}
+
+fn driver(args: &[String]) {
+    let subscribers: usize = arg(args, "--subscribers", 100);
+    let payload: usize = arg(args, "--payload", 8192);
+    let rate: u64 = arg(args, "--rate", 30);
+    let seconds: u64 = arg(args, "--seconds", 20);
+    let keyframe_every: u64 = arg(args, "--keyframe-every", 0);
+    let keyframe_bytes: usize = arg(args, "--keyframe-bytes", 0);
+    let per_endpoint: usize = arg(args, "--per-endpoint", 8);
+    let (child, addr, cert) = spawn_relay(args);
+    let pid = child.id();
+    let rt = tokio::runtime::Builder::new_multi_thread().enable_all().build().expect("a runtime");
+    rt.block_on(async move {
+        tokio::time::sleep(Duration::from_millis(300)).await;
+        let cert = rustls::pki_types::CertificateDer::from(cert);
+        let alpn = joule_moq::draft22::ALPN.to_vec();
+        let bind: std::net::SocketAddr = "127.0.0.1:0".parse().expect("an address");
+        let (space, track) = (TrackNamespace::from_strs(["load"]), TrackName::from_str("video"));
+        let base_kb = proc_kb(pid, "VmRSS");
+
+        let pub_ep = quic::client_endpoint(bind, vec![cert.clone()], vec![alpn.clone()], None).expect("an endpoint");
+        let conn = timeout(STEP, quic::connect(&pub_ep, addr, "localhost")).await.expect("in time").expect("a session");
+        let publisher = Arc::new(Client::connect(Arc::new(conn), ClientOptions::default()).await.expect("SETUP"));
+        let _published = publisher.publish(space.clone(), track.clone(), 1, Parameters::new(), Default::default()).await.expect("PUBLISH");
+
+        let seen = Arc::new(Seen { objects: AtomicU64::new(0), buckets: (0..BUCKETS).map(|_| AtomicU64::new(0)).collect() });
+        let endpoints: Vec<quinn::Endpoint> = (0..subscribers.div_ceil(per_endpoint)).map(|_| quic::client_endpoint(bind, vec![cert.clone()], vec![alpn.clone()], None).expect("an endpoint")).collect();
+        let connecting = Instant::now();
+        let gate = Arc::new(tokio::sync::Semaphore::new(128));
+        let mut joins = Vec::with_capacity(subscribers);
+        for i in 0..subscribers {
+            let ep = endpoints[i / per_endpoint].clone();
+            let (gate, space, track, seen) = (gate.clone(), space.clone(), track.clone(), seen.clone());
+            joins.push(tokio::spawn(async move {
+                let permit = gate.acquire_owned().await.expect("the gate");
+                let conn = timeout(STEP, quic::connect(&ep, addr, "localhost")).await.expect("in time").expect("a session");
+                let c = Client::connect(Arc::new(conn), ClientOptions::default()).await.expect("SETUP");
+                let held = c.subscribe(space, track, Parameters::new()).await.expect("SUBSCRIBE");
+                drop(permit);
+                tokio::spawn(async move {
+                    let _held = held;
+                    while let Some(mut r) = c.next_data(Duration::from_secs(3600)).await {
+                        while let Ok(Some(e)) = r.next().await {
+                            if let UniEvent::SubgroupObject(o) = e {
+                                let sent = u64::from_le_bytes(o.payload[..8].try_into().expect("a timestamp"));
+                                seen.note(now_us().saturating_sub(sent));
+                            }
+                        }
+                    }
+                });
+            }));
+        }
+        for j in joins {
+            j.await.expect("a subscriber");
+        }
+        let connect_s = connecting.elapsed().as_secs_f64();
+
+        // The bystander: a publisher and a subscriber of their own track.
+        let aside = Arc::new(Seen { objects: AtomicU64::new(0), buckets: (0..BUCKETS).map(|_| AtomicU64::new(0)).collect() });
+        let by_ep = quic::client_endpoint(bind, vec![cert.clone()], vec![alpn.clone()], None).expect("an endpoint");
+        let by_name = TrackName::from_str("bystander");
+        let conn = timeout(STEP, quic::connect(&by_ep, addr, "localhost")).await.expect("in time").expect("a session");
+        let by_pub = Arc::new(Client::connect(Arc::new(conn), ClientOptions::default()).await.expect("SETUP"));
+        let _by_published = by_pub.publish(space.clone(), by_name.clone(), 2, Parameters::new(), Default::default()).await.expect("PUBLISH");
+        let conn = timeout(STEP, quic::connect(&by_ep, addr, "localhost")).await.expect("in time").expect("a session");
+        let by_sub = Client::connect(Arc::new(conn), ClientOptions::default()).await.expect("SETUP");
+        let by_held = by_sub.subscribe(space.clone(), by_name, Parameters::new()).await.expect("SUBSCRIBE");
+        let by_seen = aside.clone();
+        tokio::spawn(async move {
+            let _held = by_held;
+            while let Some(mut r) = by_sub.next_data(Duration::from_secs(3600)).await {
+                while let Ok(Some(e)) = r.next().await {
+                    if let UniEvent::SubgroupObject(o) = e {
+                        by_seen.note(now_us().saturating_sub(u64::from_le_bytes(o.payload[..8].try_into().expect("a timestamp"))));
+                    }
+                }
+            }
+        });
+        tokio::time::sleep(Duration::from_secs(3)).await;
+        let idle_kb = proc_kb(pid, "VmRSS");
+        let idle_cpu_start = proc_cpu(pid);
+        tokio::time::sleep(Duration::from_secs(3)).await;
+        let idle_cores = (proc_cpu(pid) - idle_cpu_start) / 3.0;
+
+        // The run: one object to a group, at the rate, each stamped with when it left.
+        let drops_start = udp_drops(addr.port());
+        let cpu_start = proc_cpu(pid);
+        let driver_start = proc_cpu(std::process::id());
+        let started = Instant::now();
+        let by_tick = Duration::from_nanos(1_000_000_000 / rate);
+        let by_run = Duration::from_secs(seconds);
+        let by_sender = tokio::spawn(async move {
+            let mut tick = tokio::time::interval(by_tick);
+            let (t0, mut g) = (Instant::now(), 0u64);
+            while t0.elapsed() < by_run {
+                tick.tick().await;
+                let mut b = vec![0u8; 1024];
+                b[..8].copy_from_slice(&now_us().to_le_bytes());
+                let h = SubgroupHeader { track_alias: 2, group_id: g, subgroup_id: SubgroupId::Zero, publisher_priority: Some(128), properties: false, end_of_group: true, first_object: true };
+                let _ = timeout(STEP, by_pub.send_subgroup(h, &[SubgroupObject::normal(0, b)])).await;
+                g += 1;
+            }
+        });
+        let mut tick = tokio::time::interval(Duration::from_nanos(1_000_000_000 / rate));
+        let body = vec![0x5au8; payload.max(8)];
+        let mut sent = 0u64;
+        let mut lagging = 0u64;
+        let mut peak_kb = 0;
+        while started.elapsed() < Duration::from_secs(seconds) {
+            tick.tick().await;
+            let mut b = if keyframe_every != 0 && sent.is_multiple_of(keyframe_every) { vec![0x5au8; keyframe_bytes.max(8)] } else { body.clone() };
+            b[..8].copy_from_slice(&now_us().to_le_bytes());
+            let h = SubgroupHeader { track_alias: 1, group_id: sent, subgroup_id: SubgroupId::Zero, publisher_priority: Some(128), properties: false, end_of_group: true, first_object: true };
+            let t0 = Instant::now();
+            timeout(STEP, publisher.send_subgroup(h, &[SubgroupObject::normal(0, b)])).await.expect("in time").expect("a send");
+            if t0.elapsed() > Duration::from_nanos(1_000_000_000 / rate) {
+                lagging += 1;
+            }
+            sent += 1;
+            if sent.is_multiple_of(rate) {
+                peak_kb = peak_kb.max(proc_kb(pid, "VmRSS"));
+            }
+        }
+        let run_s = started.elapsed().as_secs_f64();
+        let run_cpu = proc_cpu(pid) - cpu_start;
+        let driver_cores = (proc_cpu(std::process::id()) - driver_start) / run_s;
+        let drops_end = udp_drops(addr.port());
+        let _ = by_sender.await;
+        tokio::time::sleep(Duration::from_secs(3)).await;
+        let got = seen.objects.load(Ordering::Relaxed);
+        let want = sent * subscribers as u64;
+        let l = seen.at(&[0.5, 0.9, 0.99, 0.999]);
+        let by = aside.at(&[0.5, 0.99, 0.999]);
+        println!(
+            "{{\"subscribers\":{subscribers},\"payload\":{payload},\"rate\":{rate},\"seconds\":{seconds},\"connect_s\":{connect_s:.1},\"rss_base_mb\":{:.1},\"rss_idle_mb\":{:.1},\"kb_per_idle_subscriber\":{:.1},\"idle_cores\":{idle_cores:.3},\"rss_peak_mb\":{:.1},\"relay_cores\":{:.2},\"cpu_us_per_delivery\":{:.1},\"sent\":{sent},\"send_lagged\":{lagging},\"delivered\":{got},\"delivered_ratio\":{:.4},\"gbps\":{:.2},\"p50_ms\":{:.2},\"p90_ms\":{:.2},\"p99_ms\":{:.2},\"p999_ms\":{:.2},\"driver_cores\":{driver_cores:.2},\"udp_drops_relay\":{},\"udp_drops_clients\":{},\"bystander_p50_ms\":{:.2},\"bystander_p99_ms\":{:.2},\"bystander_p999_ms\":{:.2}}}",
+            base_kb as f64 / 1024.0,
+            idle_kb as f64 / 1024.0,
+            idle_kb.saturating_sub(base_kb) as f64 / subscribers.max(1) as f64,
+            peak_kb.max(proc_kb(pid, "VmHWM")) as f64 / 1024.0,
+            run_cpu / run_s,
+            run_cpu * 1e6 / got.max(1) as f64,
+            got as f64 / want.max(1) as f64,
+            got as f64 * payload as f64 * 8.0 / run_s / 1e9,
+            l[0],
+            l[1],
+            l[2],
+            l[3],
+            drops_end.0.saturating_sub(drops_start.0),
+            drops_end.1.saturating_sub(drops_start.1),
+            by[0],
+            by[1],
+            by[2],
+        );
+        std::process::exit(0);
+    });
+}
+
+fn main() {
+    let args: Vec<String> = std::env::args().skip(1).collect();
+    if args.first().map(String::as_str) == Some("relay") {
+        relay_main(&args);
+    } else {
+        driver(&args);
+    }
+}

joule-code/joule-code-rs/crates/joule-moq-relay/relay.example.toml

@@ -47,6 +47,7 @@ max_relay_hops = 8
 upstream_reconnect_ms = 1000
 websocket_buffer_bytes = 67108864
 idle_timeout_ms = 30000
+quic_socket_buffer_bytes = 4194304  # the UDP socket's buffers; the system caps them (net.core.rmem_max on Linux)
 quic_send_window_bytes = 1048576
 quic_stream_window_bytes = 16777216  # a stream's flow-control window: more than the largest group a door holds
 quic_connection_window_bytes = 67108864  # a session's, all its streams together: what bounds the bytes it leaves unread; never below the stream window

joule-code/joule-code-rs/crates/joule-moq-relay/src/cascade.rs

@@ -349,7 +349,7 @@ async fn dial(relay: &Relay, link: &Link, url: &str) -> Result<(SharedConnection
             // among it: an upstream can leave this relay holding no more unread than a session
             // it accepted can.
             let t = crate::server::transport_config(cfg);
-            let ep = crate::transport::quic::client_endpoint_tls(bind, (*tls).clone(), Some(t)).map_err(|e| e.to_string())?;
+            let ep = crate::transport::quic::client_endpoint_tls(bind, (*tls).clone(), Some(t), cfg.limits.quic_socket_buffer_bytes).map_err(|e| e.to_string())?;
             let conn = tokio::time::timeout(step, crate::transport::quic::connect(&ep, addr, &host)).await.map_err(|_| "no QUIC session in time".to_string())?.map_err(|e| e.to_string())?;
             Ok((Arc::new(conn), Some(ep)))
         }

joule-code/joule-code-rs/crates/joule-moq-relay/src/config.rs

@@ -245,6 +245,13 @@ pub struct LimitsConfig {
     /// QUIC idle timeout.
     #[serde(default = "d_idle")]
     pub idle_timeout_ms: u64,
+    /// Bytes the kernel may hold for the relay's UDP socket in each direction, or 0 for the
+    /// system's default. The system caps it (`net.core.rmem_max` on Linux, often 212992), and the
+    /// relay logs a warning when it is given less than this. Each fan-out's acknowledgements arrive
+    /// together: at 2,000 sessions of 2 Mbit/s on loopback, a buffer of 212992 dropped about a million
+    /// packets in 20 s, and 4 MiB dropped none at that steady load.
+    #[serde(default = "d_socket_buffer")]
+    pub quic_socket_buffer_bytes: usize,
     /// Bytes a QUIC or WebTransport session may have sent and unacknowledged, or
     /// buffered, before a write waits. Kept near the path's bandwidth-delay product, so
     /// what waits waits in the relay's scheduler, where priority and delivery timeouts
@@ -332,6 +339,9 @@ fn d_upstream() -> u64 {
 fn d_fill() -> u64 {
     2_000
 }
+fn d_socket_buffer() -> usize {
+    4 << 20
+}
 fn d_writes() -> usize {
     8
 }
@@ -380,6 +390,7 @@ impl Default for LimitsConfig {
             goaway_timeout_ms: d_goaway(),
             websocket_buffer_bytes: d_ws_buffer(),
             idle_timeout_ms: d_idle(),
+            quic_socket_buffer_bytes: d_socket_buffer(),
             quic_send_window_bytes: d_send_window(),
             quic_stream_window_bytes: d_stream_window(),
             quic_connection_window_bytes: d_connection_window(),

joule-code/joule-code-rs/crates/joule-moq-relay/src/relay.rs

@@ -441,7 +441,8 @@ pub struct ObjectIn<'a> {
     pub header: &'a SubgroupHeader,
     pub subgroup_id: u64,
     pub first_in_stream: Option<u64>,
-    pub object: &'a SubgroupObject,
+    /// The object, held once: the cache, each subscriber's queue and each QUIC send buffer share it.
+    pub object: Arc<SubgroupObject>,
     pub arrived: Instant,
     pub object_timeout: u64,
     pub signed: Option<(u64, &'static str)>,
@@ -875,6 +876,7 @@ impl Relay {
     /// stream is the one the relay takes its subgroup from (§7.3): the cache counts it
     /// open from its first object, and [`Relay::cache_stream_ended`] closes the count.
     pub fn fan_out_object(&self, key: &TrackKey, o: ObjectIn<'_>, started: &mut HashSet<u64>) -> bool {
+        let payload = crate::sched::shared_payload(&o.object);
         let mut st = self.inner.state.lock().unwrap();
         let receipts_key = self.receipts_of(key);
         let is_receipts = self.is_receipts_track(&key.1);
@@ -914,7 +916,7 @@ impl Relay {
             publisher_priority: priority,
             properties: o.object.properties.clone(),
             status: o.object.status,
-            payload: Bytes::from(o.object.payload.clone()),
+            payload: payload.clone(),
         };
         let size = cache::size_of(&cached);
         let ins = track.cache.insert(group, oid, cached);
@@ -958,7 +960,7 @@ impl Relay {
             };
             sub.sched.push(
                 LaneKey { sub: sub.id, kind: LaneKind::Subgroup { group, subgroup: o.subgroup_id } },
-                Item { op: Op::Object { header, object: o.object.clone() }, prio, deadline, meta: Some(meta) },
+                Item { op: Op::Object { header, object: o.object.clone(), payload: payload.clone() }, prio, deadline, meta: Some(meta) },
             );
         }
         if matches!(o.object.status, joule_moq::draft22::data::ObjectStatus::EndOfGroup | joule_moq::draft22::data::ObjectStatus::EndOfTrack) {
@@ -1003,6 +1005,7 @@ impl Relay {
     /// One object datagram from upstream (§11.2).
     pub fn fan_out_datagram(&self, key: &TrackKey, upstream: u64, d: &ObjectDatagram, arrived: Instant, signed: Option<(u64, &'static str)>) {
         use joule_moq::draft22::data::DatagramContent;
+        let shared = Arc::new(d.clone());
         let mut st = self.inner.state.lock().unwrap();
         let receipts_key = self.receipts_of(key);
         let rsessions: HashSet<u64> = st.tracks.get(&receipts_key).map(|t| t.downs.iter().map(|d| d.session).collect()).unwrap_or_default();
@@ -1041,11 +1044,9 @@ impl Relay {
             // For a datagram SUBGROUP_DELIVERY_TIMEOUT acts as OBJECT_DELIVERY_TIMEOUT (§5.2).
             let t = effective_timeout(object_timeout, effective_timeout(p.object_timeout, p.subgroup_timeout));
             let deadline = (t != 0 && !sub.receipts_track).then(|| arrived + Duration::from_millis(t));
-            let mut out = d.clone();
-            out.track_alias = sub.alias;
             let prio = Prio { subscriber: p.subscriber_priority, publisher: priority, group_rank: Prio::group_rank(p.order, d.group_id), fill_rank: 1, mode_rank: 0, within: d.object_id };
             let meta = ObjMeta { track: sub.render.clone(), payload_len: payload.len() as u64, object_id: d.object_id, receipts_subscribed: rsessions.contains(&sub.session), fetched: false, dvr: false, signed };
-            sub.sched.push(LaneKey { sub: sub.id, kind: LaneKind::Datagram { group: d.group_id, object: d.object_id } }, Item { op: Op::Datagram { datagram: out }, prio, deadline, meta: Some(meta) });
+            sub.sched.push(LaneKey { sub: sub.id, kind: LaneKind::Datagram { group: d.group_id, object: d.object_id } }, Item { op: Op::Datagram { datagram: shared.clone(), alias: sub.alias }, prio, deadline, meta: Some(meta) });
         }
         self.charge_cache(&mut st, track_id, d.group_id, size);
     }

joule-code/joule-code-rs/crates/joule-moq-relay/src/sched.rs

@@ -30,6 +30,7 @@
 //! Receipts tracks carry no deadline (WAI moq-streaming-format §4.1), so the receipts of
 //! an object dropped here still reach a subscriber of its receipts track.
 
+use bytes::Bytes;
 use crate::summary::Tally;
 use crate::transport::{BoxSend, SharedConnection, TransportError};
 use joule_moq::draft22::data::{FetchCodec, FetchEntry, FetchHeader, ObjectDatagram, SubgroupCodec, SubgroupHeader, SubgroupObject};
@@ -106,13 +107,16 @@ pub struct ObjMeta {
 }
 
 pub enum Op {
-    Object { header: SubgroupHeader, object: SubgroupObject },
+    /// `payload` is the object's payload, shared ([`shared_payload`]), so a session's queue and its
+    /// transport hold no copy of it.
+    Object { header: SubgroupHeader, object: Arc<SubgroupObject>, payload: Bytes },
     /// The subgroup is complete upstream: FIN the stream, then hold it to
     /// SUBGROUP_DELIVERY_TIMEOUT.
     Fin { subgroup_timeout: Option<Duration> },
     /// The upstream stream was reset: reset this one.
     Reset { code: u64 },
-    Datagram { datagram: ObjectDatagram },
+    /// One datagram of the upstream's, shared, and this subscription's Track Alias for it.
+    Datagram { datagram: Arc<ObjectDatagram>, alias: u64 },
     FetchEntry { request_id: u64, order: GroupOrder, entry: FetchEntry },
     FetchFin { request_id: u64, order: GroupOrder },
     /// End a fetch stream abruptly, opening it first if it is not (a fill that fails,
@@ -120,6 +124,18 @@ pub enum Op {
     FetchReset { request_id: u64, order: GroupOrder, code: u64 },
 }
 
+/// An object's payload as a buffer that keeps the object alive: every subscriber's queue and every
+/// QUIC send buffer that takes it share the one allocation.
+pub fn shared_payload(object: &Arc<SubgroupObject>) -> Bytes {
+    struct Payload(Arc<SubgroupObject>);
+    impl AsRef<[u8]> for Payload {
+        fn as_ref(&self) -> &[u8] {
+            &self.0.payload
+        }
+    }
+    Bytes::from_owner(Payload(object.clone()))
+}
+
 pub struct Item {
     pub op: Op,
     pub prio: Prio,
@@ -234,7 +250,7 @@ enum Then {
 struct Job {
     lane: LaneKey,
     send: Option<BoxSend>,
-    bytes: Vec<u8>,
+    parts: Vec<Bytes>,
     then: Then,
     deadline: Option<Instant>,
     priority: i32,
@@ -438,9 +454,9 @@ impl Sched {
         let conn = self.conn.clone();
         let done = self.done_tx.clone();
         tokio::spawn(async move {
-            let Job { lane, send, bytes, then, deadline, priority, meta, track } = job;
+            let Job { lane, send, parts, then, deadline, priority, meta, track } = job;
             let mut send = send;
-            let n = bytes.len() as u64;
+            let n = parts.iter().map(|p| p.len() as u64).sum::<u64>();
             // Opening the stream (which can wait for the peer's stream credit) and the
             // write are both bounded by the object's deadline.
             let work = async {
@@ -450,8 +466,8 @@ impl Sched {
                     send = Some(s);
                 }
                 let s = send.as_mut().expect("a stream");
-                if !bytes.is_empty() {
-                    s.write_all(&bytes).await?;
+                if n != 0 {
+                    s.write_parts(parts).await?;
                 }
                 Ok::<(), TransportError>(())
             };
@@ -526,7 +542,7 @@ impl Sched {
         }
         let priority = prio.transport_priority();
         match op {
-            Op::Object { header, object } => {
+            Op::Object { header, object, payload } => {
                 if expired(deadline) {
                     if let Some(m) = &meta {
                         self.count_dropped(m);
@@ -547,7 +563,7 @@ impl Sched {
                     self.subs.entry(key.sub).or_default().streams_opened += 1;
                 }
                 let lane = self.lanes.get_mut(&key).expect("the lane");
-                let enc = lane.subgroup.as_mut().expect("a subgroup codec").encode_object(&object, &mut bytes);
+                let enc = lane.subgroup.as_mut().expect("a subgroup codec").encode_object_head(&object, &mut bytes);
                 if enc.is_err() {
                     self.settle(key.sub, 1);
                     self.kill(key, StreamResetCode::INTERNAL_ERROR.0);
@@ -555,7 +571,8 @@ impl Sched {
                 }
                 lane.busy = true;
                 let send = lane.send.take();
-                self.spawn_job(Job { lane: key, send, bytes, then: Then::Nothing, deadline, priority, meta, track: None });
+                let parts = if payload.is_empty() { vec![bytes.into()] } else { vec![bytes.into(), payload] };
+                self.spawn_job(Job { lane: key, send, parts, then: Then::Nothing, deadline, priority, meta, track: None });
             }
             Op::Fin { subgroup_timeout } => {
                 if !lane.opened {
@@ -566,7 +583,7 @@ impl Sched {
                 lane.busy = true;
                 let send = lane.send.take();
                 let track = lane.track.clone();
-                self.spawn_job(Job { lane: key, send, bytes: vec![], then: Then::Finish(subgroup_timeout), deadline: None, priority, meta: None, track });
+                self.spawn_job(Job { lane: key, send, parts: vec![], then: Then::Finish(subgroup_timeout), deadline: None, priority, meta: None, track });
             }
             Op::Reset { code } => {
                 if let Some(s) = lane.send.as_mut() {
@@ -575,13 +592,17 @@ impl Sched {
                 self.settle(key.sub, 1);
                 self.remove_lane(key);
             }
-            Op::Datagram { datagram } => {
+            Op::Datagram { datagram, alias } => {
                 self.settle(key.sub, 1);
                 if expired(deadline) {
                     if let Some(m) = &meta {
                         self.count_dropped(m);
                     }
-                } else if let Ok(b) = datagram.encode() {
+                } else if let Ok(b) = {
+                    let mut out = (*datagram).clone();
+                    out.track_alias = alias;
+                    out.encode()
+                } {
                     let n = b.len() as u64;
                     if self.conn.send_datagram(b.into()).is_ok()
                         && let Some(m) = &meta
@@ -607,7 +628,7 @@ impl Sched {
                 }
                 lane.busy = true;
                 let send = lane.send.take();
-                self.spawn_job(Job { lane: key, send, bytes, then: Then::Nothing, deadline: None, priority, meta, track: None });
+                self.spawn_job(Job { lane: key, send, parts: vec![bytes.into()], then: Then::Nothing, deadline: None, priority, meta, track: None });
             }
             Op::FetchFin { request_id, order } | Op::FetchReset { request_id, order, .. } => {
                 let then = match op {
@@ -626,7 +647,7 @@ impl Sched {
                 let lane = self.lanes.get_mut(&key).expect("the lane");
                 lane.busy = true;
                 let send = lane.send.take();
-                self.spawn_job(Job { lane: key, send, bytes, then, deadline: None, priority, meta: None, track: None });
+                self.spawn_job(Job { lane: key, send, parts: if bytes.is_empty() { vec![] } else { vec![bytes.into()] }, then, deadline: None, priority, meta: None, track: None });
             }
         }
     }
@@ -796,6 +817,17 @@ impl Sched {
 mod tests {
     use super::*;
 
+    #[test]
+    fn a_shared_payload_is_the_objects_own_bytes() {
+        let object = Arc::new(SubgroupObject::normal(3, vec![7u8; 4096]));
+        let (a, b) = (shared_payload(&object), shared_payload(&object));
+        assert_eq!(a.as_ptr(), object.payload.as_ptr(), "no copy of the payload");
+        assert_eq!(b.as_ptr(), object.payload.as_ptr());
+        assert_eq!(&a[..], &object.payload[..]);
+        drop(object);
+        assert_eq!(a[4095], 7, "the buffer keeps the object alive");
+    }
+
     #[test]
     fn the_order_is_section_5_1_2s() {
         let p = |s, pp, g| Prio { subscriber: s, publisher: pp, group_rank: g, fill_rank: 1, mode_rank: 1, within: 0 };

joule-code/joule-code-rs/crates/joule-moq-relay/src/server.rs

@@ -63,7 +63,9 @@ impl RelayServer {
             let crypto = quinn::crypto::rustls::QuicServerConfig::try_from(tls).map_err(|e| e.to_string())?;
             let mut sc = quinn::ServerConfig::with_crypto(Arc::new(crypto));
             sc.transport_config(Arc::new(transport_config(&cfg)));
-            let ep = quinn::Endpoint::server(sc, addr).map_err(|e| format!("binding {addr}: {e}"))?;
+            let sock = crate::transport::quic::bind_udp(addr, cfg.limits.quic_socket_buffer_bytes).map_err(|e| format!("binding {addr}: {e}"))?;
+            let runtime = quinn::default_runtime().ok_or("no async runtime")?;
+            let ep = quinn::Endpoint::new(quinn::EndpointConfig::default(), Some(sc), sock, runtime).map_err(|e| format!("binding {addr}: {e}"))?;
             quic_addr = Some(ep.local_addr().map_err(|e| e.to_string())?);
             quic_task = Some(tokio::spawn(quic_accept(relay.clone(), ep.clone(), cfg.path.clone())));
             endpoint = Some(ep);

joule-code/joule-code-rs/crates/joule-moq-relay/src/session.rs

@@ -881,9 +881,10 @@ async fn upstream_subgroup(relay: Relay, ctx: Arc<SessionCtx>, h: joule_moq::dra
                     }
                 }
                 let is_first = first == Some(o.object_id);
+                let o = Arc::new(o);
                 let taken = relay.fan_out_object(
                     &key,
-                    ObjectIn { upstream: ctx.id, header: &h, subgroup_id: sid, first_in_stream: first, object: &o, arrived, object_timeout, signed },
+                    ObjectIn { upstream: ctx.id, header: &h, subgroup_id: sid, first_in_stream: first, object: o.clone(), arrived, object_timeout, signed },
                     &mut started,
                 );
                 if is_first {

joule-code/joule-code-rs/crates/joule-moq-relay/src/transport/mod.rs

@@ -84,6 +84,15 @@ pub type TResult<T> = Result<T, TransportError>;
 pub trait SendStream: Send + Sync {
     /// Write all of `buf`, waiting for flow control as the transport requires.
     async fn write_all(&mut self, buf: &[u8]) -> TResult<()>;
+    /// Write `parts`, in order, as one write. A transport that keeps what it is given takes the
+    /// buffers themselves, so one object's payload is held once however many sessions are
+    /// sending it; any other copies them, here.
+    async fn write_parts(&mut self, parts: Vec<Bytes>) -> TResult<()> {
+        match parts.as_slice() {
+            [one] => self.write_all(one).await,
+            many => self.write_all(&many.concat()).await,
+        }
+    }
     /// End the stream gracefully (a FIN).
     fn finish(&mut self) -> TResult<()>;
     /// End the stream abruptly with an application code (RESET_STREAM).
@@ -148,3 +157,40 @@ pub trait Connection: Send + Sync {
 }
 
 pub type SharedConnection = Arc<dyn Connection>;
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    /// A transport that keeps no buffer it is given: what it writes is what a default write gives it.
+    #[derive(Default)]
+    struct Copying(Vec<Vec<u8>>);
+
+    #[async_trait]
+    impl SendStream for Copying {
+        async fn write_all(&mut self, buf: &[u8]) -> TResult<()> {
+            self.0.push(buf.to_vec());
+            Ok(())
+        }
+        fn finish(&mut self) -> TResult<()> {
+            Ok(())
+        }
+        fn reset(&mut self, _code: u64) {}
+        fn set_priority(&mut self, _priority: i32) {}
+        async fn delivered(&mut self) -> TResult<Option<u64>> {
+            Ok(None)
+        }
+    }
+
+    /// A bridge to an earlier draft reads what it is given as one write of the wire, so an object's
+    /// head and its payload reach it together.
+    #[tokio::test]
+    async fn parts_reach_a_copying_transport_as_one_write() {
+        let mut s = Copying::default();
+        s.write_parts(vec![Bytes::from_static(b"head"), Bytes::from_static(b"payload")]).await.unwrap();
+        assert_eq!(s.0, vec![b"headpayload".to_vec()]);
+        let mut s = Copying::default();
+        s.write_parts(vec![Bytes::from_static(b"one")]).await.unwrap();
+        assert_eq!(s.0, vec![b"one".to_vec()]);
+    }
+}

joule-code/joule-code-rs/crates/joule-moq-relay/src/transport/quic.rs

@@ -65,6 +65,9 @@ impl SendStream for QuicSend {
     async fn write_all(&mut self, buf: &[u8]) -> TResult<()> {
         self.0.write_all(buf).await.map_err(write_err)
     }
+    async fn write_parts(&mut self, mut parts: Vec<Bytes>) -> TResult<()> {
+        self.0.write_all_chunks(&mut parts).await.map_err(write_err)
+    }
     fn finish(&mut self) -> TResult<()> {
         self.0.finish().map_err(TransportError::other)
     }
@@ -163,6 +166,26 @@ pub fn quic_diagnostics(c: &quinn::Connection) -> String {
     )
 }
 
+/// A UDP socket for an endpoint, its receive and send buffers asked to be `buffer` bytes each, or
+/// left at the system's default when it is 0. One socket takes every peer's packets, and they
+/// arrive in bursts, each fan-out's acknowledgements together: past its buffer the kernel drops
+/// them, and a dropped acknowledgement is a retransmission and a smaller congestion window. The
+/// system caps the size (`net.core.rmem_max` on Linux), and what it gave is logged when it is less.
+pub fn bind_udp(addr: std::net::SocketAddr, buffer: usize) -> std::io::Result<std::net::UdpSocket> {
+    let sock = std::net::UdpSocket::bind(addr)?;
+    if buffer > 0 {
+        // The kernel's option is an int: a larger size would wrap to something small.
+        let buffer = buffer.min(i32::MAX as usize);
+        let state = quinn::udp::UdpSocketState::new((&sock).into())?;
+        let (r, s) = (state.set_recv_buffer_size((&sock).into(), buffer), state.set_send_buffer_size((&sock).into(), buffer));
+        let got = state.recv_buffer_size((&sock).into()).unwrap_or(0);
+        if r.is_err() || s.is_err() || got < buffer {
+            tracing::warn!(asked = buffer, got, "the UDP receive buffer is smaller than asked: raise the system's limit (net.core.rmem_max on Linux), or a burst of packets is dropped");
+        }
+    }
+    Ok(sock)
+}
+
 /// A client endpoint that offers `alpns` and trusts `roots`, with an optional
 /// transport configuration (a subscriber that limits its receive windows, say).
 pub fn client_endpoint(
@@ -196,7 +219,7 @@ pub fn client_endpoint(
 /// A client endpoint with the caller's TLS configuration (a relay dialing its upstream,
 /// [`crate::cascade`], trusts a CA or a pinned certificate): TLS 1.3, offering `moqt-22`
 /// alone, so the session speaks -22 from its first byte.
-pub fn client_endpoint_tls(bind: std::net::SocketAddr, mut tls: rustls::ClientConfig, transport: Option<quinn::TransportConfig>) -> TResult<quinn::Endpoint> {
+pub fn client_endpoint_tls(bind: std::net::SocketAddr, mut tls: rustls::ClientConfig, transport: Option<quinn::TransportConfig>, socket_buffer: usize) -> TResult<quinn::Endpoint> {
     crate::cert::install_provider();
     tls.alpn_protocols = vec![joule_moq::draft22::ALPN.to_vec()];
     let quic = quinn::crypto::rustls::QuicClientConfig::try_from(tls).map_err(TransportError::other)?;
@@ -204,7 +227,9 @@ pub fn client_endpoint_tls(bind: std::net::SocketAddr, mut tls: rustls::ClientCo
     if let Some(t) = transport {
         cc.transport_config(std::sync::Arc::new(t));
     }
-    let mut ep = quinn::Endpoint::client(bind).map_err(TransportError::other)?;
+    let sock = bind_udp(bind, socket_buffer).map_err(TransportError::other)?;
+    let runtime = quinn::default_runtime().ok_or_else(|| TransportError::other("no async runtime"))?;
+    let mut ep = quinn::Endpoint::new(quinn::EndpointConfig::default(), None, sock, runtime).map_err(TransportError::other)?;
     ep.set_default_client_config(cc);
     Ok(ep)
 }
@@ -237,3 +262,21 @@ pub async fn connect_any(ep: &quinn::Endpoint, addr: std::net::SocketAddr, serve
         }
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[tokio::test]
+    async fn a_socket_is_given_the_buffer_asked_for_within_the_systems_limit() {
+        let any: std::net::SocketAddr = "127.0.0.1:0".parse().unwrap();
+        let of = |s: &std::net::UdpSocket| quinn::udp::UdpSocketState::new(s.into()).unwrap().recv_buffer_size(s.into()).unwrap();
+        let default = of(&bind_udp(any, 0).unwrap());
+        // Well under any system's limit, and Linux reports twice what it is asked.
+        let small = 32 * 1024;
+        let asked = of(&bind_udp(any, small).unwrap());
+        assert!(asked >= small, "{asked} of {small}");
+        // A size past the limit is capped with a warning, never a refusal, and never wraps.
+        assert!(of(&bind_udp(any, 1 << 40).unwrap()) >= default);
+    }
+}

joule-code/joule-code-rs/crates/joule-moq/src/draft22/data.rs

@@ -213,6 +213,14 @@ impl SubgroupCodec {
     }
 
     pub fn encode_object(&mut self, o: &SubgroupObject, out: &mut Vec<u8>) -> WireResult<()> {
+        self.encode_object_head(o, out)?;
+        out.extend_from_slice(&o.payload);
+        Ok(())
+    }
+
+    /// [`Self::encode_object`], but for the payload's own bytes, which follow on the wire: a
+    /// sender holding the payload in a buffer it shares sends that buffer rather than a copy.
+    pub fn encode_object_head(&mut self, o: &SubgroupObject, out: &mut Vec<u8>) -> WireResult<()> {
         let unenc = |reason| Err(WireError::Unencodable { what: "subgroup object", reason });
         let delta = match self.prev {
             None => o.object_id,
@@ -235,7 +243,6 @@ impl SubgroupCodec {
         if o.payload.is_empty() {
             put_varint(out, o.status.code());
         }
-        out.extend_from_slice(&o.payload);
         self.prev = Some(o.object_id);
         Ok(())
     }
@@ -1045,6 +1052,20 @@ pub fn encode_padding_stream(n: usize) -> Vec<u8> {
 #[cfg(test)]
 mod tests {
     use super::*;
+
+    #[test]
+    fn an_objects_head_and_payload_are_its_whole_encoding() {
+        let header = SubgroupHeader { track_alias: 1, group_id: 2, subgroup_id: SubgroupId::Zero, publisher_priority: None, properties: false, end_of_group: false, first_object: true };
+        let (mut whole, mut parts) = (SubgroupCodec::new(header), SubgroupCodec::new(header));
+        for o in [SubgroupObject::normal(0, vec![9u8; 300]), SubgroupObject::normal(4, vec![1u8; 2])] {
+            let (mut a, mut b) = (Vec::new(), Vec::new());
+            whole.encode_object(&o, &mut a).unwrap();
+            parts.encode_object_head(&o, &mut b).unwrap();
+            assert!(b.len() < a.len(), "the head holds no payload");
+            b.extend_from_slice(&o.payload);
+            assert_eq!(a, b);
+        }
+    }
     use crate::draft22::kvp::KeyValuePair;
 
     fn hdr(props: bool) -> SubgroupHeader {