diff options
Diffstat (limited to 'packages/meshbay-node/src/meshbay_node')
4 files changed, 68 insertions, 25 deletions
diff --git a/packages/meshbay-node/src/meshbay_node/daemon.py b/packages/meshbay-node/src/meshbay_node/daemon.py index 69c7e66..26f73fc 100644 --- a/packages/meshbay-node/src/meshbay_node/daemon.py +++ b/packages/meshbay-node/src/meshbay_node/daemon.py @@ -45,7 +45,7 @@ from meshbay_node.audit import RETENTION_DAYS as AUDIT_RETENTION_DAYS, AuditStor from meshbay_node.bundle_store import BundleStore from meshbay_node.chat.store import ChatStore from meshbay_node.config import Config, DEFAULT_CONFIG_PATH, load_config, write_example_config -from meshbay_node.roots import RootSet, RootError, entry_abs_path +from meshbay_node.roots import RootSet, RootError, entry_abs_path, off_disk from meshbay_node.hub_client import HubClient, HubConfig from meshbay_node.indexer import DirectoryIndexer, IndexCache, GroupIndex from meshbay_node.indexer.enrich import Enricher @@ -331,7 +331,7 @@ class NodeDaemon: log.error("Group %r: %s — skipping", group_cfg.name, e) continue - roots.refresh_availability() + await off_disk(roots, roots.refresh_availability) if not any(r.available for r in roots): # Not skipped for being empty: a group whose only drive is # unplugged still exists, and its index is frozen rather @@ -826,7 +826,7 @@ class NodeDaemon: # reports "nothing changed" for exactly that edit. if _root_shape(ctx["roots"]) == _root_shape(roots): continue - roots.refresh_availability() + await off_disk(roots, roots.refresh_availability) indexer = next((i for i in self._indexers if i.group_id == group_cfg.id), None) if indexer is None: @@ -869,7 +869,7 @@ class NodeDaemon: except RootError as e: log.error("New group %r: %s — skipping", group_cfg.name, e) continue - roots.refresh_availability() + await off_disk(roots, roots.refresh_availability) gek = None if sk_x_raw and pk_x_raw: diff --git a/packages/meshbay-node/src/meshbay_node/indexer/indexer.py b/packages/meshbay-node/src/meshbay_node/indexer/indexer.py index c1b151a..887d435 100644 --- a/packages/meshbay-node/src/meshbay_node/indexer/indexer.py +++ b/packages/meshbay-node/src/meshbay_node/indexer/indexer.py @@ -40,7 +40,7 @@ from meshbay_common.paths import fold, find_fold_collisions, long_path from meshbay_common.protocol import IndexEntry from meshbay_node.indexer.cache import IndexCache from meshbay_node.indexer.group_index import GroupIndex -from meshbay_node.roots import Root, RootSet +from meshbay_node.roots import Root, RootSet, off_disk log = logging.getLogger(__name__) @@ -387,7 +387,7 @@ class DirectoryIndexer: await self._initial_scan() async def _initial_scan(self) -> None: - self.roots.refresh_availability() + await off_disk(self.roots, self.roots.refresh_availability) total = 0 waiting = [r.name for r in self.roots if r.available] self._queue(waiting) @@ -695,7 +695,7 @@ class DirectoryIndexer: self._index.remove_entry(entry.id) self.roots = roots - roots.refresh_availability() + await off_disk(roots, roots.refresh_availability) added = [r for r in roots if r.folded not in old_names and r.available] self._index.roots = roots.describe() @@ -794,7 +794,7 @@ class DirectoryIndexer: return await self._reconcile() async def _reconcile(self) -> bool: - changed = self.roots.refresh_availability() + changed = await off_disk(self.roots, self.roots.refresh_availability) touched = False # Drained before the loop below, because persisting the flag is what @@ -1042,7 +1042,7 @@ class DirectoryIndexer: if root is None: return root.ejected = False - root.available = root.is_live() + root.available = await off_disk(self.roots, root.is_live) if not root.available: await self._finish_plug(None) return @@ -1071,7 +1071,8 @@ class DirectoryIndexer: # Ejected or removed again while it waited. `_rescan_root` # drops the entries before it walks, so going ahead would # empty a root that is not there to be read. - if self._holds(root) and not root.ejected and root.is_live(): + live = await off_disk(self.roots, root.is_live) + if self._holds(root) and not root.ejected and live: await self._rescan_root(root) finally: if waiting: @@ -1143,7 +1144,7 @@ class DirectoryIndexer: if root is None: return - if deleted and not root.is_live(): + if deleted and not await off_disk(self.roots, root.is_live): # The volume went away rather than the file. Freeze: mark the # root and touch nothing. Every other event for this root # will arrive here too and be dropped the same way, which is diff --git a/packages/meshbay-node/src/meshbay_node/ops.py b/packages/meshbay-node/src/meshbay_node/ops.py index 8cd893c..2221bea 100644 --- a/packages/meshbay-node/src/meshbay_node/ops.py +++ b/packages/meshbay-node/src/meshbay_node/ops.py @@ -39,7 +39,7 @@ from meshbay_common.crypto import ( ) from meshbay_node.config import DEFAULT_CONFIG_PATH from meshbay_common.join import ROLE_MEMBER, ROLE_OPERATOR -from meshbay_node.roots import RootError, RootSet +from meshbay_node.roots import RootError, RootSet, off_disk log = logging.getLogger(__name__) @@ -1130,7 +1130,7 @@ async def plug_root(state: dict, group_id: str, root_name: str) -> dict: if not root.ejected: return {"status": "already_plugged", "name": root_name, "group_id": group_id, "roots": roots.describe()} - if not root.is_live(): + if not await off_disk(roots, root.is_live): raise OpError( f"Directory not found: {root.path}. Is the device connected?", status=409) @@ -1145,7 +1145,7 @@ async def plug_root(state: dict, group_id: str, root_name: str) -> dict: if indexer: await indexer.plug_root(root_name) root.ejected = False - root.available = root.is_live() + root.available = await off_disk(roots, root.is_live) log.info("Root plugged: %s in group %s", root_name, group_id[:8]) return {"status": "plugged", "name": root_name, "group_id": group_id, diff --git a/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py b/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py index e748be9..92f2951 100644 --- a/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py +++ b/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py @@ -644,7 +644,7 @@ class WebRTCPeerSession: elif mtype == MNP.TRANSFER_CLOSE: self._do_transfer_close(msg) elif mtype == MNP.FILE_UPLOAD: - self._do_file_upload(msg) + self._spawn(self._do_file_upload(msg)) elif mtype == MNP.DIR_CREATE: self._do_dir_create(msg) elif mtype == MNP.DIR_DELETE: @@ -5205,7 +5205,43 @@ class WebRTCPeerSession: ctx["partial_uploads"] = store return store - def _do_file_upload(self, msg: dict) -> None: + @staticmethod + def _upload_lock(ctx: dict) -> asyncio.Lock: + """ + One lock per group, beside the state it protects. + + Not per session: `partial_uploads` lives in the group context so a + client that reconnects finds its upload where it left it, which means + two sessions of the same member share the position of one `.part` file. + A lock on the session would let them interleave — and the loop no longer + serializes them for free now that a chunk write is awaited. + """ + lock = ctx.get("upload_lock") + if lock is None: + lock = asyncio.Lock() + ctx["upload_lock"] = lock + return lock + + async def _do_file_upload(self, msg: dict) -> None: + """ + One chunk of an upload, in the order it arrived. + + The chunk ordering rule — `chunk_index != state.next_index` is refused — + used to hold for free: the handler was synchronous, so nothing could run + between the check and the `advance` that answers it. Awaiting the write + opens that gap, and two chunks of one upload racing through it is a + `.part` file with a hole in it or a chunk refused for arriving on time. + So the check, the write and the advance are one critical section again. + + The order is the arrival order: `_dispatch_message` runs per message as + it arrives and creates these tasks in that order, tasks start in + creation order, and this lock is the first thing each one waits on, so + its queue of waiters is in arrival order too. + """ + async with self._upload_lock(self._group_ctx()): + await self._upload_chunk(msg) + + async def _upload_chunk(self, msg: dict) -> None: """ One chunk of an upload, sealed under the group key (MNP 2.0). @@ -5386,8 +5422,8 @@ class WebRTCPeerSession: # client names *where among the group's own folders*, never a path on # the operator's filesystem. if target_rel: - target_dir = roots.resolve(target_rel) - if target_dir is None or not target_dir.is_dir(): + target_dir = await off_disk(roots, roots.resolve, target_rel) + if target_dir is None or not await off_disk(roots, target_dir.is_dir): _refuse("Not a directory in this group", "no_such_directory") return rel_dir = target_rel @@ -5396,7 +5432,7 @@ class WebRTCPeerSession: # one destination is. target_dir = upload_root.path rel_dir = upload_root.name - if not target_dir.is_dir(): + if not await off_disk(roots, target_dir.is_dir): _refuse("That directory is currently unavailable", "root_unavailable") return @@ -5419,7 +5455,8 @@ class WebRTCPeerSession: # A shared directory means two people can send the same name. Refusing the # second is safe but silly — everyone's camera produces IMG_1234.jpg — so # a free name is found instead. Never a replacement. - stored_name = state.stored_name if state else _free_name(target_dir, filename) + stored_name = (state.stored_name if state + else await off_disk(roots, _free_name, target_dir, filename)) tmp_path = target_dir / f"{stored_name}{uploads_mod.PART_SUFFIX}" final_path = target_dir / stored_name @@ -5449,7 +5486,7 @@ class WebRTCPeerSession: if chunk_index == 0: # Backstop: _free_name already guarantees this, and it stays because # it asserts the invariant where the write happens. - if final_path.exists(): + if await off_disk(roots, final_path.exists): _refuse("File already exists", "already_exists") return state = uploads.start(user_id, rel_dir, filename, stored_name, @@ -5466,12 +5503,11 @@ class WebRTCPeerSession: if state.bytes + len(chunk_bytes) > MAX_UPLOAD_BYTES: uploads.drop(user_id, rel_dir, filename) - tmp_path.unlink(missing_ok=True) + await off_disk(roots, tmp_path.unlink, True) _refuse("Upload exceeds size limit", "too_large") return - with open(tmp_path, "wb" if chunk_index == 0 else "ab") as f: - f.write(chunk_bytes) + await off_disk(roots, _append_chunk, tmp_path, chunk_bytes, chunk_index == 0) uploads.advance(user_id, rel_dir, filename, chunk_index, len(chunk_bytes)) self._send(file_upload_ack_wire( @@ -5487,7 +5523,7 @@ class WebRTCPeerSession: if chunk_index + 1 >= total_chunks: uploads.drop(user_id, rel_dir, filename) - tmp_path.rename(final_path) + await off_disk(roots, tmp_path.rename, final_path) log.info("Upload complete: %s (%d chunks, %d bytes)", stored_name, total_chunks, state.bytes) self._audit("file_upload", f"{rel_dir}/{stored_name}") @@ -6589,6 +6625,12 @@ def _locate(roots: RootSet, entry) -> tuple[Path | None, str | None]: return path, None +def _append_chunk(tmp_path: Path, chunk_bytes: bytes, first: bool) -> None: + """Add one chunk to a partial upload. Blocking; called through `off_disk`.""" + with open(tmp_path, "wb" if first else "ab") as f: + f.write(chunk_bytes) + + def _read_and_encrypt( gek: bytes, file_path: Path, |