fix(ingestion): EPCIS/FDA export fails loudly instead of silent empty/truncated (SEL-506, SEL-508)

.semgrep/tenant-scope-allowlist.yml

@@ -144,7 +144,7 @@ baseline:
     scope: _query_tenant_events
     rule: sql-without-tenant-predicate
     target: fsma.cte_events,fsma.cte_kdes
-    qhash: "87bb48c02e1970fb"
+    qhash: "c9b46a2dd14590f8"  # pragma: allowlist secret
     reason: "WHERE list seeded with e.tenant_id = :tenant_id; verified scoped at runtime (cte_kdes joined on the scoped event)."
   - file: services/ingestion/app/incident_router.py
     scope: list_incidents

services/ingestion/app/epcis_export.py

@@ -13,7 +13,7 @@
 from datetime import datetime, timezone
 from typing import Optional
 
-from fastapi import APIRouter, Depends, Query
+from fastapi import APIRouter, Depends, HTTPException, Query
 from fastapi.responses import JSONResponse, StreamingResponse
 from pydantic import BaseModel, Field
 from shared.database import get_db_safe
@@ -32,10 +32,26 @@
 )
 
 
+# Hard cap on events buffered into one synchronous export. Exceeding it
+# returns HTTP 413 rather than silently dropping the oldest events (SEL-506),
+# matching the compliance FDA spreadsheet export (services/compliance
+# routes.py ``_MAX_EXPORT_EVENTS``).
+_MAX_EXPORT_EVENTS = 50_000
+
+
+class ExportStorageUnavailable(RuntimeError):
+    """The event store could not be read; never report this as "no data"."""
+
+
 def _query_tenant_events(
     tenant_id: str, lot_code: str | None, date_from: str | None, date_to: str | None
 ) -> list[dict]:
-    """Query real CTE events from fsma.cte_events for export."""
+    """Query real CTE events from fsma.cte_events for export.
+
+    Returns at most ``_MAX_EXPORT_EVENTS + 1`` rows so callers can detect
+    an over-cap export. Raises :class:`ExportStorageUnavailable` on any
+    database error instead of returning an empty list (SEL-508).
+    """
     db = get_db_safe()
     try:
         where = ["e.tenant_id = :tenant_id"]
@@ -80,18 +96,40 @@ def _query_tenant_events(
                 LEFT JOIN fsma.cte_kdes k_temp ON k_temp.cte_event_id = e.id AND k_temp.kde_key = 'temperature_celsius'
                 WHERE {" AND ".join(where)}
                 ORDER BY e.event_timestamp DESC
-                LIMIT 5000
+                LIMIT :row_limit
             """),
-            params,
+            {**params, "row_limit": _MAX_EXPORT_EVENTS + 1},
         ).fetchall()
         return [dict(row._mapping) for row in rows]
     except Exception as exc:
-        logger.warning("db_query_failed_for_export", extra={"error": str(exc)})
-        return []
+        logger.error("db_query_failed_for_export", extra={"error": str(exc)})
+        raise ExportStorageUnavailable(str(exc)) from exc
     finally:
         db.close()
 
 
+def _load_export_rows(tenant_id: str, request: "ExportRequest") -> list[dict]:
+    """Load rows for an export, failing loudly instead of returning partial data.
+
+    503 when storage is unavailable (never "No data yet"), 413 when the
+    result would exceed ``_MAX_EXPORT_EVENTS`` (never a silent truncation).
+    """
+    try:
+        rows = _query_tenant_events(tenant_id, request.lot_code, request.date_from, request.date_to)
+    except ExportStorageUnavailable as exc:
+        raise HTTPException(status_code=503, detail="Event storage unavailable; export not generated") from exc
+    if len(rows) > _MAX_EXPORT_EVENTS:
+        logger.warning(
+            "export_exceeded_hard_cap",
+            extra={"tenant_id": tenant_id, "cap": _MAX_EXPORT_EVENTS},
+        )
+        raise HTTPException(
+            status_code=413,
+            detail=(f"Export exceeds the {_MAX_EXPORT_EVENTS:,}-event hard cap; narrow your date range or lot code"),
+        )
+    return rows
+
+
 _CTE_TO_BIZSTEP = {
     "receiving": "urn:epcglobal:cbv:bizstep:receiving",
     "shipping": "urn:epcglobal:cbv:bizstep:shipping",
@@ -186,7 +224,7 @@ async def export_epcis(
     now = datetime.now(timezone.utc)
     tenant_id = ctx.tenant_id
 
-    rows = _query_tenant_events(tenant_id, request.lot_code, request.date_from, request.date_to)
+    rows = _load_export_rows(tenant_id, request)
     data_source = "tenant" if rows else "empty"
 
     event_list = []
@@ -284,7 +322,7 @@ async def export_fda(
     now = datetime.now(timezone.utc)
     tenant_id = ctx.tenant_id
 
-    rows = _query_tenant_events(tenant_id, request.lot_code, request.date_from, request.date_to)
+    rows = _load_export_rows(tenant_id, request)
     data_source = "tenant" if rows else "empty"
 
     if rows:

services/ingestion/tests/test_epcis_export_1342.py

@@ -103,11 +103,22 @@ def test_no_rows_returns_empty_list(self, monkeypatch: pytest.MonkeyPatch) -> No
 
         assert _query_tenant_events("t", None, None, None) == []
 
-    def test_swallow_exception_returns_empty_list(self, monkeypatch: pytest.MonkeyPatch) -> None:
-        session = _FakeSession(execute_queue=[RuntimeError()])
+    def test_db_error_raises_instead_of_empty_list(self, monkeypatch: pytest.MonkeyPatch) -> None:
+        # SEL-508: a storage failure must never look like an empty tenant.
+        session = _FakeSession(execute_queue=[RuntimeError("db down")])
         monkeypatch.setattr(epcis_export, "get_db_safe", lambda: session)
-        result = _query_tenant_events("t", None, None, None)
-        assert result == []
+        with pytest.raises(epcis_export.ExportStorageUnavailable):
+            _query_tenant_events("t", None, None, None)
+        assert session.closed
+
+    def test_query_is_bounded_by_cap_plus_one_not_5000(self, monkeypatch: pytest.MonkeyPatch) -> None:
+        # SEL-506: the old LIMIT 5000 silently dropped the oldest events.
+        session = _FakeSession(execute_queue=[[]])
+        monkeypatch.setattr(epcis_export, "get_db_safe", lambda: session)
+        _query_tenant_events("t", None, None, None)
+        sql, params = session.calls[0][1]
+        assert "LIMIT 5000" not in sql
+        assert params["row_limit"] == epcis_export._MAX_EXPORT_EVENTS + 1
 
 
 # ---------------------------------------------------------------------------
@@ -260,3 +271,42 @@ def test_lists_formats(self, client: TestClient) -> None:
             assert "name" in fmt
             assert "content_type" in fmt
             assert "retailers" in fmt
+
+
+def _raise_storage(*_a: Any, **_k: Any) -> list[dict]:
+    raise epcis_export.ExportStorageUnavailable("db down")
+
+
+class TestExportFailsLoudly:
+    """SEL-506 / SEL-508: partial or failed reads are never reported as complete."""
+
+    @pytest.mark.parametrize("path", ["/api/v1/export/epcis", "/api/v1/export/fda"])
+    def test_storage_error_returns_503_not_no_data_yet(
+        self, client: TestClient, monkeypatch: pytest.MonkeyPatch, path: str
+    ) -> None:
+        session = _FakeSession(execute_queue=[RuntimeError("db down")])
+        monkeypatch.setattr(epcis_export, "get_db_safe", lambda: session)
+        response = client.post(path, json={"tenant_id": "t"})
+        assert response.status_code == 503
+        assert "No data yet" not in response.text
+        assert response.headers.get("X-RegEngine-Data-Source") != "empty"
+
+    def test_5001_events_export_5001(self, client: TestClient, monkeypatch: pytest.MonkeyPatch) -> None:
+        rows = [_event_row() for _ in range(5001)]
+        session = _FakeSession(execute_queue=[rows])
+        monkeypatch.setattr(epcis_export, "get_db_safe", lambda: session)
+        response = client.post("/api/v1/export/epcis", json={"tenant_id": "t"})
+        assert response.status_code == 200
+        assert len(response.json()["epcisBody"]["eventList"]) == 5001
+        assert response.headers["X-RegEngine-Events-Count"] == "5001"
+
+    @pytest.mark.parametrize("path", ["/api/v1/export/epcis", "/api/v1/export/fda"])
+    def test_over_cap_returns_413_not_truncated(
+        self, client: TestClient, monkeypatch: pytest.MonkeyPatch, path: str
+    ) -> None:
+        monkeypatch.setattr(epcis_export, "_MAX_EXPORT_EVENTS", 3)
+        session = _FakeSession(execute_queue=[[_event_row() for _ in range(4)]])
+        monkeypatch.setattr(epcis_export, "get_db_safe", lambda: session)
+        response = client.post(path, json={"tenant_id": "t"})
+        assert response.status_code == 413
+        assert session.calls[0][1][1]["row_limit"] == 4