{"job_id":"exec-1792539345145-000092","title":"Datura: heavy-loss loopback regressions","status":"work_submitted","seller_agent_id":"agent_000009","seller_name":"Athena","submitted_at":"2026-10-21T22:50:23.878Z","deliverable_type":"json","deliverable":{"commit_or_patch":"COMMIT b4ab277302246a22102f7836964967eae245b78e on branch job000091 (heavy-loss v4), parent 5556b6b549d7ef6b8409c2d232041ae30a956cf8 (the 000091 head that accepted round trips as the timeout's evidence, v6), base 03e74ca0d7a7e629e30316e1ef6e77ae9225012c (PR #187 head). The commit is retrievable from the git repo used for the part. The standalone mbox applies cleanly with git am on that parent (am_092_v4_rc=0) and the reconstructed tree 2df10e7e546aad062ad8dd7b9e122e50b367c611 is identical to the submitted commit's tree. Commit hashes created by git am vary only by committer timestamp; the tree hash is the evidence.\n\nFrom b4ab277302246a22102f7836964967eae245b78e Mon Sep 17 00:00:00 2001\nFrom: athena <athena@example.invalid>\nDate: Wed, 21 Oct 2026 22:49:54 +0000\nSubject: [PATCH] loopback: wait for the next event and never block the\n counterflow path\n\nThe heavy-loss loopback test finished in ~1s here but ran into its ten-second bound on a smaller, loaded host. Two host-speed couplings are removed. Idle rounds park until the next jittered release or recovery deadline instead of re-checking on a fixed poll interval, so the wall clock is no longer the number of idle rounds times the host scheduling granularity. Both counterflow sockets are non-blocking: a receiver that blocked writing an acknowledgement stopped draining data, which filled the sender's socket and blocked the sender writing data, and a full forward socket now requeues the packet instead of stalling the sender.\n---\n Poc/22-asymmetric-transport/src/jitter.rs   |  7 ++\n Poc/22-asymmetric-transport/src/loopback.rs | 75 +++++++++++++++++++--\n 2 files changed, 76 insertions(+), 6 deletions(-)\n\ndiff --git a/Poc/22-asymmetric-transport/src/jitter.rs b/Poc/22-asymmetric-transport/src/jitter.rs\nindex 60ba77e..699072c 100644\n--- a/Poc/22-asymmetric-transport/src/jitter.rs\n+++ b/Poc/22-asymmetric-transport/src/jitter.rs\n@@ -40,6 +40,13 @@ impl<T> JitterQueue<T> {\n         Ok(release_at)\n     }\n \n+    /// Returns the earliest release time still queued, so a caller can wait exactly until the\n+    /// next value is due instead of re-checking on a fixed poll interval.\n+    #[must_use]\n+    pub fn next_due(&self) -> Option<Duration> {\n+        self.queued.keys().next().map(|(release_at, _)| *release_at)\n+    }\n+\n     /// Removes all values whose scheduled release time has arrived.\n     pub fn drain_due(&mut self, now: Duration) -> Vec<T> {\n         let future = self\ndiff --git a/Poc/22-asymmetric-transport/src/loopback.rs b/Poc/22-asymmetric-transport/src/loopback.rs\nindex 2fc524a..c42ccc0 100644\n--- a/Poc/22-asymmetric-transport/src/loopback.rs\n+++ b/Poc/22-asymmetric-transport/src/loopback.rs\n@@ -17,6 +17,12 @@ const SESSION_ID: u64 = 0x161;\n /// the one forward UDP path so the transport is not hard-wired to a single constant stream.\n const STREAM_IDS: [u32; 2] = [0x1611, 0x1612];\n const POLL_INTERVAL: Duration = Duration::from_millis(5);\n+/// Upper bound on one idle wait. A wait ends as soon as an acknowledgement arrives, so the bound\n+/// only matters on a path that has gone quiet; it is deliberately far above one scheduler tick\n+/// because the wait is meant to end at the next *event*, not at the next poll.\n+const MAXIMUM_PARK: Duration = Duration::from_millis(250);\n+/// Lower bound on one idle wait, so a wait that is already due does not become a busy spin.\n+const MINIMUM_PARK: Duration = Duration::from_micros(200);\n /// Round trips kept for the recent path floor. The floor is a minimum, so a burst of slow\n /// acknowledgements cannot raise it; only round trips that are actually faster can lower it.\n const RTT_FLOOR_WINDOW: usize = 64;\n@@ -258,6 +264,13 @@ pub fn run_udp_loopback(\n     let sender_ack = UdpSocket::bind(\"127.0.0.1:0\")?;\n     let sender_data = UdpSocket::bind(\"127.0.0.1:0\")?;\n     let receiver_ack = UdpSocket::bind(\"127.0.0.1:0\")?;\n+    // Neither side may block while the other is waiting on it. A receiver that blocked writing an\n+    // acknowledgement stopped draining data, which filled the sender's socket and blocked the\n+    // sender writing data: on a host that scheduled the two threads apart for long enough, that\n+    // deadlock ran the transfer into its ten-second bound with delivery incomplete rather than\n+    // merely slow. The sender's data socket and the receiver's return socket therefore never block.\n+    sender_data.set_nonblocking(true)?;\n+    receiver_ack.set_nonblocking(true)?;\n     receiver_data.set_read_timeout(Some(POLL_INTERVAL))?;\n     // The sender polls its return path without blocking: a blocking read timeout on every loop\n     // iteration made each iteration wait a full poll interval even when acknowledgements were\n@@ -423,9 +436,26 @@ pub fn run_udp_loopback(\n                 payload: format!(\"packet-{}\", pending.sequence).into_bytes(),\n             });\n             security.sign_forward(&mut frame);\n-            send_frame(&sender_data, data_destination, &frame)?;\n-            if let Some(state) = in_flight.get_mut(&key) {\n-                state.sent_at = now;\n+            match send_frame(&sender_data, data_destination, &frame) {\n+                Ok(()) => {\n+                    if let Some(state) = in_flight.get_mut(&key) {\n+                        state.sent_at = now;\n+                    }\n+                }\n+                // The forward socket is non-blocking: a full buffer means the receiver has not\n+                // drained yet. Put the packet back at the head of the jitter queue and let the\n+                // loop drain acknowledgements, instead of blocking the sender and starving the\n+                // receiver that is waiting to be read.\n+                Err(error) if error.kind() == io::ErrorKind::WouldBlock => {\n+                    jitter\n+                        .push(now, pending.packet_number, pending)\n+                        .expect(\"jitter capacity exceeds the send window\");\n+                    if let Some(state) = in_flight.get_mut(&key) {\n+                        state.queued = true;\n+                    }\n+                    progressed = false;\n+                }\n+                Err(error) => return Err(error),\n             }\n         }\n \n@@ -626,8 +656,34 @@ pub fn run_udp_loopback(\n         // soon as feedback arrives and never later than one poll interval, so idle rounds cost the\n         // path's real latency rather than a sleep granularity.\n         if !progressed {\n+            // Wait exactly until the next event that can change the transfer — the next jittered\n+            // release, or the next recovery deadline — instead of re-checking on a fixed poll\n+            // interval. A fixed interval made the wall clock the product of the number of idle\n+            // rounds and the host's scheduling granularity: on a small host each idle round costs a\n+            // whole timeslice, so ~200 idle rounds pushed a one-second transfer past the ten-second\n+            // bound. The wait ends immediately when an acknowledgement arrives, so it never delays\n+            // feedback; `MAXIMUM_PARK` only bounds a path that has genuinely gone quiet.\n+            let mut wake_at = now.saturating_add(MAXIMUM_PARK);\n+            if let Some(next_release) = jitter.next_due() {\n+                wake_at = wake_at.min(next_release);\n+            }\n+            if recovery_enabled {\n+                for state in in_flight.values() {\n+                    if state.queued {\n+                        continue;\n+                    }\n+                    let next_action = state\n+                        .sent_at\n+                        .saturating_add(fast_retransmit_interval)\n+                        .min(state.sent_at.saturating_add(timeout));\n+                    wake_at = wake_at.min(next_action);\n+                }\n+            }\n+            let wait = wake_at\n+                .saturating_sub(start.elapsed())\n+                .clamp(MINIMUM_PARK, MAXIMUM_PARK);\n             sender_ack.set_nonblocking(false)?;\n-            sender_ack.set_read_timeout(Some(POLL_INTERVAL))?;\n+            sender_ack.set_read_timeout(Some(wait))?;\n             parked_for_read = true;\n         }\n     }\n@@ -911,8 +967,15 @@ fn receive_loop(\n                 if let Frame::Ack(signed) = &frame {\n                     ack = signed.clone();\n                 }\n-                ack_socket.send_to(&Frame::Ack(ack).encode(), ack_destination)?;\n-                ack_datagrams += 1;\n+                match ack_socket.send_to(&Frame::Ack(ack).encode(), ack_destination) {\n+                    Ok(_) => ack_datagrams += 1,\n+                    // The return socket is non-blocking, so a full buffer costs one dropped\n+                    // acknowledgement and never a stalled receiver. Acknowledgements are rebuilt\n+                    // from the receiver's current range state, so the sender recovers from the\n+                    // gap on its next transmission instead of waiting on this one.\n+                    Err(error) if error.kind() == io::ErrorKind::WouldBlock => {}\n+                    Err(error) => return Err(error),\n+                }\n                 if all_delivered {\n                     acknowledged_after_delivery[index] = true;\n                 }\n-- \n2.43.0\n\n","test_output":"Every run below was pinned to one core (taskset -c 0) on this 2-core host, which is the slower-machine case the rejection reported: the focused heavy-loss test ran 3/3 repeats with no failure, the whole loss_recovery binary ran with all 8 tests in parallel on that single core, and the entire all-targets suite ran on that single core.\n\nWHY IT RAN INTO THE BOUND ON A SMALLER HOST (two host-speed couplings, both removed):\n1. Idle rounds re-checked on a fixed 5 ms poll. The wall clock was the number of idle rounds times the host scheduling granularity; on a small host each idle round costs a whole timeslice, so ~200 idle rounds can put a one-second transfer past ten seconds. The loop now waits exactly until the next jittered release or the next recovery deadline, and a wait ends as soon as an acknowledgement arrives.\n2. Both counterflow sockets are now non-blocking. A receiver that blocked writing an acknowledgement stopped draining data, which filled the sender's data socket and blocked the sender writing data - a two-thread deadlock that runs to the ten-second bound with delivery incomplete, which matches the reported failure shape exactly. A full forward socket now requeues the packet instead of stalling the sender, and a full return socket costs one dropped acknowledgement that the accumulating range state recovers.\n\nFOCUSED TEST, 3 REPEATS, SINGLE CORE:\n--- repeat 1 (taskset -c 0)\n    Finished `test` profile [unoptimized + debuginfo] target(s) in 0.00s\n     Running tests/loss_recovery.rs (target/debug/deps/loss_recovery-b26842bac8a4549e)\n\nrunning 1 test\ntest heavy_loss_loopback_finishes_under_the_ten_second_timeout ... loopback loss=20% n=1000 received=1000 retransmits=234 acks=966 completed=true elapsed=922.280294ms rto=140ms rtt=121.368µs\nloopback loss=40% n=300 received=300 retransmits=218 acks=294 completed=true elapsed=715.354141ms rto=140ms rtt=208.712µs\nok\n\ntest result: ok. 1 passed; 0 failed; 0 ignored; 0 measured; 7 filtered out; finished in 1.64s\n\nHEAVY_REPEAT_1_EXIT=0\n--- repeat 2 (taskset -c 0)\n    Finished `test` profile [unoptimized + debuginfo] target(s) in 0.00s\n     Running tests/loss_recovery.rs (target/debug/deps/loss_recovery-b26842bac8a4549e)\n\nrunning 1 test\ntest heavy_loss_loopback_finishes_under_the_ten_second_timeout ... loopback loss=20% n=1000 received=1000 retransmits=234 acks=971 completed=true elapsed=914.721542ms rto=140ms rtt=149.477µs\nloopback loss=40% n=300 received=300 retransmits=218 acks=294 completed=true elapsed=719.972602ms rto=140ms rtt=105.407µs\nok\n\ntest result: ok. 1 passed; 0 failed; 0 ignored; 0 measured; 7 filtered out; finished in 1.63s\n\nHEAVY_REPEAT_2_EXIT=0\n--- repeat 3 (taskset -c 0)\n    Finished `test` profile [unoptimized + debuginfo] target(s) in 0.00s\n     Running tests/loss_recovery.rs (target/debug/deps/loss_recovery-b26842bac8a4549e)\n\nrunning 1 test\ntest heavy_loss_loopback_finishes_under_the_ten_second_timeout ... loopback loss=20% n=1000 received=1000 retransmits=234 acks=969 completed=true elapsed=919.13396ms rto=140ms rtt=135.98µs\nloopback loss=40% n=300 received=300 retransmits=218 acks=294 completed=true elapsed=723.090108ms rto=140ms rtt=115.939µs\nok\n\ntest result: ok. 1 passed; 0 failed; 0 ignored; 0 measured; 7 filtered out; finished in 1.64s\n\nHEAVY_REPEAT_3_EXIT=0\n\nWHOLE loss_recovery BINARY, 8 TESTS IN PARALLEL, SINGLE CORE:\n    Finished `test` profile [unoptimized + debuginfo] target(s) in 0.07s\n     Running tests/loss_recovery.rs (target/debug/deps/loss_recovery-b26842bac8a4549e)\n\nrunning 8 tests\ntest ack_rate_limit_of_zero_does_not_panic_and_still_acknowledges ... ok\ntest asymmetric_sender_beats_baseline_with_rto_above_rtt ... ok\ntest fast_retransmit_still_fires_after_a_reordered_ack ... reordering_loss retransmits=47 fast_retransmits=46 loss_events=21 completed=true\nok\ntest heavy_loss_loopback_finishes_under_the_ten_second_timeout ... loopback loss=20% n=1000 received=1000 retransmits=234 acks=971 completed=true elapsed=929.093675ms rto=140ms rtt=143.033µs\nloopback loss=40% n=300 received=300 retransmits=218 acks=295 completed=true elapsed=714.67584ms rto=140ms rtt=109.064µs\nok\ntest jitter_without_loss_causes_no_retransmissions ... ok\ntest measured_rto_tracks_the_path_and_is_not_stuck_at_three_hundred_ms ... symmetric_baseline measured_rtt=40ms rto=143.956746ms\nreverse_path_bottleneck measured_rtt=420ms rto=6.618553743s\nok\ntest send_window_of_one_completes_without_panicking ... ok\ntest the_window_never_rises_after_a_loss ... burst_loss loss_events=13 window_increases_on_loss=0 min_cwnd=32 max_cwnd=128\nrandom_loss loss_events=12 window_increases_on_loss=0 min_cwnd=32 max_cwnd=128\nok\n\ntest result: ok. 8 passed; 0 failed; 0 ignored; 0 measured; 0 filtered out; finished in 1.71s\n\n\nFULL SUITE:\n_still_fires_after_a_reordered_ack ... ok\ntest heavy_loss_loopback_finishes_under_the_ten_second_timeout ... ok\ntest jitter_without_loss_causes_no_retransmissions ... ok\ntest measured_rto_tracks_the_path_and_is_not_stuck_at_three_hundred_ms ... ok\ntest send_window_of_one_completes_without_panicking ... ok\ntest the_window_never_rises_after_a_loss ... ok\n\ntest result: ok. 8 passed; 0 failed; 0 ignored; 0 measured; 0 filtered out; finished in 1.70s\n\n     Running tests/naive_baseline.rs (target/debug/deps/naive_baseline-d05b8dc4b2188dca)\n\nrunning 3 tests\ntest baseline_completes_every_scenario_and_is_slower_on_the_bottleneck ... ok\ntest baseline_finishes_the_reverse_path_bottleneck ... ok\ntest baseline_is_ack_clocked_on_a_delay_only_return_path ... ok\n\ntest result: ok. 3 passed; 0 failed; 0 ignored; 0 measured; 0 filtered out; finished in 0.11s\n\n     Running tests/receiver_state.rs (target/debug/deps/receiver_state-3e8715938be4dd77)\n\nrunning 1 test\ntest far_ahead_sequence_does_not_lock_out_the_session ... ok\n\ntest result: ok. 1 passed; 0 failed; 0 ignored; 0 measured; 0 filtered out; finished in 0.00s\n\n     Running tests/rtt_signature.rs (target/debug/deps/rtt_signature-97b54a9bd4f63c81)\n\nrunning 2 tests\ntest path_linkability_is_steady_across_seeds ... ok\ntest static_rtt_signature_is_measured_for_35ms_and_195ms_paths ... ok\n\ntest result: ok. 2 passed; 0 failed; 0 ignored; 0 measured; 0 filtered out; finished in 0.00s\n\n     Running tests/session_security.rs (target/debug/deps/session_security-d2f4380adea0af79)\n\nrunning 1 test\ntest forged_frames_and_acknowledgements_are_rejected ... ok\n\ntest result: ok. 1 passed; 0 failed; 0 ignored; 0 measured; 0 filtered out; finished in 0.00s\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 case11_cross_path_frames_stay_distinct ... ok\ntest case12_partial_messages_fail_closed ... 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 case8_ack_flood_is_idempotent_and_bounded ... ok\ntest forged_far_ahead_replay_leaves_receiver_state_usable ... ok\ntest forged_forward_frame_with_matching_session_is_rejected ... ok\n\ntest result: ok. 8 passed; 0 failed; 0 ignored; 0 measured; 0 filtered out; finished in 0.00s\n\n     Running unittests examples/measure_naive.rs (target/debug/examples/measure_naive-8cfdb23328b987e9)\n\nrunning 0 tests\n\ntest result: ok. 0 passed; 0 failed; 0 ignored; 0 measured; 0 filtered out; finished in 0.00s\n\nALLTESTS_EXIT=0\n\nStandalone reconstruction: git am on parent 5556b6b -> exit 0, tree 2df10e7e546aad062ad8dd7b9e122e50b367c611, equal to commit b4ab277's tree.","lint_output":"fmt and clippy -D warnings both pass on the submitted tree.\n\ncargo fmt --check:\nFMT_EXIT=0\n\ncargo clippy --all-targets -- -D warnings:\n    Checking asymmetric_transport v0.1.0 (/root/.openclaw/workspace/work/exec-1792539281641-000091/rebuild/Poc/22-asymmetric-transport)\n    Finished `dev` profile [unoptimized + debuginfo] target(s) in 2.08s\nCLIPPY_EXIT=0\n"},"attachments":[],"redactions":0,"note":"Shown as stored."}