diff --git a/src/infra/outbound/deliver.test.ts b/src/infra/outbound/deliver.test.ts index 1c700990363f..b866b9038fc3 100644 --- a/src/infra/outbound/deliver.test.ts +++ b/src/infra/outbound/deliver.test.ts @@ -1046,14 +1046,6 @@ describe("deliverOutboundPayloads", () => { }); it("marks queued delivery as unknown-after-send (not failed) when a later payload fails after an earlier one succeeded", async () => { - // Regression: required-mode batch send where an earlier payload succeeded - // (results.length > 0, OutboundDeliveryError.sentBeforeError === true) but a - // later payload throws. Previously the wrapper catch called failDelivery, - // leaving the entry in `send_attempt_started` so reconnect drain later - // replayed it as "not yet sent" — producing duplicate messages when the - // adapter's unknown-send reconciliation misreported `not_sent`. The fix - // advances to `unknown_after_send` so drain routes the entry through - // reconcileUnknownQueuedDelivery instead of blind replay. const sendMatrix = vi .fn() .mockResolvedValueOnce({ messageId: "m1" }) @@ -1072,16 +1064,11 @@ describe("deliverOutboundPayloads", () => { expect(sendMatrix).toHaveBeenCalledTimes(2); expect(queueMocks.markDeliveryPlatformOutcomeUnknown).toHaveBeenCalledWith("mock-queue-id"); - // Must NOT failDelivery — that would leave send_attempt_started for drain to replay. expect(queueMocks.failDelivery).not.toHaveBeenCalled(); - // Must NOT ack — the batch did not fully succeed; reconcile must confirm the - // partial send rather than silently dropping the queue entry. expect(queueMocks.ackDelivery).not.toHaveBeenCalled(); }); it("still calls failDelivery when a payload fails before any send succeeded", async () => { - // No send evidence (sentBeforeError === false): failDelivery is correct — - // nothing reached the channel, so leaving the entry for retry is safe. const sendMatrix = vi.fn().mockRejectedValueOnce(new Error("first payload send failed")); await expect( diff --git a/src/infra/outbound/deliver.ts b/src/infra/outbound/deliver.ts index 397f9eff27a3..68ae52060a74 100644 --- a/src/infra/outbound/deliver.ts +++ b/src/infra/outbound/deliver.ts @@ -1412,15 +1412,6 @@ async function deliverOutboundPayloadsWithQueueCleanup( if (isDeliveryAbortError(err)) { await ackDelivery(queueId).catch(() => {}); } else if (!platformResultsReturned) { - // If the platform send already started and the error carries partial - // send evidence (OutboundDeliveryError with sentBeforeError), the - // message may already have reached the channel. Calling failDelivery - // here leaves the entry in `send_attempt_started`, which reconnect - // drain later replays as "not yet sent" — producing duplicate - // messages when the adapter's unknown-send reconciliation misreports - // `not_sent`. Advance to `unknown_after_send` instead so drain routes - // the entry through reconcileUnknownQueuedDelivery (query the adapter - // for actual send state) rather than blind replay. const sendEvidence = platformSendStarted && err instanceof OutboundDeliveryError && err.sentBeforeError; if (sendEvidence) { diff --git a/src/infra/outbound/delivery-queue-recovery.ts b/src/infra/outbound/delivery-queue-recovery.ts index 743f9064cddc..5a41b3bdc5e9 100644 --- a/src/infra/outbound/delivery-queue-recovery.ts +++ b/src/infra/outbound/delivery-queue-recovery.ts @@ -389,15 +389,19 @@ async function drainQueuedEntry(opts: { return "failed"; } } - if (reconciliation?.status === "not_sent") { + const reconciliationProvedPreSendFailure = + reconciliation?.status === "not_sent" && entry.recoveryState === "send_attempt_started"; + if (reconciliationProvedPreSendFailure) { opts.log.info( `Delivery entry ${entry.id} reconciled ${entry.recoveryState} as not sent; replaying`, ); } else { - const errMsg = - reconciliation?.status === "unresolved" && reconciliation.error - ? `delivery state is ${entry.recoveryState} and reconciliation is unresolved: ${reconciliation.error}` - : `delivery state is ${entry.recoveryState}; refusing blind replay without adapter reconciliation`; + let errMsg = `delivery state is ${entry.recoveryState}; refusing blind replay without adapter reconciliation`; + if (reconciliation?.status === "not_sent") { + errMsg = `delivery state is ${entry.recoveryState}; refusing full replay after post-send evidence`; + } else if (reconciliation?.status === "unresolved" && reconciliation.error) { + errMsg = `delivery state is ${entry.recoveryState} and reconciliation is unresolved: ${reconciliation.error}`; + } opts.log.warn(`Delivery entry ${entry.id} ${errMsg}`); opts.onFailed?.(entry, errMsg); if (reconciliation?.status === "unresolved" && reconciliation.retryable === true) { diff --git a/src/infra/outbound/delivery-queue.recovery.test.ts b/src/infra/outbound/delivery-queue.recovery.test.ts index 2425d728851e..eecc8bf8c8e1 100644 --- a/src/infra/outbound/delivery-queue.recovery.test.ts +++ b/src/infra/outbound/delivery-queue.recovery.test.ts @@ -328,7 +328,7 @@ describe("delivery-queue recovery", () => { expect(await loadPendingDeliveries(tmpDir())).toHaveLength(0); }); - it("replays unknown-after-send entries only after adapter proves they were not sent", async () => { + it("moves unknown-after-send entries to failed when adapter reports not sent", async () => { const id = await enqueueDelivery( { channel: "demo-channel-a", to: "+1", payloads: [{ text: "not sent" }] }, tmpDir(), @@ -346,24 +346,19 @@ describe("delivery-queue recovery", () => { }); const deliver = vi.fn().mockResolvedValue([]); - const { result } = await runRecovery({ deliver }); + const log = createRecoveryLog(); + const { result } = await runRecovery({ deliver, log }); - expect(deliver).toHaveBeenCalledTimes(1); - const deliverInput = mockCallArg(deliver) as { - channel?: string; - to?: string; - skipQueue?: boolean; - }; - expect(deliverInput.channel).toBe("demo-channel-a"); - expect(deliverInput.to).toBe("+1"); - expect(deliverInput.skipQueue).toBe(true); + expect(deliver).not.toHaveBeenCalled(); expect(result).toEqual({ - recovered: 1, - failed: 0, + recovered: 0, + failed: 1, skippedMaxRetries: 0, deferredBackoff: 0, }); expect(await loadPendingDeliveries(tmpDir())).toHaveLength(0); + expect(readOutboundQueueStatus(tmpDir(), id)).toBe("failed"); + expectMockMessageContaining(log.warn, "refusing full replay after post-send evidence"); }); it("keeps retryable unresolved unknown-after-send entries on the queue without replaying", async () => {