feat: Add Write-Ahead Log for Crash Durability

README.md

@@ -24,6 +24,10 @@ Message brokers hide a lot of machinery behind `publish` and `subscribe`. This r
 - **Producer-facing occupancy**: capacity is ready depth. Prefetch already caps in-flight. Redelivery of accepted work is allowed to sit above capacity so a nack cannot drop a message the queue already took.
 - **Per-key FIFO ordering**: at most one in-flight delivery per key (SQS FIFO message group / Kafka key), a partial order across keys, and head-of-line blocking only within a key
 
+- **Write-ahead logging**: append-only log, **length-prefixed framing**, **CRC32 checksums**, and **fsync** before an append returns
+- **Torn-write recovery**: a truncated or checksum-mismatched tail is dropped, not replayed
+- **Crash recovery by replay**: enqueue / ack / drop records rebuild the ready set
+- **Log truncation (checkpoint)**: rewrite the file to live enqueue records so the log cannot grow without bound
 ## What's implemented
 
 - **Topic-based publish/subscribe with fan-out.** A `Broker` where subscribers register handlers against a topic and every publish fans out to all matching subscribers, with monotonic message ids, idempotent unsubscribe, and snapshot-consistent delivery (subscribing or unsubscribing during dispatch never changes who receives the in-flight message).
@@ -38,6 +42,11 @@ Message brokers hide a lot of machinery behind `publish` and `subscribe`. This r
 - **Bounded queues with backpressure signaling.** Give `WorkQueue` a finite `capacity`. New publishes that would grow the ready backlog past that bound are rejected (`enqueue` throws `QueueFullError`, `tryEnqueue` returns `{ accepted: false }`). A `WatermarkGate` watches ready depth: occupancy at or above `highWatermark` emits `paused`, occupancy at or below `lowWatermark` emits `open`, and values in between hold the last state. Subscribe with `onBackpressure` or poll `backpressure()`. Defaults: high equals capacity, low is half the capacity (or `high - 1` when high is small). Unbounded queues stay the default so existing callers do not change.
 - **Per-key FIFO ordering**: at most one in-flight delivery per key (SQS FIFO message group / Kafka key), a partial order across keys, and head-of-line blocking only within a key
 - **Per-key ordering guarantees.** A `KeyedWorkQueue` where each message carries a group key. The head of a key is exclusive: the tail stays queued until that head is acked, nacked-and-dropped, or cancelled. Different keys can be in flight on competing consumers at the same time. A nack requeues at the head of that key so a later message of the same key cannot overtake. Prefetch can hold several keys, never two messages of one key.
+- **Write-ahead logging**: append-only log, **length-prefixed framing**, **CRC32 checksums**, and **fsync** before an append returns
+- **Torn-write recovery**: a truncated or checksum-mismatched tail is dropped, not replayed
+- **Crash recovery by replay**: enqueue / ack / drop records rebuild the ready set
+- **Log truncation (checkpoint)**: rewrite the file to live enqueue records so the log cannot grow without bound
+- **Write-ahead log for crash durability.** A `WriteAheadLog` stores length-prefixed records (`u32le` length, `u32le` CRC32, payload) and fsyncs before `append` returns. On open it replays complete records and truncates a torn or corrupt tail. `DurableWorkQueue` logs enqueue, ack, and drop *before* the in-memory mutation, so a restart redelivers unacked work and does not resurrect acked work. `checkpoint()` rewrites the file to the live enqueue records (temp file, fsync, atomic rename).
 ## Usage
 
 ```ts

package.json

@@ -1,7 +1,7 @@
 {
   "name": "event-broker-lab",
   "version": "0.1.0",
-  "description": "A from-scratch in-memory message broker exploring pub/sub, work queues, delivery guarantees, and backpressure — the concepts behind Kafka, RabbitMQ, and SQS, built small enough to read.",
+  "description": "A from-scratch message broker exploring pub/sub, work queues, delivery guarantees, and write-ahead logging: the concepts behind Kafka, RabbitMQ, and SQS, built small enough to read.",
   "type": "module",
   "license": "MIT",
   "scripts": {
@@ -12,6 +12,7 @@
   },
   "packageManager": "pnpm@11.3.0",
   "devDependencies": {
+    "@types/node": "^22.20.1",
     "typescript": "^5.6.0",
     "vitest": "^2.1.0"
   }

src/durable-work-queue.ts

@@ -0,0 +1,159 @@
+import { WriteAheadLog, type WriteAheadLogOptions } from './wal.js'
+import {
+  WorkQueue,
+  type ConsumerHandler,
+  type Delivery,
+  type Unsubscribe,
+  type WorkQueueOptions,
+} from './work-queue.js'
+
+const DEFAULT_MAX = 10
+
+type Inner<T> = { walId: number; body: T }
+
+type Op<T> =
+  | { op: 'enqueue'; id: number; payload: T }
+  | { op: 'ack'; id: number }
+  | { op: 'drop'; id: number }
+
+export interface DurableWorkQueueOptions extends WorkQueueOptions, WriteAheadLogOptions {}
+
+function encode<T>(op: Op<T>): Uint8Array {
+  if (op.op === 'enqueue' && op.payload === undefined) {
+    throw new Error('enqueue payload must be JSON-serializable')
+  }
+  return new TextEncoder().encode(JSON.stringify(op))
+}
+
+function decode<T>(bytes: Uint8Array): Op<T> {
+  const value = JSON.parse(new TextDecoder().decode(bytes)) as Op<T>
+  if (value.op !== 'enqueue' && value.op !== 'ack' && value.op !== 'drop') {
+    throw new Error('invalid wal record')
+  }
+  if (!Number.isInteger(value.id) || value.id < 1) throw new Error('invalid wal record')
+  return value
+}
+
+export class DurableWorkQueue<T> {
+  private constructor(
+    private readonly log: WriteAheadLog,
+    private readonly inner: WorkQueue<Inner<T>>,
+    private readonly live: Map<number, T>,
+    private nextId: number,
+    private readonly maxDeliveryCount: number,
+  ) {}
+
+  static open<T>(path: string, options?: DurableWorkQueueOptions): DurableWorkQueue<T> {
+    return DurableWorkQueue.fromLog(WriteAheadLog.open(path, options), options)
+  }
+
+  static memory<T>(options?: DurableWorkQueueOptions): DurableWorkQueue<T> {
+    return DurableWorkQueue.fromLog(WriteAheadLog.memory(options), options)
+  }
+
+  private static fromLog<T>(
+    log: WriteAheadLog,
+    options?: DurableWorkQueueOptions,
+  ): DurableWorkQueue<T> {
+    const maxDeliveryCount = options?.maxDeliveryCount ?? DEFAULT_MAX
+    if (!Number.isInteger(maxDeliveryCount) || maxDeliveryCount < 1) {
+      throw new Error(`maxDeliveryCount must be a positive integer, got ${maxDeliveryCount}`)
+    }
+    const inner = new WorkQueue<Inner<T>>({ maxDeliveryCount })
+    const live = new Map<number, T>()
+    let nextId = 1
+    try {
+      for (const bytes of log.replay()) {
+        const rec = decode<T>(bytes)
+        if (rec.id >= nextId) nextId = rec.id + 1
+        if (rec.op === 'enqueue') live.set(rec.id, rec.payload)
+        else live.delete(rec.id)
+      }
+    } catch (error) {
+      log.close()
+      throw error
+    }
+    const queue = new DurableWorkQueue(log, inner, live, nextId, maxDeliveryCount)
+    for (const [id, payload] of live) inner.enqueue({ walId: id, body: payload })
+    return queue
+  }
+
+  enqueue(payload: T): number {
+    const id = this.nextId
+    this.log.append(encode({ op: 'enqueue', id, payload }))
+    this.nextId += 1
+    this.live.set(id, payload)
+    this.inner.enqueue({ walId: id, body: payload })
+    return id
+  }
+
+  consume(handler: ConsumerHandler<T>, options?: { prefetch?: number }): Unsubscribe {
+    const inflight = new Set<{ settled: boolean }>()
+    const off = this.inner.consume((delivery) => {
+      const gate = { settled: false }
+      inflight.add(gate)
+      const wrapped = this.wrap(delivery, gate)
+      try {
+        handler(wrapped)
+      } catch {
+        wrapped.nack()
+      } finally {
+        if (gate.settled) inflight.delete(gate)
+      }
+    }, options)
+    return () => {
+      for (const gate of inflight) gate.settled = true
+      inflight.clear()
+      off()
+    }
+  }
+
+  checkpoint(): void {
+    const kept: Uint8Array[] = []
+    for (const [id, payload] of this.live) kept.push(encode({ op: 'enqueue', id, payload }))
+    this.log.rewrite(kept)
+  }
+
+  close(): void {
+    this.log.close()
+  }
+
+  readyCount(): number {
+    return this.inner.readyCount()
+  }
+
+  private wrap(delivery: Delivery<Inner<T>>, gate: { settled: boolean }): Delivery<T> {
+    const walId = delivery.message.payload.walId
+    const finish = (fn: () => void) => {
+      if (gate.settled) return
+      gate.settled = true
+      fn()
+    }
+    return {
+      message: {
+        id: walId,
+        payload: delivery.message.payload.body,
+        deliveryCount: delivery.message.deliveryCount,
+      },
+      deliveryTag: delivery.deliveryTag,
+      ack: () =>
+        finish(() => {
+          this.commit('ack', walId)
+          delivery.ack()
+        }),
+      nack: (opts) =>
+        finish(() => {
+          if (opts?.requeue === false || delivery.message.deliveryCount >= this.maxDeliveryCount) {
+            this.commit('drop', walId)
+          }
+          delivery.nack(opts)
+        }),
+    }
+  }
+
+  private commit(op: 'ack' | 'drop', id: number): void {
+    if (!this.live.has(id)) return
+    this.log.append(encode({ op, id }))
+    this.live.delete(id)
+  }
+}

src/index.ts

@@ -57,3 +57,15 @@ export type {
   KeyedWorkQueueOptions,
   Unsubscribe as KeyedUnsubscribe,
 } from './keyed-queue.js'
+
+export { WriteAheadLog, crc32 } from './wal.js'
+
+export type {
+  WriteAheadLogOptions,
+} from './wal.js'
+
+export { DurableWorkQueue } from './durable-work-queue.js'
+
+export type {
+  DurableWorkQueueOptions,
+} from './durable-work-queue.js'