joule-code/joule-code-rs/crates/joule-contrib/src/lossy.rs

@@ -37,6 +37,22 @@ pub struct Impair {
 pub struct Counts {
     pub forwarded: AtomicU64,
     pub dropped: AtomicU64,
+    /// The datagrams it dropped, in order (the first [`DROPPED_KEPT`]), for a test to read
+    /// what was lost: which of them were first sends of data, say, rather than control.
+    pub dropped_log: std::sync::Mutex<Vec<Vec<u8>>>,
+}
+
+/// How many dropped datagrams [`Counts::dropped_log`] keeps.
+pub const DROPPED_KEPT: usize = 4096;
+
+impl Counts {
+    fn record_drop(&self, b: &[u8]) {
+        self.dropped.fetch_add(1, Ordering::Relaxed);
+        let mut log = self.dropped_log.lock().unwrap();
+        if log.len() < DROPPED_KEPT {
+            log.push(b.to_vec());
+        }
+    }
 }
 
 /// xorshift64*, from a seed: the same impairment on every run.
@@ -123,7 +139,7 @@ impl Proxy {
                         let Ok((n, from)) = r else { continue };
                         *client.lock().unwrap() = Some(from);
                         match du.decide() {
-                            None => { cu2.dropped.fetch_add(1, Ordering::Relaxed); }
+                            None => cu2.record_drop(&a[..n]),
                             Some(d) => {
                                 cu2.forwarded.fetch_add(1, Ordering::Relaxed);
                                 let (sock, data) = (back.clone(), a[..n].to_vec());
@@ -137,7 +153,7 @@ impl Proxy {
                         let Ok(n) = r else { continue };
                         let Some(to) = *client.lock().unwrap() else { continue };
                         match dd.decide() {
-                            None => { cd2.dropped.fetch_add(1, Ordering::Relaxed); }
+                            None => cd2.record_drop(&b[..n]),
                             Some(d) => {
                                 cd2.forwarded.fetch_add(1, Ordering::Relaxed);
                                 let (sock, data) = (front.clone(), b[..n].to_vec());

joule-code/joule-code-rs/crates/joule-contrib/src/rist/recv.rs

@@ -189,7 +189,13 @@ impl Receiver {
             self.account.lock().unwrap().duplicate();
             return;
         }
-        let missed_at = self.missing.remove(&seq).map(|m| m.found_us);
+        let mut missed_at = self.missing.remove(&seq).map(|m| m.found_us);
+        // A retransmission of a packet not seen yet, and not known missing: its first send was
+        // lost with nothing behind it to show the gap. It is recovered, found missing as it
+        // arrives, never a first try (live-contribution §6.1).
+        if missed_at.is_none() && p.ssrc & 1 == 1 {
+            missed_at = Some(now);
+        }
         if seq > self.highest {
             for s in self.highest + 1..seq {
                 self.missing.insert(s, Miss { found_us: now, next_req_us: now + self.cfg.reorder_us, requests: 0 });

joule-code/joule-code-rs/crates/joule-contrib/src/srt/recv.rs

@@ -236,7 +236,13 @@ impl Receiver {
             self.account.lock().unwrap().duplicate();
             return;
         }
-        let missed_at = self.missing.remove(&seq).map(|m| m.found_us);
+        let mut missed_at = self.missing.remove(&seq).map(|m| m.found_us);
+        // A retransmission of a packet not seen yet, and not known missing: its first send was
+        // lost with nothing behind it to show the gap (a probed tail, live-contribution §4.1).
+        // It is recovered, found missing as it arrives, never a first try.
+        if missed_at.is_none() && d.rexmit {
+            missed_at = Some(now);
+        }
         if seq > self.highest {
             if seq > self.highest + 1 {
                 let (a, b) = (self.highest + 1, seq - 1);

joule-code/joule-code-rs/crates/joule-contrib/tests/srt_link.rs

@@ -4,8 +4,9 @@
 //! stream ID reaching the admission and its refusal codes; and the latency window: a packet
 //! is not delivered before it.
 
-use joule_contrib::lossy::{Impair, Proxy};
-use joule_contrib::srt::packet::rej;
+use joule_contrib::lossy::{Counts, Impair, Proxy};
+use joule_contrib::srt::packet::{rej, Packet};
+use std::collections::BTreeSet;
 use joule_contrib::srt::{self, conn::CallError, Admitted, CallConfig, ConnRequest, ListenerConfig, Role};
 use std::sync::{Arc, Mutex};
 use std::time::{Duration, Instant};
@@ -61,6 +62,34 @@ async fn carry(tx: &joule_contrib::link::Link, rx: &mut joule_contrib::link::Lin
     got
 }
 
+
+/// The sequence numbers whose first send the proxy dropped on its way to the receiver:
+/// each must be found missing and recovered. Control packets and retransmissions it
+/// dropped are not among them.
+fn first_sends_lost(c: &Counts) -> BTreeSet<u32> {
+    c.dropped_log.lock().unwrap().iter().filter_map(|b| match Packet::read(b) {
+        Some(Packet::Data(d)) if !d.rexmit => Some(d.seq),
+        _ => None,
+    }).collect()
+}
+
+/// The longest run of consecutive sequence numbers in `s`, 31-bit and wrapping.
+fn longest_run(s: &BTreeSet<u32>) -> u32 {
+    let (mut best, mut run, mut prev) = (0u32, 0u32, None::<u32>);
+    for &q in s {
+        run = if prev == Some(q.wrapping_sub(1)) { run + 1 } else { 1 };
+        best = best.max(run);
+        prev = Some(q);
+    }
+    // A run across the wrap: the top of the space followed by 0.
+    if s.contains(&0x7fff_ffff) && s.contains(&0) {
+        let top = s.iter().rev().zip(0..).take_while(|(q, i)| **q == 0x7fff_ffff - i).count() as u32;
+        let low = s.iter().zip(0..).take_while(|(q, i)| **q == *i).count() as u32;
+        best = best.max(top + low);
+    }
+    best
+}
+
 #[tokio::test(flavor = "multi_thread", worker_threads = 4)]
 async fn every_payload_arrives_in_order_and_is_accounted_for() {
     let seen = Arc::new(Mutex::new(Vec::new()));
@@ -108,9 +137,14 @@ async fn losses_each_way_are_recovered_within_the_window() {
     assert_eq!(part.dropped, 0, "{part:?}");
     assert_eq!(got.len(), 1000);
     assert!(got.iter().enumerate().all(|(i, p)| *p == payload(i as u64)));
-    assert!(part.first_try < 1000 && part.delivered - part.first_try >= 5, "losses were found and recovered: {part:?}");
+    // The burst of five dropped datagrams holds as many first sends of data as the timing of
+    // the control packets between them leaves: count them, rather than assume all five.
+    let lost = first_sends_lost(&px.up);
+    assert!(lost.len() >= 2, "the path lost first sends of data: {lost:?}");
+    assert_eq!(part.delivered - part.first_try, lost.len() as u64, "each first send lost was found and recovered: {part:?}");
     assert!(part.requests > 0 && part.retransmitted > 0);
-    assert!(part.runs.iter().any(|r| r.count >= 5), "the burst of five is one run");
+    let burst = longest_run(&lost);
+    assert!(part.runs.iter().any(|r| r.count >= burst), "the longest burst of losses ({burst}) is one run: {:?}", part.runs);
     assert!(part.recovery.max < 300_000 && part.margin_min > 0);
     px.stop();
 }

joule-code/joule-code-rs/crates/joule-moq-ingest/tests/contrib_srt.rs

@@ -11,8 +11,8 @@
 mod common;
 
 use common::*;
-use joule_contrib::lossy::{Impair, Proxy};
-use joule_contrib::srt::{self, conn::CallError, packet::rej, Admitted, CallConfig, ConnRequest, ListenerConfig, Role};
+use joule_contrib::lossy::{Counts, Impair, Proxy};
+use joule_contrib::srt::{self, conn::CallError, packet::{rej, Packet}, Admitted, CallConfig, ConnRequest, ListenerConfig, Role};
 use joule_contrib::ts;
 use joule_moq::carriage::ReceiptBundle;
 use joule_moq::receipt::ReceiptVerifier;
@@ -21,6 +21,7 @@ use joule_moq_ingest::catalog;
 use joule_moq_ingest::frame::Codec;
 use joule_moq_ingest::linkrec::{self, ended, LinkRecord};
 use joule_moq_ingest::room::media;
+use std::collections::BTreeSet;
 use std::sync::Arc;
 use std::time::Duration;
 use tokio::time::timeout;
@@ -73,6 +74,17 @@ fn parts(w: &World) -> Vec<LinkRecord> {
     linkrec::verify_series(&recs).unwrap_or_else(|e| panic!("the link record's parts verify as a series: {e:?}"))
 }
 
+
+/// The sequence numbers whose first send the proxy dropped on its way to the receiver:
+/// each must be found missing and recovered. Control packets and retransmissions it
+/// dropped are not among them.
+fn first_sends_lost(c: &Counts) -> BTreeSet<u32> {
+    c.dropped_log.lock().unwrap().iter().filter_map(|b| match Packet::read(b) {
+        Some(Packet::Data(d)) if !d.rexmit => Some(d.seq),
+        _ => None,
+    }).collect()
+}
+
 #[tokio::test(flavor = "multi_thread", worker_threads = 4)]
 async fn an_srt_publisher_reaches_a_relay_subscriber_with_every_receipt() {
     timeout(Duration::from_secs(120), run()).await.expect("the test finishes in time");
@@ -139,7 +151,11 @@ async fn run() {
         count += p.account.seq_count;
     }
     assert_eq!(dropped, 0, "every loss recovered within the window");
-    assert!(delivered - first >= 6, "the losses were found and recovered: {}", delivered - first);
+    // The burst of six dropped datagrams holds as many first sends of data as the timing of the
+    // control packets between them leaves: count them, rather than assume all six.
+    let lost = first_sends_lost(&px.up);
+    assert!(lost.len() >= 2, "the path lost first sends of data: {lost:?}");
+    assert_eq!(delivered - first, lost.len() as u64, "each first send lost was found and recovered");
     assert_eq!(count, delivered + last.account.outstanding);
     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:?}");

wai/CHANGELOG.md

@@ -32,6 +32,10 @@ wire format may change and such changes are recorded here.
   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`.
+- **§6.1:** a packet whose first arrival is a retransmission, with nothing behind it yet to show
+  the gap, is **recovered** (found missing as it arrives, a recovery of 0 µs), never delivered at
+  the first try: its first send was lost. The SRT and RIST receivers count it so; before, a
+  probed tail read as a first try.
 
 ### `wai.live.message` and `wai.live.call`: conversations and calls on live rooms
 

wai/extensions/live-contribution.md

@@ -318,9 +318,12 @@ The link record is the gateway's statement of what its link carried, as a receiv
 it. A receiver passes sequence numbers in ascending order, each at its delivery time, and
 each comes to exactly one end:
 
-- **delivered at the first try**: it arrived before anything found it missing;
+- **delivered at the first try**: its first send arrived before anything found it missing;
 - **recovered**: it was found missing (a gap behind a later arrival), then arrived, by a
-  retransmission or late from reordering, before its delivery time;
+  retransmission or late from reordering, before its delivery time; or its first arrival
+  was a retransmission (SRT's R flag, RIST's retransmission SSRC) with nothing behind it
+  yet to show the gap, as when a sender probes a stalled tail (§4.1), and it is found
+  missing as it arrives (a recovery of 0 µs);
 - **dropped**: its delivery time passed without it;
 - **outstanding**: the link ended before it arrived and before its delivery time.