diff --git a/src/queue/index.ts b/src/queue/index.ts index 52f56f2..111a080 100644 --- a/src/queue/index.ts +++ b/src/queue/index.ts @@ -188,6 +188,8 @@ export interface StreamConsumerOptions { logger?: Logger; batchSize?: number; blockMs?: number; + bufferAcks?: boolean; + ackFlushMs?: number; } type XReadGroupResult = Array<[string, Array<[string, string[] | null]>]> | null; @@ -202,6 +204,10 @@ export class StreamConsumer { private readonly logger?: Logger; private readonly batchSize: number; private readonly blockMs: number; + private readonly bufferAcks: boolean; + private readonly ackFlushMs: number; + private ackBuffers = new Map(); + private ackTimer: NodeJS.Timeout | null = null; private running = false; constructor({ @@ -213,6 +219,8 @@ export class StreamConsumer { logger, batchSize = 10, blockMs = 5_000, + bufferAcks = false, + ackFlushMs = 5, }: StreamConsumerOptions) { this.redis = redis; this.blockingRedis = redis.duplicate(); @@ -223,6 +231,8 @@ export class StreamConsumer { this.logger = logger; this.batchSize = batchSize; this.blockMs = blockMs; + this.bufferAcks = bufferAcks; + this.ackFlushMs = ackFlushMs; } async ensureGroup(): Promise { @@ -336,8 +346,49 @@ export class StreamConsumer { const s = stream ?? this.streams[0]!; const ids = Array.isArray(messageId) ? messageId : [messageId]; if (ids.length === 0) return; - await this.redis.xack(s, this.group, ...ids); - this.logger?.debug({ stream: s, count: ids.length }, "messages acknowledged"); + + if (!this.bufferAcks) { + await this.redis.xack(s, this.group, ...ids); + this.logger?.debug({ stream: s, count: ids.length }, "messages acknowledged"); + return; + } + + let buf = this.ackBuffers.get(s); + if (!buf) { + buf = []; + this.ackBuffers.set(s, buf); + } + buf.push(...ids); + + if (buf.length >= this.batchSize) { + await this.flushAcksForStream(s); + } else if (!this.ackTimer) { + this.ackTimer = setTimeout(() => { + this.ackTimer = null; + void this.flushAcks(); + }, this.ackFlushMs); + } + } + + async flushAcksForStream(stream: string): Promise { + const ids = this.ackBuffers.get(stream); + if (!ids || ids.length === 0) return; + this.ackBuffers.set(stream, []); + try { + await this.redis.xack(stream, this.group, ...ids); + this.logger?.debug({ stream, count: ids.length }, "buffered messages acknowledged"); + } catch (err) { + this.logger?.error({ err, stream, count: ids.length }, "failed to flush buffered acks"); + } + } + + async flushAcks(): Promise { + if (this.ackTimer) { + clearTimeout(this.ackTimer); + this.ackTimer = null; + } + const streams = Array.from(this.ackBuffers.keys()); + await Promise.all(streams.map((s) => this.flushAcksForStream(s))); } async nack(messageId: string, event: StreamEvent, stream?: string): Promise { @@ -375,6 +426,7 @@ export class StreamConsumer { async stop(): Promise { this.running = false; + await this.flushAcks().catch(() => {}); try { await this.blockingRedis.quit(); } catch (err: any) { diff --git a/src/services/delivery/main.ts b/src/services/delivery/main.ts index 2cd96db..4536691 100644 --- a/src/services/delivery/main.ts +++ b/src/services/delivery/main.ts @@ -350,8 +350,6 @@ export class DeliveryWorker extends BaseWorker { } try { - publishPromises.push(this.outboxInsertProcessor.add(task).then(() => {})); - publishPromises.push( this.eventProcessor.add( buildStreamEvent( @@ -419,6 +417,14 @@ export class DeliveryWorker extends BaseWorker { } if (lastResult.invalidToken) { + publishPromises.push( + this.outboxUpdateProcessor.add({ + taskId: task.taskId, + channel: task.channel, + destination: task.destination, + providerMessageId: null, + }), + ); await this.contactRepo.deactivate( task.projectId, task.recipientId, @@ -528,6 +534,15 @@ export class DeliveryWorker extends BaseWorker { ); metrics.deliveryFailed.inc({ channel: task.channel, reason: "push_error" }); + publishPromises.push( + this.outboxUpdateProcessor.add({ + taskId: task.taskId, + channel: task.channel, + destination: task.destination, + providerMessageId: null, + }), + ); + publishPromises.push( this.eventProcessor.add( buildStreamEvent( @@ -644,6 +659,15 @@ export class DeliveryWorker extends BaseWorker { ), ); } else { + publishPromises.push( + this.outboxUpdateProcessor.add({ + taskId: task.taskId, + channel: task.channel, + destination: task.destination, + providerMessageId: null, + }), + ); + publishPromises.push( this.eventProcessor.add( buildStreamEvent( @@ -704,7 +728,9 @@ export class DeliveryWorker extends BaseWorker { }); const flushTime = Date.now() - tFlushStart; - await this.idempotency.markProcessed(idempotencyKey); + void this.idempotency.markProcessed(idempotencyKey).catch((err: any) => { + this.logger.warn({ err, taskId: task.taskId }, "failed to upgrade idempotency TTL"); + }); const t = ((global as any)._telemetry = (global as any)._telemetry || { count: 0, @@ -744,13 +770,15 @@ export async function startDeliveryWorker() { const dbData = createDatabase({ url: config.DATABASE_URL, applicationName: "delivery", logger }); sql = dbData.sql; db = dbData.db; + const consumerId = `delivery-${process.env.HOSTNAME || process.pid}-${Math.random().toString(36).slice(2, 8)}`; consumer = new StreamConsumer({ redis: redis.native, stream: OUTBOUND_STREAMS as unknown as StreamName[], group: CONSUMER_GROUPS.DELIVERY, - consumer: `delivery-${process.pid}`, + consumer: consumerId, dlqStream: STREAMS.DEAD_LETTER, batchSize: config.WORKER_CONCURRENCY, + bufferAcks: true, logger, }); @@ -764,7 +792,7 @@ export async function startDeliveryWorker() { redis: redis.native, stream: OUTBOUND_STREAMS as unknown as StreamName[], group: CONSUMER_GROUPS.DELIVERY, - consumer: `delivery-${process.pid}`, + consumer: consumerId, logger, }); diff --git a/src/services/engine/main.ts b/src/services/engine/main.ts index f06fc60..e6a8fb1 100644 --- a/src/services/engine/main.ts +++ b/src/services/engine/main.ts @@ -706,13 +706,15 @@ export async function startEngineWorker() { templateRepo = new TemplateRepository(db); templateCache = new TemplateCache(templateRepo); const contactRepo = new ContactRepository(db); + const consumerId = `engine-${process.env.HOSTNAME || process.pid}-${Math.random().toString(36).slice(2, 8)}`; consumer = new StreamConsumer({ redis: redis.native, stream: ENRICHED_STREAMS as unknown as StreamName[], group: CONSUMER_GROUPS.ENGINE, - consumer: `engine-${process.pid}`, + consumer: consumerId, dlqStream: STREAMS.DEAD_LETTER, batchSize: config.WORKER_CONCURRENCY, + bufferAcks: true, logger, }); @@ -720,7 +722,7 @@ export async function startEngineWorker() { redis: redis.native, stream: ENRICHED_STREAMS as unknown as StreamName[], group: CONSUMER_GROUPS.ENGINE, - consumer: `engine-${process.pid}`, + consumer: consumerId, logger, }); diff --git a/src/services/enricher/main.ts b/src/services/enricher/main.ts index 78b801c..d8b4f9d 100644 --- a/src/services/enricher/main.ts +++ b/src/services/enricher/main.ts @@ -510,13 +510,15 @@ export async function startEnricherWorker() { const dbData = createDatabase({ url: config.DATABASE_URL, applicationName: "enricher", logger }); sql = dbData.sql; db = dbData.db; + const consumerId = `enricher-${process.env.HOSTNAME || process.pid}-${Math.random().toString(36).slice(2, 8)}`; consumer = new StreamConsumer({ redis: redis.native, stream: INBOUND_STREAMS as unknown as StreamName[], group: CONSUMER_GROUPS.ENRICHER, - consumer: `enricher-${process.pid}`, + consumer: consumerId, dlqStream: STREAMS.DEAD_LETTER, batchSize: config.WORKER_CONCURRENCY, + bufferAcks: true, logger, }); @@ -524,7 +526,7 @@ export async function startEnricherWorker() { redis: redis.native, stream: INBOUND_STREAMS as unknown as StreamName[], group: CONSUMER_GROUPS.ENRICHER, - consumer: `enricher-${process.pid}`, + consumer: consumerId, logger, }); diff --git a/src/workers/index.ts b/src/workers/index.ts index 2e9ed80..0197136 100644 --- a/src/workers/index.ts +++ b/src/workers/index.ts @@ -210,7 +210,9 @@ export abstract class BaseWorker { await this.process(message, retryCount); await this.consumer.ack(message.id, stream); - await this.consumer.redis.del(retryKey); + void this.consumer.redis.del(retryKey).catch((err: any) => { + this.logger.debug({ err, retryKey }, "failed to clean up worker retry key"); + }); this.processedCount += 1; this.lastProcessedAt = new Date().toISOString(); metrics.messagesProcessed.inc({ worker: this.constructor.name, status: "success" }); diff --git a/tests/queue.test.ts b/tests/queue.test.ts index 0748a8f..c7a3324 100644 --- a/tests/queue.test.ts +++ b/tests/queue.test.ts @@ -154,6 +154,42 @@ describe("StreamConsumer.ack", () => { expect(mockRedis.xack).not.toHaveBeenCalled(); }); + + it("buffers acks and flushes when batchSize is reached", async () => { + const mockRedis = { xack: vi.fn().mockResolvedValue(2), duplicate: vi.fn().mockReturnThis() }; + const consumer = new StreamConsumer({ + redis: mockRedis as any, + stream: "stream-1" as any, + group: "g" as any, + consumer: "c", + bufferAcks: true, + batchSize: 2, + }); + + await consumer.ack("1-0"); + expect(mockRedis.xack).not.toHaveBeenCalled(); + + await consumer.ack("2-0"); + expect(mockRedis.xack).toHaveBeenCalledWith("stream-1", "g", "1-0", "2-0"); + }); + + it("flushes remaining buffered acks on flushAcks", async () => { + const mockRedis = { xack: vi.fn().mockResolvedValue(1), duplicate: vi.fn().mockReturnThis() }; + const consumer = new StreamConsumer({ + redis: mockRedis as any, + stream: "stream-1" as any, + group: "g" as any, + consumer: "c", + bufferAcks: true, + batchSize: 10, + }); + + await consumer.ack("1-0"); + expect(mockRedis.xack).not.toHaveBeenCalled(); + + await consumer.flushAcks(); + expect(mockRedis.xack).toHaveBeenCalledWith("stream-1", "g", "1-0"); + }); }); describe("StreamConsumer.nack", () => {