diff --git a/.github/workflows/architecture-check.yml b/.github/workflows/architecture-check.yml index 1240339..bfd8183 100644 --- a/.github/workflows/architecture-check.yml +++ b/.github/workflows/architecture-check.yml @@ -1,5 +1,12 @@ name: Architecture Validation +# Fork PRs run with a read-only GITHUB_TOKEN by default; the PR-comment +# step needs write access to post the architecture report. +permissions: + contents: read + issues: write + pull-requests: write + on: push: branches: @@ -130,6 +137,9 @@ jobs: - name: Comment on PR with results if: github.event_name == 'pull_request' && always() + # Fork PRs run with a read-only GITHUB_TOKEN that cannot write + # comments, so a skipped report comment must not fail the check. + continue-on-error: true uses: actions/github-script@v7 with: script: | diff --git a/README.md b/README.md index c6b70a4..5d411b2 100644 --- a/README.md +++ b/README.md @@ -275,6 +275,14 @@ sudo apt-get install redis-server redis-server ``` +> **Zero-loss messaging**: the queue's zero-loss message service persists +> message state (payloads, attempts, processing leases, replication targets) +> in Redis under the `zls:*` key namespace. All state survives process +> restarts and is shared across horizontally scaled instances. A background +> recovery sweep re-queues messages whose processing lease expired (worker +> crash) exactly once, so Redis must be reachable for zero-loss guarantees +> to hold. + ### 4. Database Setup #### Development (SQLite) diff --git a/src/queue/horizontal-scaling.controller.ts b/src/queue/horizontal-scaling.controller.ts index 023069e..896098a 100644 --- a/src/queue/horizontal-scaling.controller.ts +++ b/src/queue/horizontal-scaling.controller.ts @@ -310,7 +310,7 @@ export class HorizontalScalingController { @Get('zero-loss/stats') @ApiOperation({ summary: 'Get zero-loss message statistics' }) @ApiResponse({ status: 200, description: 'Zero-loss stats' }) - getZeroLossStats() { + async getZeroLossStats() { return this.zeroLoss.getStats(); } @@ -318,8 +318,8 @@ export class HorizontalScalingController { @ApiOperation({ summary: 'Get message by ID' }) @ApiResponse({ status: 200, description: 'Message details' }) @ApiResponse({ status: 404, description: 'Message not found' }) - getMessage(@Param('messageId') messageId: string) { - const message = this.zeroLoss.getMessage(messageId); + async getMessage(@Param('messageId') messageId: string) { + const message = await this.zeroLoss.getMessage(messageId); if (!message) { return { error: 'Message not found' }; } @@ -329,24 +329,24 @@ export class HorizontalScalingController { @Get('zero-loss/messages/queue/:queueName') @ApiOperation({ summary: 'Get messages for a queue' }) @ApiResponse({ status: 200, description: 'Queue messages' }) - getQueueMessages(@Param('queueName') queueName: string) { + async getQueueMessages(@Param('queueName') queueName: string) { return { - messages: this.zeroLoss.getQueueMessages(queueName), + messages: await this.zeroLoss.getQueueMessages(queueName), }; } @Post('zero-loss/messages/:messageId/retry') @ApiOperation({ summary: 'Retry a failed message' }) @ApiResponse({ status: 200, description: 'Retry initiated' }) - retryMessage(@Param('messageId') messageId: string) { - const success = this.zeroLoss.retryMessage(messageId); + async retryMessage(@Param('messageId') messageId: string) { + const success = await this.zeroLoss.retryMessage(messageId); return { success }; } @Get('zero-loss/messages/:messageId/verify') @ApiOperation({ summary: 'Verify message integrity' }) @ApiResponse({ status: 200, description: 'Integrity check result' }) - verifyMessageIntegrity(@Param('messageId') messageId: string) { + async verifyMessageIntegrity(@Param('messageId') messageId: string) { return this.zeroLoss.verifyMessageIntegrity(messageId); } diff --git a/src/queue/horizontal-scaling.module.ts b/src/queue/horizontal-scaling.module.ts index cb48d23..c243e0a 100644 --- a/src/queue/horizontal-scaling.module.ts +++ b/src/queue/horizontal-scaling.module.ts @@ -3,6 +3,7 @@ import { Module } from '@nestjs/common'; import { ConfigModule } from '@nestjs/config'; import { ScheduleModule } from '@nestjs/schedule'; import { EventEmitterModule } from '@nestjs/event-emitter'; +import { CustomCacheModule } from '../common/cache/cache.module'; import { QueueWorkerManagerService } from './queue-worker-manager.service'; import { QueueLoadBalancerService } from './queue-load-balancer.service'; import { QueueFaultToleranceService } from './queue-fault-tolerance.service'; @@ -23,6 +24,9 @@ import { HorizontalScalingController } from './horizontal-scaling.controller'; ConfigModule, ScheduleModule.forRoot(), EventEmitterModule.forRoot(), + // Provides RedisPoolService — the durable store backing the zero-loss + // message layer (same Redis infrastructure Bull already uses). + CustomCacheModule, ], controllers: [HorizontalScalingController], providers: [ diff --git a/src/queue/zero-loss-message.service.spec.ts b/src/queue/zero-loss-message.service.spec.ts new file mode 100644 index 0000000..911c88f --- /dev/null +++ b/src/queue/zero-loss-message.service.spec.ts @@ -0,0 +1,383 @@ +// src/queue/zero-loss-message.service.spec.ts +import { Test, TestingModule } from '@nestjs/testing'; +import { EventEmitter2 } from '@nestjs/event-emitter'; +import { ZeroLossMessageService } from './zero-loss-message.service'; +import { RedisPoolService } from '../common/cache/redis-pool.service'; + +/** + * Faithful in-memory implementation of the ioredis command surface the + * zero-loss service uses. It models Redis semantics that matter here: + * `SET ... EX ... NX` atomicity (single-threaded, so the NX guard is exact), + * per-key TTL expiry, hashes, sets and SCAN. A test "restarts" a service by + * building a new instance over the same fake — exactly like two deployment + * instances sharing one real Redis. + */ +type FakeEntry = + | { type: 'string'; value: string; expiresAtMs: number | null } + | { type: 'hash'; fields: Map; expiresAtMs: number | null } + | { type: 'set'; members: Set; expiresAtMs: number | null }; + +class FakeRedis { + private store = new Map(); + nowMs: () => number = () => Date.now(); + + private isAlive(key: string): boolean { + const entry = this.store.get(key); + if (!entry) return false; + if (entry.expiresAtMs !== null && entry.expiresAtMs <= this.nowMs()) { + this.store.delete(key); + return false; + } + return true; + } + + get(key: string): Promise { + const entry = this.store.get(key); + if (!this.isAlive(key) || !entry || entry.type !== 'string') + return Promise.resolve(null); + return Promise.resolve(entry.value); + } + + set(key: string, value: string, ...args: unknown[]): Promise<'OK' | null> { + const exIndex = args.indexOf('EX'); + const nx = args.includes('NX'); + if (nx && this.isAlive(key)) return Promise.resolve(null); + const seconds = exIndex >= 0 ? Number(args[exIndex + 1]) : 0; + this.store.set(key, { + type: 'string', + value, + expiresAtMs: seconds > 0 ? this.nowMs() + seconds * 1000 : null, + }); + return Promise.resolve('OK'); + } + + del(...keys: string[]): Promise { + let removed = 0; + for (const key of keys) { + if (this.isAlive(key)) { + this.store.delete(key); + removed++; + } + } + return Promise.resolve(removed); + } + + expire(key: string, seconds: number): Promise { + const entry = this.store.get(key); + if (!this.isAlive(key) || !entry) return Promise.resolve(0); + entry.expiresAtMs = this.nowMs() + seconds * 1000; + return Promise.resolve(1); + } + + hget(key: string, field: string): Promise { + const entry = this.store.get(key); + if (!this.isAlive(key) || !entry || entry.type !== 'hash') + return Promise.resolve(null); + return Promise.resolve(entry.fields.get(field) ?? null); + } + + hset( + key: string, + ...args: Array> + ): Promise { + // ioredis accepts both `hset(key, field, value, ...)` and the object + // form `hset(key, { field: value })`; flatten the latter like ioredis does. + const fields = + args.length === 1 && typeof args[0] === 'object' + ? Object.entries(args[0]).flat() + : (args as string[]); + let entry = this.store.get(key); + if (!this.isAlive(key) || !entry || entry.type !== 'hash') { + entry = { type: 'hash', fields: new Map(), expiresAtMs: null }; + this.store.set(key, entry); + } + let added = 0; + for (let i = 0; i < fields.length; i += 2) { + if (!entry.fields.has(fields[i])) added++; + entry.fields.set(fields[i], fields[i + 1]); + } + return Promise.resolve(added); + } + + hgetall(key: string): Promise> { + const entry = this.store.get(key); + if (!this.isAlive(key) || !entry || entry.type !== 'hash') + return Promise.resolve({}); + return Promise.resolve(Object.fromEntries(entry.fields)); + } + + hincrby(key: string, field: string, incr: number): Promise { + let entry = this.store.get(key); + if (!this.isAlive(key) || !entry || entry.type !== 'hash') { + entry = { type: 'hash', fields: new Map(), expiresAtMs: null }; + this.store.set(key, entry); + } + const next = Number(entry.fields.get(field) ?? 0) + incr; + entry.fields.set(field, String(next)); + return Promise.resolve(next); + } + + sadd(key: string, ...members: string[]): Promise { + let entry = this.store.get(key); + if (!this.isAlive(key) || !entry || entry.type !== 'set') { + entry = { type: 'set', members: new Set(), expiresAtMs: null }; + this.store.set(key, entry); + } + let added = 0; + for (const member of members) { + if (!entry.members.has(member)) { + entry.members.add(member); + added++; + } + } + return Promise.resolve(added); + } + + srem(key: string, ...members: string[]): Promise { + const entry = this.store.get(key); + if (!this.isAlive(key) || !entry || entry.type !== 'set') + return Promise.resolve(0); + let removed = 0; + for (const member of members) { + if (entry.members.delete(member)) removed++; + } + return Promise.resolve(removed); + } + + smembers(key: string): Promise { + const entry = this.store.get(key); + if (!this.isAlive(key) || !entry || entry.type !== 'set') + return Promise.resolve([]); + return Promise.resolve([...entry.members]); + } + + scard(key: string): Promise { + const entry = this.store.get(key); + if (!this.isAlive(key) || !entry || entry.type !== 'set') + return Promise.resolve(0); + return Promise.resolve(entry.members.size); + } + + scan( + cursor: string, + match: string, + pattern: string, + count: string, + limit: number, + ): Promise<[string, string[]]> { + void match; + void count; + void limit; + const prefix = pattern.slice(0, pattern.indexOf('*')); + const keys: string[] = []; + for (const key of this.store.keys()) { + if (this.isAlive(key) && key.startsWith(prefix)) keys.push(key); + } + return Promise.resolve(['0', keys]); + } + + flushAll(): void { + this.store.clear(); + } + + /** Advance the fake clock by ms (expires TTL'd keys). */ + advance(ms: number): void { + const base = Date.now(); + this.nowMs = () => base + ms; + } +} + +describe('ZeroLossMessageService (durable store)', () => { + let fake: FakeRedis; + let eventEmitter: { emit: jest.Mock }; + + const requeueCount = (): number => + eventEmitter.emit.mock.calls.filter(([name]) => name === 'message.requeued') + .length; + + const buildService = async (): Promise => { + const module: TestingModule = await Test.createTestingModule({ + providers: [ + ZeroLossMessageService, + { + provide: RedisPoolService, + useValue: { + withClient: jest.fn( + async (fn: (client: FakeRedis) => Promise | T) => fn(fake), + ), + }, + }, + { provide: EventEmitter2, useValue: eventEmitter }, + ], + }).compile(); + return module.get(ZeroLossMessageService); + }; + + beforeEach(() => { + fake = new FakeRedis(); + eventEmitter = { emit: jest.fn() }; + // Pinned clock: service timestamps (createdAt/updatedAt/completedAt) and + // lease TTLs all derive from Date.now(), so tests advance time with the + // same mechanism that expires Redis keys. + jest.useFakeTimers(); + jest.setSystemTime(new Date('2026-08-20T00:00:00.000Z')); + }); + + afterEach(() => { + jest.useRealTimers(); + }); + + it('persists messages durably across a simulated restart', async () => { + const svcA = await buildService(); + await svcA.persistMessage('m1', 'emails', { to: 'a@example.com' }); + + // "Restart": a fresh instance over the same store. + const svcB = await buildService(); + const recovered = await svcB.getMessage('m1'); + + expect(recovered).toBeDefined(); + expect(recovered!.status).toBe('pending'); + expect(recovered!.data).toEqual({ to: 'a@example.com' }); + expect(recovered!.queueName).toBe('emails'); + + // The recovered message is claimable by the new instance. + expect(await svcB.markProcessing('m1')).toBe(true); + }); + + it('re-queues an orphaned processing message exactly once', async () => { + const svcA = await buildService(); + await svcA.persistMessage('m1', 'emails', {}); + expect(await svcA.markProcessing('m1')).toBe(true); + + // Simulate a crash: the lease TTL (5s default) expires. + jest.advanceTimersByTime(6000); + + const svcB = await buildService(); + await svcB.recoverOrphanedMessages(); + + const message = await svcB.getMessage('m1'); + expect(message!.status).toBe('pending'); + expect(requeueCount()).toBe(1); + + // A fresh worker can now claim it again. + expect(await svcB.markProcessing('m1')).toBe(true); + }); + + it('never double-processes: concurrent sweeps transition exactly once', async () => { + const svcA = await buildService(); + await svcA.persistMessage('m1', 'emails', {}); + await svcA.markProcessing('m1'); + + jest.advanceTimersByTime(6000); + const svcB = await buildService(); + const svcC = await buildService(); + + await Promise.all([ + svcB.recoverOrphanedMessages(), + svcC.recoverOrphanedMessages(), + ]); + + const message = await svcB.getMessage('m1'); + expect(message!.status).toBe('pending'); + // Exactly one sweep won the lease claim; only it emitted the requeue. + expect(requeueCount()).toBe(1); + }); + + it('fails a message instead of re-queuing once maxAttempts are exhausted', async () => { + const svcA = await buildService(); + await svcA.persistMessage('m1', 'emails', {}, 1); + await svcA.markProcessing('m1'); // attempts -> 1, maxAttempts 1 + + jest.advanceTimersByTime(6000); + await svcA.recoverOrphanedMessages(); + + const message = await svcA.getMessage('m1'); + expect(message!.status).toBe('failed'); + expect(message!.error).toBe('Acknowledgment timeout'); + expect(requeueCount()).toBe(0); + }); + + it('does not re-queue a message whose worker renews its lease', async () => { + const svcA = await buildService(); + await svcA.persistMessage('m1', 'emails', {}); + await svcA.markProcessing('m1'); + + // Worker still alive: renews near the lease boundary. + jest.advanceTimersByTime(4000); + expect(await svcA.renewProcessingLease('m1')).toBe(true); + + // Past the original TTL but within the renewed lease — sweep must skip. + jest.advanceTimersByTime(4000); + await svcA.recoverOrphanedMessages(); + expect((await svcA.getMessage('m1'))!.status).toBe('processing'); + expect(requeueCount()).toBe(0); + + // Worker gone now: lease expired, message is re-queued. + jest.advanceTimersByTime(3000); + await svcA.recoverOrphanedMessages(); + expect((await svcA.getMessage('m1'))!.status).toBe('pending'); + expect(requeueCount()).toBe(1); + }); + + it('never re-queues an acknowledged message', async () => { + const svcA = await buildService(); + await svcA.persistMessage('m1', 'emails', {}); + await svcA.markProcessing('m1'); + await svcA.acknowledgeMessage('m1'); + + jest.advanceTimersByTime(6000); + await svcA.recoverOrphanedMessages(); + + const message = await svcA.getMessage('m1'); + expect(message!.status).toBe('processing'); // not re-queued + expect(message!.acknowledgedAt).toBeDefined(); + expect(requeueCount()).toBe(0); + + await svcA.markCompleted('m1'); + expect((await svcA.getMessage('m1'))!.status).toBe('completed'); + }); + + it('shares replication nodes across instances and survives restart', async () => { + const svcA = await buildService(); + await svcA.registerReplicationNode('node-1'); + + const svcB = await buildService(); + await svcB.registerReplicationNode('node-2'); + + // A third instance sees both nodes and captures both as targets. + const svcC = await buildService(); + await svcC.persistMessage('m1', 'emails', {}); + const message = await svcC.getMessage('m1'); + expect([...message!.replicationNodes].sort()).toEqual(['node-1', 'node-2']); + + expect((await svcC.getStats()).replicationNodes).toBe(2); + + // Unregistering through one instance is visible to the others. + await svcB.unregisterReplicationNode('node-1'); + expect((await svcA.getStats()).replicationNodes).toBe(1); + }); + + it('tracks stats and cleans up old completed messages', async () => { + const svcA = await buildService(); + await svcA.persistMessage('m1', 'emails', {}); + await svcA.persistMessage('m2', 'reports', {}); + + const stats = await svcA.getStats(); + expect(stats.total).toBe(2); + expect(stats.pending).toBe(2); + + await svcA.markProcessing('m1'); + const processing = await svcA.getMessagesByStatus('processing'); + expect(processing.map((m) => m.messageId)).toEqual(['m1']); + + // Force-complete m1 with an old timestamp so cleanup removes it. + await svcA.acknowledgeMessage('m1'); + await svcA.markCompleted('m1'); + jest.advanceTimersByTime(24 * 60 * 60 * 1000 + 1000); + + const cleaned = await svcA.cleanupCompletedMessages(); + expect(cleaned).toBe(1); + expect(await svcA.getMessage('m1')).toBeUndefined(); + expect(await svcA.getMessage('m2')).toBeDefined(); + }); +}); diff --git a/src/queue/zero-loss-message.service.ts b/src/queue/zero-loss-message.service.ts index 2c9bafa..8333071 100644 --- a/src/queue/zero-loss-message.service.ts +++ b/src/queue/zero-loss-message.service.ts @@ -1,18 +1,36 @@ // src/queue/zero-loss-message.service.ts import { Injectable, Logger } from '@nestjs/common'; +import { Interval } from '@nestjs/schedule'; import { EventEmitter2 } from '@nestjs/event-emitter'; +import { randomUUID } from 'crypto'; +import type { Redis } from 'ioredis'; +import { RedisPoolService } from '../common/cache/redis-pool.service'; import { HorizontalScalingConfig, DEFAULT_HORIZONTAL_SCALING_CONFIG, } from './horizontal-scaling.config'; +const KEY_PREFIX = 'zls'; +const NODES_KEY = `${KEY_PREFIX}:nodes`; +const msgKey = (messageId: string) => `${KEY_PREFIX}:msg:${messageId}`; +const queueKey = (queueName: string) => `${KEY_PREFIX}:queue:${queueName}`; +const leaseKey = (messageId: string) => `${KEY_PREFIX}:lease:${messageId}`; + +/** + * How often the recovery sweep scans for orphaned `processing` messages. + * Kept below the default acknowledgment timeout (5s) so a dead worker's + * message is re-queued within a few seconds of its lease expiring. + */ +const RECOVERY_SWEEP_INTERVAL_MS = 1000; + /** * Message persistence entry */ export interface PersistedMessage { messageId: string; queueName: string; - data: any; + /** Payload, JSON-round-tripped through the durable store. */ + data: unknown; status: 'pending' | 'processing' | 'completed' | 'failed'; attempts: number; maxAttempts: number; @@ -27,46 +45,91 @@ export interface PersistedMessage { /** * Zero-Loss Message Service - * Ensures no messages are lost through persistence, replication, and acknowledgment + * + * Durable, horizontally-shared message state backed by Redis (the same + * infrastructure Bull already uses). All state lives in Redis, so a process + * restart or scale-out loses nothing: + * + * - `zls:msg:{id}` — hash with the full PersistedMessage (status, + * attempts, payload, timestamps, replication targets) + * - `zls:queue:{name}` — set of message ids per queue (for listing) + * - `zls:nodes` — set of registered replication nodes (shared) + * - `zls:lease:{id}` — processing lease, a `SET ... EX ... NX` key. A + * worker claims it atomically on markProcessing and + * renews it while alive; when it expires (crash), a + * recovery sweep re-claims it with NX and + * transitions `processing -> pending` exactly once. + * + * A worker that is still alive keeps renewing its lease, so the sweep never + * re-queues a message whose worker is actually progressing. Delivery remains + * at-least-once: if a worker exceeds the lease without renewing, its message + * is re-queued even if the worker later finishes. */ @Injectable() export class ZeroLossMessageService { private readonly logger = new Logger(ZeroLossMessageService.name); + private readonly instanceId = randomUUID(); private config: HorizontalScalingConfig; - private persistedMessages: Map = new Map(); - private pendingAcknowledgments: Map = new Map(); - private replicationNodes: Set = new Set(); - constructor(private eventEmitter: EventEmitter2) { + constructor( + private readonly redis: RedisPoolService, + private eventEmitter: EventEmitter2, + ) { this.config = DEFAULT_HORIZONTAL_SCALING_CONFIG; - this.logger.log('Zero-Loss Message Service initialized'); + this.logger.log('Zero-Loss Message Service initialized (durable store)'); + } + + private get leaseTtlSeconds(): number { + return Math.max( + 1, + Math.ceil(this.config.zeroLoss.acknowledgmentTimeoutMs / 1000), + ); } + // ==================== Replication nodes ==================== + /** - * Register a replication node + * Register a replication node. Nodes live in a shared Redis set, so every + * instance of the deployment sees the same registration set. */ - registerReplicationNode(nodeId: string): void { - this.replicationNodes.add(nodeId); + async registerReplicationNode(nodeId: string): Promise { + await this.redis.withClient((client) => client.sadd(NODES_KEY, nodeId)); this.logger.log(`Replication node registered: ${nodeId}`); } /** - * Unregister a replication node + * Unregister a replication node. */ - unregisterReplicationNode(nodeId: string): void { - this.replicationNodes.delete(nodeId); + async unregisterReplicationNode(nodeId: string): Promise { + await this.redis.withClient((client) => client.srem(NODES_KEY, nodeId)); this.logger.log(`Replication node unregistered: ${nodeId}`); } + private async getReplicationTargets(): Promise { + const nodes = await this.redis.withClient((client) => + client.smembers(NODES_KEY), + ); + nodes.sort(); + const count = Math.min( + this.config.zeroLoss.replicationFactor, + nodes.length, + ); + return nodes.slice(0, count); + } + + // ==================== Message lifecycle ==================== + /** - * Persist a message + * Persist a message durably. The message is written to Redis (shared across + * instances) before this method resolves; replication targets are captured + * from the shared node set at persist time. */ - persistMessage( + async persistMessage( messageId: string, queueName: string, - data: any, + data: unknown, maxAttempts: number = this.config.zeroLoss.maxRetryAttempts, - ): PersistedMessage { + ): Promise { const now = new Date(); const message: PersistedMessage = { @@ -78,15 +141,13 @@ export class ZeroLossMessageService { maxAttempts, createdAt: now, updatedAt: now, - replicationNodes: this.getReplicationTargets(), + replicationNodes: await this.getReplicationTargets(), }; - this.persistedMessages.set(messageId, message); - - // Set acknowledgment timeout - if (this.config.zeroLoss.enabled) { - this.setAcknowledgmentTimeout(messageId); - } + await this.redis.withClient(async (client) => { + await client.hset(msgKey(messageId), this.toHashFields(message)); + await client.sadd(queueKey(queueName), messageId); + }); this.logger.debug(`Message persisted: ${messageId} (queue: ${queueName})`); this.eventEmitter.emit('message.persisted', { messageId, queueName }); @@ -95,273 +156,405 @@ export class ZeroLossMessageService { } /** - * Get replication targets + * Atomically claim a pending message for processing. The claim is a + * `SET zls:lease:{id} EX NX`, so at most one worker + * across all instances can hold the lease; every other concurrent caller + * gets `false`. The lease expires automatically after the acknowledgment + * timeout unless renewed with {@link renewProcessingLease}. */ - private getReplicationTargets(): string[] { - const targets: string[] = []; - const nodes = Array.from(this.replicationNodes); + async markProcessing(messageId: string): Promise { + return this.redis.withClient(async (client) => { + const claimed = await client.set( + leaseKey(messageId), + this.instanceId, + 'EX', + this.leaseTtlSeconds, + 'NX', + ); + if (!claimed) { + return false; // lease already held — another worker is processing it + } - // Select nodes based on replication factor - const count = Math.min( - this.config.zeroLoss.replicationFactor, - nodes.length, - ); + const status = await client.hget(msgKey(messageId), 'status'); + if (status !== 'pending') { + await client.del(leaseKey(messageId)); + return false; + } - for (let i = 0; i < count; i++) { - targets.push(nodes[i % nodes.length]); - } + const attempts = await client.hincrby(msgKey(messageId), 'attempts', 1); + const maxAttempts = Number( + await client.hget(msgKey(messageId), 'maxAttempts'), + ); + await client.hset( + msgKey(messageId), + 'status', + 'processing', + 'updatedAt', + new Date().toISOString(), + 'leaseUntil', + String(Date.now() + this.config.zeroLoss.acknowledgmentTimeoutMs), + ); - return targets; + this.logger.debug( + `Message marked as processing: ${messageId} (attempt ${attempts}/${maxAttempts})`, + ); + return true; + }); } /** - * Set acknowledgment timeout for a message + * Renew the processing lease of a message this instance is currently + * working on. A live worker calls this periodically so the recovery sweep + * never re-queues a message that is actually progressing. */ - private setAcknowledgmentTimeout(messageId: string): void { - const timeout = setTimeout(() => { - this.handleAcknowledgmentTimeout(messageId); - }, this.config.zeroLoss.acknowledgmentTimeoutMs); - - this.pendingAcknowledgments.set(messageId, timeout); + async renewProcessingLease(messageId: string): Promise { + return this.redis.withClient(async (client) => { + const owner = await client.get(leaseKey(messageId)); + if (owner !== this.instanceId) { + return false; // lease lost (expired/recovered) or never held + } + const renewed = await client.expire( + leaseKey(messageId), + this.leaseTtlSeconds, + ); + if (renewed === 1) { + await client.hset( + msgKey(messageId), + 'updatedAt', + new Date().toISOString(), + 'leaseUntil', + String(Date.now() + this.config.zeroLoss.acknowledgmentTimeoutMs), + ); + return true; + } + return false; + }); } /** - * Handle acknowledgment timeout + * Acknowledge message processing. Marks the message acknowledged and + * releases the lease, so the recovery sweep will not re-queue it. */ - private handleAcknowledgmentTimeout(messageId: string): void { - const message = this.persistedMessages.get(messageId); - - if (message && message.status === 'processing' && !message.acknowledgedAt) { - this.logger.warn(`Acknowledgment timeout for message: ${messageId}`); - - // Retry if attempts remaining - if (message.attempts < message.maxAttempts) { - message.status = 'pending'; - message.updatedAt = new Date(); - this.logger.log(`Message re-queued after timeout: ${messageId}`); - this.eventEmitter.emit('message.requeued', { - messageId, - reason: 'acknowledgment-timeout', - }); - } else { - message.status = 'failed'; - message.error = 'Acknowledgment timeout'; - message.failedAt = new Date(); - message.updatedAt = new Date(); - this.logger.error(`Message failed after max attempts: ${messageId}`); - this.eventEmitter.emit('message.failed', { - messageId, - reason: 'max-attempts-exceeded', - }); + async acknowledgeMessage(messageId: string): Promise { + return this.redis.withClient(async (client) => { + const status = await client.hget(msgKey(messageId), 'status'); + if (!status) { + this.logger.warn(`Message not found for acknowledgment: ${messageId}`); + return false; } - } - - this.pendingAcknowledgments.delete(messageId); + await client.hset( + msgKey(messageId), + 'acknowledgedAt', + new Date().toISOString(), + 'updatedAt', + new Date().toISOString(), + 'leaseUntil', + '', + ); + await client.del(leaseKey(messageId)); + this.logger.debug(`Message acknowledged: ${messageId}`); + this.eventEmitter.emit('message.acknowledged', { messageId }); + return true; + }); } /** - * Mark message as processing + * Mark message as completed. */ - markProcessing(messageId: string): boolean { - const message = this.persistedMessages.get(messageId); - - if (!message) { - this.logger.warn(`Message not found: ${messageId}`); - return false; - } - - message.status = 'processing'; - message.attempts++; - message.updatedAt = new Date(); - - this.logger.debug( - `Message marked as processing: ${messageId} (attempt ${message.attempts}/${message.maxAttempts})`, - ); - return true; + async markCompleted(messageId: string): Promise { + return this.redis.withClient(async (client) => { + const status = await client.hget(msgKey(messageId), 'status'); + if (!status) return false; + await client.hset( + msgKey(messageId), + 'status', + 'completed', + 'completedAt', + new Date().toISOString(), + 'updatedAt', + new Date().toISOString(), + 'leaseUntil', + '', + ); + await client.del(leaseKey(messageId)); + this.logger.debug(`Message completed: ${messageId}`); + this.eventEmitter.emit('message.completed', { messageId }); + return true; + }); } /** - * Acknowledge message processing + * Mark message as failed. */ - acknowledgeMessage(messageId: string): boolean { - const message = this.persistedMessages.get(messageId); - - if (!message) { - this.logger.warn(`Message not found for acknowledgment: ${messageId}`); - return false; - } - - // Clear timeout - const timeout = this.pendingAcknowledgments.get(messageId); - if (timeout) { - clearTimeout(timeout); - this.pendingAcknowledgments.delete(messageId); - } - - message.acknowledgedAt = new Date(); - message.updatedAt = new Date(); - - this.logger.debug(`Message acknowledged: ${messageId}`); - this.eventEmitter.emit('message.acknowledged', { messageId }); - - return true; + async markFailed(messageId: string, error: string): Promise { + return this.redis.withClient(async (client) => { + const status = await client.hget(msgKey(messageId), 'status'); + if (!status) return false; + await client.hset( + msgKey(messageId), + 'status', + 'failed', + 'error', + error, + 'failedAt', + new Date().toISOString(), + 'updatedAt', + new Date().toISOString(), + 'leaseUntil', + '', + ); + await client.del(leaseKey(messageId)); + this.logger.warn(`Message failed: ${messageId} - ${error}`); + this.eventEmitter.emit('message.failed', { messageId, error }); + return true; + }); } /** - * Mark message as completed + * Retry a failed message (pending again, lease released). No-op once the + * message has exhausted maxAttempts. */ - markCompleted(messageId: string): boolean { - const message = this.persistedMessages.get(messageId); - - if (!message) { - return false; - } - - message.status = 'completed'; - message.completedAt = new Date(); - message.updatedAt = new Date(); - - this.logger.debug(`Message completed: ${messageId}`); - this.eventEmitter.emit('message.completed', { messageId }); - - return true; + async retryMessage(messageId: string): Promise { + return this.redis.withClient(async (client) => { + const attempts = Number(await client.hget(msgKey(messageId), 'attempts')); + const maxAttempts = Number( + await client.hget(msgKey(messageId), 'maxAttempts'), + ); + if (!attempts && !maxAttempts) { + this.logger.warn(`Message not found for retry: ${messageId}`); + return false; + } + if (attempts >= maxAttempts) { + this.logger.warn(`Message ${messageId} has exceeded max attempts`); + return false; + } + await client.hset( + msgKey(messageId), + 'status', + 'pending', + 'error', + '', + 'updatedAt', + new Date().toISOString(), + 'leaseUntil', + '', + ); + await client.del(leaseKey(messageId)); + this.logger.log(`Message retry initiated: ${messageId}`); + this.eventEmitter.emit('message.retried', { messageId }); + return true; + }); } - /** - * Mark message as failed - */ - markFailed(messageId: string, error: string): boolean { - const message = this.persistedMessages.get(messageId); - - if (!message) { - return false; - } - - message.status = 'failed'; - message.error = error; - message.failedAt = new Date(); - message.updatedAt = new Date(); - - this.logger.warn(`Message failed: ${messageId} - ${error}`); - this.eventEmitter.emit('message.failed', { messageId, error }); - - return true; - } + // ==================== Recovery ==================== /** - * Retry a failed message + * Recovery sweep. Scans the durable message hashes (which never expire) + * for messages stuck in `processing` whose lease has expired — the lease is + * a TTL key, so its disappearance *is* the crash signal. Each orphan is + * atomically transitioned back to `pending` — or to `failed` once + * maxAttempts are exhausted. The `SET ... NX` re-claim guarantees the + * transition runs exactly once even when several instances sweep the same + * message concurrently, and a worker that is still alive holds the lease, + * so its message is never touched. */ - retryMessage(messageId: string): boolean { - const message = this.persistedMessages.get(messageId); - - if (!message) { - return false; + @Interval(RECOVERY_SWEEP_INTERVAL_MS) + async recoverOrphanedMessages(): Promise { + const recovered: string[] = []; + await this.redis.withClient(async (client) => { + const messageKeys = await this.scanKeys(client, `${KEY_PREFIX}:msg:*`); + for (const key of messageKeys) { + const messageId = key.slice(msgKey('').length); + if (await this.recoverOne(client, messageId)) { + recovered.push(messageId); + } + } + }); + if (recovered.length > 0) { + this.logger.log( + `Recovery sweep re-queued ${recovered.length} orphaned message(s)`, + ); } + return recovered.length; + } - if (message.attempts >= message.maxAttempts) { - this.logger.warn(`Message ${messageId} has exceeded max attempts`); - return false; + private async recoverOne(client: Redis, messageId: string): Promise { + const message = await this.getHashMessage(client, messageId); + if (!message || message.status !== 'processing' || message.acknowledgedAt) { + return false; // not orphaned (or already acknowledged) } - message.status = 'pending'; - message.error = undefined; - message.updatedAt = new Date(); + // A live lease means a worker is still processing the message — skip. + const leaseOwner = await client.get(leaseKey(messageId)); + if (leaseOwner) return false; + + // Lease expired (worker crashed). Re-claim it: only one sweep across all + // instances can win the NX, so the transition happens exactly once. + const claimed = await client.set( + leaseKey(messageId), + `${this.instanceId}:recovery`, + 'EX', + this.leaseTtlSeconds, + 'NX', + ); + if (!claimed) return false; - this.logger.log(`Message retry initiated: ${messageId}`); - this.eventEmitter.emit('message.retried', { messageId }); + try { + // Re-check under the claim: a concurrent sweep may have already + // transitioned or acknowledged the message. + const current = await this.getHashMessage(client, messageId); + if ( + !current || + current.status !== 'processing' || + current.acknowledgedAt + ) { + return false; + } - return true; + if (current.attempts < current.maxAttempts) { + await client.hset( + msgKey(messageId), + 'status', + 'pending', + 'updatedAt', + new Date().toISOString(), + 'leaseUntil', + '', + ); + this.logger.warn( + `Acknowledgment timeout for message: ${messageId} — re-queued`, + ); + this.eventEmitter.emit('message.requeued', { + messageId, + reason: 'acknowledgment-timeout', + }); + } else { + await client.hset( + msgKey(messageId), + 'status', + 'failed', + 'error', + 'Acknowledgment timeout', + 'failedAt', + new Date().toISOString(), + 'updatedAt', + new Date().toISOString(), + 'leaseUntil', + '', + ); + this.logger.error(`Message failed after max attempts: ${messageId}`); + this.eventEmitter.emit('message.failed', { + messageId, + reason: 'max-attempts-exceeded', + }); + } + return true; + } finally { + await client.del(leaseKey(messageId)); + } } + // ==================== Reads ==================== + /** - * Get message by ID + * Get message by ID. */ - getMessage(messageId: string): PersistedMessage | undefined { - return this.persistedMessages.get(messageId); + async getMessage(messageId: string): Promise { + return this.redis.withClient((client) => + this.getHashMessage(client, messageId), + ); } /** - * Get all messages for a queue + * Get all messages for a queue. */ - getQueueMessages(queueName: string): PersistedMessage[] { - const messages: PersistedMessage[] = []; - - for (const message of this.persistedMessages.values()) { - if (message.queueName === queueName) { - messages.push(message); + async getQueueMessages(queueName: string): Promise { + return this.redis.withClient(async (client) => { + const ids = await client.smembers(queueKey(queueName)); + const messages: PersistedMessage[] = []; + for (const id of ids) { + const message = await this.getHashMessage(client, id); + if (message) messages.push(message); } - } - - return messages; + return messages; + }); } /** - * Get messages by status + * Get messages by status. */ - getMessagesByStatus(status: PersistedMessage['status']): PersistedMessage[] { - const messages: PersistedMessage[] = []; - - for (const message of this.persistedMessages.values()) { - if (message.status === status) { - messages.push(message); - } - } - - return messages; + async getMessagesByStatus( + status: PersistedMessage['status'], + ): Promise { + const messages = await this.getAllMessages(); + return messages.filter((m) => m.status === status); } /** - * Get pending messages (for recovery) + * Get pending messages (for recovery). */ - getPendingMessages(): PersistedMessage[] { + async getPendingMessages(): Promise { return this.getMessagesByStatus('pending'); } /** - * Get failed messages (for manual intervention) + * Get failed messages (for manual intervention). */ - getFailedMessages(): PersistedMessage[] { + async getFailedMessages(): Promise { return this.getMessagesByStatus('failed'); } /** - * Clean up completed messages + * Clean up completed messages older than maxAgeMs. */ - cleanupCompletedMessages(maxAgeMs: number = 24 * 60 * 60 * 1000): number { + async cleanupCompletedMessages( + maxAgeMs: number = 24 * 60 * 60 * 1000, + ): Promise { const cutoffTime = Date.now() - maxAgeMs; let cleanedCount = 0; - - for (const [messageId, message] of this.persistedMessages.entries()) { - if ( - message.status === 'completed' && - message.completedAt && - message.completedAt.getTime() < cutoffTime - ) { - this.persistedMessages.delete(messageId); - cleanedCount++; + await this.redis.withClient(async (client) => { + const keys = await this.scanKeys(client, `${KEY_PREFIX}:msg:*`); + for (const key of keys) { + const messageId = key.slice(msgKey('').length); + const message = await this.getHashMessage(client, messageId); + if ( + message && + message.status === 'completed' && + message.completedAt && + message.completedAt.getTime() < cutoffTime + ) { + await client.del(key); + await client.srem(queueKey(message.queueName), messageId); + cleanedCount++; + } } - } - + }); if (cleanedCount > 0) { this.logger.debug(`Cleaned up ${cleanedCount} completed messages`); } - return cleanedCount; } /** - * Get message statistics + * Get message statistics. */ - getStats(): { + async getStats(): Promise<{ total: number; pending: number; processing: number; completed: number; failed: number; replicationNodes: number; - pendingAcknowledgments: number; - } { - const messages = Array.from(this.persistedMessages.values()); + activeLeases: number; + }> { + const messages = await this.getAllMessages(); + const replicationNodes = await this.redis.withClient((client) => + client.scard(NODES_KEY), + ); + const activeLeases = await this.redis.withClient( + async (client) => + (await this.scanKeys(client, `${KEY_PREFIX}:lease:*`)).length, + ); return { total: messages.length, @@ -369,30 +562,32 @@ export class ZeroLossMessageService { processing: messages.filter((m) => m.status === 'processing').length, completed: messages.filter((m) => m.status === 'completed').length, failed: messages.filter((m) => m.status === 'failed').length, - replicationNodes: this.replicationNodes.size, - pendingAcknowledgments: this.pendingAcknowledgments.size, + replicationNodes, + activeLeases, }; } /** - * Verify message integrity + * Verify message integrity. */ - verifyMessageIntegrity(messageId: string): { + async verifyMessageIntegrity(messageId: string): Promise<{ valid: boolean; issues: string[]; - } { - const message = this.persistedMessages.get(messageId); + }> { + const message = await this.getMessage(messageId); const issues: string[] = []; if (!message) { return { valid: false, issues: ['Message not found'] }; } - // Check for orphaned processing state + // Check for orphaned processing state (lease expired) if (message.status === 'processing' && !message.acknowledgedAt) { - const timeSinceUpdate = Date.now() - message.updatedAt.getTime(); - if (timeSinceUpdate > this.config.zeroLoss.acknowledgmentTimeoutMs * 2) { - issues.push('Message stuck in processing state'); + const leaseAlive = await this.redis.withClient((client) => + client.get(leaseKey(messageId)), + ); + if (!leaseAlive) { + issues.push('Message stuck in processing state (lease expired)'); } } @@ -416,8 +611,10 @@ export class ZeroLossMessageService { }; } + // ==================== Configuration ==================== + /** - * Update configuration + * Update configuration. */ updateConfig(newConfig: Partial): void { this.config = { @@ -428,9 +625,89 @@ export class ZeroLossMessageService { } /** - * Get current configuration + * Get current configuration. */ getConfig(): HorizontalScalingConfig { return { ...this.config }; } + + // ==================== Internals ==================== + + private async getAllMessages(): Promise { + return this.redis.withClient(async (client) => { + const keys = await this.scanKeys(client, `${KEY_PREFIX}:msg:*`); + const messages: PersistedMessage[] = []; + for (const key of keys) { + const message = await this.getHashMessage( + client, + key.slice(msgKey('').length), + ); + if (message) messages.push(message); + } + return messages; + }); + } + + private async getHashMessage( + client: Redis, + messageId: string, + ): Promise { + const fields = await client.hgetall(msgKey(messageId)); + if (!fields.messageId) return undefined; + return { + messageId: fields.messageId, + queueName: fields.queueName, + data: JSON.parse(fields.data ?? 'null') as unknown, + status: fields.status as PersistedMessage['status'], + attempts: Number(fields.attempts), + maxAttempts: Number(fields.maxAttempts), + createdAt: new Date(fields.createdAt), + updatedAt: new Date(fields.updatedAt), + acknowledgedAt: fields.acknowledgedAt + ? new Date(fields.acknowledgedAt) + : undefined, + completedAt: fields.completedAt + ? new Date(fields.completedAt) + : undefined, + failedAt: fields.failedAt ? new Date(fields.failedAt) : undefined, + error: fields.error || undefined, + replicationNodes: JSON.parse(fields.replicationNodes ?? '[]') as string[], + }; + } + + private toHashFields(message: PersistedMessage): Record { + return { + messageId: message.messageId, + queueName: message.queueName, + data: JSON.stringify(message.data ?? null), + status: message.status, + attempts: String(message.attempts), + maxAttempts: String(message.maxAttempts), + createdAt: message.createdAt.toISOString(), + updatedAt: message.updatedAt.toISOString(), + acknowledgedAt: message.acknowledgedAt?.toISOString() ?? '', + completedAt: message.completedAt?.toISOString() ?? '', + failedAt: message.failedAt?.toISOString() ?? '', + error: message.error ?? '', + replicationNodes: JSON.stringify(message.replicationNodes), + leaseUntil: '', + }; + } + + private async scanKeys(client: Redis, pattern: string): Promise { + const keys: string[] = []; + let cursor = '0'; + do { + const [nextCursor, batch] = await client.scan( + cursor, + 'MATCH', + pattern, + 'COUNT', + 100, + ); + keys.push(...batch); + cursor = nextCursor; + } while (cursor !== '0'); + return keys; + } }