diff --git a/CHANGELOG.md b/CHANGELOG.md index a2c28ac2..7dece556 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -12,7 +12,7 @@ ### 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. All four source adapters (FS, HTTP, S3, future 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. +- `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. 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 stay `claimed` and the reaper resets them on next start. 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. diff --git a/haiku_rag_slim/haiku/rag/ingester/sources/http.py b/haiku_rag_slim/haiku/rag/ingester/sources/http.py index df23182b..a0d9041b 100644 --- a/haiku_rag_slim/haiku/rag/ingester/sources/http.py +++ b/haiku_rag_slim/haiku/rag/ingester/sources/http.py @@ -50,11 +50,16 @@ class HTTPSource: return httpx.AsyncClient(headers=self.headers, transport=self._transport) async def head(self, uri: str) -> str | None: - # HTTP doesn't get a cheap revision lookup in v1: the existing - # ingestion flow always GETs and the dedup uses MD5. A HEAD-first - # optimization could land later without changing this contract — - # callers just need to tolerate the extra HEAD. - return None + """HEAD probe for the cheap revision short-circuit. Returns the + ETag (or Last-Modified) so an unchanged remote URL can skip the + full GET. None on HTTP error so the caller falls back to fetch(); + network errors propagate and the worker's classifier handles them.""" + async with self._client() as http: + response = await http.head(uri) + if response.is_error: + return None + revision, _ = _extract_revision(response.headers) + return revision async def fetch(self, uri: str) -> FetchResult: async with self._client() as http: diff --git a/tests/ingester/test_http_source.py b/tests/ingester/test_http_source.py index 9130a1a5..5c5e3a49 100644 --- a/tests/ingester/test_http_source.py +++ b/tests/ingester/test_http_source.py @@ -30,10 +30,45 @@ def test_source_id_is_user_provided(): @pytest.mark.asyncio -async def test_head_returns_none(): - # HTTP has no cheap revision lookup in v1 — the pipeline always GETs. - src = HTTPSource(source_id="default") - assert await src.head("https://example.com/a.md") is None +async def test_head_returns_etag(): + transport = _transport( + { + ("HEAD", "https://example.com/a.md"): httpx.Response( + 200, headers={"etag": '"rev-7"'} + ), + } + ) + src = HTTPSource(source_id="default", transport=transport) + assert await src.head("https://example.com/a.md") == "rev-7" + + +@pytest.mark.asyncio +async def test_head_falls_back_to_last_modified(): + transport = _transport( + { + ("HEAD", "https://example.com/a.md"): httpx.Response( + 200, headers={"last-modified": "Wed, 21 Oct 2025 07:28:00 GMT"} + ), + } + ) + src = HTTPSource(source_id="default", transport=transport) + assert await src.head("https://example.com/a.md") == "Wed, 21 Oct 2025 07:28:00 GMT" + + +@pytest.mark.asyncio +async def test_head_returns_none_on_error_status(): + transport = _transport( + {("HEAD", "https://example.com/missing"): httpx.Response(404)} + ) + src = HTTPSource(source_id="default", transport=transport) + assert await src.head("https://example.com/missing") is None + + +@pytest.mark.asyncio +async def test_head_returns_none_when_no_revision_headers(): + transport = _transport({("HEAD", "https://example.com/a"): httpx.Response(200)}) + src = HTTPSource(source_id="default", transport=transport) + assert await src.head("https://example.com/a") is None @pytest.mark.asyncio