diff --git a/application/state/sftp/globalTransferScheduler.test.ts b/application/state/sftp/globalTransferScheduler.test.ts index d53f0b008..0d7f3ae12 100644 --- a/application/state/sftp/globalTransferScheduler.test.ts +++ b/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((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((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"); +}); diff --git a/application/state/sftp/globalTransferScheduler.ts b/application/state/sftp/globalTransferScheduler.ts index a1fbe15c3..a4a73ca8a 100644 --- a/application/state/sftp/globalTransferScheduler.ts +++ b/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>(); + 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) => { @@ -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(ownerId: string, taskId: string, resourceKeys: readonly string[], readLimit: LimitReader, work: () => Promise): Promise { return new Promise((resolve, reject) => { queue.push({ ownerId, taskId, resourceKeys: normalizeResourceKeys(resourceKeys), priority: 0, readLimit, work, resolve, reject } as ScheduledJob); - 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; }, }; diff --git a/components/GlobalSftpTransferCenter.tsx b/components/GlobalSftpTransferCenter.tsx index 8c977000e..b390885e9 100644 --- a/components/GlobalSftpTransferCenter.tsx +++ b/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 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 (
setExpanded((value) => !value)} + onClick={onToggleExpanded} > {expanded ? : } @@ -755,6 +761,59 @@ function TransferRow({ ); } +function TransferList({ tasks, childrenByParent, empty }: { + tasks: readonly TransferTask[]; + childrenByParent: ReadonlyMap; + empty: React.ReactNode; +}) { + const scrollRef = useRef(null); + // Keep folder expansion when a row scrolls out of the mounted viewport. + const [expandedIds, setExpandedIds] = useState>(() => 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) => ( + setExpandedIds((previous) => { + const next = new Set(previous); + if (next.has(task.id)) next.delete(task.id); + else next.add(task.id); + return next; + })} + /> + ); + return ( +
+ {tasks.length === 0 ? empty : !virtual ? tasks.map(renderRow) : ( +
+ {virtualizer.getVirtualItems().map((row) => ( +
+ {renderRow(tasks[row.index], row.index)} +
+ ))} +
+ )} +
+ ); +} + export function GlobalSftpTransferCenter() { const { t } = useI18n(); // Badge is a stable store snapshot (identity unchanged on pure progress). @@ -866,20 +925,17 @@ export function GlobalSftpTransferCenter() { ))}
-
- {displayed.length === 0 ? ( + {badge.hasAttention && bucket !== "attention" && bucket !== "all" ? : } {t("sftp.transferCenter.empty")}
- ) : displayed.map((task) => ( - - ))} - + )} + /> {(() => { const showBackgroundToggle = collapsed.length > 0; diff --git a/components/GlobalSftpTransferCenter.virtualization.test.tsx b/components/GlobalSftpTransferCenter.virtualization.test.tsx new file mode 100644 index 000000000..87a725cc0 --- /dev/null +++ b/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(); + 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(); + 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("[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"); +}); diff --git a/docs/research/issues-3213-3155-sftp-responsiveness.md b/docs/research/issues-3213-3155-sftp-responsiveness.md new file mode 100644 index 000000000..059236eab --- /dev/null +++ b/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. diff --git a/electron/bridges/transferBridge.cjs b/electron/bridges/transferBridge.cjs index f82e08c01..b6fd5354a 100644 --- a/electron/bridges/transferBridge.cjs +++ b/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, }; diff --git a/electron/bridges/transferBridge.test.cjs b/electron/bridges/transferBridge.test.cjs index 65c2d00eb..842a75c58 100644 --- a/electron/bridges/transferBridge.test.cjs +++ b/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 () => { diff --git a/scripts/sftp-transfer-resume.live.test.cjs b/scripts/sftp-transfer-resume.live.test.cjs new file mode 100644 index 000000000..c37c6ff88 --- /dev/null +++ b/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", + }, () => {}); +}