hive-c0re: close build-log rows a dead process left in flight

A build_logs row is opened with finished_at NULL and closed only by the
process that runs the build. If hive-c0re stopped in between, the row stayed
open for good: the dashboard listed it with no status, its log stream never
sent `done`, and the retention sweep, which skips open rows, never deleted it.

Coordinator::open now closes every open row as `fail` and appends a line
saying hive-c0re stopped. It runs before this process starts any build, and
only there: BuildLogs::open itself leaves rows alone so a second opener can't
close another process's live rows.

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<u64> {
+        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();