From 2f7e2aee1584d799e3fdee679fa8158ad3903dd4 Mon Sep 17 00:00:00 2001 From: tzy-17 Date: Fri, 10 Jul 2026 13:53:29 +0800 Subject: [PATCH] fix(queue): prevent applyQueueDropPolicy from selecting in-flight items as overflow victims (#103284) * fix(queue): prevent applyQueueDropPolicy from selecting in-flight items as overflow victims When a burst of inbound messages hits the followup queue cap while the head item is mid-delivery, applyQueueDropPolicy can select that same in-flight item as an overflow victim. With the default dropPolicy: "summarize", the in-flight item ends up recorded in the overflow summary as dropped even though it is still being delivered, producing a contradictory record where the same message is both answered and reported as unanswered-due-to-overflow. The fix introduces an optional `inFlight` Set parameter to applyQueueDropPolicy and drainNextQueueItem. The followup queue state now owns a shared inFlight set that is: - Populated by drainNextQueueItem during the await run(next) window - Populated by the collect-merge drain path for activeGroupItems - Passed to applyQueueDropPolicy in enqueueFollowupRun The drop policy now computes an effective queue length that excludes in-flight items, and skips them when selecting splice victims. Fixes #103246 Co-Authored-By: Claude Opus 4.7 * fix(queue): align pending depth with active deliveries --------- Co-authored-by: Claude Opus 4.7 Co-authored-by: Peter Steinberger --- src/auto-reply/reply/queue.in-flight.test.ts | 250 +++++++++++++++++++ src/auto-reply/reply/queue/drain.ts | 11 +- src/auto-reply/reply/queue/enqueue.ts | 12 +- src/auto-reply/reply/queue/state.ts | 4 + src/utils/queue-helpers.test.ts | 68 ++++- src/utils/queue-helpers.ts | 43 +++- 6 files changed, 368 insertions(+), 20 deletions(-) create mode 100644 src/auto-reply/reply/queue.in-flight.test.ts diff --git a/src/auto-reply/reply/queue.in-flight.test.ts b/src/auto-reply/reply/queue.in-flight.test.ts new file mode 100644 index 000000000000..78c446adcb8b --- /dev/null +++ b/src/auto-reply/reply/queue.in-flight.test.ts @@ -0,0 +1,250 @@ +// Proves queue caps and depth describe pending work while active identities remain in shared state. +import { afterEach, describe, expect, it, vi } from "vitest"; +import { + clearFollowupQueue, + completeFollowupRunLifecycle, + enqueueFollowupRun, + getFollowupQueueDepth, + scheduleFollowupDrain, +} from "./queue.js"; +import { createDeferred, createQueueTestRun as createRun } from "./queue.test-helpers.js"; +import { getExistingFollowupQueue } from "./queue/state.js"; +import type { FollowupRun, QueueDropPolicy, QueueSettings } from "./queue/types.js"; + +describe("followup queue in-flight ownership", () => { + const keys = new Set(); + + afterEach(() => { + for (const key of keys) { + clearFollowupQueue(key); + } + keys.clear(); + }); + + const createKey = (suffix: string) => { + const key = `test-in-flight-${suffix}-${Date.now()}-${Math.random()}`; + keys.add(key); + return key; + }; + + const createSettings = (dropPolicy: QueueDropPolicy): QueueSettings => ({ + mode: "followup", + debounceMs: 0, + cap: 1, + dropPolicy, + }); + + it.each(["old", "summarize"] as const)( + "keeps an active single delivery out of %s overflow victims", + async (dropPolicy) => { + const key = createKey(dropPolicy); + const entered = createDeferred(); + const release = createDeferred(); + const activeComplete = vi.fn(); + const pendingComplete = vi.fn(); + const calls: FollowupRun[] = []; + const active = { + ...createRun({ prompt: "active" }), + queuedLifecycle: { onComplete: activeComplete }, + }; + const runFollowup = async (run: FollowupRun) => { + calls.push(run); + run.queuedLifecycle?.onAdmitted?.(); + if (run === active) { + entered.resolve(); + await release.promise; + } + completeFollowupRunLifecycle(run); + }; + + try { + expect( + enqueueFollowupRun(key, active, createSettings(dropPolicy), "none", runFollowup), + ).toBe(true); + await entered.promise; + + expect(getFollowupQueueDepth(key)).toBe(0); + expect( + enqueueFollowupRun( + key, + { + ...createRun({ prompt: "pending" }), + queuedLifecycle: { onComplete: pendingComplete }, + }, + createSettings(dropPolicy), + "none", + ), + ).toBe(true); + expect( + enqueueFollowupRun( + key, + createRun({ prompt: "survivor" }), + createSettings(dropPolicy), + "none", + ), + ).toBe(true); + + const queue = getExistingFollowupQueue(key); + expect(queue?.inFlight.has(active)).toBe(true); + expect(queue?.items.map((item) => item.prompt)).toEqual(["active", "survivor"]); + expect(getFollowupQueueDepth(key)).toBe(1); + expect(activeComplete).not.toHaveBeenCalled(); + expect(pendingComplete).toHaveBeenCalledTimes(dropPolicy === "old" ? 1 : 0); + expect(queue?.summarySources.map((item) => item.prompt)).toEqual( + dropPolicy === "summarize" ? ["pending"] : [], + ); + } finally { + release.resolve(); + } + + await expect.poll(() => getExistingFollowupQueue(key)).toBeUndefined(); + expect(activeComplete).toHaveBeenCalledOnce(); + expect(pendingComplete).toHaveBeenCalledOnce(); + expect(calls.at(-1)?.prompt).toBe("survivor"); + }, + ); + + it("admits one pending item under drop:new while another item is active", async () => { + const key = createKey("new"); + const entered = createDeferred(); + const release = createDeferred(); + const rejectedEnqueued = vi.fn(); + const rejectedComplete = vi.fn(); + const active = createRun({ prompt: "active" }); + const runFollowup = async (run: FollowupRun) => { + run.queuedLifecycle?.onAdmitted?.(); + if (run === active) { + entered.resolve(); + await release.promise; + } + completeFollowupRunLifecycle(run); + }; + + try { + expect(enqueueFollowupRun(key, active, createSettings("new"), "none", runFollowup)).toBe( + true, + ); + await entered.promise; + + expect(getFollowupQueueDepth(key)).toBe(0); + expect( + enqueueFollowupRun(key, createRun({ prompt: "pending" }), createSettings("new"), "none"), + ).toBe(true); + expect( + enqueueFollowupRun( + key, + { + ...createRun({ prompt: "rejected" }), + queuedLifecycle: { + onEnqueued: rejectedEnqueued, + onComplete: rejectedComplete, + }, + }, + createSettings("new"), + "none", + ), + ).toBe(false); + + expect(getFollowupQueueDepth(key)).toBe(1); + expect(getExistingFollowupQueue(key)?.items.map((item) => item.prompt)).toEqual([ + "active", + "pending", + ]); + expect(rejectedEnqueued).not.toHaveBeenCalled(); + expect(rejectedComplete).toHaveBeenCalledOnce(); + } finally { + release.resolve(); + } + + await expect.poll(() => getExistingFollowupQueue(key)).toBeUndefined(); + }); + + it("protects a collect group and counts only active identities still present", async () => { + const key = createKey("collect"); + const entered = createDeferred(); + const release = createDeferred(); + const groupCompletions = [vi.fn(), vi.fn()]; + const pendingComplete = vi.fn(); + const rejectedComplete = vi.fn(); + let aggregate: FollowupRun | undefined; + const initialSettings: QueueSettings = { + mode: "collect", + debounceMs: 0, + cap: 50, + dropPolicy: "summarize", + }; + const group = groupCompletions.map((onComplete, index) => ({ + ...createRun({ + prompt: `group-${index + 1}`, + originatingChannel: "slack" as const, + originatingTo: "channel:A", + originatingChatType: "channel", + }), + queuedLifecycle: { onComplete }, + })); + const runFollowup = async (run: FollowupRun) => { + if (!aggregate) { + aggregate = run; + entered.resolve(); + await release.promise; + } + completeFollowupRunLifecycle(run); + }; + + for (const run of group) { + expect(enqueueFollowupRun(key, run, initialSettings, "none", undefined, false)).toBe(true); + } + scheduleFollowupDrain(key, runFollowup); + + try { + await entered.promise; + const queue = getExistingFollowupQueue(key); + expect(queue?.inFlight.size).toBe(2); + expect(getFollowupQueueDepth(key)).toBe(0); + + const oldSettings: QueueSettings = { ...initialSettings, cap: 1, dropPolicy: "old" }; + expect( + enqueueFollowupRun( + key, + { + ...createRun({ prompt: "pending-old" }), + queuedLifecycle: { onComplete: pendingComplete }, + }, + oldSettings, + "none", + ), + ).toBe(true); + expect(enqueueFollowupRun(key, createRun({ prompt: "survivor" }), oldSettings, "none")).toBe( + true, + ); + + expect(queue?.items.map((item) => item.prompt)).toEqual(["group-1", "group-2", "survivor"]); + expect(pendingComplete).toHaveBeenCalledOnce(); + expect(groupCompletions.map((complete) => complete.mock.calls.length)).toEqual([0, 0]); + + aggregate?.queuedLifecycle?.onAdmitted?.(); + expect(queue?.items.map((item) => item.prompt)).toEqual(["survivor"]); + expect(queue?.inFlight.size).toBe(2); + expect(getFollowupQueueDepth(key)).toBe(1); + + expect( + enqueueFollowupRun( + key, + { + ...createRun({ prompt: "rejected-new" }), + queuedLifecycle: { onComplete: rejectedComplete }, + }, + { ...initialSettings, cap: 1, dropPolicy: "new" }, + "none", + ), + ).toBe(false); + expect(rejectedComplete).toHaveBeenCalledOnce(); + expect(getFollowupQueueDepth(key)).toBe(1); + } finally { + release.resolve(); + } + + await expect.poll(() => getExistingFollowupQueue(key)).toBeUndefined(); + expect(groupCompletions.map((complete) => complete.mock.calls.length)).toEqual([1, 1]); + }); +}); diff --git a/src/auto-reply/reply/queue/drain.ts b/src/auto-reply/reply/queue/drain.ts index e91e6cfa43b8..4cb550480f06 100644 --- a/src/auto-reply/reply/queue/drain.ts +++ b/src/auto-reply/reply/queue/drain.ts @@ -1106,6 +1106,7 @@ export function scheduleFollowupDrain( isCrossChannel, items: queue.items, run: effectiveRunFollowup, + inFlight: queue.inFlight, }); if (collectDrainResult === "empty") { break; @@ -1194,6 +1195,11 @@ export function scheduleFollowupDrain( }); }; try { + // Mark active group items as in-flight so the drop policy does not + // select them as overflow victims while the group drain is awaited. + for (const item of activeGroupItems) { + queue.inFlight.add(item); + } await drainGroup(); } catch (err) { if (admitted) { @@ -1201,6 +1207,9 @@ export function scheduleFollowupDrain( } throw err; } finally { + for (const item of activeGroupItems) { + queue.inFlight.delete(item); + } cancellation.dispose(); } if (!admitted) { @@ -1218,7 +1227,7 @@ export function scheduleFollowupDrain( continue; } - if (!(await drainNextQueueItem(queue.items, effectiveRunFollowup))) { + if (!(await drainNextQueueItem(queue.items, effectiveRunFollowup, queue.inFlight))) { break; } } diff --git a/src/auto-reply/reply/queue/enqueue.ts b/src/auto-reply/reply/queue/enqueue.ts index 24a0406968c8..6af6d296d032 100644 --- a/src/auto-reply/reply/queue/enqueue.ts +++ b/src/auto-reply/reply/queue/enqueue.ts @@ -3,7 +3,11 @@ import { normalizeOptionalString } from "@openclaw/normalization-core/string-coe import { normalizeChatType } from "../../../channels/chat-type.js"; import { resolveGlobalDedupeCache } from "../../../infra/dedupe.js"; import { channelRouteDedupeKey } from "../../../plugin-sdk/channel-route.js"; -import { applyQueueDropPolicy, shouldSkipQueueItem } from "../../../utils/queue-helpers.js"; +import { + applyQueueDropPolicy, + countPendingQueueItems, + shouldSkipQueueItem, +} from "../../../utils/queue-helpers.js"; import { createOverflowSummaryRetrySource, kickFollowupDrainIfIdle, @@ -120,7 +124,8 @@ export function enqueueFollowupRun( } // drop:new rejects this source without mutating the existing queue. Do not // publish an external queued identity for work that will never be admitted. - if (queue.dropPolicy === "new" && queue.cap > 0 && queue.items.length >= queue.cap) { + const pendingCount = countPendingQueueItems(queue.items, queue.inFlight); + if (queue.dropPolicy === "new" && queue.cap > 0 && pendingCount >= queue.cap) { completeFollowupRunLifecycle(run); return false; } @@ -132,6 +137,7 @@ export function enqueueFollowupRun( const shouldEnqueue = applyQueueDropPolicy({ queue, + inFlight: queue.inFlight, summarize: (item) => normalizeOptionalString(item.summaryLine) || item.prompt.trim(), onDrop: (dropped) => { if (queue.dropPolicy === "summarize") { @@ -201,7 +207,7 @@ export function getFollowupQueueDepth(key: string): number { if (!queue) { return 0; } - return queue.items.length; + return countPendingQueueItems(queue.items, queue.inFlight); } export function resetRecentQueuedMessageIdDedupe(): void { diff --git a/src/auto-reply/reply/queue/state.ts b/src/auto-reply/reply/queue/state.ts index a5b04013502f..5e4f1f264d91 100644 --- a/src/auto-reply/reply/queue/state.ts +++ b/src/auto-reply/reply/queue/state.ts @@ -14,6 +14,8 @@ export type FollowupQueueState = { abortController: AbortController; items: FollowupRun[]; draining: boolean; + /** Identities retained in `items` while delivery awaits; pending cap and depth must exclude them. */ + inFlight: Set; lastEnqueuedAt: number; mode: QueueMode; debounceMs: number; @@ -112,6 +114,7 @@ export function getFollowupQueue(key: string, settings: QueueSettings): Followup abortController: new AbortController(), items: [], draining: false, + inFlight: new Set(), lastEnqueuedAt: 0, mode: settings.mode, debounceMs: @@ -158,6 +161,7 @@ export function clearFollowupQueue(key: string): number { } } queue.items.length = 0; + queue.inFlight.clear(); queue.droppedCount = 0; queue.summaryLines = []; queue.summarySources = []; diff --git a/src/utils/queue-helpers.test.ts b/src/utils/queue-helpers.test.ts index 9fec04c2dd73..61b78bd6cdae 100644 --- a/src/utils/queue-helpers.test.ts +++ b/src/utils/queue-helpers.test.ts @@ -4,6 +4,7 @@ import { applyQueueDropPolicy, applyQueueRuntimeSettings, clearQueueSummaryState, + countPendingQueueItems, drainCollectQueueStep, drainNextQueueItem, hasCrossChannelItems, @@ -165,14 +166,22 @@ describe("drainCollectQueueStep", () => { }); describe("drainNextQueueItem", () => { + it("counts only in-flight identities that still intersect the queue", () => { + const active = { id: "active" }; + const pending = { id: "pending" }; + const alreadyRemoved = { id: "already-removed" }; + + expect(countPendingQueueItems([active, pending], new Set([active, alreadyRemoved]))).toBe(1); + }); + it("keeps overflow survivors when the queue mutates during an awaited drain", async () => { type Item = { id: string }; const queue = { - items: [{ id: "m1" }], + items: [{ id: "m1" }] as Item[], cap: 3, dropPolicy: "summarize" as const, droppedCount: 0, - summaryLines: [], + summaryLines: [] as string[], }; const delivered: string[] = []; const dropped: string[] = []; @@ -180,11 +189,16 @@ describe("drainNextQueueItem", () => { const gate = new Promise((resolve) => { release = resolve; }); + const inFlight = new Set(); - const firstDrain = drainNextQueueItem(queue.items, async (item: Item) => { - delivered.push(item.id); - await gate; - }); + const firstDrain = drainNextQueueItem( + queue.items, + async (item: Item) => { + delivered.push(item.id); + await gate; + }, + inFlight, + ); await Promise.resolve(); for (let index = 2; index <= 8; index += 1) { @@ -192,6 +206,7 @@ describe("drainNextQueueItem", () => { const shouldEnqueue = applyQueueDropPolicy({ queue, summarize: (queued) => queued.id, + inFlight, onDrop: (items) => { dropped.push(...items.map((queued) => queued.id)); }, @@ -204,15 +219,48 @@ describe("drainNextQueueItem", () => { release(); await firstDrain; while ( - await drainNextQueueItem(queue.items, async (item) => { - delivered.push(item.id); - }) + await drainNextQueueItem( + queue.items, + async (item) => { + delivered.push(item.id); + }, + inFlight, + ) ) {} expect(delivered).toEqual(["m1", "m6", "m7", "m8"]); - expect(dropped).toEqual(["m1", "m2", "m3", "m4", "m5"]); + expect(dropped).toEqual(["m2", "m3", "m4", "m5"]); expect(queue.items).toEqual([]); }); + + it("skips in-flight items when selecting drop victims", () => { + type Item = { id: string }; + const m1: Item = { id: "m1" }; + const m2: Item = { id: "m2" }; + const m3: Item = { id: "m3" }; + const m4: Item = { id: "m4" }; + const queue = { + items: [m1, m2, m3, m4], + cap: 2, + dropPolicy: "old" as const, + droppedCount: 0, + summaryLines: [] as string[], + }; + const inFlight = new Set([m1]); + const dropped: string[] = []; + + applyQueueDropPolicy({ + queue, + inFlight, + summarize: (item) => item.id, + onDrop: (items) => { + dropped.push(...items.map((item) => item.id)); + }, + }); + + expect(dropped).toEqual(["m2", "m3"]); + expect(queue.items).toEqual([m1, m4]); + }); }); describe("hasCrossChannelItems", () => { diff --git a/src/utils/queue-helpers.ts b/src/utils/queue-helpers.ts index e81e9594ff05..d7bba2f92035 100644 --- a/src/utils/queue-helpers.ts +++ b/src/utils/queue-helpers.ts @@ -98,22 +98,41 @@ export function shouldSkipQueueItem(params: { return params.dedupe(params.item, params.items); } +/** Count identities that are still pending in the queue, excluding active deliveries. */ +export function countPendingQueueItems(items: readonly T[], inFlight?: ReadonlySet): number { + if (!inFlight || inFlight.size === 0) { + return items.length; + } + return items.reduce((count, item) => count + (inFlight.has(item) ? 0 : 1), 0); +} + /** Apply overflow policy before enqueueing another item. */ export function applyQueueDropPolicy(params: { queue: QueueState; summarize: (item: T) => string; summaryLimit?: number; onDrop?: (items: T[]) => void; + inFlight?: ReadonlySet; }): boolean { const cap = params.queue.cap; - if (cap <= 0 || params.queue.items.length < cap) { + const pendingCount = countPendingQueueItems(params.queue.items, params.inFlight); + if (cap <= 0 || pendingCount < cap) { return true; } if (params.queue.dropPolicy === "new") { return false; } - const dropCount = params.queue.items.length - cap + 1; - const dropped = params.queue.items.splice(0, dropCount); + const dropCount = pendingCount - cap + 1; + const dropped: T[] = []; + // Active identities remain in the shared array until delivery succeeds; evict only pending work. + for (let index = 0; dropped.length < dropCount; ) { + const item = params.queue.items[index]; + if (params.inFlight?.has(item)) { + index += 1; + continue; + } + dropped.push(...params.queue.items.splice(index, 1)); + } params.onDrop?.(dropped); if (params.queue.dropPolicy === "summarize") { for (const item of dropped) { @@ -181,13 +200,22 @@ export function removeQueuedItemsByRef(items: T[], processed: readonly T[]): export async function drainNextQueueItem( items: T[], run: (item: T) => Promise, + inFlight?: Set, ): Promise { const next = items[0]; if (!next) { return false; } - await run(next); - removeQueuedItemsByRef(items, [next]); + // Mark the item as in-flight so applyQueueDropPolicy skips it during the + // await window when the shared items array is still mutated by enqueuers. + inFlight?.add(next); + try { + await run(next); + // Keep the identity protected until its successful by-reference removal. + removeQueuedItemsByRef(items, [next]); + } finally { + inFlight?.delete(next); + } return true; } @@ -198,6 +226,7 @@ async function drainCollectItemIfNeeded(params: { setForceIndividualCollect?: (next: boolean) => void; items: T[]; run: (item: T) => Promise; + inFlight?: Set; }): Promise<"skipped" | "drained" | "empty"> { if (!params.forceIndividualCollect && !params.isCrossChannel) { return "skipped"; @@ -206,7 +235,7 @@ async function drainCollectItemIfNeeded(params: { // Once cross-channel items appear, future collection stays individual to preserve ordering. params.setForceIndividualCollect?.(true); } - const drained = await drainNextQueueItem(params.items, params.run); + const drained = await drainNextQueueItem(params.items, params.run, params.inFlight); return drained ? "drained" : "empty"; } @@ -216,6 +245,7 @@ export async function drainCollectQueueStep(params: { isCrossChannel: boolean; items: T[]; run: (item: T) => Promise; + inFlight?: Set; }): Promise<"skipped" | "drained" | "empty"> { return await drainCollectItemIfNeeded({ forceIndividualCollect: params.collectState.forceIndividualCollect, @@ -225,6 +255,7 @@ export async function drainCollectQueueStep(params: { }, items: params.items, run: params.run, + inFlight: params.inFlight, }); }