Compare commits

...
10 changed files with 894 additions and 84 deletions

No files matched your search

+22 -6
View File
@@ -1,10 +1,12 @@
import type { TransferOwnerObservation } from "../state/sftp/waitForTransferOwner";
import type { TransferStatus, TransferTask } from "../../domain/models";
export const DEDICATED_RESUME_LARGE_HISTORY_THRESHOLD = 4_096;
export const DEDICATED_RESUME_CHILD_UPDATE_BATCH_SIZE = 512;
export interface DedicatedResumeChildUpdateBatcher {
push(task: TransferTask): void;
push(task: TransferTask, observation?: TransferOwnerObservation): void;
discard(taskId: string): void;
flush(): void;
}
@@ -24,25 +26,39 @@ export function createDedicatedResumeChildUpdateBatcher(deps: {
hasTask: (taskId: string) => boolean;
upsertTasks: (tasks: readonly TransferTask[]) => void;
}): DedicatedResumeChildUpdateBatcher {
const pending = new Map<string, TransferTask>();
const pending = new Map<string, { task: TransferTask; observation?: TransferOwnerObservation }>();
const flush = () => {
if (pending.size === 0) return;
const batch = [...pending.values()];
pending.clear();
deps.upsertTasks(batch);
try {
const updates = batch.filter(({ observation }) => !observation?.hasIdentityConflict()).map(({ task }) => task);
if (updates.length) deps.upsertTasks(updates);
} finally {
for (const { observation } of batch) observation?.dispose();
}
};
return {
push(task) {
push(task, observation) {
if (observation?.hasIdentityConflict()) {
pending.get(task.id)?.observation?.dispose();
pending.delete(task.id);
observation.dispose();
return;
}
const shouldBatch = !!task.parentTaskId
&& deps.getTaskCount() >= DEDICATED_RESUME_LARGE_HISTORY_THRESHOLD
&& deps.hasTask(task.id);
if (!shouldBatch) {
deps.upsertTasks([task]);
try { deps.upsertTasks([task]); } finally { observation?.dispose(); }
return;
}
pending.set(task.id, task);
const previous = pending.get(task.id);
if (previous?.observation && previous.observation !== observation) previous.observation.dispose();
pending.set(task.id, { task, observation });
if (pending.size >= DEDICATED_RESUME_CHILD_UPDATE_BATCH_SIZE) flush();
},
discard: (taskId) => { pending.get(taskId)?.observation?.dispose(); pending.delete(taskId); },
flush,
};
}
+5 -2
View File
@@ -181,11 +181,14 @@ export function useAppStartupEffects(ctx: StartupEffectsContext) {
},
{
children,
onChildUpdate: (child) => {
onChildUpdate: (child, observation) => {
if (acceptsResumeCallbacks) {
childUpdateBatcher.push({ ...child, ownerId: "dedicated-resume" });
childUpdateBatcher.push({ ...child, ownerId: "dedicated-resume" }, observation);
return true;
}
},
onChildSuperseded: (taskId) => childUpdateBatcher.discard(taskId),
flushChildUpdates: () => childUpdateBatcher.flush(),
onDirectoryCheckpointUpdate: (checkpoint) => {
if (acceptsResumeCallbacks) {
sftpTransferCenterStore.patchTask(task.id, {
@@ -41,8 +41,12 @@ const host = (id: string, label: string, hostname = label): Host => ({
for (const retainedStatus of [undefined, "interrupted", "failed", "paused"] as const) {
for (const newerPause of retainedStatus === "paused" ? [false, true] : [false]) {
for (const ownerChange of retainedStatus === "interrupted" || retainedStatus === undefined ? ["same", "active", "completed"] : ["same"]) {
for (const completedBeforeReplacement of ownerChange === "active" ? [false, true] : [false]) {
for (const flushBeforeReply of ownerChange === "same" || retainedStatus === undefined ? [false] : [false, true]) {
for (const replyRejects of ownerChange !== "same" && !completedBeforeReplacement && !flushBeforeReply ? [false, true] : [false]) {
const batchExistingIdentity = retainedStatus !== undefined;
test(`superseded folder child settles when its completion was compacted into the parent: retained=${retainedStatus ?? "none"}, newerPause=${newerPause}`, async (t) => {
test(`superseded folder child settles when its completion was compacted into the parent: retained=${retainedStatus ?? "none"}, newerPause=${newerPause}, owner=${ownerChange}, flush=${flushBeforeReply}, priorComplete=${completedBeforeReplacement}, reject=${replyRejects}`, async (t) => {
const { sftpTransferCenterStore } = await import("../sftpTransferCenterStore");
const previousLocalStorage = Object.getOwnPropertyDescriptor(globalThis, "localStorage");
Object.defineProperty(globalThis, "localStorage", {
@@ -75,14 +79,19 @@ test(`superseded folder child settles when its completion was compacted into the
...parent, id: "retained-child", parentTaskId: parent.id, isDirectory: false,
sourcePath: "/local/folder/a.bin", targetPath: "/remote/folder/a.bin",
totalBytes: 10, sourceLastModified: 1, status: retainedStatus ?? "interrupted",
directoryEntryIndex: 0, directoryEntryIdentity: "a".repeat(64),
directoryEntryIndex: 0,
directoryEntryIdentity: newerPause ? createDirectoryEntryIdentity({
sourcePath: "/local/folder/a.bin", targetPath: "/remote/folder/a.bin", size: 10, lastModified: 1,
}) : "a".repeat(64),
};
const { createDedicatedResumeChildUpdateBatcher } = await import("../../app/dedicatedResumeProgress");
// Exercise the real large-history batching branch with one relevant retained row.
const childBatcher = createDedicatedResumeChildUpdateBatcher({
getTaskCount: () => 4096,
hasTask: (id) => id === persisted.id,
upsertTasks: (updates) => sftpTransferCenterStore.upsertTasks(updates),
hasTask: (id) => id === persisted.id || id.startsWith("flush-peer-"),
// Peer rows only drive the real 512-entry flush; keep the relevant store
// fixture small so unrelated history maintenance cannot mask ownership.
upsertTasks: (updates) => sftpTransferCenterStore.upsertTasks(updates.filter((task) => !task.id.startsWith("flush-peer-"))),
});
sftpTransferCenterStore.publishOwner("compacted-owner", [parent, ...(batchExistingIdentity ? [persisted] : [])]);
(netcattyBridge as { get: () => unknown }).get = () => ({
@@ -106,20 +115,50 @@ test(`superseded folder child settles when its completion was compacted into the
directoryEntryIndex: options.directoryEntryIndex,
directoryEntryIdentity: options.directoryEntryIdentity,
});
// A newer same-id owner completes before this older invocation rejoins.
sftpTransferCenterStore.ingestBackgroundEvent({
type: "completed", transferId: childId, transferred: 10, totalBytes: 10, lifecycleEpoch: 0,
});
return { superseded: true };
if (ownerChange !== "same") {
const current = sftpTransferCenterStore.getTask(childId)!;
if (completedBeforeReplacement) {
sftpTransferCenterStore.ingestBackgroundEvent({
type: "completed", transferId: childId, transferred: 10, totalBytes: 10, lifecycleEpoch: 0,
});
}
sftpTransferCenterStore.upsertTasks([{
...current, status: "transferring", directoryEntryIdentity: "e".repeat(64), lifecycleEpoch: 1,
}]);
}
if (ownerChange !== "active") {
// A newer same-id owner completes before this older invocation rejoins.
sftpTransferCenterStore.ingestBackgroundEvent({
type: "completed", transferId: childId, transferred: 10, totalBytes: 10, lifecycleEpoch: ownerChange === "same" ? 0 : 1,
});
}
if (flushBeforeReply) {
for (let index = 0; index < 512; index += 1) {
childBatcher.push({ ...persisted, id: `flush-peer-${index}`, status: "transferring" });
}
if (ownerChange === "active") {
assert.equal(sftpTransferCenterStore.getTask(childId)?.directoryEntryIdentity, "e".repeat(64),
"automatic flush before the displaced reply must preserve the new owner");
} else assert.equal(sftpTransferCenterStore.getTask(childId), undefined,
"automatic flush must not recreate the compacted new owner");
}
if (replyRejects) throw new Error("old invocation transport failed");
return completedBeforeReplacement ? {} : { superseded: true };
},
});
running = resumeTransferWithDedicatedSession(parent, {
hosts: [host("h1", "box", "1.2.3.4")], keys: [], identities: [],
}, undefined, {
children: batchExistingIdentity ? [persisted] : [], shouldAbort: () => abort,
onChildUpdate: (child) => batchExistingIdentity
? childBatcher.push(child)
: sftpTransferCenterStore.publishOwner("compacted-owner", [parent, child]),
onChildSuperseded: (id) => childBatcher.discard(id),
flushChildUpdates: () => childBatcher.flush(),
onChildUpdate: (child, observation) => {
if (batchExistingIdentity) {
childBatcher.push(child, observation);
return true;
}
sftpTransferCenterStore.publishOwner("compacted-owner", [parent, child]);
},
});
const result = await Promise.race([
running,
@@ -133,6 +172,15 @@ test(`superseded folder child settles when its completion was compacted into the
}
assert.notEqual(result, "still-waiting", "completed compacted child must not leave its folder waiting forever");
assert.ok(childId);
if (ownerChange !== "same") {
childBatcher.flush();
assert.equal((result as { success: boolean }).success, false, "the displaced walk must not report success");
if (ownerChange === "active") {
assert.equal(sftpTransferCenterStore.getTask(childId)?.status, "transferring");
assert.equal(sftpTransferCenterStore.getTask(childId)?.directoryEntryIdentity, "e".repeat(64));
} else assert.equal(sftpTransferCenterStore.getTask(childId), undefined, "old waiter must not recreate a failed row");
return;
}
assert.equal(sftpTransferCenterStore.getTask(childId), undefined, "completed row was compacted");
assert.equal(sftpTransferCenterStore.getTask(parent.id)?.directoryResumeCheckpoint?.completedEntries, 1);
assert.notEqual(result, "still-waiting", "completed compacted child must not leave its folder waiting forever");
@@ -140,6 +188,10 @@ test(`superseded folder child settles when its completion was compacted into the
});
}
}
}
}
}
}
test("resolveDirectoryResumeTargetRoot prefers staged replace path", () => {
assert.equal(
@@ -2131,3 +2183,202 @@ for (const scenario of ["deleted-id", "ambiguous-legacy-label"] as const) {
assert.equal(uploads, 0);
});
}
for (const completeDuringSetup of [false, true]) {
test(`large-history replace recovery retransfers retained completed children into its empty stage: late=${completeDuringSetup}`, async (t) => {
const { sftpTransferCenterStore: store } = await import("../sftpTransferCenterStore");
const { createDedicatedResumeChildUpdateBatcher } = await import("../../app/dedicatedResumeProgress");
const originalGet = netcattyBridge.get;
const previousLocalStorage = Object.getOwnPropertyDescriptor(globalThis, "localStorage");
Object.defineProperty(globalThis, "localStorage", {
configurable: true, value: { getItem: () => "2", setItem: () => {}, removeItem: () => {} },
});
const { createRequire } = await import("node:module");
const tempDirBridge = createRequire(import.meta.url)("../../../electron/bridges/tempDirBridge.cjs");
const root = await fs.promises.mkdtemp(`${tempDirBridge.getTempFilePath("folder-rebuild-test")}-`);
const finalPath = path.join(root, "final");
const stage = `${finalPath}.netcatty-retained-replace.part`;
await fs.promises.mkdir(stage);
await fs.promises.writeFile(path.join(stage, "b.txt"), "old contents");
const parent: TransferTask = {
id: "retained-replace", fileName: "folder", sourcePath: "/remote/folder", targetPath: finalPath,
stagedTargetPath: stage, replaceExistingTarget: true, sourceConnectionId: "old-sftp", targetConnectionId: "local",
sourceHostId: "h1", direction: "download", status: "transferring", totalBytes: 2,
transferredBytes: 1, speed: 0, startTime: 1, isDirectory: true, progressMode: "files", reconnectRequired: true,
};
const retained: TransferTask = {
...parent, id: "retained-replace-b", parentTaskId: parent.id, isDirectory: false,
sourcePath: "/remote/folder/b.txt", targetPath: `${stage}/b.txt`, fileName: "b.txt",
status: completeDuringSetup ? "interrupted" : "completed", totalBytes: 10, transferredBytes: 10, checkpointBytes: 10,
sourceLastModified: 2, directoryEntryIndex: 1,
directoryEntryIdentity: createDirectoryEntryIdentity({
sourcePath: "/remote/folder/b.txt", targetPath: `${stage}/b.txt`, size: 10, lastModified: 2,
}),
};
store.upsertTasks([parent, retained]);
assert.ok(store.getTask(retained.id), "missing earlier manifest index retains the completed row");
const batcher = createDedicatedResumeChildUpdateBatcher({
getTaskCount: () => 4096,
hasTask: (id) => !!store.getTask(id),
upsertTasks: (updates) => store.upsertTasks(updates),
});
let releaseFirst!: () => void;
const secondDispatched = new Promise<void>((resolve) => { releaseFirst = resolve; });
t.after(async () => {
await fs.promises.rm(root, { recursive: true, force: true });
netcattyBridge.get = originalGet;
store.patchTask(parent.id, { status: "completed" });
store.dismiss(parent.id);
if (previousLocalStorage) Object.defineProperty(globalThis, "localStorage", previousLocalStorage);
else Reflect.deleteProperty(globalThis, "localStorage");
resetDedicatedSessionOpenGateForTests();
});
(netcattyBridge as { get: () => unknown }).get = () => ({
openSftp: async () => "dedicated-sftp", closeSftp: async () => {},
listSftp: async (_id: string, remotePath: string) => {
if (completeDuringSetup) store.patchTask(retained.id, { status: "completed", ownerId: "dedicated-resume" });
return remotePath === parent.sourcePath
? ["a.txt", "b.txt"].map((name) => ({ name, type: "file", size: 10, lastModified: 2 })) : [];
},
mkdirLocal: async (target: string) => { await fs.promises.mkdir(target, { recursive: true }); },
statLocal: async (target: string) => {
try {
const stat = await fs.promises.stat(target);
return { type: stat.isDirectory() ? "directory" : "file", size: stat.size, lastModified: stat.mtimeMs };
} catch (error) {
if ((error as NodeJS.ErrnoException).code === "ENOENT") return null;
throw error;
}
},
statSftp: async (_id: string, source: string) => {
if (source.endsWith("a.txt")) await secondDispatched;
return { type: "file", size: 10, lastModified: 2 };
},
deleteLocalFile: async (target: string) => { await fs.promises.rm(target, { recursive: true }); },
renameLocalFile: async (source: string, target: string) => { await fs.promises.rename(source, target); },
startStreamTransfer: async (options: { targetPath: string; checkpointBytes: number }) => {
assert.equal(options.checkpointBytes, 0, "an empty replacement stage cannot reuse byte checkpoints");
await fs.promises.writeFile(options.targetPath, `new ${path.basename(options.targetPath)}`);
return {};
},
});
const result = await resumeTransferWithDedicatedSession(parent, {
hosts: [host("h1", "box")], keys: [], identities: [],
}, undefined, {
children: [retained],
flushChildUpdates: () => batcher.flush(),
onChildUpdate: (task, observation) => {
batcher.push(task, observation);
if (task.id === retained.id && task.status === "completed") releaseFirst();
return true;
},
onDirectoryCheckpointUpdate: (value) => store.patchTask(parent.id, { directoryResumeCheckpoint: value }),
});
batcher.flush();
assert.equal(result.success, true, result.error);
assert.deepEqual((await fs.promises.readdir(finalPath)).sort(), ["a.txt", "b.txt"],
"published replacement must include the previously completed file");
assert.equal(await fs.promises.readFile(path.join(finalPath, "b.txt"), "utf8"), "new b.txt");
});
}
for (const boundary of ["before-admission", "source-missing", "source-modified", "after-completion", "after-commit"] as const) {
test(`dedicated recovery preserves replacement ownership ${boundary}`, async (t) => {
const { sftpTransferCenterStore: store } = await import("../sftpTransferCenterStore");
const { createDedicatedResumeChildUpdateBatcher } = await import("../../app/dedicatedResumeProgress");
const originalGet = netcattyBridge.get;
const originalStorage = Object.getOwnPropertyDescriptor(globalThis, "localStorage");
Object.defineProperty(globalThis, "localStorage", { configurable: true, value: {
getItem: () => "2", setItem: () => {}, removeItem: () => {},
} });
const parent: TransferTask = {
id: `ownership-${boundary}`, fileName: "folder", sourcePath: "/local/folder", targetPath: "/remote/folder",
sourceConnectionId: "local", targetConnectionId: "old", targetHostId: "h1", direction: "upload",
status: "transferring", totalBytes: 2, transferredBytes: 0, speed: 0, startTime: 1,
isDirectory: true, progressMode: "files", reconnectRequired: true,
};
const child: TransferTask = {
...parent, id: `${parent.id}-a`, parentTaskId: parent.id, isDirectory: false,
sourcePath: "/local/folder/a", targetPath: "/remote/folder/a", fileName: "a", totalBytes: 1,
status: "interrupted", directoryEntryIndex: 0, directoryEntryIdentity: "a".repeat(64), sourceLastModified: 1,
};
store.upsertTasks([parent, boundary === "before-admission"
? { ...child, status: "transferring", directoryEntryIdentity: "b".repeat(64), lifecycleEpoch: 1 }
: child]);
const batch = createDedicatedResumeChildUpdateBatcher({
getTaskCount: () => 4096, hasTask: (id) => id === child.id || id.startsWith("peer-"),
upsertTasks: (rows) => store.upsertTasks(rows.filter((row) => !row.id.startsWith("peer-"))),
});
if (boundary === "before-admission") {
for (let n = 0; n < 511; n++) batch.push({ ...child, id: `peer-${n}` });
}
let releaseB!: () => void;
const bGate = new Promise<void>((resolve) => { releaseB = resolve; });
let timedOut = false;
const watchdog = setTimeout(() => { timedOut = true; releaseB(); }, 1000);
t.after(() => {
clearTimeout(watchdog);
releaseB();
batch.flush();
store.patchTask(parent.id, { status: "completed" });
store.dismiss(parent.id);
netcattyBridge.get = originalGet;
if (originalStorage) Object.defineProperty(globalThis, "localStorage", originalStorage);
else Reflect.deleteProperty(globalThis, "localStorage");
resetDedicatedSessionOpenGateForTests();
});
let starts = 0;
let displaced = 0;
let replaced = false;
(netcattyBridge as { get: () => unknown }).get = () => ({
openSftp: async () => "dedicated", closeSftp: async () => {
if (boundary === "after-commit") {
assert.equal(store.getTask(child.id), undefined, "completion must be committed before async cleanup");
store.upsertTasks([{ ...child, status: "transferring", directoryEntryIdentity: "b".repeat(64), lifecycleEpoch: 1 }]);
}
},
listLocalTree: async () => (boundary === "after-completion" ? ["a", "b"] : ["a"]).map((name) => ({
localPath: `/local/folder/${name}`, relativePath: name, type: "file", size: 1, lastModified: 1,
})),
mkdirSftp: async () => {}, statLocal: async () => {
if (boundary === "source-missing" || boundary === "source-modified") {
store.upsertTasks([{ ...child, status: "transferring", directoryEntryIdentity: "b".repeat(64), lifecycleEpoch: 1 }]);
if (boundary === "source-missing") return null;
return { type: "file", size: 1, lastModified: 2 };
}
return { type: "file", size: 1, lastModified: 1 };
},
startStreamTransfer: async (options: { sourcePath: string }) => {
starts++;
if (options.sourcePath.endsWith("/b")) await bGate;
return {};
},
});
const result = await resumeTransferWithDedicatedSession(parent, {
hosts: [host("h1", "box")], keys: [], identities: [],
}, undefined, {
children: [child],
onChildSuperseded: (id) => { displaced++; batch.discard(id); },
flushChildUpdates: () => batch.flush(),
onChildUpdate: (task, observation) => {
batch.push(task, observation);
if (boundary === "after-completion" && task.id === child.id && task.status === "completed" && !replaced) {
replaced = true;
queueMicrotask(() => {
try {
store.upsertTasks([{ ...task, status: "transferring", directoryEntryIdentity: "b".repeat(64), lifecycleEpoch: 1 }]);
} finally { releaseB(); }
});
}
return true;
},
});
batch.flush();
assert.equal(timedOut, false);
assert.equal(result.success, boundary === "after-commit", "only a committed success may precede a legitimate replacement");
assert.equal(store.getTask(child.id)?.directoryEntryIdentity, "b".repeat(64));
assert.equal(store.getTask(child.id)?.status, "transferring");
if (boundary !== "after-completion" && boundary !== "after-commit") assert.equal(starts, 0, "source validation and a full batch must not bypass ownership");
else assert.equal(displaced, boundary === "after-commit" ? 0 : 1);
});
}
@@ -1,4 +1,4 @@
import { runTransferAndWaitForOwner } from "./waitForTransferOwner";
import { runTransferAndWaitForOwner, TransferOwnerChangedError, type TransferOwnerObservation } from "./waitForTransferOwner";
import type { Host, Identity, KnownHost, SSHKey, TerminalSettings, TransferTask } from "../../../domain/models";
import { validateTransferResumeSource } from "../../../domain/sftpTransferCenter";
import { STORAGE_KEY_SFTP_TRANSFER_CONCURRENCY } from "../../../infrastructure/config/storageKeys";
@@ -62,7 +62,10 @@ export type DedicatedResumeResult = {
export type DedicatedResumeOptions = {
children?: readonly TransferTask[];
onChildUpdate?: (child: TransferTask) => void;
/** Return true only when the consumer takes responsibility for disposing the observation. */
onChildUpdate?: (child: TransferTask, observation?: TransferOwnerObservation) => unknown;
onChildSuperseded?: (taskId: string) => void;
flushChildUpdates?: () => void;
onDirectoryCheckpointUpdate?: (checkpoint: TransferTask["directoryResumeCheckpoint"]) => void;
shouldAbort?: () => boolean;
};
@@ -1090,6 +1093,12 @@ async function resumeDirectoryWithDedicatedSession(
// the source since the interrupted attempt, so rebuild it from empty.
await resetDirectoryReplaceStage(parent, endpoints, targetSftpId);
}
// Session setup and traversal may finish more children. Capture only
// after resetting the stage/checkpoint: all these completions refer to
// the discarded destination, while later completions remain valid.
const completedAtRestart = new Map(sftpTransferCenterStore.getSnapshot().tasks
.filter((child) => child.parentTaskId === parent.id && child.status === "completed")
.map((child) => [child.id, child]));
const destRoot = resolveDirectoryResumeTargetRoot(parent);
if (endpoints.isDownload) await ensureLocalDir(destRoot);
if (targetSftpId) await ensureRemoteDir(targetSftpId, destRoot);
@@ -1188,8 +1197,25 @@ async function resumeDirectoryWithDedicatedSession(
}
if (targetSftpId) await ensureRemoteDir(targetSftpId, getParentPath(file.targetPath));
let countedCompleted = false;
const onOwnerChanged = () => {
if (countedCompleted) {
countedCompleted = false;
completedCount -= 1;
failedCount += 1;
}
options?.onChildSuperseded?.(childId);
};
let childObservation: TransferOwnerObservation | undefined = sftpTransferCenterStore.observeTaskSettlement(
childBase, resetPersistedCheckpoint ? completedAtRestart.get(childId) : undefined, onOwnerChanged,
);
const publishChild = (child: TransferTask) => {
if (childObservation?.hasIdentityConflict()) throw new TransferOwnerChangedError("Transfer identity changed before child update");
if (options?.onChildUpdate?.(child, childObservation) === true) childObservation = undefined;
};
try {
if (options?.shouldAbort?.()) throw new Error("Transfer cancelled");
if (childObservation.hasIdentityConflict()) throw new TransferOwnerChangedError("Transfer identity changed before source validation");
const sourceType = endpoints.isUpload ? "local" as const : "sftp" as const;
const targetType = endpoints.isDownload ? "local" as const : "sftp" as const;
@@ -1217,8 +1243,7 @@ async function resumeDirectoryWithDedicatedSession(
sourceLastModified: sourceStat.lastModified,
};
} else if (classified.kind === "modified") {
attentionCount += 1;
options?.onChildUpdate?.({
publishChild({
...childBase,
status: "attention",
error: classified.message || validationError || "Source was modified",
@@ -1227,6 +1252,7 @@ async function resumeDirectoryWithDedicatedSession(
phase: undefined,
retryable: true,
});
attentionCount += 1;
return;
} else if (classified.kind === "fatal") {
throw new Error(classified.message || validationError || "Resume validation failed");
@@ -1251,41 +1277,49 @@ async function resumeDirectoryWithDedicatedSession(
// Re-check abort after async stat before inserting a transferring child.
if (options?.shouldAbort?.()) throw new Error("Transfer cancelled");
options?.onChildUpdate?.(childBase);
const streamResult = await runTransferAndWaitForOwner(childBase, () => {
// Publish only after admission, while identity changes are observed.
options?.onChildUpdate?.(childBase);
return bridge.startStreamTransfer!({
transferId: childId,
// Lifecycle events must carry the current identity even while
// large-history renderer child updates remain batched.
parentTaskId: childBase.parentTaskId,
directoryEntryIndex: childBase.directoryEntryIndex,
directoryEntryIdentity: childBase.directoryEntryIdentity,
sourcePath: file.sourcePath,
targetPath: file.targetPath,
sourceType,
targetType,
sourceSftpId,
targetSftpId,
sourceHostId: endpoints.sourceHost?.id,
targetHostId: endpoints.targetHost?.id,
totalBytes: Number.isFinite(childBase.totalBytes)
? childBase.totalBytes
: (file.size || undefined),
resumable: parent.resumable !== false,
checkpointBytes: childBase.checkpointBytes ?? 0,
resumeStage: childBase.resumeStage,
downloadCheckpointBytes: childBase.downloadCheckpointBytes,
uploadCheckpointBytes: childBase.uploadCheckpointBytes,
sourceFingerprint: childBase.sourceFingerprint,
skipAdmission: true,
});
}, () => options?.shouldAbort?.() === true, pausedAtResume.get(childId),
resetPersistedCheckpoint ? completedAtRestart.get(childId) : undefined,
// Drop queued UI updates at the ownership change itself: waiting
// for this invocation's reply may allow an intervening batch flush.
onOwnerChanged,
(observation) => { childObservation = observation; return true; }, childObservation);
const streamResult = await runTransferAndWaitForOwner(childBase, () => bridge.startStreamTransfer!({
transferId: childId,
// Lifecycle events must carry the current identity even while
// large-history renderer child updates remain batched.
parentTaskId: childBase.parentTaskId,
directoryEntryIndex: childBase.directoryEntryIndex,
directoryEntryIdentity: childBase.directoryEntryIdentity,
sourcePath: file.sourcePath,
targetPath: file.targetPath,
sourceType,
targetType,
sourceSftpId,
targetSftpId,
sourceHostId: endpoints.sourceHost?.id,
targetHostId: endpoints.targetHost?.id,
totalBytes: Number.isFinite(childBase.totalBytes)
? childBase.totalBytes
: (file.size || undefined),
resumable: parent.resumable !== false,
checkpointBytes: childBase.checkpointBytes ?? 0,
resumeStage: childBase.resumeStage,
downloadCheckpointBytes: childBase.downloadCheckpointBytes,
uploadCheckpointBytes: childBase.uploadCheckpointBytes,
sourceFingerprint: childBase.sourceFingerprint,
skipAdmission: true,
}), () => options?.shouldAbort?.() === true, pausedAtResume.get(childId));
if (childObservation?.hasIdentityConflict()) throw new TransferOwnerChangedError("Transfer identity changed before child update");
if (streamResult?.error || streamResult?.cancelled) {
throw new Error(streamResult.error || "Transfer cancelled");
}
completedCount += 1;
options?.onChildUpdate?.({
publishChild({
...childBase,
status: "completed",
transferredBytes: Number.isFinite(childBase.totalBytes)
@@ -1297,13 +1331,21 @@ async function resumeDirectoryWithDedicatedSession(
reconnectRequired: false,
phase: undefined,
});
countedCompleted = true;
completedCount += 1;
bumpParentProgress(0);
} catch (error) {
if (options?.shouldAbort?.() || /cancelled|canceled/i.test(error instanceof Error ? error.message : String(error))) {
throw error instanceof Error ? error : new Error(String(error));
}
failedCount += 1;
options?.onChildUpdate?.({
// Another invocation owns this ID now. Report this walk's failure
// without replacing the winner or recreating its compacted row.
if (error instanceof TransferOwnerChangedError || childObservation?.hasIdentityConflict()) {
options?.onChildSuperseded?.(childId);
return;
}
publishChild({
...childBase,
status: "failed",
error: error instanceof Error ? error.message : String(error),
@@ -1312,6 +1354,8 @@ async function resumeDirectoryWithDedicatedSession(
reconnectRequired: false,
phase: undefined,
});
} finally {
childObservation?.dispose();
}
},
);
@@ -1322,6 +1366,10 @@ async function resumeDirectoryWithDedicatedSession(
if (result?.error) throw new Error(result.error);
// Commit deferred child outcomes before deciding success or promoting a stage.
// Async connection cleanup must not outlive provisional completion evidence.
options?.flushChildUpdates?.();
if (attentionCount > 0 && failedCount === 0 && completedCount + attentionCount >= totalFiles) {
return {
success: false,
@@ -334,7 +334,8 @@ test("live directory download rejects a Windows backslash traversal entry", asyn
}
});
test("live folder settles a superseded child whose completed row was compacted", async () => {
for (const ownerChange of ["same", "changed-active", "changed-completed"] as const) {
test(`live folder settles superseded ownership without clobbering its replacement: ${ownerChange}`, async () => {
const { sftpTransferCenterStore } = await import("../sftpTransferCenterStore");
const previousWindow = (globalThis as { window?: unknown }).window;
const previousLocalStorage = Object.getOwnPropertyDescriptor(globalThis, "localStorage");
@@ -355,10 +356,18 @@ test("live folder settles a superseded child whose completed row was compacted",
startStreamTransfer: async (options: { transferId: string }) => {
childId = options.transferId;
sftpTransferCenterStore.publishOwner("live-compacted-owner", tasks);
// The winning invocation has already completed; history compaction drops its row.
sftpTransferCenterStore.ingestBackgroundEvent({
type: "completed", transferId: childId, transferred: 1, totalBytes: 1, lifecycleEpoch: 0,
});
if (ownerChange !== "same") {
const current = sftpTransferCenterStore.getTask(childId)!;
sftpTransferCenterStore.admitTaskRun({ ...current, directoryEntryIdentity: "b".repeat(64) });
}
if (ownerChange !== "changed-active") {
// The winning invocation completed; history compaction drops its row.
sftpTransferCenterStore.ingestBackgroundEvent({
type: "completed", transferId: childId, transferred: 1, totalBytes: 1, lifecycleEpoch: 0,
});
}
tasks = [...sftpTransferCenterStore.getSnapshot().tasks];
transfersRef.current = tasks;
return { superseded: true };
},
} };
@@ -387,10 +396,15 @@ test("live folder settles a superseded child whose completed row was compacted",
new Promise<"still-waiting">((resolve) => setTimeout(() => resolve("still-waiting"), 450)),
]);
assert.ok(childId);
assert.equal(sftpTransferCenterStore.getTask(childId), undefined);
assert.equal(sftpTransferCenterStore.getTask(root.id)?.directoryResumeCheckpoint?.completedEntries, 1);
if (ownerChange === "changed-active") {
assert.equal(tasks.find((task) => task.id === childId)?.status, "transferring");
assert.equal(tasks.find((task) => task.id === childId)?.directoryEntryIdentity, "b".repeat(64));
} else {
assert.equal(tasks.find((task) => task.id === childId), undefined, "old waiter must not reinsert a compacted child");
assert.equal(sftpTransferCenterStore.getTask(root.id)?.directoryResumeCheckpoint?.completedEntries, 1);
}
assert.notEqual(result, "still-waiting", "compacted completion must settle the live folder");
assert.equal(result, 0);
assert.equal(result, ownerChange === "same" ? 0 : 1);
} finally {
cancelledTasksRef.current.add(root.id);
await running?.catch(() => {});
@@ -404,3 +418,4 @@ test("live folder settles a superseded child whose completed row was compacted",
}
});
}
@@ -1,5 +1,5 @@
import { reconcileSupersededControls } from "./globalSftpTransferControl";
import { runTransferAndWaitForOwner } from "./waitForTransferOwner";
import { runTransferAndWaitForOwner, TransferOwnerChangedError } from "./waitForTransferOwner";
import { useCallback, type Dispatch, type MutableRefObject, type SetStateAction } from "react";
import type { Host, SftpFileEntry, SftpFilenameEncoding, TransferStatus, TransferTask } from "../../../domain/models";
import {
@@ -978,6 +978,10 @@ export function useSftpDirectoryTransferOps({
errors.push(err instanceof Error ? err : new Error(message));
return;
}
if (err instanceof TransferOwnerChangedError) {
errors.push(err);
return;
}
// Mark child as failed
setTransfers((prev) =>
prev.map((t) =>
@@ -169,3 +169,284 @@ test("fresh directory recovery authorizes only an unchanged retained pause under
store.patchTask("root", { status: "paused", lifecycleEpoch: 6 });
assert.equal(store.admitTaskRun(task, currentPause), "paused", "parent pause still wins");
});
for (const compactNewOwner of [false, true]) {
test(`superseded waiter rejects a changed identity even after new owner compaction: ${compactNewOwner}`, async (t) => {
const { sftpTransferCenterStore: store } = await import("../sftpTransferCenterStore");
const { runTransferAndWaitForOwner } = await import("./waitForTransferOwner");
const task = { ...child(), id: `changed-owner-${compactNewOwner}`, parentTaskId: `changed-root-${compactNewOwner}` };
store.upsertTasks([{ ...task, id: task.parentTaskId, parentTaskId: undefined, isDirectory: true }, task]);
let abort = false;
const running = runTransferAndWaitForOwner(task, async () => {
const changed = { ...task, directoryEntryIdentity: "b".repeat(64) };
assert.equal(store.admitTaskRun(changed), "ready");
if (compactNewOwner) {
store.patchTask(changed.id, { status: "completed" });
assert.equal(store.getTask(changed.id), undefined);
}
return { superseded: true };
}, () => abort);
const settled = running.then(() => "completed", (error: Error) => error.message);
t.after(async () => {
abort = true;
await settled;
store.patchTask(task.parentTaskId, { status: "completed" });
store.dismiss(task.parentTaskId);
});
assert.match(String(await Promise.race([
settled, new Promise((resolve) => setTimeout(() => resolve("still-waiting"), 450)),
])), /identity changed/i);
});
}
test("destination rebuild restarts only the captured completion and preserves newer controls", async (t) => {
const { latchTransferPause, resetTransferPauseLatchesForTests } = await import("./transferPauseLatch");
const store = createSftpTransferCenterStore();
const task = { ...child(), parentTaskId: undefined, status: "completed" as const, checkpointBytes: 10 };
store.upsertTasks([task]);
const captured = store.getTask(task.id)!;
const restart = { ...task, status: "transferring" as const, checkpointBytes: 0, transferredBytes: 0 };
latchTransferPause(task.id);
t.after(resetTransferPauseLatchesForTests);
assert.equal(store.admitTaskRun(restart, undefined, captured), "paused");
resetTransferPauseLatchesForTests();
assert.equal(store.admitTaskRun(restart, undefined, captured), "ready");
assert.equal(store.getTask(task.id)?.checkpointBytes, 0);
store.patchTask(task.id, { status: "completed" });
assert.equal(store.admitTaskRun(restart, undefined, captured), "completed", "new completion must not be restarted");
store.patchTask(task.id, { status: "cancelled" });
assert.equal(store.admitTaskRun(restart, undefined, captured), "cancelled");
});
test("destination rebuild waiting on pause reuses a newer compacted completion", async (t) => {
const { sftpTransferCenterStore: store } = await import("../sftpTransferCenterStore");
const { runTransferAndWaitForOwner } = await import("./waitForTransferOwner");
const { latchTransferPause, resetTransferPauseLatchesForTests } = await import("./transferPauseLatch");
const task = { ...child(), id: "restart-wait-child", parentTaskId: "restart-wait-root", directoryEntryIndex: 1 };
const root = { ...task, id: task.parentTaskId, parentTaskId: undefined, isDirectory: true };
store.upsertTasks([root, { ...task, status: "completed" }]);
const captured = store.getTask(task.id)!;
latchTransferPause(task.id);
let starts = 0;
let abort = false;
const running = runTransferAndWaitForOwner(task, async () => { starts += 1; return {}; }, () => abort, undefined, captured);
const settled = running.then(() => "completed", (error: Error) => error.message);
t.after(async () => {
abort = true;
resetTransferPauseLatchesForTests();
await settled;
store.patchTask(root.id, { status: "completed" });
store.dismiss(root.id);
});
// An unrelated lifecycle emit must not pin the old completion in the observation.
store.patchTask(root.id, { reconnectRequired: true });
assert.equal(await Promise.race([
settled, new Promise((resolve) => setTimeout(() => resolve("still-waiting"), 230)),
]), "still-waiting", "old completion cannot finish recovery while a new transfer is still needed");
assert.equal(starts, 0);
resetTransferPauseLatchesForTests();
store.patchTask(task.id, { status: "transferring", lifecycleEpoch: 1 });
store.upsertTasks([{ ...task, id: "restart-wait-first", directoryEntryIndex: 0, status: "completed" }]);
store.patchTask(task.id, { status: "completed", lifecycleEpoch: 2 });
assert.equal(store.getTask(task.id), undefined);
assert.equal(await settled, "completed");
assert.equal(starts, 0, "new completion after stage rebuild must be reused even when its row was compacted");
});
test("paused admission cannot reclaim an identity replaced while it waited", async (t) => {
const { sftpTransferCenterStore: store } = await import("../sftpTransferCenterStore");
const { runTransferAndWaitForOwner } = await import("./waitForTransferOwner");
const task = { ...child(), id: "paused-displaced", parentTaskId: undefined };
store.upsertTasks([{ ...task, status: "paused" }]);
let starts = 0;
let abort = false;
const running = runTransferAndWaitForOwner(task, async () => { starts += 1; return {}; }, () => abort);
const settled = running.then(() => "completed", (error: Error) => error.message);
t.after(async () => {
abort = true;
await settled;
store.patchTask(task.id, { status: "completed" });
store.dismiss(task.id);
});
store.upsertTasks([{ ...task, status: "transferring", directoryEntryIdentity: "b".repeat(64), lifecycleEpoch: 1 }]);
assert.match(await settled, /identity changed/i);
assert.equal(starts, 0);
assert.equal(store.getTask(task.id)?.directoryEntryIdentity, "b".repeat(64));
});
for (const identityChangesDuringWait of [false, true]) {
test(`paused recovery distinguishes initial stale identity from later ownership: changed=${identityChangesDuringWait}`, async (t) => {
const { sftpTransferCenterStore: store } = await import("../sftpTransferCenterStore");
const { runTransferAndWaitForOwner } = await import("./waitForTransferOwner");
const { latchTransferPause, resetTransferPauseLatchesForTests } = await import("./transferPauseLatch");
const task = { ...child(), id: `paused-origin-${identityChangesDuringWait}`, parentTaskId: `paused-origin-root-${identityChangesDuringWait}` };
const root = { ...task, id: task.parentTaskId, parentTaskId: undefined, isDirectory: true };
store.upsertTasks([root, {
...task, status: identityChangesDuringWait ? "failed" : "paused",
directoryEntryIdentity: identityChangesDuringWait ? task.directoryEntryIdentity : "b".repeat(64),
}]);
const initialPause = store.getTask(task.id)!;
latchTransferPause(task.id);
let starts = 0;
let abort = false;
let conflicts = 0;
const running = runTransferAndWaitForOwner(task, async () => { starts += 1; return {}; }, () => abort,
identityChangesDuringWait ? undefined : initialPause, undefined, () => { conflicts += 1; });
const settled = running.then(() => "completed", (error: Error) => error.message);
t.after(async () => {
abort = true;
resetTransferPauseLatchesForTests();
await settled;
store.patchTask(root.id, { status: "completed" });
store.dismiss(root.id);
});
// This captures the prior failed result, or republishes the unchanged
// initial old identity. Neither is evidence of a replacement by itself.
store.patchTask(root.id, { reconnectRequired: true });
resetTransferPauseLatchesForTests();
if (identityChangesDuringWait) {
store.patchTask(task.id, { status: "transferring", directoryEntryIdentity: "b".repeat(64), lifecycleEpoch: 1 });
}
const outcome = await settled;
if (identityChangesDuringWait) {
assert.match(outcome, /identity changed/i);
assert.equal(starts, 0);
assert.equal(conflicts, 1);
assert.equal(store.getTask(task.id)?.directoryEntryIdentity, "b".repeat(64));
} else {
assert.equal(outcome, "completed");
assert.equal(starts, 1);
assert.equal(conflicts, 0);
}
});
}
test("paused retry observes newer same-identity completion after an old failure", async (t) => {
const { sftpTransferCenterStore: store } = await import("../sftpTransferCenterStore");
const { runTransferAndWaitForOwner } = await import("./waitForTransferOwner");
const { latchTransferPause, resetTransferPauseLatchesForTests } = await import("./transferPauseLatch");
const task = { ...child(), id: "failed-then-completed", parentTaskId: "failed-then-completed-root" };
const root = { ...task, id: task.parentTaskId, parentTaskId: undefined, isDirectory: true };
store.upsertTasks([root, { ...task, status: "failed" }]);
latchTransferPause(task.id);
let starts = 0;
let abort = false;
const running = runTransferAndWaitForOwner(task, async () => { starts += 1; return {}; }, () => abort);
const settled = running.then(() => "completed", (error: Error) => error.message);
t.after(async () => {
abort = true;
resetTransferPauseLatchesForTests();
await settled;
store.patchTask(root.id, { status: "completed" });
store.dismiss(root.id);
});
store.patchTask(root.id, { reconnectRequired: true });
resetTransferPauseLatchesForTests();
store.patchTask(task.id, { status: "transferring", lifecycleEpoch: 1 });
store.patchTask(task.id, { status: "completed", lifecycleEpoch: 2 });
assert.equal(store.getTask(task.id), undefined);
assert.equal(await settled, "completed");
assert.equal(starts, 0, "a newer completed attempt must replace the captured old failure");
});
test("an unchanged owner retains its original transport rejection", async (t) => {
const { sftpTransferCenterStore: store } = await import("../sftpTransferCenterStore");
const { runTransferAndWaitForOwner } = await import("./waitForTransferOwner");
const task = { ...child(), id: "unchanged-rejected-owner", parentTaskId: undefined };
store.upsertTasks([task]);
t.after(() => { store.patchTask(task.id, { status: "completed" }); store.dismiss(task.id); });
const error = new Error("original transport failure");
await assert.rejects(runTransferAndWaitForOwner(task, async () => { throw error; }, () => false),
(actual) => actual === error);
});
for (const activation of ["patch", "progress", "owner"] as const) {
test(`initial stale identity exemption ends when another owner activates that row: ${activation}`, async (t) => {
const { sftpTransferCenterStore: store } = await import("../sftpTransferCenterStore");
const { runTransferAndWaitForOwner } = await import("./waitForTransferOwner");
const { latchTransferPause, resetTransferPauseLatchesForTests } = await import("./transferPauseLatch");
const task = { ...child(), id: "activated-initial-identity", parentTaskId: undefined };
store.upsertTasks([{ ...task, status: "paused", directoryEntryIdentity: "b".repeat(64), lifecycleEpoch: 1 }]);
const initial = store.getTask(task.id)!;
latchTransferPause(task.id);
let starts = 0;
let abort = false;
const running = runTransferAndWaitForOwner(task, async () => { starts += 1; return {}; }, () => abort, initial);
const settled = running.then(() => "completed", (error: Error) => error.message);
t.after(async () => {
abort = true;
resetTransferPauseLatchesForTests();
await settled;
store.patchTask(task.id, { status: "completed" });
store.dismiss(task.id);
});
resetTransferPauseLatchesForTests();
if (activation === "progress") {
store.ingestBackgroundEvent({ type: "progress", transferId: task.id,
transferred: 1, lifecycleEpoch: 2, lifecycleState: "transferring" });
} else if (activation === "owner") store.patchTask(task.id, { ownerId: "replacement-owner" });
else store.patchTask(task.id, { status: "transferring", lifecycleEpoch: 2 });
assert.match(String(await Promise.race([
settled, new Promise((resolve) => setTimeout(() => resolve("still-waiting"), 450)),
])), /identity changed/i);
assert.equal(starts, 0);
assert.equal(store.getTask(task.id)?.directoryEntryIdentity, "b".repeat(64));
});
}
test("already active different identity is rejected before first admission", async (t) => {
const { sftpTransferCenterStore: store } = await import("../sftpTransferCenterStore");
const { runTransferAndWaitForOwner } = await import("./waitForTransferOwner");
const task = { ...child(), id: "already-active-identity", parentTaskId: undefined };
store.upsertTasks([{ ...task, directoryEntryIdentity: "b".repeat(64), lifecycleEpoch: 1 }]);
t.after(() => { store.patchTask(task.id, { status: "completed" }); store.dismiss(task.id); });
let starts = 0;
let conflicts = 0;
await assert.rejects(runTransferAndWaitForOwner(task, async () => { starts += 1; return {}; }, () => false,
undefined, undefined, () => { conflicts += 1; }), /identity changed/i);
assert.equal(starts, 0);
assert.equal(conflicts, 1);
assert.equal(store.getTask(task.id)?.directoryEntryIdentity, "b".repeat(64));
});
for (const replacementStatus of ["transferring", "completed"] as const) {
for (const oldStatus of ["completed", "failed"] as const) {
test(`deferred ${oldStatus} cannot overwrite a ${replacementStatus} replacement after waiter exit`, async () => {
const { sftpTransferCenterStore: store } = await import("../sftpTransferCenterStore");
const { runTransferAndWaitForOwner } = await import("./waitForTransferOwner");
const { createDedicatedResumeChildUpdateBatcher } = await import("../../app/dedicatedResumeProgress");
const task = child();
const root = { ...task, id: task.parentTaskId!, parentTaskId: undefined, isDirectory: true };
const batcher = createDedicatedResumeChildUpdateBatcher({
getTaskCount: () => 4096, hasTask: () => true,
upsertTasks: (updates) => store.upsertTasks(updates),
});
store.upsertTasks([root, task]);
let observation: ReturnType<typeof store.observeTaskSettlement> | undefined;
try {
await runTransferAndWaitForOwner(task, async () => {
store.patchTask(task.id, { status: oldStatus });
return {};
}, () => false, undefined, undefined, () => batcher.discard(task.id), (value) => {
observation = value;
return true;
});
assert.ok(observation);
batcher.push({ ...task, status: oldStatus }, observation);
store.upsertTasks([{ ...task, directoryEntryIdentity: "b".repeat(64), status: replacementStatus }]);
const replacement = store.getTask(task.id);
batcher.flush();
assert.equal(store.getTask(task.id), replacement);
if (replacementStatus === "transferring") assert.equal(replacement?.directoryEntryIdentity, "b".repeat(64));
else assert.equal(replacement, undefined, "the replacement remains compacted");
assert.equal(observation.read(), undefined, "the retained observation is disposed");
} finally {
batcher.flush();
observation?.dispose();
store.patchTask(root.id, { status: "completed" });
store.dismiss(root.id);
}
});
}
}
+24 -5
View File
@@ -1,6 +1,11 @@
import type { TransferTask } from "../../../domain/models";
import { sftpTransferCenterStore } from "../sftpTransferCenterStore";
/** The displaced invocation may fail its walk, but no longer owns the child row. */
export class TransferOwnerChangedError extends Error {}
export type TransferOwnerObservation = ReturnType<typeof sftpTransferCenterStore.observeTaskSettlement>;
type StreamResult = { error?: string; cancelled?: boolean; superseded?: boolean } | undefined;
/** A live waiter owns bounded settlement evidence; persisted history stays compact. */
@@ -9,33 +14,47 @@ export async function runTransferAndWaitForOwner(
start: () => Promise<StreamResult>,
shouldAbort: () => boolean,
pausedAtResume?: TransferTask,
completedAtRestart?: TransferTask,
onOwnerChanged?: () => void,
retainObservation?: (observation: TransferOwnerObservation) => boolean,
existingObservation?: TransferOwnerObservation,
): Promise<StreamResult> {
// Register before admission/start: an owner may finish while dispatch waits for resume.
let observation = sftpTransferCenterStore.observeTaskSettlement(task);
let observation = existingObservation ?? sftpTransferCenterStore.observeTaskSettlement(task, completedAtRestart, onOwnerChanged);
try {
for (;;) {
if (shouldAbort()) throw new Error("Transfer cancelled");
const admission = sftpTransferCenterStore.admitTaskRun(task, pausedAtResume);
if (observation.hasIdentityConflict()) throw new TransferOwnerChangedError("Transfer identity changed before dispatch");
const admission = sftpTransferCenterStore.admitTaskRun(task, pausedAtResume, completedAtRestart);
if (admission === "cancelled") throw new Error("Transfer cancelled");
if (admission === "completed" || observation.read()?.status === "completed") return {};
if (admission === "conflict") throw new Error("Transfer identity changed before dispatch");
if (admission === "conflict") throw new TransferOwnerChangedError("Transfer identity changed before dispatch");
if (admission === "ready") break;
await new Promise((resolve) => setTimeout(resolve, 200));
}
// Admission begins a new attempt; discard any previous failed settlement.
observation.dispose();
observation = sftpTransferCenterStore.observeTaskSettlement(task);
observation = sftpTransferCenterStore.observeTaskSettlement(task, undefined, onOwnerChanged);
const result = await start();
if (observation.hasIdentityConflict()) throw new TransferOwnerChangedError("Transfer identity changed while waiting for its owner");
if (!result?.superseded) return result;
for (;;) {
if (shouldAbort()) throw new Error("Transfer cancelled");
if (observation.hasIdentityConflict()) throw new TransferOwnerChangedError("Transfer identity changed while waiting for its owner");
const latest = observation.read();
if (latest?.status === "completed") return { ...result, superseded: false };
if (latest?.status === "failed") throw new Error(latest.error || "Transfer failed");
if (latest?.status === "cancelled") throw new Error("Transfer cancelled");
await new Promise((resolve) => setTimeout(resolve, 200));
}
} catch (error) {
// A displaced invocation can reject instead of returning superseded. Its
// transport error must not regain permission to write the winning row.
if (observation.hasIdentityConflict()) throw new TransferOwnerChangedError("Transfer identity changed while waiting for its owner");
throw error;
} finally {
observation.dispose();
// A deferred child update may outlive the invocation. Its caller must either
// retain this evidence through the write or release it after handling exit.
if (!retainObservation?.(observation)) observation.dispose();
}
}
+71 -16
View File
@@ -42,6 +42,14 @@ import { restoreSftpTransferHistoryCooperatively } from "./sftp/transferHistoryR
import { cancelExternalUploadRuntime } from "./sftp/externalUploadRuntime";
type Listener = () => void;
type TaskSettlementObserver = {
expected: TransferTask;
initialInactiveRow?: TransferTask;
settled?: TransferTask;
conflicted?: boolean;
ignoredCompletion?: TransferTask;
onIdentityConflict?: () => void;
};
// Ordinary bounded history restores synchronously so existing callers receive
// it immediately. Legacy directory snapshots can contain tens of thousands of
@@ -98,9 +106,9 @@ export interface SftpTransferCenterStore {
/** Single task row — same object identity until that row is patched. */
getTask(taskId: string): TransferTask | undefined;
/** Observe exact task settlement before completed child rows are compacted. */
observeTaskSettlement(task: TransferTask): { read(): TransferTask | undefined; dispose(): void };
observeTaskSettlement(task: TransferTask, ignoredCompletion?: TransferTask, onIdentityConflict?: () => void): { read(): TransferTask | undefined; hasIdentityConflict(): boolean; dispose(): void };
/** Publish an explicit dispatch identity without flushing large child-history batches. */
admitTaskRun(task: TransferTask, pausedAtResume?: TransferTask): "ready" | "paused" | "completed" | "cancelled" | "conflict";
admitTaskRun(task: TransferTask, pausedAtResume?: TransferTask, completedAtRestart?: TransferTask): "ready" | "paused" | "completed" | "cancelled" | "conflict";
getOwnerTasks(ownerId: string): TransferTask[];
publishOwner(ownerId: string, tasks: readonly TransferTask[]): void;
registerOwner(ownerId: string, controls: SftpTransferOwnerControls): () => void;
@@ -405,7 +413,7 @@ export function createSftpTransferCenterStore(persistence?: StorePersistence): S
let snapshotDirty = false;
const listeners = new Set<Listener>();
const progressListeners = new Set<Listener>();
const settlementObservers = new Map<string, Set<{ expected: TransferTask; settled?: TransferTask }>>();
const settlementObservers = new Map<string, Set<TaskSettlementObserver>>();
const matchesObservedTask = (expected: TransferTask, candidate: TransferTask) => (
expected.id === candidate.id
&& expected.sourcePath === candidate.sourcePath
@@ -414,6 +422,29 @@ export function createSftpTransferCenterStore(persistence?: StorePersistence): S
&& expected.directoryEntryIndex === candidate.directoryEntryIndex
&& expected.directoryEntryIdentity === candidate.directoryEntryIdentity
);
const captureObservedTask = (task: TransferTask) => {
for (const observer of settlementObservers.get(task.id) ?? []) {
if (observer.conflicted || task === observer.ignoredCompletion) continue;
const matchesExpected = matchesObservedTask(observer.expected, task);
// A retry may plan a new source identity while its initial stale row is
// still paused. Only the unchanged initial attempt is exempt: activating
// it or replacing its owner/epoch must not let the stale plan reclaim it.
if (!matchesExpected && observer.initialInactiveRow
&& matchesObservedTask(observer.initialInactiveRow, task)
&& observer.initialInactiveRow.status === task.status
&& observer.initialInactiveRow.lifecycleEpoch === task.lifecycleEpoch
&& observer.initialInactiveRow.ownerId === task.ownerId) continue;
observer.initialInactiveRow = undefined;
if (!matchesExpected) {
observer.conflicted = true;
observer.onIdentityConflict?.();
} else if (observer.settled?.status !== "completed") {
// Failed/cancelled evidence belongs to the latest observed attempt. A
// resumed owner can supersede it; exact successful completion stays final.
observer.settled = TERMINAL_OWNER_STATUSES.has(task.status) ? task : undefined;
}
}
};
const refreshBadgeSnapshot = () => {
const next = buildBadgeSnapshot(tasks);
if (
@@ -633,12 +664,7 @@ export function createSftpTransferCenterStore(persistence?: StorePersistence): S
// settles; everything else can release its whole tree here.
settleFinishedTransferControlState(tasks, tasks);
if (settlementObservers.size > 0) {
for (const task of tasks) {
if (!TERMINAL_OWNER_STATUSES.has(task.status)) continue;
for (const observer of settlementObservers.get(task.id) ?? []) {
if (!observer.settled && matchesObservedTask(observer.expected, task)) observer.settled = task;
}
}
for (const task of tasks) captureObservedTask(task);
}
const beforePrune = tasks;
tasks = pruneSftpTransferHistory(tasks);
@@ -1338,7 +1364,7 @@ export function createSftpTransferCenterStore(persistence?: StorePersistence): S
},
getSnapshot: () => ensureSnapshot(),
getBadgeSnapshot: () => badgeSnapshot,
admitTaskRun(incoming, pausedAtResume) {
admitTaskRun(incoming, pausedAtResume, completedAtRestart) {
const rootId = incoming.parentTaskId ?? incoming.id;
if (isTransferOrRootCancelled(rootId, incoming.id)) return "cancelled";
const parent = incoming.parentTaskId ? tasks.find((task) => task.id === rootId) : undefined;
@@ -1348,7 +1374,10 @@ export function createSftpTransferCenterStore(persistence?: StorePersistence): S
if (existing && (existing.sourcePath !== incoming.sourcePath || existing.targetPath !== incoming.targetPath
|| existing.parentTaskId !== incoming.parentTaskId)) return "conflict";
if (existing?.status === "cancelled") return "cancelled";
if (existing?.status === "completed") {
// Only the exact completion captured at the destination reset is
// stale. A newer completion or user control must keep its authority.
const restartsCompletion = existing?.status === "completed" && existing === completedAtRestart;
if (existing?.status === "completed" && !restartsCompletion) {
return matchesObservedTask(incoming, existing) ? "completed" : "conflict";
}
const resumesUnchangedPause = existing === pausedAtResume
@@ -1369,6 +1398,14 @@ export function createSftpTransferCenterStore(persistence?: StorePersistence): S
...existing,
directoryEntryIndex: incoming.directoryEntryIndex,
directoryEntryIdentity: incoming.directoryEntryIdentity,
...(restartsCompletion ? {
checkpointBytes: incoming.checkpointBytes,
transferredBytes: incoming.transferredBytes,
resumeStage: incoming.resumeStage,
downloadCheckpointBytes: incoming.downloadCheckpointBytes,
uploadCheckpointBytes: incoming.uploadCheckpointBytes,
sourceFingerprint: incoming.sourceFingerprint,
} : {}),
...(existing.status === "transferring" ? {} : {
status: "transferring" as const,
lifecycleEpoch: undefined,
@@ -1378,25 +1415,36 @@ export function createSftpTransferCenterStore(persistence?: StorePersistence): S
}),
};
tasks = next;
captureObservedTask(next[index]);
emitProgress(false);
return "ready";
},
observeTaskSettlement(expected) {
const observer: { expected: TransferTask; settled?: TransferTask } = { expected: { ...expected } };
observeTaskSettlement(expected, ignoredCompletion, onIdentityConflict) {
const initial = tasks.find((task) => task.id === expected.id);
const observer: TaskSettlementObserver = {
expected: { ...expected }, ignoredCompletion, onIdentityConflict,
initialInactiveRow: initial && ["paused", "failed", "interrupted", "attention"].includes(initial.status)
&& !matchesObservedTask(expected, initial) ? { ...initial } : undefined,
};
const observers = settlementObservers.get(expected.id) ?? new Set();
observers.add(observer);
settlementObservers.set(expected.id, observers);
// A different active owner may predate registration. Detect it now,
// before admission can publish the stale caller's identity over it.
if (initial) captureObservedTask(initial);
let disposed = false;
return {
read() {
if (disposed) return undefined;
if (observer.settled) return observer.settled;
const current = tasks.find((task) => task.id === expected.id);
return current && matchesObservedTask(observer.expected, current) ? current : undefined;
return current && current !== ignoredCompletion && matchesObservedTask(observer.expected, current) ? current : undefined;
},
hasIdentityConflict: () => !disposed && observer.conflicted === true,
dispose() {
disposed = true;
observer.settled = undefined;
observer.conflicted = undefined;
observers.delete(observer);
if (observers.size === 0) settlementObservers.delete(expected.id);
},
@@ -1662,6 +1710,7 @@ export function createSftpTransferCenterStore(persistence?: StorePersistence): S
const next = tasks.slice();
next[index] = nextTask;
tasks = next;
captureObservedTask(nextTask);
if (changed) {
const lifecycleChanged = (
(updates.status !== undefined && updates.status !== task.status)
@@ -2023,7 +2072,8 @@ export function createSftpTransferCenterStore(persistence?: StorePersistence): S
backgroundEventsDuringRestore.set(event.transferId, pending);
return;
}
let existing = tasks.find((task) => task.id === event.transferId);
let existingIndex = tasks.findIndex((task) => task.id === event.transferId);
let existing = tasks[existingIndex];
const eventParent = event.parentTaskId
? tasks.find((task) => task.id === event.parentTaskId)
: undefined;
@@ -2039,7 +2089,8 @@ export function createSftpTransferCenterStore(persistence?: StorePersistence): S
directoryEntryIdentity: event.directoryEntryIdentity ?? task.directoryEntryIdentity,
background: false,
} : task);
existing = tasks.find((task) => task.id === event.transferId);
existing = tasks[existingIndex];
captureObservedTask(existing);
}
// A delayed child progress event can arrive after the renderer compacted
// that completed child into its parent checkpoint. Never resurrect it as
@@ -2119,6 +2170,7 @@ export function createSftpTransferCenterStore(persistence?: StorePersistence): S
if ((event.type === "queued" || event.type === "started" || event.type === "progress") && !existing) {
const sourcePath = event.sourcePath ?? "";
const targetPath = event.targetPath ?? "";
existingIndex = tasks.length;
tasks.push({
id: event.transferId,
ownerId: eventParent?.ownerId
@@ -2352,6 +2404,9 @@ export function createSftpTransferCenterStore(persistence?: StorePersistence): S
phase: undefined,
} : task);
}
// Ownership changes must be observed synchronously even when a progress
// event takes the lightweight UI path. Reuse the located row, not a scan.
if (existingIndex >= 0) captureObservedTask(tasks[existingIndex]);
// Progress-only ticks skip prune + coalesce listener paints; lifecycle
// events still take the full emit path so UI never lags on pause/cancel.
if (event.type === "progress" && !persistImmediately) {
+118
View File
@@ -22,3 +22,121 @@ This addresses a separately reproduced folder-never-settles condition relevant
to #2568 and #3155. It does not establish the original reporters' precise cause.
Large-history recovery follow-up: stream lifecycle events carry the current child hierarchy identity. Before dispatch, the store admits the explicit retry into the current row without a full history compaction, so a batched old failed row cannot reject the new completion. Admission distinguishes pause waiting, cancellation, identity conflict and exact prior completion; active lifecycle epochs and newer pause/cancel intent remain protected.
## Follow-up: retained completions and changed owners (#3155)
Two remaining #3287 review cases reproduce independently of the original
Windows report:
- With the 4096-row batching path enabled, a retained completed child at manifest
index 1 can reach dispatch before index 0 is published. Rebuilding a replacement
stage deletes that child's old output, but completion admission used to skip
its new transfer. The actual recovery entrypoint returned success while the
promoted directory omitted the file. A filesystem regression now checks both
filenames and the retransferred contents; transport and source listing are
controlled fixtures, not a real remote server.
- A superseded invocation waited forever after a new same-ID owner changed file
identity, including when the new owner completed and its row was compacted.
Both bounded regressions returned `still-waiting` before the fix.
Recovery now authorizes reuse of a completed row for a fresh transfer only when
it is the exact row captured at the destination/checkpoint reset, after stage
deletion has finished. This includes files completed during session setup or
traversal. Newer
completion, pause and cancellation still win. Reset byte checkpoints and source
fingerprints are published with that admission. Ordinary recovery continues to
reuse valid completed work.
Settlement observations retain identity-conflict evidence before compaction,
including explicit admission changes. Before each admission attempt, an observed identity conflict stops the old
invocation from reclaiming an active row after a pause. A displaced invocation fails explicitly
instead of waiting for completion evidence belonging to another file. Both live
and dedicated walks report that failure without overwriting the new owner's
child row; dedicated recovery discards its deferred update synchronously when
the observation detects displacement. This also prevents an automatic 512-entry
batch flush from publishing the old snapshot while its invoke reply is delayed.
Exact completion evidence is retained independently of later ownership changes:
an old success still cannot overwrite a newer owner. Initial retained identities
are distinguished from changes observed while waiting, and newer same-file
attempts supersede old failed evidence. Missing rows alone still do not prove
success. Observations remain scoped to their waiter or its pending child update and are disposed after the final write or discard.
These cases explain specific large-history recovery failures. They do not prove
that the original Windows v1.1.82 freeze was caused by either case; its file count,
transfer direction, server and logs remain unavailable.
## Ownership rules for the follow-up
The expected identity is the tuple of task ID, source path, target path, parent
ID, manifest index and manifest identity. Successful completion evidence and
permission to update the current row are separate: an old completion never grants
permission to overwrite a different identity. Identity conflict is sticky for a
waiter, immediately discards its deferred child update, and ends the old walk
without writing that child or reporting the walk as successful.
Only two initial-state allowances exist, both bounded to one waiter:
1. An initial **inactive** row (`paused`, `failed`, `interrupted`, `attention`)
can precede a fresh recovery plan. It is tolerated only while identity,
status, lifecycle epoch and owner still match that initial snapshot. It
does not authorize dispatch: the ordinary pause/cancel/admission gates remain.
2. An exact completed row captured **after** destination/checkpoint reset is
obsolete completion evidence for the deleted destination. Only that row
object can be restarted. A newer completion cannot inherit this permission.
| Boundary | Same expected identity | Different identity | Missing row |
| --- | --- | --- | --- |
| Registration | Capture current state immediately; completion may be reused unless it is the exact reset-invalidated row | Tolerate only the unchanged initial inactive snapshot or exact reset-invalidated completion; already-active differences conflict immediately | No evidence of completion; continue to normal admission |
| Waiting for admission | A new active/paused state replaces old failed/cancelled evidence; a later success is retained. Pause/cancel gates still apply | Status, epoch or owner changes end the initial inactive allowance; any non-allowed mismatch records conflict and discards pending updates immediately | Retained exact success can settle the wait; absence alone cannot |
| Before every admission | Check cancellation and recorded conflict first. Reuse exact completion, or restart only the exact reset-invalidated row. An unchanged captured pause requires an active parent and released pause latches | Recorded conflict stops before `admitTaskRun` can overwrite the row; structural mismatch is also rejected by admission | Dispatch only if normal admission allows it and no retained completion/conflict settles the waiter |
| Invocation resolves | Check conflict before interpreting success or superseded. A superseded call follows observed same-file state | Fail with an ownership-change error; do not write old success/failure into the new row | A superseded call needs retained exact settlement evidence; ordinary success retains existing invocation semantics |
| Invocation rejects | Without conflict, preserve the original transport error | Convert to ownership-change error so caller does not overwrite the winner | Do not infer success; preserve original error unless conflict was already observed |
| Exit | Live waiters dispose on every path; dedicated child handlers keep the same observation until their final update is synchronously applied or handed to the batcher | Live/dedicated walks count failure, skip old child writes; dedicated recovery also discards deferred updates | Never recreate a compacted winning row from an obsolete invocation |
For the same expected file identity, an epoch change alone is not an identity
conflict: the bridge still owns invocation deduplication and lifecycle ordering.
Epoch/status/owner changes matter when deciding whether a *different* initial
inactive identity is still the unchanged pre-recovery record. This follow-up does
not replace the existing same-file control/epoch machinery or introduce a new
persistent attempt protocol.
Evidence maps directly to these rules: `transferSettlementObservation.test.ts`
covers registration, waiting, pre-admission, initial inactive/active states,
old failure versus new success, compacted completion, newer epochs and original
error propagation. `dedicatedTransferResume.test.ts` exercises actual recovery
with immediate/batched updates, automatic flush before delayed replies, new
active/completed owners, old success/rejection, and real filesystem stage reset
including completion during setup. `transferDirectoryOps.discovery.test.tsx`
checks the actual live hook does not overwrite replacement owners. Existing
pause/cancel, directory/store and batching tests remain in the targeted run.
Observation updates follow state writes, not UI notification type. Explicit
admission, direct row patches and background event ingestion synchronously
capture the affected row, including progress-only epoch/status/owner changes.
Lifecycle publication still captures settlement before history compaction.
The background path reuses its existing row index; progress does not gain a
full-history scan, serialization or synchronous persistence.
Deferred child writes retain the original observation, without another store
lookup or observer registration. Dedicated recovery owns it while handling the
invocation result; the update callback returns `true` only when its consumer
accepts responsibility for disposal. The batcher stores the update and observation
together, checks conflict immediately before writing, and disposes on write or
discard. A callback that does not accept ownership leaves disposal to recovery's
`finally`. This also covers a new identity appearing after one file finishes
while other directory files are still running. Evidence includes both old
success/failure and active/compacted replacement rows after waiter exit.
The write/discard is the ownership deadline for a deferred update, not the end
of the entire directory. A completed child already written successfully may
legitimately be replaced later; that does not retroactively fail the old success.
A completed child still waiting to be written is provisional: if replaced before
its write, its pending completion count is withdrawn and the old directory fails.
Observation starts before source validation and is passed into the waiter, so
validation errors and attention outcomes obey the same write-ownership rule.
Recovery flushes child updates before deciding the directory result or promoting
a replacement stage. Its later asynchronous connection cleanup cannot outlive
provisional success. The outer cleanup retains an idempotent flush as an error
fallback. A regression replaces a child during connection close and verifies its
old completion was already committed and the legitimate new owner survives.