diff --git a/tests/conftest.py b/tests/conftest.py index b0a6a9f..485dc95 100644 --- a/tests/conftest.py +++ b/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]] = [] diff --git a/tests/policy_hazard_tasks.py b/tests/policy_hazard_tasks.py index d58ae78..f6385b7 100644 --- a/tests/policy_hazard_tasks.py +++ b/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: diff --git a/tests/policy_tasks.py b/tests/policy_tasks.py index 7d21594..f15bcd8 100644 --- a/tests/policy_tasks.py +++ b/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" diff --git a/tests/tasks.py b/tests/tasks.py index 217eca1..567d8ac 100644 --- a/tests/tasks.py +++ b/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) diff --git a/tests/test_connection_pool.py b/tests/test_connection_pool.py index 1ff34b7..37f302d 100644 --- a/tests/test_connection_pool.py +++ b/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 diff --git a/tests/test_policy.py b/tests/test_policy.py index f8b9d37..2e17069 100644 --- a/tests/test_policy.py +++ b/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 diff --git a/tests/test_policy_hazards.py b/tests/test_policy_hazards.py index 2e35dc2..d52cd20 100644 --- a/tests/test_policy_hazards.py +++ b/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" diff --git a/tests/test_policy_worker.py b/tests/test_policy_worker.py index fac14cf..cfae374 100644 --- a/tests/test_policy_worker.py +++ b/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 diff --git a/tests/test_timeouts.py b/tests/test_timeouts.py index d09e92d..d3f0ed4 100644 --- a/tests/test_timeouts.py +++ b/tests/test_timeouts.py @@ -44,7 +44,11 @@ from django_ox.timeouts import MAX_SECONDS, task_timeouts_from_options from django_ox.worker import WATCHDOG_MAX_WAIT, Worker, _active_tracer -from .conftest import start_worker_thread, wait_for +from .conftest import ( + assert_counts_down_from_its_arming, + start_worker_thread, + wait_for, +) from .tasks import ( TASK_ROW, add, @@ -791,16 +795,17 @@ def row_count(status): @pytest.mark.django_db(transaction=True) class TestDeadline: - # The deadline is set when the attempt starts, so what a task reads back - # is the timeout, minus however long the worker took to get from the - # enqueue to the task body. SLACK is the room for that: a sixth of the - # timeout, which is far more than the trip costs and far less than the - # distance to any other number the worker could have used, so a report - # derived from something else still fails. + # The deadline is set when the attempt is armed, so counted from before + # run_once() it is the timeout plus however long the worker took to get + # from the enqueue to arming it. SLACK is the room for that: a sixth of + # the timeout, which is far more than the trip costs and far less than + # the distance to any other number the worker could have used, so a + # report derived from something else still fails. What the task reads + # back is checked against the arming itself, which needs no room. TIMEOUT = 30.0 SLACK = TIMEOUT / 6 - def test_deadline_and_remaining_inside_a_task(self): + def test_deadline_and_remaining_inside_a_task(self, task_state): worker = Worker( task_timeout=self.TIMEOUT, backoff_initial=0, poll_interval=0.05 ) @@ -814,17 +819,16 @@ def test_deadline_and_remaining_inside_a_task(self): # which is at or after `before`; it can only be late, never early. out = (deadline - before).total_seconds() assert self.TIMEOUT <= out <= self.TIMEOUT + self.SLACK, out - # What is left of it when the task looks can only be less. - assert self.TIMEOUT - self.SLACK <= reported["remaining"] <= self.TIMEOUT + # What is left of it when the task looks counts down to it. + assert_counts_down_from_its_arming(task_state["countdown"], self.TIMEOUT) - def test_inside_an_async_task(self): + def test_inside_an_async_task(self, task_state): worker = Worker( task_timeout=self.TIMEOUT, backoff_initial=0, poll_interval=0.05 ) - result = async_report_deadline.enqueue() + async_report_deadline.enqueue() assert worker.run_once() - result.refresh() - assert self.TIMEOUT - self.SLACK <= result.return_value <= self.TIMEOUT + assert_counts_down_from_its_arming(task_state["countdown"], self.TIMEOUT) def test_none_without_a_timeout_and_outside_a_task(self): worker = Worker(backoff_initial=0, poll_interval=0.05)