// Uploads: a whole file, sealed chunk by chunk with several in flight, and the // folders it lands in. // // Methods of MeshBayTransport, copied onto its prototype by extendTransport // (transport.js, which the shell loads first). extendTransport(class { /** * Push a whole file, several chunks in flight at once. * * One chunk per round trip is 48 KB of throughput per RTT no matter how much * bandwidth there is: 4.8 MB/s on a 10 ms path, 480 KB/s on a 100 ms one, and * the sender is idle for almost all of it — which also keeps SCTP's congestion * window shut, so the transport never gets a chance to speed up either. A * window of chunks makes the rate depend on bandwidth rather than distance. * * Order is not at risk: a DataChannel is ordered and reliable by default, and * the node refuses any chunk that is not the one it expects next. * * The node decides where this lands (uploads/) and under what name — it finds a * free one rather than replacing anything. The ack says which, and that is what * this returns. * * `dir` names the folder to upload into, as a virtual path * (`Media/Films/1999`) — where the sender is actually looking. The node * resolves it against the group's own roots, which refuses `..`, absolute * segments and anything escaping its root; it is a place among the group's * folders, never a path on the operator's filesystem. * * `root` is the older, coarser form: the root's name and nothing below it. * Kept because a node that predates `dir` reads it, and because Chat has no * folder on screen to name. Omitting both leaves the node to pick, which it * only does for a client old enough to have had one destination. */ async uploadFile(file, { chunkSize, onProgress, signal, root, dir, tr = '' } = {}) { // The same file twice at once would confuse the node, which keys its own // upload state by folder and name — and would race for the same // destination. The guard uses the same key: by name alone, a dropped folder // holding a `cover.jpg` in two albums failed the second one for nothing. const inFlightKey = `${dir || ''}/${file.name}`; if (this._inFlightUploads.has(inFlightKey)) { throw new Error(`${file.name} is already being uploaded`); } if (!this._gekRaw) throw new Error('This group has no key on this device'); const C = window.MeshBayCrypto; const groupId = (this._connectArgs && this._connectArgs.groupId) || ''; this._inFlightUploads.add(inFlightKey); const uploadId = _hex(crypto.getRandomValues(new Uint8Array(16))); const size = chunkSize || UPLOAD_CHUNK_SIZE; const total = Math.max(1, Math.ceil(file.size / size)); let acked = 0; let stored = null; let failure = null; const acks = []; const wake = () => { acked += 1; if (onProgress) onProgress(Math.min(file.size, acked * size), file.size); 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; } // The ack is sealed too — `stored_as` and the folder it landed in name // the operator's content. Opening it is what makes the result usable, so // a failure here fails the upload rather than being swallowed: a chat // attachment that cannot learn its stored name would point at nothing. C.openGroup(this._gekRaw, 'upload', 'file_upload_ack', groupId, msg) .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; }) // 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 { // 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) { if (signal && signal.aborted) throw _aborted(); await new Promise(r => setTimeout(r, 20)); } while (i - acked >= UPLOAD_WINDOW) { await nextAck(); if (failure) throw failure; } if (failure) throw failure; const buf = new Uint8Array( await file.slice(i * size, (i + 1) * size).arrayBuffer()); // The name, the destination and the bytes go inside the seal together. // Mirrors `file_upload_wire` in meshbay_common/protocol.py; only the // fields the node routes on stay outside it. const sealed = await C.sealGroup( this._gekRaw, 'upload', 'file_upload', groupId, msgpack_encode({ filename: file.name, data: buf, dir: dir || '', root: root || '' })); this._send({ type: 'file_upload', v: '0.1', upload_id: uploadId, chunk_index: i, total_chunks: total, ...(tr ? { tr } : {}), ...sealed, }); } while (acked < total) { await nextAck(); if (failure) throw failure; } } finally { this._uploaders.delete(uploadId); this._inFlightUploads.delete(inFlightKey); } return stored || {}; } /** Create a directory under the current one. Any member may. */ async createDirectory(dir, name) { const msg = await this._sendAndWait({ type: 'dir_create', v: '0.1', dir: dir || '', name, }); if (msg.type === 'error') throw new Error(msg.detail); return msg; } }); // 48 KB is what fits comfortably in one SCTP message across stacks; the window is // what makes the rate independent of the round trip. 32 × 48 KB = 1.5 MB in // 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;