land: views-4v-2-watcher-cookies

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"
+  ]
+}

crates/aft/src/context.rs

@@ -1720,6 +1720,9 @@ pub(crate) struct WatcherCounters {
     pub(crate) bash_flush: Arc<crate::watcher_backend::flush::FlushControl>,
     pub(crate) bash_changes: std::sync::OnceLock<Arc<crate::bash_changed_files::Journal>>,
     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<crate::executor::view_publication::PublicationRetry> {

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;

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<ReconcileWork>,
 }
@@ -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<CookieTicket> {
+        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(&current.composed, &held.composed));
+        assert!(Arc::ptr_eq(&current.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();