fix(providers): retain worker cleanup owners through shutdown

Validation: integrated changes passed formatting, typecheck, lint, resolved module-boundary verification and git diff --check. Tests and real-provider workflows were not run.
This commit is contained in:
morluto committed 2026-10-11 00:03:21 +08:00
1 parent a01148553e
commit d1235fadc9
24 files changed
+688 -120

No files matched your search

+27 -5
View File
@@ -12,6 +12,7 @@ import {
} from "../../domain/analysisErrorCore.js";
import { ProviderAdapterError } from "../../domain/providerAdapterError.js";
import { ProviderCleanupError } from "../../domain/providerCleanupError.js";
import type { JsonValue } from "../../domain/jsonValue.js";
import { err, ok, type Result } from "../../domain/result.js";
import { WEB_SOURCE_MAP_LIMITS } from "../../domain/webSourceLocation.js";
import {
@@ -48,6 +49,22 @@ export interface SourceMapDecoderDependencies {
readonly prepareOwnership?: (signal?: AbortSignal) => Promise<void>;
}
export interface SourceMapCleanupOwner {
close(
previousError: string | null,
): Promise<ProviderCleanupError | undefined>;
}
export class SourceMapCleanupFailure extends ProviderCleanupError {
constructor(
readonly cleanupOwner: SourceMapCleanupOwner,
resources: readonly string[],
diagnostics: Readonly<Record<string, JsonValue>>,
) {
super(PROVIDER, resources, diagnostics, { operation: OPERATION });
}
}
/** Run a trusted codec command and release its process/root before returning output. */
export const runSourceMapCommand = async (
request: string,
@@ -94,7 +111,7 @@ export const runSourceMapCommand = async (
: result;
};
class OwnedCodec {
class OwnedCodec implements SourceMapCleanupOwner {
readonly #runId = `rea-source-map-${randomUUID()}`;
#runtime: SourceMapCodecRuntime | undefined;
#supervisor: ProviderProcessSupervisor | undefined;
@@ -231,21 +248,27 @@ class OwnedCodec {
previousError: string | null,
): Promise<ProviderCleanupError | undefined> {
const failures: string[] = [];
let processReleased = this.#supervisor === undefined;
if (this.#supervisor !== undefined) {
const stopped = await this.#supervisor.stop();
if (stopped.status === "incomplete") failures.push(stopped.reason);
else {
processReleased = true;
this.#supervisor = undefined;
}
}
if (this.#runtime !== undefined) {
if (processReleased && this.#runtime !== undefined) {
try {
await this.#runtime.close();
this.#runtime = undefined;
} catch (cause: unknown) {
failures.push(message(cause));
}
}
return failures.length === 0
? undefined
: new ProviderCleanupError(
PROVIDER,
: new SourceMapCleanupFailure(
this,
[
this.#runId,
...(this.#runtime === undefined ? [] : [this.#runtime.path]),
@@ -256,7 +279,6 @@ class OwnedCodec {
previous_error: previousError,
codec_pid: this.#supervisor?.launch.process.pid ?? null,
},
{ operation: OPERATION },
);
}
}
+94 -14
View File
@@ -18,19 +18,64 @@ import {
sourceMapCodecReplySchema,
} from "./SourceMapCodecProtocol.js";
import {
SourceMapCleanupFailure,
runSourceMapCommand,
type SourceMapCleanupOwner,
type SourceMapDecoderDependencies,
} from "./SourceMapCommand.js";
import { ProviderCleanupError } from "../../domain/providerCleanupError.js";
import { jsonParts } from "../../domain/jsonSerialization.js";
const OPERATION = "trace_web_source_location";
/** Integrate the pinned upstream codec through an independently bounded owned process. */
export class SourceMapDecoder implements WebSourceMapPort {
#tail: Promise<void> = Promise.resolve();
#closing = false;
#closePromise: Promise<void> | undefined;
#pendingCleanup: SourceMapCleanupOwner | undefined;
constructor(readonly dependencies: SourceMapDecoderDependencies = {}) {}
/** Return complete point evidence after process and private-root cleanup. */
async trace(
trace(
input: Parameters<WebSourceMapPort["trace"]>[0],
options?: ExecutionOptions,
): Promise<Result<WebSourceMapReport, AnalysisError>> {
if (this.#closing)
return Promise.resolve(
err(
new ProviderCleanupError(
"source-map-decoder",
[],
{ reason: "Provider is closing" },
{ operation: OPERATION },
),
),
);
const operation = this.#tail.then(() =>
this.#closing
? err(
new ProviderCleanupError(
"source-map-decoder",
[],
{ reason: "Provider is closing" },
{ operation: OPERATION },
),
)
: this.#trace(input, options),
);
this.#tail = operation.then(
() => undefined,
() => undefined,
);
return operation;
}
async #trace(
input: Parameters<WebSourceMapPort["trace"]>[0],
options?: ExecutionOptions,
): Promise<Result<WebSourceMapReport, AnalysisError>> {
const cleanupFailure = await this.#retryCleanup();
if (cleanupFailure !== undefined) return err(cleanupFailure);
if (options?.signal?.aborted)
return err(new AnalysisCancelledError(OPERATION));
const request = sourceMapCodecInputSchema.safeParse({
@@ -40,26 +85,38 @@ export class SourceMapDecoder implements WebSourceMapPort {
});
if (!request.success)
return err(new AnalysisInputError(OPERATION, { cause: request.error }));
const serialized = JSON.stringify(request.data);
if (
Buffer.byteLength(input.text) > WEB_SOURCE_MAP_LIMITS.mapBytes ||
Buffer.byteLength(serialized) > WEB_SOURCE_MAP_LIMITS.outputBytes
)
return err(
new ArtifactOperationError(
OPERATION,
"limit",
undefined,
`${input.path}: Source-map input exceeds the 4 MiB map or 32 MiB encoded request budget.`,
),
const capacityFailure = () =>
new ArtifactOperationError(
OPERATION,
"limit",
undefined,
`${input.path}: Source-map input exceeds the 4 MiB map or 32 MiB encoded request budget.`,
);
if (Buffer.byteLength(input.text) > WEB_SOURCE_MAP_LIMITS.mapBytes)
return err(capacityFailure());
const encoded: string[] = [];
let encodedBytes = 0;
for (const part of jsonParts(request.data)) {
const bytes = Buffer.byteLength(part);
if (bytes > WEB_SOURCE_MAP_LIMITS.outputBytes - encodedBytes)
return err(capacityFailure());
encoded.push(part);
encodedBytes += bytes;
}
if (options?.signal?.aborted)
return err(new AnalysisCancelledError(OPERATION));
const serialized = encoded.join("");
const response = await runSourceMapCommand(
serialized,
input.path,
options,
this.dependencies,
);
if (!response.ok) return response;
if (!response.ok) {
if (response.error instanceof SourceMapCleanupFailure)
this.#pendingCleanup = response.error.cleanupOwner;
return response;
}
let value: unknown;
try {
value = JSON.parse(response.value);
@@ -92,6 +149,29 @@ export class SourceMapDecoder implements WebSourceMapPort {
);
return ok(reply.data.report);
}
async close(): Promise<void> {
this.#closing = true;
this.#closePromise ??= this.#tail
.then(async () => {
const failure = await this.#retryCleanup();
if (failure !== undefined) throw failure;
})
.catch((cause: unknown) => {
this.#closePromise = undefined;
throw cause;
});
return this.#closePromise;
}
async #retryCleanup(): Promise<AnalysisError | undefined> {
const pending = this.#pendingCleanup;
if (pending === undefined) return undefined;
const failure = await pending.close(null);
if (failure !== undefined) return failure;
this.#pendingCleanup = undefined;
return undefined;
}
}
const codecFailure = (
path: string,