From e3bf401f6e5a5bd1449cf3aca84bccbaa5b7c411 Mon Sep 17 00:00:00 2001 From: damocles Date: Thu, 12 Nov 2026 12:53:28 +0100 Subject: [PATCH] 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 | 32 +++++++---- hive-agent/src/reminders.rs | 99 +++++++++++++++++++++++++++++++- hive-agent/src/todos.rs | 41 +++++++++++-- 3 files changed, 155 insertions(+), 17 deletions(-) diff --git a/hive-agent/src/reminder_timer.rs b/hive-agent/src/reminder_timer.rs index 37ca8a88a..f4e590d07 100644 --- a/hive-agent/src/reminder_timer.rs +++ b/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 { 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!( - "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`." - )); + return Err(over_cap(pending, max)); } } 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 - .store(&stored_message, stored_path.as_deref(), due_at) - .map_err(|e| format!("failed to store reminder: {e:#}")) + match store + .store_within_cap(&stored_message, stored_path.as_deref(), due_at, max) + .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 diff --git a/hive-agent/src/reminders.rs b/hive-agent/src/reminders.rs index 08e1a323e..34207ee5f 100644 --- a/hive-agent/src/reminders.rs +++ b/hive-agent/src/reminders.rs @@ -47,6 +47,18 @@ pub struct Reminder { pub created_at: DateTime, } +/// 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). - /// Returns the new row id. + /// Store a new pending reminder with no cap, due at `due_at` (unix + /// 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 { 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 { + 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(); diff --git a/hive-agent/src/todos.rs b/hive-agent/src/todos.rs index 4d94e8174..54d3ad6ee 100644 --- a/hive-agent/src/todos.rs +++ b/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();