diff --git a/crates/spt-store/tests/fixtures/serving_registry_worker.rs b/crates/spt-store/tests/fixtures/serving_registry_worker.rs new file mode 100644 index 00000000..01c3b94b --- /dev/null +++ b/crates/spt-store/tests/fixtures/serving_registry_worker.rs @@ -0,0 +1,105 @@ +//! Test-only serving-registry writer. All IPC is explicit, per-child file +//! barriers under a temporary directory; polling delays never order transactions. +//! Usage: serving_registry_worker + +use std::fs::{self, File, OpenOptions}; +use std::io; +use std::path::Path; +use std::time::{Duration, Instant}; + +use spt_store::serving::{lock_registry_at, ServedKind, ServingRegistry}; + +const COMMAND_BOUND: Duration = Duration::from_secs(120); + +fn announce(barriers: &Path, event: &str) -> io::Result<()> { + fs::write(barriers.join(event), b"") +} + +fn await_command(barriers: &Path, command: &str) -> io::Result<()> { + let deadline = Instant::now() + COMMAND_BOUND; + loop { + if barriers.join(command).try_exists()? { + return Ok(()); + } + if Instant::now() >= deadline { + return Err(io::Error::new(io::ErrorKind::TimedOut, format!("waiting for {command}"))); + } + std::thread::sleep(Duration::from_millis(10)); + } +} + +/// Probe a fresh handle to the documented sentinel, not the atomically replaced +/// registry inode. Only genuine lock contention counts as the exclusion witness. +fn witness_exclusion(registry_path: &Path) -> io::Result<()> { + let mut sentinel = registry_path.as_os_str().to_os_string(); + sentinel.push(".lock"); + let probe = OpenOptions::new().read(true).write(true).open(Path::new(&sentinel))?; + match fs2::FileExt::try_lock_exclusive(&probe) { + Ok(()) => Err(io::Error::other("another process reported HELD but its sentinel was unlocked")), + Err(error) if error.raw_os_error() == fs2::lock_contended_error().raw_os_error() => Ok(()), + Err(error) => Err(error), + } +} + +/// Keep the witness and pre-acquire barrier inside this call: moving a load +/// above this call deterministically captures the holder's unpublished state, +/// rather than depending on how quickly the parent schedules the next command. +fn acquire(registry_path: &Path, barriers: &Path, wait: bool) -> io::Result { + if wait { + witness_exclusion(registry_path)?; + announce(barriers, "blocked")?; + // Re-probe after the holder has atomically replaced the registry, while + // it still owns its guard. A lock on the replaced inode is insufficient. + await_command(barriers, "probe-published")?; + witness_exclusion(registry_path)?; + announce(barriers, "blocked-after-publish")?; + await_command(barriers, "acquire")?; + } + lock_registry_at(registry_path) +} + +fn run() -> io::Result<()> { + let args: Vec<_> = std::env::args_os().collect(); + if args.len() != 6 { + return Err(io::Error::new(io::ErrorKind::InvalidInput, "expected mode, role, registry, barriers, source")); + } + let mode = args[1].to_str().ok_or_else(|| io::Error::other("invalid mode"))?; + if mode != "reap" && mode != "register" { + return Err(io::Error::new(io::ErrorKind::InvalidInput, "expected reap or register")); + } + let wait = match args[2].to_str() { + Some("hold") => false, + Some("wait") => true, + _ => return Err(io::Error::new(io::ErrorKind::InvalidInput, "expected hold or wait")), + }; + let registry_path = Path::new(&args[3]); + let barriers = Path::new(&args[4]); + let source = Path::new(&args[5]); + + let guard = acquire(registry_path, barriers, wait)?; + let mut registry = ServingRegistry::load_at(registry_path)?; + announce(barriers, "loaded")?; + await_command(barriers, "mutate")?; + if mode == "reap" { + for entry in registry.reap_expired(1_000) { + if entry.kind == ServedKind::Attachment { + fs::remove_file(&entry.path)?; + } + } + } else { + registry.add_reference(source, Some("report.txt"), None, 1_000)?; + } + registry.save_at(registry_path)?; + announce(barriers, "saved")?; + // The guard deliberately spans publication and the explicit release barrier. + await_command(barriers, "release")?; + drop(guard); + Ok(()) +} + +fn main() { + if let Err(error) = run() { + eprintln!("serving registry worker: {error}"); + std::process::exit(1); + } +} diff --git a/crates/spt-store/tests/serving_registry_two_process_int.rs b/crates/spt-store/tests/serving_registry_two_process_int.rs new file mode 100644 index 00000000..111a5e8d --- /dev/null +++ b/crates/spt-store/tests/serving_registry_two_process_int.rs @@ -0,0 +1,212 @@ +//! Cross-process storage-contract proof for releases#308. These children use the +//! public registry API, not the broker/brain entry points; daemon callsite coverage +//! belongs to their own tests. Fixed logical time makes expiry deterministic. + +use std::fs; +use std::io; +use std::path::{Path, PathBuf}; +use std::process::{Child, Command, ExitStatus, Stdio}; +use std::time::{Duration, Instant}; + +use spt_store::serving::{lock_registry_at, ServingRegistry}; + +const CHILD_BOUND: Duration = Duration::from_secs(120); +const REAP_BOUND: Duration = Duration::from_secs(10); + +/// Own exactly the child we spawned, before any fallible readiness/assertion +/// work. No image-name sweeping, inherited pipes, or descendant processes. +struct RegistryWorker { + child: Child, + barriers: PathBuf, +} + +impl RegistryWorker { + fn spawn(mode: &str, role: &str, registry: &Path, barriers: PathBuf, source: &Path) -> Self { + fs::create_dir_all(&barriers).expect("worker barrier directory"); + let child = Command::new(env!("CARGO_BIN_EXE_serving_registry_worker")) + .args([mode, role]) + .arg(registry) + .arg(&barriers) + .arg(source) + .stdin(Stdio::null()) + .stdout(Stdio::null()) + .stderr(Stdio::inherit()) + .spawn() + .expect("spawn serving registry worker"); + Self { child, barriers } + } + + fn command(&self, command: &str) { + fs::write(self.barriers.join(command), b"").expect("publish worker command"); + } + + fn await_event(&mut self, event: &str) { + let deadline = Instant::now() + CHILD_BOUND; + loop { + if self.barriers.join(event).try_exists().expect("read worker event") { + return; + } + if let Some(status) = self.child.try_wait().expect("poll worker") { + panic!("worker exited {status} before {event}"); + } + assert!(Instant::now() < deadline, "worker did not announce {event}"); + // Polling only: the event, never elapsed time, establishes ordering. + std::thread::sleep(Duration::from_millis(10)); + } + } + + fn wait_exit(&mut self, bound: Duration) -> io::Result { + let deadline = Instant::now() + bound; + loop { + if let Some(status) = self.child.try_wait()? { + return Ok(status); + } + if Instant::now() >= deadline { + return Err(io::Error::new(io::ErrorKind::TimedOut, "worker did not exit")); + } + std::thread::sleep(Duration::from_millis(10)); + } + } + + fn finish(&mut self) { + let status = self.wait_exit(CHILD_BOUND).expect("worker completion deadline"); + assert!(status.success(), "worker failed: {status}"); + } + + fn kill_and_reap(&mut self) -> io::Result { + if let Some(status) = self.child.try_wait()? { + return Ok(status); + } + if let Err(error) = self.child.kill() { + // A natural exit may race the kill; do not report that as a leak. + if let Some(status) = self.child.try_wait()? { + return Ok(status); + } + return Err(error); + } + self.wait_exit(REAP_BOUND) + } +} + +impl Drop for RegistryWorker { + fn drop(&mut self) { + if let Err(error) = self.kill_and_reap() { + // Cleanup must not double-panic during assertion unwinding. A real + // OS kill/reap failure is reported rather than silently swallowed. + eprintln!("failed to reap registry worker {}: {error}", self.child.id()); + } + } +} + +fn overlap(holder_mode: &str, abrupt_exit: bool) { + let temp = tempfile::tempdir().expect("isolated registry home"); + let home = temp.path(); + // Cover the canonical name and an extension-bearing explicit path. The + // sentinel must append .lock, not replace an existing extension. + let registry_path = home.join("serve").join(if holder_mode == "reap" { "registry" } else { "registry.json" }); + let expired_path = home.join("expired.snapshot"); + let retired_path = home.join("retired.txt"); + let registered_path = home.join("new.txt"); + let third_path = home.join("third.txt"); + for path in [&expired_path, &retired_path, ®istered_path, &third_path] { + fs::write(path, b"fixture bytes").expect("fixture source"); + } + let expired; + let retired; + { + let _guard = lock_registry_at(®istry_path).expect("seed guard"); + let mut seed = ServingRegistry::load_at(®istry_path).expect("empty registry"); + expired = seed.add_attachment(&expired_path, "report.txt", None, 10, None, 100) + .expect("expiring bare name"); + retired = seed.add_reference(&retired_path, Some("report.txt"), None, 100) + .expect("historical suffixed name"); + assert_eq!(expired.served_name, "report.txt"); + assert_eq!(retired.served_name, "report~1.txt"); + seed.remove(&retired.id).expect("retire suffix without forgetting owner"); + seed.save_at(®istry_path).expect("seed publication"); + } + + let waiter_mode = if holder_mode == "reap" { "register" } else { "reap" }; + let mut holder = RegistryWorker::spawn(holder_mode, "hold", ®istry_path, home.join("holder"), ®istered_path); + holder.await_event("loaded"); + let mut waiter = RegistryWorker::spawn(waiter_mode, "wait", ®istry_path, home.join("waiter"), ®istered_path); + // A real fs2 try-lock must fail with the platform's contention error while + // the other PROCESS confirms its loaded transaction owns the guard. + waiter.await_event("blocked"); + holder.command("mutate"); + holder.await_event("saved"); + waiter.command("probe-published"); + waiter.await_event("blocked-after-publish"); + waiter.command("acquire"); + if abrupt_exit { + assert!(holder.child.try_wait().expect("holder still alive").is_none(), "holder exited before the abrupt-exit command"); + holder.child.kill().expect("force-kill lock holder"); + let status = holder.wait_exit(REAP_BOUND).expect("reap killed lock holder"); + assert!(!status.success(), "holder must exit abruptly without releasing its Rust guard"); + } else { + holder.command("release"); + holder.finish(); + } + waiter.await_event("loaded"); + waiter.command("mutate"); + waiter.await_event("saved"); + waiter.command("release"); + waiter.finish(); + + // Both orders matter: a stale reaper erases the acknowledged registration; + // a stale registrar resurrects the expired entry after its snapshot is gone. + let final_registry = ServingRegistry::load_at(®istry_path).expect("read durable final state"); + assert!(final_registry.get(&expired.id).is_none(), "expired entry was resurrected by a stale writer"); + assert!(!expired_path.try_exists().expect("snapshot existence"), "reap must remove expired snapshot bytes"); + let registration = final_registry.get("report~2.txt").expect("acknowledged registration survives later publication"); + assert_eq!(registration.path, registered_path); + assert_eq!(final_registry.entries().count(), 1, "only the new registration remains exposed"); + let registration_id = registration.id.clone(); + + // Observe history through future assignments, not private JSON fields. A + // save/reload after retirement must retain BOTH ownership and next suffix. + { + let _guard = lock_registry_at(®istry_path).expect("retirement guard"); + let mut registry = ServingRegistry::load_at(®istry_path).expect("fresh retirement load"); + registry.remove(®istration_id).expect("retire acknowledged registration"); + registry.save_at(®istry_path).expect("persist retirement"); + } + { + let _guard = lock_registry_at(®istry_path).expect("history guard"); + let mut registry = ServingRegistry::load_at(®istry_path).expect("reload allocation history"); + let reclaimed = registry.add_reference(®istered_path, Some("report.txt"), None, 2_000) + .expect("reclaim recent owner"); + assert_eq!(reclaimed.served_name, "report~2.txt", "retiring must preserve the new owner's name"); + let old_owner = registry.add_reference(&retired_path, Some("report.txt"), None, 2_000) + .expect("reclaim historical owner"); + assert_eq!(old_owner.served_name, "report~1.txt", "reap must preserve previously retired ownership"); + let next = registry.add_reference(&third_path, Some("report.txt"), None, 2_000) + .expect("new source gets next suffix"); + assert_eq!(next.served_name, "report~3.txt", "neither expired nor retired names may be reassigned"); + registry.save_at(®istry_path).expect("persist continued history"); + } + let published = ServingRegistry::load_at(®istry_path).expect("read continued assignments"); + assert_eq!(published.get("report~1.txt").expect("old owner").path, retired_path); + assert_eq!(published.get("report~2.txt").expect("recent owner").path, registered_path); + assert_eq!(published.get("report~3.txt").expect("new owner").path, third_path); +} + +// [int->REQ-HAZARD-SERVE-REGISTRY-LOST-UPDATE] +// [int->REQ-WEB-SERVING-REGISTRY] [int->REQ-WEB-ATTACHMENT-PULL] +#[test] +fn loaded_reaper_serializes_registration_without_resurrecting_expiry() { + overlap("reap", false); +} + +// [int->REQ-HAZARD-SERVE-REGISTRY-LOST-UPDATE] +// [int->REQ-WEB-SERVING-REGISTRY] [int->REQ-WEB-ATTACHMENT-PULL] +#[test] +fn later_reap_preserves_a_registration_published_while_it_was_waiting() { + overlap("register", false); +} + +// [int->REQ-HAZARD-SERVE-REGISTRY-LOST-UPDATE] +#[test] +fn abrupt_holder_exit_releases_the_sentinel_for_the_waiting_reaper() { + overlap("register", true); +} diff --git a/crates/spt/tests/webserve_attachment_e2e.rs b/crates/spt/tests/webserve_attachment_e2e.rs index c1516a82..8c235cb7 100644 --- a/crates/spt/tests/webserve_attachment_e2e.rs +++ b/crates/spt/tests/webserve_attachment_e2e.rs @@ -661,43 +661,30 @@ registry: {}", // // `scope_entry` also sets `audience` unconditionally from its argument, so // passing None CLEARS any audience. reap-me.md has none, which makes that a - // no-op HERE — but do not lift this loop onto an audienced entry without + // no-op HERE — but do not lift this mutation onto an audienced entry without // passing its audience back in, or the scope goes with the ttl. // - // The loop re-reads because the daemon is the registry's other writer. Be - // precise about what that buys: it detects the entry being ABSENT, not a - // clobber — a daemon save landing between this load and this save would be - // overwritten by this snapshot and nothing here would notice. That hazard is - // idle in this arm by arithmetic rather than by vigilance: every other live - // entry carries 30 d or 3600 s, so the only concurrent writer has nothing to - // retire and nothing to write, and this snapshot equals its state. - let mut retired = false; - for attempt in 0..5 { + // This fixture is another PROCESS writing the live daemon's registry. + // Hold its shared guard across the fresh read and publication, not a retry + // after a stale write. An already-retired entry needs no further mutation. + // [int->REQ-HAZARD-SERVE-REGISTRY-LOST-UPDATE] + { + let _writer = spt_store::serving::lock_registry_at(®istry_path).unwrap(); let mut registry = spt_store::serving::ServingRegistry::load_at(®istry_path).unwrap(); if registry .scope_entry("reap-me.md", Some(1), None, None) .is_some() { registry.save_at(®istry_path).unwrap(); - retired = true; - break; - } - // Gone already means the daemon retired it between our read and now, - // which is the outcome this loop is trying to cause — not a failure. - if snapshot_of("reap-me.md").is_none() { - retired = true; - break; } - assert!(attempt < 4, "could not backdate the doomed entry's ttl"); } - assert!(retired, "the doomed entry was given an already-expired lifetime"); - // A reference entry, already expired, planted for the guard. Re-read after - // planting: the daemon is the registry's other writer and its own tick could - // land between this load and this save. + // Plant the already-expired reference under the same transaction guard. + // Release it before waiting for the brain to reap either entry. let kept_source = work.join("kept-by-the-guard.md"); std::fs::write(&kept_source, b"the user's own file").unwrap(); - for attempt in 0..5 { + { + let _writer = spt_store::serving::lock_registry_at(®istry_path).unwrap(); let mut registry = spt_store::serving::ServingRegistry::load_at(®istry_path).unwrap(); let planted = registry .add_reference(&kept_source, Some("kept-by-the-guard.md"), None, 1) @@ -706,10 +693,6 @@ registry: {}", .scope_entry(&planted.id, Some(1), None, None) .expect("give it an already-expired ttl"); registry.save_at(®istry_path).unwrap(); - if snapshot_of("kept-by-the-guard.md").is_some() { - break; - } - assert!(attempt < 4, "the planted reference entry kept being clobbered"); } // One pulse is LIVE_RECONCILE_INTERVAL_MS (5 s); wait past two. diff --git a/crates/spt/tests/webserve_cross_node_e2e.rs b/crates/spt/tests/webserve_cross_node_e2e.rs index a4a5a48b..be23c819 100644 --- a/crates/spt/tests/webserve_cross_node_e2e.rs +++ b/crates/spt/tests/webserve_cross_node_e2e.rs @@ -659,11 +659,15 @@ fn a_peers_url_is_served_by_its_owner_through_the_local_listener() { .expect("seed B's view of who hosts the audience"); let registry_path = spt_store::perch::serving_registry_file_in(home_b.path()); - let mut serving = spt_store::serving::ServingRegistry::load_at(®istry_path).unwrap(); - serving - .scope_entry("report.md", None, Some(AUDIENCE_ENDPOINT), None) - .expect("the served entry is there to narrow"); - serving.save_at(®istry_path).expect("persist the narrowing"); + // [int->REQ-HAZARD-SERVE-REGISTRY-LOST-UPDATE] + { + let _writer = spt_store::serving::lock_registry_at(®istry_path).unwrap(); + let mut serving = spt_store::serving::ServingRegistry::load_at(®istry_path).unwrap(); + serving + .scope_entry("report.md", None, Some(AUDIENCE_ENDPOINT), None) + .expect("the served entry is there to narrow"); + serving.save_at(®istry_path).expect("persist the narrowing"); + } let (status, _, body) = http(port_a, "GET", &url, &[]); assert_eq!( @@ -673,11 +677,14 @@ fn a_peers_url_is_served_by_its_owner_through_the_local_listener() { ); // Now the SAME entry addressed to an endpoint B cannot place on A. - let mut serving = spt_store::serving::ServingRegistry::load_at(®istry_path).unwrap(); - serving - .scope_entry("report.md", None, Some("somebody-else"), None) - .expect("the entry is still there"); - serving.save_at(®istry_path).expect("persist the second narrowing"); + { + let _writer = spt_store::serving::lock_registry_at(®istry_path).unwrap(); + let mut serving = spt_store::serving::ServingRegistry::load_at(®istry_path).unwrap(); + serving + .scope_entry("report.md", None, Some("somebody-else"), None) + .expect("the entry is still there"); + serving.save_at(®istry_path).expect("persist the second narrowing"); + } let (status, _, body) = http(port_a, "GET", &url, &[]); let body = String::from_utf8(body).unwrap(); @@ -757,9 +764,12 @@ fn a_peers_url_is_served_by_its_owner_through_the_local_listener() { ); // Restore the open entry for the arms that follow. - let mut serving = spt_store::serving::ServingRegistry::load_at(®istry_path).unwrap(); - serving.scope_entry("report.md", None, None, None).expect("clear the audience"); - serving.save_at(®istry_path).expect("persist the restore"); + { + let _writer = spt_store::serving::lock_registry_at(®istry_path).unwrap(); + let mut serving = spt_store::serving::ServingRegistry::load_at(®istry_path).unwrap(); + serving.scope_entry("report.md", None, None, None).expect("clear the audience"); + serving.save_at(®istry_path).expect("persist the restore"); + } let (status, _, _) = http(port_a, "GET", &url, &[]); assert_eq!(status, 200, "audience cleared, the fetch is open again"); }