diff --git a/electron/bridges/transferBridge.cjs b/electron/bridges/transferBridge.cjs index b98f10fe0..ac2d628ea 100644 --- a/electron/bridges/transferBridge.cjs +++ b/electron/bridges/transferBridge.cjs @@ -797,8 +797,13 @@ async function uploadFile(localPath, remotePath, client, fileSize, transfer, sen transfer.readStream = readStream; transfer.writeStream = writeStream; // Honor a pause that raced stream open (fingerprint / dir ensure window). + // Do not pipe while paused — Node pipe auto-resumes on drain. if (transfer.paused) { try { readStream.pause(); } catch { } + transfer.streamsUnpiped = true; + } else { + readStream.pipe(writeStream); + transfer.streamsUnpiped = false; } const cleanup = (err) => { @@ -833,7 +838,6 @@ async function uploadFile(localPath, remotePath, client, fileSize, transfer, sen } cleanup(null); }); - readStream.pipe(writeStream); }); await assertRemoteUploadSize(client, remotePath, fileSize); if (initialSource) { @@ -1572,6 +1576,10 @@ async function downloadFile(remotePath, localPath, client, fileSize, transfer, s transfer.writeStream = writeStream; if (transfer.paused) { try { readStream.pause(); } catch { } + transfer.streamsUnpiped = true; + } else { + readStream.pipe(writeStream); + transfer.streamsUnpiped = false; } const cleanup = (err) => { @@ -1607,7 +1615,6 @@ async function downloadFile(remotePath, localPath, client, fileSize, transfer, s writeStream.on('close', () => { if (transfer.cancelled) cleanup(new Error('Transfer cancelled')); }); - readStream.pipe(writeStream); }); if (initialSource) { const latestSource = await client.stat(remotePath); @@ -1664,6 +1671,9 @@ async function startTransferNow(event, payload, onProgress) { targetEncoding, readStream: null, writeStream: null, + // True after unpipe (or when open skipped pipe while paused). Guards + // resumeStreamPair against duplicate pipe() which doubles writes. + streamsUnpiped: false, abort: null, }; activeTransfers.set(transferId, transfer); @@ -1993,6 +2003,10 @@ async function startTransferNow(event, payload, onProgress) { transfer.writeStream = writeStream; if (transfer.paused) { try { readStream.pause(); } catch { } + transfer.streamsUnpiped = true; + } else { + readStream.pipe(writeStream); + transfer.streamsUnpiped = false; } const cleanup = (err) => { @@ -2028,7 +2042,6 @@ async function startTransferNow(event, payload, onProgress) { writeStream.on('close', () => { if (transfer.cancelled) cleanup(new Error('Transfer cancelled')); }); - readStream.pipe(writeStream); }); if (transfer.resumable && transfer.stagedLocalPath) { await promoteLocalTransfer(transfer.stagedLocalPath, targetPath); @@ -2410,6 +2423,19 @@ function clearPendingCancel(transferId) { if (transferId) pendingCancelTransferIds.delete(String(transferId)); } +/** + * Re-attach a paused stream pair and continue reading. + * pauseTransfer unpipes so destination drain cannot auto-resume the source. + * Idempotent: Node does not dedupe pipe(), so only re-pipe while unpiped. + */ +function resumeStreamPair(transfer) { + if (transfer.readStream && transfer.writeStream && transfer.streamsUnpiped) { + try { transfer.readStream.pipe(transfer.writeStream); } catch { } + transfer.streamsUnpiped = false; + } + try { transfer.readStream?.resume?.(); } catch { } +} + async function pauseTransfer(_event, payload) { const queuedResult = pauseQueuedTransfer(payload?.transferId); if (queuedResult) return queuedResult; @@ -2437,6 +2463,14 @@ async function pauseTransfer(_event, payload) { transfer.pauseSuperseded = false; const pauseOperation = (async () => { transfer.paused = true; + // Stream transfers use readStream.pipe(writeStream). Node's pipe resumes the + // source on destination 'drain', and pauseTransfer waits for that drain to + // flush durable bytes — so pause() alone is undone and upload continues while + // the UI shows paused. Unpipe first; resumeTransfer re-pipes. + if (transfer.readStream && transfer.writeStream) { + try { transfer.readStream.unpipe?.(transfer.writeStream); } catch { } + transfer.streamsUnpiped = true; + } try { transfer.readStream?.pause?.(); } catch { } const usesContiguousRangeCheckpoint = typeof transfer.waitForPause === "function"; if (usesContiguousRangeCheckpoint) { @@ -2494,7 +2528,7 @@ async function pauseTransfer(_event, payload) { } } catch { transfer.paused = false; - try { transfer.readStream?.resume?.(); } catch { } + resumeStreamPair(transfer); return { success: false, reason: "Could not verify the saved transfer checkpoint" }; } if (transfer.resumeStage === 'download') transfer.downloadCheckpointBytes = transfer.checkpointBytes; @@ -2509,7 +2543,7 @@ async function pauseTransfer(_event, payload) { }); } catch { transfer.paused = false; - try { transfer.readStream?.resume?.(); } catch { } + resumeStreamPair(transfer); return { success: false, reason: "Could not verify that the source is safe to resume" }; } } @@ -2557,9 +2591,13 @@ async function resumeTransfer(_event, payload) { if (currentTransfer !== transfer || transfer.cancelled) { return { success: false, reason: "Transfer is no longer active" }; } + // Already flowing (e.g. double-click resume): do not pipe() again. + if (!transfer.paused) { + return { success: true }; + } transfer.paused = false; transfer.pauseSuperseded = false; - try { transfer.readStream?.resume?.(); } catch { } + resumeStreamPair(transfer); return { success: true }; } diff --git a/electron/bridges/transferBridge.test.cjs b/electron/bridges/transferBridge.test.cjs index dbe88fbf7..f006e2c20 100644 --- a/electron/bridges/transferBridge.test.cjs +++ b/electron/bridges/transferBridge.test.cjs @@ -2291,6 +2291,215 @@ test("resumable stream transfers pause without losing their checkpoint and conti assert.equal((await running).error, undefined); }); +test("stream upload pause survives write-stream drain without auto-resuming the pipe", async (t) => { + // Regression for transfer-list pause: Node's .pipe() resumes the source on + // destination 'drain'. pauseTransfer also waits for drain to flush bytes, so + // without unpipe the upload keeps flowing after UI reports paused. + const tempDir = await fs.promises.mkdtemp(path.join(os.tmpdir(), "netcatty-transfer-pipe-pause-")); + t.after(async () => { + await fs.promises.rm(tempDir, { recursive: true, force: true }); + }); + + const payload = Buffer.alloc(256 * 1024, 91); + const localPath = path.join(tempDir, "upload.bin"); + await fs.promises.writeFile(localPath, payload); + + let durableBytes = 0; + const pendingWriteCallbacks = []; + let holdWrites = true; + const writeStream = new Writable({ + highWaterMark: 16, + write(chunk, _encoding, callback) { + durableBytes += chunk.length; + if (holdWrites) { + pendingWriteCallbacks.push(callback); + return; + } + setImmediate(callback); + }, + }); + + const streamSftp = createFastSftp({ + createWriteStream() { + return writeStream; + }, + }); + const client = { + sftp: streamSftp, + stat() { + return Promise.resolve({ size: durableBytes || payload.length }); + }, + rename() { + return Promise.resolve(); + }, + delete() { + return Promise.resolve(); + }, + // Force sequential stream path (no isolated fast channel). + client: { + sftp(callback) { + callback(new Error("isolated channel unavailable")); + }, + }, + }; + transferBridge.init({ sftpClients: new Map([["target", client]]) }); + + const sender = createSender(); + const running = transferBridge.startTransfer( + { sender }, + { + transferId: "upload-pipe-pause", + sourcePath: localPath, + targetPath: "/tmp/upload-pipe-pause.bin", + sourceType: "local", + targetType: "sftp", + targetSftpId: "target", + totalBytes: payload.length, + resumable: true, + }, + ); + + const backpressureDeadline = Date.now() + 2000; + while (pendingWriteCallbacks.length === 0 && Date.now() < backpressureDeadline) { + await new Promise((resolve) => setImmediate(resolve)); + } + assert.ok(pendingWriteCallbacks.length > 0, "write stream should be backpressured"); + const bytesWhenPauseRequested = durableBytes; + + const pausing = transferBridge.pauseTransfer(null, { transferId: "upload-pipe-pause" }); + // Release buffered writes so destination emits drain (the pipe auto-resume hazard). + holdWrites = false; + for (const callback of pendingWriteCallbacks.splice(0)) callback(); + const paused = await pausing; + assert.equal(paused.success, true); + const checkpoint = paused.checkpointBytes; + assert.ok(checkpoint >= bytesWhenPauseRequested); + assert.ok( + checkpoint < payload.length, + "pause must stop before the upload finishes under backpressure", + ); + + await new Promise((resolve) => setTimeout(resolve, 50)); + assert.equal( + durableBytes, + checkpoint, + "paused stream upload must not keep writing after drain", + ); + + assert.deepEqual( + await transferBridge.resumeTransfer(null, { transferId: "upload-pipe-pause" }), + { success: true }, + ); + assert.equal((await running).error, undefined); + assert.equal(durableBytes, payload.length); +}); + +test("repeated resume does not double-pipe the same stream pair", async (t) => { + // Node's Readable.pipe does not dedupe: a second pipe() on the same pair + // delivers each chunk twice and can corrupt/oversize staged copies. + const tempDir = await fs.promises.mkdtemp(path.join(os.tmpdir(), "netcatty-transfer-double-resume-")); + t.after(async () => { + await fs.promises.rm(tempDir, { recursive: true, force: true }); + }); + + const payload = Buffer.alloc(256 * 1024, 77); + const localPath = path.join(tempDir, "upload.bin"); + await fs.promises.writeFile(localPath, payload); + + let durableBytes = 0; + const pendingWriteCallbacks = []; + let holdWrites = true; + let pipeCount = 0; + const writeStream = new Writable({ + highWaterMark: 16, + write(chunk, _encoding, callback) { + durableBytes += chunk.length; + if (holdWrites) { + pendingWriteCallbacks.push(callback); + return; + } + setImmediate(callback); + }, + }); + const originalReadablePipe = Readable.prototype.pipe; + t.after(() => { + Readable.prototype.pipe = originalReadablePipe; + }); + Readable.prototype.pipe = function patchedPipe(dest, ...args) { + if (dest === writeStream) pipeCount += 1; + return originalReadablePipe.apply(this, [dest, ...args]); + }; + + const streamSftp = createFastSftp({ + createWriteStream() { + return writeStream; + }, + }); + const client = { + sftp: streamSftp, + stat() { + return Promise.resolve({ size: durableBytes || payload.length }); + }, + rename() { + return Promise.resolve(); + }, + delete() { + return Promise.resolve(); + }, + client: { + sftp(callback) { + callback(new Error("isolated channel unavailable")); + }, + }, + }; + transferBridge.init({ sftpClients: new Map([["target", client]]) }); + + const sender = createSender(); + const running = transferBridge.startTransfer( + { sender }, + { + transferId: "upload-double-resume", + sourcePath: localPath, + targetPath: "/tmp/upload-double-resume.bin", + sourceType: "local", + targetType: "sftp", + targetSftpId: "target", + totalBytes: payload.length, + resumable: true, + }, + ); + + const backpressureDeadline = Date.now() + 2000; + while (pendingWriteCallbacks.length === 0 && Date.now() < backpressureDeadline) { + await new Promise((resolve) => setImmediate(resolve)); + } + assert.ok(pendingWriteCallbacks.length > 0, "write stream should be backpressured"); + + const pausing = transferBridge.pauseTransfer(null, { transferId: "upload-double-resume" }); + holdWrites = false; + for (const callback of pendingWriteCallbacks.splice(0)) callback(); + const paused = await pausing; + assert.equal(paused.success, true); + const pipesAfterPause = pipeCount; + + assert.deepEqual( + await transferBridge.resumeTransfer(null, { transferId: "upload-double-resume" }), + { success: true }, + ); + assert.deepEqual( + await transferBridge.resumeTransfer(null, { transferId: "upload-double-resume" }), + { success: true }, + ); + assert.equal( + pipeCount, + pipesAfterPause + 1, + "second resume must not call pipe() again on the same pair", + ); + + assert.equal((await running).error, undefined); + assert.equal(durableBytes, payload.length); +}); + test("resumable downloads never promote a partial staged file", async (t) => { const tempDir = await fs.promises.mkdtemp(path.join(os.tmpdir(), "netcatty-transfer-partial-test-")); t.after(async () => {