diff --git a/CHANGELOG.md b/CHANGELOG.md index 47cafddc..f4d1e816 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,6 +1,10 @@ # Changelog ## [Unreleased] +### Fixed + +- Ingester jobs failing with an `obstore` `PermissionDeniedError`, `UnauthenticatedError`, `UnknownConfigurationKeyError` or `InvalidPathError` are dead-lettered instead of retried to `max_attempts`. + ## [0.72.1] - 2026-07-31 ### Changed diff --git a/docs/ingester.md b/docs/ingester.md index 243f2fed..81bb8cf5 100644 --- a/docs/ingester.md +++ b/docs/ingester.md @@ -351,7 +351,8 @@ job at a time. `worker_count` is therefore also the maximum number of concurrent in-flight jobs. Jobs that hit a `TransientError` are rescheduled with exponential backoff plus jitter, up to `max_attempts`, then land in the dead-letter queue. `PermanentError` (unsupported -extension, 4xx HTTP except 408/429, etc.) skips retry entirely. +extension, 4xx HTTP except 408/429, object-store credential and +configuration errors, etc.) skips retry entirely. While a worker processes a job it renews the job's lease every `heartbeat_interval_s`. A reaper task resets any claim whose lease has not diff --git a/haiku_rag_slim/haiku/rag/ingester/workers/pipeline.py b/haiku_rag_slim/haiku/rag/ingester/workers/pipeline.py index 977abbac..4b3fe3b3 100644 --- a/haiku_rag_slim/haiku/rag/ingester/workers/pipeline.py +++ b/haiku_rag_slim/haiku/rag/ingester/workers/pipeline.py @@ -3,6 +3,12 @@ from contextlib import nullcontext from typing import TYPE_CHECKING import httpx +from obstore.exceptions import ( + InvalidPathError, + PermissionDeniedError, + UnauthenticatedError, + UnknownConfigurationKeyError, +) from pydantic import BaseModel from haiku.rag.client.exceptions import UnsupportedSourceError @@ -62,6 +68,19 @@ def _classify(exc: BaseException) -> Exception: # ProxyError — every transport-layer failure that's worth retrying. return TransientError(f"network: {exc}") + if isinstance( + exc, + PermissionDeniedError + | UnauthenticatedError + | UnknownConfigurationKeyError + | InvalidPathError, + ): + # These subclass neither OSError nor httpx, so without this branch + # they'd fall through to the transient default and retry the whole + # backoff ladder. A missing object arrives as a builtin + # FileNotFoundError instead, handled below. + return PermanentError(f"object store: {exc}") + if isinstance(exc, FileNotFoundError): return PermanentError(f"file not found: {exc}") diff --git a/tests/ingester/test_pipeline.py b/tests/ingester/test_pipeline.py index 882c9ee2..2d52873f 100644 --- a/tests/ingester/test_pipeline.py +++ b/tests/ingester/test_pipeline.py @@ -3,6 +3,14 @@ from unittest.mock import AsyncMock import httpx import pytest +from obstore.exceptions import ( + GenericError, + InvalidPathError, + JoinError, + PermissionDeniedError, + UnauthenticatedError, + UnknownConfigurationKeyError, +) from haiku.rag.client import HaikuRAG from haiku.rag.ingester.exceptions import PermanentError, TransientError @@ -541,3 +549,34 @@ async def test_file_too_large_classified_as_permanent(): client.create_document_from_source.side_effect = FileTooLargeError("too big") with pytest.raises(PermanentError, match="too big"): await run_job(client, _job()) + + +@pytest.mark.parametrize( + "exc_class", + [ + PermissionDeniedError, + UnauthenticatedError, + UnknownConfigurationKeyError, + InvalidPathError, + ], +) +@pytest.mark.asyncio +async def test_obstore_config_errors_classified_permanent(exc_class): + client = _mock_client() + client.create_document_from_source.side_effect = exc_class("bad config") + with pytest.raises(PermanentError, match="object store"): + await run_job(client, _job()) + + +@pytest.mark.parametrize( + "exc_class", + [GenericError, JoinError], +) +@pytest.mark.asyncio +async def test_other_obstore_errors_classified_transient(exc_class): + """Only the credential/configuration errors are permanent; umbrellaing on + obstore's BaseError would sweep up retryable failures too.""" + client = _mock_client() + client.create_document_from_source.side_effect = exc_class("upstream hiccup") + with pytest.raises(TransientError): + await run_job(client, _job())