summaryrefslogtreecommitdiffstats
path: root/packages/meshbay-node
diff options
context:
space:
mode:
authorChristophe Besson <cbesson@gmail.com>2026-09-13 15:40:50 +0200
committerChristophe Besson <cbesson@gmail.com>2026-09-13 15:40:50 +0200
commitf2d9a026db453899e5bb50f101b1f5f1a9f91ddf (patch)
treeed3a33a20f340f18f917a30f1d73877a14a1c3cf /packages/meshbay-node
parentd917bb61e42336c38782b22da604d7ca923d484a (diff)
downloadmeshbay-f2d9a026db453899e5bb50f101b1f5f1a9f91ddf.tar.gz
fix: hold every background task, in both codebases
asyncio keeps only a weak reference to a task, so a coroutine started with `asyncio.ensure_future(...)` whose result is discarded can be collected while it is still running: the loop logs "Task was destroyed but it is pending!" and the work simply does not happen. No error reaches the caller, and what is lost is whatever that coroutine was in the middle of. The node already had a guard for this, written after an abandoned stream task lost a transcode slot for good — and it read one file, `webrtc_server.py`, because that is where the defect was found. Outside that file there were nineteen sites: the hub's `chat_notify` (a notification for every member of a group), the indexer's debounce (every real-time index update), eleven in `daemon.py` including the SIGHUP reload and each enrichment pass, two in `ops.py`, and five in the loopback API. `meshbay_common.background.spawn()` is the one door. It holds the task, drops it when it finishes, and logs what it raised under the coroutine's own name — an exception in a task nobody awaits was otherwise reported by asyncio at collection time, out of context or not at all. A peer session's `_spawn` stays as it is: that one can also *cancel* what it holds, which a module-level holder cannot, because a session ends and a process does not. `test_background_tasks.py` walks every package's source and refuses a discarded handle. It parses rather than greps, so an assignment, a comprehension or an await is not mistaken for one, and it was checked against a deliberate reintroduction. 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. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01UMxEQadpzPkYLFf5CYKhpW
Diffstat (limited to 'packages/meshbay-node')
-rw-r--r--packages/meshbay-node/src/meshbay_node/daemon.py23
-rw-r--r--packages/meshbay-node/src/meshbay_node/indexer/indexer.py3
-rw-r--r--packages/meshbay-node/src/meshbay_node/ops.py5
-rw-r--r--packages/meshbay-node/src/meshbay_node/ui/app.py16
-rw-r--r--packages/meshbay-node/tests/test_task_lifetime.py8
5 files changed, 33 insertions, 22 deletions
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