1 files changed, 22 insertions, 3 deletions

diff --git a/packages/meshbay-hub/src/meshbay_hub/static/file-utils.js b/packages/meshbay-hub/src/meshbay_hub/static/file-utils.js
index 294455a..585558f 100644
--- a/packages/meshbay-hub/src/meshbay_hub/static/file-utils.js
+++ b/packages/meshbay-hub/src/meshbay_hub/static/file-utils.js
@@ -48,6 +48,11 @@ const CHUNK_SIZE = 1024 * 1024;
 // fragmented MP4. A stream that arrives in gulps has no margin for a network
 // that hesitates, and looks like a hang while it is quiet.
 const PIPELINE_WINDOW = 8;
+// A leased download with at least this much left opens the transport's extra
+// connections (`lanes` in transport.js). Below it the handshakes, half a second
+// to a second each, cost more than they save; lanes already open are used
+// whatever the size.
+const LANE_MIN_CHUNKS = 64;
 
 // The most this code will ever collect in the page.
 //
@@ -392,10 +397,24 @@ async function pipelinedDownload(transport, gekKey, fileId, totalChunks, onChunk
   let nextSend = fromChunk, nextRecv = fromChunk;
   const inflight = new Array(totalChunks);
 
+  // Chunk i goes over lane i modulo the lanes open at that moment, two chunks
+  // in flight per lane so none waits on the slowest. Order is restored below,
+  // where each chunk is awaited by its index. A lane that fails a request is
+  // dropped and the chunk asked for again on the main connection, which has
+  // the retries and the reconnect.
   const fire = () => {
-    while (nextSend < totalChunks && nextSend - nextRecv < PIPELINE_WINDOW) {
-      inflight[nextSend] = _fetchChunkResilient(transport, fileId, nextSend, tr,
-                                                signal);
+    const lanes = (tr && transport.lanes)
+      ? transport.lanes(totalChunks - nextRecv >= LANE_MIN_CHUNKS) : [transport];
+    const depth = Math.max(PIPELINE_WINDOW, 2 * lanes.length);
+    while (nextSend < totalChunks && nextSend - nextRecv < depth) {
+      const index = nextSend;
+      const lane = lanes[index % lanes.length];
+      inflight[index] = lane === transport
+        ? _fetchChunkResilient(transport, fileId, index, tr, signal)
+        : lane.fetchChunk(fileId, index, tr).catch(() => {
+          transport.dropLane(lane);
+          return _fetchChunkResilient(transport, fileId, index, tr, signal);
+        });
       nextSend++;
     }
   };