diff --git a/README.md b/README.md index 19f5245..327cec3 100644 --- a/README.md +++ b/README.md @@ -98,3 +98,30 @@ taskbadger.init( - **`task.configure(...).defer(...)` is not tracked.** Procrastinate's `configure()` returns a separate `JobDeferrer` whose methods bypass our wrapper. Use `task.defer(...)` directly for tracked deferrals. Tasks deferred via `configure().defer()` will run normally but will not appear in TaskBadger. - **`task.batch_defer*` is not tracked.** Same reason as `configure().defer()`. - **Tasks added via `app.add_tasks_from(blueprint)` after `ProcrastinateSystemIntegration` is constructed are not auto-instrumented.** Construct the integration after all blueprints are registered, or apply `@track` to those tasks explicitly. + +### Keeping long-running tasks fresh + +A task with a `stale_timeout` is marked `stale` by Task Badger if it goes too long without an +update. Set `heartbeat_interval` (seconds) to have the SDK ping the task for you while it runs, +rather than updating it from the task body. + +For Procrastinate, on the task or on `ProcrastinateSystemIntegration(...)`: + +```python +@track(heartbeat_interval=60) +@app.task +async def slow_job(): + ... +``` + +For Celery, on `CelerySystemIntegration(...)`, on the task, or per call with +`slow_job.apply_async(taskbadger_heartbeat_interval=60)`: + +```python +@app.task(base=taskbadger.Task, taskbadger_heartbeat_interval=60) +def slow_job(): + ... +``` + +Unless `stale_timeout` is given explicitly it is set to twice the interval. All running tasks are +pinged from a single background thread, started the first time a task with a heartbeat runs. diff --git a/integration_tests/tasks.py b/integration_tests/tasks.py index c1cd9e0..e708cd4 100644 --- a/integration_tests/tasks.py +++ b/integration_tests/tasks.py @@ -1,7 +1,13 @@ +import time + from celery import shared_task import taskbadger.celery +HEARTBEAT_INTERVAL = 1 +# long enough to sample the task's `updated` time while it is still running +SLOW_ADD_DURATION = 6 + @shared_task(bind=True, base=taskbadger.celery.Task) def add(self, x, y): @@ -14,3 +20,10 @@ def add(self, x, y): def add_auto_track(self, x, y): assert self.request.taskbadger_task_id is not None, "missing task ID on self.request" return x + y + + +@shared_task(bind=True, base=taskbadger.celery.Task, taskbadger_heartbeat_interval=HEARTBEAT_INTERVAL) +def slow_add(self, x, y): + """Runs long enough to go stale without a heartbeat, and never updates itself.""" + time.sleep(SLOW_ADD_DURATION) + return x + y diff --git a/integration_tests/test_celery.py b/integration_tests/test_celery.py index 4bd24d8..c23a276 100644 --- a/integration_tests/test_celery.py +++ b/integration_tests/test_celery.py @@ -1,11 +1,13 @@ import logging import random +import time import pytest +import taskbadger from taskbadger import StatusEnum -from .tasks import add, add_auto_track +from .tasks import HEARTBEAT_INTERVAL, add, add_auto_track, slow_add @pytest.fixture(autouse=True) @@ -41,3 +43,30 @@ def test_celery_auto_track(celery_session_app, celery_session_worker): a, b = random.randint(1, 1000), random.randint(1, 1000) result = add_auto_track.delay(a, b) assert result.get(timeout=10, propagate=True) == a + b + + +def test_celery_heartbeat(celery_session_app, celery_session_worker): + """The worker pings the task while it runs, so it doesn't go stale.""" + a, b = random.randint(1, 1000), random.randint(1, 1000) + result = slow_add.delay(a, b) + + running = _wait_for_status(result.taskbadger_task_id, StatusEnum.PROCESSING) + assert running.stale_timeout == HEARTBEAT_INTERVAL * 2 + + time.sleep(HEARTBEAT_INTERVAL * 2) + pinged = taskbadger.get_task(running.id) + + assert result.get(timeout=30, propagate=True) == a + b + # still running, so the task can only have been touched by the heartbeat + assert pinged.status == StatusEnum.PROCESSING + assert pinged.updated > running.updated, "task was not pinged while it was running" + + +def _wait_for_status(task_id, status, timeout=15): + deadline = time.monotonic() + timeout + while time.monotonic() < deadline: + task = taskbadger.get_task(task_id) + if task.status == status: + return task + time.sleep(0.2) + pytest.fail(f"task '{task_id}' did not reach status '{status}'") diff --git a/integration_tests/test_procrastinate.py b/integration_tests/test_procrastinate.py index 0119f40..d8fa4da 100644 --- a/integration_tests/test_procrastinate.py +++ b/integration_tests/test_procrastinate.py @@ -8,9 +8,11 @@ pyproject.toml. """ +import datetime import logging import os import random +import time import procrastinate import psycopg @@ -26,6 +28,10 @@ "postgresql://postgres:postgres@localhost:5432/procrastinate", ) +HEARTBEAT_INTERVAL = 1 +# long enough for the heartbeat to fire while the task is still running +SLOW_TASK_DURATION = 3 + @pytest.fixture(autouse=True) def _check_log_errors(caplog): @@ -102,6 +108,38 @@ def add_manual(a, b): assert fetched.data == {"result": a + b} +def test_heartbeat(app): + """The worker pings the task while it runs, so it doesn't go stale.""" + + @track(heartbeat_interval=HEARTBEAT_INTERVAL) + @app.task(name="slow", queue="taskbadger_int_hb") + def slow(): + # `run_worker` blocks the test, so sample the task's `updated` time from + # inside the body. Fetched directly to bypass the integration's cache. + tb_id = current_task().id + before = taskbadger.get_task(tb_id).updated + time.sleep(SLOW_TASK_DURATION) + after = taskbadger.get_task(tb_id).updated + current_task().update(data={"before": before.isoformat(), "after": after.isoformat()}) + + job_id = slow.defer() + app.run_worker( + queues=["taskbadger_int_hb"], + wait=False, + install_signal_handlers=False, + listen_notify=False, + ) + + args = _fetch_job_args(job_id) + fetched = taskbadger.get_task(args["__taskbadger_task_id__"]) + + assert fetched.status == StatusEnum.SUCCESS + assert fetched.stale_timeout == HEARTBEAT_INTERVAL * 2 + before = datetime.datetime.fromisoformat(fetched.data["before"]) + after = datetime.datetime.fromisoformat(fetched.data["after"]) + assert after > before, "task was not pinged while it was running" + + def test_auto_track_via_system(app): ProcrastinateSystemIntegration(app=app, auto_track_tasks=True) diff --git a/taskbadger/_heartbeat.py b/taskbadger/_heartbeat.py new file mode 100644 index 0000000..203736e --- /dev/null +++ b/taskbadger/_heartbeat.py @@ -0,0 +1,127 @@ +"""Background heartbeat support for the system integrations (Celery, +Procrastinate). Not part of the public API. + +A task that sets ``stale_timeout`` is marked ``stale`` by the API if it goes +too long without an update. Long-running queue tasks that don't report progress +would trip that timeout while perfectly healthy, so the integrations can ping +the task periodically for the duration of the run. + +All in-flight tasks are pinged from a single daemon thread rather than one +thread per task. The thread is started lazily on first use and sleeps +indefinitely while nothing is registered. +""" + +from __future__ import annotations + +import dataclasses +import logging +import threading +import time + +from ._integrations import TERMINAL_STATES, is_valid_interval +from .mug import Badger, Settings +from .safe_sdk import update_task_safe + +log = logging.getLogger("taskbadger") + + +@dataclasses.dataclass +class _Entry: + interval: float + settings: Settings + due: float + + +class Heartbeat: + """Periodically pings registered tasks to keep them from going stale.""" + + def __init__(self): + self._lock = threading.Lock() + self._entries: dict[str, _Entry] = {} + self._wake = threading.Event() + self._thread = None + + def start(self, task_id: str, interval: float | None) -> None: + """Begin pinging ``task_id`` every ``interval`` seconds. + + No-op if there is no interval or Task Badger isn't configured in the + calling thread. Registering the same task again resets its schedule. + """ + if not task_id or not is_valid_interval(interval): + return + + settings = Badger.current.settings + if settings is None: + return + + with self._lock: + self._entries[task_id] = _Entry(interval, settings, time.monotonic() + interval) + self._ensure_thread() + self._wake.set() + + def stop(self, task_id: str) -> None: + """Stop pinging ``task_id``. No-op if it isn't registered.""" + with self._lock: + self._entries.pop(task_id, None) + + def stop_all(self) -> None: + with self._lock: + self._entries.clear() + + def _ensure_thread(self) -> None: + """Start the beat thread if it isn't running. Called with the lock held.""" + if self._thread is not None and self._thread.is_alive(): + return + + # `is_alive()` is also False for a thread inherited from a parent + # process, so a forked worker (e.g. Celery's prefork pool) starts its + # own thread the first time it runs a tracked task. + self._thread = threading.Thread(target=self._run, name="taskbadger-heartbeat", daemon=True) + self._thread.start() + + def _run(self) -> None: + while True: + try: + timeout = self._beat() + except Exception: + # Never let the thread die: every registered task would then go + # stale with nothing to restart the pings. + log.exception("heartbeat beat failed") + timeout = 1.0 + self._wake.wait(timeout) + self._wake.clear() + + def _beat(self) -> float | None: + """Ping every task that is due and return the seconds to sleep for. + + ``None`` means "sleep until a task is registered". + """ + now = time.monotonic() + with self._lock: + due = [(task_id, entry) for task_id, entry in self._entries.items() if entry.due <= now] + for _, entry in due: + entry.due = now + entry.interval + + for task_id, entry in due: + self._ping(task_id, entry) + + with self._lock: + if not self._entries: + return None + next_due = min(entry.due for entry in self._entries.values()) + return max(next_due - time.monotonic(), 0) + + def _ping(self, task_id: str, entry: _Entry) -> None: + # The beat thread has its own context, so bind the settings captured + # when the task registered rather than relying on inheritance. + if Badger.current.settings is not entry.settings: + Badger.current.bind(entry.settings) + + log.debug("heartbeat ping '%s'", task_id) + task = update_task_safe(task_id) + if task is not None and task.status in TERMINAL_STATES: + # The task finished (or was cancelled) elsewhere. + self.stop(task_id) + + +heartbeat = Heartbeat() diff --git a/taskbadger/_integrations.py b/taskbadger/_integrations.py index dc0e0f7..400ca29 100644 --- a/taskbadger/_integrations.py +++ b/taskbadger/_integrations.py @@ -12,9 +12,11 @@ import collections import logging +import math import re from . import sdk +from .exceptions import ConfigurationError from .internal.models import StatusEnum from .systems import System @@ -27,6 +29,10 @@ StatusEnum.STALE, } +# When a heartbeat is configured but no `stale_timeout` is given, the timeout +# is derived from the interval using this factor (as the CLI's `run` does). +STALE_TIMEOUT_FACTOR = 2 + class TaskCache: """Bounded LRU-ish cache for TaskBadger Task objects. @@ -91,6 +97,48 @@ def match_task_name(task_name: str, includes, excludes) -> bool: return True +def is_valid_interval(interval) -> bool: + """Return True if ``interval`` is usable as a heartbeat interval.""" + return isinstance(interval, int | float) and interval > 0 + + +def validate_interval(interval) -> None: + """Raise if ``interval`` is set but isn't usable as a heartbeat interval.""" + if interval is not None and not is_valid_interval(interval): + raise ConfigurationError(f"heartbeat_interval must be a positive number of seconds: {interval!r}") + + +def resolve_heartbeat_options(heartbeat_interval, stale_timeout, system): + """Resolve the heartbeat interval and stale timeout for a single task. + + Values set on the task win over those set on the system integration. If a + heartbeat is configured without a stale timeout, one is derived from the + interval. + + Returns: + A tuple of ``(heartbeat_interval, stale_timeout)``, either of which may + be ``None``. + """ + if system is not None: + if heartbeat_interval is None: + heartbeat_interval = system.heartbeat_interval + if stale_timeout is None: + stale_timeout = system.stale_timeout + + if heartbeat_interval is not None and not is_valid_interval(heartbeat_interval): + # Per-task values don't go through `validate_interval`, and a bad one + # shouldn't stop the task from being tracked. + log.warning("Ignoring invalid heartbeat_interval: %r", heartbeat_interval) + heartbeat_interval = None + + if stale_timeout is None and heartbeat_interval: + # `stale_timeout` is whole seconds, so round up: a sub-second interval + # must not produce a timeout of 0. + stale_timeout = max(1, math.ceil(heartbeat_interval * STALE_TIMEOUT_FACTOR)) + + return heartbeat_interval, stale_timeout + + class BaseSystemIntegration(System): """Common ctor + ``track_task`` body for system integrations. @@ -98,11 +146,22 @@ class BaseSystemIntegration(System): additional filtering (e.g. skipping built-in tasks). """ - def __init__(self, auto_track_tasks=True, includes=None, excludes=None, record_task_args=False): + def __init__( + self, + auto_track_tasks=True, + includes=None, + excludes=None, + record_task_args=False, + heartbeat_interval=None, + stale_timeout=None, + ): + validate_interval(heartbeat_interval) self.auto_track_tasks = auto_track_tasks self.includes = includes self.excludes = excludes self.record_task_args = record_task_args + self.heartbeat_interval = heartbeat_interval + self.stale_timeout = stale_timeout def track_task(self, task_name: str) -> bool: if not self.auto_track_tasks: diff --git a/taskbadger/celery.py b/taskbadger/celery.py index e30f65d..1804fd9 100644 --- a/taskbadger/celery.py +++ b/taskbadger/celery.py @@ -6,6 +6,7 @@ from celery.signals import ( before_task_publish, task_failure, + task_postrun, task_prerun, task_retry, task_success, @@ -13,7 +14,8 @@ from kombu import serialization from . import sdk -from ._integrations import TERMINAL_STATES, safe_get_task, task_cache +from ._heartbeat import heartbeat +from ._integrations import TERMINAL_STATES, resolve_heartbeat_options, safe_get_task, task_cache from .internal.models import StatusEnum from .mug import Badger from .safe_sdk import create_task_safe, update_task_safe @@ -23,6 +25,11 @@ TB_KWARGS_ARG = f"{KWARG_PREFIX}kwargs" IGNORE_ARGS = {TB_KWARGS_ARG, f"{KWARG_PREFIX}task", f"{KWARG_PREFIX}task_id", f"{KWARG_PREFIX}record_task_args"} TB_TASK_ID = f"{KWARG_PREFIX}task_id" +TB_HEARTBEAT_INTERVAL = f"{KWARG_PREFIX}heartbeat_interval" +TB_STALE_TIMEOUT = f"{KWARG_PREFIX}stale_timeout" +# Marks a request whose signal handlers opened the Task Badger session, so that +# only they close it again. +TB_OWNS_SESSION = f"{KWARG_PREFIX}owns_session" log = logging.getLogger("taskbadger") @@ -40,6 +47,12 @@ class Task(celery.Task): No tracking is done for tasks that ar executed synchronously either via `.appy()` or if Celery is configured to run tasks eagerly. + Task Badger options may be set as `taskbadger_`-prefixed arguments on the task or on + `apply_async`. `taskbadger_heartbeat_interval` is treated specially: rather than being + set on the task it makes the worker ping the task every N seconds while it runs, which + keeps tasks with a `stale_timeout` from going stale. Unless `taskbadger_stale_timeout` + is also given it is set to twice the interval. + Access to the task is provided via the `taskbadger_task` property of the Celery task. The task ID may also be accessed via the `taskbadger_task_id` property. These may be `None` if the task is not being tracked (e.g. Task Badger is not configured or @@ -152,6 +165,14 @@ def task_publish_handler(sender=None, headers=None, body=None, **kwargs): kwargs.setdefault("external_id", headers["id"]) name = kwargs.pop("name", headers["task"]) + # `heartbeat_interval` isn't a task field; it tells the worker how often to + # ping while the task runs and may set a `stale_timeout` for it to protect. + heartbeat_interval, stale_timeout = resolve_heartbeat_options( + kwargs.pop("heartbeat_interval", None), kwargs.pop("stale_timeout", None), celery_system + ) + if stale_timeout is not None: + kwargs["stale_timeout"] = stale_timeout + global_record_task_args = celery_system and celery_system.record_task_args if headers.get("taskbadger_record_task_args", global_record_task_args): data = { @@ -170,6 +191,9 @@ def task_publish_handler(sender=None, headers=None, body=None, **kwargs): if task: meta = {TB_TASK_ID: task.id} headers.update(meta) + if heartbeat_interval: + # Carried on the message so per-call intervals reach the worker. + headers[TB_HEARTBEAT_INTERVAL] = heartbeat_interval if ctask: ctask.update_state(task_id=headers["id"], state="PENDING", meta=meta) @@ -247,15 +271,28 @@ def _maybe_create_task(signal_sender): if not manual_track and not auto_track: return - enter_session() + enter_session(signal_sender) delivery_info = getattr(signal_sender.request, "delivery_info", None) or {} queue = delivery_info.get("routing_key") external_id = signal_sender.request.id - task = create_task_safe(task_name, status=StatusEnum.PENDING, data=data, queue=queue, external_id=external_id) + create_kwargs = {"status": StatusEnum.PENDING, "data": data, "queue": queue, "external_id": external_id} + # `before_task_publish` never ran for these, so per-call options are still + # sitting in the headers rather than resolved into the message. + header_kwargs = headers.get(TB_KWARGS_ARG) or {} + heartbeat_interval, stale_timeout = resolve_heartbeat_options( + header_kwargs.get("heartbeat_interval", getattr(signal_sender, TB_HEARTBEAT_INTERVAL, None)), + header_kwargs.get("stale_timeout", getattr(signal_sender, TB_STALE_TIMEOUT, None)), + celery_system, + ) + if stale_timeout is not None: + create_kwargs["stale_timeout"] = stale_timeout + task = create_task_safe(task_name, **create_kwargs) if task: # Store the task ID in the request so _update_task can find it signal_sender.request.update({TB_TASK_ID: task.id}) + if heartbeat_interval: + signal_sender.request.update({TB_HEARTBEAT_INTERVAL: heartbeat_interval}) task_cache.set(task.id, task) @@ -263,6 +300,15 @@ def _maybe_create_task(signal_sender): def task_prerun_handler(sender=None, **kwargs): _maybe_create_task(sender) _update_task(sender, StatusEnum.PROCESSING) + _start_heartbeat(sender) + + +@task_postrun.connect +def task_postrun_handler(sender=None, **kwargs): + # postrun always fires, unlike the success/failure/retry signals. + task_id = _get_taskbadger_task_id(sender.request) + if task_id: + heartbeat.stop(task_id) @task_success.connect @@ -301,7 +347,7 @@ def _update_task(signal_sender, status, einfo=None): # ignore tasks that have already been set to a terminal state (probably in the task body) return - enter_session() + enter_session(signal_sender) data = None if einfo: @@ -311,25 +357,51 @@ def _update_task(signal_sender, status, einfo=None): task_cache.set(task_id, task) -def enter_session(): +def _start_heartbeat(signal_sender): + """Ping the task periodically for the duration of the run so that tasks + with a ``stale_timeout`` don't go stale while they're still working. + + The interval is resolved wherever the task was created — at publish time + for most tasks, in ``_maybe_create_task`` for the rest — so that it always + matches the ``stale_timeout`` the task was created with. + """ + request = signal_sender.request + task_id = _get_taskbadger_task_id(request) + if not task_id: + return + + interval = request.get(TB_HEARTBEAT_INTERVAL) + if interval is None and request.headers: + interval = request.headers.get(TB_HEARTBEAT_INTERVAL) + heartbeat.start(task_id, interval) + + +def enter_session(signal_sender): if not Badger.is_configured(): return session = Badger.current.session() - if not session.client: - session.__enter__() + if session.client: + # Already open, e.g. an eager task running inside the caller's + # `taskbadger.Session()`. Not ours, so don't close it either. + return + session.__enter__() + signal_sender.request.update({TB_OWNS_SESSION: True}) def exit_session(signal_sender): - headers = signal_sender.request.headers - if not headers: - return - - task_id = headers.get(TB_TASK_ID) + request = signal_sender.request + # not `request.headers`: eager tasks carry the ID on the request itself + # (the task id is checked first so that untracked tasks never touch `Badger`) + task_id = _get_taskbadger_task_id(request) if not task_id or not Badger.is_configured(): return task_cache.unset(task_id) + if not request.get(TB_OWNS_SESSION): + return + request.update({TB_OWNS_SESSION: False}) + session = Badger.current.session() if session.client: session.__exit__() diff --git a/taskbadger/procrastinate.py b/taskbadger/procrastinate.py index 3ae6364..3db2f43 100644 --- a/taskbadger/procrastinate.py +++ b/taskbadger/procrastinate.py @@ -17,7 +17,14 @@ import logging from contextvars import ContextVar -from ._integrations import TERMINAL_STATES, safe_get_task, task_cache +from ._heartbeat import heartbeat +from ._integrations import ( + TERMINAL_STATES, + resolve_heartbeat_options, + safe_get_task, + task_cache, + validate_interval, +) from .internal.models import StatusEnum from .mug import Badger from .safe_sdk import create_task_safe, update_task_safe @@ -35,6 +42,7 @@ _INSTRUMENTED_ATTR = "_taskbadger_instrumented" _MANUAL_ATTR = "_taskbadger_manual" _OPTS_ATTR = "_taskbadger_opts" +_SYSTEM_ATTR = "_taskbadger_system" _current_tb_task_id: ContextVar[str | None] = ContextVar("_current_tb_task_id", default=None) @@ -55,6 +63,12 @@ def _instrument_task(task, system=None, manual=False, opts=None): if manual: setattr(task, _MANUAL_ATTR, True) + # Set before the idempotency check: a task already instrumented by ``@track`` + # when the system integration is constructed still has to inherit the + # system's options (heartbeat_interval, stale_timeout, record_task_args). + if system is not None or not hasattr(task, _SYSTEM_ATTR): + setattr(task, _SYSTEM_ATTR, system) + if getattr(task, _INSTRUMENTED_ATTR, False): return @@ -73,6 +87,7 @@ async def wrapped(*args, **kwargs): token = _current_tb_task_id.set(tb_id) try: _update_status(tb_id, StatusEnum.PROCESSING) + heartbeat.start(tb_id, _heartbeat_interval(task)) try: result = await original_func(*args, **kwargs) except Exception as exc: @@ -81,6 +96,7 @@ async def wrapped(*args, **kwargs): _update_status(tb_id, StatusEnum.SUCCESS) return result finally: + heartbeat.stop(tb_id) _current_tb_task_id.reset(token) else: @@ -92,6 +108,7 @@ def wrapped(*args, **kwargs): token = _current_tb_task_id.set(tb_id) try: _update_status(tb_id, StatusEnum.PROCESSING) + heartbeat.start(tb_id, _heartbeat_interval(task)) try: result = original_func(*args, **kwargs) except Exception as exc: @@ -100,12 +117,12 @@ def wrapped(*args, **kwargs): _update_status(tb_id, StatusEnum.SUCCESS) return result finally: + heartbeat.stop(tb_id) _current_tb_task_id.reset(token) _wrap_defer(task) task.func = wrapped setattr(task, _INSTRUMENTED_ATTR, True) - setattr(task, "_taskbadger_system", system) def _update_status(tb_id, status, exception=None): @@ -136,6 +153,14 @@ def _update_status(tb_id, status, exception=None): task_cache.set(tb_id, updated) +def _heartbeat_interval(task): + """Seconds between pings while ``task`` runs, or ``None`` for no heartbeat.""" + opts = getattr(task, _OPTS_ATTR, {}) or {} + system = getattr(task, _SYSTEM_ATTR, None) + interval, _ = resolve_heartbeat_options(opts.get("heartbeat_interval"), None, system) + return interval + + def _wrap_defer(task): """Wrap ``task.defer`` and ``task.defer_async`` so they create a TaskBadger task in PENDING state and inject its id into the job's task_kwargs. @@ -176,7 +201,7 @@ def _create_pending_task(task, task_kwargs, queue=None): if not Badger.is_configured(): return None - system = getattr(task, "_taskbadger_system", None) + system = getattr(task, _SYSTEM_ATTR, None) manual = getattr(task, _MANUAL_ATTR, False) auto = bool(system) and system.track_task(task.name) if not manual and not auto: @@ -192,6 +217,10 @@ def _create_pending_task(task, task_kwargs, queue=None): if key in opts and opts[key] is not None: create_kwargs[key] = opts[key] + _, stale_timeout = resolve_heartbeat_options(opts.get("heartbeat_interval"), opts.get("stale_timeout"), system) + if stale_timeout is not None: + create_kwargs["stale_timeout"] = stale_timeout + data = dict(opts.get("data") or {}) record_args = opts.get("record_task_args") @@ -243,7 +272,7 @@ def _serialize_kwargs(kwargs): return {} -_TRACK_OPT_KEYS = ("name", "value_max", "tags", "data", "record_task_args") +_TRACK_OPT_KEYS = ("name", "value_max", "tags", "data", "record_task_args", "heartbeat_interval", "stale_timeout") def track(original_task=None, **opts): @@ -267,11 +296,19 @@ async def big_job(...): ... record_task_args: If True, serialize the Procrastinate job kwargs and store them under ``data["procrastinate_task_kwargs"]``. Defaults to ``None`` meaning "inherit from system integration if any, else False". + heartbeat_interval: Seconds between automatic pings while the task is + running. This keeps long-running tasks from being marked stale + without having to update them from the task body. Unless + ``stale_timeout`` is also given it will be set to twice this value. + stale_timeout: Maximum allowed time between task updates (seconds) + before the task is considered stale. """ unknown = set(opts) - set(_TRACK_OPT_KEYS) if unknown: raise TypeError(f"track() got unexpected keyword arguments: {sorted(unknown)}") + validate_interval(opts.get("heartbeat_interval")) + def wrap(task): _instrument_task(task, system=None, manual=True, opts=opts) return task diff --git a/taskbadger/systems/celery.py b/taskbadger/systems/celery.py index dad50f4..48135ef 100644 --- a/taskbadger/systems/celery.py +++ b/taskbadger/systems/celery.py @@ -4,7 +4,15 @@ class CelerySystemIntegration(BaseSystemIntegration): identifier = "celery" - def __init__(self, auto_track_tasks=True, includes=None, excludes=None, record_task_args=False): + def __init__( + self, + auto_track_tasks=True, + includes=None, + excludes=None, + record_task_args=False, + heartbeat_interval=None, + stale_timeout=None, + ): """ Args: auto_track_tasks: Automatically track all Celery tasks regardless of whether they are using the @@ -15,12 +23,19 @@ def __init__(self, auto_track_tasks=True, includes=None, excludes=None, record_t excludes: A list of task names to exclude from tracking. As with `includes`, these can be either the full task name or a regular expression. Exclusions take precedence over inclusions. record_task_args: Record the arguments passed to each task. + heartbeat_interval: Seconds between automatic pings while a task is running. This keeps + long-running tasks from being marked stale without having to update them from the task + body. Unless `stale_timeout` is also given it will be set to twice this value. + stale_timeout: Maximum allowed time between task updates (seconds) before the task is + considered stale. Set on the task when it is created. """ super().__init__( auto_track_tasks=auto_track_tasks, includes=includes, excludes=excludes, record_task_args=record_task_args, + heartbeat_interval=heartbeat_interval, + stale_timeout=stale_timeout, ) if auto_track_tasks: diff --git a/taskbadger/systems/procrastinate.py b/taskbadger/systems/procrastinate.py index a77922f..0b19f81 100644 --- a/taskbadger/systems/procrastinate.py +++ b/taskbadger/systems/procrastinate.py @@ -16,6 +16,8 @@ def __init__( includes=None, excludes=None, record_task_args=False, + heartbeat_interval=None, + stale_timeout=None, ): """ Args: @@ -29,12 +31,21 @@ def __init__( ``includes``. Exclusions take precedence. record_task_args: Record the task's defer kwargs into the TaskBadger task's ``data`` under ``procrastinate_task_kwargs``. + heartbeat_interval: Seconds between automatic pings while a task is + running. This keeps long-running tasks from being marked stale + without having to update them from the task body. Unless + ``stale_timeout`` is also given it will be set to twice this value. + stale_timeout: Maximum allowed time between task updates (seconds) + before the task is considered stale. Set on the task when it is + created. """ super().__init__( auto_track_tasks=auto_track_tasks, includes=includes, excludes=excludes, record_task_args=record_task_args, + heartbeat_interval=heartbeat_interval, + stale_timeout=stale_timeout, ) self.app = app diff --git a/tests/conftest.py b/tests/conftest.py index 1279e2b..49a25de 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -1,9 +1,18 @@ import pytest +from taskbadger._heartbeat import heartbeat from taskbadger._integrations import task_cache from taskbadger.mug import Badger, Settings +@pytest.fixture(autouse=True) +def _clear_heartbeats(): + """Make sure no test leaves a task registered with the heartbeat thread.""" + heartbeat.stop_all() + yield + heartbeat.stop_all() + + @pytest.fixture(autouse=True) def _clear_task_cache(): """Clear the shared integrations task cache around every test so cached diff --git a/tests/test_celery_heartbeat.py b/tests/test_celery_heartbeat.py new file mode 100644 index 0000000..aba509a --- /dev/null +++ b/tests/test_celery_heartbeat.py @@ -0,0 +1,228 @@ +"""Tests for the automatic heartbeat that keeps running Celery tasks fresh. + +See the note in ``test_celery.py`` about the Celery fixture setup. +""" + +import logging +import time +from unittest import mock + +import pytest + +import taskbadger +from taskbadger import StatusEnum +from taskbadger.celery import Task +from taskbadger.mug import Badger, Settings +from taskbadger.systems.celery import CelerySystemIntegration +from tests.utils import task_for_test + + +def _wait_for_call_count(mock_obj, expected, timeout=10.0): + deadline = time.monotonic() + timeout + while mock_obj.call_count < expected and time.monotonic() < deadline: + time.sleep(0.01) + + +@pytest.fixture +def _bind_settings_with_heartbeat(): + systems = [CelerySystemIntegration(heartbeat_interval=5)] + Badger.current.bind( + Settings( + "https://taskbadger.net", + "token", + "org", + "proj", + systems={system.identifier: system for system in systems}, + ) + ) + yield + Badger.current.bind(None) + + +@pytest.fixture(autouse=True) +def _check_log_errors(caplog): + yield + errors = [r.getMessage() for r in caplog.get_records("call") if r.levelno == logging.ERROR] + if errors: + pytest.fail(f"log errors during tests: {errors}") + + +@pytest.fixture +def heartbeat(): + with mock.patch("taskbadger.celery.heartbeat") as heartbeat: + yield heartbeat + + +@pytest.mark.usefixtures("_bind_settings_with_heartbeat") +def test_heartbeat_from_system_integration(celery_session_app, celery_session_worker, heartbeat): + @celery_session_app.task + def add_auto_heartbeat(a, b): + return a + b + + celery_session_worker.reload() + + with ( + mock.patch("taskbadger.celery.create_task_safe") as create, + mock.patch("taskbadger.celery.update_task_safe"), + mock.patch("taskbadger.sdk.get_task"), + ): + tb_task = task_for_test() + create.return_value = tb_task + assert add_auto_heartbeat.delay(2, 2).get(timeout=10, propagate=True) == 4 + _wait_for_call_count(heartbeat.stop, 1) + + create.assert_called_once_with( + "tests.test_celery_heartbeat.add_auto_heartbeat", + status=StatusEnum.PENDING, + queue="celery", + external_id=mock.ANY, + stale_timeout=10, + ) + heartbeat.start.assert_called_once_with(tb_task.id, 5) + heartbeat.stop.assert_called_once_with(tb_task.id) + + +@pytest.mark.usefixtures("_bind_settings") +def test_heartbeat_from_task_option(celery_session_app, celery_session_worker, heartbeat): + @celery_session_app.task(base=Task, taskbadger_heartbeat_interval=7) + def add_task_heartbeat(a, b): + return a + b + + celery_session_worker.reload() + + with ( + mock.patch("taskbadger.celery.create_task_safe") as create, + mock.patch("taskbadger.celery.update_task_safe"), + mock.patch("taskbadger.sdk.get_task"), + ): + tb_task = task_for_test() + create.return_value = tb_task + assert add_task_heartbeat.delay(2, 2).get(timeout=10, propagate=True) == 4 + _wait_for_call_count(heartbeat.stop, 1) + + assert create.call_args.kwargs["stale_timeout"] == 14 + heartbeat.start.assert_called_once_with(tb_task.id, 7) + heartbeat.stop.assert_called_once_with(tb_task.id) + + +@pytest.mark.usefixtures("_bind_settings_with_heartbeat") +def test_heartbeat_from_apply_async_overrides_system(celery_session_app, celery_session_worker, heartbeat): + # per-call options are only picked up by tasks using the `Task` base class + @celery_session_app.task(base=Task) + def add_call_heartbeat(a, b): + return a + b + + celery_session_worker.reload() + + with ( + mock.patch("taskbadger.celery.create_task_safe") as create, + mock.patch("taskbadger.celery.update_task_safe"), + mock.patch("taskbadger.sdk.get_task"), + ): + tb_task = task_for_test() + create.return_value = tb_task + result = add_call_heartbeat.apply_async((2, 2), taskbadger_heartbeat_interval=3, taskbadger_stale_timeout=30) + assert result.get(timeout=10, propagate=True) == 4 + _wait_for_call_count(heartbeat.stop, 1) + + assert create.call_args.kwargs["stale_timeout"] == 30 + heartbeat.start.assert_called_once_with(tb_task.id, 3) + + +@pytest.mark.usefixtures("_bind_settings_with_heartbeat") +def test_heartbeat_for_eager_task(celery_session_app, heartbeat): + """Eager tasks are created worker-side by ``_maybe_create_task``.""" + + @celery_session_app.task(base=Task) + def add_eager(a, b): + return a + b + + celery_session_app.conf.task_always_eager = True + try: + with ( + mock.patch("taskbadger.celery.create_task_safe") as create, + mock.patch("taskbadger.celery.update_task_safe"), + mock.patch("taskbadger.sdk.get_task"), + ): + tb_task = task_for_test() + create.return_value = tb_task + assert add_eager.delay(2, 2).get(timeout=10, propagate=True) == 4 + finally: + celery_session_app.conf.task_always_eager = False + + assert create.call_args.kwargs["stale_timeout"] == 10 + heartbeat.start.assert_called_once_with(tb_task.id, 5) + heartbeat.stop.assert_called_once_with(tb_task.id) + + +@pytest.mark.usefixtures("_bind_settings_with_heartbeat") +def test_heartbeat_for_eager_task_from_apply_async(celery_session_app, heartbeat): + """``before_task_publish`` doesn't run for eager tasks, so the per-call + options have to be read back out of the headers.""" + + @celery_session_app.task(base=Task) + def add_eager_options(a, b): + return a + b + + celery_session_app.conf.task_always_eager = True + try: + with ( + mock.patch("taskbadger.celery.create_task_safe") as create, + mock.patch("taskbadger.celery.update_task_safe"), + mock.patch("taskbadger.sdk.get_task"), + ): + tb_task = task_for_test() + create.return_value = tb_task + result = add_eager_options.apply_async((2, 2), taskbadger_heartbeat_interval=3) + assert result.get(timeout=10, propagate=True) == 4 + finally: + celery_session_app.conf.task_always_eager = False + + assert create.call_args.kwargs["stale_timeout"] == 6 + heartbeat.start.assert_called_once_with(tb_task.id, 3) + + +@pytest.mark.usefixtures("_bind_settings_with_heartbeat") +def test_eager_task_leaves_an_open_session_alone(celery_session_app, heartbeat): + """An eager task must not close a session opened by its caller.""" + + @celery_session_app.task(base=Task) + def add_in_session(a, b): + return a + b + + celery_session_app.conf.task_always_eager = True + try: + with ( + mock.patch("taskbadger.celery.create_task_safe", return_value=task_for_test()), + mock.patch("taskbadger.celery.update_task_safe", return_value=task_for_test()), + mock.patch("taskbadger.sdk.get_task"), + mock.patch("taskbadger.mug.AuthenticatedClient"), + ): + with taskbadger.Session(): + session = Badger.current.session() + assert add_in_session.delay(2, 2).get(timeout=10, propagate=True) == 4 + assert session.client is not None, "the caller's session was closed by the task" + assert session.client is None + finally: + celery_session_app.conf.task_always_eager = False + + +@pytest.mark.usefixtures("_bind_settings") +def test_no_heartbeat_by_default(celery_session_app, celery_session_worker, heartbeat): + @celery_session_app.task(base=Task) + def add_no_heartbeat(a, b): + return a + b + + celery_session_worker.reload() + + with ( + mock.patch("taskbadger.celery.create_task_safe") as create, + mock.patch("taskbadger.celery.update_task_safe") as update, + mock.patch("taskbadger.sdk.get_task"), + ): + create.return_value = task_for_test() + assert add_no_heartbeat.delay(2, 2).get(timeout=10, propagate=True) == 4 + _wait_for_call_count(update, 2) + + assert "stale_timeout" not in create.call_args.kwargs + heartbeat.start.assert_called_once_with(mock.ANY, None) diff --git a/tests/test_heartbeat.py b/tests/test_heartbeat.py new file mode 100644 index 0000000..c629ee2 --- /dev/null +++ b/tests/test_heartbeat.py @@ -0,0 +1,153 @@ +import logging +import time +from http import HTTPStatus +from unittest import mock + +import pytest + +from taskbadger import StatusEnum +from taskbadger._heartbeat import heartbeat +from taskbadger._integrations import resolve_heartbeat_options, validate_interval +from taskbadger.exceptions import ConfigurationError +from taskbadger.internal.types import Response +from taskbadger.mug import Badger +from tests.utils import task_for_test + +INTERVAL = 0.05 + + +def _wait_for_call_count(mock_obj, expected, timeout=5.0): + deadline = time.monotonic() + timeout + while mock_obj.call_count < expected and time.monotonic() < deadline: + time.sleep(0.01) + return mock_obj.call_count + + +@pytest.fixture +def update(): + with mock.patch("taskbadger._heartbeat.update_task_safe") as update: + update.return_value = None + yield update + + +@pytest.mark.usefixtures("_bind_settings") +def test_heartbeat_pings_until_stopped(update): + heartbeat.start("tb-1", INTERVAL) + assert _wait_for_call_count(update, 2) >= 2 + update.assert_called_with("tb-1") + + heartbeat.stop("tb-1") + time.sleep(INTERVAL * 3) + calls = update.call_count + time.sleep(INTERVAL * 3) + assert update.call_count == calls + + +@pytest.mark.usefixtures("_bind_settings") +def test_heartbeat_ping_is_an_empty_update(): + """The beat thread has its own context, so it has to build its own client.""" + with mock.patch("taskbadger.sdk.task_partial_update.sync_detailed") as request: + # a terminal status unregisters the task, so only one request is made + request.return_value = Response(HTTPStatus.OK, b"", {}, task_for_test(status=StatusEnum.SUCCESS)) + heartbeat.start("tb-9", INTERVAL) + assert _wait_for_call_count(request, 1) == 1 + heartbeat.stop("tb-9") + + assert request.call_args.kwargs["id"] == "tb-9" + assert request.call_args.kwargs["body"].to_dict() == {} + + +@pytest.mark.usefixtures("_bind_settings") +def test_heartbeat_stops_when_task_reaches_terminal_state(update): + update.return_value = task_for_test(status=StatusEnum.SUCCESS) + + heartbeat.start("tb-2", INTERVAL) + assert _wait_for_call_count(update, 1) == 1 + + time.sleep(INTERVAL * 3) + assert update.call_count == 1 + + +@pytest.mark.usefixtures("_bind_settings") +def test_heartbeat_binds_settings_in_the_beat_thread(update): + seen = [] + update.side_effect = lambda task_id: seen.append(Badger.current.settings) + + heartbeat.start("tb-3", INTERVAL) + _wait_for_call_count(update, 1) + heartbeat.stop("tb-3") + + assert seen[0] is Badger.current.settings + + +@pytest.mark.usefixtures("_bind_settings") +def test_heartbeat_pings_multiple_tasks(update): + heartbeat.start("tb-4", INTERVAL) + heartbeat.start("tb-5", INTERVAL) + _wait_for_call_count(update, 4) + heartbeat.stop("tb-4") + heartbeat.stop("tb-5") + + task_ids = {call.args[0] for call in update.call_args_list} + assert task_ids == {"tb-4", "tb-5"} + + +@pytest.mark.usefixtures("_bind_settings") +@pytest.mark.parametrize("interval", [None, 0, -1]) +def test_heartbeat_ignores_missing_interval(update, interval): + heartbeat.start("tb-6", interval) + time.sleep(INTERVAL * 3) + update.assert_not_called() + + +def test_heartbeat_ignores_unconfigured_client(update): + heartbeat.start("tb-7", INTERVAL) + time.sleep(INTERVAL * 3) + update.assert_not_called() + + +@pytest.mark.parametrize("interval", [0, -1, "30"]) +def test_validate_interval_rejects_bad_values(interval): + with pytest.raises(ConfigurationError): + validate_interval(interval) + + +class FakeSystem: + def __init__(self, heartbeat_interval=None, stale_timeout=None): + self.heartbeat_interval = heartbeat_interval + self.stale_timeout = stale_timeout + + +@pytest.mark.parametrize( + ("interval", "stale_timeout", "system", "expected"), + [ + (None, None, None, (None, None)), + (None, None, FakeSystem(), (None, None)), + # a stale timeout is derived from the interval + (30, None, None, (30, 60)), + (None, None, FakeSystem(heartbeat_interval=30), (30, 60)), + # an explicit stale timeout wins over the derived one + (30, 45, None, (30, 45)), + (30, None, FakeSystem(stale_timeout=45), (30, 45)), + # task options win over the system integration + (10, None, FakeSystem(heartbeat_interval=30), (10, 20)), + (None, 45, FakeSystem(heartbeat_interval=30, stale_timeout=90), (30, 45)), + # a stale timeout on its own doesn't imply a heartbeat + (None, 45, None, (None, 45)), + ], +) +def test_resolve_heartbeat_options(interval, stale_timeout, system, expected): + assert resolve_heartbeat_options(interval, stale_timeout, system) == expected + + +@pytest.mark.usefixtures("_bind_settings") +def test_heartbeat_ignores_invalid_interval(update): + heartbeat.start("tb-8", "30") + time.sleep(INTERVAL * 3) + update.assert_not_called() + + +def test_resolve_heartbeat_options_ignores_invalid_interval(caplog): + with caplog.at_level(logging.WARNING, logger="taskbadger"): + assert resolve_heartbeat_options("30", None, None) == (None, None) + assert "Ignoring invalid heartbeat_interval" in caplog.text diff --git a/tests/test_procrastinate_heartbeat.py b/tests/test_procrastinate_heartbeat.py new file mode 100644 index 0000000..76c021f --- /dev/null +++ b/tests/test_procrastinate_heartbeat.py @@ -0,0 +1,184 @@ +"""Tests for the automatic heartbeat that keeps running Procrastinate tasks fresh.""" + +import asyncio +from unittest import mock + +import procrastinate +import pytest +from procrastinate import testing + +from taskbadger import StatusEnum +from taskbadger.exceptions import ConfigurationError +from taskbadger.procrastinate import TB_TASK_ID_KWARG, _instrument_task, track +from taskbadger.systems.procrastinate import ProcrastinateSystemIntegration +from tests.utils import task_for_test + + +@pytest.fixture +def app(): + in_memory = testing.InMemoryConnector() + app = procrastinate.App(connector=in_memory) + with app.open(): + yield app + + +@pytest.fixture +def heartbeat(): + with mock.patch("taskbadger.procrastinate.heartbeat") as heartbeat: + yield heartbeat + + +@pytest.mark.usefixtures("_bind_settings") +def test_heartbeat_runs_for_the_duration_of_the_task(app, heartbeat): + @track(heartbeat_interval=30) + @app.task(name="slow") + def slow(): + heartbeat.start.assert_called_once_with("tb-1", 30) + heartbeat.stop.assert_not_called() + + with ( + mock.patch("taskbadger.procrastinate.update_task_safe"), + mock.patch("taskbadger.sdk.get_task", return_value=task_for_test(status=StatusEnum.PROCESSING)), + ): + slow.func(**{TB_TASK_ID_KWARG: "tb-1"}) + + heartbeat.stop.assert_called_once_with("tb-1") + + +@pytest.mark.usefixtures("_bind_settings") +def test_heartbeat_stops_when_the_task_fails(app, heartbeat): + @track(heartbeat_interval=30) + @app.task(name="slow_boom") + def slow_boom(): + raise ValueError("nope") + + with ( + mock.patch("taskbadger.procrastinate.update_task_safe"), + mock.patch("taskbadger.sdk.get_task", return_value=task_for_test(status=StatusEnum.PROCESSING)), + pytest.raises(ValueError, match="nope"), + ): + slow_boom.func(**{TB_TASK_ID_KWARG: "tb-2"}) + + heartbeat.start.assert_called_once_with("tb-2", 30) + heartbeat.stop.assert_called_once_with("tb-2") + + +@pytest.mark.usefixtures("_bind_settings") +def test_heartbeat_runs_for_async_tasks(app, heartbeat): + @track(heartbeat_interval=15) + @app.task(name="slow_async") + async def slow_async(): + return "done" + + with ( + mock.patch("taskbadger.procrastinate.update_task_safe"), + mock.patch("taskbadger.sdk.get_task", return_value=task_for_test(status=StatusEnum.PROCESSING)), + ): + assert asyncio.run(slow_async.func(**{TB_TASK_ID_KWARG: "tb-3"})) == "done" + + heartbeat.start.assert_called_once_with("tb-3", 15) + heartbeat.stop.assert_called_once_with("tb-3") + + +@pytest.mark.usefixtures("_bind_settings") +def test_no_heartbeat_by_default(app, heartbeat): + @app.task(name="plain") + def plain(): + return 1 + + _instrument_task(plain, system=None, manual=True) + + with ( + mock.patch("taskbadger.procrastinate.update_task_safe"), + mock.patch("taskbadger.sdk.get_task", return_value=task_for_test(status=StatusEnum.PROCESSING)), + ): + plain.func(**{TB_TASK_ID_KWARG: "tb-4"}) + + heartbeat.start.assert_called_once_with("tb-4", None) + + +@pytest.mark.usefixtures("_bind_settings") +def test_defer_derives_stale_timeout_from_the_interval(app): + @track(heartbeat_interval=30) + @app.task(name="deferred") + def deferred(a): + return a + + with ( + mock.patch("taskbadger.procrastinate.create_task_safe", return_value=task_for_test()) as create, + mock.patch("taskbadger.procrastinate.update_task_safe"), + ): + deferred.defer(a=1) + + assert create.call_args.kwargs["stale_timeout"] == 60 + + +@pytest.mark.usefixtures("_bind_settings") +def test_defer_uses_explicit_stale_timeout(app): + @track(heartbeat_interval=30, stale_timeout=120) + @app.task(name="deferred_explicit") + def deferred_explicit(a): + return a + + with ( + mock.patch("taskbadger.procrastinate.create_task_safe", return_value=task_for_test()) as create, + mock.patch("taskbadger.procrastinate.update_task_safe"), + ): + deferred_explicit.defer(a=1) + + assert create.call_args.kwargs["stale_timeout"] == 120 + + +@pytest.mark.usefixtures("_bind_settings") +def test_system_integration_heartbeat(app, heartbeat): + @app.task(name="auto_heartbeat") + def auto_heartbeat(a): + return a + + ProcrastinateSystemIntegration(app=app, heartbeat_interval=45) + + with ( + mock.patch("taskbadger.procrastinate.create_task_safe", return_value=task_for_test(id="tb-5")) as create, + mock.patch("taskbadger.procrastinate.update_task_safe"), + mock.patch("taskbadger.sdk.get_task", return_value=task_for_test(status=StatusEnum.PROCESSING)), + ): + auto_heartbeat.defer(a=1) + auto_heartbeat.func(a=1, **{TB_TASK_ID_KWARG: "tb-5"}) + + assert create.call_args.kwargs["stale_timeout"] == 90 + heartbeat.start.assert_called_once_with("tb-5", 45) + heartbeat.stop.assert_called_once_with("tb-5") + + +@pytest.mark.usefixtures("_bind_settings") +def test_system_integration_heartbeat_for_tracked_task(app, heartbeat): + """A ``@track``ed task still inherits the system's options when the + integration is constructed after the task was registered.""" + + @track() + @app.task(name="tracked_auto_heartbeat") + def tracked_auto_heartbeat(a): + return a + + ProcrastinateSystemIntegration(app=app, heartbeat_interval=45) + + with ( + mock.patch("taskbadger.procrastinate.create_task_safe", return_value=task_for_test(id="tb-6")) as create, + mock.patch("taskbadger.procrastinate.update_task_safe"), + mock.patch("taskbadger.sdk.get_task", return_value=task_for_test(status=StatusEnum.PROCESSING)), + ): + tracked_auto_heartbeat.defer(a=1) + tracked_auto_heartbeat.func(a=1, **{TB_TASK_ID_KWARG: "tb-6"}) + + assert create.call_args.kwargs["stale_timeout"] == 90 + heartbeat.start.assert_called_once_with("tb-6", 45) + + +def test_track_rejects_bad_interval(app): + with pytest.raises(ConfigurationError): + track(heartbeat_interval=0) + + +def test_system_integration_rejects_bad_interval(app): + with pytest.raises(ConfigurationError): + ProcrastinateSystemIntegration(app=app, heartbeat_interval=-5)