fix(ingestion): close lost-event race on persist failure (#62)

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))

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

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)

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()