From c89642872c108f735a698248d1d4e1249bda3f89 Mon Sep 17 00:00:00 2001
From: damocles <damocles@hyperhive.local>
Date: Thu, 2 Jul 2026 11:55:46 +0200
Subject: [PATCH] =?UTF-8?q?recv/count/dedupe=20queries=20exclude=20acked?=
 =?UTF-8?q?=20rows=20=E2=80=94=20a=20pending=20row=20closed=20by=20ack=5Fu?=
 =?UTF-8?q?ntil=20must=20never=20pop=20(iris=20review)?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 hive-c0re/src/broker.rs | 15 ++++++++++++---
 1 file changed, 12 insertions(+), 3 deletions(-)

diff --git a/hive-c0re/src/broker.rs b/hive-c0re/src/broker.rs
index 8f103862..b2caef09 100644
--- a/hive-c0re/src/broker.rs
+++ b/hive-c0re/src/broker.rs
@@ -441,9 +441,11 @@ impl Broker {
     /// lines without popping the queue.
     pub fn count_pending(&self, recipient: &str) -> Result<u64> {
         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(
             "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],
             |row| row.get(0),
         )?;
@@ -457,9 +459,12 @@ impl Broker {
     /// bodies differ.
     pub fn has_pending_with_body(&self, recipient: &str, sender: &str, body: &str) -> Result<bool> {
         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(
             "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],
             |row| row.get(0),
         )?;
@@ -492,6 +497,7 @@ impl Broker {
                     AND sender = ?2
                     AND body LIKE ?3
                     AND delivered_at IS NULL
+                    AND acked_at IS NULL
                   LIMIT 1",
                 params![child, hive_sh4re::SYSTEM_SENDER, format!("{PREFIX}%")],
                 |row| Ok((row.get::<_, i64>(0)?, row.get::<_, String>(1)?)),
@@ -644,10 +650,13 @@ impl Broker {
         let mut inflight = self.inflight.lock().unwrap();
         let conn = self.conn.lock().unwrap();
         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(
             "SELECT id, sender, recipient, body, in_reply_to
                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
               LIMIT ?2",
         )?;