fix(sftp): abandon channel on stalled CLOSE; share abort gate

Drop the cached SFTP channel when verification CLOSE times out or fails,
and fan out concurrent prefix READs through one shared AbortSignal gate
so the bounded window does not attach dozens of listeners.

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

electron/bridges/transferBridge.cjs

@@ -21,6 +21,8 @@ const { openBoundedSftpChannel } = require("./boundedSftpOpen.cjs");
 const {
   DOWNLOAD_TRANSFER_CONCURRENCY,
   FAST_DOWNLOAD_CHANNELS_PER_SESSION,
+  SFTP_OPEN_TIMEOUT_MS,
+  SFTP_REQUEST_TIMEOUT_MS,
   TRANSFER_CHUNK_SIZE,
   UPLOAD_TRANSFER_CONCURRENCY,
 } = require("./transferLimits.cjs");
@@ -613,6 +615,9 @@ async function resolveRemoteResumeCheckpoint(client, sftpId, filePath, encoding,
 
 async function hashReadable(readable, options = {}) {
   const { signal, onProgress } = options;
+  const inactivityTimeoutMs = Number(options.inactivityTimeoutMs) > 0
+    ? Number(options.inactivityTimeoutMs)
+    : 0;
   const cancellationError = () => {
     const error = new Error("Transfer cancelled");
     error.code = "ABORT_ERR";
@@ -628,23 +633,50 @@ async function hashReadable(readable, options = {}) {
   signal?.addEventListener?.("abort", abortReadable, { once: true });
   const hash = crypto.createHash("sha256");
   let bytesRead = 0;
-  try {
-    for await (const chunk of readable) {
+  let timer = null;
+  let rejectTimeout = null;
+  const armInactivityTimer = () => {
+    if (!inactivityTimeoutMs) return;
+    if (timer) clearTimeout(timer);
+    timer = setTimeout(() => {
+      const error = new Error(`SFTP stream timed out after ${inactivityTimeoutMs} ms`);
+      error.code = "SFTP_STREAM_TIMEOUT";
+      error.sftpRequestTimedOut = true;
+      try { readable.destroy?.(error); } catch { /* ignore */ }
+      rejectTimeout?.(error);
+    }, inactivityTimeoutMs);
+  };
+  const consume = (async () => {
+    armInactivityTimer();
+    try {
+      for await (const chunk of readable) {
+        if (signal?.aborted) throw cancellationError();
+        hash.update(chunk);
+        bytesRead += chunk.length;
+        onProgress?.(bytesRead);
+        armInactivityTimer();
+      }
       if (signal?.aborted) throw cancellationError();
-      hash.update(chunk);
-      bytesRead += chunk.length;
-      onProgress?.(bytesRead);
+      return hash.digest("hex");
+    } catch (error) {
+      if (signal?.aborted) throw cancellationError();
+      throw error;
     }
-    if (signal?.aborted) throw cancellationError();
-    return hash.digest("hex");
-  } catch (error) {
-    if (signal?.aborted) throw cancellationError();
-    throw error;
+  })();
+  try {
+    if (!inactivityTimeoutMs) return await consume;
+    const timeout = new Promise((_, reject) => {
+      rejectTimeout = reject;
+    });
+    return await Promise.race([consume, timeout]);
   } finally {
+    if (timer) clearTimeout(timer);
     signal?.removeEventListener?.("abort", abortReadable);
   }
 }
 
+const EMPTY_SHA256_HEX = crypto.createHash("sha256").update("").digest("hex");
+
 function hashLocalPrefix(filePath, bytes, options) {
   if (!Number.isFinite(bytes) || bytes < 0) return Promise.resolve(null);
   if (bytes === 0) return Promise.resolve(EMPTY_SHA256_HEX);
@@ -655,6 +687,67 @@ function hashLocalFile(filePath, options = {}) {
   return hashReadable(fs.createReadStream(filePath), options);
 }
 
+async function hashRemotePrefixViaSshCommand(client, remotePath, bytes, options = {}) {
+  if (!Number.isFinite(bytes) || bytes <= 0 || isScpModeClient(client)) return null;
+  // Sudo SFTP elevates the subsystem, but exec() still runs as the login user.
+  // For a root-only source, head can fail while sha256sum/openssl still emit the
+  // empty-input digest and exit 0 — skip the command path and use elevated SFTP.
+  if (client?.__netcattySudoMode) return null;
+  const sshClient = client?.client;
+  if (!sshClient || typeof sshClient.exec !== "function") return null;
+
+  if (typeof remotePath !== "string") return null;
+  const byteCount = Math.floor(bytes);
+  const escapedPath = remotePath.replace(/'/g, "'\\''");
+  const commands = [
+    `if command -v head >/dev/null 2>&1 && command -v sha256sum >/dev/null 2>&1; then head -c ${byteCount} '${escapedPath}' | sha256sum; else exit 127; fi`,
+    `if command -v busybox >/dev/null 2>&1; then busybox head -c ${byteCount} '${escapedPath}' | busybox sha256sum; else exit 127; fi`,
+    `if command -v head >/dev/null 2>&1 && command -v openssl >/dev/null 2>&1; then head -c ${byteCount} '${escapedPath}' | openssl dgst -sha256; else exit 127; fi`,
+  ];
+
+  for (const command of commands) {
+    try {
+      const result = await executeBoundedSshCommand(sshClient, command, {
+        signal: options.signal,
+        openingTimeoutMs: Number(options.sshDigestOpeningTimeoutMs) > 0
+          ? Number(options.sshDigestOpeningTimeoutMs)
+          : 15_000,
+        runTimeoutMs: Number(options.sshDigestRunTimeoutMs) > 0
+          ? Number(options.sshDigestRunTimeoutMs)
+          : 10 * 60_000,
+        maxOutputBytes: 64 * 1024,
+      });
+      if (result.code !== 0) continue;
+      const match = String(result.stdout || "").match(/\b([a-fA-F0-9]{64})\b/);
+      if (!match) continue;
+      const digest = match[1].toLowerCase();
+      // Empty-input digest with bytes > 0 means head failed open into the hasher.
+      if (digest === EMPTY_SHA256_HEX) continue;
+      return digest;
+    } catch (error) {
+      if (options.signal?.aborted || error?.code === "ABORT_ERR") {
+        throw error;
+      }
+      // Run timeout: the exec stream opened, so the SSH transport is still valid.
+      // Treat the optional digest as a miss and fall through to SFTP verification.
+      if (error?.code === "SSH_EXEC_RUN_TIMEOUT") {
+        return null;
+      }
+      // Open timeout invalidates the physical transport in boundedSshExec. Do not
+      // continue into SFTP on this dead session — propagate so the caller fails
+      // closed instead of hanging/failing obscurely on a poisoned channel.
+      // Mark noTransferFallback so downloadFile's isolated→shared catch does not
+      // retry verification/body transfer on the invalidated transport.
+      if (error?.code === "SSH_EXEC_OPEN_TIMEOUT") {
+        error.noTransferFallback = true;
+        abandonWedgedVerificationSftpChannel(client);
+        throw error;
+      }
+    }
+  }
+  return null;
+}
+
 async function hashRemoteFile(client, sftpId, filePath, encoding, options = {}) {
   if (isScpModeClient(client)) return null;
   const sshClient = client?.client;
@@ -687,8 +780,6 @@ async function hashRemoteFile(client, sftpId, filePath, encoding, options = {})
   );
 }
 
-const EMPTY_SHA256_HEX = crypto.createHash("sha256").update("").digest("hex");
-
 /** @param {number|null|undefined} prefixBytes null = full file; >=0 = bounded prefix (incl. empty). */
 function formatSourceFingerprint(digest, prefixBytes) {
   if (!digest) return null;
@@ -3239,21 +3330,63 @@ function closeSftpHandle(sftp, handle) {
   });
 }
 
-async function readSftpRange(sftp, handle, buffer, position, length) {
+async function readSftpRange(sftp, handle, buffer, position, length, options = {}) {
+  const timeoutMs = Number(options.timeoutMs) > 0 ? Number(options.timeoutMs) : 0;
+  const signal = options.signal;
+  const abortGate = options.abortGate;
   let received = 0;
   while (received < length) {
     const bytesRead = await new Promise((resolve, reject) => {
-      sftp.read(
-        handle,
-        buffer,
-        received,
-        length - received,
-        position + received,
-        (error, count) => {
-          if (error) reject(error);
-          else resolve(Number(count) || 0);
-        },
-      );
+      let settled = false;
+      let timer = null;
+      let unwatchAbort = null;
+      const cleanup = () => {
+        if (timer) clearTimeout(timer);
+        unwatchAbort?.();
+        unwatchAbort = null;
+        signal?.removeEventListener?.("abort", onAbort);
+      };
+      const finish = (error, count) => {
+        if (settled) return;
+        settled = true;
+        cleanup();
+        if (error) reject(error);
+        else resolve(Number(count) || 0);
+      };
+      const onAbort = () => {
+        const error = new Error("Transfer cancelled");
+        error.code = "ABORT_ERR";
+        finish(error);
+      };
+      if (signal?.aborted || abortGate?.aborted) {
+        onAbort();
+        return;
+      }
+      if (abortGate) {
+        unwatchAbort = abortGate.watch(onAbort);
+      } else {
+        signal?.addEventListener?.("abort", onAbort, { once: true });
+      }
+      if (timeoutMs > 0) {
+        timer = setTimeout(() => {
+          const error = new Error(`SFTP READ timed out after ${timeoutMs} ms`);
+          error.code = "SFTP_READ_TIMEOUT";
+          error.sftpRequestTimedOut = true;
+          finish(error);
+        }, timeoutMs);
+      }
+      try {
+        sftp.read(
+          handle,
+          buffer,
+          received,
+          length - received,
+          position + received,
+          (error, count) => finish(error, count),
+        );
+      } catch (error) {
+        finish(error);
+      }
     });
     if (bytesRead <= 0) {
       throw new Error("Download stream finished before the full source was received");
@@ -3262,6 +3395,202 @@ async function readSftpRange(sftp, handle, buffer, position, length) {
   }
 }
 
+function createSharedAbortGate(signal) {
+  const waiters = new Set();
+  const notify = () => {
+    const error = new Error("Transfer cancelled");
+    error.code = "ABORT_ERR";
+    for (const reject of [...waiters]) {
+      try { reject(error); } catch { /* ignore */ }
+    }
+    waiters.clear();
+  };
+  const onAbort = () => notify();
+  if (signal?.aborted) {
+    return {
+      get aborted() { return true; },
+      watch(onAbortWatch) {
+        onAbortWatch();
+        return () => {};
+      },
+      dispose() {},
+    };
+  }
+  signal?.addEventListener?.("abort", onAbort, { once: true });
+  return {
+    get aborted() { return Boolean(signal?.aborted); },
+    watch(onAbortWatch) {
+      if (signal?.aborted) {
+        onAbortWatch();
+        return () => {};
+      }
+      waiters.add(onAbortWatch);
+      return () => waiters.delete(onAbortWatch);
+    },
+    dispose() {
+      signal?.removeEventListener?.("abort", onAbort);
+      waiters.clear();
+    },
+  };
+}
+
+async function closeSftpHandleBestEffort(sftp, handle, timeoutMs = 2_000) {
+  let timer = null;
+  let timedOut = false;
+  let failed = false;
+  try {
+    await Promise.race([
+      closeSftpHandle(sftp, handle),
+      new Promise((resolve) => {
+        timer = setTimeout(() => {
+          timedOut = true;
+          resolve();
+        }, timeoutMs);
+      }),
+    ]);
+  } catch {
+    // Verification cleanup must not mask the content check result.
+    failed = true;
+  } finally {
+    if (timer) clearTimeout(timer);
+  }
+  return { timedOut, failed, unclean: timedOut || failed };
+}
+
+function openSftpReadHandle(sftp, remotePath, signal, timeoutMs) {
+  return new Promise((resolve, reject) => {
+    let settled = false;
+    let timer = null;
+    const cleanup = () => {
+      if (timer) clearTimeout(timer);
+      signal?.removeEventListener?.("abort", onAbort);
+    };
+    const finish = (error, handle) => {
+      if (settled) return;
+      settled = true;
+      cleanup();
+      if (error) reject(error);
+      else resolve(handle);
+    };
+    const onAbort = () => {
+      const error = new Error("Transfer cancelled");
+      error.code = "ABORT_ERR";
+      finish(error);
+    };
+    if (signal?.aborted) {
+      onAbort();
+      return;
+    }
+    signal?.addEventListener?.("abort", onAbort, { once: true });
+    if (timeoutMs > 0) {
+      timer = setTimeout(() => {
+        const error = new Error(`SFTP OPEN timed out after ${timeoutMs} ms`);
+        error.code = "SFTP_OPEN_TIMEOUT";
+        error.sftpRequestTimedOut = true;
+        finish(error);
+      }, timeoutMs);
+    }
+    try {
+      sftp.open(remotePath, "r", (error, handle) => {
+        if (settled) {
+          if (handle) closeSftpHandle(sftp, handle).catch(() => {});
+          return;
+        }
+        finish(error, handle);
+      });
+    } catch (error) {
+      finish(error);
+    }
+  });
+}
+
+async function hashRemotePrefixWithSftpRanges(client, remotePath, bytes, options = {}) {
+  if (!Number.isFinite(bytes) || bytes <= 0 || isScpModeClient(client)) return null;
+  await requireSftpChannel(client, { signal: options.signal });
+  const sftp = client.sftp;
+  if (typeof sftp?.open !== "function" || typeof sftp?.read !== "function") return null;
+
+  const chunkSize = TRANSFER_CHUNK_SIZE;
+  const rangeCount = Math.ceil(bytes / chunkSize);
+  const concurrency = Math.min(DOWNLOAD_TRANSFER_CONCURRENCY, rangeCount);
+  let completedBytes = 0;
+  const openTimeoutMs = Number(options.sftpOpenTimeoutMs) > 0
+    ? Number(options.sftpOpenTimeoutMs)
+    : SFTP_OPEN_TIMEOUT_MS;
+  const readTimeoutMs = Number(options.sftpReadTimeoutMs) > 0
+    ? Number(options.sftpReadTimeoutMs)
+    : SFTP_REQUEST_TIMEOUT_MS;
+  const handle = await openSftpReadHandle(sftp, remotePath, options.signal, openTimeoutMs);
+  const abortGate = createSharedAbortGate(options.signal);
+  try {
+    // Hash windows in order so peak retained buffers stay within the concurrency
+    // fanout (~2MB), not the full multi-GB prefix.
+    const hash = crypto.createHash("sha256");
+    for (let windowStart = 0; windowStart < rangeCount; windowStart += concurrency) {
+      const windowCount = Math.min(concurrency, rangeCount - windowStart);
+      const windowBuffers = new Array(windowCount);
+      // One inactivity watchdog for the whole window. Per-request deadlines would
+      // fire together after readTimeoutMs even while earlier reads keep landing
+      // on a slow/serialized server.
+      let inactivityTimer = null;
+      let rejectInactivity = null;
+      const clearInactivity = () => {
+        if (inactivityTimer) clearTimeout(inactivityTimer);
+        inactivityTimer = null;
+      };
+      const armInactivity = () => {
+        if (!(readTimeoutMs > 0) || options.signal?.aborted || abortGate.aborted) return;
+        clearInactivity();
+        inactivityTimer = setTimeout(() => {
+          const error = new Error(`SFTP READ timed out after ${readTimeoutMs} ms`);
+          error.code = "SFTP_READ_TIMEOUT";
+          error.sftpRequestTimedOut = true;
+          rejectInactivity?.(error);
+        }, readTimeoutMs);
+      };
+      const inactivityWait = readTimeoutMs > 0
+        ? new Promise((_, reject) => { rejectInactivity = reject; })
+        : null;
+      armInactivity();
+      try {
+        await Promise.race([
+          Promise.all(Array.from({ length: windowCount }, async (_, offset) => {
+            const index = windowStart + offset;
+            const position = index * chunkSize;
+            const length = Math.min(chunkSize, bytes - position);
+            const buffer = Buffer.allocUnsafe(length);
+            await readSftpRange(sftp, handle, buffer, position, length, {
+              abortGate,
+            });
+            windowBuffers[offset] = buffer;
+            completedBytes += length;
+            options.onProgress?.(completedBytes);
+            armInactivity();
+          })),
+          ...(inactivityWait ? [inactivityWait] : []),
+        ]);
+      } finally {
+        clearInactivity();
+        rejectInactivity = null;
+      }
+      for (const buffer of windowBuffers) hash.update(buffer);
+    }
+    return hash.digest("hex");
+  } finally {
+    abortGate.dispose();
+    const closeResult = await closeSftpHandleBestEffort(
+      sftp,
+      handle,
+      Number(options.sftpCloseTimeoutMs) > 0 ? Number(options.sftpCloseTimeoutMs) : 2_000,
+    );
+    // A timed-out/failed CLOSE leaves an unresolved request on the shared
+    // channel; drop it so later browse/transfer work cannot reuse it.
+    if (closeResult?.unclean) {
+      abandonWedgedVerificationSftpChannel(client);
+    }
+  }
+}
+
 async function writeLocalRange(fileHandle, buffer, position, length) {
   let written = 0;
   while (written < length) {
@@ -3703,6 +4032,17 @@ function assertSourceMetadataUnchanged(initialSource, latestSource, expectedSize
  * @param {number} prefixBytes planned snapshot size
  * @param {{ signal?: AbortSignal, onProgress?: (n: number) => void }} [options]
  */
+function abandonWedgedVerificationSftpChannel(client) {
+  const sftp = client?.sftp;
+  if (!client || !sftp) return;
+  // Drop the cached channel first so requireSftpChannel cannot hand the wedged
+  // object back to later browse/transfer work (hasSftpChannelApi only checks
+  // method presence). Non-sudo sessions can reopen; sudo recovery stays disabled.
+  client.sftp = null;
+  try { sftp.end?.(); } catch { /* ignore */ }
+  try { sftp.destroy?.(); } catch { /* ignore */ }
+}
+
 async function assertLocalDownloadMatchesRemotePrefix(
   localPath,
   client,
@@ -3715,22 +4055,61 @@ async function assertLocalDownloadMatchesRemotePrefix(
     // SCP cannot range-hash portably; fail closed when growth needs proof.
     throw createSourceContentChangedError();
   }
-  await requireSftpChannel(client, { signal: options.signal });
-  if (typeof client.sftp?.createReadStream !== "function") {
-    throw createSourceContentChangedError();
-  }
-  const [localHash, remoteHash] = await Promise.all([
-    hashLocalFile(localPath, options),
-    hashReadable(
-      client.sftp.createReadStream(remotePath, {
-        start: 0,
-        end: prefixBytes - 1,
-      }),
-      options,
-    ),
-  ]);
-  if (!localHash || !remoteHash || localHash !== remoteHash) {
-    throw createSourceContentChangedError();
+  try {
+    await requireSftpChannel(client, { signal: options.signal });
+    const remoteHash = (async () => {
+      const commandDigest = await hashRemotePrefixViaSshCommand(
+        client,
+        remotePath,
+        prefixBytes,
+        options,
+      );
+      if (commandDigest) return commandDigest;
+      if (options.preferSftpRanges !== false) {
+        try {
+          const rangeDigest = await hashRemotePrefixWithSftpRanges(
+            client,
+            remotePath,
+            prefixBytes,
+            options,
+          );
+          if (rangeDigest) return rangeDigest;
+        } catch (error) {
+          if (options.signal?.aborted || error?.sftpRequestTimedOut) throw error;
+        }
+      }
+      if (typeof client.sftp?.createReadStream !== "function") {
+        throw createSourceContentChangedError();
+      }
+      return hashReadable(
+        client.sftp.createReadStream(remotePath, {
+          start: 0,
+          end: prefixBytes - 1,
+        }),
+        {
+          ...options,
+          inactivityTimeoutMs: Number(options.sftpReadTimeoutMs) > 0
+            ? Number(options.sftpReadTimeoutMs)
+            : SFTP_REQUEST_TIMEOUT_MS,
+        },
+      );
+    })();
+    const [localHash, verifiedRemoteHash] = await Promise.all([
+      hashLocalFile(localPath, options),
+      remoteHash,
+    ]);
+    if (!localHash || !verifiedRemoteHash || localHash !== verifiedRemoteHash) {
+      throw createSourceContentChangedError();
+    }
+  } catch (error) {
+    // Verification OPEN/READ/stream watchdogs must fail closed — downloadFile's
+    // isolated→shared catch would otherwise retry body transfer on a channel
+    // that still owns the timed-out request and can hang indefinitely.
+    if (error && typeof error === "object" && error.sftpRequestTimedOut) {
+      error.noTransferFallback = true;
+      abandonWedgedVerificationSftpChannel(client);
+    }
+    throw error;
   }
 }
 
@@ -3747,6 +4126,8 @@ async function assertDownloadSourceAfterTransfer(
     client,
     remotePath,
     signal,
+    preferSftpRanges = true,
+    ...verificationOptions
   } = {},
 ) {
   if (!initialSource) return;
@@ -3763,7 +4144,7 @@ async function assertDownloadSourceAfterTransfer(
       client,
       remotePath,
       expectedSize,
-      { signal },
+      { signal, preferSftpRanges, ...verificationOptions },
     );
     assertSourceMetadataUnchanged(initialSource, latestSource, expectedSize, {
       allowSourceGrowth: true,
@@ -4483,13 +4864,25 @@ async function downloadFile(
     }
     sendProgress(fileSize, fileSize, { force: true, checkpointBytes: fileSize });
     if (initialSource) {
-      const latestSource = await runCancelablePreflight(() => client.stat(remotePath));
-      await assertDownloadSourceAfterTransfer(initialSource, latestSource, fileSize, {
-        localPath,
-        client,
-        remotePath,
-        signal: transfer.signal,
-      });
+      // Growth verification may use concurrent prefix ranges (especially sudo,
+      // which skips unprivileged SSH digests). Hold the session fast-download
+      // slot so those READs do not race another download's fanout.
+      let holdSessionSlot = false;
+      try {
+        await acquireSessionFastDownloadSlot(client, transfer);
+        holdSessionSlot = true;
+        const latestSource = await runCancelablePreflight(() => client.stat(remotePath));
+        await assertDownloadSourceAfterTransfer(initialSource, latestSource, fileSize, {
+          localPath,
+          client,
+          remotePath,
+          signal: transfer.signal,
+        });
+      } finally {
+        if (holdSessionSlot) {
+          releaseSessionFastDownloadSlot(client);
+        }
+      }
     }
     return;
   }
@@ -4515,15 +4908,33 @@ async function downloadFile(
 
   const finishSuccessfulDownload = async () => {
     if (!initialSource) return;
-    const latestSource = await runCancelablePreflight(() => client.stat(remotePath));
-    // Downloads capture a fixed snapshot; remote appends (live logs) are OK
-    // only when the full planned prefix still matches the staged file.
-    await assertDownloadSourceAfterTransfer(initialSource, latestSource, fileSize, {
-      localPath,
-      client,
-      remotePath,
-      signal: transfer.signal,
-    });
+    const previousPhase = transfer.phase;
+    let lastVerificationProgressAt = 0;
+    transfer.phase = "verifying";
+    transfer.publishCurrentProgress?.();
+    try {
+      const latestSource = await runCancelablePreflight(() => client.stat(remotePath));
+      // Downloads capture a fixed snapshot; remote appends (live logs) are OK
+      // only when the full planned prefix still matches the staged file.
+      await assertDownloadSourceAfterTransfer(initialSource, latestSource, fileSize, {
+        localPath,
+        client,
+        remotePath,
+        signal: transfer.signal,
+        onProgress: () => {
+          const now = Date.now();
+          if (now - lastVerificationProgressAt < 250) return;
+          lastVerificationProgressAt = now;
+          transfer.publishCurrentProgress?.();
+        },
+      });
+      transfer.publishCurrentProgress?.();
+    } finally {
+      transfer.phase = previousPhase || "transferring";
+      if (!transfer.cancelled && !transfer.signal?.aborted) {
+        transfer.publishCurrentProgress?.();
+      }
+    }
   };
 
   // One session-wide fast-path slot for isolated + shared READ fanout (#1507).

electron/bridges/transferBridge.test.cjs

@@ -3721,6 +3721,574 @@ test("resumable fast downloads clear staged data after a same-second source chan
   await assert.rejects(fs.promises.stat(stagedPath), { code: "ENOENT" });
 });
 
+test("growing-download prefix verification prefers a bounded remote digest", async (t) => {
+  const tempDir = await fs.promises.mkdtemp(path.join(os.tmpdir(), "netcatty-transfer-prefix-digest-"));
+  t.after(async () => fs.promises.rm(tempDir, { recursive: true, force: true }));
+
+  const payload = Buffer.alloc(128 * 1024, 71);
+  const localPath = path.join(tempDir, "download.bin");
+  await fs.promises.writeFile(localPath, payload);
+  const expectedDigest = crypto.createHash("sha256").update(payload).digest("hex");
+  let command = "";
+  const client = {
+    sftp: createFastSftp({
+      createReadStream() {
+        throw new Error("SFTP prefix stream should not run when remote digest is available");
+      },
+    }),
+    client: {
+      exec(request, callback) {
+        command = request;
+        const stream = new EventEmitter();
+        stream.stderr = new EventEmitter();
+        stream.destroy = () => {};
+        callback(null, stream);
+        setImmediate(() => {
+          stream.emit("data", Buffer.from(`${expectedDigest}  -\n`));
+          stream.emit("close", 0);
+        });
+      },
+    },
+  };
+
+  await transferBridge._assertDownloadSourceAfterTransferForTests(
+    { size: payload.length, mtimeMs: 1 },
+    { size: payload.length + 1, mtimeMs: 2 },
+    payload.length,
+    {
+      localPath,
+      client,
+      remotePath: "/var/log/app.log",
+      signal: new AbortController().signal,
+    },
+  );
+
+  assert.match(command, new RegExp(`head -c ${payload.length}`));
+  assert.match(command, /sha256sum|busybox|openssl/);
+});
+
+test("growing-download prefix verification skips unprivileged digest for sudo SFTP", async (t) => {
+  const tempDir = await fs.promises.mkdtemp(path.join(os.tmpdir(), "netcatty-transfer-prefix-sudo-digest-"));
+  t.after(async () => fs.promises.rm(tempDir, { recursive: true, force: true }));
+
+  const payload = Buffer.alloc(96 * 1024, 72);
+  const localPath = path.join(tempDir, "download.bin");
+  await fs.promises.writeFile(localPath, payload);
+  const emptyDigest = crypto.createHash("sha256").update("").digest("hex");
+  let execCalls = 0;
+  const pipelined = createPipelinedDownloadSftp(payload, {
+    createReadStream() {
+      throw new Error("serial prefix stream should remain the final fallback");
+    },
+  });
+  const client = {
+    __netcattySudoMode: true,
+    sftp: pipelined.sftp,
+    client: {
+      exec(_request, callback) {
+        execCalls += 1;
+        const stream = new EventEmitter();
+        stream.stderr = new EventEmitter();
+        stream.destroy = () => {};
+        callback(null, stream);
+        // Unprivileged head fails open into sha256sum and still exits 0.
+        setImmediate(() => {
+          stream.emit("data", Buffer.from(`${emptyDigest}  -\n`));
+          stream.emit("close", 0);
+        });
+      },
+    },
+  };
+
+  await transferBridge._assertDownloadSourceAfterTransferForTests(
+    { size: payload.length, mtimeMs: 1 },
+    { size: payload.length + 1, mtimeMs: 2 },
+    payload.length,
+    {
+      localPath,
+      client,
+      remotePath: "/root/secure.log",
+      signal: new AbortController().signal,
+    },
+  );
+
+  assert.equal(execCalls, 0, "sudo downloads must not use unprivileged digest commands");
+});
+
+test("growing-download prefix verification falls back after digest command timeout", async (t) => {
+  const tempDir = await fs.promises.mkdtemp(path.join(os.tmpdir(), "netcatty-transfer-prefix-digest-timeout-"));
+  t.after(async () => fs.promises.rm(tempDir, { recursive: true, force: true }));
+
+  const payload = Buffer.alloc(96 * 1024, 69);
+  const localPath = path.join(tempDir, "download.bin");
+  await fs.promises.writeFile(localPath, payload);
+  let rangeReads = 0;
+  const pipelined = createPipelinedDownloadSftp(payload, {
+    read(handle, buffer, offset, length, position, callback) {
+      rangeReads += 1;
+      const end = Math.min(position + length, payload.length);
+      const slice = payload.subarray(position, end);
+      slice.copy(buffer, offset);
+      setImmediate(() => callback(null, slice.length));
+    },
+    createReadStream() {
+      throw new Error("serial prefix stream should remain the final fallback");
+    },
+  });
+  const client = {
+    sftp: pipelined.sftp,
+    client: {
+      exec(_request, callback) {
+        const stream = new EventEmitter();
+        stream.stderr = new EventEmitter();
+        stream.destroy = () => {};
+        callback(null, stream);
+        // Open succeeds but never completes — hits SSH_EXEC_RUN_TIMEOUT.
+      },
+    },
+  };
+
+  await transferBridge._assertDownloadSourceAfterTransferForTests(
+    { size: payload.length, mtimeMs: 1 },
+    { size: payload.length + 1, mtimeMs: 2 },
+    payload.length,
+    {
+      localPath,
+      client,
+      remotePath: "/var/log/app.log",
+      signal: new AbortController().signal,
+      sshDigestRunTimeoutMs: 30,
+    },
+  );
+
+  assert.ok(rangeReads > 0, "digest run timeout must fall through to SFTP range hashing");
+});
+
+test("growing-download prefix verification propagates digest open timeouts", async (t) => {
+  const tempDir = await fs.promises.mkdtemp(path.join(os.tmpdir(), "netcatty-transfer-prefix-digest-open-timeout-"));
+  t.after(async () => fs.promises.rm(tempDir, { recursive: true, force: true }));
+
+  const payload = Buffer.alloc(32 * 1024, 68);
+  const localPath = path.join(tempDir, "download.bin");
+  await fs.promises.writeFile(localPath, payload);
+  let rangeReads = 0;
+  const client = {
+    sftp: createFastSftp({
+      open(_remotePath, _flags, callback) {
+        rangeReads += 1;
+        callback(null, Buffer.from("should-not-open"));
+      },
+      read() {
+        throw new Error("SFTP fallback must not run after exec-open timeout");
+      },
+      createReadStream() {
+        throw new Error("SFTP stream fallback must not run after exec-open timeout");
+      },
+    }),
+    client: {
+      // Never invoke the exec callback — hits SSH_EXEC_OPEN_TIMEOUT and
+      // invalidates the transport in boundedSshExec.
+      exec() {},
+    },
+  };
+
+  await assert.rejects(
+    transferBridge._assertDownloadSourceAfterTransferForTests(
+      { size: payload.length, mtimeMs: 1 },
+      { size: payload.length + 1, mtimeMs: 2 },
+      payload.length,
+      {
+        localPath,
+        client,
+        remotePath: "/var/log/app.log",
+        signal: new AbortController().signal,
+        sshDigestOpeningTimeoutMs: 30,
+      },
+    ),
+    (error) => error?.code === "SSH_EXEC_OPEN_TIMEOUT" && error?.noTransferFallback === true,
+  );
+  assert.equal(rangeReads, 0, "open timeout must not fall through onto a poisoned session");
+  assert.equal(client.sftp, null, "exec-open timeout must drop the cached SFTP channel");
+});
+
+test("growing-download prefix verification rejects empty remote digests", async (t) => {
+  const tempDir = await fs.promises.mkdtemp(path.join(os.tmpdir(), "netcatty-transfer-prefix-empty-digest-"));
+  t.after(async () => fs.promises.rm(tempDir, { recursive: true, force: true }));
+
+  const payload = Buffer.alloc(64 * 1024, 70);
+  const localPath = path.join(tempDir, "download.bin");
+  await fs.promises.writeFile(localPath, payload);
+  const emptyDigest = crypto.createHash("sha256").update("").digest("hex");
+  let rangeReads = 0;
+  const pipelined = createPipelinedDownloadSftp(payload, {
+    read(handle, buffer, offset, length, position, callback) {
+      rangeReads += 1;
+      const end = Math.min(position + length, payload.length);
+      const slice = payload.subarray(position, end);
+      slice.copy(buffer, offset);
+      setImmediate(() => callback(null, slice.length));
+    },
+    createReadStream() {
+      throw new Error("serial prefix stream should remain the final fallback");
+    },
+  });
+  const client = {
+    sftp: pipelined.sftp,
+    client: {
+      exec(_request, callback) {
+        const stream = new EventEmitter();
+        stream.stderr = new EventEmitter();
+        stream.destroy = () => {};
+        callback(null, stream);
+        setImmediate(() => {
+          stream.emit("data", Buffer.from(`${emptyDigest}  -\n`));
+          stream.emit("close", 0);
+        });
+      },
+    },
+  };
+
+  await transferBridge._assertDownloadSourceAfterTransferForTests(
+    { size: payload.length, mtimeMs: 1 },
+    { size: payload.length + 1, mtimeMs: 2 },
+    payload.length,
+    {
+      localPath,
+      client,
+      remotePath: "/var/log/app.log",
+      signal: new AbortController().signal,
+    },
+  );
+
+  assert.ok(rangeReads > 0, "empty digest must fall through to elevated/SFTP hashing");
+});
+
+test("growing-download prefix verification falls back to parallel SFTP reads", async (t) => {
+  const tempDir = await fs.promises.mkdtemp(path.join(os.tmpdir(), "netcatty-transfer-prefix-ranges-"));
+  t.after(async () => fs.promises.rm(tempDir, { recursive: true, force: true }));
+
+  const payload = Buffer.alloc(256 * 1024, 73);
+  const localPath = path.join(tempDir, "download.bin");
+  await fs.promises.writeFile(localPath, payload);
+  const pipelined = createPipelinedDownloadSftp(payload, {
+    createReadStream() {
+      throw new Error("serial prefix stream should be the final fallback only");
+    },
+  });
+  const client = {
+    sftp: pipelined.sftp,
+    client: {
+      exec(_request, callback) {
+        const error = new Error("SSH exec unavailable");
+        error.code = "SSH_EXEC_UNAVAILABLE";
+        callback(error);
+      },
+    },
+  };
+
+  await transferBridge._assertDownloadSourceAfterTransferForTests(
+    { size: payload.length, mtimeMs: 1 },
+    { size: payload.length + 1, mtimeMs: 2 },
+    payload.length,
+    {
+      localPath,
+      client,
+      remotePath: "/var/log/app.log",
+      signal: new AbortController().signal,
+    },
+  );
+});
+
+test("growing-download SFTP range hashing keeps a bounded read window", async (t) => {
+  const tempDir = await fs.promises.mkdtemp(path.join(os.tmpdir(), "netcatty-transfer-prefix-bound-"));
+  t.after(async () => fs.promises.rm(tempDir, { recursive: true, force: true }));
+
+  const chunkSize = TRANSFER_CHUNK_SIZE;
+  const concurrency = DOWNLOAD_TRANSFER_CONCURRENCY;
+  const payload = Buffer.alloc((concurrency + 8) * chunkSize, 77);
+  const localPath = path.join(tempDir, "download.bin");
+  await fs.promises.writeFile(localPath, payload);
+
+  let maxIndexStarted = -1;
+  let releaseFirstWindow;
+  const firstWindowBlocked = new Promise((resolve) => { releaseFirstWindow = resolve; });
+  let firstWindowGateOpened = false;
+  const pendingFirstWindow = [];
+  const warnings = [];
+  const onWarning = (warning) => {
+    warnings.push(String(warning?.name || warning?.message || warning));
+  };
+  process.on("warning", onWarning);
+  t.after(() => process.off("warning", onWarning));
+
+  const client = {
+    sftp: createFastSftp({
+      open(_remotePath, _flags, callback) {
+        callback(null, Buffer.from("bounded-handle"));
+      },
+      read(_handle, buffer, offset, length, position, callback) {
+        const index = Math.floor(position / chunkSize);
+        maxIndexStarted = Math.max(maxIndexStarted, index);
+        const end = Math.min(position + length, payload.length);
+        const slice = payload.subarray(position, end);
+        const deliver = () => {
+          slice.copy(buffer, offset);
+          callback(null, slice.length);
+        };
+        if (!firstWindowGateOpened && index < concurrency) {
+          pendingFirstWindow.push(deliver);
+          if (pendingFirstWindow.length === 1) {
+            firstWindowBlocked.then(() => {
+              firstWindowGateOpened = true;
+              for (const pending of pendingFirstWindow.splice(0)) pending();
+            });
+          }
+          return;
+        }
+        setImmediate(deliver);
+      },
+      close(_handle, callback) {
+        callback(null);
+      },
+      createReadStream() {
+        throw new Error("serial prefix stream should not run for bounded-window test");
+      },
+    }),
+    client: {
+      exec(_request, callback) {
+        const error = new Error("SSH exec unavailable");
+        error.code = "SSH_EXEC_UNAVAILABLE";
+        callback(error);
+      },
+    },
+  };
+
+  const verifyPromise = transferBridge._assertDownloadSourceAfterTransferForTests(
+    { size: payload.length, mtimeMs: 1 },
+    { size: payload.length + 1, mtimeMs: 2 },
+    payload.length,
+    {
+      localPath,
+      client,
+      remotePath: "/var/log/app.log",
+      signal: new AbortController().signal,
+    },
+  );
+
+  await waitUntil(() => pendingFirstWindow.length >= concurrency, 2000);
+  assert.equal(
+    maxIndexStarted,
+    concurrency - 1,
+    "later windows must wait until the current concurrency window finishes",
+  );
+  releaseFirstWindow();
+  await verifyPromise;
+  assert.ok(maxIndexStarted >= concurrency, "verification must continue past the first window");
+  assert.equal(
+    warnings.some((message) => /MaxListenersExceededWarning/i.test(message)),
+    false,
+    "shared abort gate must not attach one listener per concurrent READ",
+  );
+});
+
+test("growing-download SFTP range hashing resets inactivity on window progress", async (t) => {
+  const tempDir = await fs.promises.mkdtemp(path.join(os.tmpdir(), "netcatty-transfer-prefix-window-watchdog-"));
+  t.after(async () => fs.promises.rm(tempDir, { recursive: true, force: true }));
+
+  const chunkSize = TRANSFER_CHUNK_SIZE;
+  const payload = Buffer.alloc(3 * chunkSize, 79);
+  const localPath = path.join(tempDir, "download.bin");
+  await fs.promises.writeFile(localPath, payload);
+  const pendingReads = [];
+  let pumping = false;
+
+  const client = {
+    sftp: createFastSftp({
+      open(_remotePath, _flags, callback) {
+        callback(null, Buffer.from("watchdog-handle"));
+      },
+      read(_handle, buffer, offset, length, position, callback) {
+        const end = Math.min(position + length, payload.length);
+        const slice = payload.subarray(position, end);
+        pendingReads.push(() => {
+          slice.copy(buffer, offset);
+          callback(null, slice.length);
+        });
+        if (!pumping && pendingReads.length >= 3) {
+          pumping = true;
+          // Serialize completions so wall time exceeds the inactivity budget,
+          // but keep landing progress so a window watchdog stays armed.
+          void (async () => {
+            while (pendingReads.length > 0) {
+              await new Promise((resolve) => setTimeout(resolve, 40));
+              pendingReads.shift()?.();
+            }
+          })();
+        }
+      },
+      close(_handle, callback) {
+        callback(null);
+      },
+      createReadStream() {
+        throw new Error("serial prefix stream should not hide window-watchdog behavior");
+      },
+    }),
+    client: {
+      exec(_request, callback) {
+        const error = new Error("SSH exec unavailable");
+        error.code = "SSH_EXEC_UNAVAILABLE";
+        callback(error);
+      },
+    },
+  };
+
+  await transferBridge._assertDownloadSourceAfterTransferForTests(
+    { size: payload.length, mtimeMs: 1 },
+    { size: payload.length + 1, mtimeMs: 2 },
+    payload.length,
+    {
+      localPath,
+      client,
+      remotePath: "/var/log/app.log",
+      signal: new AbortController().signal,
+      // 80ms per-request deadlines would kill the 3rd serialized read (~120ms),
+      // but window inactivity resets on each completion.
+      sftpReadTimeoutMs: 80,
+    },
+  );
+});
+
+test("growing-download SFTP range hashing drops channel when CLOSE stalls", async (t) => {
+  const tempDir = await fs.promises.mkdtemp(path.join(os.tmpdir(), "netcatty-transfer-prefix-close-stall-"));
+  t.after(async () => fs.promises.rm(tempDir, { recursive: true, force: true }));
+
+  const payload = Buffer.alloc(64 * 1024, 81);
+  const localPath = path.join(tempDir, "download.bin");
+  await fs.promises.writeFile(localPath, payload);
+  const client = {
+    sftp: createFastSftp({
+      open(_remotePath, _flags, callback) {
+        callback(null, Buffer.from("close-stall-handle"));
+      },
+      read(_handle, buffer, offset, length, position, callback) {
+        const end = Math.min(position + length, payload.length);
+        const slice = payload.subarray(position, end);
+        slice.copy(buffer, offset);
+        setImmediate(() => callback(null, slice.length));
+      },
+      close() {
+        // Never invoke the callback — CLOSE watchdog must abandon the channel.
+      },
+      createReadStream() {
+        throw new Error("serial prefix stream should not run for CLOSE-stall test");
+      },
+    }),
+    client: {
+      exec(_request, callback) {
+        const error = new Error("SSH exec unavailable");
+        error.code = "SSH_EXEC_UNAVAILABLE";
+        callback(error);
+      },
+    },
+  };
+
+  await transferBridge._assertDownloadSourceAfterTransferForTests(
+    { size: payload.length, mtimeMs: 1 },
+    { size: payload.length + 1, mtimeMs: 2 },
+    payload.length,
+    {
+      localPath,
+      client,
+      remotePath: "/var/log/app.log",
+      signal: new AbortController().signal,
+      sftpCloseTimeoutMs: 20,
+    },
+  );
+  assert.equal(client.sftp, null, "stalled CLOSE must drop the wedged SFTP channel");
+});
+
+test("growing-download prefix verification times out a stalled SFTP READ", async (t) => {
+  const tempDir = await fs.promises.mkdtemp(path.join(os.tmpdir(), "netcatty-transfer-prefix-timeout-"));
+  t.after(async () => fs.promises.rm(tempDir, { recursive: true, force: true }));
+
+  const payload = Buffer.alloc(64 * 1024, 74);
+  const localPath = path.join(tempDir, "download.bin");
+  await fs.promises.writeFile(localPath, payload);
+  const client = {
+    sftp: createFastSftp({
+      open(_path, _flags, callback) {
+        callback(null, Buffer.from("stalled-handle"));
+      },
+      read() {},
+      createReadStream() {
+        throw new Error("serial prefix stream should not hide a stalled range");
+      },
+      close(_handle, callback) {
+        callback(null);
+      },
+    }),
+  };
+
+  await assert.rejects(
+    transferBridge._assertDownloadSourceAfterTransferForTests(
+      { size: payload.length, mtimeMs: 1 },
+      { size: payload.length + 1, mtimeMs: 2 },
+      payload.length,
+      {
+        localPath,
+        client,
+        remotePath: "/var/log/app.log",
+        signal: new AbortController().signal,
+        sftpReadTimeoutMs: 20,
+      },
+    ),
+    (error) => (
+      /SFTP READ timed out/.test(error?.message || "")
+      && error?.sftpRequestTimedOut === true
+      && error?.noTransferFallback === true
+    ),
+  );
+  assert.equal(client.sftp, null, "timed-out verification must drop the wedged SFTP channel");
+});
+
+test("growing-download prefix stream fallback times out without data", async (t) => {
+  const tempDir = await fs.promises.mkdtemp(path.join(os.tmpdir(), "netcatty-transfer-prefix-stream-timeout-"));
+  t.after(async () => fs.promises.rm(tempDir, { recursive: true, force: true }));
+
+  const payload = Buffer.alloc(32 * 1024, 75);
+  const localPath = path.join(tempDir, "download.bin");
+  await fs.promises.writeFile(localPath, payload);
+  const client = {
+    sftp: createFastSftp({
+      createReadStream() {
+        return new Readable({ read() {} });
+      },
+    }),
+  };
+
+  await assert.rejects(
+    transferBridge._assertDownloadSourceAfterTransferForTests(
+      { size: payload.length, mtimeMs: 1 },
+      { size: payload.length + 1, mtimeMs: 2 },
+      payload.length,
+      {
+        localPath,
+        client,
+        remotePath: "/var/log/app.log",
+        signal: new AbortController().signal,
+        sftpReadTimeoutMs: 20,
+        preferSftpRanges: false,
+      },
+    ),
+    (error) => (
+      /SFTP stream timed out/.test(error?.message || "")
+      && error?.sftpRequestTimedOut === true
+      && error?.noTransferFallback === true
+    ),
+  );
+  assert.equal(client.sftp, null, "stream timeout must drop the wedged SFTP channel");
+});
+
 test("resumable SFTP downloads succeed when the remote source only grows (live logs)", async (t) => {
   const tempDir = await fs.promises.mkdtemp(path.join(os.tmpdir(), "netcatty-transfer-download-growth-"));
   const transferId = "download-source-growth";
@@ -3895,7 +4463,7 @@ test("resumable SFTP downloads reject growth when the planned prefix was rewritt
   }
 });
 
-test("checkpoint-complete resume skips source open when checkpoint already covers the snapshot", async (t) => {
+test("checkpoint-complete resume verifies growth with bounded prefix ranges", async (t) => {
   const tempDir = await fs.promises.mkdtemp(path.join(os.tmpdir(), "netcatty-transfer-download-complete-checkpoint-"));
   const transferId = "download-complete-checkpoint-growth";
   const targetPath = path.join(tempDir, "download.bin");
@@ -3918,14 +4486,24 @@ test("checkpoint-complete resume skips source open when checkpoint already cover
 
   let bodyOpenAtOldEof = false;
   let openCalls = 0;
+  let verifyRangeReads = 0;
   const sharedSftp = createFastSftp({
-    open() {
+    open(_remotePath, _flags, callback) {
       openCalls += 1;
-      bodyOpenAtOldEof = true;
-      throw new Error("must not open transfer body at planned EOF");
+      callback(null, Buffer.from("verify-handle"));
     },
-    read() {
-      throw new Error("must not READ after planned EOF");
+    read(_handle, buffer, offset, length, position, callback) {
+      if (position >= snapshot.length) {
+        bodyOpenAtOldEof = true;
+        throw new Error("must not READ past planned snapshot");
+      }
+      verifyRangeReads += 1;
+      const end = Math.min(position + length, snapshot.length);
+      snapshot.subarray(position, end).copy(buffer, offset);
+      setImmediate(() => callback(null, end - position));
+    },
+    close(_handle, callback) {
+      callback(null);
     },
     createReadStream(_remotePath, options = {}) {
       const start = Number.isFinite(options.start) ? options.start : 0;
@@ -3970,7 +4548,8 @@ test("checkpoint-complete resume skips source open when checkpoint already cover
   );
 
   assert.equal(result.error, undefined, result.error);
-  assert.equal(openCalls, 0, "checkpoint-complete must not open remote handle");
+  assert.ok(verifyRangeReads > 0, "checkpoint-complete growth must use bounded prefix ranges");
+  assert.ok(openCalls > 0, "prefix verification may open a read handle at offset 0");
   assert.equal(bodyOpenAtOldEof, false, "must not open transfer body at the old EOF");
   const downloaded = await fs.promises.readFile(targetPath);
   assert.deepEqual(downloaded, snapshot);

electron/bridges/transferLimits.cjs

@@ -15,6 +15,10 @@ const UPLOAD_TRANSFER_CONCURRENCY = 64;
 // ssh2's fastGet default and, with the safe 32KB request size, restores the 2MB
 // in-flight window Netcatty used before the shared chunk-size fix in #2030.
 const DOWNLOAD_TRANSFER_CONCURRENCY = 64;
+// Prefix verification must fail closed when a server stops answering, but allow
+// slow links to make progress without imposing a whole-transfer deadline.
+const SFTP_OPEN_TIMEOUT_MS = 15_000;
+const SFTP_REQUEST_TIMEOUT_MS = 30_000;
 // Only one file per SFTP session holds the 64-request concurrent READ fanout
 // (isolated channel or shared/sudo browse path). Additional downloads wait for
 // a free slot rather than degrading to serial createReadStream (#2719 / #2449
@@ -27,6 +31,8 @@ const FAST_DOWNLOAD_CHANNELS_PER_SESSION = 1;
 module.exports = {
   DOWNLOAD_TRANSFER_CONCURRENCY,
   FAST_DOWNLOAD_CHANNELS_PER_SESSION,
+  SFTP_OPEN_TIMEOUT_MS,
+  SFTP_REQUEST_TIMEOUT_MS,
   TRANSFER_CHUNK_SIZE,
   UPLOAD_TRANSFER_CONCURRENCY,
 };