From 5fa2009bf09fb15bd902ea94aaf2144ed5d2d76e Mon Sep 17 00:00:00 2001 From: JumpMaster Date: Mon, 15 Jun 2026 23:53:56 -0600 Subject: [PATCH] fix(ingestion): close lost-event race on persist failure MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The webhook handler marked the Redis idempotency key (SET NX EX 24h) before persisting the incident. A transient failure in `ingest` after the mark turned the sender's byte-identical retry into a silently dropped event — the exact failure mode ADR 0001 ("never lose an event") is meant to eliminate. Wrap `ingest` in the handler: on failure, best-effort `unmark` the key (new IdempotencyStore method) and return INFRA_UNAVAILABLE (503) so the sender re-delivers and the now-cleared key lets the retry re-process. If the unmark itself fails, still return 503 and let the key fall back to its 24h TTL. The fix is safe because `ingest` is single-transaction: a failure rolls back, so unmark never erases a persisted event. Adds regression tests for ingest-fails-after-mark (with and without a failing unmark) and a mark→unmark→processable-again round-trip. Closes #62 Co-Authored-By: Claude Opus 4.8 (1M context) --- sentinel/ingestion/idempotency.py | 18 +++++++- sentinel/ingestion/webhook.py | 28 +++++++++--- tests/unit/ingestion/test_idempotency.py | 18 ++++++++ tests/unit/ingestion/test_webhook_handler.py | 46 ++++++++++++++++++++ 4 files changed, 103 insertions(+), 7 deletions(-) diff --git a/sentinel/ingestion/idempotency.py b/sentinel/ingestion/idempotency.py index 58ab891..ef19739 100644 --- a/sentinel/ingestion/idempotency.py +++ b/sentinel/ingestion/idempotency.py @@ -18,14 +18,30 @@ async def check_and_mark(self, source: str, body: bytes) -> bool: """Return True if (source, body) was already seen; False if first.""" ... + async def unmark(self, source: str, body: bytes) -> None: + """Drop the idempotency key so a retry of (source, body) re-processes. + + Compensating action when persistence fails *after* the key was marked — + without it, the sender's byte-identical retry would dedup into silent loss + (ADR 0001, "never lose an event"). + """ + ... + + +def _idempotency_key(source: str, body: bytes) -> str: + return f"webhook:{source}:{hashlib.sha256(body).hexdigest()}" + class RedisIdempotencyStore: def __init__(self, redis: Redis) -> None: self._redis = redis async def check_and_mark(self, source: str, body: bytes) -> bool: - key = f"webhook:{source}:{hashlib.sha256(body).hexdigest()}" + key = _idempotency_key(source, body) # NX EX semantics: SET if not exists, with expiry. # Returns True on first set, None if already present. result = await self._redis.set(key, b"1", nx=True, ex=_TTL_SECONDS) return result is None + + async def unmark(self, source: str, body: bytes) -> None: + await self._redis.delete(_idempotency_key(source, body)) diff --git a/sentinel/ingestion/webhook.py b/sentinel/ingestion/webhook.py index 780a0f3..5910241 100644 --- a/sentinel/ingestion/webhook.py +++ b/sentinel/ingestion/webhook.py @@ -126,12 +126,28 @@ async def handle( payload_hash = sha256(body).hexdigest() # 7. Persist + outbox in one tx. - result = await self._repo.ingest( - alert, - fingerprint=fp, - outbox_topic=self._outbox_topic, - payload_hash=payload_hash, - ) + # + # The idempotency key is already marked (step 4). If persistence fails + # here we MUST clear that key, else the sender's byte-identical retry + # dedups into silent permanent loss — the exact failure ADR 0001 forbids. + # Compensate, then return a retryable 503 so the sender re-delivers. + try: + result = await self._repo.ingest( + alert, + fingerprint=fp, + outbox_topic=self._outbox_topic, + payload_hash=payload_hash, + ) + except Exception: + log.exception("ingest_failed_after_idempotency_mark") + try: + await self._idempotency.unmark(source, body) + except Exception: + # Best-effort: the key falls back to its 24h TTL. Still return + # 503 so the sender retries rather than treating this as done. + log.exception("idempotency_unmark_failed") + outcome = HandlerOutcome.INFRA_UNAVAILABLE + return HandlerResult(outcome=outcome, reason="persist_failed") incident_id = result.incident_id outcome = ( HandlerOutcome.RECURRED diff --git a/tests/unit/ingestion/test_idempotency.py b/tests/unit/ingestion/test_idempotency.py index 36207c5..a9715cd 100644 --- a/tests/unit/ingestion/test_idempotency.py +++ b/tests/unit/ingestion/test_idempotency.py @@ -46,6 +46,24 @@ async def test_different_source_is_not_duplicate(redis: FakeRedis) -> None: assert await store.check_and_mark("datadog", body) is False +@pytest.mark.asyncio +async def test_unmark_makes_body_processable_again(redis: FakeRedis) -> None: + """After unmark, the same body is treated as new — the compensating delete + that closes the lost-event window (issue #62).""" + store = RedisIdempotencyStore(redis) + body = b'{"a":1}' + assert await store.check_and_mark("sentry", body) is False # marked + await store.unmark("sentry", body) + assert await store.check_and_mark("sentry", body) is False # processable again + + +@pytest.mark.asyncio +async def test_unmark_absent_key_is_noop(redis: FakeRedis) -> None: + store = RedisIdempotencyStore(redis) + # Deleting a key that was never set must not raise. + await store.unmark("sentry", b'{"never":"set"}') + + @pytest.mark.asyncio async def test_ttl_is_24h(redis: FakeRedis) -> None: store = RedisIdempotencyStore(redis) diff --git a/tests/unit/ingestion/test_webhook_handler.py b/tests/unit/ingestion/test_webhook_handler.py index 26ec952..0198625 100644 --- a/tests/unit/ingestion/test_webhook_handler.py +++ b/tests/unit/ingestion/test_webhook_handler.py @@ -35,6 +35,7 @@ def _make_handler( ) idempotency = MagicMock() idempotency.check_and_mark = AsyncMock(return_value=False) + idempotency.unmark = AsyncMock() handler = WebhookHandler( incident_repo=incident_repo, idempotency=idempotency, @@ -144,6 +145,51 @@ async def test_bad_json_returns_bad_request(fixture_body: bytes) -> None: assert result.outcome == HandlerOutcome.BAD_REQUEST +@pytest.mark.asyncio +async def test_ingest_failure_unmarks_idempotency_key(fixture_body: bytes) -> None: + """Lost-event race (issue #62): if persistence fails *after* the idempotency + key is marked, the key must be cleared so the sender's byte-identical retry + re-processes the event instead of being silently deduped into oblivion. + """ + handler, repo, idem = _make_handler() + repo.ingest = AsyncMock(side_effect=RuntimeError("transient db failure")) + from sentinel.integrations.base import compute_hmac_sha256 + + sig = compute_hmac_sha256(fixture_body, b"shh") + result = await handler.handle( + source="sentry", + headers={"Sentry-Hook-Signature": sig}, + body=fixture_body, + ) + + # Sender must see a retryable status (503), not a 2xx that ends delivery. + assert result.outcome == HandlerOutcome.INFRA_UNAVAILABLE + # The compensating delete must have fired so the retry is not deduped. + idem.unmark.assert_awaited_once_with("sentry", fixture_body) + + +@pytest.mark.asyncio +async def test_ingest_failure_still_503_when_unmark_also_fails( + fixture_body: bytes, +) -> None: + """If the compensating unmark itself fails (e.g. Redis down), the handler + must still return a retryable 503 rather than surfacing the unmark error. + """ + handler, repo, idem = _make_handler() + repo.ingest = AsyncMock(side_effect=RuntimeError("transient db failure")) + idem.unmark = AsyncMock(side_effect=RuntimeError("redis down")) + from sentinel.integrations.base import compute_hmac_sha256 + + sig = compute_hmac_sha256(fixture_body, b"shh") + result = await handler.handle( + source="sentry", + headers={"Sentry-Hook-Signature": sig}, + body=fixture_body, + ) + + assert result.outcome == HandlerOutcome.INFRA_UNAVAILABLE + + @pytest.mark.asyncio async def test_handler_median_under_50ms(fixture_body: bytes) -> None: handler, _, _ = _make_handler()