diff --git a/haiku_rag_slim/haiku/rag/ingester/queue/repository.py b/haiku_rag_slim/haiku/rag/ingester/queue/repository.py index ad9a7a6d..2c7e57df 100644 --- a/haiku_rag_slim/haiku/rag/ingester/queue/repository.py +++ b/haiku_rag_slim/haiku/rag/ingester/queue/repository.py @@ -392,7 +392,9 @@ class SyncStateRepo: ingested: bool = False, ) -> None: """Insert-or-update the sync_state row. `ingested=True` stamps - last_ingested_at; otherwise only last_seen_at is bumped.""" + last_ingested_at; otherwise only last_seen_at is bumped. + `revision=None` and `content_hash=None` leave any existing values + untouched.""" now = _utcnow_iso() ingested_at = now if ingested else None async with self._lock: @@ -403,8 +405,8 @@ class SyncStateRepo: ) VALUES (?, ?, ?, ?, ?, ?) ON CONFLICT(source_id, uri) DO UPDATE SET - revision = excluded.revision, - content_hash = excluded.content_hash, + revision = COALESCE(excluded.revision, revision), + content_hash = COALESCE(excluded.content_hash, content_hash), last_seen_at = excluded.last_seen_at, last_ingested_at = COALESCE(excluded.last_ingested_at, last_ingested_at) """, diff --git a/tests/ingester/test_pollers.py b/tests/ingester/test_pollers.py index 2b527dfe..e4744813 100644 --- a/tests/ingester/test_pollers.py +++ b/tests/ingester/test_pollers.py @@ -202,6 +202,33 @@ async def test_skipped_sweep_records_pending_work_reason(fs_config, jobs, sync): assert poller.last_skip_reason is None +@pytest.mark.asyncio +async def test_dead_job_does_not_clear_sync_state_revision(fs_config, jobs, sync): + """When a job dies, the URI's previously-ingested revision must remain + in sync_state so subsequent sweeps still see the URI as known.""" + await sync.upsert("src", "file:///a.md", revision="r1", content_hash="h1") + + changed = _event("file:///a.md", revision="r2") + source = _StubSource("src", [[changed], [changed]]) + poller = _periodic(source, fs_config, jobs, sync) + + await poller._sweep_once() + queued = await jobs.list_jobs() + assert len(queued) == 1 + assert queued[0].revision == "r2" + + claimed = await jobs.claim_next("w") + assert claimed is not None + await jobs.mark_dead(claimed.id, "transient blew up") + + row = await sync.get_row("src", "file:///a.md") + assert row is not None + assert row.revision == "r1" + + await poller._sweep_once() + assert await sync.get_snapshot("src") == {"file:///a.md": "r1"} + + @pytest.mark.asyncio async def test_sweep_resumes_after_queue_drains(fs_config, jobs, sync): """Once the queue clears (success, dead, or cancel), sweeps resume.""" diff --git a/tests/ingester/test_queue.py b/tests/ingester/test_queue.py index 2e39d742..152a0b46 100644 --- a/tests/ingester/test_queue.py +++ b/tests/ingester/test_queue.py @@ -572,3 +572,27 @@ async def test_sync_state_snapshot_scoped_per_source(sync): await sync.upsert("s2", "u", revision="def", content_hash="m") assert await sync.get_snapshot("s1") == {"u": "abc"} assert await sync.get_snapshot("s2") == {"u": "def"} + + +@pytest.mark.asyncio +async def test_sync_state_upsert_preserves_revision_when_none(sync): + """upsert(revision=None) leaves an existing revision in place.""" + await sync.upsert("s", "u", revision="v1", content_hash="hash-v1") + await sync.upsert("s", "u", revision=None, content_hash=None) + row = await sync.get_row("s", "u") + assert row is not None + assert row.revision == "v1" + assert row.content_hash == "hash-v1" + assert await sync.get_snapshot("s") == {"u": "v1"} + + +@pytest.mark.asyncio +async def test_sync_state_upsert_replaces_revision_when_provided(sync): + """upsert with a non-None revision overwrites the existing one.""" + await sync.upsert("s", "u", revision="v1", content_hash="hash-v1") + await sync.upsert("s", "u", revision="v2", content_hash="hash-v2", ingested=True) + row = await sync.get_row("s", "u") + assert row is not None + assert row.revision == "v2" + assert row.content_hash == "hash-v2" + assert row.last_ingested_at is not None