From a86b2824bf053c13a7c48921afae3f098070e8dd Mon Sep 17 00:00:00 2001 From: Transaction Science Date: Thu, 29 Oct 2026 06:22:32 -0400 Subject: [PATCH] joule-moq-relay: reset a stream its lane was killed under; eoc-meter: wait on the child's CPU MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Two tests failed intermittently on loaded CI hosts. One of them was a relay bug. joule-moq-relay (age_rising.rs:138, linux-arm64). A rise of a rating ends the subscriber's groups (end_groups -> kill). kill() resets the lane's stream only when it is in the lane, but while a write is in flight the write holds it. That write then returned it to the dead lane, the lane was removed, and a stream dropped unreset is finished whole. So a group blocked on flow control when the rise was held reached a viewer below the threshold whole. That breaks WAI age-assurance §5.5 ("forwards none of that group's objects"); cancel() has the same gap against §3.1.2's "MUST reset any open streams". on_done now resets a returned stream whose lane died meanwhile, with the lane's code, and cancel() records its code for that. Reproduced on the Linux build host, 200 runs eight at a time: 14 and 11 failures before (all ten groups whole, the relay sent the slow viewer no RESET_STREAM), 0 after (it sends four to seven). The crate's whole suite passes. eoc-meter (process_energy.rs:229, macOS). The child test stopped `yes` once its counter read 0.5 J and 300 ms of wall time had passed, then required 150 ms of reaped CPU. On the loaded Studio, 300 ms of wall time gave it 130 ms. The loop now waits on the child's own CPU (rusage, as it reads the energy) to 300 ms, for at most 10 s, and the reaped CPU is at least that. 30 of 30 runs pass on the Studio five at a time; clippy (-D warnings, as CI lints eoc-meter) is clean. Co-Authored-By: Claude Opus 5.5 --- .../crates/eoc-meter/tests/process_energy.rs | 15 ++++++++++----- .../crates/joule-moq-relay/src/sched.rs | 9 +++++++++ 2 files changed, 19 insertions(+), 5 deletions(-) diff --git a/eoc/eoc-rs/crates/eoc-meter/tests/process_energy.rs b/eoc/eoc-rs/crates/eoc-meter/tests/process_energy.rs index 963f86694..69f8ba9c3 100644 --- a/eoc/eoc-rs/crates/eoc-meter/tests/process_energy.rs +++ b/eoc/eoc-rs/crates/eoc-meter/tests/process_energy.rs @@ -190,8 +190,9 @@ fn a_sleeping_span_reads_under_0_1_j() { } /// A child process's work is not this process's energy, neither while it -/// runs nor once it is reaped. The child (`yes`, to /dev/null) runs for at -/// least 300 ms and until its own counter has read at least 0.5 J (or 5 s); +/// runs nor once it is reaped. The child (`yes`, to /dev/null) runs until it +/// has used at least 300 ms of CPU and its own counter has read at least 0.5 J (or +/// 10 s); /// the parent's counter, read before the spawn and after the reap, moves by /// under 0.1 J and by under a fifth of the child's own. #[test] @@ -203,10 +204,14 @@ fn a_child_processs_work_is_not_attributed() { let mut child = Command::new("yes").stdout(Stdio::null()).spawn().expect("spawn yes"); let started = Instant::now(); let mut child_nj = None; - while started.elapsed() < Duration::from_secs(5) { + // Wait on the child's own CPU time, not the wall clock: on a loaded host 300 ms of wall + // time gave it 130 ms of CPU. What is reaped below is at least what is read here. + while started.elapsed() < Duration::from_secs(10) { std::thread::sleep(Duration::from_millis(50)); - child_nj = energy_nj(child.id()); - if child_nj.is_some_and(|nj| nj >= 500_000_000) && started.elapsed() >= Duration::from_millis(300) { + let w = rusage(child.id()); + child_nj = w.map(|w| w[ENERGY_NJ]); + let cpu = w.map(|w| mach(w[USER_TIME] + w[SYSTEM_TIME])); + if child_nj.is_some_and(|nj| nj >= 500_000_000) && cpu.is_some_and(|c| c >= Duration::from_millis(300)) { break; } } diff --git a/joule-code/joule-code-rs/crates/joule-moq-relay/src/sched.rs b/joule-code/joule-code-rs/crates/joule-moq-relay/src/sched.rs index e20ef8153..50742faa1 100644 --- a/joule-code/joule-code-rs/crates/joule-moq-relay/src/sched.rs +++ b/joule-code/joule-code-rs/crates/joule-moq-relay/src/sched.rs @@ -687,6 +687,14 @@ impl Sched { if let Some(lane) = self.lanes.get_mut(&d.lane) { lane.busy = false; lane.send = d.send; + // Killed while this write was in flight (a rise, an unsubscribe): the stream + // was with the write then, not in the lane to reset, and a stream dropped + // unreset is finished whole. It is reset now. + if lane.dead + && let Some(s) = lane.send.as_mut() + { + s.reset(lane.dead_code.unwrap_or(StreamResetCode::CANCELLED.0)); + } } self.refresh(d.lane); } @@ -723,6 +731,7 @@ impl Sched { for k in keys { if let Some(lane) = self.lanes.get_mut(&k) { lane.dead = true; + lane.dead_code.get_or_insert(code); if let Some(s) = lane.send.as_mut() { s.reset(code); } -- 2.54.0 (Apple Git-157)