diff --git a/src/transcription/services/workflows.py b/src/transcription/services/workflows.py index 67f3c9c..d0d2ebf 100644 --- a/src/transcription/services/workflows.py +++ b/src/transcription/services/workflows.py @@ -416,10 +416,46 @@ async def process_next_queued_job( if session is not None: await session.commit() - await advance_job(job=job, services=services, settings=settings, session=session) + await _advance_job_with_containment(job=job, services=services, settings=settings, session=session) return True +async def _advance_job_with_containment( + *, + job: Job, + services: ServiceBundle, + settings: Settings | None, + session: AsyncSession | None, +) -> None: + """Advance a claimed job, guaranteeing it never stays stuck in PROCESSING. + + The job has already been committed as PROCESSING at this point, and + ``claim_next_queued_job`` only ever selects QUEUED rows. So an exception + escaping ``advance_job`` used to strand the job in PROCESSING permanently, + with a single swallowed log line and no recovery path (review log [8]). + + Any escaping exception is therefore classified and the job driven to the + terminal FAILED state, which is visible in the UI and resubmittable. The + terminal write runs in its own transaction so it is atomic even when the + caller's session was left dirty by the failure. + """ + try: + await advance_job(job=job, services=services, settings=settings, session=session) + except Exception as exc: + error = exc if isinstance(exc, AppError) else classify_unexpected_error(exc, operation="worker.advance_job") + logger.exception( + "Job processing failed outside page handling operation=worker.advance_job " + "job_id=%s error_id=%s category=%s retriable=%s", + job.id, + error.error_id, + error.category.value, + error.retriable, + ) + if session is not None: + await session.rollback() + await services.jobs.update_job_state(job_id=job.id, status=JobStatus.FAILED) + + def _resolve_job_sources(job: Job) -> list[Source]: """Resolve pending linked sources for a job in deterministic page order. diff --git a/src/transcription/worker.py b/src/transcription/worker.py index d79b060..74c1fa9 100644 --- a/src/transcription/worker.py +++ b/src/transcription/worker.py @@ -94,16 +94,30 @@ async def worker_consumer_lifespan( @contextmanager def handle_worker_exceptions(operation: str = "worker.loop"): - """Context manager to log and suppress exceptions in the worker loop.""" + """Log worker-loop exceptions, suppressing only the retriable ones. + + ``classify_unexpected_error`` marks unknown exceptions non-retriable, but that + verdict used to be logged and then discarded, so a programming error raised + before a job could be claimed spun the loop at the poll interval indefinitely. + Nothing capped it, because it never reached the per-job retry machinery + (review log [8]). + + A non-retriable fault is a defect rather than a transient condition, so it is + re-raised for the caller to stop on. Faults raised after a job is claimed are + contained at the job level instead, and never reach here. + """ try: yield except Exception as exc: error = exc if isinstance(exc, AppError) else classify_unexpected_error(exc, operation=operation) logger.exception( - "Worker loop exception error_id=%s category=%s", + "Worker loop exception error_id=%s category=%s retriable=%s", error.error_id, error.category.value, + error.retriable, ) + if not error.retriable: + raise error from exc async def run_worker_loop( @@ -121,6 +135,10 @@ async def run_worker_loop( The service bundle — and with it the provider's pooled HTTP client — is built once for the lifetime of the loop, so consecutive jobs reuse one connection instead of paying a fresh TLS handshake each time. + + The loop returns early on a non-retriable error raised before a job could be + claimed. Faults raised after a claim are contained by marking that job FAILED, + so a single poison job cannot stop transcription for every other job. """ services = ServiceBundle.from_session_factory(session_factory) try: @@ -150,6 +168,15 @@ async def run_worker_loop( if wake_event is None and not processed_any: await asyncio.sleep(poll_interval_seconds) + except AppError as error: + # Stop rather than spin. A non-retriable fault here means no job could be + # claimed, so there is no row to mark FAILED and no reason to expect the + # next poll to behave differently. + logger.critical( + "Worker loop stopped after a non-retriable error error_id=%s category=%s", + error.error_id, + error.category.value, + ) finally: await services.aclose() diff --git a/tests/services/test_workflows_reliability.py b/tests/services/test_workflows_reliability.py index 5533338..abac4cc 100644 --- a/tests/services/test_workflows_reliability.py +++ b/tests/services/test_workflows_reliability.py @@ -178,6 +178,63 @@ class TestWorkflowReliability: assert duration_ms >= int(budget_seconds * 1000 * 0.9) assert duration_ms < int((budget_seconds + setup_seconds) * 1000 * 0.9) + @pytest.mark.asyncio + async def test_error_after_claim_fails_the_job_instead_of_stranding_it( + self, + default_session_factory, + monkeypatch, + ): + """A non-retriable fault after the claim drives the job terminal, not stuck. + + Regression guard for review log [8]. The claim commits PROCESSING before any + provider work, and claim_next_queued_job only ever selects QUEUED, so an + exception escaping advance_job used to strand the job in PROCESSING forever + with one swallowed log line. Measured before the fix: raised once, job left + processing, retry_count 0, never re-claimed. + """ + services = ServiceBundle.from_session_factory(default_session_factory) + async with services.jobs._session_scope() as session: + document = Document(id=uuid4(), name="strand-doc") + session.add(document) + await session.flush() + job = Job(document_id=document.id, status=JobStatus.QUEUED) + session.add(job) + await session.flush() + source = Source( + document_id=document.id, + page_number=1, + upload_name="strand.jpg", + filename="strand.jpg", + file_path=str(Path("tests/fixtures/images/real/Book Two - page 02.jpg")), + file_hash="e" * 64, + file_size_bytes=1, + ) + session.add(source) + await session.flush() + session.add(JobSource(job_id=job.id, source_id=source.id, status=JobSourceStatus.PENDING)) + await session.commit() + job_id = job.id + + async def _succeeds(*args, **kwargs): + _ = (args, kwargs) + return TranscriptionResult(text="page text", provider="test", model="test-model") + + async def _boom(**kwargs): + _ = kwargs + raise AttributeError("deliberate programming error") + + monkeypatch.setattr("transcription.services.workflows.transcribe_document_image", _succeeds) + monkeypatch.setattr("transcription.services.workflows._finalize_batch_outcome", _boom) + + processed = await workflows_module.process_next_queued_job(services=services) + + assert processed is True + async with services.jobs._session_scope() as session: + final = await session.get(Job, job_id) + assert final is not None + # Terminal and resubmittable, rather than stranded in PROCESSING. + assert final.status == JobStatus.FAILED + @pytest.mark.asyncio async def test_completed_page_is_committed_before_next_provider_call_finishes( self, diff --git a/tests/test_worker.py b/tests/test_worker.py index b31a086..9d6f7ed 100644 --- a/tests/test_worker.py +++ b/tests/test_worker.py @@ -4,6 +4,8 @@ from typing import cast import pytest +from transcription.errors import AppError +from transcription.errors import ErrorCategory from transcription.services import ServiceBundle from transcription.services.sources import SourceService from transcription.worker import process_next_queued_job @@ -11,7 +13,38 @@ from transcription.worker import run_worker_loop @pytest.mark.asyncio -async def test_run_worker_loop_survives_process_next_exception(monkeypatch, caplog): +async def test_run_worker_loop_stops_on_non_retriable_exception(monkeypatch, caplog): + """A programming error before a job is claimed stops the loop instead of spinning. + + Regression guard for review log [8]. This previously spun at the poll interval + forever: the fault was classified non-retriable, logged, and then discarded, and + it never reached the per-job retry machinery so nothing capped it. Measured at 20 + iterations in 1.2s before the fix. + """ + calls = 0 + stop_event = asyncio.Event() + + async def _fake_process_next_queued_job(*, session=None, session_factory=None, services=None): + nonlocal calls + _ = (session, session_factory, services) + calls += 1 + raise RuntimeError("boom") + + monkeypatch.setattr("transcription.worker.process_next_queued_job", _fake_process_next_queued_job) + + with caplog.at_level(logging.CRITICAL): + await asyncio.wait_for( + run_worker_loop(stop_event=stop_event, poll_interval_seconds=0), + timeout=5, + ) + + assert calls == 1 + assert "Worker loop stopped after a non-retriable error" in caplog.text + + +@pytest.mark.asyncio +async def test_run_worker_loop_survives_retriable_exception(monkeypatch, caplog): + """A retriable fault is still suppressed so transient conditions do not stop work.""" calls = 0 stop_event = asyncio.Event() @@ -20,7 +53,12 @@ async def test_run_worker_loop_survives_process_next_exception(monkeypatch, capl _ = (session, session_factory, services) calls += 1 if calls == 1: - raise RuntimeError("boom") + raise AppError( + "transient", + category=ErrorCategory.EXTERNAL_PROVIDER, + suggestion="retry", + retriable=True, + ) stop_event.set() return False