harness todos: ack reconciled todos instead of deleting them
This commit is contained in:
damocles
parent
ef1554a1b2
commit
8a81085770
3 changed files with 233 additions and 27 deletions
|
|
@ -15,7 +15,10 @@
|
|||
//! loose-ends list instead of waking the agent every tick. Only crossing
|
||||
//! into a new bucket (or a change in which directories are big enough to
|
||||
//! list) speaks up again. Dropping back under the threshold clears the
|
||||
//! todo.
|
||||
//! todo. Acknowledging the todo (`cancel_loose_end`) does not defeat this —
|
||||
//! `Todos::mark_done` acks the row rather than deleting it, so the next
|
||||
//! probe still has the prior summary to compare against instead of seeing
|
||||
//! an empty table and re-announcing an unchanged condition as new.
|
||||
//!
|
||||
//! Scoped by design: the `statvfs` reads the *whole filesystem*, which on
|
||||
//! a shared host volume (several agents' state dirs on the same
|
||||
|
|
|
|||
|
|
@ -15,9 +15,18 @@
|
|||
//! same item is idempotent, and a producer can list / clear / rebuild only
|
||||
//! its own set (e.g. matrix wipes + recreates its todos on daemon restart).
|
||||
//!
|
||||
//! Removal has two paths (mara's call): the producing subsystem `clear`s a
|
||||
//! todo it has resolved (keyed by subsystem + key), or the agent itself
|
||||
//! `mark_done`s one by id. Both delete the row.
|
||||
//! Removal/dismissal has two different paths. The **producing subsystem**
|
||||
//! `clear`s a todo it has itself resolved (keyed by subsystem + key) — a
|
||||
//! one-shot event has nothing left to represent once resolved, so this
|
||||
//! hard-deletes the row. The **agent** `mark_done`s one by id (via
|
||||
//! `cancel_loose_end`) — that only means "stop showing it to me", so it sets
|
||||
//! `acked` instead of deleting: a *reconciled* producer like `disk_watch`
|
||||
//! (re-derives its summary on a timer, not a one-shot event) needs the row
|
||||
//! to survive ack so [`Todos::upsert`] can still tell "unchanged" from
|
||||
//! "genuinely different". Acked rows stay out of [`Todos::list`] but keep
|
||||
//! anchoring that comparison; a materially different summary un-acks +
|
||||
//! wakes again. Acked rows are reaped after a retention window
|
||||
//! (`vacuum.rs`) so the table doesn't grow unbounded.
|
||||
|
||||
use std::path::Path;
|
||||
use std::sync::Mutex;
|
||||
|
|
@ -26,6 +35,10 @@ use anyhow::{Context, Result};
|
|||
use hive_sh4re::wire_time::now_unix;
|
||||
use rusqlite::{Connection, 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
|
||||
/// [`MIGRATIONS`] as try-and-ignore `ALTER TABLE`s so existing dbs catch up
|
||||
/// without a hard schema-version bump.
|
||||
const SCHEMA: &str = r"
|
||||
CREATE TABLE IF NOT EXISTS todos (
|
||||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||||
|
|
@ -42,6 +55,17 @@ CREATE UNIQUE INDEX IF NOT EXISTS idx_todos_dedup
|
|||
ON todos (subsystem, subsystem_key);
|
||||
";
|
||||
|
||||
/// Additive column migrations — see `turn_stats.rs`'s identical pattern.
|
||||
/// Each runs unconditionally and ignores `duplicate column name` errors, so
|
||||
/// re-running on an already-migrated db is a silent no-op.
|
||||
const MIGRATIONS: &[&str] = &[
|
||||
"ALTER TABLE todos ADD COLUMN acked INTEGER NOT NULL DEFAULT 0",
|
||||
// No default: only meaningful once `acked = 1`; NULL on every
|
||||
// pre-migration / never-acked row, which `reap_acked`'s `acked_at < ?`
|
||||
// comparison naturally excludes (NULL compares false, not true).
|
||||
"ALTER TABLE todos ADD COLUMN acked_at INTEGER",
|
||||
];
|
||||
|
||||
/// One dynamic, subsystem-pushed todo. Timestamps are unix seconds; the
|
||||
/// consumer derives `age_seconds` from `updated_at`.
|
||||
#[derive(Debug, Clone)]
|
||||
|
|
@ -75,6 +99,18 @@ impl Todos {
|
|||
let conn =
|
||||
Connection::open(path).with_context(|| format!("open todos db {}", path.display()))?;
|
||||
conn.execute_batch(SCHEMA).context("apply todos schema")?;
|
||||
for stmt in MIGRATIONS {
|
||||
// Ignore "duplicate column name" — the migration already ran.
|
||||
// Any other error is logged but doesn't fail open(): the base
|
||||
// schema still works and we'd rather keep the harness alive
|
||||
// than crash on an upgrade hiccup.
|
||||
if let Err(e) = conn.execute(stmt, []) {
|
||||
let msg = e.to_string();
|
||||
if !msg.contains("duplicate column name") {
|
||||
tracing::warn!(error = %msg, stmt, "todos migration failed");
|
||||
}
|
||||
}
|
||||
}
|
||||
Ok(Self {
|
||||
conn: Mutex::new(conn),
|
||||
})
|
||||
|
|
@ -84,10 +120,14 @@ impl Todos {
|
|||
/// when `key` is `Some` and already present. A `None` key never
|
||||
/// conflicts, so it always inserts a fresh row.
|
||||
///
|
||||
/// Returns `(id, changed)` where `changed` is `true` when the row is
|
||||
/// new OR its `summary`/`source` actually differed — the caller uses
|
||||
/// this to decide whether to signal the turn loop (re-pushing an
|
||||
/// identical keyed todo is a no-op and must not re-wake).
|
||||
/// Returns `(id, changed)` where `changed` is `true` when the row is new
|
||||
/// OR its `summary`/`source` actually differed — the caller uses this to
|
||||
/// decide whether to signal the turn loop (re-pushing an identical keyed
|
||||
/// todo is a no-op and must not re-wake). This holds **regardless of
|
||||
/// whether the existing row is acked**: an identical re-push of an
|
||||
/// acked row stays quiet (and stays acked/hidden), while a genuinely
|
||||
/// different summary un-acks it and reports `changed = true` — the
|
||||
/// agent's earlier dismissal doesn't suppress a real re-deterioration.
|
||||
///
|
||||
/// # Errors
|
||||
///
|
||||
|
|
@ -121,8 +161,12 @@ impl Todos {
|
|||
if unchanged {
|
||||
return Ok((id, false));
|
||||
}
|
||||
// A materially different summary clears any prior ack — this is
|
||||
// new information, not a re-announcement of what was dismissed.
|
||||
conn.execute(
|
||||
"UPDATE todos SET summary = ?1, source = ?2, updated_at = ?3 WHERE id = ?4",
|
||||
"UPDATE todos SET summary = ?1, source = ?2, updated_at = ?3, \
|
||||
acked = 0, acked_at = NULL \
|
||||
WHERE id = ?4",
|
||||
params![summary, source, now, id],
|
||||
)?;
|
||||
return Ok((id, true));
|
||||
|
|
@ -175,24 +219,35 @@ impl Todos {
|
|||
Ok(n)
|
||||
}
|
||||
|
||||
/// The agent marks one of its todos done, by id. Returns the number of
|
||||
/// rows deleted (0 when the id was unknown / already gone).
|
||||
/// The agent marks one of its todos done, by id — i.e. dismisses it from
|
||||
/// its own list. Sets `acked` rather than deleting the row (see the
|
||||
/// module doc comment for why): the row stays intact so a reconciled
|
||||
/// producer's next `upsert` can still tell "unchanged" from "genuinely
|
||||
/// different" instead of seeing an empty table and treating a re-poll of
|
||||
/// the same condition as brand new. Returns the number of rows newly
|
||||
/// acked (0 when the id was unknown, already gone, or already acked —
|
||||
/// acking twice is not a new action).
|
||||
///
|
||||
/// # Errors
|
||||
///
|
||||
/// Propagates the sqlite delete failure.
|
||||
/// Propagates the sqlite update failure.
|
||||
///
|
||||
/// # Panics
|
||||
///
|
||||
/// Panics if the connection mutex is poisoned.
|
||||
pub fn mark_done(&self, id: i64) -> Result<usize> {
|
||||
let conn = self.conn.lock().unwrap();
|
||||
let n = conn.execute("DELETE FROM todos WHERE id = ?1", params![id])?;
|
||||
let n = conn.execute(
|
||||
"UPDATE todos SET acked = 1, acked_at = ?1 WHERE id = ?2 AND acked = 0",
|
||||
params![now_unix(), id],
|
||||
)?;
|
||||
Ok(n)
|
||||
}
|
||||
|
||||
/// List todos, newest-updated first. `subsystem = Some(..)` filters to
|
||||
/// one producer's set; `None` returns all.
|
||||
/// one producer's set; `None` returns all. Excludes acked rows — once
|
||||
/// the agent has dismissed a todo it stays out of its own list, even
|
||||
/// though the row itself survives (see `mark_done`).
|
||||
///
|
||||
/// # Errors
|
||||
///
|
||||
|
|
@ -206,7 +261,7 @@ impl Todos {
|
|||
let mut stmt = conn.prepare(
|
||||
"SELECT id, subsystem, subsystem_key, summary, source, updated_at \
|
||||
FROM todos \
|
||||
WHERE (?1 IS NULL OR subsystem = ?1) \
|
||||
WHERE acked = 0 AND (?1 IS NULL OR subsystem = ?1) \
|
||||
ORDER BY updated_at DESC, id DESC",
|
||||
)?;
|
||||
let rows = stmt
|
||||
|
|
@ -224,12 +279,12 @@ impl Todos {
|
|||
Ok(rows)
|
||||
}
|
||||
|
||||
/// Cheap existence check — `true` when at least one todo row exists
|
||||
/// (across every subsystem). Used to gate a `todo_wake` notification
|
||||
/// against being turned into a turn when its payload has already been
|
||||
/// drained by an earlier turn (see the serve loop's `LocalTodo` arm):
|
||||
/// an `EXISTS` probe, not a full `list` + row materialization, since
|
||||
/// this runs on every wake.
|
||||
/// Cheap existence check — `true` when at least one un-acked todo row
|
||||
/// exists (across every subsystem). Used to gate a `todo_wake`
|
||||
/// notification against being turned into a turn when its payload has
|
||||
/// already been drained by an earlier turn (see the serve loop's
|
||||
/// `LocalTodo` arm): an `EXISTS` probe, not a full `list` + row
|
||||
/// materialization, since this runs on every wake.
|
||||
///
|
||||
/// # Errors
|
||||
///
|
||||
|
|
@ -240,10 +295,36 @@ impl Todos {
|
|||
/// Panics if the connection mutex is poisoned.
|
||||
pub fn has_any(&self) -> Result<bool> {
|
||||
let conn = self.conn.lock().unwrap();
|
||||
let any: bool =
|
||||
conn.query_row("SELECT EXISTS(SELECT 1 FROM todos)", [], |row| row.get(0))?;
|
||||
let any: bool = conn.query_row(
|
||||
"SELECT EXISTS(SELECT 1 FROM todos WHERE acked = 0)",
|
||||
[],
|
||||
|row| row.get(0),
|
||||
)?;
|
||||
Ok(any)
|
||||
}
|
||||
|
||||
/// Reap acked rows whose `acked_at` is older than `cutoff` (a unix
|
||||
/// timestamp). Called by the harness's hourly vacuum sweep
|
||||
/// (`vacuum.rs`), not on any hot path — bounds the table against acked
|
||||
/// one-shot todos (bash/matrix/forge) that will never naturally be
|
||||
/// upserted again and so would otherwise sit forever. Never touches
|
||||
/// un-acked rows. Returns the number of rows deleted.
|
||||
///
|
||||
/// # Errors
|
||||
///
|
||||
/// Propagates the sqlite delete failure.
|
||||
///
|
||||
/// # Panics
|
||||
///
|
||||
/// Panics if the connection mutex is poisoned.
|
||||
pub fn reap_acked(&self, cutoff: i64) -> Result<usize> {
|
||||
let conn = self.conn.lock().unwrap();
|
||||
let n = conn.execute(
|
||||
"DELETE FROM todos WHERE acked = 1 AND acked_at < ?1",
|
||||
params![cutoff],
|
||||
)?;
|
||||
Ok(n)
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
|
|
@ -291,13 +372,20 @@ mod tests {
|
|||
}
|
||||
|
||||
#[test]
|
||||
fn clear_and_mark_done_remove_rows() {
|
||||
fn clear_deletes_and_mark_done_hides_from_list() {
|
||||
let (_dir, s) = store();
|
||||
s.upsert("matrix", Some("!a:x"), "unread", None).unwrap();
|
||||
let (id, _) = s.upsert("forge", Some("pr-1"), "review", None).unwrap();
|
||||
assert_eq!(s.clear("matrix", Some("!a:x")).unwrap(), 1);
|
||||
assert_eq!(s.mark_done(id).unwrap(), 1);
|
||||
assert!(s.list(None).unwrap().is_empty());
|
||||
assert_eq!(s.clear("matrix", Some("!a:x")).unwrap(), 1, "clear deletes");
|
||||
assert_eq!(
|
||||
s.mark_done(id).unwrap(),
|
||||
1,
|
||||
"mark_done acks, doesn't delete"
|
||||
);
|
||||
assert!(
|
||||
s.list(None).unwrap().is_empty(),
|
||||
"both are gone from the agent-visible list either way"
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
|
|
@ -324,4 +412,101 @@ mod tests {
|
|||
assert_eq!(left.len(), 1);
|
||||
assert_eq!(left[0].subsystem, "forge");
|
||||
}
|
||||
|
||||
/// The whole point of the fix: acking a reconciled todo must not defeat
|
||||
/// the producer's own anti-nag comparison. A re-push with an identical
|
||||
/// summary after ack stays quiet AND stays out of the list — the `disk_watch`
|
||||
/// bug this closes.
|
||||
#[test]
|
||||
fn acked_row_survives_unchanged_reupsert_and_stays_hidden() {
|
||||
let (_dir, s) = store();
|
||||
let (id, changed1) = s.upsert("disk", Some("usage"), "over 85%", None).unwrap();
|
||||
assert!(changed1);
|
||||
assert_eq!(s.mark_done(id).unwrap(), 1, "first ack succeeds");
|
||||
assert!(s.list(None).unwrap().is_empty(), "acked row is hidden");
|
||||
|
||||
// Same producer, same reconcile tick, identical summary — must NOT
|
||||
// look like a new todo just because the agent dismissed it.
|
||||
let (id2, changed2) = s.upsert("disk", Some("usage"), "over 85%", None).unwrap();
|
||||
assert_eq!(id, id2, "same row, not a fresh insert");
|
||||
assert!(!changed2, "identical summary stays quiet even though acked");
|
||||
assert!(
|
||||
s.list(None).unwrap().is_empty(),
|
||||
"still hidden — an unchanged re-push does not un-ack"
|
||||
);
|
||||
}
|
||||
|
||||
/// A materially different summary on an acked row is real information —
|
||||
/// it must un-ack, re-surface in `list`, and report `changed = true`.
|
||||
#[test]
|
||||
fn acked_row_resurfaces_on_genuine_change() {
|
||||
let (_dir, s) = store();
|
||||
let (id, _) = s.upsert("disk", Some("usage"), "over 85%", None).unwrap();
|
||||
s.mark_done(id).unwrap();
|
||||
assert!(s.list(None).unwrap().is_empty());
|
||||
|
||||
let (id2, changed) = s.upsert("disk", Some("usage"), "over 90%", None).unwrap();
|
||||
assert_eq!(id, id2);
|
||||
assert!(changed, "a genuine re-deterioration must not be silenced");
|
||||
let visible = s.list(None).unwrap();
|
||||
assert_eq!(visible.len(), 1, "un-acked row is visible again");
|
||||
assert_eq!(visible[0].summary, "over 90%");
|
||||
}
|
||||
|
||||
/// Acking an already-acked (or unknown) id is a no-op, not a new action.
|
||||
#[test]
|
||||
fn mark_done_twice_is_idempotent() {
|
||||
let (_dir, s) = store();
|
||||
let (id, _) = s.upsert("bash", None, "task done", None).unwrap();
|
||||
assert_eq!(s.mark_done(id).unwrap(), 1);
|
||||
assert_eq!(s.mark_done(id).unwrap(), 0, "already acked");
|
||||
assert_eq!(s.mark_done(999_999).unwrap(), 0, "unknown id");
|
||||
}
|
||||
|
||||
/// `has_any` must also treat an acked-only table as empty — it gates a
|
||||
/// wake against a payload the agent has already drained.
|
||||
#[test]
|
||||
fn has_any_excludes_acked_rows() {
|
||||
let (_dir, s) = store();
|
||||
let (id, _) = s.upsert("bash", None, "task done", None).unwrap();
|
||||
assert!(s.has_any().unwrap());
|
||||
s.mark_done(id).unwrap();
|
||||
assert!(!s.has_any().unwrap(), "acked-only table reads as empty");
|
||||
}
|
||||
|
||||
/// `reap_acked` only removes acked rows past the cutoff — a recent ack
|
||||
/// and any un-acked row both survive.
|
||||
#[test]
|
||||
fn reap_acked_only_removes_old_acked_rows() {
|
||||
let (dir, s) = store();
|
||||
let (old_id, _) = s.upsert("bash", None, "old task", None).unwrap();
|
||||
let (recent_id, _) = s.upsert("bash", None, "recent task", None).unwrap();
|
||||
let (never_acked_id, _) = s.upsert("forge", Some("pr-1"), "review", None).unwrap();
|
||||
s.mark_done(old_id).unwrap();
|
||||
s.mark_done(recent_id).unwrap();
|
||||
|
||||
// Backdate the old row's ack past the cutoff directly — mark_done
|
||||
// always acks "now", so simulate age via the raw connection.
|
||||
{
|
||||
let conn = s.conn.lock().unwrap();
|
||||
conn.execute(
|
||||
"UPDATE todos SET acked_at = ?1 WHERE id = ?2",
|
||||
params![now_unix() - 1000, old_id],
|
||||
)
|
||||
.unwrap();
|
||||
}
|
||||
|
||||
let removed = s.reap_acked(now_unix() - 500).unwrap();
|
||||
assert_eq!(removed, 1, "only the backdated row is past the cutoff");
|
||||
|
||||
let conn = Connection::open(dir.path().join("todos.sqlite")).unwrap();
|
||||
let remaining: i64 = conn
|
||||
.query_row("SELECT COUNT(*) FROM todos", [], |r| r.get(0))
|
||||
.unwrap();
|
||||
assert_eq!(
|
||||
remaining, 2,
|
||||
"recent ack + never-acked row both survive the reap"
|
||||
);
|
||||
let _ = never_acked_id;
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -32,6 +32,12 @@ const STREAM_KEEP_SECS: i64 = 14 * 24 * 3600;
|
|||
/// before reaping them — same window as `STREAM_KEEP_SECS`, kept around
|
||||
/// only to serve the trailing-window `ReminderRollup` stats.
|
||||
const REMINDER_KEEP_SECS: i64 = 14 * 24 * 3600;
|
||||
/// Keep acked (agent-dismissed) todo rows this long before reaping them —
|
||||
/// see `todos.rs`'s module doc for why acking doesn't delete outright. Long
|
||||
/// enough that a genuinely quiet month is the only way to trigger the "one
|
||||
/// spurious re-announcement" fallback path, short enough the table doesn't
|
||||
/// grow meaningfully from one-shot todos that will never be upserted again.
|
||||
const TODO_ACKED_KEEP_SECS: i64 = 30 * 24 * 3600;
|
||||
/// Terminal bash-task statuses whose files are eligible for deletion.
|
||||
const TERMINAL_STATUSES: &[&str] = &["done", "timed_out", "interrupted"];
|
||||
|
||||
|
|
@ -72,6 +78,11 @@ fn sweep_once() {
|
|||
Ok(n) => tracing::info!(removed = n, "reminders vacuum"),
|
||||
Err(e) => tracing::warn!(error = ?e, "reminders vacuum failed"),
|
||||
}
|
||||
match vacuum_todos(&state_db) {
|
||||
Ok(0) => {}
|
||||
Ok(n) => tracing::info!(removed = n, "todos vacuum"),
|
||||
Err(e) => tracing::warn!(error = ?e, "todos vacuum failed"),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
|
|
@ -84,6 +95,13 @@ fn vacuum_reminders(path: &Path) -> anyhow::Result<usize> {
|
|||
store.prune_delivered_older_than(now_unix() - REMINDER_KEEP_SECS)
|
||||
}
|
||||
|
||||
/// Reap acked todo rows older than [`TODO_ACKED_KEEP_SECS`] via the typed
|
||||
/// store API — same own-short-lived-connection shape as `vacuum_reminders`.
|
||||
fn vacuum_todos(path: &Path) -> anyhow::Result<usize> {
|
||||
let store = crate::todos::Todos::open(path)?;
|
||||
store.reap_acked(now_unix() - TODO_ACKED_KEEP_SECS)
|
||||
}
|
||||
|
||||
/// Delete eligible bash-task trios in `dir`. Returns the count of `.json`
|
||||
/// sentinels removed (each represents one task; `.out`/`.err` deletions are
|
||||
/// not counted separately).
|
||||
|
|
|
|||
Loading…
Reference in a new issue