Merge pull request #910 from cbcoutinho/feat/worker-batch-ocr
feat(ocr): opt-in batch OCR mode via the gateway's async batch routes
This commit is contained in:
Chris Coutinho
@@ -572,6 +572,42 @@ A PDF larger than `DOCUMENT_MAX_PDF_SIZE_MB` fails fast with reason `oversize`
|
||||
of being handed to the tiers, where a 40+ MB scan would otherwise burn the full
|
||||
OCR timeout for zero recovered text.
|
||||
|
||||
#### OCR execution mode: synchronous vs batch (Deck #332)
|
||||
|
||||
The tier-3 OCR processor has two execution modes, selected by `DOCUMENT_OCR_MODE`:
|
||||
|
||||
```dotenv
|
||||
DOCUMENT_OCR_MODE=sync # "sync" (default) | "batch"
|
||||
DOCUMENT_OCR_BATCH_POLL_SECONDS=120 # re-poll cadence for a batch job (default: 120)
|
||||
DOCUMENT_OCR_BATCH_MAX_WAIT_SECONDS=86400 # give up + mark timeout after this (default: 24h)
|
||||
```
|
||||
|
||||
- **`sync`** (default) — transcribe the document inline via the backend's
|
||||
synchronous path (`POST /v1/ocr` for the gateway, or the direct Mistral OCR
|
||||
API). The document is parsed in a single call.
|
||||
- **`batch`** — submit the document to the **gateway's async Batch OCR** job
|
||||
(`POST /v1/ocr/batch`) and re-poll `GET /v1/ocr/batch/{job_id}` until it
|
||||
finishes. This trades latency (a batch job runs minutes–hours) for roughly
|
||||
**half the OCR cost**, so it suits large-corpus backfill rather than
|
||||
interactive ingest.
|
||||
|
||||
Batch mode is **opt-in and gateway-only**: it routes Mistral's Batch API
|
||||
*through* the gateway's batch routes (no provider keys in the pod). With the
|
||||
direct `mistral` backend, no `EMBEDDING_GATEWAY_URL`, or on the in-process
|
||||
(`INGEST_QUEUE=memory`) pipeline — which can't defer a poll — batch transparently
|
||||
**falls back to synchronous OCR**. So enabling it requires the Postgres ingest
|
||||
queue (the per-tier procrastinate workers) and the gateway embedding backend.
|
||||
|
||||
Mechanics: the OCR tier submits the job, records its id in the `batch_ocr_jobs`
|
||||
app-DB table (keyed on the document + its etag), and raises a re-poll deferral so
|
||||
procrastinate re-runs the tier after `DOCUMENT_OCR_BATCH_POLL_SECONDS` — releasing
|
||||
the worker slot between polls (a long batch never pins a worker or is reclaimed as
|
||||
stalled). On completion the per-page markdown is indexed exactly like the sync
|
||||
path; a failure or a job exceeding `DOCUMENT_OCR_BATCH_MAX_WAIT_SECONDS` marks the
|
||||
document parse-failed. Each poll re-fetches + re-classifies the PDF (a known v1
|
||||
inefficiency, bounded by the poll cadence); one batch job is submitted per
|
||||
document (coalescing many documents per job is a planned follow-up).
|
||||
|
||||
### Embedding Service Configuration
|
||||
|
||||
The server picks an embedding provider via auto-detection. Priority order
|
||||
|
||||
@@ -0,0 +1,64 @@
|
||||
"""Add batch_ocr_jobs table for async batch OCR job tracking.
|
||||
|
||||
Deck #332 / embedding-gateway batch OCR (astrolabe-cloud-website#372). When
|
||||
``DOCUMENT_OCR_MODE=batch`` the OCR tier submits a document to the gateway's
|
||||
async ``POST /v1/ocr/batch`` and must re-poll ``GET /v1/ocr/batch/{job_id}``
|
||||
across procrastinate retries. procrastinate job args are immutable, so the
|
||||
gateway ``job_id`` (and submit time, for the poll deadline) are persisted here,
|
||||
keyed on the document + its content version (``etag``).
|
||||
|
||||
One row per in-flight job; the row is deleted once the job reaches a terminal
|
||||
state. Empty + unused unless batch mode is enabled (gateway-only), so OSS/SQLite
|
||||
self-hosters get an idle table and zero overhead.
|
||||
|
||||
Portable types only (Text + unix-epoch BigInteger timestamps, like the rest of
|
||||
this schema except the CP-queried usage_events) so the same migration runs on
|
||||
both self-host SQLite and cloud Postgres.
|
||||
|
||||
Revision ID: 008
|
||||
Revises: 007
|
||||
Create Date: 2026-06-15 12:00:00.000000
|
||||
"""
|
||||
|
||||
import sqlalchemy as sa
|
||||
|
||||
from alembic import op
|
||||
|
||||
revision = "008"
|
||||
down_revision = "007"
|
||||
branch_labels = None
|
||||
depends_on = None
|
||||
|
||||
|
||||
def upgrade() -> None:
|
||||
op.create_table(
|
||||
"batch_ocr_jobs",
|
||||
# Document identity (the same keys the OCR tier receives via the
|
||||
# processor ``options``). ``etag`` is the content-version key: a changed
|
||||
# document (new etag) is a new job, so a stale row never serves results
|
||||
# for the wrong content. The four-column natural key IS the primary key:
|
||||
# one in-flight job per (document, content version), and the PK doubles as
|
||||
# the unique index ``insert_pending``'s ON CONFLICT target relies on. A
|
||||
# resubmit for a new etag inserts a new row; the superseded row is swept on
|
||||
# resubmit (delete_stale_for_doc).
|
||||
sa.Column("user_id", sa.Text(), nullable=False),
|
||||
sa.Column("doc_id", sa.Text(), nullable=False),
|
||||
sa.Column("doc_type", sa.Text(), nullable=False),
|
||||
sa.Column("etag", sa.Text(), nullable=False),
|
||||
# The gateway's namespaced batch job id ("<provider>/<batch_job_id>") —
|
||||
# the only handle for polling (the gateway is stateless).
|
||||
sa.Column("job_id", sa.Text(), nullable=False),
|
||||
# Unix-epoch seconds. ``submitted_at`` anchors the poll deadline
|
||||
# (DOCUMENT_OCR_BATCH_MAX_WAIT_SECONDS). No status/updated_at column: a row
|
||||
# only ever exists in the pending state (terminal jobs are deleted), and
|
||||
# the live status always comes from a fresh poll — a stored mirror would
|
||||
# be permanently "pending" and carry no information.
|
||||
sa.Column("submitted_at", sa.BigInteger(), nullable=False),
|
||||
sa.PrimaryKeyConstraint(
|
||||
"user_id", "doc_id", "doc_type", "etag", name="pk_batch_ocr_jobs"
|
||||
),
|
||||
)
|
||||
|
||||
|
||||
def downgrade() -> None:
|
||||
op.drop_table("batch_ocr_jobs")
|
||||
@@ -175,6 +175,20 @@ _DEFAULTS: dict[str, Any] = {
|
||||
# 20-60s; raise/lower per tenant. Configurable so a tenant isn't stuck with
|
||||
# the 180s default when its gateway has its own shorter ceiling.
|
||||
"document_ocr_timeout_seconds": 180.0,
|
||||
# OCR execution mode (Deck #332). "sync" (default) transcribes inline via the
|
||||
# backend's synchronous path. "batch" routes to the gateway's async Batch OCR
|
||||
# job (~50% cheaper, minutes-hours latency) for large-corpus backfill — opt-in
|
||||
# and gateway-only; with the direct mistral backend or no gateway it falls
|
||||
# back to sync. The submit->defer-poll loop runs on the per-tier procrastinate
|
||||
# path; the inline/memory pool can't defer, so batch falls back to sync there.
|
||||
"document_ocr_mode": "sync",
|
||||
# Seconds between batch-job polls (the procrastinate re-enqueue delay). Each
|
||||
# poll re-runs the tier; keep it well above a few seconds.
|
||||
"document_ocr_batch_poll_seconds": 120,
|
||||
# Hard deadline (seconds from submit) after which a still-pending batch job is
|
||||
# abandoned and the document marked parse-failed (timeout). Matches the
|
||||
# gateway's 24h Batch timeout default.
|
||||
"document_ocr_batch_max_wait_seconds": 86400,
|
||||
# Observability
|
||||
"metrics_enabled": True,
|
||||
"metrics_port": 9090,
|
||||
@@ -352,6 +366,14 @@ _dynaconf = Dynaconf(
|
||||
Validator("DOCUMENT_CHUNK_SIZE", gte=1),
|
||||
Validator("DOCUMENT_PARSE_TIMEOUT_SECONDS", gte=1),
|
||||
Validator("DOCUMENT_OCR_TIMEOUT_SECONDS", gte=1),
|
||||
# DOCUMENT_OCR_MODE is normalised + membership-checked in
|
||||
# Settings.__post_init__ via _enum_fields (case-insensitive, like
|
||||
# DOCUMENT_OCR_PROVIDER) — no strict dynaconf Validator here, so
|
||||
# "Batch"/"SYNC" normalise instead of erroring.
|
||||
# Poll cadence well above a few seconds (each poll re-runs the tier);
|
||||
# deadline at least one poll interval.
|
||||
Validator("DOCUMENT_OCR_BATCH_POLL_SECONDS", gte=5),
|
||||
Validator("DOCUMENT_OCR_BATCH_MAX_WAIT_SECONDS", gte=60),
|
||||
Validator("DOCUMENT_PARSE_MEM_LIMIT_MB", gte=128),
|
||||
# 0 disables the pre-parse PDF size cap; otherwise it must be positive.
|
||||
Validator("DOCUMENT_MAX_PDF_SIZE_MB", gte=0),
|
||||
@@ -859,6 +881,13 @@ class Settings:
|
||||
# parse timeout / httpx.Timeout; per-tenant tunable so a gateway with a
|
||||
# shorter ceiling isn't masked by the 180s default.
|
||||
document_ocr_timeout_seconds: float = 180.0
|
||||
# OCR execution mode: "sync" | "batch" (Deck #332). batch is opt-in,
|
||||
# gateway-only, and used for large-corpus backfill; it falls back to sync when
|
||||
# no gateway backend resolves or the path can't defer (inline/memory pool).
|
||||
document_ocr_mode: str = "sync"
|
||||
# Batch-job poll cadence (procrastinate re-enqueue delay) and hard deadline.
|
||||
document_ocr_batch_poll_seconds: int = 120
|
||||
document_ocr_batch_max_wait_seconds: int = 86400
|
||||
# OCR escalation triggers (tier-0), per-tenant tunable. A page is OCR-worthy
|
||||
# if near-empty (< min_page_chars) OR low text-quality (< min_text_quality)
|
||||
# OR (when detect_scanned, image-analysis only runs when OCR is enabled)
|
||||
@@ -1003,6 +1032,7 @@ class Settings:
|
||||
"collection_metadata_source": {"qdrant", "api"},
|
||||
"document_tier1_engine": {"pypdfium2", "pymupdf"},
|
||||
"document_ocr_provider": {"auto", "gateway", "mistral", "none"},
|
||||
"document_ocr_mode": {"sync", "batch"},
|
||||
}
|
||||
for _field, _allowed in _enum_fields.items():
|
||||
_val = (getattr(self, _field) or "").strip().lower()
|
||||
@@ -1495,6 +1525,9 @@ def get_settings() -> Settings:
|
||||
"document_ocr_provider": "DOCUMENT_OCR_PROVIDER",
|
||||
"document_ocr_model": "DOCUMENT_OCR_MODEL",
|
||||
"document_ocr_timeout_seconds": "DOCUMENT_OCR_TIMEOUT_SECONDS",
|
||||
"document_ocr_mode": "DOCUMENT_OCR_MODE",
|
||||
"document_ocr_batch_poll_seconds": "DOCUMENT_OCR_BATCH_POLL_SECONDS",
|
||||
"document_ocr_batch_max_wait_seconds": "DOCUMENT_OCR_BATCH_MAX_WAIT_SECONDS",
|
||||
"document_ocr_min_text_quality": "DOCUMENT_OCR_MIN_TEXT_QUALITY",
|
||||
"document_ocr_page_fraction": "DOCUMENT_OCR_PAGE_FRACTION",
|
||||
"document_ocr_min_page_chars": "DOCUMENT_OCR_MIN_PAGE_CHARS",
|
||||
|
||||
@@ -93,3 +93,26 @@ class EscalateError(Exception):
|
||||
super().__init__(
|
||||
f"escalate {from_tier}->{to_tier} (reason={reason})",
|
||||
)
|
||||
|
||||
|
||||
class BatchPending(Exception):
|
||||
"""Raised when a tier's work is in flight on an async backend and the worker
|
||||
should poll again later (Deck #332 — batch OCR).
|
||||
|
||||
Like :class:`EscalateError` it is a **control-flow signal, NOT a failure**:
|
||||
the document's batch OCR job is still running on the gateway, so the OCR tier
|
||||
submits it (or polls an existing job) and raises this to ask the procrastinate
|
||||
retry strategy to re-run the SAME job on the SAME queue after ``retry_in``
|
||||
seconds — releasing the worker slot meanwhile so a multi-minute/hour batch
|
||||
doesn't pin a worker (and isn't reclaimed as a stalled ``doing`` job).
|
||||
|
||||
It must propagate untouched to the retry strategy: never swallowed by a broad
|
||||
``except Exception`` on the indexing path, never counted as a drop/parse
|
||||
error, and never marks the placeholder failed (the doc isn't done yet).
|
||||
Unlike ``EscalateError`` it does NOT change queue — the job stays on its own
|
||||
(``ocr``) tier queue and is simply deferred.
|
||||
"""
|
||||
|
||||
def __init__(self, *, retry_in: int) -> None:
|
||||
self.retry_in = retry_in
|
||||
super().__init__(f"batch OCR pending (retry_in={retry_in}s)")
|
||||
|
||||
@@ -18,9 +18,10 @@ Mistral (if ``MISTRAL_API_KEY``). Both return GitHub-flavoured markdown + exact
|
||||
|
||||
import base64
|
||||
import logging
|
||||
import time
|
||||
from abc import ABC, abstractmethod
|
||||
from collections.abc import Awaitable, Callable
|
||||
from typing import Any
|
||||
from typing import TYPE_CHECKING, Any
|
||||
|
||||
import anyio
|
||||
import httpx
|
||||
@@ -29,12 +30,25 @@ from nextcloud_mcp_server.config import Settings, get_settings
|
||||
|
||||
from .base import DocumentProcessor, ProcessingResult
|
||||
|
||||
if TYPE_CHECKING:
|
||||
# Annotation-only import (the runtime import is lazy, inside
|
||||
# build_gateway_batch_client, to avoid a document_processors -> embedding
|
||||
# cycle at load).
|
||||
from ..embedding.gateway_batch_client import GatewayBatchOcrClient
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
# Connect timeout for the OCR backend request. The overall (read) timeout is
|
||||
# configurable via DOCUMENT_OCR_TIMEOUT_SECONDS and resolved per call.
|
||||
_OCR_CONNECT_TIMEOUT_SECONDS = 10.0
|
||||
|
||||
# Sentinel keys on a ProcessingResult.metadata that mark "batch OCR job still in
|
||||
# flight — poll again later". The processor can't raise across the registry, so
|
||||
# it returns this sentinel and ``vector/processor._parse_pdf_tier`` translates it
|
||||
# into a ``BatchPending`` control-flow raise (same site as ``EscalateError``).
|
||||
OCR_BATCH_PENDING_KEY = "ocr_batch_pending"
|
||||
OCR_BATCH_RETRY_IN_KEY = "ocr_batch_retry_in"
|
||||
|
||||
|
||||
def _pages_to_text(
|
||||
pages: list[tuple[int, str]],
|
||||
@@ -64,6 +78,24 @@ def _pages_to_text(
|
||||
return "".join(parts), boundaries
|
||||
|
||||
|
||||
def _batch_identity(
|
||||
options: dict[str, Any] | None,
|
||||
) -> tuple[str, str, str, str] | None:
|
||||
"""Extract ``(user_id, doc_id, doc_type, etag)`` from the processor options
|
||||
the per-tier path threads in, or ``None`` if identity is absent (the inline
|
||||
pool, which can't defer a poll). ``etag`` may be empty (a file with no etag is
|
||||
still one tracked job keyed on "").
|
||||
"""
|
||||
if not options:
|
||||
return None
|
||||
user_id = options.get("user_id")
|
||||
doc_id = options.get("doc_id")
|
||||
doc_type = options.get("doc_type")
|
||||
if not user_id or not doc_id or not doc_type:
|
||||
return None
|
||||
return str(user_id), str(doc_id), str(doc_type), str(options.get("etag") or "")
|
||||
|
||||
|
||||
class _OcrBackend(ABC):
|
||||
@abstractmethod
|
||||
async def ocr(
|
||||
@@ -141,6 +173,54 @@ class _MistralOcrBackend(_OcrBackend):
|
||||
return _pages_to_text(pages)
|
||||
|
||||
|
||||
def _build_gateway_token_provider(settings: Settings) -> Any:
|
||||
"""Build the M2M ``GatewayTokenProvider`` from settings, or ``None`` when no
|
||||
client-id is configured (unauthenticated gateway). Shared by the sync OCR
|
||||
backend and the batch client so the M2M-triple validation lives in one place.
|
||||
"""
|
||||
if not settings.embedding_gateway_client_id:
|
||||
return None
|
||||
# Lazy import avoids a document_processors -> embedding cycle at load.
|
||||
from ..embedding.gateway_client import GatewayTokenProvider # noqa: PLC0415
|
||||
|
||||
# Explicit (not assert -- assert is stripped under `python -O`): the M2M
|
||||
# triple is all-or-nothing.
|
||||
if not settings.embedding_gateway_token_url:
|
||||
raise ValueError(
|
||||
"EMBEDDING_GATEWAY_TOKEN_URL is required when "
|
||||
"EMBEDDING_GATEWAY_CLIENT_ID is set"
|
||||
)
|
||||
if not settings.embedding_gateway_client_secret:
|
||||
raise ValueError(
|
||||
"EMBEDDING_GATEWAY_CLIENT_SECRET is required when "
|
||||
"EMBEDDING_GATEWAY_CLIENT_ID is set"
|
||||
)
|
||||
return GatewayTokenProvider(
|
||||
token_url=settings.embedding_gateway_token_url,
|
||||
client_id=settings.embedding_gateway_client_id,
|
||||
client_secret=settings.embedding_gateway_client_secret,
|
||||
scope=settings.embedding_gateway_scope,
|
||||
)
|
||||
|
||||
|
||||
def build_gateway_batch_client(settings: Settings) -> "GatewayBatchOcrClient | None":
|
||||
"""Build a ``GatewayBatchOcrClient`` when the gateway is the OCR backend, else
|
||||
``None`` (so batch mode falls back to sync for provider=mistral / no gateway).
|
||||
Batch OCR is gateway-only — Mistral's Batch API is reached *through* the
|
||||
gateway's batch routes, never directly from the pod."""
|
||||
if settings.document_ocr_provider not in ("gateway", "auto"):
|
||||
return None
|
||||
if not settings.embedding_gateway_url:
|
||||
return None
|
||||
from ..embedding.gateway_batch_client import GatewayBatchOcrClient # noqa: PLC0415
|
||||
|
||||
return GatewayBatchOcrClient(
|
||||
settings.embedding_gateway_url,
|
||||
settings.document_ocr_model,
|
||||
_build_gateway_token_provider(settings),
|
||||
)
|
||||
|
||||
|
||||
def build_ocr_backend(settings: Settings) -> _OcrBackend | None:
|
||||
"""Select an OCR backend from settings, or None when none is available."""
|
||||
provider = settings.document_ocr_provider
|
||||
@@ -148,33 +228,10 @@ def build_ocr_backend(settings: Settings) -> _OcrBackend | None:
|
||||
return None
|
||||
|
||||
if provider in ("gateway", "auto") and settings.embedding_gateway_url:
|
||||
token_provider = None
|
||||
if settings.embedding_gateway_client_id:
|
||||
# Lazy import avoids a document_processors -> embedding cycle at load.
|
||||
from ..embedding.gateway_client import ( # noqa: PLC0415
|
||||
GatewayTokenProvider,
|
||||
)
|
||||
|
||||
# Explicit (not assert -- assert is stripped under `python -O`): the
|
||||
# M2M triple is all-or-nothing.
|
||||
if not settings.embedding_gateway_token_url:
|
||||
raise ValueError(
|
||||
"EMBEDDING_GATEWAY_TOKEN_URL is required when "
|
||||
"EMBEDDING_GATEWAY_CLIENT_ID is set"
|
||||
)
|
||||
if not settings.embedding_gateway_client_secret:
|
||||
raise ValueError(
|
||||
"EMBEDDING_GATEWAY_CLIENT_SECRET is required when "
|
||||
"EMBEDDING_GATEWAY_CLIENT_ID is set"
|
||||
)
|
||||
token_provider = GatewayTokenProvider(
|
||||
token_url=settings.embedding_gateway_token_url,
|
||||
client_id=settings.embedding_gateway_client_id,
|
||||
client_secret=settings.embedding_gateway_client_secret,
|
||||
scope=settings.embedding_gateway_scope,
|
||||
)
|
||||
return _GatewayOcrBackend(
|
||||
settings.embedding_gateway_url, settings.document_ocr_model, token_provider
|
||||
settings.embedding_gateway_url,
|
||||
settings.document_ocr_model,
|
||||
_build_gateway_token_provider(settings),
|
||||
)
|
||||
|
||||
if provider in ("mistral", "auto") and settings.mistral_api_key:
|
||||
@@ -215,6 +272,14 @@ class OcrProcessor(DocumentProcessor):
|
||||
# doesn't each build a backend (and fetch its own M2M token). Lazy-init:
|
||||
# anyio primitives must not be created at import time.
|
||||
self._backend_lock: anyio.Lock | None = None
|
||||
# Batch-mode (Deck #332): the gateway batch client is cached like the sync
|
||||
# backend so its GatewayTokenProvider keeps its M2M-token cache across
|
||||
# documents. ``_batch_fallback_warned`` rate-limits the "can't batch,
|
||||
# using sync" warning to once per pod.
|
||||
self._batch_client_resolved = False
|
||||
self._batch_client: GatewayBatchOcrClient | None = None
|
||||
self._batch_client_lock: anyio.Lock | None = None
|
||||
self._batch_fallback_warned = False
|
||||
|
||||
@property
|
||||
def name(self) -> str:
|
||||
@@ -239,6 +304,24 @@ class OcrProcessor(DocumentProcessor):
|
||||
) = None,
|
||||
) -> ProcessingResult:
|
||||
settings = get_settings()
|
||||
|
||||
# Batch mode (Deck #332): submit to the gateway's async Batch OCR job and
|
||||
# poll across procrastinate retries. Returns a result (incl. the
|
||||
# "pending" sentinel) when handled, or None to fall back to the
|
||||
# synchronous path below (no gateway backend, or no per-doc identity — the
|
||||
# inline/memory pool can't defer a poll).
|
||||
#
|
||||
# A transport error from _process_batch (e.g. the gateway briefly down)
|
||||
# is intentionally NOT caught here: it propagates to procrastinate for a
|
||||
# durable retry rather than silently falling back to sync. If you've opted
|
||||
# into batch mode you want the retry, not an unexpected sync transcription.
|
||||
if settings.document_ocr_mode == "batch":
|
||||
batch_result = await self._process_batch(
|
||||
content, content_type, filename, options, settings
|
||||
)
|
||||
if batch_result is not None:
|
||||
return batch_result
|
||||
|
||||
if not self._backend_resolved:
|
||||
if self._backend_lock is None:
|
||||
self._backend_lock = anyio.Lock()
|
||||
@@ -301,6 +384,171 @@ class OcrProcessor(DocumentProcessor):
|
||||
processor=self.name,
|
||||
)
|
||||
|
||||
async def _get_batch_client(self) -> "GatewayBatchOcrClient | None":
|
||||
"""Cached gateway batch client (or ``None`` when batch isn't applicable —
|
||||
provider=mistral / no gateway). Resolved once under the backend lock so the
|
||||
token provider's M2M cache survives across documents."""
|
||||
if not self._batch_client_resolved:
|
||||
if self._batch_client_lock is None:
|
||||
self._batch_client_lock = anyio.Lock()
|
||||
async with self._batch_client_lock:
|
||||
if not self._batch_client_resolved: # double-checked
|
||||
self._batch_client = build_gateway_batch_client(get_settings())
|
||||
self._batch_client_resolved = True
|
||||
return self._batch_client
|
||||
|
||||
def _batch_fallback(self, reason: str, filename: str | None) -> None:
|
||||
"""Warn once that batch mode is falling back to the synchronous path."""
|
||||
if not self._batch_fallback_warned:
|
||||
logger.warning(
|
||||
"DOCUMENT_OCR_MODE=batch but %s; falling back to synchronous OCR",
|
||||
reason,
|
||||
)
|
||||
self._batch_fallback_warned = True
|
||||
|
||||
async def _process_batch(
|
||||
self,
|
||||
content: bytes,
|
||||
content_type: str,
|
||||
filename: str | None,
|
||||
options: dict[str, Any] | None,
|
||||
settings: Settings,
|
||||
) -> ProcessingResult | None:
|
||||
"""Submit + poll a one-document batch OCR job.
|
||||
|
||||
Returns a :class:`ProcessingResult` when batch handled the document — the
|
||||
terminal success/failure result, or the *pending sentinel* (``success=False``
|
||||
+ ``OCR_BATCH_PENDING_KEY`` metadata) that ``_parse_pdf_tier`` turns into a
|
||||
``BatchPending`` re-poll. Returns ``None`` to fall back to synchronous OCR
|
||||
(no gateway backend, or no per-doc identity — the inline pool can't defer).
|
||||
"""
|
||||
# Per-doc identity is threaded via ``options`` only on the per-tier
|
||||
# procrastinate path; the inline/memory pool omits it and can't defer a
|
||||
# poll, so batch is inapplicable there.
|
||||
identity = _batch_identity(options)
|
||||
if identity is None:
|
||||
self._batch_fallback("no per-document identity (inline path)", filename)
|
||||
return None
|
||||
client = await self._get_batch_client()
|
||||
if client is None:
|
||||
self._batch_fallback(
|
||||
"no gateway backend (provider=mistral or EMBEDDING_GATEWAY_URL unset)",
|
||||
filename,
|
||||
)
|
||||
return None
|
||||
|
||||
# Lazy import: keep the vector/DB stack off the document_processors load
|
||||
# path (mirrors the EscalateError lazy import in vector/processor).
|
||||
from ..vector.batch_ocr_store import BatchOcrJobStore # noqa: PLC0415
|
||||
|
||||
user_id, doc_id, doc_type, etag = identity
|
||||
store = await BatchOcrJobStore.shared()
|
||||
mime = content_type.split(";")[0].strip().lower()
|
||||
poll_seconds = settings.document_ocr_batch_poll_seconds
|
||||
|
||||
job = await store.get(
|
||||
user_id=user_id, doc_id=doc_id, doc_type=doc_type, etag=etag
|
||||
)
|
||||
if job is None:
|
||||
# New submission. Drop any superseded-version rows for this doc first
|
||||
# (a re-edited file changes etag) — a no-op on the very first submit,
|
||||
# one cheap DELETE on a resubmit. Then submit + record.
|
||||
await store.delete_stale_for_doc(
|
||||
user_id=user_id, doc_id=doc_id, doc_type=doc_type, keep_etag=etag
|
||||
)
|
||||
job_id = await client.submit(content, mime, custom_id=doc_id)
|
||||
await store.insert_pending(
|
||||
user_id=user_id,
|
||||
doc_id=doc_id,
|
||||
doc_type=doc_type,
|
||||
etag=etag,
|
||||
job_id=job_id,
|
||||
)
|
||||
logger.info(
|
||||
"batch OCR job submitted for %s (job_id=%s); deferring poll",
|
||||
filename or doc_id,
|
||||
job_id,
|
||||
)
|
||||
return self._pending(poll_seconds)
|
||||
|
||||
# Existing job — poll the gateway.
|
||||
result = await client.poll(job.job_id)
|
||||
if result.is_pending:
|
||||
elapsed = int(time.time()) - job.submitted_at
|
||||
if elapsed >= settings.document_ocr_batch_max_wait_seconds:
|
||||
await store.delete(
|
||||
user_id=user_id, doc_id=doc_id, doc_type=doc_type, etag=etag
|
||||
)
|
||||
# We don't cancel the gateway-side job (there's no cancel endpoint
|
||||
# at this layer) — it keeps running and is reaped by the gateway's
|
||||
# own file purge. Dropping the row just stops us polling it.
|
||||
logger.warning(
|
||||
"batch OCR job %s exceeded max wait (%ss); marking failed",
|
||||
job.job_id,
|
||||
settings.document_ocr_batch_max_wait_seconds,
|
||||
)
|
||||
return ProcessingResult(
|
||||
text="",
|
||||
metadata={"parse_failed_reason": "timeout"},
|
||||
processor=self.name,
|
||||
success=False,
|
||||
error="batch OCR timed out",
|
||||
)
|
||||
return self._pending(poll_seconds)
|
||||
|
||||
# Terminal — drop the tracking row either way.
|
||||
await store.delete(user_id=user_id, doc_id=doc_id, doc_type=doc_type, etag=etag)
|
||||
if result.is_failed:
|
||||
logger.warning(
|
||||
"batch OCR job %s failed: %s", job.job_id, result.error or "unknown"
|
||||
)
|
||||
return ProcessingResult(
|
||||
text="",
|
||||
metadata={"parse_failed_reason": "error"},
|
||||
processor=self.name,
|
||||
success=False,
|
||||
error=f"batch OCR failed: {result.error or 'unknown'}",
|
||||
)
|
||||
if not result.is_succeeded:
|
||||
# Defensive: poll() maps anything that isn't "succeeded" to its raw
|
||||
# status, and only pending/succeeded/failed are handled above. An
|
||||
# unexpected terminal status (gateway version skew, a new lifecycle
|
||||
# state) must NOT fall through to _pages_to_text([]) -> a 0-chunk
|
||||
# "success" that silently indexes empty text and re-submits forever.
|
||||
logger.warning(
|
||||
"batch OCR job %s returned unexpected status %r; marking failed",
|
||||
job.job_id,
|
||||
result.status,
|
||||
)
|
||||
return ProcessingResult(
|
||||
text="",
|
||||
metadata={"parse_failed_reason": "error"},
|
||||
processor=self.name,
|
||||
success=False,
|
||||
error=f"unexpected batch status: {result.status}",
|
||||
)
|
||||
text, boundaries = _pages_to_text(result.pages)
|
||||
return ProcessingResult(
|
||||
text=text,
|
||||
metadata={
|
||||
"page_count": len(boundaries),
|
||||
"page_boundaries": boundaries,
|
||||
"file_size": len(content),
|
||||
},
|
||||
processor=self.name,
|
||||
)
|
||||
|
||||
def _pending(self, retry_in: int) -> ProcessingResult:
|
||||
"""The pending sentinel — ``_parse_pdf_tier`` raises ``BatchPending`` from
|
||||
it. ``success=False`` keeps it out of the index path, and the sentinel key
|
||||
keeps it out of the parse-failed path (it isn't a failure)."""
|
||||
return ProcessingResult(
|
||||
text="",
|
||||
metadata={OCR_BATCH_PENDING_KEY: True, OCR_BATCH_RETRY_IN_KEY: retry_in},
|
||||
processor=self.name,
|
||||
success=False,
|
||||
)
|
||||
|
||||
async def health_check(self) -> bool:
|
||||
# Backends are resolved lazily (and configured per tenant), so there is
|
||||
# nothing to probe here without making a billable upstream call -- the
|
||||
|
||||
@@ -0,0 +1,199 @@
|
||||
"""Client for the embedding gateway's async **batch OCR** routes (Deck #332).
|
||||
|
||||
The gateway exposes two batch routes alongside the synchronous ``POST /v1/ocr``
|
||||
(astrolabe-cloud-website#372):
|
||||
|
||||
- ``POST /v1/ocr/batch`` — submit N documents (each with a caller ``custom_id``)
|
||||
as one Mistral Batch job; returns ``202`` + a namespaced ``job_id``
|
||||
(``<provider>/<batch_job_id>``).
|
||||
- ``GET /v1/ocr/batch/{job_id}`` — poll; returns the lifecycle status and, once
|
||||
terminal, per-document results (per-page markdown, or a per-document error).
|
||||
|
||||
The gateway is a **stateless passthrough** to Mistral's Batch API — the
|
||||
``job_id`` is the only handle, so the worker persists it (see
|
||||
``vector/batch_ocr_store``) and re-polls across procrastinate retries.
|
||||
|
||||
This client submits exactly **one document per job** (the v1 unit; coalescing N
|
||||
docs/job is a follow-up). Auth + ``/v1`` base-url handling mirror the synchronous
|
||||
:class:`~nextcloud_mcp_server.embedding.gateway_client.GatewayProvider` /
|
||||
``_GatewayOcrBackend`` — same M2M :class:`GatewayTokenProvider` bearer, no
|
||||
provider keys in the pod.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import base64
|
||||
import logging
|
||||
from dataclasses import dataclass
|
||||
from typing import Any
|
||||
|
||||
import httpx
|
||||
|
||||
from .gateway_client import GatewayTokenProvider
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
# Connect timeout for the (cheap) submit/poll calls. These are control-plane-ish
|
||||
# requests — a submit returns immediately with a job id and a poll is a status
|
||||
# read — so they get a short, fixed timeout, NOT the document-OCR read timeout
|
||||
# (which sizes a synchronous transcription).
|
||||
_BATCH_CONNECT_TIMEOUT_SECONDS = 5.0
|
||||
_BATCH_REQUEST_TIMEOUT_SECONDS = 30.0
|
||||
|
||||
# Gateway-normalised batch lifecycle (OcrBatchStatus on the gateway side).
|
||||
_PENDING = "pending"
|
||||
_SUCCEEDED = "succeeded"
|
||||
_FAILED = "failed"
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class BatchPollResult:
|
||||
"""One poll of a batch OCR job.
|
||||
|
||||
``status`` is the gateway-normalised lifecycle (``pending`` | ``succeeded`` |
|
||||
``failed``). For a single-document job: on ``succeeded`` ``pages`` holds the
|
||||
document's per-page ``(index, markdown)`` (empty + ``error`` set if that one
|
||||
document errored inside an otherwise-successful job); on ``failed`` ``error``
|
||||
carries the job-level failure.
|
||||
"""
|
||||
|
||||
status: str
|
||||
pages: list[tuple[int, str]]
|
||||
error: str | None = None
|
||||
|
||||
@property
|
||||
def is_pending(self) -> bool:
|
||||
return self.status == _PENDING
|
||||
|
||||
@property
|
||||
def is_succeeded(self) -> bool:
|
||||
return self.status == _SUCCEEDED
|
||||
|
||||
@property
|
||||
def is_failed(self) -> bool:
|
||||
return self.status == _FAILED
|
||||
|
||||
|
||||
class GatewayBatchOcrClient:
|
||||
"""Submits + polls single-document batch OCR jobs against the gateway."""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
base_url: str,
|
||||
model: str,
|
||||
token_provider: GatewayTokenProvider | None = None,
|
||||
) -> None:
|
||||
# EMBEDDING_GATEWAY_URL is a bare origin; the batch routes live under /v1
|
||||
# like the rest of the gateway API. Idempotent if already /v1-suffixed.
|
||||
base = base_url.rstrip("/")
|
||||
if not base.endswith("/v1"):
|
||||
base = f"{base}/v1"
|
||||
self._base = base
|
||||
self._model = model
|
||||
self._token_provider = token_provider
|
||||
|
||||
async def _headers(self) -> dict[str, str]:
|
||||
if self._token_provider is None:
|
||||
return {}
|
||||
return {"Authorization": f"Bearer {await self._token_provider.get_token()}"}
|
||||
|
||||
async def submit(self, content: bytes, mime_type: str, custom_id: str) -> str:
|
||||
"""Submit ``content`` as a one-document batch job; return the namespaced
|
||||
``job_id`` to persist + poll. Raises on transport / non-2xx."""
|
||||
payload = {
|
||||
"model": self._model,
|
||||
"documents": [
|
||||
{
|
||||
"custom_id": custom_id,
|
||||
"mime_type": mime_type,
|
||||
"document_b64": base64.b64encode(content).decode("ascii"),
|
||||
}
|
||||
],
|
||||
}
|
||||
async with httpx.AsyncClient(
|
||||
timeout=httpx.Timeout(
|
||||
_BATCH_REQUEST_TIMEOUT_SECONDS, connect=_BATCH_CONNECT_TIMEOUT_SECONDS
|
||||
)
|
||||
) as client:
|
||||
resp = await client.post(
|
||||
f"{self._base}/ocr/batch", json=payload, headers=await self._headers()
|
||||
)
|
||||
resp.raise_for_status()
|
||||
body = resp.json()
|
||||
job_id = body.get("job_id")
|
||||
if not job_id:
|
||||
# Contract violation (2xx without a job id) — fail with an actionable
|
||||
# message rather than a bare KeyError deep in the caller.
|
||||
raise ValueError(f"gateway batch submit returned no job_id: {body!r}")
|
||||
logger.info(
|
||||
"batch OCR submitted: job_id=%s custom_id=%s status=%s",
|
||||
job_id,
|
||||
custom_id,
|
||||
body.get("status"),
|
||||
)
|
||||
return job_id
|
||||
|
||||
async def poll(self, job_id: str) -> BatchPollResult:
|
||||
"""Poll a batch job. Raises on transport / non-2xx; maps a terminal job's
|
||||
single-document result into :class:`BatchPollResult`.
|
||||
|
||||
``job_id`` is the gateway's namespaced id (``<provider>/<batch_job_id>``),
|
||||
so it embeds a ``/`` and the request path is multi-segment
|
||||
(``/v1/ocr/batch/mistral/job-1``). The gateway declares this route with a
|
||||
path-capture parameter (``GET /v1/ocr/batch/{job_id:path}``) so the slash
|
||||
is captured whole — a plain single-segment ``{job_id}`` would 404 here.
|
||||
"""
|
||||
async with httpx.AsyncClient(
|
||||
timeout=httpx.Timeout(
|
||||
_BATCH_REQUEST_TIMEOUT_SECONDS, connect=_BATCH_CONNECT_TIMEOUT_SECONDS
|
||||
)
|
||||
) as client:
|
||||
resp = await client.get(
|
||||
f"{self._base}/ocr/batch/{job_id}", headers=await self._headers()
|
||||
)
|
||||
resp.raise_for_status()
|
||||
body = resp.json()
|
||||
status = body.get("status")
|
||||
if status is None:
|
||||
# A well-formed gateway response always carries status. A 2xx without
|
||||
# it is a contract violation: fail fast rather than silently treating
|
||||
# it as pending and re-polling until the deadline.
|
||||
logger.warning("gateway batch poll returned no status: %r", body)
|
||||
return BatchPollResult(
|
||||
status=_FAILED, pages=[], error="gateway returned no status"
|
||||
)
|
||||
if status != _SUCCEEDED:
|
||||
# pending: nothing to read yet. failed: surface the job-level error.
|
||||
return BatchPollResult(status=status, pages=[], error=body.get("error"))
|
||||
return _result_from_success(body)
|
||||
|
||||
|
||||
def _result_from_success(body: dict[str, Any]) -> BatchPollResult:
|
||||
"""Extract the single document's pages from a succeeded job's results.
|
||||
|
||||
Submitting one document per job means exactly one result item; defensively
|
||||
take the first. A per-document error inside a succeeded job (the document
|
||||
failed but the job didn't) surfaces as a failed poll so the caller marks the
|
||||
doc parse-failed rather than indexing empty text.
|
||||
"""
|
||||
results = body.get("results") or []
|
||||
if not results:
|
||||
return BatchPollResult(
|
||||
status=_FAILED,
|
||||
pages=[],
|
||||
error="batch job succeeded but returned no results",
|
||||
)
|
||||
item = results[0]
|
||||
# ``not item.get("pages")`` catches both a missing key AND an empty list:
|
||||
# a succeeded job that produced zero pages is a per-document failure (nothing
|
||||
# to index), not a silent 0-chunk success.
|
||||
if item.get("error") is not None or not item.get("pages"):
|
||||
return BatchPollResult(
|
||||
status=_FAILED, pages=[], error=item.get("error") or "no pages returned"
|
||||
)
|
||||
# Defensive on both fields (the page index falls back to position) so a
|
||||
# malformed page object degrades rather than raising KeyError mid-parse.
|
||||
pages = [
|
||||
(p.get("index", i), p.get("markdown", "")) for i, p in enumerate(item["pages"])
|
||||
]
|
||||
return BatchPollResult(status=_SUCCEEDED, pages=pages)
|
||||
@@ -0,0 +1,128 @@
|
||||
"""Tracking store for in-flight async batch OCR jobs (Deck #332).
|
||||
|
||||
When ``DOCUMENT_OCR_MODE=batch`` the OCR tier submits a document to the gateway's
|
||||
async batch route and then re-polls across procrastinate retries. procrastinate
|
||||
job args are immutable, so the gateway ``job_id`` (and submit time, for the poll
|
||||
deadline) live in the ``batch_ocr_jobs`` app-DB table, keyed on the document +
|
||||
its content version (``etag``).
|
||||
|
||||
Engine reuse mirrors :class:`~nextcloud_mcp_server.usage.store.UsageEventStore`:
|
||||
rather than open its own engine this store borrows the process-wide
|
||||
:class:`RefreshTokenStorage` singleton (``get_shared_storage()``) — same app DB,
|
||||
dialect handling, ``?``-placeholder shim, and the guarantee that Alembic
|
||||
migrations (incl. ``batch_ocr_jobs``) already ran.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
import time
|
||||
from dataclasses import dataclass
|
||||
|
||||
import anyio
|
||||
|
||||
from nextcloud_mcp_server.auth.storage import RefreshTokenStorage, get_shared_storage
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class BatchOcrJob:
|
||||
"""A tracked in-flight batch OCR job. A row exists only while pending (terminal
|
||||
jobs are deleted), so there's no stored status — the live status comes from a
|
||||
fresh ``GatewayBatchOcrClient.poll``. ``submitted_at`` anchors the deadline."""
|
||||
|
||||
job_id: str
|
||||
submitted_at: int
|
||||
|
||||
|
||||
class BatchOcrJobStore:
|
||||
"""CRUD for the ``batch_ocr_jobs`` table (one row per in-flight job)."""
|
||||
|
||||
_shared_instance: BatchOcrJobStore | None = None
|
||||
# Lazy-init: anyio primitives must not be created at import time (CLAUDE.md;
|
||||
# mirrors OcrProcessor._backend_lock). Created on first shared() call.
|
||||
_shared_lock: anyio.Lock | None = None
|
||||
|
||||
def __init__(self, storage: RefreshTokenStorage) -> None:
|
||||
self._storage = storage
|
||||
|
||||
@classmethod
|
||||
async def shared(cls) -> BatchOcrJobStore:
|
||||
"""Process-wide store backed by the storage singleton. Tests should
|
||||
construct ``BatchOcrJobStore(storage)`` directly — the cache is a process
|
||||
global with no teardown hook."""
|
||||
# No await between the None-check and the assignment, so this is atomic
|
||||
# within the single event loop (anyio is cooperative) — two cold-start
|
||||
# callers can't both create a lock.
|
||||
if cls._shared_lock is None:
|
||||
cls._shared_lock = anyio.Lock()
|
||||
async with cls._shared_lock:
|
||||
if cls._shared_instance is None:
|
||||
cls._shared_instance = cls(await get_shared_storage())
|
||||
return cls._shared_instance
|
||||
|
||||
async def get(
|
||||
self, *, user_id: str, doc_id: str, doc_type: str, etag: str
|
||||
) -> BatchOcrJob | None:
|
||||
"""The in-flight job for this document+version, or ``None``."""
|
||||
async with self._storage.acquire() as db:
|
||||
async with db.execute(
|
||||
"SELECT job_id, submitted_at FROM batch_ocr_jobs "
|
||||
"WHERE user_id = ? AND doc_id = ? AND doc_type = ? AND etag = ?",
|
||||
(user_id, doc_id, doc_type, etag),
|
||||
) as cursor:
|
||||
row = await cursor.fetchone()
|
||||
if row is None:
|
||||
return None
|
||||
return BatchOcrJob(job_id=row[0], submitted_at=int(row[1]))
|
||||
|
||||
async def insert_pending(
|
||||
self,
|
||||
*,
|
||||
user_id: str,
|
||||
doc_id: str,
|
||||
doc_type: str,
|
||||
etag: str,
|
||||
job_id: str,
|
||||
submitted_at: int | None = None,
|
||||
) -> None:
|
||||
"""Record a freshly-submitted job. ``ON CONFLICT DO NOTHING`` makes a
|
||||
racing double-submit harmless (the first row wins; the loser's job id is
|
||||
abandoned and reaped by the gateway-side file purge)."""
|
||||
now = submitted_at if submitted_at is not None else int(time.time())
|
||||
async with self._storage.acquire() as db:
|
||||
await db.execute(
|
||||
"INSERT INTO batch_ocr_jobs "
|
||||
"(user_id, doc_id, doc_type, etag, job_id, submitted_at) "
|
||||
"VALUES (?, ?, ?, ?, ?, ?) "
|
||||
"ON CONFLICT (user_id, doc_id, doc_type, etag) DO NOTHING",
|
||||
(user_id, doc_id, doc_type, etag, job_id, now),
|
||||
)
|
||||
await db.commit()
|
||||
|
||||
async def delete(
|
||||
self, *, user_id: str, doc_id: str, doc_type: str, etag: str
|
||||
) -> None:
|
||||
"""Drop the row once the job is terminal (succeeded or failed)."""
|
||||
async with self._storage.acquire() as db:
|
||||
await db.execute(
|
||||
"DELETE FROM batch_ocr_jobs "
|
||||
"WHERE user_id = ? AND doc_id = ? AND doc_type = ? AND etag = ?",
|
||||
(user_id, doc_id, doc_type, etag),
|
||||
)
|
||||
await db.commit()
|
||||
|
||||
async def delete_stale_for_doc(
|
||||
self, *, user_id: str, doc_id: str, doc_type: str, keep_etag: str
|
||||
) -> None:
|
||||
"""Remove superseded-version rows for a document (any etag other than the
|
||||
current one) before a resubmit, so a re-edited file doesn't leave its
|
||||
old in-flight job tracked forever."""
|
||||
async with self._storage.acquire() as db:
|
||||
await db.execute(
|
||||
"DELETE FROM batch_ocr_jobs "
|
||||
"WHERE user_id = ? AND doc_id = ? AND doc_type = ? AND etag != ?",
|
||||
(user_id, doc_id, doc_type, keep_etag),
|
||||
)
|
||||
await db.commit()
|
||||
@@ -126,6 +126,7 @@ async def _parse_pdf_tier(
|
||||
filename: str | None,
|
||||
tier: str,
|
||||
settings: Any,
|
||||
options: dict[str, Any] | None = None,
|
||||
) -> "ProcessingResult":
|
||||
"""Run a single extraction tier and apply the post-parse escalation gate.
|
||||
|
||||
@@ -141,18 +142,31 @@ async def _parse_pdf_tier(
|
||||
the "OCR is an enhancement, never worse than off" invariant: a tenant who has
|
||||
not enabled a higher tier (or has no processor for it) simply indexes the
|
||||
cheap tier's output.
|
||||
|
||||
Batch OCR (Deck #332): when the OCR tier's batch job is still in flight the
|
||||
processor returns a *pending sentinel* result; we translate it here into a
|
||||
``BatchPending`` raise (same decision point as ``EscalateError``) so the
|
||||
retry strategy re-runs this tier after a delay instead of indexing empty text.
|
||||
"""
|
||||
# Lazy import: keep the document stack (pymupdf/_isolation) off the module
|
||||
# load path; this runs only on the per-tier worker, which needs it anyway.
|
||||
from nextcloud_mcp_server.document_processors.escalation import ( # noqa: PLC0415
|
||||
BatchPending,
|
||||
EscalateError,
|
||||
)
|
||||
from nextcloud_mcp_server.document_processors.ocr import ( # noqa: PLC0415
|
||||
OCR_BATCH_PENDING_KEY,
|
||||
OCR_BATCH_RETRY_IN_KEY,
|
||||
)
|
||||
|
||||
# options / progress_callback are not threaded here -- the indexing caller
|
||||
# passes neither today, and the inline path (registry.process) omits them
|
||||
# too. Forward them if a tier processor ever needs per-call tuning (e.g. OCR
|
||||
# DPI); keeping the two paths symmetric until then.
|
||||
result = await registry.process_tier(content, content_type, filename, tier)
|
||||
# ``options`` threads per-document identity (user_id/doc_id/doc_type/etag) to
|
||||
# the OCR tier so batch mode can key its job-tracking table (Deck #332). Other
|
||||
# tiers ignore it. The inline path (registry.process) passes None.
|
||||
result = await registry.process_tier(
|
||||
content, content_type, filename, tier, options=options
|
||||
)
|
||||
if result.metadata.get(OCR_BATCH_PENDING_KEY):
|
||||
raise BatchPending(retry_in=int(result.metadata[OCR_BATCH_RETRY_IN_KEY]))
|
||||
if result.success:
|
||||
decision = registry.evaluate_escalation(
|
||||
result, content, tier, settings, filename=filename
|
||||
@@ -525,19 +539,21 @@ async def process_document(
|
||||
re-picked on the next scan; the procrastinate path (max_retries=1) caps it at
|
||||
one outer attempt (~30s) and defers. Don't stack a third retry layer here.
|
||||
"""
|
||||
# EscalateError is a control-flow signal that arises ONLY on the per-tier
|
||||
# external path (tier set). Bind the class lazily here, and only when a tier
|
||||
# is set, so the document stack is never imported at *module load* (the #877
|
||||
# invariant) nor on the delete / text-doc call paths (file processing already
|
||||
# imports it via get_registry regardless). When tier is None it can't be
|
||||
# raised, so the guards below stay inert.
|
||||
escalate_error_cls: type[BaseException] | None = None
|
||||
# EscalateError and BatchPending are control-flow signals that arise ONLY on
|
||||
# the per-tier external path (tier set). Bind them lazily here, and only when a
|
||||
# tier is set, so the document stack is never imported at *module load* (the
|
||||
# #877 invariant) nor on the delete / text-doc call paths (file processing
|
||||
# already imports them via get_registry regardless). When tier is None neither
|
||||
# can be raised, so the guards below stay inert. Bound as a tuple so the guards
|
||||
# treat both identically: propagate untouched, never record an error/drop.
|
||||
control_flow_excs: tuple[type[BaseException], ...] = ()
|
||||
if tier is not None:
|
||||
from nextcloud_mcp_server.document_processors.escalation import ( # noqa: PLC0415
|
||||
BatchPending,
|
||||
EscalateError,
|
||||
)
|
||||
|
||||
escalate_error_cls = EscalateError
|
||||
control_flow_excs = (EscalateError, BatchPending)
|
||||
|
||||
start_time = time.time()
|
||||
|
||||
@@ -632,13 +648,11 @@ async def process_document(
|
||||
return # Success
|
||||
|
||||
except Exception as e:
|
||||
# An escalation signal is control flow, not a failure:
|
||||
# propagate it untouched so the procrastinate retry strategy
|
||||
# can hop the job to the next tier's queue. Never retry it
|
||||
# A control-flow signal (escalation hop, or batch-OCR re-poll
|
||||
# deferral) is not a failure: propagate it untouched so the
|
||||
# procrastinate retry strategy handles it. Never retry it
|
||||
# in-process and never count it as a drop.
|
||||
if escalate_error_cls is not None and isinstance(
|
||||
e, escalate_error_cls
|
||||
):
|
||||
if isinstance(e, control_flow_excs):
|
||||
raise
|
||||
if attempt < max_retries - 1:
|
||||
logger.warning(
|
||||
@@ -691,10 +705,11 @@ async def process_document(
|
||||
raise
|
||||
|
||||
except Exception as e:
|
||||
# An escalation signal must reach the procrastinate retry strategy
|
||||
# un-recorded -- it is neither a processing success nor an error
|
||||
# (the hop is its own event, counted via record_document_escalation).
|
||||
if escalate_error_cls is not None and isinstance(e, escalate_error_cls):
|
||||
# A control-flow signal must reach the procrastinate retry strategy
|
||||
# un-recorded -- it is neither a processing success nor an error (an
|
||||
# escalation hop is counted via record_document_escalation; a batch
|
||||
# re-poll deferral is not an event at all).
|
||||
if isinstance(e, control_flow_excs):
|
||||
raise
|
||||
# Single processing-error call site: catches exhausted-retry
|
||||
# re-raises, delete failures, and setup errors (get_qdrant_client /
|
||||
@@ -956,6 +971,7 @@ async def _index_document(
|
||||
get_registry,
|
||||
)
|
||||
from nextcloud_mcp_server.document_processors.escalation import ( # noqa: PLC0415
|
||||
BatchPending,
|
||||
EscalateError,
|
||||
)
|
||||
|
||||
@@ -968,6 +984,16 @@ async def _index_document(
|
||||
# and the in-process/memory pool (tier is None) -- runs the inline
|
||||
# tiered pipeline (fast -> OCR escalation in one call).
|
||||
if tier is not None and _is_pdf(content_type):
|
||||
# Per-document identity, forwarded to every tier's processor.
|
||||
# Only the OCR tier reads it (batch mode keys its job-tracking
|
||||
# table on it, Deck #332); fast/structured ignore it, so it's
|
||||
# safe to pass on all tiers.
|
||||
doc_identity_options = {
|
||||
"user_id": doc_task.user_id,
|
||||
"doc_id": doc_task.doc_id,
|
||||
"doc_type": doc_task.doc_type,
|
||||
"etag": doc_task.etag or "",
|
||||
}
|
||||
result = await _parse_pdf_tier(
|
||||
registry,
|
||||
content_bytes,
|
||||
@@ -975,6 +1001,7 @@ async def _index_document(
|
||||
file_path,
|
||||
tier,
|
||||
settings,
|
||||
options=doc_identity_options,
|
||||
)
|
||||
else:
|
||||
result = await registry.process(
|
||||
@@ -1047,10 +1074,12 @@ async def _index_document(
|
||||
)
|
||||
else:
|
||||
logger.debug("No page_boundaries in metadata for %s", file_path)
|
||||
except EscalateError:
|
||||
# Control-flow signal (per-tier path): re-raise untouched so the
|
||||
# queue hops the job to the next tier. NOT a "failed to process"
|
||||
# error -- don't log it as one.
|
||||
except (EscalateError, BatchPending):
|
||||
# Control-flow signals (per-tier path): re-raise untouched.
|
||||
# EscalateError hops the job to the next tier; BatchPending defers
|
||||
# a re-poll on the same tier (batch OCR still in flight, Deck
|
||||
# #332). Neither is a "failed to process" error -- don't log them
|
||||
# as one.
|
||||
raise
|
||||
except Exception as e:
|
||||
logger.error("Failed to process file %s: %s", file_path, e)
|
||||
|
||||
@@ -328,13 +328,26 @@ class TieredEscalationStrategy(BaseRetryStrategy):
|
||||
def get_retry_decision(
|
||||
self, *, exception: BaseException, job: Job
|
||||
) -> RetryDecision | None:
|
||||
# Lazy import: EscalateError lives in the document stack, which the API
|
||||
# pod (it also builds this App to defer) must not load. get_retry_decision
|
||||
# runs only in the worker, where the stack is already imported.
|
||||
from ...document_processors.escalation import EscalateError # noqa: PLC0415
|
||||
# Lazy import: these live in the document stack, which the API pod (it
|
||||
# also builds this App to defer) must not load. get_retry_decision runs
|
||||
# only in the worker, where the stack is already imported.
|
||||
from ...document_processors.escalation import ( # noqa: PLC0415
|
||||
BatchPending,
|
||||
EscalateError,
|
||||
)
|
||||
|
||||
exc = _first_leaf(exception)
|
||||
|
||||
if isinstance(exc, BatchPending):
|
||||
# Batch OCR job still in flight (Deck #332): defer a re-poll on the
|
||||
# SAME queue after retry_in seconds. Deliberately exempt from the
|
||||
# transient cap below — a batch job can take minutes-hours, so the
|
||||
# poll count is unbounded here; the OCR processor's own deadline
|
||||
# (DOCUMENT_OCR_BATCH_MAX_WAIT_SECONDS) is what terminates a stuck job.
|
||||
# Releasing the worker between polls keeps the job out of `doing`, so
|
||||
# it's never stall-reclaimed.
|
||||
return RetryDecision(retry_in={"seconds": exc.retry_in})
|
||||
|
||||
if isinstance(exc, EscalateError):
|
||||
queue = TIER_QUEUES.get(exc.to_tier)
|
||||
if queue is None:
|
||||
|
||||
@@ -13,10 +13,15 @@ from pathlib import Path
|
||||
import pytest
|
||||
from pact import Pact
|
||||
|
||||
# Pact participant names. These MUST match the names used on the astrolabe side
|
||||
# and in the broker, so keep them in sync with the astrolabe repo's pact tests.
|
||||
# Pact participant names. These MUST match the names used on the provider side
|
||||
# and in the broker, so keep them in sync with the provider repos' pact tests.
|
||||
CONSUMER = "nextcloud-mcp-server"
|
||||
PROVIDER = "astrolabe"
|
||||
# The embedding gateway is a *separate* provider (astrolabe-cloud-website,
|
||||
# services/embedding-gateway). Its provider-verification job
|
||||
# (test_gateway_provider_verification.py, PROVIDER_NAME="astrolabe-cloud-gateway")
|
||||
# picks up this consumer's pact from the broker.
|
||||
GATEWAY_PROVIDER = "astrolabe-cloud-gateway"
|
||||
|
||||
PACT_DIR = Path(__file__).parent / "pacts"
|
||||
|
||||
@@ -40,3 +45,16 @@ def consumer_pact():
|
||||
pact = Pact(CONSUMER, PROVIDER).with_specification("V4")
|
||||
yield pact
|
||||
pact.write_file(PACT_DIR, overwrite=False)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def gateway_consumer_pact():
|
||||
"""A fresh Pact (consumer=nextcloud-mcp-server, provider=astrolabe-cloud-gateway).
|
||||
|
||||
Separate from ``consumer_pact`` because the embedding gateway is a distinct
|
||||
provider — its interactions merge into their own pact file, verified by the
|
||||
gateway's provider job (Deck #332).
|
||||
"""
|
||||
pact = Pact(CONSUMER, GATEWAY_PROVIDER).with_specification("V4")
|
||||
yield pact
|
||||
pact.write_file(PACT_DIR, overwrite=False)
|
||||
|
||||
@@ -0,0 +1,145 @@
|
||||
"""Consumer contract: nextcloud-mcp-server -> embedding-gateway batch OCR (Deck #332).
|
||||
|
||||
When ``DOCUMENT_OCR_MODE=batch`` the ingest worker drives the gateway's async
|
||||
Batch OCR routes via :class:`GatewayBatchOcrClient`
|
||||
(``embedding/gateway_batch_client.py``):
|
||||
|
||||
- ``POST /v1/ocr/batch`` — submit one document, returns a namespaced ``job_id``.
|
||||
- ``GET /v1/ocr/batch/{job_id}`` — poll; pending until terminal, then per-page
|
||||
markdown (succeeded) or an error (failed).
|
||||
|
||||
This pact pins the request/response shapes the consumer depends on, for the
|
||||
``astrolabe-cloud-gateway`` provider (whose verification job lives in
|
||||
astrolabe-cloud-website, services/embedding-gateway). Only the fields the client
|
||||
actually reads are asserted, so the contract stays robust to additive response
|
||||
changes (the gateway's ``OcrBatchJobOut`` carries more fields — total/completed/
|
||||
counts — that the single-document client ignores).
|
||||
|
||||
The gateway is unauthenticated today, so no bearer is sent (matching the
|
||||
M2M-optional ``GatewayBatchOcrClient``). See ADR-029 for the contract-testing
|
||||
architecture.
|
||||
"""
|
||||
|
||||
import base64
|
||||
|
||||
import pytest
|
||||
from pact import match
|
||||
|
||||
from nextcloud_mcp_server.embedding.gateway_batch_client import GatewayBatchOcrClient
|
||||
|
||||
pytestmark = pytest.mark.contract
|
||||
|
||||
_MODEL = "mistral/mistral-ocr-latest"
|
||||
# A small, valid base64 PDF payload — the gateway base64-decodes + size-checks
|
||||
# the document, so the replayed request must carry decodable bytes.
|
||||
_PDF_B64 = base64.b64encode(b"%PDF-1.4 contract test").decode("ascii")
|
||||
|
||||
|
||||
async def test_submit_returns_namespaced_job_id(gateway_consumer_pact):
|
||||
(
|
||||
gateway_consumer_pact.upon_receiving("a batch OCR submission for one document")
|
||||
.given("the gateway accepts a batch OCR submission")
|
||||
.with_request("POST", "/v1/ocr/batch")
|
||||
.with_body(
|
||||
{
|
||||
"model": _MODEL,
|
||||
"documents": [
|
||||
{
|
||||
"custom_id": "0",
|
||||
"mime_type": "application/pdf",
|
||||
"document_b64": _PDF_B64,
|
||||
}
|
||||
],
|
||||
},
|
||||
content_type="application/json",
|
||||
)
|
||||
.will_respond_with(202)
|
||||
.with_body(
|
||||
{
|
||||
# Namespaced "<provider>/<batch_job_id>" — the only field submit() reads.
|
||||
"job_id": match.regex("mistral/job-abc", regex=r"[^/]+/.+"),
|
||||
"status": "pending",
|
||||
},
|
||||
content_type="application/json",
|
||||
)
|
||||
)
|
||||
|
||||
with gateway_consumer_pact.serve() as srv:
|
||||
client = GatewayBatchOcrClient(str(srv.url), _MODEL)
|
||||
job_id = await client.submit(
|
||||
b"%PDF-1.4 contract test", "application/pdf", custom_id="0"
|
||||
)
|
||||
|
||||
assert job_id == "mistral/job-abc"
|
||||
|
||||
|
||||
async def test_poll_pending(gateway_consumer_pact):
|
||||
(
|
||||
gateway_consumer_pact.upon_receiving("a poll for a still-running batch OCR job")
|
||||
.given("a pending batch OCR job mistral/job-pending exists")
|
||||
.with_request("GET", "/v1/ocr/batch/mistral/job-pending")
|
||||
.will_respond_with(200)
|
||||
.with_body({"status": "pending"}, content_type="application/json")
|
||||
)
|
||||
|
||||
with gateway_consumer_pact.serve() as srv:
|
||||
result = await GatewayBatchOcrClient(str(srv.url), _MODEL).poll(
|
||||
"mistral/job-pending"
|
||||
)
|
||||
|
||||
assert result.is_pending
|
||||
|
||||
|
||||
async def test_poll_succeeded_returns_pages(gateway_consumer_pact):
|
||||
(
|
||||
gateway_consumer_pact.upon_receiving("a poll for a succeeded batch OCR job")
|
||||
.given("a succeeded batch OCR job mistral/job-done exists")
|
||||
.with_request("GET", "/v1/ocr/batch/mistral/job-done")
|
||||
.will_respond_with(200)
|
||||
.with_body(
|
||||
{
|
||||
"status": "succeeded",
|
||||
"results": [
|
||||
{
|
||||
"custom_id": "0",
|
||||
"pages": [
|
||||
{
|
||||
"index": match.integer(0),
|
||||
"markdown": match.string("# Page one"),
|
||||
}
|
||||
],
|
||||
}
|
||||
],
|
||||
},
|
||||
content_type="application/json",
|
||||
)
|
||||
)
|
||||
|
||||
with gateway_consumer_pact.serve() as srv:
|
||||
result = await GatewayBatchOcrClient(str(srv.url), _MODEL).poll(
|
||||
"mistral/job-done"
|
||||
)
|
||||
|
||||
assert result.is_succeeded
|
||||
assert result.pages == [(0, "# Page one")]
|
||||
|
||||
|
||||
async def test_poll_failed_surfaces_error(gateway_consumer_pact):
|
||||
(
|
||||
gateway_consumer_pact.upon_receiving("a poll for a failed batch OCR job")
|
||||
.given("a failed batch OCR job mistral/job-failed exists")
|
||||
.with_request("GET", "/v1/ocr/batch/mistral/job-failed")
|
||||
.will_respond_with(200)
|
||||
.with_body(
|
||||
{"status": "failed", "error": match.string("batch job failed")},
|
||||
content_type="application/json",
|
||||
)
|
||||
)
|
||||
|
||||
with gateway_consumer_pact.serve() as srv:
|
||||
result = await GatewayBatchOcrClient(str(srv.url), _MODEL).poll(
|
||||
"mistral/job-failed"
|
||||
)
|
||||
|
||||
assert result.is_failed
|
||||
assert result.error == "batch job failed"
|
||||
@@ -0,0 +1,77 @@
|
||||
"""Unit tests for the batch OCR job-tracking store (Deck #332).
|
||||
|
||||
Runs against a real temp-SQLite ``RefreshTokenStorage`` (its ``initialize()``
|
||||
applies the migrations, incl. ``batch_ocr_jobs``).
|
||||
"""
|
||||
|
||||
import tempfile
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
|
||||
from nextcloud_mcp_server.auth.storage import RefreshTokenStorage
|
||||
from nextcloud_mcp_server.vector.batch_ocr_store import BatchOcrJobStore
|
||||
|
||||
pytestmark = pytest.mark.unit
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
async def store():
|
||||
with tempfile.TemporaryDirectory() as tmp:
|
||||
storage = RefreshTokenStorage(db_path=str(Path(tmp) / "batch.db"))
|
||||
await storage.initialize()
|
||||
yield BatchOcrJobStore(storage)
|
||||
|
||||
|
||||
_DOC = dict(user_id="u1", doc_id="d1", doc_type="file", etag="v1")
|
||||
|
||||
|
||||
async def test_get_missing_returns_none(store):
|
||||
assert await store.get(**_DOC) is None
|
||||
|
||||
|
||||
async def test_insert_then_get(store):
|
||||
await store.insert_pending(**_DOC, job_id="mistral/j1")
|
||||
job = await store.get(**_DOC)
|
||||
assert job is not None
|
||||
assert job.job_id == "mistral/j1"
|
||||
assert job.submitted_at > 0
|
||||
|
||||
|
||||
async def test_insert_is_idempotent_on_conflict(store):
|
||||
await store.insert_pending(**_DOC, job_id="mistral/j1", submitted_at=100)
|
||||
# A racing re-submit must not overwrite the first row's job id.
|
||||
await store.insert_pending(**_DOC, job_id="mistral/j2", submitted_at=200)
|
||||
job = await store.get(**_DOC)
|
||||
assert job.job_id == "mistral/j1"
|
||||
assert job.submitted_at == 100
|
||||
|
||||
|
||||
async def test_delete(store):
|
||||
await store.insert_pending(**_DOC, job_id="mistral/j1")
|
||||
await store.delete(**_DOC)
|
||||
assert await store.get(**_DOC) is None
|
||||
|
||||
|
||||
async def test_delete_stale_for_doc_keeps_current_etag(store):
|
||||
await store.insert_pending(
|
||||
user_id="u1", doc_id="d1", doc_type="file", etag="old", job_id="mistral/old"
|
||||
)
|
||||
await store.insert_pending(
|
||||
user_id="u1", doc_id="d1", doc_type="file", etag="new", job_id="mistral/new"
|
||||
)
|
||||
await store.delete_stale_for_doc(
|
||||
user_id="u1", doc_id="d1", doc_type="file", keep_etag="new"
|
||||
)
|
||||
# Old version row gone; current one kept.
|
||||
assert (
|
||||
await store.get(user_id="u1", doc_id="d1", doc_type="file", etag="old")
|
||||
) is None
|
||||
kept = await store.get(user_id="u1", doc_id="d1", doc_type="file", etag="new")
|
||||
assert kept is not None and kept.job_id == "mistral/new"
|
||||
|
||||
|
||||
async def test_rows_are_scoped_per_document(store):
|
||||
await store.insert_pending(**_DOC, job_id="mistral/j1")
|
||||
other = await store.get(user_id="u1", doc_id="d2", doc_type="file", etag="v1")
|
||||
assert other is None # different doc_id
|
||||
@@ -115,6 +115,41 @@ class TestGetSettings:
|
||||
assert settings.oidc_token_type == "jwt"
|
||||
assert settings.oidc_scopes == "openid profile"
|
||||
|
||||
@patch.dict(
|
||||
os.environ,
|
||||
{
|
||||
"DOCUMENT_OCR_MODE": "batch",
|
||||
"DOCUMENT_OCR_BATCH_POLL_SECONDS": "45",
|
||||
"DOCUMENT_OCR_BATCH_MAX_WAIT_SECONDS": "3600",
|
||||
},
|
||||
clear=True,
|
||||
)
|
||||
def test_get_settings_ocr_batch_mode_from_env(self):
|
||||
"""DOCUMENT_OCR_MODE / batch tuning must reach settings (regression).
|
||||
|
||||
These were added to _DEFAULTS + the Settings dataclass but initially
|
||||
omitted from _field_map, so dynaconf silently ignored the env vars and
|
||||
batch mode could never be enabled in production (Deck #332).
|
||||
"""
|
||||
_reload_config()
|
||||
settings = get_settings()
|
||||
assert settings.document_ocr_mode == "batch"
|
||||
assert settings.document_ocr_batch_poll_seconds == 45
|
||||
assert settings.document_ocr_batch_max_wait_seconds == 3600
|
||||
|
||||
@patch.dict(os.environ, {"DOCUMENT_OCR_MODE": "Batch"}, clear=True)
|
||||
def test_document_ocr_mode_case_normalised(self):
|
||||
"""DOCUMENT_OCR_MODE is case-insensitive (normalised in __post_init__ via
|
||||
_enum_fields, like DOCUMENT_OCR_PROVIDER) — "Batch" -> "batch"."""
|
||||
_reload_config()
|
||||
assert get_settings().document_ocr_mode == "batch"
|
||||
|
||||
@patch.dict(os.environ, {"DOCUMENT_OCR_MODE": "bogus"}, clear=True)
|
||||
def test_document_ocr_mode_invalid_rejected(self):
|
||||
_reload_config()
|
||||
with pytest.raises(ValueError, match="DOCUMENT_OCR_MODE"):
|
||||
get_settings()
|
||||
|
||||
@patch.dict(
|
||||
os.environ,
|
||||
{"QDRANT_LOCATION": "/app/data/qdrant"},
|
||||
|
||||
@@ -0,0 +1,166 @@
|
||||
"""Unit tests for the gateway batch OCR client (Deck #332).
|
||||
|
||||
HTTP is exercised via an ``httpx.MockTransport`` injected by monkeypatching
|
||||
``httpx.AsyncClient`` (the repo has no respx dependency).
|
||||
"""
|
||||
|
||||
from typing import Any, cast
|
||||
|
||||
import httpx
|
||||
import pytest
|
||||
|
||||
from nextcloud_mcp_server.embedding import gateway_batch_client as gbc
|
||||
|
||||
pytestmark = pytest.mark.unit
|
||||
|
||||
|
||||
def _patch_transport(monkeypatch, handler) -> list[httpx.Request]:
|
||||
"""Route the client's httpx calls through ``handler``; return a list that
|
||||
captures each issued request for assertions."""
|
||||
seen: list[httpx.Request] = []
|
||||
real = httpx.AsyncClient
|
||||
|
||||
def factory(*args: Any, **kwargs: Any) -> httpx.AsyncClient:
|
||||
def _wrapped(request: httpx.Request) -> httpx.Response:
|
||||
seen.append(request)
|
||||
return handler(request)
|
||||
|
||||
kwargs["transport"] = httpx.MockTransport(_wrapped)
|
||||
return real(*args, **kwargs)
|
||||
|
||||
monkeypatch.setattr(httpx, "AsyncClient", factory)
|
||||
return seen
|
||||
|
||||
|
||||
def test_base_url_normalization():
|
||||
assert gbc.GatewayBatchOcrClient("https://gw", "m")._base == "https://gw/v1"
|
||||
assert gbc.GatewayBatchOcrClient("https://gw/v1/", "m")._base == "https://gw/v1"
|
||||
|
||||
|
||||
async def test_submit_posts_one_document_and_returns_job_id(monkeypatch):
|
||||
def handler(request: httpx.Request) -> httpx.Response:
|
||||
return httpx.Response(
|
||||
202, json={"job_id": "mistral/job-1", "status": "pending"}
|
||||
)
|
||||
|
||||
seen = _patch_transport(monkeypatch, handler)
|
||||
client = gbc.GatewayBatchOcrClient("https://gw", "mistral/mistral-ocr-latest")
|
||||
|
||||
job_id = await client.submit(b"%PDF-1.7", "application/pdf", custom_id="doc-9")
|
||||
|
||||
assert job_id == "mistral/job-1"
|
||||
req = seen[0]
|
||||
assert req.method == "POST" and req.url.path == "/v1/ocr/batch"
|
||||
import json
|
||||
|
||||
body = json.loads(req.content)
|
||||
assert body["model"] == "mistral/mistral-ocr-latest"
|
||||
assert len(body["documents"]) == 1
|
||||
assert body["documents"][0]["custom_id"] == "doc-9"
|
||||
assert body["documents"][0]["mime_type"] == "application/pdf"
|
||||
assert body["documents"][0]["document_b64"] # base64 present
|
||||
|
||||
|
||||
async def test_submit_sends_bearer_when_token_provider(monkeypatch):
|
||||
class _Tok:
|
||||
async def get_token(self) -> str:
|
||||
return "tok-abc"
|
||||
|
||||
def handler(request: httpx.Request) -> httpx.Response:
|
||||
return httpx.Response(202, json={"job_id": "mistral/j", "status": "pending"})
|
||||
|
||||
seen = _patch_transport(monkeypatch, handler)
|
||||
# _Tok duck-types get_token; cast for the type checker (the client only awaits
|
||||
# get_token()).
|
||||
client = gbc.GatewayBatchOcrClient(
|
||||
"https://gw", "m", token_provider=cast(Any, _Tok())
|
||||
)
|
||||
await client.submit(b"x", "application/pdf", custom_id="d")
|
||||
assert seen[0].headers["Authorization"] == "Bearer tok-abc"
|
||||
|
||||
|
||||
async def test_submit_raises_on_missing_job_id(monkeypatch):
|
||||
# A 2xx with no job_id is a gateway contract violation -> actionable error.
|
||||
_patch_transport(monkeypatch, lambda r: httpx.Response(202, json={}))
|
||||
with pytest.raises(ValueError, match="no job_id"):
|
||||
await gbc.GatewayBatchOcrClient("https://gw", "m").submit(
|
||||
b"x", "application/pdf", custom_id="d"
|
||||
)
|
||||
|
||||
|
||||
async def test_poll_missing_status_is_failed(monkeypatch):
|
||||
# A 2xx body without a status field must fail fast, not poll forever.
|
||||
_patch_transport(monkeypatch, lambda r: httpx.Response(200, json={"total": 1}))
|
||||
result = await gbc.GatewayBatchOcrClient("https://gw", "m").poll("mistral/j")
|
||||
assert result.is_failed
|
||||
|
||||
|
||||
async def test_poll_pending(monkeypatch):
|
||||
_patch_transport(
|
||||
monkeypatch,
|
||||
lambda r: httpx.Response(200, json={"status": "pending", "total": 1}),
|
||||
)
|
||||
result = await gbc.GatewayBatchOcrClient("https://gw", "m").poll("mistral/j")
|
||||
assert result.is_pending and result.pages == []
|
||||
|
||||
|
||||
async def test_poll_succeeded_maps_pages(monkeypatch):
|
||||
body = {
|
||||
"status": "succeeded",
|
||||
"results": [
|
||||
{
|
||||
"custom_id": "d",
|
||||
"pages": [
|
||||
{"index": 1, "markdown": "two"},
|
||||
{"index": 0, "markdown": "one"},
|
||||
],
|
||||
}
|
||||
],
|
||||
}
|
||||
_patch_transport(monkeypatch, lambda r: httpx.Response(200, json=body))
|
||||
result = await gbc.GatewayBatchOcrClient("https://gw", "m").poll("mistral/j")
|
||||
assert result.is_succeeded
|
||||
# Order is preserved as returned; _pages_to_text sorts downstream.
|
||||
assert result.pages == [(1, "two"), (0, "one")]
|
||||
|
||||
|
||||
async def test_poll_failed_surfaces_error(monkeypatch):
|
||||
_patch_transport(
|
||||
monkeypatch,
|
||||
lambda r: httpx.Response(200, json={"status": "failed", "error": "quota"}),
|
||||
)
|
||||
result = await gbc.GatewayBatchOcrClient("https://gw", "m").poll("mistral/j")
|
||||
assert result.is_failed and result.error == "quota"
|
||||
|
||||
|
||||
async def test_poll_succeeded_with_per_document_error_is_failed(monkeypatch):
|
||||
body = {"status": "succeeded", "results": [{"custom_id": "d", "error": "bad page"}]}
|
||||
_patch_transport(monkeypatch, lambda r: httpx.Response(200, json=body))
|
||||
result = await gbc.GatewayBatchOcrClient("https://gw", "m").poll("mistral/j")
|
||||
assert result.is_failed and result.error == "bad page"
|
||||
|
||||
|
||||
async def test_poll_succeeded_empty_pages_is_failed(monkeypatch):
|
||||
# A succeeded job that produced zero pages is a per-document failure, not a
|
||||
# silent 0-chunk success.
|
||||
body = {"status": "succeeded", "results": [{"custom_id": "d", "pages": []}]}
|
||||
_patch_transport(monkeypatch, lambda r: httpx.Response(200, json=body))
|
||||
result = await gbc.GatewayBatchOcrClient("https://gw", "m").poll("mistral/j")
|
||||
assert result.is_failed and result.error == "no pages returned"
|
||||
|
||||
|
||||
async def test_poll_succeeded_no_results_is_failed(monkeypatch):
|
||||
_patch_transport(
|
||||
monkeypatch,
|
||||
lambda r: httpx.Response(200, json={"status": "succeeded", "results": []}),
|
||||
)
|
||||
result = await gbc.GatewayBatchOcrClient("https://gw", "m").poll("mistral/j")
|
||||
assert result.is_failed
|
||||
|
||||
|
||||
async def test_poll_raises_on_http_error(monkeypatch):
|
||||
_patch_transport(
|
||||
monkeypatch, lambda r: httpx.Response(503, json={"detail": "down"})
|
||||
)
|
||||
with pytest.raises(httpx.HTTPStatusError):
|
||||
await gbc.GatewayBatchOcrClient("https://gw", "m").poll("mistral/j")
|
||||
@@ -4,9 +4,12 @@ from types import SimpleNamespace
|
||||
from typing import Any
|
||||
|
||||
import anyio
|
||||
import httpx
|
||||
import pytest
|
||||
|
||||
from nextcloud_mcp_server.document_processors import ocr
|
||||
from nextcloud_mcp_server.embedding.gateway_batch_client import BatchPollResult
|
||||
from nextcloud_mcp_server.vector import batch_ocr_store as _bos
|
||||
|
||||
pytestmark = pytest.mark.unit
|
||||
|
||||
@@ -16,6 +19,9 @@ def _settings(**kw) -> Any: # a Settings stand-in (only the read fields matter)
|
||||
document_ocr_provider="auto",
|
||||
document_ocr_model="mistral/mistral-ocr-latest",
|
||||
document_ocr_timeout_seconds=180.0,
|
||||
document_ocr_mode="sync",
|
||||
document_ocr_batch_poll_seconds=120,
|
||||
document_ocr_batch_max_wait_seconds=86400,
|
||||
embedding_gateway_url=None,
|
||||
embedding_gateway_client_id=None,
|
||||
embedding_gateway_client_secret=None,
|
||||
@@ -73,6 +79,26 @@ def test_build_backend_auto_none_configured():
|
||||
assert ocr.build_ocr_backend(_settings()) is None
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"kw, expect_client",
|
||||
[
|
||||
# batch is gateway-only: the direct mistral backend never gets a client.
|
||||
(dict(document_ocr_provider="mistral", mistral_api_key="k"), False),
|
||||
# gateway selected but no URL -> no client (falls back to sync).
|
||||
(dict(document_ocr_provider="gateway"), False),
|
||||
(
|
||||
dict(document_ocr_provider="gateway", embedding_gateway_url="https://gw"),
|
||||
True,
|
||||
),
|
||||
(dict(document_ocr_provider="auto", embedding_gateway_url="https://gw"), True),
|
||||
(dict(document_ocr_provider="none", embedding_gateway_url="https://gw"), False),
|
||||
],
|
||||
)
|
||||
def test_build_gateway_batch_client_gateway_only(kw, expect_client):
|
||||
client = ocr.build_gateway_batch_client(_settings(**kw))
|
||||
assert (client is not None) is expect_client
|
||||
|
||||
|
||||
def test_build_backend_gateway_missing_m2m_raises():
|
||||
# client_id set but token_url/secret missing -> explicit ValueError (not a
|
||||
# stripped assert), surfaced on backend resolution.
|
||||
@@ -146,7 +172,7 @@ async def test_processor_timeout_returns_timeout_reason(monkeypatch):
|
||||
r = await ocr.OcrProcessor().process(b"%PDF-1.7", "application/pdf")
|
||||
assert r.success is False
|
||||
assert r.metadata["parse_failed_reason"] == "timeout"
|
||||
assert "timed out" in r.error
|
||||
assert "timed out" in (r.error or "")
|
||||
|
||||
|
||||
async def test_gateway_httpx_timeout_maps_to_timeout_reason(monkeypatch):
|
||||
@@ -165,7 +191,7 @@ async def test_gateway_httpx_timeout_maps_to_timeout_reason(monkeypatch):
|
||||
r = await ocr.OcrProcessor().process(b"%PDF-1.7", "application/pdf")
|
||||
assert r.success is False
|
||||
assert r.metadata["parse_failed_reason"] == "timeout"
|
||||
assert "timed out" in r.error
|
||||
assert "timed out" in (r.error or "")
|
||||
|
||||
|
||||
async def test_gateway_backend_uses_configured_timeout(mocker, monkeypatch):
|
||||
@@ -218,3 +244,269 @@ async def test_mistral_backend_applies_timeout(mocker, monkeypatch):
|
||||
|
||||
with pytest.raises(TimeoutError):
|
||||
await backend.ocr(b"%PDF-1.7", "application/pdf")
|
||||
|
||||
|
||||
# --- batch mode (Deck #332) --------------------------------------------------
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"options",
|
||||
[
|
||||
None,
|
||||
{},
|
||||
{"doc_id": "d", "doc_type": "file"}, # missing user_id
|
||||
{"user_id": "u", "doc_type": "file"}, # missing doc_id
|
||||
{"user_id": "u", "doc_id": "d"}, # missing doc_type
|
||||
{"user_id": "u", "doc_id": "d", "doc_type": ""}, # empty doc_type
|
||||
],
|
||||
)
|
||||
def test_batch_identity_returns_none_without_full_identity(options):
|
||||
assert ocr._batch_identity(options) is None
|
||||
|
||||
|
||||
def test_batch_identity_extracts_tuple_and_defaults_etag():
|
||||
assert ocr._batch_identity(
|
||||
{"user_id": "u", "doc_id": "d", "doc_type": "file", "etag": "v1"}
|
||||
) == ("u", "d", "file", "v1")
|
||||
# etag may be absent/empty -> normalised to "".
|
||||
assert ocr._batch_identity({"user_id": "u", "doc_id": "d", "doc_type": "file"}) == (
|
||||
"u",
|
||||
"d",
|
||||
"file",
|
||||
"",
|
||||
)
|
||||
|
||||
|
||||
_IDENTITY = {"user_id": "u1", "doc_id": "d1", "doc_type": "file", "etag": "v1"}
|
||||
|
||||
|
||||
class _FakeStore:
|
||||
"""In-memory stand-in for BatchOcrJobStore keyed like the real table."""
|
||||
|
||||
def __init__(self, preset=None):
|
||||
self.rows: dict[tuple, Any] = {}
|
||||
self.deleted: list[tuple] = []
|
||||
self.stale_swept: list[tuple] = []
|
||||
if preset is not None:
|
||||
self.rows[("u1", "d1", "file", "v1")] = preset
|
||||
|
||||
async def get(self, *, user_id, doc_id, doc_type, etag):
|
||||
return self.rows.get((user_id, doc_id, doc_type, etag))
|
||||
|
||||
async def insert_pending(
|
||||
self, *, user_id, doc_id, doc_type, etag, job_id, submitted_at=None
|
||||
):
|
||||
self.rows[(user_id, doc_id, doc_type, etag)] = SimpleNamespace(
|
||||
job_id=job_id, submitted_at=submitted_at or 1000
|
||||
)
|
||||
|
||||
async def delete(self, *, user_id, doc_id, doc_type, etag):
|
||||
self.deleted.append((user_id, doc_id, doc_type, etag))
|
||||
self.rows.pop((user_id, doc_id, doc_type, etag), None)
|
||||
|
||||
async def delete_stale_for_doc(self, *, user_id, doc_id, doc_type, keep_etag):
|
||||
self.stale_swept.append((user_id, doc_id, doc_type, keep_etag))
|
||||
|
||||
|
||||
class _FakeBatchClient:
|
||||
def __init__(self, *, submit_job="mistral/job-1", poll=None):
|
||||
self._submit_job = submit_job
|
||||
self._poll = poll or BatchPollResult(status="pending", pages=[])
|
||||
self.submitted: list[tuple] = []
|
||||
self.polled: list[str] = []
|
||||
|
||||
async def submit(self, content, mime_type, custom_id):
|
||||
self.submitted.append((content, mime_type, custom_id))
|
||||
return self._submit_job
|
||||
|
||||
async def poll(self, job_id):
|
||||
self.polled.append(job_id)
|
||||
return self._poll
|
||||
|
||||
|
||||
def _wire_batch(monkeypatch, *, client, store, settings=None):
|
||||
settings = settings or _settings(
|
||||
document_ocr_mode="batch",
|
||||
document_ocr_provider="gateway",
|
||||
embedding_gateway_url="https://gw",
|
||||
)
|
||||
monkeypatch.setattr(ocr, "get_settings", lambda: settings)
|
||||
monkeypatch.setattr(ocr, "build_gateway_batch_client", lambda s: client)
|
||||
|
||||
async def _shared(cls):
|
||||
return store
|
||||
|
||||
monkeypatch.setattr(_bos.BatchOcrJobStore, "shared", classmethod(_shared))
|
||||
|
||||
|
||||
async def test_batch_first_run_submits_and_returns_pending_sentinel(monkeypatch):
|
||||
client = _FakeBatchClient()
|
||||
store = _FakeStore()
|
||||
_wire_batch(monkeypatch, client=client, store=store)
|
||||
|
||||
r = await ocr.OcrProcessor().process(
|
||||
b"%PDF-1.7", "application/pdf", options=dict(_IDENTITY)
|
||||
)
|
||||
|
||||
assert r.success is False
|
||||
assert r.metadata[ocr.OCR_BATCH_PENDING_KEY] is True
|
||||
assert r.metadata[ocr.OCR_BATCH_RETRY_IN_KEY] == 120
|
||||
# submitted with the doc id as custom_id, recorded a pending row, swept stale
|
||||
assert client.submitted and client.submitted[0][2] == "d1"
|
||||
assert store.rows[("u1", "d1", "file", "v1")].job_id == "mistral/job-1"
|
||||
assert store.stale_swept == [("u1", "d1", "file", "v1")]
|
||||
|
||||
|
||||
async def test_batch_existing_pending_polls_and_defers(monkeypatch):
|
||||
preset = SimpleNamespace(job_id="mistral/j", submitted_at=1000)
|
||||
client = _FakeBatchClient(poll=BatchPollResult(status="pending", pages=[]))
|
||||
store = _FakeStore(preset=preset)
|
||||
# submitted just now -> deadline not reached
|
||||
monkeypatch.setattr(ocr.time, "time", lambda: 1000.0)
|
||||
_wire_batch(monkeypatch, client=client, store=store)
|
||||
|
||||
r = await ocr.OcrProcessor().process(
|
||||
b"%PDF", "application/pdf", options=dict(_IDENTITY)
|
||||
)
|
||||
|
||||
assert client.polled == ["mistral/j"]
|
||||
assert r.metadata[ocr.OCR_BATCH_PENDING_KEY] is True
|
||||
assert client.submitted == [] # did NOT resubmit
|
||||
|
||||
|
||||
async def test_batch_succeeded_returns_indexed_result(monkeypatch):
|
||||
preset = SimpleNamespace(job_id="mistral/j", submitted_at=1000)
|
||||
client = _FakeBatchClient(
|
||||
poll=BatchPollResult(status="succeeded", pages=[(0, "# One"), (1, "## Two")])
|
||||
)
|
||||
store = _FakeStore(preset=preset)
|
||||
_wire_batch(monkeypatch, client=client, store=store)
|
||||
|
||||
r = await ocr.OcrProcessor().process(
|
||||
b"%PDF", "application/pdf", options=dict(_IDENTITY)
|
||||
)
|
||||
|
||||
assert r.success is True
|
||||
assert r.text == "# One\n\n## Two"
|
||||
assert r.metadata["page_count"] == 2
|
||||
assert ("u1", "d1", "file", "v1") in store.deleted # row cleaned up
|
||||
|
||||
|
||||
async def test_batch_failed_marks_parse_error(monkeypatch):
|
||||
preset = SimpleNamespace(job_id="mistral/j", submitted_at=1000)
|
||||
client = _FakeBatchClient(
|
||||
poll=BatchPollResult(status="failed", pages=[], error="x")
|
||||
)
|
||||
store = _FakeStore(preset=preset)
|
||||
_wire_batch(monkeypatch, client=client, store=store)
|
||||
|
||||
r = await ocr.OcrProcessor().process(
|
||||
b"%PDF", "application/pdf", options=dict(_IDENTITY)
|
||||
)
|
||||
|
||||
assert r.success is False
|
||||
assert r.metadata["parse_failed_reason"] == "error"
|
||||
assert ("u1", "d1", "file", "v1") in store.deleted
|
||||
|
||||
|
||||
async def test_batch_unexpected_status_marks_failed_not_empty_success(monkeypatch):
|
||||
# A terminal status that isn't succeeded/failed (gateway skew) must NOT
|
||||
# produce a 0-chunk "success" that silently indexes empty text + loops.
|
||||
preset = SimpleNamespace(job_id="mistral/j", submitted_at=1000)
|
||||
client = _FakeBatchClient(poll=BatchPollResult(status="cancelled", pages=[]))
|
||||
store = _FakeStore(preset=preset)
|
||||
_wire_batch(monkeypatch, client=client, store=store)
|
||||
|
||||
r = await ocr.OcrProcessor().process(
|
||||
b"%PDF", "application/pdf", options=dict(_IDENTITY)
|
||||
)
|
||||
|
||||
assert r.success is False
|
||||
assert r.metadata["parse_failed_reason"] == "error"
|
||||
assert "cancelled" in (r.error or "")
|
||||
assert ("u1", "d1", "file", "v1") in store.deleted
|
||||
|
||||
|
||||
async def test_batch_deadline_exceeded_marks_timeout(monkeypatch):
|
||||
preset = SimpleNamespace(job_id="mistral/j", submitted_at=1000)
|
||||
client = _FakeBatchClient(poll=BatchPollResult(status="pending", pages=[]))
|
||||
store = _FakeStore(preset=preset)
|
||||
# now far past submitted_at + max_wait (86400)
|
||||
monkeypatch.setattr(ocr.time, "time", lambda: 1000.0 + 90000)
|
||||
_wire_batch(monkeypatch, client=client, store=store)
|
||||
|
||||
r = await ocr.OcrProcessor().process(
|
||||
b"%PDF", "application/pdf", options=dict(_IDENTITY)
|
||||
)
|
||||
|
||||
assert r.success is False
|
||||
assert r.metadata["parse_failed_reason"] == "timeout"
|
||||
assert ("u1", "d1", "file", "v1") in store.deleted
|
||||
|
||||
|
||||
async def test_batch_falls_back_to_sync_when_no_gateway(monkeypatch):
|
||||
class _FakeBackend:
|
||||
async def ocr(self, content, mime_type):
|
||||
return "sync text", [{"page": 1, "start_offset": 0, "end_offset": 9}]
|
||||
|
||||
settings = _settings(document_ocr_mode="batch", document_ocr_provider="mistral")
|
||||
monkeypatch.setattr(ocr, "get_settings", lambda: settings)
|
||||
monkeypatch.setattr(ocr, "build_gateway_batch_client", lambda s: None)
|
||||
monkeypatch.setattr(ocr, "build_ocr_backend", lambda s: _FakeBackend())
|
||||
|
||||
r = await ocr.OcrProcessor().process(
|
||||
b"%PDF", "application/pdf", options=dict(_IDENTITY)
|
||||
)
|
||||
assert r.success is True and r.text == "sync text"
|
||||
|
||||
|
||||
async def test_batch_falls_back_to_sync_when_no_identity(monkeypatch):
|
||||
class _FakeBackend:
|
||||
async def ocr(self, content, mime_type):
|
||||
return "sync text", [{"page": 1, "start_offset": 0, "end_offset": 9}]
|
||||
|
||||
client = _FakeBatchClient()
|
||||
settings = _settings(
|
||||
document_ocr_mode="batch",
|
||||
document_ocr_provider="gateway",
|
||||
embedding_gateway_url="https://gw",
|
||||
)
|
||||
monkeypatch.setattr(ocr, "get_settings", lambda: settings)
|
||||
monkeypatch.setattr(ocr, "build_gateway_batch_client", lambda s: client)
|
||||
monkeypatch.setattr(ocr, "build_ocr_backend", lambda s: _FakeBackend())
|
||||
|
||||
# No options -> inline path -> batch inapplicable -> sync fallback.
|
||||
r = await ocr.OcrProcessor().process(b"%PDF", "application/pdf", options=None)
|
||||
assert r.success is True and r.text == "sync text"
|
||||
assert client.submitted == [] # never attempted batch
|
||||
|
||||
|
||||
async def test_batch_submit_transport_error_propagates_not_caught(monkeypatch):
|
||||
# Opted into batch: a transport error from submit() must propagate (to
|
||||
# procrastinate for a durable retry), NOT be caught by the sync OCR
|
||||
# try/except or fall back to a surprise sync transcription. Guards the
|
||||
# intentional asymmetry documented in process().
|
||||
class _DownClient:
|
||||
submitted: list = []
|
||||
|
||||
async def submit(self, content, mime_type, custom_id):
|
||||
raise httpx.ConnectError("gateway down")
|
||||
|
||||
async def poll(self, job_id): # pragma: no cover - not reached
|
||||
raise AssertionError("poll should not be called")
|
||||
|
||||
sync_backend_used = False
|
||||
|
||||
def _build_backend(_s):
|
||||
nonlocal sync_backend_used
|
||||
sync_backend_used = True
|
||||
return None
|
||||
|
||||
monkeypatch.setattr(ocr, "build_ocr_backend", _build_backend)
|
||||
_wire_batch(monkeypatch, client=_DownClient(), store=_FakeStore())
|
||||
|
||||
with pytest.raises(httpx.ConnectError):
|
||||
await ocr.OcrProcessor().process(
|
||||
b"%PDF", "application/pdf", options=dict(_IDENTITY)
|
||||
)
|
||||
assert sync_backend_used is False # never fell back to the sync path
|
||||
|
||||
@@ -97,3 +97,40 @@ async def test_hard_failure_returns_result_without_escalating(monkeypatch):
|
||||
reg.evaluate_escalation.assert_not_called()
|
||||
rec.assert_not_called()
|
||||
sup.assert_not_called()
|
||||
|
||||
|
||||
async def test_ocr_batch_pending_sentinel_raises_batch_pending():
|
||||
"""Batch OCR (Deck #332): the OCR tier's pending sentinel result is turned
|
||||
into a BatchPending raise (same decision point as EscalateError), carrying
|
||||
the processor's retry_in, and the escalation gate is never consulted."""
|
||||
from nextcloud_mcp_server.document_processors.escalation import BatchPending
|
||||
from nextcloud_mcp_server.document_processors.ocr import (
|
||||
OCR_BATCH_PENDING_KEY,
|
||||
OCR_BATCH_RETRY_IN_KEY,
|
||||
)
|
||||
|
||||
result = ProcessingResult(
|
||||
text="",
|
||||
metadata={OCR_BATCH_PENDING_KEY: True, OCR_BATCH_RETRY_IN_KEY: 90},
|
||||
processor="ocr",
|
||||
success=False,
|
||||
)
|
||||
reg = _registry(result, decision=None)
|
||||
with pytest.raises(BatchPending) as ei:
|
||||
await processor._parse_pdf_tier(
|
||||
reg, b"%PDF", "application/pdf", "scan.pdf", "ocr", settings=object()
|
||||
)
|
||||
assert ei.value.retry_in == 90
|
||||
reg.evaluate_escalation.assert_not_called()
|
||||
|
||||
|
||||
async def test_options_threaded_to_process_tier():
|
||||
"""The OCR identity options are forwarded to process_tier (batch needs them)."""
|
||||
result = ProcessingResult(text="clean", metadata={}, processor="ocr")
|
||||
reg = _registry(result, decision=None)
|
||||
opts = {"user_id": "u", "doc_id": "d", "doc_type": "file", "etag": "v"}
|
||||
await processor._parse_pdf_tier(
|
||||
reg, b"%PDF", "application/pdf", "f.pdf", "ocr", settings=object(), options=opts
|
||||
)
|
||||
# process_tier(content, content_type, filename, tier, options=...)
|
||||
assert reg.process_tier.await_args.kwargs["options"] == opts
|
||||
|
||||
@@ -14,6 +14,7 @@ from procrastinate.jobs import Job
|
||||
import nextcloud_mcp_server.vector.queue.procrastinate as pq
|
||||
from nextcloud_mcp_server.document_processors.escalation import (
|
||||
TIER_LADDER,
|
||||
BatchPending,
|
||||
EscalateError,
|
||||
next_tier,
|
||||
)
|
||||
@@ -120,3 +121,30 @@ class TestTieredEscalationStrategy:
|
||||
exception=ValueError("permanent"), job=_job(attempts=1)
|
||||
)
|
||||
assert decision is None
|
||||
|
||||
def test_batch_pending_defers_same_queue(self):
|
||||
# Batch OCR re-poll (Deck #332): same-queue deferral after retry_in.
|
||||
before = datetime.now(timezone.utc)
|
||||
decision = self._strategy().get_retry_decision(
|
||||
exception=BatchPending(retry_in=120), job=_job(queue=pq.INGEST_QUEUE_OCR)
|
||||
)
|
||||
after = datetime.now(timezone.utc)
|
||||
assert decision is not None
|
||||
assert decision.queue is None # stays on its own tier queue
|
||||
assert decision.retry_at is not None
|
||||
lo = (decision.retry_at - after).total_seconds()
|
||||
hi = (decision.retry_at - before).total_seconds()
|
||||
assert lo <= 120 <= hi
|
||||
|
||||
def test_batch_pending_exempt_from_transient_cap(self):
|
||||
# A batch can take hours -> many polls; the transient cap must NOT stop it
|
||||
# (the OCR processor's own deadline terminates a stuck job instead).
|
||||
decision = self._strategy(max_transient=5).get_retry_decision(
|
||||
exception=BatchPending(retry_in=60), job=_job(attempts=999)
|
||||
)
|
||||
assert decision is not None and decision.retry_at is not None
|
||||
|
||||
def test_batch_pending_unwraps_exception_group(self):
|
||||
group = ExceptionGroup("wrapped", [BatchPending(retry_in=60)])
|
||||
decision = self._strategy().get_retry_decision(exception=group, job=_job())
|
||||
assert decision is not None and decision.retry_at is not None
|
||||
|
||||
Reference in New Issue
Block a user