diff --git a/extensions/telegram/src/bot-update-tracker.test.ts b/extensions/telegram/src/bot-update-tracker.test.ts index 7aa7db154d58..dde1ef570382 100644 --- a/extensions/telegram/src/bot-update-tracker.test.ts +++ b/extensions/telegram/src/bot-update-tracker.test.ts @@ -1,6 +1,6 @@ // Telegram tests cover bot update tracker plugin behavior. import { describe, expect, it, vi } from "vitest"; -import { createTelegramUpdateTracker } from "./bot-update-tracker.js"; +import { ACCEPTED_UPDATE_ID_RETENTION, createTelegramUpdateTracker } from "./bot-update-tracker.js"; type TelegramUpdateTrackerState = ReturnType< ReturnType["getState"] @@ -186,10 +186,9 @@ describe("createTelegramUpdateTracker", () => { } tracker.finishUpdate(oldReplay.update, { completed: true }); - expect(tracker.beginUpdate(updateCtx(42))).toEqual({ - accepted: false, - reason: "accepted-watermark", - }); + // Second begin is rejected (numeric set and/or semantic key). After persist + // advances, the numeric id may already be pruned below the persisted floor. + expect(tracker.beginUpdate(updateCtx(42)).accepted).toBe(false); expectTrackerState(tracker.getState(), { highestAcceptedUpdateId: 43, highestCompletedUpdateId: 43, @@ -199,6 +198,167 @@ describe("createTelegramUpdateTracker", () => { } satisfies Partial); }); + it("dispatches a delayed lower update id after newer cross-lane ids complete", () => { + const tracker = createTelegramUpdateTracker({ + initialUpdateId: null, + persistenceFloorUpdateId: 100, + ackPolicy: "after_agent_dispatch", + }); + + // Lane B finishes newer global update ids while lane A still holds N+1. + const laterA = tracker.beginUpdate(updateCtx(102)); + const laterB = tracker.beginUpdate(updateCtx(103)); + if (!laterA.accepted || !laterB.accepted) { + throw new Error("expected later cross-lane updates to be accepted"); + } + tracker.finishUpdate(laterA.update, { completed: true }); + tracker.finishUpdate(laterB.update, { completed: true }); + + // Delayed durable-spool replay of N+1 must still dispatch exactly once. + const delayed = tracker.beginUpdate(updateCtx(101)); + if (!delayed.accepted) { + throw new Error("expected delayed cross-lane spool replay to be accepted"); + } + tracker.finishUpdate(delayed.update, { completed: true }); + + expect(tracker.beginUpdate(updateCtx(101))).toEqual({ + accepted: false, + reason: "accepted-watermark", + }); + expectTrackerState(tracker.getState(), { + highestAcceptedUpdateId: 103, + highestCompletedUpdateId: 103, + pendingUpdateIds: [], + failedUpdateIds: [], + } satisfies Partial); + }); + + it("accepts a delayed group mention after newer cross-lane ids so routing can run", () => { + const tracker = createTelegramUpdateTracker({ + initialUpdateId: null, + persistenceFloorUpdateId: 200, + ackPolicy: "after_agent_dispatch", + }); + + const later = tracker.beginUpdate(updateCtx(202)); + if (!later.accepted) { + throw new Error("expected later update to be accepted"); + } + tracker.finishUpdate(later.update, { completed: true }); + + // Mention-shaped payload uses the same beginUpdate gate as any other update; + // watermark must not drop it before normal user_request mention routing. + const mention = tracker.beginUpdate({ + update: { + update_id: 201, + message: { + message_id: 10, + text: "@bot hello", + entities: [{ type: "mention", offset: 0, length: 4 }], + chat: { id: -100, type: "supergroup", title: "group" }, + date: 1, + }, + }, + }); + if (!mention.accepted) { + throw new Error("expected delayed mention update to be accepted for user_request routing"); + } + tracker.finishUpdate(mention.update, { completed: true }); + expect(tracker.beginUpdate(updateCtx(201))).toEqual({ + accepted: false, + reason: "accepted-watermark", + }); + }); + + it("bounds accepted-id memory without a persist callback via retention window", () => { + // No onAcceptedUpdateId: persisted floor never advances past the option floor. + const tracker = createTelegramUpdateTracker({ + initialUpdateId: null, + persistenceFloorUpdateId: 0, + ackPolicy: "after_agent_dispatch", + }); + const firstId = 1; + const lastId = ACCEPTED_UPDATE_ID_RETENTION + 50; + for (let updateId = firstId; updateId <= lastId; updateId += 1) { + const begun = tracker.beginUpdate(updateCtx(updateId)); + if (!begun.accepted) { + throw new Error(`expected update ${updateId} to be accepted`); + } + tracker.finishUpdate(begun.update, { completed: true }); + } + // Ids far below the retention window may be pruned; recent ids stay suppressed. + expect(tracker.beginUpdate(updateCtx(firstId)).accepted).toBe(true); + expect(tracker.beginUpdate(updateCtx(lastId))).toEqual({ + accepted: false, + reason: "accepted-watermark", + }); + }); + + it("does not prune pending or failed accepted ids from the retention window", () => { + const tracker = createTelegramUpdateTracker({ + initialUpdateId: null, + persistenceFloorUpdateId: 0, + ackPolicy: "after_agent_dispatch", + }); + const pendingId = 1; + const failedId = 2; + const pending = tracker.beginUpdate(updateCtx(pendingId)); + const failed = tracker.beginUpdate(updateCtx(failedId)); + if (!pending.accepted || !failed.accepted) { + throw new Error("expected seed updates to be accepted"); + } + tracker.finishUpdate(failed.update, { completed: false }); + + const lastId = ACCEPTED_UPDATE_ID_RETENTION + 50; + for (let updateId = 3; updateId <= lastId; updateId += 1) { + const begun = tracker.beginUpdate(updateCtx(updateId)); + if (!begun.accepted) { + throw new Error(`expected update ${updateId} to be accepted`); + } + tracker.finishUpdate(begun.update, { completed: true }); + } + + // Pending ids stay in the numeric set (never pruned) so re-begin is rejected + // as accepted-watermark, not re-dispatched. + expect(tracker.beginUpdate(updateCtx(pendingId))).toEqual({ + accepted: false, + reason: "accepted-watermark", + }); + const failedRetry = tracker.beginUpdate(updateCtx(failedId)); + if (!failedRetry.accepted) { + throw new Error("expected failed update retry to be accepted"); + } + tracker.finishUpdate(failedRetry.update, { completed: true }); + // After success, re-begin is rejected (numeric and/or semantic). + expect(tracker.beginUpdate(updateCtx(failedId)).accepted).toBe(false); + }); + + it("prunes accepted ids at or below the persisted Bot API offset", async () => { + const onAcceptedUpdateId = vi.fn(); + const tracker = createTelegramUpdateTracker({ + initialUpdateId: 100, + onAcceptedUpdateId, + }); + const early = tracker.beginUpdate(updateCtx(101)); + if (!early.accepted) { + throw new Error("expected early update to be accepted"); + } + tracker.finishUpdate(early.update, { completed: true }); + await flushTrackerMicrotasks(); + expect(onAcceptedUpdateId).toHaveBeenCalledWith(101); + + const later = tracker.beginUpdate(updateCtx(102)); + if (!later.accepted) { + throw new Error("expected later update to be accepted"); + } + tracker.finishUpdate(later.update, { completed: true }); + await flushTrackerMicrotasks(); + + // Recent completed id stays suppressed; early completed id is eligible for + // numeric prune once <= highestPersisted (semantic/spool still guard dups). + expect(tracker.beginUpdate(updateCtx(102)).accepted).toBe(false); + }); + it("serializes and coalesces accepted offset persistence", async () => { const firstWrite = deferred(); const secondWrite = deferred(); diff --git a/extensions/telegram/src/bot-update-tracker.ts b/extensions/telegram/src/bot-update-tracker.ts index 2ec1c011a5ee..4eb073ae0a51 100644 --- a/extensions/telegram/src/bot-update-tracker.ts +++ b/extensions/telegram/src/bot-update-tracker.ts @@ -55,6 +55,12 @@ function sortedIds(ids: Set): number[] { return [...ids].toSorted((a, b) => a - b); } +// Bound for per-id numeric dedupe when the persisted Bot API offset does not +// advance (no onAcceptedUpdateId) or lags. Only the realistic in-process +// redelivery window needs numeric retention; semantic keys + spool tombstones +// cover older ids. +export const ACCEPTED_UPDATE_ID_RETENTION = 10_000; + export function createTelegramUpdateTracker(options: TelegramUpdateTrackerOptions = {}) { const initialUpdateId = typeof options.initialUpdateId === "number" ? options.initialUpdateId : null; @@ -68,7 +74,9 @@ export function createTelegramUpdateTracker(options: TelegramUpdateTrackerOption const activeHandledUpdateKeys = new Map(); const pendingUpdateIds = new Set(); const failedUpdateIds = new Set(); - const completedFloorReplayUpdateIds = new Set(); + // Per-id acceptance, not a global high-water mark: multi-lane spool drains can + // finish newer update IDs before an older delayed id from another chat replays. + const acceptedUpdateIds = new Set(); let highestAcceptedUpdateId: number | null = initialUpdateId; let highestPersistedAcceptedUpdateId: number | null = persistenceFloorUpdateId; let highestPersistenceRequestedUpdateId: number | null = persistenceFloorUpdateId; @@ -80,6 +88,34 @@ export function createTelegramUpdateTracker(options: TelegramUpdateTrackerOption options.onSkip?.(key); }; + // One prune rule: drop accepted ids at or below max(persisted offset, + // highestAccepted - retention) unless still pending or failed. Persisted + // floor is safe (getUpdates cannot redeliver below it); retention bounds + // trackers that never advance a persisted floor. + const pruneAcceptedUpdateIds = () => { + if (highestAcceptedUpdateId === null && highestPersistedAcceptedUpdateId === null) { + return; + } + const windowFloor = + highestAcceptedUpdateId === null + ? Number.NEGATIVE_INFINITY + : highestAcceptedUpdateId - ACCEPTED_UPDATE_ID_RETENTION; + const persistedFloor = + highestPersistedAcceptedUpdateId === null + ? Number.NEGATIVE_INFINITY + : highestPersistedAcceptedUpdateId; + const pruneAtOrBelow = Math.max(persistedFloor, windowFloor); + for (const id of acceptedUpdateIds) { + if (id > pruneAtOrBelow) { + continue; + } + if (pendingUpdateIds.has(id) || failedUpdateIds.has(id)) { + continue; + } + acceptedUpdateIds.delete(id); + } + }; + const drainPersistQueue = async () => { const persist = options.onAcceptedUpdateId; if (persistInFlight || typeof persist !== "function") { @@ -97,6 +133,7 @@ export function createTelegramUpdateTracker(options: TelegramUpdateTrackerOption updateId > highestPersistedAcceptedUpdateId ) { highestPersistedAcceptedUpdateId = updateId; + pruneAcceptedUpdateIds(); } } catch (err) { options.onPersistError?.(err); @@ -125,17 +162,13 @@ export function createTelegramUpdateTracker(options: TelegramUpdateTrackerOption }; const acceptUpdateId = (updateId: number) => { - if (highestAcceptedUpdateId !== null && updateId <= highestAcceptedUpdateId) { - return; + acceptedUpdateIds.add(updateId); + if (highestAcceptedUpdateId === null || updateId > highestAcceptedUpdateId) { + highestAcceptedUpdateId = updateId; } - highestAcceptedUpdateId = updateId; + pruneAcceptedUpdateIds(); }; - const isFloorReplayUpdateId = (updateId: number) => - initialUpdateId === null && - persistenceFloorUpdateId !== null && - updateId <= persistenceFloorUpdateId; - function resolveSafeCompletedUpdateId() { if (highestCompletedUpdateId === null) { return null; @@ -184,18 +217,16 @@ export function createTelegramUpdateTracker(options: TelegramUpdateTrackerOption const updateId = resolveTelegramUpdateId(ctx); const updateKey = buildTelegramUpdateKey(ctx); if (typeof updateId === "number") { - if (highestAcceptedUpdateId !== null && updateId <= highestAcceptedUpdateId) { - const floorReplay = isFloorReplayUpdateId(updateId); - if (!floorReplay && !failedUpdateIds.has(updateId)) { - skip(`update:${updateId}`); - return { accepted: false, reason: "accepted-watermark" }; - } - if (floorReplay && completedFloorReplayUpdateIds.has(updateId)) { - skip(`update:${updateId}`); - return { accepted: false, reason: "accepted-watermark" }; - } - } else { + if (failedUpdateIds.has(updateId)) { failedUpdateIds.delete(updateId); + } else if (initialUpdateId !== null && updateId <= initialUpdateId) { + // Restored Bot API offset: suppress redelivery of already-persisted ids. + skip(`update:${updateId}`); + return { accepted: false, reason: "accepted-watermark" }; + } else if (acceptedUpdateIds.has(updateId)) { + // Same process already accepted this exact id (completed or in-flight). + skip(`update:${updateId}`); + return { accepted: false, reason: "accepted-watermark" }; } } if (updateKey) { @@ -241,9 +272,6 @@ export function createTelegramUpdateTracker(options: TelegramUpdateTrackerOption pendingUpdateIds.delete(update.updateId); if (finish.completed) { failedUpdateIds.delete(update.updateId); - if (isFloorReplayUpdateId(update.updateId)) { - completedFloorReplayUpdateIds.add(update.updateId); - } if (highestCompletedUpdateId === null || update.updateId > highestCompletedUpdateId) { highestCompletedUpdateId = update.updateId; } @@ -256,6 +284,7 @@ export function createTelegramUpdateTracker(options: TelegramUpdateTrackerOption options.onPersistError?.(err); }); } + pruneAcceptedUpdateIds(); } };