fix(session): isolate availability observers
This commit is contained in:
commit
43766fda69
|
|
@ -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> {
|
||||
|
|
|
|||
|
|
@ -81,5 +81,5 @@ export interface BinarySessionPort extends AnalysisOperationPort {
|
|||
UnknownRegistryError
|
||||
>;
|
||||
providerIdentity(operation?: AnalysisOperation): ProviderIdentity;
|
||||
onAvailabilityChanged?(listener: () => void): () => void;
|
||||
onAvailabilityChanged?(listener: () => void | Promise<void>): () => void;
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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[] = [];
|
||||
|
|
|
|||
Loading…
Reference in New Issue