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..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 @@ -80,6 +80,12 @@ 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, + /// 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 f3feaa574..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,12 +117,13 @@ 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(); c.sent += 1; c.bytes += bytes; + c.unacknowledged = self.held.len() as u64; } pub fn on_datagram(&mut self, b: &[u8]) { @@ -125,7 +137,12 @@ 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 { // A full ACK: ACKACK at once, and its RTT. self.ctrl(ctrl::ACKACK, c.info, Vec::new()); @@ -143,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; } @@ -175,6 +193,18 @@ impl Sender { for k in old { self.held.remove(&k); } + 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 274bfcba6..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 @@ -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)); @@ -138,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