Measure task timeouts from watchdog arming

tests/conftest.py

@@ -95,6 +95,20 @@ def wait_for(predicate, timeout=5.0, interval=0.02):
     return False
 
 
+def assert_counts_down_from_its_arming(reading, timeout):
+    """
+    A tasks.countdown() reading: the watchdog armed `timeout` seconds,
+    before the body read remaining(), and remaining() counted down to that
+    deadline. Nothing here bounds the time from the arming to the read, which
+    is the worker's own path to the task body and as long as it happens to be.
+    """
+    armed, deadline = reading["armed"], reading["deadline"]
+    assert (reading["timeout"], armed + timeout) == (timeout, deadline), reading
+    assert armed <= reading["before"], reading
+    left = reading["remaining"]
+    assert deadline - reading["after"] <= left <= deadline - reading["before"], reading
+
+
 _worker_threads: list[tuple[Worker, threading.Thread]] = []
 
 

tests/policy_hazard_tasks.py

@@ -10,6 +10,11 @@
 overlap or a write from the wrong epoch shows up as a record that should not
 be there rather than as a final status that happens to look right.
 
+A body under a timeout also records the watchdog's own account of the
+attempt (tasks.countdown, _times_out): when it armed the timeout and for
+which deadline, on the monotonic clock it enforces the deadline with, which
+is the clock a timeout is measured on.
+
 A body that has to wait for the test waits on a gate: a file beside the log
 that the test creates. Every wait has a limit, so a gate that is never opened
 costs a bounded wait rather than a hung worker, and the test then fails on
@@ -33,7 +38,7 @@
 
 from .dead_connection_tasks import end_connection
 from .policy_tasks import note, policy_task, retry_now
-from .tasks import _busy
+from .tasks import _busy, countdown, watched
 
 #: The longest any body waits for a gate or for its siblings. Far past what
 #: any of these tests needs, so reaching it means the scenario did not happen.
@@ -42,6 +47,9 @@
 #: How many concurrent attempts the mixed-policy test runs in one worker.
 MIXED = 6
 
+#: How long a body that starts late holds back its start record.
+LATE = 2.0
+
 
 def _log_path():
     return Path(os.environ["OX_TEST_POLICY_LOG"])
@@ -309,16 +317,51 @@ def outlives_its_lease(context, seconds):
     return "outlived"
 
 
+def _times_out(context, late=0.0):
+    """
+    Run until the timeout strikes, and record the attempt on the watchdog's
+    own clock: when it armed the timeout and for which deadline, when the
+    body began and noted its start, when the watchdog decided the timeout
+    (its grace starts then) and when the timeout landed here. `late` holds
+    the start record back that many seconds, as a slow first query would.
+    """
+    began = time.monotonic()
+    worker, watch = watched()
+    if late:
+        _wait(lambda: time.monotonic() - began >= late)
+    _claim(
+        context,
+        began=began,
+        noted=time.monotonic(),
+        armed=watch.started,
+        deadline=watch.deadline,
+        timeout=watch.timeout,
+    )
+    try:
+        _busy(60)
+    finally:
+        _end(
+            context,
+            struck=time.monotonic(),
+            fired=watch.fired,
+            grace_at=watch.grace_at,
+            grace=worker.timeouts.grace,
+        )
+
+
 @policy_task(takes_context=True, max_attempts=2, timeout=7, backoff=retry_now)
 def times_out_past_its_lease(context):
-    _claim(context)
     if context.attempt > 1:
+        _claim(context)
         _end(context)
         return "second"
-    try:
-        _busy(60)
-    finally:
-        _end(context)
+    _times_out(context)
+    return "never"
+
+
+@policy_task(takes_context=True, max_attempts=1, timeout=7)
+def times_out_after_a_late_start(context, late):
+    _times_out(context, late)
     return "never"
 
 
@@ -449,11 +492,12 @@ def _mixed_started():
 
 
 def _mixed(context):
-    remaining = django_ox.remaining()
+    reading = countdown()
     deadline = django_ox.deadline()
     _claim(
         context,
-        remaining=remaining,
+        remaining=reading["remaining"],
+        countdown=reading,
         deadline=None if deadline is None else deadline.timestamp(),
     )
     if context.attempt == 1:

tests/policy_tasks.py

@@ -27,7 +27,7 @@
 from django_ox.compat import HAS_CORE_TASKS, task
 from django_ox.timeouts import MAX_SECONDS
 
-from .tasks import STATE, _busy
+from .tasks import STATE, _busy, countdown
 
 #: Whether @task forwards keyword arguments to the backend's task class.
 #: Django 6.1 and the 5.2 backport do; Django 6.0 does not.
@@ -206,7 +206,7 @@ def spins_on_emails(seconds):
 def reports_its_deadline_then_fails(pause):
     note(
         attempt="deadline",
-        remaining=django_ox.remaining(),
+        countdown=countdown(),
         deadline=django_ox.deadline().timestamp(),
     )
     time.sleep(pause)
@@ -230,7 +230,7 @@ def succeeds_with_budget_two():
 
 @policy_task(timeout=5)
 def reports_deadline_in_process():
-    STATE["remaining"] = django_ox.remaining()
+    STATE["countdown"] = countdown()
     STATE["deadline"] = django_ox.deadline()
     return "reported"
 

tests/tasks.py

@@ -3,6 +3,7 @@
 import asyncio
 import os
 import sys
+import threading
 import time
 from pathlib import Path
 
@@ -14,11 +15,53 @@
 from django_ox.compat import task
 from django_ox.exceptions import TaskTimeout
 from django_ox.models import OxTask
+from django_ox.worker import _executing, _live_workers
 
 # Mutable per-test state, reset by the `task_state` fixture.
 STATE: dict[str, object] = {}
 
 
+def watched(*, by_task=False):
+    """
+    The worker running this attempt and its watchdog's entry for it, which
+    the worker wrote when it armed the attempt's timeout, or None when the
+    attempt has no timeout. Found by this thread; with `by_task`, from a
+    coroutine on a thread of asgiref's, by the task the attempt is running.
+    """
+    ident = threading.get_ident()
+    path = _executing.get()
+    found = [
+        (worker, watch)
+        for worker in list(_live_workers)
+        for key, watch in list(worker._watches.items())
+        if (watch.db_task.task_path == path if by_task else key == ident)
+    ]
+    if not found:
+        return None
+    (one,) = found
+    return one
+
+
+def countdown(*, by_task=False):
+    """
+    remaining(), between two readings of the monotonic clock it counts down
+    on, and when the watchdog armed the deadline it counts down to, for how
+    long, on that same clock. Enough to check the countdown without assuming
+    anything about how long the worker took to reach the task body.
+    """
+    before = time.monotonic()
+    left = django_ox.remaining()
+    after = time.monotonic()
+    reading = {"before": before, "remaining": left, "after": after}
+    found = watched(by_task=by_task)
+    if found is not None:
+        _, watch = found
+        reading.update(
+            armed=watch.started, deadline=watch.deadline, timeout=watch.timeout
+        )
+    return reading
+
+
 def _spin(seconds, step=0.005):
     """Sleep in short slices: a loop that keeps returning to bytecode."""
     end = time.monotonic() + seconds
@@ -211,9 +254,10 @@ def query_then_hold():
 @task
 def report_deadline():
     at = django_ox.deadline()
+    STATE["countdown"] = reading = countdown()
     return {
         "deadline": None if at is None else at.isoformat(),
-        "remaining": django_ox.remaining(),
+        "remaining": reading["remaining"],
     }
 
 
@@ -230,7 +274,8 @@ async def async_spin(seconds):
 
 @task
 async def async_report_deadline():
-    return django_ox.remaining()
+    STATE["countdown"] = reading = countdown(by_task=True)
+    return reading["remaining"]
 
 
 @task(takes_context=True)

tests/test_connection_pool.py

@@ -2110,6 +2110,13 @@ def stuck_batch(add_alias):
         )
     pool = connections[ALIAS].pool
     connections[ALIAS].close()
+    # As in `claimed`: the claims ran before the pool's first connection was
+    # ready, so the pool grows in the background, and that connection can
+    # arrive between a test's two reads of pool_available. Wait until every
+    # connection the pool holds is available before a test compares counts.
+    assert wait_for(
+        lambda: (stats := pool.get_stats())["pool_available"] == stats["pool_size"]
+    ), pool.get_stats()
     return worker, watches, pool
 
 

tests/test_policy.py

@@ -36,6 +36,7 @@
 from django_ox.worker import Worker
 
 from . import policy_tasks, signal_tasks
+from .conftest import assert_counts_down_from_its_arming
 from .policy_tasks import DECORATOR_TAKES_POLICY
 from .tasks import STATE, add, fail_always
 
@@ -473,14 +474,14 @@ def test_deadline_and_remaining_follow_the_task_timeout(self, task_state):
         assert not worker.timeouts.enabled
         policy_tasks.reports_deadline_in_process.enqueue()
         assert worker.run_once() is True
-        assert 4 < task_state["remaining"] <= 5
+        assert_counts_down_from_its_arming(task_state["countdown"], 5)
         assert task_state["deadline"] is not None
 
     def test_the_task_timeout_wins_over_the_queue_and_the_worker(self, task_state):
         worker = Worker(task_timeout=60, **NO_WAIT)
         policy_tasks.reports_deadline_in_process.enqueue()
         assert worker.run_once() is True
-        assert 4 < task_state["remaining"] <= 5
+        assert_counts_down_from_its_arming(task_state["countdown"], 5)
 
     def test_the_interpreter_notice_is_given_for_a_task_timeout(
         self, caplog, monkeypatch

tests/test_policy_hazards.py

@@ -45,6 +45,7 @@
 from django_ox.models import OxTask
 
 from . import policy_hazard_tasks as hazard
+from .conftest import assert_counts_down_from_its_arming
 from .policy_hazard_tasks import gate, read_notes
 
 TIMEOUT_PATH = f"{TaskTimeout.__module__}.{TaskTimeout.__qualname__}"
@@ -312,6 +313,30 @@ def seconds_after(when, since):
     return (when - since).total_seconds()
 
 
+def assert_struck_at_its_deadline(notes, result, timeout):
+    """
+    The first attempt's timeout, as the worker's watchdog measured it, on
+    the monotonic clock it enforces with: armed `timeout` seconds before its
+    deadline and before the body began, decided at or after the deadline,
+    and landed in the body after that decision and within the worker's
+    grace of the deadline. Returns the start and end records.
+
+    The time from the body's start record to the strike is not bounded
+    below by the timeout: whatever ran between the arming and that record
+    came out of the attempt's seconds, and nothing bounds how long it took.
+    """
+    started = starts(notes, result)[0]
+    struck = ends(notes, result)[0]
+    armed, deadline, grace = started["armed"], started["deadline"], struck["grace"]
+    assert (started["timeout"], armed + timeout) == (timeout, deadline), started
+    assert armed <= started["began"] <= started["noted"] <= struck["struck"]
+    assert struck["fired"], struck
+    # The watchdog's grace starts when it decides the timeout.
+    assert deadline + grace <= struck["grace_at"] <= struck["struck"] + grace, struck
+    assert struck["struck"] - deadline < grace, (started, struck)
+    return started, struck
+
+
 def test_two_workers_claiming_at_once_never_share_an_attempt(workers, policy_log):
     kinds = [
         (hazard.race_succeeds, OxTask.Status.SUCCESSFUL, 1),
@@ -414,9 +439,7 @@ def test_renewal_holds_attempts_that_outlast_the_lease(workers, policy_log):
     (error,) = stored.errors
     assert error["exception_class_path"] == TIMEOUT_PATH
     assert "past the 7s timeout" in error["traceback"]
-    first_attempt = starts(notes, times_out)[0]
-    struck = ends(notes, times_out)[0]
-    assert 6.9 < struck["at"] - first_attempt["at"] < 30
+    assert_struck_at_its_deadline(notes, times_out, 7)
 
     stored = row(slow)
     assert (stored.status, stored.attempts, stored.lease_epoch) == (
@@ -440,6 +463,31 @@ def test_renewal_holds_attempts_that_outlast_the_lease(workers, policy_log):
         assert "lost its lease" not in process.text(), process.text()
 
 
+def test_a_body_that_starts_late_is_struck_at_the_deadline_armed_before_it(
+    workers, policy_log
+):
+    # Whatever runs between the worker arming the timeout and the body's
+    # start record (a first query opening its connection, a busy
+    # interpreter) comes out of the attempt's seven seconds. This body holds
+    # its record back LATE seconds on purpose: counted from that record the
+    # attempt had five seconds left, and counted from the arming, as the
+    # watchdog counts, it still had exactly seven.
+    result = hazard.times_out_after_a_late_start.enqueue(hazard.LATE)
+    worker = workers.start("one", "--batch")
+    assert worker.wait() == 0, worker.text()
+
+    stored = row(result)
+    assert (stored.status, stored.attempts) == (OxTask.Status.FAILED, 1)
+    (error,) = stored.errors
+    assert error["exception_class_path"] == TIMEOUT_PATH
+    assert "past the 7s timeout" in error["traceback"]
+    notes = read_notes(policy_log)
+    assert_history(stored, notes, result, {worker.pid})
+    started, _ = assert_struck_at_its_deadline(notes, result, 7)
+    assert started["noted"] - started["armed"] >= hazard.LATE
+    assert "ran past its 7s timeout on attempt 1/1" in worker.text()
+
+
 @stops_a_worker
 def test_an_owner_back_after_losing_its_lease_writes_nothing(workers, policy_log):
     result = hazard.late_owner.enqueue()
@@ -722,7 +770,7 @@ def started(result, attempt=1):
     (error,) = stored.errors
     assert error["exception_class_path"] == TIMEOUT_PATH
     assert "past the 1s timeout" in error["traceback"]
-    assert 0 < started(times_out)["remaining"] <= 1
+    assert_counts_down_from_its_arming(started(times_out)["countdown"], 1)
 
     # Its sibling's one second never reached it.
     stored = row(runs_long)
@@ -755,8 +803,10 @@ def started(result, attempt=1):
     assert (stored.status, stored.attempts) == (OxTask.Status.SUCCESSFUL, 2)
     assert stored.return_value == "second"
     # A fresh five seconds on each attempt, not a sibling's one.
-    for attempt in (1, 2):
-        assert 4 < started(retries, attempt)["remaining"] <= 5
+    first, second = (started(retries, n)["countdown"] for n in (1, 2))
+    assert_counts_down_from_its_arming(first, 5)
+    assert_counts_down_from_its_arming(second, 5)
+    assert first["armed"] < second["armed"]
     (asked,) = callbacks(notes, retries)
     assert asked["callback"] == "now"
 

tests/test_policy_worker.py

@@ -24,6 +24,7 @@
 from django_ox.models import OxTask
 
 from . import policy_tasks, signal_tasks
+from .conftest import assert_counts_down_from_its_arming
 from .policy_tasks import notes
 from .tasks import fail_always
 from .test_policy import SIGNAL_TASKS, assert_is_the_declared_task, tasks_setting
@@ -339,10 +340,10 @@ def test_every_retry_gets_a_fresh_deadline(policy_log):
     stored = row(result)
     assert (stored.status, stored.attempts) == ("FAILED", 2)
     first, second = notes(policy_log)
-    # The whole three seconds at the start of each attempt, although the
+    # The whole three seconds from each attempt's own arming, although the
     # first spent half of them before failing.
-    assert 2.5 < first["remaining"] <= 3
-    assert 2.5 < second["remaining"] <= 3
+    assert_counts_down_from_its_arming(first["countdown"], 3)
+    assert_counts_down_from_its_arming(second["countdown"], 3)
     assert second["deadline"] - first["deadline"] > 1.4