fix(zmodem): race drain against progress poll timer

awaitUploadDrainWithProgress slept out the rest of each ~100ms progress
poll after transport drain settled, throttling fast uploads on every
ssh2 backpressure cycle. Race the poll delay against drain and clear
the timer when the wait settles.

Co-authored-by: 陈大猫 <binaricat@users.noreply.github.com>

electron/bridges/sshBridge/startSession.cjs

@@ -548,6 +548,15 @@ function createStartSessionApi(ctx) {
             ),
           });
         },
+        // Upload progress tracks bytes the transport actually accepted instead
+        // of bytes queued (issue #3663): ssh2 keeps writableLength unchanged
+        // until one queued write fully completes but shrinks _chunk as the
+        // channel window opens, so summing both approximates the outstanding
+        // backlog while it drains. Without this, an rz progress bar freezes
+        // mid-file for as long as the channel window takes to flush.
+        getTransportPendingBytes() {
+          return (Number(stream.writableLength) || 0) + (stream._chunk?.length || 0);
+        },
         interruptRemote() {
           try { stream.signal?.("INT"); } catch { /* ignore */ }
         },

electron/bridges/zmodemHelper.cjs

@@ -107,6 +107,11 @@ function buildModeRestores(dir, names, removeIndices, modes) {
  *   Optional. When writeToRemote returns false, the upload loop awaits this
  *   before sending more file data (typically waitForWritableDrain(stream)).
  *   Callers should forward `opts.signal` so cancel can abort a blocked drain.
+ * @param {() => number} [opts.getTransportPendingBytes]
+ *   Optional. Return the bytes the transport has queued but not delivered yet
+ *   (e.g. the SSH channel backlog). Enables delivered-byte upload progress:
+ *   while parked on backpressure the loop reports queued-minus-outstanding
+ *   bytes so the progress indicator tracks actual delivery (issue #3663).
  * @param {() => import('electron').WebContents | null} opts.getWebContents
  *   Returns the Electron WebContents for sending progress IPC events.
  * @param {string} [opts.label]
@@ -912,6 +917,8 @@ const UPLOAD_DRAIN_TIMEOUT_MS = 60000;
 const UPLOAD_CHUNK_SIZE = 64 * 1024;
 /** Default interval between non-final upload progress IPC events. */
 const DEFAULT_UPLOAD_PROGRESS_THROTTLE_MS = 100;
+/** Interval between delivered-byte progress reports while parked on drain. */
+const UPLOAD_DRAIN_PROGRESS_INTERVAL_MS = 100;
 
 function resolveTimeoutMs(value, fallback) {
   return Number.isFinite(value) && value >= 0 ? value : fallback;
@@ -1127,6 +1134,76 @@ function createZmodemUploadDrainWaiter(opts) {
   };
 }
 
+/**
+ * Park the upload loop on transport backpressure while keeping progress
+ * updates alive (issue #3663).
+ *
+ * Once a large write backlog fills the transport, the upload loop otherwise
+ * sits silent on the "drain" event while the transport keeps delivering the
+ * queued backlog at full link speed: on SSH the entire channel window drains
+ * between two upload-loop steps, so the progress indicator freezes mid-file
+ * and jumps when the loop resumes. While the drain wait stays pending, poll
+ * it and let the caller emit progress events showing bytes the transport has
+ * actually accepted instead.
+ *
+ * Falls back to a plain drain wait when the transport cannot report its
+ * outstanding byte count. Rejections (timeout, cancellation) propagate.
+ *
+ * @param {{
+ *   waitForDrain?: () => Promise<void>,
+ *   getTransportPendingBytes?: () => unknown,
+ *   progressThrottleMs?: number,
+ * }} opts
+ * @param {() => void} emitProgress Called at most once per throttle interval
+ *   while the drain wait stays pending.
+ * @returns {Promise<void>}
+ */
+async function awaitUploadDrainWithProgress(opts, emitProgress) {
+  if (!opts.waitForDrain) return;
+  if (typeof opts.getTransportPendingBytes !== "function") {
+    await opts.waitForDrain();
+    return;
+  }
+
+  let settled = false;
+  const drain = opts.waitForDrain().finally(() => {
+    settled = true;
+  });
+  drain.catch(() => {}); // observed at the final await; never surface early
+  // A drain wait without pending backpressure resolves after one event-loop
+  // turn; yield once so backpressure-free uploads do not gain polling delay.
+  await new Promise((resolve) => setImmediate(resolve));
+  if (settled) {
+    await drain;
+    return;
+  }
+
+  const intervalMs = Math.max(
+    UPLOAD_DRAIN_PROGRESS_INTERVAL_MS,
+    resolveProgressThrottleMs(opts.progressThrottleMs),
+  );
+  // Parked: report accepted bytes right away, then once per interval while
+  // the transport drains its backlog. Progress emissions must never break
+  // (or slow) the transfer itself.
+  try { emitProgress(); } catch { /* ignore */ }
+  // Race each poll delay against drain. A wait that settles mid-interval
+  // must resume the upload immediately; sleeping out the rest of the
+  // interval throttles every backpressure cycle by up to ~100ms.
+  while (!settled) {
+    let timer;
+    await Promise.race([
+      drain.then(() => {}, () => {}),
+      new Promise((resolve) => {
+        timer = setTimeout(resolve, intervalMs);
+      }),
+    ]);
+    clearTimeout(timer);
+    if (settled) break;
+    try { emitProgress(); } catch { /* ignore */ }
+  }
+  await drain;
+}
+
 /**
  * Resolve when `promise` settles, or reject with NETCATTY_ZMODEM_CANCELLED
  * if `signal` aborts first. Used so cancel unblocks upload drain waits even
@@ -1329,19 +1406,42 @@ async function handleUpload(zsession, opts) {
     const { originalIndex, filePath, stat, name } = offers[i];
     opts.resetUploadBackpressure?.();
 
-    safeSend(getWebContents(), "netcatty:zmodem:progress", {
-      sessionId,
-      filename: name,
-      transferred: 0,
-      total: stat.size,
-      fileIndex: i,
-      fileCount: offers.length,
-      transferType: "upload",
-    });
-
     let bytesRemaining = 0;
     for (let j = i; j < offers.length; j++) bytesRemaining += offers[j].stat.size;
 
+    // Progress should track bytes the transport accepted, not just bytes
+    // queued (#3663): on SSH the channel can buffer its whole window before
+    // write() returns false, so queued-only reporting freezes the progress
+    // bar while the backlog drains at link speed. When the transport can
+    // report outstanding bytes, show `sent - outstanding` and never move
+    // the bar backwards.
+    const getOutstandingBytes = typeof opts.getTransportPendingBytes === "function"
+      ? opts.getTransportPendingBytes
+      : null;
+    let sent = 0;
+    let displayedTransferred = 0;
+    const emitFileProgress = () => {
+      const outstanding = getOutstandingBytes ? Number(getOutstandingBytes()) : NaN;
+      const delivered = Number.isFinite(outstanding)
+        ? sent - Math.max(outstanding, 0)
+        : sent;
+      displayedTransferred = Math.min(
+        Math.max(delivered, displayedTransferred),
+        stat.size,
+      );
+      safeSend(getWebContents(), "netcatty:zmodem:progress", {
+        sessionId,
+        filename: name,
+        transferred: displayedTransferred,
+        total: stat.size,
+        fileIndex: i,
+        fileCount: offers.length,
+        transferType: "upload",
+      });
+    };
+
+    emitFileProgress();
+
     // The offer handshake is the only upload step without a built-in
     // deadline: if rz dies before answering ZFILE (crash, YMODEM fallback),
     // send_offer() would park this loop forever and block all future
@@ -1369,7 +1469,6 @@ async function handleUpload(zsession, opts) {
     // Read and send in chunks
     const fd = fs.openSync(filePath, "r");
     const buf = Buffer.alloc(UPLOAD_CHUNK_SIZE);
-    let sent = 0;
     // Progress IPC is throttled by default for every transport. Pass 0 only
     // when a caller explicitly needs one event per chunk.
     const progressThrottleMs = resolveProgressThrottleMs(opts.progressThrottleMs);
@@ -1389,21 +1488,16 @@ async function handleUpload(zsession, opts) {
 
         const now = Date.now();
         if (progressThrottleMs === 0 || now - lastProgressEmitAt >= progressThrottleMs) {
-          safeSend(getWebContents(), "netcatty:zmodem:progress", {
-            sessionId,
-            filename: name,
-            transferred: sent,
-            total: stat.size,
-            fileIndex: i,
-            fileCount: offers.length,
-            transferType: "upload",
-          });
+          emitFileProgress();
           lastProgressEmitAt = now;
         }
 
-        // Wait for transport to drain if its buffer is full, then yield
-        // so inbound ZMODEM control frames can be processed.
-        if (opts.waitForDrain) await opts.waitForDrain();
+        // Wait for transport to drain if its buffer is full (emitting
+        // delivered-byte progress while parked, #3663), then yield so
+        // inbound ZMODEM control frames can be processed.
+        if (opts.waitForDrain) {
+          await awaitUploadDrainWithProgress(opts, emitFileProgress);
+        }
         await yieldToIO();
       }
       // All data written to Node.js buffer — but TCP may still be
@@ -1688,6 +1782,8 @@ module.exports = {
   handleDownload,
   waitForWritableDrain,
   createZmodemUploadDrainWaiter,
+  awaitUploadDrainWithProgress,
   UPLOAD_CHUNK_SIZE,
   UPLOAD_DRAIN_TIMEOUT_MS,
+  UPLOAD_DRAIN_PROGRESS_INTERVAL_MS,
 };

electron/bridges/zmodemHelper.test.cjs

@@ -11,6 +11,7 @@ const {
   handleDownload,
   waitForWritableDrain,
   createZmodemUploadDrainWaiter,
+  awaitUploadDrainWithProgress,
   UPLOAD_CHUNK_SIZE,
   UPLOAD_DRAIN_TIMEOUT_MS,
 } = require("./zmodemHelper.cjs");
@@ -422,6 +423,282 @@ test("handleUpload does not read the next chunk until transport backpressure cle
   fs.rmSync(tempDir, { recursive: true, force: true });
 });
 
+test("handleUpload reports transport-accepted bytes instead of the queued backlog", async () => {
+  const tempDir = fs.mkdtempSync(path.join(os.tmpdir(), "netcatty-zmodem-delivered-"));
+  const filePath = path.join(tempDir, "upload.bin");
+  const size = UPLOAD_CHUNK_SIZE * 2;
+  fs.writeFileSync(filePath, Buffer.alloc(size, 0x5a));
+  const progress = [];
+
+  const zsession = {
+    async send_offer() {
+      return {
+        send() {},
+        async end() {},
+      };
+    },
+    async close() {},
+  };
+
+  await handleUpload(zsession, {
+    sessionId: "session-1",
+    getWebContents: () => ({
+      isDestroyed: () => false,
+      send(channel, data) {
+        if (channel === "netcatty:zmodem:progress") progress.push(data);
+      },
+    }),
+    takeDragDropUpload: () => ({
+      filePaths: [filePath],
+      remoteNames: ["upload.bin"],
+    }),
+    progressThrottleMs: 0,
+    // 4096 bytes of every queued amount are still outstanding in the transport.
+    getTransportPendingBytes: () => 4096,
+  });
+
+  assert.deepEqual(
+    progress.map((event) => [event.transferred, Boolean(event.finalizing)]),
+    [
+      [0, false],
+      [UPLOAD_CHUNK_SIZE - 4096, false],
+      [size - 4096, false],
+      [size, true],
+    ],
+  );
+  fs.rmSync(tempDir, { recursive: true, force: true });
+});
+
+test("handleUpload emits delivered-byte progress while parked on transport drain", async () => {
+  const tempDir = fs.mkdtempSync(path.join(os.tmpdir(), "netcatty-zmodem-park-"));
+  const filePath = path.join(tempDir, "upload.bin");
+  const size = UPLOAD_CHUNK_SIZE * 2 + 1;
+  fs.writeFileSync(filePath, Buffer.alloc(size, 0x5a));
+  const progress = [];
+  // Bytes queued into the transport (never consumed; delivery is simulated
+  // below on the pending-byte readout).
+  let outstanding = 0;
+  let drainStartedAt = null;
+  let releaseFirstDrain;
+  const firstDrain = new Promise((resolve) => {
+    releaseFirstDrain = resolve;
+  });
+  let drainCalls = 0;
+
+  const zsession = {
+    async send_offer() {
+      return {
+        send(chunk) {
+          outstanding += chunk.byteLength;
+        },
+        async end() {},
+      };
+    },
+    async close() {},
+  };
+
+  const upload = handleUpload(zsession, {
+    sessionId: "session-1",
+    getWebContents: () => ({
+      isDestroyed: () => false,
+      send(channel, data) {
+        if (channel === "netcatty:zmodem:progress") progress.push(data);
+      },
+    }),
+    takeDragDropUpload: () => ({
+      filePaths: [filePath],
+      remoteNames: ["upload.bin"],
+    }),
+    progressThrottleMs: 0,
+    getTransportPendingBytes: () => {
+      if (drainStartedAt === null) return outstanding;
+      // The backlog drains at 512 B/ms once the parked wait begins.
+      const delivered = Math.min(outstanding, (Date.now() - drainStartedAt) * 512);
+      return outstanding - delivered;
+    },
+    async waitForDrain() {
+      drainCalls += 1;
+      if (drainCalls === 1) {
+        drainStartedAt = Date.now();
+        await firstDrain;
+      }
+    },
+  });
+
+  while (drainCalls === 0) {
+    await new Promise((resolve) => setImmediate(resolve));
+  }
+  const progressedBefore = progress.length;
+  await new Promise((resolve) => setTimeout(resolve, 250));
+  const parkedEvents = progress.slice(progressedBefore);
+  releaseFirstDrain();
+  await upload;
+
+  // While parked, delivered-byte progress must keep flowing (previously the
+  // indicator stayed frozen until the whole backlog drained).
+  assert.ok(
+    parkedEvents.some((event) => event.transferred > 0 && !event.finalizing),
+    `expected delivered-byte emissions while parked, got ${JSON.stringify(parkedEvents.map((e) => [e.transferred, Boolean(e.finalizing)]))}`,
+  );
+
+  const values = progress.map((event) => event.transferred);
+  for (let k = 1; k < values.length; k++) {
+    assert.ok(
+      values[k] >= values[k - 1],
+      `progress must never move backwards: ${values.join(",")}`,
+    );
+    assert.ok(values[k] <= size, `progress must not exceed the file size: ${values.join(",")}`);
+  }
+  assert.equal(progress.at(-1).finalizing, true);
+  fs.rmSync(tempDir, { recursive: true, force: true });
+});
+
+test("awaitUploadDrainWithProgress falls back to a plain drain wait without pending-byte accounting", async () => {
+  let drainCalls = 0;
+  await awaitUploadDrainWithProgress(
+    {
+      waitForDrain: () => {
+        drainCalls += 1;
+        return Promise.resolve();
+      },
+    },
+    () => {
+      throw new Error("no progress emissions without pending-byte accounting");
+    },
+  );
+  assert.equal(drainCalls, 1);
+});
+
+test("awaitUploadDrainWithProgress adds no polling delay when drain resolves immediately", async () => {
+  const startedAt = Date.now();
+  await awaitUploadDrainWithProgress(
+    {
+      waitForDrain: () => Promise.resolve(),
+      getTransportPendingBytes: () => 0,
+    },
+    () => {
+      throw new Error("no emissions without parked backpressure");
+    },
+  );
+  assert.ok(Date.now() - startedAt < 50);
+});
+
+test("awaitUploadDrainWithProgress returns promptly when drain settles during the poll interval", async () => {
+  let releaseDrain;
+  const drainGate = new Promise((resolve) => {
+    releaseDrain = resolve;
+  });
+  let parked = false;
+  const done = awaitUploadDrainWithProgress(
+    {
+      waitForDrain: () => drainGate,
+      getTransportPendingBytes: () => 2048,
+    },
+    () => {
+      parked = true;
+    },
+  );
+
+  const waitStarted = Date.now();
+  while (!parked) {
+    if (Date.now() - waitStarted > 1000) {
+      throw new Error("upload drain wait never parked");
+    }
+    await new Promise((resolve) => setImmediate(resolve));
+  }
+  // The immediate park emission has run and the poll timer (~100ms) is armed.
+  // Settle on the next turn, still inside that interval.
+  await new Promise((resolve) => setImmediate(resolve));
+  const releasedAt = Date.now();
+  releaseDrain();
+  await done;
+  const elapsed = Date.now() - releasedAt;
+  assert.ok(
+    elapsed < 50,
+    `drain settlement during a poll interval resumed in ${elapsed}ms`,
+  );
+});
+
+test("awaitUploadDrainWithProgress propagates a rejection that arrives during the poll interval", async () => {
+  let rejectDrain;
+  const drainGate = new Promise((_, reject) => {
+    rejectDrain = reject;
+  });
+  const err = new Error("Transport drain timeout");
+  err.code = "NETCATTY_ZMODEM_TIMEOUT";
+  let parked = false;
+  const done = awaitUploadDrainWithProgress(
+    {
+      waitForDrain: () => drainGate,
+      getTransportPendingBytes: () => 2048,
+    },
+    () => {
+      parked = true;
+    },
+  );
+
+  const waitStarted = Date.now();
+  while (!parked) {
+    if (Date.now() - waitStarted > 1000) {
+      throw new Error("upload drain wait never parked");
+    }
+    await new Promise((resolve) => setImmediate(resolve));
+  }
+  await new Promise((resolve) => setImmediate(resolve));
+  const releasedAt = Date.now();
+  rejectDrain(err);
+  await assert.rejects(done, (caught) => caught === err);
+  const elapsed = Date.now() - releasedAt;
+  assert.ok(
+    elapsed < 50,
+    `drain rejection during a poll interval resumed in ${elapsed}ms`,
+  );
+});
+
+test("awaitUploadDrainWithProgress emits progress while parked until drain settles", async () => {
+  let releaseDrain;
+  const drainGate = new Promise((resolve) => {
+    releaseDrain = resolve;
+  });
+  let displayed = 0;
+  const emitted = [];
+  const deliverTimer = setInterval(() => {
+    displayed += 512;
+  }, 40);
+
+  try {
+    const parked = awaitUploadDrainWithProgress(
+      {
+        waitForDrain: () => drainGate,
+        getTransportPendingBytes: () => 1024,
+      },
+      () => emitted.push(displayed),
+    );
+    await new Promise((resolve) => setTimeout(resolve, 280));
+    assert.ok(emitted.length >= 2, `expected poll emissions, got ${emitted.length}`);
+    assert.ok(emitted.every((v) => v === 0 || v % 512 === 0));
+    releaseDrain();
+    await parked;
+  } finally {
+    clearInterval(deliverTimer);
+  }
+});
+
+test("awaitUploadDrainWithProgress propagates drain-wait rejections", async () => {
+  const err = new Error("Transport drain timeout");
+  err.code = "NETCATTY_ZMODEM_TIMEOUT";
+  await assert.rejects(
+    awaitUploadDrainWithProgress(
+      {
+        waitForDrain: () => Promise.reject(err),
+        getTransportPendingBytes: () => 0,
+      },
+      () => {},
+    ),
+    (caught) => caught === err,
+  );
+});
+
 test("handleUpload throttles progress IPC by default for every transport", async () => {
   const tempDir = fs.mkdtempSync(path.join(os.tmpdir(), "netcatty-zmodem-progress-"));
   const filePath = path.join(tempDir, "large-upload.bin");