From 10ea57d2884eb5a03205521924b7b170de7442d3 Mon Sep 17 00:00:00 2001 From: joshavant <830519+joshavant@users.noreply.github.com> Date: Wed, 12 Aug 2026 15:27:12 -0500 Subject: [PATCH] fix(discord): unblock ingress after retry exhaustion --- docs/plugins/sdk-channel-outbound.md | 16 +- .../discord/src/monitor/ingress.test.ts | 25 ++ extensions/discord/src/monitor/ingress.ts | 5 + .../discord/src/monitor/message-dispatcher.ts | 17 +- .../src/monitor/message-handler.queue.test.ts | 229 +++++++++++++++++- src/channels/message/ingress-drain.test.ts | 181 ++++++++++++++ src/channels/message/ingress-drain.ts | 45 ++-- src/channels/message/ingress-monitor.ts | 34 ++- .../channel-ingress-runtime.test.ts | 5 + src/plugin-sdk/channel-ingress-runtime.ts | 11 +- 10 files changed, 512 insertions(+), 56 deletions(-) diff --git a/docs/plugins/sdk-channel-outbound.md b/docs/plugins/sdk-channel-outbound.md index d701b7be096b..7ce7198e442e 100644 --- a/docs/plugins/sdk-channel-outbound.md +++ b/docs/plugins/sdk-channel-outbound.md @@ -47,13 +47,15 @@ the transport callback instead of dispatching an event that was not made durable. At claim time it decodes the versioned payload, re-runs `inspect`, and rejects an id or lane mismatch before delivery. -`deliver` receives `onAdopted`, `onDeferred`, `onAdoptionFinalizing`, -`onAbandoned`, and `abortSignal`. Returning without an explicit handoff marks a -terminal no-dispatch event adopted. `admission` is always `exclusive`. A -deferred handoff keeps the claim held, while shutdown or abort leaves unadopted -work retryable. The monitor tracks delivery independently from claim settlement -because adoption can tombstone a row before the channel's delivery promise -returns. +`deliver` receives `onAdopted`, `onDeferred`, `onAdoptionFinalizing`, `onFailed`, +`onCancelled`, `onAbandoned`, and `abortSignal`. Use `onFailed` for delivery +errors, `onCancelled` for explicit pre-adoption cancellation that must preserve +retry accounting, and `onAbandoned` when a non-adopted turn should consume a +retry attempt. Returning without an explicit handoff marks a terminal +no-dispatch event adopted. `admission` is always `exclusive`. A deferred handoff +keeps the claim held, while shutdown or abort leaves unadopted work retryable. +The monitor tracks delivery independently from claim settlement because +adoption can tombstone a row before the channel's delivery promise returns. Optional settings include custom append delays, a `drain` option block for advanced drain ordering/concurrency/retry policy, an external `abortSignal`, a diff --git a/extensions/discord/src/monitor/ingress.test.ts b/extensions/discord/src/monitor/ingress.test.ts index 18ab32ade29e..ccf7e6fee135 100644 --- a/extensions/discord/src/monitor/ingress.test.ts +++ b/extensions/discord/src/monitor/ingress.test.ts @@ -116,6 +116,31 @@ describe("Discord durable ingress", () => { }); }); + it("rejects unstable message identity before durable allocation", async () => { + await withQueue(async (queue) => { + const dispatch = vi.fn(); + const monitor = createDiscordIngressMonitor({ + accountId: "default", + client: {} as never, + runtime: runtime(), + queue, + dispatch, + }); + monitor.start(); + try { + const missingMessageId = { ...createRawMessage("missing"), id: undefined }; + const missingChannelId = { ...createRawMessage("missing"), channel_id: undefined }; + + await expect(monitor.accept(missingMessageId as never)).rejects.toThrow("snowflake"); + await expect(monitor.accept(missingChannelId as never)).rejects.toThrow("channel_id"); + expect(await queue.listPending({ limit: "all" })).toEqual([]); + expect(dispatch).not.toHaveBeenCalled(); + } finally { + await monitor.stop(); + } + }); + }); + it("recovers a claimed row with a fresh drain and dispatches it exactly once", async () => { await withQueue(async (queue) => { const monitors: DiscordIngressMonitor[] = []; diff --git a/extensions/discord/src/monitor/ingress.ts b/extensions/discord/src/monitor/ingress.ts index c5052bf539bc..a2c438a77774 100644 --- a/extensions/discord/src/monitor/ingress.ts +++ b/extensions/discord/src/monitor/ingress.ts @@ -3,6 +3,7 @@ import { GatewayDispatchEvents, type APIMessage } from "discord-api-types/v10"; import { createChannelIngressError, createChannelIngressMonitor, + DEFAULT_INGRESS_RETRY_MAX_ATTEMPTS, type ChannelIngressQueue, type ChannelIngressMonitorDeliveryResult, type ChannelIngressMonitorLifecycle, @@ -147,6 +148,10 @@ export function createDiscordIngressMonitor(params: { }, appendRetryDelaysMs: [0], drain: { + retryPolicy: { + maxAttempts: DEFAULT_INGRESS_RETRY_MAX_ATTEMPTS, + deadLetterMinAgeMs: 0, + }, resolveNonRetryableFailure: (error) => { if (error instanceof DiscordIngressPayloadError) { return { reason: "invalid-event", message: error.message }; diff --git a/extensions/discord/src/monitor/message-dispatcher.ts b/extensions/discord/src/monitor/message-dispatcher.ts index 32d284409c83..c7bbb504d897 100644 --- a/extensions/discord/src/monitor/message-dispatcher.ts +++ b/extensions/discord/src/monitor/message-dispatcher.ts @@ -139,7 +139,7 @@ export function createDiscordMessageDispatcher( } const abortSignal = last.abortSignal; if (abortSignal?.aborted) { - await admissionLifecycle.onAbandoned(); + await ingress.lifecycle?.onCancelled?.(); return; } try { @@ -157,7 +157,7 @@ export function createDiscordMessageDispatcher( turnAdoptionLifecycle: admissionLifecycle, }); if (abortSignal?.aborted) { - await ingress.abandon(abortSignal.reason); + await ingress.lifecycle?.onCancelled?.(); return; } if (!ctx) { @@ -211,7 +211,7 @@ export function createDiscordMessageDispatcher( turnAdoptionLifecycle: admissionLifecycle, }); if (abortSignal?.aborted) { - await ingress.abandon(abortSignal.reason); + await ingress.lifecycle?.onCancelled?.(); return; } if (!ctx) { @@ -232,7 +232,10 @@ export function createDiscordMessageDispatcher( } messageRunQueue.enqueue(buildDiscordInboundJob(ctx, { ingressSettlement: ingress })); } catch (error) { - await admissionLifecycle.onAbandoned(); + if (abortSignal?.aborted) { + await ingress.lifecycle?.onCancelled?.(); + return; + } throw error; } }, @@ -244,7 +247,7 @@ export function createDiscordMessageDispatcher( onCancel: (entries) => { for (const entry of entries) { pendingDebounceEntries.delete(entry); - const settlement = Promise.resolve(entry.turnAdoptionLifecycle?.onAbandoned()) + const settlement = Promise.resolve(entry.turnAdoptionLifecycle?.onCancelled?.()) .catch((error: unknown) => { params.runtime.error( danger(`discord ingress cancellation settlement failed: ${String(error)}`), @@ -271,6 +274,10 @@ export function createDiscordMessageDispatcher( const reason = dispatcherShutdown.signal.aborted ? (dispatcherShutdown.signal.reason ?? new Error("discord dispatcher shut down")) : (options?.abortSignal?.reason ?? new Error("discord dispatch aborted")); + if (options?.turnAdoptionLifecycle?.onCancelled) { + await options.turnAdoptionLifecycle.onCancelled(); + return { kind: "deferred" }; + } return { kind: "failed-retryable", error: reason }; } // Filter bot-own messages before they enter the debounce queue. diff --git a/extensions/discord/src/monitor/message-handler.queue.test.ts b/extensions/discord/src/monitor/message-handler.queue.test.ts index d6c66a8baef2..6d17ec88634d 100644 --- a/extensions/discord/src/monitor/message-handler.queue.test.ts +++ b/extensions/discord/src/monitor/message-handler.queue.test.ts @@ -1,9 +1,21 @@ // Discord tests cover message handler.queue plugin behavior. import { getEventListeners } from "node:events"; +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 type { OpenClawConfig } from "openclaw/plugin-sdk/config-contracts"; import { createDeferred } from "openclaw/plugin-sdk/extension-shared"; +import { + closeOpenClawStateDatabaseForTest, + createChannelIngressQueueForTests, +} from "openclaw/plugin-sdk/plugin-state-test-runtime"; import { beforeEach, describe, expect, it, vi } from "vitest"; -import type { DiscordIngressLifecycle } from "./ingress.js"; +import { createDiscordIngressMonitor, type DiscordIngressLifecycle } from "./ingress.js"; import { createDiscordMessageHandler as createDurableDiscordMessageHandler } from "./message-handler.js"; import { createDiscordMessageHandler, @@ -35,6 +47,8 @@ function expectStatusPatch(setStatus: MockCallSource, expected: Record; + onFailed: ReturnType; + onCancelled: ReturnType; onAbandoned: ReturnType; } { return { @@ -42,10 +56,61 @@ function createIngressLifecycle(): DiscordIngressLifecycle & { onAdopted: vi.fn(async () => {}), onDeferred: vi.fn(), onAdoptionFinalizing: vi.fn(), + onFailed: vi.fn(async () => {}), + onCancelled: vi.fn(async () => {}), onAbandoned: vi.fn(async () => {}), }; } +type DiscordIngressPayload = { + version: 1; + receivedAt: number; + rawMessage: APIMessage; +}; + +async function withDiscordQueue( + run: (queue: ChannelIngressQueue) => Promise, +): Promise { + const created = await fs.mkdtemp(path.join(os.tmpdir(), "openclaw-discord-handler-")); + const stateDir = await fs.realpath(created); + const queue = createChannelIngressQueueForTests({ + channelId: "discord", + accountId: "default", + stateDir, + }); + try { + return await run(queue); + } finally { + closeOpenClawStateDatabaseForTest(); + await fs.rm(stateDir, { recursive: true, force: true }); + } +} + +function createRawMessage(id: string, channelId = "ch-1"): 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().toISOString(), + edited_timestamp: null, + components: [], + pinned: false, + type: 0, + tts: false, + } as unknown as APIMessage; +} + async function flushQueueWork(): Promise { for (let i = 0; i < 40; i += 1) { await Promise.resolve(); @@ -314,11 +379,31 @@ describe("createDiscordMessageHandler queue behavior", () => { turnAdoptionLifecycle: lifecycle, }); - expect(result).toMatchObject({ kind: "failed-retryable" }); + expect(result).toMatchObject({ kind: "deferred" }); + expect(lifecycle.onCancelled).toHaveBeenCalledTimes(1); expect(lifecycle.onAdopted).not.toHaveBeenCalled(); }); - it("abandons a buffered ingress claim during deactivation", async () => { + it("reports a genuine pre-admission exception only through onFailed", async () => { + preflightDiscordMessageMock.mockReset(); + processDiscordMessageMock.mockReset(); + const failure = new Error("preflight failed"); + preflightDiscordMessageMock.mockRejectedValue(failure); + const handler = createDiscordMessageHandler(createDiscordHandlerParams()); + const lifecycle = createIngressLifecycle(); + + await expect( + handler(createTextMessageData("m-failed") as never, {} as never, { + turnAdoptionLifecycle: lifecycle, + }), + ).resolves.toEqual({ kind: "deferred" }); + + expect(lifecycle.onFailed).toHaveBeenCalledExactlyOnceWith(failure); + expect(lifecycle.onCancelled).not.toHaveBeenCalled(); + expect(lifecycle.onAbandoned).not.toHaveBeenCalled(); + }); + + it("cancels a buffered ingress claim during deactivation", async () => { preflightDiscordMessageMock.mockReset(); processDiscordMessageMock.mockReset(); const params = createDiscordHandlerParams(); @@ -332,11 +417,12 @@ describe("createDiscordMessageHandler queue behavior", () => { await handler.deactivate(); expect(preflightDiscordMessageMock).not.toHaveBeenCalled(); - expect(lifecycle.onAbandoned).toHaveBeenCalledTimes(1); + expect(lifecycle.onCancelled).toHaveBeenCalledTimes(1); + expect(lifecycle.onAbandoned).not.toHaveBeenCalled(); expect(lifecycle.onAdopted).not.toHaveBeenCalled(); }); - it("waits for an active debounce flush and abandons it after shutdown", async () => { + it("waits for an active debounce flush and cancels it after shutdown", async () => { preflightDiscordMessageMock.mockReset(); processDiscordMessageMock.mockReset(); const preflightGate = createDeferred(); @@ -360,7 +446,8 @@ describe("createDiscordMessageHandler queue behavior", () => { preflightGate.resolve(); await Promise.all([handling, deactivation]); - expect(lifecycle.onAbandoned).toHaveBeenCalledTimes(1); + expect(lifecycle.onCancelled).toHaveBeenCalledTimes(1); + expect(lifecycle.onAbandoned).not.toHaveBeenCalled(); expect(lifecycle.onAdopted).not.toHaveBeenCalled(); }); @@ -394,6 +481,136 @@ describe("createDiscordMessageHandler queue behavior", () => { expect(stop).toHaveBeenCalledTimes(1); }); + it("dead-letters an exhausted preflight failure and releases its Discord lane", async () => { + vi.useFakeTimers(); + vi.setSystemTime(new Date("2026-01-01T00:00:00.000Z")); + try { + await withDiscordQueue(async (queue) => { + const attempted: string[] = []; + const preflight = vi.fn(async (params: { data: { message?: { id?: string } } }) => { + const id = params.data.message?.id ?? "unknown"; + attempted.push(id); + if (id === "poison") { + throw new Error("deterministic preflight failure"); + } + return null; + }); + const params = createDiscordHandlerParams(); + const handler = createDurableDiscordMessageHandler({ + ...params, + client: {} as never, + testing: { + preflightDiscordMessage: preflight as never, + createIngressMonitor: (monitorParams) => + createDiscordIngressMonitor({ ...monitorParams, queue }), + }, + }); + try { + await handler(createRawMessage("poison", "lane-a") as never, {} as never); + await handler(createRawMessage("follower", "lane-a") as never, {} as never); + await handler(createRawMessage("independent", "lane-b") as never, {} as never); + + for (let attempt = 0; attempt < DEFAULT_INGRESS_RETRY_MAX_ATTEMPTS; attempt += 1) { + await vi.advanceTimersByTimeAsync(3 * 60_000); + } + + await vi.waitFor(() => expect(attempted).toContain("follower")); + expect(attempted.indexOf("independent")).toBeGreaterThanOrEqual(0); + expect(attempted.indexOf("independent")).toBeLessThan(attempted.indexOf("follower")); + expect(attempted.filter((id) => id === "poison")).toHaveLength( + DEFAULT_INGRESS_RETRY_MAX_ATTEMPTS, + ); + 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" }); + const runtimeErrors = mockCalls(params.runtime.error as unknown as MockCallSource).map( + ([message]) => String(message), + ); + expect(runtimeErrors.some((message) => message.includes("reached retry limit"))).toBe( + true, + ); + expect(runtimeErrors.join("\n")).not.toContain("hello"); + } finally { + await handler.deactivate(); + } + }); + } finally { + vi.useRealTimers(); + } + }); + + it("preserves retry facts when deactivation cancels a durable Discord claim", async () => { + await withDiscordQueue(async (queue) => { + const raw = createRawMessage("cancelled", "lane-a"); + await queue.enqueue( + "cancelled", + { version: 1, receivedAt: 10, rawMessage: raw }, + { laneKey: "channel:lane-a", receivedAt: 10 }, + ); + const failedClaim = await queue.claim("cancelled", { ownerId: "failed-owner" }); + expect(failedClaim).not.toBeNull(); + if (!failedClaim) { + return; + } + await queue.release(failedClaim, { + lastError: "previous genuine failure", + releasedAt: 20, + }); + const before = (await queue.listPending())[0]; + const firstPreflight = vi.fn(async () => null); + const firstParams = createDiscordHandlerParams(); + firstParams.cfg.messages = { inbound: { debounceMs: 60_000 } }; + const first = createDurableDiscordMessageHandler({ + ...firstParams, + client: {} as never, + testing: { + preflightDiscordMessage: firstPreflight as never, + createIngressMonitor: (monitorParams) => + createDiscordIngressMonitor({ ...monitorParams, queue }), + }, + }); + + await vi.waitFor(async () => expect(await queue.listClaims()).toHaveLength(1)); + await first.deactivate(); + + expect(firstPreflight).not.toHaveBeenCalled(); + expect(await queue.listPending()).toEqual([ + expect.objectContaining({ + id: "cancelled", + attempts: before?.attempts, + lastAttemptAt: before?.lastAttemptAt, + lastError: before?.lastError, + }), + ]); + + const replacementPreflight = vi.fn(async () => null); + const replacementParams = createDiscordHandlerParams(); + const replacement = createDurableDiscordMessageHandler({ + ...replacementParams, + client: {} as never, + testing: { + preflightDiscordMessage: replacementPreflight as never, + createIngressMonitor: (monitorParams) => + createDiscordIngressMonitor({ ...monitorParams, queue }), + }, + }); + try { + await vi.waitFor(() => expect(replacementPreflight).toHaveBeenCalledTimes(1)); + await expect( + queue.enqueue("cancelled", {} as DiscordIngressPayload), + ).resolves.toMatchObject({ kind: "completed" }); + } finally { + await replacement.deactivate(); + } + }); + }); + it("does not abort concurrent runs with a Discord-owned channel timeout", async () => { vi.useFakeTimers(); try { diff --git a/src/channels/message/ingress-drain.test.ts b/src/channels/message/ingress-drain.test.ts index 2d6e6cc790c3..d59f8ab33f29 100644 --- a/src/channels/message/ingress-drain.test.ts +++ b/src/channels/message/ingress-drain.test.ts @@ -170,6 +170,149 @@ describe("channel ingress drain", () => { }); }); + it("cancels unadopted work without changing its retry facts", async () => { + await withTempState(async (stateDir) => { + let clock = 100; + const queue = createTestIngressQueue(stateDir, { now: () => clock }); + await queue.enqueue("evt-cancel", { text: "x" }, { laneKey: "l1", receivedAt: 1 }); + const failedClaim = await queue.claim("evt-cancel", { ownerId: "failed-owner" }); + expect(failedClaim).not.toBeNull(); + if (!failedClaim) { + return; + } + await queue.release(failedClaim, { lastError: "previous failure", releasedAt: clock }); + const before = (await queue.listPending())[0]; + for (let cycle = 0; cycle < 3; cycle += 1) { + const lifecycles: ChannelIngressDispatchLifecycle[] = []; + clock += 1; + const drain = createChannelIngressDrain({ + queue, + now: () => clock, + retryPolicy: { baseMs: 0, maxMs: 0 }, + dispatchClaimedEvent: async (_event, lifecycle) => { + lifecycles.push(lifecycle); + return { kind: "deferred" }; + }, + }); + + await drain.drainOnce(); + await vi.waitFor(() => expect(lifecycles).toHaveLength(1)); + await expectDefined( + expectDefined(lifecycles[0], "cancelled lifecycle").onCancelled, + "cancel callback", + )(); + expect(await queue.listPending()).toEqual([ + expect.objectContaining({ + id: "evt-cancel", + attempts: before?.attempts, + lastAttemptAt: before?.lastAttemptAt, + lastError: before?.lastError, + }), + ]); + expect(await queue.listClaims()).toEqual([]); + drain.dispose(); + } + + const terminal = createChannelIngressDrain({ + queue, + now: () => clock, + retryPolicy: { maxAttempts: 2, deadLetterMinAgeMs: 0, baseMs: 0, maxMs: 0 }, + dispatchClaimedEvent: async () => { + throw new Error("final genuine failure"); + }, + }); + await terminal.drainOnce(); + await terminal.waitForIdle(); + expect(await queue.listFailed?.()).toEqual([ + expect.objectContaining({ + id: "evt-cancel", + attempts: 1, + reason: "retry-limit-exceeded", + message: "final genuine failure", + }), + ]); + terminal.dispose(); + }); + }); + + it("keeps the lane owned until a dead-letter write commits", async () => { + await withTempState(async (stateDir) => { + const queue = createTestIngressQueue(stateDir); + await queue.enqueue("poison", { text: "bad" }, { laneKey: "shared", receivedAt: 1 }); + await queue.enqueue("follower", { text: "good" }, { laneKey: "shared", receivedAt: 2 }); + const fail = queue.fail.bind(queue); + let failAttempts = 0; + queue.fail = async (...args) => { + failAttempts += 1; + if (failAttempts < 3) { + throw new Error(`transient fail write ${failAttempts}`); + } + return await fail(...args); + }; + const dispatched: string[] = []; + const drain = createChannelIngressDrain({ + queue, + retryPolicy: { maxAttempts: 1, deadLetterMinAgeMs: 0 }, + dispatchClaimedEvent: async (event, lifecycle) => { + dispatched.push(event.id); + if (event.id === "poison") { + throw new Error("poison delivery"); + } + await lifecycle.onAdopted(); + }, + }); + + await drain.drainOnce(); + const idle = drain.waitForIdle(); + await vi.advanceTimersByTimeAsync(0); + expect(failAttempts).toBe(1); + expect(drain.activeLaneKeys()).toEqual(new Set(["shared"])); + expect(await drain.drainOnce()).toEqual({ started: 0 }); + expect(dispatched).toEqual(["poison"]); + + await vi.advanceTimersByTimeAsync(5_000); + await idle; + expect(failAttempts).toBe(3); + expect(await drain.drainOnce()).toEqual({ started: 1 }); + await drain.waitForIdle(); + expect(dispatched).toEqual(["poison", "follower"]); + drain.dispose(); + }); + }); + + it("keeps ownership when every dead-letter write fails", async () => { + await withTempState(async (stateDir) => { + const queue = createTestIngressQueue(stateDir); + await queue.enqueue("poison", { text: "bad" }, { laneKey: "shared", receivedAt: 1 }); + await queue.enqueue("follower", { text: "good" }, { laneKey: "shared", receivedAt: 2 }); + queue.fail = async () => { + throw new Error("persistent fail write"); + }; + const dispatched: string[] = []; + const drain = createChannelIngressDrain({ + queue, + retryPolicy: { maxAttempts: 1, deadLetterMinAgeMs: 0 }, + dispatchClaimedEvent: async (event) => { + dispatched.push(event.id); + throw new Error("poison delivery"); + }, + }); + + await drain.drainOnce(); + const idle = drain.waitForIdle(); + for (let attempt = 0; attempt < 8; attempt += 1) { + await vi.advanceTimersByTimeAsync(180_000); + } + await idle; + + expect(dispatched).toEqual(["poison"]); + expect(drain.activeLaneKeys()).toEqual(new Set(["shared"])); + expect((await queue.listClaims()).map((claim) => claim.id)).toEqual(["poison"]); + expect(await drain.drainOnce()).toEqual({ started: 0 }); + drain.dispose(); + }); + }); + it("holds lanes by default and releases only opted-in deferred lanes", async () => { for (const occupancy of ["hold", "release"] as const) { await withTempState(async (stateDir) => { @@ -697,6 +840,39 @@ describe("channel ingress drain", () => { }); }); + it("keeps retry-accounted abandonment pending beyond the failure threshold", async () => { + await withTempState(async (stateDir) => { + let clock = 1; + const queue = createTestIngressQueue(stateDir, { now: () => clock }); + await queue.enqueue("abandoned", { text: "x" }, { laneKey: "l", receivedAt: 1 }); + + for (let attempt = 0; attempt < 3; attempt += 1) { + clock += 1; + const drain = createChannelIngressDrain({ + queue, + now: () => clock, + retryPolicy: { maxAttempts: 1, deadLetterMinAgeMs: 0, baseMs: 0, maxMs: 0 }, + dispatchClaimedEvent: async (_event, lifecycle) => { + await lifecycle.onAbandoned(); + return { kind: "deferred" }; + }, + }); + await drain.drainOnce(); + await drain.waitForIdle(); + drain.dispose(); + } + + expect(await queue.listPending()).toEqual([ + expect.objectContaining({ + id: "abandoned", + attempts: 3, + lastError: "turn-abandoned", + }), + ]); + expect(await queue.listFailed?.()).toEqual([]); + }); + }); + it("bindIngressLifecycleToReplyOptions returns only turnAdoptionLifecycle", async () => { const abort = new AbortController(); const calls: string[] = []; @@ -708,6 +884,9 @@ describe("channel ingress drain", () => { onFailed: () => { calls.push("failed"); }, + onCancelled: () => { + calls.push("cancelled"); + }, onAdopted: () => { calls.push("adopted"); }, @@ -721,6 +900,7 @@ describe("channel ingress drain", () => { expect(bound.turnAdoptionLifecycle.abortSignal).toBe(abort.signal); expect(bound.turnAdoptionLifecycle.admission).toBe("exclusive"); expect("onFailed" in bound.turnAdoptionLifecycle).toBe(false); + expect("onCancelled" in bound.turnAdoptionLifecycle).toBe(false); expect("onAdopted" in bound).toBe(false); expect(Object.keys(bound)).toEqual(["turnAdoptionLifecycle"]); bound.turnAdoptionLifecycle.onDeferred(); @@ -1150,6 +1330,7 @@ describe("channel ingress drain", () => { onDeferred: () => {}, onAdoptionFinalizing: () => {}, onFailed: () => {}, + onCancelled: () => {}, onAbandoned: () => {}, }); expect(bound.turnAdoptionLifecycle.admission).toBe("exclusive"); diff --git a/src/channels/message/ingress-drain.ts b/src/channels/message/ingress-drain.ts index b6cdd089f41f..8e7ed89f5977 100644 --- a/src/channels/message/ingress-drain.ts +++ b/src/channels/message/ingress-drain.ts @@ -69,6 +69,8 @@ type ChannelIngressDispatchLifecycle = { onAdoptionFinalizing: () => void; /** Deferred work terminally failed after dispatch returned. */ onFailed?: (error: unknown) => void | Promise; + /** Explicit cancellation before adoption; releases without consuming retry budget. */ + onCancelled?: () => void | Promise; /** * Deferred turn finished without ever owning the reply lane. * Drain releases the claim for retry. @@ -347,13 +349,12 @@ export function createChannelIngressDrain< const releaseClaim = async ( claim: ChannelIngressQueueClaim, - lastError?: string, + releaseOptions?: { lastError?: string; recordAttempt?: boolean }, ) => { await commitClaimWriteWithRetry({ claim, label: "release", - write: () => - queue.release(claim, lastError === undefined ? {} : { lastError, releasedAt: now() }), + write: () => queue.release(claim, { ...releaseOptions, releasedAt: now() }), falseMeansReclaimed: false, }); }; @@ -401,7 +402,7 @@ export function createChannelIngressDrain< } const displayId = claim.id.replace(/^0+(?=\d)/, "") || claim.id; log(`spooled update ${displayId} failed; keeping for retry: ${disposition.message}`); - await releaseClaim(claim, disposition.message); + await releaseClaim(claim, { lastError: disposition.message }); }; const createSettleOwner = ( @@ -468,6 +469,24 @@ export function createChannelIngressDrain< state.stallTimer.unref?.(); }; + const releaseUnadopted = async ( + state: ActiveHandlerState, + releaseOptions: { lastError?: string; recordAttempt?: boolean }, + ) => { + if (state.phase !== "deferred" && state.phase !== "dispatching") { + return; + } + if (state.guillotined || state.superseded) { + return; + } + clearStallTimer(state); + await state + .settleOnce(async () => { + await releaseClaim(state.claim, releaseOptions); + }) + .catch(() => undefined); + }; + const createLifecycle = ( state: ActiveHandlerState, ): ChannelIngressDispatchLifecycle => { @@ -528,19 +547,13 @@ export function createChannelIngressDrain< await applyFailureDisposition(state.claim, error); }); }, + onCancelled: async () => { + // Cancellation means ownership ended before delivery, so preserve every + // prior retry fact while reopening the canonical row for replacement. + await releaseUnadopted(state, { recordAttempt: false }); + }, onAbandoned: async () => { - if (state.phase !== "deferred" && state.phase !== "dispatching") { - return; - } - if (state.guillotined || state.superseded) { - return; - } - clearStallTimer(state); - await state - .settleOnce(async () => { - await releaseClaim(state.claim, "turn-abandoned"); - }) - .catch(() => undefined); + await releaseUnadopted(state, { lastError: "turn-abandoned" }); }, }; }; diff --git a/src/channels/message/ingress-monitor.ts b/src/channels/message/ingress-monitor.ts index 07831bdf6423..31a1739d61a4 100644 --- a/src/channels/message/ingress-monitor.ts +++ b/src/channels/message/ingress-monitor.ts @@ -65,6 +65,7 @@ export type ChannelIngressMonitorLifecycle = { onDeferred: () => void; onAdoptionFinalizing: () => void; onFailed?: (error: unknown) => void | Promise; + onCancelled?: () => void | Promise; onAbandoned: () => void | Promise; }; @@ -368,6 +369,16 @@ export function createChannelIngressMonitor void | Promise) => { + handedOff = true; + deferredHandoff = true; + try { + await settle(); + requestDrain(); + } finally { + settleDeferredClaim(); + } + }; const wrappedLifecycle: ChannelIngressMonitorLifecycle = { ...lifecycle, admission: "exclusive", @@ -393,26 +404,9 @@ export function createChannelIngressMonitor { - handedOff = true; - deferredHandoff = true; - try { - await lifecycle.onFailed?.(error); - requestDrain(); - } finally { - settleDeferredClaim(); - } - }, - onAbandoned: async () => { - handedOff = true; - deferredHandoff = true; - try { - await lifecycle.onAbandoned(); - requestDrain(); - } finally { - settleDeferredClaim(); - } - }, + onFailed: (error) => settleDeferredLifecycle(() => lifecycle.onFailed?.(error)), + onCancelled: () => settleDeferredLifecycle(() => lifecycle.onCancelled?.()), + onAbandoned: () => settleDeferredLifecycle(() => lifecycle.onAbandoned()), }; // Adoption can complete before delivery returns; track both lifetimes so stop diff --git a/src/plugin-sdk/channel-ingress-runtime.test.ts b/src/plugin-sdk/channel-ingress-runtime.test.ts index 7e1680c876e5..786f2848691a 100644 --- a/src/plugin-sdk/channel-ingress-runtime.test.ts +++ b/src/plugin-sdk/channel-ingress-runtime.test.ts @@ -33,10 +33,13 @@ describe("plugin-sdk/channel-ingress-runtime", () => { onDeferred: vi.fn(), onAdoptionFinalizing: vi.fn(), onFailed: vi.fn(async () => {}), + onCancelled: vi.fn(async () => {}), onAbandoned: vi.fn(async () => {}), }); const first = createLifecycle(); const second = createLifecycle(); + const cancellation = fanInChannelIngressLifecycles([first, second]); + await cancellation.lifecycle?.onCancelled?.(); const combined = fanInChannelIngressLifecycles([undefined, first, second]); combined.lifecycle?.onAdoptionFinalizing(); @@ -49,6 +52,8 @@ describe("plugin-sdk/channel-ingress-runtime", () => { expect(second.onAdopted).toHaveBeenCalledOnce(); expect(first.onAbandoned).not.toHaveBeenCalled(); expect(second.onAbandoned).not.toHaveBeenCalled(); + expect(first.onCancelled).toHaveBeenCalledOnce(); + expect(second.onCancelled).toHaveBeenCalledOnce(); }); it("settles or abandons claims that no reply lane adopted", async () => { diff --git a/src/plugin-sdk/channel-ingress-runtime.ts b/src/plugin-sdk/channel-ingress-runtime.ts index 3d958ac97912..480b01bbac06 100644 --- a/src/plugin-sdk/channel-ingress-runtime.ts +++ b/src/plugin-sdk/channel-ingress-runtime.ts @@ -132,7 +132,11 @@ export function fanInChannelIngressLifecycles( const lifecycles = inputs.filter((lifecycle) => lifecycle !== undefined); const first = lifecycles[0]; if (!first) { - return { lifecycle: undefined, settle: async () => {}, abandon: async () => {} }; + return { + lifecycle: undefined, + settle: async () => {}, + abandon: async () => {}, + }; } let handedOff = false; @@ -147,7 +151,6 @@ export function fanInChannelIngressLifecycles( const failAll = async (error: unknown) => { await Promise.all(lifecycles.map(async (lifecycle) => await lifecycle.onFailed?.(error))); }; - return { lifecycle: { abortSignal: @@ -173,6 +176,10 @@ export function fanInChannelIngressLifecycles( handedOff = true; await failAll(error); }, + onCancelled: async () => { + handedOff = true; + await Promise.all(lifecycles.map(async (lifecycle) => await lifecycle.onCancelled?.())); + }, onAbandoned: async () => { handedOff = true; await abandonAll();