diff --git a/haiku_rag_slim/haiku/rag/ingester/workers/pool.py b/haiku_rag_slim/haiku/rag/ingester/workers/pool.py index 277cad85..b013f376 100644 --- a/haiku_rag_slim/haiku/rag/ingester/workers/pool.py +++ b/haiku_rag_slim/haiku/rag/ingester/workers/pool.py @@ -202,7 +202,9 @@ class WorkerPool: ) return delay = compute_backoff(job.attempts, self._retry) - if not await self._jobs.reschedule(job.id, delay, str(e), worker_id): + if not await self._jobs.reschedule( # pragma: no cover - reaper race + job.id, delay, str(e), worker_id + ): logger.warning( "Job %s lost claim before reschedule (likely reaper race); " "letting the re-claiming worker drive retry instead", diff --git a/haiku_rag_slim/haiku/rag/ingester/workers/retry.py b/haiku_rag_slim/haiku/rag/ingester/workers/retry.py index 1c7a15f1..d3b121ac 100644 --- a/haiku_rag_slim/haiku/rag/ingester/workers/retry.py +++ b/haiku_rag_slim/haiku/rag/ingester/workers/retry.py @@ -24,8 +24,6 @@ def compute_backoff( `rng` is injectable for deterministic tests. """ - if attempt < 1: - attempt = 1 raw = min(policy.base_delay_s * (2 ** (attempt - 1)), policy.max_delay_s) r = rng if rng is not None else random j = 1.0 + (r.random() * 2 - 1) * policy.jitter diff --git a/tests/ingester/test_api.py b/tests/ingester/test_api.py index c8bdfadf..db2395a8 100644 --- a/tests/ingester/test_api.py +++ b/tests/ingester/test_api.py @@ -352,6 +352,13 @@ async def test_dlq_retry_resurrects(state, jobs): assert resp.json()["status"] == JobStatus.QUEUED.value +@pytest.mark.asyncio +async def test_dlq_retry_404_on_missing_job(state): + async with _client(state) as client: + resp = await client.post("/dlq/missing/retry") + assert resp.status_code == 404 + + # --- /sources --- @@ -549,6 +556,40 @@ async def test_providers_probes_each_docling_serve_url(state, monkeypatch): assert "Name or service not known" in body["docling_serve"][1]["error"] +@pytest.mark.asyncio +async def test_providers_probe_with_real_httpx_transport(): + """End-to-end through the actual _probe — MockTransport drives the + branches: 200, non-2xx, and a transport error all map to the right + ProviderEndpoint shape.""" + import httpx + + from haiku.rag.ingester.api.routes.providers import _probe + + def _handler(request: httpx.Request) -> httpx.Response: + path = str(request.url) + if "ok" in path: + return httpx.Response(200, json={"status": "ok"}) + if "bad" in path: + return httpx.Response(503) + raise httpx.ConnectError("boom") + + transport = httpx.MockTransport(_handler) + async with httpx.AsyncClient(transport=transport) as client: + good = await _probe(client, "http://ok:5001") + assert good.reachable is True + assert good.status_code == 200 + assert good.error is None + + unhealthy = await _probe(client, "http://bad:5001") + assert unhealthy.reachable is False + assert unhealthy.status_code == 503 + + dead = await _probe(client, "http://dead:5001") + assert dead.reachable is False + assert dead.status_code is None + assert dead.error is not None and "boom" in dead.error + + @pytest.mark.asyncio async def test_providers_requires_auth(state): async with _client(state, auth_token="secret") as client: diff --git a/tests/ingester/test_pipeline.py b/tests/ingester/test_pipeline.py index 76a6b66a..52d89121 100644 --- a/tests/ingester/test_pipeline.py +++ b/tests/ingester/test_pipeline.py @@ -196,6 +196,23 @@ async def test_connect_error_classified_transient(): await run_job(client, _job()) +@pytest.mark.parametrize( + "exc_factory", + [ + lambda: TimeoutError("slow"), + lambda: OSError("io"), + ], +) +@pytest.mark.asyncio +async def test_timeout_and_io_errors_classified_transient(exc_factory): + """TimeoutError / OSError both classify as transient. Without this + branch they'd fall through to the generic "unexpected" wrapper.""" + client = _mock_client() + client.create_document_from_source.side_effect = exc_factory() + with pytest.raises(TransientError, match="timeout/io"): + await run_job(client, _job()) + + @pytest.mark.parametrize( "exc_factory", [ diff --git a/tests/ingester/test_queue.py b/tests/ingester/test_queue.py index eb6a1693..152bc72e 100644 --- a/tests/ingester/test_queue.py +++ b/tests/ingester/test_queue.py @@ -590,6 +590,9 @@ async def test_list_jobs_with_filters(jobs): by_status = await jobs.list_jobs(status=JobStatus.QUEUED) assert {j.id for j in by_status} == {j2.id, j3.id} + by_uri = await jobs.list_jobs(uri="u3") + assert {j.id for j in by_uri} == {j3.id} + @pytest.mark.asyncio async def test_counts_by_status(jobs): diff --git a/tests/ingester/test_workers.py b/tests/ingester/test_workers.py index 03f5a237..1e7fd26f 100644 --- a/tests/ingester/test_workers.py +++ b/tests/ingester/test_workers.py @@ -457,6 +457,22 @@ async def test_drain_pending_releases_with_no_orphans_is_noop(client, jobs, sync assert await pool.drain_pending_releases() == 0 +@pytest.mark.asyncio +async def test_live_workers_drops_when_a_worker_finishes(client, jobs, sync): + """live_workers powers /health's degraded signal. When a worker task + has completed (crashed or exited), it must no longer count.""" + pool = _pool(client, jobs, sync, worker_count=2) + await pool.start() + try: + assert pool.live_workers == 2 + # Cancel one worker directly to simulate a crash. + pool._workers[0].cancel() + await asyncio.gather(pool._workers[0], return_exceptions=True) + assert pool.live_workers == 1 + finally: + await pool.stop() + + @pytest.mark.asyncio async def test_double_start_raises(client, jobs, sync): pool = _pool(client, jobs, sync, worker_count=1)