From 86cdb4035ca2befccd17734ea0c3e06b128a607c Mon Sep 17 00:00:00 2001 From: Jim Lancaster <40281233+zoltan57@users.noreply.github.com> Date: Sun, 23 Aug 2026 18:30:54 -0500 Subject: [PATCH] fix: gate retries by error category with backoff Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- .env.example | 1 + src/transcription/config.py | 1 + src/transcription/services/evidence.py | 20 +++++++ src/transcription/services/workflows.py | 27 ++++++++- tests/services/test_workflows_reliability.py | 63 ++++++++++++++++++++ tests/test_config.py | 1 + tests/test_meta_contract_guards.py | 1 + 7 files changed, 112 insertions(+), 2 deletions(-) diff --git a/.env.example b/.env.example index f521e0e..e7edf95 100644 --- a/.env.example +++ b/.env.example @@ -58,6 +58,7 @@ DATABASE_BACKUP_DIR=./data/backups WORKER_MAX_RETRIES=0 WORKER_PROVIDER_TIMEOUT_SECONDS=30.0 WORKER_STALE_JOB_SECONDS=30.0 +WORKER_RETRY_BACKOFF_SECONDS=1.0 WORKER_MIN_TRANSCRIPTION_CHARS=0 WORKER_MIN_TRANSCRIPTION_LINES=0 WORKER_FAIL_ON_FINISH_REASON_LENGTH=false diff --git a/src/transcription/config.py b/src/transcription/config.py index 1d6f531..99c0bd2 100644 --- a/src/transcription/config.py +++ b/src/transcription/config.py @@ -115,6 +115,7 @@ class Settings(BaseSettings): # well past twenty seconds, so an upper cap here would silently fail real work. worker_provider_timeout_seconds: float = Field(default=30.0, gt=0.0) worker_stale_job_seconds: float = Field(default=30.0, gt=0.0) + worker_retry_backoff_seconds: float = Field(default=1.0, ge=0.0) worker_min_transcription_chars: int = Field(default=0, ge=0) worker_min_transcription_lines: int = Field(default=0, ge=0) worker_fail_on_finish_reason_length: bool = False diff --git a/src/transcription/services/evidence.py b/src/transcription/services/evidence.py index 687cf59..49b7ac2 100644 --- a/src/transcription/services/evidence.py +++ b/src/transcription/services/evidence.py @@ -43,6 +43,26 @@ class LatestExecutionAttempt: class EvidenceService(ServiceBase): """Read, project, and export execution attempt evidence.""" + async def read_latest_job_error_category( + self, + *, + job_id: UUID, + session: AsyncSession | None = None, + ) -> str | None: + """Read the latest persisted execution-attempt error category for a job.""" + async with self._session_scope(session) as _session: + query = ( + select(ExecutionAttempt.error_category) + .where(ExecutionAttempt.job_id == job_id) + .where(col(ExecutionAttempt.error_category).is_not(None)) + .order_by( + col(ExecutionAttempt.created_at).desc(), + col(ExecutionAttempt.id).desc(), + ) + .limit(1) + ) + return (await _session.exec(query)).first() + async def read_latest_execution_attempt( self, *, diff --git a/src/transcription/services/workflows.py b/src/transcription/services/workflows.py index 58cc2fe..afd3345 100644 --- a/src/transcription/services/workflows.py +++ b/src/transcription/services/workflows.py @@ -40,6 +40,12 @@ from .sources import transcribe_document_image logger = logging.getLogger(__name__) +_RETRIABLE_FAILED_JOB_ERROR_CATEGORIES = { + ErrorCategory.EXTERNAL_PROVIDER.value, + ErrorCategory.EXTERNAL_TIMEOUT.value, + ErrorCategory.INFRA_TRANSIENT.value, +} + async def create_document_with_people( *, @@ -182,16 +188,33 @@ async def advance_job( # Recover mid-flight jobs by continuing the queued processing path. return await process_queued_job(job=job, services=services, settings=settings, session=session) case JobStatus.FAILED: - if job.retry_count < settings.worker_max_retries: + latest_error_category = await services.evidence.read_latest_job_error_category( + job_id=job.id, + session=session, + ) + can_retry = ( + job.retry_count < settings.worker_max_retries + and latest_error_category in _RETRIABLE_FAILED_JOB_ERROR_CATEGORIES + ) + if can_retry: + if settings.worker_retry_backoff_seconds > 0: + await asyncio.sleep(settings.worker_retry_backoff_seconds) return await services.jobs.update_job_state( job_id=job.id, status=JobStatus.QUEUED, retry_count_increment=1, session=session, ) + + if job.retry_count < settings.worker_max_retries: + logger.warning( + "Job %s failed with non-retriable category %s; skipping retry.", + job.id, + latest_error_category or "unknown", + ) else: logger.error("Job %s has failed and reached max retries.", job.id) - return + return case _: return diff --git a/tests/services/test_workflows_reliability.py b/tests/services/test_workflows_reliability.py index d546e94..9ed4e21 100644 --- a/tests/services/test_workflows_reliability.py +++ b/tests/services/test_workflows_reliability.py @@ -2,6 +2,8 @@ import asyncio import time +from datetime import UTC +from datetime import datetime from pathlib import Path from uuid import uuid4 @@ -20,6 +22,7 @@ from transcription.db.models import Source from transcription.providers.base import TranscriptionResult from transcription.services import ServiceBundle from transcription.services import workflows as workflows_module +from transcription.services.workflows import advance_job from transcription.services.workflows import process_queued_job @@ -366,3 +369,63 @@ class TestWorkflowReliability: result = await task assert result is not None assert result.status == JobStatus.TRANSCRIBED + + @pytest.mark.asyncio + async def test_failed_job_with_validation_category_is_not_requeued(self, default_session_factory): + services = ServiceBundle.from_session_factory(default_session_factory) + async with services.jobs._session_scope() as session: + document = Document(id=uuid4(), name="validation-failure-doc") + session.add(document) + await session.flush() + + source = Source( + document_id=document.id, + page_number=1, + upload_name="validation.jpg", + filename="validation.jpg", + file_path=str(Path("tests/fixtures/images/real/Book Two - page 02.jpg")), + file_hash="9" * 64, + file_size_bytes=1, + ) + session.add(source) + await session.flush() + + job = Job(document_id=document.id, status=JobStatus.FAILED, retry_count=0) + session.add(job) + await session.flush() + + job_source = JobSource(job_id=job.id, source_id=source.id, status=JobSourceStatus.FAILED) + session.add(job_source) + await session.flush() + + now = datetime.now(UTC) + session.add( + ExecutionAttempt( + job_source_id=job_source.id, + job_id=job.id, + source_id=source.id, + attempt_number=1, + status=JobSourceStatus.FAILED, + provider="fixture", + started_at=now, + finished_at=now, + duration_ms=0, + error_category="validation_error", + error_detail="invalid payload", + ) + ) + await session.commit() + failed_job = await services.jobs.read_job(job_id=job.id, session=session) + + result = await advance_job( + failed_job, + services=services, + settings=Settings(openrouter_api_key="test-key", worker_max_retries=1), + ) + + assert result is None + async with services.jobs._session_scope() as session: + persisted = await session.get(Job, failed_job.id) + assert persisted is not None + assert persisted.status == JobStatus.FAILED + assert persisted.retry_count == 0 diff --git a/tests/test_config.py b/tests/test_config.py index 9bb9627..fd71cb8 100644 --- a/tests/test_config.py +++ b/tests/test_config.py @@ -168,6 +168,7 @@ class TestWorkerReliabilitySettings: settings = _make_settings() assert settings.worker_max_retries == 0 assert settings.worker_stale_job_seconds == 30.0 + assert settings.worker_retry_backoff_seconds == 1.0 def test_provider_timeout_is_not_capped_at_twenty_seconds(): diff --git a/tests/test_meta_contract_guards.py b/tests/test_meta_contract_guards.py index 78f676e..bda4b9e 100644 --- a/tests/test_meta_contract_guards.py +++ b/tests/test_meta_contract_guards.py @@ -168,6 +168,7 @@ def test_env_example_default_values_match_settings_defaults(): "WORKER_MAX_RETRIES": str(defaults.worker_max_retries), "WORKER_PROVIDER_TIMEOUT_SECONDS": str(defaults.worker_provider_timeout_seconds), "WORKER_STALE_JOB_SECONDS": str(defaults.worker_stale_job_seconds), + "WORKER_RETRY_BACKOFF_SECONDS": str(defaults.worker_retry_backoff_seconds), "WORKER_MIN_TRANSCRIPTION_CHARS": str(defaults.worker_min_transcription_chars), "WORKER_MIN_TRANSCRIPTION_LINES": str(defaults.worker_min_transcription_lines), "WORKER_FAIL_ON_FINISH_REASON_LENGTH": str(defaults.worker_fail_on_finish_reason_length).lower(),