diff --git a/crates/spt-daemon/src/livehost.rs b/crates/spt-daemon/src/livehost.rs index 151c5e2e..f2cda242 100644 --- a/crates/spt-daemon/src/livehost.rs +++ b/crates/spt-daemon/src/livehost.rs @@ -1278,62 +1278,60 @@ fn fresh_live_bin_old(live_bin: &Path) -> PathBuf { /// containment check is belt-and-braces on that same point: a hand-edited or /// corrupted registry must not be able to aim a delete at an arbitrary path. // [impl->REQ-WEB-ATTACHMENT-PULL] +// [impl->REQ-HAZARD-SERVE-REGISTRY-LOST-UPDATE] fn reap_expired_attachments() { - // ONE WRITER. The whole load -> reap -> save runs under the registry write - // lock every other writer in this process holds, because this pass is a - // read-modify-write on a file `serve add` also writes: an entry registered - // between our load and our save would be clobbered by our stale snapshot - // and silently lost, on a 5s tick, forever. - let held = crate::servehost::with_registry_write(|| { - let registry_path = perch::serving_registry_file(); - if !registry_path.exists() { - return; // nothing has ever been served on this node + let registry_path = perch::serving_registry_file(); + if !registry_path.exists() { + return; // nothing has ever been served on this node + } + // The broker is a DIFFERENT process. Hold the shared sentinel from before + // loading until snapshot cleanup and publication have both finished. + let _writer = match spt_store::serving::lock_registry_at(®istry_path) { + Ok(writer) => writer, + Err(err) => { + eprintln!("SERVE_REAP_LOCK_FAIL: {err}"); + return; // never reap outside the lock } - let mut registry = match spt_store::serving::ServingRegistry::load_at(®istry_path) { - Ok(registry) => registry, - Err(err) => { - eprintln!("SERVE_REAP_LOAD_FAIL: {err}"); - return; - } - }; - let reaped = registry.reap_expired(now_ms()); - if reaped.is_empty() { + }; + let mut registry = match spt_store::serving::ServingRegistry::load_at(®istry_path) { + Ok(registry) => registry, + Err(err) => { + eprintln!("SERVE_REAP_LOAD_FAIL: {err}"); return; } - let snapshots = perch::serve_snapshots_dir(); - let mut bytes_removed: u64 = 0; - for entry in &reaped { - if entry.kind != spt_store::serving::ServedKind::Attachment { - continue; // a reference entry's path is the user's own file - } - if !entry.path.starts_with(&snapshots) { - eprintln!("SERVE_REAP_SKIP_FOREIGN:{}", entry.id); - continue; - } - let len = std::fs::metadata(&entry.path).map(|meta| meta.len()).unwrap_or(0); - match std::fs::remove_file(&entry.path) { - Ok(()) => bytes_removed = bytes_removed.saturating_add(len), - Err(err) if err.kind() == std::io::ErrorKind::NotFound => {} - Err(err) => eprintln!("SERVE_REAP_UNLINK_FAIL:{}: {err}", entry.id), - } + }; + let reaped = registry.reap_expired(now_ms()); + if reaped.is_empty() { + return; + } + let snapshots = perch::serve_snapshots_dir(); + let mut bytes_removed: u64 = 0; + for entry in &reaped { + if entry.kind != spt_store::serving::ServedKind::Attachment { + continue; // a reference entry's path is the user's own file } - if let Err(err) = registry.save_at(®istry_path) { - // The bytes are gone and the entries are not. Say so rather than - // reporting a reap that only half happened; the next tick retries. - eprintln!("SERVE_REAP_SAVE_FAIL: {err}"); - return; + if !entry.path.starts_with(&snapshots) { + eprintln!("SERVE_REAP_SKIP_FOREIGN:{}", entry.id); + continue; } - eprintln!( - "SERVE_REAP: reaped {} expired entries, freed {} bytes", - reaped.len(), - bytes_removed - ); - }); - if let Err(err) = held { - // A poisoned writer lock is not something a pulse may paper over: say - // it every tick rather than reaping outside the lock. - eprintln!("SERVE_REAP_LOCK_FAIL: {err}"); + let len = std::fs::metadata(&entry.path).map(|meta| meta.len()).unwrap_or(0); + match std::fs::remove_file(&entry.path) { + Ok(()) => bytes_removed = bytes_removed.saturating_add(len), + Err(err) if err.kind() == std::io::ErrorKind::NotFound => {} + Err(err) => eprintln!("SERVE_REAP_UNLINK_FAIL:{}: {err}", entry.id), + } + } + if let Err(err) = registry.save_at(®istry_path) { + // The bytes are gone and the entries are not. Say so rather than + // reporting a reap that only half happened; the next tick retries. + eprintln!("SERVE_REAP_SAVE_FAIL: {err}"); + return; } + eprintln!( + "SERVE_REAP: reaped {} expired entries, freed {} bytes", + reaped.len(), + bytes_removed + ); } /// Spawn the brain's live host: one thread sweeping [`reconcile_once`] at boot diff --git a/crates/spt-daemon/src/servehost.rs b/crates/spt-daemon/src/servehost.rs index 377f4a64..e84c2e69 100644 --- a/crates/spt-daemon/src/servehost.rs +++ b/crates/spt-daemon/src/servehost.rs @@ -3,7 +3,6 @@ use std::io; use std::path::{Path, PathBuf}; -use std::sync::Mutex; use serde::{Deserialize, Serialize}; use spt_store::serving::{ServedEntry, ServingRegistry}; @@ -15,10 +14,6 @@ use crate::transport::{send_hello, LocalSocketTransport}; pub const KIND_SERVE_REQUEST: &str = "serve_request"; pub const KIND_SERVE_RESULT: &str = "serve_result"; -// Serializes load/change/publish across the control channel's connection threads. -// The HTTP reader observes an atomic persisted snapshot, never this lock. -static REGISTRY_WRITE: Mutex<()> = Mutex::new(()); - /// An explicit local registry operation; no HTTP request can mutate exposure. // [impl->REQ-WEB-SERVING-REGISTRY] #[derive(Debug, Clone, Serialize, Deserialize)] @@ -132,29 +127,10 @@ pub enum ServeResult { LanRefused { code: String }, } -/// Run `f` holding the ONE registry write lock, so a whole read-modify-write -/// pass is atomic against every other writer in this process. -/// -/// The daemon has a second registry writer: the live host's TTL reaper, which -/// does load -> retain -> save on its own pulse. Without this it raced every -/// `serve add` — an entry registered between the reaper's load and its save was -/// clobbered by the reaper's stale snapshot and silently lost, on a 5s tick, -/// forever. The lock was already here; only this door into it was missing. -/// -/// Callers must not re-enter `apply_at` from inside `f`: the mutex is not -/// reentrant, and doing so would deadlock the pulse thread. -// [impl->REQ-WEB-SERVING-REGISTRY] -// [impl->REQ-WEB-ATTACHMENT-PULL] -pub(crate) fn with_registry_write(f: impl FnOnce() -> R) -> io::Result { - let _writer = REGISTRY_WRITE - .lock() - .map_err(|_| io::Error::other("SERVE_REGISTRY_POISONED: writer state is unproven"))?; - Ok(f()) -} - /// Apply a request at an explicit home. Publication completes before success /// is returned, and a failed publication cannot advance an in-memory counter. // [impl->REQ-WEB-SERVING-REGISTRY] +// [impl->REQ-HAZARD-SERVE-REGISTRY-LOST-UPDATE] pub(crate) fn apply_at(home: &Path, request: ServeRequest, now_ms: u64) -> io::Result { // [impl->REQ-WEB-URL-BOUND-PORT] if matches!(&request, ServeRequest::DocsStatus) { @@ -168,10 +144,8 @@ pub(crate) fn apply_at(home: &Path, request: ServeRequest, now_ms: u64) -> io::R if let Some(result) = lan_request(home, &request) { return Ok(result); } - let _writer = REGISTRY_WRITE - .lock() - .map_err(|_| io::Error::other("SERVE_REGISTRY_POISONED: writer state is unproven"))?; let path = spt_store::perch::serving_registry_file_in(home); + let _writer = spt_store::serving::lock_registry_at(&path)?; let mut registry = ServingRegistry::load_at(&path).map_err(|error| { if matches!(&request, ServeRequest::AddInputReference { .. }) { io::Error::new(error.kind(), format!("INPUT_PATH_REGISTRY: {error}")) @@ -658,7 +632,26 @@ mod tests { } } + // [unit->REQ-HAZARD-SERVE-REGISTRY-LOST-UPDATE] + #[test] + fn unavailable_registry_lock_cannot_publish_an_add() { + let tmp = tempfile::tempdir().unwrap(); + let home = tmp.path().join("home"); + let source = tmp.path().join("report.md"); + std::fs::write(&source, "must not be registered").unwrap(); + let registry_path = spt_store::perch::serving_registry_file_in(&home); + // A directory at the sentinel path deterministically refuses acquisition + // on both supported platforms, without permission or timing assumptions. + let mut lock_path = registry_path.as_os_str().to_os_string(); + lock_path.push(".lock"); + std::fs::create_dir_all(PathBuf::from(lock_path)).unwrap(); + + assert!(apply_at(&home, add(&source), 1).is_err()); + assert!(!registry_path.exists(), "a failed lock must not permit publication"); + } + // [unit->REQ-WEB-SERVING-REGISTRY] + // [unit->REQ-HAZARD-SERVE-REGISTRY-LOST-UPDATE] #[test] fn concurrent_requests_publish_distinct_entries_without_lost_updates() { let tmp = tempfile::tempdir().unwrap(); @@ -686,75 +679,6 @@ mod tests { assert_eq!(names, ["report.md", "report~1.md"].into_iter().collect()); } - /// A reap-shaped read-modify-write cannot clobber a concurrent `serve add`. - /// - /// The TTL reaper is the daemon's SECOND registry writer, and it runs on a - /// pulse: load, retain, save. Outside the write lock, an entry registered - /// between its load and its save is overwritten by its stale snapshot and - /// silently lost — no error, no log line, once every five seconds forever. - /// - /// The reaping thread holds the lock across a deliberate pause, so the - /// adding thread MUST be waiting on it. Without the lock the add lands in - /// that window and the reaper's save drops it; with the lock the two passes - /// serialize and both outcomes survive. No platform gate: the pause is a - /// delay, not a scheduling assumption, and the assertion only holds one way. - // [unit->REQ-WEB-ATTACHMENT-PULL] - #[test] - fn a_reap_shaped_pass_cannot_clobber_a_concurrent_add() { - let tmp = tempfile::tempdir().unwrap(); - let home = tmp.path().join("home"); - let registry_path = spt_store::perch::serving_registry_file_in(&home); - let doomed_source = tmp.path().join("doomed.md"); - let added_source = tmp.path().join("added.md"); - std::fs::write(&doomed_source, "doomed").unwrap(); - std::fs::write(&added_source, "added").unwrap(); - - // One entry already registered, with an already-expired lifetime. - let ServeResult::Entry { entry } = apply_at(&home, add(&doomed_source), 1).unwrap() else { - panic!("seed add did not return an entry"); - }; - with_registry_write(|| { - let mut registry = ServingRegistry::load_at(®istry_path).unwrap(); - registry.scope_entry(&entry.id, Some(1), None, None).expect("give it a spent ttl"); - registry.save_at(®istry_path).unwrap(); - }) - .unwrap(); - - std::thread::scope(|scope| { - let reaper = scope.spawn(|| { - with_registry_write(|| { - let mut registry = ServingRegistry::load_at(®istry_path).unwrap(); - let reaped = registry.reap_expired(u64::MAX); - assert_eq!(reaped.len(), 1, "the spent entry is what expired"); - // Held ACROSS the pause: this is the window the add must not - // be able to slip into. - std::thread::sleep(std::time::Duration::from_millis(300)); - registry.save_at(®istry_path).unwrap(); - }) - .unwrap(); - }); - // Give the reaper the lock first, then race it. - std::thread::sleep(std::time::Duration::from_millis(50)); - let adder = scope.spawn(|| apply_at(&home, add(&added_source), 2).unwrap()); - reaper.join().unwrap(); - adder.join().unwrap(); - }); - - let names: Vec = ServingRegistry::load_at(®istry_path) - .unwrap() - .entries() - .map(|e| e.served_name.clone()) - .collect(); - assert!( - names.iter().any(|n| n == "added.md"), - "the concurrent add SURVIVED the reap: {names:?}" - ); - assert!( - !names.iter().any(|n| n == "doomed.md"), - "and the expired entry was still reaped: {names:?}" - ); - } - /// `AddScoped` is the FILE_ACCESS_HELPER's registration, and it had no cell /// at all — neither for the file case nor this one. A quoted path can name a /// DIRECTORY, and the helper must register it as one: the two halves (add a diff --git a/crates/spt-store/Cargo.toml b/crates/spt-store/Cargo.toml index 104b8321..d2b7fbe0 100644 --- a/crates/spt-store/Cargo.toml +++ b/crates/spt-store/Cargo.toml @@ -16,6 +16,11 @@ description = "spt-core persistence layer: atomic writes, spool, perch layout, r name = "wtlock_worker" path = "tests/fixtures/wtlock_worker.rs" +# Test-only second-process fixture for the serving-registry transaction lock. +[[bin]] +name = "serving_registry_worker" +path = "tests/fixtures/serving_registry_worker.rs" + [dependencies] fs2 = "0.4" # The ONE string-trigger match rule (`matchrule`), shared by monic triggers and diff --git a/crates/spt-store/src/serving.rs b/crates/spt-store/src/serving.rs index 52317a20..a9e7c147 100644 --- a/crates/spt-store/src/serving.rs +++ b/crates/spt-store/src/serving.rs @@ -3,7 +3,8 @@ //! Schema 1 stores live entries, per-family suffix counters, and the absolute //! path and kind each name last identified. Removal changes only live entries; //! the same path and kind may reclaim their names (ADR-0057 Amendment 2). -//! The daemon is the single writer; corruption is never an empty store. +//! Broker mutations and brain reaping share a cross-process writer guard; +//! corruption is never an empty store. // [impl->REQ-WEB-SERVING-REGISTRY] use std::collections::{BTreeMap, BTreeSet}; @@ -132,6 +133,36 @@ pub struct ServingRegistry { name_owners: BTreeMap, } +/// Serialize a registry transaction across threads and processes. +/// +/// Acquire BEFORE loading and keep the returned file alive through mutation, +/// snapshot cleanup/rollback, and explicit publication. This does not load or +/// save anything itself, so no-op transactions need not publish a registry. +/// +/// The stable sibling `registry.lock` is never renamed or removed: locking the +/// registry itself would stop coordinating writers after atomic replacement. +/// Closing the returned file (including process exit) releases the lock. Do not +/// acquire it again while holding it; independently opened handles contend. +// [impl->REQ-HAZARD-SERVE-REGISTRY-LOST-UPDATE] +// [impl->REQ-WEB-SERVING-REGISTRY] +// [impl->REQ-WEB-ATTACHMENT-PULL] +pub fn lock_registry_at(registry_path: &Path) -> io::Result { + use fs2::FileExt; + + if let Some(parent) = registry_path.parent().filter(|p| !p.as_os_str().is_empty()) { + fs::create_dir_all(parent)?; + } + let mut lock_path = registry_path.as_os_str().to_os_string(); + lock_path.push(".lock"); + let lock = fs::OpenOptions::new() + .create(true) + .write(true) + .truncate(false) + .open(PathBuf::from(lock_path))?; + lock.lock_exclusive()?; + Ok(lock) +} + impl ServingRegistry { /// Missing is empty. Unreadable, malformed, or inconsistent history is an /// error: never recover by resetting the allocator and reusing stale URLs. @@ -161,7 +192,8 @@ impl ServingRegistry { } /// Publish a durable old-or-new snapshot, creating its parent directory. - /// The caller must serialize mutations and retain the previous state on error. + /// A concurrent writer must hold [`lock_registry_at`] from BEFORE its load + /// through this publication; locking only this save cannot repair stale state. pub fn save_at(&self, registry_path: &Path) -> io::Result<()> { self.validate()?; let bytes = serde_json::to_vec_pretty(self).map_err(invalid_data)?;