|
| 1 | +//! The join between resync ([`06-resync`](../docs/vrs/06-resync/spec.md)) and DING delivery. |
| 2 | +//! |
| 3 | +//! [`tests/resync.rs`] proves the ingress half and defers the wake to "the existing delivery |
| 4 | +//! suite". That suite proves the wake for a *publicly* emitted stream event |
| 5 | +//! ([`tests/event_e2e.rs::event_emit_cli_returns_a_stable_json_receipt_and_ding_marks_the_record`]), |
| 6 | +//! which reaches the inbox through a different admission path than a built-in resync record, and |
| 7 | +//! reads the inbox with `message::list_inbox` rather than the arrival scan the live loop uses. |
| 8 | +//! |
| 9 | +//! So nothing proved that a resync record survives DING's own `new_arrivals` scan — the place a |
| 10 | +//! stream predicate would silently swallow it — or that it renders as stream work. This file |
| 11 | +//! closes that seam against the real supervisor ingress. |
| 12 | +
|
| 13 | +use std::collections::HashSet; |
| 14 | +use std::fs; |
| 15 | +use std::path::{Path, PathBuf}; |
| 16 | +use std::time::{Duration, Instant}; |
| 17 | + |
| 18 | +fn write_agent(root: &Path) -> PathBuf { |
| 19 | + let dir = root.join("agents/hetz/worker"); |
| 20 | + fs::create_dir_all(&dir).unwrap(); |
| 21 | + fs::write( |
| 22 | + dir.join("agent.kdl"), |
| 23 | + r#"agent "worker" { |
| 24 | + host "hetz" |
| 25 | + command "agent" |
| 26 | + resource "goal" uri="resources/goal.md" reason="Mission." |
| 27 | +}"#, |
| 28 | + ) |
| 29 | + .unwrap(); |
| 30 | + st2::event::publish_owner_binding_for_test(root, "hetz").unwrap(); |
| 31 | + dir |
| 32 | +} |
| 33 | + |
| 34 | +fn resync_records(inbox: &Path) -> usize { |
| 35 | + fs::read_dir(inbox) |
| 36 | + .map(|entries| { |
| 37 | + entries |
| 38 | + .flatten() |
| 39 | + .filter(|entry| { |
| 40 | + fs::read_to_string(entry.path()) |
| 41 | + .is_ok_and(|contents| contents.contains("stream: resync")) |
| 42 | + }) |
| 43 | + .count() |
| 44 | + }) |
| 45 | + .unwrap_or(0) |
| 46 | +} |
| 47 | + |
| 48 | +fn wait_for(condition: impl Fn() -> usize, expected: usize) -> bool { |
| 49 | + let deadline = Instant::now() + Duration::from_secs(15); |
| 50 | + while Instant::now() < deadline { |
| 51 | + if condition() >= expected { |
| 52 | + return true; |
| 53 | + } |
| 54 | + std::thread::sleep(Duration::from_millis(50)); |
| 55 | + } |
| 56 | + condition() >= expected |
| 57 | +} |
| 58 | + |
| 59 | +/// A carrier change becomes a `[DING]` notice: the record is a new arrival to DING's own scan, and |
| 60 | +/// it renders as stream work rather than as a message from an unknown peer. |
| 61 | +#[test] |
| 62 | +fn a_resync_record_is_a_ding_arrival_and_renders_as_stream_work() { |
| 63 | + let catalog = tempfile::tempdir().unwrap(); |
| 64 | + let agent_dir = write_agent(catalog.path()); |
| 65 | + let inbox = agent_dir.join("resources/inbox"); |
| 66 | + |
| 67 | + let supervisor = |
| 68 | + st2::resync::ResyncSupervisor::spawn(catalog.path().to_path_buf(), "hetz".to_owned()); |
| 69 | + assert!( |
| 70 | + supervisor |
| 71 | + .refresh( |
| 72 | + &st2::discover_strict(catalog.path()).specs, |
| 73 | + "hetz", |
| 74 | + &[], |
| 75 | + &[], |
| 76 | + ) |
| 77 | + .is_empty() |
| 78 | + ); |
| 79 | + std::thread::sleep(Duration::from_millis(300)); |
| 80 | + |
| 81 | + // Start DING's seen-set from the current unread set, exactly as `run_ding` does at startup, so |
| 82 | + // the arrival below is the only thing this scan can report. |
| 83 | + let mut seen: HashSet<String> = HashSet::new(); |
| 84 | + let backlog = st2::ding::new_arrivals(&inbox, &mut seen); |
| 85 | + assert!( |
| 86 | + backlog.is_empty(), |
| 87 | + "seeding is silent, so there is no backlog" |
| 88 | + ); |
| 89 | + |
| 90 | + let goal = agent_dir.join("resources/goal.md"); |
| 91 | + fs::create_dir_all(goal.parent().unwrap()).unwrap(); |
| 92 | + fs::write(&goal, "ship the thing\n").unwrap(); |
| 93 | + assert!( |
| 94 | + wait_for(|| resync_records(&inbox), 1), |
| 95 | + "the goal carrier change must reach the inbox" |
| 96 | + ); |
| 97 | + |
| 98 | + let arrivals = st2::ding::new_arrivals(&inbox, &mut seen); |
| 99 | + assert_eq!( |
| 100 | + arrivals.len(), |
| 101 | + 1, |
| 102 | + "the resync record must reach DING's arrival scan, not be filtered out of it" |
| 103 | + ); |
| 104 | + let record = &arrivals[0]; |
| 105 | + assert_eq!(record.stream.as_deref(), Some("resync")); |
| 106 | + assert!( |
| 107 | + record.event_id.is_some(), |
| 108 | + "a resync record carries an event id" |
| 109 | + ); |
| 110 | + |
| 111 | + let notice = st2::ding::poke_text(catalog.path(), "hetz", "hetz.worker", record); |
| 112 | + assert!( |
| 113 | + notice.starts_with("[DING] » hetz.worker/resync: resource goal changed"), |
| 114 | + "a resync record renders as stream work: {notice}" |
| 115 | + ); |
| 116 | + |
| 117 | + // The arrival is consumed exactly once: a second scan re-poking it would duplicate the wake. |
| 118 | + assert!( |
| 119 | + st2::ding::new_arrivals(&inbox, &mut seen).is_empty(), |
| 120 | + "a delivered resync record must not be reported as a new arrival again" |
| 121 | + ); |
| 122 | +} |
0 commit comments