From c883ed69176434a47f4e22c285d742dab5405063 Mon Sep 17 00:00:00 2001 From: ualtinok <94532+ualtinok@users.noreply.github.com> Date: Mon, 26 Oct 2026 14:13:10 +0100 Subject: [PATCH 1/3] mason: audit watcher frontier coverage before cookie wiring Co-authored-by: alfonso-aft <289616620+alfonso-aft@users.noreply.github.com> --- docs/design/views-4v-frontier-audit.md | 162 +++++++++++++++++++++++++ 1 file changed, 162 insertions(+) create mode 100644 docs/design/views-4v-frontier-audit.md diff --git a/docs/design/views-4v-frontier-audit.md b/docs/design/views-4v-frontier-audit.md new file mode 100644 index 0000000000..3c171f6ff8 --- /dev/null +++ b/docs/design/views-4v-frontier-audit.md @@ -0,0 +1,162 @@ +# Watcher delivery-frontier audit + +## Decision and scope + +This audit concerns **verified bind**, not whether ordinary watching works. A +cookie acknowledgment must mean that every relevant write completed before +cookie creation has reached the views driver as reflected data or pending work. +An empty queue, a quiet interval, or a native flush returning is not that proof. +The cookie lives directly in the checkout root and is created exclusively as +`.aft-cookie--`; it must never become corpus membership. + +The audit starts from commit `a8e301bea1eb7d2cba7ecbaccfeecc4a461c6138`. +No backend is enabled for verified bind by this slice. In particular, FSEvents +is **off pending proof**, not demonstrated incorrect. Cookie/overflow +infrastructure can be built without relaxing that restriction. E0 is the start +of a fully registered watcher subscription; F is the later cookie delivery +frontier. A delivery sequence orders invalidations at the driver, and the +durable cut is the exact prefix of those changes persisted in a publication. +Connecting those points to ready publication belongs to later verified-bind work. +Ordinary strict reconciliation remains available with its existing, weaker +per-path stability and subsequent-event-invalidation contract. + +## Shared filters, batching and the driver queue + +`watcher_filter::WatcherFilterThread` receives normalized native events over an +MPSC channel. Ordinary paths are canonicalized, deduplicated, and batched before +corpus filtering. `send_dispatch` and `retry_backlog` feed a bounded FIFO driver +channel; held batches may merge, and overflow replaces paths with a rescan. +Consequently, observing a cookie at raw intake alone cannot acknowledge a +views frontier: earlier paths may still be in either batch, backlog, channel, +or `runtime_drain`'s sliced `pending_paths`. + +The additive cookie plumbing must bypass ignore/infra filters, unlink a root +cookie immediately at intake, flush earlier paths, and enqueue a distinct +control rather than a membership path. A mixed event must retain its non-cookie +paths. Only a create event may acknowledge a ticket; unlink/metadata events must +not. The shared `watcher_backend::flush::is_marker_path` admission predicate must +also reject cookies during a filesystem walk, not just during event delivery. +The command-report `.aft-flush` acknowledgment remains a separate protocol. + +The driver-side cookie arm must record all preceding queued paths as pending +before acknowledging the exact registered cookie. A rescan must enter the +driver as `WatcherState::Overflowed` and invalidate an in-flight installation; +a subsequent ordinary path must not reset that state to Healthy. A cookie held +in a backlog whose paths are discarded must be discarded too (timeout, not a +false successful frontier). This provides an ordered **delivery acknowledgment**, +not ready publication: the bind owner must still establish E0, reject any gap, +replay through the acknowledged cut, and revalidate before publication. + +The planned additive implementation includes regression tests named +`watcher_cookie_delayed_delivery_past_read_inotify`, +`watcher_cookie_delayed_delivery_past_read_fsevents`, and +`watcher_cookie_delayed_delivery_past_read_windows`, using a backend-delivery shim +which holds a write notification and its following cookie past a source read. +They must show no acknowledgment before release and a pending path before the +acknowledgment afterwards. They test the application queue, **not** a kernel's +completed-write coverage. Native macOS and Windows execution is delegated to CI. + +## Linux: inotify + +`inotify::ProjectWatcher::create_with_plan` creates one notify inotify instance, +collects directories, and synchronously installs non-recursive watches before +spawning its backend thread. Every per-directory descriptor belongs to that +instance's ordered queue; extra watches also use that instance. Successful +return is later than registration of the collected set, but is not yet a safe +E0: `collect_watch_directories` walks **before** the root watch is installed. +A directory created after its parent was enumerated but before the parent was +watched can be absent from the collected set without a queued create event. +The later bind must close that registration gap before recording E0, and must +not mistake a returned watcher handle for a complete subscription proof. + +For newly delivered directory creates/renames, `watch_new_tree` does install +each directory's watch before `exclusion_read_dir`. Already-populated files are +recovered by listing. `PendingNewDirectoryFiles` delays that synthetic file +batch; it must be flushed before a root cookie even when command-report tracking +is inactive. Otherwise the root cookie could overtake a recovered child write. +An independent metadata walk cannot assume the backend has already processed +every directory create: watch-before-list coordination with that walk remains +an enablement obligation of the bind implementation. + +Upstream notify's inotify queue-overflow flag is forwarded as `need_rescan`. +An incomplete/over-budget new-tree listing also emits a rescan. The filter maps +these to `RescanRequired`, and the context adapter must mark the installed +driver Overflowed before any subsequent cookie acknowledgment. Watch-limit and +other fatal backend errors terminate delivery; a disconnected/error stream +cannot certify a frontier. Retained watches are re-added without unwatching in +`reconcile_watch_directories`, but policy/boundary changes still require bind +revalidation. A single queue is necessary, not proof that every directory was +covered when a write occurred. + +## macOS: FSEvents + +`fsevents::FsEventsStream::start_paths` starts one recursive project stream with +FileEvents and waits for `FSEventStreamStart` on its run loop. There is no +per-directory watch-registration walk: a recursive stream covers newly created +descendants without adding directory handles. Excluded/cross-device descendants +are not thereby certified. A later bind can record E0 only after successful +startup and coverage/root/policy validation, not before the startup handshake. + +The callback translates in native array order, hands off one vector, and advances +its replay cursor only after handoff. The backend forwards each vector in order +to the filter. Stream replacement replays from the old delivered event ID rather +than SinceNow; a missing/wrapped cursor requires a rescan. MustScanSubDirs, +UserDropped and KernelDropped all emit `Flag::Rescan` in `translate_event`; +those controls must survive filters and mark the driver Overflowed. + +FSEvents is **off: completed-write-before-cookie coverage through the views +queue is unproved**. Auxiliary streams are independently ordered +(`fsevents.rs:163-164` at the audited commit). FlushSync covers only events +already in the pipeline (`fsevents.rs:214-222`), not all completed filesystem +writes. Preserving callback/event-ID order does not by itself close that gap. +No root cookie certifies an independently ordered auxiliary stream. No +quiet-window substitute is accepted. + +The later native test required to turn it on is +`fsevents_completed_writes_before_cookie_reach_views_frontier`: on a writable +same-device local root, establish E0, hold delivery beyond the stat/content read, +complete writes in multiple existing and newly created descendants, create the +cookie, and prove every pre-cookie write reflected or pending through the exact +views delivery cut before ready publication. Exercise coalesced callbacks, +stream handoff, and drops (drops must refuse verified bind); separately fence +auxiliary streams or exclude their state from the frontier claim. The shim test +above is necessary application-ordering evidence, not this native proof. + +## Windows: ReadDirectoryChangesW + +`windows::ProjectWatcher::start` blocks on `start_rx` until `open_watches` has +opened and submitted every initial read. `project_watch_targets` includes one +recursive root handle and separate non-recursive extra-path/parent handles. +Each `DirectoryWatch` has its own ordered notification buffer and pending read. +The recursive root handle already covers newly created directories; there is no +per-child registration/listing seam within that handle. E0 must follow all read +submissions and root/target validation. + +`run_completion_loop` rearms a read before translating its completed buffer, and +forwards records in buffer order. A completion port merges independent handles; +it does not give their writes a shared filesystem ordering. A root cookie +certifies neither an earlier undelivered extra-handle completion nor handle +replacement. Single-recursive-handle checkout-only coverage is a possible later +restriction, not a proof supplied by this slice. + +ERROR_NOTIFY_ENUM_DIR and successful zero-byte completions produce the typed +buffer-overflow rescan. These must mark the driver Overflowed; other fatal +completion errors end the stream and invalidate any bind. DirectoryWatcher +replacement drops its old watcher before starting the next; no verified bind +may span that registration gap. Native completed-write coverage (including +last-write/close semantics) and multi-handle fencing remain unproved. + +## Verified-bind on/off table + +These are enablement decisions, not changes to ordinary watcher selection. + +| Backend / root | Verified bind | Reason / remaining obligation | +| --- | --- | --- | +| Linux inotify, writable local same-device root | **Off pending coverage proof** | Initial collect-before-watch gap; coordinate new-directory registration before subscription E0 and during the bind walk through cookie frontier F. One instance plus ordered cookies alone is insufficient. | +| macOS FSEvents, writable local same-device root | **Off pending proof** | Completed-write-before-cookie coverage through views queue unproved; independent auxiliary streams; FlushSync only covers events already in pipeline. Native turn-on test named above belongs to later verified-bind work. | +| Windows native recursive root plus auxiliary handles | **Off pending proof** | Ordered buffers are per handle, not a global frontier; native completed-write coverage and handle lifecycle must be proved. | +| notify fallback; polling; any other unproved backend | **Off** | No proved ordered completed-write frontier. | +| Network filesystems (including NFS/SMB); FAT | **Off** | Backend/local timestamp and delivery assumptions do not apply. | +| Read-only roots | **Off** | Cannot create an exclusive cookie. | +| Cross-mount or cross-device subtrees | **Off for the subtree** | A root stream/cookie does not prove coverage on another mount/device. | +| Cookie timeout, watcher loss, root/policy change, unregistered directory | **Off for that bind** | Strict reconcile; never treat silence as continuity. | From 8db5f949395bf2534dc3d4000f9f1dc6a73caf87 Mon Sep 17 00:00:00 2001 From: ualtinok <94532+ualtinok@users.noreply.github.com> Date: Tue, 27 Oct 2026 05:25:34 +0100 Subject: [PATCH 2/3] mason: deliver ordered watcher cookies and views overflow state Rebase onto the composed InstalledBundle and persistent LiveDelta: preserve readers, install cuts, readiness and replay journals while recording every delivery gap as a new epoch. Ordinary edits retain the gap state. Allow Overflowed recovery only after a completed strict walk whose reconciled epoch matches the live epoch, matching the reviewed views-stack guard from 15049cfae1. Stopped and stale epochs remain rejected. Regression coverage checks stale cuts, stale strict tags, successful recovery and held bundle/readers. Co-authored-by: alfonso-aft <289616620+alfonso-aft@users.noreply.github.com> --- crates/aft/src/context.rs | 30 ++ crates/aft/src/runtime_drain.rs | 16 + crates/aft/src/views/first_load.rs | 171 +++++++++- crates/aft/src/watcher_backend/cookie.rs | 151 +++++++++ .../aft/src/watcher_backend/cookie_tests.rs | 299 ++++++++++++++++++ crates/aft/src/watcher_backend/flush.rs | 8 +- crates/aft/src/watcher_backend/fsevents.rs | 1 + crates/aft/src/watcher_backend/inotify.rs | 52 ++- crates/aft/src/watcher_backend/mod.rs | 3 + crates/aft/src/watcher_filter.rs | 170 +++++++++- docs/design/views-4v-frontier-audit.md | 7 +- 11 files changed, 894 insertions(+), 14 deletions(-) create mode 100644 crates/aft/src/watcher_backend/cookie.rs create mode 100644 crates/aft/src/watcher_backend/cookie_tests.rs diff --git a/crates/aft/src/context.rs b/crates/aft/src/context.rs index 48e3b14360..46c3f11d68 100644 --- a/crates/aft/src/context.rs +++ b/crates/aft/src/context.rs @@ -1720,6 +1720,9 @@ pub(crate) struct WatcherCounters { pub(crate) bash_flush: Arc, pub(crate) bash_changes: std::sync::OnceLock>, overflow_ignore_evidence: AtomicU8, + /// Legacy rescan work coalesces repeated drops, but a newly installed views + /// snapshot must still learn that delivery was lost during that work. + checkout_watcher_gap_pending: AtomicBool, fsevents_stream_creations_total: AtomicU64, raw_events_total: AtomicU64, raw_events_since_last_rescan: AtomicU64, @@ -1891,6 +1894,8 @@ impl WatcherCounters { .is_ok() { self.overflows_during_rescan.fetch_add(1, Ordering::Relaxed); + self.checkout_watcher_gap_pending + .store(true, Ordering::Release); return true; } } @@ -10522,6 +10527,31 @@ impl AppContext { self.checkout_semantic.wake(); } + pub(crate) fn record_checkout_watcher_overflow(&self) { + self.reset_view_publication_retry(); + if let Some(driver) = self.checkout_driver.get() { + driver.record_watcher_overflow(); + } + self.checkout_semantic.wake(); + } + + pub(crate) fn record_checkout_watcher_cookie(&self, path: &Path) { + self.record_checkout_pending_watcher_gap(); + if let Some(driver) = self.checkout_driver.get() { + driver.record_watcher_cookie(path); + } + } + + pub(crate) fn record_checkout_pending_watcher_gap(&self) { + if self + .watcher_counters() + .checkout_watcher_gap_pending + .swap(false, Ordering::AcqRel) + { + self.record_checkout_watcher_overflow(); + } + } + pub(crate) fn view_publication_retry( &self, ) -> &parking_lot::Mutex { diff --git a/crates/aft/src/runtime_drain.rs b/crates/aft/src/runtime_drain.rs index c8557a85c4..ffb909dbc0 100644 --- a/crates/aft/src/runtime_drain.rs +++ b/crates/aft/src/runtime_drain.rs @@ -3695,6 +3695,7 @@ pub fn drain_watcher_events_bounded(ctx: &AppContext, max_paths: usize) -> Drain { return outcome; } + ctx.record_checkout_pending_watcher_gap(); let mut state = match ctx.watcher_drain_slice().lock().take() { Some(state) if state.configure_generation == configure_generation => state, // Lifecycle-only generation change (transient unbind + equivalent @@ -3754,6 +3755,7 @@ pub fn drain_watcher_events_bounded(ctx: &AppContext, max_paths: usize) -> Drain } } Ok(WatcherDispatchEvent::RescanRequired(reason)) => { + ctx.record_checkout_watcher_overflow(); state.view_publication_reason.events |= crate::views::publication_reason::Reason::RESCAN; state.view_publication_full = true; @@ -3776,6 +3778,20 @@ pub fn drain_watcher_events_bounded(ctx: &AppContext, max_paths: usize) -> Drain state.semantic_refresh_paths.clear(); state.scheduler_changed_path_count = 0; } + Ok(WatcherDispatchEvent::CookieObserved(path)) => { + dispatch_events_received += 1; + // Earlier batches may still await sliced Tier-2 work. Record + // their intents now so the cookie cannot overtake them. + for pending in &state.pending_paths { + ctx.record_checkout_watcher_change(pending); + } + if let WatcherDrainPhase::Apply { paths, .. } = &state.phase { + for pending in paths { + ctx.record_checkout_watcher_change(pending); + } + } + ctx.record_checkout_watcher_cookie(&path); + } Ok(WatcherDispatchEvent::IgnoreRulesChanged { paths }) => { state.view_publication_reason.events |= crate::views::publication_reason::Reason::IGNORE; diff --git a/crates/aft/src/views/first_load.rs b/crates/aft/src/views/first_load.rs index c9d2bb12c8..8efe245ba1 100644 --- a/crates/aft/src/views/first_load.rs +++ b/crates/aft/src/views/first_load.rs @@ -5,6 +5,7 @@ use std::collections::BTreeMap; use std::sync::Arc; use crate::blob_store::v2::FamilyPlane; +pub use crate::watcher_backend::cookie::{CookieObservation, CookieTicket}; use super::contracts::{LoaderStage, PlaneError, ViewAccess}; use super::manifest_v2::Producers; @@ -1449,6 +1450,7 @@ pub struct CheckoutDriver { // without holding query locks during I/O or plane-reader retirement. publication_transition: std::sync::Mutex<()>, retry: std::sync::atomic::AtomicBool, + watcher_cookies: crate::watcher_backend::cookie::CookieBarrier, #[cfg(test)] work: std::sync::Mutex, } @@ -1519,6 +1521,7 @@ impl CheckoutDriver { adapters: Vec::new(), publish_fence: std::sync::RwLock::new(None), observed: Default::default(), + watcher_cookies: Default::default(), installed: std::sync::Mutex::new(Arc::new(InstalledBundle { seq: 0, composed, @@ -1581,6 +1584,8 @@ impl CheckoutDriver { delta.record_intent(path); // Membership is unchanged, so are the ignore rules it was walked under. delta.record_root_ignore_files(installed.snapshot.root_ignore_files().cloned()); + // The persistent live delta retains any stream-wide delivery gap; + // recording a known path cannot make the watcher healthy again. installed.snapshot = delta .snapshot() .with_readers(installed.readers.clone()) @@ -1593,6 +1598,57 @@ impl CheckoutDriver { }); } + /// Invalidates stale builds and indexed membership until strict reconciliation. + pub fn record_watcher_overflow(&self) { + let mut delta = self + .live + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); + let mut installed = self + .installed + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); + let installed = Arc::make_mut(&mut installed); + installed.revision += 1; + // Every loss invalidates a walk begun before it, even if the watcher + // was already overflowed when that walk started. + if delta.watcher() == super::snapshot::WatcherState::Overflowed { + delta.bump_epoch(); + } + delta.set_watcher(super::snapshot::WatcherState::Overflowed); + installed.snapshot = delta + .snapshot() + .with_readers(installed.readers.clone()) + .with_install_cut(InstallCut { + seq: installed.seq, + resident: installed.composed.clone(), + epoch: delta.epoch(), + version: delta.version(), + intent_version: delta.snapshot().intent_version(), + }); + } + + /// Creates an exclusive root cookie. This does not enable verified bind; + /// the caller must establish subscription coverage and reject delivery gaps. + pub fn begin_watcher_cookie(&self) -> std::io::Result { + self.watcher_cookies.begin(self.owner.root()) + } + + /// Called only after preceding dispatch paths have entered the driver. + pub(crate) fn record_watcher_cookie(&self, path: &std::path::Path) { + let installed = self + .installed + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); + self.watcher_cookies.observe( + path, + CookieObservation { + revision: installed.revision, + watcher: installed.snapshot.watcher(), + }, + ); + } + fn check_owner(&self, access: &ViewAccess) -> Result<(), PlaneError> { match access { ViewAccess::Owner(owner) @@ -1628,7 +1684,12 @@ impl CheckoutDriver { } match live.watcher() { super::snapshot::WatcherState::Healthy => Ok(()), - super::snapshot::WatcherState::Reconciling if strict => Ok(()), + super::snapshot::WatcherState::Reconciling + | super::snapshot::WatcherState::Overflowed + if strict => + { + Ok(()) + } _ => Err(SiblingLoader::error(format!( "{CHECKOUT_CHANGED_BEFORE_INSTALL}: watcher is not healthy or strictly reconciled" ))), @@ -2655,6 +2716,114 @@ mod composite_tests { ); } + #[test] + fn watcher_cookie_overflow_preserves_composition_readers_and_live_journals() { + let root = tempfile::tempdir().unwrap(); + let storage = tempfile::tempdir().unwrap(); + let (driver, _, _, access) = fixture(root.path(), storage.path()); + drop( + SiblingLoader::new(driver.clone(), vec![]) + .load(&access) + .unwrap(), + ); + let file = RelPath::new(b"file.rs".to_vec()).unwrap(); + driver.live.lock().unwrap().apply( + file.clone(), + LiveEntry::new(DiskState::of_bytes(b"updated content"), 1), + ); + driver.live.lock().unwrap().set_readiness( + file.clone(), + super::super::snapshot::PlanePatch { + content: crate::blob_store::v2::ContentHash::of(b"updated content"), + planes: EntryPlanes { + trigram: Some(super::super::readiness::PlaneState::pending( + "checkout repair", + )), + ..Default::default() + }, + }, + ); + driver.record_change(file.clone()); + let held = driver.installed_bundle(); + let journal = driver.live.lock().unwrap().journal().to_vec(); + assert!(!journal.is_empty()); + let ticket = driver.begin_watcher_cookie().unwrap(); + driver.record_watcher_overflow(); + driver.record_watcher_cookie(ticket.path()); + let observed = ticket.wait_timeout(std::time::Duration::ZERO).unwrap(); + let current = driver.installed_bundle(); + assert_eq!( + observed.watcher, + super::super::snapshot::WatcherState::Overflowed + ); + assert_eq!(observed.revision, current.revision); + assert!(current.revision > held.revision); + assert_eq!(current.seq, held.seq); + assert!(Arc::ptr_eq(¤t.composed, &held.composed)); + assert!(Arc::ptr_eq(¤t.readers, &held.readers)); + assert!(Arc::ptr_eq( + current.snapshot.readers().unwrap(), + &held.readers + )); + assert_eq!( + current.snapshot.readiness_patches(), + held.snapshot.readiness_patches() + ); + assert_eq!(driver.live.lock().unwrap().journal(), journal.as_slice()); + assert!(current.snapshot.pending_intent().any(|path| path == &file)); + assert_eq!( + held.snapshot.watcher(), + super::super::snapshot::WatcherState::Healthy + ); + assert!(current.snapshot.epoch() > held.snapshot.epoch()); + let cut = current.snapshot.install_cut().unwrap(); + assert_eq!(cut.seq, held.seq); + assert_eq!(cut.epoch, current.snapshot.epoch()); + assert!(Arc::ptr_eq(&cut.resident, &held.composed)); + } + + #[test] + fn watcher_cookie_each_overflow_refuses_a_walk_cut_before_the_gap() { + let root = tempfile::tempdir().unwrap(); + let storage = tempfile::tempdir().unwrap(); + let (driver, _, _, access) = fixture(root.path(), storage.path()); + let loader = SiblingLoader::new(driver.clone(), vec![]); + drop(loader.load(&access).unwrap()); + for _ in 0..2 { + let (walk, revision) = loader + .reconciled(&access, driver.installed_snapshot().generation().clone()) + .unwrap(); + driver.record_watcher_overflow(); + assert!(driver.installed_snapshot().epoch() > walk.install_cut().unwrap().epoch); + let error = driver.install(&access, &walk, revision).unwrap_err(); + assert!(error.reason.contains("EpochChanged"), "{error:?}"); + let stale_epoch_walk = walk + .clone() + .with_install_cut(driver.installed_snapshot().install_cut().unwrap().clone()); + let error = driver + .install(&access, &stale_epoch_walk, revision) + .unwrap_err(); + assert!( + error + .reason + .contains("watcher is not healthy or strictly reconciled"), + "{error:?}" + ); + assert_eq!( + driver.installed_snapshot().watcher(), + super::super::snapshot::WatcherState::Overflowed + ); + } + let (walk, revision) = loader + .reconciled(&access, driver.installed_snapshot().generation().clone()) + .unwrap(); + driver.install(&access, &walk, revision).unwrap(); + assert_eq!( + driver.installed_snapshot().watcher(), + super::super::snapshot::WatcherState::Healthy + ); + } + #[test] fn rejected_install_does_not_consume_prepared_readers() { let root = tempfile::tempdir().unwrap(); diff --git a/crates/aft/src/watcher_backend/cookie.rs b/crates/aft/src/watcher_backend/cookie.rs new file mode 100644 index 0000000000..3bc0e3ddfc --- /dev/null +++ b/crates/aft/src/watcher_backend/cookie.rs @@ -0,0 +1,151 @@ +//! In-band views delivery acknowledgments, distinct from command-report flushes. +use crate::views::snapshot::WatcherState; +use std::collections::BTreeMap; +use std::path::{Path, PathBuf}; +use std::sync::{atomic::AtomicU64, atomic::Ordering, mpsc, Arc, Mutex}; +use std::time::{Duration, SystemTime, UNIX_EPOCH}; + +const PREFIX: &str = ".aft-cookie-"; +static NEXT: AtomicU64 = AtomicU64::new(0); +type Pending = Arc>>>; + +/// A driver delivery cut, not a verified-bind or durable-publication certificate. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub struct CookieObservation { + pub revision: u64, + pub watcher: WatcherState, +} + +/// An exclusive root cookie. Dropping a timed-out ticket cleans up its file. +pub struct CookieTicket { + path: PathBuf, + pending: Pending, + rx: mpsc::Receiver, +} + +impl CookieTicket { + pub fn path(&self) -> &Path { + &self.path + } + + /// On timeout or an untrusted watcher, use strict reconciliation rather + /// than relying on the cookie acknowledgment. + pub fn wait_timeout( + &self, + timeout: Duration, + ) -> Result { + self.rx.recv_timeout(timeout) + } +} + +impl Drop for CookieTicket { + fn drop(&mut self) { + self.pending + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) + .remove(&self.path); + let _ = std::fs::remove_file(&self.path); + } +} + +#[derive(Default)] +pub(crate) struct CookieBarrier { + pending: Pending, +} + +impl CookieBarrier { + pub(crate) fn begin(&self, root: &Path) -> std::io::Result { + let nonce = SystemTime::now() + .duration_since(UNIX_EPOCH) + .unwrap_or_default() + .as_nanos(); + let path = root.join(format!( + "{PREFIX}{}-{nonce:x}{:016x}", + std::process::id(), + NEXT.fetch_add(1, Ordering::Relaxed) + )); + let (tx, rx) = mpsc::channel(); + // Hold the registry lock across creation: even immediate delivery must + // find the waiter, and create_new cannot overwrite a checkout file. + let mut pending = self + .pending + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); + std::fs::OpenOptions::new() + .write(true) + .create_new(true) + .open(&path)?; + pending.insert(path.clone(), tx); + Ok(CookieTicket { + path, + pending: Arc::clone(&self.pending), + rx, + }) + } + + pub(crate) fn observe(&self, path: &Path, observation: CookieObservation) { + if let Some(tx) = self + .pending + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) + .remove(path) + { + let _ = tx.send(observation); + } + } +} + +pub(crate) fn is_cookie_path(path: &Path) -> bool { + path.file_name() + .and_then(|name| name.to_str()) + .and_then(|name| name.strip_prefix(PREFIX)) + .and_then(|suffix| suffix.split_once('-')) + .is_some_and(|(pid, nonce)| { + !pid.is_empty() + && pid.bytes().all(|byte| byte.is_ascii_digit()) + && !nonce.is_empty() + && nonce + .bytes() + .all(|byte| byte.is_ascii_alphanumeric() || byte == b'-') + }) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn watcher_cookie_tickets_match_exact_names_and_cleanup_on_timeout() { + let root = tempfile::tempdir().unwrap(); + let barrier = CookieBarrier::default(); + let ticket = barrier.begin(root.path()).unwrap(); + let path = ticket.path().to_path_buf(); + assert!(is_cookie_path(&path)); + let observation = CookieObservation { + revision: 7, + watcher: WatcherState::Healthy, + }; + barrier.observe(&root.path().join(".aft-cookie-1-unregistered"), observation); + assert_eq!( + ticket.wait_timeout(Duration::ZERO), + Err(mpsc::RecvTimeoutError::Timeout) + ); + barrier.observe(&path, observation); + assert_eq!(ticket.wait_timeout(Duration::ZERO), Ok(observation)); + drop(ticket); + assert!(!path.exists()); + let ticket = barrier.begin(root.path()).unwrap(); + let path = ticket.path().to_path_buf(); + drop(ticket); + assert!(!path.exists()); + assert!(barrier.pending.lock().unwrap().is_empty()); + for name in [ + ".aft-cookie", + ".aft-cookie--nonce", + ".aft-cookie-pid-nonce", + ".aft-cookie-1-", + ] { + assert!(!is_cookie_path(Path::new(name))); + } + } +} diff --git a/crates/aft/src/watcher_backend/cookie_tests.rs b/crates/aft/src/watcher_backend/cookie_tests.rs new file mode 100644 index 0000000000..05c1a8b882 --- /dev/null +++ b/crates/aft/src/watcher_backend/cookie_tests.rs @@ -0,0 +1,299 @@ +//! Backend delivery shim: the native producer is held while the real filter and +//! driver drain remain live. This tests application ordering, not kernel coverage. +use crate::config::Config; +use crate::context::{default_language_provider_factory, AppContext}; +use crate::runtime_drain::{drain_watcher_events_bounded, drain_watcher_to_completion_for_test}; +use crate::views::contracts::ViewAccess; +use crate::views::first_load::{CheckoutDriver, ConfiguredMembershipWalker, FirstLoadDriver}; +use crate::views::manifest_v2::Producers; +use crate::views::registry::FamilyRegistry; +use crate::views::snapshot::WatcherState; +use crate::views::RelPath; +use crate::watcher_filter::{ + run_watcher_thread, watcher_dispatch_channel, WatcherDispatchEvent, WatcherFilterConfig, +}; +use notify::{event::CreateKind, event::DataChange, event::ModifyKind, Event, EventKind}; +use std::path::Path; +use std::sync::atomic::{AtomicBool, AtomicU64, Ordering}; +use std::sync::{mpsc, Arc, RwLock}; +use std::time::Duration; + +fn driver(root: &Path, storage: &Path) -> (Arc, ViewAccess) { + let registry = FamilyRegistry::open(storage, "family").unwrap(); + let owner = registry + .register_view_with_storage(registry.storage(), "cookie", root) + .unwrap(); + let access = ViewAccess::Owner(owner.clone()); + ( + Arc::new(CheckoutDriver::new( + owner, + Producers { + trigram: "unused".into(), + semantic: None, + callgraph: "unused".into(), + }, + None, + Arc::new(ConfiguredMembershipWalker), + vec![], + )), + access, + ) +} + +fn context(root: &Path, driver: &Arc) -> AppContext { + let ctx = AppContext::new(default_language_provider_factory(), Config::default()); + ctx.update_config(|config| config.project_root = Some(root.to_path_buf())); + ctx.set_canonical_cache_root(root.to_path_buf()); + ctx.mark_subc_bound(); + ctx.install_checkout_driver(Arc::clone(driver)); + ctx +} + +fn delayed_delivery_past_read() { + let temp = tempfile::tempdir().unwrap(); + let storage = tempfile::tempdir().unwrap(); + let root = temp.path().canonicalize().unwrap(); + let changed = root.join("source.rs"); + std::fs::write(&changed, b"old bytes").unwrap(); + let (driver, _) = driver(&root, storage.path()); + let ctx = context(&root, &driver); + let (dispatch_tx, dispatch_rx) = watcher_dispatch_channel(); + *ctx.watcher_rx().lock() = Some(dispatch_rx.clone()); + let shutdown = Arc::new(AtomicBool::new(false)); + let (attached_tx, attached_rx) = mpsc::channel(); + let (release_tx, release_rx) = mpsc::channel(); + let (script_tx, script_rx) = mpsc::channel::>(); + let config = WatcherFilterConfig::new(root.clone(), None); + let thread_shutdown = Arc::clone(&shutdown); + let filter = std::thread::spawn(move || { + run_watcher_thread( + config, + vec![], + Arc::new(RwLock::new(None)), + Arc::new(AtomicU64::new(0)), + dispatch_tx, + thread_shutdown, + move |_, _, raw_tx| { + // Holding at the backend handoff, not at the reader, reproduces + // a completed write whose invalidation arrives after the read. + let shim = std::thread::spawn(move || { + attached_tx.send(()).unwrap(); + release_rx.recv().unwrap(); + for event in script_rx.recv().unwrap() { + raw_tx.send(Ok(event)).unwrap(); + } + // Keep the raw stream connected until the filter stops. + raw_tx + }); + Ok::<_, String>(shim) + }, + ); + }); + attached_rx.recv_timeout(Duration::from_secs(5)).unwrap(); + let read = std::fs::read(&changed).unwrap(); + std::fs::write(&changed, b"new bytes").unwrap(); + assert_eq!(std::fs::read(&changed).unwrap(), b"new bytes"); + let ticket = driver.begin_watcher_cookie().unwrap(); + let cookie_path = ticket.path().to_path_buf(); + script_tx + .send(vec![ + Event::new(EventKind::Modify(ModifyKind::Data(DataChange::Content))) + .add_path(changed.clone()), + Event::new(EventKind::Create(CreateKind::File)).add_path(cookie_path.clone()), + ]) + .unwrap(); + drain_watcher_events_bounded(&ctx, 1); + assert_eq!(read, b"old bytes"); + assert_eq!( + ticket.wait_timeout(Duration::ZERO), + Err(mpsc::RecvTimeoutError::Timeout) + ); + assert!(driver + .installed_snapshot() + .pending_intent() + .next() + .is_none()); + release_tx.send(()).unwrap(); + + // Peek by consuming then restoring the complete prefix. This is a bounded + // handshake, not a timing assumption about how fast the worker is scheduled. + let first = dispatch_rx.recv_timeout(Duration::from_secs(5)).unwrap(); + let second = dispatch_rx.recv_timeout(Duration::from_secs(5)).unwrap(); + assert!( + matches!(&first, WatcherDispatchEvent::ContentOnlyPaths(paths) if paths == &vec![changed.clone()]) + ); + assert_eq!( + second, + WatcherDispatchEvent::CookieObserved(cookie_path.clone()) + ); + assert!( + !cookie_path.exists(), + "unlink must precede driver acknowledgment" + ); + assert_eq!( + ticket.wait_timeout(Duration::ZERO), + Err(mpsc::RecvTimeoutError::Timeout) + ); + let (tx, rx) = crossbeam_channel::unbounded(); + tx.send(first).unwrap(); + tx.send(second).unwrap(); + *ctx.watcher_rx().lock() = Some(rx); + drain_watcher_to_completion_for_test(&ctx, 1); + let observed = ticket.wait_timeout(Duration::ZERO).unwrap(); + assert!(observed.revision > 0); + assert!(driver + .installed_snapshot() + .pending_intent() + .any(|path| path == &RelPath::new("source.rs").unwrap())); + assert!(!driver + .installed_snapshot() + .membership() + .keys() + .any(|path| path.as_bytes().starts_with(b".aft-cookie-"))); + shutdown.store(true, Ordering::SeqCst); + filter.join().unwrap(); +} + +#[cfg(target_os = "linux")] +#[test] +fn watcher_cookie_delayed_delivery_past_read_inotify() { + delayed_delivery_past_read(); +} + +#[cfg(target_os = "macos")] +#[test] +fn watcher_cookie_delayed_delivery_past_read_fsevents() { + delayed_delivery_past_read(); +} + +#[cfg(windows)] +#[test] +fn watcher_cookie_delayed_delivery_past_read_windows() { + delayed_delivery_past_read(); +} + +#[test] +fn watcher_cookie_is_excluded_from_walk_membership_before_delivery() { + let temp = tempfile::tempdir().unwrap(); + let storage = tempfile::tempdir().unwrap(); + let root = temp.path().canonicalize().unwrap(); + std::fs::write(root.join("source.rs"), b"source").unwrap(); + let (driver, access) = driver(&root, storage.path()); + let ticket = driver.begin_watcher_cookie().unwrap(); + assert!(ticket.path().exists()); + let reconciled = driver.reconcile(&access).unwrap(); + assert_eq!(reconciled.entries.len(), 1); + assert!(reconciled + .entries + .contains_key(&RelPath::new("source.rs").unwrap())); + assert!( + ticket.path().exists(), + "only delivery or ticket cleanup unlinks the cookie" + ); + assert!(!super::flush::FlushControl::default().observe( + &Event::new(EventKind::Create(CreateKind::File)).add_path(ticket.path().to_path_buf()) + )); +} + +#[test] +fn watcher_cookie_rescan_dispatch_marks_driver_and_survives_later_paths() { + use crate::watcher_filter::RescanReason; + let temp = tempfile::tempdir().unwrap(); + let storage = tempfile::tempdir().unwrap(); + let root = temp.path().canonicalize().unwrap(); + let (driver, access) = driver(&root, storage.path()); + let ctx = context(&root, &driver); + let (tx, rx) = crossbeam_channel::unbounded(); + *ctx.watcher_rx().lock() = Some(rx); + for reason in [ + RescanReason::Unknown, + RescanReason::UserDropped, + RescanReason::KernelDropped, + RescanReason::BufferOverflow, + ] { + let before = driver.revision(&access); + let ticket = driver.begin_watcher_cookie().unwrap(); + tx.send(WatcherDispatchEvent::RescanRequired(reason)) + .unwrap(); + tx.send(WatcherDispatchEvent::CookieObserved( + ticket.path().to_path_buf(), + )) + .unwrap(); + drain_watcher_to_completion_for_test(&ctx, 1); + let observed = ticket.wait_timeout(Duration::ZERO).unwrap(); + assert_eq!(observed.watcher, WatcherState::Overflowed); + assert!(observed.revision > before); + driver.record_absolute_change(&root.join("later.rs")); + assert_eq!( + driver.installed_snapshot().watcher(), + WatcherState::Overflowed + ); + } +} + +#[test] +fn watcher_cookie_fences_paths_already_parked_in_an_apply_slice() { + use crate::context::{WatcherDrainApplyPhase, WatcherDrainPhase, WatcherDrainSliceState}; + let temp = tempfile::tempdir().unwrap(); + let storage = tempfile::tempdir().unwrap(); + let root = temp.path().canonicalize().unwrap(); + let (driver, _) = driver(&root, storage.path()); + let ctx = context(&root, &driver); + let mut state = WatcherDrainSliceState::new( + ctx.configure_generation(), + ctx.configure_content_generation(), + ); + state.phase = WatcherDrainPhase::Apply { + stage: WatcherDrainApplyPhase::PendingTier2, + paths: [root.join("parked.rs")].into(), + remaining: 1, + oversized_inline_batch: false, + }; + *ctx.watcher_drain_slice().lock() = Some(state); + let (tx, rx) = crossbeam_channel::unbounded(); + *ctx.watcher_rx().lock() = Some(rx); + let ticket = driver.begin_watcher_cookie().unwrap(); + tx.send(WatcherDispatchEvent::CookieObserved( + ticket.path().to_path_buf(), + )) + .unwrap(); + drain_watcher_to_completion_for_test(&ctx, 1); + assert!(ticket.wait_timeout(Duration::ZERO).unwrap().revision > 0); + assert!(driver + .installed_snapshot() + .pending_intent() + .any(|path| path == &RelPath::new("parked.rs").unwrap())); +} + +#[test] +fn watcher_cookie_coalesced_rescan_gap_still_invalidates_the_current_driver() { + use crate::watcher_filter::RescanReason; + let temp = tempfile::tempdir().unwrap(); + let storage = tempfile::tempdir().unwrap(); + let root = temp.path().canonicalize().unwrap(); + let (driver, access) = driver(&root, storage.path()); + let ctx = context(&root, &driver); + let (_tx, rx) = crossbeam_channel::unbounded(); + *ctx.watcher_rx().lock() = Some(rx); + ctx.watcher_counters().start_rescan(); + assert!(ctx + .watcher_counters() + .note_overflow(RescanReason::BufferOverflow, vec![])); + let before = driver.revision(&access); + drain_watcher_events_bounded(&ctx, 1); + assert!(driver.revision(&access) > before); + assert_eq!( + driver.installed_snapshot().watcher(), + WatcherState::Overflowed + ); + assert_eq!( + ctx.watcher_counters() + .snapshot() + .rescans_buffer_overflow_total, + 0 + ); + assert_eq!( + ctx.watcher_counters().finish_rescan_walk(), + Some(RescanReason::BufferOverflow) + ); +} diff --git a/crates/aft/src/watcher_backend/flush.rs b/crates/aft/src/watcher_backend/flush.rs index 8e67664552..2f0be61180 100644 --- a/crates/aft/src/watcher_backend/flush.rs +++ b/crates/aft/src/watcher_backend/flush.rs @@ -231,7 +231,7 @@ impl FlushControl { if event.need_rescan() { return false; } - let marker_event = event.paths.iter().any(|path| is_marker_path(path)); + let marker_event = event.paths.iter().any(|path| is_flush_marker_path(path)); if !marker_event { return false; } @@ -253,6 +253,12 @@ impl FlushControl { } pub(crate) fn is_marker_path(path: &Path) -> bool { + // Corpus admission excludes both protocols, but command-report intake must + // not consume a views cookie before it reaches the driver's ordered queue. + super::cookie::is_cookie_path(path) || is_flush_marker_path(path) +} + +fn is_flush_marker_path(path: &Path) -> bool { let Some(name) = path.file_name().and_then(|name| name.to_str()) else { return false; }; diff --git a/crates/aft/src/watcher_backend/fsevents.rs b/crates/aft/src/watcher_backend/fsevents.rs index 8d29737957..e28a5a50f0 100644 --- a/crates/aft/src/watcher_backend/fsevents.rs +++ b/crates/aft/src/watcher_backend/fsevents.rs @@ -1878,6 +1878,7 @@ mod tests { Ok( WatcherDispatchEvent::Paths(_) | WatcherDispatchEvent::ContentOnlyPaths(_) + | WatcherDispatchEvent::CookieObserved(_) | WatcherDispatchEvent::IgnoreRulesChanged { .. } | WatcherDispatchEvent::RootDeleted, ) => {} diff --git a/crates/aft/src/watcher_backend/inotify.rs b/crates/aft/src/watcher_backend/inotify.rs index e133f89270..fc529f5d22 100644 --- a/crates/aft/src/watcher_backend/inotify.rs +++ b/crates/aft/src/watcher_backend/inotify.rs @@ -335,12 +335,11 @@ impl ProjectWatcher { match backend_rx.recv_timeout(BACKEND_POLL_INTERVAL) { Ok(Ok(event)) => { - if counters.bash_changes.get().is_some() - && event - .paths - .iter() - .any(|path| super::flush::is_marker_path(path)) - { + if event.paths.iter().any(|path| { + super::cookie::is_cookie_path(path) + || (counters.bash_changes.get().is_some() + && super::flush::is_marker_path(path)) + }) { // New-directory recovery waits briefly for duplicate kernel // events. Publish its files before the ordered marker too. if !new_directory_files.files.is_empty() @@ -813,6 +812,47 @@ fn desired_watch_directories( #[cfg(test)] pub(super) mod tests { + #[test] + fn watcher_cookie_inotify_recovery_precedes_root_cookie_without_command_tracking() { + let temp = tempfile::tempdir().unwrap(); + let root = temp.path().canonicalize().unwrap(); + let staging = tempfile::tempdir().unwrap(); + let tree = staging.path().join("tree"); + std::fs::create_dir(&tree).unwrap(); + std::fs::write(tree.join("source.rs"), b"already populated").unwrap(); + let counters = crate::context::watcher_counters_for_root(&root); + assert!(counters.bash_changes.get().is_none()); + let (tx, rx) = mpsc::channel(); + let _watcher = ProjectWatcher::create( + root.clone(), + vec![], + tx, + Arc::new(RwLock::new(None)), + Arc::new(AtomicU64::new(0)), + ) + .unwrap(); + std::fs::rename(&tree, root.join("tree")).unwrap(); + let cookie = root.join(".aft-cookie-123-nonce"); + std::fs::write(&cookie, b"").unwrap(); + let deadline = Instant::now() + Duration::from_secs(5); + let mut recovered = false; + loop { + let event = rx + .recv_timeout(deadline.saturating_duration_since(Instant::now())) + .unwrap() + .unwrap(); + assert!(!event.need_rescan()); + recovered |= event.paths.contains(&root.join("tree/source.rs")); + if event.paths.contains(&cookie) { + assert!( + recovered, + "cookie overtook recovered new-directory membership" + ); + break; + } + } + } + use std::sync::{Condvar, Mutex, OnceLock, RwLock}; use std::time::{Duration, Instant}; diff --git a/crates/aft/src/watcher_backend/mod.rs b/crates/aft/src/watcher_backend/mod.rs index 5785c74302..1f901c54d6 100644 --- a/crates/aft/src/watcher_backend/mod.rs +++ b/crates/aft/src/watcher_backend/mod.rs @@ -3,6 +3,9 @@ use std::sync::atomic::AtomicU64; use std::sync::{mpsc, Arc}; use crate::watcher_filter::SharedGitignore; +pub(crate) mod cookie; +#[cfg(test)] +mod cookie_tests; pub(crate) mod flush; #[cfg(target_os = "macos")] diff --git a/crates/aft/src/watcher_filter.rs b/crates/aft/src/watcher_filter.rs index eb4933d394..b91224fe79 100644 --- a/crates/aft/src/watcher_filter.rs +++ b/crates/aft/src/watcher_filter.rs @@ -180,6 +180,8 @@ pub enum WatcherDispatchEvent { Paths(Vec), /// Every path had only explicit data modifications, never topology events. ContentOnlyPaths(Vec), + /// Ordered control; preceding changes must enter the driver before its ack. + CookieObserved(PathBuf), RescanRequired(RescanReason), IgnoreRulesChanged { paths: Vec, @@ -285,7 +287,9 @@ pub fn watcher_event_invalidates(kind: ¬ify::EventKind) -> bool { pub fn watcher_path_is_infra_skip(path: &Path) -> bool { // Operating-system metadata files (a Finder `.DS_Store` rewrite, say) are // never indexed by the project walks, so their events carry no change. - if crate::project_content::is_excluded_event(path) { + if crate::watcher_backend::cookie::is_cookie_path(path) + || crate::project_content::is_excluded_event(path) + { return true; } watcher_path_has_infra_component(path) @@ -2403,10 +2407,11 @@ impl WatcherFilterThread { } match raw_rx.recv_timeout(self.next_recv_timeout()) { - Ok(Ok(event)) => { + Ok(Ok(mut event)) => { #[cfg(feature = "test-timing-hooks")] hold_watcher_intake_for_test(&self.shutdown); self.config.counters.note_raw_event(); + let cookies = self.take_cookie_paths(&mut event); if self.config.counters.bash_changes.get().is_some() && self.config.counters.bash_flush.observe(&event) { @@ -2460,6 +2465,14 @@ impl WatcherFilterThread { return; } } + for cookie in cookies { + if !self.flush_pending() + || self.send_dispatch(WatcherDispatchEvent::CookieObserved(cookie)) + == Dispatch::Closed + { + return; + } + } } Ok(Err(error)) => { if let Some(journal) = self.config.counters.bash_changes.get() { @@ -2486,6 +2499,33 @@ impl WatcherFilterThread { } } + fn take_cookie_paths(&self, event: &mut notify::Event) -> Vec { + let acknowledge = + !event.need_rescan() && matches!(event.kind, notify::EventKind::Create(_)); + let mut cookies = Vec::new(); + event.paths.retain(|path| { + if !crate::watcher_backend::cookie::is_cookie_path(path) { + return true; + } + // Only root cookies are controls. Do not unlink a similarly named + // file in an auxiliary watch or a nested directory. + if path.parent() == Some(self.config.project_root.as_path()) { + let removed = std::fs::remove_file(path).or_else(|error| { + if error.kind() == std::io::ErrorKind::NotFound { + Ok(()) + } else { + Err(error) + } + }); + if acknowledge && removed.is_ok() { + cookies.push(path.clone()); + } + } + false + }); + cookies + } + fn project_root_was_deleted(&mut self, now: Instant) -> bool { // Check on a wall-clock cadence, not once per raw event. Ignored build // bursts can otherwise turn the root-deletion fallback into a stat loop. @@ -2705,6 +2745,7 @@ impl WatcherFilterThread { .strip_prefix(&self.config.project_root) .is_ok_and(watcher_path_is_high_churn_infra) || crate::os_metadata::is_os_metadata_path(&path) + || crate::watcher_backend::cookie::is_cookie_path(&path) { continue; } @@ -2981,12 +3022,21 @@ impl WatcherFilterThread { WatcherDispatchEvent::Paths(paths) => self.hold_paths(paths, false), WatcherDispatchEvent::ContentOnlyPaths(paths) => self.hold_paths(paths, true), WatcherDispatchEvent::RescanRequired(reason) => { - self.backlog.retain(|held| !is_path_batch(held)); + // Losing the paths before a held cookie also loses its proof. + self.backlog.retain(|held| { + !is_path_batch(held) && !matches!(held, WatcherDispatchEvent::CookieObserved(_)) + }); if !rescan_held { self.backlog .push_back(WatcherDispatchEvent::RescanRequired(reason)); } } + WatcherDispatchEvent::CookieObserved(path) => { + if !rescan_held { + self.backlog + .push_back(WatcherDispatchEvent::CookieObserved(path)); + } + } WatcherDispatchEvent::IgnoreRulesChanged { paths } => match self.backlog.back_mut() { Some(WatcherDispatchEvent::IgnoreRulesChanged { paths: held }) => { merge_paths(held, paths); @@ -3036,7 +3086,9 @@ impl WatcherFilterThread { // logged and counted like an operating-system event overflow. A // rescan already running is told to run again instead. let during_rescan = self.log_overflow(RescanReason::BufferOverflow); - self.backlog.retain(|held| !is_path_batch(held)); + self.backlog.retain(|held| { + !is_path_batch(held) && !matches!(held, WatcherDispatchEvent::CookieObserved(_)) + }); if !during_rescan { self.backlog.push_back(WatcherDispatchEvent::RescanRequired( RescanReason::BufferOverflow, @@ -3149,6 +3201,116 @@ pub(crate) fn take_watcher_overflow_logs_for_test() -> Vec { #[cfg(test)] mod tests { + #[test] + fn watcher_cookie_bypasses_ignore_and_flushes_mixed_paths_before_ack() { + let temp = tempfile::tempdir().unwrap(); + let root = temp.path().canonicalize().unwrap(); + let source = root.join("source.rs"); + let cookie = root.join(".aft-cookie-123-nonce"); + std::fs::write(&source, b"source").unwrap(); + std::fs::write(&cookie, b"").unwrap(); + let mut builder = ignore::gitignore::GitignoreBuilder::new(&root); + builder.add_line(None, ".aft-cookie-*").unwrap(); + let matcher = Arc::new(RwLock::new(Some(Arc::new(builder.build().unwrap())))); + let (tx, rx) = watcher_dispatch_channel(); + let (raw_tx, raw_rx) = mpsc::channel(); + let shutdown = Arc::new(AtomicBool::new(false)); + let mut filter = WatcherFilterThread::new( + WatcherFilterConfig::new(root.clone(), None), + matcher, + Arc::new(AtomicU64::new(0)), + tx, + Arc::clone(&shutdown), + ); + let worker = thread::spawn(move || filter.run(raw_rx)); + raw_tx + .send(Ok(notify::Event::new(notify::EventKind::Create( + notify::event::CreateKind::File, + )) + .add_path(source.clone()) + .add_path(cookie.clone()))) + .unwrap(); + assert_eq!( + rx.recv_timeout(Duration::from_secs(5)).unwrap(), + WatcherDispatchEvent::Paths(vec![source]) + ); + assert_eq!( + rx.recv_timeout(Duration::from_secs(5)).unwrap(), + WatcherDispatchEvent::CookieObserved(cookie.clone()) + ); + assert!(!cookie.exists()); + shutdown.store(true, Ordering::SeqCst); + drop(raw_tx); + worker.join().unwrap(); + } + + #[test] + fn watcher_cookie_unlinks_on_overflow_but_only_create_can_ack() { + let temp = tempfile::tempdir().unwrap(); + let root = temp.path().canonicalize().unwrap(); + let (filter, _rx, _) = filter_with_full_dispatch_channel(&root); + let cookie = root.join(".aft-cookie-123-nonce"); + for kind in [ + notify::EventKind::Remove(notify::event::RemoveKind::File), + notify::EventKind::Any, + ] { + std::fs::write(&cookie, b"").unwrap(); + let mut event = notify::Event::new(kind).add_path(cookie.clone()); + assert!(filter.take_cookie_paths(&mut event).is_empty()); + assert!(event.paths.is_empty()); + assert!(!cookie.exists()); + } + std::fs::write(&cookie, b"").unwrap(); + let mut event = + notify::Event::new(notify::EventKind::Create(notify::event::CreateKind::File)) + .add_path(cookie.clone()) + .set_flag(notify::event::Flag::Rescan); + assert!(filter.take_cookie_paths(&mut event).is_empty()); + assert!(event.need_rescan()); + assert!(!cookie.exists()); + let nested = root.join("nested/.aft-cookie-123-nonce"); + std::fs::create_dir_all(nested.parent().unwrap()).unwrap(); + std::fs::write(&nested, b"not a root control").unwrap(); + let mut event = + notify::Event::new(notify::EventKind::Create(notify::event::CreateKind::File)) + .add_path(nested.clone()); + assert!(filter.take_cookie_paths(&mut event).is_empty()); + assert!(nested.exists()); + } + + #[test] + fn watcher_cookie_backlog_preserves_barriers_and_drops_them_with_lost_paths() { + let temp = tempfile::tempdir().unwrap(); + let root = temp.path().canonicalize().unwrap(); + let (mut filter, _rx, _) = filter_with_full_dispatch_channel(&root); + filter.send_dispatch(WatcherDispatchEvent::Paths(vec![root.join("a.rs")])); + filter.send_dispatch(WatcherDispatchEvent::CookieObserved( + root.join(".aft-cookie-1-nonce"), + )); + filter.send_dispatch(WatcherDispatchEvent::Paths(vec![root.join("b.rs")])); + assert_eq!( + filter.backlog.len(), + 3, + "paths must not merge across the cookie" + ); + filter.send_dispatch(WatcherDispatchEvent::RescanRequired( + RescanReason::BufferOverflow, + )); + assert_eq!(filter.backlog.len(), 1); + assert!(matches!( + filter.backlog.front(), + Some(WatcherDispatchEvent::RescanRequired(_)) + )); + filter.send_dispatch(WatcherDispatchEvent::CookieObserved( + root.join(".aft-cookie-1-next"), + )); + assert_eq!( + filter.backlog.len(), + 1, + "a held rescan cannot certify a cookie" + ); + } + #[test] fn issue645_nested_git_and_local_manifests_get_kernel_exclusions() { use super::*; diff --git a/docs/design/views-4v-frontier-audit.md b/docs/design/views-4v-frontier-audit.md index 3c171f6ff8..bf4fa4a194 100644 --- a/docs/design/views-4v-frontier-audit.md +++ b/docs/design/views-4v-frontier-audit.md @@ -43,11 +43,14 @@ before acknowledging the exact registered cookie. A rescan must enter the driver as `WatcherState::Overflowed` and invalidate an in-flight installation; a subsequent ordinary path must not reset that state to Healthy. A cookie held in a backlog whose paths are discarded must be discarded too (timeout, not a -false successful frontier). This provides an ordered **delivery acknowledgment**, +false successful frontier). Drops coalesced during a legacy rescan also set a +pending driver-gap signal, consumed at drain entry and before a cookie ack; +legacy rescan work coalescing must not hide a gap from a newly installed driver. +This provides an ordered **delivery acknowledgment**, not ready publication: the bind owner must still establish E0, reject any gap, replay through the acknowledged cut, and revalidate before publication. -The planned additive implementation includes regression tests named +The additive implementation includes regression tests named `watcher_cookie_delayed_delivery_past_read_inotify`, `watcher_cookie_delayed_delivery_past_read_fsevents`, and `watcher_cookie_delayed_delivery_past_read_windows`, using a backend-delivery shim From 4323bb29610596451c53da100a3e8ef4cb96282e Mon Sep 17 00:00:00 2001 From: ualtinok <94532+ualtinok@users.noreply.github.com> Date: Mon, 26 Oct 2026 14:50:53 +0100 Subject: [PATCH 3/3] release notes and search-quality descriptor for views-4v-2-watcher-cookies Co-authored-by: alfonso-aft <289616620+alfonso-aft@users.noreply.github.com> --- .../land-views-4v-2-watcher-cookies.json | 8 ++++++++ 1 file changed, 8 insertions(+) create mode 100644 benchmarks/aft-search/slice-descriptors/land-views-4v-2-watcher-cookies.json diff --git a/benchmarks/aft-search/slice-descriptors/land-views-4v-2-watcher-cookies.json b/benchmarks/aft-search/slice-descriptors/land-views-4v-2-watcher-cookies.json new file mode 100644 index 0000000000..ebd574711e --- /dev/null +++ b/benchmarks/aft-search/slice-descriptors/land-views-4v-2-watcher-cookies.json @@ -0,0 +1,8 @@ +{ + "targeted_mechanism": "none", + "kind": "harness", + "slice_class": "non_ranking", + "fixtures": [ + "harness-goldens" + ] +}