aboutsummaryrefslogtreecommitdiffstats
path: root/packages/meshbay-hub/src
diff options
context:
space:
mode:
authorChristophe Besson <cbesson@gmail.com>2026-08-15 09:19:34 +0200
committerChristophe Besson <cbesson@gmail.com>2026-08-15 09:19:34 +0200
commit4fa546a759dc9bd2ab77f793e3f2e660acd31bdb (patch)
tree967f46a0ef0f24086587d76231224667908c6ac1 /packages/meshbay-hub/src
parent84b032c65e17267d41e04605e79eea82a6f5a59f (diff)
downloadmeshbay-4fa546a759dc9bd2ab77f793e3f2e660acd31bdb.tar.gz
perf(upload): several chunks in flight, instead of one per round trip
The uploader read a 48 KB slice, sent it, and waited for the node to acknowledge it before reading the next one. That caps throughput at one chunk per round trip regardless of available bandwidth, and it is worse than the arithmetic suggests: the sender is idle for almost the whole time, so SCTP's congestion window never opens either, and the transport stays slow even when the link is not. Measured against the real node over a 100 ms path (netem on loopback): 48 KB chunks, one at a time 0.16 MB/s 48 KB chunks, 32 in flight 3.47 MB/s On loopback with no latency both are ~32 MB/s, which is why nothing here ever caught it: the local end-to-end run cannot see a round-trip problem. transport.uploadFile() now keeps a window of chunks in flight and matches acks by arrival, with the node's own ordering rule as the guard — a DataChannel is ordered and reliable, and the node refuses any chunk that is not the one it expects next. It pauses when the channel's buffered amount gets high, so the progress bar keeps reporting what the node has taken rather than what the browser has queued. Both callers, the Files panel and chat attachments, go through it. The end-to-end harness grew an opt-in benchmark behind MESHBAY_BENCH=1 that removes its own files afterwards, and it taught me something about the harness rather than the code: it took an unsolicited index_sync push for an upload ack, because unlike app.js it had no place to put one. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Diffstat (limited to 'packages/meshbay-hub/src')
-rw-r--r--packages/meshbay-hub/src/meshbay_hub/static/app.js26
-rw-r--r--packages/meshbay-hub/src/meshbay_hub/static/transport.js102
2 files changed, 96 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;