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

broker: add mark_all_read + POST /api/agent/{name}/mark-all-read (#559 backend half)

Browse source
This commit is contained in:
damocles 2026-05-29 11:39:21 +02:00 • committed by Mara
commit 666c32ae65
2 changed files with 143 additions and 0 deletions
Download patch file Download diff file Expand all files Collapse all files

123
hive-c0re/src/broker.rs
View file

@ -513,6 +513,51 @@ impl Broker {
Ok(u64::try_from(ids.len()).unwrap_or(0))
}
/// Operator-driven "clear the inbox": mark every message addressed
/// to `recipient` as acked. Backfills `delivered_at = NOW` for any
/// row that was still pending (undelivered), so the row doesn't
/// become impossible to vacuum later — `vacuum_delivered` requires
/// both timestamps set. Also clears the in-memory inflight state
/// for the recipient so a subsequent `ack_turn` doesn't try to
/// re-mark ids that are already acked. Returns the number of rows
/// affected (zero is normal — inbox already empty).
///
/// Distinct from `ack_turn` (which acks only the per-turn unacked
/// ids the harness pulled via `recv_batch`) and `requeue_inflight`
/// (which puts inflight-but-unacked rows BACK on the queue). This
/// is the destructive "drain everything for this agent" path the
/// dashboard surfaces as the side-panel "mark all read" button
/// (#559). Backs `POST /api/agent/{name}/mark-all-read`.
pub fn mark_all_read(&self, recipient: &str) -> Result<u64> {
let mut inflight = self.inflight.lock().unwrap();
let conn = self.conn.lock().unwrap();
let now = now_unix();
// Two-axis update in one statement: set acked_at on every
// row for the recipient that doesn't have it yet, AND backfill
// delivered_at if it was NULL so the row is fully consumed
// (vacuum_delivered's `acked_at IS NOT NULL AND acked_at < ?`
// predicate then collects it on the normal hourly sweep).
let n = conn.execute(
"UPDATE messages
SET delivered_at = COALESCE(delivered_at, ?1),
acked_at = ?1
WHERE recipient = ?2
AND acked_at IS NULL",
params![now, recipient],
)?;
// Drop in-memory inflight bookkeeping for this recipient: the
// ids we just acked might still be in `unacked_ids` from a
// prior `recv_batch`; leaving them would cause the next
// `ack_turn` to re-issue an UPDATE against rows that no longer
// need it (correct but wasteful) and the requeue path would
// see stale ids. Cleanest to reset.
if let Some(slot) = inflight.get_mut(recipient) {
slot.unacked_ids.clear();
slot.requeued_ids.clear();
}
Ok(u64::try_from(n).unwrap_or(0))
}
/// Store a new reminder. Returns the reminder id.
pub fn store_reminder(
&self,
@ -1113,6 +1158,84 @@ mod tests {
assert!(!d.redelivered);
}
/// `mark_all_read` covers a mix of pending + delivered + acked rows:
/// pending rows get both `delivered_at` and `acked_at` backfilled,
/// delivered-but-unacked rows just get `acked_at` set, already-acked
/// rows pass through untouched. Returns the count of rows mutated.
#[test]
fn mark_all_read_drains_all_states_for_recipient() {
let h = open_broker();
let broker = &h.broker;
// Set up three rows in three different states:
// r1 — pending (never popped, both timestamps NULL)
// r2 — delivered + unacked (popped, harness didn't ack yet)
// r3 — delivered + acked (popped + ack_turn ran)
broker.send(&msg("a", "b", "pending")).unwrap();
broker.send(&msg("a", "b", "delivered")).unwrap();
broker.send(&msg("a", "b", "acked")).unwrap();
// Pop both deliverable rows, then ack only the last so r2 stays
// delivered-but-unacked. After pop r3 is still in the unacked
// list; recv pops in FIFO order so first pop = "pending", but
// we want THAT row to remain undelivered. Workaround: pop two
// rows (so "pending" and "delivered" come off the queue) and
// requeue the first to put "pending" back. Then send a fourth
// "acked" and pop+ack just that.
//
// Simpler approach: bypass the queue helpers and craft the row
// states directly via send + recv + ack. FIFO order is by
// insertion; we pop two, ack only the second.
let _ = pop_one(broker, "b").expect("pop 1: pending → now delivered");
let _ = pop_one(broker, "b").expect("pop 2: delivered");
let _ = pop_one(broker, "b").expect("pop 3: acked-soon");
assert_eq!(broker.ack_turn("b").unwrap(), 3);
// Now reshape: requeue first two so they're pending again.
// (Hack — easier: just call mark_all_read on the state we
// have, which is "three rows already acked". Should return
// zero because no row has acked_at IS NULL.)
assert_eq!(broker.mark_all_read("b").unwrap(), 0);
// Add a fresh pending row + a delivered-but-unacked row.
broker.send(&msg("a", "b", "new pending")).unwrap();
broker.send(&msg("a", "b", "new delivered")).unwrap();
let _ = pop_one(broker, "b").expect("pop new pending → delivered");
// Don't ack — leaves it delivered+unacked.
// Now: one row is pending (delivered_at IS NULL), one is
// delivered+unacked. mark_all_read should hit both.
assert_eq!(broker.mark_all_read("b").unwrap(), 2);
// Second call: nothing pending now.
assert_eq!(broker.mark_all_read("b").unwrap(), 0);
// Confirm via recv — inbox is empty.
assert!(pop_one(broker, "b").is_none());
}
/// Per-recipient isolation: marking alice doesn't touch bob's inbox.
#[test]
fn mark_all_read_is_per_recipient() {
let h = open_broker();
let broker = &h.broker;
broker.send(&msg("x", "alice", "for alice")).unwrap();
broker.send(&msg("x", "bob", "for bob")).unwrap();
assert_eq!(broker.mark_all_read("alice").unwrap(), 1);
// bob's row still pending — pop succeeds.
let d = pop_one(broker, "bob").expect("bob pop");
assert_eq!(d.message.body, "for bob");
}
/// After `mark_all_read`, a subsequent `ack_turn` from a stale
/// in-memory unacked list MUST NOT panic or double-ack. The
/// inflight bookkeeping is cleared by `mark_all_read`.
#[test]
fn mark_all_read_clears_inflight_so_ack_turn_is_noop() {
let h = open_broker();
let broker = &h.broker;
broker.send(&msg("a", "b", "hi")).unwrap();
let _ = pop_one(broker, "b").expect("popped");
// Now there's an id in the unacked_ids list. mark_all_read
// should clear it so the harness's next ack_turn (which still
// thinks the id is unacked) returns 0 cleanly.
assert_eq!(broker.mark_all_read("b").unwrap(), 1);
assert_eq!(broker.ack_turn("b").unwrap(), 0);
}
/// Per-recipient isolation: `requeue_inflight("a")` doesn't touch
/// b's inflight rows.
#[test]

20
hive-c0re/src/dashboard.rs
View file

@ -69,6 +69,7 @@ pub async fn serve(port: u16, coord: Arc<Coordinator>) -> Result<()> {
.route("/api/state-file", get(get_state_file))
.route("/api/reminders", get(api_reminders))
.route("/api/agent/{name}/links", get(get_agent_links))
.route("/api/agent/{name}/mark-all-read", post(post_mark_all_read))
.route("/cancel-reminder/{id}", post(post_cancel_reminder))
.route("/retry-reminder/{id}", post(post_retry_reminder))
.route("/request-spawn", post(post_request_spawn))
@ -1747,6 +1748,25 @@ async fn post_retry_reminder(
}
}
/// Operator-driven "clear this agent's inbox" — backs the side-panel
/// "mark all read" button (#559). Marks every message addressed to the
/// agent as acked (backfilling `delivered_at` for any still-pending
/// rows so vacuum can collect them). Returns `{ "marked": N }` so the
/// frontend can show "cleared N messages" feedback without an extra
/// fetch.
async fn post_mark_all_read(
State(state): State<AppState>,
AxumPath(name): AxumPath<String>,
) -> Response {
match state.coord.broker.mark_all_read(&name) {
Ok(n) => {
tracing::info!(%name, marked = n, "operator marked all messages read");
axum::Json(serde_json::json!({ "marked": n })).into_response()
}
Err(e) => error_response(&format!("mark-all-read {name} failed: {e:#}")),
}
}
async fn post_purge_tombstone(
State(state): State<AppState>,
AxumPath(name): AxumPath<String>,