mirror of
https://github.com/morluto/rea.git
synced 2026-11-12 01:41:21 +00:00
fix(binary): seal session admission during shutdown (#10012)
fix(session): seal binary work admission during shutdown
This commit is contained in:
1 parent
5b02119b69
commit
44ef32c1c0
2 files changed
+151
-5
No files matched your search
@@ -224,7 +224,10 @@ export class BinarySession
|
||||
options: Pick<ExecutionOptions, "signal"> = {},
|
||||
): Promise<Result<ResolvedAppBundleResourceRoot, AnalysisError>> {
|
||||
return this.#serialize(async (admittedThrough) => {
|
||||
const refused = this.#targetAcquisitionRefused("open_bundle_resources");
|
||||
const refused = this.#targetAcquisitionRefused(
|
||||
"open_bundle_resources",
|
||||
options.signal,
|
||||
);
|
||||
if (refused !== undefined) return refused;
|
||||
await this.#drainCalls(admittedThrough);
|
||||
if (isAborted(options.signal))
|
||||
@@ -530,7 +533,10 @@ export class BinarySession
|
||||
const attempt = async (): Promise<
|
||||
Result<BinaryTarget, AnalysisError>
|
||||
> => {
|
||||
const refused = this.#targetAcquisitionRefused("open_binary");
|
||||
const refused = this.#targetAcquisitionRefused(
|
||||
"open_binary",
|
||||
options.signal,
|
||||
);
|
||||
if (refused !== undefined) return refused;
|
||||
const admitted = await this.#withLoadedSnapshot(options);
|
||||
if (!admitted.ok) return admitted;
|
||||
@@ -790,6 +796,12 @@ export class BinarySession
|
||||
failure: AnalysisError | undefined,
|
||||
) => Promise<Result<null, AnalysisError>>,
|
||||
): Promise<Result<Result<Value, AnalysisError>, AnalysisError>> {
|
||||
if (this.#shutdownStarted)
|
||||
return Promise.resolve(
|
||||
err(
|
||||
new AnalysisCancelledError(operationName, { cause: signal?.reason }),
|
||||
),
|
||||
);
|
||||
return this.#serialize(async (admittedThrough) => {
|
||||
if (this.#activeBundleResources !== undefined)
|
||||
return err(
|
||||
@@ -921,9 +933,12 @@ export class BinarySession
|
||||
|
||||
#targetAcquisitionRefused(
|
||||
operation: "open_binary" | "open_bundle_resources",
|
||||
signal?: AbortSignal,
|
||||
): Result<never, AnalysisCancelledError> | undefined {
|
||||
if (!this.#shutdownStarted) return undefined;
|
||||
return err(new AnalysisCancelledError(operation));
|
||||
return err(
|
||||
new AnalysisCancelledError(operation, { cause: signal?.reason }),
|
||||
);
|
||||
}
|
||||
|
||||
/** Release the engine and replay state while retaining investigation history. */
|
||||
@@ -1004,7 +1019,11 @@ export class BinarySession
|
||||
): Promise<Result<ResetInvestigationReceipt, AnalysisError>> {
|
||||
if (this.#shutdownStarted)
|
||||
return Promise.resolve(
|
||||
err(new AnalysisCancelledError("reset_investigation")),
|
||||
err(
|
||||
new AnalysisCancelledError("reset_investigation", {
|
||||
cause: options.signal?.reason,
|
||||
}),
|
||||
),
|
||||
);
|
||||
return this.#serialize(async (admittedThrough) => {
|
||||
await this.#drainCalls(admittedThrough);
|
||||
@@ -1410,6 +1429,12 @@ export class BinarySession
|
||||
options?: ExecutionOptions,
|
||||
recoveryParameters?: Readonly<Record<string, JsonValue>>,
|
||||
): Promise<Result<AnalysisExecution, AnalysisError>> {
|
||||
if (this.#shutdownStarted)
|
||||
return Promise.resolve(
|
||||
err(
|
||||
new AnalysisCancelledError(name, { cause: options?.signal?.reason }),
|
||||
),
|
||||
);
|
||||
const generation = this.#transitionGeneration;
|
||||
const call = this.#execute(
|
||||
name,
|
||||
@@ -1430,6 +1455,7 @@ export class BinarySession
|
||||
): Promise<Result<AnalysisExecution, AnalysisError> | undefined> {
|
||||
const active = this.#active;
|
||||
if (
|
||||
this.#shutdownStarted ||
|
||||
this.#transitionPending > 0 ||
|
||||
active === undefined ||
|
||||
active.client.executeIfReady === undefined
|
||||
@@ -1461,7 +1487,11 @@ export class BinarySession
|
||||
operation: (analysis: AnalysisOperationPort) => Promise<Value>,
|
||||
): Promise<Result<Value, AnalysisError>> {
|
||||
if (this.#shutdownStarted)
|
||||
return Promise.resolve(err(new AnalysisCancelledError(operationName)));
|
||||
return Promise.resolve(
|
||||
err(
|
||||
new AnalysisCancelledError(operationName, { cause: signal?.reason }),
|
||||
),
|
||||
);
|
||||
const generation = this.#transitionGeneration;
|
||||
const transition = this.#transition;
|
||||
const call = (async (): Promise<Result<Value, AnalysisError>> => {
|
||||
|
||||
@@ -570,3 +570,119 @@ it("retains exact cancellation reason and a complete checkpoint during final cle
|
||||
await session.dispose();
|
||||
}
|
||||
});
|
||||
|
||||
it.each(
|
||||
(
|
||||
[
|
||||
"execute",
|
||||
"direct observation",
|
||||
"external mutation",
|
||||
"composed analysis",
|
||||
"reset",
|
||||
"open binary",
|
||||
"open resources",
|
||||
] as const
|
||||
).flatMap((entrypoint) =>
|
||||
[false, true].map((aborted) => ({ entrypoint, aborted })),
|
||||
),
|
||||
)(
|
||||
"seals new $entrypoint admission at shutdown with caller aborted=$aborted",
|
||||
async ({ entrypoint, aborted }) => {
|
||||
const [target] = await createBinarySessionTargets();
|
||||
const calls: string[] = [];
|
||||
const session = createTestBinarySession(createCacheProvider(calls));
|
||||
const entered = createDeferred<void>();
|
||||
const release = createDeferred<void>();
|
||||
const parameters = { address: "0x1000", document: "fixture" };
|
||||
const mutation = vi.fn(async () => ok(null));
|
||||
let earlier: Promise<unknown> | undefined;
|
||||
try {
|
||||
const opened = await session.open(target);
|
||||
if (!opened.ok) throw opened.error;
|
||||
earlier = session.withAdmittedAnalysis(
|
||||
"earlier",
|
||||
undefined,
|
||||
async (analysis) => {
|
||||
entered.resolve();
|
||||
await release.promise;
|
||||
return analysis.execute("address_name", parameters);
|
||||
},
|
||||
);
|
||||
await awaitAcknowledgement(
|
||||
entered.promise,
|
||||
earlier,
|
||||
"earlier analysis admission",
|
||||
);
|
||||
const controller = new AbortController();
|
||||
const reason = { caller: "opaque shutdown cancellation", entrypoint };
|
||||
if (aborted) controller.abort(reason);
|
||||
session.beginShutdown();
|
||||
const options = { signal: controller.signal };
|
||||
const late = {
|
||||
execute: () => session.execute("address_name", parameters, options),
|
||||
"direct observation": () =>
|
||||
session.executeForDirectObservation(
|
||||
"address_name",
|
||||
parameters,
|
||||
parameters,
|
||||
options,
|
||||
),
|
||||
"external mutation": () =>
|
||||
session.withAdmittedExternalMutation(
|
||||
undefined,
|
||||
controller.signal,
|
||||
mutation,
|
||||
(_value, error) => error,
|
||||
),
|
||||
"composed analysis": () =>
|
||||
session.withAdmittedAnalysis("late", controller.signal, mutation),
|
||||
reset: () => session.resetInvestigation(options),
|
||||
"open binary": () => session.open(target, options),
|
||||
"open resources": () => session.openBundleResources(target, options),
|
||||
}[entrypoint]();
|
||||
release.resolve();
|
||||
const result = await late;
|
||||
if (result.ok) throw new Error("Expected shutdown admission refusal");
|
||||
expect(result.error._tag).toBe("AnalysisCancelledError");
|
||||
expect(result.error.cause).toBe(aborted ? reason : undefined);
|
||||
expect(mutation).not.toHaveBeenCalled();
|
||||
expect(await earlier).toMatchObject({ ok: true, value: { ok: true } });
|
||||
expect(calls.filter((name) => name === "address_name")).toHaveLength(1);
|
||||
} finally {
|
||||
release.resolve();
|
||||
await earlier;
|
||||
await session.dispose();
|
||||
}
|
||||
},
|
||||
);
|
||||
|
||||
it("stops prompt completion dispatch as soon as shutdown seals admission", async () => {
|
||||
const [target] = await createBinarySessionTargets();
|
||||
const calls: string[] = [];
|
||||
const base = createCacheProvider(calls);
|
||||
const session = createTestBinarySession({
|
||||
...base,
|
||||
createClient: (...arguments_) => {
|
||||
const client = base.createClient(...arguments_);
|
||||
return { ...client, executeIfReady: client.execute.bind(client) };
|
||||
},
|
||||
});
|
||||
try {
|
||||
const opened = await session.open(target);
|
||||
if (!opened.ok) throw opened.error;
|
||||
// Instantiate the real composite's child before exercising its ready path.
|
||||
const initial = await session.execute("address_name", {});
|
||||
if (!initial.ok) throw initial.error;
|
||||
expect(
|
||||
await session.executeIfProviderReady("address_name", {}),
|
||||
).toMatchObject({ ok: true });
|
||||
const before = calls.length;
|
||||
session.beginShutdown();
|
||||
expect(
|
||||
await session.executeIfProviderReady("address_name", {}),
|
||||
).toBeUndefined();
|
||||
expect(calls).toHaveLength(before);
|
||||
} finally {
|
||||
await session.dispose();
|
||||
}
|
||||
});
|
||||
Reference in new issue
Block a user