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 589bd04..d6e4a9f 100644 --- a/furrow-core/src/repository.rs +++ b/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) 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 new file mode 100644 index 0000000..b58b1c7 --- /dev/null +++ b/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 { + 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..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. @@ -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 = (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(); + } }