1 files changed, 95 insertions, 10 deletions

diff --git a/packages/meshbay-hub/src/meshbay_hub/static/transfers.js b/packages/meshbay-hub/src/meshbay_hub/static/transfers.js
index fa34c67..46f75a2 100644
--- a/packages/meshbay-hub/src/meshbay_hub/static/transfers.js
+++ b/packages/meshbay-hub/src/meshbay_hub/static/transfers.js
@@ -22,6 +22,12 @@
 
 const SPEED_WINDOW_MS = 5000;
 
+function _abortError() {
+  const err = new Error('Cancelled');
+  err.name = 'AbortError';
+  return err;
+}
+
 let _nextId = 1;
 
 export class TransferStore {
@@ -53,8 +59,18 @@ export class TransferStore {
       total: it.total,
       done: it.done,
       status: it.status,
+      // How many are in front of this one, and whose limit is holding it up:
+      // "your own two slots are busy" and "the node is full" are different
+      // situations and the person can act on only one of them.
+      ahead: it.ahead || 0,
+      queuedByOwnLimit: Boolean(
+        it.lease && it.lease.cap && it.lease.used >= it.lease.cap),
       error: it.error || '',
       speed: this._speed(it),
+      // The ETA is drawn only once the window holds a few seconds of real
+      // measurement -- see etaSeconds.
+      settled: it.samples.length > 2
+        && (it.samples[it.samples.length - 1].t - it.samples[0].t) >= 3000,
       percent: it.total ? Math.min(100, Math.round(it.done / it.total * 100)) : 0,
       // Only for a file written into a folder the browser granted us: that is
       // the one case where the page can read its own download back.
@@ -66,6 +82,12 @@ export class TransferStore {
     return this._items.filter(it => it.status === 'running').length;
   }
 
+  /** Running or waiting for a slot — what the nav badge counts. */
+  get pending() {
+    return this._items.filter(
+      it => it.status === 'running' || it.status === 'queued').length;
+  }
+
   _speed(it) {
     // Over a window rather than since the start: a transfer that stalls should
     // read as slow immediately, not as its own historical average.
@@ -82,12 +104,18 @@ export class TransferStore {
    * `run` receives `{ signal, onProgress }`. It must poll `signal.aborted` — a
    * cancel that only sets a flag nobody reads is a button that lies.
    */
-  start({ kind, name, total = 0, transport = null, run, open = null }) {
+  start({ kind, name, total = 0, transport = null, run, open = null,
+          lease = null }) {
     const item = {
       id: _nextId++,
-      kind, name, total, transport, open,
+      kind, name, total, transport, open, lease,
       done: 0,
-      status: 'running',
+      // A transfer that has to wait for a slot starts as 'queued', not
+      // 'running'. Two different things are true of it — nothing is moving, and
+      // nothing is wrong — and a status that conflates them is what makes a
+      // queue look like a hang.
+      status: lease && lease.state !== 'granted' ? 'queued' : 'running',
+      ahead: (lease && lease.ahead) || 0,
       error: '',
       samples: [{ t: this._now(), done: 0 }],
       signal: { aborted: false },
@@ -113,8 +141,39 @@ export class TransferStore {
       this._maybeRelease(item.transport);
     };
 
+    // The slot is given back in a `finally` around everything, so it survives
+    // a throw, a cancel and a return alike. A slot not returned is a member who
+    // cannot start another transfer until the node times it out.
+    const finished = () => {
+      if (item.lease) item.lease.release(
+        item.signal.aborted ? 'cancelled' : 'done');
+    };
+
+    // Installed here and not inside the promise chain below. A state push that
+    // arrived before the first microtask ran was simply dropped, so a transfer
+    // could sit at the position it was given when it was created and never
+    // appear to move — the widget showing "3 ahead" for ever while the node
+    // quietly worked through the queue. Nothing about that looks wrong from
+    // either side, which is why it needs a test rather than a reading.
+    if (item.lease) {
+      item.lease._onState = (lease) => {
+        if (item.status !== 'queued' && item.status !== 'running') return;
+        item.ahead = lease.ahead;
+        item.status = lease.state === 'granted' ? 'running' : 'queued';
+        this._emit();
+      };
+    }
+
     const promise = Promise.resolve()
-      .then(() => run({ signal: item.signal, onProgress }))
+      .then(async () => {
+        if (item.lease) {
+          await item.lease.acquire();
+          if (item.signal.aborted) throw _abortError();
+          item.status = 'running';
+          this._emit();
+        }
+        return run({ signal: item.signal, onProgress, lease: item.lease });
+      })
       .then(() => {
         if (item.signal.aborted) finish('cancelled');
         else {
@@ -125,7 +184,8 @@ export class TransferStore {
       .catch(err => {
         if (item.signal.aborted || err.name === 'AbortError') finish('cancelled');
         else finish('failed', err.message || String(err));
-      });
+      })
+      .finally(finished);
 
     item.promise = promise;
     return item.id;
@@ -146,8 +206,12 @@ export class TransferStore {
 
   cancel(id) {
     const item = this._items.find(it => it.id === id);
-    if (!item || item.status !== 'running') return;
+    // 'queued' too: a transfer waiting for a slot is exactly the one somebody
+    // is most likely to give up on, and its queue entry has to go with it or
+    // the node grants a slot to a transfer that will never use it.
+    if (!item || (item.status !== 'running' && item.status !== 'queued')) return;
     item.signal.aborted = true;
+    if (item.lease) item.lease.release('cancelled');
     // Marked at once. The work stops when it next looks, but a cancelled
     // transfer should not keep reporting progress in the meantime.
     item.status = 'cancelled';
@@ -157,19 +221,24 @@ export class TransferStore {
 
   cancelAll() {
     for (const it of this._items) {
-      if (it.status === 'running') this.cancel(it.id);
+      if (it.status === 'running' || it.status === 'queued') this.cancel(it.id);
     }
   }
 
-  /** Drop everything finished, keeping what is still running. */
+  /** Drop everything finished, keeping what is still running or waiting. */
   clearFinished() {
-    this._items = this._items.filter(it => it.status === 'running');
+    this._items = this._items.filter(
+      it => it.status === 'running' || it.status === 'queued');
     this._emit();
   }
 
   _busy(transport) {
+    // Queued counts as busy: a transport closed while a transfer waits for a
+    // slot can never be granted one, and the transfer would sit at "waiting"
+    // for ever with nothing left to answer it.
     return this._items.some(
-      it => it.transport === transport && it.status === 'running');
+      it => it.transport === transport
+        && (it.status === 'running' || it.status === 'queued'));
   }
 
   /**
@@ -211,6 +280,22 @@ export class TransferStore {
 
 export const transfers = new TransferStore();
 
+/**
+ * Seconds left, or null when saying nothing is the honest answer.
+ *
+ * Withheld until the speed window has real samples in it: a figure computed
+ * from the first two chunks of a transfer swings between "4 seconds" and "an
+ * hour" and back, and a number that behaves like that is worse than a blank —
+ * people read the first one they see and plan around it.
+ */
+export function etaSeconds(item) {
+  if (item.status !== 'running' || !item.total || !item.speed) return null;
+  const left = item.total - item.done;
+  if (left <= 0) return null;
+  const secs = left / item.speed;
+  return Number.isFinite(secs) ? secs : null;
+}
+
 /** Human-readable rate, for a widget that updates several times a second. */
 export function formatSpeed(bytesPerSecond) {
   if (!bytesPerSecond || bytesPerSecond < 1) return '';