From d62b6a4e8985d0504e1825f6f8f663ccd64489ae Mon Sep 17 00:00:00 2001 From: Christophe Besson Date: Wed, 9 Sep 2026 11:26:16 +0200 Subject: feat(node): uploads outlive their connection, and their leftovers are reaped Stage 8 of ~/next/improve-downloads.md, first half. Two defects that are the same defect seen from two sides. An upload's progress lived on the session, keyed by `rel_dir/filename`. A dropped connection threw it away and the client's next chunk was refused with `not_started`: an upload interrupted at 99% could only be started again from zero, on a link flaky enough to have interrupted it once. It now lives in the group context, keyed by member as well -- a shared directory means two people can be sending IMG_1234.jpg at the same moment and neither may inherit, or overwrite the position of, the other's. What the lost state left behind was a `.part` nothing would ever finish, delete or look at again. It is not an index entry, so it is invisible to every member and to the operator's own file list: one abandoned film is a gigabyte of their disk, kept for ever. That leak predates this branch. A `.part` is deleted only when **both** hold: no upload is writing it, and nothing has been written to it for 24 hours. Waiting costs disk; being wrong costs somebody their upload, and is not reversible -- so a read-only root is never walked (it cannot have received an upload), an unavailable one is never walked (an unmounted drive reporting "nothing found" is how a careless janitor deletes a library), and a file whose mtime is in the future is left alone (a clock that went backwards is not evidence). The reaper matches whole paths and the state records the path it is writing, rather than both sides rebuilding one from a root name -- two implementations of one rule whose failure mode is deleting a live upload. The rules are in `uploads.py`, pure logic with no asyncio and no transport, the same shape as `transfers.py` and for the same reason. 23 cases, four of them checked against the unfixed source. Node suite 1195 passed. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_01HCGdheDLxGReuKHga3BtST --- .../meshbay-node/tests/test_partial_uploads.py | 360 +++++++++++++++++++++ 1 file changed, 360 insertions(+) create mode 100644 packages/meshbay-node/tests/test_partial_uploads.py (limited to 'packages/meshbay-node/tests/test_partial_uploads.py') diff --git a/packages/meshbay-node/tests/test_partial_uploads.py b/packages/meshbay-node/tests/test_partial_uploads.py new file mode 100644 index 0000000..ea5637c --- /dev/null +++ b/packages/meshbay-node/tests/test_partial_uploads.py @@ -0,0 +1,360 @@ +""" +Two rules about an upload that stopped in the middle. + +**It belongs to the group, not to the connection.** Progress used to be kept on +the session, so a dropped connection lost it and the client's next chunk was +refused with `not_started` — an upload interrupted at 99% could only start again +from zero, on a link flaky enough to have interrupted it once. + +**And what it leaves on disk has an owner or it has an end.** The state that was +lost left a `.part` file nothing would ever finish, delete or look at again: +invisible in the index, because `.part` is not an index entry, and a gigabyte of +somebody else's disk for one abandoned film. + +The keying is a correctness property rather than a nicety: a shared directory +means two members can be sending `IMG_1234.jpg` at the same moment, and neither +may inherit — or overwrite the position of — the other's. +""" + +import os +import time +import types +from pathlib import Path + +from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey +from meshbay_common.crypto import generate_gek +from meshbay_node.daemon import NodeDaemon +from meshbay_node.indexer.group_index import GroupIndex +from meshbay_node.roots import Root, RootSet +from meshbay_node.transport.webrtc_server import WebRTCPeerSession + +from conftest import one_root, sealed_upload + +from meshbay_node.uploads import ( + ORPHAN_AFTER_SECS, PART_SUFFIX, PartialUploads, find_parts, orphaned_parts, +) + + +# ── the state ─────────────────────────────────────────────────────────────── + +def test_an_upload_is_found_again_after_the_connection_went_away(): + """The whole point: the store outlives the session, so the position is + still there when the client comes back.""" + uploads = PartialUploads() + uploads.start("alice", "media", "film.mkv", "film.mkv") + uploads.advance("alice", "media", "film.mkv", chunk_index=0, nbytes=1024) + uploads.advance("alice", "media", "film.mkv", chunk_index=1, nbytes=1024) + + state = uploads.get("alice", "media", "film.mkv") + assert state is not None + assert state.next_index == 2 + assert state.bytes == 2048 + + +def test_two_members_uploading_the_same_name_do_not_share_a_position(): + """A shared folder makes this ordinary, not adversarial: everyone's camera + produces the same filenames. Inheriting the other's position would append + one person's chunks to another person's file.""" + uploads = PartialUploads() + uploads.start("alice", "photos", "IMG_1234.jpg", "IMG_1234.jpg") + uploads.start("bob", "photos", "IMG_1234.jpg", "IMG_1234 (2).jpg") + uploads.advance("alice", "photos", "IMG_1234.jpg", 0, 10) + + assert uploads.get("alice", "photos", "IMG_1234.jpg").next_index == 1 + assert uploads.get("bob", "photos", "IMG_1234.jpg").next_index == 0 + assert uploads.get("bob", "photos", "IMG_1234.jpg").stored_name \ + == "IMG_1234 (2).jpg" + + +def test_the_same_name_in_two_directories_is_two_uploads(): + uploads = PartialUploads() + uploads.start("alice", "media", "a.bin", "a.bin") + uploads.start("alice", "archive", "a.bin", "a.bin") + uploads.advance("alice", "media", "a.bin", 0, 5) + assert uploads.get("alice", "archive", "a.bin").next_index == 0 + + +def test_advancing_an_upload_nobody_started_says_so(): + """The caller refuses the chunk on this; silently creating the state here + would let a client append to whatever `.part` is already on disk.""" + assert PartialUploads().advance("alice", "media", "x", 0, 1) is None + + +def test_starting_again_forgets_the_old_position(): + """Chunk zero means "from the beginning" — the file is opened for writing, + not appending, so the position has to go with it.""" + uploads = PartialUploads() + uploads.start("alice", "media", "a.bin", "a.bin") + uploads.advance("alice", "media", "a.bin", 0, 500) + uploads.start("alice", "media", "a.bin", "a.bin") + assert uploads.get("alice", "media", "a.bin").next_index == 0 + assert uploads.get("alice", "media", "a.bin").bytes == 0 + + +# ── the reaper ────────────────────────────────────────────────────────────── + +def _old(seconds: float) -> float: + return 1_000_000.0 - seconds + + +NOW = 1_000_000.0 +FILM = Path("/roots/media/film.mkv.part") + + +def test_a_part_nobody_is_writing_and_nobody_has_touched_is_deleted(): + """The leak this exists to close: an abandoned upload's file, kept for ever + and invisible because `.part` is not an index entry.""" + doomed = orphaned_parts([(FILM, _old(ORPHAN_AFTER_SECS + 1))], + live=set(), now=NOW) + assert doomed == [FILM] + + +def test_an_upload_in_progress_is_never_deleted(): + """Even when its file is old: a large upload over a slow link is exactly the + one that has been on disk the longest, and it is the one that would hurt + most to lose.""" + doomed = orphaned_parts([(FILM, _old(ORPHAN_AFTER_SECS * 3))], + live={FILM}, now=NOW) + assert doomed == [] + + +def test_a_recently_written_part_is_left_alone(): + """No state and recent writes is a client that has just reconnected, or one + whose state this node has not seen yet. Waiting a day costs disk; being + wrong costs somebody their upload.""" + doomed = orphaned_parts([(FILM, _old(60))], live=set(), now=NOW) + assert doomed == [] + + +def test_the_same_name_in_another_directory_does_not_protect_it(): + """Matched on the whole path, so an upload to `media/` cannot keep an + orphan in `archive/` alive for ever. Comparing names would; comparing a + path rebuilt from a root and a relative directory would be a second + implementation that has to agree with the first for ever, and the state + records the path it is writing instead.""" + other = Path("/roots/archive/film.mkv.part") + doomed = orphaned_parts([(other, _old(ORPHAN_AFTER_SECS + 1))], + live={FILM}, now=NOW) + assert doomed == [other] + + +def test_a_finished_file_is_not_a_candidate(): + """Only `.part` is ever deleted. A bug that let this touch a real file would + be the worst one in the project, so the check is here as well as at the call + site that only offers `.part` paths.""" + doomed = orphaned_parts( + [(Path("/roots/media/film.mkv"), _old(ORPHAN_AFTER_SECS * 10))], + live=set(), now=NOW) + assert doomed == [] + + +def test_a_file_from_the_future_is_left_alone(): + """A clock that went backwards is not evidence that a file is abandoned, and + deleting is not reversible.""" + doomed = orphaned_parts([(Path("/roots/media/a.part"), NOW + 10_000)], + live=set(), now=NOW) + assert doomed == [] + + +def test_the_boundary_is_the_age_itself(): + at = [(Path("/roots/media/a.part"), _old(ORPHAN_AFTER_SECS))] + just_under = [(Path("/roots/media/a.part"), _old(ORPHAN_AFTER_SECS - 1))] + assert orphaned_parts(at, set(), NOW) == [Path("/roots/media/a.part")] + assert orphaned_parts(just_under, set(), NOW) == [] + + +def test_an_upload_records_the_file_it_is_writing(): + """What keeps the reaper honest. Without it the two sides would have to + agree on how a path is built from a root name and a relative directory — + two implementations of one rule, and the failure mode is deleting a live + upload.""" + uploads = PartialUploads() + uploads.start("alice", "media", "film.mkv", "film.mkv", part_path=FILM) + assert uploads.live_paths() == {FILM} + uploads.drop("alice", "media", "film.mkv") + assert uploads.live_paths() == set() + + +def test_the_suffix_is_named_once(): + """Two spellings of `.part` would be a bug nobody could see: the writer + would produce one and the reaper would look for the other.""" + assert PART_SUFFIX == ".part" + + +# ── the walk, and the deletion ────────────────────────────────────────────── + +def _root(tmp_path, name, *, writable=True, available=True) -> Root: + path = tmp_path / name + path.mkdir(parents=True, exist_ok=True) + return Root(name=name, path=path, writable=writable, available=available) + + +def _aged(path: Path, seconds: float, content: bytes = b"x") -> Path: + path.parent.mkdir(parents=True, exist_ok=True) + path.write_bytes(content) + when = time.time() - seconds + os.utime(path, (when, when)) + return path + + +def test_the_walk_finds_parts_in_subdirectories(tmp_path): + """Uploads go into the folder the sender was looking at, which is any + directory in the group — not a quarantine subfolder, since 2026-08-14.""" + root = _root(tmp_path, "media") + _aged(root.path / "a.part", 10) + _aged(root.path / "series" / "b.part", 10) + _aged(root.path / "series" / "kept.mkv", 10) + found = {p.name for p, _ in find_parts([root])} + assert found == {"a.part", "b.part"} + + +def test_a_read_only_root_is_not_walked(tmp_path): + """It cannot have received an upload, so anything `.part` in it belongs to + the operator and is none of this code's business.""" + root = _root(tmp_path, "library", writable=False) + _aged(root.path / "theirs.part", ORPHAN_AFTER_SECS * 2) + assert find_parts([root]) == [] + + +def test_an_unavailable_root_is_not_walked(tmp_path): + """A drive that is not mounted. Walking it finds nothing, and "nothing + found" is the input from which a careless janitor concludes everything is + gone.""" + root = _root(tmp_path, "external", available=False) + _aged(root.path / "x.part", ORPHAN_AFTER_SECS * 2) + assert find_parts([root]) == [] + + +def _daemon(groups: dict) -> NodeDaemon: + """A daemon with nothing but what `_reap_once` reads.""" + daemon = NodeDaemon.__new__(NodeDaemon) + daemon._webrtc = types.SimpleNamespace(_ctx={"groups": groups}) + return daemon + + +def test_the_janitor_deletes_the_abandoned_and_keeps_the_rest(tmp_path): + """End to end on real files: the old orphan goes, the recent one and the + one somebody is still writing stay, and a finished file is never a + candidate.""" + root = _root(tmp_path, "media") + old = _aged(root.path / "abandoned.mkv.part", ORPHAN_AFTER_SECS + 60) + recent = _aged(root.path / "fresh.mkv.part", 30) + live = _aged(root.path / "sending.mkv.part", ORPHAN_AFTER_SECS * 2) + finished = _aged(root.path / "done.mkv", ORPHAN_AFTER_SECS * 5) + + uploads = PartialUploads() + uploads.start("alice", "media", "sending.mkv", "sending.mkv", part_path=live) + + daemon = _daemon({"g1": {"roots": RootSet(roots=[root]), + "partial_uploads": uploads}}) + assert daemon._reap_once() == 1 + assert not old.exists() + assert recent.exists() and live.exists() and finished.exists() + + +def test_a_group_that_has_never_uploaded_anything_is_handled(tmp_path): + """No `partial_uploads` in the context yet — it is created on first use, so + a node that has been up for five minutes has none.""" + root = _root(tmp_path, "media") + old = _aged(root.path / "left.mkv.part", ORPHAN_AFTER_SECS + 1) + daemon = _daemon({"g1": {"roots": RootSet(roots=[root])}}) + assert daemon._reap_once() == 1 + assert not old.exists() + + +def test_a_group_with_no_roots_is_skipped(tmp_path): + assert _daemon({"g1": {}})._reap_once() == 0 + + +# ── across two connections ────────────────────────────────────────────────── + +GROUP = "g" * 32 + + +def _peer(ctx: dict, user_id: str = "user-1") -> WebRTCPeerSession: + """One connection into a group whose context is shared, as it is on a node. + + Two of these standing for the same member is the whole point: the second is + the reconnection, and it must find what the first was doing. + """ + session = WebRTCPeerSession.__new__(WebRTCPeerSession) + session._ctx = ctx + session._group_id = GROUP + session._user_id = user_id + session._pk_user = "" + session.sent = [] + session._send = session.sent.append + session._audit = lambda *a, **k: None + return session + + +def _group_ctx(tmp_path) -> dict: + shared = tmp_path / "shared" + shared.mkdir(exist_ok=True) + return {"roots": one_root(shared), + "index": GroupIndex(group_id=GROUP, + sk_node=Ed25519PrivateKey.generate()), + "gek": generate_gek()} + + +def _errors(session): + return [m for m in session.sent if m.get("type") == "error"] + + +def test_an_upload_survives_the_connection_that_started_it(tmp_path): + """The defect this stage exists to fix. + + The state used to live on the session, so the second connection saw no + upload at all and refused the chunk with `not_started`: an upload + interrupted at 99% could only be started again from zero, on a link flaky + enough to have interrupted it once. + """ + ctx = _group_ctx(tmp_path) + first = _peer(ctx) + first._do_file_upload(sealed_upload(first, filename="film.mkv", + data=b"first-half", + chunk_index=0, total_chunks=2)) + assert _errors(first) == [] + + # The link drops; the client comes back on a new connection and carries on. + second = _peer(ctx) + second._do_file_upload(sealed_upload(second, filename="film.mkv", + data=b"second-half", + chunk_index=1, total_chunks=2)) + assert _errors(second) == [], _errors(second) + + root = ctx["roots"].roots[0] + assert (root.path / "film.mkv").read_bytes() == b"first-halfsecond-half" + + +def test_another_member_cannot_continue_somebody_elses_upload(tmp_path): + """The key includes the member for a reason. Without it, a second person + sending the same name into the same folder would append their chunks to the + first person's file — which a shared folder makes an ordinary accident, not + only an attack.""" + ctx = _group_ctx(tmp_path) + alice = _peer(ctx, "alice") + alice._do_file_upload(sealed_upload(alice, filename="IMG_1234.jpg", + data=b"hers", chunk_index=0, + total_chunks=2)) + assert _errors(alice) == [] + + bob = _peer(ctx, "bob") + bob._do_file_upload(sealed_upload(bob, filename="IMG_1234.jpg", + data=b"his", chunk_index=1, + total_chunks=2)) + assert [m.get("code") for m in _errors(bob)] == ["not_started"] + + +def test_an_upload_in_flight_is_known_to_the_reaper(tmp_path): + """The two halves of this stage meeting: the state the node keeps is what + stops the janitor deleting a file somebody is still sending.""" + ctx = _group_ctx(tmp_path) + peer = _peer(ctx) + peer._do_file_upload(sealed_upload(peer, filename="film.mkv", + data=b"half", chunk_index=0, + total_chunks=2)) + live = ctx["partial_uploads"].live_paths() + assert len(live) == 1 + assert next(iter(live)).name == "film.mkv.part" + assert next(iter(live)).exists() -- cgit v1.2.3 From 4f5d3d4ac151874f03c6fcc451d6b1d5bb1efb78 Mon Sep 17 00:00:00 2001 From: Christophe Besson Date: Wed, 9 Sep 2026 12:01:06 +0200 Subject: feat: resume an interrupted upload, and pause one MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Stage 8 of ~/next/improve-downloads.md, second half, plus the gap it exposed in stage 7. **Asking where to resume.** The node identifies an upload by (member, directory, filename), so a client resuming one has to name the file — and `transfer_open`, the obvious place to ask, travels in clear. Naming it there would undo exactly what sealing this path bought in MNP 2.0: before it, the same file was ciphertext leaving a node and plaintext arriving at one. So the question is asked inside the seal that already exists, as an ordinary `file_upload` with no bytes and `chunk_index: -1`. The node writes nothing, creates no state, reserves no name, and answers with `resume_from` in the sealed ack. A node that predates it refuses the index, which the client reads as "start from the beginning" — the behaviour it had anyway — and the wait is bounded so one that answers neither does not strand an upload. The probe is answered after every check the write path makes, so it cannot ask questions about a directory the caller may not write to, and it answers only about the member who asks: otherwise one member could measure another's progress on a file they never sent, and worse, resume it. **Pausing an upload.** Reported: no pause button on an upload, even in the desktop app. Stage 7 built pause around the download path — a target declares whether it can be stopped — and an upload has no local target to ask. It was also refused by design, since a transfer handed a lease it cannot re-create must not be offered a button that would drop its slot for good. Uploads now ask for their slot rather than being handed one, and say they are pausable outright: a File is seekable and the node keeps the position. Resuming re-probes rather than trusting the client's own memory, so it works across a reconnect too. **And the slot they hold.** `_do_file_upload` never called `slots.touch(tr)`. Chunks are not gated by the lease, so the file arrived — but the node reclaimed a grant nobody appeared to be using after thirty seconds, twice, then abandoned it, and the widget follows the lease. Measured from the journal: a 3.5 GB upload read "waiting, 0 ahead" for a minute and a half while it was transferring. The download twin of this was fixed on 2026-09-08; the same omission was still here, invisible until uploads took a real lease. `test_the_upload_itself_is_sealed` now checks every message `uploadFile` sends rather than the first. Adding the probe put a second one in front of the one it was written for, and it would have kept passing while guarding nothing. Node suite 1202 passed, hub suite 850 passed. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_01HCGdheDLxGReuKHga3BtST --- .../meshbay-common/src/meshbay_common/protocol.py | 24 ++++ .../src/meshbay_hub/static/files-app.js | 10 +- .../src/meshbay_hub/static/transfers.js | 9 +- .../src/meshbay_hub/static/transport.js | 85 +++++++++++++- .../tests/harness/upload_seal_probe.mjs | 25 +++- packages/meshbay-hub/tests/test_transfers.py | 57 +++++++++ .../meshbay-hub/tests/test_transport_contracts.py | 33 ++++-- .../meshbay-hub/tests/test_upload_seal_client.py | 38 ++++++ .../src/meshbay_node/transport/webrtc_server.py | 48 ++++++++ .../meshbay-node/tests/test_partial_uploads.py | 129 +++++++++++++++++++++ 10 files changed, 440 insertions(+), 18 deletions(-) (limited to 'packages/meshbay-node/tests/test_partial_uploads.py') diff --git a/packages/meshbay-common/src/meshbay_common/protocol.py b/packages/meshbay-common/src/meshbay_common/protocol.py index 8a521bb..5bd2903 100644 --- a/packages/meshbay-common/src/meshbay_common/protocol.py +++ b/packages/meshbay-common/src/meshbay_common/protocol.py @@ -501,6 +501,22 @@ def file_upload_payload(gek: bytes, group_id: str, msg: dict) -> dict: return unseal(gek, PURPOSE_UPLOAD, MNP.FILE_UPLOAD, group_id, msg) +# "Where am I?", asked as an ordinary sealed upload chunk rather than as a new +# message. +# +# The node identifies an upload by (member, directory, filename), so a client +# resuming one has to name the file — and `transfer_open`, the obvious place to +# ask, travels in clear. Naming it there would undo exactly what sealing the +# upload path bought: before MNP 2.0 the same file was ciphertext leaving a node +# and plaintext arriving at one. +# +# So the question is asked inside the seal that already exists, as a chunk with +# no bytes and this index. The node writes nothing, changes nothing, and answers +# with `resume_from`. A node that predates this refuses the index, which the +# client reads as "start from the beginning" — the behaviour it had anyway. +UPLOAD_PROBE_INDEX = -1 + + def file_upload_ack_wire( gek: bytes, group_id: str, @@ -510,6 +526,7 @@ def file_upload_ack_wire( filename: str, stored_as: str, dir: str = "", + resume_from: int | None = None, ) -> dict: """ The node's answer to one chunk, sealed the same way. @@ -518,8 +535,15 @@ def file_upload_ack_wire( replacing anything — and `dir` is where it landed. Both name the operator's content, so both belong inside the seal; only `upload_id` and `chunk_index` stay out, because the client matches on them. + + `resume_from` answers the probe chunk (`UPLOAD_PROBE_INDEX`): how many + chunks of this file the node already holds. Inside the seal like the rest — + it is a fact about the operator's disk — and absent from an ordinary ack, so + a client can tell the two apart without looking at `chunk_index`. """ payload = {"filename": filename, "stored_as": stored_as, "dir": dir} + if resume_from is not None: + payload["resume_from"] = int(resume_from) return { "type": MNP.FILE_UPLOAD_ACK, "v": MNP_VERSION, diff --git a/packages/meshbay-hub/src/meshbay_hub/static/files-app.js b/packages/meshbay-hub/src/meshbay_hub/static/files-app.js index fb9d801..65860ec 100644 --- a/packages/meshbay-hub/src/meshbay_hub/static/files-app.js +++ b/packages/meshbay-hub/src/meshbay_hub/static/files-app.js @@ -94,7 +94,15 @@ function FilesPanel({ for (const file of files) { transfers.start({ kind: 'upload', name: file.name, total: file.size, transport, - lease: transport.openTransfer({ kind: 'upload', bytes: file.size }), + // `makeLease`, not `lease`: pausing gives the slot back, so resuming + // has to be able to ask for another one, and a transfer handed a lease + // it cannot re-create is refused the button rather than offered one + // that would drop its slot for good. + makeLease: () => transport.openTransfer({ kind: 'upload', + bytes: file.size }), + // A `File` is seekable and the node remembers how much it holds, so + // there is no target tier to consult here — unlike a download. + pausable: true, run: async ({ signal, onProgress, lease }) => { await transport.uploadFile(file, { // Bytes the node acknowledged, not bytes read locally. diff --git a/packages/meshbay-hub/src/meshbay_hub/static/transfers.js b/packages/meshbay-hub/src/meshbay_hub/static/transfers.js index 797321c..524b7f3 100644 --- a/packages/meshbay-hub/src/meshbay_hub/static/transfers.js +++ b/packages/meshbay-hub/src/meshbay_hub/static/transfers.js @@ -141,7 +141,7 @@ export class TransferStore { * there is somewhere to write — see file-utils.js's downloadEntry. */ start({ kind, name, total = 0, transport = null, run, open = null, - lease = null, prepare = null, makeLease = null }) { + lease = null, prepare = null, makeLease = null, pausable = false }) { const item = { id: _nextId++, kind, name, total, transport, open, lease, @@ -156,8 +156,11 @@ export class TransferStore { error: '', samples: [{ t: this._now(), done: 0 }], signal: { aborted: false, paused: false }, - // Set from `prepare`: whether this target can be stopped and continued. - pausable: false, + // Whether this transfer can be stopped and continued. A download learns + // it from `prepare`, because only its target knows; an upload says so + // outright, because a `File` is always seekable and the node keeps the + // position (see uploads.py). + pausable: Boolean(pausable), // Where a resumed run picks up, in chunks. Zero until something pauses. resumeFrom: 0, // Resolved by resume(); awaited by the run loop while paused. diff --git a/packages/meshbay-hub/src/meshbay_hub/static/transport.js b/packages/meshbay-hub/src/meshbay_hub/static/transport.js index 54a1302..23823ac 100644 --- a/packages/meshbay-hub/src/meshbay_hub/static/transport.js +++ b/packages/meshbay-hub/src/meshbay_hub/static/transport.js @@ -40,6 +40,16 @@ async function _pkEdFromSk(skPkcs8B64) { // flight, which saturates any path up to roughly 100 Mb/s at 100 ms. const UPLOAD_CHUNK_SIZE = 48 * 1024; const UPLOAD_WINDOW = 32; +// "Where am I?", asked as an ordinary sealed upload chunk with no bytes rather +// than on a clear message. Mirrors UPLOAD_PROBE_INDEX in +// meshbay_common/protocol.py; the node writes nothing and answers with +// `resume_from`, and one that predates it refuses the index, which reads as +// "start from the beginning". +const UPLOAD_PROBE_INDEX = -1; +// How long to wait for that answer before assuming there is none. A node that +// answers neither the probe nor its refusal must not leave an upload waiting +// for ever, and starting over is always safe. +const UPLOAD_PROBE_TIMEOUT_MS = 5000; const UPLOAD_BUFFER_HIGH = 1024 * 1024; // Segments of 256 KB: 24 in flight is 6 MB, enough to keep playback fed over a @@ -2409,8 +2419,23 @@ class MeshBayTransport { const waiter = acks.shift(); if (waiter) waiter(); }; + // "Where am I?" — resolved by the node's answer to the probe chunk below, + // or by anything that says this node cannot answer it. + let settleProbe = null; + const probed = new Promise((r) => { settleProbe = r; }); + const answerProbe = (from) => { + if (!settleProbe) return false; + const done = settleProbe; + settleProbe = null; + done(from); + return true; + }; this._uploaders.set(uploadId, (msg) => { if (msg.type === 'error') { + // A node that predates the probe refuses its index. That is not a + // failure — it is the answer "start from the beginning", which is what + // this client did before there was anything to ask. + if (answerProbe(0)) return; failure = new Error(msg.detail || 'Upload refused'); wake(); return; @@ -2423,19 +2448,75 @@ class MeshBayTransport { .then((plain) => { const payload = msgpack_decode(plain); if (payload.stored_as) stored = payload; + // Only the probe's answer carries this, so the two are told apart + // without trusting the index the node echoed back in clear. + if (typeof payload.resume_from === 'number') return answerProbe(payload.resume_from); + return false; }) .catch((e) => { failure = new Error( `The node's upload reply did not open under the group key (${e.message})`); + return false; }) - .finally(wake); + // A probe's answer is not a chunk: waking here would credit the + // progress bar with a chunk that was never sent. + .then((wasProbe) => { if (!wasProbe) wake(); }); }); const nextAck = () => new Promise(r => acks.push(r)); try { - for (let i = 0; i < total; i++) { + // Ask before sending anything. An upload interrupted at 99% used to start + // again from zero, because the node kept its position on the connection + // that was lost — see `uploads.py`. The question goes inside the seal, as + // a chunk with no bytes, because naming the file on a clear message is + // exactly what sealing this path was for. + // Sealed first, spread second — the same shape as the chunk loop below, + // and not only for symmetry: `test_the_upload_itself_is_sealed` reads + // this call and fails if a filename appears in it, which is how it can + // tell a field outside the seal from one inside it. + const probeSealed = await C.sealGroup( + this._gekRaw, 'upload', 'file_upload', groupId, + msgpack_encode({ filename: file.name, data: new Uint8Array(0), + dir: dir || '', root: root || '' })); + this._send({ + type: 'file_upload', + v: '0.1', + upload_id: uploadId, + chunk_index: UPLOAD_PROBE_INDEX, + total_chunks: total, + ...(tr ? { tr } : {}), + ...probeSealed, + }); + // Bounded: a node that answers neither the probe nor its refusal must not + // leave an upload waiting for ever. Starting over is always safe. + let from = await Promise.race([ + probed, + new Promise((r) => setTimeout(() => { answerProbe(0); r(0); }, + UPLOAD_PROBE_TIMEOUT_MS)), + ]); + // Defensive: a node reporting a position at or past the end would have + // renamed the file and dropped its state, so this cannot happen — and if + // it does, sending everything again is the answer that cannot corrupt. + if (!(from > 0) || from >= total) from = 0; + if (from > 0) { + acked = from; + if (onProgress) onProgress(Math.min(file.size, from * size), file.size); + } + + for (let i = from; i < total; i++) { if (signal && signal.aborted) throw _aborted(); + // Between two chunks, never inside one — the node refuses a chunk that + // is not the one it expects, so a position is the only thing worth + // remembering. Nothing is recorded here beyond that: the node holds the + // real position, and the probe above is what asks for it on the way + // back in, which makes resuming correct even across a reconnect. + if (signal && signal.paused) { + signal.resumeFrom = i; + const paused = new Error('Paused'); + paused.name = 'PausedError'; + throw paused; + } // Backpressure: without it the whole file lands in the browser's send // buffer in seconds and the progress bar becomes a work of fiction. while (this._channel && this._channel.bufferedAmount > UPLOAD_BUFFER_HIGH) { diff --git a/packages/meshbay-hub/tests/harness/upload_seal_probe.mjs b/packages/meshbay-hub/tests/harness/upload_seal_probe.mjs index 0b77e42..a6008c2 100644 --- a/packages/meshbay-hub/tests/harness/upload_seal_probe.mjs +++ b/packages/meshbay-hub/tests/harness/upload_seal_probe.mjs @@ -72,13 +72,34 @@ tp._nodeVersion = input.node_version; const frames = []; let uploadId = null; +let answered = 0; tp._send = (msg) => { frames.push(toHex(msgpack_encode(msg))); if (msg.upload_id) uploadId = msg.upload_id; - if (input.mode !== 'receive') return; + if (input.mode !== 'receive') { + // Nothing answers in this mode -- except the probe, which the client waits + // five seconds for. A node that predates it refuses the index, and that + // refusal is a plain error rather than a sealed ack, so the harness can + // produce it honestly. It is also the degradation path worth exercising. + if (msg.chunk_index === -1) { + // With `probe_ack`, answer it the way a node holding part of this file + // does; without, the way one that predates the probe does. + const reply = input.probe_ack + ? Object.assign(msgpack_decode(hex(input.probe_ack)), + { upload_id: uploadId }) + : { type: 'error', upload_id: uploadId, + code: 'bad_chunk_index', detail: 'Unexpected chunk index' }; + setImmediate(() => tp._dispatch(reply)); + } + return; + } // Answer as the node did, on the next turn of the loop so the send path // finishes first — which is also how a real ack arrives. - const ack = msgpack_decode(hex(input.acks[msg.chunk_index])); + // + // By position, not by `chunk_index`: the node answers every frame including + // the probe, whose index is -1, and the two lists are built from the same + // sequence of frames. + const ack = msgpack_decode(hex(input.acks[answered++])); ack.upload_id = uploadId; // Through the real `_dispatch`, so the routing under test — matching an // ack to its uploader by `upload_id` — is the shipped one. diff --git a/packages/meshbay-hub/tests/test_transfers.py b/packages/meshbay-hub/tests/test_transfers.py index 035f569..f27d4fd 100644 --- a/packages/meshbay-hub/tests/test_transfers.py +++ b/packages/meshbay-hub/tests/test_transfers.py @@ -713,3 +713,60 @@ def test_a_paused_transfer_still_counts_as_live(tmp_path): say('pending:' + t.pending); """, tmp_path) assert out == ["pending:1"] + + +def test_an_upload_can_be_paused_without_a_prepare_step(tmp_path): + """A download learns whether it can pause from its target, because only the + target knows. An upload has no target to ask: a `File` is seekable and the + node keeps the position, so it says so outright. + + This was missed when pause shipped — the button appeared on downloads and + nowhere else, including in the desktop app where everything else works. + """ + out = _run(_lease_stub() + """ + const t = new TransferStore(); + const leases = []; + t.start({ + kind: 'upload', name: 'f', total: 100, pausable: true, + makeLease: () => { const l = new L(); leases.push(l); return l; }, + run: async ({ signal, from }) => { + for (let i = from || 0; i < 10; i++) { + await new Promise(r => setTimeout(r, 5)); + if (signal.paused) { + signal.resumeFrom = i; + const e = new Error('p'); e.name = 'PausedError'; throw e; + } + } + }, + }); + await new Promise(r => setTimeout(r, 5)); + leases[0].grant(); + await new Promise(r => setTimeout(r, 20)); + say('pausable:' + t.list()[0].pausable); + t.pause(t.list()[0].id); + await new Promise(r => setTimeout(r, 30)); + say('status:' + t.list()[0].status); + say('released:' + leases[0].released.join(',')); + """, tmp_path) + assert out == ["pausable:true", "status:paused", "released:paused"] + + +def test_an_upload_handed_a_lease_it_cannot_recreate_is_not_offered_pause(tmp_path): + """Pausing gives the slot back. A transfer that cannot ask for another one + would pause once and wait for ever, so the button is refused instead.""" + out = _run(_lease_stub() + """ + const t = new TransferStore(); + const lease = new L(); + t.start({ kind: 'upload', name: 'f', total: 100, pausable: true, lease, + run: async ({ signal }) => { + while (!signal.aborted) await new Promise(r => setTimeout(r, 5)); + } }); + lease.grant(); + await new Promise(r => setTimeout(r, 20)); + const id = t.list()[0].id; + t.pause(id); + await new Promise(r => setTimeout(r, 20)); + say('status:' + t.list()[0].status); + t.cancel(id); + """, tmp_path) + assert out == ["status:running"] diff --git a/packages/meshbay-hub/tests/test_transport_contracts.py b/packages/meshbay-hub/tests/test_transport_contracts.py index 879062b..fe550f9 100644 --- a/packages/meshbay-hub/tests/test_transport_contracts.py +++ b/packages/meshbay-hub/tests/test_transport_contracts.py @@ -349,16 +349,29 @@ def test_the_upload_itself_is_sealed(transport): "the upload must be sealed under the group key") assert "openGroup(" in body and "'file_upload_ack'" in body, ( "the ack carries the stored name and must be opened, not read") - # The message the node actually receives: everything between `this._send({` - # and its close. Read on its own, because the same field names appear a few - # lines above inside `msgpack_encode({...})`, which is the sealed half. - sent = body[body.index("this._send({"):] - sent = sent[:sent.index("});")] - assert "filename" not in sent, "the filename is on the message in clear" - assert "data" not in sent, "the bytes are on the message in clear" - assert "dir" not in sent and "root" not in sent, ( - "the destination is on the message in clear") - assert "...sealed," in sent, "the message must carry the sealed pair" + # The messages the node actually receives: everything between each + # `this._send({` and its close. Read on their own, because the same field + # names appear a few lines above inside `msgpack_encode({...})`, which is + # the sealed half. + # + # Every one of them, not the first: `uploadFile` sends a probe chunk before + # the file ("where am I?", UPLOAD_PROBE_INDEX) and it names the file too, so + # a check that stopped at the first message would have moved off the one it + # was written for the day the second appeared. + sends = [] + rest = body + while "this._send({" in rest: + rest = rest[rest.index("this._send({"):] + sends.append(rest[:rest.index("});")]) + rest = rest[len("this._send({"):] + assert len(sends) >= 2, "the probe and the chunks are both sent from here" + for sent in sends: + assert "filename" not in sent, "the filename is on the message in clear" + assert "data" not in sent, "the bytes are on the message in clear" + assert "dir" not in sent and "root" not in sent, ( + "the destination is on the message in clear") + assert "...sealed," in sent or "...probeSealed," in sent, ( + "the message must carry the sealed pair") assert "supportsSealedUpload" in body, ( "an older node must be refused before a chunk is sent, not after") diff --git a/packages/meshbay-hub/tests/test_upload_seal_client.py b/packages/meshbay-hub/tests/test_upload_seal_client.py index d6f9156..2e4bfb5 100644 --- a/packages/meshbay-hub/tests/test_upload_seal_client.py +++ b/packages/meshbay-hub/tests/test_upload_seal_client.py @@ -167,3 +167,41 @@ def test_the_client_refuses_an_older_node_before_sending_a_chunk(_gek): assert result["state"] == "rejected" assert "older MeshBay" in result["message"] assert result["frames"] == [], "a chunk was sent to a node that cannot open it" + + +def test_an_interrupted_upload_resumes_where_the_node_stopped(tmp_path, _gek): + """ + The browser asks, the node answers, and the second attempt sends only what + is missing. + + Both halves are the shipped ones: the frames come from the real + `uploadFile`, the answer comes from the real node handler. What is asserted + is the thing that used to be impossible — an upload interrupted at chunk two + of five that sends three chunks instead of five. + """ + body = bytes(range(256)) * ((CHUNK * 5) // 256 + 1) + body = body[:CHUNK * 5] + first = _run_probe(_probe_input(_gek, "send", + file={"name": "film.mkv", "data": body.hex()})) + frames = [msgpack.unpackb(bytes.fromhex(f), raw=False) + for f in first["frames"]] + assert [f["chunk_index"] for f in frames] == [-1, 0, 1, 2, 3, 4] + + # The link drops after two chunks. + session = _node_session(tmp_path, _gek) + for frame in frames[1:3]: + session._do_file_upload(frame) + assert not [m for m in session.sent if m.get("type") == "error"] + + # It comes back and asks. + session.sent.clear() + session._do_file_upload(frames[0]) + probe_ack = msgpack.packb(session.sent[-1], use_bin_type=True).hex() + + second = _run_probe(_probe_input( + _gek, "send", file={"name": "film.mkv", "data": body.hex()}, + probe_ack=probe_ack)) + resumed = [msgpack.unpackb(bytes.fromhex(f), raw=False)["chunk_index"] + for f in second["frames"]] + assert resumed == [-1, 2, 3, 4], ( + f"sent {resumed} — the answer to the probe was not used") 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") -- cgit v1.2.3