From 0c1ef74f6efbdf0912bae786ba95e45408fc10f6 Mon Sep 17 00:00:00 2001 From: SAU-359 implementer Date: Fri, 30 Oct 2026 21:06:10 +0000 Subject: [PATCH] Keep file-watch observation from serializing unrelated IO Scope open-registration and rename/remove ordering to overlapping paths; let descriptor mutations and directory creation proceed without that lock. Prove blocked opens/writes cannot stall unrelated namespaces, including a queued rename, and retain post-rename descriptor notifications. Document handler-owned watch lifetimes and narrow paths; test natural idle exit. Card: SAU-359 --- PROTOCOL.md | 11 +- sdk/src/calls.rs | 14 +- sdk/tests/plugins/probe/src/main.rs | 35 ++++ sdk/tests/wasi.rs | 9 + vfs/src/observe.rs | 286 ++++++++++++++++++++++++++-- 5 files changed, 336 insertions(+), 19 deletions(-) diff --git a/PROTOCOL.md b/PROTOCOL.md index bd5635c..9adef6d 100644 --- a/PROTOCOL.md +++ b/PROTOCOL.md @@ -267,8 +267,15 @@ name visible files. External host writes are outside this contract. Changes may coalesce, including a change already visible in the initial scan, but a change after the last notification always produces another. -Delivery holds no filesystem locks. The watch ends with its caller; there -is no cancellation API or polling/retry loop. +Delivery holds no filesystem locks. Observation does not serialize descriptor +IO; only overlapping opens and rename/removal wait for one another. +Watch the narrowest useful path: delivery is not rate-limited, so watching a +busy parent can cause a rescan for every write. + +The watch ends with its caller; there is no cancellation API or polling/retry +loop. SDK callers should poll the watch future alongside their handler's work +and drop it on return, not keep it in a background task that prevents an idle +process from exiting. Dropping the future does not cancel the host watch. ## 5. Endpoints diff --git a/sdk/src/calls.rs b/sdk/src/calls.rs index bff9005..04e2c6e 100644 --- a/sdk/src/calls.rs +++ b/sdk/src/calls.rs @@ -27,10 +27,16 @@ impl Call { } /// Receive progress notifications before the result arrives. /// - /// For `core.watch`, await this call in a background task for the process's - /// lifetime. Scan only after `armed`, and rescan on `changed`; events can - /// coalesce. The call has no cancellation API. Do not use endpoint connect - /// retries to wait for file publication. + /// For `core.watch`, poll the call future alongside work in the handler, then + /// drop it when the handler returns. Do not put it in `background()` just to + /// keep the watch alive: background tasks keep the process alive, while the + /// watch only ends when the process ends. Dropping the future does not cancel + /// the host watch, but it lets an otherwise idle process exit normally. + /// + /// Scan only after `armed`, and rescan on `changed`. Events may coalesce, but + /// are not rate-limited: watch the narrowest useful path to avoid rescanning + /// for unrelated writes (for example, watch a mailbox, not all session data). + /// Do not use endpoint connect retries to wait for file publication. pub fn on_notify(mut self, f: impl FnMut(&str, Value) + 'static) -> Self { self.notify = Some(Box::new(f)); self diff --git a/sdk/tests/plugins/probe/src/main.rs b/sdk/tests/plugins/probe/src/main.rs index 581a946..ba8caa4 100644 --- a/sdk/tests/plugins/probe/src/main.rs +++ b/sdk/tests/plugins/probe/src/main.rs @@ -63,6 +63,41 @@ impl Probe { Ok(()) } + /// Poll the watch in the handler, then return idle without a background task. + async fn watch_once(&self, p: WatchPath) -> Result<()> { + use std::{ + cell::RefCell, + rc::Rc, + task::{Poll, Waker}, + }; + let state = Rc::new(RefCell::new((false, None::))); + let received = state.clone(); + let mut watch = std::pin::pin!(std::future::IntoFuture::into_future( + call("core.watch", json!({"path":p.path})).on_notify(move |name, _| { + if name == "armed" { + let mut state = received.borrow_mut(); + state.0 = true; + if let Some(waker) = state.1.take() { + waker.wake(); + } + } + }) + )); + std::future::poll_fn(|cx| { + if let Poll::Ready(result) = watch.as_mut().poll(cx) { + return Poll::Ready(result.map(|_| ())); + } + let mut state = state.borrow_mut(); + state.1 = Some(cx.waker().clone()); + if state.0 { + Poll::Ready(Ok(())) + } else { + Poll::Pending + } + }) + .await + } + /// Publish a control endpoint until EOF. async fn serve(&self) -> Result<()> { sausage_plugin::listen("/endpoint", std::rc::Rc::new(Control)).await diff --git a/sdk/tests/wasi.rs b/sdk/tests/wasi.rs index 251a3de..1c71860 100644 --- a/sdk/tests/wasi.rs +++ b/sdk/tests/wasi.rs @@ -108,6 +108,15 @@ fn protocol() { } p.finishes(); } + // A handler-owned watch does not keep an idle process alive after return, + // even though the host never answers the watch and stdin remains open. + let mut p = start(&wasm, &dir); + p.request("watch_once", json!({"path":"/mailbox"})); + let watch = p.read(); + assert_eq!(watch["method"], "core.watch"); + p.send(json!({"jsonrpc":"2.0","method":"core.notify","params":{"call":watch["id"],"name":"armed","params":{}}})); + assert_eq!(p.read()["result"], Value::Null); + p.finishes(); let mut p = start(&wasm, &dir); p.request("rpc.discover", json!({})); let d = p.read(); diff --git a/vfs/src/observe.rs b/vfs/src/observe.rs index be72b99..e7f67d0 100644 --- a/vfs/src/observe.rs +++ b/vfs/src/observe.rs @@ -1,6 +1,16 @@ //! Observes successful namespace and open-file mutations, even if their caller is cancelled. use crate::*; -use std::sync::{Mutex, Weak}; +use std::{ + collections::BTreeMap, + sync::{Mutex, Weak}, +}; +use tokio::sync::{OwnedRwLockReadGuard, OwnedRwLockWriteGuard, RwLock}; + +type Namespace = Mutex>>>; +enum Held { + Read { _guard: OwnedRwLockReadGuard<()> }, + Write { _guard: OwnedRwLockWriteGuard<()> }, +} type TrackedPath = Mutex>; type Path = Arc; @@ -11,15 +21,16 @@ type Callback = dyn Fn(&str) + Send + Sync; pub struct Observer { changed: Arc, paths: Arc>>>, - // Keep path rebasing and mutations in the same order across all namespaces. - mutations: Arc>, + // Only open-registration versus rename/remove needs ordering. Lock the affected + // subtrees, not all IO: a blocked open must not stall unrelated namespaces. + namespace: Arc, } impl Observer { pub fn new(changed: impl Fn(&str) + Send + Sync + 'static) -> Self { Self { changed: Arc::new(changed), paths: Arc::default(), - mutations: Arc::default(), + namespace: Arc::default(), } } /// Wrap a layer at its canonical namespace prefix (e.g. `/run`). @@ -31,6 +42,56 @@ impl Observer { process, }) } + /// Opens share their path and ancestors; rename/remove exclusively hold their + /// endpoints, sharing ancestors. Sorted acquisition avoids cross-rename deadlocks. + /// No descriptor IO or notification holds these guards. + async fn order(&self, endpoints: &[(&str, bool)]) -> Vec { + let mut modes = BTreeMap::new(); + for &(path, exclusive) in endpoints { + modes + .entry(path.to_owned()) + .and_modify(|mode| *mode |= exclusive) + .or_insert(exclusive); + let mut ancestor = path; + while let Some((parent, _)) = ancestor.rsplit_once('/') { + let parent = if parent.is_empty() { "/" } else { parent }; + modes.entry(parent.to_owned()).or_insert(false); + if parent == "/" { + break; + } + ancestor = parent; + } + } + let locks: Vec<_> = { + let mut namespace = self.namespace.lock().unwrap(); + namespace.retain(|_, lock| lock.strong_count() != 0); + modes + .into_iter() + .map(|(path, exclusive)| { + let lock = namespace.entry(path).or_default(); + let strong = lock.upgrade().unwrap_or_else(|| { + let strong = Arc::new(RwLock::new(())); + *lock = Arc::downgrade(&strong); + strong + }); + (strong, exclusive) + }) + .collect() + }; + let mut held = Vec::new(); + for (lock, exclusive) in locks { + held.push(if exclusive { + Held::Write { + _guard: lock.write_owned().await, + } + } else { + Held::Read { + _guard: lock.read_owned().await, + } + }); + } + held + } fn path(&self, path: String) -> Path { let path = Arc::new(Mutex::new(Some(path))); let mut paths = self.paths.lock().unwrap(); @@ -85,10 +146,8 @@ impl Handle { let (path, observer) = (self.path.clone(), self.observer.clone()); self.process .operation(async move { - let order = observer.mutations.lock().await; let result = f.await?; let path = path.lock().unwrap().clone(); - drop(order); if let Some(path) = path { (observer.changed)(&path); } @@ -118,7 +177,7 @@ impl Layer for Observed { ); self.process .operation(async move { - let order = observer.mutations.lock().await; + let order = observer.order(&[(&path, false)]).await; let file = layer.open(&local, options).await?; let handle = Arc::new(Handle { file, @@ -143,9 +202,7 @@ impl Layer for Observed { ); self.process .operation(async move { - let order = observer.mutations.lock().await; layer.mkdir(&local).await?; - drop(order); (observer.changed)(&path); Ok(()) }) @@ -168,7 +225,7 @@ impl Layer for Observed { ); self.process .operation(async move { - let order = observer.mutations.lock().await; + let order = observer.order(&[(&from, true), (&to, true)]).await; layer.rename(&a, &b).await?; if from != to { observer.moved(&from, Some(&to)); @@ -191,7 +248,7 @@ impl Observed { ); self.process .operation(async move { - let order = observer.mutations.lock().await; + let order = observer.order(&[(&path, true)]).await; if directory { layer.remove_dir(&local).await?; } else { @@ -324,6 +381,200 @@ mod tests { assert!(rx.try_recv().is_err()); } } + /// A pending exclusive namespace change must not block a sibling's opens, + /// even though Tokio's RwLock prefers queued writers over new readers. + #[tokio::test] + async fn namespace_order_is_scoped_and_shared() { + async fn ready(future: impl Future) -> T { + let mut future = std::pin::pin!(future); + std::future::poll_fn(|cx| match future.as_mut().poll(cx) { + std::task::Poll::Ready(value) => std::task::Poll::Ready(value), + std::task::Poll::Pending => panic!("unrelated namespace operation blocked"), + }) + .await + } + let observer = Observer::new(|_| {}); + let open = observer.order(&[("/run/slow/file", false)]).await; + let concurrent_open = ready(observer.order(&[("/run/slow/file", false)])).await; + let mut rename = + std::pin::pin!(observer.order(&[("/run/slow", true), ("/run/moved", true)])); + std::future::poll_fn(|cx| { + assert!(rename.as_mut().poll(cx).is_pending()); + std::task::Poll::Ready(()) + }) + .await; + // Same mounted layer, different subtree; also a different mounted layer. + let sibling_open = ready(observer.order(&[("/run/sibling/file", false)])).await; + let host_open = ready(observer.order(&[("/state/file", false)])).await; + drop(sibling_open); + let sibling_rename = + ready(observer.order(&[("/run/sibling", true), ("/run/other", true)])).await; + drop((sibling_rename, host_open, concurrent_open, open)); + let renamed = ready(rename).await; + // A descendant open waits for its ancestor's rename until rebasing is done. + let mut descendant = std::pin::pin!(observer.order(&[("/run/moved/file", false)])); + std::future::poll_fn(|cx| { + assert!(descendant.as_mut().poll(cx).is_pending()); + std::task::Poll::Ready(()) + }) + .await; + drop(renamed); + drop(ready(descendant).await); + drop( + observer + .order(&[("/run/moved/file", true), ("/run/moved", true)]) + .await, + ); + drop( + observer + .order(&[("/run/moved", true), ("/run/moved", true)]) + .await, + ); + // Dead weak entries are pruned, rather than retaining every path ever opened. + observer.order(&[("/", false)]).await; + assert_eq!(observer.namespace.lock().unwrap().len(), 1); + } + + /// Pause an open after obtaining its real identity, before registering its + /// observed handle, like a slow host syscall or /self/log query. + struct SlowOpen { + memory: Arc, + gate: Mutex< + Option<( + tokio::sync::oneshot::Sender<()>, + tokio::sync::oneshot::Receiver<()>, + )>, + >, + } + #[async_trait] + impl Layer for SlowOpen { + async fn stat(&self, p: &str, follow: bool) -> Result { + self.memory.stat(p, follow).await + } + async fn list(&self, p: &str) -> Result> { + self.memory.list(p).await + } + async fn open(&self, p: &str, options: Open) -> Result> { + let f = self.memory.open(p, options).await?; + let gate = self.gate.lock().unwrap().take(); + if let Some((started, release)) = gate { + started.send(()).unwrap(); + release.await.unwrap(); + } + Ok(f) + } + async fn mkdir(&self, p: &str) -> Result<()> { + self.memory.mkdir(p).await + } + async fn remove_file(&self, p: &str) -> Result<()> { + self.memory.remove_file(p).await + } + async fn remove_dir(&self, p: &str) -> Result<()> { + self.memory.remove_dir(p).await + } + async fn rename(&self, a: &str, b: &str) -> Result<()> { + self.memory.rename(a, b).await + } + } + #[tokio::test] + async fn blocked_io_does_not_stall_other_processes() { + let memory = Arc::new(Memory::default()); + memory.mkdir("/slow").await.unwrap(); + let options = Open { + create: true, + write: true, + ..Open::default() + }; + let raw = memory.open("/slow/file", options).await.unwrap(); + let (started, opening) = tokio::sync::oneshot::channel(); + let (release, released) = tokio::sync::oneshot::channel(); + let (tx, mut rx) = mpsc::unbounded_channel(); + let observer = Observer::new(move |p| { + tx.send(p.to_owned()).unwrap(); + }); + let slow = observer.layer( + Arc::new(SlowOpen { + memory: memory.clone(), + gate: Mutex::new(Some((started, released))), + }), + "/run", + Process::default(), + ); + let other = observer.layer(memory, "/run", Process::default()); + let opening_task = tokio::spawn(async move { + slow.open( + "/slow/file", + Open { + write: true, + ..Open::default() + }, + ) + .await + }); + opening.await.unwrap(); + let renamer = other.clone(); + let renamed = tokio::spawn(async move { renamer.rename("/slow", "/moved").await }); + // These must complete while the first process's open is still blocked, + // and a conflicting rename is waiting to rebase its registered handle. + other + .open( + "/", + Open { + directory: true, + ..Open::default() + }, + ) + .await + .unwrap(); + let f = other.open("/other", options).await.unwrap(); + assert_eq!(rx.recv().await.unwrap(), "/run/other"); + f.write(0, b"visible".to_vec()).await.unwrap(); + assert_eq!(rx.recv().await.unwrap(), "/run/other"); + other.mkdir("/sibling").await.unwrap(); + assert_eq!(rx.recv().await.unwrap(), "/run/sibling"); + other.rename("/other", "/sibling/file").await.unwrap(); + assert_eq!(rx.recv().await.unwrap(), "/run/other"); + assert_eq!(rx.recv().await.unwrap(), "/run/sibling/file"); + other.remove_file("/sibling/file").await.unwrap(); + assert_eq!(rx.recv().await.unwrap(), "/run/sibling/file"); + assert!(!renamed.is_finished()); + release.send(()).unwrap(); + let opened = opening_task.await.unwrap().unwrap(); + renamed.await.unwrap().unwrap(); + rx.recv().await.unwrap(); + rx.recv().await.unwrap(); + opened.set_size(0).await.unwrap(); + assert_eq!(rx.recv().await.unwrap(), "/run/moved/file"); + + // No namespace guard surrounds descriptor IO: a rename completes while + // this write is held in flight. Its notification uses the rebased path. + let handle = Handle { + file: raw.clone(), + path: observer.path("/run/moved/file".into()), + observer, + process: Process::default(), + }; + let (started, writing) = tokio::sync::oneshot::channel(); + let (release, released) = tokio::sync::oneshot::channel(); + let write = tokio::spawn(async move { + handle + .mutate(async move { + started.send(()).unwrap(); + released.await.unwrap(); + raw.write(0, b"after rename".to_vec()).await + }) + .await + }); + writing.await.unwrap(); + other.rename("/moved", "/again").await.unwrap(); + rx.recv().await.unwrap(); + rx.recv().await.unwrap(); + other.open("/again/file", Open::default()).await.unwrap(); + release.send(()).unwrap(); + write.await.unwrap().unwrap(); + assert_eq!(rx.recv().await.unwrap(), "/run/again/file"); + } + /// Killing a waiting writer does not discard observation of a completed mutation; /// the process's idle barrier includes the callback, not only the system call. #[tokio::test] @@ -334,14 +585,23 @@ mod tests { }); let process = Process::default(); let layer = observer.layer(Arc::new(Memory::default()), "/", process.clone()); - let order = observer.mutations.lock().await; + let order = observer.order(&[("/published", true)]).await; let (started, ready) = tokio::sync::oneshot::channel(); let p = process.clone(); let task = tokio::spawn(async move { // Signal from inside the tracked operation, not before its guard exists. p.operation(async move { started.send(()).unwrap(); - layer.mkdir("/published").await + layer + .open( + "/published", + Open { + create: true, + write: true, + ..Open::default() + }, + ) + .await }) .await });