diff options
Diffstat (limited to 'packages/meshbay-node/src/meshbay_node/transport')
| -rw-r--r-- | packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py | 64 |
1 files changed, 53 insertions, 11 deletions
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, |