fix(resync): close watch and carrier gaps

Author: schickling

docs/vrs/06-resync/spec.md

@@ -52,8 +52,9 @@ Resource projection so absence of coverage is observable, never silent.
 
 Windows are constants in one place, named as provisional, and tuned by
 observed notification volume (issue #341 rollout note). A burst inside a
-window collapses to one pass; within that pass each changed carrier still
-gets its own event so per-binding supersession stays meaningful.
+window collapses to one pass; when subscribers of different classes share a
+path, each remains dirty until its own class window. Within one due class each
+changed carrier gets its own event so per-binding supersession stays meaningful.
 
 ## Watcher mechanics
 
@@ -71,11 +72,13 @@ gets its own event so per-binding supersession stays meaningful.
   bus id, label, and class so every subscriber retains its digest and pending
   dirty state even when multiple bindings share one path; only new
   subscriptions seed silently.
-- A previously blind path is digest-diffed before its recovered parent directory is added to the
-  registered-watch set, covering changes made during the blind interval.
+- A previously blind path is digest-diffed both before and after its recovered
+  parent watch is registered, closing the poll-to-registration gap.
 - Installation failure is diagnosed once and degrades to timer-based digest
   polling over the watch set (bounded by the number of bindings), never to
   silence about the mechanism.
+- Digest reads open carriers nonblocking and accept regular files only; FIFOs and other special
+  files cannot stall the worker.
 
 ## Built-in stream
 

src/resync.rs

@@ -328,7 +328,10 @@ impl Worker {
         // Diff paths that were blind before registering newly recovered parents; otherwise the
         // new watch suppresses polling of mutations that happened during the blind interval.
         self.poll_unwatched();
-        self.refresh_watches();
+        let registered = self.refresh_watches();
+        // Registration closes the event gap first; this second digest pass covers writes between
+        // the pre-registration poll and watch installation.
+        self.poll_registered(&registered);
     }
 
     /// Digest-diff every entry in `next`, emitting observed transitions. Seeding stays silent
@@ -377,15 +380,31 @@ impl Worker {
             .cloned()
             .collect();
         for path in unwatched {
-            self.flush_path(&path);
+            self.flush_path(&path, None);
+        }
+    }
+
+    fn poll_registered(&mut self, directories: &[PathBuf]) {
+        let paths: Vec<PathBuf> = self
+            .carriers
+            .keys()
+            .filter(|path| {
+                path.parent()
+                    .is_some_and(|parent| directories.iter().any(|dir| dir == parent))
+            })
+            .cloned()
+            .collect();
+        for path in paths {
+            self.flush_path(&path, None);
         }
     }
 
     /// Re-register watches so every distinct parent directory of the current watch set is covered,
     /// dropping directories that left the set or were replaced (identity change). A replaced watch
     /// stays blind until the next pass rebuilds it — bounded by the reconcile interval, the same
     /// tradeoff `CatalogDeclarationWatcher` accepts for declarations.
-    fn refresh_watches(&mut self) {
+    fn refresh_watches(&mut self) -> Vec<PathBuf> {
+        let mut registered = Vec::new();
         let mut desired: Vec<PathBuf> = Vec::new();
         for path in self.carriers.keys() {
             if let Some(parent) = path.parent() {
@@ -418,9 +437,11 @@ impl Worker {
                 .watch(&dir, notify::RecursiveMode::NonRecursive)
                 .is_ok()
             {
+                registered.push(dir.clone());
                 self.watched.insert(dir, identity);
             }
         }
+        registered
     }
 
     fn mark_mutated(&mut self, paths: Vec<PathBuf>) {
@@ -459,7 +480,8 @@ impl Worker {
             drop(dirty_here);
         }
         if extend {
-            self.refresh_watches();
+            let registered = self.refresh_watches();
+            self.poll_registered(&registered);
         }
     }
 
@@ -483,19 +505,23 @@ impl Worker {
                 .map(|(path, _)| path.clone())
                 .collect();
             for path in targets {
-                self.flush_path(&path);
+                self.flush_path(&path, Some(class));
             }
         }
     }
 
-    /// Flush every subscriber of one path: diff digests, emit transitions, and retain failed
-    /// publications for retry with the same event identity.
-    fn flush_path(&mut self, path: &Path) {
+    /// Flush subscribers of one path whose class is due, or every subscriber for fallback polls:
+    /// diff digests, emit transitions, and retain failed publications for retry with the same
+    /// event identity.
+    fn flush_path(&mut self, path: &Path, due_class: Option<CarrierClass>) {
         let Some(entries) = self.carriers.get_mut(path) else {
             return;
         };
         let mut retries = Vec::new();
         for entry in entries.iter_mut() {
+            if due_class.is_some_and(|class| entry.class != class) {
+                continue;
+            }
             entry.dirty = false;
             let Some(new_digest) = read_digest(path) else {
                 // Unreadable right now (deleted mid-window): stay quiet and keep the previous
@@ -573,10 +599,36 @@ fn emit_resync(
 }
 
 fn read_digest(path: &Path) -> Option<String> {
-    let bytes = std::fs::read(path).ok()?;
+    let bytes = read_regular(path)?;
     Some(format!("{:x}", Sha256::digest(&bytes)))
 }
 
+#[cfg(unix)]
+fn read_regular(path: &Path) -> Option<Vec<u8>> {
+    use std::io::Read as _;
+    use std::os::unix::fs::OpenOptionsExt as _;
+
+    let mut file = std::fs::OpenOptions::new()
+        .read(true)
+        .custom_flags(libc::O_NOFOLLOW | libc::O_CLOEXEC | libc::O_NONBLOCK)
+        .open(path)
+        .ok()?;
+    if !file.metadata().ok()?.file_type().is_file() {
+        return None;
+    }
+    let mut bytes = Vec::new();
+    file.read_to_end(&mut bytes).ok()?;
+    Some(bytes)
+}
+
+#[cfg(not(unix))]
+fn read_regular(path: &Path) -> Option<Vec<u8>> {
+    if !std::fs::metadata(path).ok()?.file_type().is_file() {
+        return None;
+    }
+    std::fs::read(path).ok()
+}
+
 fn render_body(label: &str, path: &Path, old: Option<&str>, new: &str) -> String {
     format!(
         "resource `{label}` changed\n\nbinding: {label}\npath: {}\nold: {}\nnew: {new}\n",
@@ -694,6 +746,51 @@ mod tests {
         }
     }
 
+    #[cfg(unix)]
+    #[test]
+    fn digesting_a_fifo_fails_without_blocking_the_worker() {
+        use std::os::unix::ffi::OsStrExt as _;
+
+        let tmp = tempfile::tempdir().unwrap();
+        let fifo = tmp.path().join("carrier.fifo");
+        let fifo_c = std::ffi::CString::new(fifo.as_os_str().as_bytes()).unwrap();
+        // SAFETY: the path is NUL-terminated and points into the live temp directory.
+        assert_eq!(unsafe { libc::mkfifo(fifo_c.as_ptr(), 0o600) }, 0);
+        assert_eq!(read_digest(&fifo), None);
+    }
+
+    #[test]
+    fn due_flush_only_clears_subscribers_of_the_due_class() {
+        let root = tempfile::tempdir().unwrap();
+        let carrier = root.path().join("carrier.md");
+        std::fs::write(&carrier, "same bytes").unwrap();
+        let digest = read_digest(&carrier);
+        let entries = [CarrierClass::Immediate, CarrierClass::Coalesced]
+            .into_iter()
+            .map(|class| Entry {
+                bus_id: "host.worker".to_owned(),
+                label: format!("{class:?}"),
+                class,
+                digest: digest.clone(),
+                dirty: true,
+            })
+            .collect();
+        let now = Instant::now();
+        let mut worker = Worker {
+            root: root.path().to_path_buf(),
+            this_host: "host".to_owned(),
+            carriers: BTreeMap::from([(carrier, entries)]),
+            deadlines: BTreeMap::from([(CarrierClass::Immediate, now)]),
+            watched: BTreeMap::new(),
+            watcher: None,
+        };
+
+        worker.flush_due(now);
+        let entries = worker.carriers.values().next().unwrap();
+        assert!(!entries[0].dirty);
+        assert!(entries[1].dirty, "coalesced subscriber must wait for its own deadline");
+    }
+
     #[test]
     fn failed_emit_retains_digest_and_schedules_the_same_transition_for_retry() {
         let root = tempfile::tempdir().unwrap();
@@ -717,7 +814,7 @@ mod tests {
             watcher: None,
         };
 
-        worker.flush_path(&carrier);
+        worker.flush_path(&carrier, None);
         let entry = &worker.carriers[&carrier][0];
         assert_eq!(entry.digest.as_deref(), Some("old-digest"));
         assert!(entry.dirty);