diff --git a/.env.example b/.env.example index 1bbc274..f521e0e 100644 --- a/.env.example +++ b/.env.example @@ -57,6 +57,7 @@ DATABASE_BACKUP_DIR=./data/backups # --- worker reliability --- WORKER_MAX_RETRIES=0 WORKER_PROVIDER_TIMEOUT_SECONDS=30.0 +WORKER_STALE_JOB_SECONDS=30.0 WORKER_MIN_TRANSCRIPTION_CHARS=0 WORKER_MIN_TRANSCRIPTION_LINES=0 WORKER_FAIL_ON_FINISH_REASON_LENGTH=false diff --git a/src/transcription/app.py b/src/transcription/app.py index ef76513..8e21576 100644 --- a/src/transcription/app.py +++ b/src/transcription/app.py @@ -71,11 +71,11 @@ async def _lifespan(app: FastAPI): async def _recover_stale_processing_jobs(app: FastAPI) -> None: """Re-queue stale processing jobs at startup. - Any job left in PROCESSING longer than the configured provider timeout is - assumed orphaned and moved back to QUEUED before the worker starts. + Any job left in PROCESSING longer than the stale-job threshold is assumed + orphaned and moved back to QUEUED before the worker starts. """ settings = app.state.settings - stale_before = datetime.now(UTC) - timedelta(seconds=settings.worker_provider_timeout_seconds) + stale_before = datetime.now(UTC) - timedelta(seconds=settings.worker_stale_job_seconds) recovered = await app.state.services.jobs.requeue_stale_processing_jobs(stale_before=stale_before) if recovered > 0: logger.warning("Recovered %s stale processing job(s) at startup", recovered) diff --git a/src/transcription/config.py b/src/transcription/config.py index 095584a..1d6f531 100644 --- a/src/transcription/config.py +++ b/src/transcription/config.py @@ -114,6 +114,7 @@ class Settings(BaseSettings): # Bounded only from below. Vision transcription of a dense page routinely runs # 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_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/worker.py b/src/transcription/worker.py index e963b0c..b948a0e 100644 --- a/src/transcription/worker.py +++ b/src/transcription/worker.py @@ -9,6 +9,9 @@ from contextlib import asynccontextmanager from contextlib import contextmanager from contextlib import suppress from dataclasses import dataclass +from datetime import UTC +from datetime import datetime +from datetime import timedelta from typing import Literal from typing import Protocol from typing import runtime_checkable @@ -215,6 +218,14 @@ async def run_worker_loop( await asyncio.wait_for(wake_event.wait(), timeout=poll_interval_seconds) wake_event.clear() + if session_factory is not None: + with handle_worker_exceptions(operation="worker.requeue_stale_processing_jobs"): + stale_seconds = services.jobs.settings.worker_stale_job_seconds + stale_before = datetime.now(UTC) - timedelta(seconds=stale_seconds) + recovered = await services.jobs.requeue_stale_processing_jobs(stale_before=stale_before) + if recovered > 0: + logger.warning("Recovered %s stale processing job(s) in worker loop", recovered) + processed_any = False while True: with handle_worker_exceptions(operation="worker.process_next_queued_job"): diff --git a/tests/test_config.py b/tests/test_config.py index a41d52b..9bb9627 100644 --- a/tests/test_config.py +++ b/tests/test_config.py @@ -167,6 +167,7 @@ class TestWorkerReliabilitySettings: """worker retry settings default to no retries.""" settings = _make_settings() assert settings.worker_max_retries == 0 + assert settings.worker_stale_job_seconds == 30.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 db7a5dd..78f676e 100644 --- a/tests/test_meta_contract_guards.py +++ b/tests/test_meta_contract_guards.py @@ -167,6 +167,7 @@ def test_env_example_default_values_match_settings_defaults(): "DATABASE_BACKUP_DIR": str(defaults.database_backup_dir), "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_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(), diff --git a/tests/test_worker.py b/tests/test_worker.py index 48325d8..1db9ee9 100644 --- a/tests/test_worker.py +++ b/tests/test_worker.py @@ -1,9 +1,15 @@ import asyncio import logging +from datetime import UTC +from datetime import datetime +from datetime import timedelta from typing import cast import pytest +from sqlalchemy.ext.asyncio import async_sessionmaker +from sqlmodel.ext.asyncio.session import AsyncSession +from transcription.config import Settings from transcription.errors import AppError from transcription.errors import ErrorCategory from transcription.services import ServiceBundle @@ -79,6 +85,45 @@ async def test_run_worker_loop_survives_retriable_exception(monkeypatch, caplog) assert worker_health.snapshot().state == "stopped" +@pytest.mark.asyncio +async def test_run_worker_loop_periodically_requeues_stale_processing_jobs(monkeypatch): + stop_event = asyncio.Event() + stale_sweep_calls: list[datetime] = [] + + class _Jobs: + settings = Settings(openrouter_api_key="test-key", worker_stale_job_seconds=120.0) + + async def requeue_stale_processing_jobs(self, *, stale_before, session=None): + _ = session + stale_sweep_calls.append(stale_before) + return 1 + + class _Bundle: + jobs = _Jobs() + + async def aclose(self): + return + + monkeypatch.setattr( + "transcription.worker.ServiceBundle.from_session_factory", + classmethod(lambda _cls, _factory=None, **_kwargs: cast(ServiceBundle, _Bundle())), + ) + + async def _fake_process_next_queued_job(*, session=None, session_factory=None, services=None): + _ = (session, session_factory, services) + stop_event.set() + return False + + monkeypatch.setattr("transcription.worker.process_next_queued_job", _fake_process_next_queued_job) + + fake_session_factory = cast(async_sessionmaker[AsyncSession], object()) + await run_worker_loop(stop_event=stop_event, poll_interval_seconds=0, session_factory=fake_session_factory) + + assert len(stale_sweep_calls) >= 1 + expected_upper_bound = datetime.now(UTC) - timedelta(seconds=120.0) + assert stale_sweep_calls[0] <= expected_upper_bound + + @pytest.mark.asyncio async def test_run_worker_loop_reuses_one_bundle_across_jobs(monkeypatch): """HIGH-02: the provider client is built once per loop, not once per job."""