mara/hyperhive
mara/hyperhive
Watch
0
0
Fork
You've already forked hyperhive
 0
Code Activity

broker: keep unacked ids until the ack update succeeds

Browse source 
Closes #6092
This commit is contained in:
atlas 2026-11-06 02:29:46 +01:00 • committed by mara
commit 970ee83a93
1 changed files with 29 additions and 3 deletions
Download patch file Download diff file Expand all files Collapse all files

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

@ -465,8 +465,9 @@ impl Broker {
Ok(deliveries) Ok(deliveries)
} }
/// Drain the per-recipient unacked-id list and mark every row /// Mark every row in the per-recipient unacked-id list
/// `acked_at = NOW`. Fired by the harness after `TurnOutcome::Ok`. /// `acked_at = NOW`, then drain the list. On a sqlite error the list
/// is kept so a retry acks the same rows. Fired by the harness after `TurnOutcome::Ok`.
/// Returns the number of rows acked (zero is normal — claude /// Returns the number of rows acked (zero is normal — claude
/// may have not called recv during the turn). Tolerant of ids /// may have not called recv during the turn). Tolerant of ids
/// that no longer exist in the DB (vacuumed, manually deleted) /// that no longer exist in the DB (vacuumed, manually deleted)
@ -476,7 +477,7 @@ impl Broker {
let mut inflight = self.inflight.lock().unwrap(); let mut inflight = self.inflight.lock().unwrap();
let ids: Vec<i64> = inflight let ids: Vec<i64> = inflight
.get_mut(recipient) .get_mut(recipient)
.map(|s| std::mem::take(&mut s.unacked_ids)) .map(|s| s.unacked_ids.clone())
.unwrap_or_default(); .unwrap_or_default();
if ids.is_empty() { if ids.is_empty() {
return Ok(0); return Ok(0);
@ -497,6 +498,10 @@ impl Broker {
params_vec.push(id); params_vec.push(id);
} }
let n = conn.execute(&sql, params_vec.as_slice())?; let n = conn.execute(&sql, params_vec.as_slice())?;
// Clear only after the UPDATE succeeded, so a failed ack can be retried.
if let Some(s) = inflight.get_mut(recipient) {
s.unacked_ids.clear();
}
Ok(u64::try_from(n).unwrap_or(0)) Ok(u64::try_from(n).unwrap_or(0))
} }
@ -712,6 +717,27 @@ mod tests {
assert!(pop_one(broker, "b").is_none()); assert!(pop_one(broker, "b").is_none());
} }
/// A failed UPDATE keeps the unacked ids: the retry acks the same rows.
#[test]
fn ack_turn_keeps_ids_when_update_fails() {
let h = open_broker();
let broker = &h.broker;
broker.send(&msg("a", "b", "hi")).unwrap();
pop_one(broker, "b").expect("popped");
broker
.conn
.lock()
.unwrap()
.busy_timeout(std::time::Duration::ZERO)
.unwrap();
let blocker = rusqlite::Connection::open(&h.path).unwrap();
blocker.execute_batch("BEGIN IMMEDIATE").unwrap();
assert!(broker.ack_turn("b").is_err());
blocker.execute_batch("ROLLBACK").unwrap();
assert_eq!(broker.ack_turn("b").unwrap(), 1);
assert_eq!(broker.ack_turn("b").unwrap(), 0);
}
/// Bulk triage: three queued messages, agent acks up to the second /// Bulk triage: three queued messages, agent acks up to the second
/// id — the first two never deliver again, the third still pops. /// id — the first two never deliver again, the third still pops.
#[test] #[test]