RabbitMQ

Durable fan-out. Each server gets its own queue bound to a shared exchange, and that queue holds invalidations while the server is disconnected. When it comes back, it drains the backlog and catches up on every key it missed.

That is the thing Redis Pub/Sub and NATS Core cannot do. The price is broker state per server, and one option with no default that you have to set.

Run RabbitMQ

RABBITMQ_URL="amqp://guest:guest@localhost:5672"

Set it up

import Redis from 'ioredis'

export const redis = new Redis(process.env.REDIS_URL, {
  maxRetriesPerRequest: 2,
  enableReadyCheck: true,
  enableOfflineQueue: true,
})

redis.on('error', (err) => {
  console.error('[redis]', err.message)
})

import { RabbitMQEventBus } from 'lazy-layers-cache'

export const bus = new RabbitMQEventBus('lazylayers.invalidate', {
  url: process.env.RABBITMQ_URL,
  exchangeType: 'fanout',
  durable: true,
  persistent: true,
  durableInvalidationMode: true,
  // Required. The queue is asserted durable only when it is named,
  // so without this you get a transient, broker-named queue.
  queueName: `lazylayers.${process.env.INSTANCE_ID}`,
  // Stable per server. An unnamed queue is never durable.
  queueName: `${process.env.INSTANCE_ID}-cache`,
  exclusiveQueue: false,
  autoDeleteQueue: false,
  // No default. Without it, handlers run as fast as the broker delivers.
  prefetch: 32,
  retryQueue: { maxSize: 1000 },
  logging: { env: 'production' },
})

await bus.connect()

import { LazyLayersCache, RedisStore } from 'lazy-layers-cache'
import { redis } from './redis.js'
import { bus } from './bus.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,
  // The same identity that names the queue above.
  source: process.env.INSTANCE_ID,
  broadcastSetMaxBytes: 32 * 1024,
})

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
  })
}

export async function updateUser(id, patch) {
  const updated = await db.users.update(id, patch)
  await cache.delete(`user:${id}`)
  return updated
}

prefetch has no default, and it is the only concurrency dial

The consumer decodes a message and calls void Promise.resolve(handler(event)). It attaches .then() to acknowledge and .catch() to negatively acknowledge, and then it returns. It does not await the handler before taking the next message.

Because the acknowledgement happens when the handler settles, prefetch is what caps in-flight work: it bounds unacknowledged messages on the consumer channel, and unacknowledged means still running. Setting it turns unbounded concurrency into a number you chose.

This matters most in exactly the situation you bought RabbitMQ for. A server that has been down for ten minutes reconnects to a full queue, and the whole backlog arrives at once.

Start at 32 and raise it if the queue drains too slowly for you. A set handler writes to L1 and a del handler touches L1 and L2, so the work per message is small and bounded.

Every option

Failure modes

An unnamed queue in durable mode is the trap. With durableInvalidationMode: true and no queueName, the broker generates a random name. The queue is not exclusive and not auto-deleted, so it survives your disconnect, but it is also not durable and your next start asks for another random name. The old queue keeps its binding to the exchange and keeps accumulating messages that nothing will ever consume, while your restarted server has an empty queue and no backlog. You get the broker cost of durability and none of the benefit.

A handler that throws loses the event. The catch path calls nack with requeue set to false, so the message is discarded rather than retried. It is dead-lettered only if you configured a dead-letter exchange on the queue yourself. Unlike JetStream, there is no redelivery attempt here.

A message that does not decode is acknowledged and dropped, with a warning. That is deliberate: redelivering something unparseable would fail the same way forever.

Publishing before connect() throws. So does subscribing. Both check for the channel and refuse.

Publishes wait for a broker confirm. The publisher uses a confirm channel and awaits waitForConfirms(), so a delete does not resolve until the broker has taken responsibility for the message. That is stronger than Redis Pub/Sub, and it means a slow broker slows your write path.

Repeated publish failures trip the cache's circuit breaker, and then the retry queue stops helping. Once the breaker is open the cache returns before it calls eventBus.publish, so nothing reaches the retry queue to be buffered. See the sharp edges.

Operational notes

healthCheck() reports queueName as null until subscribe() has run, because the queue is asserted at subscribe time, not at connect time. A healthy result before the cache has subscribed is still healthy.

One queue exists per server, forever, unless something removes it. If your servers are ephemeral, that set grows with every deploy. Either derive queueName from a stable identity so restarts reuse the same queue, or set a queue expiry policy on the broker so an unused queue is reaped. On Kubernetes, a StatefulSet gives you the stable name. A Deployment does not.

In the management UI, open Queues. You should see one queue per server, each bound to your exchange. Stop a server and delete a key elsewhere, and its queue depth rises. Start it again and the depth falls to zero as it drains. Watch the same thing from the application side:

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

cache.on((event) => {
  if (event.type === 'invalidation:received') {
    console.log('received', event.eventType, event.eventId)
  }
})

Queue depth is the metric worth alerting on. A depth that grows and never returns to zero means a server is gone and its queue is not being reaped, or prefetch is far lower than your invalidation rate.

When to choose it

Choose RabbitMQ when a missed invalidation is a correctness problem and RabbitMQ is already something your team runs. Durable queues per server, a management UI that shows the backlog, and dead-letter routing you may already have standardised on.

Choose NATS JetStream instead if you want redelivery on handler failure rather than a discard, or if you would rather run one small binary than a broker.

Stay on Redis Pub/Sub if a missed invalidation only costs you a reload. Broker state per server is not free, and you should not pay for it without a reason.

Where to next