diff --git a/crates/feature-storage/src/identity_tests.rs b/crates/feature-storage/src/identity_tests.rs index b178a0d28..dac0cfc82 100644 --- a/crates/feature-storage/src/identity_tests.rs +++ b/crates/feature-storage/src/identity_tests.rs @@ -315,6 +315,9 @@ fn reserved_adoption_reconciles_at_each_real_identity_commit_boundary() { let lifecycle = owner::Lifecycle { root: temp.path().into(), scope: SCOPE.into(), + established: Arc::default(), + #[cfg(feature = "test-support")] + observation: Arc::default(), }; let expected = { let _lease = lifecycle.lock(true).unwrap(); diff --git a/crates/feature-storage/src/lib.rs b/crates/feature-storage/src/lib.rs index 64d513fdd..c8e058915 100644 --- a/crates/feature-storage/src/lib.rs +++ b/crates/feature-storage/src/lib.rs @@ -151,6 +151,11 @@ impl FeatureRegistration { } } +#[cfg(feature = "test-support")] +mod lifecycle_observation; +#[cfg(feature = "test-support")] +pub use lifecycle_observation::LifecycleObservation; + #[cfg(feature = "test-support")] #[path = "open_observation.rs"] mod open_observation; @@ -194,6 +199,9 @@ struct FeatureStorageInner { id: Uuid, root: PathBuf, state: Mutex, + lifecycle_established: Arc, + #[cfg(feature = "test-support")] + lifecycle_observation: Arc, } struct FeatureStorageState { @@ -251,10 +259,20 @@ impl FeatureStorage { maintenance: HashSet::new(), deletion_fences: HashSet::new(), }), + lifecycle_established: Arc::default(), + #[cfg(feature = "test-support")] + lifecycle_observation: Arc::default(), }), } } + /// Read-only cumulative lifecycle diagnostics shared by this manager's Host + /// and Feature handles, independent of per-registration open observations. + #[cfg(feature = "test-support")] + pub fn lifecycle_observation(&self) -> LifecycleObservation { + self.inner.lifecycle_observation.snapshot() + } + pub fn for_server_database(database_path: impl AsRef) -> Result { let database_path = database_path.as_ref(); let parent = database_path.parent().ok_or_else(|| { @@ -359,6 +377,9 @@ impl FeatureStorage { accepting_operations: AtomicBool::new(true), connection: Mutex::new(Some(connection)), root: self.inner.root.clone(), + lifecycle_established: self.inner.lifecycle_established.clone(), + #[cfg(feature = "test-support")] + lifecycle_observation: self.inner.lifecycle_observation.clone(), path: path.into(), binding, }), @@ -471,6 +492,9 @@ impl FeatureStorage { owner::Lifecycle { root: self.inner.root.clone(), scope: scope_id.into(), + established: self.inner.lifecycle_established.clone(), + #[cfg(feature = "test-support")] + observation: self.inner.lifecycle_observation.clone(), } } @@ -574,6 +598,9 @@ impl FeatureStorage { accepting_operations: AtomicBool::new(true), connection: Mutex::new(Some(connection)), root: self.inner.root.clone(), + lifecycle_established: self.inner.lifecycle_established.clone(), + #[cfg(feature = "test-support")] + lifecycle_observation: self.inner.lifecycle_observation.clone(), path, binding, }); @@ -897,6 +924,13 @@ pub struct ScopedFeatureStorage { pub type WorkspaceFeatureStorage = ScopedFeatureStorage; impl ScopedFeatureStorage { + /// Manager-wide lifecycle totals, including other scopes and retained handles. + /// This scope handle neither filters the observation nor changes storage I/O. + #[cfg(feature = "test-support")] + pub fn lifecycle_observation(&self) -> LifecycleObservation { + self.manager.lifecycle_observation() + } + pub fn scope_id(&self) -> &str { &self.scope_id } @@ -1142,6 +1176,9 @@ struct FeatureDatabaseInner { accepting_operations: AtomicBool, connection: Mutex>, root: PathBuf, + lifecycle_established: Arc, + #[cfg(feature = "test-support")] + lifecycle_observation: Arc, path: PathBuf, binding: owner::Binding, } @@ -1175,6 +1212,9 @@ impl FeatureDatabase { let lease = owner::Lifecycle { root: self.inner.root.clone(), scope: self.inner.scope_id.clone(), + established: self.inner.lifecycle_established.clone(), + #[cfg(feature = "test-support")] + observation: self.inner.lifecycle_observation.clone(), } .lock(false)?; self.inner.ensure_accepting()?; @@ -1232,6 +1272,9 @@ impl FeatureDatabase { let lifecycle = owner::Lifecycle { root: self.inner.root.clone(), scope: self.inner.scope_id.clone(), + established: self.inner.lifecycle_established.clone(), + #[cfg(feature = "test-support")] + observation: self.inner.lifecycle_observation.clone(), }; let _lease = lifecycle.lock(false).map_err(E::from)?; self.inner.ensure_accepting().map_err(E::from)?; @@ -1308,6 +1351,9 @@ impl FeatureDatabaseInner { owner::Lifecycle { root: self.root.clone(), scope: self.scope_id.clone(), + established: self.lifecycle_established.clone(), + #[cfg(feature = "test-support")] + observation: self.lifecycle_observation.clone(), } .verify( &self.binding, @@ -1753,6 +1799,149 @@ mod tests { FeatureRegistration::new(feature_id, MIGRATIONS) } + #[cfg(feature = "test-support")] + #[test] + fn lifecycle_observation_is_manager_local_and_counts_retained_feature_operations() { + use std::os::unix::fs::MetadataExt; + + let temp = tempfile::tempdir().unwrap(); + let root = temp.path().join("storage"); + let manager = FeatureStorage::new(&root); + let observer = manager.clone(); + // Even managers sharing a physical root must not share observation state. + let other = FeatureStorage::new(&root); + let feature = manager.register(registration("test-feature")).unwrap(); + let database = manager.scope("workspace").unwrap().open(&feature).unwrap(); + assert_eq!( + other.lifecycle_observation(), + LifecycleObservation::default() + ); + let other_feature = other.register(registration("test-feature")).unwrap(); + let other_database = other + .scope("workspace") + .unwrap() + .open(&other_feature) + .unwrap(); + assert_eq!(database.identity(), other_database.identity()); + let identity = database.identity().clone(); + let lock_path = root.join(".lifecycle/workspace.lock"); + let lock_inode = fs::metadata(&lock_path).unwrap().ino(); + let other_before = other.lifecycle_observation(); + let opens_before = feature.open_observation(); + let before = observer.lifecycle_observation(); + + database + .with_connection(|connection| { + let synchronous: u32 = + connection.query_row("PRAGMA synchronous", [], |row| row.get(0))?; + assert_eq!(synchronous, 2, "durability policy remains FULL"); + Ok(()) + }) + .unwrap(); + let after_read = observer.lifecycle_observation(); + let read = after_read.since(before); + assert_eq!(read.acquisitions, 1); + assert_eq!( + read.sync_attempts, 0, + "same durable lifecycle needs no re-sync" + ); + assert_eq!(read.lock_attempts, 1); + assert_eq!(read.sync_nanoseconds, 0); + assert!(read.lock_nanoseconds > 0); + assert_eq!((read.failed_syncs, read.failed_locks), (0, 0)); + + insert(&database.clone(), "observed commit").unwrap(); + let after_write = observer.lifecycle_observation(); + let write = after_write.since(after_read); + assert_eq!(write.acquisitions, 1); + assert_eq!(write.sync_attempts, 0); + assert_eq!(write.lock_attempts, 1); + assert_eq!(write.sync_nanoseconds, 0); + assert!(write.lock_nanoseconds > 0); + assert_eq!((write.failed_syncs, write.failed_locks), (0, 0)); + assert_eq!(after_write, manager.lifecycle_observation()); + assert_eq!(other_before, other.lifecycle_observation()); + assert_eq!(opens_before, feature.open_observation()); + assert_eq!(database.identity(), &identity); + assert_eq!(fs::metadata(&lock_path).unwrap().ino(), lock_inode); + // Re-read committed data through the independent manager's verified DB. + assert_eq!(values(&other_database).unwrap(), ["observed commit"]); + } + + #[cfg(feature = "test-support")] + #[test] + fn lifecycle_observation_includes_retained_host_catalog_operations() { + let temp = tempfile::tempdir().unwrap(); + let manager = FeatureStorage::new(temp.path().join("storage")); + let database = manager + .open_host_catalog::( + &temp.path().join("host.sqlite"), + "host-provision", + "isolated test Host", + |connection| { + connection.execute_batch( + "CREATE TABLE items (id INTEGER PRIMARY KEY, value TEXT NOT NULL UNIQUE);", + )?; + Ok(()) + }, + ) + .unwrap(); + let before = manager.lifecycle_observation(); + assert_eq!(before.acquisitions, 1); + assert_eq!(before.sync_attempts, 1); + assert_eq!(before.lock_attempts, 1); + let identity = database.identity().clone(); + insert(&database, "host commit").unwrap(); + assert_eq!(values(&database).unwrap(), ["host commit"]); + let delta = manager.lifecycle_observation().since(before); + assert_eq!(delta.acquisitions, 2); + assert_eq!(delta.sync_attempts, 0); + assert_eq!(delta.lock_attempts, 2); + assert_eq!(delta.sync_nanoseconds, 0); + assert!(delta.lock_nanoseconds > 0); + assert_eq!((delta.failed_syncs, delta.failed_locks), (0, 0)); + assert_eq!(database.identity(), &identity); + } + + #[cfg(feature = "test-support")] + #[test] + fn lifecycle_observation_counts_failed_directory_attempts_without_suppressing_retry() { + let temp = tempfile::tempdir().unwrap(); + let root = temp.path().join("storage"); + fs::create_dir(&root).unwrap(); + let obstacle = root.join(".lifecycle"); + fs::write(&obstacle, b"not a directory").unwrap(); + let manager = FeatureStorage::new(&root); + let feature = manager.register(registration("test-feature")).unwrap(); + let scoped = manager.scope("workspace").unwrap(); + for _ in 0..2 { + assert!(matches!( + scoped.open(&feature), + Err(FeatureStorageError::Io(_)) + )); + } + let failed = manager.lifecycle_observation(); + assert_eq!(failed.acquisitions, 2); + assert_eq!(failed.sync_attempts, 2); + assert_eq!(failed.failed_syncs, 2); + assert!(failed.sync_nanoseconds > 0); + assert_eq!(failed.lock_attempts, 0); + assert_eq!(failed.lock_nanoseconds, 0); + assert_eq!(failed.failed_locks, 0); + assert_eq!(feature.open_observation(), OpenObservation::default()); + assert!(!root.join("workspace/features/test-feature.sqlite").exists()); + + fs::remove_file(obstacle).unwrap(); + let database = scoped.open(&feature).unwrap(); + insert(&database, "retry committed").unwrap(); + assert_eq!(values(&database).unwrap(), ["retry committed"]); + let retried = manager.lifecycle_observation().since(failed); + assert!(retried.acquisitions > 0); + assert_eq!(retried.sync_attempts, 1); + assert_eq!(retried.lock_attempts, retried.acquisitions); + assert_eq!((retried.failed_syncs, retried.failed_locks), (0, 0)); + } + #[test] fn metadata_ancestor_links_are_synced_before_sqlite_initialization() { let temp = tempfile::tempdir().unwrap(); diff --git a/crates/feature-storage/src/lifecycle_lock.rs b/crates/feature-storage/src/lifecycle_lock.rs new file mode 100644 index 000000000..d05543cd7 --- /dev/null +++ b/crates/feature-storage/src/lifecycle_lock.rs @@ -0,0 +1,219 @@ +//! Durability establishment is local to a manager, not inferred from existence. +//! Retained descriptors pin the established ancestor chain and scoped lock inode; +//! each acquisition rechecks the configured/canonical paths before and after the +//! OS lock. A replacement is rejected for the lifetime of that proof. New +//! managers (including new processes) repeat all persistence steps. Failed +//! establishment publishes no proof. Configured Host symlinks remain supported, +//! but their mutable links require full establishment on every acquisition. +//! +//! This is only durability evidence. The caller must still verify the current +//! owner catalog, DB identity, fences and authorization under the returned lease. +//! As with the owner, advisory checks require a Host-protected filesystem; they +//! do not sandbox native writers racing after the final path check. +use super::*; + +/// Instance-local durability evidence, shared only by a manager and its handles. +/// It never substitutes for the owner catalog, DB identity, fences or authority. +#[derive(Default)] +pub(crate) struct EstablishedLocks { + scopes: Mutex>>, +} + +struct Directory { + path: PathBuf, + file: File, + symlink: bool, +} + +struct Proof { + root: PathBuf, + canonical_root: PathBuf, + directories: Vec, + lock_path: PathBuf, + lock: File, + reusable: bool, +} + +fn conflict() -> FeatureStorageError { + FeatureStorageError::Identity("canonical lifecycle lock path was replaced or redirected".into()) +} + +fn same_file(left: &fs::Metadata, right: &fs::Metadata) -> bool { + left.dev() == right.dev() && left.ino() == right.ino() +} + +fn regular_lock(path: &Path) -> Result { + let metadata = fs::symlink_metadata(path)?; + if !metadata.is_file() { + return Err(conflict()); + } + Ok(metadata) +} + +impl Proof { + fn capture(root: &Path, scope: &str) -> Result { + let root = if root.is_absolute() { + root.to_path_buf() + } else { + std::env::current_dir()?.join(root) + }; + let directory = root.join(".lifecycle"); + // A configured Host root may contain symlinks. The lock namespace itself + // must not: following a replacement there could split the advisory lock. + if !fs::symlink_metadata(&directory)?.is_dir() { + return Err(conflict()); + } + let canonical_root = fs::canonicalize(&root)?; + let resolved = fs::canonicalize(&directory)?; + if resolved != canonical_root.join(".lifecycle") { + return Err(conflict()); + } + let mut directories = Vec::new(); + let mut seen = HashSet::new(); + for path in directory.ancestors().chain(resolved.ancestors()) { + if seen.insert(path.to_path_buf()) { + let symlink = fs::symlink_metadata(path)?.is_symlink(); + let file = File::open(path)?; + if !file.metadata()?.is_dir() { + return Err(conflict()); + } + directories.push(Directory { + path: path.into(), + file, + symlink, + }); + } + } + let lock_path = directory.join(format!("{scope}.lock")); + match fs::symlink_metadata(&lock_path) { + Ok(_) => { + regular_lock(&lock_path)?; + } + Err(error) if error.kind() == std::io::ErrorKind::NotFound => {} + Err(error) => return Err(error.into()), + } + let lock = OpenOptions::new() + .read(true) + .write(true) + .create(true) + .truncate(false) + .open(&lock_path)?; + let proof = Self { + root, + canonical_root, + reusable: directories.iter().all(|entry| !entry.symlink), + directories, + lock_path, + lock, + }; + proof.check()?; + Ok(proof) + } + + fn check(&self) -> Result<()> { + if fs::canonicalize(&self.root)? != self.canonical_root { + return Err(conflict()); + } + for directory in &self.directories { + let link = fs::symlink_metadata(&directory.path)?; + if link.is_symlink() != directory.symlink + || !same_file(&fs::metadata(&directory.path)?, &directory.file.metadata()?) + { + return Err(conflict()); + } + } + if !same_file(®ular_lock(&self.lock_path)?, &self.lock.metadata()?) + || fs::canonicalize(&self.lock_path)? + != self + .canonical_root + .join(".lifecycle") + .join(self.lock_path.file_name().ok_or_else(conflict)?) + { + return Err(conflict()); + } + Ok(()) + } + + fn open(&self) -> Result { + self.check()?; + // A fresh open description is required for each flock lease. Cloning + // the pinned File would share one OS lock across independent operations. + let lock = OpenOptions::new() + .read(true) + .write(true) + .open(&self.lock_path)?; + if !same_file(&lock.metadata()?, &self.lock.metadata()?) { + return Err(conflict()); + } + Ok(lock) + } +} + +impl Lifecycle { + pub(super) fn lock_with_sync( + &self, + exclusive: bool, + sync: &mut impl FnMut(&Path) -> std::io::Result<()>, + ) -> Result { + #[cfg(feature = "test-support")] + self.observation.acquisition(); + let proof = { + let mut scopes = self.established.scopes.lock().map_err(|_| { + FeatureStorageError::Operation("lifecycle establishment mutex poisoned".into()) + })?; + let previous = scopes.get(&self.scope); + if let Some(proof) = previous { + // Replacements are rejected, not silently accepted as a new + // lock domain while an old manager/handle might still hold it. + proof.check()?; + } + if let Some(proof) = previous.filter(|proof| proof.reusable) { + proof.clone() + } else { + #[cfg(feature = "test-support")] + let started = self.observation.sync_start(); + let result = (|| { + let directory = self.root.join(".lifecycle"); + fs::create_dir_all(&directory)?; + let proof = Proof::capture(&self.root, &self.scope)?; + // Re-sync even existing ancestors: their presence is not + // evidence that an earlier attempt persisted their links. + durable_create_directory(&directory, sync)?; + proof.lock.sync_all()?; + sync(&directory)?; + proof.check()?; + Ok::<_, FeatureStorageError>(Arc::new(proof)) + })(); + #[cfg(feature = "test-support")] + self.observation.sync_finished(started, result.is_err()); + let proof = result?; + // Publish only after every durability step and path recheck. + // Pinned descriptors prevent inode reuse from reviving proof. + scopes.insert(self.scope.clone(), proof.clone()); + proof + } + }; + let lock = proof.open()?; + #[cfg(feature = "test-support")] + let started = self.observation.lock_start(); + let result = if exclusive { + lock.lock() + } else { + lock.lock_shared() + }; + #[cfg(feature = "test-support")] + self.observation.lock_finished(started, result.is_err()); + result?; + // Do not hold the instance mutex while waiting for another process. + // Check again after the OS wait, before the caller can touch its DB. + proof.check()?; + if !same_file(&lock.metadata()?, &proof.lock.metadata()?) { + return Err(conflict()); + } + Ok(lock) + } +} + +#[cfg(test)] +#[path = "lifecycle_lock_tests.rs"] +mod tests; diff --git a/crates/feature-storage/src/lifecycle_lock_tests.rs b/crates/feature-storage/src/lifecycle_lock_tests.rs new file mode 100644 index 000000000..9f7fec9e5 --- /dev/null +++ b/crates/feature-storage/src/lifecycle_lock_tests.rs @@ -0,0 +1,255 @@ +//! Component tests use the production manager, lifecycle locks and filesystem. +//! Only directory sync failure is injected; DB identity checks remain production. +use super::*; +use std::os::unix::fs::symlink; + +fn sync_paths(lifecycle: &Lifecycle) -> Vec { + let mut paths = Vec::new(); + drop( + lifecycle + .lock_with_sync(false, &mut |path| { + paths.push(path.into()); + sync_directory(path) + }) + .unwrap(), + ); + paths +} + +#[test] +fn failed_establishment_retries_every_link_even_when_directories_and_lock_exist() { + let reference = tempfile::tempdir().unwrap(); + let reference_manager = FeatureStorage::new(reference.path().join("nested/storage")); + let count = sync_paths(&reference_manager.lifecycle("scope")).len(); + assert!(count > 2, "ancestor chain and final lock-directory sync"); + for fail_at in 0..count { + let temp = tempfile::tempdir().unwrap(); + let root = temp.path().join("nested/storage"); + let manager = FeatureStorage::new(&root); + let lifecycle = manager.lifecycle("scope"); + let mut calls = 0; + let result = lifecycle.lock_with_sync(false, &mut |path| { + let index = calls; + calls += 1; + if index == fail_at { + Err(std::io::Error::other("injected directory sync failure")) + } else { + sync_directory(path) + } + }); + assert!( + matches!(result, Err(FeatureStorageError::Io(_))), + "{fail_at}" + ); + assert_eq!(calls, fail_at + 1); + assert!(root.join(".lifecycle/scope.lock").is_file()); + assert!(lifecycle.established.scopes.lock().unwrap().is_empty()); + let retried = sync_paths(&lifecycle); + assert_eq!( + retried.len(), + count, + "all durability steps retry at {fail_at}" + ); + assert_eq!( + retried.first(), + retried.last(), + "lock directory persisted last" + ); + drop( + lifecycle + .lock_with_sync(false, &mut |_| panic!("already established")) + .unwrap(), + ); + } +} + +#[test] +fn establishment_is_shared_with_handles_but_not_new_managers_or_scopes() { + let temp = tempfile::tempdir().unwrap(); + let root = temp.path().join("storage"); + let manager = FeatureStorage::new(&root); + let paths = sync_paths(&manager.lifecycle("scope")); + assert!(!paths.is_empty()); + assert!(sync_paths(&manager.clone().lifecycle("scope")).is_empty()); + assert_eq!(sync_paths(&manager.lifecycle("other")), paths); + let other = FeatureStorage::new(&root); + assert_eq!( + sync_paths(&other.lifecycle("scope")), + paths, + "existence is not proof" + ); + assert_eq!( + sync_paths(&FeatureStorage::new(&root).lifecycle("scope")), + paths + ); +} + +fn open_database(root: &Path) -> (FeatureStorage, RegisteredFeature, FeatureDatabase) { + let manager = FeatureStorage::new(root); + let registration = manager + .register(FeatureRegistration::new("feature", &[])) + .unwrap(); + let database = manager.scope("scope").unwrap().open(®istration).unwrap(); + database + .transaction(|tx| { + tx.execute_batch( + "CREATE TABLE marker(value TEXT); INSERT INTO marker VALUES('persisted');", + )?; + Ok(()) + }) + .unwrap(); + (manager, registration, database) +} + +fn rejected( + manager: &FeatureStorage, + registration: &RegisteredFeature, + database: &FeatureDatabase, +) { + assert!( + database + .with_connection(|_| -> Result<()> { panic!("replaced lock cannot enter DB closure") }) + .is_err() + ); + assert!(database.guard(database.identity()).is_err()); + assert!(manager.scope("scope").unwrap().open(registration).is_err()); +} + +#[test] +fn physical_root_directory_lock_and_ancestor_replacement_reject_retained_access() { + for replacement in ["root", "directory", "lock", "ancestor"] { + let temp = tempfile::tempdir().unwrap(); + let root = temp.path().join("parent/storage"); + let (manager, registration, database) = open_database(&root); + let path = match replacement { + "root" => root.clone(), + "directory" => root.join(".lifecycle"), + "lock" => root.join(".lifecycle/scope.lock"), + "ancestor" => root.parent().unwrap().to_path_buf(), + _ => unreachable!(), + }; + let moved = temp.path().join("moved"); + fs::rename(&path, &moved).unwrap(); + if replacement == "lock" { + fs::write(&path, b"").unwrap(); + } else { + fs::create_dir_all(&path).unwrap(); + } + rejected(&manager, ®istration, &database); + // Rejecting must not erase the old proof and accept the replacement on retry. + rejected(&manager, ®istration, &database); + fs::remove_file(&path) + .or_else(|_| fs::remove_dir_all(&path)) + .unwrap(); + fs::rename(&moved, &path).unwrap(); + let value: String = database + .with_connection(|connection| { + Ok(connection.query_row("SELECT value FROM marker", [], |r| r.get(0))?) + }) + .unwrap(); + assert_eq!(value, "persisted"); + } +} + +#[test] +fn symlink_replacement_cannot_redirect_or_split_an_established_lock() { + for replacement in ["root", "directory", "lock"] { + let temp = tempfile::tempdir().unwrap(); + let root = temp.path().join("storage"); + let (manager, registration, database) = open_database(&root); + let path = match replacement { + "root" => root.clone(), + "directory" => root.join(".lifecycle"), + "lock" => root.join(".lifecycle/scope.lock"), + _ => unreachable!(), + }; + let moved = temp.path().join("moved"); + fs::rename(&path, &moved).unwrap(); + symlink(&moved, &path).unwrap(); + // The target inode is still the original; following metadata alone is insufficient. + rejected(&manager, ®istration, &database); + fs::remove_file(&path).unwrap(); + fs::rename(&moved, &path).unwrap(); + drop(database.guard(database.identity()).unwrap()); + } +} + +#[test] +fn initial_lock_namespace_symlinks_are_rejected_without_establishment_proof() { + for directory_link in [true, false] { + let temp = tempfile::tempdir().unwrap(); + let root = temp.path().join("storage"); + fs::create_dir(&root).unwrap(); + let target = temp.path().join("target"); + if directory_link { + fs::create_dir(&target).unwrap(); + symlink(&target, root.join(".lifecycle")).unwrap(); + } else { + fs::create_dir(root.join(".lifecycle")).unwrap(); + fs::write(&target, b"").unwrap(); + symlink(&target, root.join(".lifecycle/scope.lock")).unwrap(); + } + let manager = FeatureStorage::new(&root); + let lifecycle = manager.lifecycle("scope"); + assert!(matches!( + lifecycle.lock(false), + Err(FeatureStorageError::Identity(_)) + )); + assert!(lifecycle.established.scopes.lock().unwrap().is_empty()); + } +} + +#[test] +fn configured_host_symlinks_reestablish_and_reject_changed_canonical_root() { + let temp = tempfile::tempdir().unwrap(); + let target = temp.path().join("target"); + fs::create_dir(&target).unwrap(); + let root = temp.path().join("alias"); + symlink(&target, &root).unwrap(); + let manager = FeatureStorage::new(&root); + let lifecycle = manager.lifecycle("scope"); + let paths = sync_paths(&lifecycle); + assert!(paths.contains(&target)); + // Mutable symlink links are not pinned by ordinary directory descriptors; + // therefore no successful proof can suppress their durability steps. + assert_eq!(sync_paths(&lifecycle), paths); + let next = temp.path().join("next"); + fs::create_dir(&next).unwrap(); + fs::remove_file(&root).unwrap(); + symlink(&next, &root).unwrap(); + assert!(matches!( + lifecycle.lock(false), + Err(FeatureStorageError::Identity(_)) + )); +} + +#[cfg(feature = "test-support")] +#[test] +fn replacement_during_os_lock_wait_is_rejected_before_database_access() { + use std::time::{Duration, Instant}; + + let temp = tempfile::tempdir().unwrap(); + let root = temp.path().join("storage"); + let (manager, registration, database) = open_database(&root); + let guard = database.guard(database.identity()).unwrap(); + let attempts = manager.lifecycle_observation().lock_attempts; + let lifecycle = manager.lifecycle("scope"); + let waiter = std::thread::spawn(move || lifecycle.lock(true)); + let deadline = Instant::now() + Duration::from_secs(5); + while manager.lifecycle_observation().lock_attempts == attempts { + assert!( + Instant::now() < deadline, + "waiter did not reach OS lock call" + ); + std::thread::yield_now(); + } + let lock_path = root.join(".lifecycle/scope.lock"); + fs::rename(&lock_path, root.join(".lifecycle/held.lock")).unwrap(); + fs::write(&lock_path, b"").unwrap(); + drop(guard); + assert!(matches!( + waiter.join().unwrap(), + Err(FeatureStorageError::Identity(_)) + )); + rejected(&manager, ®istration, &database); +} diff --git a/crates/feature-storage/src/lifecycle_observation.rs b/crates/feature-storage/src/lifecycle_observation.rs new file mode 100644 index 000000000..9b456609e --- /dev/null +++ b/crates/feature-storage/src/lifecycle_observation.rs @@ -0,0 +1,114 @@ +use std::sync::atomic::{AtomicU64, Ordering}; +use std::time::Instant; + +/// Cumulative lifecycle observations for one storage manager and its clones/handles. +/// Counts attempts before starting I/O, including attempts that return errors. +/// `acquisitions` counts entries to lifecycle acquisition; `sync_attempts` counts +/// lifecycle durability establishment attempts (the full ancestor sync plus +/// lockfile and lock-directory persistence), and `lock_attempts` counts OS +/// shared/exclusive lock calls, excluding lockfile open failures. Once this +/// manager has established and rechecked the same physical paths, ordinary +/// acquisitions need no establishment attempt. A fresh manager establishes again. +/// +/// Durations sum monotonic elapsed nanoseconds around those calls, including +/// errors and OS lock waits, but not lease holding time or database operations. +/// Overlapping threads contribute independently: aggregates are not wall time. +/// Fields saturate at u64::MAX and are independently sampled with relaxed loads; +/// a concurrent snapshot need not describe one atomic point in time. In-flight +/// calls contribute attempts before their duration or failure is recorded. +/// These diagnostics retain no database or authority and are not termination, +/// durability, or successful-work evidence; they never control storage I/O. +#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)] +pub struct LifecycleObservation { + pub acquisitions: u64, + pub sync_attempts: u64, + pub sync_nanoseconds: u64, + pub failed_syncs: u64, + pub lock_attempts: u64, + pub lock_nanoseconds: u64, + pub failed_locks: u64, +} + +impl LifecycleObservation { + /// Saturating differences between snapshots from the same manager. + pub fn since(self, before: Self) -> Self { + Self { + acquisitions: self.acquisitions.saturating_sub(before.acquisitions), + sync_attempts: self.sync_attempts.saturating_sub(before.sync_attempts), + sync_nanoseconds: self + .sync_nanoseconds + .saturating_sub(before.sync_nanoseconds), + failed_syncs: self.failed_syncs.saturating_sub(before.failed_syncs), + lock_attempts: self.lock_attempts.saturating_sub(before.lock_attempts), + lock_nanoseconds: self + .lock_nanoseconds + .saturating_sub(before.lock_nanoseconds), + failed_locks: self.failed_locks.saturating_sub(before.failed_locks), + } + } +} + +#[derive(Debug, Default)] +pub(super) struct Counters { + acquisitions: AtomicU64, + sync_attempts: AtomicU64, + sync_nanoseconds: AtomicU64, + failed_syncs: AtomicU64, + lock_attempts: AtomicU64, + lock_nanoseconds: AtomicU64, + failed_locks: AtomicU64, +} + +impl Counters { + pub(super) fn snapshot(&self) -> LifecycleObservation { + LifecycleObservation { + acquisitions: self.acquisitions.load(Ordering::Relaxed), + sync_attempts: self.sync_attempts.load(Ordering::Relaxed), + sync_nanoseconds: self.sync_nanoseconds.load(Ordering::Relaxed), + failed_syncs: self.failed_syncs.load(Ordering::Relaxed), + lock_attempts: self.lock_attempts.load(Ordering::Relaxed), + lock_nanoseconds: self.lock_nanoseconds.load(Ordering::Relaxed), + failed_locks: self.failed_locks.load(Ordering::Relaxed), + } + } + + pub(super) fn acquisition(&self) { + add(&self.acquisitions, 1); + } + + pub(super) fn sync_start(&self) -> Instant { + add(&self.sync_attempts, 1); + Instant::now() + } + + pub(super) fn sync_finished(&self, started: Instant, failed: bool) { + let elapsed = started.elapsed(); + add(&self.sync_nanoseconds, nanoseconds(elapsed)); + if failed { + add(&self.failed_syncs, 1); + } + } + + pub(super) fn lock_start(&self) -> Instant { + add(&self.lock_attempts, 1); + Instant::now() + } + + pub(super) fn lock_finished(&self, started: Instant, failed: bool) { + let elapsed = started.elapsed(); + add(&self.lock_nanoseconds, nanoseconds(elapsed)); + if failed { + add(&self.failed_locks, 1); + } + } +} + +fn nanoseconds(duration: std::time::Duration) -> u64 { + duration.as_nanos().min(u128::from(u64::MAX)) as u64 +} + +fn add(counter: &AtomicU64, amount: u64) { + let _ = counter.fetch_update(Ordering::Relaxed, Ordering::Relaxed, |value| { + Some(value.saturating_add(amount)) + }); +} diff --git a/crates/feature-storage/src/owner.rs b/crates/feature-storage/src/owner.rs index 8fa4216c6..7b4c71577 100644 --- a/crates/feature-storage/src/owner.rs +++ b/crates/feature-storage/src/owner.rs @@ -16,6 +16,10 @@ use std::os::unix::fs::MetadataExt; const IDENTITY_TABLE: &str = "__yoi_storage_identity"; +#[path = "lifecycle_lock.rs"] +mod lifecycle_lock; +pub(super) use lifecycle_lock::EstablishedLocks; + /// Expected storage target, retained with authorization and rechecked on resume. /// This is a physical identity, not an authorization credential or a Job lease. #[derive(Clone, Debug, PartialEq, Eq)] @@ -36,27 +40,14 @@ pub(super) struct Binding { pub(super) struct Lifecycle { pub root: PathBuf, pub scope: String, + pub established: Arc, + #[cfg(feature = "test-support")] + pub observation: Arc, } impl Lifecycle { pub fn lock(&self, exclusive: bool) -> Result { - durable_create_directory(&self.root.join(".lifecycle"), &mut sync_directory)?; - let lock = OpenOptions::new() - .read(true) - .write(true) - .create(true) - .truncate(false) - .open( - self.root - .join(".lifecycle") - .join(format!("{}.lock", self.scope)), - )?; - if exclusive { - lock.lock()?; - } else { - lock.lock_shared()?; - } - Ok(lock) + self.lock_with_sync(exclusive, &mut sync_directory) } fn catalog(&self, create: bool) -> Result { diff --git a/crates/standalone/src/subjektiv.rs b/crates/standalone/src/subjektiv.rs index 6d90e2651..51146edc4 100644 --- a/crates/standalone/src/subjektiv.rs +++ b/crates/standalone/src/subjektiv.rs @@ -381,30 +381,80 @@ enum ConsolidationRecoveryError { Service(#[from] crate::jobs::JobServiceError), #[error(transparent)] Domain(#[from] subjektiv::api::OperationError), + #[error(transparent)] + Writer(#[from] subjektiv::journal::JournalError), } impl ConsolidationRecoveryError { fn retryable_read_or_storage(&self) -> bool { use crate::jobs::{JobServiceError, JobStoreError}; - matches!( - self, - Self::Service(JobServiceError::Store( - JobStoreError::Sqlite(_) | JobStoreError::Storage(_) | JobStoreError::Io(_) - )) | Self::Domain(subjektiv::api::OperationError::Storage(_)) - // The Host's writer/Subject liveness reads wrap storage errors in - // Unavailable. Recheck them within this same finite read-only budget; - // their failure NEVER authorizes a receipt, release, or execution. - | Self::Host(SubjektivHostError::Unavailable(_)) - ) + fn sqlite(error: &rusqlite::Error) -> bool { + matches!(error, rusqlite::Error::SqliteFailure(cause, _) + if matches!(cause.code, + rusqlite::ErrorCode::DatabaseBusy | rusqlite::ErrorCode::DatabaseLocked + | rusqlite::ErrorCode::SystemIoFailure | rusqlite::ErrorCode::DiskFull + | rusqlite::ErrorCode::CannotOpen)) + } + fn io(error: &std::io::Error) -> bool { + matches!( + error.kind(), + std::io::ErrorKind::Interrupted + | std::io::ErrorKind::WouldBlock + | std::io::ErrorKind::TimedOut + ) + } + fn storage(error: &feature_storage::FeatureStorageError) -> bool { + match error { + feature_storage::FeatureStorageError::Sqlite(error) => sqlite(error), + feature_storage::FeatureStorageError::Io(error) => io(error), + _ => false, + } + } + match self { + Self::Service(JobServiceError::Store(JobStoreError::Sqlite(error))) => sqlite(error), + Self::Service(JobServiceError::Store(JobStoreError::Io(error))) => io(error), + Self::Service(JobServiceError::Store(JobStoreError::Storage(error))) + | Self::Domain(subjektiv::api::OperationError::Storage(error)) + | Self::Writer(subjektiv::journal::JournalError::Storage(error)) + | Self::Writer(subjektiv::journal::JournalError::Subject( + subjektiv::SubjektivError::Storage(error), + )) => storage(error), + // Authority loss, permanent SQL constraints, malformed data and + // untyped Host unavailability never acquire a storage retry owner. + _ => false, + } } } +#[derive(Debug, PartialEq, Eq)] +enum ConsolidationRecoveryProgress { + FastRetry, + RetainedStorage, + Blocked, +} + struct ConsolidationRecoveryOwner { - running: bool, + progress: ConsolidationRecoveryProgress, passes: usize, error: ConsolidationRecoveryError, } +impl ConsolidationRecoveryOwner { + fn next_action(&self) -> &'static str { + match self.progress { + ConsolidationRecoveryProgress::FastRetry => { + "reconcile the same attempt after short storage backoff" + } + ConsolidationRecoveryProgress::RetainedStorage => { + "recheck the original binding/attempt/input/grant and same domain operation with capped storage backoff" + } + ConsolidationRecoveryProgress::Blocked => { + "resolve the recorded permanent storage or authority cause; no automatic execution or cleanup" + } + } + } +} + pub(crate) struct LocalSubjectHost { writer: subjektiv::journal::JournalWriterBinding, self_weak: std::sync::Weak, @@ -813,6 +863,23 @@ impl LocalSubjectHost { self.recover_journals(); } + /// Recheck the actual writer store without erasing its typed storage cause. + /// A released writer/closed Host is authority loss, not a transient outage. + fn ensure_consolidation_live(&self) -> Result<(), ConsolidationRecoveryError> { + if !self.live.load(Ordering::Acquire) { + return Err( + SubjektivHostError::Unavailable("local Subject Host is closed".into()).into(), + ); + } + if !self.catalog.store.journal_writer_active(&self.writer)? { + return Err(SubjektivHostError::Unavailable( + "Subject writer binding was released".into(), + ) + .into()); + } + Ok(()) + } + /// Prefer a complete saved result. Otherwise close only this failed attempt's /// domain reservation, retaining unresolved work and the original failure. /// Neither path reruns the finalizer or proves prior execution shutdown. @@ -821,7 +888,7 @@ impl LocalSubjectHost { jobs: &crate::jobs::StandaloneJobs, observed: &crate::jobs::JobSnapshot, ) -> Result { - self.ensure_live()?; + self.ensure_consolidation_live()?; let current = jobs.get(&observed.request.job_id)?; if current.attempt.attempt_id != observed.attempt.attempt_id || current.input_digest != observed.input_digest @@ -852,7 +919,7 @@ impl LocalSubjectHost { ¤t.request.job_id, ¤t.attempt.attempt_id, )? { - self.ensure_live()?; + self.ensure_consolidation_live()?; let settled = jobs.accept_reconciled(&job::JobResultSubmission { job_id: current.request.job_id, attempt_id: current.attempt.attempt_id, @@ -891,7 +958,7 @@ impl LocalSubjectHost { ¤t.input_digest, )) .map_err(|error| SubjektivHostError::InvalidArgument(error.to_string()))?; - self.ensure_live()?; + self.ensure_consolidation_live()?; // Recheck the exact Host/grant before persistence. The service/store also // fence the current attempt and independently require ALL executions released. let checked = jobs.get(¤t.request.job_id)?; @@ -910,23 +977,16 @@ impl LocalSubjectHost { Ok(true) } - /// Keep a finite recovery owner for read/storage failure of this exact - /// snapshot. No startup-only repair, disconnect polling, new attempt or model - /// call. Exhaustion retains the typed cause and original durable obligation; - /// a fresh Host can reconcile it, never infer success from budget exhaustion. + /// Retain only typed temporary storage failures of this exact operation. + /// After eight fast passes the same owner uses capped backoff, not another + /// execution budget or a healthy-other-store probe. Each pass rereads the + /// current binding/attempt/input/grant/retirement and actual domain receipts. fn own_consolidation_recovery( &self, jobs: crate::jobs::StandaloneJobs, observed: crate::jobs::JobSnapshot, error: ConsolidationRecoveryError, ) { - eprintln!( - "standalone consolidation Job {} reconciliation blocked: {error}", - observed.request.job_id - ); - if !error.retryable_read_or_storage() { - return; - } let key = ( observed.request.job_id.clone(), observed.attempt.attempt_id.clone(), @@ -936,14 +996,8 @@ impl LocalSubjectHost { .consolidation_recovery .lock() .expect("consolidation recovery poisoned"); - if let Some(owner) = owners.get(&key) { - if !owner.running { - eprintln!( - "standalone consolidation Job {} retains exhausted reconciliation after {} passes: {}; next action: resolve cause and restore the same Host binding", - key.0, owner.passes, owner.error - ); - } - return; // Repeated disconnect cannot refill a running or exhausted budget. + if owners.contains_key(&key) { + return; // Duplicate notifications cannot refill a budget or revive a blocked owner. } let Some(host) = self.self_weak.upgrade() else { return; @@ -951,18 +1005,42 @@ impl LocalSubjectHost { let Ok(runtime) = tokio::runtime::Handle::try_current() else { return; }; - owners.insert( - key.clone(), - ConsolidationRecoveryOwner { - running: true, - passes: 1, - error, + let retry = error.retryable_read_or_storage(); + let owner = ConsolidationRecoveryOwner { + progress: if retry { + ConsolidationRecoveryProgress::FastRetry + } else { + ConsolidationRecoveryProgress::Blocked }, + passes: 1, + error, + }; + eprintln!( + "standalone consolidation Job {} reconciliation {:?} after {} passes: {}; next action: {}", + key.0, + owner.progress, + owner.passes, + owner.error, + owner.next_action() ); + owners.insert(key.clone(), owner); + if !retry { + return; + } runtime.spawn(async move { - const PASSES: usize = 8; - for pass in 2..=PASSES { - tokio::time::sleep(std::time::Duration::from_millis(100 << (pass - 2).min(5))).await; + const FAST_PASSES: usize = 8; + let mut pass = 1usize; + let mut retained_delay = std::time::Duration::from_secs(1); + loop { + let delay = if pass < FAST_PASSES { + std::time::Duration::from_millis(100 << (pass - 1).min(5)) + } else { + let delay = retained_delay; + retained_delay = (retained_delay * 2).min(std::time::Duration::from_secs(30)); + delay + }; + tokio::time::sleep(delay).await; + pass = pass.saturating_add(1); let result = host.recover_consolidation_job(&jobs, &observed); let mut owners = host.consolidation_recovery.lock().expect("consolidation recovery poisoned"); match result { @@ -973,15 +1051,18 @@ impl LocalSubjectHost { return; } Err(error) => { - let retry = error.retryable_read_or_storage() && pass < PASSES; + let retry = error.retryable_read_or_storage(); let owner = owners.get_mut(&key).expect("retained consolidation owner"); owner.passes = pass; - owner.running = retry; owner.error = error; - if !retry { - eprintln!("standalone consolidation Job {} reconciliation stopped after {} passes: {}; original Failed/input/receipts retained; next action: resolve storage/authority failure and restore the same Host binding", key.0, owner.passes, owner.error); - return; + owner.progress = if !retry { ConsolidationRecoveryProgress::Blocked } + else if pass < FAST_PASSES { ConsolidationRecoveryProgress::FastRetry } + else { ConsolidationRecoveryProgress::RetainedStorage }; + if !retry || pass == FAST_PASSES { + eprintln!("standalone consolidation Job {} reconciliation {:?} after {} passes: {}; original input/receipts retained; next action: {}", + key.0, owner.progress, owner.passes, owner.error, owner.next_action()); } + if !retry { return; } } } } @@ -991,7 +1072,7 @@ impl LocalSubjectHost { fn consolidation_grant( &self, snapshot: &crate::jobs::JobSnapshot, - ) -> Result { + ) -> Result { let grant: LocalConsolidationGrant = serde_json::from_value(snapshot.domain_grant.clone().ok_or_else(|| { SubjektivHostError::InvalidArgument("missing explicit Subject delegation".into()) @@ -1004,16 +1085,17 @@ impl LocalSubjectHost { { return Err(SubjektivHostError::InvalidArgument( "persisted Job grant/input does not match local Subject capability".into(), - )); + ) + .into()); } let subject = self .catalog .store .subject(&self.binding.subject_id) - .map_err(domain_error)? + .map_err(subjektiv::api::subjektiv_store_error)? .ok_or_else(|| SubjektivHostError::Unavailable("Subject missing".into()))?; if subject.state != subjektiv::SubjectState::Active { - return Err(SubjektivHostError::Unavailable("Subject is retired".into())); + return Err(SubjektivHostError::Unavailable("Subject is retired".into()).into()); } Ok(grant) } diff --git a/crates/standalone/src/subjektiv/tests.rs b/crates/standalone/src/subjektiv/tests.rs index 9a1280930..ecbca1fab 100644 --- a/crates/standalone/src/subjektiv/tests.rs +++ b/crates/standalone/src/subjektiv/tests.rs @@ -99,6 +99,16 @@ impl ConsolidationRecoveryFixture { } } + fn domain_path(&self) -> PathBuf { + self.body + .host + .catalog + .root + .join("features") + .join(&self.body.host.catalog.scope_id) + .join("features/subjektiv.sqlite") + } + fn domain_closure(&self) -> Option { use rusqlite::OptionalExtension; rusqlite::Connection::open(&self.body.jobs_path) @@ -356,8 +366,226 @@ async fn failed_consolidation_retains_unreleased_stale_inactive_migrated_and_era } } +fn assert_real_busy(error: &ConsolidationRecoveryError, domain: bool) { + let cause = match error { + ConsolidationRecoveryError::Domain(subjektiv::api::OperationError::Storage( + feature_storage::FeatureStorageError::Sqlite(rusqlite::Error::SqliteFailure(cause, _)), + )) if domain => cause, + ConsolidationRecoveryError::Service(crate::jobs::JobServiceError::Store( + crate::jobs::JobStoreError::Sqlite(rusqlite::Error::SqliteFailure(cause, _)), + )) if !domain => cause, + other => panic!("expected typed Busy from original domain={domain}: {other:?}"), + }; + assert_eq!(cause.code, rusqlite::ErrorCode::DatabaseBusy); +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn failed_consolidation_real_sqlite_busy_beyond_fast_budget_recovers_without_restore() { + real_sqlite_busy_consolidation_recovery(false, None).await; +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn failed_consolidation_real_domain_busy_beyond_fast_budget_recovers_without_restore() { + real_sqlite_busy_consolidation_recovery(true, None).await; +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn failed_consolidation_real_busy_retained_owner_rejects_stale_domain() { + real_sqlite_busy_consolidation_recovery(true, Some("stale")).await; +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn failed_consolidation_real_busy_retained_owner_rejects_retired_input() { + real_sqlite_busy_consolidation_recovery(true, Some("retired")).await; +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn failed_consolidation_real_busy_retained_owner_requires_actual_execution_cleanup() { + real_sqlite_busy_consolidation_recovery(true, Some("cleanup")).await; +} + +async fn real_sqlite_busy_consolidation_recovery(domain: bool, control: Option<&str>) { + let mut fixture = ConsolidationRecoveryFixture::new(); + if control != Some("cleanup") { + fixture.confirm_execution_joined(); + } + if control == Some("stale") { + fixture.generation(&fixture.saved.attempt.attempt_id); + fixture + .body + .host + .catalog + .store + .create_memory( + &fixture.body.binding.subject_id, + subjektiv::MemoryDraft::active( + memory::extract::CandidateKind::Lesson, + "changed", + "body", + "reason", + "change", + ), + ) + .unwrap(); + } + let before = fixture.saved.clone(); + let path = if domain { + fixture.domain_path() + } else { + fixture.body.jobs_path.clone() + }; + let connection = rusqlite::Connection::open(path).unwrap(); + connection.execute_batch("BEGIN IMMEDIATE;").unwrap(); + let error = fixture + .body + .host + .recover_consolidation_job(&fixture.jobs, &before) + .unwrap_err(); + assert_real_busy(&error, domain); + eprintln!( + "S1 real SQLite Busy domain={domain}; unchanged exact attempt; holding through pass8" + ); + fixture + .body + .host + .own_consolidation_recovery(fixture.jobs.clone(), before.clone(), error); + tokio::time::timeout(std::time::Duration::from_secs(60), async { + loop { + let reached = fixture + .body + .host + .consolidation_recovery + .lock() + .unwrap() + .values() + .any(|owner| owner.passes >= 8); + if reached { + break; + } + tokio::time::sleep(std::time::Duration::from_millis(10)).await; + } + }) + .await + .expect("real SQLite fault must remain through all eight passes"); + { + let owners = fixture.body.host.consolidation_recovery.lock().unwrap(); + assert_eq!(owners.len(), 1); + let owner = owners.values().next().unwrap(); + assert_eq!( + owner.progress, + ConsolidationRecoveryProgress::RetainedStorage + ); + assert_real_busy(&owner.error, domain); + assert!(owner.next_action().contains("same domain operation")); + } + assert!(fixture.domain_closure().is_none()); + assert_eq!(fixture.jobs.get(&before.request.job_id).unwrap(), before); + if control == Some("retired") { + // While the original Feature DB is STILL unavailable, explicitly erase + // the Job's original input. A healthy Job DB is not Feature recovery. + fixture + .jobs + .reclaim_input( + &before.request.job_id, + &before.input_digest, + crate::jobs::JobInputRetirementReason::ExplicitErasure, + ) + .unwrap(); + } + let expected = fixture.jobs.get(&before.request.job_id).unwrap(); + eprintln!("S1 pass8 observed while real lock still held; releasing only SQLite transaction"); + connection.execute_batch("ROLLBACK;").unwrap(); + // No restore, recover_jobs, notification, repeated own call, new Job, model or finalizer. + if let Some(control) = control { + tokio::time::timeout(std::time::Duration::from_secs(10), async { + loop { + let owners = fixture.body.host.consolidation_recovery.lock().unwrap(); + let stopped = if control == "cleanup" { + owners.is_empty() + } else { + owners + .values() + .any(|owner| owner.progress == ConsolidationRecoveryProgress::Blocked) + }; + drop(owners); + if stopped { + break; + } + tokio::time::sleep(std::time::Duration::from_millis(10)).await; + } + }) + .await + .expect("retained owner must honor current stale/retirement/cleanup refusal"); + assert!(fixture.domain_closure().is_none(), "{control}"); + assert_eq!( + fixture.jobs.get(&before.request.job_id).unwrap(), + expected, + "{control}" + ); + assert_eq!( + expected.executions, before.executions, + "no invented cleanup proof" + ); + if control == "cleanup" { + assert!(!expected.executions_released()); + assert!( + expected + .executions + .iter() + .all(|execution| !execution.writes_revoked) + ); + assert!(!fixture.jobs.unreleased_granted().unwrap().is_empty()); + } else { + let owners = fixture.body.host.consolidation_recovery.lock().unwrap(); + let owner = owners.values().next().unwrap(); + assert!(owner.passes > 8); + assert!(!owner.error.retryable_read_or_storage()); + assert!( + owner + .next_action() + .contains("no automatic execution or cleanup") + ); + } + return; + } + tokio::time::timeout(std::time::Duration::from_secs(10), async { + while fixture.domain_closure().is_none() { + tokio::time::sleep(std::time::Duration::from_millis(10)).await; + } + }) + .await + .expect("same original domain obligation must recover after only fault release"); + assert!(fixture.jobs.unreleased_granted().unwrap().is_empty()); + assert_eq!(fixture.jobs.get(&before.request.job_id).unwrap(), before); + assert!( + fixture + .body + .host + .consolidation_recovery + .lock() + .unwrap() + .is_empty() + ); + let retained = subjektiv::job::inspect_abandoned_attempt( + &fixture.body.host.catalog.store, + &fixture.body.binding.subject_id, + &["candidate".into()], + &before.request.job_id, + &before.attempt.attempt_id, + ) + .unwrap(); + assert!( + retained.prepared_generation_id.is_none(), + "no finalizer replay" + ); + assert!(retained.disposition_ids.is_empty(), "no model effects"); + assert_eq!(retained.unresolved_candidate_ids, vec!["candidate"]); + let reopened = crate::job_store::JobStore::open(&fixture.body.jobs_path).unwrap(); + assert_eq!(reopened.get(&before.request.job_id).unwrap(), before); +} + #[tokio::test] -async fn failed_consolidation_receipt_storage_error_has_autonomous_exact_attempt_recovery() { +async fn failed_consolidation_permanent_receipt_constraint_retains_cause_without_retry() { let mut fixture = ConsolidationRecoveryFixture::new(); fixture.confirm_execution_joined(); let before = fixture.saved.clone(); @@ -375,7 +603,7 @@ async fn failed_consolidation_receipt_storage_error_has_autonomous_exact_attempt .host .recover_consolidation_job(&fixture.jobs, &before) .unwrap_err(); - assert!(error.retryable_read_or_storage()); + assert!(!error.retryable_read_or_storage()); assert!(fixture.domain_closure().is_none()); fixture .body @@ -384,24 +612,23 @@ async fn failed_consolidation_receipt_storage_error_has_autonomous_exact_attempt connection .execute_batch("DROP TRIGGER fail_domain_closure;") .unwrap(); - // No disconnect, recover_jobs call, restart, new Job, model or finalizer. - tokio::time::timeout(std::time::Duration::from_secs(5), async { - while fixture.domain_closure().is_none() { - tokio::task::yield_now().await; - } - }) - .await - .unwrap(); - assert!(fixture.jobs.unreleased_granted().unwrap().is_empty()); + // Removing a permanent constraint does not reclassify it as a transient + // outage or refill the same owner's budget. Diagnose it, never blind retry. + tokio::time::sleep(std::time::Duration::from_millis(250)).await; + assert!(fixture.domain_closure().is_none()); assert_eq!(fixture.jobs.get(&before.request.job_id).unwrap(), before); + let owners = fixture.body.host.consolidation_recovery.lock().unwrap(); + let owner = owners.values().next().unwrap(); + assert_eq!(owner.progress, ConsolidationRecoveryProgress::Blocked); + assert_eq!(owner.passes, 1); + assert!(matches!(&owner.error, + ConsolidationRecoveryError::Service(crate::jobs::JobServiceError::Store( + crate::jobs::JobStoreError::Sqlite(rusqlite::Error::SqliteFailure(cause, _)))) + if cause.code == rusqlite::ErrorCode::ConstraintViolation)); assert!( - fixture - .body - .host - .consolidation_recovery - .lock() - .unwrap() - .is_empty() + owner + .next_action() + .contains("permanent storage or authority") ); } diff --git a/crates/workspace-server/src/server/tests/backend_job_tests.rs b/crates/workspace-server/src/server/tests/backend_job_tests.rs index 78016fade..bcabe718b 100644 --- a/crates/workspace-server/src/server/tests/backend_job_tests.rs +++ b/crates/workspace-server/src/server/tests/backend_job_tests.rs @@ -1161,6 +1161,21 @@ async fn backend_job_cleanup_commit_failure_retries_old_attempt_without_touching api.retry_backend_job(&request.job_id).is_err(), "retry must not overlap an incompletely cleaned attempt" ); + // This is a permanent SQL constraint, not a transient-storage recovery test. + // Wait for the failed owner's admission lease to leave before explicitly + // sweeping; otherwise that request can coalesce with the still-ending owner. + // Map absence here is only admission ordering, never execution/cleanup proof. + tokio::time::timeout(std::time::Duration::from_secs(10), async { + while api + .backend_job_recovery + .cleanup_owner_ids() + .contains(&first.attempt.attempt_id) + { + tokio::time::sleep(std::time::Duration::from_millis(10)).await; + } + }) + .await + .expect("permanent-failure cleanup owner must return before explicit sweep"); api.config_store .with_conn(|conn| { conn.execute_batch("DROP TRIGGER fail_job_cleanup_catalog_commit;")?; diff --git a/crates/workspace-server/src/server/tests/subjektiv_journal_connected_tests.rs b/crates/workspace-server/src/server/tests/subjektiv_journal_connected_tests.rs index 809010dc8..e9fbe1091 100644 --- a/crates/workspace-server/src/server/tests/subjektiv_journal_connected_tests.rs +++ b/crates/workspace-server/src/server/tests/subjektiv_journal_connected_tests.rs @@ -56,9 +56,27 @@ struct ConnectedJournalModel { dialogs: Arc>>, journal_inputs: Arc>>, journal_observations: Arc>>, + lifecycle_storage: Arc>>, + clock: Arc>>, + http_observations: Arc>>, + poll_observations: Arc>>, } impl ConnectedJournalModel { + fn elapsed_micros(&self) -> u64 { + self.clock.lock().unwrap().as_ref().map_or(0, |clock| { + clock.elapsed().as_micros().min(u64::MAX as u128) as u64 + }) + } + + fn observe_bounded(observations: &Mutex>, value: Value) { + let mut observations = observations.lock().unwrap(); + if observations.len() == 64 { + observations.remove(0); + } + observations.push(value); + } + fn journal_response(request: &Request) -> Vec { // Consume the actual Controller-rendered input, not a fixture generation or // a Source tool response. Decode also checks complete bodies/ranges/digests. @@ -264,6 +282,14 @@ impl LlmClient for ConnectedJournalModel { } observations.push(json!({ "observed_at": Utc::now().to_rfc3339(), + "elapsed_micros": self.elapsed_micros(), + "lifecycle_manager": self.lifecycle_storage.lock().unwrap().as_ref() + .map(|storage| { let timing = storage.lifecycle_observation(); json!({ + "acquisitions":timing.acquisitions,"sync_attempts":timing.sync_attempts, + "sync_nanoseconds":timing.sync_nanoseconds,"lock_attempts":timing.lock_attempts, + "lock_nanoseconds":timing.lock_nanoseconds,"failed_syncs":timing.failed_syncs, + "failed_locks":timing.failed_locks, + }) }), "tool_results": request.items.iter().rev().filter_map(|item| match item { Item::ToolResult { call_id, is_error, summary, .. } => Some(json!({"call_id":call_id,"is_error":is_error,"summary":summary.chars().take(256).collect::()})), @@ -352,15 +378,17 @@ async fn connected_wait_for_recording( ) { let host = JournalHostContext::validated_session(store, attr.clone()).unwrap(); tokio::time::timeout(std::time::Duration::from_secs(30), async { - let mut tick = tokio::time::interval(std::time::Duration::from_millis(25)); loop { - tick.tick().await; + let poll_started = model.elapsed_micros(); let jobs = api .store .list_backend_journal_jobs(TEST_WORKSPACE_ID, &attr.session_id, None, 100) .unwrap(); - let generation_closed = - store + let ready = !jobs.is_empty() + && jobs + .iter() + .all(|job| job.state == BackendJobState::Completed) + && store .journal_for_session(&host) .unwrap() .is_some_and(|journal| { @@ -368,22 +396,21 @@ async fn connected_wait_for_recording( .active_journal_generation(&host, &journal) .unwrap() .is_none() - }); - if !jobs.is_empty() - && generation_closed - && jobs - .iter() - .all(|job| job.state == BackendJobState::Completed) + }) && !store.journal_retention_required(&host).unwrap() && api .runtime .worker(&RuntimeWorkerRef::new(&attr.runtime_id, &attr.worker_id)) .unwrap() .worker_state - .is_some_and(|state| state.catalog_status() == protocol::WorkerStatus::Idle) - { + .is_some_and(|state| state.catalog_status() == protocol::WorkerStatus::Idle); + ConnectedJournalModel::observe_bounded(&model.poll_observations, + json!({"phase":phase,"start_micros":poll_started, + "end_micros":model.elapsed_micros()})); + if ready { break; } + tokio::time::sleep(std::time::Duration::from_millis(25)).await; } }) .await @@ -399,12 +426,20 @@ async fn connected_wait_for_recording( (job.request.job_id.clone(), history, workers) }).collect::>(); panic!( - "real Controller recording did not complete: phase={phase}; jobs={jobs:?}; attempts={attempts:?}; recovery_owners={:?}; cleanup_owners={:?}; model_observations={:?}", + "real Controller recording did not complete: phase={phase}; jobs={jobs:?}; attempts={attempts:?}; recovery_owners={:?}; cleanup_owners={:?}; model_observations={:?}; http_observations={:?}; poll_observations={:?}", api.backend_job_recovery.recovery_owner_ids(), api.backend_job_recovery.cleanup_owner_ids(), model.journal_observations.lock().unwrap(), + model.http_observations.lock().unwrap(), + model.poll_observations.lock().unwrap(), ) }); + eprintln!( + "recording progress timing: phase={phase}; model={:?}; http={:?}; polls={:?}", + model.journal_observations.lock().unwrap(), + model.http_observations.lock().unwrap(), + model.poll_observations.lock().unwrap() + ); } #[tokio::test(flavor = "multi_thread", worker_threads = 6)] @@ -457,14 +492,25 @@ async fn backend_real_runtime_journal_release_and_next_body_persisted_startup() .await .unwrap(); seed_test_delivery_binding(&api); + *model.clock.lock().unwrap() = Some(std::time::Instant::now()); + *model.lifecycle_storage.lock().unwrap() = Some(api.feature_storage.clone()); let requests = Arc::new(Mutex::new(Vec::new())); let observed = requests.clone(); + let timing_model = model.clone(); let app = build_router(api.clone()).layer(axum::middleware::from_fn( move |request: axum::extract::Request, next: axum::middleware::Next| { let observed = observed.clone(); + let timing_model = timing_model.clone(); async move { let path = request.uri().path().to_string(); + let started = timing_model.elapsed_micros(); + ConnectedJournalModel::observe_bounded(&timing_model.http_observations, + json!({"event":"ingress","path":path.chars().take(256).collect::(),"start_micros":started})); let response = next.run(request).await; + ConnectedJournalModel::observe_bounded(&timing_model.http_observations, + json!({"event":"return","path":path.chars().take(256).collect::(), + "start_micros":started,"end_micros":timing_model.elapsed_micros(), + "status":response.status().as_u16()})); observed.lock().unwrap().push((path, response.status())); response }