diff --git a/packages/adapters/src/messaging-delivery.test.ts b/packages/adapters/src/messaging-delivery.test.ts index 4b2e366..1a0388d 100644 --- a/packages/adapters/src/messaging-delivery.test.ts +++ b/packages/adapters/src/messaging-delivery.test.ts @@ -197,7 +197,7 @@ describe("deliverMessagingOutbound", () => { await deliverMessagingOutbound(deps, { runId: "run-1" }, context); expect(deps.sendToThread).toHaveBeenCalledWith( - { threadId: "sendblue:dm-1", body: "Hello from your bot" }, + { threadId: "sendblue:dm-1", body: "Hello from your bot", idempotencyKey: "msg:m-1" }, context, ); expect(deps.rows).toEqual([ @@ -237,12 +237,20 @@ describe("deliverMessagingOutbound", () => { }); expect(deps.sendToThread).toHaveBeenNthCalledWith( 1, - { threadId: "sendblue:opened:+15551234567", body: "first" }, + { + threadId: "sendblue:opened:+15551234567", + body: "first", + idempotencyKey: "msg:m-1", + }, context, ); expect(deps.sendToThread).toHaveBeenNthCalledWith( 2, - { threadId: "sendblue:opened:+15551234567", body: "second" }, + { + threadId: "sendblue:opened:+15551234567", + body: "second", + idempotencyKey: "msg:m-2", + }, context, ); }); @@ -271,7 +279,7 @@ describe("deliverMessagingOutbound", () => { const deps = createDeps({ run: { ...messagingRun, trigger: "bot_message" } }); await deliverMessagingOutbound(deps, { runId: "run-1" }, context); expect(deps.sendToThread).toHaveBeenCalledWith( - { threadId: "sendblue:dm-1", body: "Hello from your bot" }, + { threadId: "sendblue:dm-1", body: "Hello from your bot", idempotencyKey: "msg:m-1" }, context, ); }); @@ -305,22 +313,16 @@ describe("deliverMessagingOutbound", () => { ]); }); - it("returns a transient send failure to pending and schedules a delayed retry", async () => { + it("does not re-queue a mirrored DM when the provider send fails", async () => { const deps = createDeps({ sendError: new Error("messaging provider 500") }); await deliverMessagingOutbound(deps, { runId: "run-1" }, context); - expect(deps.rows).toEqual([ - expect.objectContaining({ - kind: "dm", - status: "pending", - attempts: 1, - nextAttemptAt: expect.any(Date), - }), - ]); + // At-most-once: keep the claim rather than risk a duplicate DM. + expect(deps.rows[0]).toEqual(expect.objectContaining({ kind: "dm", status: "sent" })); expect(deps.rows[0]!.providerHandle ?? null).toBeNull(); - expect(deps.jobs.enqueue).toHaveBeenCalledWith( - expect.objectContaining({ name: "messaging.deliver", availableAt: expect.any(Date) }), - ); + expect(deps.rows[0]!.attempts ?? 0).toBe(0); + expect(deps.jobs.enqueue).not.toHaveBeenCalled(); + expect(deps.prisma.messagingIdentity.update).not.toHaveBeenCalled(); }); it("skips pending rows whose nextAttemptAt is still in the future", async () => { @@ -364,7 +366,11 @@ describe("deliverMessagingOutbound", () => { await deliverMessagingOutbound(deps, {}, context); expect(deps.sendToThread).toHaveBeenCalledWith( - { threadId: "sendblue:dm-1", body: "pending earlier" }, + { + threadId: "sendblue:dm-1", + body: "pending earlier", + idempotencyKey: "invite:ch-1:mi-1", + }, context, ); expect(deps.rows[0]).toEqual(expect.objectContaining({ status: "sent" })); @@ -519,6 +525,28 @@ describe("deliverMessagingOutbound", () => { expect(deps.prisma.messagingIdentity.update).not.toHaveBeenCalled(); }); + it("does not re-queue a channel invite when the provider send fails", async () => { + const deps = createDeps({ + run: null, + sendError: new Error("provider down"), + outboundRows: [ + { + id: "out-invite", + idempotencyKey: "invite:ch-1:mi-1", + kind: "dm", + identityId: "mi-1", + body: "join the group", + status: "pending", + providerHandle: null, + }, + ], + }); + await deliverMessagingOutbound(deps, {}, context); + + expect(deps.rows[0]).toEqual(expect.objectContaining({ status: "sent" })); + expect(deps.jobs.enqueue).not.toHaveBeenCalled(); + }); + it("does not re-queue a connect invite after a delivery transaction timeout", async () => { const deps = createDeps({ run: null, @@ -840,7 +868,11 @@ describe("deliverMessagingOutbound channel runs", () => { }); await deliverMessagingOutbound(deps, { runId: "run-1" }, context); expect(deps.sendToThread).toHaveBeenCalledWith( - { threadId: "sendblue:grp-1", body: "Alice's agent: Public update" }, + { + threadId: "sendblue:grp-1", + body: "Alice's agent: Public update", + idempotencyKey: "msg:reply-1", + }, context, ); }); @@ -850,7 +882,11 @@ describe("deliverMessagingOutbound channel runs", () => { await deliverMessagingOutbound(deps, { runId: "run-1" }, context); expect(deps.sendToThread).toHaveBeenCalledWith( - { threadId: "sendblue:grp-1", body: "Alice's agent: found it" }, + { + threadId: "sendblue:grp-1", + body: "Alice's agent: found it", + idempotencyKey: "msg:m-1", + }, context, ); expect(deps.rows).toEqual([ @@ -949,6 +985,16 @@ describe("deliverMessagingOutbound channel runs", () => { ); }); + it("does not re-queue a group post when the provider send fails", async () => { + const deps = createChannelDeps(); + deps.sendToThread.mockRejectedValueOnce(new Error("provider down")); + await deliverMessagingOutbound(deps, { runId: "run-1" }, context); + + expect(deps.rows[0]).toEqual(expect.objectContaining({ kind: "group", status: "sent" })); + expect(deps.rows[0]!.providerHandle ?? null).toBeNull(); + expect(deps.enqueue).not.toHaveBeenCalled(); + }); + it("does not wake anyone when the hop budget is exhausted", async () => { const deps = createChannelDeps({ text: "@Helper again?", sourceHop: 6 }); await deliverMessagingOutbound(deps, { runId: "run-1" }, context); @@ -962,26 +1008,61 @@ describe("deliverMessagingOutbound channel runs", () => { }); }); -describe("deliverMessagingOutbound transient failure retry", () => { - it("marks the row failed only once the attempt budget is exhausted", async () => { +describe("deliverMessagingOutbound pre-send failure retry", () => { + it("returns a thread-resolution failure to pending and schedules a delayed retry", async () => { const deps = createDeps({ run: null, - sendError: new Error("messaging provider 500"), + identity: { dmThreadId: null }, outboundRows: [ { id: "out-9", - idempotencyKey: "msg:m-9", + idempotencyKey: "invite:ch-1:mi-1", kind: "dm", identityId: "mi-1", - body: "fourth failure", + body: "needs thread lookup", + status: "pending", + providerHandle: null, + attempts: 0, + }, + ], + }); + deps.openDirectThread.mockRejectedValueOnce(new Error("provider lookup 503")); + await deliverMessagingOutbound(deps, {}, context); + + expect(deps.sendToThread).not.toHaveBeenCalled(); + expect(deps.rows).toEqual([ + expect.objectContaining({ + status: "pending", + attempts: 1, + nextAttemptAt: expect.any(Date), + }), + ]); + expect(deps.jobs.enqueue).toHaveBeenCalledWith( + expect.objectContaining({ name: "messaging.deliver", availableAt: expect.any(Date) }), + ); + }); + + it("marks the row failed only once the pre-send attempt budget is exhausted", async () => { + const deps = createDeps({ + run: null, + identity: { dmThreadId: null }, + outboundRows: [ + { + id: "out-9", + idempotencyKey: "invite:ch-1:mi-1", + kind: "dm", + identityId: "mi-1", + body: "last lookup try", status: "pending", providerHandle: null, attempts: MESSAGING_OUTBOUND_MAX_ATTEMPTS - 1, }, ], }); + deps.openDirectThread.mockRejectedValue(new Error("provider lookup 503")); await deliverMessagingOutbound(deps, {}, context); + expect(deps.sendToThread).not.toHaveBeenCalled(); expect(deps.rows).toEqual([ expect.objectContaining({ status: "failed", attempts: MESSAGING_OUTBOUND_MAX_ATTEMPTS }), ]); @@ -991,8 +1072,8 @@ describe("deliverMessagingOutbound transient failure retry", () => { describe("deliverMessagingOutbound retry enqueue failure", () => { it("propagates the enqueue failure so the job queue retries the drain", async () => { - const deps = createDeps({}); - deps.sendToThread.mockRejectedValueOnce(new Error("messaging provider 500")); + const deps = createDeps({ identity: { dmThreadId: null } }); + deps.openDirectThread.mockRejectedValueOnce(new Error("provider lookup 503")); deps.jobs.enqueue.mockRejectedValueOnce(new Error("queue down")); // A swallowed enqueue failure would strand the row in pending forever: @@ -1000,6 +1081,7 @@ describe("deliverMessagingOutbound retry enqueue failure", () => { await expect(deliverMessagingOutbound(deps, { runId: "run-1" }, context)).rejects.toThrow( "queue down", ); + expect(deps.sendToThread).not.toHaveBeenCalled(); expect(deps.rows).toEqual([ expect.objectContaining({ kind: "dm", @@ -1011,8 +1093,9 @@ describe("deliverMessagingOutbound retry enqueue failure", () => { // The delayed messaging.deliver job fires at nextAttemptAt: make the row due. deps.rows[0]!.nextAttemptAt = new Date(Date.now() - 1); + deps.openDirectThread.mockResolvedValueOnce("sendblue:opened:+15551234567"); await deliverMessagingOutbound(deps, { runId: "run-1" }, context); - expect(deps.sendToThread).toHaveBeenCalledTimes(2); + expect(deps.sendToThread).toHaveBeenCalledTimes(1); expect(deps.rows).toEqual([ expect.objectContaining({ kind: "dm", status: "sent", providerHandle: "handle-out-1" }), ]); diff --git a/packages/adapters/src/messaging-delivery.ts b/packages/adapters/src/messaging-delivery.ts index 3036335..5215d92 100644 --- a/packages/adapters/src/messaging-delivery.ts +++ b/packages/adapters/src/messaging-delivery.ts @@ -277,6 +277,53 @@ function connectInvitePair( * restore pending and duplicate the YES/NO DM. A lost invite is fine; * reconnect starts a fresh cycle. */ +/** + * At-most-once provider send: any error after the call (or a lost response) + * keeps the outer claim as `sent` so drain does not restore pending and + * duplicate the message. A lost send is acceptable; the outbox row stays + * held without a providerHandle until status webhooks or manual cleanup. + */ +async function sendOutboundAtMostOnce( + deps: MessagingDeliveryDeps, + row: { id: string; idempotencyKey: string; body: string }, + threadId: string, + context: AdapterContext, +): Promise<"delivered" | "held"> { + try { + const sent = await deps.messaging.sendToThread( + { threadId, body: row.body, idempotencyKey: row.idempotencyKey }, + context, + ); + await deps.prisma.messagingOutbound.updateMany({ + where: { id: row.id }, + data: { providerHandle: sent.handle }, + }); + return "delivered"; + } catch { + return "held"; + } +} + +async function revertOutboundToPendingWithRetry( + deps: MessagingDeliveryDeps, + row: { id: string; attempts?: number | null }, +): Promise { + const attempts = (row.attempts ?? 0) + 1; + const exhausted = attempts >= MESSAGING_OUTBOUND_MAX_ATTEMPTS; + const retryAt = exhausted ? null : new Date(Date.now() + messagingOutboundRetryDelayMs(attempts)); + await deps.prisma.messagingOutbound.updateMany({ + where: { id: row.id }, + data: { + attempts, + status: exhausted ? "failed" : "pending", + nextAttemptAt: retryAt, + }, + }); + if (!exhausted && retryAt) { + await deps.jobs.enqueue(messagingDeliverJob(undefined, retryAt)); + } +} + async function sendConnectInvite( deps: MessagingDeliveryDeps, row: { id: string; body: string }, @@ -343,99 +390,81 @@ async function drain(deps: MessagingDeliveryDeps, context: AdapterContext): Prom data: { status: "sent", nextAttemptAt: null }, }); if (claim.count === 0) continue; - try { - if (row.kind === "group" || row.kind === "intro") { - if (!row.threadId) { - await deps.prisma.messagingOutbound.update({ - where: { id: row.id }, - data: { status: "failed" }, - }); - continue; - } - const sent = await deps.messaging.sendToThread( - { threadId: row.threadId, body: row.body }, - context, - ); - await deps.prisma.messagingOutbound.update({ - where: { id: row.id }, - data: { providerHandle: sent.handle }, - }); - continue; - } - const identity = row.identityId - ? await deps.prisma.messagingIdentity.findUnique({ where: { id: row.identityId } }) - : null; - if (!identity) { + if (row.kind === "group" || row.kind === "intro") { + if (!row.threadId) { await deps.prisma.messagingOutbound.update({ where: { id: row.id }, data: { status: "failed" }, }); continue; } - // Sendblue's consecutive-outbound vendor cap; other providers have no - // equivalent limit, so do not hold Slack/WhatsApp/Telegram DMs. - if ( - identity.provider === "sendblue" && - identity.outboundSinceInbound >= MESSAGING_DM_OUTBOUND_CAP - ) { - // Cap holds are not failures: release the claim back to pending. - await deps.prisma.messagingOutbound.update({ - where: { id: row.id }, - data: { status: "pending" }, - }); - continue; - } - const threadId = await resolveDirectThread(deps, identity, context); - const invitePair = connectInvitePair(row.idempotencyKey); - if (invitePair) { - const result = await sendConnectInvite( - deps, - { id: row.id, body: row.body }, - threadId, - invitePair, - context, - ); - if (result === "delivered") { - await deps.prisma.messagingIdentity.update({ - where: { id: identity.id }, - data: { outboundSinceInbound: { increment: 1 } }, - }); - } - continue; - } - const sent = await deps.messaging.sendToThread({ threadId, body: row.body }, context); - await deps.prisma.messagingOutbound.updateMany({ + await sendOutboundAtMostOnce( + deps, + { id: row.id, idempotencyKey: row.idempotencyKey, body: row.body }, + row.threadId, + context, + ); + continue; + } + const identity = row.identityId + ? await deps.prisma.messagingIdentity.findUnique({ where: { id: row.identityId } }) + : null; + if (!identity) { + await deps.prisma.messagingOutbound.update({ where: { id: row.id }, - data: { providerHandle: sent.handle }, + data: { status: "failed" }, }); + continue; + } + // Sendblue's consecutive-outbound vendor cap; other providers have no + // equivalent limit, so do not hold Slack/WhatsApp/Telegram DMs. + if ( + identity.provider === "sendblue" && + identity.outboundSinceInbound >= MESSAGING_DM_OUTBOUND_CAP + ) { + // Cap holds are not failures: release the claim back to pending. + await deps.prisma.messagingOutbound.update({ + where: { id: row.id }, + data: { status: "pending" }, + }); + continue; + } + let threadId: string; + try { + threadId = await resolveDirectThread(deps, identity, context); + } catch { + // Pre-send setup failed; no provider call was made, so retry is safe. + await revertOutboundToPendingWithRetry(deps, row); + continue; + } + const invitePair = connectInvitePair(row.idempotencyKey); + if (invitePair) { + const result = await sendConnectInvite( + deps, + { id: row.id, body: row.body }, + threadId, + invitePair, + context, + ); + if (result === "delivered") { + await deps.prisma.messagingIdentity.update({ + where: { id: identity.id }, + data: { outboundSinceInbound: { increment: 1 } }, + }); + } + continue; + } + const result = await sendOutboundAtMostOnce( + deps, + { id: row.id, idempotencyKey: row.idempotencyKey, body: row.body }, + threadId, + context, + ); + if (result === "delivered") { await deps.prisma.messagingIdentity.update({ where: { id: identity.id }, data: { outboundSinceInbound: { increment: 1 } }, }); - } catch { - // Transient provider errors go back to pending with a backed-off - // retry; only an exhausted budget is terminal. - const attempts = (row.attempts ?? 0) + 1; - const exhausted = attempts >= MESSAGING_OUTBOUND_MAX_ATTEMPTS; - const retryAt = exhausted - ? null - : new Date(Date.now() + messagingOutboundRetryDelayMs(attempts)); - await deps.prisma.messagingOutbound.updateMany({ - where: { id: row.id }, - data: { - attempts, - status: exhausted ? "failed" : "pending", - nextAttemptAt: retryAt, - }, - }); - if (!exhausted && retryAt) { - // Propagate an enqueue failure: the messaging.deliver job then fails - // and the queue's own retry re-runs the drain. Swallowing it would - // strand the row in pending — no reconciler reclaims outbox rows. - // Re-entry is safe: the row is pending again and nextAttemptAt keeps - // other drains from racing the backoff window. - await deps.jobs.enqueue(messagingDeliverJob(undefined, retryAt)); - } } } }