fix(#2458): [Bug] 新增的文件传输列表功能的暂停和全部暂定功能不生效

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 };
 }
 

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 () => {