diff --git a/.env.example b/.env.example index e7edf95..04a2f76 100644 --- a/.env.example +++ b/.env.example @@ -59,6 +59,7 @@ WORKER_MAX_RETRIES=0 WORKER_PROVIDER_TIMEOUT_SECONDS=30.0 WORKER_STALE_JOB_SECONDS=30.0 WORKER_RETRY_BACKOFF_SECONDS=1.0 +WORKER_SHUTDOWN_GRACE_SECONDS=5.0 WORKER_MIN_TRANSCRIPTION_CHARS=0 WORKER_MIN_TRANSCRIPTION_LINES=0 WORKER_FAIL_ON_FINISH_REASON_LENGTH=false diff --git a/docs/production-runbook.md b/docs/production-runbook.md index 8542e01..0138d0f 100644 --- a/docs/production-runbook.md +++ b/docs/production-runbook.md @@ -127,3 +127,17 @@ Every suppression must be: Do not use broad or rationale-free suppressions. If a diagnostic is not a known false positive, fix the code instead of suppressing it. + +## 8. Worker shutdown budget + +Worker shutdown waits for at most: + +`WORKER_PROVIDER_TIMEOUT_SECONDS + WORKER_SHUTDOWN_GRACE_SECONDS` + +`WORKER_PROVIDER_TIMEOUT_SECONDS` covers an in-flight provider call, and +`WORKER_SHUTDOWN_GRACE_SECONDS` is extra time for the loop to persist outcomes +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. diff --git a/src/transcription/app.py b/src/transcription/app.py index 8e21576..17cc262 100644 --- a/src/transcription/app.py +++ b/src/transcription/app.py @@ -60,6 +60,9 @@ async def _lifespan(app: FastAPI): worker_consumer_lifespan( session_factory=app.state.runtime.session_factory, poll_interval_seconds=1.0, + shutdown_timeout_seconds=( + settings.worker_provider_timeout_seconds + settings.worker_shutdown_grace_seconds + ), ) ) app.state.worker_stop_event = stop_event diff --git a/src/transcription/config.py b/src/transcription/config.py index 99c0bd2..c33b2e4 100644 --- a/src/transcription/config.py +++ b/src/transcription/config.py @@ -116,6 +116,7 @@ class Settings(BaseSettings): 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_shutdown_grace_seconds: float = Field(default=5.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/worker.py b/src/transcription/worker.py index b948a0e..4321800 100644 --- a/src/transcription/worker.py +++ b/src/transcription/worker.py @@ -123,6 +123,7 @@ async def worker_consumer_lifespan( *, session_factory: async_sessionmaker[AsyncSession] | None = None, poll_interval_seconds: float = 1.0, + shutdown_timeout_seconds: float = 2.0, ) -> AsyncGenerator[tuple[asyncio.Event, WorkerNotifier, WorkerHealth]]: """Start and stop the worker consumer loop for app lifespan.""" stop_event = asyncio.Event() @@ -146,7 +147,7 @@ async def worker_consumer_lifespan( stop_event.set() worker_notifier.notify() try: - await asyncio.wait_for(worker_task, timeout=2.0) + await asyncio.wait_for(worker_task, timeout=shutdown_timeout_seconds) except TimeoutError: worker_task.cancel() with suppress(asyncio.CancelledError): diff --git a/tests/test_app.py b/tests/test_app.py index 2f2f3ae..62718bd 100644 --- a/tests/test_app.py +++ b/tests/test_app.py @@ -28,6 +28,7 @@ class TestAppLifespan: def test_startup_initializes_runtime_dependencies(self, monkeypatch, tmp_path): """Startup initializes logging, schema, directories, and worker resources.""" calls = [] + worker_kwargs = {} monkeypatch.setattr("transcription.app.configure_logging", lambda _settings: calls.append("logging")) monkeypatch.setattr("transcription.app.register_pages", lambda _app: None) @@ -53,6 +54,7 @@ class TestAppLifespan: @asynccontextmanager async def _worker_lifespan(**_kwargs): + worker_kwargs.update(_kwargs) calls.append("worker_start") yield object(), object(), object() calls.append("worker_stop") @@ -78,6 +80,9 @@ class TestAppLifespan: assert "worker_start" in calls assert "worker_stop" in calls assert "dispose_db" in calls + assert worker_kwargs["shutdown_timeout_seconds"] == pytest.approx( + settings.worker_provider_timeout_seconds + settings.worker_shutdown_grace_seconds + ) assert settings.upload_dir.exists() assert settings.prompt_dir.exists() diff --git a/tests/test_config.py b/tests/test_config.py index fd71cb8..748dc7c 100644 --- a/tests/test_config.py +++ b/tests/test_config.py @@ -169,6 +169,7 @@ class TestWorkerReliabilitySettings: assert settings.worker_max_retries == 0 assert settings.worker_stale_job_seconds == 30.0 assert settings.worker_retry_backoff_seconds == 1.0 + assert settings.worker_shutdown_grace_seconds == 5.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 bda4b9e..9b34b98 100644 --- a/tests/test_meta_contract_guards.py +++ b/tests/test_meta_contract_guards.py @@ -169,6 +169,7 @@ def test_env_example_default_values_match_settings_defaults(): "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_SHUTDOWN_GRACE_SECONDS": str(defaults.worker_shutdown_grace_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(),