Event buses

L1 lives in one process heap. A write handled by server A does nothing to the L1 on server B, so B keeps serving its own copy of that key until the copy expires on its own. Without a bus, every server goes stale independently and your staleness window is your L1 TTL, on every key, on every server that did not handle the write.

The event bus closes that window. A delete on A becomes a message that every peer applies to its own L1, so the window is one network hop instead of a TTL.

What crosses the bus

Three things, and they are the same on every transport.

That second one is the lazy fan-out, and it is where most of the value is. One server pays for the query. The rest get the result for free.

Two things to know about it before you rely on it:

1. Only getOrSet broadcasts a value. A direct write emits a local event but never publishes to the bus.
2. broadcastSetMaxBytes has no default. Without it, a large value is broadcast in full to every server. Set a ceiling so that a 5 MB row invalidates your peers instead of shipping itself to all of them.

The three event types

Only set carries a value. del and pattern carry key names, which is why they are cheap regardless of how large the underlying values are.

Every event also carries id, source and ts. The id is what event dedupe matches on, so a redelivery is recognised and dropped rather than applied twice.

Give every server a unique source

source is this server's identity. The cache stamps it on every event it publishes, and on receipt it compares the incoming source against its own and returns early when they match. That is what stops a server applying its own broadcast and undoing the write it just made.

Read it from the environment and include the worker identity.

import cluster from 'node:cluster'

const host = process.env.INSTANCE_ID ?? process.env.HOSTNAME ?? 'local'
const worker = cluster.worker?.id ?? process.pid

export const source = `${host}-w${worker}`

Leave source unset and the cache generates a random identifier for the process. Self-filtering still works, but the value changes on every restart, which makes it useless for naming a durable queue or a JetStream consumer and hard to follow in logs.

Delivery semantics

The transports differ in exactly one way that matters: what happens to a server that was not connected when an event was published.

At-most-once is not a bug, it is a trade. A missed event costs you one reload of that key. The question is whether it also costs you correctness. A product description that is 60 seconds stale is staleness. A revoked session, a deleted record, or a key whose TTL is measured in hours is a correctness problem, and durable delivery is what fixes it.

Durable delivery has a real price: per-server state in the broker. A queue or a consumer is created for every server and it accumulates messages while that server is gone. If your servers are ephemeral, that state has to be reaped or it grows without bound.

Pick a transport

One file, four ways to fill it. The rest of your configuration does not change, because every bus satisfies the same interface.

Every bus needs await bus.connect() before use.

Wire it into the cache

The bus and the identity go in as two options. Nothing else about the configuration depends on which transport you picked.

import { LazyLayersCache, RedisStore } from 'lazy-layers-cache'
import { redis } from './redis.js'
import { bus } from './bus.js'
import { source } from './source.js'

export const cache = new LazyLayersCache({
  ttlMs: 5 * 60 * 1000,
  levels: {
    L1: { maxEntries: 10_000, ttlMs: 60 * 1000 },
    L2: { ttlMs: 60 * 60 * 1000 },
  },
  l2: new RedisStore(redis, { prefix: 'app:cache:', deleteStrategy: 'unlink' }),

  eventBus: bus,
  // Identifies this server so it can ignore its own broadcasts.
  source,

  // A value larger than this invalidates peers instead of priming them. There
  // is no default, so without this a 5 MB value goes to every server.
  broadcastSetMaxBytes: 32 * 1024,
})

From here the bus is invisible. getOrSet primes your peers, and delete invalidates them.

import { cache } from '../cache/index.js'
import { db } from '../db.js'

export async function getUser(id) {
  return cache.getOrSet(`user:${id}`, async ({ signal } = {}) => {
    const row = await db.users.findById(id, { signal })
    return row ?? undefined
  })
  // The resolved value is published as a `set` event. Every peer writes it into
  // its own L1, so nobody else pays for this query.
}

export async function updateUser(id, patch) {
  const updated = await db.users.update(id, patch)

  // Publishes a `del` event after the write commits, not before.
  await cache.delete(`user:${id}`)

  return updated
}

Check the bus before you serve traffic

A bus that never connected fails quietly. The cache keeps answering from its own L1 and simply never hears about anyone else's writes, so you get stale data rather than an error. Every bus exposes healthCheck(), which returns a result object instead of throwing.

import { bus } from './cache/bus.js'
import { app } from './app.js'

const health = await bus.healthCheck()

if (!health.ok) {
  console.error('[cache] event bus unhealthy', health)
  process.exit(1)
}

app.listen(3000)

The check connects, and on the transports that need it asserts the exchange or creates the stream and consumer. A wrong URL, missing credentials, a JetStream mode with no durableName, or a broker that is not up yet all surface here. Exiting before listen turns a configuration mistake into a failed startup your orchestrator can retry or roll back.

Two sharp edges

These are real, they are in the code, and neither one announces itself at runtime.

The circuit breaker and the retry queue do not compose

Two independent mechanisms protect a publish, and they do not chain.

The cache's event-bus circuit breaker opens after failureThreshold publish failures, three by default, and stays open for cooldownMs, thirty seconds by default. While it is open, publishInvalidation emits event-bus:publish-skipped and returns before calling eventBus.publish.

The bus retry queue lives inside eventBus.publish. It buffers an event when the publish itself fails and flushes the buffer ahead of the next successful publish.

The retry queue covers short failures that get through to the transport. The breaker covers a broker that is failing every call, and it protects local reads at the cost of those invalidations. If a broker outage must not lose invalidations, raise failureThreshold so the breaker tolerates more failures before it starts shedding, and size retryQueue.maxSize to the number of events you write during an outage you expect to survive.

RabbitMQ does not await your handler

The RabbitMQ consumer calls void Promise.resolve(handler(event)) and attaches .then(ack). It does not await before taking the next message.

Setting prefetch is what bounds that. It caps unacknowledged messages on the channel, and since the acknowledgement happens when the handler settles, the cap becomes your concurrency limit. Redis bounds this with handlerConcurrency, which defaults to 1, and both NATS modes await each handler in the message loop. RabbitMQ is the only transport where you have to supply the number yourself.

The retry queue

Every transport takes the same two options.

The bound is deliberate. During a long broker outage an unbounded queue would grow for as long as the server keeps writing, holding every buffered value in memory until the heap runs out. Dropping the oldest invalidations costs you staleness on those keys. Growing without a limit costs you the process, and with it every entry in L1.

The queue is in process memory, so a restart discards it. It covers a broker blip, not an outage that outlasts your deploy.

Where to next