From b109312cc5808259fbe36fd947660eb8367c3e36 Mon Sep 17 00:00:00 2001 From: Thomas Hart Date: Mon, 31 Aug 2026 22:07:03 +0000 Subject: [PATCH] feat: add per-topic throughput and lag metrics TopicMetrics records produced and consumed against a topic. Lag is log end offset minus the consumer's committed offset. Rates use a sliding window of circular time buckets. Restores DeadLetterQueue barrel exports dropped by the work-queue merge. --- README.md | 8 ++ src/index.ts | 7 ++ src/metrics.ts | 234 +++++++++++++++++++++++++++++++++++++++++++ test/metrics.test.ts | 221 ++++++++++++++++++++++++++++++++++++++++ 4 files changed, 470 insertions(+) create mode 100644 src/metrics.ts create mode 100644 test/metrics.test.ts diff --git a/README.md b/README.md index 6eed016..01804fa 100644 --- a/README.md +++ b/README.md @@ -28,6 +28,10 @@ Message brokers hide a lot of machinery behind `publish` and `subscribe`. This r - **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 +- **Consumer lag** as `logEndOffset - committedOffset`, tracked per topic and per consumer +- **Time lag** as the age of the oldest unconsumed produce +- **Sliding-window throughput** (circular time buckets) for produce and consume rates, so a quiet period after a burst reports ~0 + ## 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). @@ -47,6 +51,10 @@ Message brokers hide a lot of machinery behind `publish` and `subscribe`. This r - **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). +- **Consumer lag** as `logEndOffset - committedOffset`, tracked per topic and per consumer +- **Time lag** as the age of the oldest unconsumed produce +- **Sliding-window throughput** (circular time buckets) for produce and consume rates, so a quiet period after a burst reports ~0 +- **Throughput and lag metrics per topic.** `TopicMetrics` records `produced` / `consumed` per topic. Lag is `logEndOffset - committedOffset` (Kafka's formula), time lag is the age of the oldest unconsumed record, and produce/consume rates use a sliding window of circular time buckets. Two consumers on one topic keep independent commits and rates. ## Usage ```ts diff --git a/src/index.ts b/src/index.ts index 2d99aef..c08d152 100644 --- a/src/index.ts +++ b/src/index.ts @@ -69,3 +69,10 @@ export { DurableWorkQueue } from './durable-work-queue.js' export type { DurableWorkQueueOptions, } from './durable-work-queue.js' + +export { TopicMetrics, DEFAULT_CONSUMER } from './metrics.js' + +export type { + TopicSnapshot, + TopicMetricsOptions, +} from './metrics.js' diff --git a/src/metrics.ts b/src/metrics.ts new file mode 100644 index 0000000..3506054 --- /dev/null +++ b/src/metrics.ts @@ -0,0 +1,234 @@ +export const DEFAULT_CONSUMER = 'default' + +export interface TopicSnapshot { + readonly topic: string + readonly consumer: string + readonly logEndOffset: number + readonly committedOffset: number + readonly lag: number + readonly produceRate: number + readonly consumeRate: number + readonly oldestAgeMs: number +} + +export interface TopicMetricsOptions { + readonly windowMs?: number + readonly bucketMs?: number + readonly now?: () => number +} + +const DEFAULT_WINDOW_MS = 10_000 +const DEFAULT_BUCKET_MS = 1_000 +const MAX_BUCKETS = 1_000 + +class SlidingWindow { + private readonly counts: number[] + private readonly bucketMs: number + private readonly size: number + private head = 0 + private cursor = 0 + private primed = false + + constructor(windowMs: number, bucketMs: number) { + this.bucketMs = bucketMs + this.size = windowMs / bucketMs + this.counts = new Array(this.size).fill(0) + } + + add(now: number, n: number): void { + this.advance(now) + const t = now < this.cursor ? this.cursor : now + const offset = Math.min(this.size - 1, Math.floor((t - this.cursor) / this.bucketMs)) + const i = (this.head + offset) % this.size + const current = this.counts[i] + this.counts[i] = (current ?? 0) + n + } + + sum(now: number): number { + this.advance(now) + let total = 0 + for (const count of this.counts) total += count + return total + } + + private advance(now: number): void { + if (!this.primed) { + this.cursor = now - (now % this.bucketMs) + this.primed = true + return + } + if (now < this.cursor) return + const offset = Math.floor((now - this.cursor) / this.bucketMs) + if (offset < this.size) return + const shift = offset - this.size + 1 + if (shift >= this.size) { + // A jump past one full window would walk the ring more than once; reset instead. + this.counts.fill(0) + this.head = 0 + this.cursor = now - (now % this.bucketMs) - (this.size - 1) * this.bucketMs + return + } + for (let s = 0; s < shift; s++) { + this.counts[this.head] = 0 + this.head = (this.head + 1) % this.size + this.cursor += this.bucketMs + } + } +} + +interface ConsumerState { + committedOffset: number + readonly consumeWindow: SlidingWindow +} + +interface TopicState { + logEndOffset: number + readonly produceTimes: number[] + readonly produceWindow: SlidingWindow + readonly consumers: Map +} + +export class TopicMetrics { + readonly windowMs: number + readonly bucketMs: number + private readonly now: () => number + private readonly topics = new Map() + + constructor(options: TopicMetricsOptions = {}) { + const windowMs = options.windowMs ?? DEFAULT_WINDOW_MS + const bucketMs = options.bucketMs ?? DEFAULT_BUCKET_MS + if (!Number.isInteger(windowMs) || windowMs < 1) { + throw new Error(`windowMs must be a positive integer, got ${windowMs}`) + } + if (!Number.isInteger(bucketMs) || bucketMs < 1) { + throw new Error(`bucketMs must be a positive integer, got ${bucketMs}`) + } + if (windowMs % bucketMs !== 0) { + throw new Error(`windowMs (${windowMs}) must be a multiple of bucketMs (${bucketMs})`) + } + const buckets = windowMs / bucketMs + if (buckets > MAX_BUCKETS) { + throw new Error(`windowMs / bucketMs must be <= ${MAX_BUCKETS}, got ${buckets}`) + } + this.windowMs = windowMs + this.bucketMs = bucketMs + this.now = options.now ?? Date.now + } + + produced(topic: string): number { + this.assertName(topic, 'topic') + const t = this.timestamp() + const state = this.topicState(topic) + const offset = state.logEndOffset + state.logEndOffset += 1 + state.produceTimes.push(t) + state.produceWindow.add(t, 1) + return offset + } + + consumed(topic: string, consumer: string = DEFAULT_CONSUMER): boolean { + this.assertName(topic, 'topic') + this.assertName(consumer, 'consumer') + const state = this.topics.get(topic) + if (!state) return false + const group = this.consumerState(state, consumer) + if (group.committedOffset >= state.logEndOffset) return false + group.committedOffset += 1 + group.consumeWindow.add(this.timestamp(), 1) + return true + } + + snapshot(topic: string, consumer: string = DEFAULT_CONSUMER): TopicSnapshot { + this.assertName(topic, 'topic') + this.assertName(consumer, 'consumer') + const state = this.topics.get(topic) + if (!state) { + return { + topic, + consumer, + logEndOffset: 0, + committedOffset: 0, + lag: 0, + produceRate: 0, + consumeRate: 0, + oldestAgeMs: 0, + } + } + return this.capture(topic, state, consumer) + } + + snapshots(): TopicSnapshot[] { + const rows: TopicSnapshot[] = [] + for (const [topic, state] of this.topics) { + if (state.consumers.size === 0) { + rows.push(this.capture(topic, state, DEFAULT_CONSUMER)) + continue + } + for (const consumer of state.consumers.keys()) { + rows.push(this.capture(topic, state, consumer)) + } + } + rows.sort((a, b) => a.topic.localeCompare(b.topic) || a.consumer.localeCompare(b.consumer)) + return rows + } + + private capture(topic: string, state: TopicState, consumer: string): TopicSnapshot { + const t = this.timestamp() + const group = state.consumers.get(consumer) + const committedOffset = group?.committedOffset ?? 0 + const lag = state.logEndOffset - committedOffset + const producedAt = state.produceTimes[committedOffset] + const oldestAgeMs = + lag > 0 && producedAt !== undefined ? Math.max(0, t - producedAt) : 0 + return { + topic, + consumer, + logEndOffset: state.logEndOffset, + committedOffset, + lag, + produceRate: state.produceWindow.sum(t) * (1000 / this.windowMs), + consumeRate: group ? group.consumeWindow.sum(t) * (1000 / this.windowMs) : 0, + oldestAgeMs, + } + } + + private topicState(topic: string): TopicState { + let state = this.topics.get(topic) + if (!state) { + state = { + logEndOffset: 0, + produceTimes: [], + produceWindow: new SlidingWindow(this.windowMs, this.bucketMs), + consumers: new Map(), + } + this.topics.set(topic, state) + } + return state + } + + private consumerState(state: TopicState, consumer: string): ConsumerState { + let group = state.consumers.get(consumer) + if (!group) { + group = { + committedOffset: 0, + consumeWindow: new SlidingWindow(this.windowMs, this.bucketMs), + } + state.consumers.set(consumer, group) + } + return group + } + + private timestamp(): number { + const t = this.now() + if (!Number.isFinite(t)) { + throw new Error(`now() must return a finite number, got ${t}`) + } + return t + } + + private assertName(value: string, label: string): void { + if (value.length === 0) { + throw new Error(`${label} must be a non-empty string`) + } + } +} diff --git a/test/metrics.test.ts b/test/metrics.test.ts new file mode 100644 index 0000000..501a9ab --- /dev/null +++ b/test/metrics.test.ts @@ -0,0 +1,221 @@ +import { describe, it, expect } from 'vitest' +import { TopicMetrics, DEFAULT_CONSUMER } from '../src/index.js' + +function clock(start = 0) { + const t = { now: start } + const metrics = (windowMs = 1000, bucketMs = 100) => + new TopicMetrics({ windowMs, bucketMs, now: () => t.now }) + return { t, metrics } +} + +describe('TopicMetrics construction', () => { + it('rejects bad windows, empty names, and a non-finite clock', () => { + expect(() => new TopicMetrics({ windowMs: 0 })).toThrow(/windowMs/) + expect(() => new TopicMetrics({ windowMs: 1.5 })).toThrow(/windowMs/) + expect(() => new TopicMetrics({ bucketMs: 0 })).toThrow(/bucketMs/) + expect(() => new TopicMetrics({ windowMs: 1000, bucketMs: 300 })).toThrow(/multiple/) + expect(() => new TopicMetrics({ windowMs: 2000, bucketMs: 1 })).toThrow(/<= 1000/) + const metrics = new TopicMetrics({ now: () => Number.NaN }) + expect(() => metrics.produced('')).toThrow(/topic/) + expect(() => new TopicMetrics().consumed('orders', '')).toThrow(/consumer/) + expect(() => metrics.produced('orders')).toThrow(/finite/) + }) +}) + +describe('offset lag', () => { + it('assigns 0-based offsets and reports lag as log end minus committed', () => { + const metrics = new TopicMetrics({ now: () => 0 }) + expect(metrics.produced('orders')).toBe(0) + expect(metrics.produced('orders')).toBe(1) + expect(metrics.produced('orders')).toBe(2) + expect(metrics.snapshot('orders')).toMatchObject({ + topic: 'orders', + consumer: DEFAULT_CONSUMER, + logEndOffset: 3, + committedOffset: 0, + lag: 3, + oldestAgeMs: 0, + }) + }) + + it('drops lag on FIFO consume and is a no-op once caught up', () => { + const metrics = new TopicMetrics({ now: () => 0 }) + metrics.produced('orders') + metrics.produced('orders') + expect(metrics.consumed('orders')).toBe(true) + expect(metrics.snapshot('orders').lag).toBe(1) + expect(metrics.consumed('orders')).toBe(true) + expect(metrics.snapshot('orders')).toMatchObject({ + logEndOffset: 2, + committedOffset: 2, + lag: 0, + oldestAgeMs: 0, + }) + expect(metrics.consumed('orders')).toBe(false) + expect(metrics.snapshot('orders').committedOffset).toBe(2) + }) + + it('does not consume or materialize a topic that has never been produced', () => { + const metrics = new TopicMetrics() + expect(metrics.consumed('ghost')).toBe(false) + expect(metrics.snapshot('ghost')).toEqual({ + topic: 'ghost', + consumer: DEFAULT_CONSUMER, + logEndOffset: 0, + committedOffset: 0, + lag: 0, + produceRate: 0, + consumeRate: 0, + oldestAgeMs: 0, + }) + metrics.snapshot('orders') + expect(metrics.snapshots()).toEqual([]) + }) + + it('isolates offsets across topics', () => { + const metrics = new TopicMetrics({ now: () => 0 }) + metrics.produced('orders') + metrics.produced('orders') + metrics.produced('shipments') + metrics.consumed('orders') + expect(metrics.snapshot('orders')).toMatchObject({ lag: 1, logEndOffset: 2 }) + expect(metrics.snapshot('shipments')).toMatchObject({ lag: 1, logEndOffset: 1 }) + }) + + it('tracks independent committed offsets per consumer on the same topic', () => { + const { t, metrics } = clock() + const m = metrics() + m.produced('orders') + t.now = 10 + m.produced('orders') + t.now = 20 + m.produced('orders') + t.now = 50 + expect(m.consumed('orders', 'billing')).toBe(true) + expect(m.consumed('orders', 'billing')).toBe(true) + expect(m.snapshot('orders', 'billing')).toMatchObject({ + committedOffset: 2, + lag: 1, + oldestAgeMs: 30, + }) + expect(m.snapshot('orders', 'search')).toMatchObject({ + committedOffset: 0, + lag: 3, + oldestAgeMs: 50, + }) + expect(m.consumed('orders', 'search')).toBe(true) + expect(m.snapshot('orders', 'search').oldestAgeMs).toBe(40) + expect(m.snapshot('orders', 'billing').lag).toBe(1) + }) +}) + +describe('time lag', () => { + it('ages the oldest unconsumed produce and resets after catch-up', () => { + const { t, metrics } = clock() + const m = metrics() + m.produced('orders') + t.now = 40 + m.produced('orders') + t.now = 90 + expect(m.snapshot('orders').oldestAgeMs).toBe(90) + m.consumed('orders') + expect(m.snapshot('orders').oldestAgeMs).toBe(50) + m.consumed('orders') + expect(m.snapshot('orders').oldestAgeMs).toBe(0) + t.now = 10 + expect(m.snapshot('orders').oldestAgeMs).toBe(0) + }) +}) + +describe('sliding-window throughput', () => { + it('reports produce rate as window sum over window seconds', () => { + const { t, metrics } = clock() + const wide = metrics(10_000, 1_000) + for (let i = 0; i < 10; i++) wide.produced('orders') + expect(wide.snapshot('orders').produceRate).toBe(1) + expect(wide.snapshot('orders').consumeRate).toBe(0) + const burst = metrics() + for (let i = 0; i < 5; i++) burst.produced('shipments') + expect(burst.snapshot('shipments').produceRate).toBe(5) + t.now = 400 + burst.produced('shipments') + expect(burst.snapshot('shipments').produceRate).toBe(6) + }) + + it('expires produces that have left the window and keeps later ones', () => { + const { t, metrics } = clock() + const m = metrics() + m.produced('orders') + t.now = 800 + m.produced('orders') + m.produced('orders') + t.now = 1500 + expect(m.snapshot('orders')).toMatchObject({ produceRate: 2, logEndOffset: 3, lag: 3 }) + }) + + it('records consume rate only for successful commits', () => { + const { t, metrics } = clock() + const m = metrics() + m.produced('orders') + m.produced('orders') + expect(m.consumed('orders')).toBe(true) + expect(m.consumed('orders')).toBe(true) + expect(m.consumed('orders')).toBe(false) + expect(m.snapshot('orders').consumeRate).toBe(2) + t.now = 1500 + expect(m.snapshot('orders').consumeRate).toBe(0) + expect(m.snapshot('orders').committedOffset).toBe(2) + }) + + it('zeros rates after a jump larger than the window, without resetting offsets', () => { + const { t, metrics } = clock() + const m = metrics() + m.produced('orders') + m.produced('orders') + m.consumed('orders') + t.now = 50_000 + expect(m.snapshot('orders')).toMatchObject({ + produceRate: 0, + consumeRate: 0, + logEndOffset: 2, + committedOffset: 1, + lag: 1, + }) + }) + + it('keeps independent consume rates per consumer', () => { + const { t, metrics } = clock() + const m = metrics() + m.produced('orders') + m.produced('orders') + m.produced('orders') + m.consumed('orders', 'billing') + t.now = 200 + m.consumed('orders', 'billing') + m.consumed('orders', 'search') + expect(m.snapshot('orders', 'billing').consumeRate).toBe(2) + expect(m.snapshot('orders', 'search').consumeRate).toBe(1) + expect(m.snapshot('orders', 'billing').produceRate).toBe( + m.snapshot('orders', 'search').produceRate, + ) + }) +}) + +describe('snapshots listing', () => { + it('lists produced topics with the default consumer until a named one commits', () => { + const metrics = new TopicMetrics({ now: () => 0 }) + metrics.produced('zeta') + metrics.produced('alpha') + expect(metrics.snapshots().map((s) => [s.topic, s.consumer, s.lag])).toEqual([ + ['alpha', DEFAULT_CONSUMER, 1], + ['zeta', DEFAULT_CONSUMER, 1], + ]) + metrics.consumed('alpha', 'billing') + metrics.consumed('alpha', 'search') + expect(metrics.snapshots().map((s) => [s.topic, s.consumer, s.lag])).toEqual([ + ['alpha', 'billing', 0], + ['alpha', 'search', 0], + ['zeta', DEFAULT_CONSUMER, 1], + ]) + }) +})