From cbdf2f20398cade211b0baf8d4cf246de02d7857 Mon Sep 17 00:00:00 2001 From: Transaction Science Date: Tue, 3 Nov 2026 18:27:19 -0500 Subject: [PATCH 1/2] joule-moq-relay: the cascade tests wait for a refusal, not only for what got through The door test read the edge's refused_at_door as soon as its subscriber held every other group. Each group's door verdict is reached in that group's own task, and nothing the subscriber is given waits for the tampered group's, so under load the read could come first and see 0 (CI, linux-arm64). The vote test reads the edge's refused inputs the same way, after a one-second grace. Both now wait for the refusal, bounded, then count it exactly. Co-Authored-By: Claude Opus 5.5 --- .../crates/joule-moq-relay/tests/cascade.rs | 9 ++++++++- 1 file changed, 8 insertions(+), 1 deletion(-) diff --git a/joule-code/joule-code-rs/crates/joule-moq-relay/tests/cascade.rs b/joule-code/joule-code-rs/crates/joule-moq-relay/tests/cascade.rs index 34ad0d60e..a4e2c9812 100644 --- a/joule-code/joule-code-rs/crates/joule-moq-relay/tests/cascade.rs +++ b/joule-code/joule-code-rs/crates/joule-moq-relay/tests/cascade.rs @@ -403,7 +403,11 @@ async fn every_door() { let media_e: Vec = got_e.iter().filter(|x| x.header.track_alias == se.media_alias && x.reset.is_none() && !x.objects.is_empty()).map(|x| x.header.group_id).collect(); assert!(!media_e.contains(&BAD), "the edge forwarded no tampered object: {media_e:?}"); assert_eq!(media_e.len() as u64, G - 1, "every other group reached the edge's subscriber"); - assert_eq!(edge.relay().stats().refused_at_door, 1, "the edge's door refused the tampered object"); + // Each group's door verdict is reached in that group's own task, and nothing the + // subscriber was given waits for group BAD's: every other group can reach it while BAD is + // still on its way to the door. So the refusal is waited for, then counted. + assert!(until(STEP, || edge.relay().stats().refused_at_door >= 1).await, "the edge's door refused the tampered object"); + assert_eq!(edge.relay().stats().refused_at_door, 1, "the edge's door refused the tampered object, and nothing else"); let verifier = ReceiptVerifier::with_anchors(vec![key.verifying_key()]); for x in got_e.iter().filter(|x| x.header.track_alias == se.receipts_alias) { verifier.verify_group(&ReceiptBundle::from_payload(&x.objects[0].payload).unwrap().group).expect("every receipt verifies"); @@ -852,6 +856,9 @@ async fn votes() { let only_first = vec![msf(&first)]; assert_eq!(got_e.iter().filter(|p| **p != msf(&wr)).cloned().collect::>(), only_first, "the edge's watcher: seq 0 alone"); assert_eq!(got_h, only_first, "the home watcher is given exactly what the edge's is"); + // Seq 1 is refused in its own object's task, woken once seq 0 is counted: nothing the + // watchers were given waits for that, so the refusal is waited for, then counted. + assert!(until(STEP, || fe.refused_inputs() >= 1).await, "the edge refuses seq 1"); assert_eq!((fe.refused_inputs(), fh.refused_inputs()), (1, 0), "decided once, at the edge, where the member publishes"); for x in [&c, &we, &pm, &wh] { x.close(0, "done"); -- 2.54.0 (Apple Git-157) From 1bf2f7fce3a12d34f94a29fd017bc3b395fbd886 Mon Sep 17 00:00:00 2001 From: Transaction Science Date: Tue, 3 Nov 2026 19:10:52 -0500 Subject: [PATCH 2/2] joule-moq-relay: the cascade tests count what the edge gave, not only what was read Waiting for the refusal let a relay that counted it late and then forwarded the refused object pass: the door test reads its subscriber only until it holds the other groups, the vote test for a second past seq 0. Both now also count what the edge gave its viewer, from its session summaries once the sessions end: every group but the tampered one, and seq 0 alone. Co-Authored-By: Claude Opus 5.5 --- .../crates/joule-moq-relay/tests/cascade.rs | 14 ++++++++++++++ 1 file changed, 14 insertions(+) diff --git a/joule-code/joule-code-rs/crates/joule-moq-relay/tests/cascade.rs b/joule-code/joule-code-rs/crates/joule-moq-relay/tests/cascade.rs index a4e2c9812..67cb910d9 100644 --- a/joule-code/joule-code-rs/crates/joule-moq-relay/tests/cascade.rs +++ b/joule-code/joule-code-rs/crates/joule-moq-relay/tests/cascade.rs @@ -422,6 +422,12 @@ async fn every_door() { // The refusal is counted against the link it arrived on: the home relay's session. let link = er.summaries().into_iter().find(|s| s.summary.tally.peer.is_some()).expect("the link's summary"); assert_eq!(link.summary.tally.publisher.refused_at_door, 1); + // A refusal counted is not a group withheld, and the subscriber was read only until it + // held every other group: a door that counted late and then forwarded would pass both. + // What the edge gave its viewer, counted once the session ended, is every group but BAD. + let viewer = er.summaries().into_iter().find(|s| s.summary.tally.peer.is_none()).expect("the viewer's summary"); + let m = viewer.summary.tally.subscriber.tracks.get(&joule_moq::draft22::names::render(&t.ns, Some(&t.media))).cloned().unwrap_or_default(); + assert_eq!(m.objects_forwarded + m.undelivered_at_close, G - 1, "the edge gave its viewer no tampered object"); } // ── A loop ───────────────────────────────────────────────────────────────── @@ -863,9 +869,17 @@ async fn votes() { for x in [&c, &we, &pm, &wh] { x.close(0, "done"); } + let er = edge.relay().clone(); edge.shutdown().await; home.shutdown().await; let _ = std::fs::remove_file(&peer_file); + // The watchers were read for a second past seq 0, so a refusal counted and then forwarded + // later than that would pass them. What the edge gave its watcher, counted once its + // sessions ended, is seq 0 alone. + assert!(until(STEP, || er.summaries().len() >= 3).await, "the edge's sessions' summaries"); + let member_events = joule_moq::draft22::names::render(&room_track(room, &member, "events"), Some(&name("0"))); + let given: u64 = er.summaries().iter().filter(|s| s.summary.tally.peer.is_none()).filter_map(|s| s.summary.tally.subscriber.tracks.get(&member_events).cloned()).map(|t| t.objects_forwarded + t.undelivered_at_close).sum(); + assert_eq!(given, 1, "the edge gave its watcher seq 0 alone"); } /// A `WPT1` schedule (co-performance §6.1) under refuse: from `from`, one segment of two turns -- 2.54.0 (Apple Git-157)