harness stores: a failed lookup is not no-row, bulk done is one transaction, the reminder cap is checked with the insert

hive-agent/src/reminder_timer.rs

@@ -23,7 +23,7 @@ use std::time::Duration;
 
 use tokio::sync::mpsc;
 
-use crate::reminders::Reminders;
+use crate::reminders::{Reminders, Stored};
 
 /// Per-tick cap on reminders delivered — mirrors the old c0re
 /// `REMINDER_BATCH_LIMIT`, same rationale (bound a deep backlog's
@@ -109,21 +109,33 @@ pub fn store(
     file_path: Option<&str>,
 ) -> Result<i64, String> {
     let max = remind_max_pending();
+    // A cheap early check, so a rejected reminder never writes its large body
+    // to a file; the authoritative one is the count inside the insert below.
     if max > 0 {
-        let pending = store.count_pending().unwrap_or(0);
+        let pending = store
+            .count_pending()
+            .map_err(|e| format!("failed to count pending reminders: {e:#}"))?;
         if pending >= max {
-            return Err(format!(
+            return Err(over_cap(pending, max));
-                "reminder rejected: already {pending} pending reminders (cap {max}). \
-                 Cancel some via `cancel_loose_end` or wait for them to fire before \
-                 scheduling more. Override the cap with `HIVE_REMIND_MAX_PENDING_PER_AGENT`."
-            ));
         }
     }
     let due_at = resolve_due_at(timing).map_err(|e| format!("invalid reminder timing: {e:#}"))?;
     let (stored_message, stored_path) = prepare_remind_storage(message, file_path)?;
-    store
+    match store
-        .store(&stored_message, stored_path.as_deref(), due_at)
+        .store_within_cap(&stored_message, stored_path.as_deref(), due_at, max)
-        .map_err(|e| format!("failed to store reminder: {e:#}"))
+        .map_err(|e| format!("failed to store reminder: {e:#}"))?
+    {
+        Stored::Row(id) => Ok(id),
+        Stored::AtCap { pending } => Err(over_cap(pending, max)),
+    }
+}
+
+fn over_cap(pending: u64, max: u64) -> String {
+    format!(
+        "reminder rejected: already {pending} pending reminders (cap {max}). \
+         Cancel some via `cancel_loose_end` or wait for them to fire before \
+         scheduling more. Override the cap with `HIVE_REMIND_MAX_PENDING_PER_AGENT`."
+    )
 }
 
 /// Decide what to actually persist in the reminder row (see the old

hive-agent/src/reminders.rs

@@ -47,6 +47,18 @@ pub struct Reminder {
     pub created_at: DateTime<Utc>,
 }
 
+/// What [`Reminders::store_within_cap`] did.
+#[derive(Debug, PartialEq, Eq)]
+pub enum Stored {
+    /// Stored, with this row id.
+    Row(i64),
+    /// Not stored: this many reminders were already pending.
+    AtCap {
+        /// The pending count that met the cap.
+        pending: u64,
+    },
+}
+
 /// The harness-local reminder store. Cheap to share behind an `Arc`; the
 /// inner connection is guarded by a `Mutex` (reminder ops are short sqlite
 /// writes, same as `Todos`).
@@ -70,8 +82,9 @@ impl Reminders {
         })
     }
 
-    /// Store a new pending reminder, due at `due_at` (unix seconds).
+    /// Store a new pending reminder with no cap, due at `due_at` (unix
-    /// Returns the new row id.
+    /// seconds). Returns the new row id. Tests and fixtures only: the harness
+    /// stores through [`Self::store_within_cap`].
     ///
     /// # Errors
     ///
@@ -80,6 +93,7 @@ impl Reminders {
     /// # Panics
     ///
     /// Panics if the connection mutex is poisoned.
+    #[cfg(test)]
     pub fn store(&self, message: &str, file_path: Option<&str>, due_at: i64) -> Result<i64> {
         let conn = self.conn.lock().unwrap();
         let now = Utc::now().timestamp();
@@ -92,6 +106,45 @@ impl Reminders {
         Ok(conn.last_insert_rowid())
     }
 
+    /// Store a new pending reminder unless `max` or more are already pending,
+    /// counting and inserting under one lock so two concurrent requests
+    /// cannot both pass the cap. [`Stored::AtCap`] carries the pending count.
+    ///
+    /// # Errors
+    ///
+    /// Propagates a sqlite count or insert failure; a failed count is an
+    /// error, never "zero pending".
+    ///
+    /// # Panics
+    ///
+    /// Panics if the connection mutex is poisoned.
+    pub fn store_within_cap(
+        &self,
+        message: &str,
+        file_path: Option<&str>,
+        due_at: i64,
+        max: u64,
+    ) -> Result<Stored> {
+        let conn = self.conn.lock().unwrap();
+        let pending: i64 = conn.query_row(
+            "SELECT COUNT(*) FROM reminders WHERE sent_at IS NULL",
+            [],
+            |row| row.get(0),
+        )?;
+        let pending = u64::try_from(pending).unwrap_or(0);
+        if max > 0 && pending >= max {
+            return Ok(Stored::AtCap { pending });
+        }
+        let now = Utc::now().timestamp();
+        conn.execute(
+            "INSERT INTO reminders (message, file_path, due_at, created_at, sent_at) \
+             VALUES (?1, ?2, ?3, ?4, NULL)",
+            params![message, file_path, due_at, now],
+        )?;
+        crate::otel_turn_metrics::record_reminder("scheduled");
+        Ok(Stored::Row(conn.last_insert_rowid()))
+    }
+
     /// Count of currently-pending (undelivered) reminders.
     ///
     /// # Errors
@@ -285,6 +338,48 @@ mod tests {
         (dir, db)
     }
 
+    #[test]
+    fn store_within_cap_stops_at_the_cap_and_a_cap_of_zero_means_none() {
+        let (_dir, db) = store();
+        assert!(matches!(
+            db.store_within_cap("a", None, 10, 2).unwrap(),
+            Stored::Row(_)
+        ));
+        assert!(matches!(
+            db.store_within_cap("b", None, 10, 2).unwrap(),
+            Stored::Row(_)
+        ));
+        assert_eq!(
+            db.store_within_cap("c", None, 10, 2).unwrap(),
+            Stored::AtCap { pending: 2 }
+        );
+        assert!(matches!(
+            db.store_within_cap("d", None, 10, 0).unwrap(),
+            Stored::Row(_)
+        ));
+        assert_eq!(db.count_pending().unwrap(), 3);
+    }
+
+    #[test]
+    fn concurrent_stores_cannot_pass_the_cap_together() {
+        let (_dir, db) = store();
+        let stored = std::thread::scope(|scope| {
+            let handles: Vec<_> = (0..16)
+                .map(|i| {
+                    let db = &db;
+                    scope.spawn(move || db.store_within_cap(&format!("r{i}"), None, 10, 5).unwrap())
+                })
+                .collect();
+            handles
+                .into_iter()
+                .map(|h| h.join().unwrap())
+                .filter(|s| matches!(s, Stored::Row(_)))
+                .count()
+        });
+        assert_eq!(stored, 5);
+        assert_eq!(db.count_pending().unwrap(), 5);
+    }
+
     #[test]
     fn store_and_list_pending() {
         let (_dir, s) = store();

hive-agent/src/todos.rs

@@ -34,7 +34,7 @@ use std::sync::Mutex;
 use anyhow::{Context, Result};
 use chrono::{DateTime, Utc};
 use hive_sh4re::wire_time;
-use rusqlite::{Connection, params};
+use rusqlite::{Connection, OptionalExtension as _, params};
 
 /// SQL bootstrap. `CREATE TABLE IF NOT EXISTS` so first-boot agents and
 /// existing ones converge on the same base shape; additive columns land via
@@ -161,7 +161,10 @@ impl Todos {
                 params![subsystem, key],
                 |row| Ok((row.get(0)?, row.get(1)?, row.get(2)?, row.get(3)?)),
             )
-            .ok()
+            // Only "no such row" is `None`: a lookup that failed (a busy
+            // database) must not fall through to an `INSERT` that would
+            // duplicate a keyless row or trip the dedup index on a keyed one.
+            .optional()?
         } else {
             None
         };
@@ -295,7 +298,7 @@ impl Todos {
     ///
     /// # Errors
     ///
-    /// Propagates the sqlite update failure.
+    /// Propagates the sqlite update failure; nothing is acked then.
     ///
     /// # Panics
     ///
@@ -304,15 +307,19 @@ impl Todos {
         if ids.is_empty() {
             return Ok(0);
         }
-        let conn = self.conn.lock().unwrap();
+        let mut conn = self.conn.lock().unwrap();
         let now = Utc::now().timestamp();
+        // One transaction: all of the ids are acked or none, and the commit
+        // is one fsync rather than one per id.
+        let tx = conn.transaction()?;
         let mut total = 0usize;
         for id in ids {
-            total += conn.execute(
+            total += tx.execute(
                 "UPDATE todos SET acked = 1, acked_at = ?1 WHERE acked = 0 AND id = ?2",
                 params![now, id],
             )?;
         }
+        tx.commit()?;
         Ok(total)
     }
 
@@ -666,6 +673,30 @@ mod tests {
 
     /// `mark_done_many` acks exactly the ids passed, leaves the rest
     /// untouched — no threshold/range semantics.
+    #[test]
+    fn mark_done_many_is_all_or_nothing() {
+        let (_dir, todos) = store();
+        let (a, _) = todos.upsert("bash", Some("a"), "one", None, false).unwrap();
+        let (b, _) = todos.upsert("bash", Some("b"), "two", None, false).unwrap();
+        todos
+            .conn
+            .lock()
+            .unwrap()
+            .execute_batch(&format!(
+                "CREATE TRIGGER fail_one BEFORE UPDATE ON todos WHEN NEW.id = {b} \
+                 BEGIN SELECT RAISE(ABORT, 'boom'); END;"
+            ))
+            .unwrap();
+        assert!(todos.mark_done_many(&[a, b]).is_err());
+        let open: Vec<_> = todos
+            .list(None)
+            .unwrap()
+            .into_iter()
+            .map(|t| t.id)
+            .collect();
+        assert!(open.contains(&a) && open.contains(&b), "{open:?}");
+    }
+
     #[test]
     fn mark_done_many_acks_only_the_listed_ids() {
         let (_dir, s) = store();