diff options
Diffstat (limited to 'packages/meshbay-hub/src')
3 files changed, 98 insertions, 6 deletions
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) { |