From df49b17d0f77f73b19d16d38c429bfa16eafae63 Mon Sep 17 00:00:00 2001 From: Zaki Manian Date: Thu, 5 Mar 2026 18:23:11 -0800 Subject: [PATCH] fix: prevent concurrent memory hygiene passes and Windows file lock errors (#535) * fix: prevent concurrent memory hygiene passes and Windows file lock errors (#495) The heartbeat system spawns hygiene passes via tokio::spawn on every tick, creating a TOCTOU race where multiple tasks read the state file before any saves, causing all to execute concurrently. On Windows this also triggers OS error 1224 (file locked by memory-mapped section) when multiple tasks call std::fs::write on the same file. Three fixes: - AtomicBool guard (RUNNING + RunningGuard RAII) ensures only one hygiene pass runs at a time - State file is saved before cleanup (not after) to claim the cadence window early and close the TOCTOU race - Atomic file write (write to .tmp then rename) avoids Windows file-locking errors from concurrent writers Co-Authored-By: Claude Opus 4.6 * fix: add Mutex to serialize tests touching global RUNNING AtomicBool Address PR review feedback: the running_guard_prevents_reentry test manipulates a global static AtomicBool, which could cause flaky failures if future tests also touch it and run in parallel. A test-only Mutex ensures serialization. [skip-regression-check] Co-Authored-By: Claude Opus 4.6 --------- Co-authored-by: Claude Opus 4.6 --- src/workspace/hygiene.rs | 139 +++++++++++++++++++++++++++++++++++---- 1 file changed, 127 insertions(+), 12 deletions(-) diff --git a/src/workspace/hygiene.rs b/src/workspace/hygiene.rs index 8e5935fe..9e6fc852 100644 --- a/src/workspace/hygiene.rs +++ b/src/workspace/hygiene.rs @@ -4,18 +4,26 @@ //! than the retention period. Identity files (`IDENTITY.md`, `SOUL.md`, //! etc.) are never touched. //! +//! A global [`AtomicBool`] guard prevents concurrent hygiene passes, which +//! avoids TOCTOU races on the state file and Windows file-locking errors +//! (OS error 1224) when multiple heartbeat ticks fire before the first +//! pass completes. +//! //! ```text //! ┌─────────────────────────────────────────────┐ //! │ Hygiene Pass │ //! │ │ +//! │ 0. Acquire RUNNING guard (skip if held) │ //! │ 1. Check cadence (skip if ran recently) │ -//! │ 2. List daily/ documents │ -//! │ 3. Delete those older than retention_days │ -//! │ 4. Log summary │ +//! │ 2. Save state (claim the cadence window) │ +//! │ 3. List daily/ documents │ +//! │ 4. Delete those older than retention_days │ +//! │ 5. Log summary │ //! └─────────────────────────────────────────────┘ //! ``` use std::path::PathBuf; +use std::sync::atomic::{AtomicBool, Ordering}; use chrono::{DateTime, Utc}; use serde::{Deserialize, Serialize}; @@ -23,6 +31,9 @@ use serde::{Deserialize, Serialize}; use crate::bootstrap::ironclaw_base_dir; use crate::workspace::Workspace; +/// Global guard preventing concurrent hygiene passes. +static RUNNING: AtomicBool = AtomicBool::new(false); + /// Configuration for workspace hygiene. #[derive(Debug, Clone)] pub struct HygieneConfig { @@ -73,6 +84,10 @@ impl HygieneReport { /// /// This is best-effort: failures are logged but never propagate. The /// agent should not crash because cleanup failed. +/// +/// An [`AtomicBool`] guard ensures only one pass runs at a time, and the +/// state file is written *before* cleanup so that concurrent callers that +/// slip past the guard still see an up-to-date cadence timestamp. pub async fn run_if_due(workspace: &Workspace, config: &HygieneConfig) -> HygieneReport { if !config.enabled { return HygieneReport { @@ -81,6 +96,22 @@ pub async fn run_if_due(workspace: &Workspace, config: &HygieneConfig) -> Hygien }; } + // Prevent concurrent passes. If another task is already running, + // skip immediately. + if RUNNING + .compare_exchange(false, true, Ordering::SeqCst, Ordering::SeqCst) + .is_err() + { + tracing::debug!("memory hygiene: skipping (another pass is running)"); + return HygieneReport { + skipped: true, + ..Default::default() + }; + } + + // Ensure the guard is released when we return. + let _guard = RunningGuard; + let state_file = config.state_dir.join("memory_hygiene_state.json"); // Check cadence @@ -100,6 +131,10 @@ pub async fn run_if_due(workspace: &Workspace, config: &HygieneConfig) -> Hygien } } + // Save state *before* cleanup to claim the cadence window and prevent + // TOCTOU races where another task reads stale state. + save_state(&state_file); + tracing::info!( retention_days = config.retention_days, "memory hygiene: starting cleanup pass" @@ -122,12 +157,18 @@ pub async fn run_if_due(workspace: &Workspace, config: &HygieneConfig) -> Hygien tracing::debug!("memory hygiene: nothing to clean"); } - // Save state (best-effort) - save_state(&state_file); - report } +/// RAII guard that clears the [`RUNNING`] flag on drop. +struct RunningGuard; + +impl Drop for RunningGuard { + fn drop(&mut self) { + RUNNING.store(false, Ordering::SeqCst); + } +} + /// Delete daily log documents older than `retention_days`. async fn cleanup_daily_logs( workspace: &Workspace, @@ -173,24 +214,47 @@ fn load_state(path: &std::path::Path) -> Option { serde_json::from_str(&data).ok() } +/// Save state using atomic write (write to temp file, then rename). +/// +/// This avoids partial writes and Windows file-locking errors (OS error +/// 1224) when multiple processes try to write the same file. fn save_state(path: &std::path::Path) { let state = HygieneState { last_run: Utc::now(), }; - if let Some(dir) = state_path_dir(path) { - std::fs::create_dir_all(dir).ok(); - } - if let Ok(json) = serde_json::to_string_pretty(&state) - && let Err(e) = std::fs::write(path, json) + if let Some(dir) = state_path_dir(path) + && let Err(e) = std::fs::create_dir_all(dir) { - tracing::warn!("memory hygiene: failed to save state: {e}"); + tracing::warn!("memory hygiene: failed to create state dir: {e}"); + return; + } + let Ok(json) = serde_json::to_string_pretty(&state) else { + return; + }; + + // Write to a temp file in the same directory, then atomically rename. + let tmp_path = path.with_extension("json.tmp"); + if let Err(e) = std::fs::write(&tmp_path, &json) { + tracing::warn!("memory hygiene: failed to write temp state: {e}"); + return; + } + if let Err(e) = std::fs::rename(&tmp_path, path) { + tracing::warn!("memory hygiene: failed to rename state file: {e}"); + // Clean up temp file on rename failure + let _ = std::fs::remove_file(&tmp_path); } } #[cfg(test)] mod tests { + use std::sync::Mutex; + use crate::workspace::hygiene::*; + /// Serialize tests that touch the global `RUNNING` AtomicBool so they + /// don't interfere with each other when `cargo test` runs in parallel. + static RUNNING_TESTS: Mutex<()> = Mutex::new(()); + #[test] fn default_config_is_reasonable() { let cfg = HygieneConfig::default(); @@ -241,4 +305,55 @@ mod tests { save_state(&path); assert!(path.exists()); } + + #[test] + fn save_state_is_atomic_no_tmp_left_behind() { + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("state.json"); + let tmp = dir.path().join("state.json.tmp"); + + save_state(&path); + assert!(path.exists(), "state file should exist"); + assert!(!tmp.exists(), "temp file should be cleaned up after rename"); + + // Verify the content is valid JSON + let state = load_state(&path).expect("saved state should be loadable"); + let elapsed = Utc::now().signed_duration_since(state.last_run); + assert!(elapsed.num_seconds() < 2); + } + + /// Regression test for issue #495: concurrent hygiene passes should be + /// serialized by the AtomicBool guard. + #[test] + fn running_guard_prevents_reentry() { + let _lock = RUNNING_TESTS.lock().unwrap(); + + // Simulate acquiring the guard + assert!( + RUNNING + .compare_exchange(false, true, Ordering::SeqCst, Ordering::SeqCst) + .is_ok(), + "first acquisition should succeed" + ); + + // Second acquisition should fail + assert!( + RUNNING + .compare_exchange(false, true, Ordering::SeqCst, Ordering::SeqCst) + .is_err(), + "second acquisition should fail while first is held" + ); + + // Release + RUNNING.store(false, Ordering::SeqCst); + + // Now it should succeed again + assert!( + RUNNING + .compare_exchange(false, true, Ordering::SeqCst, Ordering::SeqCst) + .is_ok(), + "acquisition should succeed after release" + ); + RUNNING.store(false, Ordering::SeqCst); + } }