diff --git a/README.md b/README.md index 4ca08aa..2c23b9e 100644 --- a/README.md +++ b/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 | --- diff --git a/audit/config.py b/audit/config.py index efccffd..33eb65a 100644 --- a/audit/config.py +++ b/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 diff --git a/consumers/stream_reader.py b/consumers/stream_reader.py index 5bd60c4..a152fc6 100644 --- a/consumers/stream_reader.py +++ b/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) \ No newline at end of file + 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 \ No newline at end of file diff --git a/tests/test_stream.py b/tests/test_stream.py index c3a2bbf..3eb9b3a 100644 --- a/tests/test_stream.py +++ b/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, + ) diff --git a/tests/test_worker_pel_recovery.py b/tests/test_worker_pel_recovery.py new file mode 100644 index 0000000..469035e --- /dev/null +++ b/tests/test_worker_pel_recovery.py @@ -0,0 +1,172 @@ +"""HIPAA P0: abandoned Pending Entries List recovery -- worker.main's +sweep_pending() and its wiring into run(), against a mocked StreamReader +(see tests/test_worker_integration_real_backends.py for the real-Redis +proof of the concurrency/race claims made in StreamReader.claim_stale's +own docstring, which a mock cannot meaningfully exercise). + +Covers the specific gap this PR closes: a message delivered via +read_group() but never acked (worker crash, transient persistence +failure) was previously invisible to every future read_group() call +forever -- read_group() only ever asks Redis for ">" (strictly new +messages). sweep_pending() is what makes such an entry reachable again. +""" +import json +from unittest.mock import MagicMock, patch + +import worker.main as worker +from audit.config import AuditConfig + + +def _raw(event_id="evt-1"): + return json.dumps({ + "event_id": event_id, + "timestamp": "2026-01-01T12:00:00", + "service": "auth", + "event_type": "auth_login", + "decision": "success", + }) + + +# --------------------------------------------------------------------------- +# sweep_pending() +# --------------------------------------------------------------------------- + +def test_sweep_pending_processes_each_reclaimed_message(): + reader = MagicMock() + reader.claim_stale.return_value = ( + [("2-0", {"data": _raw("evt-reclaimed-a")}), ("2-1", {"data": _raw("evt-reclaimed-b")})], + [], + ) + + with patch("worker.main.handle_message") as mock_handle: + worker.sweep_pending(reader) + + reader.claim_stale.assert_called_once_with(AuditConfig.CONSUMER_NAME) + assert mock_handle.call_count == 2 + mock_handle.assert_any_call(reader, "2-0", {"data": _raw("evt-reclaimed-a")}) + mock_handle.assert_any_call(reader, "2-1", {"data": _raw("evt-reclaimed-b")}) + + +def test_sweep_pending_reclaimed_message_goes_through_real_handle_message_and_acks(): + """Not a mocked handle_message this time -- proves a reclaimed entry + runs through the exact same classify/persist/ack path a freshly-read + message does, ending in a real ack() call.""" + reader = MagicMock() + reader.claim_stale.return_value = ([("3-0", {"data": _raw("evt-real-path")})], []) + mock_sink_instance = MagicMock() + mock_sink_instance.write.return_value = True + + with patch("worker.main.SessionLocal") as mock_session_local, \ + patch("worker.main.Sink", return_value=mock_sink_instance): + mock_session_local.return_value = MagicMock() + worker.sweep_pending(reader) + + mock_sink_instance.write.assert_called_once() + reader.ack.assert_called_once_with("3-0") + + +def test_sweep_pending_does_not_reprocess_poison_ids(): + reader = MagicMock() + reader.claim_stale.return_value = ([], ["9-0", "9-1"]) + + with patch("worker.main.handle_message") as mock_handle: + worker.sweep_pending(reader) + + mock_handle.assert_not_called() + + +def test_sweep_pending_logs_poison_ids_loudly(capsys): + reader = MagicMock() + reader.claim_stale.return_value = ([], ["9-0"]) + + worker.sweep_pending(reader) + + captured = capsys.readouterr() + assert "POISON MESSAGE" in captured.out + assert "9-0" in captured.out + + +def test_sweep_pending_survives_claim_stale_raising(capsys): + """A Redis blip during the sweep itself must not propagate -- matches + read_group()'s own contract in run().""" + reader = MagicMock() + reader.claim_stale.side_effect = Exception("redis connection reset") + + worker.sweep_pending(reader) # must not raise + + captured = capsys.readouterr() + assert "pending-entry sweep failed" in captured.out + + +def test_sweep_pending_no_op_when_nothing_stale(): + reader = MagicMock() + reader.claim_stale.return_value = ([], []) + + with patch("worker.main.handle_message") as mock_handle: + worker.sweep_pending(reader) + + mock_handle.assert_not_called() + reader.ack.assert_not_called() + + +# --------------------------------------------------------------------------- +# run() wiring -- sweep_pending() called once per iteration, before +# read_group(), and its own failures never interrupt the read_group/ +# handle_message half of the loop. +# --------------------------------------------------------------------------- + +def test_run_calls_sweep_pending_every_iteration(): + mock_reader = MagicMock() + mock_reader.read_group.return_value = [] + mock_reader.claim_stale.return_value = ([], []) + + with patch("worker.main.StreamReader", return_value=mock_reader): + worker.run(max_iterations=3) + + assert mock_reader.claim_stale.call_count == 3 + + +def test_run_still_processes_new_messages_when_sweep_finds_nothing(): + mock_reader = MagicMock() + mock_reader.claim_stale.return_value = ([], []) + mock_reader.read_group.return_value = [ + (worker.AuditConfig.STREAM_NAME, [("1-0", {"data": _raw("evt-new")})]), + ] + + with patch("worker.main.StreamReader", return_value=mock_reader), \ + patch("worker.main.handle_message") as mock_handle: + worker.run(max_iterations=1) + + mock_handle.assert_called_once_with(mock_reader, "1-0", {"data": _raw("evt-new")}) + + +def test_run_processes_both_reclaimed_and_new_messages_in_one_iteration(): + mock_reader = MagicMock() + mock_reader.claim_stale.return_value = ([("2-0", {"data": _raw("evt-reclaimed")})], []) + mock_reader.read_group.return_value = [ + (worker.AuditConfig.STREAM_NAME, [("3-0", {"data": _raw("evt-new")})]), + ] + + with patch("worker.main.StreamReader", return_value=mock_reader), \ + patch("worker.main.handle_message") as mock_handle: + worker.run(max_iterations=1) + + assert mock_handle.call_count == 2 + mock_handle.assert_any_call(mock_reader, "2-0", {"data": _raw("evt-reclaimed")}) + mock_handle.assert_any_call(mock_reader, "3-0", {"data": _raw("evt-new")}) + + +def test_run_survives_sweep_pending_raising_and_still_reads_new_messages(): + mock_reader = MagicMock() + mock_reader.claim_stale.side_effect = Exception("redis blip during sweep") + mock_reader.read_group.return_value = [ + (worker.AuditConfig.STREAM_NAME, [("1-0", {"data": _raw("evt-after-sweep-blip")})]), + ] + + with patch("worker.main.StreamReader", return_value=mock_reader), \ + patch("worker.main.handle_message") as mock_handle: + worker.run(max_iterations=1) # must not raise + + mock_handle.assert_called_once_with( + mock_reader, "1-0", {"data": _raw("evt-after-sweep-blip")} + ) diff --git a/tests/test_worker_pel_recovery_integration.py b/tests/test_worker_pel_recovery_integration.py new file mode 100644 index 0000000..0d108cd --- /dev/null +++ b/tests/test_worker_pel_recovery_integration.py @@ -0,0 +1,406 @@ +"""HIPAA P0: end-to-end regression against REAL Redis and REAL MySQL for +the abandoned-Pending-Entries-List recovery path (StreamReader.claim_stale ++ worker/main.py::sweep_pending). Mirrors +tests/test_worker_integration_real_backends.py's own harness/isolation +conventions exactly (own throwaway stream, own throwaway database, skips +rather than fails when real backends aren't reachable) -- see that file's +module docstring for the full rationale, not repeated here. + +This is the one place in the suite that can actually prove the claims +StreamReader.claim_stale()'s docstring makes about concurrent-worker +safety: a mocked reader can assert "xclaim was called with these ids", +but only a real Redis server enforces that two XCLAIM calls racing for +the same still-idle entry can never both succeed. + +Every scenario here uses small min_idle_ms/max_deliveries overrides +(never the real 30s/5-attempt production defaults, see audit/config.py) +so this file runs in well under a second, not tens of seconds. +""" +import json +import os +import time +import uuid +from concurrent.futures import ThreadPoolExecutor +from datetime import datetime, timezone + +import pytest +import redis as redis_lib +from sqlalchemy import create_engine, text + +TEST_REDIS_URL = os.getenv("B0_TEST_REDIS_URL", "redis://localhost:6380") +TEST_MYSQL_ROOT_URL = os.getenv( + "B0_TEST_MYSQL_ROOT_URL", "mysql+pymysql://root:root@localhost:3306/mysql" +) +TEST_DB_NAME = "omnibioai_audit_p0_pel_test" +TEST_STREAM = f"audit:events:p0-pel-test-{uuid.uuid4().hex[:8]}" +TEST_GROUP = "audit-workers" + + +def _real_backends_available(): + try: + r = redis_lib.from_url(TEST_REDIS_URL, socket_connect_timeout=2) + r.ping() + except Exception: # noqa: BLE001 -- availability probe: any failure means "unavailable", never a crash + return False + try: + engine = create_engine(TEST_MYSQL_ROOT_URL, connect_args={"connect_timeout": 2}) + with engine.connect(): + pass + except Exception: # noqa: BLE001 -- same as above + return False + return True + + +pytestmark = pytest.mark.skipif( + not _real_backends_available(), + reason="real Redis/MySQL not reachable (set B0_TEST_REDIS_URL / " + "B0_TEST_MYSQL_ROOT_URL, or run against the dev docker-compose stack) " + "-- skipped, not failed, same convention as " + "test_worker_integration_real_backends.py", +) + + +@pytest.fixture +def real_redis_stream(): + from audit.config import AuditConfig + from consumers.stream_reader import StreamReader + + original_stream = AuditConfig.STREAM_NAME + original_group = AuditConfig.CONSUMER_GROUP + AuditConfig.STREAM_NAME = TEST_STREAM + AuditConfig.CONSUMER_GROUP = TEST_GROUP + try: + reader = StreamReader() + reader.ensure_group() + yield reader + finally: + try: + reader.redis.xgroup_destroy(TEST_STREAM, TEST_GROUP) + except Exception as cleanup_err: # noqa: BLE001 -- best-effort teardown must never mask the real test failure + print(f"[TEST TEARDOWN] xgroup_destroy failed (non-fatal): {cleanup_err}") + try: + reader.redis.delete(TEST_STREAM) + except Exception as cleanup_err: # noqa: BLE001 -- same as above + print(f"[TEST TEARDOWN] stream delete failed (non-fatal): {cleanup_err}") + AuditConfig.STREAM_NAME = original_stream + AuditConfig.CONSUMER_GROUP = original_group + + +@pytest.fixture +def real_mysql_url(): + root_engine = create_engine(TEST_MYSQL_ROOT_URL) + with root_engine.connect() as conn: + conn.execute(text(f"DROP DATABASE IF EXISTS {TEST_DB_NAME}")) + conn.execute(text(f"CREATE DATABASE {TEST_DB_NAME}")) + conn.commit() + + db_url = TEST_MYSQL_ROOT_URL.rsplit("/", 1)[0] + f"/{TEST_DB_NAME}" + + from pathlib import Path + + from alembic.config import Config + + from alembic import command + + repo_root = Path(__file__).resolve().parent.parent + cfg = Config(str(repo_root / "alembic.ini")) + cfg.set_main_option("script_location", str(repo_root / "alembic")) + cfg.set_main_option("sqlalchemy.url", db_url) + command.upgrade(cfg, "head") + + yield db_url + + with root_engine.connect() as conn: + conn.execute(text(f"DROP DATABASE IF EXISTS {TEST_DB_NAME}")) + conn.commit() + + +def _payload(event_id, **overrides): + payload = { + "event_id": event_id, + "timestamp": datetime.now(timezone.utc).isoformat(), + "service": "p0-pel-integration-test", + "event_type": "test", + "user_id": "test-user", + "action": "p0_pel_smoke", + "resource": None, + "decision": "success", + "reason": None, + "trace_id": "p0-pel-trace-1", + "context": {}, + } + payload.update(overrides) + return payload + + +def _session_local(real_mysql_url): + from sqlalchemy.orm import sessionmaker + + engine = create_engine(real_mysql_url) + return sessionmaker(bind=engine) + + +# --------------------------------------------------------------------------- +# 1. Crash before ack -> the entry is claimable once idle, and a *second* +# worker (distinct consumer identity, exactly like a different replica +# or a restarted process with a new pid-based name) can pick it up and +# successfully persist + ack it. Covers both "abandoned PEL entry +# becoming claimable" and "second worker successfully claiming and +# processing it" from the task's required scenario list in one +# end-to-end proof. +# --------------------------------------------------------------------------- + +def test_real_crash_before_ack_is_reclaimed_by_a_second_worker_and_persisted( + real_redis_stream, real_mysql_url, monkeypatch, +): + import worker.main as worker_module + + TestSessionLocal = _session_local(real_mysql_url) + monkeypatch.setattr(worker_module, "SessionLocal", TestSessionLocal) + + event_id = f"p0-crash-{uuid.uuid4()}" + real_redis_stream.redis.xadd(TEST_STREAM, {"data": json.dumps(_payload(event_id))}) + + # "Worker A" reads it and then crashes -- never calls handle_message, + # never acks. This is exactly what read_group() delivering a message + # that's then never processed looks like from Redis's point of view. + delivered = real_redis_stream.read_group("worker-a-crashed", block=2000) + assert delivered, "expected the event to be delivered to worker A" + + # Confirm it's genuinely sitting in the PEL, unacked, before recovery. + pending_before = real_redis_stream.redis.xpending(TEST_STREAM, TEST_GROUP) + assert pending_before["pending"] == 1 + + time.sleep(0.1) # exceed the tiny min_idle_ms used below + + # "Worker B" -- a different consumer identity entirely -- sweeps and + # reclaims it. + claimed, poison_ids = real_redis_stream.claim_stale("worker-b-recovers", min_idle_ms=50) + assert poison_ids == [] + assert len(claimed) == 1 + message_id, fields = claimed[0] + + result = worker_module.handle_message(real_redis_stream, message_id, fields) + assert result is True + + with TestSessionLocal() as session: + from db.models import AuditEventRecord + + row = session.get(AuditEventRecord, event_id) + assert row is not None + assert row.action == "p0_pel_smoke" + + pending_after = real_redis_stream.redis.xpending(TEST_STREAM, TEST_GROUP) + assert pending_after["pending"] == 0 + + +# --------------------------------------------------------------------------- +# 2. Multiple workers racing to recover the very same abandoned entry -- +# the concurrency-safety claim StreamReader.claim_stale()'s docstring +# makes. Only real Redis can prove this; a mock can't enforce atomicity. +# --------------------------------------------------------------------------- + +def test_real_concurrent_workers_racing_for_the_same_entry_only_one_wins( + real_redis_stream, real_mysql_url, +): + event_id = f"p0-race-{uuid.uuid4()}" + real_redis_stream.redis.xadd(TEST_STREAM, {"data": json.dumps(_payload(event_id))}) + + delivered = real_redis_stream.read_group("worker-a-crashed", block=2000) + assert delivered + time.sleep(0.1) + + from consumers.stream_reader import StreamReader + + reader_b = real_redis_stream + reader_c = StreamReader() + reader_c.stream = TEST_STREAM + + results = [] + with ThreadPoolExecutor(max_workers=2) as pool: + futures = [ + pool.submit(reader_b.claim_stale, "worker-b", group=TEST_GROUP, min_idle_ms=50), + pool.submit(reader_c.claim_stale, "worker-c", group=TEST_GROUP, min_idle_ms=50), + ] + for f in futures: + results.append(f.result()) + + total_claimed = sum(len(claimed) for claimed, _poison in results) + assert total_claimed == 1, ( + "exactly one of the two racing claim_stale() calls must win this " + f"entry, got {total_claimed} total claims across both" + ) + + +# --------------------------------------------------------------------------- +# 3. Transient persistence failure -- previously permanent event loss +# (see worker/main.py's old docstring claim vs. actual behavior), +# now recoverable: the first delivery fails to persist, the message +# is reclaimed once idle, and the retry succeeds. +# --------------------------------------------------------------------------- + +def test_real_transient_persistence_failure_then_reclaim_succeeds( + real_redis_stream, real_mysql_url, monkeypatch, +): + import worker.main as worker_module + from consumers.sink import Sink + + TestSessionLocal = _session_local(real_mysql_url) + + event_id = f"p0-transient-{uuid.uuid4()}" + real_redis_stream.redis.xadd(TEST_STREAM, {"data": json.dumps(_payload(event_id))}) + + response = real_redis_stream.read_group("worker-a", block=2000) + assert response + message_id, fields = response[0][1][0] + + # First attempt: simulate a transient MySQL failure during persistence. + class _FailingSink(Sink): + def write(self, event): + raise Exception("simulated transient MySQL outage") # noqa: TRY002 -- deliberately generic, standing in for "any real DB exception class" + + monkeypatch.setattr(worker_module, "SessionLocal", TestSessionLocal) + monkeypatch.setattr(worker_module, "Sink", _FailingSink) + first_result = worker_module.handle_message(real_redis_stream, message_id, fields) + assert first_result is False + + pending = real_redis_stream.redis.xpending(TEST_STREAM, TEST_GROUP) + assert pending["pending"] == 1 # still unacked after the transient failure + + time.sleep(0.1) + + # MySQL "recovers" -- restore the real Sink and reclaim + retry. + monkeypatch.setattr(worker_module, "Sink", Sink) + claimed, poison_ids = real_redis_stream.claim_stale("worker-b", min_idle_ms=50) + assert poison_ids == [] + assert len(claimed) == 1 + second_message_id, second_fields = claimed[0] + + second_result = worker_module.handle_message(real_redis_stream, second_message_id, second_fields) + assert second_result is True + + with TestSessionLocal() as session: + from db.models import AuditEventRecord + + row = session.get(AuditEventRecord, event_id) + assert row is not None + + +# --------------------------------------------------------------------------- +# 4. Duplicate delivery through the reclaim path specifically -- a +# successful persist whose ack was itself lost (crash between the +# MySQL commit and the XACK call) must not create a second row when +# a different worker reclaims and reprocesses the same entry. +# Complements test_real_duplicate_delivery_does_not_duplicate_row in +# test_worker_integration_real_backends.py, which exercises Sink +# directly rather than through claim_stale(). +# --------------------------------------------------------------------------- + +def test_real_duplicate_delivery_after_reclaim_does_not_duplicate_row( + real_redis_stream, real_mysql_url, monkeypatch, +): + import worker.main as worker_module + + TestSessionLocal = _session_local(real_mysql_url) + monkeypatch.setattr(worker_module, "SessionLocal", TestSessionLocal) + + event_id = f"p0-dup-reclaim-{uuid.uuid4()}" + real_redis_stream.redis.xadd(TEST_STREAM, {"data": json.dumps(_payload(event_id))}) + + response = real_redis_stream.read_group("worker-a", block=2000) + assert response + _message_id, fields = response[0][1][0] + + # Worker A persists successfully but "crashes" before the ack() call + # actually below -- call Sink directly to model exactly that gap. + from consumers.sink import Sink + + db = TestSessionLocal() + try: + payload = json.loads(fields["data"]) + payload["integrity_status"] = "unsigned" + Sink(db).write(payload) + finally: + db.close() + # Deliberately no ack() here -- modeling the crash-after-commit gap. + + time.sleep(0.1) + + claimed, poison_ids = real_redis_stream.claim_stale("worker-b", min_idle_ms=50) + assert poison_ids == [] + assert len(claimed) == 1 + second_message_id, second_fields = claimed[0] + + # Worker B reprocesses the same event_id end-to-end via the real + # handle_message() -- Sink.write's IntegrityError-swallow (PK on + # event_id) must make this a safe no-op, and worker B must still ack. + result = worker_module.handle_message(real_redis_stream, second_message_id, second_fields) + assert result is True + + with TestSessionLocal() as session: + from sqlalchemy import func + + from db.models import AuditEventRecord + + count = session.query(func.count(AuditEventRecord.event_id)).filter( + AuditEventRecord.event_id == event_id + ).scalar() + assert count == 1 # exactly one row, not two + + pending_after = real_redis_stream.redis.xpending(TEST_STREAM, TEST_GROUP) + assert pending_after["pending"] == 0 + + +# --------------------------------------------------------------------------- +# 5. Poison message (deterministically malformed, e.g. truly not JSON) -- +# must be retried a bounded number of times, never forever. +# --------------------------------------------------------------------------- + +def test_real_malformed_event_is_abandoned_after_max_deliveries_not_retried_forever( + real_redis_stream, real_mysql_url, monkeypatch, +): + import worker.main as worker_module + + TestSessionLocal = _session_local(real_mysql_url) + monkeypatch.setattr(worker_module, "SessionLocal", TestSessionLocal) + + MAX_DELIVERIES = 3 + real_redis_stream.redis.xadd(TEST_STREAM, {"data": "this is not valid json at all"}) + + response = real_redis_stream.read_group("worker-a", block=2000) + assert response + message_id, fields = response[0][1][0] + + # Delivery #1 (the read above) already failed to parse -- handle_message + # never acks. Reclaim it (MAX_DELIVERIES - 1) more times, each one also + # failing identically (it's not transient -- the payload is simply not + # JSON), to reach exactly MAX_DELIVERIES total deliveries. + result = worker_module.handle_message(real_redis_stream, message_id, fields) + assert result is False + + for _ in range(MAX_DELIVERIES - 1): + time.sleep(0.1) + claimed, poison_ids = real_redis_stream.claim_stale( + "worker-b", min_idle_ms=50, max_deliveries=MAX_DELIVERIES, + ) + assert poison_ids == [] # not poison yet -- still under the threshold + assert len(claimed) == 1 + cmsg_id, cfields = claimed[0] + result = worker_module.handle_message(real_redis_stream, cmsg_id, cfields) + assert result is False # still unparseable + + # One more sweep: this entry has now been delivered MAX_DELIVERIES + # times without ever succeeding -- must be abandoned as poison, ACKed + # directly by claim_stale(), never handed back for a (MAX_DELIVERIES+1)th + # attempt. + time.sleep(0.1) + claimed, poison_ids = real_redis_stream.claim_stale( + "worker-b", min_idle_ms=50, max_deliveries=MAX_DELIVERIES, + ) + assert claimed == [] + assert len(poison_ids) == 1 + + # The PEL is now empty -- proof the loop actually terminated, not + # just that this one sweep classified it correctly. + pending_after = real_redis_stream.redis.xpending(TEST_STREAM, TEST_GROUP) + assert pending_after["pending"] == 0 diff --git a/worker/main.py b/worker/main.py index e9aefeb..649f230 100644 --- a/worker/main.py +++ b/worker/main.py @@ -23,24 +23,37 @@ def handle_message(reader: StreamReader, message_id: str, fields: dict) -> bool: Deserialize -> classify -> persist -> ack, in that order, and only ack after a successful DB commit. Any failure along the way leaves the - message unacknowledged in the consumer group's pending entries list so - it can be retried -- it is never dropped. + message unacknowledged in the consumer group's pending entries list. + + HIPAA P0: "unacknowledged" alone does NOT mean "will be retried" -- + that claim was this docstring's own, and it was false. read_group() + only ever asks Redis for ">" (strictly new messages), so a delivered- + but-unacked entry is invisible to every future read_group() call, from + this consumer identity or any other, forever -- there was no retry + path. What actually makes retry happen is run()'s call to + sweep_pending() every loop iteration, via StreamReader.claim_stale() + (XPENDING+XCLAIM): once an entry has sat unacked for + AuditConfig.PEL_MIN_IDLE_MS, ANY live worker (this one after a + reconnect, or a different replica entirely) reclaims it and runs it + back through this exact function. A message is dropped only if it + exceeds AuditConfig.PEL_MAX_DELIVERIES without ever succeeding -- + see sweep_pending()'s own docstring for why that bound exists. PR2: classification (valid/invalid/unsigned -- see consumers/ processor.py::classify_event_integrity) is deliberately NOT another reason to leave a message unacked. Parse/DB failures above are retried because they may be transient or fixable; a signature is not -- it is - either right or wrong forever, so retrying it accomplishes nothing and - would only accumulate a permanent poison-pill pending entry. Every - classification is persisted (as durable evidence, including forged - attempts) and acked, exactly like today's fully-unsigned traffic. - Uses fields["data"] itself, not a re-serialization of `event` -- the - signature covers the exact transmitted bytes (see audit/signing.py). + either right or wrong forever, so retrying it accomplishes nothing. + Every classification is persisted (as durable evidence, including + forged attempts) and acked, exactly like today's fully-unsigned + traffic. Uses fields["data"] itself, not a re-serialization of + `event` -- the signature covers the exact transmitted bytes (see + audit/signing.py). """ raw_data = fields["data"] try: event = parse_audit_event(raw_data) - except Exception as e: + except Exception as e: # noqa: BLE001 -- any parse failure is retriable via sweep_pending(), never a crash print(f"[WORKER] failed to parse message {message_id}: {e}") return False @@ -62,7 +75,7 @@ def handle_message(reader: StreamReader, message_id: str, fields: dict) -> bool: payload = event.model_dump() payload["integrity_status"] = integrity_status Sink(db).write(payload) - except Exception as e: + except Exception as e: # noqa: BLE001 -- any persistence failure is retriable via sweep_pending(), never a crash print(f"[WORKER] failed to persist message {message_id}: {e}") return False finally: @@ -72,6 +85,40 @@ def handle_message(reader: StreamReader, message_id: str, fields: dict) -> bool: return True +def sweep_pending(reader: StreamReader) -> None: + """Reclaims and processes Pending Entries List messages abandoned by + a crashed or stuck consumer -- see StreamReader.claim_stale()'s own + docstring for the XPENDING+XCLAIM mechanism this wraps. Called once + per run() loop iteration, before reading new messages, so a backlog + of abandoned events isn't starved by a steady stream of new traffic. + + Never raises: a Redis blip during the sweep itself must not kill the + worker, matching read_group()'s own contract in run() below -- a + failed sweep just tries again next iteration. + + Reclaimed entries run through the exact same handle_message() used + for freshly-read messages -- same classify/persist/ack path, so a + reclaimed message that fails again (still-down MySQL, say) is simply + left pending again and picked up by the next sweep once + PEL_MIN_IDLE_MS has re-elapsed, same as any other unacked entry. + """ + try: + claimed, poison_ids = reader.claim_stale(AuditConfig.CONSUMER_NAME) + except Exception as e: # noqa: BLE001 -- a Redis blip here must never kill the worker, same "NEVER break core system" contract read_group() below already has + print(f"[WORKER] pending-entry sweep failed, will retry: {e}") + return + + for message_id in poison_ids: + print( + f"[WORKER] POISON MESSAGE {message_id} abandoned after " + f">={AuditConfig.PEL_MAX_DELIVERIES} delivery attempts -- " + f"ACKed without processing, never persisted, will not be retried again" + ) + + for message_id, fields in claimed: + handle_message(reader, message_id, fields) + + def run(max_iterations=None): """Main consumer loop. @@ -83,6 +130,7 @@ def run(max_iterations=None): iterations = 0 while max_iterations is None or iterations < max_iterations: + sweep_pending(reader) try: response = reader.read_group(AuditConfig.CONSUMER_NAME) except RedisTimeoutError: @@ -97,7 +145,7 @@ def run(max_iterations=None): # crash forever (PR-B0 follow-up). No print here: this is # normal idle-stream behavior, not an error to alarm on. response = [] - except Exception as e: + except Exception as e: # noqa: BLE001 -- a Redis blip here must never kill the worker # A genuine unexpected Redis/connection failure. Same # print-and-continue convention as handle_message()/ # audit/logger.py above: never let a transient infra blip