diff --git a/docs/agent-lifecycle/persistence.md b/docs/agent-lifecycle/persistence.md index d579fdd1..ddddcf7c 100644 --- a/docs/agent-lifecycle/persistence.md +++ b/docs/agent-lifecycle/persistence.md @@ -209,6 +209,11 @@ Writes are best-effort: `append_stdout` / `append_stderr` / `finish` log a warning on sqlite error and let the build continue. A failed log row never blocks a rebuild. +A row is open (`finished_at IS NULL`) only while its child runs. At startup +`Coordinator::open` closes every row still open as `fail` and appends a +line saying hive-c0re stopped: the previous process died mid-build, and +the retention sweep never deletes open rows. + ### `/harness/hyperhive-model` (per agent) Single-line text file holding the claude model name currently diff --git a/hive-c0re/src/coordinator.rs b/hive-c0re/src/coordinator.rs index d06f5969..9e70ad3d 100644 --- a/hive-c0re/src/coordinator.rs +++ b/hive-c0re/src/coordinator.rs @@ -462,6 +462,15 @@ impl Coordinator { let build_logs = Arc::new( crate::build_logs::BuildLogs::open(build_logs_dir).context("open build_logs")?, ); + // This process is the database's only writer and has started no + // build yet, so a row still in-flight belongs to a dead process. + let reaped = build_logs.reap_orphans()?; + if reaped > 0 { + tracing::warn!( + reaped, + "build_logs: closed rows left in-flight by a previous run" + ); + } // Install the process-wide handle so `lifecycle::run` / // `lifecycle::prebuild_toplevel` can write without us having // to thread an `Arc` through every public entry @@ -1376,3 +1385,35 @@ mod agent_paths_tests { assert_eq!(paths.agent, std::path::PathBuf::from("/run/x")); } } + +#[cfg(test)] +mod open_tests { + use super::{Coordinator, HiveEnv}; + + fn open(db: &std::path::Path) -> Coordinator { + Coordinator::open( + db, + HiveEnv::default(), + crate::hive_stats::PriceTable::new(), + 1, + ) + .expect("open coordinator") + } + + /// A build log row the previous process left open is closed as failed + /// when the next one opens the coordinator. + #[tokio::test] + async fn open_closes_build_logs_left_in_flight() { + let dir = tempfile::tempdir().expect("tempdir"); + let db = dir.path().join("broker.sqlite"); + let id = open(&db) + .build_logs + .start("alice", "run", "cmd", None) + .expect("start"); + + let coord = open(&db); + let row = coord.build_logs.get_full(id).expect("get").expect("Some"); + assert_eq!(row.header.status.as_deref(), Some("fail")); + assert!(row.header.finished_at.is_some()); + } +} diff --git a/hive-c0re/src/stores/build_logs.rs b/hive-c0re/src/stores/build_logs.rs index b7ca43b9..6af05a71 100644 --- a/hive-c0re/src/stores/build_logs.rs +++ b/hive-c0re/src/stores/build_logs.rs @@ -48,6 +48,9 @@ const KEEP_FAIL_SECS: i64 = 30 * 24 * 3600; /// `finished_at IS NULL`). const KEEP_OK_SECS: i64 = 24 * 3600; +/// Last line of a row `reap_orphans` closed. +const ORPHAN_NOTE: &str = "hive-c0re stopped before this build finished"; + const SCHEMA: &str = " CREATE TABLE IF NOT EXISTS build_logs ( id INTEGER PRIMARY KEY AUTOINCREMENT, @@ -175,6 +178,30 @@ impl BuildLogs { }) } + /// Close every row still marked in-flight as failed. Returns the number + /// closed. + /// + /// Only sound when no other writer holds this database: the caller must + /// be the one process that owns it, before it starts any build. A row + /// with `finished_at IS NULL` then belongs to a process that died, and + /// nothing will ever finish it. Kept out of `open` so a second opener + /// can't close another process's live rows. + pub fn reap_orphans(&self) -> Result { + let now = Utc::now().timestamp(); + let conn = self.conn.lock().unwrap(); + let closed = conn + .execute( + "UPDATE build_logs + SET finished_at = ?1, + status = 'fail', + stderr = stderr || ?2 || x'0a' + WHERE finished_at IS NULL", + params![now, ORPHAN_NOTE], + ) + .context("close orphaned build_logs rows")?; + Ok(u64::try_from(closed).unwrap_or(0)) + } + /// Subscribe to per-build append/finish notifications. Each emitted /// value is the `id` of the row that changed. The SSE stream handler /// calls this once and filters for its target id. @@ -522,6 +549,31 @@ mod tests { assert!(full.stderr.ends_with('\n')); } + #[test] + fn reap_orphans_fails_rows_left_in_flight_by_a_previous_process() { + let dir = tempfile::tempdir().expect("tempdir"); + let (orphan, done) = { + let db = BuildLogs::open(dir.path()).expect("open"); + let orphan = db.start("alice", "run", "cmd one", None).expect("start"); + db.append_stdout(orphan, "building"); + let done = db.start("bob", "run", "cmd two", None).expect("start"); + db.finish(done, BuildStatus::Ok); + (orphan, done) + }; + + let db = BuildLogs::open(dir.path()).expect("reopen"); + assert_eq!(db.reap_orphans().expect("reap"), 1); + + let row = db.get_full(orphan).expect("get").expect("Some"); + assert_eq!(row.header.status.as_deref(), Some("fail")); + assert!(row.header.finished_at.is_some()); + assert!(row.stdout.contains("building")); + assert!(row.stderr.ends_with(&format!("{ORPHAN_NOTE}\n"))); + let ok = db.get_full(done).expect("get").expect("Some"); + assert_eq!(ok.header.status.as_deref(), Some("ok")); + assert_eq!(db.reap_orphans().expect("reap again"), 0); + } + #[test] fn list_recent_orders_newest_first_and_clamps() { let (_d, db) = tmpdb();