From 0a705ca2535a7abd56b1bcdcdf431afafd1b252c Mon Sep 17 00:00:00 2001 From: santoshkumarradha Date: Mon, 26 Oct 2026 14:45:08 -0400 Subject: [PATCH 1/2] test: pin removal of abandoned capture plans Plans a caller never seals stay in capture-plans/ forever today. A new plan, and a quiet daemon writer for a timeline that never plans again, must remove plans older than any caller takes to seal, keep a fresh plan sealable, and leave a stalled caller a clean re-plan error. --- furrow-core/src/repository.rs | 2 + .../src/repository/plan_prune_tests.rs | 152 ++++++++++++++++++ furrow-daemon/src/router.rs | 58 +++++++ 3 files changed, 212 insertions(+) create mode 100644 furrow-core/src/repository/plan_prune_tests.rs diff --git a/furrow-core/src/repository.rs b/furrow-core/src/repository.rs index 589bd04..7eaefe5 100644 --- a/furrow-core/src/repository.rs +++ b/furrow-core/src/repository.rs @@ -6327,6 +6327,8 @@ 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) diff --git a/furrow-core/src/repository/plan_prune_tests.rs b/furrow-core/src/repository/plan_prune_tests.rs new file mode 100644 index 0000000..8da2f5c --- /dev/null +++ b/furrow-core/src/repository/plan_prune_tests.rs @@ -0,0 +1,152 @@ +//! 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 { + 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, ¤t.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 = (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(); +} diff --git a/furrow-daemon/src/router.rs b/furrow-daemon/src/router.rs index 1398c7c..541be34 100644 --- a/furrow-daemon/src/router.rs +++ b/furrow-daemon/src/router.rs @@ -637,4 +637,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 = (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(); + } } From e8e0f30912bac3c72312857ef7928b5fd64125f4 Mon Sep 17 00:00:00 2001 From: santoshkumarradha Date: Mon, 26 Oct 2026 14:46:56 -0400 Subject: [PATCH 2/2] fix: remove capture plans no caller will seal A plan is deleted only by its own seal, so every re-plan, crash, timeout or engine exit left its file behind; one machine held 98 GB of them. A plan older than ten minutes, twenty times the time a caller takes to seal, is now abandoned: each new plan removes its timeline's abandoned plans and partial writes, and an idle daemon writer does the same for timelines that never plan again. Both read one directory of plans, never the store. Sealing a removed plan still asks for a re-plan. --- README.md | 2 + furrow-core/src/repository.rs | 2 +- furrow-core/src/repository/capture_plan.rs | 56 ++++++++++++++++++- .../src/repository/plan_prune_tests.rs | 3 +- furrow-daemon/src/router.rs | 6 +- 5 files changed, 63 insertions(+), 6 deletions(-) diff --git a/README.md b/README.md index 3e37811..e4c57df 100644 --- a/README.md +++ b/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` diff --git a/furrow-core/src/repository.rs b/furrow-core/src/repository.rs index 7eaefe5..d6e4a9f 100644 --- a/furrow-core/src/repository.rs +++ b/furrow-core/src/repository.rs @@ -6322,7 +6322,7 @@ 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)] diff --git a/furrow-core/src/repository/capture_plan.rs b/furrow-core/src/repository/capture_plan.rs index 83b7802..67b82c0 100644 --- a/furrow-core/src/repository/capture_plan.rs +++ b/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>, } +/// 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 { 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!( diff --git a/furrow-core/src/repository/plan_prune_tests.rs b/furrow-core/src/repository/plan_prune_tests.rs index 8da2f5c..b58b1c7 100644 --- a/furrow-core/src/repository/plan_prune_tests.rs +++ b/furrow-core/src/repository/plan_prune_tests.rs @@ -123,8 +123,7 @@ fn sealing_while_another_handle_plans_and_prunes_keeps_every_fresh_plan() { 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(); + let other = FurrowRepository::attach_timeline_in(&data, &root, Some("chat"), None).unwrap(); for round in 0..30 { let junk: Vec = (0..20) .map(|n| { diff --git a/furrow-daemon/src/router.rs b/furrow-daemon/src/router.rs index 541be34..98169f4 100644 --- a/furrow-daemon/src/router.rs +++ b/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.