fix(eval): acknowledge sync only after preceding writes complete 🤖🤖🤖

util/eval_pipeline/src/eval_pipeline/headless_backend.py

@@ -50,48 +50,62 @@ def _make_headless_app():
     from nooa.viewer import otlp_store
     from nooa.viewer.trace_routes import router as trace_router
 
-    _ingest_queue: asyncio.Queue[bytes | asyncio.Event] = asyncio.Queue()
+    _ingest_queue: asyncio.Queue[bytes | asyncio.Future[bool]] = asyncio.Queue()
     _write_executor = ThreadPoolExecutor(max_workers=1, thread_name_prefix="headless-writer")
 
     async def _ingest_worker() -> None:
+        """Write queued batches in order and resolve barriers after preceding writes finish."""
         loop = asyncio.get_running_loop()
+        write_failed = False
+
+        def finish_barrier(barrier: asyncio.Future[bool]) -> None:
+            """Acknowledge a consumed barrier without reviving a cancelled sync request."""
+            # Timed-out or disconnected callers may have cancelled their future.
+            if not barrier.done():
+                barrier.set_result(not write_failed)
+            _ingest_queue.task_done()
+
         while True:
             item = await _ingest_queue.get()
-            # Sentinel event from /v1/sync — everything before it is written.
-            if isinstance(item, asyncio.Event):
-                item.set()
-                _ingest_queue.task_done()
+            if isinstance(item, asyncio.Future):
+                finish_barrier(item)
                 continue
             batch: list[bytes] = [item]
+            barrier: asyncio.Future[bool] | None = None
             while len(batch) < _INGEST_MAX_BATCH:
                 try:
                     next_item = _ingest_queue.get_nowait()
-                    if isinstance(next_item, asyncio.Event):
-                        next_item.set()
-                        _ingest_queue.task_done()
-                        continue
+                    if isinstance(next_item, asyncio.Future):
+                        barrier = next_item
+                        break  # Flush the preceding batch before acknowledging sync.
                     batch.append(next_item)
                 except asyncio.QueueEmpty:
                     break
             try:
-                await loop.run_in_executor(
+                results = await loop.run_in_executor(
                     _write_executor, otlp_store.ingest_batch_write_bytes, batch
                 )
+                if len(results) != len(batch):
+                    write_failed = True
             except Exception:
+                # A dropped batch invalidates subsequent sync guarantees for this backend.
+                write_failed = True
                 log.exception("headless ingest_worker: failed to write batch of %d", len(batch))
             finally:
                 for _ in batch:
                     _ingest_queue.task_done()
+            if barrier is not None:
+                finish_barrier(barrier)
 
     @asynccontextmanager
     async def lifespan(app: fastapi.FastAPI):
+        """Start the writer and drain queued or in-flight work before shutting it down."""
         otlp_store.init_db()
         worker = asyncio.create_task(_ingest_worker())
         try:
             yield
         finally:
-            if not _ingest_queue.empty():
-                await _ingest_queue.join()
+            await _ingest_queue.join()
             worker.cancel()
             try:
                 await worker
@@ -176,15 +190,17 @@ async def journal_blocks(request: _FastAPIRequest):
     async def sync():
         """Wait until all spans queued before this call are written.
 
-        Inserts a sentinel Event into the queue.  The ingest worker
-        processes items in order; when it reaches the sentinel it sets
-        the event, guaranteeing everything enqueued before it has been
-        written to SQLite.  Unlike Queue.join(), this is not affected
-        by new items arriving from other concurrent tasks.
+        The worker resolves a queued future after writing the preceding batch.
+        Unlike Queue.join(), this does not wait for items arriving later.
+        A prior write failure returns an error because the missing spans cannot
+        be recovered by another sync request.
         """
-        event = asyncio.Event()
-        await _ingest_queue.put(event)
-        await asyncio.wait_for(event.wait(), timeout=30)
+        barrier: asyncio.Future[bool] = asyncio.get_running_loop().create_future()
+        await _ingest_queue.put(barrier)
+        if not await asyncio.wait_for(barrier, timeout=30):
+            return _JSONResponse(
+                status_code=500, content={"error": "One or more trace batches failed to persist"}
+            )
         return _JSONResponse(content={"synced": True})
 
     @app.get("/health")

util/eval_pipeline/tests/test_headless_sync.py

@@ -0,0 +1,133 @@
+# SPDX-FileCopyrightText: Copyright (c) 2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+"""Deterministic persistence barriers without a server or SQLite connection."""
+
+import asyncio
+import threading
+from contextlib import asynccontextmanager
+
+import httpx
+import pytest
+
+from eval_pipeline.headless_backend import _make_headless_app
+
+pytestmark = pytest.mark.asyncio
+
+
+@asynccontextmanager
+async def _client(monkeypatch, writer):
+    """Exercise the app lifecycle and HTTP routes using a controlled in-process writer."""
+    monkeypatch.setattr("nooa.viewer.otlp_store.init_db", lambda: None)
+    monkeypatch.setattr("nooa.viewer.otlp_store.ingest_batch_write_bytes", writer)
+    app = _make_headless_app()
+    async with (
+        app.router.lifespan_context(app),
+        httpx.AsyncClient(transport=httpx.ASGITransport(app=app), base_url="http://test") as client,
+    ):
+        yield client
+
+
+def _observe_barrier(monkeypatch):
+    """Expose an event when a sync barrier enters the queue, without timing sleeps."""
+    queued = asyncio.Event()
+
+    class ObservedQueue(asyncio.Queue):
+        def put_nowait(self, item):
+            """Signal only after the barrier has actually been enqueued."""
+            super().put_nowait(item)
+            if not isinstance(item, bytes):
+                queued.set()
+
+    monkeypatch.setattr("eval_pipeline.headless_backend.asyncio.Queue", ObservedQueue)
+    return queued
+
+
+@pytest.mark.parametrize("batch_limit", [1, 32])
+async def test_sync_waits_for_preceding_batch_but_not_later_arrivals(monkeypatch, batch_limit):
+    """Sync waits for earlier payloads but excludes writes enqueued after its barrier."""
+    monkeypatch.setattr("eval_pipeline.headless_backend._INGEST_MAX_BATCH", batch_limit)
+    started = {name: threading.Event() for name in (b"first", b"second", b"later")}
+    release = {name: threading.Event() for name in started}
+    persisted = []
+    barrier_queued = _observe_barrier(monkeypatch)
+
+    def writer(batch):
+        """Block each named batch until the test explicitly allows it to persist."""
+        started[batch[0]].set()
+        assert release[batch[0]].wait(10), "test did not release writer"
+        persisted.extend(batch)
+        return [{} for _ in batch]
+
+    async with _client(monkeypatch, writer) as client:
+        task = None
+        try:
+            await client.post("/v1/traces", content=b"first")
+            assert await asyncio.to_thread(started[b"first"].wait, 5)
+            await client.post("/v1/traces", content=b"second")
+            task = asyncio.create_task(client.post("/v1/sync"))
+            await asyncio.wait_for(barrier_queued.wait(), 5)
+            await client.post("/v1/traces", content=b"later")
+            release[b"first"].set()
+            assert await asyncio.to_thread(started[b"second"].wait, 5)
+            assert not task.done(), "sync acknowledged an unwritten preceding batch"
+            release[b"second"].set()
+            response = await asyncio.wait_for(task, 5)
+            assert response.status_code == 200
+            assert response.json() == {"synced": True}
+            assert persisted == [b"first", b"second"]
+        finally:
+            for event in release.values():
+                event.set()
+            if task is not None:
+                await asyncio.gather(task, return_exceptions=True)
+
+
+@pytest.mark.parametrize("failure", ["exception", "skipped_payload"])
+async def test_sync_reports_write_failure(monkeypatch, failure):
+    """Both raised and silently skipped writes invalidate subsequent sync acknowledgements."""
+
+    def writer(batch):
+        """Emulate the store's two failure modes without a live database."""
+        if failure == "exception":
+            raise OSError("disk full")
+        return []  # The store omits results for payloads that fail during ingestion.
+
+    async with _client(monkeypatch, writer) as client:
+        await client.post("/v1/traces", content=b"payload")
+        for _ in range(2):
+            response = await client.post("/v1/sync")
+            assert response.status_code == 500
+            assert "error" in response.json()
+
+
+async def test_cancelled_sync_does_not_stop_worker(monkeypatch):
+    """Cancelling one caller leaves the worker able to persist data and handle later syncs."""
+    started, release = threading.Event(), threading.Event()
+    queued = _observe_barrier(monkeypatch)
+    persisted = []
+
+    def writer(batch):
+        """Hold persistence until the first sync caller has been cancelled."""
+        started.set()
+        assert release.wait(10), "test did not release writer"
+        persisted.extend(batch)
+        return [{} for _ in batch]
+
+    async with _client(monkeypatch, writer) as client:
+        task = None
+        try:
+            await client.post("/v1/traces", content=b"payload")
+            assert await asyncio.to_thread(started.wait, 5)
+            task = asyncio.create_task(client.post("/v1/sync"))
+            await asyncio.wait_for(queued.wait(), 5)
+            task.cancel()
+            with pytest.raises(asyncio.CancelledError):
+                await task
+            release.set()
+            response = await asyncio.wait_for(client.post("/v1/sync"), 5)
+            assert response.status_code == 200
+            assert persisted == [b"payload"]
+        finally:
+            release.set()
+            if task is not None:
+                await asyncio.gather(task, return_exceptions=True)