diff options
Diffstat (limited to 'packages/meshbay-common')
| -rw-r--r-- | packages/meshbay-common/src/meshbay_common/background.py | 70 | ||||
| -rw-r--r-- | packages/meshbay-common/tests/test_background_tasks.py | 153 |
2 files changed, 223 insertions, 0 deletions
diff --git a/packages/meshbay-common/src/meshbay_common/background.py b/packages/meshbay-common/src/meshbay_common/background.py new file mode 100644 index 0000000..1f25dfd --- /dev/null +++ b/packages/meshbay-common/src/meshbay_common/background.py @@ -0,0 +1,70 @@ +""" +Background work that outlives the call which started it. + +**asyncio keeps only a weak reference to a task.** A coroutine started with +`asyncio.ensure_future(...)` whose result is thrown away can therefore be +collected while it is still running: the loop logs "Task was destroyed but it is +pending!" and the work simply does not happen. There is no error at the caller, +nothing fails, and what was lost is whatever that coroutine was in the middle of +— a notification never written, a file never indexed, a reload half applied. + +`spawn()` is the one door for that shape. The reference in `_running` is what +keeps the task alive; the done callback is what stops the set growing. + +Two things it deliberately does **not** replace: + +- A task whose handle the caller keeps — the daemon's service tasks, the + indexer's reconcile loop — is already held, and being held is the whole point. +- A task scoped to something that can end, where the owner must be able to + **cancel** it: a peer session's spawns die with the session + (`WebRTCPeerSession._spawn`), and an enrichment run is tracked by its own + module. Those keep their own sets, because a module-level one can only ever + grow and forget, never shut down. + +It also logs what a discarded task swallowed. An exception in a task nobody +awaits is reported by asyncio at collection time, out of context and often not +at all; here it is logged where it happened, naming the work. +""" + +from __future__ import annotations + +import asyncio +import logging +from typing import Any, Coroutine + +log = logging.getLogger(__name__) + +# Strong references to everything in flight. Module-level on purpose: these are +# tasks with no owner to hold them, which is exactly why they need this. +_running: set[asyncio.Task] = set() + + +def spawn(coro: Coroutine[Any, Any, Any], *, what: str = "") -> asyncio.Task: + """Run a coroutine in the background and hold on to it until it finishes. + + The work names itself in the log: `what` defaults to the coroutine's own + qualified name, so no call site has to carry a label for the sake of a line + nobody reads until something breaks. Pass one only where the coroutine's + name is not the useful answer. + """ + if not what: + what = getattr(coro, "__qualname__", "") or repr(coro) + task = asyncio.ensure_future(coro) + _running.add(task) + task.add_done_callback(lambda t: _finished(t, what)) + return task + + +def _finished(task: asyncio.Task, what: str) -> None: + _running.discard(task) + if task.cancelled(): + return + exc = task.exception() + if exc is not None: + log.error("Background task %s failed: %s", what or "(unnamed)", exc, + exc_info=exc) + + +def pending() -> int: + """How many background tasks are in flight. For diagnostics and tests.""" + return len(_running) diff --git a/packages/meshbay-common/tests/test_background_tasks.py b/packages/meshbay-common/tests/test_background_tasks.py new file mode 100644 index 0000000..4f0ac59 --- /dev/null +++ b/packages/meshbay-common/tests/test_background_tasks.py @@ -0,0 +1,153 @@ +""" +Tasks nobody holds, across every package. + +asyncio keeps only a **weak** reference to a task, so a coroutine started with +`asyncio.ensure_future(...)` whose result is thrown away can be collected while +it is still running. Nothing raises: the loop logs "Task was destroyed but it is +pending!" and whatever that task was doing simply stops. + +The node already had a guard for this — written after an abandoned stream task +never reached `async with sem.__aexit__` and lost a transcode slot for good — +and it read one file, `webrtc_server.py`, because that is where the defect was +found. Meanwhile the hub answered `chat_notify` with a discarded +`ensure_future` (a notification for every member of a group, dropped whenever +the collector got there first), the node's indexer scheduled every real-time +index update the same way (a file copied in that never appears until a +reconciliation sweep an unbounded time later), and seventeen more sites did it +in `daemon.py`, `ops.py` and the loopback API — a reload, an enrichment pass, a +swarm registration. + +So this one walks **every package's source**. A guard that stops at the edge of +the file where the bug was found is a guard against that bug, not against its +class. +""" + +import ast +import asyncio +import logging +from pathlib import Path + +import pytest + +from meshbay_common import background + +REPO = Path(__file__).resolve().parents[3] +PACKAGES = ("meshbay-common", "meshbay-hub", "meshbay-node") + +SPAWNERS = {"ensure_future", "create_task"} + +# `spawn()` is the implementation of the rule and is the one place allowed to +# call the primitive with nothing holding the result — that is what it is for. +ALLOWED = {Path("meshbay-common/src/meshbay_common/background.py")} + + +def _sources() -> list[Path]: + out: list[Path] = [] + for pkg in PACKAGES: + src = REPO / "packages" / pkg / "src" + if src.is_dir(): + out.extend(sorted(src.rglob("*.py"))) + return out + + +def test_there_are_sources_to_check(): + """A rule applied to nothing passes forever.""" + files = _sources() + assert len(files) > 20, f"only found {len(files)} source files — wrong root?" + + +def _discarded(tree: ast.AST) -> list[int]: + """Lines where a task is started and its handle dropped on the floor. + + An `Expr` statement is a call whose value goes nowhere; a `Lambda` whose + whole body is such a call is the same thing wearing a callback's clothes + (`signal.add_signal_handler(SIGHUP, lambda: ensure_future(reload()))`). + Anything assigned, returned, awaited or collected into a list is held by + its caller and is not this defect. + """ + found: list[int] = [] + + def is_spawn(node: ast.AST) -> bool: + return (isinstance(node, ast.Call) + and isinstance(node.func, ast.Attribute) + and node.func.attr in SPAWNERS + and isinstance(node.func.value, ast.Name) + and node.func.value.id == "asyncio") + + for node in ast.walk(tree): + if isinstance(node, ast.Expr) and is_spawn(node.value): + found.append(node.lineno) + elif isinstance(node, ast.Lambda) and is_spawn(node.body): + found.append(node.lineno) + return found + + +def test_no_task_is_started_and_forgotten(): + offenders: list[str] = [] + for path in _sources(): + rel = path.relative_to(REPO / "packages") + if rel in ALLOWED: + continue + tree = ast.parse(path.read_text(encoding="utf-8"), filename=str(path)) + for line in _discarded(tree): + offenders.append(f"{rel}:{line}") + + assert not offenders, ( + "a task whose result is discarded can be collected mid-flight and the " + "work silently does not happen. Use meshbay_common.background.spawn(), " + "or keep the handle where the owner can also cancel it:\n " + + "\n ".join(offenders)) + + +# ── The helper itself ──────────────────────────────────────────────────────── + +@pytest.mark.asyncio +async def test_spawn_holds_the_task_until_it_finishes(): + started = asyncio.Event() + release = asyncio.Event() + + async def work(): + started.set() + await release.wait() + + task = background.spawn(work()) + await started.wait() + assert background.pending() == 1, "nothing is holding the task" + # The only strong reference the test itself holds, gone: if `spawn` is not + # keeping one, the task is now collectable. + del task + release.set() + await asyncio.sleep(0) + await asyncio.sleep(0) + assert background.pending() == 0, "the set grows for the life of the process" + + +@pytest.mark.asyncio +async def test_a_failure_is_logged_where_it_happened(caplog): + """Nobody awaits this task, so the log is the only place it can surface.""" + async def work(): + raise RuntimeError("the background work failed") + + with caplog.at_level(logging.ERROR, logger=background.log.name): + # Pinned rather than assumed: an earlier test in the run may have + # raised this logger's level or disabled it outright. + background.log.disabled = False + background.log.setLevel(logging.NOTSET) + task = background.spawn(work(), what="the work") + with pytest.raises(RuntimeError): + await task + + assert any("the work" in r.getMessage() for r in caplog.records), caplog.text + + +@pytest.mark.asyncio +async def test_the_work_names_itself_without_a_label(): + async def a_named_coroutine(): + raise RuntimeError("boom") + + task = background.spawn(a_named_coroutine()) + with pytest.raises(RuntimeError): + await task + # The default label is the coroutine's own name, so no call site has to + # carry one for a log line to be readable. + assert "a_named_coroutine" in a_named_coroutine.__qualname__ |