perf(pipeline): batch stream ACKs, consolidate outbox writes, and remove blocking Redis round-trips

- Implement stream-keyed ACK buffering in StreamConsumer with periodic and capacity-triggered flushes
- Consolidate delivery outbox persistence to a single post-dispatch write, cutting PostgreSQL outbox writes by 50%
- Make retry key deletion and idempotency TTL extension non-blocking on message success
- Add host-unique consumer identifiers across distributed delivery, engine, and enricher workers
- Add unit tests verifying buffered stream ACK flushing behavior

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<string, string[]>();
+  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<void> {
@@ -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<void> {
+    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<void> {
+    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<void> {
@@ -375,6 +426,7 @@ export class StreamConsumer {
 
   async stop(): Promise<void> {
     this.running = false;
+    await this.flushAcks().catch(() => {});
     try {
       await this.blockingRedis.quit();
     } catch (err: any) {

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,
   });
 

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,
   });
 

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,
   });
 

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" });