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

electron/bridges/transferBridge.cjs

@@ -797,8 +797,11 @@ 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 { }
+    } else {
+      readStream.pipe(writeStream);
     }
 
     const cleanup = (err) => {
@@ -833,7 +836,6 @@ async function uploadFile(localPath, remotePath, client, fileSize, transfer, sen
       }
       cleanup(null);
     });
-    readStream.pipe(writeStream);
   });
   await assertRemoteUploadSize(client, remotePath, fileSize);
   if (initialSource) {
@@ -1572,6 +1574,8 @@ async function downloadFile(remotePath, localPath, client, fileSize, transfer, s
     transfer.writeStream = writeStream;
     if (transfer.paused) {
       try { readStream.pause(); } catch { }
+    } else {
+      readStream.pipe(writeStream);
     }
 
     const cleanup = (err) => {
@@ -1607,7 +1611,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);
@@ -1993,6 +1996,8 @@ async function startTransferNow(event, payload, onProgress) {
         transfer.writeStream = writeStream;
         if (transfer.paused) {
           try { readStream.pause(); } catch { }
+        } else {
+          readStream.pipe(writeStream);
         }
 
         const cleanup = (err) => {
@@ -2028,7 +2033,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 +2414,17 @@ 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.
+ */
+function resumeStreamPair(transfer) {
+  if (transfer.readStream && transfer.writeStream) {
+    try { transfer.readStream.pipe(transfer.writeStream); } catch { }
+  }
+  try { transfer.readStream?.resume?.(); } catch { }
+}
+
 async function pauseTransfer(_event, payload) {
   const queuedResult = pauseQueuedTransfer(payload?.transferId);
   if (queuedResult) return queuedResult;
@@ -2437,6 +2452,13 @@ 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 { }
+  }
   try { transfer.readStream?.pause?.(); } catch { }
   const usesContiguousRangeCheckpoint = typeof transfer.waitForPause === "function";
   if (usesContiguousRangeCheckpoint) {
@@ -2494,7 +2516,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 +2531,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" };
     }
   }
@@ -2559,7 +2581,7 @@ async function resumeTransfer(_event, payload) {
   }
   transfer.paused = false;
   transfer.pauseSuperseded = false;
-  try { transfer.readStream?.resume?.(); } catch { }
+  resumeStreamPair(transfer);
   return { success: true };
 }
 

electron/bridges/transferBridge.test.cjs

@@ -2291,6 +2291,109 @@ 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("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 () => {