Merge pull request #527 from ggozad/fix/s3-error-classification

Classify obstore config errors as permanent

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

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

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}")
 

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())