diff options
| -rw-r--r-- | packages/meshbay-hub/src/meshbay_hub/static/app.js | 26 | ||||
| -rw-r--r-- | packages/meshbay-hub/src/meshbay_hub/static/transport.js | 102 | ||||
| -rw-r--r-- | packages/meshbay-hub/tests/test_spa_ordering.py | 24 |
3 files changed, 120 insertions, 32 deletions
diff --git a/packages/meshbay-hub/src/meshbay_hub/static/app.js b/packages/meshbay-hub/src/meshbay_hub/static/app.js index fee21dc..3ef0da5 100644 --- a/packages/meshbay-hub/src/meshbay_hub/static/app.js +++ b/packages/meshbay-hub/src/meshbay_hub/static/app.js @@ -832,7 +832,6 @@ function formatDate(ts) { // ── Group Page ────────────────────────────────────────────────────────────── const CHUNK_SIZE = 1024 * 1024; -const UPLOAD_CHUNK_SIZE = 48 * 1024; const PIPELINE_WINDOW = 8; async function pipelinedDownload(transport, gekKey, fileId, totalChunks, onChunk, writable) { @@ -1109,15 +1108,10 @@ function GroupPage({ groupId, group, token, username, userId, onRefreshAuth, setError(''); setUlState({ name: file.name, sent: 0, total: file.size, indexing: false }); try { - const totalChunks = Math.ceil(file.size / UPLOAD_CHUNK_SIZE); - for (let i = 0; i < totalChunks; i++) { - const slice = file.slice(i * UPLOAD_CHUNK_SIZE, (i + 1) * UPLOAD_CHUNK_SIZE); - const buf = new Uint8Array(await slice.arrayBuffer()); - await transport.uploadChunk(file.name, i, totalChunks, buf); + await transport.uploadFile(file, { // Bytes actually acknowledged by the node, not bytes read locally. - setUlState(prev => prev && { ...prev, sent: Math.min(file.size, - (i + 1) * UPLOAD_CHUNK_SIZE) }); - } + onProgress: (sent) => setUlState(prev => prev && { ...prev, sent }), + }); // The node re-indexes on a filesystem event; there is nothing to poll, so // say what is happening instead of showing a finished bar and no file. setUlState(prev => prev && { ...prev, indexing: true }); @@ -1934,16 +1928,10 @@ function ChatPanel({ transportRef, username, entries, gekRef, onRefreshIndex, on if (!transport || !transport.connected) return; setAttaching(true); try { - const totalChunks = Math.ceil(file.size / UPLOAD_CHUNK_SIZE); - let storedAs = file.name; - for (let i = 0; i < totalChunks; i++) { - const slice = file.slice(i * UPLOAD_CHUNK_SIZE, (i + 1) * UPLOAD_CHUNK_SIZE); - const buf = new Uint8Array(await slice.arrayBuffer()); - const ack = await transport.uploadChunk(file.name, i, totalChunks, buf); - // Two people sending IMG_1234.jpg both succeed; the node picks a free - // name and the message has to point at the one it chose. - if (ack && ack.stored_as) storedAs = ack.stored_as; - } + // Two people sending IMG_1234.jpg both succeed; the node picks a free name + // and the message has to point at the one it chose. + const ack = await transport.uploadFile(file); + const storedAs = (ack && ack.stored_as) || file.name; await new Promise(r => setTimeout(r, 2500)); if (onRefreshIndex) await onRefreshIndex(); const ext = file.name.split('.').pop().toLowerCase(); diff --git a/packages/meshbay-hub/src/meshbay_hub/static/transport.js b/packages/meshbay-hub/src/meshbay_hub/static/transport.js index d904286..e93c601 100644 --- a/packages/meshbay-hub/src/meshbay_hub/static/transport.js +++ b/packages/meshbay-hub/src/meshbay_hub/static/transport.js @@ -35,6 +35,13 @@ async function _pkEdFromSk(skPkcs8B64) { return pad ? b64 + '='.repeat(4 - pad) : b64; } +// 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; +const UPLOAD_BUFFER_HIGH = 1024 * 1024; + const JOIN_REFUSALS = { code_required: 'This node does not know this browser yet. Ask the node operator ' + 'for a pairing code (meshbay-node operator pair).', @@ -66,6 +73,7 @@ class MeshBayTransport { this._onStreamData = null; this._onStreamEnd = null; this._onIndexSync = null; + this._onUploadAck = null; } get connected() { return this._connected; } @@ -489,19 +497,79 @@ class MeshBayTransport { this._send({ type: 'stream_req', v: '0.1', file_id: fileId }); } - async uploadChunk(filename, chunkIndex, totalChunks, data) { - // The node decides where this lands (uploads/) and under what name — it - // finds a free one rather than replacing anything. The ack says which. - const msg = await this._sendAndWait({ - type: 'file_upload', - v: '0.1', - filename, - chunk_index: chunkIndex, - total_chunks: totalChunks, - data: data, - }); - if (msg.type === 'error') throw new Error(msg.detail); - return msg; + /** + * 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. + */ + async uploadFile(file, { chunkSize, onProgress } = {}) { + // One at a time: the acks are matched by arrival, so two uploads sharing the + // channel would credit each other's progress and finish at the wrong moment. + if (this._onUploadAck) throw new Error('Another upload is already running'); + 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 = []; + this._onUploadAck = (msg) => { + if (msg.type === 'error') { + failure = new Error(msg.detail || 'Upload refused'); + } else if (msg.stored_as) { + stored = msg; + } + acked += 1; + if (onProgress) onProgress(Math.min(file.size, acked * size), file.size); + const waiter = acks.shift(); + if (waiter) waiter(); + }; + + const nextAck = () => new Promise(r => acks.push(r)); + + try { + for (let i = 0; i < total; i++) { + // 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) { + 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()); + this._send({ + type: 'file_upload', + v: '0.1', + filename: file.name, + chunk_index: i, + total_chunks: total, + data: buf, + }); + } + while (acked < total) { + await nextAck(); + if (failure) throw failure; + } + } finally { + this._onUploadAck = null; + } + return stored || {}; } /** Create a directory under the current one. Any member may. */ @@ -681,6 +749,14 @@ class MeshBayTransport { } _dispatch(msg) { + // While an upload is in flight the acks are its own, and there are many of + // them: they must not be handed to whatever request happens to be oldest in + // the pending map. + if (this._onUploadAck + && (msg.type === 'file_upload_ack' || msg.type === 'error')) { + this._onUploadAck(msg); + return; + } if (msg.type === 'chat_msg' && this._onChat) { this._onChat(msg); return; diff --git a/packages/meshbay-hub/tests/test_spa_ordering.py b/packages/meshbay-hub/tests/test_spa_ordering.py index 9301fcc..1329b74 100644 --- a/packages/meshbay-hub/tests/test_spa_ordering.py +++ b/packages/meshbay-hub/tests/test_spa_ordering.py @@ -143,3 +143,27 @@ def test_admin_page_does_not_borrow_the_members_panel_state(): for name in ("doInvite", "adminId", "inviteCode", "setInviteUser"): assert name not in admin, \ f"AdminPage references {name}, which only exists in MembersPanel" + + +# ── Upload pipelining ─────────────────────────────────────────────────────── +# +# The upload loop waited for the node to acknowledge each 48 KB chunk before +# reading the next one, which caps throughput at one chunk per round trip no +# matter how much bandwidth there is — and keeps SCTP's congestion window shut, +# so the transport never speeds up either. Measured over a 100 ms path: 0.16 MB/s +# waiting for every ack, 3.47 MB/s with 32 chunks in flight. + +def test_the_uploader_keeps_several_chunks_in_flight(): + src = TRANSPORT.read_text() + body = src[src.index("async uploadFile("):] + body = body[:body.index("\n async ", 1)] + assert "UPLOAD_WINDOW" in body, "the send window is gone — uploads are serial again" + assert "bufferedAmount" in body, ( + "without backpressure the file lands in the send buffer in seconds and " + "the progress bar becomes fiction") + + +def test_no_caller_waits_for_one_chunk_at_a_time(): + app = APP.read_text() + assert "uploadChunk(" not in app, ( + "a per-chunk await is back in the SPA; use transport.uploadFile()") |