feat(p9): IiosDlqItem + DlqService (dead-letter + deterministic replay)

Task D.1: mandated dlq_item table (sourceType/sourceId/failureStage/errorCode/ retryCount/payloadRef/consumerName/scopeId/status). DlqService.record (idempotent per consumer+sourceId, bumps retryCount), onConsumerFailure (quarantine + clear the IiosProcessedEvent claim so it's replayable — fixes claim-then-fail), registerHandler + replay (clears claim, awaits the registered consumer handler → RESOLVED on success, QUARANTINED+retry on repeat throw; outbox-sourced items reset to PENDING). Wired into OutboxModule; resetDb truncates it. 3 tests. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-07-03 01:28:48 +05:30
parent e5d3ce5ecd
commit 208fb0b4f0
6 changed files with 219 additions and 3 deletions
@@ -0,0 +1,62 @@
+import { describe, it, expect, beforeAll, afterAll, beforeEach } from 'vitest';
+import { PrismaClient } from '@prisma/client';
+import { resetDb } from '../test-utils/reset-db';
+import type { CloudEvent } from '@insignia/iios-contracts';
+import { DlqService } from './dlq.service';
+import type { PrismaService } from '../prisma/prisma.service';
+
+const url = process.env.DATABASE_URL ?? 'postgresql://iios:iios@localhost:5434/iios?schema=public';
+const prisma = new PrismaClient({ datasources: { db: { url } } });
+const asService = prisma as unknown as PrismaService;
+
+const evt = (id: string): CloudEvent => ({
+  specversion: '1.0', id, type: 'x', source: 's', time: new Date().toISOString(),
+  insignia: { scopeSnapshotId: 'scope-A', idempotencyKey: `k:${id}` }, data: {},
+});
+
+beforeAll(async () => { await prisma.$connect(); });
+afterAll(async () => { await prisma.$disconnect(); });
+beforeEach(async () => { await resetDb(prisma); });
+
+// A DLQ item's replay needs the underlying outbox event to exist.
+async function seedOutboxEvent(eventId: string): Promise<void> {
+  await prisma.iiosOutboxEvent.create({
+    data: { eventId, aggregateType: 'x', aggregateId: eventId, eventType: 'x', cloudEvent: evt(eventId) as unknown as object, partitionKey: 'p' },
+  });
+}
+
+describe('DlqService (P9 — dead-letter + replay)', () => {
+  it('record is idempotent per (consumer, sourceId) and bumps retryCount', async () => {
+    const dlq = new DlqService(asService);
+    await dlq.record({ sourceType: 'projection', sourceId: 'e1', failureStage: 'c', consumerName: 'c', scopeId: 'scope-A' });
+    await dlq.record({ sourceType: 'projection', sourceId: 'e1', failureStage: 'c', consumerName: 'c', scopeId: 'scope-A' });
+    const items = await dlq.listDlq('scope-A');
+    expect(items).toHaveLength(1);
+    expect(items[0].retryCount).toBe(1); // 0 on create, +1 on repeat
+  });
+
+  it('onConsumerFailure quarantines the event and clears its claim (replayable)', async () => {
+    const dlq = new DlqService(asService);
+    await prisma.iiosProcessedEvent.create({ data: { consumerName: 'c', eventId: 'e2' } }); // claimed then failed
+    await dlq.onConsumerFailure('c', evt('e2'), new Error('boom'));
+    expect(await prisma.iiosDlqItem.count({ where: { sourceId: 'e2', consumerName: 'c' } })).toBe(1);
+    expect(await prisma.iiosProcessedEvent.count({ where: { consumerName: 'c', eventId: 'e2' } })).toBe(0); // claim cleared
+  });
+
+  it('replay re-runs the registered handler → RESOLVED (only when it succeeds)', async () => {
+    const dlq = new DlqService(asService);
+    await seedOutboxEvent('e3');
+    let calls = 0;
+    dlq.registerHandler('c', async () => { calls++; if (calls === 1) throw new Error('still broken'); });
+    await dlq.onConsumerFailure('c', evt('e3'), new Error('boom'));
+    const item = await prisma.iiosDlqItem.findFirstOrThrow({ where: { sourceId: 'e3' } });
+
+    const r1 = await dlq.replay(item.id); // handler throws again
+    expect(r1.status).toBe('QUARANTINED');
+    expect((await prisma.iiosDlqItem.findUniqueOrThrow({ where: { id: item.id } })).retryCount).toBe(1);
+
+    const r2 = await dlq.replay(item.id); // handler now succeeds
+    expect(r2.status).toBe('RESOLVED');
+    expect((await prisma.iiosDlqItem.findUniqueOrThrow({ where: { id: item.id } })).resolvedAt).toBeTruthy();
+  });
+});
@@ -0,0 +1,108 @@
+import { Injectable, NotFoundException } from '@nestjs/common';
+import type { CloudEvent } from '@insignia/iios-contracts';
+import { PrismaService } from '../prisma/prisma.service';
+
+type ReplayHandler = (event: CloudEvent) => Promise<void>;
+
+export interface DlqRecordInput {
+  sourceType: string;
+  sourceId: string;
+  failureStage: string;
+  errorCode?: string;
+  payloadRef?: string;
+  scopeId?: string;
+  consumerName?: string;
+}
+
+/**
+ * Dead-letter queue (P9, KG-13). Failures are quarantined here — never silently
+ * swallowed — and replayed deterministically: a consumer registers its handler, and
+ * replay clears the idempotency claim then re-runs exactly that consumer (siblings
+ * stay untouched because the claim ledger makes re-processing a no-op for them).
+ */
+@Injectable()
+export class DlqService {
+  private readonly handlers = new Map<string, ReplayHandler>();
+
+  constructor(private readonly prisma: PrismaService) {}
+
+  registerHandler(consumer: string, fn: ReplayHandler): void {
+    this.handlers.set(consumer, fn);
+  }
+
+  /** Quarantine a failure (idempotent per (consumer, sourceId) — repeats bump retryCount). */
+  async record(input: DlqRecordInput): Promise<void> {
+    const existing = await this.prisma.iiosDlqItem.findFirst({
+      where: { consumerName: input.consumerName ?? null, sourceId: input.sourceId },
+    });
+    if (existing) {
+      await this.prisma.iiosDlqItem.update({
+        where: { id: existing.id },
+        data: { retryCount: { increment: 1 }, lastFailedAt: new Date(), errorCode: input.errorCode ?? existing.errorCode, status: 'QUARANTINED' },
+      });
+      return;
+    }
+    await this.prisma.iiosDlqItem.create({
+      data: {
+        sourceType: input.sourceType,
+        sourceId: input.sourceId,
+        failureStage: input.failureStage,
+        errorCode: input.errorCode,
+        payloadRef: input.payloadRef,
+        scopeId: input.scopeId,
+        consumerName: input.consumerName,
+      },
+    });
+  }
+
+  /** A projector/consumer threw: dead-letter the event AND clear its claim so it can replay. */
+  async onConsumerFailure(consumer: string, event: CloudEvent, err: unknown): Promise<void> {
+    await this.record({
+      sourceType: 'projection',
+      sourceId: event.id,
+      failureStage: consumer,
+      errorCode: (err as Error)?.message?.slice(0, 160),
+      payloadRef: event.id,
+      scopeId: event?.insignia?.scopeSnapshotId,
+      consumerName: consumer,
+    });
+    await this.prisma.iiosProcessedEvent.deleteMany({ where: { consumerName: consumer, eventId: event.id } });
+  }
+
+  async listDlq(scopeId?: string, status?: string) {
+    return this.prisma.iiosDlqItem.findMany({
+      where: { ...(scopeId ? { scopeId } : {}), ...(status ? { status } : {}) },
+      orderBy: { lastFailedAt: 'desc' },
+      take: 200,
+    });
+  }
+
+  /** Re-run the failed consumer (deterministic) or hand an outbox event back to the relay. */
+  async replay(id: string): Promise<{ status: string }> {
+    const item = await this.prisma.iiosDlqItem.findUnique({ where: { id } });
+    if (!item) throw new NotFoundException('dlq item not found');
+
+    // Projection failure: clear the claim + await the registered handler.
+    if (item.consumerName && item.payloadRef && this.handlers.has(item.consumerName)) {
+      await this.prisma.iiosProcessedEvent.deleteMany({ where: { consumerName: item.consumerName, eventId: item.sourceId } });
+      const evt = await this.prisma.iiosOutboxEvent.findUnique({ where: { eventId: item.payloadRef } });
+      if (!evt) throw new NotFoundException('event payload not found');
+      try {
+        await this.handlers.get(item.consumerName)!(evt.cloudEvent as unknown as CloudEvent);
+      } catch (err) {
+        await this.prisma.iiosDlqItem.update({ where: { id }, data: { retryCount: { increment: 1 }, lastFailedAt: new Date(), errorCode: (err as Error)?.message?.slice(0, 160) } });
+        return { status: 'QUARANTINED' };
+      }
+      await this.prisma.iiosDlqItem.update({ where: { id }, data: { status: 'RESOLVED', resolvedAt: new Date() } });
+      return { status: 'RESOLVED' };
+    }
+
+    // Outbox publish failure: hand the event back to the relay (reset to PENDING).
+    if (item.payloadRef) {
+      await this.prisma.iiosOutboxEvent.update({ where: { eventId: item.payloadRef }, data: { status: 'PENDING', nextAttemptAt: new Date(), attempts: 0 } });
+      await this.prisma.iiosDlqItem.update({ where: { id }, data: { status: 'RESOLVED', resolvedAt: new Date() } });
+      return { status: 'RESOLVED' };
+    }
+    return { status: item.status };
+  }
+}
@@ -1,9 +1,10 @@
 import { Module } from '@nestjs/common';
 import { OutboxBus } from './outbox.bus';
 import { OutboxRelay } from './outbox.relay';
+import { DlqService } from './dlq.service';

@Module({
-  providers: [OutboxBus, OutboxRelay],
-  exports: [OutboxBus, OutboxRelay],
+  providers: [OutboxBus, OutboxRelay, DlqService],
+  exports: [OutboxBus, OutboxRelay, DlqService],
 })
 export class OutboxModule {}
@@ -8,7 +8,7 @@ import type { PrismaClient } from '@prisma/client';
 export async function resetDb(prisma: PrismaClient): Promise<void> {
  await prisma.$executeRawUnsafe(
    `TRUNCATE TABLE
-      "IiosAuditLink",
+      "IiosDlqItem","IiosAuditLink",
      "IiosMeetingActionItem","IiosActionItem","IiosTranscriptSegment","IiosMeetingTranscript","IiosMeetingSummary","IiosMeetingParticipant","IiosMeeting","IiosMeetingRequest",
      "IiosCalendarSyncCursor","IiosCalendarEvent","IiosCalendarProviderAccount","IiosAvailabilityWindow",
      "IiosAiEvidenceLink","IiosAiClaim","IiosAiToolCall","IiosAiArtifact","IiosAiModelRun","IiosAiJob","IiosEmbeddingRef",