阅读中文版本

Build a Local Event Bus

An in-process event bus with typed channels, subscriber ordering, error isolation per handler, and a replay buffer for late subscribers.

event-bus-local/
├── src/bus.ts
├── src/channel.ts
├── src/replay.ts
├── tests/bus.test.ts
└── package.json

bus.publish("user.created", user); // type error if payload wrong

for (const h of subs) await h(event);

try { await h(event); } catch (e) { bus.publish("bus.dead", {event, e}); }

bus.subscribe("user.created", handler, {replay: 10});

Project overview

A producer that calls a consumer directly is coupled to it, and adding a second consumer means changing the producer. An event bus decouples them: the producer publishes to a channel, and any number of consumers subscribe without the producer knowing. This is the smallest useful decoupling, and it runs in a single process with no network.

The typed channels are what keep the decoupling safe. A bus with string channels and any payloads is a bag of untyped messages where a typo is a runtime error. A bus with literal channel types and payload types makes a wrong payload a compile error, so the decoupling does not cost type safety.

The error isolation is what keeps the bus reliable. A bus that lets one handler exception stop the others is fragile, because a single buggy consumer breaks every consumer. The bus catches each handler in a try-catch and reports the error to a dead-letter channel, so a failure is visible and contained.

The replay buffer is what helps late subscribers. A subscriber that joins after an event was published misses it, which is a problem for a handler that needs recent context. The bus keeps a ring buffer per channel and replays it on request, so a late joiner sees the recent past without a separate query.

The value of this project is the error isolation, not the pub-sub. Publishing and subscribing is trivial; making sure one bad handler does not poison the channel is the part worth learning, because that is what lets a bus stay useful as the number of handlers grows.

What you will build

The project delivers five capabilities, each driven by a test:

• Typed channels
• Ordered delivery per channel
• Error isolation per handler
• Replay buffer for late joiners
• Sync and async handlers

A bus holds a map of channel name to subscriber list and a ring buffer per channel. Publish appends to the buffer and invokes each subscriber in order, catching errors per handler so one failure does not stop the others. Subscribe adds a handler and optionally replays the buffer tail.

Architecture and data model

The bus is a map from channel name to a channel object. The channel holds a subscriber list and a ring buffer. The subscriber list is an array of handlers, appended in subscription order, so delivery is deterministic. The ring buffer is a fixed-size circular array of recent events, overwritten oldest-first.

Publish appends the event to the ring buffer and invokes each subscriber in order. The invocation is synchronous for a sync handler and awaited for an async handler, so a sync bus call returns when all sync handlers have run and all async handlers have started. A fully async bus that awaits every handler is a separate mode, because it changes the latency profile.

The error isolation wraps each handler call in a try-catch. A throw is caught, the error and the event are published to a dead-letter channel, and the bus continues with the next handler. The dead-letter channel is a regular channel, so it can have its own subscribers for logging or retry.

The replay buffer is a ring per channel, sized at bus creation. A subscriber can request replay of the last N events, and the bus delivers them in order before new events. The replay is a snapshot of the buffer at subscribe time, so concurrent publishes do not interleave with the replay.

The typed channels are a TypeScript generic on the bus. A channel is a string literal type, and the bus maps each channel to a payload type. Publish and subscribe are generic methods that check the payload type at compile time, so a wrong payload is a compile error, not a runtime exception.

Prerequisites and setup

You need the following before you begin:

• Node.js 20 LTS or later
• TypeScript for typed channels

The dependencies cover the core needs:

• node:events (built-in) — Reference for emitter semantics

pnpm init

pnpm add -D typescript tsx vitest

Keep secrets and configuration in the environment, and validate them at boot so a misconfigured service fails loudly instead of silently dropping work.

Directory structure

event-bus-local/ is organized so each concern is testable in isolation:

event-bus-local/
├── src/bus.ts
├── src/channel.ts
├── src/replay.ts
├── tests/bus.test.ts
└── package.json

Phase 1: Typed channels

Define a channel as a string literal type and a payload type, so publish and subscribe are type-checked.

bus.publish("user.created", user); // type error if payload wrong

A wrong payload type is a compile error, not a runtime surprise.

The phase is driven by a test that asserts the expected behavior, so the implementation is correct before the next phase builds on it.

Phase 2: Ordered delivery

Invoke subscribers in subscription order, so a handler that depends on a prior handler runs after it.

for (const h of subs) await h(event);

Order is subscription order, which is deterministic but not priority-based.

The phase is driven by a test that asserts the expected behavior, so the implementation is correct before the next phase builds on it.

Phase 3: Error isolation

Catch each handler in a try-catch so one throw does not stop the rest, and report the error to a dead-letter channel.

try { await h(event); } catch (e) { bus.publish("bus.dead", {event, e}); }

A failing handler is logged, not silent, and the bus continues.

The phase is driven by a test that asserts the expected behavior, so the implementation is correct before the next phase builds on it.

Phase 4: Replay buffer

Keep a ring buffer per channel and replay it to a late subscriber on request.

bus.subscribe("user.created", handler, {replay: 10});

Replay gives a late joiner the recent context without a separate query.

The phase is driven by a test that asserts the expected behavior, so the implementation is correct before the next phase builds on it.

Testing

Publish an event and assert both handlers receive it in order; throw in the first handler and assert the second still runs; subscribe with replay and assert the handler receives the last N events before new ones.

The delivery test subscribes two handlers, publishes an event, and asserts both received it in subscription order. The error-isolation test throws in the first handler and asserts the second still received the event, and the dead-letter channel received the error.

The replay test publishes five events, subscribes with replay of three, and asserts the handler received the last three before any new event. The ordering test asserts the replay events come before new events, not interleaved.

The type test is a compile-time check: a publish with the wrong payload type is a type error, asserted by a TypeScript compilation step. The async test subscribes an async handler, publishes, and asserts the handler ran, with the bus awaiting it.

The ring-buffer test publishes more events than the buffer size and asserts the oldest are overwritten, so a replay of the full buffer returns only the recent events. The dead-letter test subscribes to the dead-letter channel and asserts it receives the event and the error when a handler throws.

The unsubscribe test removes a handler and asserts it does not receive subsequent publishes, proving the removal is effective.

pnpm vitest run

Observability

Track the typed channels rate, the event-bus resource usage, and the error count as the three primary signals for the event-bus-local system. A sudden change in any of these is the signal that something has shifted, not necessarily broken but worth a look before it becomes an incident. Chart the ordered delivery per channel alongside the typed channels so a correlation is visible at a glance, because a rising ordered delivery per channel with a stable typed channels means the system is doing more work per unit, not that it is failing.

Log every typed channels transition with the context that triggered it, because an operator who sees the symptom but not the cause will guess wrong. The log entry should carry the event-bus identifier, the input that caused the transition, and the outcome, so the operator can reconstruct what happened without re-running it. A log without context is a line of text; a log with context is a diagnosis.

Set alerts on the derived metrics, not the raw counters. A rate of change is more actionable than a total, because a total grows monotonically and never tells you when the growth accelerated. Alert on the typed channels error rate crossing a threshold, the event-bus utilization exceeding a percentage, and the ordered delivery per channel latency exceeding the SLO. Each alert should have a runbook link, so the operator who receives it at three in the morning knows what to do, not just what is wrong.

The event-bus-local system should expose its metrics in a format the existing dashboard can scrape, so the observability is additive, not a new tool. A metric that nobody reads is noise, so before adding one, confirm it has a consumer. The dashboard has three panels: throughput, error rate, and resource utilization, each with a threshold line that is the SLO, so a glance tells an operator whether the system is healthy without reading numbers.

Deployment and operations

Use one bus per process. The bus is in-memory and not network-aware, so sharing it across processes requires a transport, which is a different system. A process-local bus is the right tool for decoupling within a process, and a message queue is the right tool for decoupling across processes.

Size the replay buffer from the late-joiner need, not a default. A buffer of a hundred events is fine for a handler that needs recent context; a buffer of a million is wasteful if no one replays. The buffer is per channel, so a channel with no replay need can have a buffer of zero.

Monitor the dead-letter channel. A growing dead-letter rate means a handler is failing, and the dead-letter is the signal. The dead-letter events should be inspected and the handler fixed, not ignored, because a handler that always fails is a no-op subscriber.

Do not use the bus for backpressure. A slow handler blocks the channel, because delivery is synchronous in subscription order. A bus with backpressure, where a slow handler is skipped or queued, is a different design, and the right answer depends on the semantics the caller needs.

Security considerations

Treat every input to the event-bus-local system as untrusted, because the typed channels path is the one place an external caller controls what the system processes. Validate the input shape before any event-bus operation, and reject early with a clear error rather than letting a malformed input propagate into the typed channels phase where it causes a confusing failure that is hard to trace back to its source.

Keep secrets and configuration in the environment, never in the repository. The node:events credentials, the signing keys, and the service endpoints are all environment variables, read once at boot and validated, so a misconfigured event-bus-local fails loudly at startup instead of silently misbehaving at runtime. A system that starts with a bad configuration and runs anyway is a system that will produce wrong results for hours before anyone notices.

Rate-limit the typed channels endpoint per caller, because an unbounded event-bus operation is a resource leak as much as a security problem. A caller that can trigger the ordered delivery path without limit can exhaust the system, whether intentionally or not. The rate limit is a per-caller ceiling that costs legitimate callers nothing and stops the abuse before it reaches the typed channels logic.

Audit-log every state-changing operation with the caller identity, the input, and the timestamp. The audit log is not for debugging, it is for accountability: if something went wrong, the log says who did what and when. The event-bus-local system writes the audit log to an append-only store that the system itself cannot modify, so a compromised system cannot rewrite its history.

Do not expose internal errors to the caller. A stack trace is a map of the system internals, and an attacker who collects them learns the event-bus structure. Return a generic error with a correlation id, and log the full error internally with the same id, so an operator can trace it without leaking it.

Limitations and next steps

• In-process only; no cross-process delivery.
• No backpressure; a slow handler blocks the channel.
• The replay buffer is bounded; old events are lost.

The bus is in-process only. There is no cross-process delivery, because the bus is a map in memory. A cross-process bus needs a transport, which is a message queue, and that is a different system with different trade-offs.

There is no backpressure. A slow handler blocks the channel, because delivery is in order and awaited for async handlers. A bus that skips or queues a slow handler is a real improvement but changes the delivery semantics, and the right choice depends on the caller.

The replay buffer is bounded. Old events are overwritten when the buffer is full, so a late joiner that requests replay beyond the buffer size gets only the recent events. An unbounded buffer would grow forever, which is a memory leak, so the bound is a trade-off, not a bug.