From cfbd0b09d6b0f410316ccfb3ef4c0e9261c5a477 Mon Sep 17 00:00:00 2001 From: Yiorgis Gozadinos Date: Wed, 27 May 2026 14:07:45 +0300 Subject: [PATCH] Tighten HTTP config-removal handling --- .../haiku/rag/ingester/queue/repository.py | 11 +++++--- .../haiku/rag/ingester/sources/base.py | 2 +- .../haiku/rag/ingester/sources/fs.py | 2 +- .../haiku/rag/ingester/sources/http.py | 2 +- .../haiku/rag/ingester/sources/registry.py | 27 ++++++++++--------- .../haiku/rag/ingester/sources/s3.py | 2 +- .../haiku/rag/ingester/sources/webdav.py | 2 +- tests/ingester/test_http_source.py | 22 +++++++++++++++ tests/ingester/test_pollers.py | 5 ++-- tests/ingester/test_queue.py | 11 ++++++++ tests/ingester/test_resolve_fetcher.py | 17 ++++++++++++ 11 files changed, 80 insertions(+), 23 deletions(-) diff --git a/haiku_rag_slim/haiku/rag/ingester/queue/repository.py b/haiku_rag_slim/haiku/rag/ingester/queue/repository.py index 4156ac9a..6bff9901 100644 --- a/haiku_rag_slim/haiku/rag/ingester/queue/repository.py +++ b/haiku_rag_slim/haiku/rag/ingester/queue/repository.py @@ -399,12 +399,15 @@ class SyncStateRepo: # connection. See JobRepo for the SQLite cursor + commit constraint. self._lock = lock or asyncio.Lock() - async def get_snapshot(self, source_id: str) -> dict[str, str]: - """uri -> revision map for the source. Drops rows where revision is - NULL (the poller can't compare against an absent revision).""" + async def get_snapshot(self, source_id: str) -> dict[str, str | None]: + """uri -> revision map for the source. Revision is None for URIs + known to the source but never successfully ingested with a revision + (HTTP responses without ETag / Last-Modified, or a worker that DLQ'd + before completion). Sources compare with `previous == current`; None + compares as not-equal so the URI is treated as changed.""" async with self._lock: async with self._conn.execute( - "SELECT uri, revision FROM sync_state WHERE source_id=? AND revision IS NOT NULL", + "SELECT uri, revision FROM sync_state WHERE source_id=?", (source_id,), ) as cursor: rows = await cursor.fetchall() diff --git a/haiku_rag_slim/haiku/rag/ingester/sources/base.py b/haiku_rag_slim/haiku/rag/ingester/sources/base.py index 021515d7..7beea83f 100644 --- a/haiku_rag_slim/haiku/rag/ingester/sources/base.py +++ b/haiku_rag_slim/haiku/rag/ingester/sources/base.py @@ -9,7 +9,7 @@ from pydantic import BaseModel, Field # uri -> revision. Captures what revisions of which URIs we had last seen # for a given source. Passed to discover() so the source can yield only # UPSERT / UNCHANGED / DELETE deltas instead of a full re-scan. -RevisionSnapshot = Mapping[str, str] +RevisionSnapshot = Mapping[str, str | None] class SourceEventKind(StrEnum): diff --git a/haiku_rag_slim/haiku/rag/ingester/sources/fs.py b/haiku_rag_slim/haiku/rag/ingester/sources/fs.py index 25621772..b6cbd391 100644 --- a/haiku_rag_slim/haiku/rag/ingester/sources/fs.py +++ b/haiku_rag_slim/haiku/rag/ingester/sources/fs.py @@ -104,7 +104,7 @@ class FSSource: async def discover( self, since: RevisionSnapshot | None = None ) -> AsyncIterator[SourceEvent]: - snapshot: dict[str, str] = dict(since) if since else {} + snapshot: dict[str, str | None] = dict(since) if since else {} now = datetime.now(UTC) seen: set[str] = set() diff --git a/haiku_rag_slim/haiku/rag/ingester/sources/http.py b/haiku_rag_slim/haiku/rag/ingester/sources/http.py index ba77fa5d..82dab875 100644 --- a/haiku_rag_slim/haiku/rag/ingester/sources/http.py +++ b/haiku_rag_slim/haiku/rag/ingester/sources/http.py @@ -95,7 +95,7 @@ class HTTPSource: # ambiguous (transient outage, misconfigured URL, auth blip), so we # fall back to UPSERT with no revision and let the worker decide # via GET. - snapshot: dict[str, str] = dict(since) if since else {} + snapshot: dict[str, str | None] = dict(since) if since else {} now = datetime.now(UTC) configured = set(self.urls) diff --git a/haiku_rag_slim/haiku/rag/ingester/sources/registry.py b/haiku_rag_slim/haiku/rag/ingester/sources/registry.py index 707830b2..c102509d 100644 --- a/haiku_rag_slim/haiku/rag/ingester/sources/registry.py +++ b/haiku_rag_slim/haiku/rag/ingester/sources/registry.py @@ -25,19 +25,22 @@ def resolve_fetcher( source whose ``supports(uri)`` returns True wins, then a scheme-based ad-hoc adapter. """ + if source_id is not None: + for src in sources or (): + if src.source_id == source_id: + if not src.supports(uri): + raise UnsupportedSourceError( + f"Source {source_id!r} doesn't support URI {uri!r}" + ) + return src + raise UnsupportedSourceError( + f"No configured source with id {source_id!r} for URI {uri!r}" + ) + if sources: - if source_id is not None: - for src in sources: - if src.source_id == source_id: - if not src.supports(uri): - raise UnsupportedSourceError( - f"Source {source_id!r} doesn't support URI {uri!r}" - ) - return src - else: - for src in sources: - if src.supports(uri): - return src + for src in sources: + if src.supports(uri): + return src scheme = urlparse(uri).scheme if scheme in ("", "file"): diff --git a/haiku_rag_slim/haiku/rag/ingester/sources/s3.py b/haiku_rag_slim/haiku/rag/ingester/sources/s3.py index fe6d6917..1b5fe5d8 100644 --- a/haiku_rag_slim/haiku/rag/ingester/sources/s3.py +++ b/haiku_rag_slim/haiku/rag/ingester/sources/s3.py @@ -109,7 +109,7 @@ class S3Source: from haiku.rag.s3 import make_s3_store - snapshot: dict[str, str] = dict(since) if since else {} + snapshot: dict[str, str | None] = dict(since) if since else {} now = datetime.now(UTC) seen: set[str] = set() store = make_s3_store(self.bucket, self.storage_options) diff --git a/haiku_rag_slim/haiku/rag/ingester/sources/webdav.py b/haiku_rag_slim/haiku/rag/ingester/sources/webdav.py index 290badbb..f50cce10 100644 --- a/haiku_rag_slim/haiku/rag/ingester/sources/webdav.py +++ b/haiku_rag_slim/haiku/rag/ingester/sources/webdav.py @@ -253,7 +253,7 @@ class WebDAVSource: async def discover( self, since: RevisionSnapshot | None = None ) -> AsyncIterator[SourceEvent]: - snapshot: dict[str, str] = dict(since) if since else {} + snapshot: dict[str, str | None] = dict(since) if since else {} now = datetime.now(UTC) seen: set[str] = set() diff --git a/tests/ingester/test_http_source.py b/tests/ingester/test_http_source.py index eb08a1f1..9e30a683 100644 --- a/tests/ingester/test_http_source.py +++ b/tests/ingester/test_http_source.py @@ -296,3 +296,25 @@ async def test_discover_emits_delete_for_removed_url(): assert by_uri["https://example.com/a.md"].kind is not SourceEventKind.DELETE assert by_uri["https://example.com/gone.md"].kind is SourceEventKind.DELETE assert by_uri["https://example.com/gone.md"].revision is None + + +@pytest.mark.asyncio +async def test_discover_emits_delete_for_removed_url_with_none_revision(): + """Same as above but the snapshot entry's revision is None — an HTTP + response without ETag or Last-Modified produces that state, and these + rows must still trigger config-removal DELETE.""" + transport = _transport( + { + ("HEAD", "https://example.com/a.md"): httpx.Response( + 200, headers={"etag": '"abc"'} + ), + } + ) + src = HTTPSource( + source_id="x", urls=["https://example.com/a.md"], transport=transport + ) + events = [ + e async for e in src.discover(since={"https://example.com/no-etag.md": None}) + ] + by_uri = {e.uri: e for e in events} + assert by_uri["https://example.com/no-etag.md"].kind is SourceEventKind.DELETE diff --git a/tests/ingester/test_pollers.py b/tests/ingester/test_pollers.py index 34e62d38..c1b1d48a 100644 --- a/tests/ingester/test_pollers.py +++ b/tests/ingester/test_pollers.py @@ -130,9 +130,10 @@ async def test_upsert_event_enqueues_job_and_touches_sync_state(fs_config, jobs, assert queued[0].revision == "r1" # Pollers DO NOT write revision to sync_state — the worker does that - # after a successful ingest. But last_seen_at is bumped. + # after a successful ingest. The URI shows up in the snapshot with + # revision=None until then. snapshot = await sync.get_snapshot("src") - assert snapshot == {} # revision left empty by the poller + assert snapshot == {"file:///a.md": None} @pytest.mark.asyncio diff --git a/tests/ingester/test_queue.py b/tests/ingester/test_queue.py index 77380b44..eb6a1693 100644 --- a/tests/ingester/test_queue.py +++ b/tests/ingester/test_queue.py @@ -750,6 +750,17 @@ async def test_sync_state_snapshot_scoped_per_source(sync): assert await sync.get_snapshot("s2") == {"u": "def"} +@pytest.mark.asyncio +async def test_sync_state_snapshot_includes_null_revision_rows(sync): + """A URI known to a source but never successfully ingested with a + revision (HTTP without ETag/Last-Modified, or a worker DLQ'd before + completion) is still in the snapshot — with revision=None. Needed so + config-removal DELETE detection sees these rows.""" + await sync.upsert("s", "u", revision=None, content_hash=None) + snapshot = await sync.get_snapshot("s") + assert snapshot == {"u": None} + + @pytest.mark.asyncio async def test_sync_state_upsert_preserves_revision_when_none(sync): """upsert(revision=None) leaves an existing revision in place.""" diff --git a/tests/ingester/test_resolve_fetcher.py b/tests/ingester/test_resolve_fetcher.py index d162c7a4..51f2c30e 100644 --- a/tests/ingester/test_resolve_fetcher.py +++ b/tests/ingester/test_resolve_fetcher.py @@ -83,3 +83,20 @@ def test_source_id_raises_when_matched_source_rejects_uri(): fs = FSSource(root=Path("/tmp"), source_id="local") with pytest.raises(ValueError, match="doesn't support URI"): resolve_fetcher("https://example.com/x", sources=[fs], source_id="local") + + +def test_source_id_raises_when_no_source_matches(): + """A worker job carries source_id from when it was enqueued. If the + operator renamed/removed that source, falling through to an ad-hoc + unauthenticated adapter would silently drop credentials. Raise instead + so the job DLQs and the misconfiguration is visible.""" + arxiv = HTTPSource(source_id="arxiv", headers={"Authorization": "Bearer A"}) + with pytest.raises(ValueError, match="No configured source with id 'gone'"): + resolve_fetcher("https://example.com/x", sources=[arxiv], source_id="gone") + + +def test_source_id_with_no_sources_list_still_raises(): + """Same strict mode when sources is None: source_id is a strong claim + that must match a configured source.""" + with pytest.raises(ValueError, match="No configured source with id"): + resolve_fetcher("https://example.com/x", source_id="arxiv")