fix(session): isolate availability observers

This commit is contained in:
N0zoM1z0 2026-07-14 20:49:49 +00:00
parent 528c05e2fa
commit e5a80d4f06
3 changed files with 62 additions and 4 deletions

View File

@ -82,7 +82,7 @@ export class BinarySession implements BinarySessionPort {
string,
{ readonly available: boolean; readonly reason: string | null }
>();
readonly #availabilityListeners = new Set<() => void>();
readonly #availabilityListeners = new Set<() => void | Promise<void>>();
readonly #snapshotsEnabled: boolean;
#snapshotInvalidated = false;
@ -132,7 +132,7 @@ export class BinarySession implements BinarySessionPort {
}
/** Observe runtime provider-health changes that affect discovery metadata. */
onAvailabilityChanged(listener: () => void): () => void {
onAvailabilityChanged(listener: () => void | Promise<void>): () => void {
this.#availabilityListeners.add(listener);
return () => this.#availabilityListeners.delete(listener);
}
@ -583,7 +583,16 @@ export class BinarySession implements BinarySessionPort {
}
#emitAvailabilityChanged(): void {
for (const listener of this.#availabilityListeners) listener();
for (const listener of this.#availabilityListeners) {
try {
const notification = listener();
if (notification !== undefined)
void notification.catch(() => undefined);
} catch {
// External observers are best-effort notifications. Contain only the
// callback failure so state transitions and other listeners continue.
}
}
}
#serialize<T>(operation: () => Promise<T>): Promise<T> {

View File

@ -81,5 +81,5 @@ export interface BinarySessionPort extends AnalysisOperationPort {
UnknownRegistryError
>;
providerIdentity(operation?: AnalysisOperation): ProviderIdentity;
onAvailabilityChanged?(listener: () => void): () => void;
onAvailabilityChanged?(listener: () => void | Promise<void>): () => void;
}

View File

@ -254,6 +254,55 @@ describe("binary session", () => {
expect(changes).toBe(2);
});
it("isolates availability observers from execution results and session state", async () => {
const [first] = await targets();
const provider = cacheProvider([]);
let providerCalls = 0;
provider.createClient = () => ({
execute: (operation) => {
if (operation === "health") return Promise.resolve(ok(null));
providerCalls += 1;
return Promise.resolve(
providerCalls === 1
? err(new ProviderAdapterError("fixture", operation))
: ok(operation),
);
},
close: () => Promise.resolve(),
});
const session = new BinarySession(provider);
expect((await session.open(first)).ok).toBe(true);
const input = { address: "0x1000", document: "first" };
expect((await session.execute("address_name", input)).ok).toBe(false);
session.onAvailabilityChanged(() => {
throw new Error("external observer failed");
});
session.onAvailabilityChanged(() =>
Promise.reject(new Error("async external observer failed")),
);
let delivered = 0;
session.onAvailabilityChanged(() => {
delivered += 1;
});
await expect(session.execute("address_name", input)).resolves.toMatchObject(
{ ok: true },
);
expect(delivered).toBe(1);
expect(session.status()).toMatchObject({
capabilities: expect.arrayContaining([
expect.objectContaining({
operation: "address_name",
available: true,
reason: null,
}),
]),
});
expect((await session.execute("address_name", input)).ok).toBe(true);
expect(providerCalls).toBe(2);
});
it("does not replay operations with filesystem side effects", async () => {
const [first] = await targets();
const calls: string[] = [];