joule-code/joule-code-rs/crates/joule-moq-relay/tests/cascade.rs

@@ -403,7 +403,11 @@ async fn every_door() {
     let media_e: Vec<u64> = 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");
@@ -418,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 ─────────────────────────────────────────────────────────────────
@@ -852,13 +862,24 @@ async fn votes() {
     let only_first = vec![msf(&first)];
     assert_eq!(got_e.iter().filter(|p| **p != msf(&wr)).cloned().collect::<Vec<_>>(), 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");
     }
+    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