diff --git a/haiku_rag_slim/haiku/rag/ingester/app.py b/haiku_rag_slim/haiku/rag/ingester/app.py index 70c5a673..2e0e5d3d 100644 --- a/haiku_rag_slim/haiku/rag/ingester/app.py +++ b/haiku_rag_slim/haiku/rag/ingester/app.py @@ -205,6 +205,15 @@ class IngesterApp: counts = await self._jobs.counts_by_status() if not counts.get("queued") and not counts.get("claimed"): break + if self._pool.live_workers == 0: + outstanding = counts.get("queued", 0) + counts.get("claimed", 0) + logger.error( + "All workers have died with %d outstanding job(s) " + "— aborting batch; stranded jobs will be reaped " + "on next start", + outstanding, + ) + break await asyncio.sleep(0.1) completed = await self._jobs.counts_by_status_since(started_at) return BatchReport( diff --git a/tests/ingester/test_run_batch.py b/tests/ingester/test_run_batch.py index 8edd461d..0491916e 100644 --- a/tests/ingester/test_run_batch.py +++ b/tests/ingester/test_run_batch.py @@ -235,6 +235,39 @@ async def test_run_batch_empty_source_returns_immediately(tmp_path, use_client): client.create_document_from_source.assert_not_awaited() +@pytest.mark.asyncio +async def test_run_batch_aborts_when_all_workers_die( + tmp_path, use_client, monkeypatch, caplog +): + """If all workers crash with outstanding jobs, run_batch should break + out of the drain loop instead of spinning forever.""" + (tmp_path / "a.md").write_text("hello") + + client = _mock_client() + use_client(client) + config = _config(tmp_path) + + config.ingester.workers.worker_count = 1 + + # Patch _process to raise an unhandled exception, simulating a hard + # worker crash. _process only catches CancelledError, PermanentError, + # and TransientError — anything else propagates and kills the task. + monkeypatch.setattr( + WorkerPool, + "_process", + AsyncMock(side_effect=Exception("worker crash")), + ) + + with caplog.at_level("ERROR", logger="haiku.rag.ingester.app"): + report = await asyncio.wait_for( + IngesterApp(config=config, db_path=tmp_path / "db.lancedb").run_batch(), + timeout=10.0, + ) + + assert report is not None + assert "All workers have died" in caplog.text + + async def _wait_until(predicate, *, timeout: float = 5.0): deadline = asyncio.get_running_loop().time() + timeout while not predicate():