fix: remove capture plans no caller will seal

README.md

@@ -366,6 +366,8 @@ Plans enumerate metadata; policy and Git control files are read to determine
 eligibility. Known SQLite databases and their existing sidecars are included
 because consistent backups may read them even on a delta. Tokens are scoped to
 one timeline, folder, composed directory, and head, and are consumed by capture.
+A plan not sealed within ten minutes is treated as abandoned: the next plan on its
+timeline, or an idle daemon, removes it, and sealing it asks for a re-plan.
 
 The daemon verbs are `capture-plan` (`args.changed`, absent/null for unknown changes),
 `seal` (`args.changed`, optional `args.plan`, optional `args.durable`), and `sync`

furrow-core/src/repository.rs

@@ -6322,11 +6322,13 @@ mod unwatched_tests;
 mod windows_tests;
 
 mod capture_plan;
-pub use capture_plan::CapturePlan;
+pub use capture_plan::{prune_abandoned_plans, CapturePlan};
 mod change_tracking;
 
 #[cfg(test)]
 mod perf_tests;
+#[cfg(test)]
+mod plan_prune_tests;
 
 fn capture_rules_changed(root: &Path, path: &Path) -> bool {
     changed_path_matches(root, path, POLICY_FILE_BYTES)

furrow-core/src/repository/capture_plan.rs

@@ -1,6 +1,13 @@
 //! A persisted read scope binds a guard's screening to one capture.
 
 use super::*;
+use std::time::Duration;
+
+/// Callers seal within about thirty seconds of planning. A plan this old has
+/// been abandoned by a caller that re-planned, crashed, timed out or exited.
+const PLAN_LIFETIME: Duration = Duration::from_secs(10 * 60);
+
+const PLANS_DIR: &str = "capture-plans";
 
 #[derive(Debug, Clone, Serialize, Deserialize)]
 pub struct CapturePlan {
@@ -48,10 +55,56 @@ pub(super) struct PreparedPlan {
     candidates: BTreeMap<PathBuf, Option<Stamp>>,
 }
 
+/// Removes plans in one directory that are older than [`PLAN_LIFETIME`], and
+/// temporary files a writer left before renaming them into place. A seal that
+/// read its plan keeps the bytes; one that had not reads nothing and is asked
+/// to re-plan. Removal is best effort: a plan that remains is only disk space.
+fn prune_plans(directory: &Path, now: SystemTime) {
+    let Ok(entries) = fs::read_dir(directory) else {
+        return;
+    };
+    for entry in entries.flatten() {
+        let name = entry.file_name();
+        let name = name.to_string_lossy();
+        let ours = name.starts_with(".tmp")
+            || name.strip_suffix(".json").is_some_and(|token| {
+                token.len() == 64 && token.bytes().all(|b| b.is_ascii_hexdigit())
+            });
+        // A modification time in the future reads as fresh.
+        let expired = || {
+            entry
+                .metadata()
+                .and_then(|m| m.modified())
+                .is_ok_and(|modified| {
+                    now.duration_since(modified)
+                        .is_ok_and(|age| age > PLAN_LIFETIME)
+                })
+        };
+        if ours && expired() {
+            let _ = fs::remove_file(entry.path());
+        }
+    }
+}
+
+/// Removes abandoned plans from every timeline in a store, including ones
+/// that never plan again. Visits only the workspace directories and their
+/// plans, never the objects.
+pub fn prune_abandoned_plans(store_root: &Path) {
+    let Ok(workspaces) = fs::read_dir(store_root.join("workspaces")) else {
+        return;
+    };
+    let now = SystemTime::now();
+    for workspace in workspaces.flatten() {
+        prune_plans(&workspace.path().join(PLANS_DIR), now);
+    }
+}
+
 impl FurrowRepository {
     /// Plans enumerate names and metadata, never file contents to be captured.
     pub fn capture_plan(&self, changed: Option<&[PathBuf]>) -> anyhow::Result<CapturePlan> {
         let _operation = crate::writer_wait::Operation::start();
+        let directory = self.workspace_data_dir().join(PLANS_DIR);
+        prune_plans(&directory, SystemTime::now());
         let policy = self.load_policy()?;
         let (tracked, _) = self.tracked_changes()?;
         let watched = changed.is_none() && tracked.is_some();
@@ -110,7 +163,6 @@ impl FurrowRepository {
         }
         let bytes = serde_json::to_vec(&prepared)?;
         let token = blake3::hash(&bytes).to_hex().to_string();
-        let directory = self.workspace_data_dir().join("capture-plans");
         fs::create_dir_all(&directory)?;
         // Replacing a complete file is sufficient: plans are disposable and never authoritative state.
         let mut temporary = tempfile::NamedTempFile::new_in(&directory)?;
@@ -136,7 +188,7 @@ impl FurrowRepository {
         );
         let path = self
             .workspace_data_dir()
-            .join("capture-plans")
+            .join(PLANS_DIR)
             .join(format!("{token}.json"));
         let bytes = fs::read(&path).context("capture plan is absent; re-plan")?;
         anyhow::ensure!(

furrow-core/src/repository/plan_prune_tests.rs

@@ -0,0 +1,151 @@
+//! Plans a caller never seals are removed by later plans, without touching one
+//! a caller may still seal.
+
+use super::*;
+
+struct Fixture {
+    _temp: tempfile::TempDir,
+    root: PathBuf,
+    data: PathBuf,
+}
+
+impl Fixture {
+    fn new() -> Self {
+        let temp = tempfile::tempdir().unwrap();
+        let root = temp.path().join("tree");
+        let data = temp.path().join("data");
+        fs::create_dir(&root).unwrap();
+        fs::write(root.join("a"), b"first").unwrap();
+        fs::write(root.join("b"), b"other").unwrap();
+        Self {
+            _temp: temp,
+            root,
+            data,
+        }
+    }
+
+    fn attach(&self) -> FurrowRepository {
+        FurrowRepository::attach_timeline_in(&self.data, &self.root, Some("chat"), None).unwrap()
+    }
+}
+
+fn plans(repo: &FurrowRepository) -> PathBuf {
+    repo.workspace_data_dir().join("capture-plans")
+}
+
+fn plan_file(repo: &FurrowRepository, token: &str) -> PathBuf {
+    plans(repo).join(format!("{token}.json"))
+}
+
+/// A day is far past any time a caller takes between planning and sealing.
+fn age_by_a_day(path: &Path) {
+    let now = SystemTime::now().duration_since(UNIX_EPOCH).unwrap();
+    let old = FileTime::from_unix_time(now.as_secs() as i64 - 24 * 60 * 60, 0);
+    filetime::set_file_mtime(path, old).unwrap();
+}
+
+fn names(directory: &Path) -> BTreeSet<OsString> {
+    fs::read_dir(directory)
+        .unwrap()
+        .map(|entry| entry.unwrap().file_name())
+        .collect()
+}
+
+#[test]
+fn a_new_plan_removes_abandoned_plans_and_keeps_a_fresh_one() {
+    let f = Fixture::new();
+    let mut repo = f.attach();
+    repo.seal(None, SnapshotTrigger::AgentRun, None).unwrap();
+    let mut abandoned = Vec::new();
+    for round in 0..40 {
+        fs::write(f.root.join("a"), format!("round {round}")).unwrap();
+        let plan = repo.capture_plan(Some(&[PathBuf::from("a")])).unwrap();
+        abandoned.push(plan_file(&repo, &plan.token));
+    }
+    // A writer that died between writing its plan and renaming it into place.
+    let partial = plans(&repo).join(".tmpAbandoned");
+    fs::write(&partial, b"{").unwrap();
+    abandoned.push(partial);
+    // Something the engine did not write is not its to remove.
+    let foreign = plans(&repo).join("notes.txt");
+    fs::write(&foreign, b"kept").unwrap();
+    age_by_a_day(&foreign);
+    for path in &abandoned {
+        age_by_a_day(path);
+    }
+    fs::write(f.root.join("a"), b"settled").unwrap();
+    let fresh = repo.capture_plan(Some(&[PathBuf::from("a")])).unwrap();
+
+    let newest = repo.capture_plan(Some(&[PathBuf::from("b")])).unwrap();
+
+    for path in &abandoned {
+        assert!(!path.exists(), "{} was not removed", path.display());
+    }
+    assert_eq!(
+        names(&plans(&repo)),
+        BTreeSet::from([
+            OsString::from(format!("{}.json", fresh.token)),
+            OsString::from(format!("{}.json", newest.token)),
+            OsString::from("notes.txt"),
+        ])
+    );
+    repo.seal_plan(None, SnapshotTrigger::AgentRun, &fresh.token)
+        .unwrap();
+}
+
+#[test]
+fn a_plan_removed_before_its_seal_asks_for_a_new_plan() {
+    let f = Fixture::new();
+    let mut repo = f.attach();
+    repo.seal(None, SnapshotTrigger::AgentRun, None).unwrap();
+    let stalled = repo.capture_plan(Some(&[PathBuf::from("a")])).unwrap();
+    age_by_a_day(&plan_file(&repo, &stalled.token));
+    let current = repo.capture_plan(Some(&[PathBuf::from("b")])).unwrap();
+
+    let error = repo
+        .seal_plan(None, SnapshotTrigger::AgentRun, &stalled.token)
+        .unwrap_err();
+
+    assert!(
+        format!("{error:#}").starts_with("capture plan is absent; re-plan"),
+        "{error:#}"
+    );
+    repo.seal_plan(None, SnapshotTrigger::AgentRun, &current.token)
+        .unwrap();
+}
+
+#[test]
+fn sealing_while_another_handle_plans_and_prunes_keeps_every_fresh_plan() {
+    let f = Fixture::new();
+    let mut repo = f.attach();
+    repo.seal(None, SnapshotTrigger::AgentRun, None).unwrap();
+    let directory = plans(&repo);
+    fs::create_dir_all(&directory).unwrap();
+    let (root, data) = (f.root.clone(), f.data.clone());
+    let planner = std::thread::spawn(move || {
+        let other = FurrowRepository::attach_timeline_in(&data, &root, Some("chat"), None).unwrap();
+        for round in 0..30 {
+            let junk: Vec<PathBuf> = (0..20)
+                .map(|n| {
+                    let token = blake3::hash(format!("{round}/{n}").as_bytes()).to_hex();
+                    let path = directory.join(format!("{token}.json"));
+                    fs::write(&path, b"{}").unwrap();
+                    age_by_a_day(&path);
+                    path
+                })
+                .collect();
+            other.capture_plan(Some(&[PathBuf::from("b")])).unwrap();
+            for path in junk {
+                assert!(!path.exists(), "round {round}: {} survived", path.display());
+            }
+        }
+    });
+
+    for round in 0..30 {
+        fs::write(f.root.join("a"), format!("round {round}")).unwrap();
+        let plan = repo.capture_plan(Some(&[PathBuf::from("a")])).unwrap();
+        repo.seal_plan(None, SnapshotTrigger::AgentRun, &plan.token)
+            .unwrap_or_else(|error| panic!("round {round}: {error:#}"));
+    }
+    planner.join().unwrap();
+}

furrow-daemon/src/router.rs

@@ -103,6 +103,9 @@ impl Writer {
                         {
                             last_check = Instant::now();
                             check_index(&data_dir);
+                            furrow::repository::prune_abandoned_plans(
+                                &furrow::repository::store_path(&data_dir),
+                            );
                         }
                     }
                     Err(RecvTimeoutError::Disconnected) => {
@@ -174,7 +177,8 @@ impl Drop for SyncGuard<'_> {
     }
 }
 
-/// When a quiet writer checks its store's index against the pack.
+/// When a quiet writer checks its store's index against the pack and removes
+/// capture plans no caller will seal.
 #[derive(Debug, Clone, Copy)]
 pub struct Maintenance {
     /// How long the writer must have had no request.
@@ -637,4 +641,62 @@ mod tests {
         assert_eq!(rows(), whole, "the quiet writer never rebuilt the index");
         router.close();
     }
+
+    #[test]
+    fn a_quiet_writer_removes_plans_no_caller_will_seal() {
+        let temp = tempfile::tempdir().unwrap();
+        let tree = temp.path().join("tree");
+        std::fs::create_dir(&tree).unwrap();
+        std::fs::write(tree.join("a"), b"completed capture").unwrap();
+        let target = Target {
+            data_dir: temp.path().join("data"),
+            tree,
+            cell_dir: None,
+            timeline: Some("gone".into()),
+        };
+        let router = Router::with_maintenance(Maintenance {
+            idle: Duration::from_millis(100),
+            every: Duration::from_millis(200),
+        });
+        router
+            .submit(
+                target.clone(),
+                crate::verbs::lookup("seal").unwrap(),
+                json!({}),
+            )
+            .unwrap();
+        // The timeline never plans again, so only the writer can remove these.
+        let plans = furrow::repository::store_path(&target.data_dir)
+            .join("workspaces")
+            .join(furrow::repository::timeline_workspace_id("gone").unwrap())
+            .join("capture-plans");
+        std::fs::create_dir_all(&plans).unwrap();
+        let day_ago = std::time::SystemTime::now() - Duration::from_secs(24 * 60 * 60);
+        let abandoned: Vec<PathBuf> = (0..5)
+            .map(|n| {
+                let path = plans.join(format!("{}.json", blake3::hash(&[n]).to_hex()));
+                std::fs::write(&path, b"{}").unwrap();
+                std::fs::File::options()
+                    .write(true)
+                    .open(&path)
+                    .unwrap()
+                    .set_modified(day_ago)
+                    .unwrap();
+                path
+            })
+            .collect();
+        let fresh = plans.join(format!("{}.json", blake3::hash(b"fresh").to_hex()));
+        std::fs::write(&fresh, b"{}").unwrap();
+
+        let deadline = Instant::now() + Duration::from_secs(10);
+        while abandoned.iter().any(|path| path.exists()) && Instant::now() < deadline {
+            std::thread::sleep(Duration::from_millis(50));
+        }
+        assert!(
+            abandoned.iter().all(|path| !path.exists()),
+            "the quiet writer never removed abandoned plans"
+        );
+        assert!(fresh.exists(), "a plan a caller may still seal was removed");
+        router.close();
+    }
 }