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

[man4ish]

Problem

security-audit-worker — the authoritative consumer for the signed audit:events HIPAA audit trail — consumed Redis Streams with XREADGROUP ... ">" only: strictly new, never-before-delivered messages. A message that was delivered but never acked (worker crash before persistence, a transient MySQL failure, the process getting killed mid-write) sat in the consumer group's Pending Entries List and was never revisited — nothing in the repo called XCLAIM/XAUTOCLAIM/XPENDING. CONSUMER_NAME defaults to worker-{pid} and the service runs restart: on-failure, so every crash/restart also minted a brand-new consumer identity with no path back to its predecessor's orphaned entries.

handle_message()'s own docstring claimed "leaves the message unacknowledged... so it can be retried — it is never dropped." That was false as implemented. This is the fix.

Fix

• StreamReader.claim_stale() — XPENDING's extended form (to read each stale entry's times_delivered) followed by XCLAIM for the ones still worth retrying. XPENDING+XCLAIM chosen over the newer single-call XAUTOCLAIM specifically because XAUTOCLAIM doesn't expose per-entry delivery counts, which is what distinguishes "abandoned, retry" from "poison, give up."
• worker/main.py::sweep_pending() — called once per run() loop iteration, before read_group(), so an abandoned-entry backlog isn't starved by steady new traffic. Reclaimed entries run through the exact same handle_message() classify/persist/ack path as freshly-read messages.
• An entry delivered AUDIT_PEL_MAX_DELIVERIES times (default 5) without succeeding is ACKed directly as poison, never retried again — bounds the retry loop so a deterministically-malformed payload can't loop forever.
• New env vars, all optional with safe defaults: AUDIT_PEL_MIN_IDLE_MS=30000, AUDIT_PEL_MAX_DELIVERIES=5, AUDIT_PEL_SWEEP_BATCH=100. No compose/deployment changes — defaults are production-safe as shipped.
• Corrected handle_message()'s docstring to describe what actually makes retry happen now.

Explicitly out of scope (per task brief)

No changes to the six HMAC signing implementations, Control Center's public-read-only work, LIMS, or Redis network topology.

Concurrency safety

XCLAIM only reassigns an entry that is still idle ≥ min_idle_ms at the exact moment it runs — two workers racing for the same entry can never both succeed. Proven against a real Redis server, not just asserted against a mock: test_real_concurrent_workers_racing_for_the_same_entry_only_one_wins races two live claim_stale() calls via ThreadPoolExecutor and asserts total claimed == 1.

Idempotency / HMAC verification

Untouched. consumers/sink.py's PK-on-event_id dedup and audit/signing.py's HMAC verification are not modified. The reclaim path is re-verified against both specifically:

• test_real_duplicate_delivery_after_reclaim_does_not_duplicate_row — a message persisted successfully by "worker A" but never acked before it "crashes" is reclaimed and reprocessed by "worker B"; exactly one row results.
• test_real_valid_signed_event_persists_as_valid / existing signature tests all still pass unmodified.

Tests

273 passed (252 baseline + 21 new), 0 regressions:

• 6 unit tests for StreamReader.claim_stale() (mocked Redis) — tests/test_stream.py
• 10 unit tests for sweep_pending()/run() wiring (mocked reader) — tests/test_worker_pel_recovery.py
• 5 real-Redis + real-MySQL integration tests — tests/test_worker_pel_recovery_integration.py:
  • crash-before-ack → reclaimed by a second worker → persisted
  • two workers racing for the same entry → exactly one wins
  • transient persistence failure → reclaim → retry succeeds
  • duplicate delivery through the reclaim path → no duplicate row
  • malformed/poison event → bounded retries → abandoned, PEL empty (no infinite loop)

ruff check clean on every touched/new file. Repo-wide ruff count actually dropped (80 → 66) by cleaning up dead imports encountered along the way; 3 pre-existing intentional broad-except sites in worker/main.py got the same # noqa: BLE001 treatment this repo's own audit/logger.py already established.

Remaining risks (documented, not fixed here)

• A MySQL outage longer than PEL_MAX_DELIVERIES × sweep interval still eventually abandons a message as poison — a deliberate bounded-retry tradeoff, not unlimited durability.
• This PR makes multi-replica operation safe; it doesn't turn on multiple replicas itself (compose still runs one security-audit-worker) — that's a separate deploy decision.
• Poison messages are dropped with only a log line, no dead-letter persistence — matches this repo's existing "no dead-letter mechanism" convention rather than inventing one unilaterally.

Do not merge until CI and review pass.

🤖 Generated with Claude Code