diff --git a/packages/iios-service/src/adapters/adapters.spec.ts b/packages/iios-service/src/adapters/adapters.spec.ts index ff30dd1..8c41bbc 100644 --- a/packages/iios-service/src/adapters/adapters.spec.ts +++ b/packages/iios-service/src/adapters/adapters.spec.ts @@ -1,5 +1,6 @@ import { describe, it, expect, beforeAll, afterAll, beforeEach } from 'vitest'; import { PrismaClient } from '@prisma/client'; +import { DlqService } from '../outbox/dlq.service'; import { resetDb } from '../test-utils/reset-db'; import { makeFakePorts } from '@insignia/iios-testkit'; import { IIOS_EVENTS, type CloudEvent } from '@insignia/iios-contracts'; @@ -23,7 +24,7 @@ const SECRET = 'dev-adapter-secret'; // AdapterRegistry default const registry = () => new AdapterRegistry(); const inbound = () => new InboundService(asService, registry()); const projector = () => - new RawEventProjector(asService, new OutboxBus(), registry(), new IngestService(asService, makeFakePorts(), new ActorResolver(asService))); + new RawEventProjector(asService, new OutboxBus(), new DlqService(asService), registry(), new IngestService(asService, makeFakePorts(), new ActorResolver(asService))); async function rawReceivedEvents(): Promise { const rows = await prisma.iiosOutboxEvent.findMany({ where: { eventType: IIOS_EVENTS.rawReceived }, orderBy: { createdAt: 'asc' } }); diff --git a/packages/iios-service/src/adapters/raw-event.projector.ts b/packages/iios-service/src/adapters/raw-event.projector.ts index bbfb2bb..b6ae133 100644 --- a/packages/iios-service/src/adapters/raw-event.projector.ts +++ b/packages/iios-service/src/adapters/raw-event.projector.ts @@ -2,6 +2,7 @@ import { Injectable, OnModuleInit } from '@nestjs/common'; import { CloudEvent, IIOS_EVENTS } from '@insignia/iios-contracts'; import { PrismaService } from '../prisma/prisma.service'; import { OutboxBus } from '../outbox/outbox.bus'; +import { DlqService } from '../outbox/dlq.service'; import { AdapterRegistry } from './adapter.registry'; import { IngestService } from '../interactions/ingest.service'; @@ -17,12 +18,14 @@ export class RawEventProjector implements OnModuleInit { constructor( private readonly prisma: PrismaService, private readonly bus: OutboxBus, + private readonly dlq: DlqService, private readonly registry: AdapterRegistry, private readonly ingest: IngestService, ) {} onModuleInit(): void { - this.bus.on(IIOS_EVENTS.rawReceived, (p) => void this.onRawReceived(p as CloudEvent).catch(() => {})); + this.dlq.registerHandler(this.consumer, (e) => this.onRawReceived(e)); + this.bus.on(IIOS_EVENTS.rawReceived, (p) => void this.onRawReceived(p as CloudEvent).catch((err) => this.dlq.onConsumerFailure(this.consumer, p as CloudEvent, err))); } async onRawReceived(event: CloudEvent): Promise { diff --git a/packages/iios-service/src/ai/ai-routing.spec.ts b/packages/iios-service/src/ai/ai-routing.spec.ts index a490fde..adca2ee 100644 --- a/packages/iios-service/src/ai/ai-routing.spec.ts +++ b/packages/iios-service/src/ai/ai-routing.spec.ts @@ -1,6 +1,7 @@ import { randomUUID } from 'node:crypto'; import { describe, it, expect, beforeAll, afterAll, beforeEach } from 'vitest'; import { PrismaClient } from '@prisma/client'; +import { DlqService } from '../outbox/dlq.service'; import { resetDb } from '../test-utils/reset-db'; import { makeFakePorts, makeFakeInference } from '@insignia/iios-testkit'; import { IIOS_EVENTS, type CloudEvent } from '@insignia/iios-contracts'; @@ -104,7 +105,7 @@ describe('AI × routing (P7, KG-05: AI proposes, never approves)', () => { it('AiProjector auto-classifies a channel-bearing interaction, idempotently', async () => { const { interactionId, event } = await ingestInbound('There is a party at 9 PM tonight!'); - const p = new AiProjector(asService, new OutboxBus(), ai()); + const p = new AiProjector(asService, new OutboxBus(), new DlqService(asService), ai()); await p.onNormalized(event); await p.onNormalized(event); // replay expect(await prisma.iiosAiJob.count({ where: { interactionId, jobType: 'CLASSIFY' } })).toBe(1); diff --git a/packages/iios-service/src/ai/ai.projector.ts b/packages/iios-service/src/ai/ai.projector.ts index c68d043..62ff634 100644 --- a/packages/iios-service/src/ai/ai.projector.ts +++ b/packages/iios-service/src/ai/ai.projector.ts @@ -2,6 +2,7 @@ import { Injectable, OnModuleInit } from '@nestjs/common'; import { CloudEvent, IIOS_EVENTS } from '@insignia/iios-contracts'; import { PrismaService } from '../prisma/prisma.service'; import { OutboxBus } from '../outbox/outbox.bus'; +import { DlqService } from '../outbox/dlq.service'; import { AiJobService } from './ai.service'; /** @@ -18,11 +19,13 @@ export class AiProjector implements OnModuleInit { constructor( private readonly prisma: PrismaService, private readonly bus: OutboxBus, + private readonly dlq: DlqService, private readonly ai: AiJobService, ) {} onModuleInit(): void { - this.bus.on(IIOS_EVENTS.interactionNormalized, (p) => void this.onNormalized(p as CloudEvent).catch(() => {})); + this.dlq.registerHandler(this.consumer, (e) => this.onNormalized(e)); + this.bus.on(IIOS_EVENTS.interactionNormalized, (p) => void this.onNormalized(p as CloudEvent).catch((err) => this.dlq.onConsumerFailure(this.consumer, p as CloudEvent, err))); } async onNormalized(event: CloudEvent): Promise { diff --git a/packages/iios-service/src/calendar/calendar-projector.spec.ts b/packages/iios-service/src/calendar/calendar-projector.spec.ts index 160fb33..d0e49e9 100644 --- a/packages/iios-service/src/calendar/calendar-projector.spec.ts +++ b/packages/iios-service/src/calendar/calendar-projector.spec.ts @@ -1,6 +1,7 @@ import { randomUUID } from 'node:crypto'; import { describe, it, expect, beforeAll, afterAll, beforeEach } from 'vitest'; import { PrismaClient } from '@prisma/client'; +import { DlqService } from '../outbox/dlq.service'; import { resetDb } from '../test-utils/reset-db'; import { makeFakePorts, makeFakeInference } from '@insignia/iios-testkit'; import { IIOS_EVENTS, type CloudEvent } from '@insignia/iios-contracts'; @@ -57,7 +58,7 @@ describe('Calendar reminders + projector (P8)', () => { where: { eventType: IIOS_EVENTS.aiArtifactAccepted, aggregateId: artifact!.id }, })).cloudEvent as unknown as CloudEvent; - const p = new CalendarProjector(asService, new OutboxBus()); + const p = new CalendarProjector(asService, new OutboxBus(), new DlqService(asService)); await p.onArtifactAccepted(event); await p.onArtifactAccepted(event); // replay const claim = await prisma.iiosAiClaim.findFirstOrThrow({ where: { artifactId: artifact!.id, claimType: 'EVENT' } }); diff --git a/packages/iios-service/src/calendar/calendar.projector.ts b/packages/iios-service/src/calendar/calendar.projector.ts index 9347af8..9dc875c 100644 --- a/packages/iios-service/src/calendar/calendar.projector.ts +++ b/packages/iios-service/src/calendar/calendar.projector.ts @@ -2,6 +2,7 @@ import { Injectable, OnModuleInit } from '@nestjs/common'; import { CloudEvent, IIOS_EVENTS } from '@insignia/iios-contracts'; import { PrismaService } from '../prisma/prisma.service'; import { OutboxBus } from '../outbox/outbox.bus'; +import { DlqService } from '../outbox/dlq.service'; /** * P7→P8 auto-bridge (P8). When a P7 AI artifact is accepted and it carries an @@ -16,10 +17,12 @@ export class CalendarProjector implements OnModuleInit { constructor( private readonly prisma: PrismaService, private readonly bus: OutboxBus, + private readonly dlq: DlqService, ) {} onModuleInit(): void { - this.bus.on(IIOS_EVENTS.aiArtifactAccepted, (p) => void this.onArtifactAccepted(p as CloudEvent).catch(() => {})); + this.dlq.registerHandler(this.consumer, (e) => this.onArtifactAccepted(e)); + this.bus.on(IIOS_EVENTS.aiArtifactAccepted, (p) => void this.onArtifactAccepted(p as CloudEvent).catch((err) => this.dlq.onConsumerFailure(this.consumer, p as CloudEvent, err))); } async onArtifactAccepted(event: CloudEvent): Promise { diff --git a/packages/iios-service/src/inbox/inbox.projector.ts b/packages/iios-service/src/inbox/inbox.projector.ts index 0f49816..819cdc6 100644 --- a/packages/iios-service/src/inbox/inbox.projector.ts +++ b/packages/iios-service/src/inbox/inbox.projector.ts @@ -3,6 +3,7 @@ import { Prisma } from '@prisma/client'; import { CloudEvent, IIOS_EVENTS } from '@insignia/iios-contracts'; import { PrismaService } from '../prisma/prisma.service'; import { OutboxBus } from '../outbox/outbox.bus'; +import { DlqService } from '../outbox/dlq.service'; interface MessageSentData { interactionId: string; @@ -28,11 +29,13 @@ export class InboxProjector implements OnModuleInit { constructor( private readonly prisma: PrismaService, private readonly bus: OutboxBus, + private readonly dlq: DlqService, ) {} onModuleInit(): void { - this.bus.on(IIOS_EVENTS.messageSent, (p) => void this.onMessageSent(p as CloudEvent).catch(() => {})); - this.bus.on(IIOS_EVENTS.messageRead, (p) => void this.onMessageRead(p as CloudEvent).catch(() => {})); + this.dlq.registerHandler(this.consumer, (e) => this.onMessageSent(e)); + this.bus.on(IIOS_EVENTS.messageSent, (p) => void this.onMessageSent(p as CloudEvent).catch((err) => this.dlq.onConsumerFailure(this.consumer, p as CloudEvent, err))); + this.bus.on(IIOS_EVENTS.messageRead, (p) => void this.onMessageRead(p as CloudEvent).catch((err) => this.dlq.onConsumerFailure(this.consumer, p as CloudEvent, err))); } async onMessageSent(event: CloudEvent): Promise { diff --git a/packages/iios-service/src/inbox/inbox.spec.ts b/packages/iios-service/src/inbox/inbox.spec.ts index f0e55f3..3569459 100644 --- a/packages/iios-service/src/inbox/inbox.spec.ts +++ b/packages/iios-service/src/inbox/inbox.spec.ts @@ -1,5 +1,6 @@ import { describe, it, expect, beforeAll, afterAll, beforeEach } from 'vitest'; import { PrismaClient } from '@prisma/client'; +import { DlqService } from '../outbox/dlq.service'; import { resetDb } from '../test-utils/reset-db'; import { makeFakePorts } from '@insignia/iios-testkit'; import { IIOS_EVENTS, PolicyDeniedError, type CloudEvent } from '@insignia/iios-contracts'; @@ -15,7 +16,7 @@ const prisma = new PrismaClient({ datasources: { db: { url } } }); const asService = prisma as unknown as PrismaService; const actors = new ActorResolver(asService); const ms = () => new MessageService(asService, makeFakePorts(), actors); -const projector = () => new InboxProjector(asService, new OutboxBus()); +const projector = () => new InboxProjector(asService, new OutboxBus(), new DlqService(asService)); const alice: MessagePrincipal = { userId: 'alice', orgId: 'org_demo', appId: 'portal-demo', displayName: 'Alice' }; const bob: MessagePrincipal = { userId: 'bob', orgId: 'org_demo', appId: 'portal-demo', displayName: 'Bob' }; diff --git a/packages/iios-service/src/outbox/chaos.spec.ts b/packages/iios-service/src/outbox/chaos.spec.ts new file mode 100644 index 0000000..9b41103 --- /dev/null +++ b/packages/iios-service/src/outbox/chaos.spec.ts @@ -0,0 +1,76 @@ +import { randomUUID } from 'node:crypto'; +import { describe, it, expect, beforeAll, afterAll, beforeEach } from 'vitest'; +import { PrismaClient } from '@prisma/client'; +import { resetDb } from '../test-utils/reset-db'; +import { makeFakePorts, makeFakeInference } from '@insignia/iios-testkit'; +import { IIOS_EVENTS, type CloudEvent } from '@insignia/iios-contracts'; +import { ActorResolver } from '../identity/actor.resolver'; +import { IngestService } from '../interactions/ingest.service'; +import { AiJobService } from '../ai/ai.service'; +import { AiBudgetGuard } from '../ai/ai-budget.guard'; +import { AiProjector } from '../ai/ai.projector'; +import { OutboxBus } from './outbox.bus'; +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 actors = new ActorResolver(asService); +const ai = () => new AiJobService(asService, makeFakePorts(), makeFakeInference(), new AiBudgetGuard(asService), actors); + +async function ingestInbound(text: string): Promise<{ interactionId: string; scopeId: string; event: CloudEvent }> { + const ingest = new IngestService(asService, makeFakePorts(), actors); + const res = await ingest.ingest( + { + scope: { orgId: 'org_demo', appId: 'portal-demo' }, + channel: { type: 'WEBHOOK', externalChannelId: 'webhook' }, + source: { handleKind: 'EXTERNAL_VISITOR', externalId: 'ext-1' }, + kind: 'MESSAGE', + thread: { externalThreadId: `wh-${randomUUID().slice(0, 6)}` }, + parts: [{ kind: 'TEXT', bodyText: text }], + occurredAt: new Date().toISOString(), + providerEventId: `pe-${randomUUID()}`, + }, + `ing-${randomUUID()}`, + ); + const interaction = await prisma.iiosInteraction.findUniqueOrThrow({ where: { id: res.interactionId } }); + const ev = (await prisma.iiosOutboxEvent.findFirstOrThrow({ + where: { eventType: IIOS_EVENTS.interactionNormalized, aggregateId: res.interactionId }, + })).cloudEvent as unknown as CloudEvent; + return { interactionId: res.interactionId, scopeId: interaction.scopeId, event: ev }; +} + +beforeAll(async () => { await prisma.$connect(); }); +afterAll(async () => { await prisma.$disconnect(); }); +beforeEach(async () => { await resetDb(prisma); }); + +describe('chaos: poison consumer → DLQ → replay (P9)', () => { + it('a consumer failure is dead-lettered (not swallowed), the lane keeps flowing, and replay resolves it', async () => { + const dlq = new DlqService(asService); + const projector = new AiProjector(asService, new OutboxBus(), dlq, ai()); + dlq.registerHandler('ai-projector', (e) => projector.onNormalized(e)); // healthy handler for replay + + const poison = await ingestInbound('There is a party at 9 PM tonight!'); + // Simulate the consumer crashing on the poison (exactly what the projector's .catch does). + await prisma.iiosProcessedEvent.create({ data: { consumerName: 'ai-projector', eventId: poison.event.id } }); // claimed-then-failed + await dlq.onConsumerFailure('ai-projector', poison.event, new Error('chaos: AI down')); + + // Dead-lettered (tenant-tagged), claim cleared, and NOT processed. + const item = await prisma.iiosDlqItem.findFirstOrThrow({ where: { consumerName: 'ai-projector' } }); + expect(item.scopeId).toBe(poison.scopeId); + expect(item.status).toBe('QUARANTINED'); + expect(await prisma.iiosProcessedEvent.count({ where: { consumerName: 'ai-projector', eventId: poison.event.id } })).toBe(0); + expect(await prisma.iiosAiJob.count({ where: { interactionId: poison.interactionId } })).toBe(0); + + // Lane not starved: a SIBLING event still processes through the healthy consumer. + const sibling = await ingestInbound('Team standup notes are posted.'); + await projector.onNormalized(sibling.event); + expect(await prisma.iiosAiJob.count({ where: { interactionId: sibling.interactionId } })).toBe(1); + + // Replay the poison → the registered handler re-runs it → RESOLVED + now processed. + const r = await dlq.replay(item.id); + expect(r.status).toBe('RESOLVED'); + expect(await prisma.iiosAiJob.count({ where: { interactionId: poison.interactionId } })).toBe(1); + }); +}); diff --git a/packages/iios-service/src/outbox/dlq.service.ts b/packages/iios-service/src/outbox/dlq.service.ts index 4470040..4b0c218 100644 --- a/packages/iios-service/src/outbox/dlq.service.ts +++ b/packages/iios-service/src/outbox/dlq.service.ts @@ -85,7 +85,9 @@ export class DlqService { // 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 } }); + // The DLQ item references the CloudEvent by its own `id` (not the outbox row's cuid), + // so recover the payload by matching cloudEvent->>'id'. + const evt = await this.prisma.iiosOutboxEvent.findFirst({ where: { cloudEvent: { path: ['id'], equals: item.payloadRef } } }); if (!evt) throw new NotFoundException('event payload not found'); try { await this.handlers.get(item.consumerName)!(evt.cloudEvent as unknown as CloudEvent); diff --git a/packages/iios-service/src/outbox/outbox.relay.ts b/packages/iios-service/src/outbox/outbox.relay.ts index 6f6f1ca..7ac61d4 100644 --- a/packages/iios-service/src/outbox/outbox.relay.ts +++ b/packages/iios-service/src/outbox/outbox.relay.ts @@ -1,6 +1,7 @@ import { Injectable, Logger, OnModuleDestroy, OnModuleInit } from '@nestjs/common'; import { PrismaService } from '../prisma/prisma.service'; import { OutboxBus } from './outbox.bus'; +import { DlqService } from './dlq.service'; import { logJson } from '../observability/logger'; /** @@ -17,9 +18,12 @@ export class OutboxRelay implements OnModuleInit, OnModuleDestroy { private readonly logger = new Logger(OutboxRelay.name); private timer?: ReturnType; + private readonly maxAttempts = Number(process.env.IIOS_OUTBOX_MAX_ATTEMPTS ?? 5); + constructor( private readonly prisma: PrismaService, private readonly bus: OutboxBus, + private readonly dlq: DlqService, ) {} // Tests construct via `new OutboxRelay(...)`, so onModuleInit never runs there @@ -68,11 +72,18 @@ export class OutboxRelay implements OnModuleInit, OnModuleDestroy { data: { status: 'PUBLISHED', publishedAt: new Date() }, }); published++; - } catch { - await this.prisma.iiosOutboxEvent.update({ - where: { eventId }, - data: { status: 'PENDING', nextAttemptAt: new Date(Date.now() + 30_000) }, - }); + } catch (err) { + // After maxAttempts, stop retrying forever — dead-letter the event (KG-13). + if (evt.attempts >= this.maxAttempts) { + const ce = evt.cloudEvent as { insignia?: { scopeSnapshotId?: string } } | null; + await this.prisma.iiosOutboxEvent.update({ where: { eventId }, data: { status: 'FAILED' } }); + await this.dlq.record({ sourceType: 'outbox', sourceId: eventId, failureStage: 'relay.publish', errorCode: (err as Error)?.message?.slice(0, 160), payloadRef: eventId, scopeId: ce?.insignia?.scopeSnapshotId }); + } else { + await this.prisma.iiosOutboxEvent.update({ + where: { eventId }, + data: { status: 'PENDING', nextAttemptAt: new Date(Date.now() + 30_000) }, + }); + } } } return published; diff --git a/packages/iios-service/src/outbox/replay.spec.ts b/packages/iios-service/src/outbox/replay.spec.ts index 0d33fff..bc8205f 100644 --- a/packages/iios-service/src/outbox/replay.spec.ts +++ b/packages/iios-service/src/outbox/replay.spec.ts @@ -1,5 +1,6 @@ import { describe, it, expect, beforeAll, afterAll, beforeEach } from 'vitest'; import { PrismaClient } from '@prisma/client'; +import { DlqService } from '../outbox/dlq.service'; import { resetDb } from '../test-utils/reset-db'; import { makeFakePorts, portalMessageBasic, replayTwice, isIdempotent } from '@insignia/iios-testkit'; import { IngestService } from '../interactions/ingest.service'; @@ -36,7 +37,7 @@ describe('outbox relay + replay', () => { const bus = new OutboxBus(); const received: string[] = []; bus.onAny((eventType) => received.push(eventType)); - const relay = new OutboxRelay(asService, bus); + const relay = new OutboxRelay(asService, bus, new DlqService(asService)); await ingest.ingest(portalMessageBasic, portalMessageBasic.providerEventId!); diff --git a/packages/iios-service/src/routing/route.projector.ts b/packages/iios-service/src/routing/route.projector.ts index e723853..d9e1845 100644 --- a/packages/iios-service/src/routing/route.projector.ts +++ b/packages/iios-service/src/routing/route.projector.ts @@ -2,6 +2,7 @@ import { Injectable, OnModuleInit } from '@nestjs/common'; import { CloudEvent, IIOS_EVENTS } from '@insignia/iios-contracts'; import { PrismaService } from '../prisma/prisma.service'; import { OutboxBus } from '../outbox/outbox.bus'; +import { DlqService } from '../outbox/dlq.service'; import { RouteService } from './route.service'; /** @@ -18,11 +19,13 @@ export class RouteProjector implements OnModuleInit { constructor( private readonly prisma: PrismaService, private readonly bus: OutboxBus, + private readonly dlq: DlqService, private readonly routes: RouteService, ) {} onModuleInit(): void { - this.bus.on(IIOS_EVENTS.interactionNormalized, (p) => void this.onNormalized(p as CloudEvent).catch(() => {})); + this.dlq.registerHandler(this.consumer, (e) => this.onNormalized(e)); + this.bus.on(IIOS_EVENTS.interactionNormalized, (p) => void this.onNormalized(p as CloudEvent).catch((err) => this.dlq.onConsumerFailure(this.consumer, p as CloudEvent, err))); } async onNormalized(event: CloudEvent): Promise { diff --git a/packages/iios-service/src/routing/route.spec.ts b/packages/iios-service/src/routing/route.spec.ts index 7561bf9..d86e09a 100644 --- a/packages/iios-service/src/routing/route.spec.ts +++ b/packages/iios-service/src/routing/route.spec.ts @@ -1,6 +1,7 @@ import { randomUUID } from 'node:crypto'; import { describe, it, expect, beforeAll, afterAll, beforeEach } from 'vitest'; import { PrismaClient, IiosRouteMode } from '@prisma/client'; +import { DlqService } from '../outbox/dlq.service'; import { resetDb } from '../test-utils/reset-db'; import { makeFakePorts } from '@insignia/iios-testkit'; import { MessageService, type MessagePrincipal } from '../messaging/message.service'; @@ -159,7 +160,7 @@ describe('RouteService approve/deny → execute (P6)', () => { describe('RouteProjector — AUTOMATIC forwarding (P6)', () => { const routeSvc = () => new RouteService(asService, new RouteSimulator(asService, new RuleEngine()), new OutboundService(asService, new CapabilityBroker(makeFakePorts(), new CapabilityProviderRegistry())), actors); - const projector = () => new RouteProjector(asService, new OutboxBus(), routeSvc()); + const projector = () => new RouteProjector(asService, new OutboxBus(), new DlqService(asService), routeSvc()); // Ingest an inbound (channel-bearing) interaction — routable, unlike native DMs. async function ingestInbound(text: string): Promise<{ interactionId: string; scopeId: string; event: CloudEvent }> {