fix(extraction): harden daemon termination and drain deadlines

Prevent aborted daemon waiters and pre-aborted requests from spawning one-shot JVM children. Ignore buffered request completion after abort or timeout so permits and disk leases remain owned until real daemon close. Keep incomplete shutdown operations and leases alive past the shared deadline while returning explicit drain status, and make multipart reservations fail closed whenever any deduplicated member size is unknown. Add real daemon boot, busy, abort, timeout, DONE race, deadline, and reservation regression coverage.
This commit is contained in:
Sucukdeluxe
2026-08-22 17:10:01 +02:00
parent 648e7fed3a
commit 3e18ec57ea
5 changed files with 372 additions and 60 deletions
+3 -3
View File
@@ -83,7 +83,7 @@ import { mergeKnownTotalBytes } from "./download-size";
import { DiskCapacityError, DiskReservationCoordinator, type DiskReservationLease } from "./disk-space"; import { DiskCapacityError, DiskReservationCoordinator, type DiskReservationLease } from "./disk-space";
import { createRendererState } from "./renderer-state"; import { createRendererState } from "./renderer-state";
import { PackageOutputScope } from "./package-output-scope"; import { PackageOutputScope } from "./package-output-scope";
import { ExtractionCoordinator, type ExtractionArchiveMember } from "./extraction-coordinator"; import { ExtractionCoordinator, type ExtractionArchiveMember, type ExtractionShutdownResult } from "./extraction-coordinator";
import { import {
RollingAccountStatisticsAccumulator, RollingAccountStatisticsAccumulator,
addStatisticsActiveIntervalInPlace, addStatisticsActiveIntervalInPlace,
@@ -7300,8 +7300,8 @@ export class DownloadManager extends EventEmitter {
logger.info(`Shutdown-Vorbereitung beendet: requeued=${requeuedItems}`); logger.info(`Shutdown-Vorbereitung beendet: requeued=${requeuedItems}`);
} }
public async shutdownAndDrain(deadlineAt: number): Promise<void> { public async shutdownAndDrain(deadlineAt: number): Promise<ExtractionShutdownResult> {
await this.extractionCoordinator.shutdownAndDrain(deadlineAt); return this.extractionCoordinator.shutdownAndDrain(deadlineAt);
} }
public persistForShutdown(): void { public persistForShutdown(): void {
+23 -8
View File
@@ -35,6 +35,14 @@ export type ExtractionOperation = {
finalize(finalizeScope?: () => void | Promise<void>): Promise<void>; finalize(finalizeScope?: () => void | Promise<void>): Promise<void>;
}; };
export type ExtractionShutdownResult = Readonly<{
completed: boolean;
timedOut: boolean;
activeJobs: number;
pendingOperations: number;
queuedJobs: number;
}>;
type ArchiveJob = { type ArchiveJob = {
archiveId: string; archiveId: string;
execute: (signal: AbortSignal) => Promise<unknown>; execute: (signal: AbortSignal) => Promise<unknown>;
@@ -97,10 +105,10 @@ function deduplicateMembers(members: readonly ExtractionArchiveMember[]): Extrac
} }
function reservationBytes(members: readonly ExtractionArchiveMember[]): number | null { function reservationBytes(members: readonly ExtractionArchiveMember[]): number | null {
const known = members if (members.length === 0 || members.some((member) => member.size === null)) {
.map((member) => member.size) return null;
.filter((size): size is number => typeof size === "number" && Number.isFinite(size)); }
return known.length > 0 ? known.reduce((total, size) => total + Math.max(0, Math.floor(size)), 0) : null; return members.reduce((total, member) => total + Math.max(0, Math.floor(member.size as number)), 0);
} }
export class ExtractionCancelledError extends Error { export class ExtractionCancelledError extends Error {
@@ -216,7 +224,7 @@ export class ExtractionCoordinator {
await this.cancelMatching((state) => state.context.packageId === packageId, reason); await this.cancelMatching((state) => state.context.packageId === packageId, reason);
} }
public async shutdownAndDrain(deadlineAt: number): Promise<void> { public async shutdownAndDrain(deadlineAt: number): Promise<ExtractionShutdownResult> {
this.closed = true; this.closed = true;
const states = [...this.operations.values()]; const states = [...this.operations.values()];
for (const state of states) { for (const state of states) {
@@ -234,9 +242,16 @@ export class ExtractionCoordinator {
} }
await this.waitUntilDeadline(Promise.all(states.map((state) => state.drain.promise)), deadlineAt); await this.waitUntilDeadline(Promise.all(states.map((state) => state.drain.promise)), deadlineAt);
await this.waitForFinalization(states, deadlineAt); await this.waitForFinalization(states, deadlineAt);
for (const state of states) { const pendingOperations = states.filter((state) => this.operations.get(state.context.operationId) === state).length;
this.releaseLease(state); const queuedJobs = states.reduce((total, state) => total + state.queued.length, 0);
} const completed = this.activeCount === 0 && pendingOperations === 0 && queuedJobs === 0;
return Object.freeze({
completed,
timedOut: !completed && Date.now() >= deadlineAt,
activeJobs: this.activeCount,
pendingOperations,
queuedJobs
});
} }
private async finalizeOperation(state: OperationState, finalizeScope?: () => void | Promise<void>): Promise<void> { private async finalizeOperation(state: OperationState, finalizeScope?: () => void | Promise<void>): Promise<void> {
+91 -47
View File
@@ -1815,14 +1815,14 @@ function handleDaemonLine(line: string): void {
return; return;
} }
if (trimmed.startsWith("RD_REQUEST_DONE ")) { if (trimmed.startsWith("RD_REQUEST_DONE ")) {
const code = parseInt(trimmed.slice("RD_REQUEST_DONE ".length).trim(), 10); const code = parseInt(trimmed.slice("RD_REQUEST_DONE ".length).trim(), 10);
const req = daemonCurrentRequest; const req = daemonCurrentRequest;
if (!req) return; if (!req || req.terminationStarted) return;
const finalize = (): void => { const finalize = (): void => {
if (daemonCurrentRequest !== req) { if (daemonCurrentRequest !== req || req.terminationStarted) {
return; return;
} }
flushDaemonParseBuffers(req); flushDaemonParseBuffers(req);
if (req.parseState.outputError) { if (req.parseState.outputError) {
failDaemonOutputCallback(req); failDaemonOutputCallback(req);
@@ -1978,26 +1978,59 @@ function startDaemon(layout: JvmExtractorLayout): boolean {
} }
} }
function isDaemonAvailable(layout: JvmExtractorLayout): boolean { function isDaemonAvailable(layout: JvmExtractorLayout): boolean {
if (!daemonProcess || !daemonReady) { if (!daemonProcess || !daemonReady) {
startDaemon(layout); startDaemon(layout);
} }
return Boolean(daemonProcess && daemonReady && !daemonBusy); return Boolean(daemonProcess && daemonReady && !daemonBusy);
} }
function waitForDaemonReady(maxWaitMs: number, signal?: AbortSignal): Promise<boolean> { function abortedJvmExtractResult(): JvmExtractResult {
return new Promise((resolve) => { return {
const start = Date.now(); ok: false,
const check = () => { missingCommand: false,
if (signal?.aborted) { resolve(false); return; } missingRuntime: false,
if (daemonProcess && daemonReady && !daemonBusy) { resolve(true); return; } aborted: true,
if (!daemonProcess) { resolve(false); return; } timedOut: false,
if (Date.now() - start >= maxWaitMs) { resolve(false); return; } errorText: "aborted:extract",
setTimeout(check, 50); usedPassword: "",
}; backend: ""
check(); };
}); }
}
function waitForDaemonReady(maxWaitMs: number, signal?: AbortSignal): Promise<boolean> {
return new Promise((resolve) => {
const start = Date.now();
let settled = false;
let timer: NodeJS.Timeout | null = null;
const finish = (ready: boolean): void => {
if (settled) {
return;
}
settled = true;
if (timer) {
clearTimeout(timer);
timer = null;
}
signal?.removeEventListener("abort", onAbort);
resolve(ready);
};
const onAbort = (): void => finish(false);
const check = () => {
if (signal?.aborted) { finish(false); return; }
if (daemonProcess && daemonReady && !daemonBusy) { finish(true); return; }
if (!daemonProcess) { finish(false); return; }
if (Date.now() - start >= maxWaitMs) { finish(false); return; }
timer = setTimeout(check, 50);
};
if (signal?.aborted) {
finish(false);
return;
}
signal?.addEventListener("abort", onAbort, { once: true });
check();
});
}
function sendDaemonRequest( function sendDaemonRequest(
archivePath: string, archivePath: string,
@@ -2008,8 +2041,11 @@ function sendDaemonRequest(
signal?: AbortSignal, signal?: AbortSignal,
timeoutMs?: number, timeoutMs?: number,
onOutput?: (event: ExtractOutputEvent) => void onOutput?: (event: ExtractOutputEvent) => void
): Promise<JvmExtractResult> { ): Promise<JvmExtractResult> {
return new Promise((resolve) => { if (signal?.aborted) {
return Promise.resolve(abortedJvmExtractResult());
}
return new Promise((resolve) => {
const mode = effectiveConflictMode(conflictMode); const mode = effectiveConflictMode(conflictMode);
const parseState = { bestPercent: 0, usedPassword: "", backend: "", reportedError: "" }; const parseState = { bestPercent: 0, usedPassword: "", backend: "", reportedError: "" };
const archiveName = path.basename(archivePath); const archiveName = path.basename(archivePath);
@@ -2094,13 +2130,9 @@ async function runJvmExtractCommand(
signal?: AbortSignal, signal?: AbortSignal,
timeoutMs?: number, timeoutMs?: number,
onOutput?: (event: ExtractOutputEvent) => void onOutput?: (event: ExtractOutputEvent) => void
): Promise<JvmExtractResult> { ): Promise<JvmExtractResult> {
if (signal?.aborted) { if (signal?.aborted) {
return Promise.resolve({ return Promise.resolve(abortedJvmExtractResult());
ok: false, missingCommand: false, missingRuntime: false,
aborted: true, timedOut: false, errorText: "aborted:extract",
usedPassword: "", backend: ""
});
} }
if (isDaemonAvailable(layout)) { if (isDaemonAvailable(layout)) {
@@ -2112,18 +2144,25 @@ async function runJvmExtractCommand(
if (daemonProcess) { if (daemonProcess) {
const reason = !daemonReady ? "booting" : "busy"; const reason = !daemonReady ? "booting" : "busy";
const waitStartedAt = Date.now(); const waitStartedAt = Date.now();
logger.info(`JVM Daemon: Warte auf ${reason} Daemon für ${path.basename(archivePath)}...`); logger.info(`JVM Daemon: Warte auf ${reason} Daemon für ${path.basename(archivePath)}...`);
const ready = await waitForDaemonReady(15_000, signal); const ready = await waitForDaemonReady(15_000, signal);
const waitedMs = Date.now() - waitStartedAt; const waitedMs = Date.now() - waitStartedAt;
if (ready) { if (signal?.aborted) {
return abortedJvmExtractResult();
}
if (ready) {
lowerExtractProcessPriority(daemonProcess?.pid, currentExtractCpuPriority); lowerExtractProcessPriority(daemonProcess?.pid, currentExtractCpuPriority);
logger.info(`JVM Daemon: Bereit nach ${waitedMs}ms — sende Request für ${path.basename(archivePath)}`); logger.info(`JVM Daemon: Bereit nach ${waitedMs}ms — sende Request für ${path.basename(archivePath)}`);
return sendDaemonRequest(archivePath, targetDir, conflictMode, passwordCandidates, onArchiveProgress, signal, timeoutMs, onOutput); return sendDaemonRequest(archivePath, targetDir, conflictMode, passwordCandidates, onArchiveProgress, signal, timeoutMs, onOutput);
} }
logger.warn(`JVM Daemon: Timeout nach ${waitedMs}ms beim Warten — Fallback auf neuen Prozess für ${path.basename(archivePath)}`); logger.warn(`JVM Daemon: Timeout nach ${waitedMs}ms beim Warten — Fallback auf neuen Prozess für ${path.basename(archivePath)}`);
} }
logger.info(`JVM Spawn: Neuer Prozess für ${path.basename(archivePath)}`); if (signal?.aborted) {
return abortedJvmExtractResult();
}
logger.info(`JVM Spawn: Neuer Prozess für ${path.basename(archivePath)}`);
const mode = effectiveConflictMode(conflictMode); const mode = effectiveConflictMode(conflictMode);
const jvmTmpDir = path.join(os.tmpdir(), `rd-extract-${crypto.randomUUID()}`); const jvmTmpDir = path.join(os.tmpdir(), `rd-extract-${crypto.randomUUID()}`);
@@ -2147,11 +2186,16 @@ async function runJvmExtractCommand(
"--backend", "--backend",
"auto" "auto"
]; ];
for (const password of passwordCandidates) { for (const password of passwordCandidates) {
args.push("--password", password); args.push("--password", password);
} }
return new Promise((resolve) => { if (signal?.aborted) {
fs.rm(jvmTmpDir, { recursive: true, force: true }, () => {});
return abortedJvmExtractResult();
}
return new Promise((resolve) => {
let settled = false; let settled = false;
let output = ""; let output = "";
let timeoutId: NodeJS.Timeout | null = null; let timeoutId: NodeJS.Timeout | null = null;
+72
View File
@@ -296,6 +296,48 @@ describe("ExtractionCoordinator", () => {
await operation.finalize(); await operation.finalize();
}); });
it("keeps the reservation unknown when any deduplicated multipart member has unknown size", async () => {
const coordinator = new ExtractionCoordinator(1);
const heldLease = lease();
const requiredBytes: Array<number | null> = [];
const operation = await coordinator.beginOperation({
context: context("multipart-partial-unknown", "package-a", "run"),
targetPath: "C:\\target",
members: [
{ path: "C:\\archives\\show.part1.rar", size: 100 },
{ path: "C:\\archives\\show.part2.rar", size: null }
],
acquireLease: async (request) => {
requiredBytes.push(request.requiredBytes);
return heldLease;
}
});
expect(requiredBytes).toEqual([null]);
await operation.finalize();
});
it("keeps the reservation unknown when all multipart member sizes are unknown", async () => {
const coordinator = new ExtractionCoordinator(1);
const heldLease = lease();
const requiredBytes: Array<number | null> = [];
const operation = await coordinator.beginOperation({
context: context("multipart-all-unknown", "package-a", "run"),
targetPath: "C:\\target",
members: [
{ path: "C:\\archives\\show.part1.rar", size: null },
{ path: "C:\\archives\\show.part2.rar", size: null }
],
acquireLease: async (request) => {
requiredBytes.push(request.requiredBytes);
return heldLease;
}
});
expect(requiredBytes).toEqual([null]);
await operation.finalize();
});
it("shuts down in queue-close, waiter-cancel, active-abort, child-drain, finalization and lease-release order", async () => { it("shuts down in queue-close, waiter-cancel, active-abort, child-drain, finalization and lease-release order", async () => {
const coordinator = new ExtractionCoordinator(1); const coordinator = new ExtractionCoordinator(1);
const events: string[] = []; const events: string[] = [];
@@ -362,4 +404,34 @@ describe("ExtractionCoordinator", () => {
await Promise.all([finalization, shutdown]); await Promise.all([finalization, shutdown]);
expect(heldLease.release).toHaveBeenCalledTimes(1); expect(heldLease.release).toHaveBeenCalledTimes(1);
}); });
it("returns incomplete at the deadline without releasing an active job lease or reopening the queue", async () => {
const coordinator = new ExtractionCoordinator(1);
const heldLease = lease();
const operation = await coordinator.beginOperation({
context: context("deadline-active", "package-a", "run"),
targetPath: "C:\\target",
members: [{ path: "C:\\archives\\one.rar", size: 100 }],
acquireLease: async () => heldLease
});
const childClose = deferred<void>();
const job = coordinator.scheduleArchive(operation, "active", async () => childClose.promise);
const finalization = operation.finalize();
await flush();
const result = await coordinator.shutdownAndDrain(Date.now() + 10);
expect(result).toEqual(expect.objectContaining({
completed: false,
timedOut: true,
activeJobs: 1,
pendingOperations: 1
}));
expect(heldLease.release).not.toHaveBeenCalled();
await expect(coordinator.scheduleArchive(operation, "late", async () => undefined)).rejects.toBeInstanceOf(ExtractionCancelledError);
childClose.resolve();
await Promise.all([job, finalization]);
expect(heldLease.release).toHaveBeenCalledTimes(1);
});
}); });
+183 -2
View File
@@ -38,9 +38,19 @@ const childProcesses = vi.hoisted(() => {
return { return {
nextPid: 10_000, nextPid: 10_000,
activeExtraction: null as FakeChild | null, activeExtraction: null as FakeChild | null,
daemonChild: null as FakeChild | null,
oneShotJvmChildren: [] as FakeChild[],
autoDaemonReady: true,
spawn: vi.fn((_command: string, args: string[]) => { spawn: vi.fn((_command: string, args: string[]) => {
const child = new FakeChild(childProcesses.nextPid++); const child = new FakeChild(childProcesses.nextPid++);
if (args[0] === "?") { if (args.includes("--daemon")) {
childProcesses.daemonChild = child;
if (childProcesses.autoDaemonReady) {
queueMicrotask(() => child.stdout.emit("data", "RD_DAEMON_READY\n"));
}
} else if (args.includes("--archive")) {
childProcesses.oneShotJvmChildren.push(child);
} else if (args[0] === "?") {
queueMicrotask(() => child.emit("close", 0)); queueMicrotask(() => child.emit("close", 0));
} else if (args[0] === "l") { } else if (args[0] === "l") {
queueMicrotask(() => { queueMicrotask(() => {
@@ -63,18 +73,60 @@ vi.mock("node:child_process", () => ({
spawnSync: childProcesses.spawnSync spawnSync: childProcesses.spawnSync
})); }));
import { extractPackageArchives } from "../src/main/extractor"; import { extractPackageArchives, shutdownDaemon } from "../src/main/extractor";
import { ExtractionCoordinator } from "../src/main/extraction-coordinator";
const tempDirs: string[] = []; const tempDirs: string[] = [];
const originalBackend = process.env.RD_EXTRACT_BACKEND; const originalBackend = process.env.RD_EXTRACT_BACKEND;
const originalSevenZip = process.env.RD_7Z_BIN; const originalSevenZip = process.env.RD_7Z_BIN;
const originalJava = process.env.RD_JAVA_BIN;
const originalJvmRoot = process.env.RD_EXTRACTOR_JVM_DIR;
function createJvmFixture(prefix: string) {
const root = fs.mkdtempSync(path.join(os.tmpdir(), prefix));
tempDirs.push(root);
const packageDir = path.join(root, "package");
const targetDir = path.join(root, "target");
const javaPath = path.join(root, "java.exe");
const jvmRoot = path.join(root, "extractor-jvm");
const classesDir = path.join(jvmRoot, "classes");
const libDir = path.join(jvmRoot, "lib");
fs.mkdirSync(packageDir, { recursive: true });
fs.mkdirSync(classesDir, { recursive: true });
fs.mkdirSync(libDir, { recursive: true });
fs.writeFileSync(javaPath, "fake");
for (const name of ["sevenzipjbinding.jar", "sevenzipjbinding-all-platforms.jar", "zip4j.jar"]) {
fs.writeFileSync(path.join(libDir, name), "fake");
}
fs.writeFileSync(path.join(packageDir, "release.7z"), Buffer.from("377abcaf271c", "hex"));
process.env.RD_EXTRACT_BACKEND = "jvm";
process.env.RD_JAVA_BIN = javaPath;
process.env.RD_EXTRACTOR_JVM_DIR = jvmRoot;
return { packageDir, targetDir };
}
function jvmExtractionOptions(fixture: ReturnType<typeof createJvmFixture>, signal?: AbortSignal) {
return {
...fixture,
cleanupMode: "none" as const,
conflictMode: "overwrite" as const,
removeLinks: false,
removeSamples: false,
passwordList: "",
signal
};
}
afterEach(() => { afterEach(() => {
shutdownDaemon();
vi.useRealTimers(); vi.useRealTimers();
for (const directory of tempDirs.splice(0)) { for (const directory of tempDirs.splice(0)) {
fs.rmSync(directory, { recursive: true, force: true }); fs.rmSync(directory, { recursive: true, force: true });
} }
childProcesses.activeExtraction = null; childProcesses.activeExtraction = null;
childProcesses.daemonChild = null;
childProcesses.oneShotJvmChildren = [];
childProcesses.autoDaemonReady = true;
childProcesses.spawn.mockClear(); childProcesses.spawn.mockClear();
if (originalBackend === undefined) { if (originalBackend === undefined) {
delete process.env.RD_EXTRACT_BACKEND; delete process.env.RD_EXTRACT_BACKEND;
@@ -86,6 +138,16 @@ afterEach(() => {
} else { } else {
process.env.RD_7Z_BIN = originalSevenZip; process.env.RD_7Z_BIN = originalSevenZip;
} }
if (originalJava === undefined) {
delete process.env.RD_JAVA_BIN;
} else {
process.env.RD_JAVA_BIN = originalJava;
}
if (originalJvmRoot === undefined) {
delete process.env.RD_EXTRACTOR_JVM_DIR;
} else {
process.env.RD_EXTRACTOR_JVM_DIR = originalJvmRoot;
}
}); });
describe("extractor child close lifecycle", () => { describe("extractor child close lifecycle", () => {
@@ -167,4 +229,123 @@ describe("extractor child close lifecycle", () => {
expect(results[0]).toEqual(expect.objectContaining({ failed: 1 })); expect(results[0]).toEqual(expect.objectContaining({ failed: 1 }));
expect(results[0]?.lastError).toContain("Timeout"); expect(results[0]?.lastError).toContain("Timeout");
}); });
it("does not spawn a JVM one-shot after abort while the daemon is booting", async () => {
const fixture = createJvmFixture("rd-jvm-boot-abort-");
childProcesses.autoDaemonReady = false;
const controller = new AbortController();
const extraction = extractPackageArchives(jvmExtractionOptions(fixture, controller.signal)).catch(() => undefined);
await vi.waitFor(() => expect(childProcesses.daemonChild).not.toBeNull());
controller.abort("boot-abort");
await new Promise<void>((resolve) => setTimeout(resolve, 80));
const oneShotCount = childProcesses.oneShotJvmChildren.length;
for (const child of childProcesses.oneShotJvmChildren) {
child.emit("close", 1);
}
childProcesses.daemonChild?.emit("close", 1);
await extraction;
expect(oneShotCount).toBe(0);
});
it("does not spawn a JVM one-shot after abort while the daemon is busy", async () => {
const fixture = createJvmFixture("rd-jvm-busy-abort-");
const firstController = new AbortController();
const first = extractPackageArchives(jvmExtractionOptions(fixture, firstController.signal)).catch(() => undefined);
await vi.waitFor(() => expect(childProcesses.daemonChild?.stdin.write).toHaveBeenCalledTimes(1));
const secondController = new AbortController();
const second = extractPackageArchives(jvmExtractionOptions(fixture, secondController.signal)).catch(() => undefined);
await new Promise<void>((resolve) => setTimeout(resolve, 20));
secondController.abort("busy-abort");
await new Promise<void>((resolve) => setTimeout(resolve, 80));
const oneShotCount = childProcesses.oneShotJvmChildren.length;
firstController.abort("cleanup");
for (const child of childProcesses.oneShotJvmChildren) {
child.emit("close", 1);
}
childProcesses.daemonChild?.emit("close", 1);
await Promise.all([first, second]);
expect(oneShotCount).toBe(0);
});
it("keeps daemon abort permits and leases until close even when DONE arrives after termination", async () => {
const fixture = createJvmFixture("rd-jvm-abort-done-");
const coordinator = new ExtractionCoordinator(1);
const lease = { release: vi.fn() };
const operation = await coordinator.beginOperation({
context: { operationId: "abort-operation", packageId: "package-a", generation: 1, runOwnerId: "run" },
targetPath: fixture.targetDir,
members: [{ path: path.join(fixture.packageDir, "release.7z"), size: 6 }],
acquireLease: async () => lease
});
const nextOperation = await coordinator.beginOperation({
context: { operationId: "next-operation", packageId: "package-b", generation: 1, runOwnerId: "run" }
});
const job = coordinator.scheduleArchive(operation, "release.7z", (signal) =>
extractPackageArchives(jvmExtractionOptions(fixture, signal))
);
const finalization = operation.finalize();
let nextStarted = false;
const nextJob = coordinator.scheduleArchive(nextOperation, "next", async () => {
nextStarted = true;
});
const nextFinalization = nextOperation.finalize();
await vi.waitFor(() => expect(childProcesses.daemonChild?.stdin.write).toHaveBeenCalledTimes(1));
childProcesses.daemonChild?.stdout.emit("data", "RD_REQUEST_DONE 0");
const cancellation = coordinator.cancelPackage("package-a", "abort");
childProcesses.daemonChild?.stdout.emit("data", "\n");
await new Promise<void>((resolve) => setTimeout(resolve, 20));
const releaseBeforeClose = lease.release.mock.calls.length;
const nextBeforeClose = nextStarted;
childProcesses.daemonChild?.emit("close", 1);
await Promise.all([job.catch(() => undefined), cancellation, finalization, nextJob, nextFinalization]);
expect(releaseBeforeClose).toBe(0);
expect(nextBeforeClose).toBe(false);
expect(lease.release).toHaveBeenCalledTimes(1);
});
it("keeps daemon timeout permits and leases until close even when DONE arrives after termination", async () => {
vi.useFakeTimers();
const fixture = createJvmFixture("rd-jvm-timeout-done-");
const coordinator = new ExtractionCoordinator(1);
const lease = { release: vi.fn() };
const operation = await coordinator.beginOperation({
context: { operationId: "timeout-operation", packageId: "package-a", generation: 1, runOwnerId: "run" },
targetPath: fixture.targetDir,
members: [{ path: path.join(fixture.packageDir, "release.7z"), size: 6 }],
acquireLease: async () => lease
});
const nextOperation = await coordinator.beginOperation({
context: { operationId: "next-timeout-operation", packageId: "package-b", generation: 1, runOwnerId: "run" }
});
const job = coordinator.scheduleArchive(operation, "release.7z", (signal) =>
extractPackageArchives(jvmExtractionOptions(fixture, signal))
);
const finalization = operation.finalize();
let nextStarted = false;
const nextJob = coordinator.scheduleArchive(nextOperation, "next", async () => {
nextStarted = true;
});
const nextFinalization = nextOperation.finalize();
await vi.waitFor(() => expect(childProcesses.daemonChild?.stdin.write).toHaveBeenCalledTimes(1));
childProcesses.daemonChild?.stdout.emit("data", "RD_REQUEST_DONE 0");
await vi.advanceTimersByTimeAsync(6 * 60 * 1000);
expect(childProcesses.spawn.mock.calls.some(([command]) => command === "taskkill")).toBe(true);
childProcesses.daemonChild?.stdout.emit("data", "\n");
await vi.waitFor(() => expect(lease.release).toHaveBeenCalledTimes(1), { timeout: 500 }).catch(() => undefined);
const releaseBeforeClose = lease.release.mock.calls.length;
const nextBeforeClose = nextStarted;
childProcesses.daemonChild?.emit("close", 1);
await Promise.all([job, finalization, nextJob, nextFinalization]);
expect(releaseBeforeClose).toBe(0);
expect(nextBeforeClose).toBe(false);
expect(lease.release).toHaveBeenCalledTimes(1);
});
}); });