fix(browser): bind CDP cleanup failures to exact transport leases

Refs #4467.

Retain failed socket owners across translated failures and resolve aggregate cleanup only after every required transport closes. Preserve concurrently active peer sessions.
This commit is contained in:
morluto committed 2026-10-17 00:44:02 +08:00
1 parent effb71f59c
commit ce0d6790ed
6 files changed
+350 -64

No files matched your search

+24 -15
View File
@@ -78,7 +78,10 @@ import { comparePngScreenshots } from "./PngVisualDiff.js";
import { CDP_BROWSER_PROVIDER_IDENTITY } from "./providerIdentities.js";
import { compareUnicodeCodePoints } from "../domain/unicodeCodePointOrder.js";
import { CdpTargetSessionOwner } from "./CdpTargetSessionOwner.js";
import {
CdpTargetSessionOwner,
bindCdpTransportCleanup,
} from "./CdpTargetSessionOwner.js";
const CLEANUP_DOMAINS = [
"Network",
"Debugger",
@@ -421,10 +424,13 @@ const withTargetSession = async <
if (cleanupFailure !== undefined) {
if (!outcome.ok && outcome.error instanceof AnalysisError)
throw analysisErrorWithCleanupFailure(
outcome.error,
throw bindCdpTransportCleanup(
cleanupFailure,
operation,
analysisErrorWithCleanupFailure(
outcome.error,
cleanupFailure,
operation,
),
);
if (!outcome.ok)
throw new AggregateError(
@@ -432,17 +438,20 @@ const withTargetSession = async <
`CDP operation ${operation} failed and cleanup remains incomplete.`,
{ cause: outcome.error },
);
throw new ProviderCleanupError(
"rea-cdp-browser",
cleanupFailure.cleanupResources,
{ cleanup_error: cleanupFailure.message },
{
operation,
cause: cleanupFailure,
...(partialObservation === undefined
? {}
: { partialObservation: partialObservation(outcome.value) }),
},
throw bindCdpTransportCleanup(
cleanupFailure,
new ProviderCleanupError(
"rea-cdp-browser",
cleanupFailure.cleanupResources,
{ cleanup_error: cleanupFailure.message },
{
operation,
cause: cleanupFailure,
...(partialObservation === undefined
? {}
: { partialObservation: partialObservation(outcome.value) }),
},
),
);
}
if (!outcome.ok) throw outcome.error;
+31 -19
View File
@@ -34,7 +34,10 @@ import { CDP_ELECTRON_PROVIDER_IDENTITY } from "./providerIdentities.js";
import { compareUnicodeCodePoints } from "../domain/unicodeCodePointOrder.js";
import { analysisErrorWithCleanupFailure } from "../domain/analysisErrorCleanup.js";
import { ProviderCleanupError } from "../domain/providerCleanupError.js";
import { CdpTargetSessionOwner } from "./CdpTargetSessionOwner.js";
import {
CdpTargetSessionOwner,
bindCdpTransportCleanup,
} from "./CdpTargetSessionOwner.js";
/** Passive Electron provider for local file pages exposed by loopback CDP. */
export class CdpElectronProvider implements ElectronObservationPort {
readonly #owner = new CdpTargetSessionOwner();
@@ -194,34 +197,43 @@ export class CdpElectronProvider implements ElectronObservationPort {
if (cleanupFailure !== undefined) {
if (finalCancellation !== undefined)
return err(
analysisErrorWithCleanupFailure(
finalCancellation,
bindCdpTransportCleanup(
cleanupFailure,
"inspect_electron_page",
analysisErrorWithCleanupFailure(
finalCancellation,
cleanupFailure,
"inspect_electron_page",
),
),
);
if (!result.ok)
return err(
analysisErrorWithCleanupFailure(
result.error,
bindCdpTransportCleanup(
cleanupFailure,
"inspect_electron_page",
analysisErrorWithCleanupFailure(
result.error,
cleanupFailure,
"inspect_electron_page",
),
),
);
return err(
new ProviderCleanupError(
CDP_ELECTRON_PROVIDER_IDENTITY.id,
cleanupFailure.cleanupResources,
{ cleanup_error: cleanupFailure.message },
{
operation: "inspect_electron_page",
cause: cleanupFailure,
partialObservation: {
kind: "electron-page-observation",
inspection: result.value,
cleanup: "incomplete",
bindCdpTransportCleanup(
cleanupFailure,
new ProviderCleanupError(
CDP_ELECTRON_PROVIDER_IDENTITY.id,
cleanupFailure.cleanupResources,
{ cleanup_error: cleanupFailure.message },
{
operation: "inspect_electron_page",
cause: cleanupFailure,
partialObservation: {
kind: "electron-page-observation",
inspection: result.value,
cleanup: "incomplete",
},
},
},
),
),
);
}
+8 -2
View File
@@ -4,7 +4,10 @@ import { analysisErrorWithCleanupFailure } from "../domain/analysisErrorCleanup.
import { BrowserObservationError } from "../domain/browserObservationError.js";
import type { BrowserObservationOperation } from "../domain/browserObservationErrors.js";
import { CdpConnection } from "./CdpConnection.js";
import { CdpTargetSessionOwner } from "./CdpTargetSessionOwner.js";
import {
CdpTargetSessionOwner,
bindCdpTransportCleanup,
} from "./CdpTargetSessionOwner.js";
import {
cdpTargetWebSocket,
discoverCdpBrowserSocket,
@@ -157,7 +160,10 @@ const combineOpenFailure = (
operation: BrowserObservationOperation,
): unknown =>
primary instanceof AnalysisError && cleanup instanceof AnalysisError
? analysisErrorWithCleanupFailure(primary, cleanup, operation)
? bindCdpTransportCleanup(
cleanup,
analysisErrorWithCleanupFailure(primary, cleanup, operation),
)
: new AggregateError(
[primary, cleanup],
`CDP operation ${operation} failed and its transport cleanup is incomplete.`,
+63 -27
View File
@@ -1,3 +1,8 @@
import {
AnalysisCleanupLease,
AnalysisCleanupLeaseGroup,
} from "../artifacts/AnalysisCleanupLease.js";
import type { JsonValue } from "../domain/jsonValue.js";
import { AnalysisError } from "../domain/analysisErrorBase.js";
import type { BrowserObservationOperation } from "../domain/browserObservationErrors.js";
import { ProviderCleanupError } from "../domain/providerCleanupError.js";
@@ -6,10 +11,40 @@ import { analysisErrorWithCleanupFailure } from "../domain/analysisErrorCleanup.
import { CdpConnection } from "./CdpConnection.js";
import type { ExecutionOptions } from "../application/AnalysisProvider.js";
/** Typed ownership facts shared by transport errors and their adapter projections. */
class CdpTransportCleanupFailure extends ProviderCleanupError {
constructor(
readonly cleanupLeases: readonly AnalysisCleanupLease[],
failure: {
readonly resources: readonly string[];
readonly operation: string;
readonly diagnostics: Readonly<Record<string, JsonValue>>;
readonly cause: unknown;
},
) {
super("rea-cdp-browser", failure.resources, failure.diagnostics, {
operation: failure.operation,
cause: failure.cause,
});
new AnalysisCleanupLeaseGroup(this, cleanupLeases);
}
}
/** Bind a translated error without interpreting diagnostic cause chains. */
export const bindCdpTransportCleanup = <Failure extends AnalysisError>(
cause: unknown,
failure: Failure,
): Failure => {
if (cause instanceof CdpTransportCleanupFailure)
new AnalysisCleanupLeaseGroup(failure, cause.cleanupLeases);
return failure;
};
interface OwnedConnection {
readonly connection: CdpConnection;
readonly operation: BrowserObservationOperation;
readonly resource: string;
readonly cleanupLease: AnalysisCleanupLease;
active: boolean;
readonly settled: Promise<void>;
settle: () => void;
@@ -75,7 +110,10 @@ export class CdpTargetSessionOwner {
cause instanceof AnalysisError &&
cleanupCause instanceof AnalysisError
)
throw analysisErrorWithCleanupFailure(cause, cleanupCause, operation);
throw bindCdpTransportCleanup(
cleanupCause,
analysisErrorWithCleanupFailure(cause, cleanupCause, operation),
);
throw new AggregateError(
[cause, cleanupCause],
`CDP operation ${operation} failed during handshake and cleanup remains incomplete.`,
@@ -106,6 +144,7 @@ export class CdpTargetSessionOwner {
connection,
operation,
resource,
cleanupLease: new AnalysisCleanupLease(() => connection.close()),
active: true,
settled,
settle,
@@ -129,7 +168,7 @@ export class CdpTargetSessionOwner {
record.cleanupAttempted = true;
await record.cleanup();
}
await connection.close();
await record.cleanupLease.release();
this.#connections.delete(connection);
} catch (cause: unknown) {
const failure = cleanupError(record, cause);
@@ -153,11 +192,11 @@ export class CdpTargetSessionOwner {
/** Retry retained failures without touching concurrently active sessions. */
async retryPending(): Promise<void> {
const failures: AnalysisError[] = [];
const failures: CdpTransportCleanupFailure[] = [];
for (const record of this.#connections.values()) {
if (record.active || record.failure === undefined) continue;
try {
await record.connection.close();
await record.cleanupLease.release();
this.#connections.delete(record.connection);
} catch (cause: unknown) {
const failure = cleanupError(record, cause);
@@ -179,10 +218,10 @@ export class CdpTargetSessionOwner {
if (active.length === 0) break;
await Promise.all(active.map(({ settled }) => settled));
}
const failures: AnalysisError[] = [];
const failures: CdpTransportCleanupFailure[] = [];
for (const record of this.#connections.values()) {
try {
await record.connection.close();
await record.cleanupLease.release();
this.#connections.delete(record.connection);
} catch (cause: unknown) {
const failure = cleanupError(record, cause);
@@ -198,11 +237,12 @@ export class CdpTargetSessionOwner {
const cleanupError = (
record: OwnedConnection,
cause: unknown,
): ProviderCleanupError =>
new ProviderCleanupError(
"rea-cdp-browser",
[record.resource],
{
): CdpTransportCleanupFailure =>
new CdpTransportCleanupFailure([record.cleanupLease], {
resources: [record.resource],
operation: record.operation,
cause,
diagnostics: {
reason:
cause instanceof AnalysisError
? (cause.userMessage ?? cause.message)
@@ -210,29 +250,25 @@ const cleanupError = (
? cause.message
: String(cause),
},
{
operation: record.operation,
cause,
},
);
});
const aggregateCleanupFailures = (
failures: readonly AnalysisError[],
): ProviderCleanupError =>
new ProviderCleanupError(
"rea-cdp-browser",
failures.flatMap(({ cleanupResources }) => cleanupResources),
{
failures: failures.map(({ message, diagnostics }) => ({
message,
diagnostics: diagnostics ?? {},
})),
},
failures: readonly CdpTransportCleanupFailure[],
): CdpTransportCleanupFailure =>
new CdpTransportCleanupFailure(
failures.flatMap((failure) => failure.cleanupLeases),
{
resources: failures.flatMap(({ cleanupResources }) => cleanupResources),
operation: "close_cdp_transports",
cause: new AggregateError(
failures,
"One or more REA-owned CDP transports remain open.",
),
diagnostics: {
failures: failures.map(({ message, diagnostics }) => ({
message,
diagnostics: diagnostics ?? {},
})),
},
},
);
@@ -4,6 +4,11 @@ import { pathToFileURL } from "node:url";
import { expect, it, onTestFinished, vi } from "vitest";
import { CdpCommandRejection } from "../../../src/browser/CdpCommandRejection.js";
import { CdpConnection } from "../../../src/browser/CdpConnection.js";
import { AnalysisError } from "../../../src/domain/analysisErrorBase.js";
import { AnalysisCancelledError } from "../../../src/domain/analysisErrorCore.js";
import { inspectWebPageInputSchema } from "../../../src/domain/browserObservation.js";
import { CdpBrowserProvider } from "../../../src/browser/CdpBrowserProvider.js";
import { CdpElectronProvider } from "../../../src/browser/CdpElectronProvider.js";
import { observeWebSessionInputSchema } from "../../../src/domain/browserSession.js";
@@ -234,3 +239,101 @@ it("classifies cancellation that arrives during final Electron detach", async ()
await vi.waitFor(() => expect(browser.activeConnectionCount()).toBe(0));
expect((await fetch(`${browser.endpoint}/json/version`)).ok).toBe(true);
});
it.each(
(["browser", "electron"] as const).flatMap((kind) =>
(["completed", "failed", "attach-failed"] as const).map((outcome) => ({
kind,
outcome,
})),
),
)(
"resolves $kind $outcome inspection and cleanup composition after confirmed socket retry",
async ({ kind, outcome }) => {
const root = await createTestTempDirectory("rea-cdp-cleanup-binding-");
const page = join(root, "index.html");
await writeFile(page, "<main>completed fixture</main>");
const browser = await startFakeCdpBrowser({
...(kind === "electron"
? { electronFileUrl: pathToFileURL(page).href }
: {}),
commandError(command) {
return (outcome === "failed" &&
command.method === "Page.getFrameTree") ||
(outcome === "attach-failed" &&
command.method === "Target.attachToTarget")
? { code: -32_000, message: "controlled primary frame read failure" }
: undefined;
},
});
const provider =
kind === "browser" ? new CdpBrowserProvider() : new CdpElectronProvider();
const controller = new AbortController();
const reason = new Error("cancel during failed transport close");
const originalClose = CdpConnection.prototype.close;
let denied = true;
const closing = vi
.spyOn(CdpConnection.prototype, "close")
.mockImplementation(function (this: CdpConnection) {
if (!denied) return originalClose.call(this);
controller.abort(reason);
return Promise.reject(new Error("controlled transport close failure"));
});
onTestFinished(async () => {
closing.mockRestore();
await provider.close();
await browser.close();
await rm(root, { recursive: true, force: true });
});
const result =
provider instanceof CdpBrowserProvider
? await provider.inspectPage(
inspectWebPageInputSchema.parse({
cdp_endpoint: browser.endpoint,
target_id: "allowed-page",
allowed_origins: [browser.allowedOrigin],
observation_ms: 0,
}),
{ signal: controller.signal },
)
: await provider.inspectPage(
inspectElectronPageInputSchema.parse({
cdp_endpoint: browser.endpoint,
target_id: "electron-page",
observation_ms: 0,
}),
{ signal: controller.signal },
);
if (result.ok)
throw new Error("Expected cancelled inspection with unresolved cleanup");
expect(result.error.cleanupIncomplete).toBe(true);
if (outcome === "completed")
expect(result.error.partialObservation).toBeDefined();
const observation = result.error.partialObservation;
const aggregate = result.error.cause;
if (!(aggregate instanceof AggregateError))
throw new Error("Missing cancellation/cleanup aggregate");
const primary = aggregate.errors[0];
if (outcome === "completed") {
expect(primary).toBeInstanceOf(AnalysisCancelledError);
expect(primary.cause).toBe(reason);
} else {
expect(primary).toBeInstanceOf(CdpCommandRejection);
expect(primary.reportedMessage).toBe(
"controlled primary frame read failure",
);
}
const primaryCause = primary.cause;
const retry = await provider.close().catch((cause: unknown) => cause);
if (!(retry instanceof AnalysisError))
throw new Error("Missing typed retry failure");
expect(retry.cleanupIncomplete).toBe(true);
denied = false;
await provider.close();
await vi.waitFor(() => expect(browser.activeConnectionCount()).toBe(0));
expect(result.error.cleanupIncomplete).toBe(false);
expect(retry.cleanupIncomplete).toBe(false);
expect(result.error.partialObservation).toBe(observation);
expect(primary.cause).toBe(primaryCause);
},
);
@@ -4,7 +4,10 @@ import { WebSocket, WebSocketServer } from "ws";
import { expect, it, onTestFinished, vi } from "vitest";
import { CdpTargetSessionOwner } from "../../../src/browser/CdpTargetSessionOwner.js";
import type { CdpConnection } from "../../../src/browser/CdpConnection.js";
import { CdpConnection } from "../../../src/browser/CdpConnection.js";
import { CdpBrowserProvider } from "../../../src/browser/CdpBrowserProvider.js";
import { inspectWebPageInputSchema } from "../../../src/domain/browserObservation.js";
import { AnalysisError } from "../../../src/domain/analysisErrorBase.js";
import { ProviderCleanupError } from "../../../src/domain/providerCleanupError.js";
import { startFakeCdpBrowser } from "../../fixtures/fakeCdpBrowser.js";
@@ -156,3 +159,120 @@ it("accepts asynchronous terminate success only after the close event", async ()
await owner.close();
await vi.waitFor(() => expect(browser.activeConnectionCount()).toBe(0));
});
it("clears returned inspection and shutdown cleanup status only after socket release", async () => {
const browser = await startFakeCdpBrowser();
const provider = new CdpBrowserProvider();
const originalClose = CdpConnection.prototype.close;
let denied = true;
const closing = vi
.spyOn(CdpConnection.prototype, "close")
.mockImplementation(function (this: CdpConnection) {
return denied
? Promise.reject(new Error("controlled transport close failure"))
: originalClose.call(this);
});
onTestFinished(async () => {
closing.mockRestore();
await provider.close();
await browser.close();
});
const result = await provider.inspectPage(
inspectWebPageInputSchema.parse({
cdp_endpoint: browser.endpoint,
allowed_origins: [browser.allowedOrigin],
target_id: "allowed-page",
observation_ms: 0,
}),
);
if (result.ok) throw new Error("Expected retained socket cleanup failure");
expect(result.error.cleanupIncomplete).toBe(true);
expect(result.error.partialObservation).toBeDefined();
const observation = result.error.partialObservation;
const retry = await provider.close().catch((cause: unknown) => cause);
expect(retry).toBeInstanceOf(AnalysisError);
if (!(retry instanceof AnalysisError))
throw new Error("Missing retry failure");
expect(retry.cleanupIncomplete).toBe(true);
expect(browser.activeConnectionCount()).toBe(1);
denied = false;
await provider.close();
await vi.waitFor(() => expect(browser.activeConnectionCount()).toBe(0));
expect(result.error.cleanupIncomplete).toBe(false);
expect(retry.cleanupIncomplete).toBe(false);
expect(result.error.partialObservation).toBe(observation);
expect(browser.commands.map(({ method }) => method)).not.toContain(
"Browser.close",
);
expect(browser.commands.map(({ method }) => method)).not.toContain(
"Target.closeTarget",
);
});
it("keeps retry aggregates incomplete until every reported socket is released", async () => {
const browser = await startFakeCdpBrowser();
const owner = new CdpTargetSessionOwner();
const first = await owner.connect(
browser.browserWebSocketUrl,
"inspect_web_page",
{},
"first-socket",
);
const second = await owner.connect(
browser.browserWebSocketUrl,
"inspect_web_page",
{},
"second-socket",
);
const active = await owner.connect(
browser.browserWebSocketUrl,
"inspect_web_page",
{},
"active-peer",
);
const originalClose = CdpConnection.prototype.close;
const denied = new Set([first, second]);
const closing = vi
.spyOn(CdpConnection.prototype, "close")
.mockImplementation(function (this: CdpConnection) {
return denied.has(this)
? Promise.reject(new Error("controlled mixed close failure"))
: originalClose.call(this);
});
onTestFinished(async () => {
closing.mockRestore();
await owner.finish(active);
await owner.close();
await browser.close();
});
const errors = await Promise.all(
[first, second].map((connection) =>
owner.finish(connection).catch((cause: unknown) => cause),
),
);
const aggregate = await owner.retryPending().catch((cause: unknown) => cause);
if (
!(aggregate instanceof AnalysisError) ||
!errors.every((error) => error instanceof AnalysisError)
)
throw new Error("Missing typed cleanup errors");
expect(aggregate.cleanupResources).toEqual(["first-socket", "second-socket"]);
denied.delete(first);
const mixed = await owner.retryPending().catch((cause: unknown) => cause);
if (!(mixed instanceof AnalysisError))
throw new Error("Missing mixed retry error");
expect(errors.map((error) => error.cleanupIncomplete)).toEqual([false, true]);
expect(aggregate.cleanupIncomplete).toBe(true);
expect(mixed.cleanupResources).toEqual(["second-socket"]);
await vi.waitFor(() => expect(browser.activeConnectionCount()).toBe(2));
denied.clear();
await owner.retryPending();
expect(errors.map((error) => error.cleanupIncomplete)).toEqual([
false,
false,
]);
expect(aggregate.cleanupIncomplete).toBe(false);
expect(mixed.cleanupIncomplete).toBe(false);
await vi.waitFor(() => expect(browser.activeConnectionCount()).toBe(1));
await owner.finish(active);
});