From 995e810d8992af04e417e83fddab00dfbe9ed0fc Mon Sep 17 00:00:00 2001 From: Chris Coutinho Date: Mon, 15 Jun 2026 10:45:10 +0200 Subject: [PATCH] =?UTF-8?q?fix(ocr):=20round-2=20review=20=E2=80=94=20lazy?= =?UTF-8?q?=20store=20lock,=20mode=20enum=20normalization,=20type=20hints?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Round 2 review (PR #910): - BLOCKING: BatchOcrJobStore._shared_lock is now lazy-init (anyio.Lock | None, created on first shared() call) instead of at class-definition time — matches the CLAUDE.md "no anyio primitives at import time" rule and OcrProcessor's pattern. The None-check->assign has no await between, so it's race-free. - document_ocr_mode now normalizes via _enum_fields (case-insensitive, like document_ocr_provider) instead of a strict dynaconf is_in Validator, so DOCUMENT_OCR_MODE=Batch normalizes to "batch" rather than erroring. Tests for case-normalization + invalid-value rejection. - TYPE_CHECKING-gated GatewayBatchOcrClient import so build_gateway_batch_client / _get_batch_client are typed `GatewayBatchOcrClient | None` instead of Any (runtime import stays lazy to avoid the import cycle). - Rename ocr_options -> doc_identity_options (it's threaded to all tiers; only OCR reads it) + clarify the comment. - Drop the redundant forward-ref quotes on _shared_instance. - Add direct _batch_identity unit tests (partial/empty options branches). Left as follow-up: reusing one httpx.AsyncClient across submit/poll (same per-call pattern as the existing sync _GatewayOcrBackend; no clean aclose hook on the cached client today). 1653 unit tests pass; ruff + ty green. Co-Authored-By: Claude Opus 4.8 (1M context) --- nextcloud_mcp_server/config.py | 6 +++- .../document_processors/ocr.py | 12 ++++++-- .../vector/batch_ocr_store.py | 20 ++++++++----- nextcloud_mcp_server/vector/processor.py | 11 +++---- tests/unit/test_config.py | 13 +++++++++ tests/unit/test_ocr_processor.py | 29 +++++++++++++++++++ 6 files changed, 75 insertions(+), 16 deletions(-) diff --git a/nextcloud_mcp_server/config.py b/nextcloud_mcp_server/config.py index 6bf0c903..0265b04a 100644 --- a/nextcloud_mcp_server/config.py +++ b/nextcloud_mcp_server/config.py @@ -366,7 +366,10 @@ _dynaconf = Dynaconf( Validator("DOCUMENT_CHUNK_SIZE", gte=1), Validator("DOCUMENT_PARSE_TIMEOUT_SECONDS", gte=1), Validator("DOCUMENT_OCR_TIMEOUT_SECONDS", gte=1), - Validator("DOCUMENT_OCR_MODE", is_in=("sync", "batch")), + # 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), @@ -1029,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() diff --git a/nextcloud_mcp_server/document_processors/ocr.py b/nextcloud_mcp_server/document_processors/ocr.py index 9f9a0ab2..21a072b9 100644 --- a/nextcloud_mcp_server/document_processors/ocr.py +++ b/nextcloud_mcp_server/document_processors/ocr.py @@ -21,7 +21,7 @@ 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 @@ -30,6 +30,12 @@ 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 @@ -197,7 +203,7 @@ def _build_gateway_token_provider(settings: Settings) -> Any: ) -def build_gateway_batch_client(settings: Settings) -> Any: +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 @@ -373,7 +379,7 @@ class OcrProcessor(DocumentProcessor): processor=self.name, ) - async def _get_batch_client(self) -> Any: + 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.""" diff --git a/nextcloud_mcp_server/vector/batch_ocr_store.py b/nextcloud_mcp_server/vector/batch_ocr_store.py index f6bfc099..97867295 100644 --- a/nextcloud_mcp_server/vector/batch_ocr_store.py +++ b/nextcloud_mcp_server/vector/batch_ocr_store.py @@ -38,18 +38,24 @@ class BatchOcrJob: class BatchOcrJobStore: """CRUD for the ``batch_ocr_jobs`` table (one row per in-flight job).""" - _shared_instance: "BatchOcrJobStore | None" = None - _shared_lock: anyio.Lock = anyio.Lock() + _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 (mirrors - ``UsageEventStore.shared``). Tests should construct - ``BatchOcrJobStore(storage)`` directly — the cache is a process global - with no teardown hook.""" + 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()) diff --git a/nextcloud_mcp_server/vector/processor.py b/nextcloud_mcp_server/vector/processor.py index cd9cd490..2ed2ccce 100644 --- a/nextcloud_mcp_server/vector/processor.py +++ b/nextcloud_mcp_server/vector/processor.py @@ -984,10 +984,11 @@ 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): - # Thread per-document identity to the OCR tier so batch mode - # (Deck #332) can key its job-tracking table; other tiers - # ignore it. - ocr_options = { + # 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, @@ -1000,7 +1001,7 @@ async def _index_document( file_path, tier, settings, - options=ocr_options, + options=doc_identity_options, ) else: result = await registry.process( diff --git a/tests/unit/test_config.py b/tests/unit/test_config.py index 3abcef40..4a15d77c 100644 --- a/tests/unit/test_config.py +++ b/tests/unit/test_config.py @@ -137,6 +137,19 @@ class TestGetSettings: 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"}, diff --git a/tests/unit/test_ocr_processor.py b/tests/unit/test_ocr_processor.py index cdbc98e5..2f12f6e6 100644 --- a/tests/unit/test_ocr_processor.py +++ b/tests/unit/test_ocr_processor.py @@ -225,6 +225,35 @@ async def test_mistral_backend_applies_timeout(mocker, monkeypatch): # --- 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", + "", + ) + + from nextcloud_mcp_server.embedding.gateway_batch_client import ( # noqa: E402 BatchPollResult, )