Merge forge/main into atlas/5034-s4-remove-hive-model

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<BuildLogs>` 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());
+    }
+}

hive-c0re/src/snapshot_push.rs

@@ -18,6 +18,8 @@
 //! stores would split that chain in two, which is the case the store
 //! exists to serve.
 
+use std::os::fd::OwnedFd;
+
 use anyhow::{Context as _, Result, bail};
 use hive_priv_sock::PrivRequest;
 use tokio::io::AsyncWriteExt as _;
@@ -61,10 +63,7 @@ pub async fn push_agent_snapshot(agent: &str, snapshot: &str, parent: Option<&st
     // closes our copy as soon as the kernel has it, so the receiver sees
     // EOF when `btrfs send` finishes rather than hanging on a descriptor
     // this process is still holding.
-    let fd = sock
-        .into_std()
-        .context("detach snapshot-store socket for hand-off")?
-        .into();
+    let fd = detach_blocking(sock)?;
 
     let resp = priv_client::call_with_fd(
         &PrivRequest::SendAgentSnapshotToFd {
@@ -93,6 +92,22 @@ pub async fn push_agent_snapshot(agent: &str, snapshot: &str, parent: Option<&st
     Ok(())
 }
 
+/// Turn a tokio socket into a descriptor safe to hand to `btrfs send`.
+///
+/// tokio sockets are `O_NONBLOCK`, and `into_std` does not clear it. The
+/// flag lives on the open file description, which `hive-priv` shares with
+/// us through `SCM_RIGHTS`, so `btrfs send` would write to a
+/// non-blocking socket and get `EAGAIN` whenever the send buffer is full.
+fn detach_blocking(sock: TcpStream) -> Result<OwnedFd> {
+    let std_sock = sock
+        .into_std()
+        .context("detach snapshot-store socket for hand-off")?;
+    std_sock
+        .set_nonblocking(false)
+        .context("switch snapshot-store socket to blocking mode for hand-off")?;
+    Ok(std_sock.into())
+}
+
 /// Require a configured store address.
 ///
 /// Split out of [`push_agent_snapshot`] so the "not configured" path is
@@ -118,7 +133,30 @@ fn require_store_addr(configured: Option<&str>) -> Result<&str> {
 
 #[cfg(test)]
 mod tests {
-    use super::require_store_addr;
+    use std::os::fd::AsRawFd as _;
+
+    use tokio::net::{TcpListener, TcpStream};
+
+    use super::{detach_blocking, require_store_addr};
+
+    #[tokio::test]
+    async fn the_handed_off_descriptor_is_blocking() {
+        let listener = TcpListener::bind("127.0.0.1:0").await.expect("bind");
+        let sock = TcpStream::connect(listener.local_addr().expect("addr"))
+            .await
+            .expect("connect");
+
+        let fd = detach_blocking(sock).expect("detach");
+
+        // SAFETY: `fd` is a live descriptor owned by this test.
+        let flags = unsafe { libc::fcntl(fd.as_raw_fd(), libc::F_GETFL) };
+        assert!(flags >= 0, "F_GETFL failed");
+        assert_eq!(
+            flags & libc::O_NONBLOCK,
+            0,
+            "btrfs send would get EAGAIN on a full socket buffer"
+        );
+    }
 
     #[test]
     fn a_configured_address_passes_through() {

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();