diff --git a/ARCHITECTURE.md b/ARCHITECTURE.md index 71ae553af..63d18aab8 100644 --- a/ARCHITECTURE.md +++ b/ARCHITECTURE.md @@ -100,6 +100,26 @@ Parsed email body/subject, address, and attachment display text strips active HTML/script markup at the parser boundary while preserving message/thread identifiers and angle-address headers separately. +The proposed [bounded attachment-source contract](docs/adr/0023-bounded-attachment-parse-source-contract.md) +retains PDF source bytes through 64 MiB while enforcing the separate verified +20 MiB NewsDOM request bound before network validation. Missing configuration +leaves the source pending; size rejection records failure without discarding +bytes. Real PostgreSQL tests cover committed outcomes and rollback with an +unchanged published PDF. Sweep IDs are captured before transactions; after +rollback, remaining pending records are explicitly reloaded with their required +relationships so one failed item does not abort the prefetched batch. Normal +successful batches retain their existing query path. +The recognition sweep holds one physical connection across both phases and +their per-item transactions. A disconnect aborts the cycle rather than +reconnecting without its lease; cancellation or uncertain lease release +invalidates the connection before reuse. Real PostgreSQL tests cover pooled +readers and one-slot recovery. Pool sizing must allow this whole-sweep checkout; +it is not exactly-once external provider execution. See ADR-0023 for alternatives +and the sibling scheduler/import repair boundaries. +This does not establish released recognition capacity; +the [PDF upload owner](docs/adr/0021-bounded-pdf-dom-upload-contract.md), immutable +provider release, exact consumer pin, and capacity evidence remain prerequisites. + Customer-owned mail, CalDAV/CardDAV, and WebDAV systems remain the durable source-of-truth. Naruon can cache/index metadata and generate writeback intents, but provider writes must use server-authoritative source records, ownership diff --git a/CHANGELOG.md b/CHANGELOG.md index 5cfc3246e..da19c1a94 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,4 +1,13 @@ ## [Unreleased] +- Proposed: retain PDF attachments up to 64 MiB for deferred processing and + preserve admitted source bytes when the processing service rejects the size. + Files above 64 MiB and unsupported binary formats remain metadata-only. + Recognition above the processing service's verified limit is not released; + split the PDF to fit that limit or wait for an approved service upgrade. + A processing failure for one file no longer stops the remaining files in its + batch; the original files remain available for a later attempt. + Background processing also releases its coordination claim when interrupted, + so a later worker can resume pending files. This repair remains proposed. - Proposed: accept manual PDF uploads up to 64MiB after the required processing service release is verified and pinned. Larger files are rejected before storage or processing with HTTP 413; split the file and upload it again. diff --git a/backend/api/data.py b/backend/api/data.py index e7f84162f..18301cc90 100644 --- a/backend/api/data.py +++ b/backend/api/data.py @@ -41,10 +41,9 @@ router = APIRouter(prefix="/api/data", tags=["data"]) DATA_VECTOR_DIMENSIONS = 1536 -# Upper bound for the binary PDF DOM recognition upload variant. Keep this in -# step with NewsDOM's MAX_PARSE_UPLOAD_BYTES and the signed email-import -# transport ceiling so large customer PDFs are not accepted by one path and -# rejected by the next. +# Proposed Naruon admission ceiling inherited from the PDF upload owner. +# Provider calls enforce their separate bound in request_pdf_dom; this constant +# is not evidence of a released NewsDOM 64 MiB recognition capability. _MAX_PDF_DOM_UPLOAD_BYTES = 64 * 1024 * 1024 ATTACHMENT_PARSE_BREAKDOWN_EVIDENCE_SOURCE = ( "email_attachments.content_type, " diff --git a/backend/services/attachment_parser.py b/backend/services/attachment_parser.py index 7359d6b2f..178b9e7ca 100644 --- a/backend/services/attachment_parser.py +++ b/backend/services/attachment_parser.py @@ -16,7 +16,10 @@ "application/x-binary", } MAX_ATTACHMENT_PARSE_SOURCE_CHARS = 1_000_000 -MAX_ATTACHMENT_PARSE_SOURCE_BYTES = 20 * 1024 * 1024 +# Keep deferred attachment recognition aligned with the authenticated upload +# transport. Unsupported binaries remain metadata-only; recognized/deferred +# formats may retain at most this bounded source payload for a worker. +MAX_ATTACHMENT_PARSE_SOURCE_BYTES = 64 * 1024 * 1024 MAX_ATTACHMENT_FILENAME_DECODE_ROUNDS = 3 diff --git a/backend/services/newsdom_client.py b/backend/services/newsdom_client.py index 09b94fc8a..0984a1938 100644 --- a/backend/services/newsdom_client.py +++ b/backend/services/newsdom_client.py @@ -34,6 +34,9 @@ _LOCAL_DEV_HOSTNAMES = {"localhost", "localhost.localdomain"} _LOCAL_DEV_IP_LITERALS = {"127.0.0.1", "::1"} _DEFAULT_PARSE_TIMEOUT_SECONDS = 300.0 +# NewsDOM's verified protected-source ``/parse`` guard is 20 MiB; this does not +# prove a deployed version. Raise only after an immutable owner release is pinned. +NEWSDOM_MAX_PARSE_UPLOAD_BYTES = 20 * 1024 * 1024 class NewsdomConfigurationError(RuntimeError): @@ -44,6 +47,10 @@ class NewsdomRequestError(RuntimeError): """Raised when the NewsDOM sidecar cannot fulfil a parse request.""" +class NewsdomPayloadTooLargeError(NewsdomRequestError): + """Raised when the payload exceeds the local or provider-side contract.""" + + class NewsdomEmptyRecognitionError(NewsdomRequestError): """Raised when a 200 sidecar response carries no usable recognized text. @@ -393,6 +400,10 @@ async def request_pdf_dom( """ if not pdf_bytes: raise NewsdomRequestError("Cannot recognize an empty PDF payload") + if len(pdf_bytes) > NEWSDOM_MAX_PARSE_UPLOAD_BYTES: + raise NewsdomPayloadTooLargeError( + "NewsDOM PDF payload exceeds the 20 MiB parse upload contract" + ) validated = await validate_newsdom_base_url_details_async(base_url) if validated is None: @@ -423,6 +434,8 @@ async def request_pdf_dom( except httpx.HTTPError as exc: raise NewsdomRequestError(f"NewsDOM request failed: {exc}") from exc + if response.status_code == 413: + raise NewsdomPayloadTooLargeError("NewsDOM returned HTTP 413 for /parse") if response.status_code >= 400: raise NewsdomRequestError( f"NewsDOM returned HTTP {response.status_code} for /parse" diff --git a/backend/services/newsdom_worker.py b/backend/services/newsdom_worker.py index 26e4cffbb..a6fcdb88e 100644 --- a/backend/services/newsdom_worker.py +++ b/backend/services/newsdom_worker.py @@ -18,6 +18,7 @@ from collections.abc import Awaitable, Callable from sqlalchemy import bindparam, func, select +from sqlalchemy.exc import DBAPIError from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy.orm import selectinload @@ -28,11 +29,12 @@ Document, Email, ) -from db.session import AsyncSessionLocal +from db.session import AsyncSessionLocal, engine from services.attachment_parser import decode_deferred_attachment_payload from services.content_graph import ParseResult from services.newsdom_client import ( NewsdomConfigurationError, + NewsdomPayloadTooLargeError, NewsdomRequestError, request_pdf_dom, ) @@ -257,6 +259,18 @@ async def process_pending_attachment( exc, ) return RESULT_PENDING + except NewsdomPayloadTooLargeError as exc: + attachment.parse_status = PDF_DOM_RECOGNITION_FAILED_STATUS + attachment.parse_error_code = "provider_payload_size_exceeded" + # A bounded, expected admission rejection is operational information, + # not an infrastructure warning; the persisted error code remains the + # customer-visible source of truth. + logger.info( + "NewsDOM attachment %s exceeds the provider payload contract: %s", + getattr(attachment, "id", "?"), + exc, + ) + return RESULT_FAILED except (NewsdomRequestError, ValueError) as exc: attachment.parse_status = PDF_DOM_RECOGNITION_FAILED_STATUS attachment.parse_error_code = "recognition_failed" @@ -315,6 +329,14 @@ async def process_pending_document( exc, ) return RESULT_PENDING + except NewsdomPayloadTooLargeError as exc: + document.document_status = PDF_DOM_RECOGNITION_FAILED_STATUS + logger.info( + "NewsDOM document %s exceeds the provider payload contract: %s", + getattr(document, "document_id", "?"), + exc, + ) + return RESULT_FAILED except (NewsdomRequestError, ValueError) as exc: document.document_status = PDF_DOM_RECOGNITION_FAILED_STATUS logger.warning( @@ -352,8 +374,8 @@ async def _try_acquire_sweep_lease(session: AsyncSession) -> bool | None: async def _release_sweep_lease(session: AsyncSession) -> None: - """Release the PostgreSQL advisory lock for a recognition sweep.""" - await session.scalar( + """Release the sweep lock and require confirmation from its owning backend.""" + released = await session.scalar( select( func.pg_advisory_unlock( func.hashtext(bindparam("namespace_key")), @@ -362,6 +384,8 @@ async def _release_sweep_lease(session: AsyncSession) -> None: ), _SWEEP_LOCK_PARAMS, ) + if released is not True: + raise RuntimeError("NewsDOM sweep lease release could not be confirmed.") class NewsdomRecognitionWorker: @@ -439,29 +463,46 @@ async def _run_loop(self) -> None: break async def _sweep(self) -> None: - """Process one leased attachment and document sweep.""" - async with AsyncSessionLocal() as session: - lease = await _try_acquire_sweep_lease(session) - if lease is False: - logger.debug( - "NewsDOM recognition sweep skipped: another replica holds " - "the lease." - ) - return + """Keep one physical connection through item transactions and lease release.""" + async with ( + engine.connect() as connection, + AsyncSessionLocal(bind=connection) as session, + ): try: + lease = await _try_acquire_sweep_lease(session) + if lease is False: + logger.debug( + "NewsDOM recognition sweep skipped: another replica holds " + "the lease." + ) + return await self._sweep_attachments(session) await self._sweep_documents(session) - finally: if lease is True: + await session.rollback() await _release_sweep_lease(session) + except BaseException: + # Acquisition cancellation and uncertain release may leave a + # session lock. End that backend before session-close rollback. + await connection.invalidate() + raise async def _sweep_attachments(self, session: AsyncSession) -> None: """Process a bounded, starvation-free batch of pending attachments.""" rows = await self._load_pending_attachments(session) - if rows: - self._attachment_cursor = rows[-1].id - for attachment in rows: + pending_rows = [(attachment.id, attachment) for attachment in rows] + reload_pending_rows = False + for attachment_id, attachment in pending_rows: try: + if reload_pending_rows: + result_rows = await session.execute( + self._pending_attachment_statement(None).where( + Attachment.id == attachment_id + ) + ) + attachment = result_rows.scalar_one_or_none() + if attachment is None: + continue result = await process_pending_attachment( session=session, attachment=attachment, @@ -472,16 +513,22 @@ async def _sweep_attachments(self, session: AsyncSession) -> None: if result != RESULT_PENDING: logger.info( "NewsDOM attachment %s recognition result: %s", - attachment.id, + attachment_id, result, ) - except Exception: + except Exception as exc: + if isinstance(exc, DBAPIError) and exc.connection_invalidated: + raise await session.rollback() + # Rollback expires the entire prefetched batch, including IDs + # and email relationships; later rows need explicit async loads. + reload_pending_rows = True logger.error( "NewsDOM attachment %s recognition raised.", - getattr(attachment, "id", "?"), + attachment_id, exc_info=True, ) + self._attachment_cursor = attachment_id def _pending_attachment_statement(self, after_id: int | None): """Build the next deterministic attachment batch query.""" @@ -525,10 +572,19 @@ async def _load_pending_attachments( async def _sweep_documents(self, session: AsyncSession) -> None: """Process a bounded, starvation-free batch of pending documents.""" rows = await self._load_pending_documents(session) - if rows: - self._document_cursor = rows[-1].document_id - for document in rows: + pending_rows = [(document.document_id, document) for document in rows] + reload_pending_rows = False + for document_id, document in pending_rows: try: + if reload_pending_rows: + result_rows = await session.execute( + self._pending_document_statement(None).where( + Document.document_id == document_id + ) + ) + document = result_rows.scalar_one_or_none() + if document is None: + continue result = await process_pending_document( session=session, document=document, @@ -539,16 +595,20 @@ async def _sweep_documents(self, session: AsyncSession) -> None: if result != RESULT_PENDING: logger.info( "NewsDOM document %s recognition result: %s", - document.document_id, + document_id, result, ) - except Exception: + except Exception as exc: + if isinstance(exc, DBAPIError) and exc.connection_invalidated: + raise await session.rollback() + reload_pending_rows = True logger.error( "NewsDOM document %s recognition raised.", - getattr(document, "document_id", "?"), + document_id, exc_info=True, ) + self._document_cursor = document_id def _pending_document_statement(self, after_id: str | None): """Build the next deterministic workspace-document batch query.""" diff --git a/backend/tests/test_attachment_parser.py b/backend/tests/test_attachment_parser.py index 4eeb27228..c29098399 100644 --- a/backend/tests/test_attachment_parser.py +++ b/backend/tests/test_attachment_parser.py @@ -1,10 +1,10 @@ import base64 +from random import Random import pytest from services.attachment_parser import ( _safe_filename, - MAX_ATTACHMENT_PARSE_SOURCE_BYTES, MAX_ATTACHMENT_PARSE_SOURCE_CHARS, decode_deferred_attachment_payload, get_attachment_parser_manifest, @@ -161,6 +161,20 @@ def test_unsupported_binary_attachment_is_visible_without_raw_bytes(): assert result.parse_error_code == "unsupported_content_type" +@pytest.mark.parametrize("payload_size", [20 * 1024 * 1024 + 1, 64 * 1024 * 1024]) +def test_large_pdf_source_round_trips_at_real_admission_boundaries(payload_size): + """Retain every byte above the former limit and at the actual ceiling.""" + payload = b"%PDF-" + Random(1469).randbytes(payload_size - 5) + result = parse_email_attachment( + filename="boundary.pdf", content_type="application/pdf", raw_content=payload + ) + + assert result.parse_status == "pdf_dom_recognition_pending" + assert result.parse_error_code is None + assert result.parse_content == "" + assert decode_deferred_attachment_payload(result.content) == payload + + def test_pdf_attachment_is_deferred_pending_newsdom_recognition(): raw = b"%PDF-1.7 raw bytes" result = parse_email_attachment( @@ -207,10 +221,11 @@ def test_invalid_pdf_payload_is_rejected_before_deferred_recognition(): def test_oversized_pdf_payload_is_not_retained(): + """Reject one byte above 64 MiB without persisting the rejected source.""" result = parse_email_attachment( filename="huge.pdf", content_type="application/pdf", - raw_content=b"%PDF-" + b"A" * MAX_ATTACHMENT_PARSE_SOURCE_BYTES, + raw_content=b"%PDF-" + Random(1469).randbytes(64 * 1024 * 1024 - 4), ) assert result.content == "" diff --git a/backend/tests/test_attachment_source_postgres.py b/backend/tests/test_attachment_source_postgres.py new file mode 100644 index 000000000..20a3d35a3 --- /dev/null +++ b/backend/tests/test_attachment_source_postgres.py @@ -0,0 +1,462 @@ +"""Real published PDF retention on a migrated database, without recognition claims.""" + +import asyncio +import base64 +import hashlib +import logging +from datetime import datetime, timezone +from pathlib import Path + +import httpx +import pytest +from sqlalchemy import select, text +from sqlalchemy.exc import DBAPIError +from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine +from sqlalchemy.orm import selectinload + +from db.models import Attachment, Document, Email, Workspace +from services.attachment_parser import parse_email_attachment +import services.newsdom_client as newsdom_client_module +import services.newsdom_worker as newsdom_worker_module +from services.newsdom_pdf_recognition import NewsdomRuntimeConfig +from services.newsdom_worker import process_pending_attachment, process_pending_document +from tests.test_email_read_state_migration_postgres import ( + _run_migrations, + fresh_database_url as fresh_database_url, +) + +pytestmark = pytest.mark.postgres + +NASA_PDF_URL = "https://www.nasa.gov/wp-content/uploads/2019/11/earth_at_night_508.pdf" +NASA_PDF_SIZE = 40_758_835 +NASA_PDF_SHA256 = "8e622ca8f6d1ba0cf809549bddfee69e6754c3a3480d151c1fb54baf49b09be0" + + +@pytest.fixture(scope="session") +def published_pdf_bytes(pytestconfig): + """Reuse a verified public-domain corpus; reject changed or oversized downloads.""" + cache_file = Path(pytestconfig.cache.makedir("attachment_source")) / "earth_at_night_508.pdf" + if cache_file.exists(): + with cache_file.open("rb") as cache_stream: + pdf_bytes = cache_stream.read(NASA_PDF_SIZE + 1) + else: + payload_buffer = bytearray() + with httpx.stream( + "GET", NASA_PDF_URL, headers={"Accept-Encoding": "identity"}, + follow_redirects=False, trust_env=False, timeout=60, + ) as response: + assert response.status_code == 200 + assert response.headers["content-type"].split(";", 1)[0] == "application/pdf" + for response_chunk in response.iter_bytes(chunk_size=64 * 1024): + assert len(payload_buffer) + len(response_chunk) <= NASA_PDF_SIZE + payload_buffer.extend(response_chunk) + pdf_bytes = bytes(payload_buffer) + assert len(pdf_bytes) == NASA_PDF_SIZE + assert hashlib.sha256(pdf_bytes).hexdigest() == NASA_PDF_SHA256 + if not cache_file.exists(): + cache_file.write_bytes(pdf_bytes) + return pdf_bytes + + +@pytest.mark.parametrize("source_kind", ["attachment", "document"]) +@pytest.mark.asyncio +async def test_published_pdf_survives_pending_rejection_and_transaction_rollback( + fresh_database_url, published_pdf_bytes, source_kind, monkeypatch, caplog, +): + """Catch discarded/truncated bytes and identity changes across committed sessions.""" + _run_migrations(fresh_database_url) + engine = create_async_engine(fresh_database_url) + session_factory = async_sessionmaker(engine, expire_on_commit=False) + parsed_source = parse_email_attachment( + filename="earth_at_night_508.pdf", content_type="application/pdf", + raw_content=published_pdf_bytes, + ) + assert parsed_source.parse_status == "pdf_dom_recognition_pending" + workspace = Workspace(workspace_id="workspace-pdf-retention", workspace_name="PDF retention") + if source_kind == "attachment": + email = Email( + user_id="pdf-retention-user", organization_id="pdf-retention-org", + message_id="pdf-retention-source", sender="NASA", subject="Earth at Night", + body="Earth at Night", date=datetime(2019, 12, 9, tzinfo=timezone.utc), + ) + source_record = Attachment( + email=email, filename=parsed_source.filename, content=parsed_source.content, + content_type=parsed_source.content_type, parse_status=parsed_source.parse_status, + parser_key=parsed_source.parser_key, parse_content_type=parsed_source.parse_content_type, + ) + source_key = Attachment.id + content_field, status_field = "content", "parse_status" + processor, argument_name = process_pending_attachment, "attachment" + else: + source_record = Document( + workspace_entity=workspace, organization_id="pdf-retention-org", + document_name=parsed_source.filename, document_type="pdf", + document_content=parsed_source.content, document_status=parsed_source.parse_status, + ) + source_key = Document.document_id + content_field, status_field = "document_content", "document_status" + processor, argument_name = process_pending_document, "document" + + async def no_provider(_session, _organization_id): + """Exercise the real worker's unavailable-provider branch.""" + return None + + async def configured_provider(_session, _organization_id): + """Select a provider; its real client must reject before network validation.""" + return NewsdomRuntimeConfig( + base_url="https://newsdom.example.com", api_token=None, + request_language="auto", recognition_mode="auto", provider_name="primary", + ) + + async def forbidden_network(*_args, **_kwargs): + """Fail if oversized recognition reaches DNS or outbound transport.""" + pytest.fail("oversized source reached provider network validation") + + monkeypatch.setattr( + newsdom_client_module, "validate_newsdom_base_url_details_async", forbidden_network + ) + + try: + async with session_factory.begin() as session: + session.add_all([workspace, source_record]) + await session.flush() + source_identity = getattr(source_record, source_key.key) + source_statement = select(type(source_record)).where(source_key == source_identity) + if source_kind == "attachment": + source_statement = source_statement.options(selectinload(Attachment.email)) + async with engine.connect() as connection: + index_valid = await connection.scalar(text( + "SELECT bool_and(index_entry.indisvalid) FROM pg_index AS index_entry " + "JOIN pg_class AS index_object ON index_object.oid = index_entry.indexrelid " + "JOIN pg_am AS access_method ON access_method.oid = index_object.relam " + "WHERE index_entry.indrelid = 'email_attachments'::regclass " + "AND access_method.amname = 'gin'" + )) + assert index_valid is True + + for config_resolver, expected_outcome, expected_status in ( + (no_provider, "pending", "pdf_dom_recognition_pending"), + (configured_provider, "failed", "pdf_dom_recognition_failed"), + ): + async with session_factory.begin() as session: + persisted_source = (await session.scalars(source_statement)).one() + result = await processor( + session=session, **{argument_name: persisted_source}, + config_resolver=config_resolver, + ) + assert result == expected_outcome + assert getattr(persisted_source, status_field) == expected_status + if source_kind == "attachment" and expected_outcome == "failed": + assert persisted_source.parse_error_code == "provider_payload_size_exceeded" + async with session_factory() as session: + persisted_source = (await session.scalars(source_statement)).one() + retained_bytes = base64.b64decode(getattr(persisted_source, content_field), validate=True) + assert retained_bytes == published_pdf_bytes + assert getattr(persisted_source, source_key.key) == source_identity + assert getattr(persisted_source, status_field) == expected_status + + async with session_factory() as session: + persisted_source = (await session.scalars(source_statement)).one() + setattr(persisted_source, content_field, "") + await session.flush() + await session.rollback() + async with session_factory() as session: + persisted_source = (await session.scalars(source_statement)).one() + retained_bytes = base64.b64decode(getattr(persisted_source, content_field), validate=True) + assert retained_bytes == published_pdf_bytes + assert getattr(persisted_source, source_key.key) == source_identity + assert getattr(persisted_source, status_field) == "pdf_dom_recognition_failed" + + async with session_factory.begin() as session: + persisted_source = (await session.scalars(source_statement)).one() + setattr(persisted_source, status_field, "pdf_dom_recognition_pending") + if source_kind == "attachment": + persisted_source.parse_error_code = None + next_source = Attachment( + email=persisted_source.email, filename=parsed_source.filename, + content=parsed_source.content, content_type="application/pdf", + parse_status="pdf_dom_recognition_pending", parser_key=parsed_source.parser_key, + parse_content_type=parsed_source.parse_content_type, + ) + else: + next_source = Document( + document_id=f"{source_identity}_next", workspace_id=persisted_source.workspace_id, + organization_id=persisted_source.organization_id, + document_name=parsed_source.filename, document_type="pdf", + document_content=parsed_source.content, document_status="pdf_dom_recognition_pending", + ) + session.add(next_source) + await session.flush() + next_identity = getattr(next_source, source_key.key) + + resolve_count = 0 + + async def fail_first_transaction(session, organization_id): + """Abort the real first transaction; the next item must still be processed.""" + nonlocal resolve_count + resolve_count += 1 + if resolve_count == 1: + await session.execute(text("SELECT 1 / 0")) + return await configured_provider(session, organization_id) + + monkeypatch.setattr(newsdom_worker_module, "AsyncSessionLocal", session_factory) + monkeypatch.setattr(newsdom_worker_module, "engine", engine) + worker = newsdom_worker_module.NewsdomRecognitionWorker( + config_resolver=fail_first_transaction, + ) + with caplog.at_level(logging.ERROR, logger="services.newsdom_worker"): + await worker._sweep() + assert resolve_count == 2 + error_records = [ + record for record in caplog.records + if record.name == "services.newsdom_worker" and record.levelno >= logging.ERROR + ] + assert len(error_records) == 1 + assert "MissingGreenlet" not in caplog.text + for record_identity, expected_status in ( + (source_identity, "pdf_dom_recognition_pending"), + (next_identity, "pdf_dom_recognition_failed"), + ): + async with session_factory() as session: + persisted_source = await session.get(type(source_record), record_identity) + retained_bytes = base64.b64decode(getattr(persisted_source, content_field), validate=True) + assert retained_bytes == published_pdf_bytes + assert getattr(persisted_source, source_key.key) == record_identity + assert getattr(persisted_source, status_field) == expected_status + finally: + await engine.dispose() + + +async def _persist_pending_published_pdf(session_factory, published_pdf_bytes): + """Store the unchanged corpus using the real parser and migrated source schema.""" + parsed_source = parse_email_attachment( + filename="earth_at_night_508.pdf", content_type="application/pdf", + raw_content=published_pdf_bytes, + ) + async with session_factory.begin() as session: + source_record = Attachment( + email=Email( + user_id="pdf-lease-user", organization_id="pdf-lease-org", + message_id="pdf-lease-source", sender="NASA", subject="Earth at Night", + body="Earth at Night", date=datetime(2019, 12, 9, tzinfo=timezone.utc), + ), + filename=parsed_source.filename, content=parsed_source.content, + content_type=parsed_source.content_type, parse_status=parsed_source.parse_status, + parser_key=parsed_source.parser_key, parse_content_type=parsed_source.parse_content_type, + ) + session.add(source_record) + await session.flush() + return source_record.id + + +@pytest.mark.parametrize("transaction_outcome", ["commit", "rollback"]) +@pytest.mark.asyncio +async def test_sweep_lease_survives_item_transaction_and_releases_on_owning_backend( + fresh_database_url, published_pdf_bytes, transaction_outcome, monkeypatch, caplog, +): + """An unrelated pooled reader must not inherit or strand the worker's lease.""" + _run_migrations(fresh_database_url) + engine = create_async_engine(fresh_database_url, pool_size=2, max_overflow=0) + session_factory = async_sessionmaker(engine, expire_on_commit=False) + probe_engine = create_async_engine(fresh_database_url, pool_size=1, max_overflow=0) + probe_factory = async_sessionmaker(probe_engine, expire_on_commit=False) + document_phase_entered = asyncio.Event() + continue_document_phase = asyncio.Event() + worker_backend_ids = [] + + async def resolve_unavailable_provider(session, _organization_id): + """Exercise a real per-item commit or an aborted PostgreSQL transaction.""" + worker_backend_ids.append(await session.scalar(text("SELECT pg_backend_pid()"))) + if transaction_outcome == "rollback": + await session.execute(text("SELECT 1 / 0")) + return None + + monkeypatch.setattr(newsdom_worker_module, "AsyncSessionLocal", session_factory) + monkeypatch.setattr(newsdom_worker_module, "engine", engine) + worker = newsdom_worker_module.NewsdomRecognitionWorker( + config_resolver=resolve_unavailable_provider, + ) + original_document_sweep = worker._sweep_documents + + async def pause_before_document_phase(session): + """Let a concurrent reader borrow a pool connection between real phases.""" + document_phase_entered.set() + await continue_document_phase.wait() + await original_document_sweep(session) + + monkeypatch.setattr(worker, "_sweep_documents", pause_before_document_phase) + sweep_task = None + try: + source_identity = await _persist_pending_published_pdf(session_factory, published_pdf_bytes) + + with caplog.at_level(logging.ERROR, logger="services.newsdom_worker"): + sweep_task = asyncio.create_task(worker._sweep()) + await asyncio.wait_for(document_phase_entered.wait(), timeout=30) + async with session_factory() as pooled_reader, probe_factory() as replica_probe: + reader_backend_id = await pooled_reader.scalar(text("SELECT pg_backend_pid()")) + try: + assert await newsdom_worker_module._try_acquire_sweep_lease(replica_probe) is False + continue_document_phase.set() + await asyncio.wait_for(sweep_task, timeout=30) + released_after_sweep = await newsdom_worker_module._try_acquire_sweep_lease( + replica_probe + ) + if released_after_sweep: + await newsdom_worker_module._release_sweep_lease(replica_probe) + assert released_after_sweep is True, ( + "completed sweep stranded its lease on a pooled reader: " + f"worker backends={worker_backend_ids}, reader backend={reader_backend_id}" + ) + assert worker_backend_ids and reader_backend_id not in worker_backend_ids + finally: + # Only this test's reader can inherit the known worker lock on a failing build. + if reader_backend_id in worker_backend_ids: + await newsdom_worker_module._release_sweep_lease(pooled_reader) + + async with session_factory() as session: + retained_source = await session.get(Attachment, source_identity) + assert retained_source.id == source_identity + assert retained_source.parse_status == "pdf_dom_recognition_pending" + assert base64.b64decode(retained_source.content, validate=True) == published_pdf_bytes + worker_errors = [ + record for record in caplog.records + if record.name == "services.newsdom_worker" and record.levelno >= logging.ERROR + ] + assert len(worker_errors) == (1 if transaction_outcome == "rollback" else 0) + finally: + continue_document_phase.set() + if sweep_task is not None: + await asyncio.gather(sweep_task, return_exceptions=True) + await probe_engine.dispose() + await engine.dispose() + + +@pytest.mark.parametrize( + "exit_mode", [ + "complete", "acquire_cancel", "processing_cancel", "disconnect", "unlock_error", + "close_before_invalidation", + ], +) +@pytest.mark.asyncio +async def test_single_connection_sweep_releases_lease_after_completion_or_interruption( + fresh_database_url, published_pdf_bytes, exit_mode, monkeypatch, +): + """A real one-slot pool must recover without stranded locks or lost PDF bytes.""" + _run_migrations(fresh_database_url) + engine = create_async_engine( + fresh_database_url, pool_size=1, max_overflow=0, pool_timeout=1, + ) + provider_entered = asyncio.Event() + close_entered = asyncio.Event() + allow_session_close = asyncio.Event() + close_observations = [] + + class ObservedCloseSession(AsyncSession): + """Keep real SQLAlchemy exit behavior and observe its pre-rollback connection.""" + + async def close(self): + """A gated cleanup models rollback waiting for an unresponsive backend.""" + if self.info.get("lease_cleanup_probe"): + close_observations.append(self.bind.invalidated) + close_entered.set() + await allow_session_close.wait() + await super().close() + + session_factory = async_sessionmaker(engine, class_=ObservedCloseSession, expire_on_commit=False) + probe_engine = create_async_engine(fresh_database_url, pool_size=1, max_overflow=0) + probe_factory = async_sessionmaker(probe_engine, expire_on_commit=False) + worker_backend_ids = [] + document_phase_calls = [] + monkeypatch.setattr(newsdom_worker_module, "engine", engine) + monkeypatch.setattr(newsdom_worker_module, "AsyncSessionLocal", session_factory) + original_acquire = newsdom_worker_module._try_acquire_sweep_lease + original_release = newsdom_worker_module._release_sweep_lease + + async def acquire_then_record(session): + """Model a lost acknowledgement only after PostgreSQL actually grants the lock.""" + acquired = await original_acquire(session) + assert acquired is True + worker_backend_ids.append(await session.scalar(text("SELECT pg_backend_pid()"))) + if exit_mode == "acquire_cancel": + raise asyncio.CancelledError("controlled acquisition cancellation") + return acquired + + async def resolve_unavailable_provider(session, _organization_id): + """Cancel or disconnect the real work transaction without calling a model.""" + if exit_mode == "close_before_invalidation": + session.info["lease_cleanup_probe"] = True + provider_entered.set() + await asyncio.Event().wait() + if exit_mode == "processing_cancel": + raise asyncio.CancelledError("controlled processing cancellation") + if exit_mode == "disconnect": + async with probe_engine.connect() as probe_connection: + assert await probe_connection.scalar( + text("SELECT pg_terminate_backend(:worker_backend_id)"), + {"worker_backend_id": worker_backend_ids[0]}, + ) is True + await session.scalar(text("SELECT 1")) + pytest.fail("terminated worker backend unexpectedly remained usable") + return None + + async def abort_unlock_transaction(session): + """Leave release unconfirmed using an actual PostgreSQL statement failure.""" + await session.execute(text("SELECT 1 / 0")) + + worker = newsdom_worker_module.NewsdomRecognitionWorker( + config_resolver=resolve_unavailable_provider, + ) + original_document_sweep = worker._sweep_documents + + async def record_document_phase(session): + """Track whether a lost lease incorrectly permits the second worker phase.""" + document_phase_calls.append(True) + await original_document_sweep(session) + + monkeypatch.setattr(worker, "_sweep_documents", record_document_phase) + monkeypatch.setattr(newsdom_worker_module, "_try_acquire_sweep_lease", acquire_then_record) + if exit_mode == "unlock_error": + monkeypatch.setattr(newsdom_worker_module, "_release_sweep_lease", abort_unlock_transaction) + sweep_task = None + try: + source_identity = await _persist_pending_published_pdf(session_factory, published_pdf_bytes) + if exit_mode == "close_before_invalidation": + sweep_task = asyncio.create_task(worker._sweep()) + await asyncio.wait_for(provider_entered.wait(), timeout=30) + sweep_task.cancel() + await asyncio.wait_for(close_entered.wait(), timeout=30) + try: + assert close_observations == [True], "session close started before lease invalidation" + finally: + allow_session_close.set() + with pytest.raises(asyncio.CancelledError): + await asyncio.wait_for(sweep_task, timeout=30) + elif exit_mode == "complete": + await asyncio.wait_for(worker._sweep(), timeout=30) + elif exit_mode in {"acquire_cancel", "processing_cancel"}: + with pytest.raises(asyncio.CancelledError): + await asyncio.wait_for(worker._sweep(), timeout=30) + else: + with pytest.raises(DBAPIError) as raised_error: + await asyncio.wait_for(worker._sweep(), timeout=30) + if exit_mode == "disconnect": + assert raised_error.value.connection_invalidated is True + assert len(document_phase_calls) == (1 if exit_mode in {"complete", "unlock_error"} else 0) + + async with probe_factory() as replica_probe: + acquired_after_sweep = await original_acquire(replica_probe) + assert acquired_after_sweep is True + await original_release(replica_probe) + async with session_factory() as session: + next_backend_id = await session.scalar(text("SELECT pg_backend_pid()")) + assert (next_backend_id == worker_backend_ids[0]) == (exit_mode == "complete") + retained_source = await session.get(Attachment, source_identity) + assert retained_source.parse_status == "pdf_dom_recognition_pending" + assert base64.b64decode(retained_source.content, validate=True) == published_pdf_bytes + finally: + allow_session_close.set() + if sweep_task is not None: + sweep_task.cancel() + await asyncio.gather(sweep_task, return_exceptions=True) + await probe_engine.dispose() + await engine.dispose() diff --git a/backend/tests/test_newsdom_client.py b/backend/tests/test_newsdom_client.py index cda13ec35..46b7e5d9b 100644 --- a/backend/tests/test_newsdom_client.py +++ b/backend/tests/test_newsdom_client.py @@ -4,12 +4,17 @@ request-time configuration guards. """ +from random import Random + +import httpx import pytest from core.config import settings +import services.newsdom_client as newsdom_client_module from services.newsdom_client import ( NEWSDOM_BASE_URL_NOT_ALLOWED, NewsdomConfigurationError, + NewsdomPayloadTooLargeError, NewsdomRequestError, _normalize_newsdom_base_url, request_pdf_dom, @@ -86,6 +91,101 @@ async def test_request_pdf_dom_rejects_empty_payload(newsdom_allowlist): ) +@pytest.mark.asyncio +async def test_request_pdf_dom_rejects_payload_above_sidecar_contract_before_network( + newsdom_allowlist, monkeypatch, +): + """One byte over the real client limit must not start DNS or HTTP.""" + async def reject_validation(_base_url): + """Fail if admission reaches address validation.""" + pytest.fail("oversized payload reached address validation") + + def reject_transport(_validated): + """Fail if admission constructs an HTTP transport.""" + pytest.fail("oversized payload reached HTTP transport") + + monkeypatch.setattr( + newsdom_client_module, "validate_newsdom_base_url_details_async", reject_validation + ) + monkeypatch.setattr(newsdom_client_module, "_PinnedNewsdomAsyncTransport", reject_transport) + with pytest.raises(NewsdomPayloadTooLargeError): + await request_pdf_dom( + base_url="https://newsdom.example.com", + api_token=None, + pdf_bytes=b"%PDF-" + Random(1469).randbytes(20 * 1024 * 1024 - 4), + ) + + +@pytest.mark.asyncio +async def test_request_pdf_dom_sends_exactly_twenty_mib_to_transport(monkeypatch): + """Keep the supported upper boundary intact in the real multipart request.""" + payload = b"%PDF-" + Random(1469).randbytes(20 * 1024 * 1024 - 5) + seen_requests = [] + + async def validated_base_url(base_url): + """Return a fixed validated identity without external DNS.""" + return newsdom_client_module.ValidatedNewsdomBaseURL( + normalized_url=base_url, + hostname="newsdom.example.com", + port=443, + addresses=("93.184.216.34",), + ) + + async def receive_request(request): + """Observe the outgoing body without a provider or fabricated parse.""" + seen_requests.append((request.method, str(request.url))) + request_body = await request.aread() + assert payload in request_body + assert len(request_body) > len(payload) + assert request.headers["content-type"].startswith("multipart/form-data;") + return httpx.Response(200, json={"pages": []}) + + monkeypatch.setattr( + newsdom_client_module, "validate_newsdom_base_url_details_async", validated_base_url + ) + monkeypatch.setattr( + newsdom_client_module, "_PinnedNewsdomAsyncTransport", + lambda _validated: httpx.MockTransport(receive_request), + ) + assert await request_pdf_dom( + base_url="https://newsdom.example.com", api_token=None, pdf_bytes=payload + ) == {"pages": []} + assert seen_requests == [("POST", "https://newsdom.example.com/parse")] + + +@pytest.mark.asyncio +async def test_request_pdf_dom_maps_provider_413_to_payload_limit(monkeypatch): + """A stricter deployed provider limit must remain a persistent size rejection.""" + async def validated_base_url(base_url): + """Return a fixed validated identity without external DNS.""" + return newsdom_client_module.ValidatedNewsdomBaseURL( + normalized_url=base_url, + hostname="newsdom.example.com", + port=443, + addresses=("93.184.216.34",), + ) + + async def reject_payload(request): + """Model a deployed provider whose accepted payload is smaller than ours.""" + await request.aread() + return httpx.Response(413) + + monkeypatch.setattr( + newsdom_client_module, "validate_newsdom_base_url_details_async", validated_base_url + ) + monkeypatch.setattr( + newsdom_client_module, "_PinnedNewsdomAsyncTransport", + lambda _validated: httpx.MockTransport(reject_payload), + ) + + with pytest.raises(NewsdomPayloadTooLargeError): + await request_pdf_dom( + base_url="https://newsdom.example.com", + api_token=None, + pdf_bytes=b"%PDF-1.7", + ) + + @pytest.mark.asyncio async def test_request_pdf_dom_raises_config_error_without_base_url(newsdom_allowlist): with pytest.raises(NewsdomConfigurationError): diff --git a/backend/tests/test_newsdom_worker.py b/backend/tests/test_newsdom_worker.py index 0693d395c..638970f4b 100644 --- a/backend/tests/test_newsdom_worker.py +++ b/backend/tests/test_newsdom_worker.py @@ -1,20 +1,25 @@ """Unit tests for the NewsDOM recognition worker's per-item processing. -Fully mocked: in-memory models, an injected async config resolver, and a canned -sidecar ``request_fn`` — no database, no network. Covers the fail-closed +In-memory models, an injected async config resolver, and canned sidecar +responses or the real pre-network size guard — no database, no network. Covers the fail-closed outcomes (unconfigured -> pending, bad payload -> failed, empty response -> failed) that keep a pending PDF from ever masquerading as parsed. """ import asyncio import base64 +import logging +from random import Random from types import SimpleNamespace +from unittest.mock import AsyncMock import pytest +from sqlalchemy.exc import DBAPIError from db.models import Attachment, Document, Email from services.content_graph import ContentSegment, ParseResult from services.newsdom_client import NewsdomConfigurationError +import services.newsdom_client as newsdom_client_module from services.newsdom_pdf_recognition import ( PDF_DOM_RECOGNITION_FAILED_STATUS, PDF_DOM_RECOGNITION_PENDING_STATUS, @@ -77,14 +82,17 @@ def _pending_attachment( return attachment -def _pending_document(document_id: str, *, organization_id: str = "org-1") -> Document: +def _pending_document( + document_id: str, *, organization_id: str = "org-1", payload: bytes = b"%PDF-1.7 fake" +) -> Document: + """Create an in-memory pending document with the supplied unit-test bytes.""" return Document( document_id=document_id, workspace_id="ws-1", organization_id=organization_id, document_name="news.pdf", document_type="pdf", - document_content=base64.b64encode(b"%PDF-1.7 fake").decode("ascii"), + document_content=base64.b64encode(payload).decode("ascii"), document_status=PDF_DOM_RECOGNITION_PENDING_STATUS, ) @@ -99,6 +107,10 @@ def scalars(self): def all(self): return self._rows + def scalar_one_or_none(self): + """Return the optional row from a controlled pending-record reload.""" + return self._rows[0] if self._rows else None + class _SequenceSession: def __init__(self, row_batches): @@ -220,6 +232,85 @@ async def request_fn(**_kwargs): assert attachment.parse_status != "parsed" +@pytest.mark.asyncio +@pytest.mark.parametrize("source_kind", ["attachment", "document"]) +async def test_large_source_is_retained_when_real_provider_guard_rejects( + source_kind, caplog, monkeypatch +): + """Both worker paths retain actual over-limit bytes after bounded rejection.""" + payload = b"%PDF-" + Random(1469).randbytes(20 * 1024 * 1024 - 4) + source_record = ( + _pending_attachment(payload) + if source_kind == "attachment" + else _pending_document("doc-size-boundary", payload=payload) + ) + process_source = ( + process_pending_attachment if source_kind == "attachment" else process_pending_document + ) + + async def reject_validation(_base_url): + """Fail if an oversized retained source reaches DNS validation.""" + pytest.fail("oversized retained source reached address validation") + + monkeypatch.setattr( + newsdom_client_module, "validate_newsdom_base_url_details_async", reject_validation + ) + + with caplog.at_level(logging.INFO, logger="services.newsdom_worker"): + result = await process_source( + session=object(), + **{source_kind: source_record}, + config_resolver=await _resolver_with(_config()), + request_fn=newsdom_client_module.request_pdf_dom, + ) + + assert result == RESULT_FAILED + if source_kind == "attachment": + assert source_record.parse_status == PDF_DOM_RECOGNITION_FAILED_STATUS + assert source_record.parse_error_code == "provider_payload_size_exceeded" + retained_content = source_record.content + else: + assert source_record.document_status == PDF_DOM_RECOGNITION_FAILED_STATUS + retained_content = source_record.document_content + assert base64.b64decode(retained_content, validate=True) == payload + records = [record for record in caplog.records if record.name == "services.newsdom_worker"] + assert records and all(record.levelno == logging.INFO for record in records) + + +@pytest.mark.asyncio +@pytest.mark.parametrize("source_kind", ["attachment", "document"]) +async def test_full_size_source_stays_pending_and_intact_without_provider(source_kind): + """Absence of a provider must not destroy or process a retained 64 MiB source.""" + payload = b"%PDF-" + Random(1469).randbytes(64 * 1024 * 1024 - 5) + source_record = ( + _pending_attachment(payload) + if source_kind == "attachment" + else _pending_document("doc-unconfigured-boundary", payload=payload) + ) + process_source = ( + process_pending_attachment if source_kind == "attachment" else process_pending_document + ) + + async def reject_request(**_kwargs): + """Fail if an unconfigured record reaches recognition.""" + pytest.fail("unconfigured source reached recognition") + + result = await process_source( + session=object(), + **{source_kind: source_record}, + config_resolver=await _resolver_with(None), + request_fn=reject_request, + ) + assert result == RESULT_PENDING + if source_kind == "attachment": + assert source_record.parse_status == PDF_DOM_RECOGNITION_PENDING_STATUS + retained_content = source_record.content + else: + assert source_record.document_status == PDF_DOM_RECOGNITION_PENDING_STATUS + retained_content = source_record.document_content + assert base64.b64decode(retained_content, validate=True) == payload + + @pytest.mark.asyncio async def test_attachment_orphan_and_rejected_configuration_stay_visible(): orphan = Attachment( @@ -498,7 +589,7 @@ async def broken_resolver(_session, _organization_id): @pytest.mark.asyncio async def test_postgresql_lease_helpers_and_non_postgresql_fallback(): - postgres = _LeaseSession(scalar_result=1) + postgres = _LeaseSession(scalar_result=True) sqlite = _LeaseSession(dialect_name="sqlite") assert await newsdom_worker_module._try_acquire_sweep_lease(postgres) is True @@ -522,7 +613,8 @@ def get_bind(self): async def test_worker_sweep_honors_lease_outcome( monkeypatch, lease, expected_sweeps, expected_releases ): - session = object() + session = SimpleNamespace(rollback=AsyncMock()) + connection = SimpleNamespace(invalidate=AsyncMock()) calls = [] releases = [] worker = NewsdomRecognitionWorker() @@ -530,7 +622,11 @@ async def test_worker_sweep_honors_lease_outcome( monkeypatch.setattr( newsdom_worker_module, "AsyncSessionLocal", - lambda: _AsyncSessionContext(session), + lambda **kwargs: _AsyncSessionContext(session) if kwargs.get("bind", connection) is connection else None, + ) + monkeypatch.setattr( + newsdom_worker_module, "engine", + SimpleNamespace(connect=lambda: _AsyncSessionContext(connection)), raising=False, ) async def acquire(actual_session): @@ -555,6 +651,137 @@ async def sweep_documents(actual_session): assert len(calls) == expected_sweeps assert len(releases) == expected_releases + assert session.rollback.await_count == expected_releases + connection.invalidate.assert_not_awaited() + + +@pytest.mark.asyncio +@pytest.mark.parametrize("failure_phase", ["acquire", "attachments", "documents", "release"]) +@pytest.mark.parametrize("failure_type", [RuntimeError, asyncio.CancelledError]) +async def test_worker_discards_connection_when_lease_lifecycle_is_uncertain( + monkeypatch, failure_phase, failure_type, +): + """Errors and cancellation must not return a potentially locked backend to the pool.""" + session = SimpleNamespace(rollback=AsyncMock()) + connection = SimpleNamespace(invalidate=AsyncMock()) + monkeypatch.setattr( + newsdom_worker_module, "engine", + SimpleNamespace(connect=lambda: _AsyncSessionContext(connection)), raising=False, + ) + monkeypatch.setattr( + newsdom_worker_module, "AsyncSessionLocal", + lambda **kwargs: _AsyncSessionContext(session) if kwargs.get("bind", connection) is connection else None, + ) + worker = NewsdomRecognitionWorker() + acquire = AsyncMock(return_value=True) + attachments = AsyncMock() + documents = AsyncMock() + release = AsyncMock() + failure = failure_type("controlled lease lifecycle failure") + phases = {"acquire": acquire, "attachments": attachments, "documents": documents, "release": release} + phases[failure_phase].side_effect = failure + monkeypatch.setattr(newsdom_worker_module, "_try_acquire_sweep_lease", acquire) + monkeypatch.setattr(newsdom_worker_module, "_release_sweep_lease", release) + monkeypatch.setattr(worker, "_sweep_attachments", attachments) + monkeypatch.setattr(worker, "_sweep_documents", documents) + + with pytest.raises(failure_type) as raised_error: + await worker._sweep() + + assert raised_error.value is failure + connection.invalidate.assert_awaited_once_with() + if failure_phase in {"acquire", "attachments"}: + documents.assert_not_awaited() + + +@pytest.mark.asyncio +@pytest.mark.parametrize("unlock_result", [False, None, 1, "true"]) +async def test_sweep_unlock_requires_explicit_ownership_confirmation(unlock_result): + """A missing lease or unverifiable response must not count as a successful unlock.""" + with pytest.raises(RuntimeError, match="lease release could not be confirmed"): + await newsdom_worker_module._release_sweep_lease(_LeaseSession(scalar_result=unlock_result)) + + +@pytest.mark.asyncio +@pytest.mark.parametrize("source_kind", ["attachment", "document"]) +async def test_item_disconnect_aborts_sweep_instead_of_reconnecting_without_lease(source_kind): + """A lost backend loses its session lock; later rows need a new leased cycle.""" + failure = DBAPIError(None, None, RuntimeError("controlled disconnect"), connection_invalidated=True) + resolver = AsyncMock(side_effect=failure) + worker = NewsdomRecognitionWorker(config_resolver=resolver) + if source_kind == "attachment": + session = _SequenceSession([[_pending_attachment(attachment_id=1), _pending_attachment(attachment_id=2)]]) + sweep = worker._sweep_attachments + else: + session = _SequenceSession([[_pending_document("doc-1"), _pending_document("doc-2")]]) + sweep = worker._sweep_documents + + with pytest.raises(DBAPIError) as raised_error: + await sweep(session) + + assert raised_error.value is failure + resolver.assert_awaited_once() + assert session.commit_count == session.rollback_count == 0 + + +@pytest.mark.asyncio +@pytest.mark.parametrize("source_kind", ["attachment", "document"]) +@pytest.mark.parametrize("completed_count", [0, 1]) +async def test_disconnect_cursor_retries_unattempted_rows_before_newer_work( + source_kind, completed_count, +): + """Resume at the last completed item, not the tail of an abandoned prefetched batch.""" + worker = NewsdomRecognitionWorker() + if source_kind == "attachment": + pending_rows = [_pending_attachment(attachment_id=index) for index in (1, 2, 3)] + cursor_name, completed_identity = "_attachment_cursor", 1 + sweep = worker._sweep_attachments + else: + pending_rows = [_pending_document(f"doc-{index}") for index in (1, 2, 3)] + cursor_name, completed_identity = "_document_cursor", "doc-1" + sweep = worker._sweep_documents + session = _SequenceSession([pending_rows, pending_rows[completed_count:]]) + failure = DBAPIError(None, None, RuntimeError("controlled disconnect"), connection_invalidated=True) + worker._config_resolver = AsyncMock(side_effect=[None] * completed_count + [failure]) + + with pytest.raises(DBAPIError): + await sweep(session) + + assert getattr(worker, cursor_name) == (completed_identity if completed_count else None) + worker._config_resolver = AsyncMock(return_value=None) + await sweep(session) + assert worker._config_resolver.await_count == 3 - completed_count + assert session.commit_count == 3 + resumed_query = session.statements[1].compile() + if completed_count: + assert completed_identity in resumed_query.params.values() + else: + assert " > " not in str(resumed_query) + + +@pytest.mark.asyncio +@pytest.mark.parametrize("source_kind", ["attachment", "document"]) +async def test_rollback_reload_skips_missing_or_no_longer_pending_source(source_kind): + """Do not process a vanished pending row; continue with the next valid cached identity.""" + worker = NewsdomRecognitionWorker( + config_resolver=AsyncMock(side_effect=[RuntimeError("controlled item failure"), None]), + ) + if source_kind == "attachment": + pending_rows = [_pending_attachment(attachment_id=index) for index in (1, 2, 3)] + cursor_name, final_identity = "_attachment_cursor", 3 + sweep = worker._sweep_attachments + else: + pending_rows = [_pending_document(f"doc-{index}") for index in (1, 2, 3)] + cursor_name, final_identity = "_document_cursor", "doc-3" + sweep = worker._sweep_documents + session = _SequenceSession([pending_rows, [], [pending_rows[2]]]) + + await sweep(session) + + assert worker._config_resolver.await_count == 2 + assert session.commit_count == session.rollback_count == 1 + assert getattr(worker, cursor_name) == final_identity + assert final_identity in session.statements[2].compile().params.values() @pytest.mark.asyncio diff --git a/docs/adr/0021-bounded-pdf-dom-upload-contract.md b/docs/adr/0021-bounded-pdf-dom-upload-contract.md index d70dcdcd7..4562c88d5 100644 --- a/docs/adr/0021-bounded-pdf-dom-upload-contract.md +++ b/docs/adr/0021-bounded-pdf-dom-upload-contract.md @@ -9,6 +9,9 @@ open-PR inventory found unrelated 0005 proposals; the decision content and immutable-release prerequisite are retained. Existing #1469 inherits this proposal and must take the rename by ancestry, not publish a second decision. +The attachment branch's complete earlier deferral wording is preserved as a +[historical proposal](../doctoring/pdf_dom_proposal_history.md), not a second +numbered ADR or a new accepted decision. ## Context diff --git a/docs/adr/0023-bounded-attachment-parse-source-contract.md b/docs/adr/0023-bounded-attachment-parse-source-contract.md new file mode 100644 index 000000000..9fd8a32e2 --- /dev/null +++ b/docs/adr/0023-bounded-attachment-parse-source-contract.md @@ -0,0 +1,185 @@ +# ADR-0023: Bounded attachment parse-source contract + +- **Status:** Proposed; #1469 is unmerged and depends on the owner stack +- **Date:** 2026-08-25 +- **Evidence updated:** 2026-09-05; remains Proposed +- **Figma file ID:** N/A — backend ingestion and evidence contract; no UX surface +- **Owners:** Naruon ingestion and data-quality maintainers +- **Depends on:** ADR-0021 and a verified immutable NewsDOM release before + increasing the provider transport bound. +- **Former proposal ID:** ADR-0006 in #1469, renamed because unrelated open + proposals already use 0006. This is the same proposal, not an accepted + superseding decision. No external service decision is accepted here. + +## Context and Problem Statement + +Naruon accepts authenticated email imports up to 64 MiB, while the deferred +attachment parser previously rejected source payloads above 20 MiB. That split +made a valid upload fail later in parsing and prevented a customer from knowing +whether a file was rejected by transport, parser admission, or an unsupported +format. Unsupported binaries are intentionally not parsed inline and must remain +metadata-only until a separately reviewed parser is available. + +A customer importing a 40 MiB published PDF must not lose the admitted source +when recognition is unavailable or rejects its size. Admission, durable storage, +provider capability, and successful extraction are distinct boundaries. The +verified NewsDOM protected-source guard is 20 MiB; its unmerged 64 MiB proposal +does not authorize Naruon to call an unreleased API. Protected integration, +immutable release, exact consumer pin, and runtime compatibility remain gates. + +## Decision Drivers + +- Preserve admitted source bytes and stable record identity across committed + worker outcomes and rollback; never relabel metadata as recognized text. +- Bound memory, transport, and storage for untrusted input while retaining + signed-session, tenant, and workspace ownership. +- Keep the provider contract under NewsDOM ownership and migration/search + storage repairs under the existing prerequisite stack. + +## Considered Options + +1. Keep 20 MiB attachment admission: bounded, but rejects files already admitted + by the email transport. Retain the old limit on protected branches until the + complete proposal satisfies release and capacity gates. +2. Retain sources up to 64 MiB, separately enforce the verified provider bound, + and publish only after owner prerequisites are complete: chosen proposal. +3. Send 64 MiB directly to an unreleased provider branch: rejected because it + bypasses the owner release and can turn accepted work into failed recognition. +4. Remove bounds or discard sources after rejection: rejected because this + exposes resource exhaustion or irreversible customer data loss. + +## Decision Outcome + +Use one 64 MiB upper bound for attachment source bytes retained for a deferred +recognition worker. The parser continues to: + +1. accept only the existing authenticated import transport; +2. retain validated PDF bytes only for the deferred NewsDOM path; +3. return `parse_size_limit_exceeded` without raw content above 64 MiB; +4. return `unsupported_content_type` with `unsupported_binary` and no raw bytes + for an unparseable content type; and +5. preserve the parser key, parse status, and error code for the Data quality + evidence surface. + +This is a bounded admission contract, not a promise that every binary format +is parseable. Adding a new parser requires its own dependency, sandbox, +provenance, and regression review. + +No provider configuration leaves the complete source pending. A real provider +size guard fails the record without changing its source bytes and before DNS or +HTTP transport. Attachments retain `provider_payload_size_exceeded`; workspace +documents currently have only a failure status, not an error-code column. +Both paths classify this expected admission rejection as operational information; +unexpected configuration/recognition failures retain their distinct handling. +Customer guidance should ask for a smaller PDF or an approved service upgrade; +actionable document error detail and retry after an upgrade remain follow-up work. + +One failed database transaction must not stop the remaining admitted sources in +the current bounded batch. Cache primitive record IDs before processing and, +only after rollback, explicitly reload the remaining pending records with their +required relationships. This preserves per-item transaction isolation without +adding queries to successful batches. A new session framework or whole-queue +reload is unnecessary; retaining expired ORM attributes reproduces the failure. + +### Sweep connection ownership + +At #1469 `1b757d5aa25c469157f8f03301964eb3061ed0fe`, real PostgreSQL +regressions reproduced a second failure: after either an item commit or rollback, +an unrelated reader borrowed the worker's still-locked backend. The worker ended +without releasing that backend's lease, so another replica could not take over. +The affected components are `NewsdomRecognitionWorker._sweep`, both per-item +sweeps, and the lease helpers; the shared database pool configuration is unchanged. + +Choose an externally held connection and bind the existing work session to it +for both phases. An ordinary item transaction error remains isolated. A lost +database connection aborts the cycle; it cannot continue on a new backend without +reacquiring leadership. Release requires an explicit true response. Cancellation, +an acquisition error, or unconfirmed release invalidates the connection before +it can return to the pool. See the real failure and recovery checks in doctoring. + +| Option | Benefit | Constraint or rejection reason | +|---|---|---| +| Keep an engine-bound work session | Existing API; no additional resource owner | Reproduced stranded lease after both transaction outcomes; rejected | +| Separate lease-only connection | Work transactions cannot return the lease holder | Needs a second pool slot and can deadlock a configured one-slot pool; rejected for this worker | +| Transaction-level lock | Automatic transaction cleanup | Per-item commits end coordination before the complete sweep; rejected | +| One externally held connection | Preserves item commits and works with one pool slot | Chosen; occupies that slot for the whole sweep and discards it after uncertain cleanup | + +This decision concerns availability, correctness, and pool capacity. It does not +promise exactly-once external recognition: loss of a database connection during +an already-running provider request can permit another replica to begin work. +Provider idempotency or durable fenced claims require separate contract evidence. + +## Consequences + +- The proposal retains attachments larger than 20 MiB through 64 MiB, but does + not claim that the current provider can recognize them. The PR stays Draft + until the prerequisites in ADR-0021 and its owner release are satisfied. +- A 64 MiB raw source can expand when base64-encoded in the existing deferred + content column; the database/object-lifecycle work must move this payload to + object storage before materially increasing the bound again. +- RFC 4648 encoding makes a 64 MiB source 89,478,488 ASCII bytes before database + and HTTP overhead. Complete-byte integrity is necessary but does not prove + acceptable heap usage, index cost, p95 latency, concurrency, or workspace quotas. +- Unsupported binaries remain visible in scoped quality counts without exposing + their bytes, identifiers, or provider content. +- The contract is independent of any Figma design and has no Storybook scene. +- Pool capacity must allow the recognition worker to retain one connection + throughout a sweep, including external recognition. Interruption can cause a + new connection on the next cycle. A rollback-triggering item error keeps the + original backend when it remains healthy. +- If runtime verification fails, stop the worker using its existing lifecycle + control and preserve pending sources. Do not restore the known leaking + lease path or discard admitted bytes. Repair and revalidate before restart. + +## Confirmation + +- `backend/tests/test_attachment_parser.py` round-trips actual 20 MiB + 1 byte + and 64 MiB unit payloads, rejects 64 MiB + 1 byte, and preserves + unsupported-binary metadata-only behavior. Synthetic bytes are unit-only. +- `backend/tests/test_newsdom_client.py` sends actual 20 MiB through multipart + transport and rejects 20 MiB + 1 byte before network validation. +- `backend/tests/test_newsdom_worker.py` uses the real size guard and verifies + complete pending/failed source retention for attachment and document paths. +- `backend/tests/test_attachment_source_postgres.py` uses the unchanged, + hash-verified 40,758,835-byte NASA *Earth at Night* PDF on freshly migrated + PostgreSQL, preserving full bytes and identity through committed pending and + rejected outcomes and transaction rollback. Two-record worker sweeps also + abort the first real transaction and verify that the next record is processed + without expired-attribute I/O. This is not recognition evidence. +- The same real corpus suite checks another pool user's backend identity, + independent-replica lease acquisition, one-slot completion, acquisition and + processing cancellation, backend termination, and failed release. Unit tests + cover both source types' disconnect handling and strict unlock responses. +- The import transport remains covered by + `backend/tests/test_email_import_service.py`. +- The PDF DOM upload contract is being integrated separately by stacked PR + #1427 / ADR-0021; #1469 inherits it by normal merge. Historical ADR-0005's + deferral rationale remains in + [the complete archived proposal](../doctoring/pdf_dom_proposal_history.md). +- Commands, source/release evidence, failure diagnoses, and remaining release + gates are recorded in [doctoring](../doctoring/bounded-attachment-parse-source-contract.md). + +## References (APA 7th) + +Fielding, R. T., Nottingham, M., & Reschke, J. (Eds.). (2022). *HTTP +semantics* (RFC 9110). Internet Engineering Task Force. +https://www.rfc-editor.org/rfc/rfc9110.html + +Josefsson, S. (2006). *The Base16, Base32, and Base64 data encodings* +(RFC 4648). Internet Engineering Task Force. +https://www.rfc-editor.org/rfc/rfc4648.html + +National Aeronautics and Space Administration. (2019). *Earth at night*. +https://www.nasa.gov/ebooks/earth-at-night/ + +PostgreSQL Global Development Group. (n.d.). *Explicit locking: Advisory locks*. +PostgreSQL 16 documentation. +https://www.postgresql.org/docs/16/explicit-locking.html#ADVISORY-LOCKS + +SQLAlchemy Authors. (n.d.). *Session basics: Committing*. SQLAlchemy 2.0 documentation. +https://docs.sqlalchemy.org/en/20/orm/session_basics.html#committing + +Souppaya, M., Scarfone, K., & Dodson, D. (2022). *Secure software development +framework (SSDF) version 1.1: Recommendations for mitigating the risk of +software vulnerabilities* (NIST Special Publication 800-218). National Institute +of Standards and Technology. https://doi.org/10.6028/NIST.SP.800-218 diff --git a/docs/adr/README.md b/docs/adr/README.md index da8978de8..c8b271730 100644 --- a/docs/adr/README.md +++ b/docs/adr/README.md @@ -15,6 +15,7 @@ govern implementation. | [ADR-0004](0004-status-weighted-calendar-conflicts.md) | Evaluate CalDAV VEVENT overlaps by occupying status; cancelled does not occupy | Accepted | `ACCEPTED-NARUON-POLICY`; advisory evaluate API only | | [ADR-0020](0020-full-document-trigram-storage.md) | Repair whole-document trigram persistence without changing ranking; measure latency before rollout | Proposed | Storage candidate; no protected integration or performance acceptance | | [ADR-0021](0021-bounded-pdf-dom-upload-contract.md) | Propose 64MiB direct PDF admission only after the required NewsDOM release is pinned | Proposed | `BLOCKED-UPSTREAM`; requires an immutable NewsDOM 64MiB release and exact Naruon pin | +| [ADR-0023](0023-bounded-attachment-parse-source-contract.md) | Retain bounded deferred sources across pending/rejected outcomes without claiming provider recognition | Proposed | Former ADR-0006; requires owner release, capacity evidence, and protected integration | The complete topic-intelligence requirements, architecture, contract, UML, conceptual ERD, security, test, and operability graph is indexed at diff --git a/docs/doctoring/bounded-attachment-parse-source-contract.md b/docs/doctoring/bounded-attachment-parse-source-contract.md new file mode 100644 index 000000000..0518625b8 --- /dev/null +++ b/docs/doctoring/bounded-attachment-parse-source-contract.md @@ -0,0 +1,382 @@ +# Bounded attachment parse-source contract + +## Customer outcome + +An email attachment between 20 MiB and 64 MiB is not rejected by a hidden +parser-only limit in the proposed ingestion change. This is not yet a released +capability. Recognition can still reject the source above the verified provider +limit; the complete admitted bytes must survive that rejection. The Data workspace reports +unsupported formats explicitly, so the next action is to add a reviewed parser +or use the original provider file rather than treating metadata as extracted +content. + +## Contract + +`MAX_ATTACHMENT_PARSE_SOURCE_BYTES` is a proposed 64 MiB Naruon retention +budget, matching the authenticated email-import ceiling. The parser is +fail-closed: + +- supported text formats are parsed inline within the existing character bound; +- validated PDF bytes may be retained only for bounded deferred NewsDOM recognition; +- unsupported binary formats return `unsupported_content_type`, + `unsupported_binary`, and empty content; +- oversized Naruon source bytes return `parse_size_limit_exceeded` and empty content; +- the NewsDOM client independently enforces the verified protected-source + 20 MiB guard, and the attachment worker persists + `provider_payload_size_exceeded` when that boundary rejects a retained PDF. + This source guard is not evidence of a released or deployed capacity. + +This separation preserves provenance without claiming that a source retained by +Naruon was accepted by an external provider. An open owner PR is evidence of a +candidate contract only; provider-size parity becomes consumable only after an +immutable NewsDOM release is verified and pinned. ADR-0021 owns the separate +direct PDF-DOM upload proposal and the same immutable-release prerequisite. + +The quality surface exposes parser key and status, not raw attachment bytes, +message IDs, attachment IDs, credentials, or customer payloads. + +## Evidence and next action + +- `backend/tests/test_attachment_parser.py` covers the Naruon 64 MiB retention + boundary and metadata-only unsupported binaries. +- `backend/tests/test_newsdom_client.py` covers provider-size preflight before + network I/O. +- `backend/tests/test_newsdom_worker.py` covers persistent provider-size failure + evidence instead of silent pending state. +- `backend/tests/test_email_import_service.py` retains import-transport coverage. +- [ADR-0021](../adr/0021-bounded-pdf-dom-upload-contract.md) remains Proposed and + blocked on the immutable owner release/pin; this document does not upgrade it. + +If a customer needs a currently unsupported format, add a dedicated parser +proposal with sandbox, dependency, provenance, and exact-head regression +evidence before changing the registry. If provider transport is raised, verify +the immutable owner release first, then bump the Naruon released-contract +boundary and rerun exact-head integration evidence. + +## 2026-09-05 owner-stack repair + +The repair starts at #1469 +`ed4bebeddf05ce1da0c76aca77448deef6254fbb` and normally merges #1427 +`cb08b1c3ea2aba8844fc29ef703c34368cc55e47`. It inherits the existing dependency, +forward migration, complete-text search storage, and PDF admission proposals. +No migration is stamped around a failure, no legacy table is fabricated, and +no index or constraint is removed to make the source fit. + +Concurrent remote merge `05c6fb2460ee69c3ce7dccd66b2b2ec0e2c66658` and +follow-up `facadfb1ce535c5d124e2a463a844942a7704ba5` independently restacked +the same original attachment delta on #1427 and restored the prerequisite +CHANGELOG. Preserve both commits by normal integration. Their admission versus +provider distinction, Proposed ADR-0023/0021 dependency, pre-network size guard, +and prerequisite release notes remain; the full-payload/database/rollback tests +and historical proposal snapshot supplement that intent. Statements calling a +source-only 20 MiB guard deployed or released are corrected using the explicit +provider receipt below, not retained as runtime claims. + +### Failure, cause, correction + +| Observation | Cause and correction | Evidence boundary | +|---|---|---| +| Fresh #1469 migration failed in `0001_initial_control_plane.py` with `relation "emails" does not exist` | Inherit the existing database-owner repair through #1427, including forward read-state and search-storage revisions; do not duplicate a local schema workaround | The pre-merge migration log is RED before any attachment test runs | +| Actual 20 MiB + 1 byte document rejection logged a warning; attachment sibling already handled it as expected admission | `process_pending_document` lacked the `NewsdomPayloadTooLargeError` branch. Add the specific failure/INFO branch before general request errors, retaining bytes | Reproduced 1 failed / 62 passed; corrected 63 passed in the strict unit suite | +| A first-item database failure stopped the actual attachment and document sweeps with `MissingGreenlet` | Rollback expired all prefetched ORM objects; reading an ID in error logging and reading later items attempted implicit async I/O. Cache primitive IDs before transactions and reload only remaining pending items after rollback | Two real-PDF PostgreSQL cases failed on checkpoint `8664cf7`; the corrected real-DB/worker suite passed 29 tests | +| New real-PDF test setup errored on `LocalPath.write_bytes` | Convert the existing pytest cache path to stdlib `Path` | Harness error, not a product RED or passing DB test | +| New real-PDF test referenced a nonexistent validation function | Use the actual `validate_newsdom_base_url_details_async` boundary, matching the existing client and unit tests | Harness error; test must subsequently reach real persistence and worker execution | + +### Provider authority + +Git refs rechecked at 2026-09-05 09:33 UTC: + +- NewsDOM protected-source `develop` is + `e06b1f3fb10903569124af011da213951e6e2473`; its `/parse` guard is 20 MiB. +- Proposed NewsDOM #665 head is + `14eb886a91702074b4a0ae1b2fc21f84cec88d37`. A PR ref proves source identity, + not merge state, deployment, or availability. +- Latest release metadata still names immutable `v0.2.0`; its tag resolves to + `c26f3db7e9176b6e698b4e686aeda79b15a010b9`. The earlier source audit found + unbounded `file.read()` there, not an immutable bounded 64 MiB API contract. + +The Naruon guard stays 20 MiB. NewsDOM must provide its protected merge, +immutable bounded release, and compatibility evidence before Naruon pins and +adopts a larger contract. No deployed capacity is inferred from these refs. +Earlier API quota failures were unknown evidence, not an empty PR set. The +completed REST inventory at 2026-09-05 09:40:34–09:41:58 UTC covered all 161 open +PRs, 162 changed-file pages, and 161 non-truncated head trees. Initial/final +head/base snapshots matched, with zero API errors or file caps. ADR-0023 was +unused; 0005 and 0006 each named multiple unrelated proposals. Recheck changed +heads and newly opened PRs before publishing; the snapshot is not an ID lock. +The 10:00:25–10:00:31 UTC incremental recheck found all 160 other PR head/base +pairs unchanged. Only #1469 moved to `facadfb1`; its ADR-0023 is the same proposal, +not a new unrelated identity collision. Its complete tree and changed-file list +were inspected, and the ending open-PR snapshot again matched the starting one. + +### Real corpus and reproducibility + +`backend/tests/test_attachment_source_postgres.py` reuses the existing isolated +`fresh_database_url` migration harness. It downloads only the fixed NASA HTTPS +URL with redirects and environment proxy inheritance disabled, bounds the read, +and checks length and SHA-256 before caching or using the book. A changed, +truncated, oversized, or unavailable corpus fails instead of silently replacing +it with synthetic content. Existing cached bytes are revalidated on every run. + +- Source: NASA, *Earth at Night* (2019), + . +- Original length: **40,758,835 bytes**, PDF 1.7, **200 pages**, not encrypted. +- SHA-256: `8e622ca8f6d1ba0cf809549bddfee69e6754c3a3480d151c1fb54baf49b09be0`. +- The acknowledgments, printed p. xiv, state that the material is public domain + and free to use. The book stays in the ignored pytest cache, not the Git tree. +- Neither truncation, repetition, compression changes, page removal, nor fake + recognized output is used. Test-scope record identifiers are isolated technical + identifiers, not customer identities. + +Run from `backend` against an isolated PostgreSQL/pgvector test service with +ephemeral test credentials in `DATABASE_URL` and `AUTH_SESSION_HMAC_SECRET`: + +```sh +uv sync --locked +uv run --frozen python scripts/migrate_db.py +uv run --frozen python scripts/migrate_db.py +uv run --frozen python -m pytest -q -W error -ra --tb=short \ + tests/test_attachment_source_postgres.py \ + tests/test_attachment_parser.py tests/test_newsdom_client.py \ + tests/test_newsdom_worker.py tests/test_email_import_service.py \ + tests/test_email_parser.py +``` + +Both attachment and workspace-document cases commit the original source, load +it in a new session, keep it pending without a provider, then commit the actual +client's pre-network size rejection. Fresh sessions compare every decoded byte, +status, and record identity. A flushed destructive content change is rolled +back, and another session must see the original source and failed status. +The migrated GIN attachment search index remains valid during the real writes. +This verifies retention, not successful PDF recognition, signed browser upload, +cross-tenant authorization, provider-network behavior, or a latency target. + +### Actual worker rollback regression + +An independent read-only review identified a gap in the manual rollback check: +it did not run the worker's exception handler. The existing one-item mock sweep +test also had no SQLAlchemy expiration behavior. With two persisted real-PDF +records, the first transaction is intentionally aborted by `SELECT 1 / 0` and +the actual `NewsdomRecognitionWorker._sweep()` must continue to the second item. +Both attachment and document variants failed with `MissingGreenlet` against +checkpoint `8664cf7fdaa60c81e34f056f0031fd12fd92adb2`: **2 failed in 41.98 s**. +This is a new product RED, separate from the earlier fixture setup errors. + +SQLAlchemy rollback expires loaded objects even with `expire_on_commit=False`; +ordinary attribute access can therefore require forbidden implicit I/O under +AsyncSession. Both sweep methods now capture primitive IDs before processing. +After a rollback, each remaining cached ID is queried explicitly with the +existing pending-status filter and attachment email eager loading. Deleted or +no-longer-pending items are skipped. The failed item is not retried within that +sweep. That initial correction kept the cursor at the prefetched batch tail; +the later disconnect repair below advances it only after an item finishes. +Normal successful batches incur no new queries. Requerying the whole queue or changing the global +session configuration was rejected: either broadens a bounded sweep or leaves +rollback expiration unresolved (SQLAlchemy Authors, n.d.-a, n.d.-b). + +After the fix, the two real-DB cases and the existing worker suite passed +**29 tests, 0 failures, 0 errors, 0 skips in 52.83 s** with `-W error`. Each case +asserts one captured, intentional database-fault log, no `MissingGreenlet`, two +processor attempts, first-source pending status, second-source actual size +rejection, and complete bytes/identity in fresh sessions. The controlled fault +is not clean provider execution or a hidden warning waiver. PostgreSQL/network +cleanup completed. The broader exact-head rerun is recorded in the PR receipt; +the older 276-test receipt below did not cover this worker error path. + +- RED JUnit SHA-256: `58f37fb95be510995b3cf4df6a5158d3c4e3cca5a0e4e68ca0daa801a8f0677e`. +- GREEN JUnit SHA-256: `13cca9c4fac9f9053719fde2d3578f7c52d63cd68ff3068a87d82bfd44756874`. + +Exact 20 MiB, 20 MiB + 1 byte, 64 MiB, and 64 MiB + 1 byte payloads are separately +exercised by the parser/client/worker **unit** tests. They use deterministic +synthetic bytes and are not presented as structurally valid real-PDF evidence. + +### Physical-connection lease regression + +The new real PostgreSQL test failed twice against runtime head +`1b757d5aa25c469157f8f03301964eb3061ed0fe`: **2 failed in 16.31 s**. +After the real worker committed its pending attachment, its recorded backend and +an unrelated pooled reader's backend were both `79`; the rollback case recorded +`149` for both. An independent replica could not acquire the lease after either +sweep completed. These PIDs identify ephemeral test backends only, not stable +runtime identities. The full NASA PDF, normal migration chain, and search +indexes were retained. RED JUnit SHA-256: +`ce3f52eebe36323639ed80ad7e5517428f2f58efa527300b14b5e94af2ef927b`. + +SQLAlchemy returns an engine-bound session's connection to its pool when its +transaction ends. PostgreSQL session advisory locks instead remain with their +backend until explicit release or session termination; acquisition on that same +backend is reentrant. The prior unlock ignored its false response on another +backend (SQLAlchemy Authors, n.d.-c; PostgreSQL Global Development Group, n.d.). +Keeping the Python session object therefore did not preserve lease ownership. + +The repair uses `engine.connect()` around the complete worker cycle and passes +that connection to the existing `AsyncSessionLocal(bind=connection)`. Per-item +commit and healthy rollback stay intact. A SQLAlchemy `DBAPIError` marked +`connection_invalidated` escapes either per-item handler and stops the cycle. +After successful phases, rollback clears the final read transaction before an +explicitly confirmed unlock. An error or cancellation during acquisition, work, +or release invalidates the held connection before session close, including +cancellation while acquisition may have succeeded. +No new dependency, pool setting, service, retry loop, or model time limit is added. + +Independent review then exposed two defects in the first connection repair. +Advancing a cursor to the prefetched batch tail before processing could skip +unattempted records after disconnect; a continuously growing queue might never +revisit them. Both cursors now advance only after completed work or a healthy +item rollback. Second, an outer error handler ran only after SQLAlchemy's +shielded session close, whose rollback could wait on an unresponsive backend. +The handler now invalidates inside the session context, before that close starts. +Four attachment/document resume cases and one real-PG cancellation case with a +controlled close gate failed before these changes: **5 failed in 12.06 s**. +The close gate tests ordering; it does not claim a real network black-hole test. +After correction the same cases passed **5 tests in 11.01 s**. RED/GREEN JUnit +SHA-256 values are respectively +`928fd980b386358a22aa79f392336d199f401fd14b52c86f2e36d32e64c13e73` and +`098492308dc900cdb86f28cee3dfe43c53963f9baa70839c378837a0517655f0`. + +```mermaid +sequenceDiagram + participant Worker as Recognition worker + participant Backend as Held PostgreSQL backend + participant Reader as Concurrent pool reader + Worker->>Backend: Acquire sweep lease + loop Each admitted source + Worker->>Backend: Process, then commit or rollback + end + Reader->>Backend: Cannot borrow the held connection + Worker->>Backend: Confirm unlock, then return connection + Note over Worker,Backend: On uncertain ownership, invalidate and stop +``` + +The first corrected run passed **43 tests in 43.92 s**, including the two real +contention cases and existing worker/retention tests. This intermediate result +predates the extended lifecycle suite; final exact-head evidence belongs in the +PR receipt. Additional checks exercise a one-slot pool with the real corpus, +completed work, cancellation after actual acquisition, processing cancellation, +termination of the test-owned backend, and an actual failed unlock transaction. +They verify fresh-replica acquisition, connection replacement only after failure, +and unchanged source bytes/status. Unit tests also require explicit true unlock +responses and prohibit either source handler from continuing after disconnect. + +An initial unit harness revision expected the new bind keyword too early and +failed in setup; it is not the product RED. After correcting that harness while +leaving runtime unchanged, **12 tests failed** on missing invalidation, ignored +unlock confirmation, or continued processing after disconnect. Corrected unit +RED JUnit SHA-256: +`bfc4fe4e2ade14ae9df144a92629c459fb83f65edfb09534c464dbfda75f5797`. + +Read-only sibling tracing found the same engine-bound lock pattern in reply SLA +scheduling and email import. Import owner [#1317](https://github.com/ContextualWisdomLab/naruon/pull/1317) +at `1b422f15e6e5f56be679f691c8ff925c9a420fb1` already proposes a separate +connection and NUL-safe owner key; scheduler [#1486](https://github.com/ContextualWisdomLab/naruon/pull/1486) +at `b32954dbf6066bc0d953887e8ca06820588f2c5f` changes workspace iteration but +retains the old lease lifetime. This is a repair dependency, not permission to +copy or overwrite those owners. Their actual contention and cancellation paths +still need independent RED/GREEN evidence; this worker result does not prove +either sibling fixed. No exactly-once provider-call or throughput claim follows. + +The complete 17-file suite on checkpoint +`47b2347dd5285081d24b13b38c6b85190dd0aa36` passed **303 tests, zero +failures/errors/skips, in 303.72 s**, after fresh and repeated migrations. Its +worker coverage was 256/258 statements and 56/58 branches, with zero exclusions; +22/22 function/class definitions had docstrings. The two uncovered paths were +pending-record reloads that return no row after rollback. Follow-up unit cases +cover both source kinds, skip that missing or no-longer-pending record, and +continue with the next cached identity. The final changed-head suite must rerun; +the 303-pass result is not automatically transferred to it. Checkpoint JUnit +SHA-256: `b7bb102a7fcd1ea1da0099ad348964e1d96fed0c122e71d762d01cc7e4694e7f`. + +### Earlier combined local execution receipt + +On 2026-09-05 the merged working tree passed **276 tests, 0 failures, 0 errors, +0 skips**, in **98.81 seconds**, with `-W error`. This combines the six files in +the command above with the inherited owner suite: + +```text +tests/test_email_read_state_migration_postgres.py +tests/test_alembic_migrations.py +tests/test_bootstrap_db.py +tests/test_data_api.py +tests/test_legacy_document_scope_postgres.py +tests/test_workspace_document_migration.py +tests/test_container_dependency_pin_contract.py +tests/test_search.py +tests/test_search_postgres.py +tests/test_search_answer.py +tests/test_hybrid_retrieval_fusion.py +``` + +Fresh and repeated migrations reached `0020_search_trigram_storage` on isolated +PostgreSQL 16.15 / pgvector. The service retained read-only root, explicit tmpfs, +256 MiB shared memory, loopback-only port publication, and no-new-privileges. +Both container and test network were removed after the terminal-success run. +The two real-PDF cases took 34.864 s (attachment) and 15.693 s (document), +including their migration/transaction work; these are not endpoint latency +measurements. The corpus was not made smaller and the indexes remained enabled. +The JUnit artifact SHA-256 is +`19bc54cec6bdc308b816fe2814b1999a64e9a0d2a7e47b8707d1b74210800c2e`. +Source-file Ruff and staged/unstaged whitespace checks also passed. This receipt +is local integration evidence, not hosted Checks, approval, merge, or deployment. + +### Preserved ADR lineage and remaining work + +ADR-0021 is inherited from the PDF-upload owner. The attachment branch's entire +earlier ADR-0005 proposal is preserved in +[the historical snapshot](pdf_dom_proposal_history.md); moving that snapshot out +of the numbered ADR directory removes a duplicate identity without losing its +deferral rationale. The attachment proposal formerly numbered 0006 becomes +ADR-0023, **Proposed**, subject to a complete current open-PR identity check before +push. Neither rename makes a decision Accepted or a PR merged. + +Keep #1469 Draft until its owner stack, current-head reviews/Checks, released +provider pin, and real capacity evidence are complete. Still required: actual +64 MiB PDF/provider recognition, realistic concurrent memory/storage/index and +latency measurements, tenant quotas, document-specific actionable error detail, +and a governed retry after provider upgrade. Full byte preservation alone does +not satisfy p95 ≤ 20 ms. The existing Python service is not a reason to choose +Python for a new hot path; profile and implement any required runtime change in +the canonical owner with contract-preserving Rust priority. + +## Research traceability + +The bounded transport and fail-closed error contract are aligned with HTTP +representation semantics (Fielding et al., 2022) and secure development +verification practices (Souppaya et al., 2022). See +[`ADR-0023`](../adr/0023-bounded-attachment-parse-source-contract.md). + +Josefsson, S. (2006). *The Base16, Base32, and Base64 data encodings* (RFC 4648). +Internet Engineering Task Force. https://www.rfc-editor.org/rfc/rfc4648.html + +National Aeronautics and Space Administration. (2019). *Earth at night*. +https://www.nasa.gov/ebooks/earth-at-night/ + +Encode OSS. (n.d.). *QuickStart: Streaming responses*. HTTPX. +https://www.python-httpx.org/quickstart/#streaming-responses + +SQLAlchemy Authors. (n.d.-a). *Asynchronous I/O (asyncio): Preventing implicit IO +when using AsyncSession*. SQLAlchemy 2.0 documentation. +https://docs.sqlalchemy.org/en/20/orm/extensions/asyncio.html#preventing-implicit-io-when-using-asyncsession + +SQLAlchemy Authors. (n.d.-b). *Session basics: Rolling back*. SQLAlchemy 2.0 documentation. +https://docs.sqlalchemy.org/en/20/orm/session_basics.html#rolling-back + +SQLAlchemy Authors. (n.d.-c). *Session basics: Committing*. SQLAlchemy 2.0 documentation. +https://docs.sqlalchemy.org/en/20/orm/session_basics.html#committing + +PostgreSQL Global Development Group. (n.d.). *Explicit locking: Advisory locks*. +PostgreSQL 16 documentation. +https://www.postgresql.org/docs/16/explicit-locking.html#ADVISORY-LOCKS + +The lease repair again encountered Context7's quota limit and used the official +documents above plus the pinned SQLAlchemy 2.0.51 runtime and real PostgreSQL +regression. The installed `adr-author` package lacked its required +`adr-identity.instructions.md`; the existing MADR-shaped Proposed ADR was amended +without allocating an ID, generating tracking state, or claiming acceptance. + +Context7 quota was exhausted and DeepWiki had no repository wiki during the +initial repair. Official HTTPX/RFC/NASA sources and exact Git refs were used +instead. The later rollback review also checked SQLAlchemy's official 2.0 +documentation; the project pins SQLAlchemy 2.0.51. Documentation supports the +causal explanation but does not substitute for the real database regression. +Context7 became available during the continuation and returned the same +SQLAlchemy rollback/explicit-loading contract; no private source was submitted. diff --git a/docs/doctoring/pdf_dom_proposal_history.md b/docs/doctoring/pdf_dom_proposal_history.md new file mode 100644 index 000000000..1eb554614 --- /dev/null +++ b/docs/doctoring/pdf_dom_proposal_history.md @@ -0,0 +1,65 @@ +# Historical ADR-0005 proposal from the attachment branch + +This is the complete historical proposal from #1469 at +`ed4bebeddf05ce1da0c76aca77448deef6254fbb`, not current implementation guidance. +The same proposal's canonical identity is now +[ADR-0021](../adr/0021-bounded-pdf-dom-upload-contract.md), inherited from #1427. +Moving this snapshot out of the numbered ADR directory removes the duplicate +identity without discarding its deferral rationale or claiming acceptance. +The original content follows unchanged. + +## ADR-0005: Bounded PDF DOM upload contract + +**Status:** Proposed +**Date:** 2026-08-20 +**Decision owner:** Naruon maintainers +**Scope:** Signed `POST /api/data/documents/pdf-dom-recognition` uploads +**Figma File ID:** N/A — backend upload contract; no visual surface. + +## Context + +Naruon email imports and deferred attachment admission use a 64 MiB bounded +transport budget. The direct Data workspace PDF-DOM endpoint and its NewsDOM +sidecar contract remain independently bounded at 20 MiB on the current +protected branch; this record preserves the proposed alignment without +claiming that the separate transport change has shipped. + +## Decision + +Retain the current 20 MiB direct PDF-DOM upload and decoder boundary until the +separate transport change is reviewed and integrated. Keep the signed-session +boundary, PDF signature validation, one-byte-over-limit read, base64 +persistence contract, and `413` response unchanged. A future alignment to the +64 MiB import budget requires sidecar confirmation, capacity evidence, and a +new current-head review; this ADR does not authorize that change. + +## Consequences + +- Email and manual PDF ingestion currently have explicit, separately governed + bounded contracts (64 MiB import/deferred admission; 20 MiB direct DOM). +- Workspace quotas, background-worker limits, and database-capacity monitoring + remain required because temporary content can be larger. +- No unbounded upload is introduced; malformed or non-PDF payloads continue to + fail closed before recognition. + +## Alternatives rejected + +### Align the manual endpoint immediately + +Deferred until the sidecar and storage capacity contract are independently +verified; changing only the Naruon endpoint would create a customer-visible +failure later in recognition. + +### Remove the upload limit + +Rejected because request and database resource use must remain bounded at the +authenticated trust boundary. + +## References (APA 7th) + +Internet Engineering Task Force. (2022). *HTTP semantics (RFC 9110).* RFC +Editor. https://www.rfc-editor.org/rfc/rfc9110 + +National Institute of Standards and Technology. (2025). *Secure software +development framework (SSDF) version 1.2* (NIST Special Publication 800-218 +Rev. 1, Initial Public Draft). https://doi.org/10.6028/NIST.SP.800-218r1.ipd