fix(sftp): keep bulk queues and large-file resume responsive (#3226)

* fix(sftp): keep bulk queues and large-file resume responsive

* fix(sftp): preserve resume compatibility and timeout cleanup

* test(sftp): verify fixture cleanup in fresh environments

* fix: address Codex review on PR #3226

* fix(sftp): remove blocking periodic history persistence

---------

Co-authored-by: netcatty-bot <308658023+netcatty-bot@users.noreply.github.com>

application/state/sftp/globalTransferScheduler.test.ts

@@ -84,3 +84,49 @@ test("queued work stays paused until resumed and can be cancelled", async () =>
   await paused;
   assert.deepEqual(order, ["b1"]);
 });
+
+test("enqueuing a large blocked batch does not repeatedly inspect the existing queue", async () => {
+  const scheduler = createGlobalSftpTransferScheduler();
+  let release: (() => void) | undefined;
+  let limitReads = 0;
+  const readLimit = () => { limitReads += 1; return 1; };
+  const blocker = scheduler.run("panel", "active", ["host"], readLimit, () => (
+    new Promise<void>((resolve) => { release = resolve; })
+  ));
+  const count = 2_000;
+  const completed: number[] = [];
+  const jobs = Array.from({ length: count }, (_, index) => scheduler.run(
+    "panel", `queued-${index}`, ["host"], readLimit,
+    async () => { completed.push(index); },
+  ));
+  await new Promise((resolve) => setImmediate(resolve));
+  const readsWhileBlocked = limitReads;
+  release?.();
+  await Promise.all([blocker, ...jobs]);
+
+  assert.deepEqual(completed, Array.from({ length: count }, (_, index) => index));
+  assert.ok(readsWhileBlocked <= count * 3,
+    `a blocked batch should be inspected linearly, got ${readsWhileBlocked} limit reads for ${count} files`);
+});
+
+test("large batches of immediately completed files yield to user input", async () => {
+  const scheduler = createGlobalSftpTransferScheduler();
+  let completed = 0;
+  const count = 1_000;
+  const inputTurn = new Promise<number>((resolve) => setTimeout(() => resolve(completed), 0));
+  const jobs = Array.from({ length: count }, (_, index) => scheduler.run(
+    "panel", `tiny-${index}`, ["host"], () => 2, async () => { completed += 1; },
+  ));
+  const completedAtInput = await inputTurn;
+  await Promise.all(jobs);
+  assert.ok(completedAtInput < count, "input must run before the entire batch drains");
+  assert.equal(completed, count);
+});
+
+test("a synchronous job failure releases its slot for queued work", async () => {
+  const scheduler = createGlobalSftpTransferScheduler();
+  const failed = scheduler.run("panel", "failed", ["host"], () => 1, () => { throw new Error("read failed"); });
+  const next = scheduler.run("panel", "next", ["host"], () => 1, async () => "completed");
+  await assert.rejects(failed, /read failed/);
+  assert.equal(await next, "completed");
+});

application/state/sftp/globalTransferScheduler.ts

@@ -44,6 +44,25 @@ export function createGlobalSftpTransferScheduler(): GlobalSftpTransferScheduler
   let lastOwnerId: string | null = null;
   let prioritySequence = 0;
   const pausedJobs = new Map<string, ScheduledJob<unknown>>();
+  let pumpScheduled = false;
+  let startsSinceYield = 0;
+
+  const schedulePump = () => {
+    if (pumpScheduled || queue.length === 0) return;
+    pumpScheduled = true;
+    const run = () => {
+      pumpScheduled = false;
+      pump();
+    };
+    // Coalesce a discovery/enqueue burst into one scan. Periodically yield to
+    // input/paint as well, including when many tiny jobs resolve immediately.
+    if (startsSinceYield >= 64) {
+      startsSinceYield = 0;
+      setTimeout(run, 0);
+    } else {
+      queueMicrotask(run);
+    }
+  };
 
   const normalizeResourceKeys = (keys: readonly string[]) => [...new Set(keys.length > 0 ? keys : ["local"])];
   const canRun = (job: ScheduledJob<unknown>) => {
@@ -60,24 +79,31 @@ export function createGlobalSftpTransferScheduler(): GlobalSftpTransferScheduler
 
   const pump = () => {
     while (queue.length > 0) {
-      const runnable = queue.map((job, index) => ({ job, index })).filter(({ job }) => canRun(job));
-      if (runnable.length === 0) return;
-      const highestPriority = runnable.reduce((max, { job }) => Math.max(max, job.priority), 0);
-      const prioritizedIndexes = queue
-        .map((job, index) => ({ job, index }))
-        .filter(({ job }) => job.priority === highestPriority && canRun(job));
-      const alternate = lastOwnerId === null
-        ? undefined
-        : prioritizedIndexes.find(({ job }) => job.ownerId !== lastOwnerId);
-      const index = alternate?.index ?? prioritizedIndexes[0]?.index ?? 0;
+      let index = -1;
+      for (let candidateIndex = 0; candidateIndex < queue.length; candidateIndex += 1) {
+        const candidate = queue[candidateIndex];
+        if (!canRun(candidate)) continue;
+        const selected = queue[index];
+        if (!selected || candidate.priority > selected.priority || (
+          candidate.priority === selected.priority
+          && selected.ownerId === lastOwnerId
+          && candidate.ownerId !== lastOwnerId
+        )) index = candidateIndex;
+      }
+      if (index < 0) return;
       const [job] = queue.splice(index, 1);
       if (!job) return;
       adjustActive(job, 1);
       lastOwnerId = job.ownerId;
-      void job.work().then(job.resolve, job.reject).finally(() => {
+      startsSinceYield += 1;
+      void (async () => job.work())().then(job.resolve, job.reject).finally(() => {
         adjustActive(job, -1);
-        pump();
+        schedulePump();
       });
+      if (startsSinceYield >= 64) {
+        schedulePump();
+        return;
+      }
     }
   };
 
@@ -85,7 +111,8 @@ export function createGlobalSftpTransferScheduler(): GlobalSftpTransferScheduler
     run<T>(ownerId: string, taskId: string, resourceKeys: readonly string[], readLimit: LimitReader, work: () => Promise<T>): Promise<T> {
       return new Promise<T>((resolve, reject) => {
         queue.push({ ownerId, taskId, resourceKeys: normalizeResourceKeys(resourceKeys), priority: 0, readLimit, work, resolve, reject } as ScheduledJob<unknown>);
-        pump();
+        if (queue.length === 1 && !pumpScheduled && startsSinceYield < 64) pump();
+        else schedulePump();
       });
     },
     prioritize(taskId: string) {
@@ -93,7 +120,7 @@ export function createGlobalSftpTransferScheduler(): GlobalSftpTransferScheduler
       if (!job) return;
       prioritySequence += 1;
       job.priority = prioritySequence;
-      pump();
+      schedulePump();
     },
     pause(taskId: string) {
       const index = queue.findIndex((job) => job.taskId === taskId);
@@ -108,7 +135,7 @@ export function createGlobalSftpTransferScheduler(): GlobalSftpTransferScheduler
       if (!job) return false;
       pausedJobs.delete(taskId);
       queue.push(job);
-      pump();
+      schedulePump();
       return true;
     },
     cancel(taskId: string) {
@@ -117,7 +144,7 @@ export function createGlobalSftpTransferScheduler(): GlobalSftpTransferScheduler
       if (!job) return false;
       pausedJobs.delete(taskId);
       job.reject(new Error("Transfer cancelled"));
-      pump();
+      schedulePump();
       return true;
     },
   };

components/GlobalSftpTransferCenter.tsx

@@ -17,6 +17,7 @@ import {
   X,
 } from "lucide-react";
 import React, { useEffect, useMemo, useRef, useState, useSyncExternalStore } from "react";
+import { useVirtualizer } from "@tanstack/react-virtual";
 
 import { useI18n } from "../application/i18n/I18nProvider";
 import {
@@ -341,12 +342,17 @@ function formatTransferPathLine(task: Pick<TransferTask, "sourcePath" | "targetP
 function TransferRow({
   task,
   childTasks = [],
+  expanded,
+  onToggleExpanded,
+  isLast,
 }: {
   task: TransferTask;
   childTasks?: readonly TransferTask[];
+  expanded: boolean;
+  onToggleExpanded: () => void;
+  isLast: boolean;
 }) {
   const { t } = useI18n();
-  const [expanded, setExpanded] = useState(false);
   const folderReplaceWarningId = React.useId();
   // Optimistic spinner from click until store status moves off paused/interrupted.
   const [resumeClicked, setResumeClicked] = useState(false);
@@ -478,7 +484,7 @@ function TransferRow({
 
   return (
     <div
-      className="border-b border-border/40 px-3 py-2.5 last:border-b-0 hover:bg-muted/30"
+      className={cn("border-border/40 px-3 py-2.5 hover:bg-muted/30", isLast ? "border-b-0" : "border-b")}
       data-section="global-sftp-transfer-row"
       data-transfer-status={task.status}
       data-directory-parent={isDirParent ? "true" : undefined}
@@ -506,7 +512,7 @@ function TransferRow({
           {canToggleChildren && (
             <TransferAction
               label={expanded ? t("sftp.transfers.collapseChildren") : t("sftp.transfers.expandChildren")}
-              onClick={() => setExpanded((value) => !value)}
+              onClick={onToggleExpanded}
             >
               {expanded ? <ChevronUp size={13} /> : <ChevronDown size={13} />}
             </TransferAction>
@@ -755,6 +761,59 @@ function TransferRow({
   );
 }
 
+function TransferList({ tasks, childrenByParent, empty }: {
+  tasks: readonly TransferTask[];
+  childrenByParent: ReadonlyMap<string, TransferTask[]>;
+  empty: React.ReactNode;
+}) {
+  const scrollRef = useRef<HTMLDivElement>(null);
+  // Keep folder expansion when a row scrolls out of the mounted viewport.
+  const [expandedIds, setExpandedIds] = useState<ReadonlySet<string>>(() => new Set());
+  const virtual = tasks.length > 20;
+  const virtualizer = useVirtualizer({
+    count: tasks.length,
+    getScrollElement: () => scrollRef.current,
+    estimateSize: () => 112,
+    getItemKey: (index) => tasks[index].id,
+    overscan: 4,
+    enabled: virtual,
+  });
+  const renderRow = (task: TransferTask, index: number) => (
+    <TransferRow
+      key={task.id}
+      task={task}
+      childTasks={childrenByParent.get(task.id) ?? []}
+      expanded={expandedIds.has(task.id)}
+      isLast={index === tasks.length - 1}
+      onToggleExpanded={() => setExpandedIds((previous) => {
+        const next = new Set(previous);
+        if (next.has(task.id)) next.delete(task.id);
+        else next.add(task.id);
+        return next;
+      })}
+    />
+  );
+  return (
+    <div ref={scrollRef} className="max-h-[460px] overflow-auto" data-section="global-sftp-transfer-list">
+      {tasks.length === 0 ? empty : !virtual ? tasks.map(renderRow) : (
+        <div className="relative w-full" style={{ height: virtualizer.getTotalSize() }}>
+          {virtualizer.getVirtualItems().map((row) => (
+            <div
+              key={row.key}
+              ref={virtualizer.measureElement}
+              data-index={row.index}
+              className="absolute left-0 top-0 w-full"
+              style={{ transform: `translateY(${row.start}px)` }}
+            >
+              {renderRow(tasks[row.index], row.index)}
+            </div>
+          ))}
+        </div>
+      )}
+    </div>
+  );
+}
+
 export function GlobalSftpTransferCenter() {
   const { t } = useI18n();
   // Badge is a stable store snapshot (identity unchanged on pure progress).
@@ -866,20 +925,17 @@ export function GlobalSftpTransferCenter() {
           ))}
         </div>
 
-        <div className="max-h-[460px] overflow-auto">
-          {displayed.length === 0 ? (
+        <TransferList
+          key={`${bucket}-${showBackground}`}
+          tasks={displayed}
+          childrenByParent={childrenByParent}
+          empty={(
             <div className="flex h-40 flex-col items-center justify-center text-muted-foreground">
               {badge.hasAttention && bucket !== "attention" && bucket !== "all" ? <AlertCircle size={22} /> : <ArrowDownUp size={22} />}
               <span className="mt-2 text-xs">{t("sftp.transferCenter.empty")}</span>
             </div>
-          ) : displayed.map((task) => (
-            <TransferRow
-              key={task.id}
-              task={task}
-              childTasks={childrenByParent.get(task.id) ?? []}
-            />
-          ))}
-        </div>
+          )}
+        />
 
         {(() => {
           const showBackgroundToggle = collapsed.length > 0;

components/GlobalSftpTransferCenter.virtualization.test.tsx

@@ -0,0 +1,67 @@
+import assert from "node:assert/strict";
+import test from "node:test";
+import React from "react";
+import { createDomRenderer, dispatchDomEvent, flushEffects, installDomEnvironment } from "./test-support/renderReactDom.tsx";
+
+test("large transfer histories mount only visible rows and can scroll to the last file", async (t) => {
+  const env = installDomEnvironment();
+  const previous = new Map<string, PropertyDescriptor | undefined>();
+  const globals = {
+    MutationObserver: env.window.MutationObserver,
+    NodeFilter: env.window.NodeFilter,
+    HTMLInputElement: env.window.HTMLInputElement,
+    Element: env.window.Element,
+    requestAnimationFrame: (callback: FrameRequestCallback) => setTimeout(() => callback(Date.now()), 0),
+    cancelAnimationFrame: clearTimeout,
+  };
+  for (const [key, value] of Object.entries(globals)) {
+    previous.set(key, Object.getOwnPropertyDescriptor(globalThis, key));
+    Object.defineProperty(globalThis, key, { configurable: true, writable: true, value });
+  }
+  Object.defineProperties(env.window.HTMLElement.prototype, {
+    offsetWidth: { configurable: true, get: () => 460 },
+    offsetHeight: { configurable: true, get() { return this.dataset.section === "global-sftp-transfer-list" ? 460 : 112; } },
+  });
+  const { I18nProvider } = await import("../application/i18n/I18nProvider.tsx");
+  const { sftpTransferCenterStore: store } = await import("../application/state/sftpTransferCenterStore.ts");
+  const { GlobalSftpTransferCenter } = await import("./GlobalSftpTransferCenter.tsx");
+  const { TooltipProvider } = await import("./ui/tooltip.tsx");
+  const renderer = await createDomRenderer(env.document);
+  t.after(async () => {
+    await renderer.unmount();
+    for (const [key, descriptor] of previous) {
+      if (descriptor) Object.defineProperty(globalThis, key, descriptor);
+      else Reflect.deleteProperty(globalThis, key);
+    }
+    env.cleanup();
+  });
+  store.publishOwner("virtual-list-test", Array.from({ length: 1_000 }, (_, index) => ({
+    id: `virtual-file-${index}`, fileName: `file-${index}.bin`,
+    sourcePath: `/source/${index}`, targetPath: `/target/${index}`,
+    sourceConnectionId: "source", targetConnectionId: "local", direction: "download" as const,
+    status: "queued" as const, totalBytes: 1_000, transferredBytes: 0, speed: 0,
+    startTime: index + 1, isDirectory: false,
+  })));
+  await renderer.render(<I18nProvider locale="en"><TooltipProvider><GlobalSftpTransferCenter /></TooltipProvider></I18nProvider>);
+  const toggle = env.document.querySelector("[data-section=global-sftp-transfer-toggle]");
+  assert.ok(toggle);
+  await dispatchDomEvent(toggle, new env.window.MouseEvent("click", { bubbles: true }));
+  await flushEffects();
+  const rows = () => env.document.querySelectorAll("[role=progressbar]");
+  assert.ok(rows().length > 0 && rows().length < 40, `rendered ${rows().length} rows`);
+  assert.ok(env.document.querySelector('[role=progressbar][aria-label="file-999.bin"]'));
+  const firstRow = env.document.querySelector('[data-transfer-status="queued"]');
+  assert.ok(firstRow);
+  assert.equal(firstRow.matches('.last\\:border-b-0:last-child'), false, "a virtual row must not lose its divider just because it has a wrapper");
+
+  const scroll = env.document.querySelector<HTMLElement>("[data-section=global-sftp-transfer-list]");
+  assert.ok(scroll);
+  scroll.scrollTop = 112 * 1_000 - 460;
+  await dispatchDomEvent(scroll, new env.window.Event("scroll"));
+  await flushEffects();
+  assert.ok(rows().length > 0 && rows().length < 40);
+  assert.ok(env.document.querySelector('[role=progressbar][aria-label="file-0.bin"]'));
+  const lastRow = env.document.querySelector('[role=progressbar][aria-label="file-0.bin"]')?.closest('[data-transfer-status]');
+  assert.ok(lastRow?.classList.contains("border-b-0"), "only the actual final row omits its divider");
+  assert.equal(store.getSnapshot().tasks.length, 1_000, "offscreen files must remain queued");
+});

docs/research/issues-3213-3155-sftp-responsiveness.md

@@ -0,0 +1,114 @@
+# SFTP responsiveness: issues #3213 and #3155
+
+Investigated on 2026-08-31 against `39d7c38a6acea2f59524566d117345e1ced21fbd`.
+
+## Evidence and scope
+
+[#3213](https://github.com/binaricat/Netcatty/issues/3213) reports slow/unstable
+large-file transfers, ineffective pause/resume and failed recovery after killing
+the app. [#3155](https://github.com/binaricat/Netcatty/issues/3155) reports a freeze
+with many files. Neither report supplies a direction, server configuration or
+transfer log. The following are reproduced code defects, not proof that every
+reported symptom has one cause. Keep both issues open for reporter confirmation.
+
+1. `globalTransferScheduler.run` scanned the entire waiting queue on every
+   insertion. Enqueuing 10,000 jobs behind two active jobs took 1,237 ms and
+   49,985,005 limit checks locally. Coalescing queue pumps reduces the same case
+   to 10,001 checks (3-6 ms in local runs). Priority, owner fairness and per-host
+   limits remain unchanged; immediately completed batches also yield to input.
+2. The transfer-center popover mounted every top-level row, even far below the
+   viewport. The actual Electron component took about 1,885 ms to display 1,000
+   rows. A measured, bounded viewport mounts around nine rows instead. All jobs
+   remain in the store; scrolling and bucket selection still expose them.
+   Folder expansion is held outside the recycled row.
+3. Remote source/prefix verification used ssh2's serial `createReadStream`,
+   even though body downloads already used pipelined reads. Pause captures a
+   complete identity; resume verifies it before continuing. A 128 MiB loopback
+   SFTP test with delayed READ replies spent 41,493 ms in resume on the old code.
+   Reuse the existing 64-request, 32 KiB, ordered SHA-256 verification helper.
+   This still checks every required byte with bounded memory, cancellation and
+   inactivity deadlines. It does not substitute metadata, sampling or file size
+   for content verification.
+
+## Correctness boundaries retained
+
+- Only contiguous acknowledged ranges are checkpoints; aggregate displayed
+  progress and sparse file length are not safe offsets.
+- First-run force-kill recovery still restarts at zero when no complete source
+  identity was captured. Recovery persistence is unchanged: progress since the
+  last lifecycle save can still be lost. This PR improves the verified resume
+  path's latency, not every whole-app force-kill recovery scenario.
+- SCP and legacy fastPut paths do not gain unsupported pause/resume.
+- Source changes, staged-prefix mismatch, missing staging, cancellation,
+  replacement, permissions and conflict handling retain their existing checks.
+- No claim is made about arbitrary server power-loss durability, every Windows
+  server, or a universal throughput multiplier.
+
+## Mature-client comparison
+
+- **FileZilla:** official SVN revision 11556 consumes directory results in
+  batches and posts a continuation to the UI loop; its queue fills available
+  transfer slots under total/directional/site limits. Adopt cooperative work
+  and bounded admission, not an unrestricted `Promise.all` or a larger packet.
+  Sources: [directory consumption, lines 69-129](https://svn.filezilla-project.org/svn/!svn/bc/11556/FileZilla3/trunk/src/interface/local_recursive_operation.cpp),
+  [queue admission, lines 544-632 and 2452-2493](https://svn.filezilla-project.org/svn/!svn/bc/11556/FileZilla3/trunk/src/interface/QueueView.cpp).
+- **OpenSSH:** 32 KiB requests and a 64-request default window; interruption
+  stops new requests, drains replies and tracks the contiguous acknowledged
+  prefix separately from the highest acknowledged position. Its manual warns
+  that mismatched partial content can corrupt a resumed file. Netcatty keeps
+  its stronger content checks. Sources: [defaults and transfer loops](https://github.com/openssh/openssh-portable/blob/0ef0f5a839831c213f24e3f2ae434765c607fb50/sftp-client.c#L59-L63),
+  [resume warning](https://man.openbsd.org/sftp.1).
+- **WinSCP:** two simultaneous background operations is also its default;
+  eligible transfers use temporary files before publication. These choices do
+  not establish that two files is Netcatty's bottleneck, or that every killed
+  process can safely resume. Sources: [background queue](https://winscp.net/eng/docs/transfer_queue),
+  [resume requirements and temporary-file tradeoffs](https://winscp.net/eng/docs/resume).
+- **rclone:** documents the same 32 KiB/64-request defaults, server compatibility
+  concerns and possible deadlock when checks/transfers compete for a capped
+  connection pool. Future adaptive windows or a server compatibility matrix
+  should be separate measured work. Source: [SFTP documentation](https://rclone.org/sftp/).
+
+## Reproduction and verification
+
+The regression tests exercise scheduler admission/order/yielding, complete remote prefix
+verification, and actual transfer-center DOM bounds/scrolling. Each new defect
+was observed failing before its fix.
+
+The opt-in real SSH/SFTP fixture listens only on loopback, uses generated test
+credentials and isolates all temp/home state. It creates a saved prefix, checks
+resume and (for sufficiently long transfers) live pause/resume, then checks the
+complete output SHA-256. It is not a simulation of a whole-app force-kill.
+
+```sh
+NETCATTY_SFTP_LIVE=1 SFTP_LIVE_MIB=128 SFTP_LIVE_FILES=12 \
+  node scripts/sftp-transfer-resume.live.test.cjs
+NETCATTY_SFTP_LIVE=1 SFTP_LIVE_MIB=128 \
+  SFTP_LIVE_BASELINE_REF=39d7c38a6 \
+  node scripts/sftp-transfer-resume.live.test.cjs
+```
+
+Twelve 128 MiB files completed with matching hashes, two submitted at a time.
+Each exercised pause/resume: pause acknowledgements took 6-15 ms; resume took
+about 1.45-3.18 seconds in this run. These are fixture-specific observations,
+not performance promises. Electron verification also exercised the production
+popover, pause-all/resume-all, scrolling to the last file, bucket switching and
+20,000-job admission while the popover was open.
+
+## Review follow-up
+
+- Preserve serial prefix verification when a server rejects range OPEN/READ
+  with an ordinary protocol error. Cancellation and request timeouts do not
+  fall back; timed-out channels are abandoned before another attempt.
+- Count every positive short READ as inactivity-watchdog activity, without
+  changing ordered full-range hashing. Finished windows cannot publish late
+  progress, rearm their watchdog or issue another partial READ.
+- Allocate the live fixture inside the managed temp directory, then isolate
+  its own staging/home state. Cleanup removes only that unique fixture child,
+  including when fixture setup fails.
+- Keep transfer-row separators based on the actual list position, not the
+  temporary viewport wrappers. The final list row alone omits its separator.
+- Remove the proposed periodic full-history save: with 20,000 queued files,
+  serialization alone occupied about 80 ms every five seconds. A compact journal
+  would additionally need cross-window ownership and retry-attempt coordination.
+  Those recovery semantics are not changed in this focused responsiveness PR.
+  There is no new persistence timer, storage key or schema migration.

electron/bridges/transferBridge.cjs

@@ -900,6 +900,24 @@ async function hashRemotePrefix(client, sftpId, filePath, encoding, bytes, optio
   if (isScpModeClient(client)) return null;
   await requireSftpChannel(client, { signal: options?.signal });
   const encodedPath = encodePathForSession(sftpId, filePath, encoding);
+  // Resume still hashes the entire required prefix. Use the same bounded READ
+  // window as downloads instead of ssh2's serial stream, which makes a large
+  // pause/restart verification pay one network round trip per read.
+  try {
+    const rangeDigest = await hashRemotePrefixWithSftpRanges(client, encodedPath, bytes, options);
+    if (rangeDigest !== null) return rangeDigest;
+  } catch (error) {
+    if (error?.sftpRequestTimedOut) {
+      error.noTransferFallback = true;
+      abandonWedgedVerificationSftpChannel(client);
+      throw error;
+    }
+    if (options?.signal?.aborted || error?.code === "ABORT_ERR") throw error;
+    // Some servers reject concurrent READs while supporting serial streams.
+    // Retain the existing compatibility path, without bypassing cancellation
+    // or retrying a request timeout on a channel that may still own the request.
+  }
+  await requireSftpChannel(client, { signal: options?.signal });
   return hashReadable(
     client.sftp.createReadStream(encodedPath, { start: 0, end: bytes - 1 }),
     options,
@@ -3327,6 +3345,7 @@ async function readSftpRange(sftp, handle, buffer, position, length, options = {
     if (bytesRead <= 0) {
       throw new Error("Download stream finished before the full source was received");
     }
+    options.onRead?.(bytesRead);
     received += bytesRead;
   }
 }
@@ -3470,12 +3489,13 @@ async function hashRemotePrefixWithSftpRanges(client, remotePath, bytes, options
       // on a slow/serialized server.
       let inactivityTimer = null;
       let rejectInactivity = null;
+      let windowActive = true;
       const clearInactivity = () => {
         if (inactivityTimer) clearTimeout(inactivityTimer);
         inactivityTimer = null;
       };
       const armInactivity = () => {
-        if (!(readTimeoutMs > 0) || options.signal?.aborted || abortGate.aborted) return;
+        if (!windowActive || !(readTimeoutMs > 0) || options.signal?.aborted || abortGate.aborted) return;
         clearInactivity();
         inactivityTimer = setTimeout(() => {
           const error = new Error(`SFTP READ timed out after ${readTimeoutMs} ms`);
@@ -3497,15 +3517,22 @@ async function hashRemotePrefixWithSftpRanges(client, remotePath, bytes, options
             const buffer = Buffer.allocUnsafe(length);
             await readSftpRange(sftp, handle, buffer, position, length, {
               abortGate,
+              onRead: () => {
+                // Short READ replies are real activity too. Do not start the
+                // next partial read after this window has already failed.
+                if (!windowActive) throw new Error("SFTP verification window ended");
+                armInactivity();
+              },
             });
+            if (!windowActive) return;
             windowBuffers[offset] = buffer;
             completedBytes += length;
             options.onProgress?.(completedBytes);
-            armInactivity();
           })),
           ...(inactivityWait ? [inactivityWait] : []),
         ]);
       } finally {
+        windowActive = false;
         clearInactivity();
         rejectInactivity = null;
       }
@@ -7447,5 +7474,6 @@ module.exports = {
   _assertSourceMetadataUnchangedForTests: assertSourceMetadataUnchanged,
   _assertDownloadSourceAfterTransferForTests: assertDownloadSourceAfterTransfer,
   _assertLocalDownloadMatchesRemotePrefixForTests: assertLocalDownloadMatchesRemotePrefix,
+  _hashRemotePrefixForTests: hashRemotePrefix,
   _remoteOpenPathMatchesStagedForTests: remoteOpenPathMatchesStaged,
 };

electron/bridges/transferBridge.test.cjs

@@ -1876,6 +1876,167 @@ test("pause acknowledges quickly then publishes a full source identity", async (
   assert.equal((await running).error, undefined);
 });
 
+test("remote restart verifies the complete source with bounded concurrent reads", async (t) => {
+  const tempDir = await fs.promises.mkdtemp(`${tempDirBridge.getTempFilePath("resume-window")}-`);
+  t.after(async () => { await fs.promises.rm(tempDir, { recursive: true, force: true }); });
+  const transferId = `resume-window-${crypto.randomUUID()}`;
+  const payload = Buffer.alloc(2 * 1024 * 1024 + 17);
+  for (let index = 0; index < payload.length; index += 1) payload[index] = index % 251;
+  const checkpoint = 512 * 1024;
+  const targetPath = path.join(tempDir, "target.bin");
+  const stagePath = tempDirBridge.getTransferTempFilePath(transferId, "target.bin");
+  await fs.promises.writeFile(stagePath, payload.subarray(0, checkpoint));
+  t.after(async () => { await fs.promises.unlink(stagePath).catch(() => {}); });
+  const sender = createSender();
+  let verificationReads = 0;
+  let maxVerificationReads = 0;
+  let verificationBytes = 0;
+  const { sftp } = createPipelinedDownloadSftp(payload, {
+    read(_handle, buffer, offset, length, position, callback) {
+      const verifying = sender.sent.at(-1)?.payload.phase === "verifying";
+      if (verifying) {
+        verificationReads += 1;
+        maxVerificationReads = Math.max(maxVerificationReads, verificationReads);
+      }
+      setTimeout(() => {
+        const bytes = Math.min(length, payload.length - position);
+        payload.copy(buffer, offset, position, position + bytes);
+        if (verifying) {
+          verificationReads -= 1;
+          verificationBytes += bytes;
+        }
+        callback(null, bytes);
+      }, 2);
+    },
+  });
+  const client = { sftp, stat: async () => ({ size: payload.length }) };
+  transferBridge.init({ sftpClients: new Map([["source", client]]) });
+  const result = await transferBridge.startTransfer({ sender }, {
+    transferId, sourcePath: "/source.bin", targetPath,
+    sourceType: "sftp", targetType: "local", sourceSftpId: "source",
+    totalBytes: payload.length, resumable: true, checkpointBytes: checkpoint,
+    sourceFingerprint: `sha256:p${payload.length}:${crypto.createHash("sha256").update(payload).digest("hex")}`,
+  });
+  assert.equal(result.error, undefined, result.error);
+  assert.deepEqual(await fs.promises.readFile(targetPath), payload);
+  assert.ok(maxVerificationReads > 1, "large resume verification must not wait for one network read at a time");
+  assert.ok(maxVerificationReads <= DOWNLOAD_TRANSFER_CONCURRENCY);
+  assert.ok(verificationBytes >= payload.length, "verify every source byte, not a sample");
+});
+
+for (const scenario of ["read-error", "open-error", "changed-content", "cancelled", "timeout"]) {
+  test(`remote restart prefix compatibility: ${scenario}`, async (t) => {
+    const tempDir = await fs.promises.mkdtemp(`${tempDirBridge.getTempFilePath("resume-compat")}-`);
+    const transferId = `resume-compat-${crypto.randomUUID()}`;
+    const payload = Buffer.alloc(3 * TRANSFER_CHUNK_SIZE, 71);
+    const targetPath = path.join(tempDir, "target.bin");
+    const stagePath = tempDirBridge.getTransferTempFilePath(transferId, "target.bin");
+    t.after(async () => {
+      await fs.promises.rm(tempDir, { recursive: true, force: true });
+      await fs.promises.rm(stagePath, { force: true });
+    });
+    await fs.promises.writeFile(stagePath, payload);
+    let streamOpens = 0;
+    const protocolError = new Error(scenario === "cancelled" ? "Transfer cancelled" : `Range ${scenario}`);
+    protocolError.code = scenario === "cancelled" ? "ABORT_ERR" : scenario === "timeout" ? "SFTP_READ_TIMEOUT" : 4;
+    if (scenario === "timeout") protocolError.sftpRequestTimedOut = true;
+    const { sftp } = createPipelinedDownloadSftp(payload, {
+      open(_path, _flags, callback) {
+        if (scenario === "open-error") callback(protocolError);
+        else callback(null, Buffer.from("prefix-handle"));
+      },
+      read(_handle, _buffer, _offset, _length, _position, callback) {
+        setImmediate(() => callback(protocolError));
+      },
+      createReadStream(_path, options) {
+        streamOpens += 1;
+        assert.equal(options.start, 0);
+        assert.equal(options.end, payload.length - 1);
+        const current = Buffer.from(payload);
+        if (scenario === "changed-content") current[current.length - 1] ^= 1;
+        return Readable.from([current]);
+      },
+    });
+    transferBridge.init({ sftpClients: new Map([["source", {
+      sftp, stat: async () => ({ size: payload.length }),
+    }]]) });
+    const result = await transferBridge.startTransfer({ sender: createSender() }, {
+      transferId, sourcePath: "/source.bin", targetPath,
+      sourceType: "sftp", targetType: "local", sourceSftpId: "source",
+      totalBytes: payload.length, resumable: true, checkpointBytes: payload.length,
+      sourceFingerprint: `sha256:p${payload.length}:${crypto.createHash("sha256").update(payload).digest("hex")}`,
+    });
+    if (scenario === "cancelled" || scenario === "timeout") {
+      assert.match(result.error || "", /cancelled|Range timeout/);
+      assert.equal(streamOpens, 0, "cancellation and timeouts must not start a fallback");
+    } else {
+      assert.ok(streamOpens > 0, "ordinary protocol errors must retain stream compatibility");
+      if (scenario === "changed-content") {
+        assert.match(result.error || "", /source.*changed|source.*match/i);
+        assert.equal(fs.existsSync(targetPath), false, "fallback must still reject changed content");
+      } else {
+        assert.equal(result.error, undefined, result.error);
+        assert.deepEqual(await fs.promises.readFile(targetPath), payload);
+      }
+    }
+  });
+}
+
+test("remote prefix timeout abandons its channel and ignores late replies", async () => {
+  const pending = [];
+  let ended = false;
+  let progress = 0;
+  const { sftp } = createPipelinedDownloadSftp(Buffer.alloc(2 * TRANSFER_CHUNK_SIZE), {
+    read(_handle, buffer, _offset, length, _position, callback) {
+      pending.push(() => { buffer.fill(0); callback(null, length); });
+    },
+    end() { ended = true; },
+    createReadStream() { throw new Error("must not fall back after a timeout"); },
+  });
+  const client = { sftp };
+  await assert.rejects(transferBridge._hashRemotePrefixForTests(
+    client, "source", "/source.bin", undefined, 2 * TRANSFER_CHUNK_SIZE,
+    { sftpReadTimeoutMs: 20, onProgress: () => { progress += 1; } },
+  ), (error) => error.code === "SFTP_READ_TIMEOUT" && error.noTransferFallback === true);
+  assert.equal(ended, true);
+  assert.equal(client.sftp, null);
+  assert.equal(pending.length, 2);
+  for (const reply of pending) reply();
+  await new Promise((resolve) => setImmediate(resolve));
+  assert.equal(progress, 0, "finished verification must ignore late window progress");
+});
+
+test("remote prefix watchdog observes every positive short READ", async (t) => {
+  t.mock.timers.enable({ apis: ["setTimeout"] });
+  const payload = Buffer.alloc(TRANSFER_CHUNK_SIZE, 73);
+  let replies = 0;
+  const { sftp } = createPipelinedDownloadSftp(payload, {
+    read(_handle, buffer, offset, length, position, callback) {
+      setTimeout(() => {
+        const count = Math.min(1024, length);
+        payload.copy(buffer, offset, position, position + count);
+        replies += 1;
+        callback(null, count);
+      }, 5);
+    },
+    createReadStream() { throw new Error("responsive range reads must not fall back"); },
+  });
+  const hashing = transferBridge._hashRemotePrefixForTests(
+    { sftp }, "source", "/source.bin", undefined, payload.length, { sftpReadTimeoutMs: 25 },
+  );
+  // Handle rejection immediately so advancing the mock clock can expose a bad
+  // watchdog without generating an unrelated unhandled-rejection failure.
+  const outcome = hashing.then((digest) => ({ digest }), (error) => ({ error }));
+  for (let index = 0; index < 40; index += 1) {
+    await new Promise((resolve) => setImmediate(resolve));
+    t.mock.timers.tick(5);
+  }
+  const result = await outcome;
+  assert.equal(result.error, undefined, result.error?.message);
+  assert.equal(replies, 32);
+  assert.equal(result.digest, crypto.createHash("sha256").update(payload).digest("hex"));
+});
+
 test("remote resume identity rejects a same-size source rewrite", async (t) => {
   const tempDir = await fs.promises.mkdtemp(path.join(os.tmpdir(), "netcatty-transfer-modifytime-id-"));
   t.after(async () => {

scripts/sftp-transfer-resume.live.test.cjs

@@ -0,0 +1,184 @@
+"use strict";
+
+// Opt-in loopback SSH/SFTP fixture; never connects to a user's server.
+// NETCATTY_SFTP_LIVE=1 SFTP_LIVE_MIB=128 SFTP_LIVE_FILES=12 node scripts/sftp-transfer-resume.live.test.cjs
+
+const assert = require("node:assert/strict");
+const crypto = require("node:crypto");
+const fs = require("node:fs");
+const path = require("node:path");
+const { Server } = require("ssh2");
+const SftpClient = require("ssh2-sftp-client");
+
+async function main() {
+  const tempBridgePath = require.resolve("../electron/bridges/tempDirBridge.cjs");
+  const managedTempBridge = require(tempBridgePath);
+  const root = fs.mkdtempSync(`${managedTempBridge.getTempFilePath("sftp-live")}-`);
+  try {
+    console.log("SFTP_LIVE_ROOT", root);
+    // Keep the fixture visible to managed cleanup, but isolate its staging and
+    // identity paths from the user's app. Reload only this process's cached
+    // bridge after changing the environment; never delete the managed parent.
+    for (const key of ["TMPDIR", "TMP", "TEMP", "HOME"]) process.env[key] = root;
+    delete require.cache[tempBridgePath];
+    await runFixture(root);
+  } finally {
+    fs.rmSync(root, { recursive: true, force: true });
+  }
+}
+
+async function runFixture(root) {
+  let bridge;
+  if (process.env.SFTP_LIVE_BASELINE_REF) {
+    const Module = require("node:module");
+    const filename = path.resolve(__dirname, "../electron/bridges/transferBridge.cjs");
+    const baseline = new Module(filename, module);
+    baseline.filename = filename;
+    baseline.paths = Module._nodeModulePaths(path.dirname(filename));
+    baseline._compile(require("node:child_process").execFileSync("git", ["show", `${process.env.SFTP_LIVE_BASELINE_REF}:electron/bridges/transferBridge.cjs`], { cwd: path.resolve(__dirname, ".."), encoding: "utf8" }), filename);
+    bridge = baseline.exports;
+  } else {
+    bridge = require("../electron/bridges/transferBridge.cjs");
+  }
+  const tempDirBridge = require("../electron/bridges/tempDirBridge.cjs");
+  assert.equal(path.dirname(tempDirBridge.getTempDir()), root, "fixture staging must remain isolated");
+  const bytes = Number(process.env.SFTP_LIVE_MIB || 8) * 1024 * 1024;
+  const fileCount = Number(process.env.SFTP_LIVE_FILES || 1);
+  const payload = Buffer.allocUnsafe(bytes);
+  for (let i = 0; i < bytes; i += 1) payload[i] = i % 251;
+  const digest = crypto.createHash("sha256").update(payload).digest("hex");
+  const privateKey = crypto.generateKeyPairSync("rsa", { modulusLength: 2048 }).privateKey.export({ type: "pkcs1", format: "pem" });
+  const connections = new Set();
+  let reads = 0;
+  let activeReads = 0;
+  let peakReads = 0;
+  const server = new Server({ hostKeys: [privateKey] }, (connection) => {
+    connections.add(connection);
+    connection.on("error", () => {});
+    connection.on("close", () => connections.delete(connection));
+    connection.on("authentication", (context) => context.username === "fixture" ? context.accept() : context.reject());
+    connection.on("ready", () => connection.on("session", (accept) => {
+      const session = accept();
+      session.on("sftp", (acceptSftp) => {
+        const sftp = acceptSftp();
+        let nextHandle = 0;
+        const attrs = { size: bytes, mode: 0o100644, uid: 1, gid: 1, atime: 1700000000, mtime: 1700000000 };
+        sftp.on("error", () => {});
+        sftp.on("REALPATH", (id) => sftp.name(id, [{ filename: "/", longname: "/", attrs }]));
+        for (const operation of ["STAT", "LSTAT", "FSTAT"]) sftp.on(operation, (id) => sftp.attrs(id, attrs));
+        sftp.on("OPEN", (id) => sftp.handle(id, Buffer.from(String(nextHandle++))));
+        sftp.on("CLOSE", (id) => sftp.status(id, 0));
+        sftp.on("READ", (id, _handle, position, length) => {
+          reads += 1;
+          activeReads += 1;
+          peakReads = Math.max(peakReads, activeReads);
+          setTimeout(() => {
+            activeReads -= 1;
+            if (sftp.destroyed) return;
+            if (position >= bytes) sftp.status(id, 1);
+            else sftp.data(id, payload.subarray(position, Math.min(bytes, position + length)));
+          }, Number(process.env.SFTP_LIVE_DELAY_MS || 5));
+        });
+      });
+    }));
+  });
+  const client = new SftpClient();
+  try {
+    await new Promise((resolve, reject) => {
+      server.once("error", reject);
+      server.listen(0, "127.0.0.1", resolve);
+    });
+    await client.connect({ host: "127.0.0.1", port: server.address().port, username: "fixture", password: "fixture" });
+    bridge.init({ sftpClients: new Map([["source", client]]) });
+    const runFile = async (index) => {
+      const transferId = `live-${crypto.randomUUID()}`;
+      const targetPath = path.join(root, `output-${index}.bin`);
+      const stagePath = tempDirBridge.getTransferTempFilePath(transferId, path.basename(targetPath));
+      const checkpoint = Math.floor(bytes / 4);
+      fs.writeFileSync(stagePath, payload.subarray(0, checkpoint));
+      let verificationStart = 0;
+      let verificationMs = 0;
+      let requestedPause = false;
+      let pauseResult;
+      const startedAt = performance.now();
+      const running = bridge.startTransfer({ sender: { send(_channel, event) {
+        if (event.phase === "verifying" && !verificationStart) verificationStart = performance.now();
+        if (event.phase !== "verifying" && verificationStart) {
+          verificationMs += performance.now() - verificationStart;
+          verificationStart = 0;
+        }
+        if (!requestedPause && event.transferred > bytes / 2 && event.transferred < bytes && event.phase === "transferring") {
+          requestedPause = true;
+          const pauseStarted = performance.now();
+          pauseResult = bridge.pauseTransfer(null, { transferId }).then(async (result) => {
+            assert.equal(result.success, true, result.reason);
+            const pauseMs = performance.now() - pauseStarted;
+            const resumeStarted = performance.now();
+            const resumed = await bridge.resumeTransfer(null, { transferId });
+            assert.equal(resumed.success, true, resumed.reason);
+            return { pauseMs: Math.round(pauseMs), resumeMs: Math.round(performance.now() - resumeStarted) };
+          });
+        }
+      } } }, {
+        transferId, sourcePath: "/source.bin", targetPath,
+        sourceType: "sftp", targetType: "local", sourceSftpId: "source",
+        totalBytes: bytes, resumable: true, checkpointBytes: checkpoint,
+        sourceFingerprint: `sha256:p${bytes}:${digest}`,
+      });
+      const result = await running;
+      assert.equal(result.error, undefined, result.error);
+      const control = await pauseResult;
+      const outputDigest = crypto.createHash("sha256");
+      for await (const chunk of fs.createReadStream(targetPath)) outputDigest.update(chunk);
+      assert.equal(outputDigest.digest("hex"), digest);
+      fs.unlinkSync(targetPath);
+      return { index, bytes, elapsedMs: Math.round(performance.now() - startedAt), verificationMs: Math.round(verificationMs), ...control };
+    };
+    const results = [];
+    for (let index = 0; index < fileCount; index += 2) {
+      results.push(...await Promise.all(Array.from({ length: Math.min(2, fileCount - index) }, (_, offset) => runFile(index + offset))));
+    }
+    console.log("SFTP_LIVE_PASS", JSON.stringify({ results, reads, peakReads }));
+  } finally {
+    await client.end().catch(() => {});
+    for (const connection of connections) connection.end();
+    await new Promise((resolve) => server.close(resolve));
+  }
+}
+
+if (process.env.NETCATTY_SFTP_LIVE === "1") {
+  const watchdog = setTimeout(() => { console.error("SFTP_LIVE_FAIL timeout"); process.exit(1); }, 120_000);
+  main().catch((error) => { console.error(error); process.exitCode = 1; }).finally(() => clearTimeout(watchdog));
+} else {
+  const test = require("node:test");
+  for (const existingParent of [false, true]) {
+    test(`live SFTP fixture cleans setup failure with ${existingParent ? "existing" : "fresh"} managed parent`, (t) => {
+      const tempDirBridge = require("../electron/bridges/tempDirBridge.cjs");
+      const testRoot = fs.mkdtempSync(`${tempDirBridge.getTempFilePath("sftp-live-setup-test")}-`);
+      t.after(() => fs.rmSync(testRoot, { recursive: true, force: true }));
+      const managedParent = path.join(testRoot, "Netcatty");
+      if (existingParent) fs.mkdirSync(managedParent, { mode: 0o700 });
+      const result = require("node:child_process").spawnSync(process.execPath, [__filename], {
+        env: {
+          ...process.env,
+          ...Object.fromEntries(["TMPDIR", "TMP", "TEMP", "HOME"].map((key) => [key, testRoot])),
+          NETCATTY_SFTP_LIVE: "1", SFTP_LIVE_MIB: "-1", SFTP_LIVE_BASELINE_REF: "",
+        },
+        encoding: "utf8",
+        timeout: 10_000,
+      });
+      assert.equal(result.status, 1, result.stderr);
+      assert.match(result.stderr, /RangeError/);
+      const match = result.stdout.match(/^SFTP_LIVE_ROOT ([^\r\n]+)/m);
+      assert.ok(match, result.stdout);
+      const fixtureRoot = match[1];
+      assert.match(path.basename(fixtureRoot), /sftp-live/);
+      assert.equal(path.dirname(fixtureRoot), managedParent);
+      assert.equal(fs.existsSync(fixtureRoot), false);
+      assert.equal(fs.existsSync(path.dirname(fixtureRoot)), true, "must not delete the shared managed parent");
+    });
+  }
+  test("large-file SFTP resume over a loopback SSH connection", {
+    skip: "set NETCATTY_SFTP_LIVE=1 to run the real connection fixture",
+  }, () => {});
+}