fix(extraction): isolate explicit jobs and run cancellation
Use validated explicit archive paths without recursively discovering a shared package root. Route selective post-processing stops through the coordinator run owner so another generation of the same package remains active, while preserving unfiltered package cancellation behavior.
This commit is contained in:
@@ -7178,9 +7178,6 @@ export class DownloadManager extends EventEmitter {
|
|||||||
this.speedBytesPerPackage.clear();
|
this.speedBytesPerPackage.clear();
|
||||||
this.speedEventsHead = 0;
|
this.speedEventsHead = 0;
|
||||||
this.abortPostProcessing("stop", stoppedRunContext?.id);
|
this.abortPostProcessing("stop", stoppedRunContext?.id);
|
||||||
if (stoppedRunContext) {
|
|
||||||
void this.extractionCoordinator.cancelRun(stoppedRunContext.id, "stop");
|
|
||||||
}
|
|
||||||
for (const active of this.activeTasks.values()) {
|
for (const active of this.activeTasks.values()) {
|
||||||
active.abortReason = abortReason;
|
active.abortReason = abortReason;
|
||||||
active.abortController.abort(abortReason);
|
active.abortController.abort(abortReason);
|
||||||
@@ -8761,12 +8758,17 @@ export class DownloadManager extends EventEmitter {
|
|||||||
}
|
}
|
||||||
|
|
||||||
private abortPostProcessing(reason: string, runContextId?: string): void {
|
private abortPostProcessing(reason: string, runContextId?: string): void {
|
||||||
|
if (runContextId !== undefined) {
|
||||||
|
void this.extractionCoordinator.cancelRun(runContextId, reason);
|
||||||
|
}
|
||||||
for (const [packageId, controller] of this.packagePostProcessAbortControllers.entries()) {
|
for (const [packageId, controller] of this.packagePostProcessAbortControllers.entries()) {
|
||||||
const owner = this.packagePostProcessRunOwnerByController.get(controller);
|
const owner = this.packagePostProcessRunOwnerByController.get(controller);
|
||||||
if (runContextId !== undefined && owner !== runContextId) {
|
if (runContextId !== undefined && owner !== runContextId) {
|
||||||
continue;
|
continue;
|
||||||
}
|
}
|
||||||
void this.extractionCoordinator.cancelPackage(packageId, reason);
|
if (runContextId === undefined) {
|
||||||
|
void this.extractionCoordinator.cancelPackage(packageId, reason);
|
||||||
|
}
|
||||||
if (!controller.signal.aborted) {
|
if (!controller.signal.aborted) {
|
||||||
controller.abort(reason);
|
controller.abort(reason);
|
||||||
}
|
}
|
||||||
@@ -8801,7 +8803,9 @@ export class DownloadManager extends EventEmitter {
|
|||||||
if (runContextId !== undefined && owner !== runContextId) {
|
if (runContextId !== undefined && owner !== runContextId) {
|
||||||
continue;
|
continue;
|
||||||
}
|
}
|
||||||
void this.extractionCoordinator.cancelPackage(packageId, reason);
|
if (runContextId === undefined) {
|
||||||
|
void this.extractionCoordinator.cancelPackage(packageId, reason);
|
||||||
|
}
|
||||||
if (!controller.signal.aborted) {
|
if (!controller.signal.aborted) {
|
||||||
controller.abort(reason);
|
controller.abort(reason);
|
||||||
}
|
}
|
||||||
@@ -8812,7 +8816,9 @@ export class DownloadManager extends EventEmitter {
|
|||||||
if (runContextId !== undefined && owner !== runContextId) {
|
if (runContextId !== undefined && owner !== runContextId) {
|
||||||
continue;
|
continue;
|
||||||
}
|
}
|
||||||
void this.extractionCoordinator.cancelPackage(packageId, reason);
|
if (runContextId === undefined) {
|
||||||
|
void this.extractionCoordinator.cancelPackage(packageId, reason);
|
||||||
|
}
|
||||||
if (!controller.signal.aborted) {
|
if (!controller.signal.aborted) {
|
||||||
controller.abort(reason);
|
controller.abort(reason);
|
||||||
}
|
}
|
||||||
|
|||||||
+28
-11
@@ -3656,16 +3656,33 @@ export async function extractPackageArchives(options: ExtractOptions): Promise<E
|
|||||||
const validateOutputTarget = (entryPath: string, outputPath: string): void => {
|
const validateOutputTarget = (entryPath: string, outputPath: string): void => {
|
||||||
outputScope.validateTarget(entryPath, outputPath);
|
outputScope.validateTarget(entryPath, outputPath);
|
||||||
};
|
};
|
||||||
options.onProgress?.({ current: 0, total: 0, percent: 0, archiveName: "Archive scannen...", phase: "preparing" });
|
options.onProgress?.({ current: 0, total: 0, percent: 0, archiveName: "Archive scannen...", phase: "preparing" });
|
||||||
const allCandidates = await findArchiveCandidates(options.packageDir);
|
const candidates: string[] = [];
|
||||||
const candidates = options.onlyArchives
|
if (options.onlyArchives) {
|
||||||
? allCandidates.filter((archivePath) => {
|
const packageRoot = path.resolve(options.packageDir);
|
||||||
const key = process.platform === "win32" ? path.resolve(archivePath).toLowerCase() : path.resolve(archivePath);
|
const inputScope = new PackageOutputScope([packageRoot]);
|
||||||
return options.onlyArchives!.has(key);
|
const seen = new Set<string>();
|
||||||
})
|
for (const archivePath of options.onlyArchives) {
|
||||||
: allCandidates;
|
const candidate = path.resolve(String(archivePath));
|
||||||
logger.info(`Entpacken gestartet: packageDir=${options.packageDir}, targetDir=${options.targetDir}, archives=${candidates.length}${options.onlyArchives ? ` (hybrid, gesamt=${allCandidates.length})` : ""}, cleanupMode=${options.cleanupMode}, conflictMode=${options.conflictMode}`);
|
const entryPath = path.relative(packageRoot, candidate).replace(/\\/g, "/");
|
||||||
options.onLog?.("INFO", `Entpacken gestartet: packageDir=${options.packageDir}, targetDir=${options.targetDir}, archives=${candidates.length}${options.onlyArchives ? ` (hybrid, gesamt=${allCandidates.length})` : ""}, cleanupMode=${options.cleanupMode}, conflictMode=${options.conflictMode}`);
|
inputScope.validateTarget(entryPath, candidate);
|
||||||
|
let stat: fs.Stats;
|
||||||
|
try {
|
||||||
|
stat = await fs.promises.lstat(candidate);
|
||||||
|
} catch {
|
||||||
|
continue;
|
||||||
|
}
|
||||||
|
const key = pathSetKey(candidate);
|
||||||
|
if (stat.isFile() && !stat.isSymbolicLink() && !seen.has(key)) {
|
||||||
|
seen.add(key);
|
||||||
|
candidates.push(candidate);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
} else {
|
||||||
|
candidates.push(...await findArchiveCandidates(options.packageDir));
|
||||||
|
}
|
||||||
|
logger.info(`Entpacken gestartet: packageDir=${options.packageDir}, targetDir=${options.targetDir}, archives=${candidates.length}${options.onlyArchives ? " (explizite Liste)" : ""}, cleanupMode=${options.cleanupMode}, conflictMode=${options.conflictMode}`);
|
||||||
|
options.onLog?.("INFO", `Entpacken gestartet: packageDir=${options.packageDir}, targetDir=${options.targetDir}, archives=${candidates.length}${options.onlyArchives ? " (explizite Liste)" : ""}, cleanupMode=${options.cleanupMode}, conflictMode=${options.conflictMode}`);
|
||||||
|
|
||||||
if (candidates.length > 0) {
|
if (candidates.length > 0) {
|
||||||
options.onProgress?.({ current: 0, total: candidates.length, percent: 0, archiveName: "Speicherplatz prüfen...", phase: "preparing" });
|
options.onProgress?.({ current: 0, total: candidates.length, percent: 0, archiveName: "Speicherplatz prüfen...", phase: "preparing" });
|
||||||
@@ -4303,7 +4320,7 @@ export async function extractPackageArchives(options: ExtractOptions): Promise<E
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
if (failed === 0 && resumeCompleted.size >= allCandidates.length && !options.skipPostCleanup) {
|
if (!options.onlyArchives && failed === 0 && resumeCompleted.size >= candidates.length && !options.skipPostCleanup) {
|
||||||
await clearExtractResumeState(options.packageDir, options.packageId);
|
await clearExtractResumeState(options.packageDir, options.packageId);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
@@ -15491,6 +15491,56 @@ describe("package priority ordering", () => {
|
|||||||
});
|
});
|
||||||
});
|
});
|
||||||
|
|
||||||
|
describe("selective extraction cancellation", () => {
|
||||||
|
it("stops only the selected run when two generations share one package id", async () => {
|
||||||
|
const root = fs.mkdtempSync(path.join(os.tmpdir(), "rd-run-cancel-"));
|
||||||
|
tempDirs.push(root);
|
||||||
|
const manager = new DownloadManager(defaultSettings(), emptySession(), createStoragePaths(path.join(root, "state")));
|
||||||
|
const state = manager as any;
|
||||||
|
const coordinator = new ExtractionCoordinator(2);
|
||||||
|
state.extractionCoordinator = coordinator;
|
||||||
|
const packageId = "shared-package";
|
||||||
|
const operationA = await coordinator.beginOperation({
|
||||||
|
context: { operationId: "operation-a", packageId, generation: 1, runOwnerId: "run-a" },
|
||||||
|
targetPath: path.join(root, "out"),
|
||||||
|
members: []
|
||||||
|
});
|
||||||
|
const operationB = await coordinator.beginOperation({
|
||||||
|
context: { operationId: "operation-b", packageId, generation: 2, runOwnerId: "run-b" },
|
||||||
|
targetPath: path.join(root, "out"),
|
||||||
|
members: []
|
||||||
|
});
|
||||||
|
const started = new Set<string>();
|
||||||
|
let completeRunA = (): void => {};
|
||||||
|
const jobA = coordinator.scheduleArchive(operationA, "archive-a", (signal) => new Promise<void>((resolve, reject) => {
|
||||||
|
started.add("run-a");
|
||||||
|
completeRunA = resolve;
|
||||||
|
signal.addEventListener("abort", () => reject(new Error(String(signal.reason || "aborted"))), { once: true });
|
||||||
|
}));
|
||||||
|
const jobB = coordinator.scheduleArchive(operationB, "archive-b", (signal) => new Promise<void>((_resolve, reject) => {
|
||||||
|
started.add("run-b");
|
||||||
|
signal.addEventListener("abort", () => reject(new Error(String(signal.reason || "aborted"))), { once: true });
|
||||||
|
}));
|
||||||
|
const outcomeA = jobA.then(() => "completed", (error) => `cancelled:${String(error)}`);
|
||||||
|
const outcomeB = jobB.then(() => "completed", (error) => `cancelled:${String(error)}`);
|
||||||
|
await vi.waitFor(() => expect(started).toEqual(new Set(["run-a", "run-b"])));
|
||||||
|
const controllerA = new AbortController();
|
||||||
|
const controllerB = new AbortController();
|
||||||
|
state.packageHybridPostProcessControllers.set(packageId, new Set([controllerA, controllerB]));
|
||||||
|
state.packageHybridRunOwnerByController.set(controllerA, "run-a");
|
||||||
|
state.packageHybridRunOwnerByController.set(controllerB, "run-b");
|
||||||
|
|
||||||
|
state.abortPostProcessing("stop", "run-b");
|
||||||
|
|
||||||
|
await vi.waitFor(() => expect(controllerB.signal.aborted).toBe(true));
|
||||||
|
expect(controllerA.signal.aborted).toBe(false);
|
||||||
|
expect(await outcomeB).toContain("stop");
|
||||||
|
completeRunA();
|
||||||
|
expect(await outcomeA).toBe("completed");
|
||||||
|
await Promise.all([operationA.finalize(), operationB.finalize()]);
|
||||||
|
});
|
||||||
|
});
|
||||||
|
|
||||||
describe("package lifecycle telemetry boundaries", () => {
|
describe("package lifecycle telemetry boundaries", () => {
|
||||||
it("captures direct package output scopes concurrently without scanning a shared root", async () => {
|
it("captures direct package output scopes concurrently without scanning a shared root", async () => {
|
||||||
const root = fs.mkdtempSync(path.join(os.tmpdir(), "rd-output-provenance-lock-"));
|
const root = fs.mkdtempSync(path.join(os.tmpdir(), "rd-output-provenance-lock-"));
|
||||||
|
|||||||
+44
-1
@@ -3,7 +3,7 @@ import { createRequire } from "node:module";
|
|||||||
import os from "node:os";
|
import os from "node:os";
|
||||||
import path from "node:path";
|
import path from "node:path";
|
||||||
import AdmZip from "adm-zip";
|
import AdmZip from "adm-zip";
|
||||||
import { afterEach, beforeEach, describe, expect, it } from "vitest";
|
import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
|
||||||
import {
|
import {
|
||||||
buildExternalExtractArgs,
|
buildExternalExtractArgs,
|
||||||
buildExternalListArgs,
|
buildExternalListArgs,
|
||||||
@@ -1427,6 +1427,49 @@ describe("extractor", () => {
|
|||||||
expect(fs.existsSync(path.join(targetDir, "foreign.txt"))).toBe(false);
|
expect(fs.existsSync(path.join(targetDir, "foreign.txt"))).toBe(false);
|
||||||
});
|
});
|
||||||
|
|
||||||
|
it("uses an explicit archive list without traversing a shared package root", async () => {
|
||||||
|
const root = fs.mkdtempSync(path.join(os.tmpdir(), "rd-explicit-archives-"));
|
||||||
|
tempDirs.push(root);
|
||||||
|
const packageDir = path.join(root, "shared");
|
||||||
|
const foreignDir = path.join(packageDir, "foreign-package");
|
||||||
|
const targetDir = path.join(root, "out");
|
||||||
|
fs.mkdirSync(foreignDir, { recursive: true });
|
||||||
|
const ownedArchive = path.join(packageDir, "owned.zip");
|
||||||
|
const foreignArchive = path.join(foreignDir, "foreign.zip");
|
||||||
|
const owned = new AdmZip();
|
||||||
|
owned.addFile("owned.txt", Buffer.from("owned"));
|
||||||
|
owned.writeZip(ownedArchive);
|
||||||
|
const foreign = new AdmZip();
|
||||||
|
foreign.addFile("foreign.txt", Buffer.from("foreign"));
|
||||||
|
foreign.writeZip(foreignArchive);
|
||||||
|
const readdirSpy = vi.spyOn(fs.promises, "readdir");
|
||||||
|
let sharedRootTraversals = 0;
|
||||||
|
|
||||||
|
try {
|
||||||
|
const result = await extractPackageArchives({
|
||||||
|
packageDir,
|
||||||
|
targetDir,
|
||||||
|
cleanupMode: "none",
|
||||||
|
conflictMode: "overwrite",
|
||||||
|
removeLinks: false,
|
||||||
|
removeSamples: false,
|
||||||
|
onlyArchives: new Set([path.resolve(ownedArchive).toLowerCase()])
|
||||||
|
});
|
||||||
|
const sharedRoot = path.resolve(packageDir);
|
||||||
|
sharedRootTraversals = readdirSpy.mock.calls.filter(([directory]) => {
|
||||||
|
const candidate = path.resolve(String(directory));
|
||||||
|
return candidate === sharedRoot || candidate.startsWith(`${sharedRoot}${path.sep}`);
|
||||||
|
}).length;
|
||||||
|
|
||||||
|
expect(result.extracted).toBe(1);
|
||||||
|
expect(fs.readFileSync(path.join(targetDir, "owned.txt"), "utf8")).toBe("owned");
|
||||||
|
expect(fs.existsSync(path.join(targetDir, "foreign.txt"))).toBe(false);
|
||||||
|
expect(sharedRootTraversals).toBe(0);
|
||||||
|
} finally {
|
||||||
|
readdirSpy.mockRestore();
|
||||||
|
}
|
||||||
|
});
|
||||||
|
|
||||||
it("delegates top-level and nested archive jobs through one scheduler", async () => {
|
it("delegates top-level and nested archive jobs through one scheduler", async () => {
|
||||||
const root = fs.mkdtempSync(path.join(os.tmpdir(), "rd-global-extract-scheduler-"));
|
const root = fs.mkdtempSync(path.join(os.tmpdir(), "rd-global-extract-scheduler-"));
|
||||||
tempDirs.push(root);
|
tempDirs.push(root);
|
||||||
|
|||||||
Reference in New Issue
Block a user