diff --git a/src/common/utils/db-errors.ts b/src/common/utils/db-errors.ts new file mode 100644 index 000000000..01c5657d0 --- /dev/null +++ b/src/common/utils/db-errors.ts @@ -0,0 +1,15 @@ +import { QueryFailedError } from 'typeorm'; + +/** + * Cross-dialect unique-constraint-violation check by driver code/message, for the two dialects we ship + * (sqlite dev, postgres prod). Lets insert-or-converge (RMW) paths distinguish a real duplicate from an + * unrelated failure without depending on a specific driver. Add another branch if a third driver is ever + * supported. + */ +export function isUniqueViolation(err: unknown): boolean { + if (!(err instanceof QueryFailedError)) return false; + const driver = err.driverError as { code?: string; message?: string } | undefined; + const code = driver?.code ?? ''; + const message = driver?.message ?? err.message ?? ''; + return code === '23505' /* postgres */ || /UNIQUE constraint failed|SQLITE_CONSTRAINT/i.test(message); +} diff --git a/src/database/migrations/1782100000000-WidenIngressDedupKey.ts b/src/database/migrations/1782100000000-WidenIngressDedupKey.ts new file mode 100644 index 000000000..faee86c07 --- /dev/null +++ b/src/database/migrations/1782100000000-WidenIngressDedupKey.ts @@ -0,0 +1,28 @@ +import { MigrationInterface, QueryRunner } from 'typeorm'; + +/** + * Widen the inbound ingress dedup key from (instanceId, providerDeliveryId) to + * (pluginId, instanceId, providerDeliveryId). + * + * instanceId is a caller-supplied string that is only unique WITHIN a plugin, so two different plugins + * sharing the same instanceId string would collide on the old 2-column unique index and a legitimate + * delivery for the second plugin would be dropped as a false "duplicate". pluginId is already stored on + * every row (recordOrSkip inserts it), so this is a pure loosening of the constraint — no data loss, no + * false-negative risk. Kept as a stable-named DROP/CREATE unique index (portable to sqlite + postgres), + * mirroring AddIntegrationFabric's index style. + */ +export class WidenIngressDedupKey1782100000000 implements MigrationInterface { + public async up(queryRunner: QueryRunner): Promise { + await queryRunner.query(`DROP INDEX IF EXISTS "UQ_ingress_events_instance_delivery"`); + await queryRunner.query( + `CREATE UNIQUE INDEX "UQ_ingress_events_instance_delivery" ON "ingress_events" ("pluginId", "instanceId", "providerDeliveryId")`, + ); + } + + public async down(queryRunner: QueryRunner): Promise { + await queryRunner.query(`DROP INDEX IF EXISTS "UQ_ingress_events_instance_delivery"`); + await queryRunner.query( + `CREATE UNIQUE INDEX "UQ_ingress_events_instance_delivery" ON "ingress_events" ("instanceId", "providerDeliveryId")`, + ); + } +} diff --git a/src/modules/integration/conversation-mapping.service.spec.ts b/src/modules/integration/conversation-mapping.service.spec.ts index 55b4c87c6..2f498e365 100644 --- a/src/modules/integration/conversation-mapping.service.spec.ts +++ b/src/modules/integration/conversation-mapping.service.spec.ts @@ -1,6 +1,6 @@ import { DataSource } from 'typeorm'; import { ConversationMapping } from './entities/conversation-mapping.entity'; -import { ConversationMappingService, MappingKey } from './conversation-mapping.service'; +import { ConversationMappingConflict, ConversationMappingService, MappingKey } from './conversation-mapping.service'; import { AddIntegrationFabric1781900000000 } from '../../database/migrations/1781900000000-AddIntegrationFabric'; describe('ConversationMappingService', () => { @@ -51,6 +51,22 @@ describe('ConversationMappingService', () => { expect(stale).toBeNull(); }); + it('rethrows ConversationMappingConflict when a providerConversationId is already bound to a different chat', async () => { + // Reverse unique key (pluginId, instanceId, providerConversationId): binding conv-1 to chat-1 then to + // a different chat-2 for the same plugin+instance is a genuine conflict with no forward row to + // converge onto — it must surface, not silently corrupt or fail-soft to a nonexistent row. + await service.upsert(key, 'conv-1'); + await expect( + service.upsert({ sessionId: 'sess-1', chatId: 'chat-2', pluginId: 'chatwoot', instanceId: 'acct1' }, 'conv-1'), + ).rejects.toBeInstanceOf(ConversationMappingConflict); + }); + + it('converges (updates, does not throw) when the same forward key already exists', async () => { + await service.upsert(key, 'conv-1'); + await expect(service.upsert(key, 'conv-9')).resolves.toBeUndefined(); + expect((await service.get(key))?.providerConversationId).toBe('conv-9'); + }); + it('findHandoverForChat returns any human/closed row for the chat, ignoring pluginId', async () => { // faq-bot keeps a bot mapping; chatwoot-adapter takes the same chat over (human). await service.upsert({ sessionId: 'sess-1', chatId: 'chat-1', pluginId: 'faq-bot', instanceId: 'i1' }, 'convA'); diff --git a/src/modules/integration/conversation-mapping.service.ts b/src/modules/integration/conversation-mapping.service.ts index 55fbf8990..1d4442940 100644 --- a/src/modules/integration/conversation-mapping.service.ts +++ b/src/modules/integration/conversation-mapping.service.ts @@ -2,6 +2,7 @@ import { Injectable } from '@nestjs/common'; import { InjectRepository } from '@nestjs/typeorm'; import { QueryDeepPartialEntity, Repository } from 'typeorm'; import { ConversationMapping, HandoverState } from './entities/conversation-mapping.entity'; +import { isUniqueViolation } from '../../common/utils/db-errors'; export interface MappingKey { sessionId: string; @@ -10,6 +11,24 @@ export interface MappingKey { instanceId: string; } +/** + * Thrown when a providerConversationId is already bound to a DIFFERENT chat for the same plugin+instance + * (the reverse unique key). Unlike a forward-key race — which converges by updating the existing row — + * this is a genuine conflict with no row to fall back to, so it surfaces instead of corrupting state. + */ +export class ConversationMappingConflict extends Error { + constructor( + readonly key: MappingKey, + readonly providerConversationId: string, + ) { + super( + `conversation mapping conflict: providerConversationId "${providerConversationId}" is already bound to ` + + `a different chat for plugin "${key.pluginId}" instance "${key.instanceId}"`, + ); + this.name = 'ConversationMappingConflict'; + } +} + @Injectable() export class ConversationMappingService { constructor(@InjectRepository(ConversationMapping, 'data') private readonly repo: Repository) {} @@ -17,13 +36,42 @@ export class ConversationMappingService { async upsert(key: MappingKey, providerConversationId: string, patch?: Partial): Promise { const existing = await this.repo.findOne({ where: key }); if (existing) { - await this.repo.update({ id: existing.id }, { + await this.updateById(existing.id, key, providerConversationId, patch); + return; + } + try { + await this.repo.save(this.repo.create({ ...key, providerConversationId, handoverState: 'bot', ...patch })); + } catch (err) { + if (!isUniqueViolation(err)) throw err; + // A concurrent writer inserted between our findOne and save (forward-key race) OR the reverse + // unique (pluginId,instanceId,providerConversationId) is bound to another chat. Re-read the + // FORWARD key: found → converge by updating it; not found → genuine reverse conflict → surface. + const raced = await this.repo.findOne({ where: key }); + if (raced) { + await this.updateById(raced.id, key, providerConversationId, patch); + return; + } + throw new ConversationMappingConflict(key, providerConversationId); + } + } + + // Update guarded against a reverse-unique collision: moving a row's providerConversationId onto a value + // already bound to another chat throws ConversationMappingConflict rather than a raw QueryFailedError. + private async updateById( + id: string, + key: MappingKey, + providerConversationId: string, + patch?: Partial, + ): Promise { + try { + await this.repo.update({ id }, { providerConversationId, ...patch, } as QueryDeepPartialEntity); - return; + } catch (err) { + if (isUniqueViolation(err)) throw new ConversationMappingConflict(key, providerConversationId); + throw err; } - await this.repo.save(this.repo.create({ ...key, providerConversationId, handoverState: 'bot', ...patch })); } get(key: MappingKey): Promise { diff --git a/src/modules/integration/entities/ingress-event.entity.ts b/src/modules/integration/entities/ingress-event.entity.ts index 5582b1975..02d28ea44 100644 --- a/src/modules/integration/entities/ingress-event.entity.ts +++ b/src/modules/integration/entities/ingress-event.entity.ts @@ -1,9 +1,11 @@ import { Column, CreateDateColumn, Entity, Index, PrimaryColumn } from 'typeorm'; import { jsonColumnType } from '../../../common/utils/column-types'; -// Persist-before-ack durable row + inbound dedup oracle. UNIQUE(instanceId, providerDeliveryId). +// Persist-before-ack durable row + inbound dedup oracle. UNIQUE(pluginId, instanceId, providerDeliveryId): +// instanceId is only unique within a plugin, so pluginId must be part of the key or two plugins sharing an +// instanceId string would drop each other's deliveries as false duplicates. @Entity('ingress_events') -@Index('UQ_ingress_events_instance_delivery', ['instanceId', 'providerDeliveryId'], { unique: true }) +@Index('UQ_ingress_events_instance_delivery', ['pluginId', 'instanceId', 'providerDeliveryId'], { unique: true }) @Index('IDX_ingress_events_createdAt', ['createdAt']) export class IngressEvent { // Host-minted uuid (crypto.randomUUID()), NOT DB-generated — the id and the jobId (= deliveryId) diff --git a/src/modules/integration/ingress-enqueue.service.spec.ts b/src/modules/integration/ingress-enqueue.service.spec.ts index ba5423b0b..31b5c8a73 100644 --- a/src/modules/integration/ingress-enqueue.service.spec.ts +++ b/src/modules/integration/ingress-enqueue.service.spec.ts @@ -26,7 +26,7 @@ describe('IngressEnqueueService', () => { (config.get as jest.Mock).mockReturnValue(true); const svc = new IngressEnqueueService(loader as PluginLoaderService, config as ConfigService, queue as never); - await svc.enqueue(data, 'd1'); + expect(await svc.enqueue(data, 'd1')).toEqual({ outcome: 'queued' }); expect(queue.add).toHaveBeenCalledWith('ingress', data, { jobId: 'd1' }); expect(loader.dispatchWebhookForInstance).not.toHaveBeenCalled(); @@ -36,7 +36,7 @@ describe('IngressEnqueueService', () => { (config.get as jest.Mock).mockReturnValue(false); const svc = new IngressEnqueueService(loader as PluginLoaderService, config as ConfigService, queue as never); - await svc.enqueue(data, 'd1'); + expect(await svc.enqueue(data, 'd1')).toEqual({ outcome: 'dispatched' }); expect(queue.add).not.toHaveBeenCalled(); expect(loader.dispatchWebhookForInstance).toHaveBeenCalledWith(data); @@ -46,17 +46,17 @@ describe('IngressEnqueueService', () => { (config.get as jest.Mock).mockReturnValue(true); const svc = new IngressEnqueueService(loader as PluginLoaderService, config as ConfigService, undefined); - await svc.enqueue(data, 'd1'); + expect(await svc.enqueue(data, 'd1')).toEqual({ outcome: 'dispatched' }); expect(loader.dispatchWebhookForInstance).toHaveBeenCalledWith(data); }); - it('swallows an inline dispatch error and logs it rather than throwing (row already persisted for redrive)', async () => { + it('swallows an inline dispatch error and returns outcome "failed" rather than throwing (row stays redrivable)', async () => { (loader.dispatchWebhookForInstance as jest.Mock).mockRejectedValue(new Error('boom')); (config.get as jest.Mock).mockReturnValue(false); const svc = new IngressEnqueueService(loader as PluginLoaderService, config as ConfigService, undefined); - await expect(svc.enqueue(data, 'd1')).resolves.toBeUndefined(); + expect(await svc.enqueue(data, 'd1')).toEqual({ outcome: 'failed' }); }); it('falls back to inline dispatch (never throws) when queue.add() fails, e.g. Redis unreachable', async () => { @@ -66,7 +66,7 @@ describe('IngressEnqueueService', () => { queue.add.mockRejectedValue(new Error('Redis connection is closed')); const svc = new IngressEnqueueService(loader as PluginLoaderService, config as ConfigService, queue as never); - await expect(svc.enqueue(data, 'd1')).resolves.toBeUndefined(); + expect(await svc.enqueue(data, 'd1')).toEqual({ outcome: 'dispatched' }); expect(queue.add).toHaveBeenCalledWith('ingress', data, { jobId: 'd1' }); expect(loader.dispatchWebhookForInstance).toHaveBeenCalledWith(data); }); diff --git a/src/modules/integration/ingress-enqueue.service.ts b/src/modules/integration/ingress-enqueue.service.ts index a74dcc198..392a2a4cd 100644 --- a/src/modules/integration/ingress-enqueue.service.ts +++ b/src/modules/integration/ingress-enqueue.service.ts @@ -7,6 +7,13 @@ import { IngressJobData } from '../queue/processors/ingress.processor'; import { QUEUE_NAMES } from '../queue/queue-names'; import { createLogger } from '../../common/services/logger.service'; +/** + * Outcome of an enqueue attempt. 'queued' = handed to BullMQ; 'dispatched' = delivered inline; 'failed' + * = inline dispatch threw and was swallowed (the row stays durable for a redrive). enqueue() never + * throws, so callers use the outcome (not exceptions) to decide durability follow-up (e.g. redrive). + */ +export type EnqueueOutcome = { outcome: 'queued' | 'dispatched' | 'failed' }; + /** * Shared queue-or-inline enqueue for inbound ingress jobs. Extracted out of IngressService's DI * factory (integration.module.ts) so RedriveService can reuse the exact same behavior when replaying @@ -24,7 +31,7 @@ export class IngressEnqueueService { @Optional() @InjectQueue(QUEUE_NAMES.INGRESS) private readonly ingressQueue?: Queue, ) {} - async enqueue(data: IngressJobData, jobId: string): Promise { + async enqueue(data: IngressJobData, jobId: string): Promise { const queueEnabled = this.config.get('queue.enabled', false); const useQueue = queueEnabled && !!this.ingressQueue; @@ -32,7 +39,7 @@ export class IngressEnqueueService { try { // jobId = deliveryId gives BullMQ exactly-once enqueue semantics. await this.ingressQueue.add('ingress', data, { jobId }); - return; + return { outcome: 'queued' }; } catch (err) { // Redis unreachable (enableOfflineQueue:false makes add() reject) — fall through to inline // dispatch. Without this, the already-persisted event would be lost forever: the throw would @@ -55,10 +62,12 @@ export class IngressEnqueueService { // (persist-before-dispatch still holds), mirroring the webhook direct-delivery fallback. try { await this.loader.dispatchWebhookForInstance(data); + return { outcome: 'dispatched' }; } catch (err) { // A duplicate delivery already 200s before this point, so a failure here is a real // dispatch error. Log and swallow: the row is durably persisted for a later redrive, // and the provider still gets its 202 (at-least-once, like the webhook fallback). + // The 'failed' outcome lets RedriveService avoid marking a DLQ row handled on a silent drop. this.logger.error('Inline ingress dispatch failed', err instanceof Error ? err.message : String(err), { pluginId: data.pluginId, instanceId: data.instanceId, @@ -66,6 +75,7 @@ export class IngressEnqueueService { deliveryId: data.deliveryId, action: 'ingress_inline_dispatch_failed', }); + return { outcome: 'failed' }; } } } diff --git a/src/modules/integration/ingress-event.service.spec.ts b/src/modules/integration/ingress-event.service.spec.ts index b3de5163d..c745bef61 100644 --- a/src/modules/integration/ingress-event.service.spec.ts +++ b/src/modules/integration/ingress-event.service.spec.ts @@ -2,6 +2,7 @@ import { DataSource } from 'typeorm'; import { IngressEvent } from './entities/ingress-event.entity'; import { IngressEventService } from './ingress-event.service'; import { AddIntegrationFabric1781900000000 } from '../../database/migrations/1781900000000-AddIntegrationFabric'; +import { WidenIngressDedupKey1782100000000 } from '../../database/migrations/1782100000000-WidenIngressDedupKey'; describe('IngressEventService.recordOrSkip', () => { let ds: DataSource; @@ -11,6 +12,7 @@ describe('IngressEventService.recordOrSkip', () => { await ds.initialize(); const runner = ds.createQueryRunner(); await new AddIntegrationFabric1781900000000().up(runner); + await new WidenIngressDedupKey1782100000000().up(runner); await runner.release(); service = new IngressEventService(ds.getRepository(IngressEvent)); }); @@ -36,4 +38,10 @@ describe('IngressEventService.recordOrSkip', () => { expect(await service.recordOrSkip(row())).toBe(true); expect(await service.recordOrSkip({ ...row(), instanceId: 'inst2' })).toBe(true); }); + + it('treats the same instance+delivery id under a different plugin as new (dedup key includes pluginId)', async () => { + // instanceId is only unique within a plugin; two plugins sharing the string must not collide. + expect(await service.recordOrSkip(row())).toBe(true); + expect(await service.recordOrSkip({ ...row(), pluginId: 'other-plug' })).toBe(true); + }); }); diff --git a/src/modules/integration/ingress-event.service.ts b/src/modules/integration/ingress-event.service.ts index 4ef09fed2..73fc64074 100644 --- a/src/modules/integration/ingress-event.service.ts +++ b/src/modules/integration/ingress-event.service.ts @@ -1,18 +1,9 @@ import { randomUUID } from 'node:crypto'; import { Injectable } from '@nestjs/common'; import { InjectRepository } from '@nestjs/typeorm'; -import { QueryFailedError, Repository } from 'typeorm'; +import { Repository } from 'typeorm'; import { IngressEvent } from './entities/ingress-event.entity'; - -// Cross-dialect unique-violation check by driver code/message — the two dialects we ship -// (sqlite dev, postgres prod). Add another branch if a third driver is ever supported. -export function isUniqueViolation(err: unknown): boolean { - if (!(err instanceof QueryFailedError)) return false; - const driver = err.driverError as { code?: string; message?: string } | undefined; - const code = driver?.code ?? ''; - const message = driver?.message ?? err.message ?? ''; - return code === '23505' /* postgres */ || /UNIQUE constraint failed|SQLITE_CONSTRAINT/i.test(message); -} +import { isUniqueViolation } from '../../common/utils/db-errors'; export interface IngressEventInput { instanceId: string; diff --git a/src/modules/integration/ingress.service.spec.ts b/src/modules/integration/ingress.service.spec.ts index 252c21576..16f3367ef 100644 --- a/src/modules/integration/ingress.service.spec.ts +++ b/src/modules/integration/ingress.service.spec.ts @@ -193,7 +193,9 @@ describe('IngressService.handle', () => { expect((await svc.handle(req)).status).toBe(404); }); - it('falls back to a generated delivery id when the dedup header is absent', async () => { + it('derives a DETERMINISTIC delivery id from the body when the dedup header is absent', async () => { + // A random UUID here would defeat both persist-dedup and BullMQ jobId idempotency, so a provider + // retry of the same body must produce the SAME id, and a different body a DIFFERENT id. const d = deps(); const svc = new IngressService(d); const res = await svc.handle({ ...req, headers: {} }); @@ -202,6 +204,16 @@ describe('IngressService.handle', () => { expect(typeof jobId).toBe('string'); expect(jobId.length).toBeGreaterThan(0); expect(jobData.deliveryId).toBe(jobId); + + // same body → same id (retry dedups) + const d2 = deps(); + await new IngressService(d2).handle({ ...req, headers: {} }); + expect((d2.enqueue.mock.calls[0] as [unknown, string])[1]).toBe(jobId); + + // different body → different id + const d3 = deps(); + await new IngressService(d3).handle({ ...req, headers: {}, rawBody: '{"a":1}' }); + expect((d3.enqueue.mock.calls[0] as [unknown, string])[1]).not.toBe(jobId); }); }); diff --git a/src/modules/integration/ingress.service.ts b/src/modules/integration/ingress.service.ts index 780608bbd..45df7fa66 100644 --- a/src/modules/integration/ingress.service.ts +++ b/src/modules/integration/ingress.service.ts @@ -1,4 +1,4 @@ -import { randomUUID } from 'node:crypto'; +import { createHash } from 'node:crypto'; import { verifyIngressSignature } from './ingress-signature'; import { PluginIngressRoute } from '../../core/plugins/plugin.interfaces'; import { IngressJobData } from '../queue/processors/ingress.processor'; @@ -40,7 +40,9 @@ export interface IngressDeps { sessionId: string | null; }): Promise; }; - enqueue: (data: IngressJobData, jobId: string) => Promise; + // Returns an enqueue outcome (queued/dispatched/failed); handle() ignores it — only durability + // follow-up paths like redrive act on it. Typed as unknown here to keep this pure module decoupled. + enqueue: (data: IngressJobData, jobId: string) => Promise; now: () => number; } @@ -79,7 +81,7 @@ export class IngressService { if (!verdict.ok) return { status: 401, body: verdict.reason ?? 'signature verification failed' }; const dedupHeader = (route.dedupHeader ?? route.signature.dedupHeader ?? 'x-delivery').toLowerCase(); - const deliveryId = req.headers[dedupHeader] ?? randomUUID(); + const deliveryId = req.headers[dedupHeader] ?? deriveDeliveryId(req); const payload = { headers: req.headers, query: req.query, body: req.rawBody, rawBody: req.rawBody }; const isNew = await this.deps.events.recordOrSkip({ instanceId: req.instanceId, @@ -110,6 +112,16 @@ export class IngressService { } } +/** + * Derives a DETERMINISTIC delivery id when the provider sends no dedup header, so a provider retry of + * the same delivery dedups instead of being treated as new. A random UUID would silently disable both + * the persist-dedup and BullMQ's jobId idempotency, causing duplicate downstream WhatsApp sends. Keyed + * on pluginId + instanceId + route + rawBody ONLY — never a server timestamp, which would defeat dedup. + */ +function deriveDeliveryId(req: IngressRequest): string { + return createHash('sha256').update([req.pluginId, req.instanceId, req.route, req.rawBody].join('\0')).digest('hex'); +} + /** * Extracts the provider conversation id from a declared header or a JSON pointer into the body. * Returns undefined when no pointer is declared or extraction fails — the P1 lock then keys per diff --git a/src/modules/integration/redrive.service.spec.ts b/src/modules/integration/redrive.service.spec.ts index 267aaaabd..9f5d0ef15 100644 --- a/src/modules/integration/redrive.service.spec.ts +++ b/src/modules/integration/redrive.service.spec.ts @@ -9,7 +9,7 @@ describe('RedriveService', () => { beforeEach(() => { repo = { find: jest.fn(), update: jest.fn().mockResolvedValue(undefined) }; - ingressEnqueue = { enqueue: jest.fn().mockResolvedValue(undefined) }; + ingressEnqueue = { enqueue: jest.fn().mockResolvedValue({ outcome: 'queued' }) }; }); function makeSvc(): RedriveService { @@ -67,6 +67,32 @@ describe('RedriveService', () => { expect(repo.update).toHaveBeenCalledWith({ id: 'f2' }, { redriven: true }); }); + it('leaves the row redrivable (does not mark redriven) when the inline re-dispatch is swallowed as failed', async () => { + // A queue-disabled fallback that silently fails must NOT retire the DLQ row, or the event is lost + // permanently with no way to redrive it again. + const rows = [ + { + id: 'f9', + direction: 'inbound', + pluginId: 'p', + instanceId: 'i', + deliveryId: 'd9', + payload: { route: 'chatwoot', ingress: { headers: {}, query: {}, body: '{}', rawBody: '{}' } }, + sessionId: 's', + redriven: false, + }, + ]; + (repo.find as jest.Mock).mockResolvedValue(rows); + (ingressEnqueue.enqueue as jest.Mock).mockResolvedValue({ outcome: 'failed' }); + const svc = makeSvc(); + + const res = await svc.redriveInstance('p', 'i'); + + expect(res.redriven).toBe(0); + expect(ingressEnqueue.enqueue).toHaveBeenCalledTimes(1); + expect(repo.update).not.toHaveBeenCalled(); + }); + it('queries only non-redriven inbound rows for the instance and returns 0 when none are found', async () => { (repo.find as jest.Mock).mockResolvedValue([]); const svc = makeSvc(); diff --git a/src/modules/integration/redrive.service.ts b/src/modules/integration/redrive.service.ts index f31a8c5ed..911166732 100644 --- a/src/modules/integration/redrive.service.ts +++ b/src/modules/integration/redrive.service.ts @@ -29,7 +29,7 @@ export class RedriveService { const stored = (row.payload ?? {}) as StoredDlqPayload; // Re-mint a jobId so BullMQ accepts the replay even if the original jobId lingers. const jobId = `redrive:${row.id}`; - await this.ingressEnqueue.enqueue( + const { outcome } = await this.ingressEnqueue.enqueue( { pluginId, instanceId, @@ -41,8 +41,13 @@ export class RedriveService { }, jobId, ); - await this.repo.update({ id: row.id }, { redriven: true }); - redriven++; + // Only retire the DLQ row once the replay was actually accepted (queued) or delivered. A swallowed + // inline-dispatch failure ('failed') leaves the row redriven=false so it stays redrivable, instead + // of silently marking it handled and permanently losing the event. + if (outcome !== 'failed') { + await this.repo.update({ id: row.id }, { redriven: true }); + redriven++; + } } return { redriven }; }