vinzenz/hyperhive
vinzenz/hyperhive
Watch
0
0
Fork
You've already forked hyperhive
 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions 1

delete c0re-side reminder plumbing (#2635 inc 1 commit 6)

Browse source
This commit is contained in:
damocles 2026-07-22 23:01:16 +02:00 • committed by mara
commit a80d0b0fed
16 changed files with 70 additions and 1136 deletions
Download patch file Download diff file Expand all files Collapse all files

365
hive-c0re/src/stores/broker.rs
View file

@ -7,7 +7,6 @@ use std::sync::Mutex;
use anyhow::{Context, Result};
use chrono::{DateTime, Utc};
use hive_sh4re::wire_time::now_unix;
use hive_sh4re::{InboxRow, Message};
@ -30,18 +29,6 @@ CREATE TABLE IF NOT EXISTS messages (
CREATE INDEX IF NOT EXISTS idx_messages_undelivered
ON messages (recipient, priority DESC, id) WHERE delivered_at IS NULL;
CREATE TABLE IF NOT EXISTS reminders (
id INTEGER PRIMARY KEY AUTOINCREMENT,
agent TEXT NOT NULL,
message TEXT NOT NULL,
file_path TEXT,
due_at INTEGER NOT NULL,
created_at INTEGER NOT NULL,
sent_at INTEGER
);
CREATE INDEX IF NOT EXISTS idx_reminders_due
ON reminders (agent, due_at) WHERE sent_at IS NULL;
CREATE TABLE IF NOT EXISTS kv (
key TEXT PRIMARY KEY,
value TEXT NOT NULL
@ -52,12 +39,6 @@ CREATE TABLE IF NOT EXISTS kv (
/// may drop events past this; we send a `lagged` notice in their stream.
const EVENT_CHANNEL: usize = 256;
/// Row shape returned by [`Broker::get_due_reminders`]:
/// `(agent, reminder_id, message, file_path)`. Type alias keeps
/// `clippy::type_complexity` quiet and makes the scheduler call site
/// self-documenting.
pub type DueReminder = (String, i64, String, Option<String>);
/// A single message hand-off from broker to recipient. Carries the
/// broker's row id (so the harness can drive `ack_turn` later) and
/// the redelivery flag (so the harness can prepend the
@ -70,37 +51,6 @@ pub struct Delivery {
pub message: Message,
}
/// Row shape for [`Broker::list_pending_reminders`], shipped on the
/// dashboard `/api/reminders` response.
#[derive(Debug, Clone, Serialize)]
pub struct PendingReminder {
pub id: i64,
pub agent: String,
pub message: String,
#[serde(skip_serializing_if = "Option::is_none")]
pub file_path: Option<String>,
pub due_at: DateTime<Utc>,
pub created_at: DateTime<Utc>,
/// Most recent delivery failure for this row, if any. Cleared
/// to NULL on operator retry. Surfaced inline in the dashboard
/// so a stuck reminder doesn't just silently retry forever.
#[serde(skip_serializing_if = "Option::is_none")]
pub last_error: Option<String>,
/// Number of failed delivery attempts since the row was
/// created or last retried. After `MAX_REMINDER_ATTEMPTS` the
/// scheduler stops trying (the row stays in `pending` with the
/// error so the operator can decide between retry + cancel).
#[serde(default)]
pub attempt_count: u32,
}
/// Stop retrying a row after this many consecutive failures. The
/// scheduler quits scheduling it until an operator explicitly
/// retries (which resets the counter) or cancels (which deletes
/// the row). Below the cap the existing 5s tick re-attempts each
/// time the row is due.
pub const MAX_REMINDER_ATTEMPTS: u32 = 5;
/// Intra-process broker event. `recv_blocking_batch` listens on the
/// same channel as the dashboard forwarder; the forwarder re-emits
/// each event as a `DashboardEvent` with a freshly-stamped seq from
@ -193,18 +143,6 @@ const BROKER_MIGRATIONS: &[Migration] = &[
COMMIT;",
adds_column: Some(("messages", "priority")),
},
// v4: attempt_count on reminders for the MAX_REMINDER_ATTEMPTS cap.
Migration {
sql: "ALTER TABLE reminders ADD COLUMN \
attempt_count INTEGER NOT NULL DEFAULT 0",
adds_column: Some(("reminders", "attempt_count")),
},
// v5: last_error on reminders — last delivery failure surfaced on the
// dashboard so a stuck reminder is visible without digging in logs.
Migration {
sql: "ALTER TABLE reminders ADD COLUMN last_error TEXT",
adds_column: Some(("reminders", "last_error")),
},
];
impl Broker {
@ -850,309 +788,6 @@ impl Broker {
}
Ok(u64::try_from(n).unwrap_or(0))
}
/// Store a new reminder. Returns the reminder id.
pub fn store_reminder(
&self,
agent: &str,
message: &str,
file_path: Option<&str>,
due_at: i64,
) -> Result<i64> {
let conn = self.conn.lock().unwrap();
conn.execute(
"INSERT INTO reminders (agent, message, file_path, due_at, created_at) VALUES (?1, ?2, ?3, ?4, ?5)",
params![agent, message, file_path, due_at, now_unix()],
)?;
let id = conn.last_insert_rowid();
Ok(id)
}
/// Every reminder still pending delivery, newest-first. Used by the
/// dashboard's reminders pane so the operator can see what's queued
/// + cancel rows that are no longer wanted.
pub fn list_pending_reminders(&self) -> Result<Vec<PendingReminder>> {
let conn = self.conn.lock().unwrap();
let mut stmt = conn.prepare(
"SELECT id, agent, message, file_path, due_at, created_at, \
last_error, attempt_count \
FROM reminders \
WHERE sent_at IS NULL \
ORDER BY due_at ASC",
)?;
let rows = stmt.query_map([], |row| {
let attempts: i64 = row.get(7)?;
Ok(PendingReminder {
id: row.get(0)?,
agent: row.get(1)?,
message: row.get(2)?,
file_path: row.get(3)?,
due_at: hive_sh4re::wire_time::from_secs(row.get(4)?),
created_at: hive_sh4re::wire_time::from_secs(row.get(5)?),
last_error: row.get(6)?,
attempt_count: u32::try_from(attempts).unwrap_or(0),
})
})?;
rows.collect::<rusqlite::Result<Vec<_>>>()
.context("list pending reminders")
}
/// Mark a delivery attempt as failed: bump `attempt_count` and
/// stash the error string. Called by `reminder_scheduler::tick`
/// when `deliver_reminder` returns Err. Soft-cap behaviour
/// lives in `get_due_reminders` (rows over the cap drop out
/// of the due-list and stop being attempted until retry).
pub fn record_reminder_failure(&self, id: i64, reason: &str) -> Result<()> {
let conn = self.conn.lock().unwrap();
conn.execute(
"UPDATE reminders \
SET attempt_count = attempt_count + 1, last_error = ?1 \
WHERE id = ?2 AND sent_at IS NULL",
params![reason, id],
)?;
Ok(())
}
/// Clear the failure state on a pending reminder so the
/// scheduler picks it up again. No-op when the row is already
/// fresh (`attempt_count == 0`). Returns the number of rows
/// affected so callers can distinguish "retried" from "no
/// such pending reminder" (already delivered, or wrong id).
pub fn reset_reminder_failure(&self, id: i64) -> Result<usize> {
let conn = self.conn.lock().unwrap();
let n = conn.execute(
"UPDATE reminders \
SET attempt_count = 0, last_error = NULL \
WHERE id = ?1 AND sent_at IS NULL",
params![id],
)?;
Ok(n)
}
/// Count this agent's still-pending (un-delivered) reminders.
/// Used by the per-turn stats sink for a cheap "what was queued
/// at turn-end" snapshot.
pub fn count_pending_reminders_for(&self, agent: &str) -> Result<u64> {
let conn = self.conn.lock().unwrap();
let n: i64 = conn.query_row(
"SELECT COUNT(*) FROM reminders WHERE agent = ?1 AND sent_at IS NULL",
params![agent],
|row| row.get(0),
)?;
Ok(u64::try_from(n).unwrap_or(0))
}
/// Reminder rollup stats for an agent over a time window. Returns
/// counts of scheduled, delivered, and pending reminders created
/// in the last `since_secs` seconds (0 = all reminders).
pub fn reminder_rollup_for(
&self,
agent: &str,
since_secs: u64,
) -> Result<hive_sh4re::ReminderStats> {
let conn = self.conn.lock().unwrap();
let cutoff_time = if since_secs > 0 {
let now = std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.ok()
.and_then(|d| i64::try_from(d.as_secs()).ok())
.unwrap_or(0);
now.saturating_sub(i64::try_from(since_secs).unwrap_or(i64::MAX))
} else {
i64::MIN
};
let scheduled: i64 = conn.query_row(
"SELECT COUNT(*) FROM reminders WHERE agent = ?1 AND created_at >= ?2",
params![agent, cutoff_time],
|row| row.get(0),
)?;
let delivered: i64 = conn.query_row(
"SELECT COUNT(*) FROM reminders WHERE agent = ?1 AND created_at >= ?2 AND sent_at IS NOT NULL",
params![agent, cutoff_time],
|row| row.get(0),
)?;
let pending: i64 = conn.query_row(
"SELECT COUNT(*) FROM reminders WHERE agent = ?1 AND created_at >= ?2 AND sent_at IS NULL",
params![agent, cutoff_time],
|row| row.get(0),
)?;
Ok(hive_sh4re::ReminderStats {
scheduled: u64::try_from(scheduled).unwrap_or(0),
delivered: u64::try_from(delivered).unwrap_or(0),
pending: u64::try_from(pending).unwrap_or(0),
})
}
/// Delete a reminder by id. Returns the number of rows removed (0
/// when the id never existed or was already delivered). Hard
/// delete rather than soft so the row doesn't linger and confuse a
/// re-creation under the same id.
pub fn cancel_reminder(&self, id: i64) -> Result<usize> {
let conn = self.conn.lock().unwrap();
let n = conn.execute(
"DELETE FROM reminders WHERE id = ?1 AND sent_at IS NULL",
params![id],
)?;
Ok(n)
}
/// Cancel a pending reminder on behalf of `canceller`. Returns
/// the owner agent name on success (handy for logging). Auth
/// rules mirror `OperatorQuestions::cancel`: the owner, the
/// operator, or a `privileged` caller (one that arrived on the
/// manager socket — the trust boundary, not a name match).
pub fn cancel_reminder_as(&self, id: i64, canceller: &str, privileged: bool) -> Result<String> {
let conn = self.conn.lock().unwrap();
let owner: Option<String> = conn
.query_row(
"SELECT agent FROM reminders WHERE id = ?1 AND sent_at IS NULL",
params![id],
|row| row.get(0),
)
.optional()?;
let Some(owner) = owner else {
anyhow::bail!("reminder {id} not pending (already delivered or unknown)");
};
let authorised =
privileged || canceller == owner || canceller == hive_sh4re::OPERATOR_RECIPIENT;
if !authorised {
anyhow::bail!("reminder {id}: '{canceller}' not allowed to cancel (owner = '{owner}')");
}
let n = conn.execute(
"DELETE FROM reminders WHERE id = ?1 AND sent_at IS NULL",
params![id],
)?;
if n == 0 {
anyhow::bail!("reminder {id} vanished between auth check and delete");
}
Ok(owner)
}
/// Get up to `limit` due reminders across all agents in a single query.
/// Returns `(agent, id, message, file_path)` tuples. Pass a small limit
/// (e.g. 100) so a burst of overdue reminders doesn't flood the broker
/// in one cycle — leftovers stay due and get picked up on the next tick.
pub fn get_due_reminders(&self, limit: u64) -> Result<Vec<DueReminder>> {
let conn = self.conn.lock().unwrap();
let limit_i = i64::try_from(limit.min(i64::MAX as u64)).unwrap_or(i64::MAX);
let max_attempts = i64::from(MAX_REMINDER_ATTEMPTS);
// attempt_count >= cap = give up; row stays pending so the
// operator sees + can retry/cancel via the dashboard.
let mut stmt = conn.prepare(
"SELECT agent, id, message, file_path FROM reminders \
WHERE due_at <= ?1 AND sent_at IS NULL AND attempt_count < ?3 \
ORDER BY agent, due_at ASC \
LIMIT ?2",
)?;
let rows = stmt.query_map(params![now_unix(), limit_i, max_attempts], |row| {
Ok((
row.get::<_, String>(0)?,
row.get::<_, i64>(1)?,
row.get::<_, String>(2)?,
row.get::<_, Option<String>>(3)?,
))
})?;
rows.collect::<rusqlite::Result<Vec<_>>>()
.context("query due reminders")
}
/// Atomic reminder delivery: insert the inbox message AND mark the
/// reminder as sent in a single sqlite transaction. Prevents the
/// orphan-reminder duplicate-delivery class of bugs that two separate
/// calls (send + `mark_reminder_sent`) could produce if the second one
/// failed transiently — the next scheduler tick would see the reminder
/// still due and redeliver. Either both writes commit or neither does;
/// re-running on failure is safe.
///
/// Emits a `Sent` event on the broadcast channel after the transaction
/// commits (so subscribers see the inbox message but never see a
/// "phantom" send for a transaction that rolled back).
/// Deliver a batch of reminders in a single transaction, reducing
/// lock contention on the shared sqlite connection under high
/// reminder volume. Returns per-item results so the scheduler can
/// record individual failures without aborting successful ones.
///
/// Items where the INSERT+UPDATE succeeds get a `MessageEvent::Sent`
/// emitted after the transaction commits. Items that fail are
/// returned as `Err` in the output vec (index-aligned with input).
pub fn deliver_reminders_batch(
&self,
items: &[(i64, String, String)], // (reminder_id, agent, body)
) -> Vec<Result<()>> {
if items.is_empty() {
return Vec::new();
}
let now = now_unix();
let mut conn = self.conn.lock().unwrap();
// Build one transaction for all deliveries so we hold the lock
// once rather than N times. On a batch-level error (e.g. DB
// corruption), fall back to returning per-item errors so the
// scheduler records the failure cleanly.
let tx = match conn.transaction() {
Ok(t) => t,
Err(e) => {
let err_str = format!("{e:#}");
return items
.iter()
.map(|_| Err(anyhow::anyhow!("{}", err_str.clone())))
.collect();
}
};
let mut results: Vec<Result<()>> = Vec::with_capacity(items.len());
// Per-item broker row ids — collected inside the transaction so
// we can emit Sent events with the correct id after commit.
let mut msg_ids: Vec<i64> = Vec::with_capacity(items.len());
for (id, agent, body) in items {
let r = (|| -> Result<i64> {
tx.execute(
"INSERT INTO messages (sender, recipient, body, sent_at) \
VALUES (?1, ?2, ?3, ?4)",
params!["reminder", agent, body, now],
)?;
let msg_id = tx.last_insert_rowid();
tx.execute(
"UPDATE reminders SET sent_at = ?1 WHERE id = ?2",
params![now, id],
)?;
Ok(msg_id)
})();
match r {
Ok(msg_id) => {
msg_ids.push(msg_id);
results.push(Ok(()));
}
Err(e) => {
msg_ids.push(-1);
results.push(Err(e));
}
}
}
if let Err(e) = tx.commit() {
let err_str = format!("{e:#}");
return items
.iter()
.map(|_| Err(anyhow::anyhow!("{}", err_str.clone())))
.collect();
}
drop(conn);
// Emit per-row Sent events (only for rows that succeeded).
for (((id, agent, body), result), msg_id) in
items.iter().zip(results.iter()).zip(msg_ids.iter())
{
if result.is_ok() {
let _ = self.events.send(MessageEvent::Sent {
id: *msg_id,
from: "reminder".to_owned(),
to: agent.clone(),
body: body.clone(),
at: now,
in_reply_to: None,
});
tracing::debug!(reminder_id = id, %agent, "reminder delivered");
}
}
results
}
}
#[cfg(test)]