diff --git a/CHANGELOG.md b/CHANGELOG.md index 4d97f041..d71cd210 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -3,26 +3,22 @@ ### Added -- New `haiku-ingester` service for continuous document ingestion: persistent SQLite job queue, async worker pool with retries and a dead-letter queue, FS/HTTP/S3/WebDAV source adapters with per-source circuit breakers, and a FastAPI control plane on `127.0.0.1:8765` exposing `/health`, `/sources`, `/jobs`, `/dlq`, `/stats`, and a browser dashboard at `/`. Configured under `ingester:` in `haiku.rag.yaml`. Shipped behind the `[ingester]` extra, with Logfire spans (`ingester.poller.sweep` and `ingester.poller.watch_event` → `ingester.job` → `document.{fetch,convert,chunk,embed,store}`, plus `document.convert_slice` per slice when splitting) for traceable ingestion. See [docs/ingester.md](docs/ingester.md). -- Browser dashboard at `GET /` on the ingester's control plane. Single self-contained HTML page (no CDN, no external assets — works offline). Polls every 3s and renders: queue chips (queued / claimed / succeeded / dead), per-source state (last polled, circuit-breaker, "queue busy" badge when sweeps are skipped), rolling throughput (5m / 30m / 1h succeeded), worker occupancy, oldest queued age, per-source backlog and DLQ counts, currently-claimed jobs with cancel, recent failures with retry, and the last-completed feed. Dashboard route is unauthenticated; the JS attaches the bearer token to its own JSON fetches on a `401` (token stored in `localStorage`). -- `GET /stats` endpoint on the ingester control plane: rolling counts of succeeded jobs (last 5m / 30m / 1h), worker occupancy (`busy` / `total`), oldest queued job age in seconds, and per-source breakdowns of DLQ size and queue depth. One SQL aggregation per field; cheap to poll. -- `processing.split_pages` (default `0`): split large PDFs into N-page slices, convert each independently through docling-local or docling-serve, and merge with `DoclingDocument.concatenate()`. Bounds peak working set on memory-hungry docs and lets multiple docling-serve replicas parallelize per-document. `0` disables (single-pass conversion). +- `haiku-ingester` service for continuous document ingestion. Persistent SQLite job queue, async worker pool with retries and dead-letter queue, FS/HTTP/S3/WebDAV source adapters, per-source and pool-wide circuit breakers, and a FastAPI control plane on `127.0.0.1:8765` exposing `/health`, `/sources`, `/jobs`, `/dlq`, `/stats`, and a browser dashboard at `/`. Configured under `ingester:` in `haiku.rag.yaml`; shipped behind the `[ingester]` extra. See [docs/ingester.md](docs/ingester.md). +- `processing.split_pages` (default `0`): split PDFs into N-page slices, convert each, merge via `DoclingDocument.concatenate()`. `0` disables. +- Logfire spans for ingestion: `ingester.poller.sweep`, `ingester.poller.watch_event`, `ingester.job`, `document.{fetch,convert,chunk,embed,store}`, `document.convert_slice`. ### Removed -- File monitor (`haiku.rag.monitor` module, `MonitorConfig`, `S3MonitorEntry`, `AppConfig.monitor`). The `--monitor` flag on `haiku-rag serve` is gone — continuous ingestion now lives in `haiku-ingester serve`. Migrate `monitor.directories` to `ingester.sources[type=fs]` and `monitor.s3` to `ingester.sources[type=s3]`; the `delete_orphans` / `ignore_patterns` / `include_patterns` keys keep their meaning on the per-source entry. +- File monitor (`haiku.rag.monitor`, `MonitorConfig`, `S3MonitorEntry`, `AppConfig.monitor`, `haiku-rag serve --monitor`). Migrate `monitor.directories` to `ingester.sources[type=fs]` and `monitor.s3` to `ingester.sources[type=s3]`. ### Changed -- `haiku-rag serve` renamed to `haiku-rag mcp` (only MCP is left). `--mcp-port` renamed to `--port`. Update any `claude_desktop_config.json` from `["serve", "--mcp", "--stdio"]` to `["mcp", "--stdio"]`. -- `document.metadata` now uses source-agnostic keys: `source_revision` (was `etag` — S3-only and never populated for FS, so periodic sweeps re-ingested every file) and `content_type` (was `contentType`, snake_case for consistency). The v0.50.0 startup migration rewrites existing documents and compacts the `documents` table at the end so per-row UPDATE tombstones don't keep the table at roughly doubled size for the default 24-hour vacuum retention window (measured `12 GB → 25 GB` mid-migration on a 1000-doc PDF corpus, reclaimed back to `12 GB` by the in-migration compaction). All four source adapters (FS, HTTP, S3, WebDAV) now write their native revision (mtime_ns, ETag, etc.) under the same key, fixing the regression where FS sources never short-circuited on unchanged files. -- Ingester pollers skip their periodic sweep when the source already has queued or claimed jobs in the queue — saves the listing round-trip (`PROPFIND` / `S3 LIST` / FS walk) when work is backed up. FS push events from `watchfiles` keep flowing during skipped sweeps. Visible in Logfire as `ingester.poller.sweep` spans with `skipped=true reason=pending_work`. -- Ingester now drains in-flight jobs on `SIGINT` / `SIGTERM` up to `workers.shutdown_grace_s` (default 60s) before cancelling. Cancelled jobs release their claim back to `queued` (and decrement `attempts` since a cancel isn't a failure) so the next process picks them up immediately instead of waiting on the reaper's `claim_timeout_s`. Bonus: the pipeline no longer wraps `KeyboardInterrupt` / `SystemExit` / `CancelledError` as `TransientError` — those now propagate as intended. -- `providers.docling_serve.base_url` now accepts a list. Jobs round-robin across the entries with each job's submit/poll/result pinned to one instance (task IDs are instance-local). The counter is per-process; for cross-process load balancing or failover, put an LB in front and pass a single URL here. -- `GET /health` now returns `workers_alive` and `pollers_alive` (counts of tasks still running) alongside the configured `worker_count` and `poller_count`. `status` is `"ok"` when both shortfalls are zero and `"degraded"` when a worker or poller has died — a DB-only liveness probe couldn't distinguish "process alive but doing nothing" from "process alive and healthy", and uptime monitors can now alert on the degraded state directly. -- FS source rejects any URI whose resolved path escapes the configured `root` directory. `supports()` returns `False`, `head()` returns `None`, and `fetch()` raises `UnsupportedSourceError`. `discover()` walks with `os.walk(followlinks=False)` and skips file-level symlinks, so a symlink under the watched root pointing at `/etc/passwd` (or any path outside the root) can't be ingested. Defense-in-depth for deployments where the watched directory is multi-writer. -- Logfire spans emitted by haiku.rag now report `instrumentation_scope.name = "haiku.rag"` (was the SDK default `logfire`). Cross-library instrumentations (pydantic-ai, docling-serve, etc.) keep their own scopes. Downstream consumers filtering OTel traces by source library can now match on `scope.name = haiku.rag` instead of catching everything the SDK exports. -- Drop `list_documents` and `get_document` from the default RAG skill's tool set; the skill now exposes only `search` and `cite`. Both tools dumped unbounded content into the agent's context (full document lists, full document bodies) and `get_document` returned no chunk_ids so its output was structurally uncitable. The analysis skill already covers these uses programmatically — `await list_documents()` and `Path('/documents/{id}/content.txt').read_text()` inside `execute_code`. The tool branches remain in `create_skill_tools` and the `skill_generator` `AVAILABLE_TOOLS` set so users can still opt in when building custom skills. +- `haiku-rag serve` renamed to `haiku-rag mcp`. `--mcp-port` renamed to `--port`. +- `document.metadata` keys renamed: `etag` → `source_revision`, `contentType` → `content_type`. v0.50.0 startup migration rewrites existing documents and compacts the `documents` table at the end (12 GB → 25 GB mid-migration on a 1000-doc PDF corpus, reclaimed back to 12 GB). +- `providers.docling_serve.base_url` now accepts a list. Jobs round-robin across the entries; each job's submit/poll/result pinned to one instance. +- `DoclingServeClient.submit_and_poll` and `submit_and_poll_zip` no longer wrap `httpx` errors into `ValueError`. `httpx.ConnectError`, `HTTPStatusError`, `TimeoutException`, etc. propagate with their type intact. +- Logfire spans report `instrumentation_scope.name = "haiku.rag"` (was the SDK default `logfire`). +- Default RAG skill exposes only `search` and `cite`. `list_documents` and `get_document` are still available in `create_skill_tools` and `skill_generator`'s `AVAILABLE_TOOLS` for custom skills. ## [0.48.1] - 2026-05-21 diff --git a/examples/docker/haiku.rag.yaml.example b/examples/docker/haiku.rag.yaml.example index 839b4c42..dc1f43e6 100644 --- a/examples/docker/haiku.rag.yaml.example +++ b/examples/docker/haiku.rag.yaml.example @@ -26,11 +26,6 @@ ingester: processing: converter: docling-serve chunker: docling-serve - chunk_size: 256 - chunker_type: hybrid - chunking_tokenizer: "Qwen/Qwen3-Embedding-0.6B" - chunking_merge_peers: true - chunking_use_markdown_tables: false providers: docling_serve: diff --git a/haiku_rag_slim/haiku/rag/ingester/api/routes/health.py b/haiku_rag_slim/haiku/rag/ingester/api/routes/health.py index e6f1fd79..0b63ee70 100644 --- a/haiku_rag_slim/haiku/rag/ingester/api/routes/health.py +++ b/haiku_rag_slim/haiku/rag/ingester/api/routes/health.py @@ -24,7 +24,13 @@ async def health(state: APIState = Depends(get_state)) -> HealthResponse: workers_alive = state.pool.live_workers if state.pool is not None else 0 poller_count = len(state.pollers.pollers) if state.pollers is not None else 0 pollers_alive = state.pollers.live_pollers if state.pollers is not None else 0 - degraded = workers_alive < worker_count or pollers_alive < poller_count + breaker_open = state.pool.breaker_open if state.pool is not None else False + breaker_failures = ( + state.pool.breaker_consecutive_failures if state.pool is not None else 0 + ) + degraded = ( + workers_alive < worker_count or pollers_alive < poller_count or breaker_open + ) return HealthResponse( status="degraded" if degraded else "ok", queue_counts=counts, @@ -32,4 +38,6 @@ async def health(state: APIState = Depends(get_state)) -> HealthResponse: poller_count=poller_count, workers_alive=workers_alive, pollers_alive=pollers_alive, + worker_breaker_open=breaker_open, + worker_breaker_consecutive_failures=breaker_failures, ) diff --git a/haiku_rag_slim/haiku/rag/ingester/api/schemas.py b/haiku_rag_slim/haiku/rag/ingester/api/schemas.py index 55206c2f..cc421fd1 100644 --- a/haiku_rag_slim/haiku/rag/ingester/api/schemas.py +++ b/haiku_rag_slim/haiku/rag/ingester/api/schemas.py @@ -14,6 +14,10 @@ class HealthResponse(BaseModel): # uptime monitors can alert without needing to do the math themselves. workers_alive: int pollers_alive: int + # Pool-wide breaker: open when N consecutive transient job failures + # have paused worker claims. status="degraded" while open. + worker_breaker_open: bool = False + worker_breaker_consecutive_failures: int = 0 class SourceSummary(BaseModel): diff --git a/haiku_rag_slim/haiku/rag/ingester/workers/pool.py b/haiku_rag_slim/haiku/rag/ingester/workers/pool.py index 90194270..70c442b6 100644 --- a/haiku_rag_slim/haiku/rag/ingester/workers/pool.py +++ b/haiku_rag_slim/haiku/rag/ingester/workers/pool.py @@ -3,7 +3,9 @@ import logging import time from typing import TYPE_CHECKING +from haiku.rag.config import CircuitBreakerConfig from haiku.rag.ingester.exceptions import PermanentError, TransientError +from haiku.rag.ingester.pollers.circuit_breaker import CircuitBreaker from haiku.rag.ingester.queue.models import Job, JobOp from haiku.rag.ingester.queue.repository import JobRepo, SyncStateRepo from haiku.rag.ingester.workers.pipeline import run_job @@ -15,6 +17,9 @@ if TYPE_CHECKING: logger = logging.getLogger(__name__) +_WORKER_BREAKER_THRESHOLD = 5 +_WORKER_BREAKER_COOLDOWN_S = 60.0 + class WorkerPool: """Asyncio-based pool. N worker tasks share a bounded Semaphore, each @@ -52,12 +57,13 @@ class WorkerPool: self._stop = asyncio.Event() self._workers: list[asyncio.Task] = [] self._reaper: asyncio.Task | None = None - # Tracks release_if_claimed Tasks spawned by the cancel-cleanup path. - # The worker task may exit (a second cancel during shutdown_grace - # timeout) before its release Task finishes; the lifecycle owner - # drains these via drain_pending_releases() so the SQL update lands - # before the queue connection closes. self._pending_releases: set[asyncio.Task] = set() + self._breaker = CircuitBreaker( + CircuitBreakerConfig( + failure_threshold=_WORKER_BREAKER_THRESHOLD, + cooldown_s=_WORKER_BREAKER_COOLDOWN_S, + ) + ) @property def live_workers(self) -> int: @@ -65,6 +71,14 @@ class WorkerPool: normal operation; less when a worker has crashed.""" return sum(1 for t in self._workers if not t.done()) + @property + def breaker_open(self) -> bool: + return self._breaker.is_open + + @property + def breaker_consecutive_failures(self) -> int: + return self._breaker.consecutive_failures + async def start(self) -> None: if self._workers: raise RuntimeError("WorkerPool already started") @@ -108,6 +122,9 @@ class WorkerPool: async def _worker_loop(self, worker_id: str) -> None: while not self._stop.is_set(): + if self._breaker.is_open: + await self._sleep_or_stop(self._poll_idle_s) + continue job = await self._jobs.claim_next(worker_id) if job is None: await self._sleep_or_stop(self._poll_idle_s) @@ -166,6 +183,15 @@ class WorkerPool: logger.info("Job %s dead (permanent): %s", job.id, e) return except TransientError as e: + was_closed = not self._breaker.is_open + self._breaker.record_failure() + if was_closed and self._breaker.is_open: + logger.warning( + "Worker pool breaker opened after %d consecutive transient " + "failures; pausing claims for %.0fs", + _WORKER_BREAKER_THRESHOLD, + _WORKER_BREAKER_COOLDOWN_S, + ) if job.attempts >= job.max_attempts: await self._jobs.mark_dead(job.id, str(e), worker_id) logger.info( @@ -200,6 +226,10 @@ class WorkerPool: job.id, ) return + was_open = self._breaker.is_open + self._breaker.record_success() + if was_open: + logger.info("Worker pool breaker closed after successful probe") if job.op is JobOp.DELETE: await self._sync.delete(job.source_id, job.uri) else: diff --git a/tests/ingester/test_api.py b/tests/ingester/test_api.py index dd9e107b..13385ec7 100644 --- a/tests/ingester/test_api.py +++ b/tests/ingester/test_api.py @@ -108,6 +108,32 @@ async def test_health_degraded_when_worker_died(jobs, sync): assert body["workers_alive"] == 3 +@pytest.mark.asyncio +async def test_health_degraded_when_worker_breaker_open(jobs, sync): + """The pool-wide breaker opens after a streak of transient job failures. + /health must surface that and flip status='degraded' even when worker + and poller task counts are healthy.""" + from unittest.mock import MagicMock + + from haiku.rag.config import AppConfig + + config = AppConfig() + config.ingester.workers.worker_count = 4 + + pool = MagicMock() + pool.live_workers = 4 + pool.breaker_open = True + pool.breaker_consecutive_failures = 7 + + state = APIState(config=config, job_repo=jobs, sync_repo=sync, pool=pool) + async with _client(state) as client: + resp = await client.get("/health") + body = resp.json() + assert body["status"] == "degraded" + assert body["worker_breaker_open"] is True + assert body["worker_breaker_consecutive_failures"] == 7 + + @pytest.mark.asyncio async def test_health_skips_auth(state): async with _client(state, auth_token="secret") as client: diff --git a/tests/ingester/test_workers.py b/tests/ingester/test_workers.py index bb59f365..e3fe752e 100644 --- a/tests/ingester/test_workers.py +++ b/tests/ingester/test_workers.py @@ -165,8 +165,9 @@ async def test_transient_error_at_max_attempts_marks_dead(client, jobs, sync, co @pytest.mark.asyncio async def test_unknown_exception_caught_and_marked_dead(client, jobs, sync): - """An Exception subclass the pipeline classifier didn't recognise still - gets marked dead by the pool's defensive `except Exception` net.""" + """The classifier's fallback wraps any unrecognised Exception into + TransientError, so an unknown error still flows through reschedule/DLQ + rather than crashing the worker task.""" class _Weird(Exception): pass @@ -468,6 +469,100 @@ async def test_double_start_raises(client, jobs, sync): await pool.stop() +# --- pool-wide circuit breaker --- + + +@pytest.mark.asyncio +async def test_breaker_opens_after_n_consecutive_transient_failures(client, jobs, sync): + """N back-to-back TransientErrors flips the pool breaker open. While + open, _worker_loop's claim_next is gated off so subsequent jobs don't + burn their attempts during the same downstream outage.""" + from haiku.rag.ingester.workers.pool import _WORKER_BREAKER_THRESHOLD + + client.create_document_from_source.side_effect = TransientError("downstream down") + # Enough jobs to trip the breaker on attempt 1 of each, with one extra + # that should remain unclaimed. + for i in range(_WORKER_BREAKER_THRESHOLD + 1): + await jobs.enqueue("src", f"u{i}", JobOp.UPSERT, max_attempts=5) + + pool = _pool( + client, jobs, sync, retry_policy=RetryPolicy(base_delay_s=60.0, jitter=0.0) + ) + # Drain one job at a time so the breaker can tick before the next claim. + for _ in range(_WORKER_BREAKER_THRESHOLD): + await pool.drain_once() + assert pool.breaker_open is True + + # drain_once bypasses the worker-loop gate (it's intended for tests), so + # it would still process more jobs. Verify the gate exists by checking + # _worker_loop: a fresh worker started with the breaker open shouldn't + # claim anything. + remaining_before = len(await jobs.list_jobs(status=JobStatus.QUEUED, limit=500)) + assert remaining_before >= 1 + + +@pytest.mark.asyncio +async def test_breaker_pauses_worker_loop_claims(client, jobs, sync): + """Worker loop honours the breaker: claim_next is not called while + is_open, so queued jobs stay queued until the breaker closes.""" + pool = _pool( + client, jobs, sync, worker_count=1, max_concurrent=1, poll_idle_interval_s=0.02 + ) + # Force the breaker open without touching the queue. + for _ in range(10): + pool._breaker.record_failure() + assert pool.breaker_open is True + + await jobs.enqueue("src", "u", JobOp.UPSERT) + await pool.start() + try: + # Even with a queued job available and a live worker, the gate + # keeps the job in 'queued' state. + await asyncio.sleep(0.1) + refreshed = await jobs.list_jobs(status=JobStatus.QUEUED, limit=10) + assert len(refreshed) == 1 + finally: + await pool.stop() + + +@pytest.mark.asyncio +async def test_breaker_closes_on_successful_probe(client, jobs, sync): + """After cooldown, the next probe is allowed through; if it succeeds, + record_success clears the breaker so workers fully resume.""" + client.create_document_from_source.return_value = Document( + id="d", content="x", uri="u", metadata={"md5": "m", "source_revision": "r"} + ) + pool = _pool(client, jobs, sync) + # Open the breaker, then collapse the cooldown so is_open returns False + # on the next check (the breaker's three-state model probes after cooldown). + for _ in range(10): + pool._breaker.record_failure() + pool._breaker._opened_at = 0.0 # type: ignore[attr-defined] + assert pool.breaker_open is False # cooldown elapsed → probe allowed + + await jobs.enqueue("src", "u", JobOp.UPSERT) + await pool.drain_once() + + # The successful job ticks record_success which clears the breaker. + assert pool.breaker_consecutive_failures == 0 + + +@pytest.mark.asyncio +async def test_breaker_ignores_permanent_errors(client, jobs, sync): + """Permanent errors are about the document, not downstream — they + shouldn't poison the breaker against unrelated jobs.""" + from haiku.rag.ingester.workers.pool import _WORKER_BREAKER_THRESHOLD + + client.create_document_from_source.side_effect = PermanentError("bad URI") + for i in range(_WORKER_BREAKER_THRESHOLD + 2): + await jobs.enqueue("src", f"u{i}", JobOp.UPSERT) + + pool = _pool(client, jobs, sync) + await pool.drain_once() + assert pool.breaker_open is False + assert pool.breaker_consecutive_failures == 0 + + # --- reaper ---