refactor web pagination and add partitioned events, flap/timeout showcase agents

server/monlet_server/services/ingestion.py

@@ -8,7 +8,7 @@ from sqlalchemy.ext.asyncio import AsyncSession
 
 from .. import metrics
 from ..logging_config import get_logger
-from ..models import Agent, Check, Event, Incident, NotificationOutbox
+from ..models import Agent, Check, Event, EventIngestDedup, Incident, NotificationOutbox
 from ..redaction import redact
 from ..schemas import CheckResultEvent, HeartbeatRequest
 from ..settings import get_settings
@@ -155,26 +155,45 @@ async def ingest_event(
     agent_id: str,
     ev: CheckResultEvent,
 ) -> bool:
-    """Return True if a new change row was written, False if not (duplicate/unchanged/late)."""
+    """Return True if a new change row was written, False if not (duplicate/unchanged/late).
+
+    Idempotency is enforced first via the standalone event_ingest_dedup table — a
+    duplicate event_id never touches checks/incidents/outbox/events. The full pipeline
+    (dedup + events + check + incident + outbox) commits as a single transaction by
+    the caller, so a partial failure cannot leave a dedup row without its side-effects.
+    """
     observed = to_utc(ev.observed_at)
     output = redact(ev.output)
     check_summary = (output or "")[:200] if output else None
     incident_key = _incident_key(agent_id, ev)
+    event_uuid = UUID(ev.event_id)
 
+    # Order matters: late check before dedup insert so that repeated late events stay
+    # observable as `late`, not silently collapsed into `deduplicated`.
     chk_res = await session.execute(
         select(Check).where(Check.agent_id == agent_id, Check.check_id == ev.check_id)
     )
     cur = chk_res.scalar_one_or_none()
     is_late = cur is not None and cur.last_observed_at > observed
-    changed = cur is None or cur.status != ev.status or cur.exit_code != ev.exit_code
-
     if is_late:
         metrics.events_total.labels(result="late").inc()
         return False
 
+    dedup_stmt = (
+        pg_insert(EventIngestDedup)
+        .values(event_id=event_uuid, received_at=datetime.now(UTC))
+        .on_conflict_do_nothing(index_elements=[EventIngestDedup.event_id])
+        .returning(EventIngestDedup.event_id)
+    )
+    if (await session.execute(dedup_stmt)).scalar_one_or_none() is None:
+        metrics.events_total.labels(result="deduplicated").inc()
+        return False
+
+    changed = cur is None or cur.status != ev.status or cur.exit_code != ev.exit_code
+
     if changed:
         stmt = pg_insert(Event).values(
-            event_id=UUID(ev.event_id),
+            event_id=event_uuid,
             agent_id=agent_id,
             check_id=ev.check_id,
             observed_at=observed,
@@ -186,14 +205,7 @@ async def ingest_event(
             notifications_enabled=ev.notifications_enabled,
             incident_key=incident_key,
         )
-        stmt = stmt.on_conflict_do_nothing(index_elements=[Event.event_id]).returning(
-            Event.event_id
-        )
-        result = await session.execute(stmt)
-        inserted = result.scalar_one_or_none()
-        if inserted is None:
-            metrics.events_total.labels(result="deduplicated").inc()
-            return False
+        await session.execute(stmt)
         metrics.events_total.labels(result="accepted").inc()
     else:
         metrics.events_total.labels(result="unchanged").inc()
@@ -218,7 +230,7 @@ async def ingest_event(
         "summary": check_summary,
     }
     if changed:
-        update_set["last_event_id"] = UUID(ev.event_id)
+        update_set["last_event_id"] = event_uuid
         update_set["incident_key"] = incident_key
     chk_stmt = pg_insert(Check).values(**check_values)
     chk_stmt = chk_stmt.on_conflict_do_update(