diff --git a/docs/production-runbook.md b/docs/production-runbook.md index 0138d0f..61f129d 100644 --- a/docs/production-runbook.md +++ b/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. diff --git a/src/transcription/services/sources.py b/src/transcription/services/sources.py index 93e895c..9f70264 100644 --- a/src/transcription/services/sources.py +++ b/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( - job_source_id=job_source.id, - job_id=job_id, - source_id=source_id, - attempt_number=(attempt_number or 0) + 1, - status=outcome, - provider=provider or job.provider or self.settings.provider.value, - model=model or job.model, - request_manifest=manifest_payload, - request_manifest_sha256=request_manifest.digest() if request_manifest is not None else None, - request_manifest_schema_version=( - request_manifest.schema_version if request_manifest is not None else None - ), - response_received=transport.response_received, - transport_status_code=transport.status_code, - transport_body=transport.body, - transport_content_type=transport.content_type, - transport_content_encoding=transport.content_encoding, - transport_safe_headers=transport.safe_headers or None, - router_request_id=transport.request_id, - router_generation_id=transport.generation_id, - sdk_response_snapshot=raw_response_payload, - normalized_metadata=attempt_metadata, - software_context=software_payload, - raw_transcription=text, - error_category=error_category, - 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)), - ) - _session.add(attempt) - await _session.flush() + attempt: ExecutionAttempt | None = None + for attempt_retry in range(1, MAX_EXECUTION_ATTEMPT_NUMBER_RETRIES + 1): + latest_attempt_number = ( + await _session.exec( + select(func.max(ExecutionAttempt.attempt_number)) + .where(ExecutionAttempt.job_id == job_id) + .where(ExecutionAttempt.source_id == source_id) + ) + ).one() + candidate = ExecutionAttempt( + job_source_id=job_source.id, + job_id=job_id, + source_id=source_id, + attempt_number=(latest_attempt_number or 0) + 1, + status=outcome, + provider=provider or job.provider or self.settings.provider.value, + model=model or job.model, + request_manifest=manifest_payload, + request_manifest_sha256=request_manifest.digest() if request_manifest is not None else None, + request_manifest_schema_version=( + request_manifest.schema_version if request_manifest is not None else None + ), + response_received=transport.response_received, + transport_status_code=transport.status_code, + transport_body=transport.body, + transport_content_type=transport.content_type, + transport_content_encoding=transport.content_encoding, + transport_safe_headers=transport.safe_headers or None, + router_request_id=transport.request_id, + router_generation_id=transport.generation_id, + sdk_response_snapshot=raw_response_payload, + normalized_metadata=attempt_metadata, + software_context=software_payload, + raw_transcription=text, + error_category=error_category, + 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, *, diff --git a/tests/services/test_v2_crud.py b/tests/services/test_v2_crud.py index 54fac13..2700f53 100644 --- a/tests/services/test_v2_crud.py +++ b/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()