fix(core-testing): align RecordingEventBus with real bus semantics

Handlers and the published record now receive the zod-PARSED payload
(defaults/coercions applied) instead of the raw input, and fan-out uses
Promise.allSettled with errors swallowed by default plus an opt-in
failFast — matching the InMemoryEventBus the events core-package
generator scaffolds. The previous sequential fail-fast divergence was
undocumented, so it is aligned rather than kept. Regression tests pin
parsed-payload delivery, non-short-circuiting fan-out, and failFast
rethrow. (S4; the generator bus templates already carry the fix.)

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
2026-07-10 17:21:52 +02:00
parent c9db7c8cd7
commit 27787193c0
3 changed files with 73 additions and 5 deletions

View File

@@ -2,7 +2,10 @@ export { RecordingTracer, type RecordedSpan } from "./recording-tracer";
export { RecordingLogger, type RecordedCapture } from "./recording-logger"; export { RecordingLogger, type RecordedCapture } from "./recording-logger";
export { RecordingMetrics, type RecordedMetric } from "./recording-metrics"; export { RecordingMetrics, type RecordedMetric } from "./recording-metrics";
export { RecordingJobQueue } from "./recording-job-queue"; export { RecordingJobQueue } from "./recording-job-queue";
export { RecordingEventBus } from "./recording-event-bus"; export {
RecordingEventBus,
type RecordingEventBusOptions,
} from "./recording-event-bus";
export { RecordingRealtimeBroadcaster } from "./recording-realtime-broadcaster"; export { RecordingRealtimeBroadcaster } from "./recording-realtime-broadcaster";
export { RecordingAuditLog } from "./recording-audit-log"; export { RecordingAuditLog } from "./recording-audit-log";
export { export {

View File

@@ -28,7 +28,7 @@ describe("RecordingEventBus", () => {
expect(bus.published).toHaveLength(0); expect(bus.published).toHaveLength(0);
}); });
it("invokes registered handlers sequentially in subscription order", async () => { it("invokes all registered handlers (started in subscription order)", async () => {
const bus = new RecordingEventBus(); const bus = new RecordingEventBus();
const order: string[] = []; const order: string[] = [];
bus.subscribe(evt, "consumer-a", async () => { bus.subscribe(evt, "consumer-a", async () => {
@@ -40,4 +40,48 @@ describe("RecordingEventBus", () => {
await bus.publish(evt, { id: "x" }); await bus.publish(evt, { id: "x" });
expect(order).toEqual(["a", "b"]); expect(order).toEqual(["a", "b"]);
}); });
it("delivers and records the zod-parsed payload, not the raw input", async () => {
const evtNormalized = {
name: "test.normalized" as const,
schema: z.object({ id: z.string(), mode: z.string().catch("auto") }),
};
const bus = new RecordingEventBus();
const received: unknown[] = [];
bus.subscribe(evtNormalized, "consumer", async (e) => {
received.push(e);
});
await bus.publish(evtNormalized, {
id: "x",
mode: 42 as unknown as string,
});
expect(received).toEqual([{ id: "x", mode: "auto" }]);
expect(bus.published).toEqual([
{ name: "test.normalized", payload: { id: "x", mode: "auto" } },
]);
});
it("swallows handler errors by default, like the real InMemoryEventBus", async () => {
const bus = new RecordingEventBus();
const order: string[] = [];
bus.subscribe(evt, "boom", async () => {
throw new Error("subscriber blew up");
});
bus.subscribe(evt, "after-boom", async () => {
order.push("after-boom");
});
await expect(bus.publish(evt, { id: "x" })).resolves.toBeUndefined();
expect(order).toEqual(["after-boom"]); // fan-out not short-circuited
expect(bus.published).toHaveLength(1);
});
it("rethrows the first handler error when failFast is true", async () => {
const bus = new RecordingEventBus({ failFast: true });
bus.subscribe(evt, "first", async () => {
throw new Error("first failure");
});
await expect(bus.publish(evt, { id: "x" })).rejects.toThrow(
"first failure",
);
});
}); });

View File

@@ -22,17 +22,38 @@ interface IEventBus {
): void; ): void;
} }
export type RecordingEventBusOptions = {
/** When true, rethrow the first handler error (default: false — errors swallowed). */
failFast?: boolean;
};
/**
* Recording test double for IEventBus. Delivery semantics deliberately mirror
* the real InMemoryEventBus (@repo/core-events, scaffolded via
* `gen core-package events`): handlers receive the zod-PARSED payload
* (defaults/coercions applied), fan-out uses Promise.allSettled, and handler
* errors are swallowed unless `failFast`.
*/
export class RecordingEventBus implements IEventBus { export class RecordingEventBus implements IEventBus {
readonly published: { name: string; payload: unknown }[] = []; readonly published: { name: string; payload: unknown }[] = [];
private readonly handlers = new Map<string, EventHandler<unknown>[]>(); private readonly handlers = new Map<string, EventHandler<unknown>[]>();
constructor(private readonly options: RecordingEventBusOptions = {}) {}
async publish<T>( async publish<T>(
descriptor: EventDescriptor<string, z.ZodType<T>>, descriptor: EventDescriptor<string, z.ZodType<T>>,
payload: T, payload: T,
): Promise<void> { ): Promise<void> {
descriptor.schema.parse(payload); const parsed = descriptor.schema.parse(payload);
this.published.push({ name: descriptor.name, payload }); this.published.push({ name: descriptor.name, payload: parsed });
for (const h of this.handlers.get(descriptor.name) ?? []) await h(payload); const subscribers = this.handlers.get(descriptor.name) ?? [];
if (subscribers.length === 0) return;
const settled = await Promise.allSettled(subscribers.map((h) => h(parsed)));
if (this.options.failFast) {
const failure = settled.find((s) => s.status === "rejected");
// Only the first rejection is rethrown — matching InMemoryEventBus.
if (failure && failure.status === "rejected") throw failure.reason;
}
} }
subscribe<T>( subscribe<T>(