fix(daemon): stream the dirty check past any number of no-op rows

engine/crates/layerfs-api/tests/unmount_dirty.rs

@@ -43,6 +43,7 @@ fn name(value: &str) -> PathName {
 struct Serials {
     root: u64,
     git: u64,
+    git_mode: u32,
     index: u64,
 }
 
@@ -71,13 +72,17 @@ fn refusal(tag: u8, operations: impl Fn(Serials) -> Vec<Operation>) -> Unmounted
     let root = view.root_serial();
     let git = view
         .lookup(operation.overlay(), root, &name(".git"))
-        .unwrap()
-        .serial;
+        .unwrap();
     let index = view
-        .lookup(operation.overlay(), git, &name("index"))
+        .lookup(operation.overlay(), git.serial, &name("index"))
         .unwrap()
         .serial;
-    for change in operations(Serials { root, git, index }) {
+    for change in operations(Serials {
+        root,
+        git: git.serial,
+        git_mode: git.metadata.mode,
+        index,
+    }) {
         let changed = operation
             .workspace()
             .mutate(operation.overlay(), operation.ports(), &view, change, NOW)
@@ -196,3 +201,122 @@ fn a_refusal_names_every_change_it_counts() {
     let touched = refusal(54, |at| vec![mtime(at.index)]);
     assert_eq!(named(touched), (vec!["/.git/index".to_owned()], false));
 }
+
+/// A created inode no name binds any more: one no-op row per pair.
+fn no_op_rows(count: usize, root: u64) -> Vec<Operation> {
+    let mut operations = Vec::with_capacity(count * 2);
+    for _ in 0..count {
+        operations.push(Operation::Create {
+            parent: root,
+            name: name(".probe"),
+            mode: 0o644,
+        });
+        operations.push(Operation::Unlink {
+            parent: root,
+            name: name(".probe"),
+        });
+    }
+    operations
+}
+
+#[test]
+fn no_op_rows_past_the_old_cap_do_not_refuse_an_unmount() {
+    let refused = refusal(61, |at| no_op_rows(4100, at.root));
+    assert!(matches!(refused, Unmounted::Closed), "{refused:?}");
+}
+
+#[test]
+fn a_change_after_many_no_op_rows_is_named() {
+    let refused = refusal(62, |at| {
+        let mut operations = no_op_rows(4100, at.root);
+        operations.push(Operation::Create {
+            parent: at.root,
+            name: name("kept"),
+            mode: 0o644,
+        });
+        operations
+    });
+    assert_eq!(named(refused), (vec!["/kept".to_owned()], false));
+}
+
+#[test]
+fn exactly_the_old_cap_of_no_op_rows_is_not_truncated() {
+    let refused = refusal(63, |at| {
+        let mut operations = no_op_rows(4096, at.root);
+        operations.push(Operation::Create {
+            parent: at.root,
+            name: name("kept"),
+            mode: 0o644,
+        });
+        operations
+    });
+    assert_eq!(named(refused), (vec!["/kept".to_owned()], false));
+}
+
+#[test]
+fn a_directory_restored_to_its_base_mode_does_not_refuse() {
+    let clean = refusal(64, |at| {
+        vec![
+            Operation::SetAttributes {
+                serial: at.git,
+                mode: Some(0o711),
+                mtime: None,
+                size: None,
+            },
+            Operation::SetAttributes {
+                serial: at.git,
+                mode: Some(at.git_mode),
+                mtime: None,
+                size: None,
+            },
+        ]
+    });
+    assert!(matches!(clean, Unmounted::Closed), "{clean:?}");
+}
+
+#[test]
+fn more_changes_than_the_report_names_set_truncated() {
+    let refused = refusal(66, |at| {
+        (0..300)
+            .map(|n| Operation::Create {
+                parent: at.root,
+                name: name(&format!(".f{n:03}")),
+                mode: 0o644,
+            })
+            .collect()
+    });
+    let (paths, truncated) = named(refused);
+    assert_eq!(paths.len(), 256);
+    assert!(truncated, "300 changes do not fit one report");
+}
+
+#[test]
+fn removing_a_base_directory_refuses_and_names_it() {
+    let refused = refusal(65, |at| {
+        vec![
+            Operation::Unlink {
+                parent: at.git,
+                name: name("index"),
+            },
+            Operation::Unlink {
+                parent: at.git,
+                name: name("link"),
+            },
+            Operation::Rmdir {
+                parent: at.root,
+                name: name(".git"),
+            },
+        ]
+    });
+    assert_eq!(
+        named(refused),
+        (
+            vec![
+                "/.git".to_owned(),
+                "/.git/index".to_owned(),
+                "/.git/link".to_owned(),
+            ],
+            false,
+        )
+    );
+}

engine/crates/layerfs-daemon/src/control/dirty.rs

@@ -9,7 +9,9 @@
 //! no change: an inode created locally that no name binds any more, and a
 //! base directory whose mode equals the base, since its names are their own
 //! rows and Commit ignores its mtime. A path is found among the active
-//! bindings first, then by a bounded walk of the local base. At most
+//! bindings first, then by a bounded walk of the local base. Both indexed
+//! windows are streamed to the end in bounded pages and only paths are
+//! retained, so the number of no-op rows is not a cap on the check. At most
 //! [`PATHS`] paths are named; the report says when some are not. The check is
 //! an observation before terminal admission, so a write that lands after it
 //! is closed with the Workspace.
@@ -23,8 +25,6 @@ use std::collections::{BTreeMap, BTreeSet};
 
 /// Paths one report names.
 const PATHS: usize = 256;
-/// Changed rows one report reads, inodes and names each.
-const ROWS: usize = 4096;
 /// Base directories one path walk may list.
 const WALK: usize = 4096;
 /// Deepest local path resolved.
@@ -82,25 +82,40 @@ impl Service {
             return Ok(None);
         }
         let mut truncated = false;
+        // A changed name binding is a change a Commit keeps: a new name, a
+        // rename or a whiteout. One row names one path, so more than [`PATHS`]
+        // rows prove the report cannot list them all and the scan stops.
         let mut names = Vec::new();
         let mut after: Option<(u64, Vec<u8>)> = None;
-        while names.len() < ROWS {
+        while !truncated {
             let window = self.job(route, |client, route| {
                 client.active_directory_entries(route, after.clone())
             })?;
             let Some(last) = window.last() else { break };
             after = Some((last.parent, last.name.clone()));
             names.extend(window);
+            truncated = names.len() > PATHS;
         }
+        let reader = self.store.ports(binding.scope);
+        let failed = |error: layerfs_content::ContentError| {
+            Failure::Workspace(layerfs_workspace::WorkspaceError::Service(Box::new(error)))
+        };
+        let base = ObjectId::from_bytes(&state.base_root).map_err(failed)?;
+        let tree = TreeTable::open(reader.as_ref(), FilesystemRootId(base)).map_err(failed)?;
+        let mut work = AttributeReadWork::default();
+        // Changed inodes are streamed to the end too. A locally created inode
+        // nothing binds any more, and a base directory whose mode equals the
+        // base, are no change, and however many such rows exist they neither
+        // refuse the Unmount nor mark the report truncated. Base directories
+        // are looked up one window of serials at a time.
         let mut files = Vec::new();
-        let mut directories = Vec::new();
-        let mut rows = 0;
+        let mut modes = Vec::new();
         let mut cursor = 0;
-        while rows < ROWS {
+        while !truncated {
             let window = self.job(route, |client, route| client.active_inodes(route, cursor))?;
             let Some(last) = window.last() else { break };
             cursor = last.serial;
-            rows += window.len();
+            let mut directories = Vec::new();
             for inode in window {
                 let created = inode.born > state.installed as u64;
                 if inode.kind != layerfs_overlay::InodeKind::Directory {
@@ -111,29 +126,26 @@ impl Service {
                     directories.push((inode.serial, u32::from(inode.mode)));
                 }
             }
-        }
-        truncated |= names.len() >= ROWS || rows >= ROWS;
-        let reader = self.store.ports(binding.scope);
-        let failed = |error: layerfs_content::ContentError| {
-            Failure::Workspace(layerfs_workspace::WorkspaceError::Service(Box::new(error)))
-        };
-        let base = ObjectId::from_bytes(&state.base_root).map_err(failed)?;
-        let tree = TreeTable::open(reader.as_ref(), FilesystemRootId(base)).map_err(failed)?;
-        let mut work = AttributeReadWork::default();
-        let mut modes = Vec::new();
-        for (serial, mode) in directories {
-            let same = match tree.value(reader.as_ref(), serial).map_err(failed)? {
-                Some(value) => {
-                    read_portable(reader.as_ref(), value.metadata_root, value.kind, &mut work)
-                        .map_err(failed)?
-                        .mode
-                        == mode
+            let serials = directories
+                .iter()
+                .map(|(serial, _)| *serial)
+                .collect::<Vec<_>>();
+            let values = tree.values(reader.as_ref(), &serials).map_err(failed)?;
+            for ((serial, mode), value) in directories.into_iter().zip(values) {
+                let same = match value {
+                    Some(value) => {
+                        read_portable(reader.as_ref(), value.metadata_root, value.kind, &mut work)
+                            .map_err(failed)?
+                            .mode
+                            == mode
+                    }
+                    None => false,
+                };
+                if !same {
+                    modes.push(serial);
                 }
-                None => false,
-            };
-            if !same {
-                modes.push(serial);
             }
+            truncated = files.len() + modes.len() > PATHS;
         }
         if names.is_empty()
             && files.is_empty()