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( {