From 208fb0b4f005fb3a10ae22a031f61cb4c8abc564 Mon Sep 17 00:00:00 2001 From: maaz519 Date: Fri, 3 Jul 2026 01:28:48 +0530 Subject: [PATCH] feat(p9): IiosDlqItem + DlqService (dead-letter + deterministic replay) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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) --- .../20260702195657_dlq/migration.sql | 24 ++++ packages/iios-service/prisma/schema.prisma | 21 ++++ .../src/outbox/dlq.service.spec.ts | 62 ++++++++++ .../iios-service/src/outbox/dlq.service.ts | 108 ++++++++++++++++++ .../iios-service/src/outbox/outbox.module.ts | 5 +- .../iios-service/src/test-utils/reset-db.ts | 2 +- 6 files changed, 219 insertions(+), 3 deletions(-) create mode 100644 packages/iios-service/prisma/migrations/20260702195657_dlq/migration.sql create mode 100644 packages/iios-service/src/outbox/dlq.service.spec.ts create mode 100644 packages/iios-service/src/outbox/dlq.service.ts diff --git a/packages/iios-service/prisma/migrations/20260702195657_dlq/migration.sql b/packages/iios-service/prisma/migrations/20260702195657_dlq/migration.sql new file mode 100644 index 0000000..338c6bc --- /dev/null +++ b/packages/iios-service/prisma/migrations/20260702195657_dlq/migration.sql @@ -0,0 +1,24 @@ +-- CreateTable +CREATE TABLE "IiosDlqItem" ( + "id" TEXT NOT NULL, + "sourceType" TEXT NOT NULL, + "sourceId" TEXT NOT NULL, + "failureStage" TEXT NOT NULL, + "errorCode" TEXT, + "retryCount" INTEGER NOT NULL DEFAULT 0, + "payloadRef" TEXT, + "consumerName" TEXT, + "scopeId" TEXT, + "status" TEXT NOT NULL DEFAULT 'QUARANTINED', + "createdAt" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP, + "lastFailedAt" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP, + "resolvedAt" TIMESTAMP(3), + + CONSTRAINT "IiosDlqItem_pkey" PRIMARY KEY ("id") +); + +-- CreateIndex +CREATE INDEX "IiosDlqItem_scopeId_status_idx" ON "IiosDlqItem"("scopeId", "status"); + +-- CreateIndex +CREATE UNIQUE INDEX "IiosDlqItem_consumerName_sourceId_key" ON "IiosDlqItem"("consumerName", "sourceId"); diff --git a/packages/iios-service/prisma/schema.prisma b/packages/iios-service/prisma/schema.prisma index 9977329..c277e5e 100644 --- a/packages/iios-service/prisma/schema.prisma +++ b/packages/iios-service/prisma/schema.prisma @@ -1178,3 +1178,24 @@ model IiosAuditLink { @@index([resourceType, resourceId]) @@index([scopeId]) } + +/// Dead-letter / replay item (P9, mandated dlq_item). A consumer/relay failure is +/// quarantined here (never silently swallowed — KG-13) and can be replayed by ops. +model IiosDlqItem { + id String @id @default(cuid()) + sourceType String // 'projection' | 'outbox' | 'outbound' | ... + sourceId String + failureStage String + errorCode String? + retryCount Int @default(0) + payloadRef String? // the outbox eventId to re-dispatch + consumerName String? + scopeId String? // tenant quarantine + status String @default("QUARANTINED") // QUARANTINED | RESOLVED | ARCHIVED + createdAt DateTime @default(now()) + lastFailedAt DateTime @default(now()) + resolvedAt DateTime? + + @@unique([consumerName, sourceId]) + @@index([scopeId, status]) +} diff --git a/packages/iios-service/src/outbox/dlq.service.spec.ts b/packages/iios-service/src/outbox/dlq.service.spec.ts new file mode 100644 index 0000000..35484a6 --- /dev/null +++ b/packages/iios-service/src/outbox/dlq.service.spec.ts @@ -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 { + 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(); + }); +}); diff --git a/packages/iios-service/src/outbox/dlq.service.ts b/packages/iios-service/src/outbox/dlq.service.ts new file mode 100644 index 0000000..4470040 --- /dev/null +++ b/packages/iios-service/src/outbox/dlq.service.ts @@ -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; + +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(); + + 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 { + 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 { + 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 }; + } +} diff --git a/packages/iios-service/src/outbox/outbox.module.ts b/packages/iios-service/src/outbox/outbox.module.ts index abdcf2b..e68547e 100644 --- a/packages/iios-service/src/outbox/outbox.module.ts +++ b/packages/iios-service/src/outbox/outbox.module.ts @@ -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 {} diff --git a/packages/iios-service/src/test-utils/reset-db.ts b/packages/iios-service/src/test-utils/reset-db.ts index 6913cb3..5ce8371 100644 --- a/packages/iios-service/src/test-utils/reset-db.ts +++ b/packages/iios-service/src/test-utils/reset-db.ts @@ -8,7 +8,7 @@ import type { PrismaClient } from '@prisma/client'; export async function resetDb(prisma: PrismaClient): Promise { 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",