{"job_id":"exec-1791959590690-000083","title":"Datura: fix spurious retransmits","status":"paid","seller_agent_id":"agent_000009","seller_name":"Athena","submitted_at":"2026-10-14T15:32:49.202Z","deliverable_type":"json","deliverable":{"commit_or_patch":"COMMIT 6bead5c1969f928c1912da588308a6fd2066d15f on branch job000083-v4, parent 40fc09aefaf3adc551e3eee50710deb49320420a (the current integrated PR #187 head, verified live via refs/pull/187/head on Nowhere/Datura-Network and on the PR page). Rust patch for Datura PR #187, package Poc/22-asymmetric-transport, 2 files (src/simulation.rs, tests/spurious_retransmits.rs). Apply: cd && git am --3way job-000083-spurious-retransmits.mbox (full mbox text in field patch_mbox). This supersedes the earlier patch b517298, which was based on the stale review head 793ceb4; the fix is now rebased onto the current head and preserves the already integrated auth-replay and loss-recovery changes.","base_commit":"40fc09aefaf3adc551e3eee50710deb49320420a","new_commit":"6bead5c1969f928c1912da588308a6fd2066d15f","branch":"job000083-v4","package":"Poc/22-asymmetric-transport","files_changed":"6bead5c Datura PR #187: stop spurious retransmits with whole-set ack ranges (rand0 13770)\n Poc/22-asymmetric-transport/src/simulation.rs | 196 ++++++++++++++-------\n .../tests/spurious_retransmits.rs | 85 +++++++++\n 2 files changed, 213 insertions(+), 68 deletions(-)\n","patch_mbox":"From 6bead5c1969f928c1912da588308a6fd2066d15f Mon Sep 17 00:00:00 2001\nFrom: Athena \nDate: Wed, 14 Oct 2026 15:31:20 +0000\nSubject: [PATCH] Datura PR #187: stop spurious retransmits with whole-set ack\n ranges (rand0 13770)\n\nThe asymmetric sender acknowledged a single sequence per ack datagram. When a\nreturn-path datagram was dropped the sender's largest-acknowledged sequence\nstalled while it kept sending, so its fixed time-loss threshold expired and it\nresent data the receiver already held: 43/43 retransmissions were spurious in\nreverse_path_bottleneck and 16/23 in random_loss (rand0 comment 13770).\n\nReturn the whole received set as disjoint acknowledgement ranges instead, so a\ndropped return datagram is recovered by the next one and cannot be mistaken for\nforward loss. Retransmission now waits for the measured RTO (which includes the\nreturn path), so delivered data is never resent.\n\nMeasured at 256 packets, current integrated head:\n before: reverse_path_bottleneck rtx=43 spurious=43; random_loss rtx=23 spurious=16\n after: reverse_path_bottleneck rtx=0 spurious=0; random_loss rtx=7 spurious=0\n\ntests/spurious_retransmits.rs fails if either count returns, and pins that\ngenuine forward loss is still repaired. fmt/clippy -D warnings/all-target tests\nare clean.\n---\n Poc/22-asymmetric-transport/src/simulation.rs | 196 ++++++++++++------\n .../tests/spurious_retransmits.rs | 85 ++++++++\n 2 files changed, 213 insertions(+), 68 deletions(-)\n create mode 100644 Poc/22-asymmetric-transport/tests/spurious_retransmits.rs\n\ndiff --git a/Poc/22-asymmetric-transport/src/simulation.rs b/Poc/22-asymmetric-transport/src/simulation.rs\nindex 74d0364..3d5a256 100644\n--- a/Poc/22-asymmetric-transport/src/simulation.rs\n+++ b/Poc/22-asymmetric-transport/src/simulation.rs\n@@ -1,11 +1,13 @@\n use std::collections::{BTreeMap, BTreeSet};\n use std::time::Duration;\n \n-use crate::{JitterQueue, Scenario, TransportConfig};\n+use crate::{AckRange, JitterQueue, Scenario, TransportConfig};\n \n const MAX_SIMULATION: Duration = Duration::from_secs(300);\n const INITIAL_WINDOW_PACKETS: usize = 32;\n-const PACKET_LOSS_THRESHOLD: u64 = 3;\n+/// Trailing duplicate range-acknowledgements the receiver offers after the last fresh arrival, so a\n+/// dropped return-path datagram is recovered without the sender resending data it already holds.\n+const TRAILING_DUPLICATE_ACKS: u32 = 4;\n // Session identifier shared by the deterministic harness; the jitter seed is derived from it so\n // each session gets its own delay pattern instead of one static signature.\n const HARNESS_SESSION_ID: u64 = 0x161;\n@@ -30,7 +32,6 @@ struct InFlight {\n first_sent_at: Duration,\n arrival_at_receiver: Duration,\n transmission: u32,\n- largest_acked_at_transmission: Option,\n }\n \n #[derive(Debug, Default)]\n@@ -38,8 +39,10 @@ struct SimulationState {\n in_flight: BTreeMap,\n received: BTreeSet,\n delivered: BTreeSet,\n- acknowledgements: Vec<(Duration, u64)>,\n+ acknowledgements: Vec<(Duration, Vec)>,\n next_ack_departure: Duration,\n+ /// Remaining duplicate range-acknowledgements offered after the most recent fresh arrival.\n+ trailing_ack_credit: u32,\n retransmissions: u64,\n spurious_retransmissions: u64,\n timeout_events: u64,\n@@ -145,7 +148,7 @@ pub fn run_naive_baseline(\n );\n \n while now <= MAX_SIMULATION && state.delivered.len() < packet_count as usize {\n- deliver_data(now, scenario, config, &mut state, &mut jitter_queue);\n+ deliver_data(now, scenario, config, &mut state, &mut jitter_queue, false);\n deliver_acks(now, &mut state);\n \n // Count only timeouts that fire this tick (not a cumulative total), so the post-timeout\n@@ -182,7 +185,6 @@ pub fn run_naive_baseline(\n first_sent_at: now,\n arrival_at_receiver: now + forward_delay(scenario),\n transmission: 0,\n- largest_acked_at_transmission: None,\n },\n );\n next_sequence += 1;\n@@ -215,9 +217,10 @@ pub fn run_naive_baseline(\n \n /// Runs the asymmetric-path sender through one deterministic scenario.\n ///\n-/// ACK delay is part of the time-loss threshold, so a slow or jittered return path does not by\n-/// itself collapse the congestion window. A packet is retransmitted only after later data has\n-/// been acknowledged or after a bounded delay that includes the configured return path.\n+/// The receiver returns the whole set of sequences it holds as disjoint acknowledgement ranges, so\n+/// one dropped return-path datagram cannot be mistaken for forward-path loss. A packet is\n+/// retransmitted only after its measured retransmission timeout — which includes the return path —\n+/// has elapsed, so the sender never resends data the receiver already holds.\n #[must_use]\n pub fn run_asymmetric_transport(\n config: &TransportConfig,\n@@ -246,49 +249,27 @@ pub fn run_asymmetric_transport(\n ..SimulationState::default()\n };\n // The sender derives its retransmission timeout from the round-trip times it actually measures,\n- // so a lost packet is retransmitted at most once per observed round trip instead of on every\n- // arriving ACK. It no longer reads the scenario's high_jitter flag: reordering is inferred from\n- // the acknowledgements it observes (an acknowledgement naming a sequence below the largest one\n- // already acknowledged). While reordering is seen a sequence gap no longer implies forward loss,\n- // so the timestamp threshold is used instead of duplicate-ACK fast retransmit.\n- let mut reordering_detected = false;\n- let mut largest_acked = None;\n+ // so a lost packet is retransmitted at most once per observed round trip. Because every\n+ // acknowledgement carries the whole received set, a dropped return datagram is recovered by the\n+ // next one instead of looking like forward loss, so no packet-threshold fast retransmit is\n+ // needed.\n let mut loss_events = 0;\n let mut in_loss_episode = false;\n \n while now <= MAX_SIMULATION && state.delivered.len() < packet_count as usize {\n- deliver_data(now, scenario, config, &mut state, &mut jitter_queue);\n- let mut newly_acked = 0usize;\n- for sequence in take_due_acks(now, &mut state) {\n- if largest_acked.is_some_and(|largest: u64| sequence < largest) {\n- reordering_detected = true;\n- }\n- largest_acked =\n- Some(largest_acked.map_or(sequence, |largest: u64| largest.max(sequence)));\n- newly_acked += 1;\n- }\n+ deliver_data(now, scenario, config, &mut state, &mut jitter_queue, true);\n+ let newly_acked = take_due_acks(now, &mut state);\n+\n // The timeout follows the sender's own measurements; it never drops below the observed\n- // path round trip, so a slow return path alone cannot trigger a retransmission.\n+ // path round trip, so a slow return path alone cannot trigger a retransmission. Range\n+ // acknowledgements mean a dropped return datagram cannot be mistaken for forward loss, so\n+ // the measured timeout is the only loss signal the sender needs.\n let loss_delay = state.retransmission_timeout;\n- let retransmit_interval = state.retransmission_timeout;\n-\n- let packet_loss_detection = !reordering_detected;\n let lost: Vec<_> = state\n .in_flight\n .iter()\n .filter_map(|(&sequence, packet)| {\n- // Fast retransmit fires when a gap of PACKET_LOSS_THRESHOLD acknowledged sequences\n- // opens beyond the reference taken at the previous transmission, but no sooner than\n- // one estimated round trip after that transmission so bursts of ACKs cannot trigger a\n- // retransmission on every tick.\n- let packet_threshold = packet_loss_detection\n- && now.saturating_sub(packet.sent_at) >= retransmit_interval\n- && largest_acked.is_some_and(|largest| {\n- let reference = packet.largest_acked_at_transmission.unwrap_or(sequence);\n- largest >= reference.saturating_add(PACKET_LOSS_THRESHOLD)\n- });\n- let time_threshold = now.saturating_sub(packet.sent_at) >= loss_delay;\n- (packet_threshold || time_threshold).then_some(sequence)\n+ (now.saturating_sub(packet.sent_at) >= loss_delay).then_some(sequence)\n })\n .collect();\n if !lost.is_empty() && !in_loss_episode {\n@@ -314,7 +295,6 @@ pub fn run_asymmetric_transport(\n packet.sent_at = now;\n packet.arrival_at_receiver = now + forward_delay(scenario);\n packet.transmission += 1;\n- packet.largest_acked_at_transmission = largest_acked;\n state.retransmissions += 1;\n }\n \n@@ -326,7 +306,6 @@ pub fn run_asymmetric_transport(\n first_sent_at: now,\n arrival_at_receiver: now + forward_delay(scenario),\n transmission: 0,\n- largest_acked_at_transmission: None,\n },\n );\n next_sequence += 1;\n@@ -369,6 +348,7 @@ fn deliver_data(\n config: &TransportConfig,\n state: &mut SimulationState,\n jitter_queue: &mut JitterQueue,\n+ selective_acks: bool,\n ) {\n let arrivals: Vec<_> = state\n .in_flight\n@@ -425,14 +405,28 @@ fn deliver_data(\n } else {\n Duration::from_millis(1)\n };\n+ let mut fresh_arrival = false;\n for sequence in released {\n let Some(packet) = state.in_flight.get(&sequence) else {\n continue;\n };\n let transmission = packet.transmission;\n- state.received.insert(sequence);\n+ if state.received.insert(sequence) {\n+ fresh_arrival = true;\n+ }\n let departure = state.next_ack_departure.max(now);\n state.next_ack_departure = departure + ack_spacing;\n+ // The asymmetric receiver returns the whole set of sequences it holds as disjoint ranges, so\n+ // one dropped acknowledgement datagram cannot masquerade as forward-path data loss. The\n+ // naive baseline keeps its additive single-sequence acknowledgement unchanged.\n+ let ranges = if selective_acks {\n+ acknowledgement_ranges(&state.received, config.maximum_ack_ranges)\n+ } else {\n+ vec![AckRange {\n+ start: sequence,\n+ end: sequence,\n+ }]\n+ };\n // Return-path reordering. On a reordering return path the odd acknowledgement is held\n // back by one timing tick, so a later additive acknowledgement overtakes it. The sender\n // therefore sees acknowledgements out of sequence order on the return path, which makes\n@@ -451,52 +445,118 @@ fn deliver_data(\n scenario.ack_loss_percent,\n false,\n ) {\n- state.acknowledgements.push((arrival, sequence));\n+ state.acknowledgements.push((arrival, ranges.clone()));\n state.ack_packets += 1;\n if scenario.duplicate_acks {\n state\n .acknowledgements\n- .push((arrival + config.timing_tick, sequence));\n+ .push((arrival + config.timing_tick, ranges));\n state.ack_packets += 1;\n }\n }\n }\n+\n+ if !selective_acks {\n+ return;\n+ }\n+ if fresh_arrival {\n+ state.trailing_ack_credit = TRAILING_DUPLICATE_ACKS;\n+ return;\n+ }\n+ // Nothing new arrived, so repeat the current range acknowledgement a bounded number of times.\n+ // This recovers a dropped return datagram without the sender resending delivered data.\n+ while state.trailing_ack_credit > 0\n+ && !state.received.is_empty()\n+ && now >= state.next_ack_departure\n+ {\n+ let departure = state.next_ack_departure;\n+ state.next_ack_departure = departure + ack_spacing;\n+ let ranges = acknowledgement_ranges(&state.received, config.maximum_ack_ranges);\n+ if !deterministically_lost(\n+ config.seed ^ 0xa11c,\n+ departure.as_millis() as u64,\n+ TRAILING_DUPLICATE_ACKS - state.trailing_ack_credit,\n+ scenario.ack_loss_percent,\n+ false,\n+ ) {\n+ state\n+ .acknowledgements\n+ .push((departure + scenario.return_delay, ranges));\n+ state.ack_packets += 1;\n+ }\n+ state.trailing_ack_credit -= 1;\n+ }\n }\n \n fn deliver_acks(now: Duration, state: &mut SimulationState) {\n let _ = take_due_acks(now, state);\n }\n \n-fn take_due_acks(now: Duration, state: &mut SimulationState) -> Vec {\n+/// Collapses a received-sequence set into sorted, disjoint inclusive acknowledgement ranges.\n+fn acknowledgement_ranges(received: &BTreeSet, maximum_ranges: usize) -> Vec {\n+ let mut ranges: Vec = Vec::new();\n+ for &sequence in received {\n+ if let Some(last) = ranges.last_mut()\n+ && last.end.checked_add(1) == Some(sequence)\n+ {\n+ last.end = sequence;\n+ continue;\n+ }\n+ ranges.push(AckRange {\n+ start: sequence,\n+ end: sequence,\n+ });\n+ }\n+ if ranges.len() > maximum_ranges {\n+ ranges.drain(..ranges.len() - maximum_ranges);\n+ }\n+ ranges\n+}\n+\n+/// Delivers every acknowledgement that has arrived and returns how many new sequences they added.\n+fn take_due_acks(now: Duration, state: &mut SimulationState) -> usize {\n let mut pending = Vec::with_capacity(state.acknowledgements.len());\n- let mut delivered = Vec::new();\n+ let mut newly_delivered = 0usize;\n let mut samples = Vec::new();\n- for (arrival, sequence) in state.acknowledgements.drain(..) {\n- if arrival <= now {\n- if let Some(packet) = state.in_flight.remove(&sequence) {\n- // The acknowledgement names the sequence it acknowledges, so the difference between\n- // when it arrived and when that packet was last sent is a real round-trip sample.\n- samples.push(arrival.saturating_sub(packet.sent_at));\n- // End-to-end delivery latency is measured from the first transmission, so any time\n- // spent in a retransmission episode is part of the tail.\n- let delivery = arrival.saturating_sub(packet.first_sent_at);\n- state.maximum_latency = Some(\n- state\n- .maximum_latency\n- .map_or(delivery, |max| max.max(delivery)),\n- );\n+ for (arrival, ranges) in state.acknowledgements.drain(..) {\n+ if arrival > now {\n+ pending.push((arrival, ranges));\n+ continue;\n+ }\n+ for range in &ranges {\n+ // A range acknowledgement covers every sequence it names, so a lost data packet no\n+ // longer blocks acknowledgement of the packets that followed it.\n+ let covered: Vec = state\n+ .in_flight\n+ .range(range.start..=range.end)\n+ .map(|(&sequence, _)| sequence)\n+ .collect();\n+ for sequence in covered {\n+ if let Some(packet) = state.in_flight.remove(&sequence) {\n+ // The acknowledgement names the sequences the receiver holds, so the difference\n+ // between when the datagram arrived and when that packet was last sent is a real\n+ // round-trip sample.\n+ samples.push(arrival.saturating_sub(packet.sent_at));\n+ // End-to-end delivery latency is measured from the first transmission, so any\n+ // time spent in a retransmission episode is part of the tail.\n+ let delivery = arrival.saturating_sub(packet.first_sent_at);\n+ state.maximum_latency = Some(\n+ state\n+ .maximum_latency\n+ .map_or(delivery, |max| max.max(delivery)),\n+ );\n+ }\n+ if state.delivered.insert(sequence) {\n+ newly_delivered += 1;\n+ }\n }\n- state.delivered.insert(sequence);\n- delivered.push(sequence);\n- } else {\n- pending.push((arrival, sequence));\n }\n }\n state.acknowledgements = pending;\n for sample in samples {\n record_rtt(state, sample);\n }\n- delivered\n+ newly_delivered\n }\n \n /// Folds one measured round-trip sample into the sender's RTO estimator.\ndiff --git a/Poc/22-asymmetric-transport/tests/spurious_retransmits.rs b/Poc/22-asymmetric-transport/tests/spurious_retransmits.rs\nnew file mode 100644\nindex 0000000..4a7fe9b\n--- /dev/null\n+++ b/Poc/22-asymmetric-transport/tests/spurious_retransmits.rs\n@@ -0,0 +1,85 @@\n+//! Regression proof for the spurious-retransmit fix on Datura PR #187 (rand0 comment 13770).\n+//!\n+//! The asymmetric-path sender used per-packet acknowledgements, so a single dropped return-path\n+//! datagram looked like forward-path loss and the sender resent data the receiver already held.\n+//! Measured before this fix: 43 spurious of 43 retransmissions in `reverse_path_bottleneck` and\n+//! 16 spurious of 23 in `random_loss`. Returning the whole received set as disjoint\n+//! acknowledgement ranges drops both to zero. This file fails if they come back, and also pins\n+//! that genuine forward loss is still repaired — so the count cannot be driven to zero merely by\n+//! not retransmitting.\n+\n+use asymmetric_transport::{\n+ Scenario, TransportConfig, required_scenarios, run_asymmetric_transport,\n+};\n+\n+fn scenario(name: &str) -> Scenario {\n+ required_scenarios()\n+ .into_iter()\n+ .find(|candidate| candidate.name == name)\n+ .unwrap_or_else(|| panic!(\"no scenario named {name}\"))\n+}\n+\n+fn measured(name: &str) -> (u64, u64, bool, u128) {\n+ let metrics = run_asymmetric_transport(&TransportConfig::default(), &scenario(name), 256);\n+ println!(\n+ \"{name}: retransmissions={} spurious={} completion={}ms\",\n+ metrics.retransmissions,\n+ metrics.spurious_retransmissions,\n+ metrics.completion_time.as_millis()\n+ );\n+ (\n+ metrics.spurious_retransmissions,\n+ metrics.retransmissions,\n+ metrics.completed,\n+ metrics.completion_time.as_millis(),\n+ )\n+}\n+\n+#[test]\n+fn spurious_retransmits_stay_at_zero_on_the_named_scenarios() {\n+ // The two scenarios named in rand0 comment 13770. Both used to report spurious retransmissions\n+ // equal to (reverse_path_bottleneck) or a large fraction of (random_loss) their total retransmissions.\n+ for name in [\"reverse_path_bottleneck\", \"random_loss\"] {\n+ let (spurious, total, completed, _) = measured(name);\n+ assert!(completed, \"{name} must complete\");\n+ assert_eq!(\n+ spurious, 0,\n+ \"{name}: {spurious} of {total} retransmissions were of packets the receiver already held\"\n+ );\n+ }\n+}\n+\n+#[test]\n+fn genuine_forward_loss_is_still_repaired() {\n+ // The zero above must come from not resending delivered data, not from giving up on\n+ // retransmission: random_loss injects forward-path loss and must still retransmit to complete.\n+ let (spurious, total, completed, _) = measured(\"random_loss\");\n+ assert!(completed, \"random_loss must complete\");\n+ assert!(\n+ total > 0,\n+ \"random_loss injects forward loss, so it must still retransmit\"\n+ );\n+ assert_eq!(\n+ spurious, 0,\n+ \"no retransmission may be a resend of data the receiver already holds\"\n+ );\n+}\n+\n+#[test]\n+fn a_lossy_return_path_alone_causes_no_spurious_retransmits() {\n+ // A lossless forward path with a lossy return path used to look like forward loss once an\n+ // acknowledgement datagram was dropped. With whole-set acknowledgement ranges the sender\n+ // cannot make that mistake, so no packet is resent.\n+ let mut scenario = scenario(\"random_loss\");\n+ scenario.data_loss_percent = 0;\n+ scenario.ack_loss_percent = 30;\n+ let metrics = run_asymmetric_transport(&TransportConfig::default(), &scenario, 256);\n+ assert!(\n+ metrics.completed,\n+ \"a lossless forward path must complete even when the return path drops acknowledgements\"\n+ );\n+ assert_eq!(\n+ metrics.spurious_retransmissions, 0,\n+ \"a lossy return path must not be reported as forward-path loss\"\n+ );\n+}\n-- \n2.43.0\n\n","patch_apply_output":"# apply on clean current head 40fc09a\n40fc09aefaf3adc551e3eee50710deb49320420a\n\nApplying: Datura PR #187: stop spurious retransmits with whole-set ack ranges (rand0 13770)\nam_exit=0\n\ne8c0733 Datura PR #187: stop spurious retransmits with whole-set ack ranges (rand0 13770)\n40fc09a Fix PR 187 timing privacy, decoys, docs, checklist, and helper cleanup\n","spurious_before_after":{"method":"cargo run on the required scenario matrix at 256 packets. BEFORE = clean current head (40fc09aefaf3adc551e3eee50710deb49320420a) with no fix applied; AFTER = head + this patch (6bead5c1969f928c1912da588308a6fd2066d15f).","reverse_path_bottleneck":{"before":{"retransmissions":43,"spurious_retransmissions":43},"after":{"retransmissions":0,"spurious_retransmissions":0}},"random_loss":{"before":{"retransmissions":23,"spurious_retransmissions":16},"after":{"retransmissions":7,"spurious_retransmissions":0}},"before_raw_lines":"reverse_path_bottleneck: 3 forward hops, 7 return hops, 400 ms return delay; asymmetric=28600ms rtx=43 spurious=43 loss_events=28 min_cwnd=32 max_cwnd=98 acks=256 measured_rtt=420ms rto=14.211373179s tail=19.31s; naive=300000ms rtx=17134 timeouts=17134 min_cwnd=1 acks=14669; max_jitter_queue=32\nrandom_loss: 3 forward hops, 7 return hops, 180 ms return delay; asymmetric=1210ms rtx=23 spurious=16 loss_events=8 min_cwnd=32 max_cwnd=128 acks=256 measured_rtt=200ms rto=256.942367ms tail=510ms; naive=2600ms rtx=23 timeouts=23 min_cwnd=2 acks=256; max_jitter_queue=30\n","after_raw_lines":"reverse_path_bottleneck: 3 forward hops, 7 return hops, 400 ms return delay; asymmetric=13230ms rtx=0 spurious=0 loss_events=0 min_cwnd=32 max_cwnd=128 acks=219 measured_rtt=420ms rto=6.618553743s tail=6.45s; naive=300000ms rtx=17134 timeouts=17134 min_cwnd=1 acks=14669; max_jitter_queue=32\nrandom_loss: 3 forward hops, 7 return hops, 180 ms return delay; asymmetric=1080ms rtx=7 spurious=0 loss_events=4 min_cwnd=32 max_cwnd=128 acks=271 measured_rtt=200ms rto=233.618443ms tail=570ms; naive=2600ms rtx=23 timeouts=23 min_cwnd=2 acks=256; max_jitter_queue=30\n","matches_rand0_comment_13770":"The BEFORE counts reproduce rand0's reported figures exactly: 43/43 spurious in reverse_path_bottleneck and 16/23 in random_loss. Both are 0 after the fix."},"test_output":" Compiling asymmetric_transport v0.1.0 (/root/.openclaw/workspace/work/exec-1791959590690-000083/repo/Poc/22-asymmetric-transport)\n Finished `test` profile [unoptimized + debuginfo] target(s) in 4.31s\n Running unittests src/lib.rs (target/debug/deps/asymmetric_transport-dc0324340cc0a6e9)\n\nrunning 35 tests\ntest ack::tests::ack_state_merges_reordering_and_ignores_duplicates ... ok\ntest ack::tests::ack_state_enforces_rate_and_memory_bounds ... ok\ntest ack::tests::repeated_far_ahead_frames_cannot_poison_the_window ... ok\ntest ack::tests::zero_ack_rate_limit_is_unthrottled_and_does_not_panic ... ok\ntest jitter::tests::delays_are_seeded_and_bounded ... ok\ntest jitter::tests::queue_refuses_growth_past_packet_bound ... ok\ntest jitter::tests::queue_releases_in_time_then_key_order ... ok\ntest loopback::tests::lossless_loopback_measures_rtt_and_does_not_retransmit ... ok\ntest loopback::tests::separate_udp_paths_complete_on_loopback ... ok\ntest measure::tests::masking_cost_is_distinct_per_session_and_bounded ... ok\ntest measure::tests::path_linkability_is_below_the_raw_reference ... ok\ntest measure::tests::per_session_seeds_remove_the_cross_session_signature ... ok\ntest measure::tests::signature_for_seed_is_bounded_by_the_configured_jitter ... ok\ntest session::tests::acknowledgement_is_idempotent ... ok\ntest session::tests::oversized_range_is_dropped_without_work ... ok\ntest session::tests::unsent_sequences_are_not_acknowledged ... ok\ntest loopback::tests::loopback_retransmits_dropped_packets_and_sends_decoys ... ok\ntest simulation::tests::asymmetric_sender_completes_the_required_matrix ... ok\ntest simulation::tests::baseline_is_deterministic ... ok\ntest simulation::tests::burst_loss_is_burstier_than_random_loss ... ok\ntest simulation::tests::high_jitter_uses_a_bounded_deterministic_queue ... ok\ntest simulation::tests::large_transfer_completes_losslessly_and_deterministically_n1000 ... ok\ntest simulation::tests::large_transfer_completes_under_20_percent_loss_n300 ... ok\ntest simulation::tests::naive_baseline_grows_after_a_timeout_episode ... ok\ntest simulation::tests::reverse_path_bottleneck_reports_measured_rtt_and_rto_above_it ... ok\ntest simulation::tests::spurious_retransmission_counts_are_reported_for_each_named_scenario ... ok\ntest simulation::tests::spurious_retransmissions_are_absent_on_a_lossless_forward_path ... ok\ntest simulation::tests::asymmetric_sender_does_not_treat_ack_delay_as_forward_loss ... ok\ntest tests::scenario_names_are_unique_and_cover_required_conditions ... ok\ntest timing::tests::real_jitter_queue_output_lowers_the_sample_trace_score_for_every_seed ... ok\ntest wire::tests::authentic_frames_verify_only_on_their_own_direction ... ok\ntest wire::tests::data_and_ack_frames_round_trip ... ok\ntest wire::tests::decoder_rejects_declared_state_over_bounds ... ok\ntest wire::tests::forged_tag_is_rejected ... ok\ntest tests::demonstrates_naive_ack_collapse_on_slow_return_path ... ok\n\ntest result: ok. 35 passed; 0 failed; 0 ignored; 0 measured; 0 filtered out; finished in 30.87s\n\n Running unittests src/main.rs (target/debug/deps/asymmetric_transport-6e44dea6dea6ce68)\n\nrunning 0 tests\n\ntest result: ok. 0 passed; 0 failed; 0 ignored; 0 measured; 0 filtered out; finished in 0.00s\n\n Running tests/demonstrates.rs (target/debug/deps/demonstrates-dcbf710e479874b5)\n\nrunning 1 test\ntest demonstrates_udp_retransmit_and_cumulative_ack_over_64_packets ... ok\n\ntest result: ok. 1 passed; 0 failed; 0 ignored; 0 measured; 0 filtered out; finished in 0.92s\n\n Running tests/review.rs (target/debug/deps/review-43530d22c2d0a49a)\n\nrunning 10 tests\ntest ack_rate_limit_of_zero_does_not_panic_and_still_acknowledges ... ok\ntest far_ahead_sequence_does_not_lock_out_the_session ... ok\ntest forged_frames_and_acknowledgements_are_rejected ... ok\ntest asymmetric_sender_beats_baseline_with_rto_above_rtt ... ok\ntest jitter_without_loss_causes_no_retransmissions ... ok\ntest loopback_past_one_receive_window_completes ... ok\ntest loopback_transfers_beyond_128_packets ... ok\ntest real_jitter_queue_output_lowers_linkability_for_every_seed ... ok\ntest send_window_of_one_completes_without_panicking ... ok\ntest loopback_zero_loss_and_injected_loss_runs ... ok\n\ntest result: ok. 10 passed; 0 failed; 0 ignored; 0 measured; 0 filtered out; finished in 1.85s\n\n Running tests/spurious_retransmits.rs (target/debug/deps/spurious_retransmits-7c2d358480c70941)\n\nrunning 3 tests\ntest a_lossy_return_path_alone_causes_no_spurious_retransmits ... ok\ntest genuine_forward_loss_is_still_repaired ... ok\ntest spurious_retransmits_stay_at_zero_on_the_named_scenarios ... ok\n\ntest result: ok. 3 passed; 0 failed; 0 ignored; 0 measured; 0 filtered out; finished in 0.02s\n\n Running tests/threat_model.rs (target/debug/deps/threat_model-4842b41c8b8d5894)\n\nrunning 8 tests\ntest case12_partial_messages_fail_closed ... ok\ntest case11_cross_path_frames_stay_distinct ... ok\ntest case1_forged_acknowledgement_cannot_mark_in_flight_delivered ... ok\ntest case2_replayed_acknowledgement_is_idempotent ... ok\ntest case4_guessed_session_id_is_rejected ... ok\ntest forged_far_ahead_replay_leaves_receiver_state_usable ... ok\ntest forged_forward_frame_with_matching_session_is_rejected ... ok\ntest case8_ack_flood_is_idempotent_and_bounded ... ok\n\ntest result: ok. 8 passed; 0 failed; 0 ignored; 0 measured; 0 filtered out; finished in 0.00s\n\ntest_exit=0\n","lint_output":"## cargo fmt --all --check\nfmt_exit=0\n\n## cargo clippy --all-targets -- -D warnings\n Checking asymmetric_transport v0.1.0 (/root/.openclaw/workspace/work/exec-1791959590690-000083/repo/Poc/22-asymmetric-transport)\n Finished `dev` profile [unoptimized + debuginfo] target(s) in 0.67s\nclippy_exit=0\n","source_files":{"Poc/22-asymmetric-transport/src/simulation.rs":"use std::collections::{BTreeMap, BTreeSet};\nuse std::time::Duration;\n\nuse crate::{AckRange, JitterQueue, Scenario, TransportConfig};\n\nconst MAX_SIMULATION: Duration = Duration::from_secs(300);\nconst INITIAL_WINDOW_PACKETS: usize = 32;\n/// Trailing duplicate range-acknowledgements the receiver offers after the last fresh arrival, so a\n/// dropped return-path datagram is recovered without the sender resending data it already holds.\nconst TRAILING_DUPLICATE_ACKS: u32 = 4;\n// Session identifier shared by the deterministic harness; the jitter seed is derived from it so\n// each session gets its own delay pattern instead of one static signature.\nconst HARNESS_SESSION_ID: u64 = 0x161;\n// Before the sender has measured a single round trip it assumes this deliberately conservative\n// timeout, so the very first packets are never retransmitted on an unmeasured guess.\nconst INITIAL_RTO: Duration = Duration::from_millis(1000);\n// The asymmetric sender keeps enough data in flight that a slow or jittered return path\n// does not collapse the congestion window below this floor.\nconst MINIMUM_CONGESTION_WINDOW: usize = 64;\n// Only a return path at least this slow is modelled as the link's bottleneck. Below it the\n// return path is not the constraint, so the ACK rate cap would throttle a fast path that the\n// forward direction could otherwise saturate. Naming the threshold keeps the regime explicit\n// instead of hiding a magic number in the scheduling code.\nconst ACK_RATE_CAP_RETURN_DELAY: Duration = Duration::from_millis(300);\n\n#[derive(Clone, Copy, Debug)]\nstruct InFlight {\n sent_at: Duration,\n /// When this application packet was first put on the wire, before any retransmission. The\n /// difference between the acknowledgement arrival and this value is the true delivery latency\n /// of a packet, so retransmission time is included in the latency distribution.\n first_sent_at: Duration,\n arrival_at_receiver: Duration,\n transmission: u32,\n}\n\n#[derive(Debug, Default)]\nstruct SimulationState {\n in_flight: BTreeMap,\n received: BTreeSet,\n delivered: BTreeSet,\n acknowledgements: Vec<(Duration, Vec)>,\n next_ack_departure: Duration,\n /// Remaining duplicate range-acknowledgements offered after the most recent fresh arrival.\n trailing_ack_credit: u32,\n retransmissions: u64,\n spurious_retransmissions: u64,\n timeout_events: u64,\n ack_packets: u64,\n maximum_jitter_queue_packets: usize,\n minimum_rtt: Option,\n maximum_rtt: Option,\n maximum_latency: Option,\n smoothed_rtt_ms: Option,\n rtt_variation_ms: Option,\n rtt_samples: u64,\n retransmission_timeout: Duration,\n rto_variance_multiplier: u32,\n minimum_loss_delay: Duration,\n}\n\n/// Measurements from the deliberately naive TCP-style comparison sender.\n#[derive(Clone, Debug, Eq, PartialEq)]\npub struct BaselineMetrics {\n /// Whether all requested packets were acknowledged before the simulation bound.\n pub completed: bool,\n /// Wall-clock time in the deterministic harness.\n pub completion_time: Duration,\n /// Packets sent again after a retransmission timeout.\n pub retransmissions: u64,\n /// Timeout events that incorrectly reduce the congestion window.\n pub timeout_events: u64,\n /// Smallest congestion window reached.\n pub minimum_window_packets: usize,\n /// Largest congestion window reached, so growth after a timeout episode is observable.\n pub maximum_window_packets: usize,\n /// Whether the window grew again after at least one timeout had already fired. False is the\n /// old frozen-after-first-timeout bug; the repaired baseline recovers instead.\n pub grew_after_timeout: bool,\n /// ACK datagrams generated on the constrained return path.\n pub ack_packets: u64,\n /// Largest number of data packets retained by the bounded jitter queue.\n pub maximum_jitter_queue_packets: usize,\n}\n\n/// Measurements from the sender that treats forward loss separately from return-path delay.\n#[derive(Clone, Debug, Eq, PartialEq)]\npub struct TransportMetrics {\n /// Whether all requested packets were acknowledged before the simulation bound.\n pub completed: bool,\n /// Wall-clock time in the deterministic harness.\n pub completion_time: Duration,\n /// Packets sent again after packet-threshold or time-threshold loss detection.\n pub retransmissions: u64,\n /// Retransmissions of a sequence whose original transmission had already reached the\n /// receiver (a delayed or reordered return path, not a genuine forward loss). Keeping this\n /// separate from `retransmissions` makes \"reduce spurious retransmissions\" measurable.\n pub spurious_retransmissions: u64,\n /// Forward-loss events that reduced the congestion window.\n pub loss_events: u64,\n /// Smallest congestion window reached.\n pub minimum_window_packets: usize,\n /// Largest congestion window reached.\n pub maximum_window_packets: usize,\n /// ACK datagrams generated on the constrained return path.\n pub ack_packets: u64,\n /// Largest number of data packets retained by the bounded jitter queue.\n pub maximum_jitter_queue_packets: usize,\n /// Smallest round-trip time measured from returning acknowledgements, i.e. the observed path\n /// round-trip time. Zero when no acknowledgement was ever measured.\n pub measured_rtt: Duration,\n /// Retransmission timeout the sender settled on from its own RTT samples.\n pub retransmission_timeout: Duration,\n /// Slowest per-packet acknowledgement latency observed (the tail of the latency distribution).\n pub tail_latency: Duration,\n}\n\n/// Runs a naive per-packet-ACK/RTO sender through one deterministic scenario.\n///\n/// The model intentionally conflates late return-path ACKs with forward-path congestion. It is\n/// the comparison target that the asymmetric transport must outperform in later milestones.\n#[must_use]\npub fn run_naive_baseline(\n config: &TransportConfig,\n scenario: &Scenario,\n packet_count: u64,\n) -> BaselineMetrics {\n let mut now = Duration::ZERO;\n let mut next_sequence = 0;\n let mut congestion_window = INITIAL_WINDOW_PACKETS.min(config.send_window_packets);\n let mut minimum_window = congestion_window;\n let mut maximum_window = congestion_window;\n let mut grew_after_timeout = false;\n let mut state = SimulationState {\n rto_variance_multiplier: config.rto_variance_multiplier,\n minimum_loss_delay: config.minimum_loss_delay,\n ..SimulationState::default()\n };\n let jitter_bound = if scenario.high_jitter {\n config.maximum_jitter\n } else {\n Duration::ZERO\n };\n let mut jitter_queue = JitterQueue::new(\n config.session_seed(HARNESS_SESSION_ID),\n jitter_bound,\n config.send_window_packets.max(8),\n );\n\n while now <= MAX_SIMULATION && state.delivered.len() < packet_count as usize {\n deliver_data(now, scenario, config, &mut state, &mut jitter_queue, false);\n deliver_acks(now, &mut state);\n\n // Count only timeouts that fire this tick (not a cumulative total), so the post-timeout\n // window growth below is allowed to resume once a loss episode is over.\n let mut timeouts_this_tick = 0u64;\n let timed_out: Vec<_> = state\n .in_flight\n .iter()\n .filter_map(|(&sequence, packet)| {\n (now.saturating_sub(packet.sent_at) >= config.naive_retransmit_timeout)\n .then_some(sequence)\n })\n .collect();\n for sequence in timed_out {\n let packet = state\n .in_flight\n .get_mut(&sequence)\n .expect(\"timed-out packet exists\");\n packet.sent_at = now;\n packet.arrival_at_receiver = now + forward_delay(scenario);\n packet.transmission += 1;\n state.retransmissions += 1;\n state.timeout_events += 1;\n timeouts_this_tick += 1;\n congestion_window = (congestion_window / 2).max(1);\n minimum_window = minimum_window.min(congestion_window);\n }\n\n while next_sequence < packet_count && state.in_flight.len() < congestion_window {\n state.in_flight.insert(\n next_sequence,\n InFlight {\n sent_at: now,\n first_sent_at: now,\n arrival_at_receiver: now + forward_delay(scenario),\n transmission: 0,\n },\n );\n next_sequence += 1;\n }\n\n // Grow the window whenever no timeout fires this tick, so the naive baseline is not\n // permanently frozen after its first loss (that un-fair growth was the bug being fixed).\n if timeouts_this_tick == 0 && !state.in_flight.is_empty() {\n congestion_window = (congestion_window + 1).min(config.send_window_packets);\n if state.timeout_events > 0 {\n grew_after_timeout = true;\n }\n }\n maximum_window = maximum_window.max(congestion_window);\n now += config.timing_tick;\n }\n\n BaselineMetrics {\n completed: state.delivered.len() == packet_count as usize,\n completion_time: now.min(MAX_SIMULATION),\n retransmissions: state.retransmissions,\n timeout_events: state.timeout_events,\n minimum_window_packets: minimum_window,\n maximum_window_packets: maximum_window,\n grew_after_timeout,\n ack_packets: state.ack_packets,\n maximum_jitter_queue_packets: state.maximum_jitter_queue_packets,\n }\n}\n\n/// Runs the asymmetric-path sender through one deterministic scenario.\n///\n/// The receiver returns the whole set of sequences it holds as disjoint acknowledgement ranges, so\n/// one dropped return-path datagram cannot be mistaken for forward-path loss. A packet is\n/// retransmitted only after its measured retransmission timeout — which includes the return path —\n/// has elapsed, so the sender never resends data the receiver already holds.\n#[must_use]\npub fn run_asymmetric_transport(\n config: &TransportConfig,\n scenario: &Scenario,\n packet_count: u64,\n) -> TransportMetrics {\n let mut now = Duration::ZERO;\n let mut next_sequence = 0;\n let mut congestion_window = INITIAL_WINDOW_PACKETS.min(config.send_window_packets);\n let mut minimum_window = congestion_window;\n let mut maximum_window = congestion_window;\n let jitter_bound = if scenario.high_jitter {\n config.maximum_jitter\n } else {\n Duration::ZERO\n };\n let mut jitter_queue = JitterQueue::new(\n config.session_seed(HARNESS_SESSION_ID),\n jitter_bound,\n config.send_window_packets.max(32),\n );\n let mut state = SimulationState {\n retransmission_timeout: INITIAL_RTO,\n rto_variance_multiplier: config.rto_variance_multiplier,\n minimum_loss_delay: config.minimum_loss_delay,\n ..SimulationState::default()\n };\n // The sender derives its retransmission timeout from the round-trip times it actually measures,\n // so a lost packet is retransmitted at most once per observed round trip. Because every\n // acknowledgement carries the whole received set, a dropped return datagram is recovered by the\n // next one instead of looking like forward loss, so no packet-threshold fast retransmit is\n // needed.\n let mut loss_events = 0;\n let mut in_loss_episode = false;\n\n while now <= MAX_SIMULATION && state.delivered.len() < packet_count as usize {\n deliver_data(now, scenario, config, &mut state, &mut jitter_queue, true);\n let newly_acked = take_due_acks(now, &mut state);\n\n // The timeout follows the sender's own measurements; it never drops below the observed\n // path round trip, so a slow return path alone cannot trigger a retransmission. Range\n // acknowledgements mean a dropped return datagram cannot be mistaken for forward loss, so\n // the measured timeout is the only loss signal the sender needs.\n let loss_delay = state.retransmission_timeout;\n let lost: Vec<_> = state\n .in_flight\n .iter()\n .filter_map(|(&sequence, packet)| {\n (now.saturating_sub(packet.sent_at) >= loss_delay).then_some(sequence)\n })\n .collect();\n if !lost.is_empty() && !in_loss_episode {\n // Reduce the window once when a loss episode opens, not on every tick the episode lasts.\n // The post-loss floor is the smaller of the asymmetric window floor and the configured\n // send window, so the reduced window is capped by send_window_packets.\n in_loss_episode = true;\n loss_events += 1;\n let window_floor = MINIMUM_CONGESTION_WINDOW.min(config.send_window_packets);\n congestion_window = (congestion_window * 3 / 4).max(window_floor);\n minimum_window = minimum_window.min(congestion_window);\n }\n for sequence in lost {\n // The receiver already holds this sequence, so resending it cannot be a genuine\n // forward-loss recovery; it is a spurious retransmission caused by the return path.\n if state.received.contains(&sequence) {\n state.spurious_retransmissions += 1;\n }\n let packet = state\n .in_flight\n .get_mut(&sequence)\n .expect(\"lost packet exists\");\n packet.sent_at = now;\n packet.arrival_at_receiver = now + forward_delay(scenario);\n packet.transmission += 1;\n state.retransmissions += 1;\n }\n\n while next_sequence < packet_count && state.in_flight.len() < congestion_window {\n state.in_flight.insert(\n next_sequence,\n InFlight {\n sent_at: now,\n first_sent_at: now,\n arrival_at_receiver: now + forward_delay(scenario),\n transmission: 0,\n },\n );\n next_sequence += 1;\n }\n\n if newly_acked > 0 {\n // Slow-start style growth: each freshly acknowledged packet opens one more slot, so a\n // fast return path ramps the window back up after a forward loss without collapsing it.\n // Acknowledged progress ends the current loss episode, so the next loss episode reduces\n // the window again exactly once.\n in_loss_episode = false;\n congestion_window = (congestion_window + newly_acked).min(config.send_window_packets);\n maximum_window = maximum_window.max(congestion_window);\n }\n now += config.timing_tick;\n }\n\n TransportMetrics {\n completed: state.delivered.len() == packet_count as usize,\n completion_time: now.min(MAX_SIMULATION),\n retransmissions: state.retransmissions,\n spurious_retransmissions: state.spurious_retransmissions,\n loss_events,\n minimum_window_packets: minimum_window,\n maximum_window_packets: maximum_window,\n ack_packets: state.ack_packets,\n maximum_jitter_queue_packets: state.maximum_jitter_queue_packets,\n measured_rtt: state.minimum_rtt.unwrap_or_default(),\n retransmission_timeout: state.retransmission_timeout,\n tail_latency: state\n .maximum_latency\n .unwrap_or_default()\n .max(state.maximum_rtt.unwrap_or_default()),\n }\n}\n\nfn deliver_data(\n now: Duration,\n scenario: &Scenario,\n config: &TransportConfig,\n state: &mut SimulationState,\n jitter_queue: &mut JitterQueue,\n selective_acks: bool,\n) {\n let arrivals: Vec<_> = state\n .in_flight\n .iter_mut()\n .filter_map(|(&sequence, packet)| {\n if packet.arrival_at_receiver > now {\n return None;\n }\n packet.arrival_at_receiver = MAX_SIMULATION + config.timing_tick;\n (!deterministically_lost(\n config.seed,\n sequence,\n packet.transmission,\n scenario.data_loss_percent,\n scenario.burst_loss,\n ))\n .then_some(sequence)\n })\n .collect();\n\n for sequence in arrivals {\n let transmission = state\n .in_flight\n .get(&sequence)\n .expect(\"arriving packet remains in flight\")\n .transmission;\n let jitter_key = if scenario.high_jitter {\n sequence\n .wrapping_mul(0x9e37_79b9_7f4a_7c15)\n .wrapping_add(u64::from(transmission))\n } else if scenario.reorder && sequence % 2 == 0 {\n sequence.saturating_add(1)\n } else if scenario.reorder {\n sequence.saturating_sub(1)\n } else {\n sequence\n };\n jitter_queue.push(now, jitter_key, sequence).map_or_else(\n |_| {\n // Queue full: apply backpressure by leaving the packet in flight and retrying it\n // next tick instead of panicking on a large configured jitter bound.\n if let Some(packet) = state.in_flight.get_mut(&sequence) {\n packet.arrival_at_receiver = now;\n }\n },\n |_| {},\n );\n }\n state.maximum_jitter_queue_packets = state.maximum_jitter_queue_packets.max(jitter_queue.len());\n\n let released = jitter_queue.drain_due(now);\n let ack_spacing = if scenario.return_delay >= ACK_RATE_CAP_RETURN_DELAY {\n Duration::from_secs_f64(1.0 / f64::from(config.ack_rate_limit))\n } else {\n Duration::from_millis(1)\n };\n let mut fresh_arrival = false;\n for sequence in released {\n let Some(packet) = state.in_flight.get(&sequence) else {\n continue;\n };\n let transmission = packet.transmission;\n if state.received.insert(sequence) {\n fresh_arrival = true;\n }\n let departure = state.next_ack_departure.max(now);\n state.next_ack_departure = departure + ack_spacing;\n // The asymmetric receiver returns the whole set of sequences it holds as disjoint ranges, so\n // one dropped acknowledgement datagram cannot masquerade as forward-path data loss. The\n // naive baseline keeps its additive single-sequence acknowledgement unchanged.\n let ranges = if selective_acks {\n acknowledgement_ranges(&state.received, config.maximum_ack_ranges)\n } else {\n vec![AckRange {\n start: sequence,\n end: sequence,\n }]\n };\n // Return-path reordering. On a reordering return path the odd acknowledgement is held\n // back by one timing tick, so a later additive acknowledgement overtakes it. The sender\n // therefore sees acknowledgements out of sequence order on the return path, which makes\n // `return_reordering` a real property of the return path instead of a relabelled\n // lossless link that merely perturbs the forward queue key.\n let reorder_offset = if scenario.reorder && sequence % 2 == 1 {\n config.timing_tick\n } else {\n Duration::ZERO\n };\n let arrival = departure + scenario.return_delay + reorder_offset;\n if !deterministically_lost(\n config.seed ^ 0xa11c,\n sequence,\n transmission,\n scenario.ack_loss_percent,\n false,\n ) {\n state.acknowledgements.push((arrival, ranges.clone()));\n state.ack_packets += 1;\n if scenario.duplicate_acks {\n state\n .acknowledgements\n .push((arrival + config.timing_tick, ranges));\n state.ack_packets += 1;\n }\n }\n }\n\n if !selective_acks {\n return;\n }\n if fresh_arrival {\n state.trailing_ack_credit = TRAILING_DUPLICATE_ACKS;\n return;\n }\n // Nothing new arrived, so repeat the current range acknowledgement a bounded number of times.\n // This recovers a dropped return datagram without the sender resending delivered data.\n while state.trailing_ack_credit > 0\n && !state.received.is_empty()\n && now >= state.next_ack_departure\n {\n let departure = state.next_ack_departure;\n state.next_ack_departure = departure + ack_spacing;\n let ranges = acknowledgement_ranges(&state.received, config.maximum_ack_ranges);\n if !deterministically_lost(\n config.seed ^ 0xa11c,\n departure.as_millis() as u64,\n TRAILING_DUPLICATE_ACKS - state.trailing_ack_credit,\n scenario.ack_loss_percent,\n false,\n ) {\n state\n .acknowledgements\n .push((departure + scenario.return_delay, ranges));\n state.ack_packets += 1;\n }\n state.trailing_ack_credit -= 1;\n }\n}\n\nfn deliver_acks(now: Duration, state: &mut SimulationState) {\n let _ = take_due_acks(now, state);\n}\n\n/// Collapses a received-sequence set into sorted, disjoint inclusive acknowledgement ranges.\nfn acknowledgement_ranges(received: &BTreeSet, maximum_ranges: usize) -> Vec {\n let mut ranges: Vec = Vec::new();\n for &sequence in received {\n if let Some(last) = ranges.last_mut()\n && last.end.checked_add(1) == Some(sequence)\n {\n last.end = sequence;\n continue;\n }\n ranges.push(AckRange {\n start: sequence,\n end: sequence,\n });\n }\n if ranges.len() > maximum_ranges {\n ranges.drain(..ranges.len() - maximum_ranges);\n }\n ranges\n}\n\n/// Delivers every acknowledgement that has arrived and returns how many new sequences they added.\nfn take_due_acks(now: Duration, state: &mut SimulationState) -> usize {\n let mut pending = Vec::with_capacity(state.acknowledgements.len());\n let mut newly_delivered = 0usize;\n let mut samples = Vec::new();\n for (arrival, ranges) in state.acknowledgements.drain(..) {\n if arrival > now {\n pending.push((arrival, ranges));\n continue;\n }\n for range in &ranges {\n // A range acknowledgement covers every sequence it names, so a lost data packet no\n // longer blocks acknowledgement of the packets that followed it.\n let covered: Vec = state\n .in_flight\n .range(range.start..=range.end)\n .map(|(&sequence, _)| sequence)\n .collect();\n for sequence in covered {\n if let Some(packet) = state.in_flight.remove(&sequence) {\n // The acknowledgement names the sequences the receiver holds, so the difference\n // between when the datagram arrived and when that packet was last sent is a real\n // round-trip sample.\n samples.push(arrival.saturating_sub(packet.sent_at));\n // End-to-end delivery latency is measured from the first transmission, so any\n // time spent in a retransmission episode is part of the tail.\n let delivery = arrival.saturating_sub(packet.first_sent_at);\n state.maximum_latency = Some(\n state\n .maximum_latency\n .map_or(delivery, |max| max.max(delivery)),\n );\n }\n if state.delivered.insert(sequence) {\n newly_delivered += 1;\n }\n }\n }\n }\n state.acknowledgements = pending;\n for sample in samples {\n record_rtt(state, sample);\n }\n newly_delivered\n}\n\n/// Folds one measured round-trip sample into the sender's RTO estimator.\n///\n/// The timeout is `srtt + 8 * rttvar` (a deliberately generous multiple of the mean deviation so\n/// that the serialization spread of a slow return path is covered), floored by the slowest round\n/// trip seen so far and by the observed path round trip. It therefore can never fall below the\n/// path RTT and never produces a spurious retransmission on a lossless path.\nfn record_rtt(state: &mut SimulationState, sample: Duration) {\n state.rtt_samples += 1;\n state.minimum_rtt = Some(state.minimum_rtt.map_or(sample, |min| min.min(sample)));\n state.maximum_rtt = Some(state.maximum_rtt.map_or(sample, |max| max.max(sample)));\n let sample_ms = sample.as_secs_f64() * 1000.0;\n match state.smoothed_rtt_ms {\n None => {\n state.smoothed_rtt_ms = Some(sample_ms);\n state.rtt_variation_ms = Some(sample_ms / 2.0);\n }\n Some(srtt) => {\n let variation = state.rtt_variation_ms.unwrap_or(sample_ms / 2.0);\n state.rtt_variation_ms = Some(variation * 0.75 + (srtt - sample_ms).abs() * 0.25);\n state.smoothed_rtt_ms = Some(srtt * 0.875 + sample_ms * 0.125);\n }\n }\n let srtt = state.smoothed_rtt_ms.unwrap_or(sample_ms);\n let variation = state.rtt_variation_ms.unwrap_or(sample_ms / 2.0);\n let multiplier = f64::from(state.rto_variance_multiplier.max(1));\n let candidate_ms = (srtt + multiplier * variation).max(1.0);\n let candidate = Duration::from_secs_f64(candidate_ms / 1000.0);\n // The timeout tracks the current smoothed estimate. It is floored only by the observed path\n // round trip (the minimum, not the running maximum) and the configured loss floor, so after an\n // old slow sample ages out of the EWMA the timeout can decrease again instead of being pinned\n // to the worst round trip ever seen.\n state.retransmission_timeout = candidate\n .max(state.minimum_rtt.unwrap_or_default())\n .max(state.minimum_loss_delay);\n}\n\nfn forward_delay(scenario: &Scenario) -> Duration {\n Duration::from_millis(u64::from(scenario.forward_hops) * 5)\n}\n\npub(crate) fn deterministically_lost(\n seed: u64,\n sequence: u64,\n transmission: u32,\n percent: u8,\n burst: bool,\n) -> bool {\n if percent == 0 {\n return false;\n }\n if burst {\n return burst_group_lost(seed, sequence, transmission, percent);\n }\n let mut value = seed ^ sequence.wrapping_mul(0x9e37_79b9_7f4a_7c15);\n value ^= u64::from(transmission).wrapping_mul(0xbf58_476d_1ce4_e5b9);\n value ^= value >> 30;\n value = value.wrapping_mul(0xbf58_476d_1ce4_e5b9);\n value ^= value >> 27;\n value = value.wrapping_mul(0x94d0_49bb_1331_11eb);\n value ^= value >> 31;\n value % 100 < u64::from(percent)\n}\n\n// The `burst_loss` scenario drops whole runs of consecutive sequences at once. Every packet in a\n// run shares one loss decision, so the reported loss pattern is bursty rather than the uniformly\n// spaced random drops the other lossy scenarios produce.\nfn burst_group_lost(seed: u64, sequence: u64, transmission: u32, percent: u8) -> bool {\n const BURST_SEQUENCES: u64 = 4;\n let group = sequence / BURST_SEQUENCES;\n let mut value = seed ^ group.wrapping_mul(0x9e37_79b9_7f4a_7c15);\n value ^= u64::from(transmission).wrapping_mul(0xbf58_476d_1ce4_e5b9);\n value ^= value >> 30;\n value = value.wrapping_mul(0xbf58_476d_1ce4_e5b9);\n value ^= value >> 27;\n value = value.wrapping_mul(0x94d0_49bb_1331_11eb);\n value ^= value >> 31;\n value % 100 < u64::from(percent)\n}\n\n/// Returns the deterministic per-packet forward-loss pattern for a scenario.\n///\n/// This is the raw sequence of drop decisions the sender sees, so callers can show that the burst\n/// scenario really is burstier than the random-loss scenario instead of merely dropping more.\n#[must_use]\npub fn data_loss_pattern(seed: u64, scenario: &Scenario, packet_count: u64) -> Vec {\n (0..packet_count)\n .map(|sequence| {\n deterministically_lost(\n seed,\n sequence,\n 0,\n scenario.data_loss_percent,\n scenario.burst_loss,\n )\n })\n .collect()\n}\n\n/// Length of the longest run of `true` values in a loss pattern.\n///\n/// A burst-loss pattern drops consecutive sequences together, so its longest run is longer than an\n/// independent-loss pattern that drops a similar percentage.\n#[must_use]\npub fn longest_run(pattern: &[bool]) -> usize {\n let mut longest = 0;\n let mut current = 0;\n for &dropped in pattern {\n current = if dropped { current + 1 } else { 0 };\n longest = longest.max(current);\n }\n longest\n}\n\n#[cfg(test)]\nmod tests {\n use super::{data_loss_pattern, longest_run, run_asymmetric_transport, run_naive_baseline};\n use crate::{TransportConfig, required_scenarios};\n use std::time::Duration;\n\n #[test]\n fn baseline_is_deterministic() {\n let config = TransportConfig::default();\n let scenario = &required_scenarios()[3];\n assert_eq!(\n run_naive_baseline(&config, scenario, 64),\n run_naive_baseline(&config, scenario, 64)\n );\n }\n\n #[test]\n fn high_jitter_uses_a_bounded_deterministic_queue() {\n let config = TransportConfig::default();\n let scenario = required_scenarios()\n .into_iter()\n .find(|scenario| scenario.name == \"high_seeded_jitter\")\n .unwrap();\n\n let first = run_naive_baseline(&config, &scenario, 64);\n let second = run_naive_baseline(&config, &scenario, 64);\n\n assert_eq!(first, second);\n assert!(first.maximum_jitter_queue_packets > 0);\n assert!(first.maximum_jitter_queue_packets <= config.send_window_packets);\n }\n\n #[test]\n fn asymmetric_sender_does_not_treat_ack_delay_as_forward_loss() {\n let config = TransportConfig::default();\n let mut scenario = required_scenarios()[3].clone();\n scenario.ack_loss_percent = 0;\n\n let baseline = run_naive_baseline(&config, &scenario, 256);\n let asymmetric = run_asymmetric_transport(&config, &scenario, 256);\n\n assert!(asymmetric.completed);\n assert!(asymmetric.retransmissions < baseline.retransmissions / 10);\n assert!(asymmetric.minimum_window_packets >= 16);\n assert!(asymmetric.completion_time < baseline.completion_time / 4);\n }\n\n #[test]\n fn asymmetric_sender_completes_the_required_matrix() {\n let config = TransportConfig::default();\n\n for scenario in required_scenarios() {\n let first = run_asymmetric_transport(&config, &scenario, 256);\n let second = run_asymmetric_transport(&config, &scenario, 256);\n assert_eq!(first, second, \"{} must be deterministic\", scenario.name);\n assert!(first.completed, \"{} must complete\", scenario.name);\n assert!(\n first.maximum_jitter_queue_packets <= config.send_window_packets,\n \"{} exceeded its jitter queue bound\",\n scenario.name\n );\n }\n }\n\n #[test]\n fn burst_loss_is_burstier_than_random_loss() {\n let config = TransportConfig::default();\n let scenarios = required_scenarios();\n let burst = scenarios\n .iter()\n .find(|scenario| scenario.name == \"burst_loss\")\n .unwrap();\n let random = scenarios\n .iter()\n .find(|scenario| scenario.name == \"random_loss\")\n .unwrap();\n\n let burst_pattern = data_loss_pattern(config.seed, burst, 256);\n let random_pattern = data_loss_pattern(config.seed, random, 256);\n\n assert!(\n longest_run(&burst_pattern) >= 3,\n \"burst loss should drop runs of consecutive packets: {burst_pattern:?}\"\n );\n assert!(\n longest_run(&random_pattern) <= 2,\n \"random loss should not produce long runs: {random_pattern:?}\"\n );\n assert_ne!(burst_pattern, random_pattern);\n }\n\n #[test]\n fn spurious_retransmissions_are_absent_on_a_lossless_forward_path() {\n // A high-jitter / reordering return path must not provoke retransmissions of packets the\n // receiver already holds. The count is now measured explicitly.\n let config = TransportConfig::default();\n let mut scenario = required_scenarios()[7].clone(); // high_seeded_jitter\n scenario.data_loss_percent = 0;\n scenario.ack_loss_percent = 0;\n let metrics = run_asymmetric_transport(&config, &scenario, 256);\n assert!(metrics.completed);\n assert_eq!(\n metrics.spurious_retransmissions, 0,\n \"no packet the receiver already holds should be retransmitted\"\n );\n }\n\n #[test]\n fn spurious_retransmission_counts_are_reported_for_each_named_scenario() {\n // The job requires the spurious-retransmission count for these three scenarios.\n let config = TransportConfig::default();\n for name in [\n \"reverse_path_bottleneck\",\n \"random_loss\",\n \"high_seeded_jitter\",\n ] {\n let scenario = required_scenarios()\n .into_iter()\n .find(|scenario| scenario.name == name)\n .unwrap();\n let metrics = run_asymmetric_transport(&config, &scenario, 256);\n assert!(metrics.completed, \"{name} must complete\");\n assert!(\n metrics.spurious_retransmissions <= metrics.retransmissions,\n \"{name}: spurious {} exceeds total {}\",\n metrics.spurious_retransmissions,\n metrics.retransmissions\n );\n println!(\n \"spurious {}: {} of {} retransmissions\",\n name, metrics.spurious_retransmissions, metrics.retransmissions\n );\n }\n }\n\n #[test]\n fn naive_baseline_grows_after_a_timeout_episode() {\n // The baseline used to freeze its window after the first timeout (it tested the cumulative\n // `timeout_events` counter). With growth repaired it recovers to the full configured send\n // window on a scenario whose losses are transient, and it still completes.\n let config = TransportConfig::default();\n let scenario = required_scenarios()\n .into_iter()\n .find(|scenario| scenario.name == \"random_loss\")\n .unwrap();\n let baseline = run_naive_baseline(&config, &scenario, 256);\n assert!(baseline.completed);\n assert!(baseline.timeout_events > 0);\n assert!(\n baseline.grew_after_timeout,\n \"post-timeout growth must resume after a loss episode, not freeze the window\"\n );\n }\n\n #[test]\n fn reverse_path_bottleneck_reports_measured_rtt_and_rto_above_it() {\n let config = TransportConfig::default();\n let scenario = required_scenarios()\n .into_iter()\n .find(|scenario| scenario.name == \"reverse_path_bottleneck\")\n .unwrap();\n\n let metrics = run_asymmetric_transport(&config, &scenario, 64);\n\n assert!(metrics.completed);\n assert!(metrics.measured_rtt > Duration::ZERO);\n assert!(metrics.retransmission_timeout >= metrics.measured_rtt);\n assert!(metrics.tail_latency >= metrics.measured_rtt);\n }\n\n #[test]\n fn large_transfer_completes_losslessly_and_deterministically_n1000() {\n let config = TransportConfig::default();\n let mut scenario = required_scenarios()[2].clone();\n scenario.data_loss_percent = 0;\n scenario.ack_loss_percent = 0;\n\n let first = run_asymmetric_transport(&config, &scenario, 1000);\n let second = run_asymmetric_transport(&config, &scenario, 1000);\n\n println!(\n \"large_transfer lossless n=1000: completed={} time={}ms rtx={} acks={} tail={:?} measured_rtt={:?} rto={:?}\",\n first.completed,\n first.completion_time.as_millis(),\n first.retransmissions,\n first.ack_packets,\n first.tail_latency,\n first.measured_rtt,\n first.retransmission_timeout\n );\n assert_eq!(\n first, second,\n \"lossless large transfer must be deterministic\"\n );\n assert!(\n first.completed,\n \"lossless transfer of 1000 packets must complete\"\n );\n assert_eq!(\n first.retransmissions, 0,\n \"no packet should be retransmitted on a lossless path\"\n );\n }\n\n #[test]\n fn large_transfer_completes_under_20_percent_loss_n300() {\n let config = TransportConfig::default();\n let mut scenario = required_scenarios()[2].clone();\n scenario.data_loss_percent = 20;\n scenario.ack_loss_percent = 0;\n\n let first = run_asymmetric_transport(&config, &scenario, 300);\n let second = run_asymmetric_transport(&config, &scenario, 300);\n\n println!(\n \"large_transfer lossy n=300 p=20%: completed={} time={}ms rtx={} acks={} tail={:?} measured_rtt={:?} rto={:?}\",\n first.completed,\n first.completion_time.as_millis(),\n first.retransmissions,\n first.ack_packets,\n first.tail_latency,\n first.measured_rtt,\n first.retransmission_timeout\n );\n assert_eq!(first, second, \"lossy large transfer must be deterministic\");\n assert!(\n first.completed,\n \"transfer of 300 packets with 20% loss must complete\"\n );\n assert!(\n first.retransmissions > 0,\n \"20% loss must force retransmissions\"\n );\n }\n}\n","Poc/22-asymmetric-transport/tests/spurious_retransmits.rs":"//! Regression proof for the spurious-retransmit fix on Datura PR #187 (rand0 comment 13770).\n//!\n//! The asymmetric-path sender used per-packet acknowledgements, so a single dropped return-path\n//! datagram looked like forward-path loss and the sender resent data the receiver already held.\n//! Measured before this fix: 43 spurious of 43 retransmissions in `reverse_path_bottleneck` and\n//! 16 spurious of 23 in `random_loss`. Returning the whole received set as disjoint\n//! acknowledgement ranges drops both to zero. This file fails if they come back, and also pins\n//! that genuine forward loss is still repaired — so the count cannot be driven to zero merely by\n//! not retransmitting.\n\nuse asymmetric_transport::{\n Scenario, TransportConfig, required_scenarios, run_asymmetric_transport,\n};\n\nfn scenario(name: &str) -> Scenario {\n required_scenarios()\n .into_iter()\n .find(|candidate| candidate.name == name)\n .unwrap_or_else(|| panic!(\"no scenario named {name}\"))\n}\n\nfn measured(name: &str) -> (u64, u64, bool, u128) {\n let metrics = run_asymmetric_transport(&TransportConfig::default(), &scenario(name), 256);\n println!(\n \"{name}: retransmissions={} spurious={} completion={}ms\",\n metrics.retransmissions,\n metrics.spurious_retransmissions,\n metrics.completion_time.as_millis()\n );\n (\n metrics.spurious_retransmissions,\n metrics.retransmissions,\n metrics.completed,\n metrics.completion_time.as_millis(),\n )\n}\n\n#[test]\nfn spurious_retransmits_stay_at_zero_on_the_named_scenarios() {\n // The two scenarios named in rand0 comment 13770. Both used to report spurious retransmissions\n // equal to (reverse_path_bottleneck) or a large fraction of (random_loss) their total retransmissions.\n for name in [\"reverse_path_bottleneck\", \"random_loss\"] {\n let (spurious, total, completed, _) = measured(name);\n assert!(completed, \"{name} must complete\");\n assert_eq!(\n spurious, 0,\n \"{name}: {spurious} of {total} retransmissions were of packets the receiver already held\"\n );\n }\n}\n\n#[test]\nfn genuine_forward_loss_is_still_repaired() {\n // The zero above must come from not resending delivered data, not from giving up on\n // retransmission: random_loss injects forward-path loss and must still retransmit to complete.\n let (spurious, total, completed, _) = measured(\"random_loss\");\n assert!(completed, \"random_loss must complete\");\n assert!(\n total > 0,\n \"random_loss injects forward loss, so it must still retransmit\"\n );\n assert_eq!(\n spurious, 0,\n \"no retransmission may be a resend of data the receiver already holds\"\n );\n}\n\n#[test]\nfn a_lossy_return_path_alone_causes_no_spurious_retransmits() {\n // A lossless forward path with a lossy return path used to look like forward loss once an\n // acknowledgement datagram was dropped. With whole-set acknowledgement ranges the sender\n // cannot make that mistake, so no packet is resent.\n let mut scenario = scenario(\"random_loss\");\n scenario.data_loss_percent = 0;\n scenario.ack_loss_percent = 30;\n let metrics = run_asymmetric_transport(&TransportConfig::default(), &scenario, 256);\n assert!(\n metrics.completed,\n \"a lossless forward path must complete even when the return path drops acknowledgements\"\n );\n assert_eq!(\n metrics.spurious_retransmissions, 0,\n \"a lossy return path must not be reported as forward-path loss\"\n );\n}\n"},"notes":"Fix for rand0 comment 13770 (spurious retransmits). Root cause: the asymmetric sender acknowledged a single sequence per ack datagram, so a dropped return-path datagram stalled its largest-acknowledged view while it kept sending; its fixed time-loss threshold then expired and it resent data the receiver already held. The fix returns the whole received set as disjoint acknowledgement ranges (AckRange) and retransmits only after the measured RTO (which includes the return path). tests/spurious_retransmits.rs fails if either count returns and pins that genuine forward loss is still repaired. fmt --check clean, clippy -D warnings clean, `cargo test --all-targets --no-fail-fast` all suites pass (35+0+1+10+3+8), including tests/review.rs (the previous submission's three review.rs failures are gone because review.rs on the current head matches the integrated transport). No secrets; no wallet or pricing changes; no maintainer contact."},"attachments":[],"redactions":0,"note":"Shown as stored."}