fix(worker): recover abandoned Pending Entries List entries (HIPAA P0)

README.md

@@ -211,6 +211,9 @@ authentication" above), the latter has no auth at all today.
 | `AUDIT_DATABASE_URL` | `mysql+pymysql://root:root@localhost:3306/omnibioai_audit` | Durable audit-event store the consumer writes into and `/audit/events` queries |
 | `AUDIT_CONSUMER_GROUP` | `audit-workers` | Redis Streams consumer group name |
 | `AUDIT_CONSUMER_NAME` | `worker-{pid}` | Per-process consumer identity within the group |
+| `AUDIT_PEL_MIN_IDLE_MS` | `30000` | How long a delivered-but-unacked message must sit idle before any worker (this one or another replica) may reclaim it — see `worker/main.py::sweep_pending` |
+| `AUDIT_PEL_MAX_DELIVERIES` | `5` | Delivery attempts (original + reclaims) before an entry is treated as poison and ACKed without further processing |
+| `AUDIT_PEL_SWEEP_BATCH` | `100` | Max stale Pending Entries List entries inspected per sweep |
 
 ---
 

audit/config.py

@@ -19,6 +19,36 @@ class AuditConfig:
     CONSUMER_GROUP = os.getenv("AUDIT_CONSUMER_GROUP", "audit-workers")
     CONSUMER_NAME = os.getenv("AUDIT_CONSUMER_NAME", f"worker-{os.getpid()}")
 
+    # HIPAA P0 (abandoned-PEL-entry recovery): a message this consumer
+    # group delivered but never acked -- crash before persistence, a
+    # transient MySQL failure, the process getting killed mid-write --
+    # sits in the group's Pending Entries List. CONSUMER_NAME above is
+    # per-process (pid-based) by design (multiple worker replicas must
+    # never collide on one identity), which means a crashed worker's
+    # pending entries are *never* revisited by that same identity again --
+    # nothing "restarts" a dead pid. Recovery has to come from any live
+    # worker sweeping the whole group's PEL, not from self-continuity, so
+    # these thresholds are named generically (PEL_*, not "own pending").
+    #
+    # PEL_MIN_IDLE_MS: how long an entry must sit unacked before ANY
+    # worker (including the one that originally received it, if it's
+    # still alive and just slow) is allowed to reclaim it. Must safely
+    # exceed one full handle_message() -- including a real MySQL
+    # round-trip -- under normal load, or a live-but-slow worker would
+    # have its own in-flight message reclaimed out from under it.
+    PEL_MIN_IDLE_MS = int(os.getenv("AUDIT_PEL_MIN_IDLE_MS", "30000"))
+    # PEL_MAX_DELIVERIES: once an entry has been *delivered* (original
+    # read + every reclaim) this many times without a successful ack, it
+    # is treated as poison -- a deterministically-unparseable payload,
+    # not a transient failure -- and ACKed without further processing so
+    # it can never loop forever. See worker/main.py::sweep_pending.
+    PEL_MAX_DELIVERIES = int(os.getenv("AUDIT_PEL_MAX_DELIVERIES", "5"))
+    # PEL_SWEEP_BATCH: cap on stale entries inspected per sweep call, same
+    # "bounded work per loop iteration" shape MAX_ENTRIES-style caps use
+    # elsewhere in this platform -- a PEL of unbounded size must not turn
+    # one sweep into an unbounded-latency Redis call.
+    PEL_SWEEP_BATCH = int(os.getenv("AUDIT_PEL_SWEEP_BATCH", "100"))
+
     # PR2 of the audit:events integrity remediation (see audit/signing.py,
     # PR1): reuses the same JWT_SECRET every other platform service (and
     # this repo's own audit/jwt_verify.py) already reads -- not a new

consumers/stream_reader.py

@@ -1,6 +1,6 @@
 import redis
-import json
 from redis.exceptions import ResponseError
+
 from audit.config import AuditConfig
 
 
@@ -41,4 +41,69 @@ def read_group(self, consumer_name, group=None, count=10, block=5000):
 
     def ack(self, message_id, group=None):
         group = group or AuditConfig.CONSUMER_GROUP
-        self.redis.xack(self.stream, group, message_id)
+        self.redis.xack(self.stream, group, message_id)
+
+    # -----------------------------------------------------------------
+    # HIPAA P0: abandoned Pending Entries List recovery. read_group()
+    # above only ever asks Redis for ">" -- strictly new, never-before-
+    # delivered messages -- so a message that was delivered but never
+    # acked (worker crash, transient persistence failure) is otherwise
+    # invisible to every future read_group() call, from this consumer
+    # identity or any other, forever. See worker/main.py::sweep_pending
+    # for the caller that makes this reachable from the main loop.
+    # -----------------------------------------------------------------
+
+    def claim_stale(self, consumer_name, group=None, min_idle_ms=None, max_deliveries=None, batch=None):
+        """Reclaims entries abandoned by a crashed or stuck consumer.
+
+        Two Redis calls (XPENDING's extended form, then XCLAIM), not the
+        newer single-call XAUTOCLAIM: XPENDING's extended form is the
+        only way to read each entry's `times_delivered`, which is what
+        distinguishes "abandoned, still worth retrying" from "poison --
+        has already failed this many times, stop retrying it" (see
+        AuditConfig.PEL_MAX_DELIVERIES). XAUTOCLAIM's combined
+        claim-in-one-call doesn't expose that count.
+
+        Concurrency-safe for multiple worker replicas without any extra
+        coordination: XCLAIM only reassigns an entry that is *still*
+        idle >= min_idle_ms at the exact moment it runs. Two workers
+        racing to reclaim the same entry each issue XCLAIM for it, but
+        Redis processes them serially -- whichever runs second finds the
+        entry already claimed (idle time just reset to ~0 by the first),
+        so XCLAIM returns nothing for that id, never a duplicate
+        delivery to both. See
+        tests/test_worker_pel_recovery_integration.py::
+        test_real_concurrent_workers_racing_for_the_same_entry_only_one_wins
+        for a real-Redis proof (two live consumers racing to reclaim the
+        same entry).
+
+        Returns (claimed, poison_ids):
+          claimed -- list of (message_id, fields) tuples now owned by
+            `consumer_name`, ready for the normal handle_message() path,
+            exactly like a message just read via read_group().
+          poison_ids -- message ids that exceeded max_deliveries and were
+            ACKed directly here (removed from the PEL, never handed back
+            for reprocessing) -- returned only so the caller can log them,
+            not for further action.
+        """
+        group = group or AuditConfig.CONSUMER_GROUP
+        min_idle_ms = AuditConfig.PEL_MIN_IDLE_MS if min_idle_ms is None else min_idle_ms
+        max_deliveries = AuditConfig.PEL_MAX_DELIVERIES if max_deliveries is None else max_deliveries
+        batch = AuditConfig.PEL_SWEEP_BATCH if batch is None else batch
+
+        stale = self.redis.xpending_range(
+            self.stream, group, min="-", max="+", count=batch, idle=min_idle_ms
+        )
+        if not stale:
+            return [], []
+
+        poison_ids = [e["message_id"] for e in stale if e["times_delivered"] >= max_deliveries]
+        reclaimable_ids = [e["message_id"] for e in stale if e["times_delivered"] < max_deliveries]
+
+        if poison_ids:
+            self.redis.xack(self.stream, group, *poison_ids)
+
+        claimed = []
+        if reclaimable_ids:
+            claimed = self.redis.xclaim(self.stream, group, consumer_name, min_idle_ms, reclaimable_ids)
+        return claimed, poison_ids

tests/test_stream.py

@@ -1,8 +1,7 @@
 import pytest
-from unittest.mock import MagicMock, patch
 from redis.exceptions import ResponseError
-from audit.config import AuditConfig
 
+from audit.config import AuditConfig
 
 # ---------------------------------------------------------------------------
 # StreamReader.read()
@@ -129,3 +128,119 @@ def test_ack_calls_xack(stream_reader):
     mock_redis.xack.assert_called_once_with(
         AuditConfig.STREAM_NAME, AuditConfig.CONSUMER_GROUP, "1-0"
     )
+
+
+# ---------------------------------------------------------------------------
+# HIPAA P0: StreamReader.claim_stale() -- abandoned-PEL-entry recovery.
+# ---------------------------------------------------------------------------
+
+def _pending_entry(message_id, times_delivered):
+    return {
+        "message_id": message_id,
+        "consumer": "some-dead-consumer",
+        "time_since_delivered": 60000,
+        "times_delivered": times_delivered,
+    }
+
+
+def test_claim_stale_queries_xpending_range_with_config_defaults(stream_reader):
+    reader, mock_redis = stream_reader
+    mock_redis.xpending_range.return_value = []
+
+    claimed, poison_ids = reader.claim_stale("worker-2")
+
+    mock_redis.xpending_range.assert_called_once_with(
+        AuditConfig.STREAM_NAME,
+        AuditConfig.CONSUMER_GROUP,
+        min="-",
+        max="+",
+        count=AuditConfig.PEL_SWEEP_BATCH,
+        idle=AuditConfig.PEL_MIN_IDLE_MS,
+    )
+    assert claimed == []
+    assert poison_ids == []
+
+
+def test_claim_stale_returns_empty_when_nothing_stale(stream_reader):
+    reader, mock_redis = stream_reader
+    mock_redis.xpending_range.return_value = []
+
+    claimed, poison_ids = reader.claim_stale("worker-2")
+
+    assert claimed == []
+    assert poison_ids == []
+    mock_redis.xclaim.assert_not_called()
+    mock_redis.xack.assert_not_called()
+
+
+def test_claim_stale_reclaims_entries_under_max_deliveries(stream_reader):
+    reader, mock_redis = stream_reader
+    mock_redis.xpending_range.return_value = [_pending_entry("5-0", times_delivered=2)]
+    mock_redis.xclaim.return_value = [("5-0", {"data": "{}"})]
+
+    claimed, poison_ids = reader.claim_stale("worker-2")
+
+    mock_redis.xclaim.assert_called_once_with(
+        AuditConfig.STREAM_NAME,
+        AuditConfig.CONSUMER_GROUP,
+        "worker-2",
+        AuditConfig.PEL_MIN_IDLE_MS,
+        ["5-0"],
+    )
+    mock_redis.xack.assert_not_called()
+    assert claimed == [("5-0", {"data": "{}"})]
+    assert poison_ids == []
+
+
+def test_claim_stale_acks_poison_entries_without_reclaiming(stream_reader):
+    reader, mock_redis = stream_reader
+    mock_redis.xpending_range.return_value = [
+        _pending_entry("6-0", times_delivered=AuditConfig.PEL_MAX_DELIVERIES)
+    ]
+
+    claimed, poison_ids = reader.claim_stale("worker-2")
+
+    mock_redis.xack.assert_called_once_with(
+        AuditConfig.STREAM_NAME, AuditConfig.CONSUMER_GROUP, "6-0"
+    )
+    mock_redis.xclaim.assert_not_called()
+    assert claimed == []
+    assert poison_ids == ["6-0"]
+
+
+def test_claim_stale_splits_a_mixed_batch_correctly(stream_reader):
+    reader, mock_redis = stream_reader
+    mock_redis.xpending_range.return_value = [
+        _pending_entry("7-0", times_delivered=1),
+        _pending_entry("7-1", times_delivered=AuditConfig.PEL_MAX_DELIVERIES + 3),
+        _pending_entry("7-2", times_delivered=AuditConfig.PEL_MAX_DELIVERIES - 1),
+    ]
+    mock_redis.xclaim.return_value = [("7-0", {"data": "a"}), ("7-2", {"data": "c"})]
+
+    claimed, poison_ids = reader.claim_stale("worker-2")
+
+    mock_redis.xack.assert_called_once_with(
+        AuditConfig.STREAM_NAME, AuditConfig.CONSUMER_GROUP, "7-1"
+    )
+    mock_redis.xclaim.assert_called_once_with(
+        AuditConfig.STREAM_NAME,
+        AuditConfig.CONSUMER_GROUP,
+        "worker-2",
+        AuditConfig.PEL_MIN_IDLE_MS,
+        ["7-0", "7-2"],
+    )
+    assert poison_ids == ["7-1"]
+    assert claimed == [("7-0", {"data": "a"}), ("7-2", {"data": "c"})]
+
+
+def test_claim_stale_honors_explicit_overrides_over_config_defaults(stream_reader):
+    reader, mock_redis = stream_reader
+    mock_redis.xpending_range.return_value = []
+
+    reader.claim_stale(
+        "worker-2", group="other-group", min_idle_ms=999, max_deliveries=1, batch=7,
+    )
+
+    mock_redis.xpending_range.assert_called_once_with(
+        AuditConfig.STREAM_NAME, "other-group", min="-", max="+", count=7, idle=999,
+    )