diff --git a/CHANGELOG.md b/CHANGELOG.md index 37573e4e..77699779 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -16,9 +16,14 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/). `respond_bytes_async` for output completion. ### Added +- **Native P3 directory-build diagnostics (#780, #753).** Owner-local filesystem + snapshots distinguish dropped waiters, running or completed native directory + builds, retained transaction receipts, and cleanup phases. Eight records per + root bound storage; omitted observations remain explicit. Launcher trace logs + report guest-exit snapshots and the independent filesystem completion result. - **Owner-local filesystem snapshots (#779, #756).** Host code can read bounded root admission, retained-work counts, cleanup phases, and recorded-failure - presence through `FsCompletion`. Launcher debug logs sample this state after + presence through `FsCompletion`. Launcher trace logs sample this state after guest exit. Reads report contention or unavailable state explicitly; guest cancellation does not imply native work or cleanup has completed. - **Wasmtime lifecycle metrics (#777, #753).** `/metrics` reports completed diff --git a/crates/cache/src/fs_work.rs b/crates/cache/src/fs_work.rs index f0b6d01d..edbb8a8e 100644 --- a/crates/cache/src/fs_work.rs +++ b/crates/cache/src/fs_work.rs @@ -4,6 +4,7 @@ use std::future::Future; use std::io; use std::panic::AssertUnwindSafe; use std::pin::Pin; +use std::sync::atomic::{AtomicU8, Ordering}; use std::sync::{Arc, Mutex, TryLockError, Weak}; use std::time::Instant; @@ -135,6 +136,75 @@ pub enum FsCleanupState { Completed, } +/// Maximum retained directory-build observations per root, including active work. +pub const FS_DIRECTORY_BUILD_CAPACITY: usize = 8; + +/// Lifetime of the selected directory-build waiter, not of the Cell or P3 call. +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub enum FsOperationWaiter { + Waiting, + /// The wrapper returned a result, including a possible execution error. + Returned, + /// The wrapper was dropped before returning. The backend can still run. + Dropped, +} + +/// Evidence from the selected blocking closure. No state implies OS cancellation. +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub enum FsOperationBackend { + /// Submitted to the retained driver; the closure has not entered. + Queued, + /// The closure entered and has not returned or unwound. + Running, + Succeeded, + Failed, + Panicked, + /// The closure was destroyed without being invoked. + NotExecuted, +} + +/// Lifetime of the existing shared transaction's root receipt. +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub enum FsOperationReceipt { + /// Includes subsequent serial phases and retained results after backend return. + Outstanding, + Released, +} + +/// One selected native directory build, identified within this root generation. +/// No paths, payloads, capability identifiers, or error text are retained. +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub struct FsDirectoryBuildSnapshot { + id: u64, + waiter: FsOperationWaiter, + backend: FsOperationBackend, + receipt: FsOperationReceipt, +} + +impl FsDirectoryBuildSnapshot { + pub fn id(&self) -> u64 { + self.id + } + pub fn waiter(&self) -> FsOperationWaiter { + self.waiter + } + pub fn backend(&self) -> FsOperationBackend { + self.backend + } + pub fn receipt(&self) -> FsOperationReceipt { + self.receipt + } + + fn reusable(&self) -> bool { + self.waiter != FsOperationWaiter::Waiting + && !matches!( + self.backend, + FsOperationBackend::Queued | FsOperationBackend::Running + ) + && self.receipt == FsOperationReceipt::Released + } +} + /// One coherent, detached read of a root's lifecycle. Contains no paths or error text. /// /// This is a point-in-time observation, not an admission or completion certificate. @@ -148,6 +218,8 @@ pub struct FsSnapshot { outstanding_work: usize, cleanup: FsCleanupState, failure_recorded: bool, + directory_builds: [Option; FS_DIRECTORY_BUILD_CAPACITY], + omitted_directory_builds: u64, } impl FsSnapshot { @@ -185,6 +257,21 @@ impl FsSnapshot { pub fn failure_recorded(&self) -> bool { self.failure_recorded } + + /// Bounded selected-operation observations, not a census of all filesystem work. + /// Fully retired records can be replaced by subsequent builds. Slot order is + /// not chronological; correlate with `id()` and this snapshot's root identity. + pub fn directory_builds( + &self, + ) -> &[Option; FS_DIRECTORY_BUILD_CAPACITY] { + &self.directory_builds + } + + /// Saturating lifetime count of builds omitted due to full slots or exhausted IDs. + /// A nonzero value means these records cannot certify absence of active builds. + pub fn omitted_directory_builds(&self) -> u64 { + self.omitted_directory_builds + } } /// A failed immediate read. No cached or partial snapshot is returned. @@ -231,6 +318,9 @@ struct Lifecycle { cleanups: BTreeMap, first_failure: Option, finished: bool, + directory_builds: [Option; FS_DIRECTORY_BUILD_CAPACITY], + next_directory_build: u64, + omitted_directory_builds: u64, } pub(crate) struct State { @@ -266,6 +356,9 @@ impl State { cleanups: BTreeMap::new(), first_failure: None, finished: false, + directory_builds: [None; FS_DIRECTORY_BUILD_CAPACITY], + next_directory_build: 0, + omitted_directory_builds: 0, }), completion_tx, #[cfg(feature = "test-support")] @@ -299,6 +392,8 @@ impl State { outstanding_work: lifecycle.work, cleanup, failure_recorded: lifecycle.first_failure.is_some(), + directory_builds: lifecycle.directory_builds, + omitted_directory_builds: lifecycle.omitted_directory_builds, }) } @@ -324,6 +419,7 @@ impl State { state: Arc::clone(self), active: true, cleanup_authorized: cleanup_bearing, + diagnostic_slots: AtomicU8::new(0), }) } @@ -580,6 +676,9 @@ pub struct FsWorkToken { state: Arc, active: bool, cleanup_authorized: bool, + // Each bit owns one diagnostic slot until this actual receipt is released. + // Registration borrows this token and holds the root lifecycle lock. + diagnostic_slots: AtomicU8, } impl fmt::Debug for FsWorkToken { @@ -593,6 +692,46 @@ impl fmt::Debug for FsWorkToken { } impl FsWorkToken { + fn observe_directory_build(&self) -> Option { + let mut lifecycle = self.state.lifecycle.lock().ok()?; + let slot = lifecycle + .directory_builds + .iter() + .position(Option::is_none) + .or_else(|| { + lifecycle + .directory_builds + .iter() + .enumerate() + .filter_map(|(slot, record)| { + record + .filter(FsDirectoryBuildSnapshot::reusable) + .map(|record| (slot, record.id)) + }) + .min_by_key(|(_, id)| *id) + .map(|(slot, _)| slot) + }); + let id = lifecycle.next_directory_build.checked_add(1); + let (Some(slot), Some(id)) = (slot, id) else { + lifecycle.omitted_directory_builds = + lifecycle.omitted_directory_builds.saturating_add(1); + return None; + }; + lifecycle.next_directory_build = id; + lifecycle.directory_builds[slot] = Some(FsDirectoryBuildSnapshot { + id, + waiter: FsOperationWaiter::Waiting, + backend: FsOperationBackend::Queued, + receipt: FsOperationReceipt::Outstanding, + }); + self.diagnostic_slots.fetch_or(1 << slot, Ordering::Relaxed); + Some(OperationObservation { + state: Arc::downgrade(&self.state), + slot, + id, + }) + } + pub(crate) fn cleanup_authorized(&self) -> bool { self.cleanup_authorized } @@ -630,6 +769,14 @@ impl Drop for FsWorkToken { .work .checked_sub(1) .expect("filesystem work token underflow"); + let mut slots = *self.diagnostic_slots.get_mut(); + while slots != 0 { + let slot = slots.trailing_zeros() as usize; + if let Some(record) = &mut lifecycle.directory_builds[slot] { + record.receipt = FsOperationReceipt::Released; + } + slots &= slots - 1; + } !lifecycle.accepting && lifecycle.work == 0 && !lifecycle.cleanup_started }; if ready { @@ -638,6 +785,86 @@ impl Drop for FsWorkToken { } } +#[derive(Clone)] +struct OperationObservation { + state: Weak, + slot: usize, + id: u64, +} + +impl OperationObservation { + fn update(&self, update: impl FnOnce(&mut FsDirectoryBuildSnapshot)) { + let Some(state) = self.state.upgrade() else { + return; + }; + let Ok(mut lifecycle) = state.lifecycle.lock() else { + return; + }; + if let Some(record) = &mut lifecycle.directory_builds[self.slot] { + if record.id == self.id { + update(record); + } + } + } +} + +struct OperationWaiter(Option); + +impl OperationWaiter { + fn returned(&mut self) { + if let Some(observation) = self.0.take() { + observation.update(|record| record.waiter = FsOperationWaiter::Returned); + } + } +} + +impl Drop for OperationWaiter { + fn drop(&mut self) { + if let Some(observation) = &self.0 { + observation.update(|record| record.waiter = FsOperationWaiter::Dropped); + } + } +} + +struct OperationBackend { + observation: Option, + entered: bool, +} + +impl OperationBackend { + fn run(mut self, operation: impl FnOnce() -> Result) -> Result { + self.entered = true; + if let Some(observation) = &self.observation { + observation.update(|record| record.backend = FsOperationBackend::Running); + } + let result = operation(); + if let Some(observation) = self.observation.take() { + observation.update(|record| { + record.backend = if result.is_ok() { + FsOperationBackend::Succeeded + } else { + FsOperationBackend::Failed + } + }); + } + result + } +} + +impl Drop for OperationBackend { + fn drop(&mut self) { + if let Some(observation) = &self.observation { + observation.update(|record| { + record.backend = if self.entered { + FsOperationBackend::Panicked + } else { + FsOperationBackend::NotExecuted + } + }); + } + } +} + pub struct FsResultEnvelope { value: Option, ownership: Option, @@ -853,6 +1080,37 @@ impl FsScope { } impl ForegroundTransaction { + /// Runs the selected native P3 directory-build slice with bounded lifecycle evidence. + /// Uses the existing retained blocking execution and receipt without adding ownership. + /// The scope must belong to this transaction to receive observations. Diagnostic + /// capacity exhaustion omits observations; it never rejects filesystem execution. + pub async fn run_directory_build( + &self, + scope: &FsScope, + operation: F, + ) -> Result>, FsWorkError> + where + T: Send + 'static, + E: Send + 'static, + F: FnOnce() -> Result + Send + 'static, + { + let observation = self + .work_for(scope) + .and_then(FsWorkToken::observe_directory_build); + let mut waiter = OperationWaiter(observation.clone()); + let backend = OperationBackend { + observation, + entered: false, + }; + let result = self + .run_blocking(FilesystemOperation::DirectoryBuild, move || { + backend.run(operation) + }) + .await; + waiter.returned(); + result + } + /// Runs an admitted, resource-free owner transaction independently of any /// waiter. The terminal callback runs only after logical/F/root ownership /// has been released, so it can remove an inflight marker and notify @@ -1078,6 +1336,174 @@ mod tests { use super::*; use crate::{HostFsOwner, HostLimits}; + #[tokio::test] + async fn directory_observation_exhaustion_and_stale_handles_do_not_change_work() { + let (host_owner, host) = HostFsOwner::start( + tokio::runtime::Handle::current(), + HostLimits::new(1, 2, 1).unwrap(), + ) + .unwrap(); + let (scope, owner, finalizer, completion) = host.try_filesystem_scope().unwrap(); + let work = scope.try_work().unwrap(); + let old = work.observe_directory_build().unwrap(); + // A destroyed, uninvoked closure is evidence of NotExecuted, regardless + // of the waiter's state. This does not change ownership admission. + drop(OperationBackend { + observation: Some(old.clone()), + entered: false, + }); + drop(OperationWaiter(Some(old.clone()))); + drop(work); + let retired = completion.try_snapshot().unwrap(); + let retired = retired.directory_builds()[0].unwrap(); + assert_eq!(retired.backend(), FsOperationBackend::NotExecuted); + assert_eq!(retired.receipt(), FsOperationReceipt::Released); + + // Fill empty slots before forcing reuse of the oldest retired slot. + for _ in 1..=FS_DIRECTORY_BUILD_CAPACITY { + let work = scope.try_work().unwrap(); + let observation = work.observe_directory_build().unwrap(); + observation.update(|record| { + record.waiter = FsOperationWaiter::Returned; + record.backend = FsOperationBackend::Succeeded; + }); + drop(work); + } + let current = completion.try_snapshot().unwrap(); + assert_ne!(current.directory_builds()[0].unwrap().id(), old.id); + old.update(|record| record.backend = FsOperationBackend::Failed); + assert_eq!( + completion.try_snapshot().unwrap().directory_builds(), + current.directory_builds() + ); + + scope.state.lifecycle.lock().unwrap().next_directory_build = u64::MAX; + let transaction = host + .try_foreground_transaction(host.try_logical_io().unwrap(), &[&scope]) + .unwrap(); + assert_eq!( + transaction + .run_directory_build(&scope, || Ok::<_, ()>(42)) + .await + .unwrap() + .into_inner(), + Ok(42) + ); + assert_eq!( + completion + .try_snapshot() + .unwrap() + .omitted_directory_builds(), + 1 + ); + scope + .state + .lifecycle + .lock() + .unwrap() + .omitted_directory_builds = u64::MAX; + assert_eq!( + transaction + .run_directory_build(&scope, || Ok::<_, ()>(43)) + .await + .unwrap() + .into_inner(), + Ok(43) + ); + assert_eq!( + completion + .try_snapshot() + .unwrap() + .omitted_directory_builds(), + u64::MAX + ); + drop(transaction); + drop(owner); + finalizer.run().await.unwrap(); + host_owner.shutdown().await.unwrap(); + } + + #[test] + fn waiter_drop_before_backend_execution_does_not_cancel_queued_work() { + use std::sync::atomic::AtomicBool; + use std::task::Poll; + let runtime = tokio::runtime::Builder::new_current_thread() + .enable_all() + .max_blocking_threads(1) + .build() + .unwrap(); + runtime.block_on(async { + let (host_owner, host) = HostFsOwner::start( + tokio::runtime::Handle::current(), + HostLimits::new(1, 2, 1).unwrap(), + ) + .unwrap(); + let (scope, owner, finalizer, completion) = host.try_filesystem_scope().unwrap(); + let transaction = host + .try_foreground_transaction(host.try_logical_io().unwrap(), &[&scope]) + .unwrap(); + let ran = Arc::new(AtomicBool::new(false)); + let witness = Arc::clone(&ran); + // An unpolled async call never submitted a backend operation. + drop(transaction.run_directory_build(&scope, move || { + witness.store(true, Ordering::SeqCst); + Ok::<_, ()>(()) + })); + assert!(completion + .try_snapshot() + .unwrap() + .directory_builds() + .iter() + .all(Option::is_none)); + assert!(!ran.load(Ordering::SeqCst)); + + let (entered_tx, entered_rx) = oneshot::channel(); + let (release_tx, release_rx) = oneshot::channel(); + let blocker = tokio::task::spawn_blocking(move || { + entered_tx.send(()).unwrap(); + release_rx.blocking_recv().unwrap(); + }); + entered_rx.await.unwrap(); + let witness = Arc::clone(&ran); + let mut waiter = Box::pin(transaction.run_directory_build(&scope, move || { + witness.store(true, Ordering::SeqCst); + Ok::<_, ()>(()) + })); + std::future::poll_fn(|cx| { + assert!(waiter.as_mut().poll(cx).is_pending()); + Poll::Ready(()) + }) + .await; + drop(waiter); + drop(transaction); + drop(owner); + let queued = completion.try_snapshot().unwrap(); + let build = queued.directory_builds()[0].unwrap(); + assert_eq!(build.backend(), FsOperationBackend::Queued); + assert_eq!(build.waiter(), FsOperationWaiter::Dropped); + assert_eq!(build.receipt(), FsOperationReceipt::Outstanding); + assert!(!ran.load(Ordering::SeqCst)); + release_tx.send(()).unwrap(); + blocker.await.unwrap(); + tokio::time::timeout(std::time::Duration::from_secs(5), completion.wait()) + .await + .unwrap() + .unwrap(); + let completed = completion.try_snapshot().unwrap(); + assert_eq!( + completed.directory_builds()[0].unwrap().backend(), + FsOperationBackend::Succeeded + ); + assert_eq!( + completed.directory_builds()[0].unwrap().receipt(), + FsOperationReceipt::Released + ); + assert!(ran.load(Ordering::SeqCst)); + finalizer.run().await.unwrap(); + host_owner.shutdown().await.unwrap(); + }); + } + #[tokio::test] async fn snapshot_returns_busy_without_waiting_for_lifecycle_lock() { let (host_owner, host) = HostFsOwner::start( @@ -1086,12 +1512,20 @@ mod tests { ) .unwrap(); let (scope, owner, finalizer, completion) = host.try_filesystem_scope().unwrap(); + let work = scope.try_work().unwrap(); + let observation = work.observe_directory_build().unwrap(); { let _guard = scope.state.lifecycle.lock().unwrap(); for _ in 0..128 { assert_eq!(completion.try_snapshot(), Err(FsSnapshotError::Busy)); } } + drop(OperationBackend { + observation: Some(observation.clone()), + entered: false, + }); + drop(OperationWaiter(Some(observation))); + drop(work); drop(owner); finalizer.run().await.unwrap(); host_owner.shutdown().await.unwrap(); @@ -1118,13 +1552,22 @@ mod tests { entered_rx.await.unwrap(); let (scope, owner, finalizer, completion) = host.try_filesystem_scope().unwrap(); + let work = scope.try_work().unwrap(); + let observation = work.observe_directory_build().unwrap(); assert!(std::panic::catch_unwind(AssertUnwindSafe(|| { let _guard = scope.state.lifecycle.lock().unwrap(); panic!("deliberate lifecycle poison"); })) .is_err()); assert_eq!(completion.try_snapshot(), Err(FsSnapshotError::Poisoned)); + // Diagnostic destructors must not turn poisoned state into a second panic. + drop(OperationBackend { + observation: Some(observation.clone()), + entered: false, + }); + drop(OperationWaiter(Some(observation))); scope.state.lifecycle.clear_poison(); + drop(work); release_tx.send(()).unwrap(); drop(owner); finalizer.run().await.unwrap(); diff --git a/crates/cache/src/lib.rs b/crates/cache/src/lib.rs index c0c1142d..7e996c86 100644 --- a/crates/cache/src/lib.rs +++ b/crates/cache/src/lib.rs @@ -11,8 +11,9 @@ pub use bloom::AtomicBloom; pub use fs_driver::{HostFsOwner, HostFsStartError}; pub use fs_work::{ CleanupPhase, CleanupRegistration, FilesystemOperation, FsCleanupState, FsCompletion, - FsFinalizer, FsOwner, FsResultEnvelope, FsRootId, FsScope, FsSnapshot, FsSnapshotError, - FsWorkError, FsWorkToken, + FsDirectoryBuildSnapshot, FsFinalizer, FsOperationBackend, FsOperationReceipt, + FsOperationWaiter, FsOwner, FsResultEnvelope, FsRootId, FsScope, FsSnapshot, FsSnapshotError, + FsWorkError, FsWorkToken, FS_DIRECTORY_BUILD_CAPACITY, }; #[cfg(feature = "test-support")] pub use fs_work::{FsTestEvent, FsTestStage}; diff --git a/crates/cache/src/resources.rs b/crates/cache/src/resources.rs index a5e40bd2..c01b5c83 100644 --- a/crates/cache/src/resources.rs +++ b/crates/cache/src/resources.rs @@ -519,4 +519,8 @@ impl ForegroundTransaction { pub(crate) fn owns_scope(&self, scope: &FsScope) -> bool { self.inner.work.iter().any(|token| token.belongs_to(scope)) } + + pub(crate) fn work_for(&self, scope: &FsScope) -> Option<&FsWorkToken> { + self.inner.work.iter().find(|token| token.belongs_to(scope)) + } } diff --git a/crates/cache/tests/fs_operation_snapshot.rs b/crates/cache/tests/fs_operation_snapshot.rs new file mode 100644 index 00000000..0013c074 --- /dev/null +++ b/crates/cache/tests/fs_operation_snapshot.rs @@ -0,0 +1,729 @@ +use std::future::Future; +use std::io; +use std::sync::atomic::{AtomicBool, Ordering}; +use std::sync::{Arc, Barrier}; +use std::task::Poll; +use std::time::Duration; + +use tokio::sync::oneshot; +use ww_cache::{ + AdmissionKind, FilesystemOperation, ForegroundTransaction, FsCleanupState, FsCompletion, + FsDirectoryBuildSnapshot, FsOperationBackend, FsOperationReceipt, FsOperationWaiter, FsScope, + FsSnapshot, FsSnapshotError, HostFsOwner, HostLimits, HostResources, + FS_DIRECTORY_BUILD_CAPACITY, +}; + +async fn deadline(future: F) -> F::Output { + tokio::time::timeout(Duration::from_secs(5), future) + .await + .expect("directory lifecycle test exceeded its deadline") +} + +async fn snapshot(completion: &FsCompletion) -> FsSnapshot { + deadline(async { + loop { + match completion.try_snapshot() { + Ok(snapshot) => return snapshot, + Err(FsSnapshotError::Busy) => tokio::task::yield_now().await, + Err(error) => panic!("unexpected snapshot error: {error:?}"), + } + } + }) + .await +} + +fn start_host(work: usize, roots: usize) -> (HostFsOwner, Arc) { + HostFsOwner::start( + tokio::runtime::Handle::current(), + HostLimits::new(work, work + 1, roots).unwrap(), + ) + .unwrap() +} + +fn admit_transaction(host: &Arc, scope: &FsScope) -> ForegroundTransaction { + host.try_foreground_transaction(host.try_logical_io().unwrap(), &[scope]) + .unwrap() +} + +fn only_operation(snapshot: &FsSnapshot) -> FsDirectoryBuildSnapshot { + let mut operations = snapshot.directory_builds().iter().flatten(); + let operation = *operations.next().expect("one directory build"); + assert!(operations.next().is_none()); + operation +} + +fn operation(snapshot: &FsSnapshot, id: u64) -> FsDirectoryBuildSnapshot { + *snapshot + .directory_builds() + .iter() + .flatten() + .find(|operation| operation.id() == id) + .expect("operation remains in the bounded snapshot") +} + +#[tokio::test] +async fn backend_return_does_not_release_a_retained_result_receipt() { + struct ResultResource(Arc); + impl Drop for ResultResource { + fn drop(&mut self) { + self.0.store(true, Ordering::SeqCst); + } + } + + let (host_owner, host) = start_host(1, 1); + let (scope, owner, finalizer, completion) = host.try_filesystem_scope().unwrap(); + let transaction = admit_transaction(&host, &scope); + let disposed = Arc::new(AtomicBool::new(false)); + let resource = ResultResource(Arc::clone(&disposed)); + let result = + deadline(transaction.run_directory_build(&scope, move || Ok::<_, io::Error>(resource))) + .await + .unwrap(); + + let returned = snapshot(&completion).await; + let operation = only_operation(&returned); + assert_eq!(operation.waiter(), FsOperationWaiter::Returned); + assert_eq!(operation.backend(), FsOperationBackend::Succeeded); + assert_eq!(operation.receipt(), FsOperationReceipt::Outstanding); + assert_eq!(returned.outstanding_work(), 1); + assert!(!disposed.load(Ordering::SeqCst)); + + drop(transaction); + drop(owner); + let retained = snapshot(&completion).await; + assert_eq!(retained.cleanup(), FsCleanupState::Pending); + assert_eq!( + only_operation(&retained).receipt(), + FsOperationReceipt::Outstanding + ); + assert!(matches!( + host.try_foreground(), + Err(error) if error.kind() == AdmissionKind::ForegroundProducer + )); + + result.dispose(); + deadline(completion.wait()).await.unwrap(); + assert!(disposed.load(Ordering::SeqCst)); + let completed = snapshot(&completion).await; + assert_eq!( + only_operation(&completed).receipt(), + FsOperationReceipt::Released + ); + assert_eq!(completed.outstanding_work(), 0); + assert_eq!(completed.cleanup(), FsCleanupState::Completed); + // The capacity witness is independent of the diagnostic fields. + drop(host.try_foreground().unwrap()); + deadline(finalizer.run()).await.unwrap(); + deadline(host_owner.shutdown()).await.unwrap(); +} + +#[tokio::test] +async fn dropped_waiter_preserves_native_execution_until_cleanup_can_start() { + let (host_owner, host) = start_host(1, 1); + let (scope, owner, finalizer, completion) = host.try_filesystem_scope().unwrap(); + let directory = tempfile::tempdir().unwrap(); + let path = directory.path().join("native-output"); + let cleanup_path = path.clone(); + let (cleanup_entered_tx, cleanup_entered_rx) = oneshot::channel(); + let (cleanup_release_tx, cleanup_release_rx) = oneshot::channel(); + scope + .own_cleanup(FilesystemOperation::ProcessCleanup, move || { + assert_eq!( + std::fs::read(&cleanup_path)?, + b"native work survived waiter drop" + ); + cleanup_entered_tx.send(()).unwrap(); + cleanup_release_rx.blocking_recv().unwrap(); + std::fs::remove_file(cleanup_path) + }) + .unwrap(); + let transaction = admit_transaction(&host, &scope); + let operation_scope = scope.clone(); + let operation_path = path.clone(); + let (entered_tx, entered_rx) = oneshot::channel(); + let (release_tx, release_rx) = oneshot::channel(); + let waiter = tokio::spawn(async move { + transaction + .run_directory_build(&operation_scope, move || { + entered_tx.send(()).unwrap(); + release_rx.blocking_recv().unwrap(); + std::fs::write(operation_path, b"native work survived waiter drop") + }) + .await + }); + deadline(entered_rx).await.unwrap(); + waiter.abort(); + assert!(deadline(waiter).await.err().unwrap().is_cancelled()); + drop(owner); + + let cancelled = snapshot(&completion).await; + let operation = only_operation(&cancelled); + assert_eq!(operation.waiter(), FsOperationWaiter::Dropped); + assert_eq!(operation.backend(), FsOperationBackend::Running); + assert_eq!(operation.receipt(), FsOperationReceipt::Outstanding); + assert_eq!(cancelled.outstanding_work(), 1); + assert!(!cancelled.root_admission_open()); + assert_eq!(cancelled.cleanup(), FsCleanupState::Pending); + assert!(!path.exists()); + assert!(matches!( + host.try_foreground(), + Err(error) if error.kind() == AdmissionKind::ForegroundProducer + )); + assert_eq!(scope.try_work().unwrap_err().kind(), AdmissionKind::Closed); + + release_tx.send(()).unwrap(); + deadline(cleanup_entered_rx).await.unwrap(); + let cleaning = snapshot(&completion).await; + let operation = only_operation(&cleaning); + assert_eq!(operation.waiter(), FsOperationWaiter::Dropped); + assert_eq!(operation.backend(), FsOperationBackend::Succeeded); + assert_eq!(operation.receipt(), FsOperationReceipt::Released); + assert_eq!(cleaning.outstanding_work(), 0); + assert_eq!(cleaning.cleanup(), FsCleanupState::Running); + assert!(path.exists()); + let mut completion_wait = Box::pin(completion.wait()); + std::future::poll_fn(|cx| { + assert!(completion_wait.as_mut().poll(cx).is_pending()); + Poll::Ready(()) + }) + .await; + + cleanup_release_tx.send(()).unwrap(); + deadline(completion_wait).await.unwrap(); + assert!(!path.exists()); + assert_eq!( + snapshot(&completion).await.cleanup(), + FsCleanupState::Completed + ); + deadline(finalizer.run()).await.unwrap(); + deadline(host_owner.shutdown()).await.unwrap(); +} + +#[tokio::test] +async fn backend_error_is_visible_without_changing_root_failure_semantics() { + let (host_owner, host) = start_host(1, 1); + let (scope, owner, finalizer, completion) = host.try_filesystem_scope().unwrap(); + let transaction = admit_transaction(&host, &scope); + let error = deadline(transaction.run_directory_build(&scope, || { + Err::<(), _>(io::Error::other("/private/capability-sensitive-path")) + })) + .await + .unwrap() + .into_inner() + .unwrap_err(); + assert_eq!(error.kind(), io::ErrorKind::Other); + let failed = snapshot(&completion).await; + assert_eq!( + only_operation(&failed).backend(), + FsOperationBackend::Failed + ); + assert_eq!( + only_operation(&failed).waiter(), + FsOperationWaiter::Returned + ); + assert!(!failed.failure_recorded()); + assert!(!failed.cleanup_poisoned()); + assert!(!format!("{failed:?}").contains("capability-sensitive-path")); + drop(transaction); + drop(owner); + deadline(completion.wait()).await.unwrap(); + deadline(finalizer.run()).await.unwrap(); + deadline(host_owner.shutdown()).await.unwrap(); +} + +#[tokio::test] +async fn backend_panic_keeps_the_existing_recorded_failure() { + let (host_owner, host) = start_host(1, 1); + let (scope, owner, finalizer, completion) = host.try_filesystem_scope().unwrap(); + let transaction = admit_transaction(&host, &scope); + let result = deadline( + transaction.run_directory_build(&scope, || -> io::Result<()> { + panic!("deliberate native backend panic") + }), + ) + .await; + assert!(result.is_err()); + let failed = snapshot(&completion).await; + assert_eq!( + only_operation(&failed).backend(), + FsOperationBackend::Panicked + ); + assert_eq!( + only_operation(&failed).waiter(), + FsOperationWaiter::Returned + ); + assert!(failed.failure_recorded()); + assert!(!failed.cleanup_poisoned()); + drop(transaction); + drop(owner); + assert!(deadline(completion.wait()).await.is_err()); + assert!(deadline(finalizer.run()).await.is_err()); + deadline(host_owner.shutdown()).await.unwrap(); +} + +#[tokio::test] +async fn independent_operations_keep_distinct_waiter_backend_and_receipt_states() { + let (host_owner, host) = start_host(2, 1); + let (scope, owner, finalizer, completion) = host.try_filesystem_scope().unwrap(); + let first_transaction = admit_transaction(&host, &scope); + let first_scope = scope.clone(); + let (first_entered_tx, first_entered_rx) = oneshot::channel(); + let (first_release_tx, first_release_rx) = oneshot::channel(); + let first_finished = Arc::new(AtomicBool::new(false)); + let first_flag = Arc::clone(&first_finished); + let first = tokio::spawn(async move { + first_transaction + .run_directory_build(&first_scope, move || { + first_entered_tx.send(()).unwrap(); + first_release_rx.blocking_recv().unwrap(); + first_flag.store(true, Ordering::SeqCst); + Ok::<_, io::Error>(11) + }) + .await + }); + deadline(first_entered_rx).await.unwrap(); + let first_id = only_operation(&snapshot(&completion).await).id(); + let second_transaction = admit_transaction(&host, &scope); + let second_scope = scope.clone(); + let (second_entered_tx, second_entered_rx) = oneshot::channel(); + let (second_release_tx, second_release_rx) = oneshot::channel(); + let second = tokio::spawn(async move { + second_transaction + .run_directory_build(&second_scope, move || { + second_entered_tx.send(()).unwrap(); + second_release_rx.blocking_recv().unwrap(); + Ok::<_, io::Error>(22) + }) + .await + }); + deadline(second_entered_rx).await.unwrap(); + let concurrent = snapshot(&completion).await; + assert_eq!(concurrent.outstanding_work(), 2); + let second_id = concurrent + .directory_builds() + .iter() + .flatten() + .find(|operation| operation.id() != first_id) + .unwrap() + .id(); + first.abort(); + assert!(deadline(first).await.err().unwrap().is_cancelled()); + second_release_tx.send(()).unwrap(); + let second_result = deadline(second).await.unwrap().unwrap(); + let mixed = snapshot(&completion).await; + assert_eq!( + operation(&mixed, first_id).waiter(), + FsOperationWaiter::Dropped + ); + assert_eq!( + operation(&mixed, first_id).backend(), + FsOperationBackend::Running + ); + assert_eq!( + operation(&mixed, second_id).waiter(), + FsOperationWaiter::Returned + ); + assert_eq!( + operation(&mixed, second_id).backend(), + FsOperationBackend::Succeeded + ); + assert_eq!( + operation(&mixed, second_id).receipt(), + FsOperationReceipt::Outstanding + ); + assert!(!first_finished.load(Ordering::SeqCst)); + assert_eq!(second_result.into_inner().unwrap(), 22); + let one_left = snapshot(&completion).await; + assert_eq!(one_left.outstanding_work(), 1); + assert_eq!( + operation(&one_left, second_id).receipt(), + FsOperationReceipt::Released + ); + assert_eq!( + operation(&one_left, first_id).receipt(), + FsOperationReceipt::Outstanding + ); + drop(owner); + first_release_tx.send(()).unwrap(); + deadline(completion.wait()).await.unwrap(); + assert!(first_finished.load(Ordering::SeqCst)); + assert_eq!( + operation(&snapshot(&completion).await, first_id).receipt(), + FsOperationReceipt::Released + ); + deadline(finalizer.run()).await.unwrap(); + deadline(host_owner.shutdown()).await.unwrap(); +} + +#[tokio::test] +async fn full_observation_slots_do_not_reject_work_or_replace_live_receipts() { + let (host_owner, host) = start_host(1, 1); + let (scope, owner, finalizer, completion) = host.try_filesystem_scope().unwrap(); + let transaction = admit_transaction(&host, &scope); + for value in 0..FS_DIRECTORY_BUILD_CAPACITY { + let actual = + deadline(transaction.run_directory_build(&scope, move || Ok::<_, io::Error>(value))) + .await + .unwrap() + .into_inner() + .unwrap(); + assert_eq!(actual, value); + } + let full = snapshot(&completion).await; + let ids: Vec<_> = full + .directory_builds() + .iter() + .flatten() + .map(|record| record.id()) + .collect(); + assert_eq!(ids.len(), FS_DIRECTORY_BUILD_CAPACITY); + assert_eq!(full.omitted_directory_builds(), 0); + assert!(full + .directory_builds() + .iter() + .flatten() + .all(|record| record.receipt() == FsOperationReceipt::Outstanding)); + // These serial operations share one transaction receipt. They must not be + // fabricated from the aggregate receipt count. + assert_eq!(full.outstanding_work(), 1); + let executed = Arc::new(AtomicBool::new(false)); + let execution_flag = Arc::clone(&executed); + deadline(transaction.run_directory_build(&scope, move || { + execution_flag.store(true, Ordering::SeqCst); + Ok::<_, io::Error>(()) + })) + .await + .unwrap() + .into_inner() + .unwrap(); + assert!(executed.load(Ordering::SeqCst)); + let omitted = snapshot(&completion).await; + assert_eq!(omitted.omitted_directory_builds(), 1); + assert_eq!(omitted.directory_builds(), full.directory_builds()); + drop(transaction); + let released = snapshot(&completion).await; + assert!(released + .directory_builds() + .iter() + .flatten() + .all(|record| record.receipt() == FsOperationReceipt::Released)); + + let replacement = admit_transaction(&host, &scope); + deadline(replacement.run_directory_build(&scope, || Ok::<_, io::Error>(()))) + .await + .unwrap() + .dispose(); + let reused = snapshot(&completion).await; + assert_eq!( + reused.directory_builds().iter().flatten().count(), + FS_DIRECTORY_BUILD_CAPACITY + ); + assert_eq!(reused.omitted_directory_builds(), 1); + let fresh = reused + .directory_builds() + .iter() + .flatten() + .find(|record| !ids.contains(&record.id())) + .unwrap(); + assert!(fresh.id() > *ids.iter().max().unwrap()); + assert_eq!(fresh.receipt(), FsOperationReceipt::Outstanding); + assert!(reused + .directory_builds() + .iter() + .flatten() + .filter(|record| record.id() != fresh.id()) + .all(|record| record.receipt() == FsOperationReceipt::Released)); + drop(replacement); + drop(owner); + deadline(completion.wait()).await.unwrap(); + deadline(finalizer.run()).await.unwrap(); + deadline(host_owner.shutdown()).await.unwrap(); +} + +#[tokio::test] +async fn receipt_release_makes_cleanup_ready_before_reaper_claims_it() { + let (host_owner, host) = start_host(1, 2); + let (blocker, blocker_owner, blocker_finalizer, _) = host.try_filesystem_scope().unwrap(); + let (blocker_entered_tx, blocker_entered_rx) = oneshot::channel(); + let (blocker_release_tx, blocker_release_rx) = oneshot::channel(); + blocker + .own_cleanup(FilesystemOperation::ProcessCleanup, move || { + blocker_entered_tx.send(()).unwrap(); + blocker_release_rx.blocking_recv().unwrap(); + Ok(()) + }) + .unwrap(); + drop(blocker_owner); + deadline(blocker_entered_rx).await.unwrap(); + let (scope, owner, finalizer, completion) = host.try_filesystem_scope().unwrap(); + let (cleanup_entered_tx, cleanup_entered_rx) = oneshot::channel(); + let (cleanup_release_tx, cleanup_release_rx) = oneshot::channel(); + scope + .own_cleanup(FilesystemOperation::ProcessCleanup, move || { + cleanup_entered_tx.send(()).unwrap(); + cleanup_release_rx.blocking_recv().unwrap(); + Ok(()) + }) + .unwrap(); + let transaction = admit_transaction(&host, &scope); + deadline(transaction.run_directory_build(&scope, || Ok::<_, io::Error>(()))) + .await + .unwrap() + .dispose(); + drop(owner); + let retained = snapshot(&completion).await; + assert_eq!(retained.cleanup(), FsCleanupState::Pending); + assert_eq!( + only_operation(&retained).backend(), + FsOperationBackend::Succeeded + ); + drop(transaction); + let ready = snapshot(&completion).await; + assert_eq!(ready.cleanup(), FsCleanupState::Ready); + assert_eq!( + only_operation(&ready).receipt(), + FsOperationReceipt::Released + ); + assert_eq!(ready.outstanding_work(), 0); + blocker_release_tx.send(()).unwrap(); + deadline(cleanup_entered_rx).await.unwrap(); + assert_eq!( + snapshot(&completion).await.cleanup(), + FsCleanupState::Running + ); + cleanup_release_tx.send(()).unwrap(); + deadline(completion.wait()).await.unwrap(); + assert_eq!( + snapshot(&completion).await.cleanup(), + FsCleanupState::Completed + ); + deadline(finalizer.run()).await.unwrap(); + deadline(blocker_finalizer.run()).await.unwrap(); + deadline(host_owner.shutdown()).await.unwrap(); +} + +#[tokio::test] +async fn operation_records_do_not_cross_reused_root_generations() { + let (host_owner, host) = start_host(1, 1); + let (old_scope, old_owner, old_finalizer, old_completion) = + host.try_filesystem_scope().unwrap(); + let old_transaction = admit_transaction(&host, &old_scope); + deadline(old_transaction.run_directory_build(&old_scope, || Ok::<_, io::Error>(()))) + .await + .unwrap() + .dispose(); + drop(old_transaction); + drop(old_owner); + deadline(old_finalizer.run()).await.unwrap(); + let old_snapshot = snapshot(&old_completion).await; + let (scope, owner, finalizer, completion) = deadline(async { + loop { + match host.try_filesystem_scope() { + Ok(root) => break root, + Err(error) if error.kind() == AdmissionKind::CleanupObligation => { + tokio::task::yield_now().await + } + Err(error) => panic!("unexpected admission error: {error}"), + } + } + }) + .await; + let empty = snapshot(&completion).await; + assert_ne!(old_snapshot.root(), empty.root()); + assert_eq!( + empty.root().generation(), + old_snapshot.root().generation() + 1 + ); + assert!(empty.directory_builds().iter().all(Option::is_none)); + assert_eq!(empty.omitted_directory_builds(), 0); + let transaction = admit_transaction(&host, &scope); + deadline(transaction.run_directory_build(&scope, || { + Err::<(), _>(io::Error::other("new root failure")) + })) + .await + .unwrap() + .dispose(); + let current = snapshot(&completion).await; + assert_eq!( + only_operation(¤t).backend(), + FsOperationBackend::Failed + ); + assert_eq!( + only_operation(&snapshot(&old_completion).await).backend(), + FsOperationBackend::Succeeded + ); + assert_eq!( + only_operation(&old_snapshot).receipt(), + FsOperationReceipt::Released + ); + drop(transaction); + drop(owner); + deadline(finalizer.run()).await.unwrap(); + deadline(host_owner.shutdown()).await.unwrap(); + drop(old_scope); + assert_eq!( + old_completion.try_snapshot(), + Err(FsSnapshotError::Unavailable) + ); + assert_eq!(snapshot(&completion).await.root(), current.root()); +} + +#[tokio::test] +async fn owner_shutdown_waits_for_native_resource_disposal_and_cleanup() { + struct NativeResult { + file: Option, + released: Arc, + } + impl Drop for NativeResult { + fn drop(&mut self) { + drop(self.file.take()); + self.released.store(true, Ordering::SeqCst); + } + } + + let (host_owner, host) = start_host(1, 1); + let (scope, owner, finalizer, completion) = host.try_filesystem_scope().unwrap(); + let directory = tempfile::tempdir().unwrap(); + let path = directory.path().join("native-result"); + let released = Arc::new(AtomicBool::new(false)); + let result = NativeResult { + file: Some(std::fs::File::create(&path).unwrap()), + released: Arc::clone(&released), + }; + let cleanup_path = path.clone(); + let cleanup_released = Arc::clone(&released); + scope + .own_cleanup(FilesystemOperation::ProcessCleanup, move || { + assert!( + cleanup_released.load(Ordering::SeqCst), + "native result must be disposed before cleanup" + ); + std::fs::remove_file(cleanup_path) + }) + .unwrap(); + let transaction = admit_transaction(&host, &scope); + let operation_scope = scope.clone(); + let (entered_tx, entered_rx) = oneshot::channel(); + let (release_tx, release_rx) = oneshot::channel(); + let waiter = tokio::spawn(async move { + transaction + .run_directory_build(&operation_scope, move || { + entered_tx.send(()).unwrap(); + release_rx.blocking_recv().unwrap(); + Ok::<_, io::Error>(result) + }) + .await + }); + deadline(entered_rx).await.unwrap(); + waiter.abort(); + assert!(deadline(waiter).await.err().unwrap().is_cancelled()); + drop(owner); + drop(finalizer); + let mut shutdown = Box::pin(host_owner.shutdown()); + std::future::poll_fn(|cx| { + assert!(shutdown.as_mut().poll(cx).is_pending()); + Poll::Ready(()) + }) + .await; + assert!(!released.load(Ordering::SeqCst)); + assert!(path.exists()); + assert_eq!( + only_operation(&snapshot(&completion).await).backend(), + FsOperationBackend::Running + ); + release_tx.send(()).unwrap(); + deadline(shutdown).await.unwrap(); + assert!(released.load(Ordering::SeqCst)); + assert!(!path.exists()); + let stopped = snapshot(&completion).await; + assert_eq!(stopped.cleanup(), FsCleanupState::Completed); + assert_eq!(stopped.outstanding_work(), 0); + assert_eq!( + only_operation(&stopped).receipt(), + FsOperationReceipt::Released + ); + drop(scope); + assert_eq!(completion.try_snapshot(), Err(FsSnapshotError::Unavailable)); + deadline(completion.wait()).await.unwrap(); +} + +#[tokio::test] +async fn snapshots_remain_coherent_when_backend_completion_races_observation() { + let (host_owner, host) = start_host(1, 1); + let (scope, owner, finalizer, completion) = host.try_filesystem_scope().unwrap(); + let transaction = admit_transaction(&host, &scope); + let operation_scope = scope.clone(); + let (entered_tx, entered_rx) = oneshot::channel(); + let (release_tx, release_rx) = oneshot::channel(); + let waiter = tokio::spawn(async move { + transaction + .run_directory_build(&operation_scope, move || { + entered_tx.send(()).unwrap(); + release_rx.blocking_recv().unwrap(); + Ok::<_, io::Error>(()) + }) + .await + }); + deadline(entered_rx).await.unwrap(); + let before = snapshot(&completion).await; + assert_eq!( + only_operation(&before).backend(), + FsOperationBackend::Running + ); + waiter.abort(); + assert!(deadline(waiter).await.err().unwrap().is_cancelled()); + drop(owner); + let barrier = Arc::new(Barrier::new(2)); + let reader_barrier = Arc::clone(&barrier); + let reader_completion = completion.clone(); + let (reader_done_tx, reader_done_rx) = oneshot::channel(); + let reader = std::thread::spawn(move || { + reader_barrier.wait(); + for _ in 0..10_000 { + match reader_completion.try_snapshot() { + Ok(snapshot) => { + let operation = only_operation(&snapshot); + assert_eq!(operation.waiter(), FsOperationWaiter::Dropped); + match operation.receipt() { + FsOperationReceipt::Outstanding => { + assert_eq!(snapshot.outstanding_work(), 1); + assert_eq!(snapshot.cleanup(), FsCleanupState::Pending); + } + FsOperationReceipt::Released => { + assert_eq!(snapshot.outstanding_work(), 0); + assert_eq!(operation.backend(), FsOperationBackend::Succeeded); + } + } + if operation.backend() == FsOperationBackend::Running { + assert_eq!(operation.receipt(), FsOperationReceipt::Outstanding); + } + } + Err(FsSnapshotError::Busy) => {} + Err(error) => panic!("unexpected concurrent snapshot error: {error:?}"), + } + } + reader_done_tx.send(()).unwrap(); + }); + barrier.wait(); + release_tx.send(()).unwrap(); + deadline(completion.wait()).await.unwrap(); + deadline(reader_done_rx).await.unwrap(); + reader.join().unwrap(); + let after = snapshot(&completion).await; + assert_eq!(after.cleanup(), FsCleanupState::Completed); + assert_eq!( + only_operation(&after).backend(), + FsOperationBackend::Succeeded + ); + assert_eq!( + only_operation(&after).receipt(), + FsOperationReceipt::Released + ); + // Earlier detached observations cannot change when the backend completes. + assert_eq!( + only_operation(&before).backend(), + FsOperationBackend::Running + ); + deadline(finalizer.run()).await.unwrap(); + deadline(host_owner.shutdown()).await.unwrap(); +} diff --git a/crates/cell/src/fs_intercept.rs b/crates/cell/src/fs_intercept.rs index 74b76ff7..01ac792e 100644 --- a/crates/cell/src/fs_intercept.rs +++ b/crates/cell/src/fs_intercept.rs @@ -550,7 +550,7 @@ async fn materialize_resolved_descriptor( let cleanup_hook = tests::take_directory_cleanup_hook(); let build_scope = filesystem_scope.clone(); transaction - .run_blocking(cache::FilesystemOperation::DirectoryBuild, move || { + .run_directory_build(filesystem_scope, move || { build_directory( staging_root, publish_path, diff --git a/doc/filesystem-snapshots.md b/doc/filesystem-snapshots.md index 3977a633..d6c4e1d7 100644 --- a/doc/filesystem-snapshots.md +++ b/doc/filesystem-snapshots.md @@ -20,10 +20,10 @@ native Wasmtime filesystem work. `ForegroundTransaction` continuations share receipts; result envelopes can retain those receipts after a backend returns. `FsCompletion::wait()` retains the existing frozen completion result. -The snapshot copies existing fields. It adds no operation counters or lifecycle -recording. The owner-local publication principle does not require Osaka's TLS -enrollment, grant routing, or worker counter sharding. Existing fuel collectors -remain independent of this API. +The snapshot copies existing ownership fields and bounded directory-build records +from the same lifecycle mutex. It uses no TLS enrollment, grant routing, worker +counter sharding, or separate operation registry. Existing fuel collectors remain +independent of this API. ## Contract @@ -39,6 +39,8 @@ remain independent of this API. | `cleanup()` | `Pending`, `Ready`, `Running`, or `Completed`, as defined below. | | `cleanup_poisoned()` | A recorded cleanup failure poisoned new cleanup-bearing admission. | | `failure_recorded()` | Whether the root recorded any failure. No failure text is copied. | +| `directory_builds()` | Eight optional selected native directory-build records; see below. | +| `omitted_directory_builds()` | Saturating lifetime count of observations omitted because all slots were occupied or operation IDs were exhausted. | Cleanup phase derives from the existing fields, in this order: @@ -57,8 +59,71 @@ can differ from the frozen `wait()` result. Guest cancellation does not prove native completion. Zero work does not prove cleanup completion. A nonzero receipt count can describe retained results after native completion. Ordinary I/O errors are not necessarily recorded failures. -The API cannot identify individual operations, guest cancellation requests, -backend start/end times, or the cause of a failure. +The API identifies individual directory builds within a root generation. It does +not observe guest cancellation requests, backend timestamps, or failure text. + +## Native P3 directory-build slice + +P3 `open-at` materializes a CID-backed directory through `build_directory` in +`crates/cell/src/fs_intercept.rs`. The selected closure creates a temporary +directory, populates entries, then publishes with `renameat_with`. It performs +rollback before returning an error. `ForegroundTransaction::run_directory_build` +observes this closure through the existing retained `run_blocking` execution. + +Each `FsDirectoryBuildSnapshot` reports four independent facts: + +| Accessor | Producer and meaning | +| --- | --- | +| `id()` | Monotonic operation sequence within this root generation. Correlate with `root()`, not across host owners. | +| `waiter()` | `Waiting` until the selected wrapper returns (`Returned`) or is dropped first (`Dropped`). This describes the directory-build waiter, not the entire P3 call or Cell. | +| `backend()` | `Queued` before closure entry, `Running` during the closure, `Succeeded` or `Failed` after the closure returns, `Panicked` on unwind, or `NotExecuted` when the closure is destroyed without entry. | +| `receipt()` | `Outstanding` until the existing shared transaction's root receipt is dropped, then `Released`. Subsequent serial phases or result envelopes can retain the receipt after backend completion. | + +```mermaid +flowchart TD + A[Admitted foreground transaction] --> Q[Directory build queued] + Q --> B[Blocking closure running] + Q --> W[Waiter waiting] + W --> D[Waiter dropped] + W --> R[Waiter returned] + B --> E[Backend returned or unwound] + E --> F[Result delivered or disposed] + F --> G[Final transaction receipt released] + C[Root admission closed] --> H[All root receipts released: cleanup Ready] + G --> H + H --> I[Reaper claims cleanup: Running] + I --> J[Callbacks finish: completion recorded and published] + J --> K[Root slot and cleanup obligation released] +``` + +Dropping the waiter does not cancel retained backend work. Dropping the async +wrapper before its first poll submits no work and creates no record. Ordinary +backend errors set `Failed` without changing existing root-failure semantics. +Panic handling and root failure recording remain those of `run_blocking`. + +Guest cancellation currently does not cancel this selected native backend +operation. [#785](https://github.com/wetware/ww/issues/785) owns the correctness +decision and any cancellation changes. This diagnostic observes existing behavior; +continued execution is not evidence that an independent Host owner needs it. + +The pinned Wasmtime `FilesystemOperation` hook exposes admission and final +opaque-owner release only. A stream session or retained result can outlive a +backend return. `NativeFsLease` therefore cannot supply these backend phases; +delegated native operations remain outside this slice. No aggregate receipt +count is interpreted as an individual operation. + +The eight slots belong to the existing root `Lifecycle`. Every receipt carries +an eight-bit slot mask. `FsWorkToken::drop` marks its records `Released` under +the same mutex that decrements `outstanding_work`. Observation handles hold weak +references and add no receipt, permit, or cleanup authority. + +A slot remains reserved until its waiter and backend terminate and its receipt +is released. New builds replace the oldest fully retired record when no empty +slot remains. Full slots or exhausted sequence IDs omit diagnostics without +rejecting execution. A nonzero omission count means the records cannot certify +absence of active builds. Replacement does not increment the omission count; +the records are bounded observations, not a complete event history. Root reuse +starts empty records and a new root generation. ## Consistency, freshness, and resource bounds @@ -79,7 +144,7 @@ The call returns one of three errors instead of cached or partial data: available through `wait()` independently. - `Poisoned`: the mutex cannot provide certified state. -There is no stale fallback, timeout queue, root scan, background task, history, +There is no stale fallback, timeout queue, root scan, background task, event history, or diagnostic allocation per read. Only one reader holds the lifecycle mutex at a time. Other concurrent reads return `Busy`. The critical section copies a constant number of fields and captures one clock value. It does not await, @@ -103,20 +168,39 @@ the reaper releases the cleanup obligation through its separate mutex. `OwnedChildLifecycle::run_until_shutdown` in `src/launcher.rs` reads the existing completion handle after guest exit and before waiting for filesystem cleanup. -With launcher debug tracing enabled, one event contains the snapshot or explicit -read error. For example, set `RUST_LOG=ww::launcher=debug` for the host process. +With launcher trace logging enabled, one event contains the snapshot or explicit +read error. For example, set `RUST_LOG=ww::launcher=trace` for the host process. The tracing macro does not evaluate the read when the callsite is disabled. -This is a single lifecycle observation, not a polling service. - -The event contains only the owner-local identity, timestamp, receipt count, -phase, and booleans. It does not contain paths, CIDs, capabilities, or error text. +After `wait()` resolves, a second event reports +`executor: child filesystem completion observed` with `succeeded`. Both events +inherit the existing Cell correlation span. No polling service is added. +Both filesystem events use trace level; normal completion emits no additional +info/debug filesystem event. The typed snapshot remains available on demand. +Cleanup certification failures retain their existing error event. + +The events contain owner-local identity, timestamp, receipt count, typed phases, +bounded operation records, omission count, and booleans. They contain no paths, +CIDs, capabilities, or error text. Root identities are not Prometheus labels. The event does not change the existing completion wait or delegated-authority release sequence. +For example, a guest-exit snapshot containing `waiter: Dropped`, +`backend: Running`, and `receipt: Outstanding` proves that selected backend work +survived its waiter. A later typed read can show `Succeeded` with cleanup +`Running`. The completion event reports the independent final wait result even +when the weak snapshot source has become `Unavailable`. Snapshot reads can +return `Busy`; consumers must preserve that uncertainty. + +The real compiled P3 test +`native_p3_directory_cancellation_distinguishes_backend_and_cleanup_completion` +gates directory publication, kills the Cell, verifies Store destruction, then +holds cleanup separately. Native file contents and deletion independently +witness publication and cleanup. Cache tests cover queued cancellation, +failure, panic, concurrency, retention, bounds, generation reuse, and shutdown. + ## Extension boundary -[#780](https://github.com/wetware/ww/issues/780) can reuse this read contract and -existing root ownership handles when adding authoritative native-P3 operation -admission, cancellation, and backend-completion evidence. Those distinctions need -evidence at the native operation boundary; timestamps and receipt counts cannot -reconstruct them. The broader agent interface remains under #756. +[#780](https://github.com/wetware/ww/issues/780) extends #779's read contract for +one directory-build lifecycle. Broader filesystem, stream, and transport +operations, bytes/backpressure, memory/resource visibility, trap classification, +and profiling remain under #753. The broader agent interface remains under #756. diff --git a/src/launcher.rs b/src/launcher.rs index 0e68937f..f4a351cf 100644 --- a/src/launcher.rs +++ b/src/launcher.rs @@ -409,7 +409,7 @@ impl OwnedChildLifecycle { let filesystem_completion = self.filesystem_completion.clone(); // The tracing macro evaluates this bounded read only when enabled. // Guest exit can precede native completion; do not infer completion here. - tracing::debug!( + tracing::trace!( filesystem = ?filesystem_completion.try_snapshot(), "executor: child filesystem ownership after guest exit" ); @@ -422,6 +422,10 @@ impl OwnedChildLifecycle { false } }; + tracing::trace!( + succeeded = filesystem_completed, + "executor: child filesystem completion observed" + ); // Release the delegated Membrane after child RPC and the stored guest // bootstrap are gone but before reporting exit to the parent. drop(self.membrane.take()); @@ -1924,6 +1928,238 @@ mod tests { drop(pool); } + async fn operation_snapshot(completion: &cache::FsCompletion) -> cache::FsSnapshot { + tokio::time::timeout(Duration::from_secs(10), async { + loop { + match completion.try_snapshot() { + Ok(snapshot) => break snapshot, + Err(cache::FsSnapshotError::Busy) => tokio::task::yield_now().await, + Err(error) => panic!("operation snapshot failed: {error}"), + } + } + }) + .await + .expect("operation snapshot deadline") + } + + #[test] + fn native_p3_directory_cancellation_distinguishes_backend_and_cleanup_completion() { + use cache::{FsCleanupState, FsOperationBackend, FsOperationReceipt, FsOperationWaiter}; + use std::sync::atomic::{AtomicBool, Ordering}; + + let (_shutdown_tx, shutdown_rx) = tokio::sync::watch::channel(()); + let pool = crate::services::ExecutorPool::new(1, shutdown_rx, test_filesystem_host()); + let runtime_engine = pool.runtime_engine(); + let component = native_kernel_fixture_component(&runtime_engine); + let (build_entered_tx, build_entered_rx) = tokio::sync::oneshot::channel(); + let build_entered_tx = Arc::new(std::sync::Mutex::new(Some(build_entered_tx))); + let (build_release_tx, build_release_rx) = std::sync::mpsc::channel(); + let build_release_rx = Arc::new(std::sync::Mutex::new(Some(build_release_rx))); + let hook: Arc = Arc::new(move |event| { + if event.operation == cache::FilesystemOperation::DirectoryBuild + && event.stage == cache::FsTestStage::BeforePublish + { + build_entered_tx + .lock() + .unwrap() + .take() + .unwrap() + .send(event.path.unwrap()) + .unwrap(); + build_release_rx + .lock() + .unwrap() + .take() + .unwrap() + .recv_timeout(Duration::from_secs(10)) + .unwrap(); + } + }); + let (cancelled_tx, cancelled_rx) = std::sync::mpsc::channel(); + let (cleanup_observed_tx, cleanup_observed_rx) = std::sync::mpsc::channel(); + let (cleanup_release_tx, cleanup_release_rx) = std::sync::mpsc::channel(); + let (finished_tx, finished_rx) = std::sync::mpsc::channel(); + let cleanup_finished = Arc::new(AtomicBool::new(false)); + let cleanup_witness = Arc::clone(&cleanup_finished); + assert!(pool + .spawn(crate::services::SpawnRequest { + name: "native-directory-cancellation".into(), + factory: Box::new(move |_| { + Box::pin(async move { + use std::future::Future; + use std::task::Poll; + + let (mut lifecycle, process, dropped, _, observer) = + lifecycle_loss_fixture(); + if let Some(old_rpc) = lifecycle.rpc_task.take() { + old_rpc.shutdown(); + } + let filesystem = test_filesystem_ownership(); + let scope = filesystem.test_scope().clone(); + let (tree_staging, tree, cache) = + native_filesystem_fixture(&filesystem, None).await; + let nested = tree_staging + .path() + .join(format!("dir-{}", native_fixture_cid(2))); + let (cleanup_entered_tx, cleanup_entered_rx) = + tokio::sync::oneshot::channel(); + let cleanup_path = nested.clone(); + scope + .own_cleanup(cache::FilesystemOperation::ProcessCleanup, move || { + // This witness reads the native filesystem independently + // of the new operation diagnostic. + assert_eq!(cleanup_path.join("child").metadata()?.len(), 12); + assert_eq!(std::fs::read_dir(&cleanup_path)?.count(), 2_049); + cleanup_entered_tx.send(()).unwrap(); + cleanup_release_rx + .recv_timeout(Duration::from_secs(10)) + .unwrap(); + std::fs::remove_dir_all(cleanup_path)?; + cleanup_witness.store(true, Ordering::SeqCst); + Ok(()) + }) + .unwrap(); + let (_epoch_tx, epoch_rx) = tokio::sync::watch::channel(authority::Epoch { + seq: 1, + head: Vec::new(), + root: None, + }); + let readiness = Arc::new(authority::KernelReadyGate::new(epoch_rx)); + readiness.bind_generation(1); + let (mut builder, mut handles) = Builder::kernel( + Program::Precompiled(component), + io::empty(), + io::sink(), + io::sink(), + tree, + readiness, + filesystem, + ); + builder = builder + .with_runtime_engine(runtime_engine) + .with_args(vec!["native-p3-fixture".into(), "directory".into()]) + .with_cache(cache) + .with_filesystem_test_hook(hook); + let completion = builder.filesystem_completion(); + let finalizer = builder.take_filesystem_finalizer().unwrap(); + let finalizer = + tokio::task::spawn_local(rpc::local_tasks::track(finalizer.run())); + lifecycle.proc = Some(builder.build().await.unwrap()); + lifecycle.filesystem_completion = completion.clone(); + let (reader, writer) = handles.take_host_split().unwrap(); + let (rpc, _) = graft::build_child_membrane_rpc( + reader, + writer, + lifecycle.membrane.as_ref().unwrap().clone(), + ); + lifecycle.rpc_task = Some(rpc); + let owner = tokio::task::spawn_local(lifecycle.run()); + + let temporary = build_entered_rx.await.unwrap(); + let active = operation_snapshot(&completion).await; + let builds: Vec<_> = active.directory_builds().iter().flatten().collect(); + assert_eq!(builds.len(), 1); + let operation_id = builds[0].id(); + assert_eq!(builds[0].waiter(), FsOperationWaiter::Waiting); + assert_eq!(builds[0].backend(), FsOperationBackend::Running); + assert_eq!(builds[0].receipt(), FsOperationReceipt::Outstanding); + assert!(active.root_admission_open()); + assert_eq!(active.omitted_directory_builds(), 0); + + let mut wait = Box::pin(process.wait_request().send().promise); + process.kill_request().send().promise.await.unwrap(); + // clear() follows drop(proc_run) in OwnedChildLifecycle. + // Bootstrap destruction therefore witnesses guest Store + // destruction without using the operation diagnostic. + tokio::time::timeout(Duration::from_secs(10), async { + while dropped.get() == 0 { + tokio::task::yield_now().await; + } + }) + .await + .expect("launcher did not release the guest bootstrap"); + assert_eq!(dropped.get(), 1); + let cancelled = operation_snapshot(&completion).await; + let build = cancelled + .directory_builds() + .iter() + .flatten() + .find(|build| build.id() == operation_id) + .unwrap(); + assert_eq!(build.waiter(), FsOperationWaiter::Dropped); + assert_eq!(build.backend(), FsOperationBackend::Running); + assert_eq!(build.receipt(), FsOperationReceipt::Outstanding); + assert!(!cancelled.root_admission_open()); + assert!(cancelled.outstanding_work() > 0); + assert_eq!(cancelled.cleanup(), FsCleanupState::Pending); + std::future::poll_fn(|cx| { + assert!(matches!(wait.as_mut().poll(cx), Poll::Pending)); + Poll::Ready(()) + }) + .await; + cancelled_tx.send((temporary, nested.clone())).unwrap(); + + cleanup_entered_rx.await.unwrap(); + let cleaning = operation_snapshot(&completion).await; + let build = cleaning + .directory_builds() + .iter() + .flatten() + .find(|build| build.id() == operation_id) + .unwrap(); + assert_eq!(build.waiter(), FsOperationWaiter::Dropped); + assert_eq!(build.backend(), FsOperationBackend::Succeeded); + assert_eq!(build.receipt(), FsOperationReceipt::Released); + assert_eq!(cleaning.outstanding_work(), 0); + assert_eq!(cleaning.cleanup(), FsCleanupState::Running); + assert_eq!(observer.state(), rpc::CleanupState::Running); + let mut filesystem_wait = Box::pin(completion.wait()); + std::future::poll_fn(|cx| { + assert!(matches!(wait.as_mut().poll(cx), Poll::Pending)); + assert!(matches!(filesystem_wait.as_mut().poll(cx), Poll::Pending)); + Poll::Ready(()) + }) + .await; + cleanup_observed_tx.send(nested.clone()).unwrap(); + + assert_eq!(wait.await.unwrap().get().unwrap().get_exit_code(), 137); + filesystem_wait.await.unwrap(); + owner.await.unwrap(); + finalizer.await.unwrap().unwrap(); + assert_eq!(observer.state(), rpc::CleanupState::Cleaned(137)); + assert_eq!(dropped.get(), 2); + let completed = operation_snapshot(&completion).await; + assert_eq!(completed.cleanup(), FsCleanupState::Completed); + assert_eq!(completed.outstanding_work(), 0); + assert!(!completed.failure_recorded()); + assert!(!nested.exists()); + drop(scope); + finished_tx.send(()).unwrap(); + }) + }), + result_tx: None, + }) + .is_ok()); + + let (temporary, nested) = cancelled_rx.recv_timeout(Duration::from_secs(20)).unwrap(); + assert!(temporary.join("child").is_file()); + assert!(!nested.exists()); + assert!(!cleanup_finished.load(Ordering::SeqCst)); + build_release_tx.send(()).unwrap(); + let published = cleanup_observed_rx + .recv_timeout(Duration::from_secs(10)) + .unwrap(); + assert_eq!(published, nested); + assert_eq!(published.join("child").metadata().unwrap().len(), 12); + assert_eq!(std::fs::read_dir(&published).unwrap().count(), 2_049); + assert!(!cleanup_finished.load(Ordering::SeqCst)); + cleanup_release_tx.send(()).unwrap(); + finished_rx.recv_timeout(Duration::from_secs(10)).unwrap(); + assert!(cleanup_finished.load(Ordering::SeqCst)); + assert!(!published.exists()); + drop(pool); + } + #[test] fn slow_scratch_teardown_keeps_process_wait_pending_and_sibling_running() { let (_shutdown_tx, shutdown_rx) = tokio::sync::watch::channel(()); diff --git a/src/launcher_tracing_tests.rs b/src/launcher_tracing_tests.rs index 6ddafbf7..41e2213f 100644 --- a/src/launcher_tracing_tests.rs +++ b/src/launcher_tracing_tests.rs @@ -40,18 +40,23 @@ impl Write for Output { #[tokio::test(flavor = "current_thread")] async fn spawned_children_keep_distinct_identity_through_execution_and_teardown() { - exercise_children(true).await; + exercise_children(true, true).await; } #[tokio::test(flavor = "current_thread")] async fn filtered_child_spans_do_not_retain_the_spawn_caller() { - exercise_children(false).await; + exercise_children(false, true).await; +} + +#[tokio::test(flavor = "current_thread")] +async fn filesystem_lifecycle_diagnostics_require_trace_filter() { + exercise_children(true, false).await; } // Value: protects=bounded filesystem diagnostics before child exit with correct // attribution; fails_when=the event/field is absent, late, or retains the caller; // why_new=existing assertions covered fuel and exit only; seam=none. -async fn exercise_children(child_span_enabled: bool) { +async fn exercise_children(child_span_enabled: bool, launcher_trace_enabled: bool) { let output = Output::default(); let writer = output.clone(); let layer = tracing_subscriber::fmt::layer() @@ -60,7 +65,10 @@ async fn exercise_children(child_span_enabled: bool) { .with_span_list(true) .with_writer(move || writer.clone()) .with_filter(tracing_subscriber::filter::filter_fn(move |metadata| { - child_span_enabled || !(metadata.is_span() && metadata.name() == "cell") + (launcher_trace_enabled + || metadata.target() != "ww::launcher" + || *metadata.level() <= tracing::Level::DEBUG) + && (child_span_enabled || !(metadata.is_span() && metadata.name() == "cell")) })); let links = CausalLinks::default(); // Both layers must reject disabled child spans, or the capture layer would @@ -68,7 +76,10 @@ async fn exercise_children(child_span_enabled: bool) { let capture = links .clone() .with_filter(tracing_subscriber::filter::filter_fn(move |metadata| { - child_span_enabled || !(metadata.is_span() && metadata.name() == "cell") + (launcher_trace_enabled + || metadata.target() != "ww::launcher" + || *metadata.level() <= tracing::Level::DEBUG) + && (child_span_enabled || !(metadata.is_span() && metadata.name() == "cell")) })); let subscriber = tracing_subscriber::registry().with(layer).with(capture); let _subscriber = tracing::subscriber::set_default(subscriber); @@ -155,6 +166,7 @@ async fn exercise_children(child_span_enabled: bool) { let mut started = Vec::new(); let mut stopped = Vec::new(); let mut filesystem_snapshots = Vec::new(); + let mut filesystem_completions = Vec::new(); for event in events { let message = event["fields"]["message"].as_str().unwrap_or_default(); if message == "unrelated-child-probe" { @@ -162,8 +174,17 @@ async fn exercise_children(child_span_enabled: bool) { } let filesystem_snapshot = message == "executor: child filesystem ownership after guest exit"; + let filesystem_completion = message == "executor: child filesystem completion observed"; + if filesystem_snapshot || filesystem_completion { + assert!( + launcher_trace_enabled, + "filesystem diagnostic below trace: {event}" + ); + assert_eq!(event["level"], "TRACE", "{event}"); + } if message == "fuel.initial" || filesystem_snapshot + || filesystem_completion || message.starts_with("executor: child process exited") { let id = if child_span_enabled { @@ -185,23 +206,41 @@ async fn exercise_children(child_span_enabled: bool) { let value = event["fields"]["filesystem"] .as_str() .expect("filesystem snapshot or explicit read error"); - assert!(value.len() < 1024, "unbounded snapshot event: {event}"); + assert!(value.len() < 4096, "unbounded snapshot event: {event}"); assert!( value.starts_with("Ok(FsSnapshot {") || matches!(value, "Err(Busy)" | "Err(Unavailable)" | "Err(Poisoned)"), "unexpected snapshot diagnostic: {event}" ); + if value.starts_with("Ok(FsSnapshot {") { + assert!(value.contains("directory_builds:"), "{event}"); + } filesystem_snapshots.push(id); - } else { - assert_eq!(filesystem_snapshots.len(), stopped.len() + 1); + } else if filesystem_completion { + assert_eq!(event["fields"]["succeeded"], true, "{event}"); + assert_eq!(filesystem_snapshots.len(), filesystem_completions.len() + 1); assert_eq!(filesystem_snapshots.last(), Some(&id)); + filesystem_completions.push(id); + } else { + if launcher_trace_enabled { + assert_eq!(filesystem_snapshots.len(), stopped.len() + 1); + assert_eq!(filesystem_snapshots.last(), Some(&id)); + assert_eq!(filesystem_completions.len(), stopped.len() + 1); + assert_eq!(filesystem_completions.last(), Some(&id)); + } stopped.push(id); } } } assert_eq!(started.len(), 2); assert_eq!(stopped, started); - assert_eq!(filesystem_snapshots, started); + if launcher_trace_enabled { + assert_eq!(filesystem_snapshots, started); + assert_eq!(filesystem_completions, started); + } else { + assert!(filesystem_snapshots.is_empty()); + assert!(filesystem_completions.is_empty()); + } if child_span_enabled { assert_ne!(started[0], started[1]); }