diff --git a/CLAUDE.md b/CLAUDE.md index 73ea2bc..7287ece 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -76,9 +76,9 @@ Schema is managed with **Alembic**. There is no `alembic.ini` — config lives i - `enqueue(task_name, branch_id, **kwargs)` pushes a `JobPayload` (task_name, branch_id, kwargs, enqueued_at, trace_context) and adds branch to the set. - `peek(branch_id)`, `ack(branch_id)`, `get_active_branches()`, `get_cursor()` / `set_cursor()`, `get_celery_queue_length()` (length of Redis list `celery`). -- **`chronos/tasks/dispatcher.py`**: `dispatch_cycle(batch_limit)` runs one round: get active branches, rotate by cursor (bisect_right), for each branch peek → validate → `task.apply_async(kwargs)` → ack → set cursor, until batch_limit or no more branches. Poison payloads (e.g. invalid JSON) are acked and skipped. Trace context is restored when dispatching. +- **`chronos/tasks/dispatcher.py`**: `dispatch_cycle(batch_limit)` runs one round: get active branches, rotate by cursor (bisect_right), for each branch peek → validate → `task.apply_async(kwargs)` → ack → set cursor, until batch_limit or no more branches. Poison payloads (e.g. invalid JSON) are acked and skipped. Trace context is restored when dispatching. `check_job_wait()` peeks every branch's head job and logs an error if the oldest `JobPayload.enqueued_at` is older than `dispatcher_max_job_wait_seconds` — the "round-robin webhooks are late" alert. -- **`chronos/worker.py`**: `job_dispatcher_task` runs in a loop: if no active jobs, sleep idle_delay; if Celery queue length ≥ `dispatcher_max_celery_queue`, sleep cycle_delay; else run `dispatch_cycle()` then sleep cycle_delay. Uses `acks_late=False` so the broker doesn't redeliver this never-ending task. +- **`chronos/worker.py`**: `job_dispatcher_task` runs in a loop: if no active jobs, sleep idle_delay; if Celery queue length ≥ `dispatcher_max_celery_queue`, sleep cycle_delay; else run `dispatch_cycle()` with `batch_limit` capped to the headroom left under `dispatcher_max_celery_queue` (so a cycle can't overshoot the ceiling), then sleep cycle_delay. While there are active jobs it also calls `check_job_wait()` at most once every `JOB_WAIT_CHECK_INTERVAL_SECONDS` (60s), before the backpressure check so stalled workers still alert. Uses `acks_late=False` so the broker doesn't redeliver this never-ending task. - **`chronos/tasks/worker_startup.py`**: On `worker_ready`, if the worker consumes the `dispatcher` queue, it starts `job_dispatcher_task.apply_async(countdown=60)`. @@ -92,7 +92,7 @@ Dispatcher must be run with `--soft-time-limit=0 --time-limit=0` so it never tim - **Databases**: `pg_dsn`, `test_pg_dsn`, `redis_url`. - **Auth**: `tc2_shared_key` (Bearer token for TC). - **Observability**: `logfire_token`, `sentry_dsn`. -- **Round-robin**: `use_round_robin`; `dispatcher_max_celery_queue`, `dispatcher_batch_limit`, `dispatcher_cycle_delay_seconds`, `dispatcher_idle_delay_seconds`. +- **Round-robin**: `use_round_robin`; `dispatcher_max_celery_queue`, `dispatcher_batch_limit`, `dispatcher_cycle_delay_seconds`, `dispatcher_idle_delay_seconds`, `dispatcher_max_job_wait_seconds`. `celery_queue_alert_length` (1000): `task_send_webhooks` logs "Queue is too long" above this. It is kept well above `dispatcher_max_celery_queue` (which the dispatcher holds the queue at during bursts) so it only catches backlogs from producers that bypass the dispatcher (Bobbin, `use_round_robin=False`, retries). - **HTTP client**: `webhook_http_timeout_seconds`, `webhook_http_max_connections`. CORS: in production, origins are beta or secure TutorCruncher; in `dev_mode`, `*`. diff --git a/chronos/settings.py b/chronos/settings.py index a08b198..45ac55d 100644 --- a/chronos/settings.py +++ b/chronos/settings.py @@ -47,6 +47,13 @@ class Settings(BaseSettings): dispatcher_cycle_delay_seconds: float = 0.01 # Sleep duration when no active branches have queued jobs. dispatcher_idle_delay_seconds: float = 1.0 + # The dispatcher logs an error (checked at most once a minute) when the oldest job in the branch queues has + # waited longer than this. + dispatcher_max_job_wait_seconds: int = 300 + # A send task logs an error when the broker queue is longer than this. Kept well above + # dispatcher_max_celery_queue, which the dispatcher holds the queue at during bursts, so it only catches + # backlogs from producers that bypass the dispatcher (Bobbin, use_round_robin=False, retries). + celery_queue_alert_length: int = 1000 # Webhook HTTP client tuning diff --git a/chronos/tasks/dispatcher.py b/chronos/tasks/dispatcher.py index 0fd3d6b..2fed80b 100644 --- a/chronos/tasks/dispatcher.py +++ b/chronos/tasks/dispatcher.py @@ -22,6 +22,7 @@ import json import logging from bisect import bisect_right +from datetime import UTC, datetime from opentelemetry import context as otel_context from opentelemetry.propagate import extract @@ -139,3 +140,34 @@ def dispatch_cycle(batch_limit: int = settings.dispatcher_batch_limit): dispatch_logger.info('Dispatched %s for branch %d', payload.task_name, branch_id) return dispatched + + +def check_job_wait() -> None: + """ + Log an error if the oldest job in the branch queues has waited longer than dispatcher_max_job_wait_seconds. + + The Celery queue is expected to sit at the backpressure ceiling during bursts, so its length says little; how long + a job has waited is what tells us webhooks are reaching customers late. Each branch queue is FIFO, so its head is + its oldest job. + """ + # avoids circular import here + from chronos.worker import job_queue + + heads = [] + for branch_id in job_queue.get_active_branches(): + try: + payload = job_queue.peek(branch_id) + except (json.JSONDecodeError, ValidationError): + # dispatch_cycle discards poison payloads + continue + if payload: + heads.append((payload.enqueued_at, branch_id)) + if not heads: + return + + oldest_enqueued_at, branch_id = min(heads) + wait_seconds = (datetime.now(UTC) - oldest_enqueued_at).total_seconds() + if wait_seconds > settings.dispatcher_max_job_wait_seconds: + dispatch_logger.error( + 'Oldest queued job (branch %d) has waited %ds. Check workers and speeds.', branch_id, wait_seconds + ) diff --git a/chronos/worker.py b/chronos/worker.py index 79c4d90..966240e 100644 --- a/chronos/worker.py +++ b/chronos/worker.py @@ -495,7 +495,7 @@ def task_send_webhooks(payload: str | dict, url_extension: str = None): provider, org_id = _resolve_send_target(loaded_payload) qlength = job_queue.get_celery_queue_length() - if qlength > settings.dispatcher_max_celery_queue: + if qlength > settings.celery_queue_alert_length: app_logger.error('Queue is too long, qlength=%s. Check workers and speeds.', qlength) app_logger.info('Starting send webhook task for %s org %s. qlength=%s.', provider, org_id, qlength) @@ -660,11 +660,15 @@ def dispatch_branch_task(task, branch_id: int, **kwargs) -> None: ) +JOB_WAIT_CHECK_INTERVAL_SECONDS = 60 + + @celery_app.task(name='job_dispatcher', acks_late=False, trail=False) def job_dispatcher_task( max_celery_queue: int = settings.dispatcher_max_celery_queue, cycle_delay: float = settings.dispatcher_cycle_delay_seconds, # at most the webhooks need to wait for 10ms idle_delay: float = settings.dispatcher_idle_delay_seconds, # this is for when no active branches, so doesn't need to be as frequent + batch_limit: int = settings.dispatcher_batch_limit, ) -> None: """ Celery task that runs round robin dispatcher. @@ -686,15 +690,22 @@ def job_dispatcher_task( """ from billiard.exceptions import SoftTimeLimitExceeded - from chronos.tasks.dispatcher import dispatch_cycle + from chronos.tasks.dispatcher import check_job_wait, dispatch_cycle app_logger.info('Job dispatcher started') + last_wait_check = 0.0 while True: try: if not job_queue.has_active_jobs(): time.sleep(idle_delay) continue + # Checked before backpressure so jobs piling up behind stalled workers still alert, and on an + # interval so a long backlog logs one error a minute rather than one per job. + if time.monotonic() - last_wait_check >= JOB_WAIT_CHECK_INTERVAL_SECONDS: + last_wait_check = time.monotonic() + check_job_wait() + # LLEN celery: measures pending broker queue only, not in-flight tasks. celery_queue_len = job_queue.get_celery_queue_length() if celery_queue_len >= max_celery_queue: @@ -706,7 +717,8 @@ def job_dispatcher_task( cycle_token = otel_context.attach(otel_context.Context()) try: with logfire.span('Dispatching jobs') as span: - dispatched = dispatch_cycle() + # Only fill the headroom left, otherwise one cycle can push the queue to ~2x the ceiling. + dispatched = dispatch_cycle(batch_limit=min(batch_limit, max_celery_queue - celery_queue_len)) if dispatched > 0: # without this gaurd the cycle will log every 10ms it finds nothing # in the dispatcher queue which can be noisy diff --git a/tests/test_retries.py b/tests/test_retries.py index 3be145b..8525fcb 100644 --- a/tests/test_retries.py +++ b/tests/test_retries.py @@ -336,17 +336,24 @@ def test_retry_countdown_values_match_backoff_formula(app_db: Session): @respx.mock -def test_queue_too_long_warning(client: TestClient, app_db: Session): - """When Celery queue length exceeds threshold, app_logger.error is called.""" +def test_queue_too_long_error_only_above_alert_length(client: TestClient, app_db: Session) -> None: + """ + The dispatcher holds the Celery queue around dispatcher_max_celery_queue during bursts, so only a queue past + celery_queue_alert_length logs an error. + """ ep = _create_endpoint(app_db, tc_id=108) respx.post(ep.webhook_url).mock(return_value=httpx.Response(200)) - payload = get_dft_webhook_data() - threshold = settings.dispatcher_max_celery_queue + payload = json.dumps(get_dft_webhook_data()) + + with patch.object(job_queue, 'get_celery_queue_length', return_value=settings.dispatcher_max_celery_queue + 50): + with patch('chronos.worker.app_logger') as mock_logger: + task_send_webhooks(payload=payload, url_extension=None) + mock_logger.error.assert_not_called() - with patch.object(job_queue, 'get_celery_queue_length', return_value=threshold + 50): + alert_length = settings.celery_queue_alert_length + with patch.object(job_queue, 'get_celery_queue_length', return_value=alert_length + 1): with patch('chronos.worker.app_logger') as mock_logger: - task_send_webhooks(payload=json.dumps(payload), url_extension=None) - mock_logger.error.assert_called_once() - call_args = mock_logger.error.call_args[0] - assert 'Queue is too long' in call_args[0] - assert call_args[1] == threshold + 50 + task_send_webhooks(payload=payload, url_extension=None) + mock_logger.error.assert_called_once_with( + 'Queue is too long, qlength=%s. Check workers and speeds.', alert_length + 1 + ) diff --git a/tests/test_round_robin.py b/tests/test_round_robin.py index a3c42c1..8e1e64f 100644 --- a/tests/test_round_robin.py +++ b/tests/test_round_robin.py @@ -4,6 +4,7 @@ import hashlib import hmac import json +from datetime import UTC, datetime, timedelta from unittest.mock import MagicMock, patch import httpx @@ -14,10 +15,18 @@ from sqlmodel import Session, select from chronos.sql_models import Provider, WebhookEndpoint, WebhookLog, WebhookStatus -from chronos.tasks.dispatcher import dispatch_cycle +from chronos.tasks.dispatcher import check_job_wait, dispatch_cycle from chronos.tasks.queue import ACTIVE_BRANCHES_KEY, BRANCH_KEY_TEMPLATE, JobQueue +from chronos.utils import settings from chronos.views.tutorcruncher import _extract_branch_id -from chronos.worker import _async_post_webhooks, cache, dispatch_branch_task, job_queue, task_send_webhooks +from chronos.worker import ( + _async_post_webhooks, + cache, + dispatch_branch_task, + job_dispatcher_task, + job_queue, + task_send_webhooks, +) from tests.test_helpers import ( _get_webhook_headers, get_dft_con_webhook_data, @@ -238,7 +247,7 @@ def test_job_dispatcher_task_catches_generic_exception_and_keeps_running(): call_count = 0 - def dispatch_side_effect(): + def dispatch_side_effect(batch_limit: int): nonlocal call_count call_count += 1 if call_count == 1: @@ -544,6 +553,85 @@ def sleep_side_effect(seconds): assert mock_span.message == 'Dispatched 1 jobs' +def test_job_dispatcher_task_only_fills_remaining_headroom() -> None: + """A cycle never dispatches more than the space left under the Celery queue ceiling.""" + with ( + patch.object(job_queue, 'has_active_jobs', return_value=True), + patch.object(job_queue, 'get_celery_queue_length', return_value=97), + patch('chronos.tasks.dispatcher.dispatch_cycle', return_value=0) as mock_dispatch, + patch('time.sleep', side_effect=SystemExit), + ): + with pytest.raises(SystemExit): + job_dispatcher_task(max_celery_queue=100, batch_limit=100) + + mock_dispatch.assert_called_once_with(batch_limit=3) + + +def test_job_dispatcher_task_checks_job_wait_once_per_interval() -> None: + """The wait check runs even while backpressure pauses dispatch, and at most once per interval.""" + sleep_calls = [] + + def sleep_side_effect(seconds: float) -> None: + sleep_calls.append(seconds) + if len(sleep_calls) >= 3: + raise SystemExit + + with ( + patch.object(job_queue, 'has_active_jobs', return_value=True), + patch.object(job_queue, 'get_celery_queue_length', return_value=100), + patch('chronos.tasks.dispatcher.check_job_wait') as mock_check, + patch('chronos.tasks.dispatcher.dispatch_cycle') as mock_dispatch, + patch('time.sleep', side_effect=sleep_side_effect), + ): + with pytest.raises(SystemExit): + job_dispatcher_task(max_celery_queue=100) + + mock_dispatch.assert_not_called() + mock_check.assert_called_once() + + +def _enqueue_job_with_age(branch_id: int, age: timedelta) -> None: + """Enqueue a job as if it had been enqueued `age` ago.""" + with patch('chronos.tasks.queue.datetime') as mock_datetime: + mock_datetime.now.return_value = datetime.now(UTC) - age + job_queue.enqueue(task_send_webhooks.name, branch_id=branch_id, payload='p') + + +def test_check_job_wait_logs_oldest_late_job() -> None: + """The error names the branch with the oldest job; empty and poison branches are skipped.""" + max_wait = settings.dispatcher_max_job_wait_seconds + _enqueue_job_with_age(10, timedelta(seconds=1)) + _enqueue_job_with_age(20, timedelta(seconds=max_wait + 60)) + _enqueue_job_with_age(30, timedelta(seconds=max_wait + 120)) + cache.rpush(BRANCH_KEY_TEMPLATE.format(40), 'not valid json{{{') + cache.sadd(ACTIVE_BRANCHES_KEY, '40', '50') + + with patch('chronos.tasks.dispatcher.dispatch_logger') as mock_logger: + check_job_wait() + + mock_logger.error.assert_called_once() + log_args = mock_logger.error.call_args[0] + assert 'has waited' in log_args[0] + assert log_args[1] == 30 + assert log_args[2] >= max_wait + 120 + + +def test_check_job_wait_no_error_for_recent_jobs() -> None: + _enqueue_job_with_age(10, timedelta(seconds=1)) + + with patch('chronos.tasks.dispatcher.dispatch_logger') as mock_logger: + check_job_wait() + + mock_logger.error.assert_not_called() + + +def test_check_job_wait_no_jobs() -> None: + with patch('chronos.tasks.dispatcher.dispatch_logger') as mock_logger: + check_job_wait() + + mock_logger.error.assert_not_called() + + def test_job_dispatcher_task_dispatch_cycle_returns_zero(): """When dispatch_cycle returns 0, the dispatcher does not set span.message.""" from chronos.worker import job_dispatcher_task