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 | 55 |
1 files changed, 41 insertions, 14 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 ce599cc..6bae27c 100644 --- a/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py +++ b/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py @@ -129,6 +129,7 @@ from meshbay_node.indexer import GroupIndex from meshbay_node.indexer.indexer import DirectoryIndexer from meshbay_node import linkpreview, ops, platform from meshbay_node import transfers as transfers_mod +from meshbay_node import uploads as uploads_mod from meshbay_node.transfers import TransferSlots # Re-imported under its original name: every call site and existing test in # this module still refers to it as `_probe_video`. The implementation lives @@ -426,7 +427,8 @@ class WebRTCPeerSession: self._join_attempts = 0 self._nonce_client: bytes = b"" self._admin_ops: dict[str, dict] = {} # op_id → pending admin operation - self._uploads: dict[str, dict] = {} # filename → {next_index, bytes} + # Uploads in progress live in the group context, not here: see + # `_partial_uploads` and `uploads.py`. # Diagnostics only (_WEBRTC_TRACE): when the last DataChannel message # arrived, so the heartbeat can report silence duration. self._last_msg_at: float = 0.0 @@ -4757,6 +4759,19 @@ class WebRTCPeerSession: if k not in ("type", "v")}) self._send(resp) + def _partial_uploads(self, ctx: dict) -> uploads_mod.PartialUploads: + """This group's uploads in progress, created on first use. + + In the group context rather than on the session, so a client that + reconnects finds its own upload where it left it — and so the reaper has + something to ask "is anyone still writing this?". + """ + store = ctx.get("partial_uploads") + if store is None: + store = uploads_mod.PartialUploads() + ctx["partial_uploads"] = store + return store + def _do_file_upload(self, msg: dict) -> None: """ One chunk of an upload, sealed under the group key (MNP 2.0). @@ -4912,13 +4927,26 @@ class WebRTCPeerSession: "root_unavailable") return - upload_key = f"{rel_dir}/{filename}" - state = self._uploads.get(upload_key) + # Held by the group, not by this connection. + # + # This used to be `self._uploads`, on the session. A dropped link threw + # the position away and the next chunk was refused with `not_started`: + # an upload interrupted at 99% could only be started again from zero, on + # a connection flaky enough to have interrupted it once. And the state + # it lost was the only thing that knew about the `.part` file left + # behind — see `uploads.orphaned_parts`, which is the other half of this. + # + # Keyed by member as well as by name, because a shared directory means + # two people can be sending IMG_1234.jpg at the same moment and neither + # may inherit the other's position. + uploads = self._partial_uploads(ctx) + user_id = self._user_id or "" + state = uploads.get(user_id, rel_dir, filename) # 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) - tmp_path = target_dir / f"{stored_name}.part" + stored_name = state.stored_name if state else _free_name(target_dir, filename) + tmp_path = target_dir / f"{stored_name}{uploads_mod.PART_SUFFIX}" final_path = target_dir / stored_name if chunk_index == 0: @@ -4927,28 +4955,27 @@ class WebRTCPeerSession: if final_path.exists(): _refuse("File already exists", "already_exists") return - state = {"next_index": 0, "bytes": 0, "stored_name": stored_name} - self._uploads[upload_key] = state + state = uploads.start(user_id, rel_dir, filename, stored_name, + part_path=tmp_path) elif state is None: _refuse("Upload not started", "not_started") return # Reject out-of-order or replayed chunks — otherwise chunk_index>0 appends # blindly to whatever .part file is already on disk. - if chunk_index != state["next_index"]: + if chunk_index != state.next_index: _refuse("Unexpected chunk index", "bad_chunk_index") return - if state["bytes"] + len(chunk_bytes) > MAX_UPLOAD_BYTES: - self._uploads.pop(upload_key, None) + if state.bytes + len(chunk_bytes) > MAX_UPLOAD_BYTES: + uploads.drop(user_id, rel_dir, filename) tmp_path.unlink(missing_ok=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) - state["next_index"] = chunk_index + 1 - state["bytes"] += len(chunk_bytes) + uploads.advance(user_id, rel_dir, filename, chunk_index, len(chunk_bytes)) self._send(file_upload_ack_wire( gek, self._group_id or "", @@ -4962,10 +4989,10 @@ class WebRTCPeerSession: )) if chunk_index + 1 >= total_chunks: - self._uploads.pop(upload_key, None) + uploads.drop(user_id, rel_dir, filename) tmp_path.rename(final_path) log.info("Upload complete: %s (%d chunks, %d bytes)", - stored_name, total_chunks, state["bytes"]) + stored_name, total_chunks, state.bytes) self._audit("file_upload", f"{rel_dir}/{stored_name}") self._register_uploader(ctx, rel_dir, stored_name) |