Fix document updates stalled by retained ingestion job results

core/database/postgres_database.py

@@ -1,12 +1,14 @@
 import json
 import logging
+from contextlib import asynccontextmanager
 from datetime import UTC, datetime
 from typing import Any, Dict, List, Optional
 
 from sqlalchemy import desc, select, text
 from sqlalchemy.exc import ProgrammingError
 from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
 from sqlalchemy.orm import sessionmaker
+from sqlalchemy.pool import NullPool
 
 from core.config import get_settings
 from core.utils.folder_utils import normalize_folder_path
@@ -172,8 +174,30 @@ def __init__(
             connect_args={"server_settings": {"statement_timeout": "30000"}},  # 30 second timeout
         )
         self.async_session = sessionmaker(self.engine, class_=AsyncSession, expire_on_commit=False)
+        # Ingestion holds a lock across parsing and embedding. Keep those connections
+        # out of the query pool so busy workers cannot exhaust it and deadlock writes.
+        self._ingestion_lock_engine = create_async_engine(uri, poolclass=NullPool)
         self._initialized = False
 
+    @asynccontextmanager
+    async def document_ingestion_lock(self, document_id: str, *, wait: bool = False):
+        """Serialize content replacement and all worker writes for one document.
+
+        Transaction-scoped advisory locks are released on cancellation/disconnect,
+        with no expiring lease that could admit another writer during a long ingest.
+        Callers must read the document/revision *after* acquiring the lock.
+        """
+        async with self._ingestion_lock_engine.begin() as connection:
+            key = {"key": f"document-ingestion:{document_id}"}
+            if wait:
+                await connection.execute(text("SELECT pg_advisory_xact_lock(hashtextextended(:key, 0))"), key)
+                acquired = True
+            else:
+                acquired = await connection.scalar(
+                    text("SELECT pg_try_advisory_xact_lock(hashtextextended(:key, 0))"), key
+                )
+            yield acquired
+
     async def initialize(self):
         """Initialize database tables and indexes."""
         if self._initialized:
@@ -296,7 +320,9 @@ async def store_document(
             logger.error(f"Error storing document metadata: {str(e)}")
             return False
 
-    async def get_document(self, document_id: str, auth: AuthContext) -> Optional[Document]:
+    async def get_document(
+        self, document_id: str, auth: AuthContext, *, raise_on_error: bool = False
+    ) -> Optional[Document]:
         """Retrieve document metadata by ID if user has access."""
         try:
             async with self.async_session() as session:
@@ -320,6 +346,8 @@ async def get_document(self, document_id: str, auth: AuthContext) -> Optional[Do
 
         except Exception as e:
             logger.error(f"Error retrieving document metadata: {str(e)}")
+            if raise_on_error:
+                raise
             return None
 
     async def get_document_by_filename(

core/routes/ingest.py

@@ -3,6 +3,7 @@
 from typing import Any, Dict, List, Optional, Set
 
 import arq
+from arq.jobs import Job, JobStatus
 from fastapi import APIRouter, Depends, File, Form, HTTPException, Request, UploadFile
 
 from core.auth_utils import verify_token
@@ -287,11 +288,39 @@ async def requeue_ingest_jobs(
     results: List[RequeueIngestionResult] = []
 
     async def _process_document(doc: Document, override_flag: Optional[bool]) -> None:
+        if doc.external_id in processed_ids:
+            return
+        async with ingestion_service.db.document_ingestion_lock(doc.external_id) as acquired:
+            if not acquired:
+                results.append(
+                    RequeueIngestionResult(
+                        external_id=doc.external_id,
+                        status="already_queued",
+                        message="Document ingestion is in progress",
+                    )
+                )
+                processed_ids.add(doc.external_id)
+                return
+            current = await ingestion_service.db.get_document(doc.external_id, auth, raise_on_error=True)
+            if current is None:
+                results.append(
+                    RequeueIngestionResult(
+                        external_id=doc.external_id,
+                        status="error",
+                        message="Document no longer exists",
+                    )
+                )
+                processed_ids.add(doc.external_id)
+                return
+            await _process_locked_document(current, override_flag)
+
+    async def _process_locked_document(doc: Document, override_flag: Optional[bool]) -> None:
         ext_id = doc.external_id
         if ext_id in processed_ids:
             return
 
         processed_ids.add(ext_id)
+        revision_persisted = False
 
         try:
             auth_for_doc = AuthContext(
@@ -332,11 +361,28 @@ async def _process_document(doc: Document, override_flag: Optional[bool]) -> Non
             if isinstance(system_metadata, str):
                 system_metadata = json.loads(system_metadata)
             sanitized_system_metadata = IngestionService._reset_processing_metadata(system_metadata)
-            await ingestion_service.db.update_document(
+            revision = int(system_metadata.get("ingestion_revision", 0))
+            current_job = Job(f"ingest:{ext_id}:{revision}", redis, _queue_name=redis.default_queue_name)
+            if await current_job.status() in {JobStatus.queued, JobStatus.deferred, JobStatus.in_progress}:
+                results.append(
+                    RequeueIngestionResult(
+                        external_id=ext_id,
+                        status="already_queued",
+                        message="An ingestion job is already pending",
+                    )
+                )
+                return
+            # A retained result may represent a failed attempt. A manual requeue
+            # gets a new revision, fencing off any delayed attempt of the old job.
+            sanitized_system_metadata["ingestion_revision"] = revision + 1
+            success = await ingestion_service.db.update_document(
                 document_id=ext_id,
                 updates={"system_metadata": sanitized_system_metadata},
                 auth=auth_for_doc,
             )
+            if not success:
+                raise RuntimeError("Failed to persist requeue revision")
+            revision_persisted = True
             job_payload = IngestionService._build_ingestion_job_payload(
                 document_id=ext_id,
                 file_key=key,
@@ -349,17 +395,12 @@ async def _process_document(doc: Document, override_flag: Optional[bool]) -> Non
                 folder_path=doc.folder_path,
                 folder_leaf=doc.folder_name,
                 end_user_id=doc.end_user_id,
+                ingestion_revision=revision + 1,
             )
             job = await redis.enqueue_job("process_ingestion_job", **job_payload)
 
             if job is None:
-                results.append(
-                    RequeueIngestionResult(
-                        external_id=ext_id,
-                        status="already_queued",
-                        message="An ingestion job is already pending for this document",
-                    )
-                )
+                raise RuntimeError("Requeue job ID is already present in Redis; no new job was queued")
             else:
                 results.append(
                     RequeueIngestionResult(
@@ -372,6 +413,8 @@ async def _process_document(doc: Document, override_flag: Optional[bool]) -> Non
             raise
         except Exception as exc:  # noqa: BLE001
             logger.error("Failed to requeue ingestion for document %s: %s", ext_id, exc, exc_info=True)
+            if revision_persisted:
+                await ingestion_service._mark_document_failed(doc, auth, f"Failed to requeue ingestion: {exc}")
             results.append(
                 RequeueIngestionResult(
                     external_id=ext_id,

core/services/ingestion_service.py

@@ -428,9 +428,11 @@ def _build_ingestion_job_payload(
         folder_path: Optional[str] = None,
         folder_leaf: Optional[str] = None,
         end_user_id: Optional[str] = None,
+        ingestion_revision: int = 0,
     ) -> Dict[str, Any]:
         return {
-            "_job_id": f"ingest:{document_id}",
+            "_job_id": f"ingest:{document_id}:{ingestion_revision}",
+            "ingestion_revision": ingestion_revision,
             "_expires": timedelta(days=7),
             "document_id": document_id,
             "file_key": file_key,
@@ -562,6 +564,38 @@ async def ingest_file_content(
         end_user_id: Optional[str] = None,
         use_colpali: Optional[bool] = False,
         external_id: Optional[str] = None,
+    ) -> Document:
+        document_id = external_id or str(uuid.uuid4())
+        async with self.db.document_ingestion_lock(document_id) as acquired:
+            if not acquired:
+                raise HTTPException(status_code=409, detail="Document ingestion is already in progress; retry later")
+            return await self._ingest_file_content_locked(
+                file_content_bytes,
+                filename,
+                content_type,
+                metadata,
+                auth,
+                redis,
+                metadata_types,
+                folder_name,
+                end_user_id,
+                use_colpali,
+                document_id,
+            )
+
+    async def _ingest_file_content_locked(
+        self,
+        file_content_bytes: bytes,
+        filename: str,
+        content_type: Optional[str],
+        metadata: Optional[Dict[str, Any]],
+        auth: AuthContext,
+        redis: arq.ArqRedis,
+        metadata_types: Optional[Dict[str, str]],
+        folder_name: Optional[Union[str, List[str]]],
+        end_user_id: Optional[str],
+        use_colpali: Optional[bool],
+        external_id: str,
     ) -> Document:
         """
         Ingests file content from bytes. Saves to storage, creates document record,
@@ -594,6 +628,7 @@ async def ingest_file_content(
             folder_path=folder_path,
         )
         doc.system_metadata = self._reset_processing_metadata(doc.system_metadata)
+        doc.system_metadata["ingestion_revision"] = 0
 
         await self._verify_ingest_and_storage_limits(auth, len(file_content_bytes), doc.external_id)
 
@@ -685,7 +720,7 @@ async def ingest_file_content(
             )
             job = await redis.enqueue_job("process_ingestion_job", **job_payload)
             if job is None:
-                logger.info("Connector file ingestion job already queued (doc_id=%s)", doc.external_id)
+                raise RuntimeError("Ingestion job ID is already present in Redis; no new job was queued")
             else:
                 logger.info(
                     "Connector file ingestion job queued with ID: %s for document: %s", job.job_id, doc.external_id
@@ -712,6 +747,33 @@ async def queue_document_update(
         metadata: Optional[Dict[str, Any]] = None,
         metadata_types: Optional[Dict[str, str]] = None,
         use_colpali: Optional[bool] = None,
+    ) -> Optional[Document]:
+        async with self.db.document_ingestion_lock(document_id) as acquired:
+            if not acquired:
+                raise HTTPException(status_code=409, detail="Document ingestion is already in progress; retry later")
+            return await self._queue_document_update_locked(
+                document_id,
+                auth,
+                redis,
+                content,
+                file,
+                filename,
+                metadata,
+                metadata_types,
+                use_colpali,
+            )
+
+    async def _queue_document_update_locked(
+        self,
+        document_id: str,
+        auth: AuthContext,
+        redis: arq.ArqRedis,
+        content: Optional[str],
+        file: Optional[UploadFile],
+        filename: Optional[str],
+        metadata: Optional[Dict[str, Any]],
+        metadata_types: Optional[Dict[str, str]],
+        use_colpali: Optional[bool],
     ) -> Optional[Document]:
         """
         Update a document by replacing its content and re-queueing ingestion.
@@ -781,6 +843,7 @@ async def queue_document_update(
             raise HTTPException(status_code=500, detail=f"Failed to upload updated file to storage: {str(e)}")
 
         doc.system_metadata = self._reset_processing_metadata(doc.system_metadata)
+        doc.system_metadata["ingestion_revision"] = int(doc.system_metadata.get("ingestion_revision", 0)) + 1
 
         updates = {
             "metadata": doc.metadata,
@@ -832,10 +895,11 @@ async def queue_document_update(
                 folder_path=doc.folder_path,
                 folder_leaf=doc.folder_name,
                 end_user_id=doc.end_user_id,
+                ingestion_revision=doc.system_metadata["ingestion_revision"],
             )
             job = await redis.enqueue_job("process_ingestion_job", **job_payload)
             if job is None:
-                logger.info("Update ingestion job already queued (doc_id=%s)", doc.external_id)
+                raise RuntimeError("Update job ID is already present in Redis; no new job was queued")
             else:
                 logger.info("Update ingestion job queued (job_id=%s, doc=%s)", job.job_id, doc.external_id)
         except Exception as e: