fix(attachments): align deferred parse budget with import transport

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

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.

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, "

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
 
 

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"

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."""

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 == ""