diff --git a/nextcloud_mcp_server/alembic/versions/20260615_1200_008_add_batch_ocr_jobs.py b/nextcloud_mcp_server/alembic/versions/20260615_1200_008_add_batch_ocr_jobs.py index 56e70521..af85c6c7 100644 --- a/nextcloud_mcp_server/alembic/versions/20260615_1200_008_add_batch_ocr_jobs.py +++ b/nextcloud_mcp_server/alembic/versions/20260615_1200_008_add_batch_ocr_jobs.py @@ -36,7 +36,11 @@ def upgrade() -> None: # 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. + # 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), @@ -51,14 +55,8 @@ def upgrade() -> None: # (DOCUMENT_OCR_BATCH_MAX_WAIT_SECONDS). sa.Column("submitted_at", sa.BigInteger(), nullable=False), sa.Column("updated_at", sa.BigInteger(), nullable=False), - # One in-flight job per (document, content version). A resubmit for a new - # etag inserts a new row; the superseded row is swept on resubmit. - sa.UniqueConstraint( - "user_id", - "doc_id", - "doc_type", - "etag", - name="uq_batch_ocr_jobs_doc", + sa.PrimaryKeyConstraint( + "user_id", "doc_id", "doc_type", "etag", name="pk_batch_ocr_jobs" ), ) diff --git a/nextcloud_mcp_server/config.py b/nextcloud_mcp_server/config.py index 684ffc67..6bf0c903 100644 --- a/nextcloud_mcp_server/config.py +++ b/nextcloud_mcp_server/config.py @@ -1521,6 +1521,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", diff --git a/nextcloud_mcp_server/document_processors/ocr.py b/nextcloud_mcp_server/document_processors/ocr.py index 5a988b73..9f9a0ab2 100644 --- a/nextcloud_mcp_server/document_processors/ocr.py +++ b/nextcloud_mcp_server/document_processors/ocr.py @@ -272,6 +272,7 @@ class OcrProcessor(DocumentProcessor): # using sync" warning to once per pod. self._batch_client_resolved = False self._batch_client: Any = None + self._batch_client_lock: anyio.Lock | None = None self._batch_fallback_warned = False @property @@ -377,9 +378,9 @@ class OcrProcessor(DocumentProcessor): 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._backend_lock is None: - self._backend_lock = anyio.Lock() - async with self._backend_lock: + 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 diff --git a/tests/unit/test_config.py b/tests/unit/test_config.py index 9856798d..3abcef40 100644 --- a/tests/unit/test_config.py +++ b/tests/unit/test_config.py @@ -115,6 +115,28 @@ 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, {"QDRANT_LOCATION": "/app/data/qdrant"}, diff --git a/tests/unit/test_gateway_batch_client.py b/tests/unit/test_gateway_batch_client.py index 67b293a2..2e2fd45b 100644 --- a/tests/unit/test_gateway_batch_client.py +++ b/tests/unit/test_gateway_batch_client.py @@ -33,8 +33,8 @@ def _patch_transport(monkeypatch, handler) -> list[httpx.Request]: def test_base_url_normalization(): - assert gbc.GatewayBatchOcrClient("http://gw", "m")._base == "http://gw/v1" - assert gbc.GatewayBatchOcrClient("http://gw/v1/", "m")._base == "http://gw/v1" + 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): @@ -44,7 +44,7 @@ async def test_submit_posts_one_document_and_returns_job_id(monkeypatch): ) seen = _patch_transport(monkeypatch, handler) - client = gbc.GatewayBatchOcrClient("http://gw", "mistral/mistral-ocr-latest") + client = gbc.GatewayBatchOcrClient("https://gw", "mistral/mistral-ocr-latest") job_id = await client.submit(b"%PDF-1.7", "application/pdf", custom_id="doc-9") @@ -73,7 +73,7 @@ async def test_submit_sends_bearer_when_token_provider(monkeypatch): # _Tok duck-types get_token; cast for the type checker (the client only awaits # get_token()). client = gbc.GatewayBatchOcrClient( - "http://gw", "m", token_provider=cast(Any, _Tok()) + "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" @@ -84,7 +84,7 @@ async def test_poll_pending(monkeypatch): monkeypatch, lambda r: httpx.Response(200, json={"status": "pending", "total": 1}), ) - result = await gbc.GatewayBatchOcrClient("http://gw", "m").poll("mistral/j") + result = await gbc.GatewayBatchOcrClient("https://gw", "m").poll("mistral/j") assert result.is_pending and result.pages == [] @@ -102,7 +102,7 @@ async def test_poll_succeeded_maps_pages(monkeypatch): ], } _patch_transport(monkeypatch, lambda r: httpx.Response(200, json=body)) - result = await gbc.GatewayBatchOcrClient("http://gw", "m").poll("mistral/j") + 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")] @@ -113,14 +113,14 @@ async def test_poll_failed_surfaces_error(monkeypatch): monkeypatch, lambda r: httpx.Response(200, json={"status": "failed", "error": "quota"}), ) - result = await gbc.GatewayBatchOcrClient("http://gw", "m").poll("mistral/j") + 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("http://gw", "m").poll("mistral/j") + result = await gbc.GatewayBatchOcrClient("https://gw", "m").poll("mistral/j") assert result.is_failed and result.error == "bad page" @@ -129,7 +129,7 @@ async def test_poll_succeeded_no_results_is_failed(monkeypatch): monkeypatch, lambda r: httpx.Response(200, json={"status": "succeeded", "results": []}), ) - result = await gbc.GatewayBatchOcrClient("http://gw", "m").poll("mistral/j") + result = await gbc.GatewayBatchOcrClient("https://gw", "m").poll("mistral/j") assert result.is_failed @@ -138,4 +138,4 @@ async def test_poll_raises_on_http_error(monkeypatch): monkeypatch, lambda r: httpx.Response(503, json={"detail": "down"}) ) with pytest.raises(httpx.HTTPStatusError): - await gbc.GatewayBatchOcrClient("http://gw", "m").poll("mistral/j") + await gbc.GatewayBatchOcrClient("https://gw", "m").poll("mistral/j") diff --git a/tests/unit/test_ocr_processor.py b/tests/unit/test_ocr_processor.py index a92115d7..cdbc98e5 100644 --- a/tests/unit/test_ocr_processor.py +++ b/tests/unit/test_ocr_processor.py @@ -281,7 +281,7 @@ def _wire_batch(monkeypatch, *, client, store, settings=None): settings = settings or _settings( document_ocr_mode="batch", document_ocr_provider="gateway", - embedding_gateway_url="http://gw", + embedding_gateway_url="https://gw", ) monkeypatch.setattr(ocr, "get_settings", lambda: settings) monkeypatch.setattr(ocr, "build_gateway_batch_client", lambda s: client) @@ -404,7 +404,7 @@ async def test_batch_falls_back_to_sync_when_no_identity(monkeypatch): settings = _settings( document_ocr_mode="batch", document_ocr_provider="gateway", - embedding_gateway_url="http://gw", + embedding_gateway_url="https://gw", ) monkeypatch.setattr(ocr, "get_settings", lambda: settings) monkeypatch.setattr(ocr, "build_gateway_batch_client", lambda s: client)