import io P = "crates/spt-daemon/src/registryhost.rs" NL = "\r\n" src = io.open(P, "r", encoding="utf-8", newline="").read() anchor = NL.join([ " fn host(dir: &Path) -> RegistryHost {", " RegistryHost::new_at(", ' "aa11",', ' EpochSource::load_from(&dir.join("epoch.json")),', ' dir.join("registry"),', " )", " }", ]) assert src.count(anchor) == 1, ("host helper", src.count(anchor)) cells = NL.join([ "", " // [unit->REQ-REGISTRY-SNAPSHOT-HYDRATE] #281 face 2, the titled defect: a", " // daemon that restarts must open with the peer rows it last knew, not with", " // an empty map it can only refill from the next gossip round. The second", " // host() at the SAME dir IS the restart -- a new process reading the same", " // snapshot dir is exactly what the daemon does at boot.", " #[test]", " fn restart_hydrates_peer_rows_instead_of_forgetting_them() {", " let dir = tempfile::tempdir().unwrap();", " let h1 = host(dir.path());", " h1.apply_feed(", ' "bb22",', " &[RegistryUpdate {", ' subnet: "home".into(),', ' endpoint_id: "ling".into(),', ' instance: identified("bb22", "PEERBOX", "mid-1", 7),', " }],", ' &policy("home", "bb22"),', " );", ' assert!(!h1.rows("home", "ling").is_empty(), "feed admitted");', " drop(h1);", "", " // The restart.", " let h2 = host(dir.path());", ' let seated = h2.rows("home", "ling");', " assert_eq!(", " seated.len(),", " 1,", ' "restarted host seats the peer row from its own snapshot, before any feed"', " );", ' assert_eq!(seated[0].node, "bb22");', " assert_eq!(", " seated[0].epoch, 7,", ' "the epoch lease is seated at the level the snapshot recorded, not reset"', " );", " // REQ-MSG-6's resolver reads this accessor: blind here is what re-stamped", " // every cross-node user-msg down to a plain msg for a whole cadence.", " assert!(", ' !h2.instances_of("ling").is_empty(),', ' "the WAN user-msg origin resolver can see the row across the restart"', " );", "", " // Hydration seats the lease, it does not bypass it: an equal epoch is", " // still Stale, a strictly-newer one still Updates.", " let stale = h2.apply_feed(", ' "bb22",', " &[RegistryUpdate {", ' subnet: "home".into(),', ' endpoint_id: "ling".into(),', ' instance: identified("bb22", "PEERBOX", "mid-1", 7),', " }],", ' &policy("home", "bb22"),', " );", " assert_eq!(", " stale,", " vec![RegistryApplyVerdict::Merged(MergeOutcome::Stale)],", ' "equal epoch against a HYDRATED row is dropped by the lease"', " );", " let fresh = h2.apply_feed(", ' "bb22",', " &[RegistryUpdate {", ' subnet: "home".into(),', ' endpoint_id: "ling".into(),', ' instance: identified("bb22", "PEERBOX", "mid-1", 8),', " }],", ' &policy("home", "bb22"),', " );", " assert_eq!(", " fresh,", " vec![RegistryApplyVerdict::Merged(MergeOutcome::Updated)],", ' "a strictly-newer epoch supersedes the hydrated row"', " );", " }", "", " // [unit->REQ-REGISTRY-SNAPSHOT-HYDRATE] rows only. The heard map is the", " // gossip-recency half of the M7 D2 hybrid liveness -- a hydrated stamp would", " // render a peer online WITHOUT a probe on the strength of a reading taken", " // before the restart. Observed through the mirror: write_snapshots writes", " // heard.meta only when the in-memory map is non-empty, so a removed file that", " // does not come back is the map being empty.", " #[test]", " fn restart_leaves_the_heard_map_unhydrated() {", " let dir = tempfile::tempdir().unwrap();", ' let snap_dir = dir.path().join("registry");', " let h1 = host(dir.path());", " h1.apply_feed(", ' "bb22",', " &[RegistryUpdate {", ' subnet: "home".into(),', ' endpoint_id: "ling".into(),', ' instance: identified("bb22", "PEERBOX", "mid-1", 3),', " }],", ' &policy("home", "bb22"),', " );", " let heard = RegistryHost::heard_path(&snap_dir);", ' assert!(heard.exists(), "an admitted feed stamps the heard map");', " drop(h1);", "", " let h2 = host(dir.path());", ' assert!(!h2.rows("home", "ling").is_empty(), "rows DID hydrate");', " std::fs::remove_file(&heard).unwrap();", " h2.write_snapshots();", " assert!(", " !heard.exists(),", ' "heard map stays empty across the restart: liveness is re-earned by a probe"', " );", " assert!(", ' RegistryHost::snapshot_path(&snap_dir, "home").exists(),', ' "the same mirror pass DID rewrite the rows -- the absence above is the map, not a skipped write"', " );", " }", "", " // [unit->REQ-REGISTRY-SNAPSHOT-HYDRATE] the reader is total by contract, so", " // an absent dir and an unparseable snapshot hydrate EMPTY rather than failing", " // construction -- a daemon must boot past a corrupt mirror, not die on it.", " #[test]", " fn absent_or_unparseable_snapshot_hydrates_empty_without_failing() {", " let dir = tempfile::tempdir().unwrap();", ' let missing = host(&dir.path().join("no-such-home"));', ' assert!(missing.rows("home", "ling").is_empty());', " assert_eq!(missing.snapshot_write_count(), 0);", "", ' let torn = dir.path().join("torn");', ' std::fs::create_dir_all(torn.join("registry")).unwrap();', ' std::fs::write(torn.join("registry").join("home.json"), b"{not json").unwrap();', " let h = host(&torn);", " assert!(", ' h.rows("home", "ling").is_empty(),', ' "an unparseable snapshot is skipped, not fatal"', " );", " }", "", " // [unit->REQ-REGISTRY-SNAPSHOT-HYDRATE] the breadcrumb: it names the counts,", " // the dir it read them from, and a wall clock (IR-84 -- a line a field reader", " // must order against a restart cannot be ordered without one). The zero case", " // is emitted too: subnets=0 rows=0 is the evidence that the host looked.", " #[test]", " fn hydrate_breadcrumb_names_counts_dir_and_a_clock() {", ' let line = snapshot_hydrate_line("aa11", 2, 5, Path::new("/tmp/reg"), 1_788_000_000_000);', ' assert!(line.starts_with("REGISTRY_SNAPSHOT_HYDRATE:aa11:"), "{line}");', ' assert!(line.contains("subnets=2"), "{line}");', ' assert!(line.contains("rows=5"), "{line}");', ' assert!(line.contains("wall_ms=1788000000000"), "{line}");', ' assert!(line.contains("reg"), "names the dir it read: {line}");', ' let zero = snapshot_hydrate_line("aa11", 0, 0, Path::new("/tmp/reg"), 1);', ' assert!(zero.contains("subnets=0 rows=0"), "the zero case is a reading: {zero}");', " }", "", " // [unit->REQ-REGISTRY-SNAPSHOT-HYDRATE] own rows hydrate too, and cannot", " // shadow the node's own next advertisement: advertise_local stamps a bumped", " // epoch from the DURABLE counter, and the per-(endpoint,node) lease supersedes", " // on strictly-newer. This is the restart path end to end.", " #[test]", " fn own_rows_hydrate_and_the_next_advertisement_supersedes_them() {", " with_home(|home| {", ' let owlery = home.join("owlery");', ' let p = perch::resolve_perch_path_in(&owlery, "self-ag", ParentHint::Infer);', " std::fs::create_dir_all(&p).unwrap();", ' let rec = InfoJson::new("self-ag", "now", std::process::id(), "s", "ready_agent");', " info::write_info(&p, &rec).unwrap();", " let mut subnets = SubnetStore::load();", ' subnets.create_subnet("adv", spt_store::access::Mode::Open).unwrap();', " subnets.save().unwrap();", "", " let h1 = host(home);", " h1.advertise_local(&owlery);", ' let e1 = h1.rows("adv", "self-ag")[0].epoch;', " drop(h1);", "", " // The restart: the own row comes back at the epoch it was left at.", " let h2 = host(home);", ' let seated = h2.rows("adv", "self-ag");', ' assert_eq!(seated.len(), 1, "own row hydrated across the restart");', ' assert_eq!(seated[0].epoch, e1, "at the epoch the snapshot recorded");', "", " h2.advertise_local(&owlery);", ' let e2 = h2.rows("adv", "self-ag")[0].epoch;', " assert!(", " e2 > e1,", ' "the fresh advertisement SUPERSEDES the hydrated own row ({e1} -> {e2}), never loses the lease race to it"', " );", " });", " }", ]) src = src.replace(anchor, anchor + cells, 1) io.open(P, "w", encoding="utf-8", newline="").write(src) b = io.open(P, "rb").read() print("cells added | CR", b.count(b"\r"), "LF", b.count(b"\n"))