Classify obstore config errors as permanent
This commit is contained in:
parent
5a023a24ad
commit
292835190a
4 changed files with 64 additions and 1 deletions
|
|
@ -1,6 +1,10 @@
|
||||||
# Changelog
|
# Changelog
|
||||||
## [Unreleased]
|
## [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
|
## [0.72.1] - 2026-07-31
|
||||||
|
|
||||||
### Changed
|
### Changed
|
||||||
|
|
|
||||||
|
|
@ -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
|
concurrent in-flight jobs. Jobs that hit a `TransientError` are
|
||||||
rescheduled with exponential backoff plus jitter, up to `max_attempts`,
|
rescheduled with exponential backoff plus jitter, up to `max_attempts`,
|
||||||
then land in the dead-letter queue. `PermanentError` (unsupported
|
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
|
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
|
`heartbeat_interval_s`. A reaper task resets any claim whose lease has not
|
||||||
|
|
|
||||||
|
|
@ -3,6 +3,12 @@ from contextlib import nullcontext
|
||||||
from typing import TYPE_CHECKING
|
from typing import TYPE_CHECKING
|
||||||
|
|
||||||
import httpx
|
import httpx
|
||||||
|
from obstore.exceptions import (
|
||||||
|
InvalidPathError,
|
||||||
|
PermissionDeniedError,
|
||||||
|
UnauthenticatedError,
|
||||||
|
UnknownConfigurationKeyError,
|
||||||
|
)
|
||||||
from pydantic import BaseModel
|
from pydantic import BaseModel
|
||||||
|
|
||||||
from haiku.rag.client.exceptions import UnsupportedSourceError
|
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.
|
# ProxyError — every transport-layer failure that's worth retrying.
|
||||||
return TransientError(f"network: {exc}")
|
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):
|
if isinstance(exc, FileNotFoundError):
|
||||||
return PermanentError(f"file not found: {exc}")
|
return PermanentError(f"file not found: {exc}")
|
||||||
|
|
||||||
|
|
|
||||||
|
|
@ -3,6 +3,14 @@ from unittest.mock import AsyncMock
|
||||||
|
|
||||||
import httpx
|
import httpx
|
||||||
import pytest
|
import pytest
|
||||||
|
from obstore.exceptions import (
|
||||||
|
GenericError,
|
||||||
|
InvalidPathError,
|
||||||
|
JoinError,
|
||||||
|
PermissionDeniedError,
|
||||||
|
UnauthenticatedError,
|
||||||
|
UnknownConfigurationKeyError,
|
||||||
|
)
|
||||||
|
|
||||||
from haiku.rag.client import HaikuRAG
|
from haiku.rag.client import HaikuRAG
|
||||||
from haiku.rag.ingester.exceptions import PermanentError, TransientError
|
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")
|
client.create_document_from_source.side_effect = FileTooLargeError("too big")
|
||||||
with pytest.raises(PermanentError, match="too big"):
|
with pytest.raises(PermanentError, match="too big"):
|
||||||
await run_job(client, _job())
|
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())
|
||||||
|
|
|
||||||
Loading…
Reference in a new issue