eventrail

A Redis-backed event log for Bun. Sits between Kafka and BullMQ: you get a single ordered stream of events, fan-out to independent consumer groups, and competing workers inside each group -- all backed by a plain Redis list.

Quick start

Start Redis, then publish your first event:

import { publish } from 'eventrail'

await publish('payment.completed', {
  orderId: 'ord_123',
  amount: 49_99,
  currency: 'USD',
})

In another process (or the same one), consume it:

import { consume } from 'eventrail'

await consume('mailer', 'payment.completed', async ({ id, event, data }) => {
  console.log(`[${id}] ${event}`, data)
  // send a receipt email...
})

That's it. The mailer group now tracks its own position in the event log. Every time a payment.completed is published, your handler runs.

How it works

Producer                      Redis                          Consumers
   │                            │                               │
   │─── RPUSH events ──────────▶│  events list (append-only)    │
   │─── PUBLISH wake:group ────▶│                               │
   │                            │◀── LINDEX + offset ───────────│
   │                            │◀── distributed lock ──────────│
   │                            │◀── INCR offset:group ─────────│

• One event log -- a single Redis events list. Every event from every producer ends up here, in order.
• Fan-out via groups -- each group (mailer, ledger, notifications, ...) maintains its own cursor (offset:{group}). All groups see all events independently.
• Competing workers -- within a group, a distributed lock ensures only one worker processes an event at a time. Spin up more processes for parallelism; they share the workload automatically.
• Pull-based -- consumers pull from Redis at their own pace. A pub/sub wake channel reduces latency when new events arrive.

Multiple groups

Imagine a payment.completed event that needs to trigger several independent side effects. Each concern gets its own group:

import { consume } from 'eventrail'

// Group 1 — send a receipt email
await consume('mailer', 'payment.completed', async ({ data }) => {
  await sendReceiptEmail(data.orderId)
})

// Group 2 — record the transaction
await consume('ledger', 'payment.completed', async ({ data }) => {
  await recordTransaction(data.orderId, data.amount)
})

// Group 3 — push a notification
await consume('notifications', 'payment.completed', async ({ data }) => {
  await pushNotification(data.userId, 'Payment received!')
})

Each group processes payment.completed independently. If the mailer falls behind, the ledger and notifications keep going.

Multiple events per group

A group can listen to many event types. They all share one ordered cursor, so events are processed in the order they were published:

await consume('billing', 'payment.completed', async ({ data }) => {
  await finalizeInvoice(data.orderId)
})

await consume('billing', 'payment.refunded', async ({ data }) => {
  await reverseInvoice(data.orderId)
})

The billing group will see payment.completed and payment.refunded interleaved in their original publish order -- a refund can never be processed before the payment it refers to.

Competing workers

For throughput, run the same group across multiple processes. They compete for the lock and share work automatically:

// process-1.ts and process-2.ts — identical code
await consume('mailer', 'payment.completed', async ({ data }) => {
  await sendReceiptEmail(data.orderId) // only one process handles each event
})

Use workCompleteJitterMs (default 80) to prevent one fast worker from starving the others. Set it to 0 in tests for speed:

await consume('mailer', 'payment.completed', handler, {
  workCompleteJitterMs: 120,
})

Retry and dead-letter

If a handler throws, the message is retried with exponential backoff. After maxRetries failures, it moves to a dead-letter list so the group keeps advancing:

await consume('ledger', 'payment.completed', handler, {
  maxRetries: 3,         // default 5
  retryBackoffMs: 500,   // default 1000; doubles each attempt
  retryBackoffMaxMs: 10_000, // default 30_000
})

Dead-lettered messages are stored in dead-letter:{group} as raw event lines. You can inspect or replay them later.

Idempotency

Enable per-message deduplication so a handler never runs twice for the same event (useful after crashes or restarts):

await consume('ledger', 'payment.completed', handler, {
  enableIdempotency: true,
  idempotencyTtlMs: 24 * 60 * 60 * 1000, // default — 24 hours
})

Under the hood, each processed message ID is stored with a TTL (processed:{group}:{messageId}). On restart, if the offset hasn't advanced but the message was already handled, it gets skipped.

Initial offset

When a group starts for the first time (no offset:{group} key exists), you choose where it begins reading:

// Default — process everything from the beginning
await consume('backfill', 'user.created', handler, {
  initialOffset: { mode: 'fromStart' },
})

// Only new events from this point forward
await consume('realtime', 'user.created', handler, {
  initialOffset: { mode: 'fromEnd' },
})

// Resume from a known position
await consume('replay', 'user.created', handler, {
  initialOffset: { mode: 'fromIndex', index: 4200 },
})

Shutdown and resume

Eventrail installs SIGINT and SIGTERM handlers automatically. On shutdown, pending Redis operations (offset increments, lock releases) finish before the process exits.

For programmatic control:

import { stopConsumers } from 'eventrail'

// gracefully stop all consumer loops without exiting
await stopConsumers()

By default, stopConsumers keeps the group registered in interests:{event} so other workers for the same group continue matching. To unregister on shutdown (useful for single-process deployments):

await consume('mailer', 'payment.completed', handler, {
  unregisterInterestOnShutdown: true,
})

On restart, the group resumes from its last committed offset. Nothing is lost.

Stale group cleanup

Each consumer refreshes a heartbeat key (heartbeat:{group}) with a configurable TTL. When a group's processes are all gone, the heartbeat expires. Call pruneStaleGroups periodically (e.g., from a cron or admin endpoint) to remove dead groups from interest sets:

import { pruneStaleGroups } from 'eventrail'

const removed = await pruneStaleGroups()
// ['old-service', 'abandoned-worker']

Configure the heartbeat TTL per consumer:

await consume('mailer', 'payment.completed', handler, {
  heartbeatTtlMs: 60_000, // default 30_000; set 0 to disable
})

Logging

Eventrail uses pino for structured JSON logging. The default logger writes to stdout with { name: 'eventrail' }. Replace it with your own pino instance:

import pino from 'pino'
import { setLogger } from 'eventrail'

setLogger(pino({
  name: 'my-app',
  level: 'warn', // quieter — only warnings and errors
}))

Metrics

Hook into processing events for observability. All methods are optional -- implement only what you need:

import { setMetrics } from 'eventrail'

setMetrics({
  onMessageProcessed(group, event, durationMs) {
    histogram.observe({ group, event }, durationMs)
  },
  onHandlerError(group, event, error, attempt) {
    errorCounter.inc({ group, event })
  },
  onMessageDeadLettered(group, event, offset, messageId) {
    deadLetterCounter.inc({ group })
    alerting.fire(`dead-letter in ${group}: ${messageId}`)
  },
  onLag(group, lag) {
    gauge.set({ group }, lag)
  },
  onLockContention(group) {
    contentionCounter.inc({ group })
  },
  onMessageSkipped(group, reason, offset) {
    skipCounter.inc({ group, reason })
  },
})

Redis keys

| Key | Type | Purpose | |-----|------|---------| | events | LIST | Append-only event log | | offset:{group} | STRING | Next index for this group | | lock:group:{group} | STRING | Distributed lock (NX + PX) | | interests:{event} | SET | Groups interested in this event | | wake:{group} | PUB/SUB | Low-latency notification channel | | dead-letter:{group} | LIST | Failed messages after max retries | | processed:{group}:{id} | STRING | Idempotency record (with PX TTL) | | retries:{group}:{offset} | STRING | Retry counter (with TTL) | | heartbeat:{group} | STRING | Liveness signal (with PX TTL) | | known_groups | SET | All registered group names | | group_events:{group} | SET | Events this group subscribes to |

Wire format

Messages use a length-prefixed binary format for fast O(1) parsing:

<36-char UUID><4-char hex event-name length><event name><JSON payload>

Encode and decode are exported if you need to inspect the raw list:

import { encode, decode } from 'eventrail'

const raw = encode(crypto.randomUUID(), 'user.created', '{"name":"Ada"}')
const { id, event, dataJson } = decode(raw)

Configuration

All options with their defaults:

await consume('group', 'event', handler, {
  initialOffset: { mode: 'fromStart' },
  lockTtlMs: 60_000,
  contentionPollMs: 50,
  idleWaitMs: 2_000,
  workCompleteJitterMs: 80,
  unregisterInterestOnShutdown: false,
  maxRetries: 5,
  retryBackoffMs: 1_000,
  retryBackoffMaxMs: 30_000,
  enableIdempotency: false,
  idempotencyTtlMs: 86_400_000,
  heartbeatTtlMs: 30_000,
})

Dashboard

Eventrail ships with a monitoring dashboard -- a React SPA backed by a lightweight Bun API server that reads Redis state directly. It shows all your consumer groups, their lag, dead-letter queues, and a browsable event log.

Setup

cd dashboard
bun install

Development

Start the API server and Vite dev server in two terminals:

bun --watch server.ts     # API on http://localhost:3001
npx vite                  # UI  on http://localhost:5173

Vite proxies /api requests to the Bun server automatically. Open http://localhost:5173 in your browser.

Production

Build the static assets and serve everything from the Bun server:

npx vite build            # outputs to dist/
bun server.ts             # serves API + static files on :3001

Configuration

The dashboard API server reads the same EVENTRAIL_REDIS_URL and EVENTRAIL_REDIS_DB environment variables as eventrail itself. You can also set DASHBOARD_PORT (default 3001) to change the API port.

For production:

• set DASHBOARD_API_TOKEN and send Authorization: Bearer <token> on every API request
• set DASHBOARD_ALLOWED_ORIGINS to trusted origins only (comma-separated)

Environment variables

| Variable | Default | Description | |----------|---------|-------------| | EVENTRAIL_REDIS_URL | redis://localhost:6379 | Redis connection string (required when NODE_ENV=production) | | EVENTRAIL_REDIS_DB | (none) | Logical database (SELECT after connect) | | DASHBOARD_PORT | 3001 | Dashboard API server port | | DASHBOARD_API_TOKEN | (none) | Required in production for dashboard API auth (Authorization: Bearer ...) | | DASHBOARD_ALLOWED_ORIGINS | http://localhost:5173 | Comma-separated allowed CORS origins for dashboard API |

Retention and replay

Eventrail supports configuration-driven retention with an optional built-in scheduler.

Config-first retention setup

import {
  loadRetentionConfigFromEnv,
  setRetentionConfig,
  startRetentionScheduler,
  stopRetentionScheduler,
} from 'eventrail'

setRetentionConfig(loadRetentionConfigFromEnv())
await startRetentionScheduler()

// on app shutdown:
await stopRetentionScheduler()

Manual helpers

You can still call retention helpers directly:

import { pruneAllDeadLetters, pruneDeadLetters, replayDeadLetter, trimEventStream } from 'eventrail'

// keep stream bounded (approximate trim for lower overhead)
await trimEventStream({ maxLen: 1_000_000, approximate: true })

// keep only latest 500 dead letters for a group
await pruneDeadLetters('mailer', 500)

// apply dead-letter retention across all known groups
await pruneAllDeadLetters(500)

// replay oldest dead-letter entry back into stream
await replayDeadLetter('mailer')

Retention env configuration

| Variable | Default | Description | |----------|---------|-------------| | RETENTION_ENABLED | false | Enables retention cycles | | RETENTION_INTERVAL_MS | 60000 | Scheduler interval | | RETENTION_STREAM_MODE | maxLen | maxLen or timeWindow | | RETENTION_STREAM_MAX_LEN | (none) | Max stream length when mode=maxLen | | RETENTION_STREAM_KEEP_MS | (none) | Keep window in ms when mode=timeWindow | | RETENTION_STREAM_APPROXIMATE | true | Uses approximate XTRIM when true | | RETENTION_DLQ_MAX_LEN | 500 | Max dead-letter size per group | | RETENTION_MAX_TRIM_PER_CYCLE | (none) | Optional warning threshold for trim count | | RETENTION_FAIL_CLOSED_ON_ARCHIVE_ERROR | true | Skip trim if archival fails | | RETENTION_ARCHIVE_ENABLED | false | Enables archival-before-trim | | RETENTION_ARCHIVE_BATCH_SIZE | 1000 | Batch size for XRANGE archival windows | | RETENTION_S3_BUCKET | (none) | S3-compatible bucket name | | RETENTION_S3_PREFIX | eventrail-retention | Archive object key prefix | | RETENTION_S3_ENDPOINT | (none) | S3-compatible endpoint | | RETENTION_S3_REGION | (none) | S3 region | | RETENTION_S3_ACCESS_KEY_ID | (none) | S3 access key | | RETENTION_S3_SECRET_ACCESS_KEY | (none) | S3 secret key |

When archival is enabled, Eventrail exports stream slices to JSONL objects before trimming. Keep object lifecycle/retention in your S3-compatible layer aligned with compliance policy.

Performance validation

Use the built-in benchmark for repeatable throughput/latency checks:

bun run bench:throughput

Useful knobs:

• BENCH_MESSAGES (default 5000)
• BENCH_MAX_SECONDS (default 30)
• BENCH_MIN_THROUGHPUT (default 200)
• BENCH_MAX_P95_MS (default 500)
• EVENTRAIL_BENCH_ALLOW_FLUSH=1 (optional, flushes current Redis DB before run)

The command exits non-zero if thresholds fail, so it can be used in CI/perf gates.

CI quality gates

CI now enforces:

• bun run lint:check
• bun run format:check
• bun run typecheck
• bun run test:ci

Local hooks mirror these checks via lefthook.

Operations runbook

Production operating guidance (SLOs, alerts, incidents, and replay steps) lives in docs/production-runbook.md.

Testing

Tests run against a real Redis instance on logical database 15:

bun test

The preload script (test/preload.ts) sets EVENTRAIL_REDIS_DB=15 and EVENTRAIL_TESTS=1 so FLUSHDB only runs in the test database. For the full stress test:

CI_FULL=1 bun test

Design notes

• At-least-once delivery -- a crash after side effects but before offset advance can repeat a message. Use idempotent handlers or enable enableIdempotency.
• Unbounded list -- the events list grows forever. Add retention or archival separately.
• No separate broker -- consumers pull directly from Redis. No extra process to deploy or monitor.
• Lock auto-renewal -- while a handler runs, the lock TTL is extended every lockTtlMs / 3 so long-running handlers don't lose their lock.