fix: retry execution attempt number conflicts

Co-authored-by: Copilot App <[email protected]>

docs/production-runbook.md

@@ -141,3 +141,13 @@ and exit cleanly after the call returns.
 Set the container or service termination grace period **above this total**
 budget. If termination grace is shorter, the process may be killed before
 terminal status and evidence writes are finalized.
+
+## 9. Horizontal scaling precondition
+
+Multiple worker replicas can race on execution-attempt numbering for the same
+`(job_id, source_id)` pair. The runtime now retries boundedly on unique-key
+conflicts (`uq_execution_attempt_number`) and surfaces a conflict-domain error
+if retries are exhausted.
+
+Do not deploy additional worker replicas unless this conflict-retry path and its
+tests are present and green in the target build.

src/transcription/services/sources.py

@@ -62,6 +62,7 @@ logger = logging.getLogger(__name__)
 
 DEFAULT_PROMPT_FILE = "transcribe_document.md"
 JSON_OBJECT_ADAPTER = TypeAdapter(dict[str, JsonValue])
+MAX_EXECUTION_ATTEMPT_NUMBER_RETRIES = 3
 
 
 class PromptExecution(BaseModel):
@@ -533,54 +534,72 @@ class SourceService(ServiceBase):
 
             finish_time = finished_at or datetime.now(UTC)
             start_time = started_at or finish_time
-            attempt_number = (
-                await _session.exec(
-                    select(func.max(ExecutionAttempt.attempt_number))
-                    .where(ExecutionAttempt.job_id == job_id)
-                    .where(ExecutionAttempt.source_id == source_id)
-                )
-            ).one()
             transport = transport_evidence or TransportEvidence(response_received=False)
             manifest_payload = request_manifest.model_dump(mode="json") if request_manifest is not None else None
             software_payload = (
                 request_manifest.software.model_dump(mode="json") if request_manifest is not None else None
             )
-            attempt = ExecutionAttempt(
+            attempt: ExecutionAttempt | None = None
-                job_source_id=job_source.id,
+            for attempt_retry in range(1, MAX_EXECUTION_ATTEMPT_NUMBER_RETRIES + 1):
-                job_id=job_id,
+                latest_attempt_number = (
-                source_id=source_id,
+                    await _session.exec(
-                attempt_number=(attempt_number or 0) + 1,
+                        select(func.max(ExecutionAttempt.attempt_number))
-                status=outcome,
+                        .where(ExecutionAttempt.job_id == job_id)
-                provider=provider or job.provider or self.settings.provider.value,
+                        .where(ExecutionAttempt.source_id == source_id)
-                model=model or job.model,
+                    )
-                request_manifest=manifest_payload,
+                ).one()
-                request_manifest_sha256=request_manifest.digest() if request_manifest is not None else None,
+                candidate = ExecutionAttempt(
-                request_manifest_schema_version=(
+                    job_source_id=job_source.id,
-                    request_manifest.schema_version if request_manifest is not None else None
+                    job_id=job_id,
-                ),
+                    source_id=source_id,
-                response_received=transport.response_received,
+                    attempt_number=(latest_attempt_number or 0) + 1,
-                transport_status_code=transport.status_code,
+                    status=outcome,
-                transport_body=transport.body,
+                    provider=provider or job.provider or self.settings.provider.value,
-                transport_content_type=transport.content_type,
+                    model=model or job.model,
-                transport_content_encoding=transport.content_encoding,
+                    request_manifest=manifest_payload,
-                transport_safe_headers=transport.safe_headers or None,
+                    request_manifest_sha256=request_manifest.digest() if request_manifest is not None else None,
-                router_request_id=transport.request_id,
+                    request_manifest_schema_version=(
-                router_generation_id=transport.generation_id,
+                        request_manifest.schema_version if request_manifest is not None else None
-                sdk_response_snapshot=raw_response_payload,
+                    ),
-                normalized_metadata=attempt_metadata,
+                    response_received=transport.response_received,
-                software_context=software_payload,
+                    transport_status_code=transport.status_code,
-                raw_transcription=text,
+                    transport_body=transport.body,
-                error_category=error_category,
+                    transport_content_type=transport.content_type,
-                error_detail=error_detail,
+                    transport_content_encoding=transport.content_encoding,
-                failure_phase=failure_phase,
+                    transport_safe_headers=transport.safe_headers or None,
-                started_at=start_time,
+                    router_request_id=transport.request_id,
-                finished_at=finish_time,
+                    router_generation_id=transport.generation_id,
-                duration_ms=duration_ms
+                    sdk_response_snapshot=raw_response_payload,
-                if duration_ms is not None
+                    normalized_metadata=attempt_metadata,
-                else max(0, int((finish_time - start_time).total_seconds() * 1000)),
+                    software_context=software_payload,
-            )
+                    raw_transcription=text,
-            _session.add(attempt)
+                    error_category=error_category,
-            await _session.flush()
+                    error_detail=error_detail,
+                    failure_phase=failure_phase,
+                    started_at=start_time,
+                    finished_at=finish_time,
+                    duration_ms=duration_ms
+                    if duration_ms is not None
+                    else max(0, int((finish_time - start_time).total_seconds() * 1000)),
+                )
+                try:
+                    async with _session.begin_nested():
+                        _session.add(candidate)
+                        await _session.flush()
+                    attempt = candidate
+                    break
+                except IntegrityError:
+                    logger.warning(
+                        "Execution attempt number conflict job_id=%s source_id=%s retry=%s/%s",
+                        job_id,
+                        source_id,
+                        attempt_retry,
+                        MAX_EXECUTION_ATTEMPT_NUMBER_RETRIES,
+                    )
+                    continue
+
+            if attempt is None:
+                raise self._execution_attempt_conflict(job_id=job_id, source_id=source_id)
 
             if text is not None and source.raw_transcription is None and source.preferred_execution_attempt_id is None:
                 source.raw_transcription = text
@@ -597,6 +616,17 @@ class SourceService(ServiceBase):
             suggestion="Use the existing job-source link instead of creating a duplicate.",
         )
 
+    @staticmethod
+    def _execution_attempt_conflict(*, job_id: UUID, source_id: UUID) -> TranscriptionError:
+        return TranscriptionError(
+            (
+                f"Failed to allocate an execution attempt number for Source {source_id} in Job {job_id} "
+                "after bounded retries"
+            ),
+            category=ErrorCategory.CONFLICT,
+            suggestion="Retry the transcription. If it repeats, investigate concurrent worker activity.",
+        )
+
     async def upsert_revision_for_source(
         self,
         *,

tests/services/test_v2_crud.py

@@ -2,9 +2,12 @@ from uuid import uuid4
 
 import pytest
 from pydantic import JsonValue
+from sqlalchemy.exc import IntegrityError
+from sqlmodel.ext.asyncio.session import AsyncSession
 
 from transcription.db.models import Document
 from transcription.db.models import DocumentPerson
+from transcription.db.models import ExecutionAttempt
 from transcription.db.models import Job
 from transcription.db.models import JobSource
 from transcription.db.models import JobSourceStatus
@@ -369,3 +372,104 @@ async def test_update_job_source_transcription_persists_provider_json_payloads(d
     assert attempt.attempt.raw_transcription == "provider transcript"
     assert attempt.attempt.normalized_metadata == metadata
     assert attempt.attempt.sdk_response_snapshot == raw_payload
+
+
+@pytest.mark.asyncio
+async def test_update_job_source_transcription_retries_on_execution_attempt_integrity_conflict(
+    default_session_factory,
+    monkeypatch,
+):
+    documents = DocumentService(session_factory=default_session_factory)
+    jobs = JobService(session_factory=default_session_factory)
+    transcriptions = SourceService(session_factory=default_session_factory)
+
+    document = await documents.create_document(Document(id=uuid4(), name="attempt-retry-doc"))
+    job = await jobs.create_job(Job(document_id=document.id))
+    source = await transcriptions.create_source(
+        Source(
+            document_id=document.id,
+            page_number=1,
+            upload_name="attempt-retry.jpg",
+            filename="attempt-retry.jpg",
+            file_path="uploads/attempt-retry.jpg",
+            file_hash="e" * 64,
+            file_size_bytes=1,
+        )
+    )
+    await transcriptions.create_job_source(
+        JobSource(job_id=job.id, source_id=source.id, status=JobSourceStatus.PENDING)
+    )
+
+    original_flush = AsyncSession.flush
+    execution_attempt_flushes = 0
+
+    async def _flaky_flush(self, *args, **kwargs):
+        nonlocal execution_attempt_flushes
+        if any(isinstance(item, ExecutionAttempt) for item in self.new):
+            execution_attempt_flushes += 1
+            if execution_attempt_flushes == 1:
+                raise IntegrityError("insert execution_attempt", {}, Exception("duplicate attempt number"))
+        return await original_flush(self, *args, **kwargs)
+
+    monkeypatch.setattr(AsyncSession, "flush", _flaky_flush)
+
+    await transcriptions.update_job_source_transcription(
+        job_id=job.id,
+        source_id=source.id,
+        text="retry succeeds",
+        provider="openrouter",
+        model="test-model",
+    )
+
+    assert execution_attempt_flushes == 2
+    rows = await transcriptions.list_job_sources(job_id=job.id)
+    assert len(rows) == 1
+    assert rows[0].status == JobSourceStatus.TRANSCRIBED
+
+
+@pytest.mark.asyncio
+async def test_update_job_source_transcription_raises_domain_error_after_attempt_retry_exhaustion(
+    default_session_factory,
+    monkeypatch,
+):
+    documents = DocumentService(session_factory=default_session_factory)
+    jobs = JobService(session_factory=default_session_factory)
+    transcriptions = SourceService(session_factory=default_session_factory)
+
+    document = await documents.create_document(Document(id=uuid4(), name="attempt-exhaustion-doc"))
+    job = await jobs.create_job(Job(document_id=document.id))
+    source = await transcriptions.create_source(
+        Source(
+            document_id=document.id,
+            page_number=1,
+            upload_name="attempt-exhaustion.jpg",
+            filename="attempt-exhaustion.jpg",
+            file_path="uploads/attempt-exhaustion.jpg",
+            file_hash="f" * 64,
+            file_size_bytes=1,
+        )
+    )
+    await transcriptions.create_job_source(
+        JobSource(job_id=job.id, source_id=source.id, status=JobSourceStatus.PENDING)
+    )
+
+    original_flush = AsyncSession.flush
+
+    async def _always_conflict_flush(self, *args, **kwargs):
+        if any(isinstance(item, ExecutionAttempt) for item in self.new):
+            raise IntegrityError("insert execution_attempt", {}, Exception("duplicate attempt number"))
+        return await original_flush(self, *args, **kwargs)
+
+    monkeypatch.setattr(AsyncSession, "flush", _always_conflict_flush)
+
+    with pytest.raises(TranscriptionError) as exc_info:
+        await transcriptions.update_job_source_transcription(
+            job_id=job.id,
+            source_id=source.id,
+            text="will not persist",
+            provider="openrouter",
+            model="test-model",
+        )
+
+    assert exc_info.value.category == ErrorCategory.CONFLICT
+    assert "attempt number" in exc_info.value.message.lower()