From 2d75a2d32814c2396f0b8e7d049794ff0d645f22 Mon Sep 17 00:00:00 2001 From: Vincent Koc Date: Wed, 12 Aug 2026 11:21:59 +0800 Subject: [PATCH] fix(delivery): settle exact sends dispatched under a producer claim (#121587) Exact-reconciliation sends skipped pre-send promotion, so dispatch cleared the producer claim without publishing the platform send attempt identity. The post-send ack then failed its ownership guard with "Delivery platform claim was lost", settling provably delivered sends as unknown; since #121833 that ambiguity surfaces as a spurious "couldn't confirm my previous reply" recovery notice. Dispatch now atomically promotes the producer claim into the active attempt and publishes the attempt identity, so successful sends settle and lease renewals keep working. Adds a boundary regression pinning unknown_after_send evidence across a later unclaimed batch dispatch. Co-authored-by: Vincent Koc Co-authored-by: Ayaan Zaidi --- .../agent-harness-runtime.json | 2 +- .../agent-harness.json | 2 +- .../plugin-sdk-api-baseline/channel-core.json | 2 +- .../channel-entry-contract.json | 2 +- .../channel-message.json | 2 +- .../channel-outbound.json | 2 +- .../channel-plugin-common.json | 2 +- .../plugin-sdk-api-baseline/core.json | 2 +- .../plugin-sdk-api-baseline/discord.json | 2 +- .../inbound-reply-dispatch.json | 2 +- .../meeting-runtime.json | 2 +- .../plugin-sdk-api-baseline/plugin-entry.json | 2 +- .../plugin-runtime.json | 2 +- .../provider-catalog-runtime.json | 2 +- .../plugin-sdk-api-baseline/tool-plugin.json | 2 +- .../webhook-ingress.json | 2 +- src/infra/delivery-queue-sqlite-claim.test.ts | 87 +++++++++++++++ src/infra/delivery-queue-sqlite-claim.ts | 69 ++++++++++-- ...e.exact-reconciliation.integration.test.ts | 101 ++++++++++++++++++ .../outbound/delivery-queue-platform-lease.ts | 36 +++++++ src/infra/outbound/delivery-queue-storage.ts | 26 ++--- .../outbound/delivery-queue.storage.test.ts | 19 ++++ 22 files changed, 325 insertions(+), 45 deletions(-) create mode 100644 src/infra/delivery-queue-sqlite-claim.test.ts create mode 100644 src/infra/outbound/deliver-queue.exact-reconciliation.integration.test.ts diff --git a/docs/.generated/plugin-sdk-api-baseline/agent-harness-runtime.json b/docs/.generated/plugin-sdk-api-baseline/agent-harness-runtime.json index 4fb7c774405c..2a530c0f36a0 100644 --- a/docs/.generated/plugin-sdk-api-baseline/agent-harness-runtime.json +++ b/docs/.generated/plugin-sdk-api-baseline/agent-harness-runtime.json @@ -1 +1 @@ -{"contentHash":"7438fb02b3bfbdb5a52aa953d68a92ce61149c457d77db2466fe08130ef62286","entrypoint":"agent-harness-runtime","importSpecifier":"openclaw/plugin-sdk/agent-harness-runtime"} +{"contentHash":"5e739e6ec9fd1a63b1fad71781d063c47b0e2169d8e39cc8cbf21352ed1333f4","entrypoint":"agent-harness-runtime","importSpecifier":"openclaw/plugin-sdk/agent-harness-runtime"} diff --git a/docs/.generated/plugin-sdk-api-baseline/agent-harness.json b/docs/.generated/plugin-sdk-api-baseline/agent-harness.json index f9d009dbda95..6a77527735c6 100644 --- a/docs/.generated/plugin-sdk-api-baseline/agent-harness.json +++ b/docs/.generated/plugin-sdk-api-baseline/agent-harness.json @@ -1 +1 @@ -{"contentHash":"dfb557dbe8f946017d425d2c8b90277b728720247996bf40e1f62e99ca4ceeda","entrypoint":"agent-harness","importSpecifier":"openclaw/plugin-sdk/agent-harness"} +{"contentHash":"7a620697c8689b8ddd08f9d9ec31e54240455158e43277946178caa8b81c3872","entrypoint":"agent-harness","importSpecifier":"openclaw/plugin-sdk/agent-harness"} diff --git a/docs/.generated/plugin-sdk-api-baseline/channel-core.json b/docs/.generated/plugin-sdk-api-baseline/channel-core.json index d230b6922683..00d2a0a258c3 100644 --- a/docs/.generated/plugin-sdk-api-baseline/channel-core.json +++ b/docs/.generated/plugin-sdk-api-baseline/channel-core.json @@ -1 +1 @@ -{"contentHash":"025a57f23dc0ec9671c3c30cbe41e4fb5dfd94be09f00f9a3f6a648923255540","entrypoint":"channel-core","importSpecifier":"openclaw/plugin-sdk/channel-core"} +{"contentHash":"c462572277db06da0e31193b91fef1ff87682602665148d89fbbf848929f11a9","entrypoint":"channel-core","importSpecifier":"openclaw/plugin-sdk/channel-core"} diff --git a/docs/.generated/plugin-sdk-api-baseline/channel-entry-contract.json b/docs/.generated/plugin-sdk-api-baseline/channel-entry-contract.json index b6a8ccd2f5e1..f402b1376699 100644 --- a/docs/.generated/plugin-sdk-api-baseline/channel-entry-contract.json +++ b/docs/.generated/plugin-sdk-api-baseline/channel-entry-contract.json @@ -1 +1 @@ -{"contentHash":"987b5195f5fe4c52c4fd99c36679aa5cab1fbb93249f9e5704c6bb7e9ff560b2","entrypoint":"channel-entry-contract","importSpecifier":"openclaw/plugin-sdk/channel-entry-contract"} +{"contentHash":"c202b9e8bbbcc1d35d29e9ce92e0d9fd5a08a5a9c2c4c6a4eb6f621a52839da9","entrypoint":"channel-entry-contract","importSpecifier":"openclaw/plugin-sdk/channel-entry-contract"} diff --git a/docs/.generated/plugin-sdk-api-baseline/channel-message.json b/docs/.generated/plugin-sdk-api-baseline/channel-message.json index 6dd4baf58d41..4d2e61f084b9 100644 --- a/docs/.generated/plugin-sdk-api-baseline/channel-message.json +++ b/docs/.generated/plugin-sdk-api-baseline/channel-message.json @@ -1 +1 @@ -{"contentHash":"8a42f14a4a29a75dfcec8b1f8eb78201bb8b308fe565286ece5d8f32bf548437","entrypoint":"channel-message","importSpecifier":"openclaw/plugin-sdk/channel-message"} +{"contentHash":"22a0413c4e79e1c1cd51e122681bf7ad3e7c867668e61dd2f510ea9e14968891","entrypoint":"channel-message","importSpecifier":"openclaw/plugin-sdk/channel-message"} diff --git a/docs/.generated/plugin-sdk-api-baseline/channel-outbound.json b/docs/.generated/plugin-sdk-api-baseline/channel-outbound.json index a8ccef6ae842..1ded94d41b1d 100644 --- a/docs/.generated/plugin-sdk-api-baseline/channel-outbound.json +++ b/docs/.generated/plugin-sdk-api-baseline/channel-outbound.json @@ -1 +1 @@ -{"contentHash":"9a57b7d2bf46a92dcbb8ff35cb3e034f8effd7ff4271f57d44f68ab2e2439062","entrypoint":"channel-outbound","importSpecifier":"openclaw/plugin-sdk/channel-outbound"} +{"contentHash":"810a5da4a06925554ab89f9462a17f1fb9fa3196277227eaa383aa82b3b81e58","entrypoint":"channel-outbound","importSpecifier":"openclaw/plugin-sdk/channel-outbound"} diff --git a/docs/.generated/plugin-sdk-api-baseline/channel-plugin-common.json b/docs/.generated/plugin-sdk-api-baseline/channel-plugin-common.json index 7d816165431d..0ebc205b6c16 100644 --- a/docs/.generated/plugin-sdk-api-baseline/channel-plugin-common.json +++ b/docs/.generated/plugin-sdk-api-baseline/channel-plugin-common.json @@ -1 +1 @@ -{"contentHash":"d121e1874d6de3f42801aa37da787279750e8abbebcd38d490177890df43712f","entrypoint":"channel-plugin-common","importSpecifier":"openclaw/plugin-sdk/channel-plugin-common"} +{"contentHash":"604289da3812346c5a080a9f867a32f4fcc0e06a9837f61395c961d8363d8b9d","entrypoint":"channel-plugin-common","importSpecifier":"openclaw/plugin-sdk/channel-plugin-common"} diff --git a/docs/.generated/plugin-sdk-api-baseline/core.json b/docs/.generated/plugin-sdk-api-baseline/core.json index b455abad7ec7..581e18aa177d 100644 --- a/docs/.generated/plugin-sdk-api-baseline/core.json +++ b/docs/.generated/plugin-sdk-api-baseline/core.json @@ -1 +1 @@ -{"contentHash":"04b433495b96cc4cb4051822b4535deee45d1194211bb73ab0ea6b9fade6a51f","entrypoint":"core","importSpecifier":"openclaw/plugin-sdk/core"} +{"contentHash":"1ecacce44a31327293a1ab1b2e24e3f085e526764f90fbb367687c42e029cce4","entrypoint":"core","importSpecifier":"openclaw/plugin-sdk/core"} diff --git a/docs/.generated/plugin-sdk-api-baseline/discord.json b/docs/.generated/plugin-sdk-api-baseline/discord.json index 0d4f7155d24d..7c1501dcd9a8 100644 --- a/docs/.generated/plugin-sdk-api-baseline/discord.json +++ b/docs/.generated/plugin-sdk-api-baseline/discord.json @@ -1 +1 @@ -{"contentHash":"7ac1385ce7667bf15cb604af3d935a87f90a573ea975ffe3fe9b5addcb25d707","entrypoint":"discord","importSpecifier":"openclaw/plugin-sdk/discord"} +{"contentHash":"b9936060e6ca111906bdca8409097589f8548304b947a7762dc068992f0d2d95","entrypoint":"discord","importSpecifier":"openclaw/plugin-sdk/discord"} diff --git a/docs/.generated/plugin-sdk-api-baseline/inbound-reply-dispatch.json b/docs/.generated/plugin-sdk-api-baseline/inbound-reply-dispatch.json index fa5fa79b7e19..ae7c44a68622 100644 --- a/docs/.generated/plugin-sdk-api-baseline/inbound-reply-dispatch.json +++ b/docs/.generated/plugin-sdk-api-baseline/inbound-reply-dispatch.json @@ -1 +1 @@ -{"contentHash":"e9de7ab92ebf34761afd5f38be47bb55beb1f6e849b9dd89d5f3f824cd823bc9","entrypoint":"inbound-reply-dispatch","importSpecifier":"openclaw/plugin-sdk/inbound-reply-dispatch"} +{"contentHash":"c7fbb54e9ad926e520df85c396531eadbab78e042ffe315dd3311a3508eb6e55","entrypoint":"inbound-reply-dispatch","importSpecifier":"openclaw/plugin-sdk/inbound-reply-dispatch"} diff --git a/docs/.generated/plugin-sdk-api-baseline/meeting-runtime.json b/docs/.generated/plugin-sdk-api-baseline/meeting-runtime.json index 411a34d82bec..e13f7ffbc91c 100644 --- a/docs/.generated/plugin-sdk-api-baseline/meeting-runtime.json +++ b/docs/.generated/plugin-sdk-api-baseline/meeting-runtime.json @@ -1 +1 @@ -{"contentHash":"35a2d593c9622675de2522a7394fd8b2e19ea1c5719d122e882b4f7f4ed2f319","entrypoint":"meeting-runtime","importSpecifier":"openclaw/plugin-sdk/meeting-runtime"} +{"contentHash":"4ea730b41856a414960f55940197485c5b9a80782f7584425eedd8fca585a177","entrypoint":"meeting-runtime","importSpecifier":"openclaw/plugin-sdk/meeting-runtime"} diff --git a/docs/.generated/plugin-sdk-api-baseline/plugin-entry.json b/docs/.generated/plugin-sdk-api-baseline/plugin-entry.json index 2362fa77b87d..af3a0f2510ff 100644 --- a/docs/.generated/plugin-sdk-api-baseline/plugin-entry.json +++ b/docs/.generated/plugin-sdk-api-baseline/plugin-entry.json @@ -1 +1 @@ -{"contentHash":"df5612f8e2eaf69285c818d31c760dbf259e234c9cd8f1e48a9ad36c99c9e110","entrypoint":"plugin-entry","importSpecifier":"openclaw/plugin-sdk/plugin-entry"} +{"contentHash":"6d3f6a64b8ad7459763f4c62a5f731b679ddc64eba4b3463be122659546b2b58","entrypoint":"plugin-entry","importSpecifier":"openclaw/plugin-sdk/plugin-entry"} diff --git a/docs/.generated/plugin-sdk-api-baseline/plugin-runtime.json b/docs/.generated/plugin-sdk-api-baseline/plugin-runtime.json index a35c1d7b57d8..9e081752c72e 100644 --- a/docs/.generated/plugin-sdk-api-baseline/plugin-runtime.json +++ b/docs/.generated/plugin-sdk-api-baseline/plugin-runtime.json @@ -1 +1 @@ -{"contentHash":"c3e5839b6da58bc4eaa9ca77fb4352351a914b98d52c001caecff16c9e7ef930","entrypoint":"plugin-runtime","importSpecifier":"openclaw/plugin-sdk/plugin-runtime"} +{"contentHash":"e7a9ba2f6a48e5c3c2f7cc4a3d6be9d4776deb0c55b4a22084968ad19ad64fbc","entrypoint":"plugin-runtime","importSpecifier":"openclaw/plugin-sdk/plugin-runtime"} diff --git a/docs/.generated/plugin-sdk-api-baseline/provider-catalog-runtime.json b/docs/.generated/plugin-sdk-api-baseline/provider-catalog-runtime.json index 9ad9e4e5bd9d..406cd885b820 100644 --- a/docs/.generated/plugin-sdk-api-baseline/provider-catalog-runtime.json +++ b/docs/.generated/plugin-sdk-api-baseline/provider-catalog-runtime.json @@ -1 +1 @@ -{"contentHash":"3ddb3a7d3e0d1a283a9844278f0643d74dae2b0e0f79e1581e97c03716fd9b53","entrypoint":"provider-catalog-runtime","importSpecifier":"openclaw/plugin-sdk/provider-catalog-runtime"} +{"contentHash":"a0bbe80276278db6a7f9981d5fb0b2e990438995e76db85d40bf695924e2ad19","entrypoint":"provider-catalog-runtime","importSpecifier":"openclaw/plugin-sdk/provider-catalog-runtime"} diff --git a/docs/.generated/plugin-sdk-api-baseline/tool-plugin.json b/docs/.generated/plugin-sdk-api-baseline/tool-plugin.json index 392da8d57ce0..a52f4f826d4b 100644 --- a/docs/.generated/plugin-sdk-api-baseline/tool-plugin.json +++ b/docs/.generated/plugin-sdk-api-baseline/tool-plugin.json @@ -1 +1 @@ -{"contentHash":"ab174d984c51f9f9512b16256a1da9f624ae6afafb3e2def2ac3d40b9fc0a468","entrypoint":"tool-plugin","importSpecifier":"openclaw/plugin-sdk/tool-plugin"} +{"contentHash":"1759143daf318471c30e816f8651b52ee01c1edbade4f0a9f87b4260b11a21a6","entrypoint":"tool-plugin","importSpecifier":"openclaw/plugin-sdk/tool-plugin"} diff --git a/docs/.generated/plugin-sdk-api-baseline/webhook-ingress.json b/docs/.generated/plugin-sdk-api-baseline/webhook-ingress.json index 00aa42cd81e4..1d2b55e775cf 100644 --- a/docs/.generated/plugin-sdk-api-baseline/webhook-ingress.json +++ b/docs/.generated/plugin-sdk-api-baseline/webhook-ingress.json @@ -1 +1 @@ -{"contentHash":"3c08887d7a84b6b30472f25e5edbeefa42c15c4310b45f6a0c9a9e02ddfe16aa","entrypoint":"webhook-ingress","importSpecifier":"openclaw/plugin-sdk/webhook-ingress"} +{"contentHash":"1eff4a94da8119d071af4884b96f77fe2b3a73650b9be79bfb2703af510e0a8e","entrypoint":"webhook-ingress","importSpecifier":"openclaw/plugin-sdk/webhook-ingress"} diff --git a/src/infra/delivery-queue-sqlite-claim.test.ts b/src/infra/delivery-queue-sqlite-claim.test.ts new file mode 100644 index 000000000000..a0741c4e6f63 --- /dev/null +++ b/src/infra/delivery-queue-sqlite-claim.test.ts @@ -0,0 +1,87 @@ +import { describe, expect, it, vi } from "vitest"; +import { + claimDeliveryQueueEntryPlatformSend, + dispatchDeliveryQueueEntryPlatformSend, +} from "./delivery-queue-sqlite-claim.js"; +import { loadDeliveryQueueEntry, upsertDeliveryQueueEntry } from "./delivery-queue-sqlite.js"; +import { installDeliveryQueueTmpDirHooks } from "./outbound/delivery-queue.test-helpers.js"; + +describe("delivery queue SQLite dispatch ownership", () => { + const { tmpDir } = installDeliveryQueueTmpDirHooks(); + const queueName = "test-dispatch-owner"; + + it("atomically promotes dispatch ownership and rejects expired or replaced claims", () => { + vi.useFakeTimers(); + try { + vi.setSystemTime(new Date("2026-08-10T10:00:00.000Z")); + const stateDir = tmpDir(); + const id = "cron-direct-delivery:v1:dispatch-owner"; + upsertDeliveryQueueEntry({ + queueName, + entry: { + id, + enqueuedAt: Date.now(), + retryCount: 0, + completionRetention: { + idPrefix: "cron-direct-delivery:v1:", + maxAgeMs: 24 * 60 * 60_000, + maxEntries: 2, + }, + requiresProducerClaim: true, + }, + stateDir, + }); + + const expiredClaimId = claimDeliveryQueueEntryPlatformSend({ queueName, id, stateDir }); + if (!expiredClaimId) { + throw new Error("test invariant: the first producer claim must be available"); + } + vi.advanceTimersByTime(30_001); + expect( + dispatchDeliveryQueueEntryPlatformSend({ + queueName, + id, + claimId: expiredClaimId, + stateDir, + }), + ).toBe(false); + + const claimId = claimDeliveryQueueEntryPlatformSend({ queueName, id, stateDir }); + if (!claimId) { + throw new Error("test invariant: the replacement producer claim must be available"); + } + expect( + dispatchDeliveryQueueEntryPlatformSend({ + queueName, + id, + claimId: expiredClaimId, + stateDir, + }), + ).toBe(false); + expect( + dispatchDeliveryQueueEntryPlatformSend({ + queueName, + id, + claimId, + stateDir, + route: { replyToId: "thread-1" }, + }), + ).toBe(true); + expect(loadDeliveryQueueEntry(queueName, id, stateDir)).toMatchObject({ + recoveryState: "send_attempt_started", + platformSendAttemptId: claimId, + platformSendStartedAt: Date.now(), + effectiveReplyToId: "thread-1", + availableAt: Date.now() + 30_000, + }); + expect(loadDeliveryQueueEntry(queueName, id, stateDir)?.producerClaimId).toBeUndefined(); + + vi.advanceTimersByTime(30_001); + expect(dispatchDeliveryQueueEntryPlatformSend({ queueName, id, claimId, stateDir })).toBe( + false, + ); + } finally { + vi.useRealTimers(); + } + }); +}); diff --git a/src/infra/delivery-queue-sqlite-claim.ts b/src/infra/delivery-queue-sqlite-claim.ts index 1ebcf0e4a0f7..911b2bf1cdd5 100644 --- a/src/infra/delivery-queue-sqlite-claim.ts +++ b/src/infra/delivery-queue-sqlite-claim.ts @@ -68,9 +68,9 @@ export function transitionOwnedDeliveryQueueEntry( ); } -function transitionUnsentDeliveryQueueEntry( +function transitionDeliveryQueueEntryPlatformSend( params: PlatformClaimParams, - operation: "claim" | "promote", + operation: "claim" | "promote" | "dispatch", transition: (entry: DeliveryQueueEntryState, now: number) => DeliveryQueueEntryState | undefined, ): boolean { // State-database opens reuse the canonical path-owned connection, so both @@ -82,13 +82,16 @@ function transitionUnsentDeliveryQueueEntry( database.db, () => { const current = loadDeliveryQueueEntry(params.queueName, params.id, params.stateDir); + if (!current) { + return false; + } if ( - !current || - (current.platformSendStartedAt !== undefined && - (operation !== "claim" || - current.platformSendStartedAt !== params.reconciledPlatformSendStartedAt || - current.platformSendAttemptId !== params.reconciledPlatformSendAttemptId || - typeof current.platformSendAttemptId !== "string")) + current.platformSendStartedAt !== undefined && + (operation === "promote" || + (operation === "claim" && + (current.platformSendStartedAt !== params.reconciledPlatformSendStartedAt || + current.platformSendAttemptId !== params.reconciledPlatformSendAttemptId || + typeof current.platformSendAttemptId !== "string"))) ) { return false; } @@ -114,7 +117,7 @@ export function claimDeliveryQueueEntryPlatformSend( params: PlatformClaimParams, ): string | undefined { const claimId = generateSecureUuid(); - return transitionUnsentDeliveryQueueEntry(params, "claim", (entry, now) => { + return transitionDeliveryQueueEntryPlatformSend(params, "claim", (entry, now) => { const reconciledNotSent = entry.recoveryState === "send_attempt_started" && typeof params.reconciledPlatformSendStartedAt === "number" && @@ -197,7 +200,7 @@ export function promoteDeliveryQueueEntryPlatformSend( route?: { replyToId?: string | null }; }, ): boolean { - return transitionUnsentDeliveryQueueEntry(params, "promote", (entry, now) => + return transitionDeliveryQueueEntryPlatformSend(params, "promote", (entry, now) => entry.recoveryState === "producer_claimed" && entry.producerClaimId === params.claimId && typeof entry.availableAt === "number" && @@ -219,3 +222,49 @@ export function promoteDeliveryQueueEntryPlatformSend( : undefined, ); } + +/** Atomically authorize dispatch, promoting a producer claim into the active attempt. */ +export function dispatchDeliveryQueueEntryPlatformSend( + params: PlatformClaimParams & { + claimId: string; + route?: { replyToId?: string | null }; + }, +): boolean { + return transitionDeliveryQueueEntryPlatformSend(params, "dispatch", (entry, now) => { + const producerOwned = + entry.recoveryState === "producer_claimed" && + entry.producerClaimId === params.claimId && + typeof entry.availableAt === "number" && + entry.availableAt > now; + const attemptOwned = + (entry.recoveryState === "send_attempt_started" || + entry.recoveryState === "unknown_after_send") && + entry.platformSendAttemptId === params.claimId && + (entry.requiresProducerClaim !== true || + (typeof entry.availableAt === "number" && entry.availableAt > now)); + if (!producerOwned && !attemptOwned) { + return undefined; + } + return { + ...entry, + // Exact reconciliation can skip pre-send promotion, so publish attempt identity + // atomically; later batch dispatches retain stronger unknown-after-send evidence. + availableAt: + entry.requiresProducerClaim === true + ? producerOwned + ? now + PLATFORM_SEND_OWNER_LEASE_MS + : entry.availableAt + : undefined, + producerClaimId: undefined, + platformSendAttemptId: params.claimId, + platformSendStartedAt: now, + ...(params.route && "replyToId" in params.route + ? { effectiveReplyToId: params.route.replyToId ?? null } + : {}), + recoveryState: + entry.recoveryState === "unknown_after_send" + ? "unknown_after_send" + : "send_attempt_started", + }; + }); +} diff --git a/src/infra/outbound/deliver-queue.exact-reconciliation.integration.test.ts b/src/infra/outbound/deliver-queue.exact-reconciliation.integration.test.ts new file mode 100644 index 000000000000..60314adcead8 --- /dev/null +++ b/src/infra/outbound/deliver-queue.exact-reconciliation.integration.test.ts @@ -0,0 +1,101 @@ +import { afterEach, beforeAll, beforeEach, describe, expect, it, vi } from "vitest"; +import { createMessageReceiptFromOutboundResults } from "../../channels/message/receipt.js"; +import type { ChannelMessageSendTextContext } from "../../channels/message/types.js"; +import type { OpenClawConfig } from "../../config/config.js"; +import { createEmptyPluginRegistry } from "../../plugins/registry.js"; +import { resetPluginRuntimeStateForTest, setActivePluginRegistry } from "../../plugins/runtime.js"; +import { createOutboundTestPlugin, createTestRegistry } from "../../test-utils/channel-plugins.js"; +import { getDeliveryQueueEntryStatus } from "../delivery-queue-sqlite.js"; +import { + boundedCronCompletionRetention, + drainMatrixReconnect, + matrixOutboundForQueueTest, +} from "./deliver.queue-integration.test-support.js"; +import { OUTBOUND_DELIVERY_QUEUE_NAME } from "./delivery-queue-media-staging.js"; +import type { DeliverFn } from "./delivery-queue.js"; +import { installDeliveryQueueTmpDirHooks } from "./delivery-queue.test-helpers.js"; + +let deliverOutboundPayloads: typeof import("./deliver.js").deliverOutboundPayloads; + +describe("exact Matrix delivery queue reconciliation", () => { + const fixtures = installDeliveryQueueTmpDirHooks(); + let tmpDir: string; + + beforeAll(async () => { + ({ deliverOutboundPayloads } = await import("./deliver.js")); + }); + + beforeEach(() => { + tmpDir = fixtures.tmpDir(); + }); + + afterEach(() => { + resetPluginRuntimeStateForTest(); + setActivePluginRegistry(createEmptyPluginRegistry()); + }); + + it.each(["required", "best_effort"] as const)( + "settles one exact Matrix %s send without restart replay", + async (queuePolicy) => { + process.env.OPENCLAW_STATE_DIR = tmpDir; + const deliveryIntentId = `cron-direct-delivery:v1:exact-${queuePolicy}-completion`; + const messageId = `exact-${queuePolicy}-message`; + const reconcileUnknownSend = vi.fn(); + const sendText = vi.fn(async (ctx: ChannelMessageSendTextContext) => { + expect(ctx.deliveryQueueId).toBe(deliveryIntentId); + await ctx.onPlatformSendDispatch?.(); + return { + messageId, + receipt: createMessageReceiptFromOutboundResults({ + results: [{ channel: "matrix", messageId }], + kind: "text", + }), + }; + }); + setActivePluginRegistry( + createTestRegistry([ + { + pluginId: "matrix", + source: "test", + plugin: { + ...createOutboundTestPlugin({ id: "matrix", outbound: matrixOutboundForQueueTest }), + message: { + id: "matrix", + durableFinal: { + capabilities: { text: true, reconcileUnknownSend: true }, + reconcileUnknownSendKinds: { text: true }, + reconcileUnknownSend, + }, + send: { text: sendText }, + }, + }, + }, + ]), + ); + const params = { + cfg: {} as OpenClawConfig, + channel: "matrix" as const, + to: "!room:example", + payloads: [{ text: "send exactly once with durable platform identity" }], + queuePolicy, + ...(queuePolicy === "best_effort" ? { bestEffort: true } : {}), + deliveryIntentId, + completionRetention: boundedCronCompletionRetention, + reusePendingDeliveryIntent: true, + requireUnknownSendReconciliation: true, + }; + + await expect(deliverOutboundPayloads(params)).resolves.toMatchObject([{ messageId }]); + expect(sendText).toHaveBeenCalledOnce(); + expect(reconcileUnknownSend).not.toHaveBeenCalled(); + expect( + getDeliveryQueueEntryStatus(OUTBOUND_DELIVERY_QUEUE_NAME, deliveryIntentId, tmpDir), + ).toBe("completed"); + + const recoveryDeliver = vi.fn(async () => []); + await drainMatrixReconnect({ deliver: recoveryDeliver, stateDir: tmpDir }); + expect(recoveryDeliver).not.toHaveBeenCalled(); + expect(sendText).toHaveBeenCalledOnce(); + }, + ); +}); diff --git a/src/infra/outbound/delivery-queue-platform-lease.ts b/src/infra/outbound/delivery-queue-platform-lease.ts index 7d011c061776..73b4b9eb3962 100644 --- a/src/infra/outbound/delivery-queue-platform-lease.ts +++ b/src/infra/outbound/delivery-queue-platform-lease.ts @@ -1,9 +1,26 @@ import { claimDeliveryQueueEntryPlatformSend, + dispatchDeliveryQueueEntryPlatformSend, renewDeliveryQueueEntryPlatformSendLease, } from "../delivery-queue-sqlite-claim.js"; import { OUTBOUND_DELIVERY_QUEUE_NAME } from "./delivery-queue-media-staging.js"; +/** Atomically transfer a stable pending producer intent to one platform sender. */ +export async function claimDeliveryPlatformSendAttempt( + id: string, + stateDir?: string, + reconciledPlatformSendStartedAt?: number, + reconciledPlatformSendAttemptId?: string, +): Promise { + return claimDeliveryQueueEntryPlatformSend({ + queueName: OUTBOUND_DELIVERY_QUEUE_NAME, + id, + stateDir, + ...(reconciledPlatformSendStartedAt !== undefined ? { reconciledPlatformSendStartedAt } : {}), + ...(reconciledPlatformSendAttemptId !== undefined ? { reconciledPlatformSendAttemptId } : {}), + }); +} + /** Claim and atomically upgrade a live reusable producer to renewable ownership. */ export async function claimReusableDeliveryPlatformSendAttempt( id: string, @@ -30,3 +47,22 @@ export async function renewDeliveryPlatformSendLease( claimId, }); } + +/** Promote or refresh the exact live owner at recipient-visible dispatch. */ +export function markOwnedDeliveryPlatformSendDispatched( + id: string, + stateDir: string | undefined, + route: { replyToId?: string | null } | undefined, + claimId: string, +): void { + const dispatched = dispatchDeliveryQueueEntryPlatformSend({ + queueName: OUTBOUND_DELIVERY_QUEUE_NAME, + id, + stateDir, + route, + claimId, + }); + if (!dispatched) { + throw new Error(`Delivery platform claim was lost: ${id}`); + } +} diff --git a/src/infra/outbound/delivery-queue-storage.ts b/src/infra/outbound/delivery-queue-storage.ts index a4c289137e4d..480558a45f84 100644 --- a/src/infra/outbound/delivery-queue-storage.ts +++ b/src/infra/outbound/delivery-queue-storage.ts @@ -9,7 +9,6 @@ import type { import type { ReplyToMode } from "../../config/types.js"; import type { PluginHookReplyPayloadSendingContext } from "../../plugins/hook-types.js"; import { - claimDeliveryQueueEntryPlatformSend, promoteDeliveryQueueEntryPlatformSend, transitionOwnedDeliveryQueueEntry, type InitialDeliveryProducerClaim, @@ -44,6 +43,7 @@ import { OUTBOUND_DELIVERY_QUEUE_NAME, OUTBOUND_LEGACY_PREPARATION_QUEUE_NAME, } from "./delivery-queue-media-staging.js"; +import { markOwnedDeliveryPlatformSendDispatched } from "./delivery-queue-platform-lease.js"; import { StableDeliveryPreparationLostError, type StableDeliveryPreparation, @@ -481,21 +481,7 @@ export async function failDeliveryAfterPlatformSend( ); } -/** Atomically transfer a stable pending producer intent to one platform sender. */ -export async function claimDeliveryPlatformSendAttempt( - id: string, - stateDir?: string, - reconciledPlatformSendStartedAt?: number, - reconciledPlatformSendAttemptId?: string, -): Promise { - return claimDeliveryQueueEntryPlatformSend({ - queueName: OUTBOUND_DELIVERY_QUEUE_NAME, - id, - stateDir, - ...(reconciledPlatformSendStartedAt !== undefined ? { reconciledPlatformSendStartedAt } : {}), - ...(reconciledPlatformSendAttemptId !== undefined ? { reconciledPlatformSendAttemptId } : {}), - }); -} +export { claimDeliveryPlatformSendAttempt } from "./delivery-queue-platform-lease.js"; /** Reserve one durable delivery call before invoking the provider path. */ export async function reserveDeliveryAttempt( @@ -579,14 +565,16 @@ export async function markDeliveryPlatformSendDispatched( route?: { replyToId?: string | null }, expectedPlatformSendAttemptId?: string | null, ): Promise { + if (typeof expectedPlatformSendAttemptId === "string") { + markOwnedDeliveryPlatformSendDispatched(id, stateDir, route, expectedPlatformSendAttemptId); + return; + } updateQueuedDelivery( id, stateDir, (entry) => ({ ...entry, - // Dispatch still belongs to the promoted producer until provider I/O - // settles; clearing its lease lets another process replay an active send. - availableAt: expectedPlatformSendAttemptId ? entry.availableAt : undefined, + availableAt: undefined, producerClaimId: undefined, platformSendStartedAt: Date.now(), ...(route && "replyToId" in route ? { effectiveReplyToId: route.replyToId ?? null } : {}), diff --git a/src/infra/outbound/delivery-queue.storage.test.ts b/src/infra/outbound/delivery-queue.storage.test.ts index 23ec9a60878e..3bf7cce2f06c 100644 --- a/src/infra/outbound/delivery-queue.storage.test.ts +++ b/src/infra/outbound/delivery-queue.storage.test.ts @@ -547,6 +547,25 @@ describe("delivery-queue storage", () => { expect(entry.recoveryState).toBe("send_attempt_started"); }); + it("keeps ambiguous post-send evidence across a later unclaimed batch dispatch", async () => { + const id = await enqueueTextDelivery( + { + channel: "forum", + to: "123", + payloads: [{ text: "test" }], + }, + tmpDir(), + ); + + await markDeliveryPlatformSendAttemptStarted(id, tmpDir()); + await markDeliveryPlatformOutcomeUnknown(id, tmpDir()); + await markDeliveryPlatformSendDispatched(id, tmpDir()); + + // Downgrading to send_attempt_started would let recovery replay the whole + // batch as not_sent and duplicate the payload that already reached the platform. + expect(readQueuedEntry(tmpDir(), id).recoveryState).toBe("unknown_after_send"); + }); + it("increments retryCount, records attempt time, and sets lastError", async () => { const id = await enqueueTextDelivery( {