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

recv/count/dedupe queries exclude acked rows — a pending row closed by ack_until must never pop (iris review)

Browse source
This commit is contained in:
damocles 2026-07-02 11:55:46 +02:00 • committed by mara
commit c89642872c
1 changed files with 12 additions and 3 deletions
Download patch file Download diff file Expand all files Collapse all files

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

@ -441,9 +441,11 @@ impl Broker {
/// lines without popping the queue. /// lines without popping the queue.
pub fn count_pending(&self, recipient: &str) -> Result<u64> { pub fn count_pending(&self, recipient: &str) -> Result<u64> {
let conn = self.conn.lock().unwrap(); let conn = self.conn.lock().unwrap();
// Skip rows closed by `ack_until` while still pending — they
// will never pop, so counting them would show phantom unread.
let n: i64 = conn.query_row( let n: i64 = conn.query_row(
"SELECT COUNT(*) FROM messages "SELECT COUNT(*) FROM messages
WHERE recipient = ?1 AND delivered_at IS NULL", WHERE recipient = ?1 AND delivered_at IS NULL AND acked_at IS NULL",
params![recipient], params![recipient],
|row| row.get(0), |row| row.get(0),
)?; )?;
@ -457,9 +459,12 @@ impl Broker {
/// bodies differ. /// bodies differ.
pub fn has_pending_with_body(&self, recipient: &str, sender: &str, body: &str) -> Result<bool> { pub fn has_pending_with_body(&self, recipient: &str, sender: &str, body: &str) -> Result<bool> {
let conn = self.conn.lock().unwrap(); let conn = self.conn.lock().unwrap();
// An `ack_until`-closed pending row is dead — it must not
// suppress a fresh scheduled delivery of the same body.
let n: i64 = conn.query_row( let n: i64 = conn.query_row(
"SELECT COUNT(*) FROM messages "SELECT COUNT(*) FROM messages
WHERE recipient = ?1 AND sender = ?2 AND body = ?3 AND delivered_at IS NULL", WHERE recipient = ?1 AND sender = ?2 AND body = ?3
AND delivered_at IS NULL AND acked_at IS NULL",
params![recipient, sender, body], params![recipient, sender, body],
|row| row.get(0), |row| row.get(0),
)?; )?;
@ -492,6 +497,7 @@ impl Broker {
AND sender = ?2 AND sender = ?2
AND body LIKE ?3 AND body LIKE ?3
AND delivered_at IS NULL AND delivered_at IS NULL
AND acked_at IS NULL
LIMIT 1", LIMIT 1",
params![child, hive_sh4re::SYSTEM_SENDER, format!("{PREFIX}%")], params![child, hive_sh4re::SYSTEM_SENDER, format!("{PREFIX}%")],
|row| Ok((row.get::<_, i64>(0)?, row.get::<_, String>(1)?)), |row| Ok((row.get::<_, i64>(0)?, row.get::<_, String>(1)?)),
@ -644,10 +650,13 @@ impl Broker {
let mut inflight = self.inflight.lock().unwrap(); let mut inflight = self.inflight.lock().unwrap();
let conn = self.conn.lock().unwrap(); let conn = self.conn.lock().unwrap();
let max_i = i64::try_from(max).unwrap_or(i64::MAX); let max_i = i64::try_from(max).unwrap_or(i64::MAX);
// `acked_at IS NULL` matters for rows closed by `ack_until`
// while still pending (never delivered): they carry an ack but
// no `delivered_at`, and must not pop.
let mut stmt = conn.prepare( let mut stmt = conn.prepare(
"SELECT id, sender, recipient, body, in_reply_to "SELECT id, sender, recipient, body, in_reply_to
FROM messages FROM messages
WHERE recipient = ?1 AND delivered_at IS NULL WHERE recipient = ?1 AND delivered_at IS NULL AND acked_at IS NULL
ORDER BY id ASC ORDER BY id ASC
LIMIT ?2", LIMIT ?2",
)?; )?;