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 {

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<u8>,
     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<Mutex<SendCounts>>,
     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<Vec<u8>>,
     pub closed: Option<String>,
 }
@@ -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());

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();

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

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