feat(hub): client-side transfer leases and the transfers panel

Steps 5 and 6 of ~/next/improve-downloads.md. The node has handed out slots
since step 2 and nothing asked for one; now the client does, and the panel shows
what is happening.

`transport.openTransfer()` returns a Lease: `acquire()` resolves when the node
grants, `release()` gives it back exactly once, and nothing else in the client
speaks to the node about slots. Whether a node hands out slots is read from the
handshake ack rather than guessed from a timeout — "no answer yet" and "this
node will never answer" are indistinguishable in time, and guessing wrong either
stalls every download or defeats the cap.

Two things exist only because a queue can lie: a watchdog re-asks when a pushed
grant does not arrive (the node is idempotent on `tr`, so asking again is free),
and a grant for a transfer the page has forgotten is handed straight back rather
than held until the node's deadline.

The slot is asked for **after** there is somewhere to write, and that ordering
is load-bearing: opening a target takes thirty seconds of streamed-download
timeouts, or as long as somebody leaves a Save As dialog open, and a grant not
taken up in time is revoked. Moving it earlier looked better and broke three
downloads into one. Pinned by a test.

The panel groups by state — running, waiting, finished — rather than re-sorting a
flat list, so a row moves only when its own state does. The ETA is withheld until
the speed window holds real measurement: a figure from the first two chunks
swings between four seconds and an hour, and people plan around the first number
they see. One live region announces state changes and not progress.

Three silent paths closed on the way: a download refused for want of a user
gesture (a browser grants one file picker per gesture, and downloading three
files is one gesture) now falls back to the streamed path, which needs none; a
click with no connection says so instead of doing nothing at all; and a queued
transfer counts as busy, so a transport is never closed under one that is
waiting for a grant that could then never arrive.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01HCGdheDLxGReuKHga3BtST

1 files changed, 211 insertions, 2 deletions

diff --git a/packages/meshbay-hub/src/meshbay_hub/static/transport.js b/packages/meshbay-hub/src/meshbay_hub/static/transport.js
index c4a24c5..54a1302 100644
--- a/packages/meshbay-hub/src/meshbay_hub/static/transport.js
+++ b/packages/meshbay-hub/src/meshbay_hub/static/transport.js
@@ -283,6 +283,129 @@ const JOIN_REFUSALS = {
   group_mismatch: 'The node refused a request naming a different group.',
 };
 
+/**
+ * One transfer's slot on the node, from this side.
+ *
+ * The contract the transfer store depends on: `acquire()` resolves when the
+ * node has granted the slot (immediately on a node that hands out none), and
+ * `release(reason)` gives it back exactly once. Nothing else in the client
+ * speaks to the node about slots.
+ *
+ * Two things here exist only because a queue can lie, and both are the
+ * difference between "waiting" and "waiting for ever":
+ *
+ *   - **the watchdog.** A grant is pushed, not polled, so a lost push leaves
+ *     this side waiting on a node that believes it has started. Re-asking is
+ *     free — the node is idempotent on `tr` — and it is the only thing that
+ *     recovers a message that did not arrive.
+ *   - **release is idempotent and unconditional.** A slot given back twice
+ *     costs nothing; one never given back is a member who cannot transfer
+ *     again until a timeout the node runs on its own.
+ */
+const LEASE_WATCHDOG_MS = 60000;
+
+class Lease {
+  constructor(transport, tr, kind, bytes, chunks, onState) {
+    this.transport = transport;
+    this.tr = tr;
+    this.kind = kind;
+    this.bytes = bytes;
+    this.chunks = chunks;
+    this.state = 'opening';
+    this.ahead = 0;
+    this.closed = false;
+    this._onState = onState;
+    this._granted = null;
+    this._watchdog = 0;
+    this._wait = new Promise((resolve) => { this._granted = resolve; });
+  }
+
+  /** No slots on this node: behave as though one was granted at once. */
+  _skip() {
+    this.state = 'granted';
+    this._granted();
+  }
+
+  _request() {
+    // A closed channel is not a failure here, and must not throw: the transport
+    // reconnects on its own, `_reopenTransfers` re-asks for every live lease
+    // when it does, and the watchdog below asks again meanwhile.
+    //
+    // This is the same tolerance `_fetchChunkResilient` already gives a chunk
+    // request — and before leases existed, a chunk request was the first thing
+    // to touch the channel, so a download started on a briefly dead connection
+    // simply retried. Asking for a slot first made `_send` the first contact
+    // and threw "DataChannel not open (state: closed)" out of `downloadEntry`,
+    // where nothing catches it: a download that used to recover became an
+    // error with no row in the widget to show it. Found by downloading a file
+    // right after a connection dropped.
+    try {
+      this.transport._send({
+        type: 'transfer_open', v: '0.1', tr: this.tr, kind: this.kind,
+        bytes: this.bytes, chunks: this.chunks,
+      });
+    } catch (err) {
+      console.warn('[MeshBay] could not ask for a slot yet:', err.message);
+    }
+    this._arm();
+  }
+
+  _arm() {
+    clearTimeout(this._watchdog);
+    if (this.closed || this.state === 'granted') return;
+    this._watchdog = setTimeout(() => {
+      if (this.closed || this.state === 'granted') return;
+      console.warn('[MeshBay] no answer for transfer', this.tr.slice(0, 8),
+                   '- asking again');
+      this._request();
+    }, LEASE_WATCHDOG_MS);
+  }
+
+  _apply(msg) {
+    if (this.closed) return;
+    this.state = msg.state;
+    this.ahead = msg.ahead || 0;
+    this.used = msg.used;
+    this.cap = msg.cap;
+    if (msg.state === 'granted') {
+      clearTimeout(this._watchdog);
+      this._granted();
+    } else if (msg.state === 'closed') {
+      // The node ended it: reclaimed as idle, or revoked. Not an error here —
+      // whoever is running the transfer finds out through its own failure — but
+      // the slot is gone and asking again is the only way back.
+      clearTimeout(this._watchdog);
+    } else {
+      this._arm();
+    }
+    if (this._onState) {
+      try { this._onState(this); } catch (e) {
+        console.error('[MeshBay] lease state handler threw:', e);
+      }
+    }
+  }
+
+  /** Resolves once the node has granted the slot. */
+  acquire() { return this._wait; }
+
+  /**
+   * Give the slot back. Safe to call twice, and safe on a dead transport: a
+   * lease that is not released is a member who cannot start another transfer
+   * until the node times it out, so this must never be conditional on anything.
+   */
+  release(reason = 'done') {
+    if (this.closed) return;
+    this.closed = true;
+    clearTimeout(this._watchdog);
+    this.transport._leases.delete(this.tr);
+    if (!this.transport.supportsTransferSlots) return;
+    try {
+      this.transport._send({ type: 'transfer_close', v: '0.1', tr: this.tr,
+                             reason });
+    } catch { /* the connection is gone, and so is the lease with it */ }
+  }
+}
+
 class MeshBayTransport {
   constructor(hubUrl, accessToken) {
     this._hubUrl = hubUrl;
@@ -315,6 +438,13 @@ class MeshBayTransport {
     // Names, not ids: the "already being uploaded" guard is about the file the
     // caller passed, and two `uploadFile` calls for one file draw two ids.
     this._inFlightUploads = new Set();
+    // tr → Lease. A transfer's slot on the node, from the client's side.
+    this._leases = new Map();
+    // Set from the handshake ack: a node that answers with `transfer_limits`
+    // speaks transfer slots. Used instead of a timeout, because "no answer
+    // yet" and "this node will never answer" are indistinguishable in time and
+    // guessing wrong either stalls every download or defeats the cap.
+    this._transferLimits = null;
     // Set once close() runs — stops the automatic reconnect from firing on a
     // connection the caller tore down on purpose (leaving the group, page
     // unload), which would otherwise race back in right as everything else
@@ -377,6 +507,20 @@ class MeshBayTransport {
   get nodeVersion() { return this._nodeVersion || ''; }
 
   /**
+   * Whether this node hands out transfer slots.
+   *
+   * Read from the handshake ack rather than from the MNP version: the caps
+   * shipped before the version bump that will make leases compulsory, so for
+   * now a node either answers with `transfer_limits` or it predates all of
+   * this. A node that does not is asked for nothing and enforces nothing —
+   * every download behaves exactly as it did.
+   */
+  get supportsTransferSlots() { return this._transferLimits !== null; }
+
+  /** This member's own caps in this group, or null when the node said nothing. */
+  get transferLimits() { return this._transferLimits; }
+
+  /**
    * Whether the node speaks the per-root and per-app operations MNP 1.1 added:
    * `root_update`/`root_eject`/`root_plug`, `app_directories`,
    * `chat_directory`, `chat_link_preview`.
@@ -880,6 +1024,7 @@ class MeshBayTransport {
       delete ack.nonce;
       delete ack.ct;
       Object.assign(ack, config);
+      this._transferLimits = ack.transfer_limits || null;
 
       // Tell the node which of this account's devices is on this connection.
       // Deliberately after the ack, and gated on the node's own version rather
@@ -1016,6 +1161,10 @@ class MeshBayTransport {
         }
         trace('reconnect_ok', { attempt: this._reconnectAttempts });
         console.log('[MeshBay] Reconnected after', this._reconnectAttempts, 'attempt(s)');
+        // Before the caller's own hook: a transfer that resumes mid-chunk must
+        // have asked for its slot back first, or its next `file_req` carries a
+        // `tr` the node has never heard of.
+        this._reopenTransfers();
         if (this._onReconnected) {
           try { this._onReconnected(); } catch (e) {
             console.error('[MeshBay] onReconnected handler threw:', e);
@@ -1100,12 +1249,52 @@ class MeshBayTransport {
     return msg;
   }
 
-  async fetchChunk(fileId, chunkIndex) {
+  // ── Transfer slots ─────────────────────────────────────────────────────────
+
+  /**
+   * Ask the node for a slot, and wait until it says yes.
+   *
+   * `tr` is drawn here, not by the node — 16 random bytes, exactly like
+   * `upload_id` — which is what makes re-opening after a reconnect idempotent
+   * rather than a second charge against the member's cap.
+   *
+   * On a node that predates transfer slots this resolves at once and costs
+   * nothing: there is no cap to respect and no message that would be
+   * understood.
+   */
+  openTransfer({ kind = 'download', bytes = 0, chunks = 0, onState = null } = {}) {
+    const tr = _hex(crypto.getRandomValues(new Uint8Array(16)));
+    const lease = new Lease(this, tr, kind, bytes, chunks, onState);
+    if (!this.supportsTransferSlots) {
+      lease._skip();
+      return lease;
+    }
+    this._leases.set(tr, lease);
+    lease._request();
+    return lease;
+  }
+
+  /** Re-ask for every live lease. Called after a reconnect. */
+  _reopenTransfers() {
+    if (!this.supportsTransferSlots) return;
+    for (const lease of this._leases.values()) {
+      // The node lost the lease with the session, so this is a fresh request
+      // for the same `tr` — which the node treats as the same transfer rather
+      // than a second one.
+      if (!lease.closed) lease._request();
+    }
+  }
+
+  async fetchChunk(fileId, chunkIndex, tr = '') {
     const msg = await this._sendAndWait({
       type: 'file_req',
       v: '0.1',
       file_id: fileId,
       chunk_index: chunkIndex,
+      // Present only when this download holds a slot. The node does not require
+      // it yet; carrying it is what lets the node see the transfer is alive and
+      // not reclaim its slot as idle.
+      ...(tr ? { tr } : {}),
     });
     if (msg.type === 'error') throw new Error(msg.detail);
     return msg;
@@ -2189,7 +2378,8 @@ class MeshBayTransport {
    * 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 } = {}) {
+  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 name — and would race for the same destination. The guard
     // is by name for that reason, even though the map below is keyed by id.
@@ -2273,6 +2463,7 @@ class MeshBayTransport {
           upload_id: uploadId,
           chunk_index: i,
           total_chunks: total,
+          ...(tr ? { tr } : {}),
           ...sealed,
         });
       }
@@ -2992,6 +3183,24 @@ class MeshBayTransport {
     // "arrived" — and every message after that is one slot off too. Found
     // live: a group mid-scan corrupted its own handshake and chat history
     // this way, arriving roughly every 2s for as long as scanning ran.
+    // Routed by `tr`, and only by `tr`. A grant arrives unsolicited, minutes
+    // after the request that produced it, so falling through to "the oldest
+    // pending request" would hand a chat send or a handshake somebody else's
+    // slot — the class of defect `req_id` was introduced for.
+    if (msg.type === 'transfer_state') {
+      const lease = this._leases.get(msg.tr);
+      if (lease) lease._apply(msg);
+      else if (msg.state === 'granted') {
+        // A grant for a transfer this page has forgotten (a reload, a cancel
+        // that raced the grant). Handing it back at once matters: otherwise the
+        // node holds it until the 30 s acceptance deadline, and everyone behind
+        // it waits for nothing.
+        this._send({ type: 'transfer_close', v: '0.1', tr: msg.tr,
+                     reason: 'cancelled' });
+      }
+      return;
+    }
+
     if (msg.type === 'index_progress') {
       if (this._onIndexProgress) {
         this._onIndexProgress({