feat(extraction): add global archive coordinator
Add a fair operation-aware queue that owns the global archive child limit across packages and extraction modes. Support live resizing, immutable operation identity, selective run and package cancellation, ghost-waiter cleanup, idempotent permits, deduplicated multipart disk reservations, held operation leases, and deadline-bounded shutdown draining. Cover peak concurrency, fairness, resize, cancellation, multi-batch drain renewal, terminal lease paths, multipart accounting, and shutdown ordering with focused RED-to-GREEN tests.
This commit is contained in:
@@ -0,0 +1,405 @@
|
|||||||
|
export type ExtractionOperationContext = Readonly<{
|
||||||
|
operationId: string;
|
||||||
|
packageId: string;
|
||||||
|
generation: number;
|
||||||
|
runOwnerId: string;
|
||||||
|
}>;
|
||||||
|
|
||||||
|
export type ExtractionArchiveMember = Readonly<{
|
||||||
|
path: string;
|
||||||
|
size: number | null;
|
||||||
|
}>;
|
||||||
|
|
||||||
|
export type ExtractionLeaseRequest = Readonly<{
|
||||||
|
phase: "extract";
|
||||||
|
ownerId: string;
|
||||||
|
targetPath: string;
|
||||||
|
requiredBytes: number | null;
|
||||||
|
memberPaths: readonly string[];
|
||||||
|
}>;
|
||||||
|
|
||||||
|
export type ExtractionDiskLease = {
|
||||||
|
readonly released?: boolean;
|
||||||
|
release(): void;
|
||||||
|
};
|
||||||
|
|
||||||
|
export type BeginExtractionOperationOptions = {
|
||||||
|
context: ExtractionOperationContext;
|
||||||
|
targetPath?: string;
|
||||||
|
members?: readonly ExtractionArchiveMember[];
|
||||||
|
acquireLease?: (request: ExtractionLeaseRequest) => Promise<ExtractionDiskLease>;
|
||||||
|
};
|
||||||
|
|
||||||
|
export type ExtractionOperation = {
|
||||||
|
readonly context: ExtractionOperationContext;
|
||||||
|
finalize(finalizeScope?: () => void | Promise<void>): Promise<void>;
|
||||||
|
};
|
||||||
|
|
||||||
|
type ArchiveJob = {
|
||||||
|
archiveId: string;
|
||||||
|
execute: (signal: AbortSignal) => Promise<unknown>;
|
||||||
|
resolve: (value: unknown) => void;
|
||||||
|
reject: (reason?: unknown) => void;
|
||||||
|
};
|
||||||
|
|
||||||
|
type OperationState = {
|
||||||
|
handle: ExtractionOperation;
|
||||||
|
context: ExtractionOperationContext;
|
||||||
|
queued: ArchiveJob[];
|
||||||
|
active: Set<Promise<void>>;
|
||||||
|
controllers: Set<AbortController>;
|
||||||
|
drain: Deferred<void>;
|
||||||
|
drained: boolean;
|
||||||
|
lease: ExtractionDiskLease | null;
|
||||||
|
leaseReleased: boolean;
|
||||||
|
cancelReason: string | null;
|
||||||
|
closing: boolean;
|
||||||
|
finalizePromise: Promise<void> | null;
|
||||||
|
};
|
||||||
|
|
||||||
|
type Deferred<T> = {
|
||||||
|
promise: Promise<T>;
|
||||||
|
resolve: (value: T | PromiseLike<T>) => void;
|
||||||
|
};
|
||||||
|
|
||||||
|
function deferred<T>(): Deferred<T> {
|
||||||
|
let resolve!: Deferred<T>["resolve"];
|
||||||
|
const promise = new Promise<T>((resolvePromise) => {
|
||||||
|
resolve = resolvePromise;
|
||||||
|
});
|
||||||
|
return { promise, resolve };
|
||||||
|
}
|
||||||
|
|
||||||
|
function normalizedLimit(value: number): number {
|
||||||
|
const parsed = Number(value);
|
||||||
|
return Number.isFinite(parsed) ? Math.max(1, Math.floor(parsed)) : 1;
|
||||||
|
}
|
||||||
|
|
||||||
|
function memberKey(memberPath: string): string {
|
||||||
|
return String(memberPath || "").replace(/\//g, "\\").toLocaleLowerCase("en-US");
|
||||||
|
}
|
||||||
|
|
||||||
|
function deduplicateMembers(members: readonly ExtractionArchiveMember[]): ExtractionArchiveMember[] {
|
||||||
|
const unique = new Map<string, ExtractionArchiveMember>();
|
||||||
|
for (const member of members) {
|
||||||
|
const key = memberKey(member.path);
|
||||||
|
if (!key || unique.has(key)) {
|
||||||
|
continue;
|
||||||
|
}
|
||||||
|
unique.set(key, Object.freeze({
|
||||||
|
path: member.path,
|
||||||
|
size: typeof member.size === "number" && Number.isFinite(member.size)
|
||||||
|
? Math.max(0, Math.floor(member.size))
|
||||||
|
: null
|
||||||
|
}));
|
||||||
|
}
|
||||||
|
return [...unique.values()];
|
||||||
|
}
|
||||||
|
|
||||||
|
function reservationBytes(members: readonly ExtractionArchiveMember[]): number | null {
|
||||||
|
const known = members
|
||||||
|
.map((member) => member.size)
|
||||||
|
.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;
|
||||||
|
}
|
||||||
|
|
||||||
|
export class ExtractionCancelledError extends Error {
|
||||||
|
public constructor(reason: string) {
|
||||||
|
super(`Extraction cancelled: ${reason}`);
|
||||||
|
this.name = "ExtractionCancelledError";
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
export class ExtractionCoordinator {
|
||||||
|
private limit: number;
|
||||||
|
private activeCount = 0;
|
||||||
|
private closed = false;
|
||||||
|
private lastServedOperationId: string | null = null;
|
||||||
|
private readonly operations = new Map<string, OperationState>();
|
||||||
|
private readonly readyOperationIds: string[] = [];
|
||||||
|
private readonly readyOperationSet = new Set<string>();
|
||||||
|
|
||||||
|
public constructor(limit: number) {
|
||||||
|
this.limit = normalizedLimit(limit);
|
||||||
|
}
|
||||||
|
|
||||||
|
public async beginOperation(options: BeginExtractionOperationOptions): Promise<ExtractionOperation> {
|
||||||
|
if (this.closed) {
|
||||||
|
throw new ExtractionCancelledError("shutdown");
|
||||||
|
}
|
||||||
|
const context = Object.freeze({
|
||||||
|
operationId: String(options.context.operationId),
|
||||||
|
packageId: String(options.context.packageId),
|
||||||
|
generation: Math.max(0, Math.floor(Number(options.context.generation) || 0)),
|
||||||
|
runOwnerId: String(options.context.runOwnerId)
|
||||||
|
});
|
||||||
|
if (!context.operationId || !context.packageId || this.operations.has(context.operationId)) {
|
||||||
|
throw new Error(`Invalid extraction operation: ${context.operationId}`);
|
||||||
|
}
|
||||||
|
const state = {} as OperationState;
|
||||||
|
const handle: ExtractionOperation = Object.freeze({
|
||||||
|
context,
|
||||||
|
finalize: (finalizeScope?: () => void | Promise<void>) => this.finalizeOperation(state, finalizeScope)
|
||||||
|
});
|
||||||
|
Object.assign(state, {
|
||||||
|
handle,
|
||||||
|
context,
|
||||||
|
queued: [],
|
||||||
|
active: new Set<Promise<void>>(),
|
||||||
|
controllers: new Set<AbortController>(),
|
||||||
|
drain: deferred<void>(),
|
||||||
|
drained: false,
|
||||||
|
lease: null,
|
||||||
|
leaseReleased: false,
|
||||||
|
cancelReason: null,
|
||||||
|
closing: false,
|
||||||
|
finalizePromise: null
|
||||||
|
});
|
||||||
|
this.operations.set(context.operationId, state);
|
||||||
|
|
||||||
|
try {
|
||||||
|
if (options.acquireLease) {
|
||||||
|
const members = deduplicateMembers(options.members || []);
|
||||||
|
state.lease = await options.acquireLease({
|
||||||
|
phase: "extract",
|
||||||
|
ownerId: context.operationId,
|
||||||
|
targetPath: String(options.targetPath || ""),
|
||||||
|
requiredBytes: reservationBytes(members),
|
||||||
|
memberPaths: Object.freeze(members.map((member) => member.path))
|
||||||
|
});
|
||||||
|
}
|
||||||
|
if (this.closed || state.cancelReason) {
|
||||||
|
this.releaseLease(state);
|
||||||
|
this.operations.delete(context.operationId);
|
||||||
|
throw new ExtractionCancelledError(state.cancelReason || "shutdown");
|
||||||
|
}
|
||||||
|
return handle;
|
||||||
|
} catch (error) {
|
||||||
|
if (this.operations.get(context.operationId) === state) {
|
||||||
|
this.operations.delete(context.operationId);
|
||||||
|
}
|
||||||
|
throw error;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
public scheduleArchive<T>(
|
||||||
|
operation: ExtractionOperation,
|
||||||
|
archiveId: string,
|
||||||
|
execute: (signal: AbortSignal) => Promise<T>
|
||||||
|
): Promise<T> {
|
||||||
|
const state = this.operations.get(operation.context.operationId);
|
||||||
|
if (!state || state.handle !== operation || state.closing || state.cancelReason || this.closed) {
|
||||||
|
return Promise.reject(new ExtractionCancelledError(state?.cancelReason || (this.closed ? "shutdown" : "operation_closed")));
|
||||||
|
}
|
||||||
|
if (state.drained) {
|
||||||
|
state.drain = deferred<void>();
|
||||||
|
state.drained = false;
|
||||||
|
}
|
||||||
|
const promise = new Promise<T>((resolve, reject) => {
|
||||||
|
state.queued.push({ archiveId, execute, resolve: (value) => resolve(value as T), reject });
|
||||||
|
});
|
||||||
|
this.markReady(state);
|
||||||
|
this.pump();
|
||||||
|
return promise;
|
||||||
|
}
|
||||||
|
|
||||||
|
public resize(limit: number): void {
|
||||||
|
this.limit = normalizedLimit(limit);
|
||||||
|
this.pump();
|
||||||
|
}
|
||||||
|
|
||||||
|
public async cancelRun(runOwnerId: string, reason = "run_cancelled"): Promise<void> {
|
||||||
|
await this.cancelMatching((state) => state.context.runOwnerId === runOwnerId, reason);
|
||||||
|
}
|
||||||
|
|
||||||
|
public async cancelPackage(packageId: string, reason = "package_cancelled"): Promise<void> {
|
||||||
|
await this.cancelMatching((state) => state.context.packageId === packageId, reason);
|
||||||
|
}
|
||||||
|
|
||||||
|
public async shutdownAndDrain(deadlineAt: number): Promise<void> {
|
||||||
|
this.closed = true;
|
||||||
|
const states = [...this.operations.values()];
|
||||||
|
for (const state of states) {
|
||||||
|
this.cancelQueued(state, "shutdown");
|
||||||
|
state.cancelReason ||= "shutdown";
|
||||||
|
}
|
||||||
|
await Promise.resolve();
|
||||||
|
for (const state of states) {
|
||||||
|
for (const controller of state.controllers) {
|
||||||
|
if (!controller.signal.aborted) {
|
||||||
|
controller.abort("shutdown");
|
||||||
|
}
|
||||||
|
}
|
||||||
|
this.resolveDrainIfIdle(state);
|
||||||
|
}
|
||||||
|
await this.waitUntilDeadline(Promise.all(states.map((state) => state.drain.promise)), deadlineAt);
|
||||||
|
const finalizers = states.map((state) => state.finalizePromise).filter((value): value is Promise<void> => Boolean(value));
|
||||||
|
if (finalizers.length > 0) {
|
||||||
|
await this.waitUntilDeadline(Promise.allSettled(finalizers), deadlineAt);
|
||||||
|
}
|
||||||
|
for (const state of states) {
|
||||||
|
this.releaseLease(state);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
private async finalizeOperation(state: OperationState, finalizeScope?: () => void | Promise<void>): Promise<void> {
|
||||||
|
if (state.finalizePromise) {
|
||||||
|
return state.finalizePromise;
|
||||||
|
}
|
||||||
|
state.closing = true;
|
||||||
|
this.resolveDrainIfIdle(state);
|
||||||
|
state.finalizePromise = (async () => {
|
||||||
|
try {
|
||||||
|
await state.drain.promise;
|
||||||
|
await finalizeScope?.();
|
||||||
|
} finally {
|
||||||
|
this.releaseLease(state);
|
||||||
|
if (this.operations.get(state.context.operationId) === state) {
|
||||||
|
this.operations.delete(state.context.operationId);
|
||||||
|
}
|
||||||
|
this.removeReady(state.context.operationId);
|
||||||
|
}
|
||||||
|
})();
|
||||||
|
return state.finalizePromise;
|
||||||
|
}
|
||||||
|
|
||||||
|
private async cancelMatching(predicate: (state: OperationState) => boolean, reason: string): Promise<void> {
|
||||||
|
const matches = [...this.operations.values()].filter(predicate);
|
||||||
|
for (const state of matches) {
|
||||||
|
state.cancelReason ||= reason;
|
||||||
|
this.cancelQueued(state, reason);
|
||||||
|
for (const controller of state.controllers) {
|
||||||
|
if (!controller.signal.aborted) {
|
||||||
|
controller.abort(reason);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
this.resolveDrainIfIdle(state);
|
||||||
|
}
|
||||||
|
await Promise.all(matches.map((state) => state.drain.promise));
|
||||||
|
}
|
||||||
|
|
||||||
|
private cancelQueued(state: OperationState, reason: string): void {
|
||||||
|
const error = new ExtractionCancelledError(reason);
|
||||||
|
for (const job of state.queued.splice(0)) {
|
||||||
|
job.reject(error);
|
||||||
|
}
|
||||||
|
this.removeReady(state.context.operationId);
|
||||||
|
}
|
||||||
|
|
||||||
|
private markReady(state: OperationState): void {
|
||||||
|
const operationId = state.context.operationId;
|
||||||
|
if (state.queued.length === 0 || this.readyOperationSet.has(operationId)) {
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
this.readyOperationSet.add(operationId);
|
||||||
|
this.readyOperationIds.push(operationId);
|
||||||
|
}
|
||||||
|
|
||||||
|
private removeReady(operationId: string): void {
|
||||||
|
if (!this.readyOperationSet.delete(operationId)) {
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
let index = this.readyOperationIds.indexOf(operationId);
|
||||||
|
while (index >= 0) {
|
||||||
|
this.readyOperationIds.splice(index, 1);
|
||||||
|
index = this.readyOperationIds.indexOf(operationId);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
private takeNextState(): OperationState | null {
|
||||||
|
if (this.readyOperationIds.length === 0) {
|
||||||
|
return null;
|
||||||
|
}
|
||||||
|
let selectedIndex = 0;
|
||||||
|
if (this.lastServedOperationId) {
|
||||||
|
const differentIndex = this.readyOperationIds.findIndex((operationId) => operationId !== this.lastServedOperationId);
|
||||||
|
if (differentIndex >= 0) {
|
||||||
|
selectedIndex = differentIndex;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
const [operationId] = this.readyOperationIds.splice(selectedIndex, 1);
|
||||||
|
this.readyOperationSet.delete(operationId);
|
||||||
|
const state = this.operations.get(operationId);
|
||||||
|
if (!state || state.queued.length === 0 || state.cancelReason) {
|
||||||
|
return this.takeNextState();
|
||||||
|
}
|
||||||
|
this.lastServedOperationId = operationId;
|
||||||
|
return state;
|
||||||
|
}
|
||||||
|
|
||||||
|
private pump(): void {
|
||||||
|
if (this.closed) {
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
while (this.activeCount < this.limit) {
|
||||||
|
const state = this.takeNextState();
|
||||||
|
if (!state) {
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
const job = state.queued.shift();
|
||||||
|
if (!job) {
|
||||||
|
this.resolveDrainIfIdle(state);
|
||||||
|
continue;
|
||||||
|
}
|
||||||
|
if (state.queued.length > 0) {
|
||||||
|
this.markReady(state);
|
||||||
|
}
|
||||||
|
this.startJob(state, job);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
private startJob(state: OperationState, job: ArchiveJob): void {
|
||||||
|
const controller = new AbortController();
|
||||||
|
state.controllers.add(controller);
|
||||||
|
this.activeCount += 1;
|
||||||
|
let released = false;
|
||||||
|
const releasePermit = (): void => {
|
||||||
|
if (released) {
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
released = true;
|
||||||
|
this.activeCount = Math.max(0, this.activeCount - 1);
|
||||||
|
};
|
||||||
|
const active = Promise.resolve()
|
||||||
|
.then(() => job.execute(controller.signal))
|
||||||
|
.then(job.resolve, job.reject)
|
||||||
|
.finally(() => {
|
||||||
|
releasePermit();
|
||||||
|
state.controllers.delete(controller);
|
||||||
|
state.active.delete(active);
|
||||||
|
this.resolveDrainIfIdle(state);
|
||||||
|
this.pump();
|
||||||
|
});
|
||||||
|
state.active.add(active);
|
||||||
|
}
|
||||||
|
|
||||||
|
private resolveDrainIfIdle(state: OperationState): void {
|
||||||
|
if (!state.drained && state.queued.length === 0 && state.active.size === 0) {
|
||||||
|
state.drained = true;
|
||||||
|
state.drain.resolve();
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
private releaseLease(state: OperationState): void {
|
||||||
|
if (state.leaseReleased) {
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
state.leaseReleased = true;
|
||||||
|
state.lease?.release();
|
||||||
|
}
|
||||||
|
|
||||||
|
private async waitUntilDeadline<T>(task: Promise<T>, deadlineAt: number): Promise<void> {
|
||||||
|
const remainingMs = Math.max(0, deadlineAt - Date.now());
|
||||||
|
if (remainingMs <= 0) {
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
let timeout: ReturnType<typeof setTimeout> | null = null;
|
||||||
|
const elapsed = new Promise<void>((resolve) => {
|
||||||
|
timeout = setTimeout(resolve, remainingMs);
|
||||||
|
});
|
||||||
|
await Promise.race([task.then(() => undefined, () => undefined), elapsed]);
|
||||||
|
if (timeout) {
|
||||||
|
clearTimeout(timeout);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -0,0 +1,334 @@
|
|||||||
|
import { describe, expect, it, vi } from "vitest";
|
||||||
|
import {
|
||||||
|
ExtractionCancelledError,
|
||||||
|
ExtractionCoordinator,
|
||||||
|
type ExtractionOperationContext
|
||||||
|
} from "../src/main/extraction-coordinator";
|
||||||
|
|
||||||
|
type Deferred<T> = {
|
||||||
|
promise: Promise<T>;
|
||||||
|
resolve: (value: T | PromiseLike<T>) => void;
|
||||||
|
reject: (reason?: unknown) => void;
|
||||||
|
};
|
||||||
|
|
||||||
|
function deferred<T>(): Deferred<T> {
|
||||||
|
let resolve!: Deferred<T>["resolve"];
|
||||||
|
let reject!: Deferred<T>["reject"];
|
||||||
|
const promise = new Promise<T>((resolvePromise, rejectPromise) => {
|
||||||
|
resolve = resolvePromise;
|
||||||
|
reject = rejectPromise;
|
||||||
|
});
|
||||||
|
return { promise, resolve, reject };
|
||||||
|
}
|
||||||
|
|
||||||
|
function context(operationId: string, packageId: string, runOwnerId: string): ExtractionOperationContext {
|
||||||
|
return {
|
||||||
|
operationId,
|
||||||
|
packageId,
|
||||||
|
generation: 1,
|
||||||
|
runOwnerId
|
||||||
|
};
|
||||||
|
}
|
||||||
|
|
||||||
|
function lease(events: string[] = []) {
|
||||||
|
let released = false;
|
||||||
|
return {
|
||||||
|
get released() {
|
||||||
|
return released;
|
||||||
|
},
|
||||||
|
release: vi.fn(() => {
|
||||||
|
if (released) {
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
released = true;
|
||||||
|
events.push("lease-release");
|
||||||
|
})
|
||||||
|
};
|
||||||
|
}
|
||||||
|
|
||||||
|
async function flush(): Promise<void> {
|
||||||
|
await Promise.resolve();
|
||||||
|
await Promise.resolve();
|
||||||
|
}
|
||||||
|
|
||||||
|
describe("ExtractionCoordinator", () => {
|
||||||
|
it("enforces one global archive-child peak across full, hybrid and nested operations", async () => {
|
||||||
|
const coordinator = new ExtractionCoordinator(2);
|
||||||
|
const operations = await Promise.all([
|
||||||
|
coordinator.beginOperation({ context: context("full", "package-a", "run-a") }),
|
||||||
|
coordinator.beginOperation({ context: context("hybrid", "package-b", "run-a") }),
|
||||||
|
coordinator.beginOperation({ context: context("nested", "package-c", "run-b") })
|
||||||
|
]);
|
||||||
|
let active = 0;
|
||||||
|
let peak = 0;
|
||||||
|
const gates = Array.from({ length: 6 }, () => deferred<void>());
|
||||||
|
const jobs = gates.map((gate, index) => coordinator.scheduleArchive(
|
||||||
|
operations[index % operations.length],
|
||||||
|
`archive-${index}`,
|
||||||
|
async () => {
|
||||||
|
active += 1;
|
||||||
|
peak = Math.max(peak, active);
|
||||||
|
await gate.promise;
|
||||||
|
active -= 1;
|
||||||
|
}
|
||||||
|
));
|
||||||
|
|
||||||
|
await flush();
|
||||||
|
expect(active).toBe(2);
|
||||||
|
expect(peak).toBe(2);
|
||||||
|
|
||||||
|
for (const gate of gates) {
|
||||||
|
gate.resolve();
|
||||||
|
await flush();
|
||||||
|
}
|
||||||
|
await Promise.all(jobs);
|
||||||
|
await Promise.all(operations.map((operation) => operation.finalize()));
|
||||||
|
expect(peak).toBe(2);
|
||||||
|
});
|
||||||
|
|
||||||
|
it("rotates fairly between operation queues", async () => {
|
||||||
|
const coordinator = new ExtractionCoordinator(1);
|
||||||
|
const first = await coordinator.beginOperation({ context: context("first", "package-a", "run") });
|
||||||
|
const second = await coordinator.beginOperation({ context: context("second", "package-b", "run") });
|
||||||
|
const gate = deferred<void>();
|
||||||
|
const order: string[] = [];
|
||||||
|
|
||||||
|
const jobs = [
|
||||||
|
coordinator.scheduleArchive(first, "a1", async () => {
|
||||||
|
order.push("a1");
|
||||||
|
await gate.promise;
|
||||||
|
}),
|
||||||
|
coordinator.scheduleArchive(first, "a2", async () => { order.push("a2"); }),
|
||||||
|
coordinator.scheduleArchive(first, "a3", async () => { order.push("a3"); }),
|
||||||
|
coordinator.scheduleArchive(second, "b1", async () => { order.push("b1"); })
|
||||||
|
];
|
||||||
|
|
||||||
|
await flush();
|
||||||
|
expect(order).toEqual(["a1"]);
|
||||||
|
gate.resolve();
|
||||||
|
await Promise.all(jobs);
|
||||||
|
expect(order).toEqual(["a1", "b1", "a2", "a3"]);
|
||||||
|
await Promise.all([first.finalize(), second.finalize()]);
|
||||||
|
});
|
||||||
|
|
||||||
|
it("shrinks without revoking active permits and grows by filling the new capacity", async () => {
|
||||||
|
const coordinator = new ExtractionCoordinator(3);
|
||||||
|
const operation = await coordinator.beginOperation({ context: context("resize", "package-a", "run") });
|
||||||
|
const gates = Array.from({ length: 5 }, () => deferred<void>());
|
||||||
|
let active = 0;
|
||||||
|
const starts: number[] = [];
|
||||||
|
const jobs = gates.map((gate, index) => coordinator.scheduleArchive(operation, `archive-${index}`, async () => {
|
||||||
|
active += 1;
|
||||||
|
starts.push(index);
|
||||||
|
await gate.promise;
|
||||||
|
active -= 1;
|
||||||
|
}));
|
||||||
|
|
||||||
|
await flush();
|
||||||
|
expect(active).toBe(3);
|
||||||
|
coordinator.resize(1);
|
||||||
|
gates[0].resolve();
|
||||||
|
gates[1].resolve();
|
||||||
|
await flush();
|
||||||
|
expect(active).toBe(1);
|
||||||
|
expect(starts).toEqual([0, 1, 2]);
|
||||||
|
|
||||||
|
coordinator.resize(3);
|
||||||
|
await flush();
|
||||||
|
expect(active).toBe(3);
|
||||||
|
expect(starts).toEqual([0, 1, 2, 3, 4]);
|
||||||
|
|
||||||
|
for (const gate of gates) {
|
||||||
|
gate.resolve();
|
||||||
|
}
|
||||||
|
await Promise.all(jobs);
|
||||||
|
await operation.finalize();
|
||||||
|
});
|
||||||
|
|
||||||
|
it("cancels only jobs owned by the selected run", async () => {
|
||||||
|
const coordinator = new ExtractionCoordinator(2);
|
||||||
|
const cancelled = await coordinator.beginOperation({ context: context("cancelled", "package-a", "run-a") });
|
||||||
|
const retained = await coordinator.beginOperation({ context: context("retained", "package-b", "run-b") });
|
||||||
|
const cancelledClose = deferred<void>();
|
||||||
|
const retainedClose = deferred<void>();
|
||||||
|
let cancelledSignal: AbortSignal | null = null;
|
||||||
|
let retainedSignal: AbortSignal | null = null;
|
||||||
|
const cancelledActive = coordinator.scheduleArchive(cancelled, "a1", async (signal) => {
|
||||||
|
cancelledSignal = signal;
|
||||||
|
await cancelledClose.promise;
|
||||||
|
});
|
||||||
|
const retainedActive = coordinator.scheduleArchive(retained, "b1", async (signal) => {
|
||||||
|
retainedSignal = signal;
|
||||||
|
await retainedClose.promise;
|
||||||
|
});
|
||||||
|
const cancelledQueued = coordinator.scheduleArchive(cancelled, "a2", async () => undefined);
|
||||||
|
|
||||||
|
await flush();
|
||||||
|
const cancellation = coordinator.cancelRun("run-a", "stop");
|
||||||
|
await expect(cancelledQueued).rejects.toBeInstanceOf(ExtractionCancelledError);
|
||||||
|
expect((cancelledSignal as AbortSignal | null)?.aborted).toBe(true);
|
||||||
|
expect((retainedSignal as AbortSignal | null)?.aborted).toBe(false);
|
||||||
|
|
||||||
|
cancelledClose.resolve();
|
||||||
|
await expect(cancelledActive).resolves.toBeUndefined();
|
||||||
|
await cancellation;
|
||||||
|
retainedClose.resolve();
|
||||||
|
await retainedActive;
|
||||||
|
await Promise.all([cancelled.finalize(), retained.finalize()]);
|
||||||
|
});
|
||||||
|
|
||||||
|
it("cancels only jobs owned by the selected package and leaves no ghost waiter", async () => {
|
||||||
|
const coordinator = new ExtractionCoordinator(1);
|
||||||
|
const selected = await coordinator.beginOperation({ context: context("selected", "package-a", "run") });
|
||||||
|
const retained = await coordinator.beginOperation({ context: context("retained", "package-b", "run") });
|
||||||
|
const activeClose = deferred<void>();
|
||||||
|
const active = coordinator.scheduleArchive(selected, "a1", async () => activeClose.promise);
|
||||||
|
const ghost = coordinator.scheduleArchive(selected, "a2", async () => undefined);
|
||||||
|
const foreign = coordinator.scheduleArchive(retained, "b1", async () => "foreign");
|
||||||
|
|
||||||
|
await flush();
|
||||||
|
const cancellation = coordinator.cancelPackage("package-a", "cancel");
|
||||||
|
await expect(ghost).rejects.toBeInstanceOf(ExtractionCancelledError);
|
||||||
|
activeClose.resolve();
|
||||||
|
await active;
|
||||||
|
await cancellation;
|
||||||
|
await expect(foreign).resolves.toBe("foreign");
|
||||||
|
await Promise.all([selected.finalize(), retained.finalize()]);
|
||||||
|
});
|
||||||
|
|
||||||
|
it("releases an archive permit only once when completion races cancellation", async () => {
|
||||||
|
const coordinator = new ExtractionCoordinator(1);
|
||||||
|
const first = await coordinator.beginOperation({ context: context("first", "package-a", "run") });
|
||||||
|
const second = await coordinator.beginOperation({ context: context("second", "package-b", "run") });
|
||||||
|
const close = deferred<void>();
|
||||||
|
let secondStarts = 0;
|
||||||
|
const firstJob = coordinator.scheduleArchive(first, "a1", async () => close.promise);
|
||||||
|
const secondJob = coordinator.scheduleArchive(second, "b1", async () => {
|
||||||
|
secondStarts += 1;
|
||||||
|
});
|
||||||
|
|
||||||
|
await flush();
|
||||||
|
const cancelled = coordinator.cancelPackage("package-a", "cancel");
|
||||||
|
close.resolve();
|
||||||
|
await Promise.all([firstJob, cancelled, secondJob]);
|
||||||
|
expect(secondStarts).toBe(1);
|
||||||
|
await Promise.all([first.finalize(), second.finalize()]);
|
||||||
|
});
|
||||||
|
|
||||||
|
it("renews its drain barrier when one operation schedules another archive batch", async () => {
|
||||||
|
const coordinator = new ExtractionCoordinator(1);
|
||||||
|
const operation = await coordinator.beginOperation({ context: context("batches", "package-a", "run") });
|
||||||
|
await coordinator.scheduleArchive(operation, "first", async () => undefined);
|
||||||
|
await flush();
|
||||||
|
const secondClose = deferred<void>();
|
||||||
|
const second = coordinator.scheduleArchive(operation, "second", async () => secondClose.promise);
|
||||||
|
let finalized = false;
|
||||||
|
const finalization = operation.finalize().then(() => {
|
||||||
|
finalized = true;
|
||||||
|
});
|
||||||
|
|
||||||
|
await new Promise<void>((resolve) => setTimeout(resolve, 0));
|
||||||
|
expect(finalized).toBe(false);
|
||||||
|
secondClose.resolve();
|
||||||
|
await Promise.all([second, finalization]);
|
||||||
|
expect(finalized).toBe(true);
|
||||||
|
});
|
||||||
|
|
||||||
|
it.each(["success", "error", "timeout", "abort"] as const)("holds and releases the operation lease once on %s", async (terminal) => {
|
||||||
|
const coordinator = new ExtractionCoordinator(1);
|
||||||
|
const heldLease = lease();
|
||||||
|
const operation = await coordinator.beginOperation({
|
||||||
|
context: context(`lease-${terminal}`, "package-a", "run"),
|
||||||
|
targetPath: "C:\\target",
|
||||||
|
members: [{ path: "C:\\archives\\one.rar", size: 100 }],
|
||||||
|
acquireLease: async () => heldLease
|
||||||
|
});
|
||||||
|
const close = deferred<void>();
|
||||||
|
const job = coordinator.scheduleArchive(operation, "archive", async (signal) => {
|
||||||
|
if (terminal === "abort") {
|
||||||
|
await new Promise<void>((resolve) => signal.addEventListener("abort", () => resolve(), { once: true }));
|
||||||
|
}
|
||||||
|
await close.promise;
|
||||||
|
if (terminal === "error") {
|
||||||
|
throw new Error("extract failed");
|
||||||
|
}
|
||||||
|
if (terminal === "timeout") {
|
||||||
|
throw new Error("extract timeout");
|
||||||
|
}
|
||||||
|
});
|
||||||
|
|
||||||
|
await flush();
|
||||||
|
const finalized = operation.finalize();
|
||||||
|
if (terminal === "abort") {
|
||||||
|
void coordinator.cancelPackage("package-a", "abort");
|
||||||
|
}
|
||||||
|
expect(heldLease.release).not.toHaveBeenCalled();
|
||||||
|
close.resolve();
|
||||||
|
await job.catch(() => undefined);
|
||||||
|
await finalized;
|
||||||
|
expect(heldLease.release).toHaveBeenCalledTimes(1);
|
||||||
|
await operation.finalize();
|
||||||
|
expect(heldLease.release).toHaveBeenCalledTimes(1);
|
||||||
|
});
|
||||||
|
|
||||||
|
it("deduplicates multipart members before calculating the disk reservation", async () => {
|
||||||
|
const coordinator = new ExtractionCoordinator(1);
|
||||||
|
const heldLease = lease();
|
||||||
|
const requests: Array<{ requiredBytes: number | null; memberPaths: readonly string[] }> = [];
|
||||||
|
const operation = await coordinator.beginOperation({
|
||||||
|
context: context("multipart", "package-a", "run"),
|
||||||
|
targetPath: "C:\\target",
|
||||||
|
members: [
|
||||||
|
{ path: "C:\\archives\\show.part1.rar", size: 100 },
|
||||||
|
{ path: "c:\\ARCHIVES\\show.part1.rar", size: 100 },
|
||||||
|
{ path: "C:\\archives\\show.part2.rar", size: 200 }
|
||||||
|
],
|
||||||
|
acquireLease: async (request) => {
|
||||||
|
requests.push({ requiredBytes: request.requiredBytes, memberPaths: request.memberPaths });
|
||||||
|
return heldLease;
|
||||||
|
}
|
||||||
|
});
|
||||||
|
|
||||||
|
expect(requests).toEqual([{
|
||||||
|
requiredBytes: 300,
|
||||||
|
memberPaths: ["C:\\archives\\show.part1.rar", "C:\\archives\\show.part2.rar"]
|
||||||
|
}]);
|
||||||
|
await operation.finalize();
|
||||||
|
});
|
||||||
|
|
||||||
|
it("shuts down in queue-close, waiter-cancel, active-abort, child-drain, finalization and lease-release order", async () => {
|
||||||
|
const coordinator = new ExtractionCoordinator(1);
|
||||||
|
const events: string[] = [];
|
||||||
|
const heldLease = lease(events);
|
||||||
|
const operation = await coordinator.beginOperation({
|
||||||
|
context: context("shutdown", "package-a", "run"),
|
||||||
|
targetPath: "C:\\target",
|
||||||
|
members: [{ path: "C:\\archives\\one.rar", size: 100 }],
|
||||||
|
acquireLease: async () => heldLease
|
||||||
|
});
|
||||||
|
const childClose = deferred<void>();
|
||||||
|
const active = coordinator.scheduleArchive(operation, "active", async (signal) => {
|
||||||
|
signal.addEventListener("abort", () => events.push("active-abort"), { once: true });
|
||||||
|
await childClose.promise;
|
||||||
|
events.push("child-close");
|
||||||
|
});
|
||||||
|
const waiter = coordinator.scheduleArchive(operation, "waiter", async () => undefined).catch((error) => {
|
||||||
|
events.push("waiter-cancel");
|
||||||
|
throw error;
|
||||||
|
});
|
||||||
|
const finalized = operation.finalize(async () => {
|
||||||
|
events.push("scope-finalize");
|
||||||
|
});
|
||||||
|
|
||||||
|
await flush();
|
||||||
|
const shutdown = coordinator.shutdownAndDrain(Date.now() + 1000);
|
||||||
|
await expect(waiter).rejects.toBeInstanceOf(ExtractionCancelledError);
|
||||||
|
expect(events).toEqual(["waiter-cancel", "active-abort"]);
|
||||||
|
await expect(coordinator.beginOperation({ context: context("late", "package-b", "run") })).rejects.toBeInstanceOf(ExtractionCancelledError);
|
||||||
|
expect(heldLease.release).not.toHaveBeenCalled();
|
||||||
|
|
||||||
|
childClose.resolve();
|
||||||
|
await Promise.all([active, finalized, shutdown]);
|
||||||
|
expect(events).toEqual(["waiter-cancel", "active-abort", "child-close", "scope-finalize", "lease-release"]);
|
||||||
|
});
|
||||||
|
});
|
||||||
Reference in New Issue
Block a user