diff --git a/README.md b/README.md index bf16bf0..6eed016 100644 --- a/README.md +++ b/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 diff --git a/package.json b/package.json index 688cf62..745d28d 100644 --- a/package.json +++ b/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" } diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index f227230..03328ec 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -8,12 +8,15 @@ importers: .: devDependencies: + '@types/node': + specifier: ^22.20.1 + version: 22.20.1 typescript: specifier: ^5.6.0 version: 5.9.3 vitest: specifier: ^2.1.0 - version: 2.1.9 + version: 2.1.9(@types/node@22.20.1) packages: @@ -299,6 +302,9 @@ packages: '@types/estree@1.0.9': resolution: {integrity: sha512-GhdPgy1el4/ImP05X05Uw4cw2/M93BCUmnEvWZNStlCzEKME4Fkk+YpoA5OiHNQmoS7Cafb8Xa3Pya8m1Qrzeg==} + '@types/node@22.20.1': + resolution: {integrity: sha512-EANqOCF9QFyra+4pfxUcX9STKJpCLjMbObVzljIJomAWSnuSIEAvyzEU53GaajbXJEgdh0iEcPL+DGvpUd4k1Q==} + '@vitest/expect@2.1.9': resolution: {integrity: sha512-UJCIkTBenHeKT1TTlKMJWy1laZewsRIzYighyYiJKZreqtdxSos/S1t+ktRMQWu2CKqaarrkeszJx1cgC5tGZw==} @@ -446,6 +452,9 @@ packages: engines: {node: '>=14.17'} hasBin: true + undici-types@6.21.0: + resolution: {integrity: sha512-iwDZqg0QAGrg9Rav5H4n0M64c3mkR59cJ6wQp+7C4nI0gsmExaedaYLNO44eT4AtBBwjbTiGPMlt2Md0T9H9JQ==} + vite-node@2.1.9: resolution: {integrity: sha512-AM9aQ/IPrW/6ENLQg3AGY4K1N2TGZdR5e4gu/MmmR2xR3Ll1+dib+nook92g4TV3PXVyeyxdWwtaCAiUL0hMxA==} engines: {node: ^18.0.0 || >=20.0.0} @@ -662,6 +671,10 @@ snapshots: '@types/estree@1.0.9': {} + '@types/node@22.20.1': + dependencies: + undici-types: 6.21.0 + '@vitest/expect@2.1.9': dependencies: '@vitest/spy': 2.1.9 @@ -669,13 +682,13 @@ snapshots: chai: 5.3.3 tinyrainbow: 1.2.0 - '@vitest/mocker@2.1.9(vite@5.4.21)': + '@vitest/mocker@2.1.9(vite@5.4.21(@types/node@22.20.1))': dependencies: '@vitest/spy': 2.1.9 estree-walker: 3.0.3 magic-string: 0.30.21 optionalDependencies: - vite: 5.4.21 + vite: 5.4.21(@types/node@22.20.1) '@vitest/pretty-format@2.1.9': dependencies: @@ -832,13 +845,15 @@ snapshots: typescript@5.9.3: {} - vite-node@2.1.9: + undici-types@6.21.0: {} + + vite-node@2.1.9(@types/node@22.20.1): dependencies: cac: 6.7.14 debug: 4.4.3 es-module-lexer: 1.7.0 pathe: 1.1.2 - vite: 5.4.21 + vite: 5.4.21(@types/node@22.20.1) transitivePeerDependencies: - '@types/node' - less @@ -850,18 +865,19 @@ snapshots: - supports-color - terser - vite@5.4.21: + vite@5.4.21(@types/node@22.20.1): dependencies: esbuild: 0.21.5 postcss: 8.5.19 rollup: 4.62.2 optionalDependencies: + '@types/node': 22.20.1 fsevents: 2.3.3 - vitest@2.1.9: + vitest@2.1.9(@types/node@22.20.1): dependencies: '@vitest/expect': 2.1.9 - '@vitest/mocker': 2.1.9(vite@5.4.21) + '@vitest/mocker': 2.1.9(vite@5.4.21(@types/node@22.20.1)) '@vitest/pretty-format': 2.1.9 '@vitest/runner': 2.1.9 '@vitest/snapshot': 2.1.9 @@ -877,9 +893,11 @@ snapshots: tinyexec: 0.3.2 tinypool: 1.1.1 tinyrainbow: 1.2.0 - vite: 5.4.21 - vite-node: 2.1.9 + vite: 5.4.21(@types/node@22.20.1) + vite-node: 2.1.9(@types/node@22.20.1) why-is-node-running: 2.3.0 + optionalDependencies: + '@types/node': 22.20.1 transitivePeerDependencies: - less - lightningcss diff --git a/src/durable-work-queue.ts b/src/durable-work-queue.ts new file mode 100644 index 0000000..0fbfab4 --- /dev/null +++ b/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 = { walId: number; body: T } + +type Op = + | { op: 'enqueue'; id: number; payload: T } + | { op: 'ack'; id: number } + | { op: 'drop'; id: number } + +export interface DurableWorkQueueOptions extends WorkQueueOptions, WriteAheadLogOptions {} + +function encode(op: Op): 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(bytes: Uint8Array): Op { + const value = JSON.parse(new TextDecoder().decode(bytes)) as Op + 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 { + private constructor( + private readonly log: WriteAheadLog, + private readonly inner: WorkQueue>, + private readonly live: Map, + private nextId: number, + private readonly maxDeliveryCount: number, + ) {} + + static open(path: string, options?: DurableWorkQueueOptions): DurableWorkQueue { + return DurableWorkQueue.fromLog(WriteAheadLog.open(path, options), options) + } + + static memory(options?: DurableWorkQueueOptions): DurableWorkQueue { + return DurableWorkQueue.fromLog(WriteAheadLog.memory(options), options) + } + + private static fromLog( + log: WriteAheadLog, + options?: DurableWorkQueueOptions, + ): DurableWorkQueue { + 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>({ maxDeliveryCount }) + const live = new Map() + let nextId = 1 + try { + for (const bytes of log.replay()) { + const rec = decode(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, 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>, gate: { settled: boolean }): Delivery { + 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) + } +} diff --git a/src/index.ts b/src/index.ts index fdfcd93..2d99aef 100644 --- a/src/index.ts +++ b/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' diff --git a/src/wal.ts b/src/wal.ts new file mode 100644 index 0000000..a5313eb --- /dev/null +++ b/src/wal.ts @@ -0,0 +1,211 @@ +import fs from 'node:fs' + +const MAGIC = Buffer.from('EBL1') +const MAX_PAYLOAD = 16 * 1024 * 1024 + +const CRC_TABLE = new Uint32Array(256) +for (let n = 0; n < 256; n++) { + let c = n + for (let k = 0; k < 8; k++) c = c & 1 ? 0xedb88320 ^ (c >>> 1) : c >>> 1 + CRC_TABLE[n] = c >>> 0 +} + +export function crc32(bytes: Uint8Array): number { + let c = 0xffffffff + for (const b of bytes) c = CRC_TABLE[(c ^ b) & 0xff]! ^ (c >>> 8) + return (c ^ 0xffffffff) >>> 0 +} + +export interface WriteAheadLogOptions { + readonly fsync?: boolean + readonly sync?: () => void +} + +function crc32LengthAndPayload(length: number, payload: Uint8Array): number { + const body = Buffer.allocUnsafe(4 + payload.byteLength) + body.writeUInt32LE(length, 0) + Buffer.from(payload).copy(body, 4) + return crc32(body) +} + +function frame(payload: Uint8Array): Buffer { + const out = Buffer.allocUnsafe(8 + payload.byteLength) + out.writeUInt32LE(payload.byteLength, 0) + out.writeUInt32LE(crc32LengthAndPayload(payload.byteLength, payload), 4) + Buffer.from(payload).copy(out, 8) + return out +} + +function parse(buf: Buffer): { payloads: Buffer[]; consumed: number } { + if (buf.length < 4 || !buf.subarray(0, 4).equals(MAGIC)) throw new Error('invalid wal magic') + const payloads: Buffer[] = [] + let offset = 4 + while (offset + 8 <= buf.length) { + const length = buf.readUInt32LE(offset) + const expected = buf.readUInt32LE(offset + 4) + const start = offset + 8 + const end = start + length + if (length > MAX_PAYLOAD || end > buf.length) break + const payload = buf.subarray(start, end) + // First bad CRC or incomplete header is a torn tail. Nothing after it is trusted. + if (crc32LengthAndPayload(length, payload) !== expected) break + payloads.push(Buffer.from(payload)) + offset = end + } + return { payloads, consumed: offset } +} + +function writeFully(fd: number, bytes: Buffer, position: number): number { + let offset = 0 + while (offset < bytes.length) { + const n = fs.writeSync(fd, bytes, offset, bytes.length - offset, position + offset) + if (n <= 0) throw new Error('short write') + offset += n + } + return offset +} + +function readFully(fd: number, size: number): Buffer { + if (size <= 0) return Buffer.alloc(0) + const buf = Buffer.allocUnsafe(size) + let offset = 0 + while (offset < size) { + const n = fs.readSync(fd, buf, offset, size - offset, offset) + if (n <= 0) return buf.subarray(0, offset) + offset += n + } + return buf +} + +export class WriteAheadLog { + private payloads: Buffer[] = [] + private mem = Buffer.alloc(0) + private fd: number | undefined + private path = '' + private size = 0 + private closed = false + + private constructor( + private readonly doSync: boolean, + private readonly injectedSync: (() => void) | undefined, + ) {} + + static open(path: string, options?: WriteAheadLogOptions): WriteAheadLog { + const fd = fs.existsSync(path) ? fs.openSync(path, 'r+') : fs.openSync(path, 'w+') + const log = new WriteAheadLog(options?.fsync !== false, options?.sync) + log.fd = fd + log.path = path + log.size = fs.fstatSync(fd).size + return log.boot() + } + + static memory(options?: WriteAheadLogOptions): WriteAheadLog { + return new WriteAheadLog(options?.fsync !== false, options?.sync).boot() + } + + append(payload: Uint8Array): number { + this.assertOpen() + if (payload.byteLength > MAX_PAYLOAD) throw new Error(`payload exceeds ${MAX_PAYLOAD} bytes`) + this.write(frame(payload)) + this.flush() + this.payloads.push(Buffer.from(payload)) + return this.payloads.length - 1 + } + + replay(): readonly Uint8Array[] { + this.assertOpen() + return this.payloads + } + + rewrite(payloads: Uint8Array[]): void { + this.assertOpen() + const kept = payloads.map((p) => Buffer.from(p)) + this.replace(Buffer.concat([MAGIC, ...kept.map(frame)])) + this.flush() + this.payloads = kept + } + + close(): void { + if (this.closed) return + this.closed = true + if (this.fd !== undefined) fs.closeSync(this.fd) + } + + private boot(): this { + try { + this.recover() + return this + } catch (error) { + this.close() + throw error + } + } + + private recover(): void { + const buf = this.readAll() + if (buf.length < 4) { + this.replace(MAGIC) + this.flush() + return + } + const { payloads, consumed } = parse(buf) + this.payloads = payloads + if (consumed < buf.length) this.truncate(consumed) + } + + private readAll(): Buffer { + if (this.fd === undefined) return this.mem + return readFully(this.fd, this.size) + } + + private write(bytes: Buffer): void { + if (this.fd === undefined) { + this.mem = Buffer.concat([this.mem, bytes]) + return + } + let offset = 0 + while (offset < bytes.length) { + const n = fs.writeSync(this.fd, bytes, offset, bytes.length - offset, this.size) + if (n <= 0) throw new Error('short write') + this.size += n + offset += n + } + } + + private replace(bytes: Buffer): void { + if (this.fd === undefined) { + this.mem = Buffer.from(bytes) + return + } + const tmp = `${this.path}.tmp` + const tfd = fs.openSync(tmp, 'w') + try { + writeFully(tfd, bytes, 0) + fs.fsyncSync(tfd) + } finally { + fs.closeSync(tfd) + } + fs.closeSync(this.fd) + fs.renameSync(tmp, this.path) + this.fd = fs.openSync(this.path, 'r+') + this.size = bytes.length + } + + private truncate(length: number): void { + if (this.fd === undefined) this.mem = this.mem.subarray(0, length) + else { + fs.ftruncateSync(this.fd, length) + this.size = length + } + } + + private flush(): void { + if (!this.doSync) return + if (this.injectedSync) this.injectedSync() + else if (this.fd !== undefined) fs.fsyncSync(this.fd) + } + + private assertOpen(): void { + if (this.closed) throw new Error('write-ahead log is closed') + } +} diff --git a/test/wal.test.ts b/test/wal.test.ts new file mode 100644 index 0000000..c8a70ec --- /dev/null +++ b/test/wal.test.ts @@ -0,0 +1,257 @@ +import fs from 'node:fs' +import os from 'node:os' +import path from 'node:path' +import { afterEach, describe, expect, it } from 'vitest' +import { crc32, DurableWorkQueue, WriteAheadLog, type Delivery } from '../src/index.js' + +const enc = (s: string) => Buffer.from(s) +const payloads = (log: WriteAheadLog) => [...log.replay()].map((b) => Buffer.from(b).toString()) + +const cleanups: Array<() => void> = [] +afterEach(() => { + while (cleanups.length > 0) cleanups.pop()?.() +}) + +function tmpFile(): string { + const dir = fs.mkdtempSync(path.join(os.tmpdir(), 'ebl-wal-')) + cleanups.push(() => fs.rmSync(dir, { recursive: true, force: true })) + return path.join(dir, 'log') +} + +function drain(q: DurableWorkQueue): T[] { + const seen: T[] = [] + q.consume((d) => { + seen.push(d.message.payload) + d.ack() + }) + return seen +} + +describe('WriteAheadLog', () => { + it('checksums, fsyncs on append, and preserves order including empty payloads', () => { + expect(crc32(new Uint8Array())).toBe(0) + expect(crc32(Buffer.alloc(4))).not.toBe(0) + expect(crc32(enc('123456789'))).toBe(0xcbf43926) + let syncs = 0 + const log = WriteAheadLog.memory({ sync: () => { syncs += 1 } }) + expect(payloads(log)).toEqual([]) + expect(log.append(enc('a'))).toBe(0) + expect(log.append(enc('b'))).toBe(1) + expect(log.append(enc(''))).toBe(2) + expect(payloads(log)).toEqual(['a', 'b', '']) + expect(syncs).toBe(4) + let skipped = 0 + WriteAheadLog.memory({ fsync: false, sync: () => { skipped += 1 } }).append(enc('y')) + expect(skipped).toBe(0) + log.close() + expect(() => log.append(enc('x'))).toThrow(/closed/) + }) + + it('survives reopen, truncates a torn or corrupt tail, and rewrite keeps the suffix', () => { + const file = tmpFile() + const log = WriteAheadLog.open(file) + log.append(enc('one')) + log.append(enc('two')) + log.close() + + const torn = tmpFile() + fs.copyFileSync(file, torn) + fs.appendFileSync(torn, Buffer.from([8, 0, 0, 0, 1, 2, 3])) + const afterTorn = WriteAheadLog.open(torn) + cleanups.push(() => afterTorn.close()) + expect(payloads(afterTorn)).toEqual(['one', 'two']) + afterTorn.append(enc('three')) + expect(payloads(afterTorn)).toEqual(['one', 'two', 'three']) + + const buf = fs.readFileSync(file) + const at = buf.lastIndexOf(Buffer.from('two')) + buf[at] = (buf[at] ?? 0) ^ 0xff + fs.writeFileSync(file, buf) + const afterCrc = WriteAheadLog.open(file) + cleanups.push(() => afterCrc.close()) + expect(payloads(afterCrc)).toEqual(['one']) + + const bad = tmpFile() + fs.writeFileSync(bad, 'XXXX') + expect(() => WriteAheadLog.open(bad)).toThrow(/magic/) + const stub = tmpFile() + fs.writeFileSync(stub, 'EB') + const reset = WriteAheadLog.open(stub) + cleanups.push(() => reset.close()) + expect(payloads(reset)).toEqual([]) + + const compacted = tmpFile() + const src = WriteAheadLog.open(compacted) + src.append(enc('a')) + src.append(enc('b')) + src.append(enc('c')) + src.rewrite([enc('b'), enc('c')]) + src.close() + const reopened = WriteAheadLog.open(compacted) + cleanups.push(() => reopened.close()) + expect(payloads(reopened)).toEqual(['b', 'c']) + }) + + it('truncates an eight-byte zero tail and keeps only real payloads', () => { + const file = tmpFile() + const log = WriteAheadLog.open(file) + log.append(enc('one')) + log.append(enc('two')) + log.close() + const size = fs.statSync(file).size + fs.appendFileSync(file, Buffer.alloc(8)) + const reopened = WriteAheadLog.open(file) + cleanups.push(() => reopened.close()) + expect(payloads(reopened)).toEqual(['one', 'two']) + expect(fs.statSync(file).size).toBe(size) + }) + + it('does not replay a well-formed record after an eight-byte zero hole', () => { + const prefix = tmpFile() + const first = WriteAheadLog.open(prefix) + first.append(enc('one')) + first.close() + const prefixSize = fs.statSync(prefix).size + + const full = tmpFile() + const second = WriteAheadLog.open(full) + second.append(enc('one')) + second.append(enc('sneak')) + second.close() + const sneak = fs.readFileSync(full).subarray(prefixSize) + + const file = tmpFile() + fs.copyFileSync(prefix, file) + fs.appendFileSync(file, Buffer.alloc(8)) + fs.appendFileSync(file, sneak) + const reopened = WriteAheadLog.open(file) + cleanups.push(() => reopened.close()) + expect(payloads(reopened)).toEqual(['one']) + expect(fs.statSync(file).size).toBe(prefixSize) + }) + + it('keeps a record appended after torn-tail recovery across reopen', () => { + const file = tmpFile() + const log = WriteAheadLog.open(file) + log.append(enc('one')) + log.append(enc('two')) + log.close() + fs.appendFileSync(file, Buffer.from([8, 0, 0, 0, 1, 2, 3])) + const recovered = WriteAheadLog.open(file) + recovered.append(enc('three')) + recovered.close() + const reopened = WriteAheadLog.open(file) + cleanups.push(() => reopened.close()) + expect(payloads(reopened)).toEqual(['one', 'two', 'three']) + }) +}) + +describe('DurableWorkQueue crash recovery', () => { + it('redelivers unacked work with stable ids', () => { + const file = tmpFile() + const q = DurableWorkQueue.open(file) + expect(q.enqueue('a')).toBe(1) + expect(q.enqueue('b')).toBe(2) + const held: string[] = [] + q.consume((d) => held.push(d.message.payload)) + expect(held).toEqual(['a']) + q.close() + const recovered = DurableWorkQueue.open(file) + cleanups.push(() => recovered.close()) + const seen: Array<{ id: number; payload: string }> = [] + recovered.consume((d) => { + seen.push({ id: d.message.id, payload: d.message.payload }) + d.ack() + }) + expect(seen).toEqual([ + { id: 1, payload: 'a' }, + { id: 2, payload: 'b' }, + ]) + expect(recovered.enqueue('c')).toBe(3) + }) + + it('does not resurrect acked, dropped, retried-then-acked, or max-delivery work', () => { + const file = tmpFile() + const q = DurableWorkQueue.open(file, { maxDeliveryCount: 2 }) + let threw = false + q.consume((d) => { + if (d.message.payload === 'drop-me') d.nack({ requeue: false }) + else if (d.message.payload === 'poison') d.nack() + else if (d.message.payload === 'boom' && !threw) { + threw = true + throw new Error('boom') + } else d.ack() + }) + q.enqueue('done') + q.enqueue('drop-me') + q.enqueue('poison') + q.enqueue('boom') + q.close() + const recovered = DurableWorkQueue.open(file) + cleanups.push(() => recovered.close()) + expect(drain(recovered)).toEqual([]) + expect(() => DurableWorkQueue.memory().enqueue(undefined)).toThrow(/JSON-serializable/) + expect(() => DurableWorkQueue.memory({ maxDeliveryCount: 0 })).toThrow(/maxDeliveryCount/) + }) + + it('checkpoint rewrites to live enqueue records only', () => { + const file = tmpFile() + const q = DurableWorkQueue.open<{ job: string }>(file) + q.enqueue({ job: 'a' }) + q.enqueue({ job: 'b' }) + q.consume((d) => { + if (d.message.payload.job === 'a') d.ack() + }) + q.checkpoint() + q.close() + const recovered = DurableWorkQueue.open<{ job: string }>(file) + cleanups.push(() => recovered.close()) + expect(drain(recovered).map((m) => m.job)).toEqual(['b']) + }) + + it('redelivers a prefix enqueue when the log tail is eight zeros', () => { + const file = tmpFile() + const q = DurableWorkQueue.open(file) + q.enqueue('keep-me') + q.close() + fs.appendFileSync(file, Buffer.alloc(8)) + const recovered = DurableWorkQueue.open(file) + cleanups.push(() => recovered.close()) + expect(drain(recovered)).toEqual(['keep-me']) + }) + + it('stale ack after unsubscribe does not drop requeued work from the log', () => { + const file = tmpFile() + const q = DurableWorkQueue.open(file) + let stolen: Delivery | undefined + const off = q.consume((d) => { + stolen = d + }) + q.enqueue('held') + off() + expect(q.readyCount()).toBe(1) + stolen!.ack() + expect(q.readyCount()).toBe(1) + q.close() + const recovered = DurableWorkQueue.open(file) + cleanups.push(() => recovered.close()) + expect(drain(recovered)).toEqual(['held']) + }) + + it('stale drop-nack after unsubscribe does not drop requeued work from the log', () => { + const file = tmpFile() + const q = DurableWorkQueue.open(file) + let stolen: Delivery | undefined + const off = q.consume((d) => { + stolen = d + }) + q.enqueue('held') + off() + stolen!.nack({ requeue: false }) + expect(q.readyCount()).toBe(1) + q.close() + const recovered = DurableWorkQueue.open(file) + cleanups.push(() => recovered.close()) + expect(drain(recovered)).toEqual(['held']) + }) +}) diff --git a/tsconfig.json b/tsconfig.json index f9c273e..6d1f997 100644 --- a/tsconfig.json +++ b/tsconfig.json @@ -8,7 +8,7 @@ "noImplicitOverride": true, "esModuleInterop": true, "skipLibCheck": true, - "types": ["vitest/globals"], + "types": ["vitest/globals", "node"], "outDir": "dist", "declaration": true },