diff --git a/apps/worker/tests/contract/test_worker_shutdown_contract.py b/apps/worker/tests/contract/test_worker_shutdown_contract.py index b970baca..34e39827 100644 --- a/apps/worker/tests/contract/test_worker_shutdown_contract.py +++ b/apps/worker/tests/contract/test_worker_shutdown_contract.py @@ -3,6 +3,9 @@ import json from pathlib import Path +import pytest +from pydantic import ValidationError + def test_should_preserve_fargate_worker_sigterm_shutdown_contract( worker_contract_environment: None, @@ -33,3 +36,40 @@ def test_should_preserve_fargate_worker_sigterm_shutdown_contract( assert celery_app.conf.worker_enable_soft_shutdown_on_idle is True assert "REMAP_SIGTERM" not in environment_values assert worker_container["stopTimeout"] == 120 + + +def test_should_redeliver_interrupted_tasks_before_processing_jobs_expire( + worker_contract_environment: None, +) -> None: + from shared.core.celery_app import celery_app + from shared.core.config.job import JobConfig + + task_time_limit_seconds: int = celery_app.conf.task_time_limit + visibility_timeout_seconds: int = celery_app.conf.broker_transport_options[ + "visibility_timeout" + ] + processing_expiry_seconds: int = JobConfig().JOB_PROCESSING_EXPIRE_SECONDS + + assert task_time_limit_seconds == 3600 + assert visibility_timeout_seconds == 4500 + assert processing_expiry_seconds == 14400 + assert ( + task_time_limit_seconds + < visibility_timeout_seconds + < processing_expiry_seconds + ) + + +def test_should_reject_processing_expiry_before_interrupted_task_redelivery( + monkeypatch: pytest.MonkeyPatch, + worker_contract_environment: None, +) -> None: + from shared.core.config import AppConfig + + monkeypatch.setenv("JOB_PROCESSING_EXPIRE_SECONDS", "4500") + + with pytest.raises( + ValidationError, + match="TASK_TIME_LIMIT_SECONDS < BROKER_VISIBILITY_TIMEOUT_SECONDS", + ): + AppConfig() diff --git a/packages/shared-python/shared/core/celery_app.py b/packages/shared-python/shared/core/celery_app.py index 9a40c9de..d4b5e39b 100644 --- a/packages/shared-python/shared/core/celery_app.py +++ b/packages/shared-python/shared/core/celery_app.py @@ -55,7 +55,7 @@ def get_unique_node_name() -> str: enable_utc=True, # Task execution task_track_started=True, - task_time_limit=3600, # 1 hour hard limit + task_time_limit=app_config.TASK_TIME_LIMIT_SECONDS, task_soft_time_limit=3300, # 55 minutes soft limit worker_prefetch_multiplier=1, task_acks_late=True, @@ -77,7 +77,7 @@ def get_unique_node_name() -> str: broker_pool_limit=app_config.BROKER_POOL_LIMIT, # Redis transport — visibility_timeout must exceed task_time_limit broker_transport_options={ - "visibility_timeout": 43200, # 12 hours + "visibility_timeout": app_config.BROKER_VISIBILITY_TIMEOUT_SECONDS, "retry_on_timeout": True, # Keep all Kombu Redis broker keys in one Redis Cluster hash slot. "global_keyprefix": "{celery}", diff --git a/packages/shared-python/shared/core/config/app.py b/packages/shared-python/shared/core/config/app.py index 9b03f788..2ef0d043 100644 --- a/packages/shared-python/shared/core/config/app.py +++ b/packages/shared-python/shared/core/config/app.py @@ -2,6 +2,7 @@ from typing import Callable, cast +from pydantic import model_validator from pydantic_settings import SettingsConfigDict from .ai import AIConfig @@ -32,6 +33,22 @@ class AppConfig( ): """Application configuration — all config components merged.""" + @model_validator(mode="after") + def validate_worker_redelivery_window(self) -> "AppConfig": + """Ensure interrupted tasks are redelivered before stale-job expiry.""" + if not ( + self.TASK_TIME_LIMIT_SECONDS + < self.BROKER_VISIBILITY_TIMEOUT_SECONDS + < self.JOB_PROCESSING_EXPIRE_SECONDS + ): + raise ValueError( + "Worker timing must satisfy TASK_TIME_LIMIT_SECONDS " + "< BROKER_VISIBILITY_TIMEOUT_SECONDS " + "< JOB_PROCESSING_EXPIRE_SECONDS" + ) + + return self + def validate_all(self) -> bool: """Validate the combined application configuration.""" validations = [ diff --git a/packages/shared-python/shared/core/config/celery.py b/packages/shared-python/shared/core/config/celery.py index e2205378..4d0c4a8f 100644 --- a/packages/shared-python/shared/core/config/celery.py +++ b/packages/shared-python/shared/core/config/celery.py @@ -2,7 +2,7 @@ Celery configuration — Redis-backed broker and result backend. """ -from typing import Dict +from typing import ClassVar, Dict from pydantic import BaseModel, Field @@ -10,6 +10,9 @@ class CeleryConfig(BaseModel): """Celery configuration backed by a dedicated Redis instance.""" + TASK_TIME_LIMIT_SECONDS: ClassVar[int] = 3600 + BROKER_VISIBILITY_TIMEOUT_SECONDS: ClassVar[int] = 4500 + # Dedicated Redis instance for Celery broker / result backend / RedBeat. # Separate from the application Redis (REDIS_*) to isolate connection pools. CELERY_REDIS_URL: str = Field(