diff --git a/joule-code/joule-code-rs/crates/joule-contrib/src/lossy.rs b/joule-code/joule-code-rs/crates/joule-contrib/src/lossy.rs index 71934e721..7aa1a131a 100644 --- a/joule-code/joule-code-rs/crates/joule-contrib/src/lossy.rs +++ b/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>>, +} + +/// 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()); diff --git a/joule-code/joule-code-rs/crates/joule-contrib/src/rist/recv.rs b/joule-code/joule-code-rs/crates/joule-contrib/src/rist/recv.rs index 0acae3ca7..75dfe66e6 100644 --- a/joule-code/joule-code-rs/crates/joule-contrib/src/rist/recv.rs +++ b/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 }); diff --git a/joule-code/joule-code-rs/crates/joule-contrib/src/srt/recv.rs b/joule-code/joule-code-rs/crates/joule-contrib/src/srt/recv.rs index 1ba35241b..ea6c38d0b 100644 --- a/joule-code/joule-code-rs/crates/joule-contrib/src/srt/recv.rs +++ b/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); diff --git a/joule-code/joule-code-rs/crates/joule-contrib/tests/srt_link.rs b/joule-code/joule-code-rs/crates/joule-contrib/tests/srt_link.rs index afe55f202..9e702ab53 100644 --- a/joule-code/joule-code-rs/crates/joule-contrib/tests/srt_link.rs +++ b/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 { + 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 { + let (mut best, mut run, mut prev) = (0u32, 0u32, None::); + 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(); } 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 6bdce06d6..8ee5ac745 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 @@ -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 { 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 { + 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:?}"); diff --git a/wai/CHANGELOG.md b/wai/CHANGELOG.md index 2241bf598..01af49091 100644 --- a/wai/CHANGELOG.md +++ b/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 diff --git a/wai/extensions/live-contribution.md b/wai/extensions/live-contribution.md index a54418752..a8ea00439 100644 --- a/wai/extensions/live-contribution.md +++ b/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.