From 07b61d4adc6629f9241571fc46eb7544bb05fc66 Mon Sep 17 00:00:00 2001 From: joshavant <830519+joshavant@users.noreply.github.com> Date: Wed, 12 Aug 2026 16:25:33 -0500 Subject: [PATCH] fix(plugin-sdk): make ingress cancellation fan-in safe --- .../message-handler.ingress-recovery.test.ts | 263 ++++++++++++++++++ .../monitor.message-handler.ingress.test.ts | 155 ++++++++++- .../monitor.inbound-system-event.test.ts | 203 ++++++++++++-- .../message-handler.ingress-lifecycle.test.ts | 130 +++++++++ ...ent-handler.reply-session-conflict.test.ts | 132 ++++++++- src/channels/message/ingress-queue.test.ts | 50 +++- .../channel-ingress-runtime.test.ts | 21 ++ src/plugin-sdk/channel-ingress-runtime.ts | 17 +- 8 files changed, 934 insertions(+), 37 deletions(-) create mode 100644 extensions/discord/src/monitor/message-handler.ingress-recovery.test.ts diff --git a/extensions/discord/src/monitor/message-handler.ingress-recovery.test.ts b/extensions/discord/src/monitor/message-handler.ingress-recovery.test.ts new file mode 100644 index 000000000000..7119e8b62e49 --- /dev/null +++ b/extensions/discord/src/monitor/message-handler.ingress-recovery.test.ts @@ -0,0 +1,263 @@ +// Discord tests cover durable retry recovery through full handler replacement. +import fs from "node:fs/promises"; +import os from "node:os"; +import path from "node:path"; +import type { APIMessage } from "discord-api-types/v10"; +import { + type ChannelIngressQueue, + DEFAULT_INGRESS_RETRY_MAX_ATTEMPTS, +} from "openclaw/plugin-sdk/channel-outbound"; +import { createDeferred } from "openclaw/plugin-sdk/extension-shared"; +import { + closeOpenClawStateDatabaseForTest, + createChannelIngressQueueForTests, +} from "openclaw/plugin-sdk/plugin-state-test-runtime"; +import { describe, expect, it, vi } from "vitest"; +import { createDiscordIngressMonitor } from "./ingress.js"; +import { createDiscordMessageHandler } from "./message-handler.js"; +import { createDiscordHandlerParams } from "./message-handler.test-helpers.js"; + +type DiscordIngressPayload = { + version: 1; + receivedAt: number; + rawMessage: APIMessage; +}; + +type DiscordQueue = ChannelIngressQueue; + +function rawMessage(id: string, channelId = "lane-a"): APIMessage { + return { + id, + channel_id: channelId, + content: "hello", + author: { + id: "user-1", + username: "alice", + discriminator: "0", + avatar: null, + }, + attachments: [], + embeds: [], + mentions: [], + mention_roles: [], + mention_everyone: false, + timestamp: new Date(0).toISOString(), + edited_timestamp: null, + components: [], + pinned: false, + type: 0, + tts: false, + } as unknown as APIMessage; +} + +async function withQueue(run: (queue: DiscordQueue) => Promise): Promise { + const created = await fs.mkdtemp(path.join(os.tmpdir(), "openclaw-discord-recovery-")); + const stateDir = await fs.realpath(created); + const queue = createChannelIngressQueueForTests({ + channelId: "discord", + accountId: "default", + stateDir, + }); + try { + await run(queue); + } finally { + closeOpenClawStateDatabaseForTest(); + await fs.rm(stateDir, { recursive: true, force: true }); + } +} + +async function seedPendingFailure(params: { + queue: DiscordQueue; + id: string; + attempts: number; + laneKey?: string; +}): Promise { + await params.queue.enqueue( + params.id, + { version: 1, receivedAt: 1, rawMessage: rawMessage(params.id) }, + { laneKey: params.laneKey ?? "channel:lane-a", receivedAt: 1 }, + ); + for (let attempt = 1; attempt <= params.attempts; attempt += 1) { + const claim = await params.queue.claim(params.id, { ownerId: `seed-${attempt}` }); + if (!claim) { + throw new Error(`Expected ${params.id} to be claimable for seed attempt ${attempt}`); + } + await params.queue.release(claim, { + lastError: `prior genuine failure ${attempt}`, + releasedAt: 10 + attempt, + }); + } +} + +async function retryFacts(queue: DiscordQueue, id: string) { + const record = (await queue.listPending({ limit: "all" })).find((entry) => entry.id === id); + if (!record) { + throw new Error(`Expected pending Discord ingress row ${id}`); + } + return { + attempts: record.attempts, + lastAttemptAt: record.lastAttemptAt, + lastError: record.lastError, + }; +} + +function createHandler(params: { + queue: DiscordQueue; + preflight: (input: { data: { message?: { id?: string } } }) => Promise; + debounceMs?: number; + beforeDispatch?: () => Promise; +}) { + const handlerParams = createDiscordHandlerParams(); + handlerParams.cfg.messages = { inbound: { debounceMs: params.debounceMs ?? 0 } }; + return createDiscordMessageHandler({ + ...handlerParams, + client: {} as never, + testing: { + preflightDiscordMessage: params.preflight as never, + createIngressMonitor: (monitorParams) => + createDiscordIngressMonitor({ + ...monitorParams, + queue: params.queue, + dispatch: params.beforeDispatch + ? async (event, lifecycle) => { + await params.beforeDispatch?.(); + return await monitorParams.dispatch(event, lifecycle); + } + : monitorParams.dispatch, + }), + }, + }); +} + +describe("Discord durable ingress replacement recovery", () => { + it("terminally settles a preexisting exhausted poison row before its follower", async () => { + await withQueue(async (queue) => { + await seedPendingFailure({ + queue, + id: "poison", + attempts: DEFAULT_INGRESS_RETRY_MAX_ATTEMPTS, + }); + await queue.enqueue( + "follower", + { version: 1, receivedAt: 2, rawMessage: rawMessage("follower") }, + { laneKey: "channel:lane-a", receivedAt: 2 }, + ); + const dispatched: string[] = []; + const handler = createHandler({ + queue, + preflight: vi.fn(async ({ data }) => { + const id = data.message?.id ?? "unknown"; + dispatched.push(id); + if (id === "poison") { + throw new Error("recovered poison failure"); + } + return null; + }), + }); + try { + await vi.waitFor(async () => { + await expect(queue.enqueue("poison", {} as DiscordIngressPayload)).resolves.toMatchObject( + { kind: "failed", record: { reason: "retry-limit-exceeded" } }, + ); + await expect( + queue.enqueue("follower", {} as DiscordIngressPayload), + ).resolves.toMatchObject({ kind: "completed" }); + }); + expect(dispatched).toEqual(["poison", "follower"]); + expect(await queue.listPending({ limit: "all" })).toEqual([]); + expect(await queue.listClaims()).toEqual([]); + } finally { + await handler.deactivate(); + } + }); + }); + + it("preserves retry facts across every Discord cancellation route and replacement", async () => { + await withQueue(async (queue) => { + await seedPendingFailure({ + queue, + id: "poison", + attempts: DEFAULT_INGRESS_RETRY_MAX_ATTEMPTS - 1, + }); + await queue.enqueue( + "follower", + { version: 1, receivedAt: 2, rawMessage: rawMessage("follower") }, + { laneKey: "channel:lane-a", receivedAt: 2 }, + ); + const expectedFacts = await retryFacts(queue, "poison"); + + const dispatchEntered = createDeferred(); + const releaseDispatch = createDeferred(); + const beforeDispatch = async () => { + dispatchEntered.resolve(); + await releaseDispatch.promise; + }; + const beforeDispatchPreflight = vi.fn(async () => null); + const beforeDispatchHandler = createHandler({ + queue, + preflight: beforeDispatchPreflight, + beforeDispatch, + }); + await dispatchEntered.promise; + const beforeDispatchStop = beforeDispatchHandler.deactivate(); + await Promise.resolve(); + releaseDispatch.resolve(); + await beforeDispatchStop; + expect(beforeDispatchPreflight).not.toHaveBeenCalled(); + expect(await retryFacts(queue, "poison")).toEqual(expectedFacts); + + const bufferedPreflight = vi.fn(async () => null); + const bufferedHandler = createHandler({ + queue, + preflight: bufferedPreflight, + debounceMs: 60_000, + }); + await vi.waitFor(async () => expect(await queue.listClaims()).toHaveLength(1)); + await bufferedHandler.deactivate(); + expect(bufferedPreflight).not.toHaveBeenCalled(); + expect(await retryFacts(queue, "poison")).toEqual(expectedFacts); + + const preflightEntered = createDeferred(); + const releasePreflight = createDeferred(); + const activePreflight = vi.fn(async () => { + preflightEntered.resolve(); + await releasePreflight.promise; + return null; + }); + const activeHandler = createHandler({ queue, preflight: activePreflight }); + await preflightEntered.promise; + const activeStop = activeHandler.deactivate(); + await Promise.resolve(); + releasePreflight.resolve(); + await activeStop; + expect(activePreflight).toHaveBeenCalledTimes(1); + expect(await retryFacts(queue, "poison")).toEqual(expectedFacts); + + const finalDispatches: string[] = []; + const replacement = createHandler({ + queue, + preflight: vi.fn(async ({ data }) => { + const id = data.message?.id ?? "unknown"; + finalDispatches.push(id); + if (id === "poison") { + throw new Error("final genuine failure"); + } + return null; + }), + }); + try { + await vi.waitFor(async () => { + await expect(queue.enqueue("poison", {} as DiscordIngressPayload)).resolves.toMatchObject( + { kind: "failed", record: { reason: "retry-limit-exceeded" } }, + ); + await expect( + queue.enqueue("follower", {} as DiscordIngressPayload), + ).resolves.toMatchObject({ kind: "completed" }); + }); + expect(finalDispatches).toEqual(["poison", "follower"]); + } finally { + await replacement.deactivate(); + } + }); + }); +}); diff --git a/extensions/feishu/src/monitor.message-handler.ingress.test.ts b/extensions/feishu/src/monitor.message-handler.ingress.test.ts index 1390a6c22be5..25f7308b4702 100644 --- a/extensions/feishu/src/monitor.message-handler.ingress.test.ts +++ b/extensions/feishu/src/monitor.message-handler.ingress.test.ts @@ -1,11 +1,20 @@ -import { createTestInboundDebounceFlush } from "openclaw/plugin-sdk/channel-test-helpers"; -import { createNonExitingRuntimeEnv } from "openclaw/plugin-sdk/plugin-test-runtime"; // Feishu ingress tests cover debounce ownership and constituent claim settlement. +import fs from "node:fs/promises"; +import os from "node:os"; +import path from "node:path"; +import { createInboundDebouncer } from "openclaw/plugin-sdk/channel-inbound-debounce"; +import { DEFAULT_INGRESS_RETRY_MAX_ATTEMPTS } from "openclaw/plugin-sdk/channel-outbound"; +import { createTestInboundDebounceFlush } from "openclaw/plugin-sdk/channel-test-helpers"; +import { + closeOpenClawStateDatabaseForTest, + createChannelIngressQueueForTests, +} from "openclaw/plugin-sdk/plugin-state-test-runtime"; +import { createNonExitingRuntimeEnv } from "openclaw/plugin-sdk/plugin-test-runtime"; import { afterEach, describe, expect, it, vi } from "vitest"; import type { ClawdbotConfig, PluginRuntime, RuntimeEnv } from "../runtime-api.js"; import * as dedup from "./dedup.js"; import type { FeishuMessageEvent } from "./event-types.js"; -import type { FeishuIngressLifecycle } from "./feishu-ingress.js"; +import { createFeishuDurableIngress, type FeishuIngressLifecycle } from "./feishu-ingress.js"; import { createFeishuMessageReceiveHandler } from "./monitor.message-handler.js"; type MessageReceiveHandlerContext = Parameters[0]; @@ -370,4 +379,144 @@ describe("Feishu durable ingress debounce lifecycle", () => { expect(harness.handleMessage).toHaveBeenCalledTimes(1); expect(second.calls.adopted).not.toHaveBeenCalled(); }); + + it("preserves abandon retry accounting, backoff, threshold, and restart behavior", async () => { + vi.useFakeTimers(); + const now = Date.UTC(2026, 0, 2); + vi.setSystemTime(now); + const created = await fs.mkdtemp(path.join(os.tmpdir(), "openclaw-feishu-abandon-")); + const stateDir = await fs.realpath(created); + type Queue = NonNullable[0]["queue"]>; + type Payload = Parameters[1]; + const queue = createChannelIngressQueueForTests({ + channelId: "feishu", + accountId: "default", + stateDir, + }); + const event = { + ...createTextEvent("evt-abandon-retry", "om-abandon-retry", "retry me"), + event_type: "im.message.receive_v1", + }; + const handleMessage = vi.fn(async () => { + throw new Error("Feishu dispatch failed before adoption"); + }); + vi.spyOn(dedup, "claimUnprocessedFeishuMessage").mockImplementation(async () => ({ + kind: "claimed", + handle: createClaim(`retry-${handleMessage.mock.calls.length}`), + })); + + const createIntegratedIngress = () => { + const channelRuntime = { + commands: { isControlCommandMessage: () => false }, + debounce: { + resolveInboundDebounceMs: () => 0, + createInboundDebouncer, + }, + } as unknown as PluginRuntime["channel"]; + const handler = createFeishuMessageReceiveHandler({ + cfg: {} as ClawdbotConfig, + channelRuntime, + accountId: "default", + runtime: createNonExitingRuntimeEnv(), + chatHistories: new Map(), + handleMessage, + resolveDebounceText: () => "retry me", + hasProcessedMessage: vi.fn(async () => false), + getBotOpenId: () => "ou-bot", + resolveIngressLifecycle: (data) => ingress.resolveLifecycle(data), + }); + const ingress = createFeishuDurableIngress({ + accountId: "default", + queue, + dispatcher: { invoke: async (data: unknown) => await handler(data as never) } as never, + runtime: { error: vi.fn(), log: vi.fn() }, + pollIntervalMs: 500, + }); + return ingress; + }; + const pendingAttempt = async (attempts: number) => { + let observed: Awaited>[number] | undefined; + await vi.waitFor(async () => { + const pending = await queue.listPending({ limit: "all" }); + expect(pending).toEqual([ + expect.objectContaining({ + id: "evt-abandon-retry", + attempts, + lastAttemptAt: expect.any(Number), + lastError: "turn-abandoned", + }), + ]); + observed = pending[0]; + }); + const lastAttemptAt = observed?.lastAttemptAt; + if (lastAttemptAt === undefined) { + throw new Error(`Missing Feishu retry timestamp for attempt ${attempts}`); + } + return { ...observed, lastAttemptAt }; + }; + + try { + const first = createIntegratedIngress(); + first.start(); + await first.invokeWebhook(event); + const firstAttempt = await pendingAttempt(1); + expect(handleMessage).toHaveBeenCalledTimes(1); + await first.stop(); + + vi.setSystemTime(firstAttempt.lastAttemptAt + 999); + const blocked = createIntegratedIngress(); + blocked.start(); + await blocked.invokeWebhook(event); + await vi.advanceTimersByTimeAsync(0); + expect(handleMessage).toHaveBeenCalledTimes(1); + await blocked.stop(); + + vi.setSystemTime(firstAttempt.lastAttemptAt + 1_001); + const second = createIntegratedIngress(); + second.start(); + await second.invokeWebhook(event); + const secondAttempt = await pendingAttempt(2); + expect(handleMessage).toHaveBeenCalledTimes(2); + await second.stop(); + + for (let attempt = 3; attempt < DEFAULT_INGRESS_RETRY_MAX_ATTEMPTS; attempt += 1) { + const claim = await queue.claim("evt-abandon-retry", { ownerId: `seed-${attempt}` }); + if (!claim) { + throw new Error(`Expected Feishu seed claim ${attempt}`); + } + await queue.release(claim, { + lastError: "turn-abandoned", + releasedAt: secondAttempt.lastAttemptAt, + }); + } + + vi.setSystemTime(secondAttempt.lastAttemptAt + 64_001); + const threshold = createIntegratedIngress(); + threshold.start(); + await threshold.invokeWebhook(event); + const thresholdAttempt = await pendingAttempt(DEFAULT_INGRESS_RETRY_MAX_ATTEMPTS); + expect(handleMessage).toHaveBeenCalledTimes(3); + await threshold.stop(); + + vi.setSystemTime(thresholdAttempt.lastAttemptAt + 128_001); + const beyond = createIntegratedIngress(); + beyond.start(); + await beyond.invokeWebhook(event); + const beyondAttempt = await pendingAttempt(DEFAULT_INGRESS_RETRY_MAX_ATTEMPTS + 1); + expect(handleMessage).toHaveBeenCalledTimes(4); + await beyond.stop(); + + vi.setSystemTime(beyondAttempt.lastAttemptAt + 1_000); + const blockedRestart = createIntegratedIngress(); + blockedRestart.start(); + await blockedRestart.invokeWebhook(event); + await vi.advanceTimersByTimeAsync(0); + expect(handleMessage).toHaveBeenCalledTimes(4); + await blockedRestart.stop(); + } finally { + closeOpenClawStateDatabaseForTest(); + await fs.rm(stateDir, { recursive: true, force: true }); + vi.useRealTimers(); + } + }); }); diff --git a/extensions/mattermost/src/mattermost/monitor.inbound-system-event.test.ts b/extensions/mattermost/src/mattermost/monitor.inbound-system-event.test.ts index dcb8e3830613..01d7a60b933b 100644 --- a/extensions/mattermost/src/mattermost/monitor.inbound-system-event.test.ts +++ b/extensions/mattermost/src/mattermost/monitor.inbound-system-event.test.ts @@ -1,10 +1,20 @@ // Mattermost tests cover monitor.inbound system event plugin behavior. import { once } from "node:events"; +import fs from "node:fs/promises"; import { createServer } from "node:http"; +import os from "node:os"; +import path from "node:path"; import { createChannelPartialDeliveryError } from "openclaw/plugin-sdk/channel-inbound"; import { createInboundDebouncer } from "openclaw/plugin-sdk/channel-inbound-debounce"; -import { createMessageReceiptFromOutboundResults } from "openclaw/plugin-sdk/channel-outbound"; +import { + createMessageReceiptFromOutboundResults, + DEFAULT_INGRESS_RETRY_MAX_ATTEMPTS, +} from "openclaw/plugin-sdk/channel-outbound"; import { createTestInboundDebounceFlush } from "openclaw/plugin-sdk/channel-test-helpers"; +import { + closeOpenClawStateDatabaseForTest, + createChannelIngressQueueForTests, +} from "openclaw/plugin-sdk/plugin-state-test-runtime"; import { beforeEach, describe, expect, it, vi } from "vitest"; import { WebSocketServer } from "ws"; import type { MattermostPost } from "./client.js"; @@ -96,6 +106,7 @@ const mockState = vi.hoisted(() => ({ enqueueSystemEvent: vi.fn(), fetchMattermostMe: vi.fn(), getGlobalHookRunner: vi.fn(), + ingressQueue: undefined as unknown, progressDrafts: [] as Array<{ getSnapshot: () => { lines: readonly unknown[] } }>, registerMattermostMonitorSlashCommands: vi.fn(), registerPluginHttpRoute: vi.fn(), @@ -180,27 +191,36 @@ vi.mock("./monitor-ingress.js", async (importOriginal) => { ...actual, createMattermostIngressMonitor: ( options: Parameters[0], - ) => ({ - receive: async (rawEvent: string) => { - const payload = JSON.parse(rawEvent) as MattermostEventPayload; - const post = - typeof payload.data?.post === "string" - ? (JSON.parse(payload.data.post) as MattermostPost) - : (payload.data?.post as MattermostPost | undefined); - if (payload.event !== "posted" || !post) { - return; - } - await options.dispatch(post, payload, { - abortSignal: new AbortController().signal, - onAdopted: async () => {}, - onDeferred: () => {}, - onAdoptionFinalizing: () => {}, - onAbandoned: async () => {}, + ) => { + if (mockState.ingressQueue) { + return actual.createMattermostIngressMonitor({ + ...options, + queue: mockState.ingressQueue as NonNullable, + pollIntervalMs: 60_000, }); - }, - stop: async () => {}, - waitForIdle: async () => {}, - }), + } + return { + receive: async (rawEvent: string) => { + const payload = JSON.parse(rawEvent) as MattermostEventPayload; + const post = + typeof payload.data?.post === "string" + ? (JSON.parse(payload.data.post) as MattermostPost) + : (payload.data?.post as MattermostPost | undefined); + if (payload.event !== "posted" || !post) { + return; + } + await options.dispatch(post, payload, { + abortSignal: new AbortController().signal, + onAdopted: async () => {}, + onDeferred: () => {}, + onAdoptionFinalizing: () => {}, + onAbandoned: async () => {}, + }); + }, + stop: async () => {}, + waitForIdle: async () => {}, + }; + }, }; }); @@ -534,6 +554,7 @@ describe("mattermost inbound user posts", () => { beforeEach(() => { vi.clearAllMocks(); mockState.abortController = undefined; + mockState.ingressQueue = undefined; mockState.progressDrafts.length = 0; mockState.getGlobalHookRunner.mockReturnValue(null); mockState.runtimeCore = createRuntimeCore(testConfig); @@ -568,6 +589,146 @@ describe("mattermost inbound user posts", () => { }); }); + it("preserves abandon retry accounting, backoff, threshold, and restart behavior", async () => { + vi.useFakeTimers(); + const now = Date.UTC(2026, 0, 2); + vi.setSystemTime(now); + const created = await fs.mkdtemp(path.join(os.tmpdir(), "openclaw-mattermost-abandon-")); + const stateDir = await fs.realpath(created); + type Payload = { version: 1; receivedAt: number; rawEvent: string }; + const queue = createChannelIngressQueueForTests({ + channelId: "mattermost", + accountId: "default", + stateDir, + }); + mockState.ingressQueue = queue; + mockState.runtimeCore = createRuntimeCore(testConfig, undefined, { + inboundDebounceMs: 0, + createInboundDebouncer, + }); + mockState.dispatchInboundMessage.mockRejectedValue( + new Error("Mattermost dispatch failed before adoption"), + ); + + const activeProviders: Array<{ stop: () => Promise }> = []; + const startProvider = async () => { + const socket = new FakeWebSocket(); + const abortController = new AbortController(); + const monitor = monitorMattermostProvider({ + config: testConfig, + runtime: testRuntime(), + abortSignal: abortController.signal, + webSocketFactory: () => socket, + }); + for (let tick = 0; tick < 20 && socket.openListenerCount === 0; tick += 1) { + await Promise.resolve(); + } + expect(socket.openListenerCount).toBeGreaterThan(0); + socket.emitOpen(); + let stopped = false; + const provider = { + socket, + stop: async () => { + if (stopped) { + return; + } + stopped = true; + abortController.abort(); + socket.emitClose(1000); + await monitor; + }, + }; + activeProviders.push(provider); + return provider; + }; + const send = async (provider: Awaited>) => { + await emitMattermostChannelPost(provider.socket, { + id: "post-abandon-retry", + message: "retry me", + }); + }; + const pendingAttempt = async (attempts: number) => { + let observed: Awaited>[number] | undefined; + await vi.waitFor(async () => { + const pending = await queue.listPending({ limit: "all" }); + expect(pending).toEqual([ + expect.objectContaining({ + id: "post-abandon-retry", + attempts, + lastAttemptAt: expect.any(Number), + lastError: "turn-abandoned", + }), + ]); + observed = pending[0]; + }); + const lastAttemptAt = observed?.lastAttemptAt; + if (lastAttemptAt === undefined) { + throw new Error(`Missing Mattermost retry timestamp for attempt ${attempts}`); + } + return { ...observed, lastAttemptAt }; + }; + + try { + const first = await startProvider(); + await send(first); + const firstAttempt = await pendingAttempt(1); + expect(mockState.dispatchInboundMessage).toHaveBeenCalledTimes(1); + await first.stop(); + + vi.setSystemTime(firstAttempt.lastAttemptAt + 999); + const blocked = await startProvider(); + await send(blocked); + await vi.advanceTimersByTimeAsync(0); + expect(mockState.dispatchInboundMessage).toHaveBeenCalledTimes(1); + await blocked.stop(); + + vi.setSystemTime(firstAttempt.lastAttemptAt + 1_001); + const second = await startProvider(); + await send(second); + const secondAttempt = await pendingAttempt(2); + expect(mockState.dispatchInboundMessage).toHaveBeenCalledTimes(2); + await second.stop(); + + for (let attempt = 3; attempt < DEFAULT_INGRESS_RETRY_MAX_ATTEMPTS; attempt += 1) { + const claim = await queue.claim("post-abandon-retry", { ownerId: `seed-${attempt}` }); + if (!claim) { + throw new Error(`Expected Mattermost seed claim ${attempt}`); + } + await queue.release(claim, { + lastError: "turn-abandoned", + releasedAt: secondAttempt.lastAttemptAt, + }); + } + + vi.setSystemTime(secondAttempt.lastAttemptAt + 64_001); + const threshold = await startProvider(); + await send(threshold); + const thresholdAttempt = await pendingAttempt(DEFAULT_INGRESS_RETRY_MAX_ATTEMPTS); + expect(mockState.dispatchInboundMessage).toHaveBeenCalledTimes(3); + await threshold.stop(); + + vi.setSystemTime(thresholdAttempt.lastAttemptAt + 128_001); + const beyond = await startProvider(); + await send(beyond); + const beyondAttempt = await pendingAttempt(DEFAULT_INGRESS_RETRY_MAX_ATTEMPTS + 1); + expect(mockState.dispatchInboundMessage).toHaveBeenCalledTimes(4); + await beyond.stop(); + + vi.setSystemTime(beyondAttempt.lastAttemptAt + 1_000); + const blockedRestart = await startProvider(); + await send(blockedRestart); + await vi.advanceTimersByTimeAsync(0); + expect(mockState.dispatchInboundMessage).toHaveBeenCalledTimes(4); + await blockedRestart.stop(); + } finally { + await Promise.allSettled(activeProviders.map(async (provider) => await provider.stop())); + mockState.ingressQueue = undefined; + closeOpenClawStateDatabaseForTest(); + await fs.rm(stateDir, { recursive: true, force: true }); + vi.useRealTimers(); + } + }); + it("publishes recovering while API authentication retries, including 401", async () => { const abortController = new AbortController(); const statusSink = vi.fn(); diff --git a/extensions/msteams/src/monitor-handler/message-handler.ingress-lifecycle.test.ts b/extensions/msteams/src/monitor-handler/message-handler.ingress-lifecycle.test.ts index 662217fb5ccc..1c9c5afc4031 100644 --- a/extensions/msteams/src/monitor-handler/message-handler.ingress-lifecycle.test.ts +++ b/extensions/msteams/src/monitor-handler/message-handler.ingress-lifecycle.test.ts @@ -1,7 +1,16 @@ // Microsoft Teams tests cover durable claim ownership through inbound debounce. +import fs from "node:fs/promises"; +import os from "node:os"; +import path from "node:path"; import { createInboundDebouncer } from "openclaw/plugin-sdk/channel-inbound-debounce"; +import { DEFAULT_INGRESS_RETRY_MAX_ATTEMPTS } from "openclaw/plugin-sdk/channel-outbound"; +import { + closeOpenClawStateDatabaseForTest, + createChannelIngressQueueForTests, +} from "openclaw/plugin-sdk/plugin-state-test-runtime"; import { beforeEach, describe, expect, it, vi } from "vitest"; import type { OpenClawConfig } from "../../runtime-api.js"; +import { createMSTeamsIngress } from "../msteams-ingress.js"; import type { MSTeamsIngressLifecycle } from "../msteams-ingress.js"; import type { MSTeamsTurnContext } from "../sdk-types.js"; import "./message-handler-mock-support.test-support.js"; @@ -156,4 +165,125 @@ describe("Microsoft Teams drain claim ownership", () => { expect(runtimeApiMockState.dispatchReplyWithBufferedBlockDispatcher).not.toHaveBeenCalled(); expect(lifecycle.abandonedCount()).toBe(0); }); + + it("preserves abandon retry accounting, backoff, threshold, and restart behavior", async () => { + vi.useFakeTimers(); + const now = Date.UTC(2026, 0, 2); + vi.setSystemTime(now); + const created = await fs.mkdtemp(path.join(os.tmpdir(), "openclaw-msteams-abandon-")); + const stateDir = await fs.realpath(created); + type Queue = NonNullable[0]["queue"]>; + type Payload = Parameters[1]; + const queue = createChannelIngressQueueForTests({ + channelId: "msteams", + accountId: "test-app", + stateDir, + }); + const incoming = directActivity("activity-abandon", "retry me"); + await queue.enqueue( + "activity-abandon", + { version: 1, receivedAt: now - 2 * 24 * 60 * 60_000, rawActivity: JSON.stringify(incoming) }, + { laneKey: "dm-conversation", receivedAt: now - 2 * 24 * 60 * 60_000 }, + ); + const dispatchMock = runtimeApiMockState.dispatchReplyWithBufferedBlockDispatcher; + const priorImplementation = dispatchMock.getMockImplementation(); + dispatchMock.mockRejectedValue(new Error("Microsoft Teams dispatch failed before adoption")); + + const createIntegratedIngress = () => { + const handler = createHandler({ + channels: { msteams: { dmPolicy: "open", allowFrom: ["*"] } }, + } as OpenClawConfig); + return createMSTeamsIngress({ + accountId: "test-app", + queue, + runtime: { error: vi.fn(), log: vi.fn() }, + dispatch: async (activity, lifecycle) => await handler(context(activity), lifecycle), + }); + }; + const expectPendingAttempt = async (attempts: number) => { + let observed: Awaited>[number] | undefined; + await vi.waitFor(async () => { + const pending = await queue.listPending({ limit: "all" }); + expect(pending).toEqual([ + expect.objectContaining({ + id: "activity-abandon", + attempts, + lastAttemptAt: expect.any(Number), + lastError: "turn-abandoned", + }), + ]); + observed = pending[0]; + }); + const lastAttemptAt = observed?.lastAttemptAt; + if (lastAttemptAt === undefined) { + throw new Error(`Missing Microsoft Teams retry timestamp for attempt ${attempts}`); + } + return { ...observed, lastAttemptAt }; + }; + + try { + const first = createIntegratedIngress(); + first.start(); + const firstAttempt = await expectPendingAttempt(1); + expect(dispatchMock).toHaveBeenCalledTimes(1); + await first.stop(); + + vi.setSystemTime(firstAttempt.lastAttemptAt + 999); + const second = createIntegratedIngress(); + second.start(); + await second.accept(incoming); + await vi.advanceTimersByTimeAsync(0); + expect(dispatchMock).toHaveBeenCalledTimes(1); + await second.stop(); + vi.setSystemTime(firstAttempt.lastAttemptAt + 1_001); + const afterBackoff = createIntegratedIngress(); + afterBackoff.start(); + await afterBackoff.accept(incoming); + const secondAttempt = await expectPendingAttempt(2); + expect(dispatchMock).toHaveBeenCalledTimes(2); + await afterBackoff.stop(); + + for (let attempt = 3; attempt < DEFAULT_INGRESS_RETRY_MAX_ATTEMPTS; attempt += 1) { + const claim = await queue.claim("activity-abandon", { ownerId: `seed-${attempt}` }); + if (!claim) { + throw new Error(`Expected Microsoft Teams seed claim ${attempt}`); + } + await queue.release(claim, { + lastError: "turn-abandoned", + releasedAt: secondAttempt.lastAttemptAt, + }); + } + vi.setSystemTime(secondAttempt.lastAttemptAt + 64_001); + const threshold = createIntegratedIngress(); + threshold.start(); + await threshold.accept(incoming); + const thresholdAttempt = await expectPendingAttempt(DEFAULT_INGRESS_RETRY_MAX_ATTEMPTS); + expect(dispatchMock).toHaveBeenCalledTimes(3); + await threshold.stop(); + + vi.setSystemTime(thresholdAttempt.lastAttemptAt + 128_001); + const beyond = createIntegratedIngress(); + beyond.start(); + await beyond.accept(incoming); + const beyondAttempt = await expectPendingAttempt(DEFAULT_INGRESS_RETRY_MAX_ATTEMPTS + 1); + expect(dispatchMock).toHaveBeenCalledTimes(4); + await beyond.stop(); + + vi.setSystemTime(beyondAttempt.lastAttemptAt + 1_000); + const blockedRestart = createIntegratedIngress(); + blockedRestart.start(); + await blockedRestart.accept(incoming); + await vi.advanceTimersByTimeAsync(0); + expect(dispatchMock).toHaveBeenCalledTimes(4); + await blockedRestart.stop(); + } finally { + dispatchMock.mockReset(); + if (priorImplementation) { + dispatchMock.mockImplementation(priorImplementation); + } + closeOpenClawStateDatabaseForTest(); + await fs.rm(stateDir, { recursive: true, force: true }); + vi.useRealTimers(); + } + }); }); diff --git a/extensions/signal/src/monitor/event-handler.reply-session-conflict.test.ts b/extensions/signal/src/monitor/event-handler.reply-session-conflict.test.ts index 81b079006673..00cd07e05580 100644 --- a/extensions/signal/src/monitor/event-handler.reply-session-conflict.test.ts +++ b/extensions/signal/src/monitor/event-handler.reply-session-conflict.test.ts @@ -1,6 +1,15 @@ -import type { OpenClawConfig } from "openclaw/plugin-sdk/config-contracts"; // Signal tests cover retry behavior for reply session initialization conflicts. +import fs from "node:fs/promises"; +import os from "node:os"; +import path from "node:path"; +import { DEFAULT_INGRESS_RETRY_MAX_ATTEMPTS } from "openclaw/plugin-sdk/channel-outbound"; +import type { OpenClawConfig } from "openclaw/plugin-sdk/config-contracts"; +import { + closeOpenClawStateDatabaseForTest, + createChannelIngressQueueForTests, +} from "openclaw/plugin-sdk/plugin-state-test-runtime"; import { beforeEach, describe, expect, it, vi } from "vitest"; +import { startSignalIngressMonitor } from "../signal-ingress.js"; import type { SignalEventHandlerDeps } from "./event-handler.types.js"; const [ @@ -326,6 +335,127 @@ describe("signal reply session init conflict retry", () => { } }); + it("preserves durable abandon accounting through backoff, threshold, and restart", async () => { + vi.useFakeTimers(); + const now = Date.UTC(2026, 0, 2); + vi.setSystemTime(now); + const created = await fs.mkdtemp(path.join(os.tmpdir(), "openclaw-signal-abandon-")); + const stateDir = await fs.realpath(created); + type Queue = NonNullable[0]["queue"]>; + type Payload = Parameters[1]; + const queue = createChannelIngressQueueForTests({ + channelId: "signal", + accountId: "default", + stateDir, + }); + const timestamp = 1_700_000_000_777; + const event = createSignalReceiveEvent({ + timestamp, + dataMessage: { timestamp, message: "retry through durable ingress", attachments: [] }, + }); + const eventId = JSON.stringify(["number:+15550001111", timestamp]); + dispatchInboundMessageMock.mockRejectedValue(CONFLICT_ERROR); + + const createIntegratedMonitor = async () => { + const tracked = createTrackedTaskHarness(); + const handler = createSignalEventHandler( + createBaseSignalEventHandlerDeps({ + cfg: { messages: { inbound: { debounceMs: 10 } } }, + runTrackedTask: tracked.runTrackedTask, + }), + ); + const monitor = await startSignalIngressMonitor({ + accountId: "default", + queue, + dispatch: async (incoming, lifecycle) => await handler(incoming, lifecycle), + runtime: { error: vi.fn(), log: vi.fn() }, + }); + return { monitor, tracked }; + }; + const finishOuterAttempt = async (tracked: ReturnType) => { + await vi.advanceTimersByTimeAsync(10); + expect(tracked.tasks).toHaveLength(1); + await vi.advanceTimersByTimeAsync(7_000); + await Promise.all(tracked.tasks); + }; + const pendingAttempt = async (attempts: number) => { + const pending = await queue.listPending({ limit: "all" }); + expect(pending).toEqual([ + expect.objectContaining({ + id: eventId, + attempts, + lastAttemptAt: expect.any(Number), + lastError: "turn-abandoned", + }), + ]); + const record = pending[0]; + const lastAttemptAt = record?.lastAttemptAt; + if (lastAttemptAt === undefined) { + throw new Error(`Missing Signal retry timestamp for attempt ${attempts}`); + } + return { ...record, lastAttemptAt }; + }; + + try { + const first = await createIntegratedMonitor(); + await first.monitor.receive(event); + await finishOuterAttempt(first.tracked); + const firstAttempt = await pendingAttempt(1); + expect(dispatchInboundMessageMock).toHaveBeenCalledTimes(4); + await first.monitor.stop(); + + vi.setSystemTime(firstAttempt.lastAttemptAt + 999); + const blocked = await createIntegratedMonitor(); + await vi.advanceTimersByTimeAsync(10); + expect(dispatchInboundMessageMock).toHaveBeenCalledTimes(4); + expect(blocked.tracked.tasks).toHaveLength(0); + await blocked.monitor.stop(); + + vi.setSystemTime(firstAttempt.lastAttemptAt + 1_001); + const second = await createIntegratedMonitor(); + await finishOuterAttempt(second.tracked); + const secondAttempt = await pendingAttempt(2); + expect(dispatchInboundMessageMock).toHaveBeenCalledTimes(8); + await second.monitor.stop(); + + for (let attempt = 3; attempt < DEFAULT_INGRESS_RETRY_MAX_ATTEMPTS; attempt += 1) { + const claim = await queue.claim(eventId, { ownerId: `seed-${attempt}` }); + if (!claim) { + throw new Error(`Expected Signal seed claim ${attempt}`); + } + await queue.release(claim, { + lastError: "turn-abandoned", + releasedAt: secondAttempt.lastAttemptAt, + }); + } + + vi.setSystemTime(secondAttempt.lastAttemptAt + 64_001); + const threshold = await createIntegratedMonitor(); + await finishOuterAttempt(threshold.tracked); + const thresholdAttempt = await pendingAttempt(DEFAULT_INGRESS_RETRY_MAX_ATTEMPTS); + expect(dispatchInboundMessageMock).toHaveBeenCalledTimes(12); + await threshold.monitor.stop(); + + vi.setSystemTime(thresholdAttempt.lastAttemptAt + 128_001); + const beyond = await createIntegratedMonitor(); + await finishOuterAttempt(beyond.tracked); + const beyondAttempt = await pendingAttempt(DEFAULT_INGRESS_RETRY_MAX_ATTEMPTS + 1); + expect(dispatchInboundMessageMock).toHaveBeenCalledTimes(16); + await beyond.monitor.stop(); + + vi.setSystemTime(beyondAttempt.lastAttemptAt + 1_000); + const blockedRestart = await createIntegratedMonitor(); + await vi.advanceTimersByTimeAsync(10); + expect(dispatchInboundMessageMock).toHaveBeenCalledTimes(16); + expect(blockedRestart.tracked.tasks).toHaveLength(0); + await blockedRestart.monitor.stop(); + } finally { + closeOpenClawStateDatabaseForTest(); + await fs.rm(stateDir, { recursive: true, force: true }); + vi.useRealTimers(); + } + }); + it("does not retry non-conflict flush failures", async () => { dispatchInboundMessageMock.mockRejectedValue(new Error("some other dispatch failure")); diff --git a/src/channels/message/ingress-queue.test.ts b/src/channels/message/ingress-queue.test.ts index 79dce21b52ea..8fd517b11c1a 100644 --- a/src/channels/message/ingress-queue.test.ts +++ b/src/channels/message/ingress-queue.test.ts @@ -119,32 +119,66 @@ describe("channel ingress queue", () => { it("keeps channel and account queue identities unambiguous", async () => { await withTempState(async (stateDir) => { const first = createChannelIngressQueue<{ text: string }>({ - channelId: "a", - accountId: "b:c", + channelId: "discord", + accountId: "account-a", stateDir, }); const second = createChannelIngressQueue<{ text: string }>({ - channelId: "a:b", - accountId: "c", + channelId: "discord", + accountId: "account-b", stateDir, }); - expect(await first.enqueue("same-id", { text: "first" })).toMatchObject({ + expect( + await first.enqueue("same-id", { text: "first" }, { laneKey: "channel:same-lane" }), + ).toMatchObject({ kind: "accepted", }); - expect(await second.enqueue("same-id", { text: "second" })).toMatchObject({ + expect( + await second.enqueue("same-id", { text: "second" }, { laneKey: "channel:same-lane" }), + ).toMatchObject({ kind: "accepted", }); - await first.complete("same-id"); + const firstClaim = await first.claim("same-id", { ownerId: "first-worker" }); + expect(firstClaim).not.toBeNull(); + if (!firstClaim) { + return; + } + await first.fail(firstClaim, { reason: "poison", failedAt: 20 }); expect(await first.enqueue("same-id", { text: "first duplicate" })).toMatchObject({ - kind: "completed", + kind: "failed", }); expect(await second.enqueue("same-id", { text: "second duplicate" })).toMatchObject({ kind: "pending", record: { payload: { text: "second" } }, }); + + if (!first.resubmit) { + return; + } + await expect(first.resubmit("same-id", { resubmittedAt: 30 })).resolves.toMatchObject({ + kind: "resubmitted", + record: { attempts: 0, laneKey: "channel:same-lane", payload: { text: "first" } }, + }); + const resubmittedClaim = await first.claim("same-id", { ownerId: "replacement" }); + const secondClaim = await second.claim("same-id", { ownerId: "second-worker" }); + expect(resubmittedClaim).not.toBeNull(); + expect(secondClaim).not.toBeNull(); + if (!resubmittedClaim || !secondClaim) { + return; + } + await first.fail(resubmittedClaim, { reason: "poison-again", failedAt: 40 }); + await second.complete(secondClaim, { completedAt: 40 }); + + expect(await first.prune({ failedTtlMs: 1, now: 42 })).toBe(1); + expect(await first.enqueue("same-id", { text: "fresh after prune" })).toMatchObject({ + kind: "accepted", + }); + expect(await second.enqueue("same-id", { text: "completed duplicate" })).toMatchObject({ + kind: "completed", + }); }); }); diff --git a/src/plugin-sdk/channel-ingress-runtime.test.ts b/src/plugin-sdk/channel-ingress-runtime.test.ts index 786f2848691a..393b84df0d0d 100644 --- a/src/plugin-sdk/channel-ingress-runtime.test.ts +++ b/src/plugin-sdk/channel-ingress-runtime.test.ts @@ -76,6 +76,27 @@ describe("plugin-sdk/channel-ingress-runtime", () => { expect(fanInChannelIngressLifecycles([]).lifecycle).toBeUndefined(); }); + it("does not expose cancellation when a source cannot cancellation-settle", async () => { + const adopted = vi.fn(async () => {}); + const cancelled = vi.fn(async () => {}); + const createLifecycle = (onCancelled?: () => Promise) => ({ + abortSignal: new AbortController().signal, + onAdopted: adopted, + onDeferred: vi.fn(), + onAdoptionFinalizing: vi.fn(), + onFailed: vi.fn(async () => {}), + onAbandoned: vi.fn(async () => {}), + ...(onCancelled ? { onCancelled } : {}), + }); + const combined = fanInChannelIngressLifecycles([createLifecycle(cancelled), createLifecycle()]); + + expect(combined.lifecycle).not.toHaveProperty("onCancelled"); + await combined.settle(); + + expect(adopted).toHaveBeenCalledTimes(2); + expect(cancelled).not.toHaveBeenCalled(); + }); + it("can abandon claims after terminal settlement adoption fails", async () => { const abandoned = vi.fn(async () => {}); const combined = fanInChannelIngressLifecycles([ diff --git a/src/plugin-sdk/channel-ingress-runtime.ts b/src/plugin-sdk/channel-ingress-runtime.ts index 480b01bbac06..262ee639849e 100644 --- a/src/plugin-sdk/channel-ingress-runtime.ts +++ b/src/plugin-sdk/channel-ingress-runtime.ts @@ -151,6 +151,18 @@ export function fanInChannelIngressLifecycles( const failAll = async (error: unknown) => { await Promise.all(lifecycles.map(async (lifecycle) => await lifecycle.onFailed?.(error))); }; + const cancellationHandlers = lifecycles.flatMap((lifecycle) => + lifecycle.onCancelled ? [lifecycle.onCancelled] : [], + ); + // Omit aggregate cancellation unless every durable source supports it. Callers + // can then use settle/abandon without an acknowledged-but-unsettled claim. + const cancelAll = + cancellationHandlers.length === lifecycles.length + ? async () => { + handedOff = true; + await Promise.all(cancellationHandlers.map(async (cancel) => await cancel())); + } + : undefined; return { lifecycle: { abortSignal: @@ -176,10 +188,7 @@ export function fanInChannelIngressLifecycles( handedOff = true; await failAll(error); }, - onCancelled: async () => { - handedOff = true; - await Promise.all(lifecycles.map(async (lifecycle) => await lifecycle.onCancelled?.())); - }, + ...(cancelAll ? { onCancelled: cancelAll } : {}), onAbandoned: async () => { handedOff = true; await abandonAll();