From 260566dce10764bd75bad8dcfaa82aa2bcd2845f Mon Sep 17 00:00:00 2001 From: Transaction Science Date: Mon, 26 Oct 2026 20:40:45 -0400 Subject: [PATCH 1/2] joule-contrib: count what an SRT sender holds unacknowledged; the SRT ingest test closes after the drain an_srt_publisher_reaches_a_relay_subscriber_with_every_receipt closed the publisher a fixed 800 ms after its last frame. On a loaded host a loss on the 3%-lossy path was still unrecovered then: the gateway counted 107 of 108 payloads delivered and one outstanding (4 failures in 32 runs, four at a time on two cores; CI's linux-arm64 leg failed it on #499). The sender now keeps `SendCounts::unacknowledged` (SRT; RIST acknowledges nothing and leaves it 0), and the test closes once it is 0. The end-code assertion now says which code it read. Co-Authored-By: Claude Opus 5.5 --- joule-code/joule-code-rs/crates/joule-contrib/src/link.rs | 3 +++ .../joule-code-rs/crates/joule-contrib/src/srt/send.rs | 3 +++ .../crates/joule-moq-ingest/tests/contrib_srt.rs | 5 ++++- 3 files changed, 10 insertions(+), 1 deletion(-) diff --git a/joule-code/joule-code-rs/crates/joule-contrib/src/link.rs b/joule-code/joule-code-rs/crates/joule-contrib/src/link.rs index 7a680f8ca..98e137c71 100644 --- a/joule-code/joule-code-rs/crates/joule-contrib/src/link.rs +++ b/joule-code/joule-code-rs/crates/joule-contrib/src/link.rs @@ -80,6 +80,9 @@ pub struct SendCounts { /// Packets dropped from the send buffer before they were acknowledged, too old to be /// of use. pub expired: u64, + /// SRT: packets sent and not yet acknowledged or expired, so a publisher can close + /// once the far end holds everything. RIST acknowledges nothing and leaves it 0. + pub unacknowledged: u64, } pub enum Cmd { diff --git a/joule-code/joule-code-rs/crates/joule-contrib/src/srt/send.rs b/joule-code/joule-code-rs/crates/joule-contrib/src/srt/send.rs index f3feaa574..eb615d5f1 100644 --- a/joule-code/joule-code-rs/crates/joule-contrib/src/srt/send.rs +++ b/joule-code/joule-code-rs/crates/joule-contrib/src/srt/send.rs @@ -112,6 +112,7 @@ impl Sender { let mut c = self.counts.lock().unwrap(); c.sent += 1; c.bytes += bytes; + c.unacknowledged = self.held.len() as u64; } pub fn on_datagram(&mut self, b: &[u8]) { @@ -126,6 +127,7 @@ impl Sender { // What is acknowledged leaves the buffer. let keep = self.held.split_off(&pos); self.held = keep; + self.counts.lock().unwrap().unacknowledged = self.held.len() as u64; if c.info != 0 { // A full ACK: ACKACK at once, and its RTT. self.ctrl(ctrl::ACKACK, c.info, Vec::new()); @@ -175,6 +177,7 @@ impl Sender { for k in old { self.held.remove(&k); } + self.counts.lock().unwrap().unacknowledged = self.held.len() as u64; } if now.saturating_sub(self.last_sent_us) >= KEEPALIVE_US { self.ctrl(ctrl::KEEPALIVE, 0, Vec::new()); diff --git a/joule-code/joule-code-rs/crates/joule-moq-ingest/tests/contrib_srt.rs b/joule-code/joule-code-rs/crates/joule-moq-ingest/tests/contrib_srt.rs index 274bfcba6..684ffca04 100644 --- a/joule-code/joule-code-rs/crates/joule-moq-ingest/tests/contrib_srt.rs +++ b/joule-code/joule-code-rs/crates/joule-moq-ingest/tests/contrib_srt.rs @@ -103,6 +103,9 @@ async fn run() { let sent = send_ts(&link, 3, 30, ts::ST_H264).await; assert!(px.up.dropped.load(std::sync::atomic::Ordering::Relaxed) > 10, "the path dropped datagrams"); tokio::time::sleep(Duration::from_millis(800)).await; + // Close once the gateway has acknowledged every packet: a fixed wait alone left a loss + // unrecovered on a loaded host, and the accounting below counted it outstanding. + assert!(until(STEP, || link.sent.lock().unwrap().unacknowledged == 0).await, "the gateway acknowledged every packet before the publisher closed"); link.close(); assert!(until(STEP, || watcher.ended.lock().unwrap().len() == 2).await, "PUBLISH_DONE on both media tracks"); settle(&watcher, Duration::from_millis(500)).await; @@ -125,7 +128,7 @@ async fn run() { let ps = parts(&w); assert!(ps.len() >= 3, "a part a second, and the last: {}", ps.len()); let last = ps.last().unwrap(); - assert!(last.last && last.ended == ended::PEER_SHUTDOWN); + assert!(last.last && last.ended == ended::PEER_SHUTDOWN, "the last part ends the session as the publisher ended it: last={} ended={}", last.last, last.ended); let (mut delivered, mut first, mut dropped, mut count) = (0, 0, 0, 0); for p in &ps { assert_eq!((p.protocol, p.mode, p.cipher), (1, 1, 128)); -- 2.54.0 (Apple Git-157) From 1fd032243626c77b6fa06b44e2e0c47fb4f87627 Mon Sep 17 00:00:00 2001 From: Transaction Science Date: Mon, 26 Oct 2026 20:47:53 -0400 Subject: [PATCH 2/2] =?UTF-8?q?joule-contrib:=20an=20SRT=20sender=20probes?= =?UTF-8?q?=20a=20stalled=20tail;=20live-contribution=20=C2=A74.1=20says?= =?UTF-8?q?=20so?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A receiver finds a gap only behind a later arrival, so a lost newest packet went unrecovered until the sender expired it. Under contention (four at a time on two cores) an_srt_publisher_reaches_a_relay_subscriber_with_every_receipt failed 4 of 32 and 17 of 48: the sender's counts read expired=1 and the gateway's parts covered 107 of the 108 sequence numbers sent. CI's linux-arm64 leg failed the same test on #499. When the acknowledged position has not moved for 2 x RTT + 20 ms and the newest packet held went out that long ago, the sender sends it again with the R flag, the wait doubling per probe to 16 times the first, reset when an ACK takes a packet. SendCounts::probes counts them apart from retransmitted, which answers requests. The test now also requires that the publisher gave nothing up (expired == 0). Co-Authored-By: Claude Opus 5.5 --- .../crates/joule-contrib/src/link.rs | 3 ++ .../crates/joule-contrib/src/srt/send.rs | 33 +++++++++++++++++-- .../joule-moq-ingest/tests/contrib_srt.rs | 4 +-- wai/CHANGELOG.md | 8 +++++ wai/extensions/live-contribution.md | 4 +++ 5 files changed, 47 insertions(+), 5 deletions(-) diff --git a/joule-code/joule-code-rs/crates/joule-contrib/src/link.rs b/joule-code/joule-code-rs/crates/joule-contrib/src/link.rs index 98e137c71..780a9abb5 100644 --- a/joule-code/joule-code-rs/crates/joule-contrib/src/link.rs +++ b/joule-code/joule-code-rs/crates/joule-contrib/src/link.rs @@ -83,6 +83,9 @@ pub struct SendCounts { /// SRT: packets sent and not yet acknowledged or expired, so a publisher can close /// once the far end holds everything. RIST acknowledges nothing and leaves it 0. pub unacknowledged: u64, + /// SRT: the newest packet sent again because the acknowledged position stalled, not + /// counted in `retransmitted`, which answers requests. + pub probes: u64, } pub enum Cmd { diff --git a/joule-code/joule-code-rs/crates/joule-contrib/src/srt/send.rs b/joule-code/joule-code-rs/crates/joule-contrib/src/srt/send.rs index eb615d5f1..5ea3618e0 100644 --- a/joule-code/joule-code-rs/crates/joule-contrib/src/srt/send.rs +++ b/joule-code/joule-code-rs/crates/joule-contrib/src/srt/send.rs @@ -5,7 +5,10 @@ //! kept until it is acknowledged or too old (§4.6: 1.25 × the latency, and at least 1 s). //! A NAK's sequence numbers still held are sent again with the R flag set and their first //! timestamp, ahead of new packets (§4.8.2); a full ACK is answered with an ACKACK at once -//! (§4.8.1), and its RTT taken (§4.10). +//! (§4.8.1), and its RTT taken (§4.10). A receiver finds a gap only behind a later arrival, +//! so a lost newest packet would wait out its time unasked for: when the acknowledged +//! position has not moved for 2 × RTT + 20 ms and the newest packet was sent that long ago, +//! it is sent again, the wait doubling with each probe to at most 16 times the first. use super::crypt::Keys; use super::packet::{ctrl, loss_list_read, Control, Data, Packet}; @@ -26,6 +29,8 @@ struct Held { /// As it went on the wire: encrypted where the link is. payload: Vec, origin_us: u64, + /// When it last went on the wire. + tx_us: u64, } pub struct Sender { @@ -43,6 +48,10 @@ pub struct Sender { pub counts: Arc>, last_sent_us: u64, last_recv_us: u64, + /// When an ACK last took a packet out of the buffer. + last_ack_us: u64, + /// Probes sent since then. + probes: u32, pub out: Vec>, pub closed: Option, } @@ -67,6 +76,8 @@ impl Sender { counts, last_sent_us: 0, last_recv_us: 0, + last_ack_us: 0, + probes: 0, out: Vec::new(), closed: None, } @@ -106,7 +117,7 @@ impl Sender { let d = Data { seq, pp: 3, order: false, kk, rexmit: false, msgno, ts: now as u32, dst: self.peer_id, payload }; let bytes = d.payload.len() as u64; self.out.push(d.write()); - self.held.insert(self.next_seq, Held { seq, ts: d.ts, kk, msgno, payload: d.payload, origin_us: now }); + self.held.insert(self.next_seq, Held { seq, ts: d.ts, kk, msgno, payload: d.payload, origin_us: now, tx_us: now }); self.next_seq += 1; self.last_sent_us = now; let mut c = self.counts.lock().unwrap(); @@ -126,6 +137,10 @@ impl Sender { let pos = self.seq_ext.extend(pos as u64); // What is acknowledged leaves the buffer. let keep = self.held.split_off(&pos); + if keep.len() < self.held.len() { + self.last_ack_us = now; + self.probes = 0; + } self.held = keep; self.counts.lock().unwrap().unacknowledged = self.held.len() as u64; if c.info != 0 { @@ -145,8 +160,9 @@ impl Sender { let a = self.seq_ext.peek(a as u64); let b = self.seq_ext.peek(b as u64).max(a); named += b - a + 1; - for (_, h) in self.held.range(a..=b) { + for (_, h) in self.held.range_mut(a..=b) { let d = Data { seq: h.seq, pp: 3, order: false, kk: h.kk, rexmit: true, msgno: h.msgno, ts: h.ts, dst: self.peer_id, payload: h.payload.clone() }; + h.tx_us = now; self.out.push(d.write()); again += 1; } @@ -179,6 +195,17 @@ impl Sender { } self.counts.lock().unwrap().unacknowledged = self.held.len() as u64; } + let wait = (2 * self.rtt + 20_000) << self.probes.min(4); + if let Some((_, h)) = self.held.iter_mut().next_back() + && now.saturating_sub(self.last_ack_us.max(h.tx_us)) >= wait + { + let d = Data { seq: h.seq, pp: 3, order: false, kk: h.kk, rexmit: true, msgno: h.msgno, ts: h.ts, dst: self.peer_id, payload: h.payload.clone() }; + h.tx_us = now; + self.out.push(d.write()); + self.last_sent_us = now; + self.probes += 1; + self.counts.lock().unwrap().probes += 1; + } if now.saturating_sub(self.last_sent_us) >= KEEPALIVE_US { self.ctrl(ctrl::KEEPALIVE, 0, Vec::new()); } diff --git a/joule-code/joule-code-rs/crates/joule-moq-ingest/tests/contrib_srt.rs b/joule-code/joule-code-rs/crates/joule-moq-ingest/tests/contrib_srt.rs index 684ffca04..6bdce06d6 100644 --- a/joule-code/joule-code-rs/crates/joule-moq-ingest/tests/contrib_srt.rs +++ b/joule-code/joule-code-rs/crates/joule-moq-ingest/tests/contrib_srt.rs @@ -141,8 +141,8 @@ async fn run() { assert_eq!(dropped, 0, "every loss recovered within the window"); assert!(delivered - first >= 6, "the losses were found and recovered: {}", delivered - first); assert_eq!(count, delivered + last.account.outstanding); - let sent_payloads = link.sent.lock().unwrap().sent; - assert_eq!(delivered, sent_payloads, "every payload the publisher sent, accounted for"); + let counts = link.sent.lock().unwrap().clone(); + assert_eq!((delivered, counts.expired), (counts.sent, 0), "every payload the publisher sent, accounted for, none given up: {counts:?}"); // Each track's head is the link hash of the last group receipt the subscriber verified. for t in [video, audio] { let objs = watcher.objects.lock().unwrap(); diff --git a/wai/CHANGELOG.md b/wai/CHANGELOG.md index c00a58c25..95ac2c65b 100644 --- a/wai/CHANGELOG.md +++ b/wai/CHANGELOG.md @@ -6,6 +6,14 @@ wire format may change and such changes are recorded here. ## v0.4.0 — current +### Contribution links: an SRT sender probes a stalled tail + +- **[live-contribution](extensions/live-contribution.md) §4.1:** a receiver finds a gap only behind a + later arrival, so a lost newest packet went unrecovered until it expired. When the acknowledged + position has not moved for 2 × RTT + 20 ms, the sender sends its newest packet again, backing off + to 16 times that wait. The reference sender counts these as `probes`, apart from + `retransmitted`, and reports what it holds `unacknowledged`. + ### `wai.live.message` and `wai.live.call`: conversations and calls on live rooms - **The extension** ([live-messaging](extensions/live-messaging.md)): a conversation is an diff --git a/wai/extensions/live-contribution.md b/wai/extensions/live-contribution.md index b859b8949..a54418752 100644 --- a/wai/extensions/live-contribution.md +++ b/wai/extensions/live-contribution.md @@ -186,6 +186,10 @@ deployment SHOULD set it to several round trips of the paths it serves (live-ing least every 100 ms; a light ACK every 64 packets. - **NAK** (§4.8.2): at once for each gap, and periodically every (RTT + 4 × RTTVar) / 2 with a 20 ms floor, for every packet still missing. +- **A stalled tail**, sending: a receiver finds a gap only behind a later arrival, so when + the acknowledged position has not moved for 2 × RTT + 20 ms and the newest packet held + went out that long ago, the newest is sent again with the R flag; the wait doubles with + each such probe, to at most 16 times the first, until an ACK takes a packet. - **RTT** (§4.10) from ACK/ACKACK pairs, from 100 ms and 50 ms. - **Encryption** (§6): AES-CTR, the SEK of 16, 24 or 32 bytes wrapped (RFC 3394) under a KEK from PBKDF2-HMAC-SHA1 over the last 64 bits of the salt with 2 048 iterations; the -- 2.54.0 (Apple Git-157)