diff options
Diffstat (limited to 'packages')
10 files changed, 440 insertions, 18 deletions
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") |