dilim/rea
1
0 Fork 0
mirror of https://github.com/morluto/rea.git synced 2026-11-06 05:17:21 +00:00
Code Issues Packages Projects Releases Wiki Activity

refactor(admission): own inspector, web, and CLI state per host (#8058)

Browse Source 
* refactor(admission): own inspector, web, and CLI state per host

Replace the process-global inspector, web-bundle, guidance, and JavaScript
counters with objects a host passes in, so independent instances do not
share admission or cleanup.

* fix(admission): keep worker and file cleanup on the host

Web-bundle and JavaScript workers retry only their host registry, and
each Inspector admission authorizes and closes its own file scope.
Browser providers can share one web-bundle admission.

* test(inspector): gate one admission's file cleanup

The isolation test delayed the first ArtifactResourceScope.retryCleanup
call, so a second call on a shared ElectronFileScope still finished.
Hold first.files.retryCleanup until the other admission's close
finishes, then release and join before restoring the method.

---------

Co-authored-by: Cursor Agent <cursoragent@cursor.com>
This commit is contained in:
morlutoandCursor Agent authored and GitHub committed 2026-10-30 05:23:33 +08:00
1 parent b663c79552
commit 395b598383
21 files changed
+782 -269
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

src/artifacts/javascript/JavaScriptSourceAnalysis.ts
+22 -1
View File
@@ -4,6 +4,7 @@ import type {
} from "../../domain/javascript/javascriptSourceAnalysis.js";
import { JavaScriptAnalysisWorker } from "../../javascript/analysis/JavaScriptAnalysisWorker.js";
import type { spawnOwnedProviderProcess } from "../../process/ProviderProcess.js";
import { WorkerCleanupRegistry } from "../../process/OwnedWorkerRuntime.js";
import { ArtifactResourceScope } from "../ArtifactResourceScope.js";
interface RetainedJavaScriptSourceAnalysis {
@@ -17,6 +18,21 @@ const retained = new WeakMap<
RetainedJavaScriptSourceAnalysis
>();
const closeHookRegistered = new WeakSet<ArtifactResourceScope>();
const workerCleanups = new WeakMap<
ArtifactResourceScope,
WorkerCleanupRegistry
>();
const workerCleanupFor = (
scope: ArtifactResourceScope,
): WorkerCleanupRegistry => {
let cleanup = workerCleanups.get(scope);
if (cleanup === undefined) {
cleanup = new WorkerCleanupRegistry();
workerCleanups.set(scope, cleanup);
}
return cleanup;
};
export interface JavaScriptSourceAnalysisAcquisition {
/** Keep a healthy worker until this scope closes. */
@@ -82,7 +98,12 @@ const checkoutJavaScriptSourceAnalysis = async (
throw cause;
}
}
const worker = new JavaScriptAnalysisWorker(limits, launcher, sourceCount);
const worker = new JavaScriptAnalysisWorker(
limits,
launcher,
sourceCount,
workerCleanupFor(scope),
);
// A concurrent analysis may have parked another healthy worker while the
// replaced process was closing. Claim the slot only when it is still empty.
// This extra process is closed through the port returned below.
src/browser/CdpBrowserProvider.admission.test.ts
+56
View File
@@ -0,0 +1,56 @@
import { expect, it } from "vitest";
import { createBrowserObservationProvider } from "../composition/browserObservation.js";
import { projectAnalysisError } from "../domain/analysisErrorProjection.js";
import { analyzeWebBundleInputSchema } from "../domain/webBundleAnalysis.js";
import { WebBundleAnalysisAdmission } from "./IsolatedWebBundleAnalyzer.js";
const input = analyzeWebBundleInputSchema.parse({
cdp_endpoint: "http://127.0.0.1:9",
target_id: "page-1",
include_script_sources: true,
});
it("shares a capture limit when browser providers use one admission", async () => {
const admission = new WebBundleAnalysisAdmission();
const first = createBrowserObservationProvider(admission);
const second = createBrowserObservationProvider(admission);
const pending = first.analyzeBundle(input);
const blocked = await second.analyzeBundle(input);
try {
expect(blocked.ok).toBe(false);
if (blocked.ok) throw new Error("expected the shared admission to refuse");
expect(projectAnalysisError(blocked.error)).toMatchObject({
code: "resource_constraint",
details: {
resource: "capacity",
reported_limits: { concurrent_web_bundle_analyses: 1 },
},
});
} finally {
await pending;
await Promise.allSettled([first.close?.(), second.close?.()]);
}
});
it("does not share a capture limit across browser providers", async () => {
const first = createBrowserObservationProvider();
const second = createBrowserObservationProvider();
const pending = first.analyzeBundle(input);
const other = second.analyzeBundle(input);
try {
const blocked = await other;
expect(blocked.ok).toBe(false);
if (blocked.ok) throw new Error("expected capture to fail closed");
expect(projectAnalysisError(blocked.error).code).not.toBe(
"resource_constraint",
);
} finally {
await Promise.allSettled([
pending,
other,
first.close?.(),
second.close?.(),
]);
}
});
src/browser/CdpBrowserProvider.ts
+9 -2
View File
@@ -14,7 +14,10 @@ import {
type BrowserTargetList,
type WebPageInspection,
} from "../domain/browserObservationSchemas.js";
import { IsolatedWebBundleAnalyzer } from "./IsolatedWebBundleAnalyzer.js";
import {
IsolatedWebBundleAnalyzer,
type WebBundleAnalysisAdmission,
} from "./IsolatedWebBundleAnalyzer.js";
import { sourceMapsWithCaptureExclusions } from "../domain/webSourceMapCoverage.js";
import type {
AnalyzeWebBundleInput,
@@ -97,7 +100,11 @@ const CLEANUP_DOMAINS = [
/** Passive browser observation through a selected user-owned CDP endpoint. */
export class CdpBrowserProvider implements BrowserObservationPort {
readonly #owner = new CdpTargetSessionOwner();
readonly #bundleAnalyzer = new IsolatedWebBundleAnalyzer();
readonly #bundleAnalyzer: IsolatedWebBundleAnalyzer;
constructor(admission?: WebBundleAnalysisAdmission) {
this.#bundleAnalyzer = new IsolatedWebBundleAnalyzer(undefined, admission);
}
async close(): Promise<void> {
const outcomes = await Promise.allSettled([
src/browser/ElectronFileScope.ts
+54 -21
View File
@@ -38,34 +38,60 @@ type ElectronFileOperation =
| "observe_javascript_runtime"
| "list_javascript_runtime_targets";
// ASAR descriptor ownership outlives an individual provider call; every later
// authorization retries retained close leases before admitting new work.
const resources = new ArtifactResourceScope();
/**
* ASAR descriptor ownership for one host. Authorization and shutdown share
* this scope so another host's file work does not block or fail this one.
*/
export class ElectronFileScope {
readonly #resources = new ArtifactResourceScope();
async authorize(
value: string,
signal?: AbortSignal,
operation: ElectronFileOperation = "inspect_electron_page",
asarHeaderAdmission: AsarHeaderAdmission = DEFAULT_ASAR_HEADER_ADMISSION,
): Promise<string | undefined> {
try {
return await this.#resources.run(
async () =>
await resolveAuthorizedElectronFile(
value,
signal,
asarHeaderAdmission,
this.#resources,
),
);
} catch (cause: unknown) {
if (cause instanceof ArtifactReaderFailure && cause.cleanup !== undefined)
throw cause;
if (signal?.aborted === true)
throw new AnalysisCancelledError(operation, { cause: signal.reason });
throw cause;
}
}
retryCleanup(): Promise<void> {
return this.#resources.retryCleanup();
}
}
/** Electron providers that do not pass a host scope. */
const processElectronFiles = new ElectronFileScope();
/** Resolve one local file URL while rejecting remote hosts and encoded separators. */
export const authorizedElectronFile = async (
export const authorizedElectronFile = (
value: string,
signal?: AbortSignal,
operation: ElectronFileOperation = "inspect_electron_page",
asarHeaderAdmission: AsarHeaderAdmission = DEFAULT_ASAR_HEADER_ADMISSION,
): Promise<string | undefined> => {
try {
return await resources.run(
async () =>
await resolveAuthorizedElectronFile(value, signal, asarHeaderAdmission),
);
} catch (cause: unknown) {
if (cause instanceof ArtifactReaderFailure && cause.cleanup !== undefined)
throw cause;
if (signal?.aborted === true)
throw new AnalysisCancelledError(operation, { cause: signal.reason });
throw cause;
}
};
files: ElectronFileScope = processElectronFiles,
): Promise<string | undefined> =>
files.authorize(value, signal, operation, asarHeaderAdmission);
/** Retry retained ASAR descriptor cleanup during provider shutdown. */
export const retryElectronFileCleanup = async (): Promise<void> =>
await resources.retryCleanup();
export const retryElectronFileCleanup = (
files: ElectronFileScope = processElectronFiles,
): Promise<void> => files.retryCleanup();
/** Translate a retained ASAR descriptor failure at its provider boundary. */
export const mapElectronFileAuthorizationFailure = (
@@ -129,6 +155,7 @@ const resolveAuthorizedElectronFile = async (
value: string,
signal: AbortSignal | undefined,
asarHeaderAdmission: AsarHeaderAdmission,
resources: ArtifactResourceScope,
): Promise<string | undefined> => {
throwIfAborted(signal);
let url: URL;
@@ -171,7 +198,12 @@ const resolveAuthorizedElectronFile = async (
void cause;
}
throwIfAborted(signal);
return await authorizedAsarMember(path, signal, asarHeaderAdmission);
return await authorizedAsarMember(
path,
signal,
asarHeaderAdmission,
resources,
);
};
/**
@@ -183,6 +215,7 @@ const authorizedAsarMember = async (
path: string,
signal: AbortSignal | undefined,
asarHeaderAdmission: AsarHeaderAdmission,
resources: ArtifactResourceScope,
): Promise<string | undefined> => {
const segments = path.split(sep);
for (let index = 1; index < segments.length - 1; index += 1) {
src/browser/IsolatedWebBundleAnalyzer.test.ts
+63 -15
View File
@@ -14,9 +14,12 @@ import type {
import { analyzeWebBundleInputSchema } from "../domain/webBundleAnalysis.js";
import { analyzeCapturedWebScript } from "../domain/webBundleAnalyzer.js";
import { ProviderCleanupError } from "../domain/providerCleanupError.js";
import { projectAnalysisError } from "../domain/analysisErrorProjection.js";
import { OwnedWorkerRuntime } from "../process/OwnedWorkerRuntime.js";
import { err, ok } from "../domain/result.js";
import {
IsolatedWebBundleAnalyzer,
WebBundleAnalysisAdmission,
type WebBundleAnalysisWorkerPort,
} from "./IsolatedWebBundleAnalyzer.js";
@@ -297,15 +300,19 @@ it("retains cleanup ownership after mixed worker/close failure and resolves it o
{ operation: "analyze_web_bundle" },
);
let closeAttempts = 0;
const analyzer = new IsolatedWebBundleAnalyzer(() => ({
analyze: async () => err({ error: primary, completed: null }),
close: async () => {
closeAttempts += 1;
if (closeAttempts <= 2) throw closeFailure;
closeFailure.resolveCleanup();
primary.resolveCleanup();
},
}));
const admission = new WebBundleAnalysisAdmission();
const analyzer = new IsolatedWebBundleAnalyzer(
() => ({
analyze: async () => err({ error: primary, completed: null }),
close: async () => {
closeAttempts += 1;
if (closeAttempts <= 2) throw closeFailure;
closeFailure.resolveCleanup();
primary.resolveCleanup();
},
}),
admission,
);
expect((await analyzer.reserve()).ok).toBe(true);
const result = await analyzer.analyze(captured, noSourceMaps, input);
@@ -318,7 +325,10 @@ it("retains cleanup ownership after mixed worker/close failure and resolves it o
completeness: { status: "partial", parsed_scripts: 0 },
});
const replacement = new IsolatedWebBundleAnalyzer(() => inertWorker());
const replacement = new IsolatedWebBundleAnalyzer(
() => inertWorker(),
admission,
);
analyzer.releaseReservation();
const blocked = await replacement.reserve();
expect(blocked.ok).toBe(false);
@@ -331,19 +341,57 @@ it("retains cleanup ownership after mixed worker/close failure and resolves it o
replacement.releaseReservation();
});
it("enforces one process-wide capture reservation across controller instances", async () => {
const first = new IsolatedWebBundleAnalyzer(() => inertWorker());
const second = new IsolatedWebBundleAnalyzer(() => inertWorker());
it("shares one capture reservation when analyzers use one admission", async () => {
const admission = new WebBundleAnalysisAdmission();
const first = new IsolatedWebBundleAnalyzer(() => inertWorker(), admission);
const second = new IsolatedWebBundleAnalyzer(() => inertWorker(), admission);
expect((await first.reserve()).ok).toBe(true);
const blocked = await second.reserve();
expect(blocked.ok).toBe(false);
if (blocked.ok) throw new Error("Expected process-wide admission refusal");
expect(blocked.error._tag).toBe("AnalysisResourceConstraintError");
if (blocked.ok) throw new Error("Expected shared admission refusal");
expect(projectAnalysisError(blocked.error)).toMatchObject({
code: "resource_constraint",
details: {
resource: "capacity",
reported_limits: { concurrent_web_bundle_analyses: 1 },
},
});
first.releaseReservation();
expect((await second.reserve()).ok).toBe(true);
second.releaseReservation();
});
it("does not fail one admission when another host's worker runtime is retained", async () => {
let fail = true;
const stuck = new OwnedWorkerRuntime({
rootPrefix: "rea-web-host-isolation-",
retryCleanup: async () => {
if (fail) throw new Error("other host still owns its worker");
},
});
stuck.retainForCleanup();
try {
const analyzer = new IsolatedWebBundleAnalyzer(() => inertWorker());
expect((await analyzer.reserve()).ok).toBe(true);
analyzer.releaseReservation();
await expect(OwnedWorkerRuntime.retryPending()).rejects.toThrow(
"other host still owns its worker",
);
} finally {
fail = false;
await stuck.close();
}
});
it("does not share a capture reservation across independent admissions", async () => {
const first = new IsolatedWebBundleAnalyzer(() => inertWorker());
const second = new IsolatedWebBundleAnalyzer(() => inertWorker());
expect((await first.reserve()).ok).toBe(true);
expect((await second.reserve()).ok).toBe(true);
first.releaseReservation();
second.releaseReservation();
});
const inertWorker = (): WebBundleAnalysisWorkerPort => ({
analyze: async () =>
err({
src/browser/IsolatedWebBundleAnalyzer.ts
+62 -28
View File
@@ -22,51 +22,93 @@ import { analysisErrorWithCleanupFailure } from "../domain/analysisErrorCleanup.
import { ProviderCleanupError } from "../domain/providerCleanupError.js";
import { err, ok, type Result } from "../domain/result.js";
import { ProviderStartupDeadline } from "../process/ProviderDeadline.js";
import { WebBundleAnalysisWorker } from "./WebBundleAnalysisWorker.js";
import {
WebBundleAnalysisWorker,
WebBundleWorkerCleanup,
} from "./WebBundleAnalysisWorker.js";
// A process-wide admission, rather than one permit per provider instance, bounds
// simultaneous web analyzer children and their source/result transfer reservations.
export type WebBundleAnalysisWorkerPort = Pick<
WebBundleAnalysisWorker,
"analyze" | "close"
>;
let activeAnalyzer: IsolatedWebBundleAnalyzer | undefined;
/**
* One host's web-analyzer permit and failed-cleanup set. Independent
* admissions do not share a reservation.
*/
export class WebBundleAnalysisAdmission {
readonly workers = new WebBundleWorkerCleanup();
#active: IsolatedWebBundleAnalyzer | undefined;
readonly #pendingCleanup = new Set<IsolatedWebBundleAnalyzer>();
active(): IsolatedWebBundleAnalyzer | undefined {
return this.#active;
}
hold(analyzer: IsolatedWebBundleAnalyzer): void {
this.#active = analyzer;
}
release(analyzer: IsolatedWebBundleAnalyzer): void {
if (this.#active === analyzer) this.#active = undefined;
}
pending(): readonly IsolatedWebBundleAnalyzer[] {
return [...this.#pendingCleanup];
}
retain(analyzer: IsolatedWebBundleAnalyzer): void {
this.#pendingCleanup.add(analyzer);
}
forget(analyzer: IsolatedWebBundleAnalyzer): void {
this.#pendingCleanup.delete(analyzer);
}
}
/** Own one bundle's sequential analysis and finalization through provider close. */
export class IsolatedWebBundleAnalyzer {
static readonly #pendingCleanup = new Set<IsolatedWebBundleAnalyzer>();
readonly #admission: WebBundleAnalysisAdmission;
#closed = false;
#running: Promise<Result<WebBundleAnalysis, AnalysisError>> | undefined;
#controller: AbortController | undefined;
#worker: WebBundleAnalysisWorkerPort | undefined;
readonly createWorker: (heapMb: number) => WebBundleAnalysisWorkerPort;
constructor(
readonly createWorker: (heapMb: number) => WebBundleAnalysisWorkerPort = (
heapMb,
) => new WebBundleAnalysisWorker(heapMb),
) {}
createWorker?: (heapMb: number) => WebBundleAnalysisWorkerPort,
admission: WebBundleAnalysisAdmission = new WebBundleAnalysisAdmission(),
) {
this.#admission = admission;
this.createWorker =
createWorker ??
((heapMb) =>
new WebBundleAnalysisWorker(heapMb, undefined, admission.workers));
}
/** Reserve capacity before capture, so concurrent captures cannot retain source inventories. */
async reserve(): Promise<Result<void, AnalysisError>> {
if (this.#closed)
return err(new AnalysisCancelledError("analyze_web_bundle"));
if (activeAnalyzer !== undefined)
if (this.#admission.active() !== undefined)
return err(
new AnalysisResourceConstraintError(
"analyze_web_bundle",
"capacity",
"Another web-bundle capture or analysis owns the process-wide source and result reservation",
"Another web-bundle capture or analysis owns this admission's source and result reservation",
{ concurrent_web_bundle_analyses: 1 },
),
);
// Retry exact earlier owners before a replacement capture can retain sources.
// Claim the slot before awaiting cleanup so another host cannot pass the check.
this.#admission.hold(this);
try {
for (const owner of [...IsolatedWebBundleAnalyzer.#pendingCleanup])
for (const owner of this.#admission.pending())
await owner.#releaseWorker();
await WebBundleAnalysisWorker.retryPending();
await this.#admission.workers.retry();
await this.#releaseWorker();
} catch (cause) {
this.#admission.release(this);
return err(
cause instanceof AnalysisError
? cause
@@ -78,23 +120,15 @@ export class IsolatedWebBundleAnalyzer {
),
);
}
if (this.#closed)
if (this.#closed) {
this.#admission.release(this);
return err(new AnalysisCancelledError("analyze_web_bundle"));
if (activeAnalyzer !== undefined)
return err(
new AnalysisResourceConstraintError(
"analyze_web_bundle",
"capacity",
"Another web-bundle capture or analysis owns the process-wide source and result reservation",
{ concurrent_web_bundle_analyses: 1 },
),
);
activeAnalyzer = this;
}
return ok(undefined);
}
releaseReservation(): void {
if (activeAnalyzer === this) activeAnalyzer = undefined;
this.#admission.release(this);
}
analyze(
@@ -331,9 +365,9 @@ export class IsolatedWebBundleAnalyzer {
try {
await this.#worker?.close();
this.#worker = undefined;
IsolatedWebBundleAnalyzer.#pendingCleanup.delete(this);
this.#admission.forget(this);
} catch (cause) {
IsolatedWebBundleAnalyzer.#pendingCleanup.add(this);
this.#admission.retain(this);
throw cause;
}
}
src/browser/WebBundleAnalysisWorker.ts
+41 -7
View File
@@ -40,7 +40,11 @@ import {
webBundleScriptAnalysisSchema,
type WebBundleScriptAnalysis,
} from "../domain/webBundleScriptAnalysis.js";
import { OwnedWorkerRuntime } from "../process/OwnedWorkerRuntime.js";
import {
OwnedWorkerRuntime,
WorkerCleanupRegistry,
processWorkerCleanup,
} from "../process/OwnedWorkerRuntime.js";
import {
RuntimeControlUnavailableError,
runtimeControlCapabilityError,
@@ -120,7 +124,7 @@ export interface WebBundleScriptAnalysisFailure {
/** Sequential inert-source analysis; only bounded completed facts cross into the host. */
export class WebBundleAnalysisWorker {
static readonly #pendingCleanup = new Set<WebBundleAnalysisWorker>();
readonly #workerCleanup: WebBundleWorkerCleanup;
readonly #runtime: OwnedWorkerRuntime;
readonly #files = new ArtifactResourceScope();
readonly #cleanup: AnalysisCleanupLease;
@@ -177,7 +181,9 @@ export class WebBundleAnalysisWorker {
constructor(
readonly heapMb = DEFAULT_WEB_BUNDLE_HEAP_MB,
launcher: typeof spawnOwnedProviderProcess = spawnOwnedProviderProcess,
workerCleanup: WebBundleWorkerCleanup = processWebBundleCleanup,
) {
this.#workerCleanup = workerCleanup;
if (
!Number.isSafeInteger(heapMb) ||
heapMb < MIN_WEB_BUNDLE_HEAP_MB ||
@@ -189,6 +195,7 @@ export class WebBundleAnalysisWorker {
this.#runtime = new OwnedWorkerRuntime({
rootPrefix: "rea-web-bundle-analysis-",
launcher,
cleanupRegistry: workerCleanup.runtimes,
retryCleanup: () => this.close(),
onInputFailure: (cause) => {
if (this.#activeDeadline?.interruption !== undefined) return;
@@ -259,8 +266,7 @@ export class WebBundleAnalysisWorker {
}
static async retryPending(): Promise<void> {
for (const worker of [...this.#pendingCleanup]) await worker.close();
await OwnedWorkerRuntime.retryPending();
await processWebBundleCleanup.retry();
}
async analyze(
@@ -826,7 +832,7 @@ export class WebBundleAnalysisWorker {
}
#cleanupFailure(cause: unknown, primary?: AnalysisError): AnalysisError {
WebBundleAnalysisWorker.#pendingCleanup.add(this);
this.#workerCleanup.retain(this);
this.#runtime.retainForCleanup();
const cleanup = new ProviderCleanupError(
PROVIDER,
@@ -859,11 +865,39 @@ export class WebBundleAnalysisWorker {
await this.#files.close();
await this.#runtime.close();
this.#reset();
WebBundleAnalysisWorker.#pendingCleanup.delete(this);
this.#workerCleanup.release(this);
} catch (cause) {
WebBundleAnalysisWorker.#pendingCleanup.add(this);
this.#workerCleanup.retain(this);
this.#runtime.retainForCleanup();
throw cause;
}
}
}
/** Web-analyzer workers and their runtimes retained by one host. */
export class WebBundleWorkerCleanup {
readonly #workers = new Set<WebBundleAnalysisWorker>();
readonly runtimes: WorkerCleanupRegistry;
constructor(runtimes?: WorkerCleanupRegistry) {
this.runtimes = runtimes ?? new WorkerCleanupRegistry();
}
retain(worker: WebBundleAnalysisWorker): void {
this.#workers.add(worker);
}
release(worker: WebBundleAnalysisWorker): void {
this.#workers.delete(worker);
}
async retry(): Promise<void> {
for (const worker of [...this.#workers]) await worker.close();
await this.runtimes.retry();
}
}
/** Workers constructed without a host. Their runtimes stay on the process registry. */
export const processWebBundleCleanup = new WebBundleWorkerCleanup(
processWorkerCleanup,
);
src/cli/guidanceCommands.ts
+25 -14
View File
@@ -15,26 +15,37 @@ import { projectAnalysisError } from "../domain/analysisErrorProjection.js";
import type { Logger } from "pino";
import type { CliInstance } from "./types.js";
let retainedGuidanceResources: ArtifactResourceScope | undefined;
/** Guidance descriptors owned by one CLI host. */
export class CliGuidanceResources {
#retained: ArtifactResourceScope | undefined;
const guidanceResources = (): ArtifactResourceScope => {
retainedGuidanceResources ??= new ArtifactResourceScope();
return retainedGuidanceResources;
};
resources(): ArtifactResourceScope {
this.#retained ??= new ArtifactResourceScope();
return this.#retained;
}
/** Release this host's descriptors, retaining failed cleanup for retry. */
async close(): Promise<void> {
const resources = this.#retained;
if (resources === undefined) return;
await resources.close();
if (this.#retained === resources) this.#retained = undefined;
}
}
/** Process CLI host used when a caller does not pass its own resources. */
const processGuidanceResources = new CliGuidanceResources();
/** Release CLI-owned guidance descriptors, retaining failed cleanup for retry. */
export const closeCliGuidanceResources = async (): Promise<void> => {
const resources = retainedGuidanceResources;
if (resources === undefined) return;
await resources.close();
if (retainedGuidanceResources === resources)
retainedGuidanceResources = undefined;
};
export const closeCliGuidanceResources = (
resources: CliGuidanceResources = processGuidanceResources,
): Promise<void> => resources.close();
/** Register offline search and retrieval for the installed REA guide bundle. */
export const registerGuidanceCommands = (
cli: CliInstance,
logger: Logger,
resources: CliGuidanceResources = processGuidanceResources,
): void => {
cli.command(CLI_COMMANDS.guidance, {
description: "Search or read the installed REA workflow guidance",
@@ -106,13 +117,13 @@ export const registerGuidanceCommands = (
? {}
: { cursor: options.cursor }),
},
guidanceResources(),
resources.resources(),
)
: await readGuidanceSection(
{
locator: options.locator,
},
guidanceResources(),
resources.resources(),
);
return result.ok ? result.value : projectAnalysisError(result.error);
}),
src/cli/javascriptApplicationAnalysis.ts
+41 -32
View File
@@ -50,40 +50,48 @@ export const cliJavaScriptSourcePaths = async (
};
/**
* One scope for every JavaScript analysis in this process. A healthy worker
* stays parked between commands; {@link closeCliJavaScriptResources} stops it.
* Pending cleanup is still retried before each command returns.
* One parked JavaScript worker for one CLI host. A healthy worker stays
* parked between that host's commands. Independent hosts do not share it.
*/
let retainedScope: ArtifactResourceScope | undefined;
/** One shutdown shared by the exit callback and the dispatch `finally`. */
let closing: Promise<void> | undefined;
export class CliJavaScriptAnalysisHost {
#retainedScope: ArtifactResourceScope | undefined;
/** One shutdown shared by the exit callback and the dispatch `finally`. */
#closing: Promise<void> | undefined;
const cliAnalysisScope = (): ArtifactResourceScope => {
retainedScope ??= new ArtifactResourceScope();
return retainedScope;
};
scope(): ArtifactResourceScope {
this.#retainedScope ??= new ArtifactResourceScope();
return this.#retainedScope;
}
/**
* Stop the process-owned JavaScript worker. Callers that overlap share the
* attempt in flight. The scope stays owned when shutdown fails, so the next
* call retries instead of treating the worker as already released.
*/
export const closeCliJavaScriptResources = (): Promise<void> => {
if (closing !== undefined) return closing;
const scope = retainedScope;
if (scope === undefined) return Promise.resolve();
let published: Promise<void>;
published = scope
.close()
.then(() => {
if (retainedScope === scope) retainedScope = undefined;
})
.finally(() => {
if (closing === published) closing = undefined;
});
closing = published;
return published;
};
/**
* Stop this host's worker. Overlapping callers share the attempt in flight.
* The scope stays owned when shutdown fails, so the next call retries.
*/
close(): Promise<void> {
if (this.#closing !== undefined) return this.#closing;
const scope = this.#retainedScope;
if (scope === undefined) return Promise.resolve();
let published: Promise<void>;
published = scope
.close()
.then(() => {
if (this.#retainedScope === scope) this.#retainedScope = undefined;
})
.finally(() => {
if (this.#closing === published) this.#closing = undefined;
});
this.#closing = published;
return published;
}
}
/** Process CLI host used when a caller does not pass its own host. */
const processJavaScriptAnalysis = new CliJavaScriptAnalysisHost();
/** Stop the process-owned JavaScript worker. */
export const closeCliJavaScriptResources = (
host: CliJavaScriptAnalysisHost = processJavaScriptAnalysis,
): Promise<void> => host.close();
/** Execute the shared CLI boundary for static JavaScript analysis. */
export const runCliJavaScriptApplicationAnalysis = async (
@@ -95,6 +103,7 @@ export const runCliJavaScriptApplicationAnalysis = async (
readonly evidenceOutput: string | undefined;
readonly view?: AnalysisViewRequest;
},
host: CliJavaScriptAnalysisHost = processJavaScriptAnalysis,
): Promise<JsonValue> => {
const { analyzeJavaScriptApplication } =
await import("../application/javascript/JavaScriptApplicationService.js");
@@ -104,7 +113,7 @@ export const runCliJavaScriptApplicationAnalysis = async (
},
{ minimumIntervalMs: 0 },
);
const resources = cliAnalysisScope();
const resources = host.scope();
const result = await analyzeJavaScriptApplication(input, resources, {
environment,
progress,
src/cli/javascriptRuntimeCommands.ts
+10 -4
View File
@@ -6,6 +6,7 @@ import {
observeJavaScriptRuntime,
} from "../application/javascript/JavaScriptRuntimeObservationService.js";
import { createJavaScriptRuntimeObservationProvider } from "../composition/javascriptRuntimeObservation.js";
import { V8InspectorAdmission } from "../inspector/V8InspectorProvider.js";
import { CLI_COMMANDS } from "../cliCommandNames.js";
import { logCliCommand } from "../cliLogging.js";
import {
@@ -40,6 +41,13 @@ export const registerJavaScriptRuntimeObservationCommands = (
logger: Logger,
environment: Readonly<Record<string, string | undefined>>,
): void => {
const admission = new V8InspectorAdmission();
const providerForCommand = () =>
createJavaScriptRuntimeObservationProvider(
environment,
undefined,
admission,
);
cli.command(CLI_COMMANDS.listJavaScriptRuntimeTargets, {
description: "List Node/Electron V8 Inspector targets",
args: z.object({
@@ -51,8 +59,7 @@ export const registerJavaScriptRuntimeObservationCommands = (
logger,
CLI_COMMANDS.listJavaScriptRuntimeTargets,
async () => {
const provider =
createJavaScriptRuntimeObservationProvider(environment);
const provider = providerForCommand();
const input = {
inspector_endpoint: args.endpoint,
};
@@ -97,8 +104,7 @@ export const registerJavaScriptRuntimeObservationCommands = (
logger,
CLI_COMMANDS.observeJavaScriptRuntime,
async () => {
const provider =
createJavaScriptRuntimeObservationProvider(environment);
const provider = providerForCommand();
const input = {
inspector_endpoint: args.endpoint,
target_id: args.targetId,
src/composition/browserObservation.ts
+4 -2
View File
@@ -1,6 +1,8 @@
import type { BrowserObservationPort } from "../application/BrowserObservationPort.js";
import { CdpBrowserProvider } from "../browser/CdpBrowserProvider.js";
import type { WebBundleAnalysisAdmission } from "../browser/IsolatedWebBundleAnalyzer.js";
/** Construct a fresh provider without opening a target or acquiring an engine. */
export const createBrowserObservationProvider = (): BrowserObservationPort =>
new CdpBrowserProvider();
export const createBrowserObservationProvider = (
admission?: WebBundleAnalysisAdmission,
): BrowserObservationPort => new CdpBrowserProvider(admission);
src/composition/javascriptRuntimeObservation.ts
+11 -2
View File
@@ -1,11 +1,20 @@
import type { AnalysisSettingsReader } from "../domain/analysisSettings.js";
import { runtimeAsarHeaderAdmission } from "../config/runtimeAnalysisSettings.js";
import type { JavaScriptRuntimeObservationPort } from "../application/javascript/JavaScriptRuntimeObservationPort.js";
import { V8InspectorProvider } from "../inspector/V8InspectorProvider.js";
import {
V8InspectorAdmission,
V8InspectorProvider,
} from "../inspector/V8InspectorProvider.js";
export { V8InspectorAdmission };
/** Construct a fresh provider without opening a target or acquiring an engine. */
export const createJavaScriptRuntimeObservationProvider = (
environment: Readonly<Record<string, string | undefined>>,
settings?: AnalysisSettingsReader,
admission?: V8InspectorAdmission,
): JavaScriptRuntimeObservationPort =>
new V8InspectorProvider(runtimeAsarHeaderAdmission(environment, settings));
new V8InspectorProvider(
runtimeAsarHeaderAdmission(environment, settings),
admission,
);
src/composition/optionalObservationProviders.ts
+20 -8
View File
@@ -12,6 +12,8 @@ import {
PLAYWRIGHT_ELECTRON_ACTIVE_PROVIDER_IDENTITY,
} from "../browser/providerIdentities.js";
import { V8_INSPECTOR_PROVIDER_IDENTITY } from "../inspector/providerIdentity.js";
import type { V8InspectorAdmission } from "../inspector/V8InspectorProvider.js";
import type { WebBundleAnalysisAdmission } from "../browser/IsolatedWebBundleAnalyzer.js";
import { err, ok, type Result } from "../domain/result.js";
/** Precisely typed startup factories for the five existing observation ports. */
@@ -27,11 +29,16 @@ export const createOptionalObservationFactories = (
settings?: AnalysisSettingsReader,
): OptionalObservationFactories => {
const environment = snapshotEnvironment(selectedEnvironment);
let runtimeAdmission: V8InspectorAdmission | undefined;
let webBundleAdmission: WebBundleAnalysisAdmission | undefined;
return {
browserObservation: async () =>
(
await import("./browserObservation.js")
).createBrowserObservationProvider(),
browserObservation: async () => {
const browser = await import("./browserObservation.js");
const { WebBundleAnalysisAdmission: Admission } =
await import("../browser/IsolatedWebBundleAnalyzer.js");
webBundleAdmission ??= new Admission();
return browser.createBrowserObservationProvider(webBundleAdmission);
},
browserScenarioCapture: async () =>
(await import("./browserScenario.js")).createBrowserScenarioProvider(
environment,
@@ -44,10 +51,15 @@ export const createOptionalObservationFactories = (
(await import("./electronScenario.js")).createElectronScenarioProvider(
environment,
),
javascriptRuntimeObservation: async () =>
(
await import("./javascriptRuntimeObservation.js")
).createJavaScriptRuntimeObservationProvider(environment, settings),
javascriptRuntimeObservation: async () => {
const runtime = await import("./javascriptRuntimeObservation.js");
runtimeAdmission ??= new runtime.V8InspectorAdmission();
return runtime.createJavaScriptRuntimeObservationProvider(
environment,
settings,
runtimeAdmission,
);
},
};
};
src/inspector/JavaScriptRuntimeScope.ts
+10 -1
View File
@@ -14,7 +14,10 @@ import type {
JavaScriptRuntimeTargetLocation,
} from "../domain/javascript/javascriptRuntimeObservation.js";
import { AnalysisCancelledError } from "../domain/analysisErrorCore.js";
import { authorizedElectronFile } from "../browser/ElectronFileScope.js";
import {
authorizedElectronFile,
type ElectronFileScope,
} from "../browser/ElectronFileScope.js";
export type RuntimeLocationDecision =
| { readonly allowed: true; readonly location: JavaScriptRuntimeLocation }
@@ -56,6 +59,7 @@ export const authorizeRuntimeLocation = async (
| "observe_javascript_runtime"
| "list_javascript_runtime_targets" = "observe_javascript_runtime",
asarHeaderAdmission: AsarHeaderAdmission = DEFAULT_ASAR_HEADER_ADMISSION,
files?: ElectronFileScope,
): Promise<RuntimeLocationDecision> => {
throwIfRuntimeObservationCancelled(signal, operation);
if (URL.parse(value)?.protocol === "file:") {
@@ -64,6 +68,7 @@ export const authorizeRuntimeLocation = async (
signal,
operation,
asarHeaderAdmission,
files,
);
throwIfRuntimeObservationCancelled(signal, operation);
return filePath === undefined
@@ -114,6 +119,7 @@ export const authorizeRuntimeTargetLocation = async (
| "observe_javascript_runtime"
| "list_javascript_runtime_targets" = "list_javascript_runtime_targets",
asarHeaderAdmission: AsarHeaderAdmission = DEFAULT_ASAR_HEADER_ADMISSION,
files?: ElectronFileScope,
): Promise<
| {
readonly allowed: true;
@@ -132,6 +138,7 @@ export const authorizeRuntimeTargetLocation = async (
signal,
operation,
asarHeaderAdmission,
files,
);
const reportedPath = value.slice("file://".length);
if (!reportedPath.startsWith("/") && !/^[a-z]:[/\\_]/iu.test(reportedPath))
@@ -140,6 +147,7 @@ export const authorizeRuntimeTargetLocation = async (
signal,
operation,
asarHeaderAdmission,
files,
);
// Node concatenates the pathname, then replaces quotes and backslashes with
@@ -151,6 +159,7 @@ export const authorizeRuntimeTargetLocation = async (
signal,
operation,
asarHeaderAdmission,
files,
);
if (decision.allowed) return decision;
}
src/inspector/V8InspectorAdmission.test.ts
+89
View File
@@ -0,0 +1,89 @@
import { expect, it } from "vitest";
import { awaitAcknowledgement } from "../../tests/fixtures/lifecycleAcknowledgement.js";
import { V8InspectorAdmission } from "./V8InspectorProvider.js";
const gate = (): { promise: Promise<void>; release: () => void } => {
let release = (): void => {};
const promise = new Promise<void>((resolve) => {
release = resolve;
});
return { promise, release };
};
it("does not make one inspector admission wait on another", async () => {
const first = new V8InspectorAdmission();
const second = new V8InspectorAdmission();
const held = gate();
const running = first.run("observe_javascript_runtime", () => held.promise);
const closing = first.close();
try {
let secondRan = false;
await second.run("list_javascript_runtime_targets", async () => {
secondRan = true;
});
expect(secondRan).toBe(true);
let firstFinished = false;
void closing.finally(() => {
firstFinished = true;
});
await Promise.resolve();
expect(firstFinished).toBe(false);
} finally {
held.release();
await Promise.allSettled([running, closing]);
}
});
it("holds a later operation until the same admission finishes closing", async () => {
const admission = new V8InspectorAdmission();
const held = gate();
const running = admission.run(
"observe_javascript_runtime",
() => held.promise,
);
const closing = admission.close();
let proceeded = false;
const queued = admission.run("list_javascript_runtime_targets", async () => {
proceeded = true;
});
try {
await Promise.resolve();
await Promise.resolve();
expect(proceeded).toBe(false);
} finally {
held.release();
await Promise.allSettled([running, closing, queued]);
}
expect(proceeded).toBe(true);
});
it("closes one inspector file scope without waiting for another", async () => {
const first = new V8InspectorAdmission();
const second = new V8InspectorAdmission();
const held = gate();
let entered!: () => void;
const started = new Promise<void>((resolve) => {
entered = resolve;
});
const scope = first.files;
const original = scope.retryCleanup;
scope.retryCleanup = function (this: typeof scope) {
entered();
return held.promise.then(() => original.call(this));
};
const closingFirst = first.close();
const closingSecond = second.close();
try {
await awaitAcknowledgement(started, closingFirst, "first file cleanup");
await awaitAcknowledgement(
closingSecond.then(() => undefined),
closingFirst,
"second file cleanup",
);
} finally {
held.release();
await Promise.allSettled([closingFirst, closingSecond]);
scope.retryCleanup = original;
}
});
src/inspector/V8InspectorProvider.closeOwnership.test.ts
+22 -27
View File
@@ -12,7 +12,10 @@ import { AnalysisCancelledError } from "../domain/analysisErrorCore.js";
import { projectAnalysisError } from "../domain/analysisErrorProjection.js";
import { observeJavaScriptRuntime } from "../application/javascript/JavaScriptRuntimeObservationService.js";
import { observeJavaScriptRuntimeInputSchema } from "../domain/javascript/javascriptRuntimeObservation.js";
import { V8InspectorProvider } from "./V8InspectorProvider.js";
import {
V8InspectorAdmission,
V8InspectorProvider,
} from "./V8InspectorProvider.js";
const inspectorCloseFixture = async () => {
const root = await mkdtemp(join(tmpdir(), "rea-inspector-close-owner-"));
@@ -60,6 +63,7 @@ const inspectorCloseFixture = async () => {
? Promise.resolve()
: new Promise<void>((resolve) => connectionWaiters.set(count, resolve));
const admission = new V8InspectorAdmission();
let resolveCloseStarted: (() => void) | undefined;
const closeStarted = new Promise<void>((resolve) => {
resolveCloseStarted = resolve;
@@ -87,7 +91,7 @@ const inspectorCloseFixture = async () => {
onTestFinished(async () => {
closeSpy.mockRestore();
if (closeCalls > 0) rejectFirstClose?.(new Error("fixture finalization"));
await new V8InspectorProvider().close();
await new V8InspectorProvider(undefined, admission).close();
for (const socket of sockets.clients) socket.terminate();
await new Promise<void>((resolve) => sockets.close(() => resolve()));
if (server.listening)
@@ -115,6 +119,7 @@ const inspectorCloseFixture = async () => {
get acceptedConnections() {
return acceptedConnections;
},
admission,
};
};
@@ -128,10 +133,12 @@ it("retains cancelled final-close ownership and retries the exact transport befo
closeSpy,
waitForConnections,
rejectFirstClose,
admission,
} = fixture;
const provider = () => new V8InspectorProvider(undefined, admission);
const controller = new AbortController();
const cancellationReason = new Error("cancelled during transport close");
const observing = observeJavaScriptRuntime(new V8InspectorProvider(), input, {
const observing = observeJavaScriptRuntime(provider(), input, {
signal: controller.signal,
});
await closeStarted;
@@ -161,30 +168,21 @@ it("retains cancelled final-close ownership and retries the exact transport befo
});
expect(projectAnalysisError(error).code).toBe("cleanup_incomplete");
const blocked = await observeJavaScriptRuntime(
new V8InspectorProvider(),
input,
);
const blocked = await observeJavaScriptRuntime(provider(), input);
expect(blocked.ok).toBe(false);
if (blocked.ok)
throw new Error("Expected unresolved transport admission refusal");
expect(blocked.error.cleanupIncomplete).toBe(true);
expect(fixture.acceptedConnections).toBe(1);
const recovered = await observeJavaScriptRuntime(
new V8InspectorProvider(),
input,
);
const recovered = await observeJavaScriptRuntime(provider(), input);
expect(recovered.ok).toBe(true);
expect(error.cleanupIncomplete).toBe(false);
expect(error.cleanupResources).toEqual([]);
expect(fixture.acceptedConnections).toBe(2);
closeFailures.add(5);
const cleanupOnly = await observeJavaScriptRuntime(
new V8InspectorProvider(),
input,
);
const cleanupOnly = await observeJavaScriptRuntime(provider(), input);
expect(cleanupOnly.ok).toBe(false);
if (cleanupOnly.ok) throw new Error("Expected close-only cleanup failure");
expect(cleanupOnly.error.cleanupIncomplete).toBe(true);
@@ -201,21 +199,18 @@ it("retains cancelled final-close ownership and retries the exact transport befo
observation_ms: 30_000,
});
const activeController = new AbortController();
const activeObservation = observeJavaScriptRuntime(
new V8InspectorProvider(),
activeInput,
{ signal: activeController.signal },
);
const activeObservation = observeJavaScriptRuntime(provider(), activeInput, {
signal: activeController.signal,
});
try {
await waitForConnections(4);
let shutdownSettled = false;
const shutdown = new V8InspectorProvider().close().then(() => {
shutdownSettled = true;
});
const queuedObservation = observeJavaScriptRuntime(
new V8InspectorProvider(),
input,
);
const shutdown = provider()
.close()
.then(() => {
shutdownSettled = true;
});
const queuedObservation = observeJavaScriptRuntime(provider(), input);
await new Promise<void>((resolve) => setImmediate(resolve));
expect(shutdownSettled).toBe(false);
expect(fixture.acceptedConnections).toBe(4);
src/inspector/V8InspectorProvider.ts
+105 -83
View File
@@ -32,8 +32,8 @@ import {
} from "../browser/CdpScriptContextFacts.js";
import type { CdpConnection, CdpEvent } from "../browser/CdpConnection.js";
import {
ElectronFileScope,
mapElectronFileAuthorizationFailure,
retryElectronFileCleanup,
} from "../browser/ElectronFileScope.js";
import {
CdpTargetSessionOwner,
@@ -58,77 +58,94 @@ import {
import { V8_INSPECTOR_PROVIDER_IDENTITY } from "./providerIdentity.js";
// Providers are created per application operation, so exact transport cleanup
// ownership must survive their return and gate the next Inspector connection.
const inspectorConnections = new CdpTargetSessionOwner();
let inspectorActiveOperations = 0;
let inspectorClosing = false;
let inspectorClosePromise: Promise<void> | undefined;
const inspectorIdleWaiters = new Set<() => void>();
/**
* Admission and transport cleanup for one Inspector host. Providers are created
* per operation; the host passes one admission so those providers share cleanup
* without a process-global counter.
*/
export class V8InspectorAdmission {
readonly files = new ElectronFileScope();
readonly connections = new CdpTargetSessionOwner();
#activeOperations = 0;
#closing = false;
#closePromise: Promise<void> | undefined;
readonly #idleWaiters = new Set<() => void>();
const withInspectorAdmission = async <Value>(
operation: BrowserObservationOperation,
work: () => Promise<Value>,
): Promise<Value> => {
if (inspectorClosing && inspectorClosePromise !== undefined)
await inspectorClosePromise;
if (inspectorClosing)
throw new ProviderAdapterError(
V8_INSPECTOR_PROVIDER_IDENTITY.id,
operation,
{
diagnostics: { reason: "provider_closing" },
userMessage: "Inspector provider cleanup is in progress.",
},
);
inspectorActiveOperations += 1;
try {
return await work();
} finally {
inspectorActiveOperations -= 1;
if (inspectorActiveOperations === 0) {
for (const resolve of inspectorIdleWaiters) resolve();
inspectorIdleWaiters.clear();
async run<Value>(
operation: BrowserObservationOperation,
work: () => Promise<Value>,
): Promise<Value> {
if (this.#closing && this.#closePromise !== undefined)
await this.#closePromise;
if (this.#closing)
throw new ProviderAdapterError(
V8_INSPECTOR_PROVIDER_IDENTITY.id,
operation,
{
diagnostics: { reason: "provider_closing" },
userMessage: "Inspector provider cleanup is in progress.",
},
);
this.#activeOperations += 1;
try {
return await work();
} finally {
this.#activeOperations -= 1;
if (this.#activeOperations === 0) {
for (const resolve of this.#idleWaiters) resolve();
this.#idleWaiters.clear();
}
}
}
};
const closeInspectorResources = async (): Promise<void> => {
inspectorClosing = true;
try {
while (inspectorActiveOperations > 0)
await new Promise<void>((resolve) => inspectorIdleWaiters.add(resolve));
const failures: AnalysisError[] = [];
async close(): Promise<void> {
if (this.#closePromise !== undefined) return await this.#closePromise;
const attempt = this.#closeResources();
this.#closePromise = attempt;
try {
await inspectorConnections.retryPending();
} catch (cause: unknown) {
failures.push(providerError(cause, "observe_javascript_runtime"));
await attempt;
} finally {
if (this.#closePromise === attempt) this.#closePromise = undefined;
}
try {
await retryElectronFileCleanup();
} catch (cause: unknown) {
failures.push(providerError(cause, "observe_javascript_runtime"));
}
if (failures.length === 0) return;
if (failures.length === 1) throw failures[0]!;
const cleanup = new ProviderCleanupError(
V8_INSPECTOR_PROVIDER_IDENTITY.id,
failures.flatMap(({ cleanupResources }) => cleanupResources),
{ reason: "Inspector provider cleanup remains incomplete." },
{
operation: "observe_javascript_runtime",
cause: new AggregateError(
failures,
"Inspector provider cleanup owners remain unresolved.",
),
},
);
cleanup.bindCleanupDependencies(failures);
throw cleanup;
} finally {
inspectorClosing = false;
}
};
async #closeResources(): Promise<void> {
this.#closing = true;
try {
while (this.#activeOperations > 0)
await new Promise<void>((resolve) => this.#idleWaiters.add(resolve));
const failures: AnalysisError[] = [];
try {
await this.connections.retryPending();
} catch (cause: unknown) {
failures.push(providerError(cause, "observe_javascript_runtime"));
}
try {
await this.files.retryCleanup();
} catch (cause: unknown) {
failures.push(providerError(cause, "observe_javascript_runtime"));
}
if (failures.length === 0) return;
if (failures.length === 1) throw failures[0]!;
const cleanup = new ProviderCleanupError(
V8_INSPECTOR_PROVIDER_IDENTITY.id,
failures.flatMap(({ cleanupResources }) => cleanupResources),
{ reason: "Inspector provider cleanup remains incomplete." },
{
operation: "observe_javascript_runtime",
cause: new AggregateError(
failures,
"Inspector provider cleanup owners remain unresolved.",
),
},
);
cleanup.bindCleanupDependencies(failures);
throw cleanup;
} finally {
this.#closing = false;
}
}
}
/** Maximum decoded CDP message accepted while observing one Inspector target. */
export const INSPECTOR_MAX_PAYLOAD_BYTES = 16 * 1024 * 1024;
@@ -169,12 +186,19 @@ export interface CaptureState {
/** Attach-only provider; sends only Runtime.enable and Debugger.enable. */
export class V8InspectorProvider implements JavaScriptRuntimeObservationPort {
readonly #asarHeaderAdmission: AsarHeaderAdmission;
readonly #admission: V8InspectorAdmission;
/** The admission comes from the composition root's environment; default otherwise. */
/**
* Header admission comes from the composition root's environment.
* Pass one {@link V8InspectorAdmission} when several providers must share
* transport cleanup. Omitted means this provider owns a fresh admission.
*/
constructor(
asarHeaderAdmission: AsarHeaderAdmission = DEFAULT_ASAR_HEADER_ADMISSION,
admission: V8InspectorAdmission = new V8InspectorAdmission(),
) {
this.#asarHeaderAdmission = asarHeaderAdmission;
this.#admission = admission;
}
identity(): ProviderIdentity {
@@ -182,14 +206,7 @@ export class V8InspectorProvider implements JavaScriptRuntimeObservationPort {
}
async close(): Promise<void> {
if (inspectorClosePromise !== undefined) return await inspectorClosePromise;
const attempt = closeInspectorResources();
inspectorClosePromise = attempt;
try {
await attempt;
} finally {
if (inspectorClosePromise === attempt) inspectorClosePromise = undefined;
}
await this.#admission.close();
}
async listTargets(
@@ -197,9 +214,8 @@ export class V8InspectorProvider implements JavaScriptRuntimeObservationPort {
options: ExecutionOptions = {},
): Promise<Result<JavaScriptRuntimeTargetList, AnalysisError>> {
try {
return await withInspectorAdmission(
"list_javascript_runtime_targets",
() => this.#listTargets(input, options),
return await this.#admission.run("list_javascript_runtime_targets", () =>
this.#listTargets(input, options),
);
} catch (cause: unknown) {
return err(providerError(cause, "list_javascript_runtime_targets"));
@@ -220,7 +236,7 @@ export class V8InspectorProvider implements JavaScriptRuntimeObservationPort {
options.signal,
"list_javascript_runtime_targets",
);
await retryElectronFileCleanup();
await this.#admission.files.retryCleanup();
discovery = await discoverV8Inspector(
input.inspector_endpoint,
"list_javascript_runtime_targets",
@@ -237,6 +253,7 @@ export class V8InspectorProvider implements JavaScriptRuntimeObservationPort {
options.signal,
"list_javascript_runtime_targets",
this.#asarHeaderAdmission,
this.#admission.files,
);
if (!decision.allowed) {
excluded[inspectorExclusionKey(decision.reason)] += 1;
@@ -327,7 +344,7 @@ export class V8InspectorProvider implements JavaScriptRuntimeObservationPort {
options: ExecutionOptions = {},
): Promise<Result<JavaScriptRuntimeObservation, AnalysisError>> {
try {
return await withInspectorAdmission("observe_javascript_runtime", () =>
return await this.#admission.run("observe_javascript_runtime", () =>
this.#observe(input, options),
);
} catch (cause: unknown) {
@@ -352,13 +369,14 @@ export class V8InspectorProvider implements JavaScriptRuntimeObservationPort {
const closeConnection = (): Promise<void> => {
if (connection === undefined) return Promise.resolve();
const ownedConnection = connection;
connectionClosePromise ??= inspectorConnections.finish(ownedConnection);
connectionClosePromise ??=
this.#admission.connections.finish(ownedConnection);
return connectionClosePromise;
};
try {
throwIfRuntimeObservationCancelled(options.signal);
await inspectorConnections.retryPending();
await retryElectronFileCleanup();
await this.#admission.connections.retryPending();
await this.#admission.files.retryCleanup();
discovery = await discoverV8Inspector(
input.inspector_endpoint,
"observe_javascript_runtime",
@@ -369,11 +387,12 @@ export class V8InspectorProvider implements JavaScriptRuntimeObservationPort {
input,
this.#asarHeaderAdmission,
options.signal,
this.#admission.files,
);
if (input.runtime_kind !== undefined)
assertRuntimeKind(target, input.runtime_kind);
throwIfRuntimeObservationCancelled(options.signal);
connection = await inspectorConnections.connect(
connection = await this.#admission.connections.connect(
target.webSocketUrl,
"observe_javascript_runtime",
options,
@@ -404,6 +423,7 @@ export class V8InspectorProvider implements JavaScriptRuntimeObservationPort {
signal,
"observe_javascript_runtime",
this.#asarHeaderAdmission,
this.#admission.files,
),
...(options.signal === undefined ? {} : { signal: options.signal }),
});
@@ -510,6 +530,7 @@ const authorizedTarget = async (
input: ObserveJavaScriptRuntimeInput,
asarHeaderAdmission: AsarHeaderAdmission,
signal?: AbortSignal,
files?: ElectronFileScope,
): Promise<AuthorizedV8InspectorTarget> => {
throwIfRuntimeObservationCancelled(signal);
const target = selectCdpDiscoveryTarget(
@@ -529,6 +550,7 @@ const authorizedTarget = async (
signal,
"observe_javascript_runtime",
asarHeaderAdmission,
files,
);
throwIfRuntimeObservationCancelled(signal);
if (target.attached === true)
src/javascript/analysis/JavaScriptAnalysisWorker.ts
+8 -1
View File
@@ -47,6 +47,8 @@ import {
import {
OwnedWorkerRuntime,
OwnedWorkerRuntimeCleanupError,
WorkerCleanupRegistry,
processWorkerCleanup,
} from "../../process/OwnedWorkerRuntime.js";
import { prepareProcessOwnershipInspection } from "../../process/ProcessOwnershipObservation.js";
import {
@@ -168,14 +170,19 @@ export class JavaScriptAnalysisWorker {
}
};
readonly #workerCleanup: WorkerCleanupRegistry;
constructor(
readonly limits: JavaScriptSourceAnalysisLimits = DEFAULTS,
readonly launcher: typeof spawnOwnedProviderProcess = spawnOwnedProviderProcess,
sourceCount?: number,
workerCleanup: WorkerCleanupRegistry = processWorkerCleanup,
) {
this.#workerCleanup = workerCleanup;
this.#runtime = new OwnedWorkerRuntime({
rootPrefix: "rea-javascript-analysis-",
launcher,
cleanupRegistry: workerCleanup,
supervisorOptions: {
captureStdout: false,
maxDiagnosticBytes: 256 * 1024,
@@ -301,7 +308,7 @@ export class JavaScriptAnalysisWorker {
);
// Host ownership preparation is outside the parser execution deadline.
await prepareProcessOwnershipInspection(signal);
await OwnedWorkerRuntime.retryPending();
await this.#workerCleanup.retry();
deadline = new ProviderStartupDeadline(this.limits.timeoutMs, signal);
this.#deadline = deadline;
await this.#start(deadline);
src/process/OwnedWorkerRuntime.ts
+53 -21
View File
@@ -18,8 +18,35 @@ export interface OwnedWorkerRuntimeOptions {
readonly onInputFailure?: (cause: Error) => void;
/** Let a higher-level owner release its dependent resources before this root. */
readonly retryCleanup?: () => Promise<void>;
/** Host that retries this runtime. Omitted runtimes share the process registry. */
readonly cleanupRegistry?: WorkerCleanupRegistry;
}
/** Retained worker runtimes one host retries before admitting another child. */
export class WorkerCleanupRegistry {
readonly #pending = new Set<OwnedWorkerRuntime>();
has(runtime: OwnedWorkerRuntime): boolean {
return this.#pending.has(runtime);
}
add(runtime: OwnedWorkerRuntime): void {
this.#pending.add(runtime);
}
delete(runtime: OwnedWorkerRuntime): void {
this.#pending.delete(runtime);
}
async retry(except?: OwnedWorkerRuntime): Promise<void> {
for (const owner of [...this.#pending])
if (owner !== except) await owner.retryRetainedCleanup();
}
}
/** Runtimes that were not given a host registry. */
export const processWorkerCleanup = new WorkerCleanupRegistry();
export class OwnedWorkerRuntimeCleanupError extends Error {
readonly #cleanupEpoch: OwnedWorkerRuntimeCleanupEpoch | undefined;
@@ -92,7 +119,7 @@ class OwnedWorkerRuntimeCleanupEpoch {
/** Own the private root and process lifecycle shared by sequential workers. */
export class OwnedWorkerRuntime {
static readonly #pending = new Set<OwnedWorkerRuntime>();
readonly #registry: WorkerCleanupRegistry;
#root: PrivateRuntimeRoot | undefined;
#cleanupEpoch: OwnedWorkerRuntimeCleanupEpoch | undefined;
@@ -112,7 +139,9 @@ export class OwnedWorkerRuntime {
readonly listener: (chunk: Buffer | string) => void;
}>();
constructor(readonly options: OwnedWorkerRuntimeOptions) {}
constructor(readonly options: OwnedWorkerRuntimeOptions) {
this.#registry = options.cleanupRegistry ?? processWorkerCleanup;
}
get rootPath(): string | undefined {
return this.#root?.path;
@@ -135,7 +164,7 @@ export class OwnedWorkerRuntime {
/** Whether this exact owner still retains an unconfirmed resource release. */
get hasPendingCleanup(): boolean {
return OwnedWorkerRuntime.#pending.has(this);
return this.#registry.has(this);
}
/** First error observed on the current child's request stream, if any. */
@@ -155,13 +184,22 @@ export class OwnedWorkerRuntime {
/** Mark this still-owned runtime for retry before any consumer starts a child. */
retainForCleanup(): void {
OwnedWorkerRuntime.#pending.add(this);
this.#registry.add(this);
}
/** Retry every earlier worker owner before any consumer admits a new child. */
/** Retry every retained runtime in this host's registry. */
retryCohort(): Promise<void> {
return this.#registry.retry();
}
/** Called by the owning registry. Not a process-wide sweep. */
retryRetainedCleanup(): Promise<void> {
return this.#retryPending();
}
/** Retry runtimes left on the process registry by owners that named no host. */
static async retryPending(except?: OwnedWorkerRuntime): Promise<void> {
for (const owner of [...this.#pending])
if (owner !== except) await owner.#retryPending();
await processWorkerCleanup.retry(except);
}
/**
@@ -175,21 +213,15 @@ export class OwnedWorkerRuntime {
if (this.#terminalCloseRequested || this.#closing)
return Promise.reject(new Error("Owned worker runtime is closing"));
if (this.#acquisition !== undefined) return this.#acquisition;
if (
this.#supervisor !== undefined &&
!OwnedWorkerRuntime.#pending.has(this)
)
if (this.#supervisor !== undefined && !this.#registry.has(this))
return Promise.resolve();
if (
this.options.retryCleanup !== undefined &&
OwnedWorkerRuntime.#pending.has(this)
)
if (this.options.retryCleanup !== undefined && this.#registry.has(this))
return this.#retryOwnedCleanup().then(() => {
if (this.#terminalCloseRequested || this.#closing)
throw new Error("Owned worker runtime is closing");
return this.start(spawnOptions, initialize);
});
const ownRetry = OwnedWorkerRuntime.#pending.has(this)
const ownRetry = this.#registry.has(this)
? this.#retryPending()
: Promise.resolve();
const acquisition = ownRetry.then(async () => {
@@ -276,7 +308,7 @@ export class OwnedWorkerRuntime {
const attempt = Promise.resolve()
.then(() => this.options.retryCleanup?.())
.then(() => {
if (OwnedWorkerRuntime.#pending.has(this))
if (this.#registry.has(this))
throw this.#cleanupError(
"Higher-level cleanup retry returned while ownership remains pending",
);
@@ -297,7 +329,7 @@ export class OwnedWorkerRuntime {
| ((supervisor: ProviderProcessSupervisor) => Promise<void>)
| undefined,
): Promise<void> {
await OwnedWorkerRuntime.retryPending(this);
await this.#registry.retry(this);
if (this.#closing) throw new Error("Owned worker runtime is closing");
this.#root ??= await PrivateRuntimeRoot.create({
prefix: this.options.rootPrefix,
@@ -337,7 +369,7 @@ export class OwnedWorkerRuntime {
this.#detachStdout();
const stopped = await supervisor.stop();
if (stopped.status === "incomplete") {
OwnedWorkerRuntime.#pending.add(this);
this.#registry.add(this);
const stdin = supervisor.launch.process.stdin;
if (
stdin !== undefined &&
@@ -372,12 +404,12 @@ export class OwnedWorkerRuntime {
this.#cleanupEpoch?.release();
this.#cleanupEpoch = undefined;
} catch (cause: unknown) {
OwnedWorkerRuntime.#pending.add(this);
this.#registry.add(this);
throw this.#cleanupError(message(cause), cause);
}
}
this.#closed = true;
OwnedWorkerRuntime.#pending.delete(this);
this.#registry.delete(this);
}
#detachStdout(): void {
src/process/WorkerCleanupRegistry.test.ts
+27
View File
@@ -0,0 +1,27 @@
import { expect, it } from "vitest";
import {
OwnedWorkerRuntime,
WorkerCleanupRegistry,
} from "./OwnedWorkerRuntime.js";
it("retries one host's retained runtime without failing another host", async () => {
const first = new WorkerCleanupRegistry();
const second = new WorkerCleanupRegistry();
let fail = true;
const stuck = new OwnedWorkerRuntime({
rootPrefix: "rea-worker-host-",
cleanupRegistry: first,
retryCleanup: async () => {
if (fail) throw new Error("host still owns its worker");
},
});
stuck.retainForCleanup();
try {
await expect(second.retry()).resolves.toBeUndefined();
await expect(first.retry()).rejects.toThrow("host still owns its worker");
} finally {
fail = false;
await stuck.close();
}
});
tests/boundary/cli/cliOwnedResources.test.ts
+50
View File
@@ -0,0 +1,50 @@
import { afterEach, expect, it, vi } from "vitest";
import { ArtifactResourceScope } from "../../../src/artifacts/ArtifactResourceScope.js";
import { CliGuidanceResources } from "../../../src/cli/guidanceCommands.js";
import { CliJavaScriptAnalysisHost } from "../../../src/cli/javascriptApplicationAnalysis.js";
const originalClose = ArtifactResourceScope.prototype.close;
afterEach(() => {
ArtifactResourceScope.prototype.close = originalClose;
vi.restoreAllMocks();
});
it("keeps guidance descriptors on the host that created them", () => {
const first = new CliGuidanceResources();
const second = new CliGuidanceResources();
expect(first.resources()).toBe(first.resources());
expect(second.resources()).toBe(second.resources());
expect(first.resources()).not.toBe(second.resources());
});
it("does not share a JavaScript worker scope or its in-flight close", async () => {
let release = (): void => {};
const gate = new Promise<void>((resolve) => {
release = resolve;
});
ArtifactResourceScope.prototype.close = function (
this: ArtifactResourceScope,
) {
return gate.then(() => originalClose.call(this));
};
const first = new CliJavaScriptAnalysisHost();
const second = new CliJavaScriptAnalysisHost();
let firstClose: Promise<void> | undefined;
let secondClose: Promise<void> | undefined;
const firstScope = first.scope();
const secondScope = second.scope();
try {
expect(firstScope).not.toBe(secondScope);
firstClose = first.close();
expect(first.close()).toBe(firstClose);
secondClose = second.close();
expect(secondClose).not.toBe(firstClose);
} finally {
release();
await Promise.allSettled([firstClose, secondClose]);
}
expect(first.scope()).not.toBe(firstScope);
expect(second.scope()).not.toBe(secondScope);
});