diff options
Diffstat (limited to 'packages/meshbay-node')
| -rw-r--r-- | packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py | 48 | ||||
| -rw-r--r-- | packages/meshbay-node/tests/test_partial_uploads.py | 129 |
2 files changed, 177 insertions, 0 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 6bae27c..9de799c 100644 --- a/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py +++ b/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py @@ -120,6 +120,7 @@ from meshbay_common.protocol import ( MNP, chunk_ciphertext, file_chunk_wire, + UPLOAD_PROBE_INDEX, file_upload_ack_wire, file_upload_payload, ) @@ -4792,6 +4793,30 @@ class WebRTCPeerSession: ctx = self._group_ctx() upload_id = str(msg.get("upload_id") or "")[:64] + # Say the slot is being used, chunk by chunk, exactly as `_do_file_req` + # does for a download. + # + # A grant nobody takes up is reclaimed after GRANT_DEADLINE_SECS and, on + # the third miss, abandoned. Uploads were not gated by the lease, so the + # file still arrived — but the widget follows the lease, so a 3.5 GB + # upload showed "waiting, 0 ahead" for a minute and a half while it was + # in fact transferring, and the node logged three reclaims against a + # transfer that never stopped. Measured, from the journal: + # + # 11:52:49 open upload 919ebf54 -> granted + # 11:53:19 reclaimed 919ebf54 (not_taken_up) + # 11:54:19 reclaimed 919ebf54 (abandoned) + # 11:55:48 Upload complete: ... (3 522 297 517 bytes) + # + # The download twin of this was fixed on 2026-09-08 (§12.1 of + # ~/next/improve-downloads.md); the same omission was still here, + # invisible until uploads started taking a real lease. + tr = msg.get("tr") + if tr: + slots = self._ctx.get("_transfer_slots") + if slots is not None: + slots.touch(str(tr)[:64]) + gek = ctx.get("gek") if not gek: self._send({"type": "error", "detail": "Group encryption not initialized", @@ -4949,6 +4974,29 @@ class WebRTCPeerSession: tmp_path = target_dir / f"{stored_name}{uploads_mod.PART_SUFFIX}" final_path = target_dir / stored_name + if chunk_index == UPLOAD_PROBE_INDEX: + # "Where am I?", asked inside the seal rather than on a clear + # message, because the answer is about a file whose name is exactly + # what sealing this path was for. + # + # It writes nothing, creates no state and reserves no name: a client + # that asks and then goes away has cost this node one reply. Every + # check above has already run, so it cannot be used to ask questions + # about a directory the caller may not write to. + self._send(file_upload_ack_wire( + gek, self._group_id or "", + upload_id=upload_id, + chunk_index=UPLOAD_PROBE_INDEX, + filename=filename, + # Only what is really on disk. Without state, `_free_name` above + # picked a name nothing has claimed yet, and reporting it would + # promise a destination the real chunk 0 may not choose. + stored_as=state.stored_name if state else "", + dir=rel_dir, + resume_from=state.next_index if state else 0, + )) + return + if chunk_index == 0: # Backstop: _free_name already guarantees this, and it stays because # it asserts the invariant where the write happens. diff --git a/packages/meshbay-node/tests/test_partial_uploads.py b/packages/meshbay-node/tests/test_partial_uploads.py index ea5637c..f5b6602 100644 --- a/packages/meshbay-node/tests/test_partial_uploads.py +++ b/packages/meshbay-node/tests/test_partial_uploads.py @@ -28,6 +28,10 @@ from meshbay_node.indexer.group_index import GroupIndex from meshbay_node.roots import Root, RootSet from meshbay_node.transport.webrtc_server import WebRTCPeerSession +from meshbay_common.protocol import ( + UPLOAD_PROBE_INDEX, file_upload_ack_payload, +) + from conftest import one_root, sealed_upload from meshbay_node.uploads import ( @@ -358,3 +362,128 @@ def test_an_upload_in_flight_is_known_to_the_reaper(tmp_path): assert len(live) == 1 assert next(iter(live)).name == "film.mkv.part" assert next(iter(live)).exists() + + +# ── asking where to resume ────────────────────────────────────────────────── + + +def _acks(session, ctx): + return [file_upload_ack_payload(ctx["gek"], GROUP, m) + for m in session.sent if m.get("type") == "file_upload_ack"] + + +def _probe(session, filename: str) -> dict: + """The question, asked exactly as the client asks it: an ordinary sealed + upload chunk with no bytes and the probe index.""" + return sealed_upload(session, filename=filename, data=b"", + chunk_index=UPLOAD_PROBE_INDEX, total_chunks=1) + + +def test_a_probe_for_an_unknown_file_says_start_at_the_beginning(tmp_path): + ctx = _group_ctx(tmp_path) + peer = _peer(ctx) + peer._do_file_upload(_probe(peer, "film.mkv")) + assert _errors(peer) == [] + assert _acks(peer, ctx)[0]["resume_from"] == 0 + + +def test_a_probe_reports_what_the_node_already_holds(tmp_path): + """The point of the whole stage: the client learns it has 2 chunks there and + sends the third, instead of sending a film again.""" + ctx = _group_ctx(tmp_path) + first = _peer(ctx) + for i in range(2): + first._do_file_upload(sealed_upload(first, filename="film.mkv", + data=b"xxxx", chunk_index=i, + total_chunks=5)) + assert _errors(first) == [] + + reconnected = _peer(ctx) + reconnected._do_file_upload(_probe(reconnected, "film.mkv")) + ack = _acks(reconnected, ctx)[0] + assert ack["resume_from"] == 2 + assert ack["stored_as"] == "film.mkv" + + +def test_a_probe_writes_nothing_and_reserves_nothing(tmp_path): + """It has to be free of consequence: a client that asks and goes away must + leave no file, no state and no name taken.""" + ctx = _group_ctx(tmp_path) + peer = _peer(ctx) + peer._do_file_upload(_probe(peer, "film.mkv")) + root = ctx["roots"].roots[0] + assert list(root.path.iterdir()) == [] + assert len(ctx.get("partial_uploads") or []) == 0 + # And it promises no destination it has not taken. + assert _acks(peer, ctx)[0]["stored_as"] == "" + + +def test_a_probe_answers_only_about_the_member_who_asks(tmp_path): + """Same keying as the upload itself. Otherwise one member could measure + another's progress on a file they never sent — and worse, resume it.""" + ctx = _group_ctx(tmp_path) + alice = _peer(ctx, "alice") + alice._do_file_upload(sealed_upload(alice, filename="film.mkv", + data=b"xxxx", chunk_index=0, + total_chunks=5)) + bob = _peer(ctx, "bob") + bob._do_file_upload(_probe(bob, "film.mkv")) + assert _acks(bob, ctx)[0]["resume_from"] == 0 + + +def test_an_ordinary_ack_carries_no_resume_field(tmp_path): + """So a client can tell a probe's answer from a chunk's without looking at + the index it echoed.""" + ctx = _group_ctx(tmp_path) + peer = _peer(ctx) + peer._do_file_upload(sealed_upload(peer, filename="a.bin", data=b"x", + chunk_index=0, total_chunks=2)) + assert "resume_from" not in _acks(peer, ctx)[0] + + +def test_a_probe_is_refused_where_an_upload_would_be(tmp_path): + """Every check the write path makes has already run when the probe is + answered, so it cannot be used to ask questions about somewhere the caller + may not write.""" + ctx = _group_ctx(tmp_path) + peer = _peer(ctx) + peer._do_file_upload(sealed_upload(peer, filename="../escape", + data=b"", chunk_index=UPLOAD_PROBE_INDEX, + total_chunks=1)) + assert [m.get("code") for m in _errors(peer)] == ["invalid_filename"] + assert _acks(peer, ctx) == [] + + +# ── the slot an upload holds ──────────────────────────────────────────────── + +def test_an_upload_chunk_says_its_slot_is_in_use(tmp_path): + """A grant nobody takes up is reclaimed after thirty seconds and abandoned + on the third miss. Uploads are not gated by the lease, so the file arrived + anyway — but the widget follows the lease, and a 3.5 GB upload therefore + read "waiting, 0 ahead" for a minute and a half while it was transferring, + with three reclaims logged against it. + + The download twin of this was fixed a day earlier; the same omission was + still here, invisible until uploads took a real lease. + """ + from meshbay_node.transfers import TransferSlots, UPLOAD + + ctx = _group_ctx(tmp_path) + peer = _peer(ctx) + slots = TransferSlots() + peer._ctx = dict(ctx) + peer._ctx["_transfer_slots"] = slots + peer._registry_key = "session-1" + lease, err = slots.open(tr="up-1", kind=UPLOAD, session_key="session-1", + user_id="user-1", group_id=GROUP, bytes=10, chunks=2) + assert not err and lease.state == "granted" + assert lease.used is False + + msg = sealed_upload(peer, filename="film.mkv", data=b"xxxx", + chunk_index=0, total_chunks=2) + msg["tr"] = "up-1" + peer._do_file_upload(msg) + + assert _errors(peer) == [] + assert slots.leases["up-1"].used is True, ( + "the node still believes nobody took this slot up, and will reclaim it") |