From 8a3e583128487f238d4d95ba7566ffc63b93d7cd Mon Sep 17 00:00:00 2001 From: Tom Hamilton Stubber Date: Mon, 28 Sep 2026 11:55:08 +0100 Subject: [PATCH 1/2] Alert on job wait time instead of Celery queue length The dispatcher deliberately holds the Celery queue at dispatcher_max_celery_queue during bursts, so erroring whenever it exceeded that fired constantly (CHRONOS-3R) without indicating a real problem. Log an error when a dispatched job waited longer than dispatcher_max_job_wait_seconds instead, and cap each dispatch cycle to the headroom left under the ceiling so it can't overshoot. Fixes CHRONOS-3R Co-Authored-By: Claude Opus 5.5 --- CLAUDE.md | 6 ++-- chronos/settings.py | 2 ++ chronos/tasks/dispatcher.py | 9 ++++++ chronos/worker.py | 7 ++--- tests/test_retries.py | 9 ++---- tests/test_round_robin.py | 63 +++++++++++++++++++++++++++++++++++-- 6 files changed, 81 insertions(+), 15 deletions(-) diff --git a/CLAUDE.md b/CLAUDE.md index 73ea2bc..6a9bfe7 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. After dispatch, if the job waited longer than `dispatcher_max_job_wait_seconds` since `JobPayload.enqueued_at`, it logs an error — this is the "webhooks are late" alert (the Celery queue length is expected to sit at the ceiling during bursts, so it isn't alerted on). -- **`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. 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`. - **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..d4bcdf0 100644 --- a/chronos/settings.py +++ b/chronos/settings.py @@ -47,6 +47,8 @@ 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 + # A job that sat in its branch queue longer than this before dispatch logs an error. + dispatcher_max_job_wait_seconds: int = 300 # Webhook HTTP client tuning diff --git a/chronos/tasks/dispatcher.py b/chronos/tasks/dispatcher.py index 0fd3d6b..768cb9c 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 @@ -138,4 +139,12 @@ def dispatch_cycle(batch_limit: int = settings.dispatcher_batch_limit): dispatched += 1 dispatch_logger.info('Dispatched %s for branch %d', payload.task_name, branch_id) + # Queue length alone is expected to sit at the backpressure ceiling during bursts; how long + # a job waited is what tells us webhooks are reaching customers late. + wait_seconds = (datetime.now(UTC) - payload.enqueued_at).total_seconds() + if wait_seconds > settings.dispatcher_max_job_wait_seconds: + dispatch_logger.error( + 'Job for branch %d waited %ds before dispatch. Check workers and speeds.', branch_id, wait_seconds + ) + return dispatched diff --git a/chronos/worker.py b/chronos/worker.py index 79c4d90..4824514 100644 --- a/chronos/worker.py +++ b/chronos/worker.py @@ -495,9 +495,6 @@ 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: - 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) with logfire.span('Sending webhooks for {provider=} {org_id=}', provider=provider, org_id=org_id): with Session(engine) as db: @@ -706,7 +703,9 @@ 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. + batch_limit = min(settings.dispatcher_batch_limit, max_celery_queue - celery_queue_len) + dispatched = dispatch_cycle(batch_limit=batch_limit) 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..f15855a 100644 --- a/tests/test_retries.py +++ b/tests/test_retries.py @@ -336,8 +336,8 @@ 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_long_celery_queue_does_not_log_error(client: TestClient, app_db: Session): + """The dispatcher holds the Celery queue at its ceiling during bursts, so a long queue is not 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() @@ -346,7 +346,4 @@ def test_queue_too_long_warning(client: TestClient, app_db: Session): with patch.object(job_queue, 'get_celery_queue_length', return_value=threshold + 50): 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 + mock_logger.error.assert_not_called() diff --git a/tests/test_round_robin.py b/tests/test_round_robin.py index a3c42c1..d6f0cf0 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 @@ -15,7 +16,8 @@ from chronos.sql_models import Provider, WebhookEndpoint, WebhookLog, WebhookStatus from chronos.tasks.dispatcher import dispatch_cycle -from chronos.tasks.queue import ACTIVE_BRANCHES_KEY, BRANCH_KEY_TEMPLATE, JobQueue +from chronos.tasks.queue import ACTIVE_BRANCHES_KEY, BRANCH_KEY_TEMPLATE, JobPayload, 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 tests.test_helpers import ( @@ -238,7 +240,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 +546,63 @@ def sleep_side_effect(seconds): assert mock_span.message == 'Dispatched 1 jobs' +def test_job_dispatcher_task_only_fills_remaining_headroom(): + """A cycle never dispatches more than the space left under the Celery queue ceiling.""" + from chronos.worker import job_dispatcher_task + + with ( + patch.object(job_queue, 'has_active_jobs', return_value=True), + patch.object(job_queue, 'get_celery_queue_length', return_value=settings.dispatcher_max_celery_queue - 3), + 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() + + mock_dispatch.assert_called_once_with(batch_limit=3) + + +def _enqueue_job_with_age(branch_id: int, age: timedelta) -> None: + """Push a job onto a branch queue as if it had been enqueued `age` ago.""" + payload = JobPayload( + task_name=task_send_webhooks.name, + branch_id=branch_id, + kwargs={'payload': 'p'}, + enqueued_at=datetime.now(UTC) - age, + ) + cache.rpush(BRANCH_KEY_TEMPLATE.format(branch_id), payload.model_dump_json()) + cache.sadd(ACTIVE_BRANCHES_KEY, str(branch_id)) + + +@patch.object(task_send_webhooks, 'apply_async') +def test_dispatch_cycle_logs_error_when_job_waited_too_long(mock_apply): + """A job that waited past dispatcher_max_job_wait_seconds is still dispatched but logs an error.""" + _enqueue_job_with_age(42, timedelta(seconds=settings.dispatcher_max_job_wait_seconds + 60)) + + with patch('chronos.tasks.dispatcher.dispatch_logger') as mock_logger: + dispatched = dispatch_cycle() + + assert dispatched == 1 + mock_apply.assert_called_once() + mock_logger.error.assert_called_once() + log_args = mock_logger.error.call_args[0] + assert 'waited' in log_args[0] + assert log_args[1] == 42 + assert log_args[2] >= settings.dispatcher_max_job_wait_seconds + 60 + + +@patch.object(task_send_webhooks, 'apply_async') +def test_dispatch_cycle_no_error_for_recent_job(mock_apply): + """A job dispatched within dispatcher_max_job_wait_seconds does not log an error.""" + _enqueue_job_with_age(42, timedelta(seconds=1)) + + with patch('chronos.tasks.dispatcher.dispatch_logger') as mock_logger: + dispatched = dispatch_cycle() + + assert dispatched == 1 + 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 From d515d683e0d8cbd5d6086ba117a7eb8f476c9d1c Mon Sep 17 00:00:00 2001 From: Tom Hamilton Stubber Date: Mon, 28 Sep 2026 12:09:49 +0100 Subject: [PATCH 2/2] Check job wait periodically and keep a high queue-length alert - Move the wait check out of dispatch_cycle into check_job_wait(), run by the dispatcher at most once a minute and before the backpressure check, so stalled workers still alert and a backlog logs one error a minute, not one per job. - Keep the queue-length error in task_send_webhooks at celery_queue_alert_length (1000) so producers that bypass the dispatcher (Bobbin, use_round_robin=False, retries) are still covered. - Make batch_limit a job_dispatcher_task parameter. Co-Authored-By: Claude Opus 5.5 --- CLAUDE.md | 6 +-- chronos/settings.py | 7 ++- chronos/tasks/dispatcher.py | 39 +++++++++++---- chronos/worker.py | 19 ++++++-- tests/test_retries.py | 22 ++++++--- tests/test_round_robin.py | 95 ++++++++++++++++++++++++------------- 6 files changed, 134 insertions(+), 54 deletions(-) diff --git a/CLAUDE.md b/CLAUDE.md index 6a9bfe7..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. After dispatch, if the job waited longer than `dispatcher_max_job_wait_seconds` since `JobPayload.enqueued_at`, it logs an error — this is the "webhooks are late" alert (the Celery queue length is expected to sit at the ceiling during bursts, so it isn't alerted on). +- **`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()` 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. 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`, `dispatcher_max_job_wait_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 d4bcdf0..45ac55d 100644 --- a/chronos/settings.py +++ b/chronos/settings.py @@ -47,8 +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 - # A job that sat in its branch queue longer than this before dispatch logs an error. + # 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 768cb9c..2fed80b 100644 --- a/chronos/tasks/dispatcher.py +++ b/chronos/tasks/dispatcher.py @@ -139,12 +139,35 @@ def dispatch_cycle(batch_limit: int = settings.dispatcher_batch_limit): dispatched += 1 dispatch_logger.info('Dispatched %s for branch %d', payload.task_name, branch_id) - # Queue length alone is expected to sit at the backpressure ceiling during bursts; how long - # a job waited is what tells us webhooks are reaching customers late. - wait_seconds = (datetime.now(UTC) - payload.enqueued_at).total_seconds() - if wait_seconds > settings.dispatcher_max_job_wait_seconds: - dispatch_logger.error( - 'Job for branch %d waited %ds before dispatch. Check workers and speeds.', branch_id, wait_seconds - ) - 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 4824514..966240e 100644 --- a/chronos/worker.py +++ b/chronos/worker.py @@ -495,6 +495,9 @@ 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.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) with logfire.span('Sending webhooks for {provider=} {org_id=}', provider=provider, org_id=org_id): with Session(engine) as db: @@ -657,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. @@ -683,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: @@ -704,8 +718,7 @@ def job_dispatcher_task( try: with logfire.span('Dispatching jobs') as span: # Only fill the headroom left, otherwise one cycle can push the queue to ~2x the ceiling. - batch_limit = min(settings.dispatcher_batch_limit, max_celery_queue - celery_queue_len) - dispatched = dispatch_cycle(batch_limit=batch_limit) + 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 f15855a..8525fcb 100644 --- a/tests/test_retries.py +++ b/tests/test_retries.py @@ -336,14 +336,24 @@ def test_retry_countdown_values_match_backoff_formula(app_db: Session): @respx.mock -def test_long_celery_queue_does_not_log_error(client: TestClient, app_db: Session): - """The dispatcher holds the Celery queue at its ceiling during bursts, so a long queue is not an error.""" +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=threshold + 50): + 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=json.dumps(payload), url_extension=None) + task_send_webhooks(payload=payload, url_extension=None) mock_logger.error.assert_not_called() + + 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=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 d6f0cf0..8e1e64f 100644 --- a/tests/test_round_robin.py +++ b/tests/test_round_robin.py @@ -15,11 +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.queue import ACTIVE_BRANCHES_KEY, BRANCH_KEY_TEMPLATE, JobPayload, JobQueue +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, @@ -546,60 +553,82 @@ def sleep_side_effect(seconds): assert mock_span.message == 'Dispatched 1 jobs' -def test_job_dispatcher_task_only_fills_remaining_headroom(): +def test_job_dispatcher_task_only_fills_remaining_headroom() -> None: """A cycle never dispatches more than the space left under the Celery queue ceiling.""" - from chronos.worker import job_dispatcher_task - with ( patch.object(job_queue, 'has_active_jobs', return_value=True), - patch.object(job_queue, 'get_celery_queue_length', return_value=settings.dispatcher_max_celery_queue - 3), + 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() + 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: - """Push a job onto a branch queue as if it had been enqueued `age` ago.""" - payload = JobPayload( - task_name=task_send_webhooks.name, - branch_id=branch_id, - kwargs={'payload': 'p'}, - enqueued_at=datetime.now(UTC) - age, - ) - cache.rpush(BRANCH_KEY_TEMPLATE.format(branch_id), payload.model_dump_json()) - cache.sadd(ACTIVE_BRANCHES_KEY, str(branch_id)) + """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') -@patch.object(task_send_webhooks, 'apply_async') -def test_dispatch_cycle_logs_error_when_job_waited_too_long(mock_apply): - """A job that waited past dispatcher_max_job_wait_seconds is still dispatched but logs an error.""" - _enqueue_job_with_age(42, timedelta(seconds=settings.dispatcher_max_job_wait_seconds + 60)) +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: - dispatched = dispatch_cycle() + check_job_wait() - assert dispatched == 1 - mock_apply.assert_called_once() mock_logger.error.assert_called_once() log_args = mock_logger.error.call_args[0] - assert 'waited' in log_args[0] - assert log_args[1] == 42 - assert log_args[2] >= settings.dispatcher_max_job_wait_seconds + 60 + assert 'has waited' in log_args[0] + assert log_args[1] == 30 + assert log_args[2] >= max_wait + 120 -@patch.object(task_send_webhooks, 'apply_async') -def test_dispatch_cycle_no_error_for_recent_job(mock_apply): - """A job dispatched within dispatcher_max_job_wait_seconds does not log an error.""" - _enqueue_job_with_age(42, timedelta(seconds=1)) +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: - dispatched = dispatch_cycle() + 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() - assert dispatched == 1 mock_logger.error.assert_not_called()