From 5a23e4eda6f6ab710cfef34b1a0f8e24b10a9efa Mon Sep 17 00:00:00 2001 From: Yiorgis Gozadinos Date: Mon, 1 Jun 2026 10:27:31 +0300 Subject: [PATCH] Surface failed discovery sweeps in run-batch --- docs/ingester.md | 5 ++-- haiku_rag_slim/haiku/rag/ingester/app.py | 6 ++-- haiku_rag_slim/haiku/rag/ingester/cli.py | 10 +++++-- .../haiku/rag/ingester/pollers/manager.py | 11 ++++++-- tests/ingester/test_cli.py | 9 ++++++ tests/ingester/test_run_batch.py | 28 +++++++++++++++++++ 6 files changed, 60 insertions(+), 9 deletions(-) diff --git a/docs/ingester.md b/docs/ingester.md index ec44847c..891f0667 100644 --- a/docs/ingester.md +++ b/docs/ingester.md @@ -322,8 +322,9 @@ haiku-ingester run-batch --db rag.lancedb ``` Orphan deletion compares each source against `sync_state` in the queue DB, -so persist `ingester.db` between runs for deletions to be detected. It -exits non-zero if any job dead-letters. +so persist `ingester.db` between runs for deletions to be detected. It exits +non-zero if any job dead-letters or a source's discovery sweep does not +complete. ### The queue diff --git a/haiku_rag_slim/haiku/rag/ingester/app.py b/haiku_rag_slim/haiku/rag/ingester/app.py index a74b8287..7b64556f 100644 --- a/haiku_rag_slim/haiku/rag/ingester/app.py +++ b/haiku_rag_slim/haiku/rag/ingester/app.py @@ -23,10 +23,11 @@ logger = logging.getLogger(__name__) class BatchReport(BaseModel): """Outcome of a one-shot batch run: terminal job counts after the queue - drained.""" + drained, plus any sources whose discovery sweep did not complete.""" succeeded: int = 0 dead: int = 0 + failed_sweeps: list[str] = [] class IngesterApp: @@ -198,7 +199,7 @@ class IngesterApp: started_at = datetime.now(UTC) await self._pool.start() try: - await self._pollers.sweep_all() + failed_sweeps = await self._pollers.sweep_all() while True: counts = await self._jobs.counts_by_status() if not counts.get("queued") and not counts.get("claimed"): @@ -208,6 +209,7 @@ class IngesterApp: return BatchReport( succeeded=completed.get("succeeded", 0), dead=completed.get("dead", 0), + failed_sweeps=failed_sweeps, ) finally: await self._stop_pool() diff --git a/haiku_rag_slim/haiku/rag/ingester/cli.py b/haiku_rag_slim/haiku/rag/ingester/cli.py index 91430516..b4a1fff4 100644 --- a/haiku_rag_slim/haiku/rag/ingester/cli.py +++ b/haiku_rag_slim/haiku/rag/ingester/cli.py @@ -167,7 +167,8 @@ def run_batch( ) -> None: """Run one discover sweep across every configured source, drain the queue, then exit. New and changed resources are ingested, resources that vanished - from a source are deleted. Exits non-zero if any job dead-letters.""" + from a source are deleted. Exits non-zero if any job dead-letters or a + source's sweep does not complete.""" asyncio.run(_run_batch(get_config(), db)) @@ -176,5 +177,10 @@ async def _run_batch(app_config: AppConfig, db_path: Path | None) -> None: app = IngesterApp(config=app_config, db_path=db) report = await app.run_batch() typer.echo(f"Batch complete: {report.succeeded} succeeded, {report.dead} dead") - if report.dead: + if report.failed_sweeps: + typer.echo( + f"Sources that failed to sweep: {', '.join(report.failed_sweeps)}", + err=True, + ) + if report.dead or report.failed_sweeps: raise typer.Exit(1) diff --git a/haiku_rag_slim/haiku/rag/ingester/pollers/manager.py b/haiku_rag_slim/haiku/rag/ingester/pollers/manager.py index 62689eba..6339a5c8 100644 --- a/haiku_rag_slim/haiku/rag/ingester/pollers/manager.py +++ b/haiku_rag_slim/haiku/rag/ingester/pollers/manager.py @@ -77,12 +77,17 @@ class PollerManager: poller._stop.clear() self._tasks.append(asyncio.create_task(poller.run())) - async def sweep_all(self) -> None: + async def sweep_all(self) -> list[str]: """Run one discover() sweep on every poller, sequentially. Used by one-shot batch runs that drive discovery explicitly rather than - through the periodic loop.""" + through the periodic loop. Returns the source ids whose sweep did not + complete (discovery failed, circuit open, or pending work already + queued) so callers can treat a one-shot run as failed.""" + failed: list[str] = [] for poller in self._pollers: - await poller._sweep_once() + if not await poller._sweep_once(): + failed.append(poller.source_id) + return failed async def stop(self) -> None: for poller in self._pollers: diff --git a/tests/ingester/test_cli.py b/tests/ingester/test_cli.py index 06a67505..b03fd445 100644 --- a/tests/ingester/test_cli.py +++ b/tests/ingester/test_cli.py @@ -35,3 +35,12 @@ def test_run_batch_exits_nonzero_when_dead(monkeypatch): assert result.exit_code == 1 assert "2 dead" in result.output + + +def test_run_batch_exits_nonzero_when_sweep_fails(monkeypatch): + _fake_app(BatchReport(succeeded=2, dead=0, failed_sweeps=["docs"]), monkeypatch) + + result = runner.invoke(cli, ["run-batch", "--db", "x.lancedb"]) + + assert result.exit_code == 1 + assert "failed to sweep: docs" in result.output diff --git a/tests/ingester/test_run_batch.py b/tests/ingester/test_run_batch.py index f4126d3a..c646d5d6 100644 --- a/tests/ingester/test_run_batch.py +++ b/tests/ingester/test_run_batch.py @@ -189,6 +189,34 @@ async def test_run_batch_recovered_doc_is_not_counted_as_dead(tmp_path, use_clie assert second.succeeded == 1 +@pytest.mark.asyncio +async def test_run_batch_reports_failed_sweep( + tmp_path, use_client, monkeypatch, caplog +): + """A source whose discover() raises is reported in failed_sweeps so the + run can be treated as failed rather than a silent empty success.""" + (tmp_path / "a.md").write_text("hello") + use_client(_mock_client()) + + async def _failing_discover(self, **kwargs): + raise RuntimeError("discover blew up") + yield # unreachable; makes this an async generator + + monkeypatch.setattr( + "haiku.rag.ingester.sources.fs.FSSource.discover", _failing_discover + ) + + with caplog.at_level("ERROR", logger="haiku.rag.ingester.pollers.base"): + report = await IngesterApp( + config=_config(tmp_path), db_path=tmp_path / "db.lancedb" + ).run_batch() + + assert report.failed_sweeps == ["local"] + assert report.succeeded == 0 + assert report.dead == 0 + assert "discover() failed" in caplog.text + + @pytest.mark.asyncio async def test_run_batch_empty_source_returns_immediately(tmp_path, use_client): client = _mock_client()