fix(cancellation): retain caller reasons across digest and startup

This commit is contained in:
morluto
2026-10-25 07:30:50 +08:00
parent 9140deae63
commit f2fa951c00
5 changed files with 99 additions and 20 deletions
@@ -1057,7 +1057,11 @@ export const digestFile = async (
remainingMs,
);
const abort = (): void => {
stream.destroy(new AnalysisCancelledError("artifact_digest"));
stream.destroy(
new AnalysisCancelledError("artifact_digest", {
cause: signal?.reason,
}),
);
};
const cleanup = (): void => {
clearTimeout(timer);
@@ -1066,7 +1070,11 @@ export const digestFile = async (
if (signal?.aborted) {
cleanup();
stream.destroy();
rejectDigest(new AnalysisCancelledError("artifact_digest"));
rejectDigest(
new AnalysisCancelledError("artifact_digest", {
cause: signal.reason,
}),
);
return;
}
signal?.addEventListener("abort", abort, { once: true });
@@ -1139,7 +1147,9 @@ const statWithDeadline = (
timeoutMs: number,
): Promise<Stats> => {
if (signal?.aborted)
return Promise.reject(new AnalysisCancelledError("artifact_digest"));
return Promise.reject(
new AnalysisCancelledError("artifact_digest", { cause: signal.reason }),
);
return new Promise<Stats>((resolveStat, rejectStat) => {
let settled = false;
const finish = (outcome: Result<Stats, unknown>): void => {
@@ -1151,7 +1161,13 @@ const statWithDeadline = (
else rejectStat(outcome.error);
};
const abort = (): void =>
finish(err(new AnalysisCancelledError("artifact_digest")));
finish(
err(
new AnalysisCancelledError("artifact_digest", {
cause: signal?.reason,
}),
),
);
const timer = setTimeout(
() =>
finish(
+18 -9
View File
@@ -124,8 +124,12 @@ interface ServeInputs {
readonly served: ServedState;
}
const cancelledServe = (): JsonValue =>
cliError(new AnalysisCancelledError(CLI_COMMANDS.environmentServe));
const cancelledServe = (signal: AbortSignal): JsonValue =>
cliError(
new AnalysisCancelledError(CLI_COMMANDS.environmentServe, {
cause: signal.reason,
}),
);
/** Pure replay, as `--replay-only` asks for: no state is derived. */
const REPLAY_ONLY: ServedState = {
@@ -149,14 +153,14 @@ const loadServeInputs = async (
if (!evidence.ok) return evidence;
const { loadEnvironmentSource } =
await import("../application/environment/environmentSource.js");
if (signal.aborted) return err(cancelledServe());
if (signal.aborted) return err(cancelledServe(signal));
const source = loadEnvironmentSource(manifest.value, evidence.value);
if (!source.ok) return err(cliError(source.error));
if (input.replayOnly)
return ok({ source: source.value, served: REPLAY_ONLY });
const { prepareServedState } =
await import("../application/environment/environmentState.js");
if (signal.aborted) return err(cancelledServe());
if (signal.aborted) return err(cancelledServe(signal));
return ok({
source: source.value,
served: prepareServedState(source.value, evidence.value, {
@@ -172,7 +176,7 @@ const serveEnvironmentCommand = async (
logger: Logger,
signal: AbortSignal,
): Promise<JsonValue> => {
if (signal.aborted) return cancelledServe();
if (signal.aborted) return cancelledServe(signal);
const endpoint = parseMcpHttpEndpoint(environment, {
...(input.host === undefined ? {} : { host: input.host }),
...(input.port === undefined ? {} : { port: input.port }),
@@ -192,13 +196,13 @@ const serveEnvironmentCommand = async (
{ code: report.code, error: report.message },
"Environment state unavailable; serving recorded replay",
);
if (signal.aborted) return cancelledServe();
if (signal.aborted) return cancelledServe(signal);
// Loaded here so registering the command catalog never loads the HTTP listener.
const { serveEnvironment } =
await import("../server/environment/serveEnvironment.js");
const { StreamableHttpListenError } =
await import("../server/http/streamableHttpListenError.js");
if (signal.aborted) return cancelledServe();
if (signal.aborted) return cancelledServe(signal);
// A listener that fails after binding ends the command instead of serving nothing.
let failListener: (error: StreamableHttpListenError) => void = () =>
undefined;
@@ -221,7 +225,8 @@ const serveEnvironmentCommand = async (
if (!started.ok) return cliError(started.error);
served = started.value;
} catch (cause: unknown) {
if (signal.aborted && cause === signal.reason) return cancelledServe();
if (signal.aborted && cause === signal.reason)
return cancelledServe(signal);
if (!(cause instanceof StreamableHttpListenError)) throw cause;
return cliError(transportError(endpoint.value, cause));
}
@@ -449,7 +454,11 @@ const buildEnvironmentCommand = async (
capture = read.value;
}
if (signal.aborted)
return cliError(new AnalysisCancelledError(CLI_COMMANDS.environmentBuild));
return cliError(
new AnalysisCancelledError(CLI_COMMANDS.environmentBuild, {
cause: signal.reason,
}),
);
const captureInput: EnvironmentBuildCaptureInput = kind.value;
const built = await buildEnvironment({
capture,
+19 -5
View File
@@ -696,7 +696,11 @@ export class GdbSessionManager {
>
> {
if (signal?.aborted)
return err(new AnalysisCancelledError("start_gdb_session"));
return err(
new AnalysisCancelledError("start_gdb_session", {
cause: signal.reason,
}),
);
const occupied = this.#sessions.size + this.#sessionReservations;
if (occupied >= this.#maxSessions)
return err(
@@ -896,7 +900,11 @@ export class GdbSessionManager {
session.startupResolve = resolve;
session.startupReject = reject;
startupAbortListener = () =>
reject(new AnalysisCancelledError("start_gdb_session"));
reject(
new AnalysisCancelledError("start_gdb_session", {
cause: signal?.reason,
}),
);
signal?.addEventListener("abort", startupAbortListener, {
once: true,
});
@@ -923,7 +931,9 @@ export class GdbSessionManager {
startupTimer.unref();
});
if (signal?.aborted)
throw new AnalysisCancelledError("start_gdb_session");
throw new AnalysisCancelledError("start_gdb_session", {
cause: signal.reason,
});
const autoLoad = await this.#request(
session,
"-gdb-set auto-load off",
@@ -977,11 +987,15 @@ export class GdbSessionManager {
session.inferiorTtyVerified = true;
}
if (signal?.aborted)
throw new AnalysisCancelledError("start_gdb_session");
throw new AnalysisCancelledError("start_gdb_session", {
cause: signal.reason,
});
// Give queued exit notifications a turn before publishing startup.
await new Promise<void>((resolve) => setImmediate(resolve));
if (signal?.aborted)
throw new AnalysisCancelledError("start_gdb_session");
throw new AnalysisCancelledError("start_gdb_session", {
cause: signal.reason,
});
if (session.exited || session.protocolError !== undefined)
throw new Error(
session.protocolError ?? "GDB exited during startup readiness.",
@@ -1970,19 +1970,23 @@ it("handles a start signal that becomes aborted during listener registration", a
it("handles cancellation after the automatic-loading response", async () => {
const controller = new AbortController();
const reason = { source: "caller after automatic loading" };
const fake = createFakeSpawner((child, token, command) => {
child.stdout.write(
command.includes("list-thread-groups")
? `${token}^done,groups=[]\n`
: `${token}^done\n`,
);
if (command.includes("auto-load off")) controller.abort();
if (command.includes("auto-load off")) controller.abort(reason);
});
const manager = managerWithFakeSpawner(fake.spawnProcess);
const started = await manager.start(controller.signal);
expect(started.ok).toBe(false);
if (!started.ok) expect(started.error._tag).toBe("AnalysisCancelledError");
if (!started.ok) {
expect(started.error._tag).toBe("AnalysisCancelledError");
expect(started.error.cause).toBe(reason);
}
});
it("reports an MI stream failure when the child has no stdin pipe", async () => {
@@ -12,6 +12,7 @@ import { TestReverseEngineeringService as ReverseEngineeringService } from "../s
import { providerSnapshot } from "../support/reverseEngineeringCompositionFixtures.js";
import { createReverseEngineeringService } from "../../src/composition/reverseEngineering.js";
import { projectAnalysisError } from "../../src/domain/analysisErrorProjection.js";
import { AnalysisCancelledError } from "../../src/domain/analysisErrorCore.js";
import { REVERSE_ENGINEERING_TOOL_CONTRACTS } from "../../src/contracts/reverseEngineeringToolContracts.js";
describe("reverse engineering command services", () => {
@@ -574,6 +575,41 @@ it("shares one preflight deadline across stat and digest", async () => {
expect(performance.now() - started).toBeLessThan(280);
});
it("retains an explicit null reason when digest admission is already cancelled", async () => {
const controller = new AbortController();
controller.abort(null);
await expect(
digestFile("/tmp/rea-cancelled-digest", controller.signal),
).rejects.toMatchObject({
_tag: "AnalysisCancelledError",
cause: null,
});
});
it("retains an object reason when cancellation interrupts an active digest stream", async () => {
const controller = new AbortController();
const reason = { source: "caller digest cancellation" };
const stat = async (): Promise<Stats> => ({ isFile: () => true }) as Stats;
const open = () => {
const stream = new PassThrough();
queueMicrotask(() => controller.abort(reason));
return stream as unknown as ReturnType<
typeof import("node:fs").createReadStream
>;
};
const failure: unknown = await digestFile(
"/tmp/rea-cancelled-stream",
controller.signal,
stat,
1_000,
open,
).catch((cause: unknown) => cause);
expect(failure).toBeInstanceOf(AnalysisCancelledError);
if (!(failure instanceof AnalysisCancelledError))
throw new Error("Expected digest cancellation");
expect(failure.cause).toBe(reason);
});
it("forwards selected and omitted deadlines through the production command runner", async () => {
const directory = await mkdtemp(join(tmpdir(), "rea-owned-timeout-"));
const path = join(directory, "fixture.bin");