diff options
Diffstat (limited to 'packages')
8 files changed, 258 insertions, 23 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__ diff --git a/packages/meshbay-hub/src/meshbay_hub/api/revocation.py b/packages/meshbay-hub/src/meshbay_hub/api/revocation.py index 0e274f9..60b0c88 100644 --- a/packages/meshbay-hub/src/meshbay_hub/api/revocation.py +++ b/packages/meshbay-hub/src/meshbay_hub/api/revocation.py @@ -40,6 +40,7 @@ from sqlalchemy import select, update from sqlalchemy.ext.asyncio import AsyncSession import jwt +from meshbay_common.background import spawn from meshbay_hub.auth import hub_public_key_pem, decode_access_token from meshbay_hub.api.deps import get_current_user, require_admin from meshbay_hub.db.engine import get_db @@ -397,7 +398,7 @@ async def node_websocket(ws: WebSocket): if not _notify_budget(node_id): log.warning("Node %s exceeded its chat_notify rate", node_id[:8]) continue - asyncio.ensure_future(_handle_chat_notify( + spawn(_handle_chat_notify( msg.get("group_id", ""), msg.get("sender_name", ""), # The author, as the node authenticated them — not diff --git a/packages/meshbay-node/src/meshbay_node/daemon.py b/packages/meshbay-node/src/meshbay_node/daemon.py index 9428ddd..518f221 100644 --- a/packages/meshbay-node/src/meshbay_node/daemon.py +++ b/packages/meshbay-node/src/meshbay_node/daemon.py @@ -37,6 +37,7 @@ from pathlib import Path import uvicorn +from meshbay_common.background import spawn from meshbay_common.paths import fold from meshbay_common import MNP_VERSION from meshbay_common.protocol import MNP @@ -738,7 +739,7 @@ class NodeDaemon: try: loop.add_signal_handler( signal.SIGHUP, - lambda: asyncio.ensure_future(self._reload_config())) + lambda: spawn(self._reload_config())) except (NotImplementedError, AttributeError): pass # no SIGHUP on Windows; `reload` says so there await stop_event.wait() @@ -1257,7 +1258,7 @@ class NodeDaemon: def fire() -> None: self._pending_broadcasts.pop(group_id, None) - asyncio.ensure_future(self._broadcast_index_change(indexer)) + spawn(self._broadcast_index_change(indexer)) self._pending_broadcasts[group_id] = loop.call_later( self._broadcast_coalesce_secs, fire) @@ -1309,14 +1310,14 @@ class NodeDaemon: self._enriched_attempted.discard((group_id, entry.id)) seen = {e.id for e in new_entries} new_entries = new_entries + [e for e in rebuilt if e.id not in seen] - asyncio.ensure_future(self._enrich_new_video_entries(indexer, new_entries)) + spawn(self._enrich_new_video_entries(indexer, new_entries)) # Music app (docs/musicbay.md §6): same shape, gated on audio_root # exactly like video_root above (added later — musicbay.md's # original "no root, whole shared tree" call didn't hold up). - asyncio.ensure_future(self._enrich_new_audio_entries(indexer, new_entries)) + spawn(self._enrich_new_audio_entries(indexer, new_entries)) # Photos app (docs/photos.md §5): same shape, gated on photo_roots # (a list, not a single string — §2.1). - asyncio.ensure_future(self._enrich_new_photo_entries(indexer, new_entries)) + spawn(self._enrich_new_photo_entries(indexer, new_entries)) # A rename/move changes the very filename (or season folder) that # §3.3/§3.4's title-parse read display_title/season/episode from, @@ -1327,11 +1328,11 @@ class NodeDaemon: # (duration/thumb_hash/... landing via _on_enriched below) leaves # name/path alone and must not re-trigger itself forever. if delta is not None and delta.updates and previous is not None: - asyncio.ensure_future( + spawn( self._reenrich_renamed_video_entries(indexer, delta.updates, previous)) - asyncio.ensure_future( + spawn( self._reenrich_renamed_audio_entries(indexer, delta.updates, previous)) - asyncio.ensure_future( + spawn( self._reenrich_renamed_photo_entries(indexer, delta.updates, previous)) # Videos/Music/Photos apps: a file that leaves the index also loses @@ -1366,7 +1367,7 @@ class NodeDaemon: still_referenced = any( i.index.get_entry(file_id) is not None for i in self._indexers) if not still_referenced: - asyncio.ensure_future(self._media_cache.prune_file(file_id)) + spawn(self._media_cache.prune_file(file_id)) # 11.5 — Push to connected WebRTC peers in this group if self._webrtc: @@ -1404,7 +1405,7 @@ class NodeDaemon: else [e.id for e in idx.entries]) if hashes: endpoint = f"webrtc:{self._config.node.quic_port}" - asyncio.ensure_future(self._register_swarm(hashes, endpoint)) + spawn(self._register_swarm(hashes, endpoint)) async def _enrich_new_video_entries(self, indexer: DirectoryIndexer, entries: list) -> None: """ @@ -1666,7 +1667,7 @@ class NodeDaemon: return for session in list(self._webrtc._sessions.values()): if session._group_id == group_id: - asyncio.ensure_future(session.close()) + spawn(session.close()) log.info("Dropped session for revoked group %s", group_id[:8]) async def _register_swarm(self, hashes: list[str], endpoint: str) -> None: diff --git a/packages/meshbay-node/src/meshbay_node/indexer/indexer.py b/packages/meshbay-node/src/meshbay_node/indexer/indexer.py index 852c061..da8ea82 100644 --- a/packages/meshbay-node/src/meshbay_node/indexer/indexer.py +++ b/packages/meshbay-node/src/meshbay_node/indexer/indexer.py @@ -35,6 +35,7 @@ from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey from watchdog.events import FileSystemEvent, FileSystemEventHandler from watchdog.observers import Observer +from meshbay_common.background import spawn from meshbay_common.paths import fold, find_fold_collisions, long_path from meshbay_common.protocol import IndexEntry from meshbay_node.indexer.cache import IndexCache @@ -933,7 +934,7 @@ class DirectoryIndexer: def fire() -> None: self._pending_timers.pop(key, None) - asyncio.ensure_future(self._update_entry(file_path, deleted)) + spawn(self._update_entry(file_path, deleted)) self._pending_timers[key] = self._loop.call_later(self.debounce_secs, fire) diff --git a/packages/meshbay-node/src/meshbay_node/ops.py b/packages/meshbay-node/src/meshbay_node/ops.py index 4d3422d..91a922a 100644 --- a/packages/meshbay-node/src/meshbay_node/ops.py +++ b/packages/meshbay-node/src/meshbay_node/ops.py @@ -30,6 +30,7 @@ from dataclasses import asdict from pathlib import Path from typing import Any +from meshbay_common.background import spawn from meshbay_common.chatbox import new_epoch_key from meshbay_common.crypto import ( generate_gek, @@ -1673,7 +1674,7 @@ async def set_app_directories(state: dict, group_id: str, app_key: str, enrich = (state.get("enrich_app_dirs_fns") or {}).get(app_key) if enrich: - asyncio.ensure_future(enrich(group_id)) + spawn(enrich(group_id)) return {"app": app_key, "directories": clean, "group_id": group_id} @@ -1881,5 +1882,5 @@ async def start_reload(state: dict) -> dict: reload_fn = state.get("reload_fn") if not reload_fn: raise OpError("Reload not available", status=503) - asyncio.ensure_future(reload_fn()) + spawn(reload_fn()) return {"status": "reloading"} diff --git a/packages/meshbay-node/src/meshbay_node/ui/app.py b/packages/meshbay-node/src/meshbay_node/ui/app.py index 4ad3787..de2542e 100644 --- a/packages/meshbay-node/src/meshbay_node/ui/app.py +++ b/packages/meshbay-node/src/meshbay_node/ui/app.py @@ -12,12 +12,12 @@ no server-rendered UI: the Node page ships in the desktop client (see `docs/refactor-node-ui.md`). """ -import asyncio import logging from fastapi import FastAPI, HTTPException, Query from fastapi.responses import JSONResponse +from meshbay_common.background import spawn from meshbay_node import __version__, ops from meshbay_node.indexer.indexer import DirectoryIndexer @@ -179,7 +179,7 @@ def create_ui_app(state: dict) -> FastAPI: )) reload_fn = state.get("reload_fn") if reload_fn: - asyncio.ensure_future(reload_fn()) + spawn(reload_fn()) return result @app.post("/api/groups/detach") @@ -190,7 +190,7 @@ def create_ui_app(state: dict) -> FastAPI: )) reload_fn = state.get("reload_fn") if reload_fn: - asyncio.ensure_future(reload_fn()) + spawn(reload_fn()) return result @app.delete("/api/groups/{group_id}/files/{file_id}") @@ -373,7 +373,7 @@ def create_ui_app(state: dict) -> FastAPI: )) reload_fn = state.get("reload_fn") if reload_fn: - asyncio.ensure_future(reload_fn()) + spawn(reload_fn()) return result @app.patch("/api/groups/{group_id}/roots/{root_name}") @@ -385,7 +385,7 @@ def create_ui_app(state: dict) -> FastAPI: )) reload_fn = state.get("reload_fn") if reload_fn: - asyncio.ensure_future(reload_fn()) + spawn(reload_fn()) return result @app.put("/api/groups/{group_id}/roots/{root_name}/eject") @@ -401,11 +401,11 @@ def create_ui_app(state: dict) -> FastAPI: result = await _op(lambda: ops.remove_root(state, group_id, root_name)) reload_fn = state.get("reload_fn") if reload_fn: - asyncio.ensure_future(reload_fn()) + spawn(reload_fn()) return result - # asyncio.ensure_future above schedules the reload (and whatever initial - # scan it triggers) on the daemon's own event loop — it has no link to + # `spawn` above schedules the reload (and whatever initial scan it + # triggers) on the daemon's own event loop — it has no link to # this HTTP request or to any browser tab. Closing the client that made # this call does not cancel it: the scan is the node's own background # work, not something borrowed from the request that started it. diff --git a/packages/meshbay-node/tests/test_task_lifetime.py b/packages/meshbay-node/tests/test_task_lifetime.py index 9dffedb..133c030 100644 --- a/packages/meshbay-node/tests/test_task_lifetime.py +++ b/packages/meshbay-node/tests/test_task_lifetime.py @@ -14,6 +14,14 @@ first try included, until the daemon was restarted. It was two slots at the time, which is how few it took. Seen in the wild on 2026-08-16 after a viewer switched films mid-stream. + +**What stays here is the session's own contract**: a peer session holds what it +starts *and can cancel it* on the way out, which a module-level holder cannot — +a session ends, a process does not. The same rule for every other +fire-and-forget task, in every package, is +`meshbay-common/tests/test_background_tasks.py`. This file used to state it for +one file, and therefore did not state it for the nineteen sites outside that +file. """ import re |