Ingestion reliability: SDK backoff and spool, idempotent batch, silence alert

README.md

@@ -242,6 +242,7 @@ Base URL: `http://localhost:8000`
 | `GET`    | `/v1/datasets/{id}/compare`       | Case-by-case diff (`?base=&candidate=`) |
 | `GET`    | `/v1/datasets/runs/{id}`          | One run with every case result    |
 | `DELETE` | `/v1/datasets/runs/{id}`          | Delete a run                      |
+| `GET`    | `/v1/ingest/health`               | Last trace and counts per project |
 | `GET`    | `/v1/evaluations/by-model`        | Turn health by model and provider |
 | `POST`   | `/v1/analytics/query`             | Flexible metrics query            |
 | `GET`    | `/v1/analytics/query/schema`      | Views, measures and dimensions    |
@@ -311,6 +312,18 @@ failed traces with reasons, one trace's spans, tools and provider chain, alerts,
 comparisons and metric queries. Read-only, and trace input/output are never returned unless
 `OBSERVER_MCP_CONTENT=1`. Setup and tools: [docs/mcp.md](docs/mcp.md).
 
+### Ingestion reliability
+
+The Python SDK exporter retries only what can succeed later (network errors, 408/425/429/5xx) with
+exponential backoff and jitter, honours `Retry-After`, and counts permanent 4xx answers as
+rejected instead of retrying them. Pass `spool_dir` to keep batches that still fail on disk (mode
+0600, unencrypted, bounded by `spool_max_bytes`) and resend them oldest first after the backend
+returns or the process restarts; `exporter.stats()` reports `exported, retried, rejected, dropped,
+spooled, recovered, spool_evicted, pending`. `/v1/traces/batch` is idempotent (a retried batch does
+not fail on spans already stored) and rate-limited by `RATE_LIMIT_BATCH` (default `120/minute`).
+`GET /v1/ingest/health` and the Overview card show when each project last sent a trace; the alert
+metric **Minutes without traces** (`ingest_silence_minutes`) notifies when one goes quiet.
+
 ### Test cases (datasets)
 
 On a trace page, **Save as case** stores a failure as a reusable test case in a dataset. A case is

backend/app/alerts.py

@@ -191,7 +191,14 @@ async def evaluate_rules(session: AsyncSession, *, now: datetime | None = None)
         if spec is None or rule.comparator not in COMPARATORS:
             continue
         details: dict[str, Any] = {}
-        if spec["kind"] == "quota":
+        if spec["kind"] == "ingest":
+            from app.api.ingest_health import silence_minutes
+
+            value = await silence_minutes(session, rule.project_id, rule.environment, now)
+            if value is None:
+                continue  # the project never sent anything: nothing to be silent about
+            samples, details = 1, {"project_id": rule.project_id}
+        elif spec["kind"] == "quota":
             value, samples, details = _quota_value(rule.scope or "")
             if value is None:
                 continue  # no trustworthy snapshot: neither fire nor resolve
@@ -207,7 +214,10 @@ async def evaluate_rules(session: AsyncSession, *, now: datetime | None = None)
                 )
             value, samples = compute_metric(rule.metric, cache[key], scope=rule.scope or "")
         result.evaluated += 1
-        enough = value is not None and samples >= int(rule.min_samples)
+        # Snapshot-style metrics (quota, ingest silence) have one reading, not a sample count.
+        enough = value is not None and (
+            spec["kind"] in {"quota", "ingest"} or samples >= int(rule.min_samples)
+        )
         open_event = (
             (
                 await session.execute(

backend/app/api/ingest_health.py

@@ -0,0 +1,71 @@
+"""Is telemetry still arriving? (``GET /v1/ingest/health``)
+
+Per project: when the last trace was stored and how many arrived in the last 5 minutes, hour and
+day. A sender that stops (crashed, blocked, spool full) shows up as growing silence, which the
+``ingest_silence_minutes`` alert metric turns into a notification.
+"""
+
+from __future__ import annotations
+
+from datetime import datetime, timedelta, timezone
+from typing import Any
+
+from fastapi import APIRouter, Depends
+from sqlalchemy import func, select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.db import get_session
+from app.models.trace import Trace
+
+router = APIRouter()
+
+
+def _aware(value: datetime | None) -> datetime | None:
+    if value is not None and value.tzinfo is None:
+        return value.replace(tzinfo=timezone.utc)
+    return value
+
+
+async def silence_minutes(
+    session: AsyncSession, project_id: str, environment: str | None, now: datetime
+) -> float | None:
+    """Minutes since the project's last stored trace; ``None`` if it never sent one."""
+    query = select(func.max(Trace.created_at)).where(Trace.project_id == project_id)
+    if environment:
+        query = query.where(Trace.environment == environment)
+    last = _aware((await session.execute(query)).scalar_one_or_none())
+    return None if last is None else max(0.0, (now - last).total_seconds() / 60.0)
+
+
+@router.get("/health")
+async def ingest_health(session: AsyncSession = Depends(get_session)) -> dict[str, Any]:
+    now = datetime.now(timezone.utc)
+    t5, t60, t24 = (now - timedelta(minutes=5), now - timedelta(hours=1), now - timedelta(days=1))
+    rows = (
+        await session.execute(
+            select(
+                Trace.project_id,
+                func.max(Trace.created_at),
+                func.count().filter(Trace.created_at >= t5),
+                func.count().filter(Trace.created_at >= t60),
+                func.count().filter(Trace.created_at >= t24),
+            )
+            .where(Trace.project_id.is_not(None))
+            .group_by(Trace.project_id)
+        )
+    ).all()
+    projects = []
+    for project_id, last, n5, n60, n24 in rows:
+        last = _aware(last)
+        projects.append(
+            {
+                "project_id": project_id,
+                "last_trace_at": last.isoformat() if last else None,
+                "silence_minutes": round((now - last).total_seconds() / 60.0, 1) if last else None,
+                "traces_5m": n5,
+                "traces_1h": n60,
+                "traces_24h": n24,
+            }
+        )
+    projects.sort(key=lambda p: p["silence_minutes"] if p["silence_minutes"] is not None else 1e9)
+    return {"checked_at": now.isoformat(), "projects": projects}

backend/app/api/traces.py

@@ -2,6 +2,7 @@
 
 from __future__ import annotations
 
+import os
 import uuid
 from datetime import datetime, timezone
 
@@ -19,6 +20,9 @@
 
 router = APIRouter()
 
+# The SDK flushes every few seconds, so the old fixed 10/minute rejected healthy senders.
+_BATCH_LIMIT = os.getenv("RATE_LIMIT_BATCH", "120/minute")
+
 
 def _safe_uuid(value: str | None) -> str | None:
     """Validate and return a UUID string, or None if invalid."""
@@ -249,7 +253,7 @@ async def create_trace(
 
 
 @router.post("/batch", response_model=list[TraceResponse])
-@limiter.limit("10/minute")
+@limiter.limit(_BATCH_LIMIT)
 async def create_traces_batch(
     request: Request,
     batch_data: BatchSpansCreate,
@@ -270,6 +274,13 @@ async def create_traces_batch(
 
     created_traces: list[Trace] = []
 
+    # A retried batch (the first answer was lost) must not fail on spans already stored.
+    wanted = [s.id for spans in spans_by_trace.values() for s in spans if s.id]
+    stored_ids: set[str] = set()
+    if wanted:
+        rows = await session.execute(select(Span.id).where(Span.id.in_(wanted)))
+        stored_ids = {str(i) for i in rows.scalars()}
+
     for trace_id, spans in spans_by_trace.items():
         # Check if trace already exists
         query = select(Trace).where(Trace.id == trace_id)
@@ -295,6 +306,8 @@ async def create_traces_batch(
         # Add spans
         mode = await capture_mode(session, trace.project_id, "full")
         for span_data in spans:
+            if span_data.id and span_data.id in stored_ids:
+                continue
             captured = apply_capture(
                 mode,
                 inp=span_data.input,

backend/app/main.py

@@ -15,6 +15,7 @@
 from app.api import analytics, evaluations, manitos_ingest, providers, traces
 from app.api import dataset_runs as dataset_runs_api
 from app.api import datasets as datasets_api
+from app.api import ingest_health as ingest_health_api
 from app.api import otlp as otlp_api
 from app.api import privacy as privacy_api
 from app.auth import require_api_key
@@ -121,6 +122,12 @@ def _rate_limit_handler(request: Request, exc: RateLimitExceeded) -> JSONRespons
     tags=["privacy"],
     dependencies=[_app_auth],
 )
+app.include_router(
+    ingest_health_api.router,
+    prefix="/v1/ingest",
+    tags=["ingest"],
+    dependencies=[_app_auth],
+)
 app.include_router(
     otlp_api.router,
     prefix="/v1/otlp",

backend/app/metrics_core.py

@@ -36,6 +36,11 @@
         "unit": "ratio",
         "kind": "provider",
     },
+    "ingest_silence_minutes": {
+        "label": "Minutos sin recibir trazas",
+        "unit": "count",
+        "kind": "ingest",
+    },
     "provider_quota_out": {
         "label": "Proveedores sin cuota",
         "unit": "count",

backend/tests/test_ingest_reliability.py

@@ -0,0 +1,119 @@
+import uuid
+from datetime import datetime, timedelta, timezone
+
+import pytest
+from sqlalchemy import func, select
+
+from app.alerts import evaluate_rules
+from app.models import AlertEvent, AlertRule, Span, Trace
+
+
+def _batch(span_id, trace_id):
+    return {
+        "spans": [
+            {
+                "id": span_id,
+                "trace_id": trace_id,
+                "name": "s",
+                "span_type": "llm",
+                "start_time": 1_750_000_000.0,
+                "end_time": 1_750_000_001.0,
+                "status": "ok",
+            }
+        ]
+    }
+
+
+@pytest.mark.anyio
+async def test_a_retried_batch_does_not_fail_or_duplicate(client, db_session):
+    span_id, trace_id = str(uuid.uuid4()), str(uuid.uuid4())
+    first = await client.post("/v1/traces/batch", json=_batch(span_id, trace_id))
+    assert first.status_code == 200
+    again = await client.post("/v1/traces/batch", json=_batch(span_id, trace_id))
+    assert again.status_code == 200  # the lost-answer retry used to be a 500
+    count = (await db_session.execute(select(func.count()).select_from(Span))).scalar()
+    assert count == 1
+
+
+@pytest.mark.anyio
+async def test_a_batch_with_old_and_new_spans_stores_only_the_new(client, db_session):
+    trace_id, old, new = str(uuid.uuid4()), str(uuid.uuid4()), str(uuid.uuid4())
+    await client.post("/v1/traces/batch", json=_batch(old, trace_id))
+    both = {"spans": _batch(old, trace_id)["spans"] + _batch(new, trace_id)["spans"]}
+    assert (await client.post("/v1/traces/batch", json=both)).status_code == 200
+    ids = set((await db_session.execute(select(Span.id))).scalars())
+    assert ids == {old, new}
+
+
+async def _trace(db, project, minutes_ago):
+    created = datetime.now(timezone.utc) - timedelta(minutes=minutes_ago)
+    db.add(
+        Trace(
+            name="t",
+            start_time=created,
+            status="ok",
+            project_id=project,
+            environment="e",
+            created_at=created,
+        )
+    )
+    await db.commit()
+
+
+@pytest.mark.anyio
+async def test_ingest_health_reports_last_trace_and_counts(client, db_session):
+    await _trace(db_session, "alive", 2)
+    await _trace(db_session, "alive", 30)
+    await _trace(db_session, "quiet", 600)
+    data = (await client.get("/v1/ingest/health")).json()
+    rows = {p["project_id"]: p for p in data["projects"]}
+    assert rows["alive"]["traces_5m"] == 1 and rows["alive"]["traces_1h"] == 2
+    assert rows["alive"]["silence_minutes"] < 5
+    assert rows["quiet"]["traces_24h"] == 1 and rows["quiet"]["silence_minutes"] > 500
+    assert data["projects"][0]["project_id"] == "alive"  # most recently heard first
+
+
+@pytest.mark.anyio
+async def test_silence_alert_fires_then_resolves_when_traces_return(db_session):
+    await _trace(db_session, "svc", 90)
+    rule = AlertRule(
+        name="svc silent",
+        metric="ingest_silence_minutes",
+        comparator=">",
+        threshold=30.0,
+        window_minutes=60,
+        min_samples=10,  # min_samples is ignored
+        project_id="svc",
+        severity="warning",
+        cooldown_minutes=0,
+    )
+    db_session.add(rule)
+    await db_session.commit()
+    result = await evaluate_rules(db_session)
+    assert result.opened == 1
+    event = (await db_session.execute(select(AlertEvent))).scalars().one()
+    assert event.state == "open" and event.value > 80
+    await _trace(db_session, "svc", 0)  # the sender is back
+    again = await evaluate_rules(db_session)
+    assert again.resolved == 1
+    await db_session.refresh(event)
+    assert event.state == "resolved"
+
+
+@pytest.mark.anyio
+async def test_a_project_that_never_sent_anything_is_not_silent(db_session):
+    db_session.add(
+        AlertRule(
+            name="ghost",
+            metric="ingest_silence_minutes",
+            comparator=">",
+            threshold=1.0,
+            window_minutes=60,
+            min_samples=1,
+            project_id="never",
+            severity="info",
+            cooldown_minutes=0,
+        )
+    )
+    await db_session.commit()
+    assert (await evaluate_rules(db_session)).opened == 0

dashboard/src/api.ts

@@ -26,6 +26,7 @@ import type {
   RunComparison,
   CaptureMode,
   PrivacyOverview,
+  IngestHealth,
 } from './types';
 
 const BASE_URL = '/api';
@@ -176,6 +177,11 @@ export const api = {
     return request(`/v1/privacy/projects/${encodeURIComponent(project)}/enforce`, { method: 'POST', body: JSON.stringify({ confirm: true }) });
   },
 
+  // Is telemetry arriving?
+  async getIngestHealth(): Promise<IngestHealth> {
+    return request('/v1/ingest/health');
+  },
+
   // Alerts
   async getAlertMetrics(): Promise<AlertMetric[]> {
     return request('/v1/alerts/metrics');

dashboard/src/components/IngestHealth.tsx

@@ -0,0 +1,40 @@
+import { useEffect, useState } from 'react';
+import { api } from '../api';
+import type { IngestHealth as Health } from '../types';
+import { Card, EmptyState } from './ui';
+import { silenceText, silenceTone } from '../helpers/ingest';
+import { fmtInt } from '../helpers/format';
+import { t } from '../i18n';
+
+const DOT: Record<string, string> = { ok: 'bg-ok', warn: 'bg-warn', bad: 'bg-bad', neutral: 'bg-faint' };
+
+/** Is telemetry still arriving? Fails silently: it is a hint, never a reason to show an error. */
+export default function IngestHealth() {
+  const [health, setHealth] = useState<Health | null>(null);
+
+  useEffect(() => {
+    let live = true;
+    const load = () => api.getIngestHealth().then((h) => { if (live) setHealth(h); }).catch(() => undefined);
+    load();
+    const id = setInterval(load, 30_000);
+    return () => { live = false; clearInterval(id); };
+  }, []);
+
+  if (!health) return null;
+  return (
+    <Card title={t('Recepción de telemetría')} subtitle={t('Cuándo llegó la última traza de cada proyecto')}>
+      {health.projects.length === 0 ? <EmptyState title={t('Aún no llega telemetría')} /> : (
+        <ul className="divide-y divide-line text-xs">
+          {health.projects.map((p) => (
+            <li key={p.project_id} className="flex flex-wrap items-center gap-3 py-2">
+              <span aria-hidden className={`size-2 rounded-full ${DOT[silenceTone(p.silence_minutes, p.traces_1h)]}`} />
+              <span className="font-mono text-ink">{p.project_id}</span>
+              <span className="text-dim">{t('hace {x}', { x: silenceText(p.silence_minutes) })}</span>
+              <span className="ml-auto text-faint">{t('{a} en 5 min · {b} en 1 h · {c} en 24 h', { a: fmtInt(p.traces_5m), b: fmtInt(p.traces_1h), c: fmtInt(p.traces_24h) })}</span>
+            </li>
+          ))}
+        </ul>
+      )}
+    </Card>
+  );
+}