diff --git a/electron/bridges/sshBridge/startSession.cjs b/electron/bridges/sshBridge/startSession.cjs index 4bfca29f9..49bad79bb 100644 --- a/electron/bridges/sshBridge/startSession.cjs +++ b/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 */ } }, diff --git a/electron/bridges/zmodemHelper.cjs b/electron/bridges/zmodemHelper.cjs index ede558b15..551b40e3b 100644 --- a/electron/bridges/zmodemHelper.cjs +++ b/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,66 @@ 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, + * getTransportPendingBytes?: () => unknown, + * progressThrottleMs?: number, + * }} opts + * @param {() => void} emitProgress Called at most once per throttle interval + * while the drain wait stays pending. + * @returns {Promise} + */ +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 */ } + while (!settled) { + await new Promise((resolve) => setTimeout(resolve, intervalMs)); + 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 +1396,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 +1459,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 +1478,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 +1772,8 @@ module.exports = { handleDownload, waitForWritableDrain, createZmodemUploadDrainWaiter, + awaitUploadDrainWithProgress, UPLOAD_CHUNK_SIZE, UPLOAD_DRAIN_TIMEOUT_MS, + UPLOAD_DRAIN_PROGRESS_INTERVAL_MS, }; diff --git a/electron/bridges/zmodemHelper.test.cjs b/electron/bridges/zmodemHelper.test.cjs index 14c1cb25a..8d6698804 100644 --- a/electron/bridges/zmodemHelper.test.cjs +++ b/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,210 @@ 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 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");