diff options
Diffstat (limited to 'packages/meshbay-node')
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 |