diff --git a/CHANGELOG.md b/CHANGELOG.md index ba10337a..74ebde3f 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -8,6 +8,7 @@ ### Fixed - `rebuild --embed-only` re-embeds picture chunks through the image path instead of overwriting their vectors with a text embedding of the caption. +- DELETE jobs re-check the source with `head()` before deleting and skip when the resource is back, so an atomic-rename save (vim, `git checkout`) no longer blackholes a live document. ## [0.52.0] - 2026-06-01 diff --git a/haiku_rag_slim/haiku/rag/ingester/workers/pipeline.py b/haiku_rag_slim/haiku/rag/ingester/workers/pipeline.py index 105bc417..b59e105c 100644 --- a/haiku_rag_slim/haiku/rag/ingester/workers/pipeline.py +++ b/haiku_rag_slim/haiku/rag/ingester/workers/pipeline.py @@ -9,6 +9,7 @@ from haiku.rag.client.exceptions import UnsupportedSourceError from haiku.rag.ingester.exceptions import PermanentError, TransientError from haiku.rag.ingester.queue.models import Job, JobOp from haiku.rag.ingester.sources.base import FileTooLargeError +from haiku.rag.ingester.sources.registry import resolve_configured_source from haiku.rag.telemetry import attach_context, logfire if TYPE_CHECKING: @@ -105,6 +106,17 @@ async def run_job( ): try: if job.op is JobOp.DELETE: + # An atomic-rename save can let a spurious DELETE win the + # enqueue race while the file is mid-rewrite. If the resource + # is already back, skip the delete (it would blackhole a live + # document) and let the next sweep re-ingest it. + try: + source = resolve_configured_source(job.uri, job.source_id, sources) + restored = await source.head(job.uri) is not None + except Exception: + restored = False + if restored: + return JobResult(deleted=False) doc = await client.get_document_by_uri(job.uri) if doc is not None and doc.id is not None: await client.delete_document(doc.id) diff --git a/tests/ingester/test_pipeline.py b/tests/ingester/test_pipeline.py index 59110d71..62d2cedf 100644 --- a/tests/ingester/test_pipeline.py +++ b/tests/ingester/test_pipeline.py @@ -7,7 +7,7 @@ import pytest from haiku.rag.client import HaikuRAG from haiku.rag.ingester.exceptions import PermanentError, TransientError from haiku.rag.ingester.queue.models import Job, JobOp, JobStatus -from haiku.rag.ingester.sources.base import FileTooLargeError +from haiku.rag.ingester.sources.base import FetchResult, FileTooLargeError, Source from haiku.rag.ingester.workers.pipeline import run_job from haiku.rag.store.models.document import Document @@ -104,6 +104,91 @@ async def test_delete_is_noop_when_document_missing(): client.delete_document.assert_not_awaited() +class _StubSource: + """Source double whose head() returns a scripted revision or raises.""" + + def __init__( + self, + source_id: str, + revision: str | None, + *, + head_error: Exception | None = None, + ): + self.source_id = source_id + self._revision = revision + self._head_error = head_error + + def supports(self, uri: str) -> bool: + return True + + async def head(self, uri: str) -> str | None: + if self._head_error is not None: + raise self._head_error + return self._revision + + async def aclose(self) -> None: ... + + async def fetch(self, uri: str) -> FetchResult: # pragma: no cover - unused + raise NotImplementedError + + async def discover(self, since=None, *, known_uris=None): # pragma: no cover + raise NotImplementedError + yield + + +@pytest.mark.asyncio +async def test_delete_skipped_when_resource_restored_on_source(): + """The file is back on disk by the time the DELETE runs, so head() + returns a revision and the delete is skipped — otherwise a live document + gets blackholed until the next sweep.""" + client = _mock_client() + client.get_document_by_uri.return_value = Document(id="doc-9", content="", uri="u") + sources: list[Source] = [_StubSource("src", "12345")] + + result = await run_job(client, _job(op=JobOp.DELETE), sources=sources) + + assert result.deleted is False + client.delete_document.assert_not_awaited() + + +@pytest.mark.asyncio +async def test_delete_proceeds_when_resource_absent_on_source(): + client = _mock_client() + client.get_document_by_uri.return_value = Document(id="doc-9", content="", uri="u") + sources: list[Source] = [_StubSource("src", None)] + + result = await run_job(client, _job(op=JobOp.DELETE), sources=sources) + + assert result.deleted is True + client.delete_document.assert_awaited_once_with("doc-9") + + +@pytest.mark.asyncio +async def test_delete_proceeds_when_source_unresolvable(): + """No configured source for the job: the probe can't run, so the delete + proceeds exactly as before.""" + client = _mock_client() + client.get_document_by_uri.return_value = Document(id="doc-9", content="", uri="u") + + result = await run_job(client, _job(op=JobOp.DELETE), sources=[]) + + assert result.deleted is True + client.delete_document.assert_awaited_once_with("doc-9") + + +@pytest.mark.asyncio +async def test_delete_proceeds_when_head_probe_raises(): + """A failing head() probe must not block the delete.""" + client = _mock_client() + client.get_document_by_uri.return_value = Document(id="doc-9", content="", uri="u") + sources: list[Source] = [_StubSource("src", None, head_error=OSError("boom"))] + + result = await run_job(client, _job(op=JobOp.DELETE), sources=sources) + + assert result.deleted is True + client.delete_document.assert_awaited_once_with("doc-9") + + @pytest.mark.asyncio async def test_unsupported_extension_classified_permanent(): from haiku.rag.client.exceptions import UnsupportedSourceError