diff --git a/application/state/sftp/globalSftpTransferControl.test.ts b/application/state/sftp/globalSftpTransferControl.test.ts index f02a19023..b31d98adb 100644 --- a/application/state/sftp/globalSftpTransferControl.test.ts +++ b/application/state/sftp/globalSftpTransferControl.test.ts @@ -51,6 +51,46 @@ function createHost(initial: TransferTask[], bridge?: TransferControlHost["getBr }; } +test("an older pause response must not resume a newer pause", async (t) => { + t.after(resetTransferPauseLatchesForTests); + let finishFirstPause!: (value: { success: boolean; lifecycleEpoch: number }) => void; + let calls = 0; + let backendPaused = true; + const { host, getTasks } = createHost([makeTask("overlapping-pause")], () => ({ + pauseTransfer: async () => { + backendPaused = true; + if (++calls === 1) return new Promise((resolve) => { finishFirstPause = resolve; }); + return { success: true, lifecycleEpoch: 2 }; + }, + resumeTransfer: async () => { + backendPaused = false; + return { success: true, lifecycleEpoch: 3 }; + }, + })); + const first = softPauseTransfer(host, "overlapping-pause"); + await softPauseTransfer(host, "overlapping-pause"); + finishFirstPause({ success: true, lifecycleEpoch: 1 }); + await first; + assert.equal(getTasks()[0].status, "paused"); + assert.equal(backendPaused, true, "late pause acknowledgement must not restart file writes"); +}); + +test("a delayed resume response must not repaint a newer pause", async (t) => { + t.after(resetTransferPauseLatchesForTests); + let finishResume!: (value: { success: boolean; lifecycleEpoch: number }) => void; + const { host, getTasks } = createHost([makeTask("resume-then-pause", "paused")], () => ({ + resumeTransfer: () => new Promise((resolve) => { finishResume = resolve; }), + pauseTransfer: async () => ({ success: true, lifecycleEpoch: 3 }), + })); + const resume = softResumeTransfer(host, "resume-then-pause"); + await softPauseTransfer(host, "resume-then-pause"); + finishResume({ success: true, lifecycleEpoch: 2 }); + await resume; + assert.equal(isTransferPauseLatched("resume-then-pause"), true); + assert.equal(getTasks()[0].status, "paused", "latest user intent must win over an older response"); + assert.equal(getTasks()[0].lifecycleEpoch, 3); +}); + test("softPauseTransfer latches and paints paused for a live directory walk without a panel", async () => { resetTransferPauseLatchesForTests(); resetTransferWalkRegistryForTests(); @@ -245,3 +285,218 @@ test("directory softResume stamps bridge epoch only on successIds; queued siblin resetTransferPauseLatchesForTests(); resetTransferWalkRegistryForTests(); }); + +for (const isDirectory of [false, true]) { + for (const newerLocalPause of [false, true]) { + test(`cross-window resume releases ${isDirectory ? "folder" : "file"} latches unless local pause is newer: ${newerLocalPause}`, async (t) => { + t.after(resetTransferPauseLatchesForTests); + let finish!: (result: { success: boolean; superseded: true; supersededBy: "resume" }) => void; + let pauses = 0; + const id = `remote-resume-${isDirectory}-${newerLocalPause}`; + const initial: TransferTask[] = [{ ...makeTask(id), isDirectory }]; + if (isDirectory) initial.push({ ...makeTask(`${id}-child`), parentTaskId: id }); + const { host, getTasks } = createHost(initial, () => ({ + pauseTransfer: () => ++pauses === 1 + ? new Promise((resolve) => { finish = resolve; }) + : Promise.resolve({ success: true, lifecycleEpoch: 9 }), + })); + const pending = softPauseTransfer(host, id); + host.setTasks(getTasks().map(task => isDirectory && task.id === id ? task : ({ ...task, status: "transferring", lifecycleEpoch: 8 }))); + if (newerLocalPause) await softPauseTransfer(host, id); + finish({ success: false, superseded: true, supersededBy: "resume" }); + await pending; + await new Promise((resolve) => setImmediate(resolve)); + assert.equal(isTransferPauseLatched(id), newerLocalPause, "authoritative resume must release the old root pause barrier"); + assert.equal(getTasks().find(task => task.id === id)?.status, newerLocalPause ? "paused" : "transferring"); + if (isDirectory) assert.equal(isTransferPauseLatched(`${id}-child`), newerLocalPause); + }); + } +} + +for (const action of ["pause", "resume"] as const) { + test(`superseded cross-window ${action} preserves authoritative paused state`, async (t) => { + t.after(resetTransferPauseLatchesForTests); + let finish!: (result: { success: boolean; superseded: boolean }) => void; + const deferred = () => new Promise<{ success: boolean; superseded: boolean }>((resolve) => { finish = resolve; }); + const { host, getTasks } = createHost([makeTask(`cross-window-${action}`, action === "pause" ? "transferring" : "paused")], () => ({ pauseTransfer: deferred, resumeTransfer: deferred })); + const id = getTasks()[0].id; + const operation = action === "pause" ? softPauseTransfer(host, id) : softResumeTransfer(host, id); + // A global event from another window changes lifecycle, not this window's local control epoch. + host.setTasks(getTasks().map(task => ({ ...task, status: "paused", lifecycleEpoch: 8 }))); + finish({ success: false, superseded: true }); + const result = await operation; + assert.equal(getTasks()[0].status, "paused"); + assert.equal(getTasks()[0].lifecycleEpoch, 8); + if (action === "pause") assert.equal(isTransferPauseLatched(id), true); + else assert.deepEqual(result, { handled: true }, "obsolete response must not trigger dedicated recovery"); + }); +} + +for (const isDirectory of [false, true]) { + test(`remote pause restores released ${isDirectory ? "folder" : "file"} barriers after stale resume`, async (t) => { + t.after(resetTransferPauseLatchesForTests); + const id = `remote-pause-${isDirectory}`; + const tasks: TransferTask[] = [{ ...makeTask(id, "paused"), isDirectory }]; + if (isDirectory) tasks.push({ ...makeTask(`${id}-child`, "paused"), parentTaskId: id }); + const { host, getTasks } = createHost(tasks, () => ({ + resumeTransfer: async () => ({ success: false, superseded: true, supersededBy: "pause" }), + })); + assert.deepEqual(await softResumeTransfer(host, id), { handled: true }); + assert.equal(isTransferPauseLatched(id), true); + if (isDirectory) assert.equal(isTransferPauseLatched(`${id}-child`), true); + assert.equal(getTasks()[0].status, "paused"); + }); +} + +test("a child-only remote resume does not release the folder pause", async (t) => { + t.after(resetTransferPauseLatchesForTests); + const { host } = createHost([ + { ...makeTask("mixed-root"), isDirectory: true }, + { ...makeTask("mixed-one"), parentTaskId: "mixed-root" }, + { ...makeTask("mixed-two"), parentTaskId: "mixed-root" }, + ], () => ({ pauseTransfer: async id => id === "mixed-one" + ? { success: false, superseded: true, supersededBy: "resume" } + : { success: true } })); + await softPauseTransfer(host, "mixed-root"); + await new Promise(resolve => setImmediate(resolve)); + assert.equal(isTransferPauseLatched("mixed-root"), true); + assert.equal(isTransferPauseLatched("mixed-two"), true); +}); + +test("directory resume joins successful and remotely resumed children", async (t) => { + t.after(resetTransferPauseLatchesForTests); + const id = "mixed-success-resume"; + const { host, getTasks } = createHost([ + { ...makeTask(id, "paused"), isDirectory: true }, + { ...makeTask(`${id}-one`, "paused"), parentTaskId: id }, + { ...makeTask(`${id}-two`, "paused"), parentTaskId: id }, + ], () => ({ resumeTransfer: async childId => childId === `${id}-one` + ? { success: true, lifecycleEpoch: 3 } + : { success: false, superseded: true, supersededBy: "resume" } })); + assert.deepEqual(await softResumeTransfer(host, id), { handled: true }); + assert.equal(getTasks().find(task => task.id === id)?.status, "transferring"); + assert.equal(isTransferPauseLatched(id), false); +}); + +for (const newerPause of [false, true]) { + test(`directory live resume rejection is ignored only when superseded: ${newerPause}`, async (t) => { + t.after(resetTransferPauseLatchesForTests); + t.after(resetTransferWalkRegistryForTests); + const id = `rejected-folder-resume-${newerPause}`; + registerTransferWalk(id); + let rejectResume!: (error: Error) => void; + let backendPaused = true; + const { host, getTasks } = createHost([ + { ...makeTask(id, "paused"), isDirectory: true }, + { ...makeTask(`${id}-child`, "paused"), parentTaskId: id }, + ], () => ({ + resumeTransfer: () => new Promise((_, reject) => { rejectResume = reject; }), + pauseTransfer: async () => { backendPaused = true; return { success: true, lifecycleEpoch: 4 }; }, + })); + const running = softResumeTransfer(host, id); + if (newerPause) await softPauseTransfer(host, id); + rejectResume(new Error("resume transport disconnected")); + const result = await running; + assert.equal(result.handled, newerPause); + if (!newerPause) assert.match(result.reason || "", /resume transport disconnected/); + assert.equal(backendPaused, true); + assert.equal(getTasks().find(task => task.id === id)?.status, "paused"); + if (newerPause) assert.equal(isTransferPauseLatched(id), true); + }); +} + +for (const newerResume of [false, true]) { + test(`partial folder resume rejection reports running and paused children unless superseded: ${newerResume}`, async (t) => { + t.after(resetTransferPauseLatchesForTests); + const id = `partial-reject-${newerResume}`; + const successfulId = `${id}-one`; + let rejectResume!: (error: Error) => void; + let successfulBackendPaused = true; + let round = 0; + let rollbackCalls = 0; + const { host, getTasks } = createHost([ + { ...makeTask(id, "paused"), isDirectory: true }, + { ...makeTask(successfulId, "paused"), parentTaskId: id }, + { ...makeTask(`${id}-two`, "paused"), parentTaskId: id }, + ], () => ({ + resumeTransfer: async childId => { + if (childId === successfulId) { successfulBackendPaused = false; return { success: true }; } + if (round > 0) return { success: true }; + return new Promise((_, reject) => { rejectResume = reject; }); + }, + pauseTransfer: async childId => { + rollbackCalls++; + if (childId === successfulId) successfulBackendPaused = true; + return { success: true }; + }, + })); + const running = softResumeTransfer(host, id); + if (newerResume) { round++; await softResumeTransfer(host, id); } + rejectResume(new Error("second child IPC rejected")); + const result = await running; + assert.equal(result.handled, true); + assert.equal(successfulBackendPaused, false, "successful child keeps running after partial resume"); + assert.equal(rollbackCalls, 0, "partial reporting must not introduce compensating controls"); + assert.equal(getTasks()[0].status, "transferring", "root must report the successful child still running"); + assert.equal(isTransferPauseLatched(id), false); + const rejected = getTasks().find(task => task.id === `${id}-two`); + assert.equal(rejected?.status, newerResume ? "transferring" : "paused"); + assert.equal(isTransferPauseLatched(`${id}-two`), !newerResume); + if (!newerResume) assert.match(rejected?.error || "", /second child IPC rejected/); + }); +} + +for (const resolvedFailure of [false, true]) { +test(`remote-resumed child remains visibly running when sibling resume fails: resolved=${resolvedFailure}`, async (t) => { + t.after(resetTransferPauseLatchesForTests); + const id = "remote-partial-reject"; + const runningId = `${id}-one`; + const rejectedId = `${id}-two`; + const { host, getTasks } = createHost([ + { ...makeTask(id, "paused"), isDirectory: true }, + { ...makeTask(runningId, "paused"), parentTaskId: id }, + { ...makeTask(rejectedId, "paused"), parentTaskId: id }, + ], () => ({ resumeTransfer: async childId => { + if (childId === rejectedId) { + if (resolvedFailure) return { success: false, reason: "sibling resume rejected" }; + throw new Error("sibling resume rejected"); + } + host.setTasks(getTasks().map(task => task.id === runningId ? { ...task, status: "transferring", lifecycleEpoch: 8 } : task)); + return { success: false, superseded: true, supersededBy: "resume" }; + } })); + assert.equal((await softResumeTransfer(host, id)).handled, true); + assert.equal(getTasks()[0].status, "transferring"); + assert.equal(isTransferPauseLatched(id), false); + assert.equal(getTasks().find(task => task.id === runningId)?.status, "transferring"); + assert.equal(getTasks().find(task => task.id === runningId)?.lifecycleEpoch, 8); + assert.equal(getTasks().find(task => task.id === rejectedId)?.status, "paused"); + assert.equal(isTransferPauseLatched(rejectedId), true); + assert.match(getTasks().find(task => task.id === rejectedId)?.error || "", /sibling resume rejected/); +}); + +} + +for (const newerResume of [false, true]) { + test(`resolved verification failure holds live folder unless newer resume won: ${newerResume}`, async (t) => { + t.after(resetTransferPauseLatchesForTests); + t.after(resetTransferWalkRegistryForTests); + const id = `resolved-verification-${newerResume}`; + registerTransferWalk(id); + let finish!: (value: { success: boolean; reason: string }) => void; + let calls = 0; + const { host, getTasks } = createHost([ + { ...makeTask(id, "paused"), isDirectory: true }, + { ...makeTask(`${id}-child`, "paused"), parentTaskId: id }, + ], () => ({ resumeTransfer: () => ++calls === 1 + ? new Promise(resolve => { finish = resolve; }) : Promise.resolve({ success: true }) })); + const running = softResumeTransfer(host, id); + if (newerResume) await softResumeTransfer(host, id); + finish({ success: false, reason: "Could not verify the source file for resume" }); + const result = await running; + assert.equal(result.handled, newerResume); + if (!newerResume) assert.match(result.reason || "", /verify the source file/); + assert.equal(getTasks()[0].status, newerResume ? "transferring" : "paused"); + assert.equal(isTransferPauseLatched(id), !newerResume); + assert.equal(isTransferPauseLatched(`${id}-child`), !newerResume); + }); +} diff --git a/application/state/sftp/globalSftpTransferControl.ts b/application/state/sftp/globalSftpTransferControl.ts index d390e6e7f..bdae2f9cf 100644 --- a/application/state/sftp/globalSftpTransferControl.ts +++ b/application/state/sftp/globalSftpTransferControl.ts @@ -20,6 +20,7 @@ import { isTransferControlEpochCurrent, } from "./transferControlEpoch"; import { + isTransferOrRootPauseLatched, latchTransferPauseTree, releaseTransferPauseTree, } from "./transferPauseLatch"; @@ -28,6 +29,9 @@ import { isTransferWalkInFlight } from "./transferWalkRegistry"; export type TransferControlBridge = { pauseTransfer?: (id: string) => Promise<{ success: boolean; + /** A newer control in another window owns the authoritative state. */ + superseded?: boolean; + supersededBy?: "pause" | "resume" | "cancel"; reason?: string; checkpointBytes?: number; resumeStage?: TransferTask["resumeStage"]; @@ -39,12 +43,67 @@ export type TransferControlBridge = { }>; resumeTransfer?: (id: string) => Promise<{ success: boolean; + /** A newer control in another window owns the authoritative state. */ + superseded?: boolean; + supersededBy?: "pause" | "resume" | "cancel"; reason?: string; lifecycleEpoch?: number; }>; cancelTransfer?: (id: string) => Promise; }; +function wasSuperseded(result: { success?: boolean; superseded?: boolean } | undefined | null): boolean { + return result?.superseded === true; +} + +type SupersededControlResult = { + success?: boolean; + superseded?: boolean; + supersededBy?: "pause" | "resume" | "cancel"; +}; + +export function reconcileSupersededControls( + host: TransferControlHost, + taskId: string, + childIds: string[], + backendIds: string[], + results: ReadonlyArray, + epoch: number, + requestedAction: "pause" | "resume", + controlTaskId = taskId, +): void { + if (!isTransferControlEpochCurrent(controlTaskId, epoch)) return; + const task = host.getTasks().find((candidate) => candidate.id === taskId); + if (!task || ["completed", "cancelled", "failed"].includes(task.status)) return; + const apply = (id: string, descendants: string[], action: "pause" | "resume" | "cancel") => { + if (action === "resume") releaseTransferPauseTree(id, descendants); + else latchTransferPauseTree(id, descendants); + for (const affectedId of [id, ...descendants]) { + try { + if (action === "resume") globalSftpTransferScheduler.resume(affectedId); + else globalSftpTransferScheduler.pause(affectedId); + } catch { /* best-effort */ } + } + }; + const decisions = results.map((result) => result?.superseded + ? result.supersededBy + : result?.success ? requestedAction : undefined); + // A child-only resume does not establish a folder-wide decision. Require all + // relevant children to agree, or an already-authoritative resumed root row. + if (decisions.length > 0 && decisions.every((action) => action === "resume")) { + apply(taskId, childIds, "resume"); + host.setTasks(host.getTasks().map((row) => row.id === taskId ? { ...row, status: "transferring" } : row)); + } else if (decisions.length > 0 && decisions.every((action) => action === "pause")) { + apply(taskId, childIds, "pause"); + host.setTasks(host.getTasks().map((row) => row.id === taskId ? { ...row, status: "paused" } : row)); + } else if (task.status === "transferring" && decisions.includes("resume")) { + apply(taskId, [], "resume"); + } + decisions.forEach((action, index) => { + if (action) apply(backendIds[index], [], action); + }); +} + /** Prefer the highest bridge lifecycleEpoch from successful pause/resume results. */ function maxBridgeLifecycleEpoch( results: ReadonlyArray<{ success?: boolean; lifecycleEpoch?: number } | undefined | null>, @@ -170,14 +229,23 @@ export async function softPauseTransfer( } const backendIds = treeIds.length > 0 ? treeIds : [taskId]; + // An obsolete pause can be superseded by another pause or cancellation, + // not only by resume. Compensate only while the tree still wants to run. + const undoObsoletePause = async (id: string) => { + const live = host.getTasks().find((candidate) => candidate.id === taskId); + if (!live || ["completed", "cancelled", "failed", "interrupted"].includes(live.status)) return; + if (isTransferOrRootPauseLatched(taskId, id)) return; + try { await bridge!.resumeTransfer?.(id); } catch { /* best-effort */ } + }; const pauseOne = async (id: string) => { let result = await bridge!.pauseTransfer?.(id) ?? { success: false, reason: "Pause unavailable" }; const maxAttempts = task.isDirectory ? 4 : 16; for (let attempt = 0; attempt < maxAttempts; attempt += 1) { + if (wasSuperseded(result)) return result; if (!isTransferControlEpochCurrent(taskId, pauseEpoch)) { if (result.success) { - try { await bridge!.resumeTransfer?.(id); } catch { /* best-effort */ } + await undoObsoletePause(id); } return { success: false, reason: "Pause superseded by resume" }; } @@ -186,6 +254,9 @@ export async function softPauseTransfer( return result; } await new Promise((resolve) => setTimeout(resolve, 40)); + if (!isTransferControlEpochCurrent(taskId, pauseEpoch)) { + return { success: false, reason: "Pause superseded by newer control" }; + } result = await bridge!.pauseTransfer?.(id) ?? { success: false, reason: "Pause unavailable" }; } @@ -200,17 +271,22 @@ export async function softPauseTransfer( return { id, result: { success: false, reason: "Pause superseded by resume" } }; } const result = await pauseOne(id); + if (wasSuperseded(result)) return { id, result }; const live = host.getTasks().find((candidate) => candidate.id === taskId); const userResumed = !pauseStillCurrent() || !live || (live.status !== "paused" && live.status !== "pausing"); if (userResumed) { - try { await bridge!.resumeTransfer?.(id); } catch { /* best-effort */ } + await undoObsoletePause(id); return { id, result: { success: false, reason: "Pause superseded by resume" } }; } return { id, result }; })); if (!pauseStillCurrent()) return; + if (pauseResults.some(({ result }) => wasSuperseded(result))) { + reconcileSupersededControls(host, taskId, childIds, backendIds, pauseResults.map(({ result }) => result), pauseEpoch, "pause"); + return; + } const after = host.getTasks().find((candidate) => candidate.id === taskId); if (!after || after.status === "cancelled") return; if (after.status !== "paused" && after.status !== "pausing") return; @@ -259,6 +335,10 @@ export async function softPauseTransfer( id, result: await pauseOne(id), }))); + if (pauseResults.some(({ result }) => wasSuperseded(result))) { + reconcileSupersededControls(host, taskId, childIds, backendIds, pauseResults.map(({ result }) => result), pauseEpoch, "pause"); + return "noop"; + } const afterLivePause = host.getTasks().find((candidate) => candidate.id === taskId); if (afterLivePause?.status === "cancelled") { releaseTransferPauseTree(taskId, childIds); @@ -271,7 +351,7 @@ export async function softPauseTransfer( if (userAlreadyResumed) { for (const { id, result } of pauseResults) { if (result?.success) { - try { await bridge.resumeTransfer?.(id); } catch { /* best-effort */ } + await undoObsoletePause(id); } } return "noop"; @@ -284,7 +364,7 @@ export async function softPauseTransfer( if (!isTransferControlEpochCurrent(taskId, pauseEpoch)) { for (const { id, result } of pauseResults) { if (result?.success) { - try { await bridge.resumeTransfer?.(id); } catch { /* best-effort */ } + await undoObsoletePause(id); } } return "noop"; @@ -367,8 +447,9 @@ export async function softPauseTransfer( bridgeResults: pauseResults.map((row) => row.result), }); for (const id of rollback.bridgeIdsToResume) { - try { await bridge.resumeTransfer?.(id); } catch { /* best-effort */ } + await undoObsoletePause(id); } + if (!isTransferControlEpochCurrent(taskId, pauseEpoch)) return "noop"; const hard = pauseResults.find(({ result }) => result && !result.success && !isBenignPauseMiss(result.reason), )?.result; @@ -428,7 +509,7 @@ export async function softResumeTransfer( : [taskId, ...childIds.filter((id) => id !== taskId)]; // Supersede in-flight soft-drain / pauseWatch only — not a bridge lifecycle stamp. - bumpTransferControlEpoch(taskId); + const resumeEpoch = bumpTransferControlEpoch(taskId); releaseTransferPauseTree(taskId, releaseIds); for (const id of [taskId, ...releaseIds]) { try { globalSftpTransferScheduler.resume(id); } catch { /* best-effort */ } @@ -438,17 +519,50 @@ export async function softResumeTransfer( try { bridge = host.getBridge(); } catch { bridge = undefined; } const resumeIds = treeIds.length > 0 ? treeIds : [taskId]; - const results = await Promise.all(resumeIds.map(async (id) => - bridge?.resumeTransfer?.(id) ?? { success: false, reason: "Resume unavailable" }, - )); + const failedIds = new Set(); + const results = await Promise.all(resumeIds.map(async (id) => { + try { + return await bridge?.resumeTransfer?.(id) ?? { success: false, reason: "Resume unavailable" }; + } catch (error) { + // Rejections must pass the same stale-control check as ordinary failures. + failedIds.add(id); + return { success: false, reason: error instanceof Error && error.message ? error.message : "Resume request failed" }; + } + })); const after = host.getTasks().find((candidate) => candidate.id === taskId); - if (after?.status === "cancelled") return { handled: true }; + if (!isTransferControlEpochCurrent(taskId, resumeEpoch)) return { handled: true }; + if (!after || ["completed", "cancelled", "failed"].includes(after.status)) return { handled: true }; + results.forEach((result, index) => { + // A missing stream can belong to queued directory work. Verification and + // other explicit failures still own a paused stream and must remain held. + const benignMiss = /^(?:Transfer is no longer active|not active|Resume unavailable)$/i.test(result.reason || ""); + if (!result.success && !result.superseded && !benignMiss) failedIds.add(resumeIds[index]); + }); + if (results.some(wasSuperseded)) { + reconcileSupersededControls(host, taskId, releaseIds, resumeIds, results, resumeEpoch, "resume"); + } const successIds = resumeIds.filter((_, index) => results[index]?.success); + const effectiveRunning = results.some((result) => result?.success || (result?.superseded && result.supersededBy === "resume")); + const failedIndex = resumeIds.findIndex((id) => failedIds.has(id)); + const failureReason = failedIndex >= 0 ? results[failedIndex]?.reason || "Resume request failed" : undefined; + if (failedIds.size > 0) { + // A partial resume stays visibly active; hold only failed children. Do not + // hide successfully running streams behind a paused root barrier. + const heldIds = effectiveRunning ? [...failedIds] : [taskId, ...failedIds]; + for (const id of heldIds) { + latchTransferPauseTree(id, []); + try { globalSftpTransferScheduler.pause(id); } catch { /* best-effort */ } + } + if (!effectiveRunning) return { handled: false, reason: failureReason }; + } + if (results.some(wasSuperseded) && failedIds.size === 0) { + return { handled: true }; + } const walkAlive = isTransferWalkInFlight(taskId); // Directory walk can continue after unlatch without bridge resume on every child. // Single-file must not claim success when every bridge resume fails (stuck bar). - if (successIds.length === 0) { + if (!effectiveRunning) { if (task.isDirectory && walkAlive) { host.setTasks(paintTreeStatus( host.getTasks(), @@ -491,6 +605,12 @@ export async function softResumeTransfer( return candidate; } + if (failedIds.has(candidate.id)) { + return { ...candidate, status: "paused" as const, speed: 0, error: failureReason }; + } + const candidateResult = results[resumeIds.indexOf(candidate.id)]; + if (candidate.id !== taskId && candidateResult?.superseded) return candidate; + // Parent: transferring. Prefer bridge epoch; never wipe to undefined or a // late pause fanout re-applies "paused" (acceptsLifecycle treats missing as any). if (candidate.id === taskId) { @@ -502,7 +622,7 @@ export async function softResumeTransfer( return { ...candidate, status: "transferring" as const, - error: undefined, + error: failureReason, reconnectRequired: false, pauseUnavailableReason: undefined, phase: undefined, diff --git a/application/state/sftp/transferDirectoryOps.discovery.test.tsx b/application/state/sftp/transferDirectoryOps.discovery.test.tsx index 4acb4171f..a3cbfee7d 100644 --- a/application/state/sftp/transferDirectoryOps.discovery.test.tsx +++ b/application/state/sftp/transferDirectoryOps.discovery.test.tsx @@ -41,6 +41,106 @@ const rootTask = (): TransferTask => ({ progressMode: "files", }); +for (const newestAction of ["pause", "cancel", "resume", "remote-resume", "remote-child-resume"] as const) { + test(`directory pause watcher respects a newer ${newestAction}`, async () => { + const { bumpTransferControlEpoch, resetTransferControlEpochsForTests } = await import("./transferControlEpoch"); + const { latchTransferPauseTree, isTransferPauseLatched, resetTransferPauseLatchesForTests } = await import("./transferPauseLatch"); + const previousWindow = (globalThis as { window?: unknown }).window; + const previousLocalStorage = (globalThis as { localStorage?: unknown }).localStorage; + const previousActEnvironment = (globalThis as { IS_REACT_ACT_ENVIRONMENT?: boolean }).IS_REACT_ACT_ENVIRONMENT; + (globalThis as { IS_REACT_ACT_ENVIRONMENT?: boolean }).IS_REACT_ACT_ENVIRONMENT = true; + const root = { ...rootTask(), id: `watch-root-${newestAction}` }; + let tasks: TransferTask[] = [root]; + const transfersRef = { current: tasks }; + const cancelledTasksRef = { current: new Set() }; + const pausedTasksRef = { current: new Set() }; + let pauseCalls = 0; + let finishPause!: (value: { success: boolean; superseded?: boolean; supersededBy?: "resume" }) => void; + let finishTransfer!: (value: { error?: string }) => void; + let pauseStarted!: () => void; + const pauseGate = new Promise((resolve) => { pauseStarted = resolve; }); + const transferGate = new Promise<{ error?: string }>((resolve) => { finishTransfer = resolve; }); + let resumeCalls = 0; + let childId = ""; + (globalThis as { window?: unknown }).window = { netcatty: { + mkdirLocal: async () => undefined, + statLocal: async () => ({ type: "directory" }), + startStreamTransfer: (options: { transferId: string }) => { + childId = options.transferId; + // Reproduce a stream arming during the parent's initial pause round. + bumpTransferControlEpoch(root.id); + latchTransferPauseTree(root.id, [childId]); + if (newestAction === "remote-resume" || newestAction === "remote-child-resume") { pausedTasksRef.current.add(root.id); pausedTasksRef.current.add(childId); } + return transferGate; + }, + pauseTransfer: () => { + pauseCalls++; + pauseStarted(); + return new Promise<{ success: boolean; superseded?: boolean; supersededBy?: "resume" }>((resolve) => { finishPause = resolve; }); + }, + resumeTransfer: async () => { resumeCalls++; return { success: true }; }, + } }; + (globalThis as { localStorage?: unknown }).localStorage = { + getItem: () => null, setItem: () => undefined, removeItem: () => undefined, + }; + let operations: ReturnType | undefined; + let renderer: ReactTestRenderer | null = null; + let running: Promise | undefined; + const Probe = () => { + operations = useSftpDirectoryTransferOps({ + ownerId: `watch-owner-${newestAction}`, cancelledTasksRef, + pausedTasksRef, + waitUntilTransferResumed: async () => undefined, + activeChildIdsRef: { current: new Map() }, transfersRef, + setTransfers: (update) => { + tasks = typeof update === "function" ? update(tasks) : update; + transfersRef.current = tasks; + }, + listLocalFiles: async () => [], listRemoteFiles: async () => [fileEntry("file.txt")], + }); + return null; + }; + try { + await act(async () => { renderer = create(React.createElement(Probe)); }); + assert.ok(operations); + running = operations.transferDirectory(root, "source-sftp", null, false, true, "auto", "auto", root.id); + await pauseGate; + if (newestAction === "remote-resume" || newestAction === "remote-child-resume") { + // Another window's resumed event updates rows without bumping this local epoch. + transfersRef.current = tasks = tasks.map(task => ({ ...task, status: task.id === root.id && newestAction === "remote-child-resume" ? "paused" : "transferring", lifecycleEpoch: 8 })); + finishPause({ success: false, superseded: true, supersededBy: "resume" }); + await new Promise((resolve) => setTimeout(resolve, 110)); + assert.equal(pauseCalls, 1, "superseded watcher must not re-pause the resumed backend"); + assert.equal(isTransferPauseLatched(root.id), newestAction === "remote-child-resume"); + assert.equal(isTransferPauseLatched(childId), false); + assert.equal(pausedTasksRef.current.has(root.id), newestAction === "remote-child-resume"); + assert.equal(pausedTasksRef.current.has(childId), false); + return; + } + bumpTransferControlEpoch(root.id); + if (newestAction === "pause") latchTransferPauseTree(root.id, [childId]); + else { + if (newestAction === "cancel") cancelledTasksRef.current.add(root.id); + resetTransferPauseLatchesForTests(); + } + finishPause({ success: true }); + await new Promise((resolve) => setImmediate(resolve)); + assert.equal(resumeCalls, newestAction === "resume" ? 1 : 0, "compensation must follow the latest decision"); + } finally { + pausedTasksRef.current.clear(); + resetTransferPauseLatchesForTests(); + finishPause?.({ success: true }); + finishTransfer({}); + await running?.catch(() => {}); + await act(async () => { renderer?.unmount(); }); + resetTransferControlEpochsForTests(); + (globalThis as { window?: unknown }).window = previousWindow; + (globalThis as { localStorage?: unknown }).localStorage = previousLocalStorage; + (globalThis as { IS_REACT_ACT_ENVIRONMENT?: boolean }).IS_REACT_ACT_ENVIRONMENT = previousActEnvironment; + } + }); +} + test("directory transfer discovers each directory once with bounded listing concurrency", async () => { const previousWindow = (globalThis as { window?: unknown }).window; const previousLocalStorage = (globalThis as { localStorage?: unknown }).localStorage; diff --git a/application/state/sftp/transferDirectoryOps.ts b/application/state/sftp/transferDirectoryOps.ts index cc165a29a..c6208f5e3 100644 --- a/application/state/sftp/transferDirectoryOps.ts +++ b/application/state/sftp/transferDirectoryOps.ts @@ -1,3 +1,4 @@ +import { reconcileSupersededControls } from "./globalSftpTransferControl"; import { runTransferAndWaitForOwner } from "./waitForTransferOwner"; import { useCallback, type Dispatch, type MutableRefObject, type SetStateAction } from "react"; import type { Host, SftpFileEntry, SftpFilenameEncoding, TransferStatus, TransferTask } from "../../../domain/models"; @@ -432,12 +433,33 @@ export function useSftpDirectoryTransferOps({ ) ) { const epochAtAttempt = getTransferControlEpoch(rootTaskId); + const childEpochAtAttempt = getTransferControlEpoch(task.id); try { const result = await netcattyBridge.get()?.pauseTransfer?.(task.id); - // Resume won while we were awaiting pause — undo. + if ( + result?.superseded && result.supersededBy === "resume" + && isTransferControlEpochCurrent(rootTaskId, epochAtAttempt) + && getTransferControlEpoch(task.id) === childEpochAtAttempt + && !isCancelledLocalOrGlobal(cancelledTasksRef, rootTaskId, task.id) + ) { + const rootAlreadyRunning = transfersRef.current.find((row) => row.id === rootTaskId)?.status === "transferring"; + reconcileSupersededControls({ + getTasks: () => transfersRef.current, + setTasks: (next) => setTransfers(next), + getBridge: () => netcattyBridge.get(), + }, rootAlreadyRunning ? rootTaskId : task.id, rootAlreadyRunning ? [task.id] : [], [task.id], [result], + epochAtAttempt, "pause", rootTaskId); + if (rootAlreadyRunning) pausedTasksRef.current.delete(rootTaskId); + pausedTasksRef.current.delete(task.id); + break; + } + // Compensate only if the newest decision still wants this file running. + // A new pause or cancellation also changes the epoch. if ( result?.success && !isTransferControlEpochCurrent(rootTaskId, epochAtAttempt) + && !isPauseLatched(rootTaskId, task.id) + && !isCancelledLocalOrGlobal(cancelledTasksRef, rootTaskId, task.id) ) { try { await netcattyBridge.get()?.resumeTransfer?.(task.id); diff --git a/application/state/sftpTransferCenterStore.test.ts b/application/state/sftpTransferCenterStore.test.ts index 2cbfcc822..b5884b927 100644 --- a/application/state/sftpTransferCenterStore.test.ts +++ b/application/state/sftpTransferCenterStore.test.ts @@ -3762,3 +3762,108 @@ test("hard directory resume failure falls back to bounded fresh retry history", assert.equal(retainedParent?.checkpointBytes, 0); assert.ok(snapshot.length <= 200, `failed directory history must stay bounded, got ${snapshot.length}`); }); + + +test("held dedicated file resume cannot overwrite a subsequent pause", async (t) => { + const store = createSftpTransferCenterStore(); + const previousWindow = Object.getOwnPropertyDescriptor(globalThis, "window"); + let releaseRun!: () => void; + let started!: () => void; + let releaseResume!: (result: { success: boolean; lifecycleEpoch: number }) => void; + const runGate = new Promise((resolve) => { releaseRun = resolve; }); + const startGate = new Promise((resolve) => { started = resolve; }); + const resumeGate = new Promise<{ success: boolean; lifecycleEpoch: number }>((resolve) => { releaseResume = resolve; }); + t.after(async () => { + releaseRun(); + if (previousWindow) Object.defineProperty(globalThis, "window", previousWindow); + else Reflect.deleteProperty(globalThis, "window"); + const { resetTransferPauseLatchesForTests } = await import("./sftp/transferPauseLatch"); + resetTransferPauseLatchesForTests(); + }); + Object.defineProperty(globalThis, "window", { configurable: true, value: { netcatty: { + pauseTransfer: async () => ({ success: true, checkpointBytes: 2, lifecycleEpoch: 3 }), + resumeTransfer: () => resumeGate, + } } }); + store.publishOwner("dedicated-resume", [{ + ...makeTask("held-file", "interrupted"), + ownerId: "dedicated-resume", targetHostId: "host-a", reconnectRequired: true, + }]); + store.setDedicatedResumeHandler(async () => { + store.patchTask("held-file", { status: "transferring", reconnectRequired: false }); + started(); + await runGate; + return { success: true }; + }); + const first = store.resume("held-file"); + await startGate; + await store.pause("held-file"); + const second = store.resume("held-file"); + await store.pause("held-file"); + releaseResume({ success: true, lifecycleEpoch: 2 }); + await new Promise((resolve) => setImmediate(resolve)); + const row = store.getSnapshot().tasks.find((task) => task.id === "held-file"); + releaseRun(); + await Promise.all([first, second]); + assert.equal(row?.status, "paused", "held transfer must respect the latest pause"); + assert.equal(row?.lifecycleEpoch, 3); +}); + +for (const outcome of ["rejected", "stream-gone"] as const) { + test(`held file ${outcome} resume cannot restart after a newer pause during wind-down`, async (t) => { + const store = createSftpTransferCenterStore(); + const previousWindow = Object.getOwnPropertyDescriptor(globalThis, "window"); + let releaseRun!: () => void; + let started!: () => void; + let resolveResume!: (result: { success: boolean; reason: string }) => void; + let rejectResume!: (error: Error) => void; + let resumeCalls = 0; + const runGate = new Promise((resolve) => { releaseRun = resolve; }); + const startGate = new Promise((resolve) => { started = resolve; }); + const resumeGate = new Promise<{ success: boolean; reason: string }>((resolve, reject) => { + resolveResume = resolve; + rejectResume = reject; + }); + t.after(async () => { + releaseRun(); + if (previousWindow) Object.defineProperty(globalThis, "window", previousWindow); + else Reflect.deleteProperty(globalThis, "window"); + const { resetTransferPauseLatchesForTests } = await import("./sftp/transferPauseLatch"); + resetTransferPauseLatchesForTests(); + }); + Object.defineProperty(globalThis, "window", { configurable: true, value: { netcatty: { + pauseTransfer: async () => ({ success: true, checkpointBytes: 2, lifecycleEpoch: 3 }), + resumeTransfer: () => ++resumeCalls === 1 + ? resumeGate + : Promise.resolve({ success: true, lifecycleEpoch: 4 }), + } } }); + const id = `held-file-${outcome}`; + store.publishOwner("dedicated-resume", [{ + ...makeTask(id, "interrupted"), + ownerId: "dedicated-resume", targetHostId: "host-a", reconnectRequired: true, + }]); + store.setDedicatedResumeHandler(async () => { + store.patchTask(id, { status: "transferring", reconnectRequired: false }); + started(); + await runGate; + return { success: false, error: "Transfer cancelled" }; + }); + const first = store.resume(id); + await startGate; + await store.pause(id); + const second = store.resume(id); + await new Promise((resolve) => setImmediate(resolve)); + if (outcome === "stream-gone") { + resolveResume({ success: false, reason: "Transfer is no longer active" }); + await new Promise((resolve) => setImmediate(resolve)); + } + await store.pause(id); + if (outcome === "rejected") { + rejectResume(new Error("worker channel unavailable")); + await new Promise((resolve) => setImmediate(resolve)); + } + releaseRun(); + await Promise.all([first, second]); + assert.equal(store.getSnapshot().tasks.find((task) => task.id === id)?.status, "paused"); + assert.equal(resumeCalls, 1, "obsolete resume must not issue another resume after the held run ends"); + }); +} diff --git a/application/state/sftpTransferCenterStore.ts b/application/state/sftpTransferCenterStore.ts index 9e1fb6a9c..0c51a10d8 100644 --- a/application/state/sftpTransferCenterStore.ts +++ b/application/state/sftpTransferCenterStore.ts @@ -1757,7 +1757,7 @@ export function createSftpTransferCenterStore(persistence?: StorePersistence): S // Always clear process-global latches + control epoch so walks wake and // late soft-drain / pauseWatch cannot re-pause streams. Do not stamp // control epoch as task.lifecycleEpoch (bridge-aligned only). - bumpTransferControlEpoch(taskId); + let resumeEpoch = bumpTransferControlEpoch(taskId); releaseTransferPauseTree(taskId, childIds); if (task.ownerId === "dedicated-resume" && task.isDirectory) { @@ -1779,64 +1779,53 @@ export function createSftpTransferCenterStore(persistence?: StorePersistence): S try { await bridge?.clearPendingTransferCancel?.(id); } catch { /* best-effort */ } } try { await bridge?.clearPendingTransferCancel?.(taskId); } catch { /* best-effort */ } + if (!isTransferControlEpochCurrent(taskId, resumeEpoch)) return existing; return resumeInvocations.get(taskId) ?? startFresh(); } try { - const resumeIds = [taskId, ...childIds.filter((id) => id !== taskId)]; - const results = await Promise.all(resumeIds.map(async (id) => - netcattyBridge.get()?.resumeTransfer?.(id) ?? { success: false }, - )); - const after = tasks.find((candidate) => candidate.id === taskId); - if (after?.status === "cancelled") return existing; - // Only rejoin when at least one backend stream actually resumed. - // Empty/all-fail must not paint transferring over a dead held run. - const successIds = resumeIds.filter((_, index) => results[index]?.success); - if (successIds.length > 0) { - const resumed = new Set(successIds); - // Align with softResumeTransfer: prefer bridge lifecycleEpoch; clear - // if omitted so main-process progress is not stale-dropped. - let bridgeEpoch: number | undefined; - for (let index = 0; index < results.length; index += 1) { - if (!results[index]?.success) continue; - const epoch = (results[index] as { lifecycleEpoch?: number } | undefined)?.lifecycleEpoch; - if (!Number.isFinite(epoch)) continue; - bridgeEpoch = bridgeEpoch === undefined - ? (epoch as number) - : Math.max(bridgeEpoch, epoch as number); - } - tasks = tasks.map((candidate) => { - if (candidate.id === taskId || resumed.has(candidate.id)) { - return { - ...candidate, - status: "transferring" as const, - error: undefined, - reconnectRequired: false, - pauseUnavailableReason: undefined, - phase: undefined, - lifecycleEpoch: bridgeEpoch, - }; - } - return candidate; - }); + // Reuse the live control path: held dedicated transfers must obey + // the same ordering and per-child lifecycle rules as panel jobs. + const softOperation = softResumeTransfer({ + getTasks: () => tasks, + setTasks: (next) => { tasks = next; emit(); }, + getBridge: defaultTransferControlBridge, + }, taskId); + resumeEpoch = getTransferControlEpoch(taskId); + const soft = await softOperation; + if (soft.handled) return existing; + const streamGone = /no longer active|not active|not found|session is no longer|Resume unavailable|Transfer not found/i + .test(soft.reason ?? ""); + if (!streamGone) { + tasks = tasks.map((candidate) => candidate.id === taskId ? { + ...candidate, + status: "paused" as const, + speed: 0, + phase: undefined, + error: soft.reason, + } : candidate); emit(); return existing; } } catch { // Fall through to await + restart. } + if (!isTransferControlEpochCurrent(taskId, resumeEpoch)) return existing; try { await existing; } catch { /* previous aborted */ } + if (!isTransferControlEpochCurrent(taskId, resumeEpoch)) return existing; return resumeInvocations.get(taskId) ?? startFresh(); } // After demotion to interrupted/attention/failed while work unwinds: // wait then re-invoke (do not rejoin a dying canceling promise). if (task && (task.status === "interrupted" || task.status === "attention" || task.status === "failed")) { + const resumeEpoch = getTransferControlEpoch(taskId); try { await existing; } catch { /* previous aborted */ } + if (getTransferControlEpoch(taskId) !== resumeEpoch) return existing; return resumeInvocations.get(taskId) ?? startFresh(); } return existing; diff --git a/docs/research/sftp-transfer-audit-2026-09.md b/docs/research/sftp-transfer-audit-2026-09.md new file mode 100644 index 000000000..85178a4a1 --- /dev/null +++ b/docs/research/sftp-transfer-audit-2026-09.md @@ -0,0 +1,248 @@ +# SFTP transfer audit (September 2026) + +Baseline: `7b964ead21c1e176999557147914f4975a63f5c8`. +Status: scoped audit completed, with five independently reproduced defects +fixed and submitted. This ledger distinguishes verified paths from reporter +environments unavailable to this audit. + +## Completion requirements + +Review the global transfer center, live and restored execution, admission and +connection ownership, pause/resume/cancel ordering, durable checkpoints, source +and destination integrity, folder replacement/traversal, history retention, +transfer responsiveness, and architectural duplication. Cross-reference actual +issue reports and mature open-source clients. Reproduce material findings before +fixing them. Submit verified fixes as PRs and retain evidence for unresolved +reporter-specific conditions. + +## Confirmed finding: stale transfer control can defeat a newer pause + +Priority: high. The user can pause a transfer and have an older request silently +restart it, or see the state return to transferring while the pause latch remains +set. File transfers and folder children share the affected control path. + +Reproductions (each failed on the baseline before the implementation change): + +- Two overlapping pauses: the first reply incorrectly compensates by calling + resume, even though the second pause is still intended. +- Resume reply delayed until after a newer pause: old success overwrites the row + and its lifecycle epoch. +- Dedicated single-file recovery held for the stream lifetime: its duplicated + soft-resume implementation independently has the same stale reply problem. +- Pipelined upload: resume waits for outstanding writes, receives a newer pause, + then resumes anyway when draining finishes. +- Worker reply fan-out: old successful resume, failed pause, or rejected pause + emits a new resumed event after a later pause succeeded. + +Fix contract: the newest pause/cancel decision wins over older replies; no late +result can revive a terminal task. Keep checkpoints and source verification. +Do not serialize controls behind a stream lifetime. Reuse the shared soft-resume +path for held single-file recovery instead of maintaining another state writer. + +Verification so far: + +- Original selected SFTP suite: 747 passed. +- Initial fixes with the three new regressions: 750 passed. +- Store and control suite after dedicated-path consolidation: 107 passed. +- Worker fan-out suite including three out-of-order cases: 9 passed. +- Loopback SSH/SFTP: 12 files of 128 MiB, two file jobs concurrently, each starts + from a 32 MiB saved checkpoint and pauses/resumes during the remaining download. + All final SHA-256 digests match. Pause acknowledgements 6-15 ms, resume checks + 1391-1493 ms. These are fixture measurements, not WAN throughput claims. +- Independent review found another held-run failure path: rejection or a dead + stream response followed by a newer pause during wind-down could start a fresh + resume. Two regressions reproduced it; shared rejection handling and epoch + checks after wind-down now pass (109 store/control tests). +- First full suite: 11377 passed, 7 plugin archive failures traced to missing + worktree-local nested dependency (yauzl 3.x). Reinstalled with npm ci; all 29 + plugin CLI tests then pass. Full rerun passed: 11386 passed, 0 failed, + 18 skipped. +- GitHub review found a cross-window gap: the obsolete worker result was a hard + failure to a renderer whose local control epoch had not changed. Explicit + superseded outcomes now bypass rollback, compensation and dedicated recovery, + including rejected worker requests. Updated focused suite: 122 passed. +- Production build and lint pass after that follow-up; two independent reviewers + found no actionable follow-up issues. +- Browser fixture exercised the actual transfer-center component/store with + simulated transport: pause all, resume all, individual pause, paused filter, + and cancel. State/buttons matched; no browser console errors. This is not a + full Electron connection-path test. +- Separate real loopback SSH/SFTP audit experiments killed the transferring child + process with SIGKILL and resumed the saved checkpoint in a fresh child process. + Both 32 MiB download and upload completed with matching SHA-256. Persisted + checkpoints were 30670848 and 14024704 bytes respectively. This validates the + transfer engine and disk staging across process death, not the renderer's + automatic history restoration or a real VPN/jump-host environment. + +## Submitted fixes + +| PR | Confirmed failure | Architectural change | +| --- | --- | --- | +| [3284](https://github.com/binaricat/Netcatty/pull/3284) | Late pause/resume replies revive a newer paused/cancelled task or overwrite its visible state, including cross-window and folder watcher paths | Shared held-file resume handling; obsolete controls carry the winning action to reconcile local barriers; compensation checks current intent rather than assuming every epoch change means resume. | +| [3285](https://github.com/binaricat/Netcatty/pull/3285) | Deleted recorded host or duplicate legacy display names can resume an upload against another saved server | Exact host-ID recovery; unique-match-only legacy resolution; preserve live-session recovery. | +| [3286](https://github.com/binaricat/Netcatty/pull/3286) | Publication/restoration can overwrite a concurrently saved local file; rollback can delete a replacement | One exclusive publication helper, a clear commit boundary, preserved recovery artifacts on conflicts or incomplete fallback copying. | +| [3287](https://github.com/binaricat/Netcatty/pull/3287) | Folder stays active after child completion was compacted out of visible history | One bounded settlement observer/helper for both live transfer and recovery; no persistent tombstone history. | +| [3288](https://github.com/binaricat/Netcatty/pull/3288) | Cancelling or timing out channel initialization disconnects shared SSH users | Channel cancellation is separated from shared-transport ownership; abandoned initialization is bounded until settlement. | + +Additional engine experiments used actual loopback SSH/SFTP, killed the child +process, and resumed in a fresh process for remote-to-remote transfers. Both the +download phase and upload phase passed final 32 MiB SHA-256 comparison. The +upload-phase checkpoint was 16777216 bytes. These complement direct upload and +download recovery, not full-app history restoration or reporter confirmation. + +## Historical issue evidence fetched in this audit + +| Issue | Reported condition | Audit treatment | +| --- | --- | --- | +| [3213](https://github.com/binaricat/Netcatty/issues/3213) | macOS, 10+ files of 100-200 MB; pause/resume and force-quit recovery unreliable | Control ordering reproduced separately; source direction and server still absent from report. Do not claim reporter confirmation. | +| [3155](https://github.com/binaricat/Netcatty/issues/3155) | Windows, many-file transfer freezes; no count or logs | Recheck bounded discovery, scheduling, publication and history work. | +| [2973](https://github.com/binaricat/Netcatty/issues/2973) | VPN uploads disconnect SSH and SFTP; transfer spinner continues; later inode VPN report | Check transport loss and settlement. Network/security cause not established by the available logs. | +| [3186](https://github.com/binaricat/Netcatty/issues/3186) | Replacement changes permissions on 1.1.82 | Check mode/owner behavior on each replacement path; existing bot explanations are not proof. | +| [3149](https://github.com/binaricat/Netcatty/issues/3149) | Windows proxy + terminal drag-upload reports No such file | Check target pinning, path encoding, session and retry behavior. | +| [2832](https://github.com/binaricat/Netcatty/issues/2832) | Browsing works through VPN/jump host, transfers wait indefinitely, cancel works | Check dedicated connection admission/authentication/timeout. | +| [2568](https://github.com/binaricat/Netcatty/issues/2568) | Folder copy reaches 100% but remains active; pause ineffective | Check parent settlement and directory checkpoints. | +| [2458](https://github.com/binaricat/Netcatty/issues/2458) | Windows 1 GiB upload continues after Pause/Pause all from both terminal sidebar and SFTP tab | Motivates transport-plus-visible-state regressions; current delayed-control defects independently reproduced. | +| [3031](https://github.com/binaricat/Netcatty/issues/3031) | macOS jump-host/proxy drag upload: no such file | Missing full error, protocol, target path and direct-connect comparison prevent attribution. | +| [2556](https://github.com/binaricat/Netcatty/issues/2556) | Windows download of 1.9 GiB from local Linux VM; separate many-small-files progress complaint | Preserve verification correctness; distinguish network payload from verification and incremental discovery. No comparative reporter throughput available. | +| [2886](https://github.com/binaricat/Netcatty/issues/2886) | sudo terminal drop denied while SFTP upload succeeds | Existing terminal fallback fix is separate; contradictory bot explanations are not evidence of identity or permission correctness. | +| [2638](https://github.com/binaricat/Netcatty/issues/2638) | Recovery after network failure | Verify restore end to end; UI availability alone is insufficient. | + +Issue state (open/closed) and automated comments do not substitute for runtime +proof. Initial title search hit its 100-result cap. Expanded SFTP search returned +308 issue matches, including reports without SFTP in the title. The reports +listed above were read as representative symptom clusters; this was not a claim +to have investigated all 308 matching issues individually. + +## External reference points + +- [Tabby SFTP implementation](https://github.com/Eugeny/tabby/blob/master/tabby-ssh/src/session/sftp.ts): + stream transfer and temporary upload destination before rename. Useful as a + separation-of-concerns comparison; this file does not establish durable + restart recovery and must not be treated as a complete replacement design. +- [WinSCP resume documentation](https://winscp.net/eng/docs/resume): partial-file + discovery and temporary filenames support interruption recovery; temporary + creation can be unavailable under some permission layouts. +- [Electerm transfer implementation](https://github.com/electerm/electerm/blob/master/src/app/server/transfer.js): + separates a per-transfer object from queue/UI state, opens a separate SFTP + channel on the existing SSH connection when available, and uses 32 KiB chunks + with 64 requests. Its live pause flag stops scheduling additional reads. Its + ordinary transfer opens the destination with `w`; it is not evidence for + durable checkpoint recovery. Retain Netcatty's staging and contiguous-offset + protections when simplifying ownership. +- [Electerm action store](https://github.com/electerm/electerm/blob/master/src/client/components/file-transfer/transports-action-store.jsx) + counts pending initializations toward admission; its + [mutation queue](https://github.com/electerm/electerm/blob/master/src/client/components/file-transfer/transfer-queue.jsx) + distinguishes completion of a state update from completion of transfer I/O. + This supports keeping control requests independent of long-lived transfer runs. + +## Coverage and practical limits + +| Area | Evidence and result | +| --- | --- | +| Transfer controls and global center | Actual component/store browser interactions; delayed replies, direct/worker, folder watcher and cross-window action regressions. PR 3284 fixes confirmed ordering failures, including root-state reconciliation and failed resume settlement. | +| Upload/download and remote-to-remote durability | Four real SSH/SFTP fresh-process recovery experiments, including both remote-to-remote phases, all compare final bytes by SHA-256. Existing transfer tests cover changed source, sparse ranges, cancellation and publication. | +| Local/SCP publication | Real filesystem conflict regressions and SCP abort tests; one shared no-overwrite publication helper. Copy fallback retains mode/timestamps and recovery files on failure. Actual FAT/exFAT hardware was unavailable. | +| Folder traversal and final state | Discovery concurrency, replacement/rollback, manifest, pause latch, skip/conflict and history tests; actual live/recovery entrypoints reproduce compacted-child hang and pass after PR 3287. | +| History, restart and ownership | Store history/large-manifest and dedicated recovery tests; PR 3285 rejects missing or ambiguous saved host identity. Full Electron quit/relaunch with restored user credentials was not exercised; engine-process recovery was. Editing endpoint details under an unchanged saved host ID remains a documented identity-model limitation. | +| Connection sharing and cleanup | Connection pool, lease and initialization tests; real delayed SFTP OPEN followed by cancellation leaves browsing alive and closes the late channel after PR 3288. VPN/MFA/security-product and original jump-host environments were unavailable. | +| Responsiveness | Actual browser controls, list virtualization, bounded directory discovery and 50k history cases; scheduler smoke figures below are diagnostic, not end-user throughput or Windows runtime proof. | + +No additional data-loss defect was established by these checks. User reports of +VPN-specific hangs, proxy drag-upload path errors and Windows throughput/freezes +still require their original environments and discriminating logs. They are not +marked resolved solely because a related mechanism was repaired here. + +## Validation summary + +Each fix received two independent local reviews. Review-discovered gaps were +corrected and checked again. PR 3284 latest focused control/worker tests: 35 pass; +two actual direct-resume tests include the full undrained timeout and pass. +PR 3285 full suite: 11,379 pass, 18 skip. PR 3286 full suite before its metadata +follow-up: 11,386 pass, 18 skip; latest publication/abort tests: 34 pass. +PR 3287 serial full suite: 11,383 pass, 18 skip; production build passes. +PR 3288 real SSH cancellation experiment and nine channel tests pass; its full +suite has 11,380 passes and one unrelated terminal write-queue timing failure, +independently reproduced on the unchanged baseline. That entire test file passes +alone (81 tests). A separate integration worktree combines all five branches; +one test-only insertion conflict was resolved by retaining both regressions. +The folder test was subsequently relocated in its own PR; a fresh merge-tree +check confirms it combines with the other control tests without a conflict. + +## Architectural assessment + +The staged-file and contiguous-checkpoint design is worth retaining. A simpler +client that writes directly into the destination is not an equivalent safety +reference. The most important simplification is ownership: a long-lived stream +must not block its own control requests, and one authoritative lifecycle must +inform every window. PR 3284 removes a duplicate state-writing recovery path +and makes obsolete controls explicit. + +Publication is a second useful module boundary. PR 3286 replaces repeated +check/rename/check/rollback branches with one no-overwrite operation shared by +publication and restoration. On hardlink-capable filesystems its commit is +atomic; the documented exclusive-copy fallback trades atomic visibility for +compatibility while preserving recoverable data and cancellation. + +The current transfer list virtualizes beyond 20 visible tasks, directory listing +has a tree-wide concurrency gate, and history migration yields cooperatively. +A scheduler-only smoke workload of 1000 and 10000 immediately completing jobs +finished in 87 ms and 4362 ms, with maximum timer gaps of 15 ms and 111 ms during +other validation activity. The queue still scans for eligibility/priority/fairness; +this is a follow-up performance lead, not proof of the Windows freeze reports or +a standalone throughput benchmark. Folder fan-out bounds ordinary queue growth. + +Do not weaken full saved-prefix verification merely to improve resume timings. +The loopback experiments include extra verification reads; those bytes are not +payload throughput. A future optimization needs proof that changed sources and +out-of-order durable ranges still cannot produce mixed file contents. + +PR 3287 makes final transfer ownership independent of visible history retention: +observers capture terminal settlement before compaction and are disposed after +the waiting invocation exits. PR 3288 keeps per-channel cancellation from +claiming ownership of a shared SSH connection. These targeted boundaries reduce +duplicated policy without replacing the working durable-transfer engine. + +Combined-engine live checks also passed: shared-connection cancellation retains +working browsing and closes the late channel; a 32 MiB download killed at a +16809984-byte checkpoint resumes in a new process to the expected SHA-256. An +initial 8 MiB fixture completed before an intermediate checkpoint was sampled; +the larger paced fixture supplies the intended process-death evidence. + +## Final combined validation + +The combined full suite passed: 11,439 passed, 0 failed, 18 skipped. Lint and +production build passed. This run includes all five fixes and the failed-child +admission follow-up. A final control-aggregation follow-up then corrected mixed +success/superseded resume outcomes and current IPC rejection handling; both +independent reviewers approved it. After bringing that follow-up and the +test-only relocation into the integration checkout, all 202 affected control, +store, folder, observation, recovery and worker tests passed. Final lint/build +are recorded in the PR descriptions. No production changes were made in the +integration checkout; the only merge edits preserve independent regression tests. + +The initial combined run stopped after an older exact-result assertion rejected +the newly structured superseded response and left its fixture alive. The +assertion was updated, its fixture completed, and the successful full run above +supersedes that aborted run. Tests were not removed or weakened. + +## Delivery follow-ups + +The remote publication branch received an automated patch during final review. +It was fetched and checked rather than assuming earlier validation covered it. +Two actual filesystem regressions showed lost normal mtime preservation and a +remaining final-path stamping race. The repair prepares staged timestamps before +publication and restrictive chmod, skips post-commit stamping of that pathname, +and uses a verified file handle on non-promoted local paths. Fallback close +errors also retain recovery artifacts. Publication/bridge/SCP checks: 31 pass; +independent targeted review: 10 pass. + +Fresh dedicated recovery now recognizes only the exact unchanged paused child +rows captured at its entry as old pauses eligible to resume. A newer row, +pausing state, root/child latch or cancellation still blocks admission. This +avoids a 4096-row batching deadlock without removing later pause protection. +Actual batched paused-history and later-pause regressions pass; 163 relevant +tests and an independent 60-test review pass. + +The delivery rerun includes all follow-ups above: 11,452 tests passed, 0 failed, +18 skipped (182.9 seconds, four test processes). This is the final combined +full-suite result and supersedes the earlier intermediate counts. diff --git a/electron/bridges/transferBridge.cjs b/electron/bridges/transferBridge.cjs index b6fd5354a..644aff969 100644 --- a/electron/bridges/transferBridge.cjs +++ b/electron/bridges/transferBridge.cjs @@ -6792,7 +6792,13 @@ async function pauseTransfer(_event, payload) { ) { return { success: false, reason: "This transfer cannot be paused yet" }; } - if (transfer.pauseOperation) return transfer.pauseOperation; + // A repeated pause is still a newer user decision: invalidate any resume + // waiting for verification or outstanding ranges, even if already paused. + transfer.pauseRequestToken = Symbol("pause"); + if (transfer.pauseOperation) { + transfer.pauseSuperseded = false; + return transfer.pauseOperation; + } if (transfer.paused && transfer.lifecycleState === "paused") { const result = { success: true, @@ -6844,7 +6850,7 @@ async function pauseTransfer(_event, payload) { if (usesContiguousRangeCheckpoint) { await transfer.waitForPause(); if (!transfer.paused || transfer.pauseSuperseded) { - return { success: false, reason: "Pause was superseded by resume" }; + return { success: false, superseded: true, supersededBy: "resume", reason: "Pause was superseded by resume" }; } // Concurrent path already tracks contiguous durable bytes — do not spend // hundreds of ms waiting for writeStream drain before acknowledging pause. @@ -6959,7 +6965,7 @@ async function pauseTransfer(_event, payload) { return { success: false, reason: "Transfer is no longer active" }; } if (!transfer.paused || transfer.pauseSuperseded) { - return { success: false, reason: "Pause was superseded by resume" }; + return { success: false, superseded: true, supersededBy: "resume", reason: "Pause was superseded by resume" }; } // Confirm pause as soon as soft-drain + durable checkpoint are ready. // Source identity (remote sample reads on download) used to block this IPC @@ -7014,6 +7020,20 @@ async function resumeTransfer(_event, payload) { reason: transfer.pauseUnavailableReason || "This transfer cannot be resumed safely", }; } + const pauseRequestToken = transfer.pauseRequestToken; + const invalidResumeResult = (reason) => ({ + success: false, reason, + ...(transfer.pauseRequestToken !== pauseRequestToken ? { superseded: true, supersededBy: "pause" } : {}), + }); + const resumeInvalidReason = () => { + if (activeTransfers.get(payload?.transferId) !== transfer || transfer.cancelled) { + return "Transfer is no longer active"; + } + if (transfer.pauseRequestToken !== pauseRequestToken) { + return "Resume was superseded by a newer pause"; + } + return null; + }; if (transfer.pauseOperation) { transfer.pauseSuperseded = true; try { transfer.cancelPauseWait?.(); } catch { } @@ -7023,6 +7043,8 @@ async function resumeTransfer(_event, payload) { if (currentTransfer !== transfer || transfer.cancelled) { return { success: false, reason: "Transfer is no longer active" }; } + const initialInvalidReason = resumeInvalidReason(); + if (initialInvalidReason) return invalidResumeResult(initialInvalidReason); // Already flowing (e.g. double-click resume): do not pipe() again. if (!transfer.paused) { transfer.lifecycleState = "transferring"; @@ -7057,7 +7079,7 @@ async function resumeTransfer(_event, payload) { await transfer.captureSourceFingerprint?.(); } if (!transfer.sourceFingerprint) { - return { success: false, reason: "Could not verify the source file for resume" }; + return invalidResumeResult("Could not verify the source file for resume"); } await transfer.verifySourceFingerprint(transfer.sourceFingerprint); } @@ -7065,12 +7087,11 @@ async function resumeTransfer(_event, payload) { if (transfer.cancelled || activeTransfers.get(payload?.transferId) !== transfer) { return { success: false, reason: "Transfer is no longer active" }; } - return { - success: false, - reason: error?.message || "Could not verify the source file for resume", - }; + return invalidResumeResult(error?.message || "Could not verify the source file for resume"); } } + const verifiedInvalidReason = resumeInvalidReason(); + if (verifiedInvalidReason) return invalidResumeResult(verifiedInvalidReason); // Soft-drained concurrent pause may leave a sparse tail past the contiguous // checkpoint. Wait with Resume's own budget, then single-flight truncate so // background settle cannot race new writes after unpause. @@ -7081,15 +7102,14 @@ async function resumeTransfer(_event, payload) { { maxWaitMs: RESUME_RANGE_SETTLE_MS }, ); if (!settled?.ok) { - return { - success: false, - reason: settled?.reason || "The current file is still finishing. Try resume again.", - }; + return invalidResumeResult(settled?.reason || "The current file is still finishing. Try resume again."); } if (transfer.cancelled || activeTransfers.get(payload?.transferId) !== transfer) { return { success: false, reason: "Transfer is no longer active" }; } } + const finalInvalidReason = resumeInvalidReason(); + if (finalInvalidReason) return invalidResumeResult(finalInvalidReason); transfer.paused = false; transfer.pauseSuperseded = false; transfer.lifecycleEpoch += 1; @@ -7264,6 +7284,37 @@ function registerWorkerHandle(ipcMain, terminalWorkerManager, channel) { function registerHandlers(ipcMain, options = {}) { const terminalWorkerManager = options.terminalWorkerManager || null; if (terminalWorkerManager) { + // Control replies can arrive out of order across windows. Only the latest + // request may publish lifecycle state; retain tokens only while in flight. + const workerControlRequests = new Map(); + const withWorkerControl = async (transferId, action, work) => { + const token = Symbol("control"); + const state = workerControlRequests.get(transferId) || { token, action, pending: 0 }; + state.token = token; + state.action = action; + state.pending += 1; + workerControlRequests.set(transferId, state); + const isCurrent = () => workerControlRequests.get(transferId) === state && state.token === token; + const superseded = () => ({ success: false, superseded: true, supersededBy: state.action }); + try { + const result = await work(isCurrent, superseded); + if (isCurrent() && !result?.success && !result?.superseded) { + if (action === "pause") state.action = "resume"; + else if (action === "resume") state.action = "pause"; + } + return result; + } catch (error) { + if (!isCurrent()) return superseded(); + if (action === "pause") state.action = "resume"; + else if (action === "resume") state.action = "pause"; + throw error; + } finally { + state.pending -= 1; + if (state.pending === 0 && workerControlRequests.get(transferId) === state) { + workerControlRequests.delete(transferId); + } + } + }; const nextWorkerLifecycleEpoch = (transferId, suggestedEpoch) => { const entry = workerTransferLifecycleEpochs.get(transferId); const current = Math.max(0, Number(entry?.epoch) || 0); @@ -7289,6 +7340,7 @@ function registerHandlers(ipcMain, options = {}) { && workerTransferLifecycleEpochs.get(payload.transferId) === lifecycleEntry ) { workerTransferLifecycleEpochs.delete(payload.transferId); + workerControlRequests.delete(payload.transferId); } }; // Renderer (or outer main) already admitted — skip a second queue so @@ -7314,20 +7366,22 @@ function registerHandlers(ipcMain, options = {}) { ipcMain.handle("netcatty:transfer:cancel", (event, payload) => ( cancelQueuedTransfer(payload?.transferId) ? { success: true } - : workerRequest(event, "netcatty:transfer:cancel", payload) + : withWorkerControl(payload?.transferId, "cancel", () => workerRequest(event, "netcatty:transfer:cancel", payload)) )); ipcMain.handle("netcatty:transfer:pause", async (event, payload) => { const queued = pauseQueuedTransfer(payload?.transferId); if (queued) return queued; - const lifecycleEpoch = nextWorkerLifecycleEpoch(payload?.transferId); - broadcastGlobalTransferEvent({ - type: "pausing", - transferId: payload?.transferId, - lifecycleEpoch, - lifecycleState: "pausing", + return withWorkerControl(payload?.transferId, "pause", async (isCurrent, superseded) => { + const lifecycleEpoch = nextWorkerLifecycleEpoch(payload?.transferId); + broadcastGlobalTransferEvent({ + type: "pausing", + transferId: payload?.transferId, + lifecycleEpoch, + lifecycleState: "pausing", }); try { const result = await workerRequest(event, "netcatty:transfer:pause", payload); + if (!isCurrent()) return superseded(); if (!result?.success) { const rollbackEpoch = nextWorkerLifecycleEpoch(payload?.transferId); broadcastGlobalTransferEvent({ @@ -7351,6 +7405,7 @@ function registerHandlers(ipcMain, options = {}) { }); return { ...result, lifecycleEpoch }; } catch (error) { + if (!isCurrent()) throw error; const rollbackEpoch = nextWorkerLifecycleEpoch(payload?.transferId); broadcastGlobalTransferEvent({ type: "resumed", @@ -7360,24 +7415,28 @@ function registerHandlers(ipcMain, options = {}) { }); throw error; } + }); }); ipcMain.handle("netcatty:transfer:resume", async (event, payload) => { const queuedResume = resumeQueuedTransfer(payload?.transferId); if (queuedResume) return queuedResume; - const result = await workerRequest(event, "netcatty:transfer:resume", payload); - if (result?.success) { - // Normalize into main-process epoch space (may advance past worker-local). - // Soft-resume UI must stamp THIS epoch or later worker progress is stale. - const lifecycleEpoch = nextWorkerLifecycleEpoch(payload?.transferId, result.lifecycleEpoch); - broadcastGlobalTransferEvent({ - type: "resumed", - transferId: payload?.transferId, - lifecycleEpoch, - lifecycleState: "transferring", - }); - return { ...result, lifecycleEpoch }; - } - return result; + return withWorkerControl(payload?.transferId, "resume", async (isCurrent, superseded) => { + const result = await workerRequest(event, "netcatty:transfer:resume", payload); + if (!isCurrent()) return superseded(); + if (result?.success) { + // Normalize into main-process epoch space (may advance past worker-local). + // Soft-resume UI must stamp THIS epoch or later worker progress is stale. + const lifecycleEpoch = nextWorkerLifecycleEpoch(payload?.transferId, result.lifecycleEpoch); + broadcastGlobalTransferEvent({ + type: "resumed", + transferId: payload?.transferId, + lifecycleEpoch, + lifecycleState: "transferring", + }); + return { ...result, lifecycleEpoch }; + } + return result; + }); }); ipcMain.handle("netcatty:transfer:prioritize", (event, payload) => ( prioritizeQueuedTransfer(payload?.transferId) diff --git a/electron/bridges/transferBridge.globalFanout.test.cjs b/electron/bridges/transferBridge.globalFanout.test.cjs index dd9ebdc4e..b88cc6eab 100644 --- a/electron/bridges/transferBridge.globalFanout.test.cjs +++ b/electron/bridges/transferBridge.globalFanout.test.cjs @@ -299,3 +299,85 @@ test("worker-backed pause and resume fan authoritative lifecycle to every window delete require.cache[require.resolve(bridgePath)]; } }); + +for (const latestAction of ["pause", "resume", "cancel"]) { + test(`stale worker pause retains latest ${latestAction} after that newer request has settled`, async (t) => { + const bridgePath = require.resolve("./transferBridge.cjs"); + delete require.cache[bridgePath]; + t.after(() => { delete require.cache[bridgePath]; }); + const bridge = require(bridgePath); + const handlers = new Map(); + let finishFirst; + const firstGate = new Promise((resolve) => { finishFirst = resolve; }); + let requests = 0; + bridge.registerHandlers({ handle: (channel, fn) => handlers.set(channel, fn) }, { + terminalWorkerManager: { request: async () => ++requests === 1 ? firstGate : { success: true } }, + }); + const payload = { transferId: `worker-last-action-${latestAction}` }; + const first = handlers.get("netcatty:transfer:pause")(null, payload); + await handlers.get(`netcatty:transfer:${latestAction}`)(null, payload); + finishFirst({ success: true }); + const stale = await first; + assert.equal(stale.superseded, true); + assert.equal(stale.supersededBy, latestAction, "completed newer request must retain intent for older pending replies"); + }); +} + +for (const outcome of ["resume-success", "pause-success", "pause-failure", "pause-error", "resume-error"]) { + test(`worker ${outcome} arriving after a newer pause cannot broadcast resumed`, async (t) => { + const sent = []; + const restore = withElectronVersionStub(); + const originalLoad = Module._load; + Module._load = function (request, parent, isMain) { + if (request === "electron") return { BrowserWindow: { getAllWindows: () => [{ + isDestroyed: () => false, + webContents: { isDestroyed: () => false, send: (_channel, payload) => sent.push(payload) }, + }] } }; + return originalLoad(request, parent, isMain); + }; + const bridgePath = require.resolve("./transferBridge.cjs"); + delete require.cache[bridgePath]; + t.after(() => { Module._load = originalLoad; restore(); delete require.cache[bridgePath]; }); + const bridge = require(bridgePath); + const handlers = new Map(); + let resolveFirst; + let rejectFirst; + const firstGate = new Promise((resolve, reject) => { resolveFirst = resolve; rejectFirst = reject; }); + let requests = 0; + bridge.registerHandlers({ handle: (channel, fn) => handlers.set(channel, fn) }, { + terminalWorkerManager: { request: async () => ++requests === 1 ? firstGate : { success: true } }, + }); + const payload = { transferId: `worker-order-${outcome}` }; + const channel = outcome.startsWith("resume") ? "resume" : "pause"; + const first = handlers.get(`netcatty:transfer:${channel}`)(null, payload).catch(() => null); + await handlers.get("netcatty:transfer:pause")(null, payload); + const afterPause = sent.length; + if (outcome.endsWith("error")) rejectFirst(new Error("channel closed")); + else resolveFirst({ success: outcome.endsWith("success"), reason: "pause unavailable" }); + assert.deepEqual(await first, { success: false, superseded: true, supersededBy: "pause" }); + assert.equal(sent.slice(afterPause).some((event) => event.type === "resumed"), false); + assert.equal(sent.at(-1).type, "paused"); + }); +} + +for (const action of ["pause", "resume"]) { + test(`superseded result follows rollback after latest ${action} fails`, async (t) => { + const bridgePath = require.resolve("./transferBridge.cjs"); + delete require.cache[bridgePath]; + t.after(() => { delete require.cache[bridgePath]; }); + const bridge = require(bridgePath); + const handlers = new Map(); + let finish; + let calls = 0; + bridge.registerHandlers({ handle: (channel, fn) => handlers.set(channel, fn) }, { + terminalWorkerManager: { request: () => ++calls === 1 + ? new Promise(resolve => { finish = resolve; }) + : Promise.resolve({ success: false, reason: "unavailable" }) }, + }); + const payload = { transferId: `rollback-${action}` }; + const older = handlers.get("netcatty:transfer:pause")(null, payload); + await handlers.get(`netcatty:transfer:${action}`)(null, payload); + finish({ success: true }); + assert.equal((await older).supersededBy, action === "pause" ? "resume" : "pause"); + }); +} diff --git a/electron/bridges/transferBridge.test.cjs b/electron/bridges/transferBridge.test.cjs index 3d273efd8..5d08eb797 100644 --- a/electron/bridges/transferBridge.test.cjs +++ b/electron/bridges/transferBridge.test.cjs @@ -1607,6 +1607,125 @@ test("pause soft-drains concurrent ranges but resume waits before truncating", a assert.equal(durableBytes, payload.length); }); +for (const settleWrites of [true, false]) { +test(`a newer pause supersedes resume while concurrent writes are draining: settle=${settleWrites}`, async (t) => { + const tempDir = await fs.promises.mkdtemp(path.join(os.tmpdir(), "netcatty-transfer-resume-repause-")); + t.after(async () => { + await fs.promises.rm(tempDir, { recursive: true, force: true }); + }); + + // Several MB so concurrent fanout stays saturated while we hold writes. + const payload = Buffer.alloc(UPLOAD_TRANSFER_CONCURRENCY * TRANSFER_CHUNK_SIZE * 4, 65); + const localPath = path.join(tempDir, "upload.bin"); + await fs.promises.writeFile(localPath, payload); + let holdWrites = true; + const pendingWrites = []; + let durableBytes = 0; + const truncateCalls = []; + const fastSftp = createFastSftp({ + open(_remotePath, _flags, callback) { + callback(null, Buffer.from("remote-handle")); + }, + write(_handle, _buffer, _offset, length, position, callback) { + const complete = () => { + durableBytes = Math.max(durableBytes, position + length); + callback(null); + }; + if (holdWrites) pendingWrites.push({ position, complete }); + else setImmediate(complete); + }, + close(_handle, callback) { + callback(null); + }, + }); + const client = { + sftp: createFastSftp({}), + stat() { + return Promise.resolve({ size: durableBytes }); + }, + truncate(_remotePath, size) { + truncateCalls.push(size); + durableBytes = size; + return Promise.resolve(); + }, + rename() { + return Promise.resolve(); + }, + delete() { + return Promise.resolve(); + }, + client: { + sftp(callback) { + callback(null, fastSftp); + }, + }, + }; + transferBridge.init({ sftpClients: new Map([["target", client]]) }); + + const running = transferBridge.startTransfer( + { sender: createSender() }, + { + transferId: "upload-resume-repause", + sourcePath: localPath, + targetPath: "/tmp/upload-soft.bin", + sourceType: "local", + targetType: "sftp", + targetSftpId: "target", + totalBytes: payload.length, + resumable: true, + }, + ); + + const readyDeadline = Date.now() + 1000; + while (pendingWrites.length < UPLOAD_TRANSFER_CONCURRENCY && Date.now() < readyDeadline) { + await new Promise((resolve) => setImmediate(resolve)); + } + assert.ok(pendingWrites.length >= 1, "expected in-flight concurrent writes"); + + const started = Date.now(); + // Hold writes so active ranges never drain — soft-drain must still resolve. + const paused = await transferBridge.pauseTransfer(null, { transferId: "upload-resume-repause" }); + const elapsed = Date.now() - started; + assert.equal(paused.success, true); + // Soft drain is PAUSE_RANGE_DRAIN_MS (~50ms); allow headroom without full drain. + assert.ok(elapsed < 1500, `soft pause took too long: ${elapsed}ms`); + + const outOfOrderWrite = pendingWrites.pop(); + assert.ok(outOfOrderWrite?.position > 0, "expected a range beyond the contiguous checkpoint"); + outOfOrderWrite.complete(); + await new Promise((resolve) => setImmediate(resolve)); + + let resumeSettled = false; + const resuming = transferBridge.resumeTransfer(null, { transferId: "upload-resume-repause" }) + .then((result) => { + resumeSettled = true; + return result; + }); + await new Promise((resolve) => setTimeout(resolve, 500)); + const truncatedBeforeDrain = truncateCalls.length > 0; + const resumedBeforeDrain = resumeSettled; + + const newerPause = await transferBridge.pauseTransfer(null, { transferId: "upload-resume-repause" }); + assert.equal(newerPause.success, true); + if (settleWrites) { + holdWrites = false; + for (const { complete } of pendingWrites.splice(0)) complete(); + } + const resumeResult = await resuming; + holdWrites = false; + for (const { complete } of pendingWrites.splice(0)) complete(); + await transferBridge.cancelTransfer(null, { transferId: "upload-resume-repause" }); + await running; + assert.equal(resumeResult.success, false, "old resume must not restart writes after a newer pause"); + if (settleWrites) assert.match(resumeResult.reason, /superseded.*pause/i); + assert.equal(resumeResult.superseded, true, "direct callers need a structured superseded result too"); + assert.equal(resumeResult.supersededBy, "pause"); + assert.equal(truncatedBeforeDrain, false); + assert.equal(resumedBeforeDrain, false); +}); + +} + test("cancelling resume during soft-drain does not truncate the staged file", async (t) => { const tempDir = await fs.promises.mkdtemp(path.join(os.tmpdir(), "netcatty-transfer-resume-cancel-")); t.after(async () => { @@ -1766,6 +1885,8 @@ test("resuming while a fast pause is pending settles the pause request", async ( assert.deepEqual(await pausing, { success: false, reason: "Pause was superseded by resume", + superseded: true, + supersededBy: "resume", }); finishWrite(); diff --git a/types/global/netcatty-bridge-sftp.d.ts b/types/global/netcatty-bridge-sftp.d.ts index c75c986a4..990c68238 100644 --- a/types/global/netcatty-bridge-sftp.d.ts +++ b/types/global/netcatty-bridge-sftp.d.ts @@ -92,6 +92,8 @@ declare global { ): Promise<{ transferId: string; totalBytes?: number; error?: string; cancelled?: boolean }>; pauseTransfer?(transferId: string): Promise<{ success: boolean; + superseded?: boolean; + supersededBy?: "pause" | "resume" | "cancel"; checkpointBytes?: number; resumeStage?: 'direct' | 'download' | 'upload'; downloadCheckpointBytes?: number; @@ -100,7 +102,7 @@ declare global { lifecycleEpoch?: number; reason?: string; }>; - resumeTransfer?(transferId: string): Promise<{ success: boolean; reason?: string; lifecycleEpoch?: number }>; + resumeTransfer?(transferId: string): Promise<{ success: boolean; reason?: string; lifecycleEpoch?: number; superseded?: boolean; supersededBy?: "pause" | "resume" | "cancel" }>; prioritizeTransfer?(transferId: string): Promise<{ success: boolean }>; setGlobalTransferConcurrency?(limit: number): Promise<{ success: boolean; limit: number }>; cleanupTransferArtifacts?(payload: {