1 files changed, 30 insertions, 11 deletions

diff --git a/packages/meshbay-hub/src/meshbay_hub/static/transport.js b/packages/meshbay-hub/src/meshbay_hub/static/transport.js
index 326aa8e..24c08b9 100644
--- a/packages/meshbay-hub/src/meshbay_hub/static/transport.js
+++ b/packages/meshbay-hub/src/meshbay_hub/static/transport.js
@@ -42,6 +42,12 @@ const UPLOAD_CHUNK_SIZE = 48 * 1024;
 const UPLOAD_WINDOW = 32;
 const UPLOAD_BUFFER_HIGH = 1024 * 1024;
 
+function _aborted() {
+  const err = new Error('Cancelled');
+  err.name = 'AbortError';
+  return err;
+}
+
 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).',
@@ -73,7 +79,10 @@ class MeshBayTransport {
     this._onStreamData = null;
     this._onStreamEnd = null;
     this._onIndexSync = null;
-    this._onUploadAck = null;
+    // filename → the uploader waiting on it. Keyed rather than FIFO because
+    // several uploads may be in flight at once and their acks interleave; the
+    // node names the file in every one.
+    this._uploaders = new Map();
   }
 
   get connected() { return this._connected; }
@@ -533,10 +542,12 @@ class MeshBayTransport {
    * 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');
+  async uploadFile(file, { chunkSize, onProgress, signal } = {}) {
+    // The same file twice at once would confuse the node, which keys its own
+    // upload state by name — and would race for the same destination.
+    if (this._uploaders.has(file.name)) {
+      throw new Error(`${file.name} is already being uploaded`);
+    }
     const size = chunkSize || UPLOAD_CHUNK_SIZE;
     const total = Math.max(1, Math.ceil(file.size / size));
     let acked = 0;
@@ -544,7 +555,7 @@ class MeshBayTransport {
     let failure = null;
 
     const acks = [];
-    this._onUploadAck = (msg) => {
+    this._uploaders.set(file.name, (msg) => {
       if (msg.type === 'error') {
         failure = new Error(msg.detail || 'Upload refused');
       } else if (msg.stored_as) {
@@ -554,15 +565,17 @@ class MeshBayTransport {
       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++) {
+        if (signal && signal.aborted) throw _aborted();
         // 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) {
@@ -587,7 +600,7 @@ class MeshBayTransport {
         if (failure) throw failure;
       }
     } finally {
-      this._onUploadAck = null;
+      this._uploaders.delete(file.name);
     }
     return stored || {};
   }
@@ -772,9 +785,15 @@ class MeshBayTransport {
     // 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);
+    if (msg.type === 'file_upload_ack' && this._uploaders.has(msg.filename)) {
+      this._uploaders.get(msg.filename)(msg);
+      return;
+    }
+    // An error carries no filename. With one upload running it is that
+    // upload's; with several there is no way to tell, so they all hear it and
+    // stop — which is the safe reading of an error on a shared channel.
+    if (msg.type === 'error' && this._uploaders.size) {
+      for (const handler of [...this._uploaders.values()]) handler(msg);
       return;
     }
     if (msg.type === 'chat_msg' && this._onChat) {