From 27787193c05e50ba3f95bc3a62f371d703a7c9b7 Mon Sep 17 00:00:00 2001 From: Danijel Martinek Date: Fri, 10 Jul 2026 17:21:52 +0200 Subject: [PATCH] fix(core-testing): align RecordingEventBus with real bus semantics MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 --- .../core-testing/src/instrumentation/index.ts | 5 +- .../recording-event-bus.test.ts | 46 ++++++++++++++++++- .../instrumentation/recording-event-bus.ts | 27 +++++++++-- 3 files changed, 73 insertions(+), 5 deletions(-) diff --git a/packages/core-testing/src/instrumentation/index.ts b/packages/core-testing/src/instrumentation/index.ts index 599c647..a10cd7e 100644 --- a/packages/core-testing/src/instrumentation/index.ts +++ b/packages/core-testing/src/instrumentation/index.ts @@ -2,7 +2,10 @@ export { RecordingTracer, type RecordedSpan } from "./recording-tracer"; export { RecordingLogger, type RecordedCapture } from "./recording-logger"; export { RecordingMetrics, type RecordedMetric } from "./recording-metrics"; 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 { RecordingAuditLog } from "./recording-audit-log"; export { diff --git a/packages/core-testing/src/instrumentation/recording-event-bus.test.ts b/packages/core-testing/src/instrumentation/recording-event-bus.test.ts index c330822..33736ad 100644 --- a/packages/core-testing/src/instrumentation/recording-event-bus.test.ts +++ b/packages/core-testing/src/instrumentation/recording-event-bus.test.ts @@ -28,7 +28,7 @@ describe("RecordingEventBus", () => { 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 order: string[] = []; bus.subscribe(evt, "consumer-a", async () => { @@ -40,4 +40,48 @@ describe("RecordingEventBus", () => { await bus.publish(evt, { id: "x" }); 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", + ); + }); }); diff --git a/packages/core-testing/src/instrumentation/recording-event-bus.ts b/packages/core-testing/src/instrumentation/recording-event-bus.ts index 2bcde70..db51c5d 100644 --- a/packages/core-testing/src/instrumentation/recording-event-bus.ts +++ b/packages/core-testing/src/instrumentation/recording-event-bus.ts @@ -22,17 +22,38 @@ interface IEventBus { ): 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 { readonly published: { name: string; payload: unknown }[] = []; private readonly handlers = new Map[]>(); + constructor(private readonly options: RecordingEventBusOptions = {}) {} + async publish( descriptor: EventDescriptor>, payload: T, ): Promise { - descriptor.schema.parse(payload); - this.published.push({ name: descriptor.name, payload }); - for (const h of this.handlers.get(descriptor.name) ?? []) await h(payload); + const parsed = descriptor.schema.parse(payload); + this.published.push({ name: descriptor.name, payload: parsed }); + 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(