Alert on webhook job wait time instead of Celery queue length

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`, `*`.

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
 

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
+        )

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

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
+    )

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