From 27513d21ca55445216b063dd5db85dd7bc41cf8f Mon Sep 17 00:00:00 2001 From: rosenlo Date: Sun, 21 Jun 2026 11:30:44 +0800 Subject: [PATCH] fix(outbound): advance queue entry to unknown_after_send on mid-batch failure with send evidence When a required-mode batch send fails mid-batch after an earlier payload already succeeded, the wrapper catch in deliverOutboundPayloadsWithQueueCleanup called failDelivery. failDelivery only bumps retryCount/lastError; it does not advance recoveryState, so the entry stayed in send_attempt_started (set earlier by markDeliveryPlatformSendAttemptStarted via onPlatformSendStart). On the next Telegram reconnect, drainQueuedEntry sees send_attempt_started and calls reconcileUnknownQueuedDelivery. When adapter reconciliation misreports not_sent (the message was actually sent, per the outbound send ok / messageId evidence), the entry is replayed and the user receives a duplicate. Fix: when the error carries send evidence (OutboundDeliveryError with sentBeforeError === true and platformSendStarted === true), call markQueuedPlatformOutcomeUnknown instead of failDelivery. This advances the entry to unknown_after_send, which drain already routes through reconcileUnknownQueuedDelivery, preserving the entry for adapter reconciliation rather than leaving it in send_attempt_started for replay. When there is no send evidence (sentBeforeError === false), failDelivery remains correct: nothing reached the channel, so retrying is safe. This is a third duplicate path distinct from #89812 (mirror best-effort) and #92274 (subagent-announce-delivery retry); it is the outbound/deliver wrapper catch, which neither prior fix covers. Tests: - regression: two payloads, first succeeds, second throws; asserts markDeliveryPlatformOutcomeUnknown called, failDelivery/ackDelivery not. - guard: no send evidence; failDelivery still called. (cherry picked from commit 71422a9a5a5829d8e5bcc6bfe5b28b2a70f7cf91) --- src/infra/outbound/deliver.test.ts | 58 ++++++++++++++++++++++++++++++ src/infra/outbound/deliver.ts | 39 ++++++++++++++++---- 2 files changed, 91 insertions(+), 6 deletions(-) diff --git a/src/infra/outbound/deliver.test.ts b/src/infra/outbound/deliver.test.ts index 4d58a11c24da..9f59ce12c5ab 100644 --- a/src/infra/outbound/deliver.test.ts +++ b/src/infra/outbound/deliver.test.ts @@ -1045,6 +1045,64 @@ describe("deliverOutboundPayloads", () => { expect(queueMocks.ackDelivery).not.toHaveBeenCalled(); }); + 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" }) + .mockRejectedValueOnce(new Error("second payload send failed")); + + await expect( + deliverOutboundPayloads({ + cfg: {}, + channel: "matrix", + to: "!room:example", + payloads: [{ text: "first" }, { text: "second" }], + deps: { matrix: sendMatrix }, + queuePolicy: "required", + }), + ).rejects.toThrow("second payload send failed"); + + 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( + deliverOutboundPayloads({ + cfg: {}, + channel: "matrix", + to: "!room:example", + payloads: [{ text: "first" }], + deps: { matrix: sendMatrix }, + queuePolicy: "required", + }), + ).rejects.toThrow("first payload send failed"); + + expect(queueMocks.failDelivery).toHaveBeenCalledWith( + "mock-queue-id", + expect.stringContaining("first payload send failed"), + ); + expect(queueMocks.markDeliveryPlatformOutcomeUnknown).not.toHaveBeenCalled(); + expect(queueMocks.ackDelivery).not.toHaveBeenCalled(); + }); + it("fails required delivery when the post-send unknown marker cannot be written", async () => { queueMocks.markDeliveryPlatformOutcomeUnknown.mockRejectedValueOnce( new Error("unknown marker offline"), diff --git a/src/infra/outbound/deliver.ts b/src/infra/outbound/deliver.ts index 1c1b9499c715..5e912098a526 100644 --- a/src/infra/outbound/deliver.ts +++ b/src/infra/outbound/deliver.ts @@ -1326,9 +1326,9 @@ async function deliverOutboundPayloadsWithQueueCleanup( }; const queuePolicy = params.queuePolicy ?? "best_effort"; let platformResultsReturned = false; + let platformSendStarted = false; try { - let platformSendStarted = false; const results = await deliverOutboundPayloadsCore({ ...wrappedParams, ...(queueId @@ -1390,11 +1390,38 @@ async function deliverOutboundPayloadsWithQueueCleanup( if (isDeliveryAbortError(err)) { await ackDelivery(queueId).catch(() => {}); } else if (!platformResultsReturned) { - await failDelivery(queueId, formatErrorMessage(err)).catch((failErr: unknown) => { - log.warn( - `failed to mark queued delivery ${queueId} as failed: ${formatErrorMessage(failErr)}`, - ); - }); + // 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) { + await markQueuedPlatformOutcomeUnknown({ + queueId, + queuePolicy, + }).catch((markErr: unknown) => { + log.warn( + `failed to mark queued delivery ${queueId} as platform-outcome-unknown after mid-send error; falling back to fail: ${formatErrorMessage(markErr)}`, + ); + return failDelivery(queueId, formatErrorMessage(err)).catch((failErr: unknown) => { + log.warn( + `failed to mark queued delivery ${queueId} as failed: ${formatErrorMessage(failErr)}`, + ); + }); + }); + } else { + await failDelivery(queueId, formatErrorMessage(err)).catch((failErr: unknown) => { + log.warn( + `failed to mark queued delivery ${queueId} as failed: ${formatErrorMessage(failErr)}`, + ); + }); + } } } throw err;