From f7786a16cfe81d2d619ff760df20387d6579c307 Mon Sep 17 00:00:00 2001 From: Ayaan Zaidi Date: Thu, 16 Jul 2026 21:18:10 +0530 Subject: [PATCH] refactor(telegram): drive ingress through the core drain Both transports now enqueue durably and pump the shared drain: polling keeps offset-after-committed-enqueue, webhook keeps 200-after-spool-write. Deletes the reply fence, private claim-owner, retry policy, and per-transport claim/replay/watchdog loops. Dispatch outcomes propagate: failed-retryable releases for replay instead of tombstoning. Supersede policy stays Telegram-owned, authorization-gated with pairing-store and owner identities via the same resolver as normal ingress; room_event ambient pending remains supersedable by a later user turn (#108656). --- .../src/bot-message-dispatch-fence.ts | 117 --- .../src/bot-message-dispatch-reply.ts | 3 +- .../telegram/src/bot-message-dispatch-turn.ts | 24 +- .../telegram/src/bot-message-dispatch.ts | 33 +- .../src/bot-message-dispatch.types.ts | 19 +- extensions/telegram/src/bot-message.ts | 84 +- .../telegram/src/bot-processing-outcome.ts | 23 +- extensions/telegram/src/polling-session.ts | 885 +----------------- .../src/spooled-update-retry-policy.ts | 87 -- .../src/telegram-ingress-claim-owner.ts | 185 ---- .../src/telegram-ingress-drain-factory.ts | 65 ++ .../telegram/src/telegram-ingress-drain.ts | 176 ++++ .../src/telegram-ingress-non-retryable.ts | 39 + .../src/telegram-ingress-spool.payload.ts | 9 + .../telegram/src/telegram-ingress-spool.ts | 351 +++---- .../src/telegram-ingress-supersede.ts | 374 ++++++++ .../telegram/src/telegram-reply-fence.ts | 255 ----- extensions/telegram/src/webhook.ts | 500 +--------- 18 files changed, 974 insertions(+), 2255 deletions(-) delete mode 100644 extensions/telegram/src/bot-message-dispatch-fence.ts delete mode 100644 extensions/telegram/src/spooled-update-retry-policy.ts delete mode 100644 extensions/telegram/src/telegram-ingress-claim-owner.ts create mode 100644 extensions/telegram/src/telegram-ingress-drain-factory.ts create mode 100644 extensions/telegram/src/telegram-ingress-drain.ts create mode 100644 extensions/telegram/src/telegram-ingress-non-retryable.ts create mode 100644 extensions/telegram/src/telegram-ingress-spool.payload.ts create mode 100644 extensions/telegram/src/telegram-ingress-supersede.ts delete mode 100644 extensions/telegram/src/telegram-reply-fence.ts diff --git a/extensions/telegram/src/bot-message-dispatch-fence.ts b/extensions/telegram/src/bot-message-dispatch-fence.ts deleted file mode 100644 index 4607337bb784..000000000000 --- a/extensions/telegram/src/bot-message-dispatch-fence.ts +++ /dev/null @@ -1,117 +0,0 @@ -// Telegram plugin module owns pre-adoption reply-fence authority. -import type { TelegramMessageContext } from "./bot-message-context.js"; -import type { DispatchTelegramMessageParams } from "./bot-message-dispatch.types.js"; -import { getTelegramSequentialKey } from "./sequential-key.js"; -import { - beginTelegramReplyFence, - buildTelegramNonInterruptingReplyFenceKey, - buildTelegramReplyFenceLaneKey, - endTelegramReplyFence, - isTelegramReplyFenceSuperseded, - releaseTelegramReplyFenceAbortController, - resolveTelegramReplyFenceKey, - shouldSupersedeTelegramReplyFence, - supersedeTelegramReplyFence, -} from "./telegram-reply-fence.js"; - -type CreateTelegramReplyFenceParams = Pick< - DispatchTelegramMessageParams, - "onTurnAdopted" | "onTurnDeferred" | "onTurnAbandoned" | "turnAbortSignal" -> & { - context: TelegramMessageContext; -}; - -export function createTelegramReplyFenceController(params: CreateTelegramReplyFenceParams) { - const { context } = params; - const replyFenceKey = resolveTelegramReplyFenceKey({ - ctxPayload: context.ctxPayload, - chatId: context.chatId, - threadSpec: context.threadSpec, - }); - const sequentialKey = getTelegramSequentialKey({ - message: context.msg, - ...(context.primaryCtx.me ? { me: context.primaryCtx.me } : {}), - }); - const laneKey = buildTelegramReplyFenceLaneKey({ - accountId: context.route.accountId, - sequentialKey, - }); - const supersedes = shouldSupersedeTelegramReplyFence(context.ctxPayload); - const activeKey = supersedes - ? replyFenceKey.activeKey - : buildTelegramNonInterruptingReplyFenceKey({ - activeKey: replyFenceKey.activeKey, - laneKey, - }); - // Ambient room-event work uses a separate fence key. Any non-room-event - // inbound may cancel it without owning abort authority over adopted user turns. - if (context.ctxPayload.InboundEventKind !== "room_event") { - supersedeTelegramReplyFence(replyFenceKey.roomEventKey); - } - - const abortController = new AbortController(); - const abortSignal = params.turnAbortSignal - ? AbortSignal.any([abortController.signal, params.turnAbortSignal]) - : abortController.signal; - let generation: number | undefined = beginTelegramReplyFence({ - key: activeKey, - supersede: supersedes, - abortController, - laneKey, - }); - let abortControllerQueued = false; - let queuedTurnAdmitted = false; - - const isSuperseded = () => - abortController.signal.aborted || - (generation !== undefined && isTelegramReplyFenceSuperseded({ key: activeKey, generation })); - - const release = () => { - if (generation === undefined) { - return; - } - endTelegramReplyFence(activeKey, abortControllerQueued ? undefined : abortController); - generation = undefined; - }; - - const adoptTurn = async () => { - await params.onTurnAdopted?.(); - // Fence abort and supersession authority end after durable adoption. - // Core then owns all interruption of the adopted run. - release(); - releaseTelegramReplyFenceAbortController(activeKey, abortController); - }; - - return { - abortSignal, - adoptTurn, - generation: () => generation, - isSuperseded, - release, - queuedFollowupLifecycle: - context.ctxPayload.InboundEventKind === "room_event" || - params.onTurnAdopted || - params.onTurnDeferred || - params.onTurnAbandoned - ? { - onEnqueued: () => { - abortControllerQueued = true; - params.onTurnDeferred?.(); - }, - onAdmitted: async () => { - await adoptTurn(); - queuedTurnAdmitted = true; - }, - onComplete: () => { - abortControllerQueued = false; - releaseTelegramReplyFenceAbortController(activeKey, abortController); - if (!queuedTurnAdmitted) { - params.onTurnAbandoned?.(); - } - }, - } - : undefined, - }; -} - -export type TelegramReplyFenceController = ReturnType; diff --git a/extensions/telegram/src/bot-message-dispatch-reply.ts b/extensions/telegram/src/bot-message-dispatch-reply.ts index 3ade3da55305..d032c9420a06 100644 --- a/extensions/telegram/src/bot-message-dispatch-reply.ts +++ b/extensions/telegram/src/bot-message-dispatch-reply.ts @@ -13,7 +13,6 @@ import type { TelegramBotDeps } from "./bot-deps.js"; import type { TelegramMessageContext } from "./bot-message-context.js"; import type { TelegramDeliveryController } from "./bot-message-dispatch-delivery.js"; import type { TelegramDraftController } from "./bot-message-dispatch-draft.js"; -import type { TelegramReplyFenceController } from "./bot-message-dispatch-fence.js"; import type { TelegramProgressController } from "./bot-message-dispatch-progress.js"; import { deduplicateBlockSentMedia } from "./bot-message-dispatch.media-dedup.js"; import type { TelegramDispatchTurnState } from "./bot-message-dispatch.types.js"; @@ -59,7 +58,7 @@ export function createTelegramReplyDelivery(params: { context: TelegramMessageContext; delivery: TelegramDeliveryController; draft: TelegramDraftController; - fence: Pick; + fence: { generation: () => number; isSuperseded: () => boolean }; progress: TelegramProgressController; runtime: RuntimeEnv; state: TelegramDispatchTurnState; diff --git a/extensions/telegram/src/bot-message-dispatch-turn.ts b/extensions/telegram/src/bot-message-dispatch-turn.ts index fab3240a8b31..3bdf119b4481 100644 --- a/extensions/telegram/src/bot-message-dispatch-turn.ts +++ b/extensions/telegram/src/bot-message-dispatch-turn.ts @@ -12,7 +12,6 @@ import type { TelegramBotDeps } from "./bot-deps.js"; import type { TelegramMessageContext } from "./bot-message-context.js"; import type { TelegramDeliveryController } from "./bot-message-dispatch-delivery.js"; import type { TelegramDraftController } from "./bot-message-dispatch-draft.js"; -import type { TelegramReplyFenceController } from "./bot-message-dispatch-fence.js"; import type { TelegramProgressController } from "./bot-message-dispatch-progress.js"; import type { TelegramReplyDelivery } from "./bot-message-dispatch-reply.js"; import type { TelegramDispatchTurnState } from "./bot-message-dispatch.types.js"; @@ -26,7 +25,15 @@ export async function runTelegramDispatchTurn(params: { context: TelegramMessageContext; delivery: TelegramDeliveryController; draft: TelegramDraftController; - fence: TelegramReplyFenceController; + /** Pre-adoption abort + lifecycle from durable ingress (optional for non-spooled). */ + turnAdoptionLifecycle?: { + admission?: "exclusive" | "cancel-only"; + onAdopted: () => void | Promise; + onDeferred?: () => void; + onAbandoned?: () => void; + abortSignal?: AbortSignal; + }; + isSuperseded: () => boolean; progress: TelegramProgressController; reply: TelegramReplyDelivery; state: TelegramDispatchTurnState; @@ -110,13 +117,20 @@ export async function runTelegramDispatchTurn(params: { replyOptions: { skillFilter: context.skillFilter, disableBlockStreaming: params.draft.disableBlockStreaming, - abortSignal: params.fence.abortSignal, - onTurnAdopted: params.fence.adoptTurn, + abortSignal: params.turnAdoptionLifecycle?.abortSignal, + turnAdoptionLifecycle: params.turnAdoptionLifecycle + ? { + admission: params.turnAdoptionLifecycle.admission ?? "exclusive", + onAdopted: params.turnAdoptionLifecycle.onAdopted, + onDeferred: params.turnAdoptionLifecycle.onDeferred, + onAbandoned: params.turnAdoptionLifecycle.onAbandoned, + abortSignal: params.turnAdoptionLifecycle.abortSignal, + } + : undefined, sourceReplyDeliveryMode: isRoomEvent ? "message_tool_only" : undefined, queuedDeliveryCorrelations: isRoomEvent ? [{ begin: beginDeliveryCorrelation }] : undefined, - queuedFollowupLifecycle: params.fence.queuedFollowupLifecycle, suppressTyping: isRoomEvent, onPartialReply: params.draft.answerLane.stream || params.draft.reasoningLane.stream diff --git a/extensions/telegram/src/bot-message-dispatch.ts b/extensions/telegram/src/bot-message-dispatch.ts index ae39e87c5917..68e6610e93da 100644 --- a/extensions/telegram/src/bot-message-dispatch.ts +++ b/extensions/telegram/src/bot-message-dispatch.ts @@ -8,7 +8,6 @@ import { truncateUtf16Safe } from "openclaw/plugin-sdk/text-utility-runtime"; import { resolveDispatchTelegramContext } from "./bot-message-dispatch-context.js"; import { createTelegramDeliveryController } from "./bot-message-dispatch-delivery.js"; import { createTelegramDraftController } from "./bot-message-dispatch-draft.js"; -import { createTelegramReplyFenceController } from "./bot-message-dispatch-fence.js"; import { createTelegramProgressController } from "./bot-message-dispatch-progress.js"; import { createTelegramReplyDelivery } from "./bot-message-dispatch-reply.js"; import { @@ -272,10 +271,7 @@ export const dispatchTelegramMessage = async ({ opts, retryDispatchErrors = false, suppressFailureFallback = false, - onTurnAdopted, - onTurnDeferred, - onTurnAbandoned, - turnAbortSignal, + turnAdoptionLifecycle, }: DispatchTelegramMessageParams): Promise => { const dispatchStartedAt = Date.now(); const dispatchContext = resolveDispatchTelegramContext({ context }); @@ -299,9 +295,9 @@ export const dispatchTelegramMessage = async ({ const forceBlockStreamingForReasoning = resolvedReasoningLevel === "on" && streamMode !== "progress"; const quote = resolveTelegramQuoteContext({ context: dispatchContext, replyToMode }); - // Controllers retain this callback but cannot run it before turn dispatch. - // Acquire the fence afterward so controller setup failures claim no ownership. - const isDispatchSuperseded = () => fence.isSuperseded(); + // Pre-adoption abort is drain-owned via turnAdoptionLifecycle.abortSignal. + const isDispatchSuperseded = () => turnAdoptionLifecycle?.abortSignal?.aborted === true; + const dispatchGeneration = 0; const draft = createTelegramDraftController({ accountId: dispatchContext.route.accountId, bot, @@ -369,7 +365,7 @@ export const dispatchTelegramMessage = async ({ delivery, draft, fence: { - generation: () => fence.generation(), + generation: () => dispatchGeneration, isSuperseded: isDispatchSuperseded, }, progress, @@ -380,19 +376,12 @@ export const dispatchTelegramMessage = async ({ }); let isFirstTurnInSession = false; - let dispatchWasSuperseded: boolean; + let dispatchWasSuperseded = false; let turnDispatched: boolean | undefined; const isDmTopic = !dispatchContext.isGroup && dispatchContext.threadSpec.scope === "dm" && dispatchContext.threadSpec.id != null; - const fence = createTelegramReplyFenceController({ - context: dispatchContext, - onTurnAdopted, - onTurnDeferred, - onTurnAbandoned, - turnAbortSignal, - }); try { await prepareTelegramSticker({ cfg, context: dispatchContext }); if (isDmTopic) { @@ -418,7 +407,8 @@ export const dispatchTelegramMessage = async ({ context: dispatchContext, delivery, draft, - fence, + turnAdoptionLifecycle, + isSuperseded: isDispatchSuperseded, progress, reply, state, @@ -439,20 +429,19 @@ export const dispatchTelegramMessage = async ({ state.dispatchError ??= err; runtime.error?.(danger(`telegram terminal block delivery failed: ${String(err)}`)); } - await draft.cleanup(fence.isSuperseded()); + await draft.cleanup(isDispatchSuperseded()); if ( streamMode === "progress" && progress.sawProgressFinal() && !state.dispatchError && !state.hadErrorReplyFailureOrSkip && - !fence.isSuperseded() + !isDispatchSuperseded() ) { await delivery.deliverProgressCollapseSummary(); } } } finally { - dispatchWasSuperseded = fence.isSuperseded(); - fence.release(); + dispatchWasSuperseded = isDispatchSuperseded(); } if (turnDispatched === false) { diff --git a/extensions/telegram/src/bot-message-dispatch.types.ts b/extensions/telegram/src/bot-message-dispatch.types.ts index 3ab531a3987d..1a85b4432ce3 100644 --- a/extensions/telegram/src/bot-message-dispatch.types.ts +++ b/extensions/telegram/src/bot-message-dispatch.types.ts @@ -26,14 +26,17 @@ export type DispatchTelegramMessageParams = { opts: Pick; retryDispatchErrors?: boolean; suppressFailureFallback?: boolean; - /** Fires after recovery-relevant session/run state is durably persisted. */ - onTurnAdopted?: () => void | Promise; - /** Marks a queued follow-up whose adoption will happen at reply-lane admission. */ - onTurnDeferred?: () => void; - /** Releases a deferred turn that completed without ever owning the reply lane. */ - onTurnAbandoned?: () => void; - /** Cancels queued/model work when ingress ownership fails before adoption. */ - turnAbortSignal?: AbortSignal; + /** + * Canonical turn ownership lifecycle from the durable ingress drain + * (or a test double). Pre-adoption abort + adopt/defer/abandon. + */ + turnAdoptionLifecycle?: { + admission?: "exclusive" | "cancel-only"; + onAdopted: () => void | Promise; + onDeferred?: () => void; + onAbandoned?: () => void; + abortSignal?: AbortSignal; + }; }; export type TelegramDispatchResult = diff --git a/extensions/telegram/src/bot-message.ts b/extensions/telegram/src/bot-message.ts index 7135bd8569a9..95e263407ed0 100644 --- a/extensions/telegram/src/bot-message.ts +++ b/extensions/telegram/src/bot-message.ts @@ -23,6 +23,7 @@ import { createTelegramSpooledReplayParticipant, createTelegramSpooledReplayDeferredParticipant, getTelegramSpooledReplayDeferredParticipant, + getTelegramSpooledReplayLifecycle, isTelegramSpooledReplayUpdate, recordTelegramMessageProcessingResult, type TelegramMessageProcessingResult, @@ -34,7 +35,7 @@ import type { TelegramContext } from "./bot/types.js"; import type { TelegramReplyChainEntry } from "./message-cache.js"; import { TELEGRAM_TEXT_CHUNK_LIMIT } from "./outbound-adapter.js"; import { TELEGRAM_RICH_TEXT_LIMIT } from "./rich-message.js"; -import { resolveSpooledUpdatePersistenceRetryDelayMs } from "./spooled-update-retry-policy.js"; +import { resolveSpooledUpdatePersistenceRetryDelayMs } from "./telegram-ingress-spool.js"; const telegramInboundLog = createSubsystemLogger("gateway/channels/telegram").child("inbound"); @@ -262,10 +263,13 @@ export const createTelegramMessageProcessor = (deps: TelegramMessageProcessorDep await turnContext.onDispatchStart?.(); } const runDispatch = async (params: { - onTurnAdopted?: () => void | Promise; - onTurnDeferred?: () => void; - onTurnAbandoned?: () => void; - turnAbortSignal?: AbortSignal; + turnAdoptionLifecycle?: { + admission?: "exclusive" | "cancel-only"; + onAdopted: () => void | Promise; + onDeferred?: () => void; + onAbandoned?: () => void; + abortSignal?: AbortSignal; + }; }): Promise => { try { const dispatchResult = await dispatchTelegramMessage({ @@ -281,10 +285,7 @@ export const createTelegramMessageProcessor = (deps: TelegramMessageProcessorDep opts, retryDispatchErrors: spooledReplay, suppressFailureFallback: spooledReplay, - onTurnAdopted: params.onTurnAdopted, - onTurnDeferred: params.onTurnDeferred, - onTurnAbandoned: params.onTurnAbandoned, - turnAbortSignal: params.turnAbortSignal, + turnAdoptionLifecycle: params.turnAdoptionLifecycle, }); if (dispatchResult?.kind === "failed-retryable") { const result: TelegramMessageProcessingResult = { @@ -389,34 +390,49 @@ export const createTelegramMessageProcessor = (deps: TelegramMessageProcessorDep } }; const run = async () => { - const turnAbortSignal = turnContext.spooledReplayAbortSignal - ? AbortSignal.any([participant.abortSignal, turnContext.spooledReplayAbortSignal]) - : participant.abortSignal; + const drainLifecycle = getTelegramSpooledReplayLifecycle(); + // Participant always owns an AbortSignal on the spooled-replay path; + // merge optional drain/context signals without widening to undefined. + const turnAbortSignal: AbortSignal = (() => { + const extras = [turnContext.spooledReplayAbortSignal, drainLifecycle?.abortSignal].filter( + (signal): signal is AbortSignal => signal !== undefined, + ); + if (extras.length === 0) { + return participant.abortSignal; + } + return AbortSignal.any([participant.abortSignal, ...extras]); + })(); const result = await runDispatch({ - turnAbortSignal, - onTurnAdopted: async () => { - if (adopted) { - return; - } - adoptionAttempted = true; - const adoptedResult = await settle({ kind: "completed" }, "adopted"); - if (adoptedResult.kind !== "completed") { - adoptionFinalizationError = - adoptedResult.kind === "failed-retryable" + turnAdoptionLifecycle: { + admission: "exclusive", + abortSignal: turnAbortSignal, + onAdopted: async () => { + if (adopted) { + return; + } + adoptionAttempted = true; + const adoptedResult = await settle({ kind: "completed" }, "adopted"); + if (adoptedResult.kind !== "completed") { + adoptionFinalizationError = + adoptedResult.kind === "failed-retryable" + ? adoptedResult.error + : new Error("telegram spooled turn adoption was not completed"); + throw adoptedResult.kind === "failed-retryable" ? adoptedResult.error : new Error("telegram spooled turn adoption was not completed"); - throw adoptedResult.kind === "failed-retryable" - ? adoptedResult.error - : new Error("telegram spooled turn adoption was not completed"); - } - }, - onTurnDeferred: () => { - deferred = true; - }, - onTurnAbandoned: () => { - if (!adopted) { - void settle({ kind: "skipped" }, "terminal"); - } + } + await drainLifecycle?.onAdopted(); + }, + onDeferred: () => { + deferred = true; + drainLifecycle?.onDeferred(); + }, + onAbandoned: () => { + if (!adopted) { + void settle({ kind: "skipped" }, "terminal"); + } + drainLifecycle?.onAbandoned(); + }, }, }); if (adopted) { diff --git a/extensions/telegram/src/bot-processing-outcome.ts b/extensions/telegram/src/bot-processing-outcome.ts index 4e1c4b488d3b..b3b24729871e 100644 --- a/extensions/telegram/src/bot-processing-outcome.ts +++ b/extensions/telegram/src/bot-processing-outcome.ts @@ -10,8 +10,18 @@ type TelegramUpdateProcessingFrame = { result?: TelegramMessageProcessingResult; }; +export type TelegramSpooledReplayLifecycle = { + abortSignal: AbortSignal; + onAdopted: () => void | Promise; + onDeferred: () => void; + /** Clears pre-adoption stall while durable adoption finalization is held. */ + onAdoptionFinalizing?: () => void; + onAbandoned: () => void; +}; + type TelegramSpooledReplayFrame = { deferredWork?: TelegramSpooledReplayDeferredParticipant; + lifecycle?: TelegramSpooledReplayLifecycle; }; export type TelegramSpooledReplayDeferredParticipant = { @@ -95,6 +105,9 @@ export function createTelegramSpooledReplayParticipant( return undefined; } settlementHeld = true; + // Timeout settlement must wait for durable adoption finalization: pause + // the drain stall watchdog while the hold is active. + telegramSpooledReplayFrames.getStore()?.lifecycle?.onAdoptionFinalizing?.(); let released = false; return { release: (mode) => { @@ -145,8 +158,11 @@ export function getTelegramSpooledReplayDeferredParticipant(): export async function runWithTelegramSpooledReplayUpdate( update: object, fn: () => Promise, + lifecycle?: TelegramSpooledReplayLifecycle, ): Promise<{ value: T; deferredWork?: TelegramSpooledReplayDeferredParticipant }> { - const frame: TelegramSpooledReplayFrame = {}; + const frame: TelegramSpooledReplayFrame = { + ...(lifecycle ? { lifecycle } : {}), + }; telegramSpooledReplayUpdates.add(update); try { const value = await telegramSpooledReplayFrames.run(frame, fn); @@ -156,6 +172,11 @@ export async function runWithTelegramSpooledReplayUpdate( } } +/** Drain lifecycle for the active spooled-replay ALS frame, if any. */ +export function getTelegramSpooledReplayLifecycle(): TelegramSpooledReplayLifecycle | undefined { + return telegramSpooledReplayFrames.getStore()?.lifecycle; +} + export function isTelegramSpooledReplayUpdate(update: unknown): boolean { return ( telegramSpooledReplayFrames.getStore() !== undefined || diff --git a/extensions/telegram/src/polling-session.ts b/extensions/telegram/src/polling-session.ts index 261f941ef62d..67a1a89c9924 100644 --- a/extensions/telegram/src/polling-session.ts +++ b/extensions/telegram/src/polling-session.ts @@ -4,18 +4,9 @@ import type { ChannelAccountSnapshot } from "openclaw/plugin-sdk/channel-contrac import type { TelegramNetworkConfig } from "openclaw/plugin-sdk/config-contracts"; import { drainPendingDeliveries } from "openclaw/plugin-sdk/delivery-queue-runtime"; import { formatErrorMessage } from "openclaw/plugin-sdk/error-runtime"; -import { - clampPositiveTimerTimeoutMs, - resolvePositiveTimerTimeoutMs, -} from "openclaw/plugin-sdk/number-runtime"; import { formatDurationPrecise, sleepWithAbort } from "openclaw/plugin-sdk/runtime-env"; import { normalizeLowercaseStringOrEmpty } from "openclaw/plugin-sdk/string-coerce-runtime"; import { withTelegramApiErrorLogging } from "./api-logging.js"; -import { - runWithTelegramSpooledReplayUpdate, - type TelegramMessageProcessingResult, - type TelegramSpooledReplayDeferredParticipant, -} from "./bot-processing-outcome.js"; import { createTelegramBot } from "./bot.js"; import type { TelegramTransport } from "./fetch.js"; import { isRecoverableTelegramNetworkError } from "./network-errors.js"; @@ -28,41 +19,17 @@ import { import { createTelegramPollingStatusPublisher } from "./polling-status.js"; import { TelegramPollingTransportState } from "./polling-transport-state.js"; import { TELEGRAM_GET_UPDATES_REQUEST_TIMEOUT_MS } from "./request-timeouts.js"; -import { getTelegramSequentialKey } from "./sequential-key.js"; +import { createTelegramTransportIngressDrain } from "./telegram-ingress-drain-factory.js"; +import { resolveTelegramAdoptionStallTimeoutMs } from "./telegram-ingress-drain.js"; import { - resolveNonRetryableSpooledUpdateFailure, - resolveSpooledUpdateAttemptNumber, - resolveSpooledUpdateRetryDelayMs, - shouldDeadLetterRetryableSpooledUpdate, - TELEGRAM_SPOOLED_RETRY_MAX_ATTEMPTS, -} from "./spooled-update-retry-policy.js"; -import { - isTelegramSpooledCorruptClaimOwnedByOtherLiveProcess, - isTelegramSpooledUpdateClaimOwnedByOtherLiveProcess, -} from "./telegram-ingress-claim-owner.js"; -import { - abandonTelegramSpooledUpdateClaim, - claimNextTelegramSpooledUpdate, - completeTelegramSpooledUpdateWithRetry, - failTelegramSpooledUpdateClaim, - listTelegramSpooledUpdateClaims, - listTelegramSpooledUpdates, - recoverStaleTelegramSpooledUpdateClaims, - refreshTelegramSpooledUpdateClaim, - releaseTelegramSpooledUpdateClaim, resolveTelegramIngressSpoolDir, + telegramSpooledUpdateLaneKey, writeTelegramSpooledUpdate, - type ClaimedTelegramSpooledUpdate, - type TelegramSpooledUpdate, } from "./telegram-ingress-spool.js"; import { createTelegramIngressWorker, type TelegramIngressWorkerFactory, } from "./telegram-ingress-worker.js"; -import { - buildTelegramReplyFenceLaneKey, - supersedeTelegramReplyFenceLane, -} from "./telegram-reply-fence.js"; // Surfaced in logs and channel status when getUpdates returns 409; the only // user-fixable causes are a second poller on the same token or a stale webhook. @@ -76,15 +43,6 @@ const MAX_POLL_STALL_THRESHOLD_MS = 600_000; const POLL_WATCHDOG_INTERVAL_MS = 30_000; const POLL_STOP_GRACE_MS = 15_000; // Status-only backlog note threshold (unrelated to adoption timeout). -const ISOLATED_INGRESS_BACKLOG_STALL_MS = 25 * 60_000; -// claim→adoption only; once adopted, run lifecycle owns the turn. -const ISOLATED_INGRESS_ADOPTION_STALL_MS = 5 * 60_000; -const TELEGRAM_SPOOLED_HANDLER_ABORT_GRACE_MS = 5_000; -const TELEGRAM_SPOOLED_HANDLER_TIMEOUT_ENV = "OPENCLAW_TELEGRAM_SPOOLED_HANDLER_TIMEOUT_MS"; -const TELEGRAM_SPOOLED_DRAIN_START_LIMIT = 100; -const TELEGRAM_SPOOLED_DRAIN_SCAN_LIMIT = TELEGRAM_SPOOLED_DRAIN_START_LIMIT * 10; -const TELEGRAM_SPOOLED_CLAIM_REFRESH_INTERVAL_MS = 5 * 60 * 1000; -const TELEGRAM_SPOOLED_CLAIM_HEALTH_GRACE_MS = 2 * TELEGRAM_SPOOLED_CLAIM_REFRESH_INTERVAL_MS; const TELEGRAM_POLLING_CLIENT_TIMEOUT_FLOOR_SECONDS = Math.ceil( TELEGRAM_GET_UPDATES_REQUEST_TIMEOUT_MS / 1000, ); @@ -112,38 +70,6 @@ const waitForGracefulStop = async (stop: () => Promise) => { } }; -const waitForSpooledHandlerTaskSettlement = async (params: { - task: Promise; - timeoutMs: number; - abortSignal?: AbortSignal; -}): Promise => { - if (params.abortSignal?.aborted) { - return false; - } - let timer: ReturnType | undefined; - let removeAbortListener: (() => void) | undefined; - try { - return await Promise.race([ - params.task.then( - () => true, - () => true, - ), - new Promise((resolve) => { - timer = setTimeout(() => resolve(false), params.timeoutMs); - timer.unref?.(); - const abort = () => resolve(false); - params.abortSignal?.addEventListener("abort", abort, { once: true }); - removeAbortListener = () => params.abortSignal?.removeEventListener("abort", abort); - }), - ]); - } finally { - if (timer) { - clearTimeout(timer); - } - removeAbortListener?.(); - } -}; - const resolvePollingStallThresholdMs = (value: number | undefined): number => { if (typeof value !== "number" || !Number.isFinite(value)) { return DEFAULT_POLL_STALL_THRESHOLD_MS; @@ -187,123 +113,16 @@ type TelegramPollingSessionOpts = { }; }; -type SpooledUpdateHandlerState = { - handlerKey: string; - laneKey: string; - task: Promise; - update: ClaimedTelegramSpooledUpdate; - updateId: number; - startedAt: number; - stopClaimRefresh: () => void; - backlogStatusMessage?: string; - timedOutAt?: number; - timeoutMessage?: string; -}; - -type DeferredSpooledUpdateClaimState = { - claimKey: string; - laneKey: string; - task: Promise; - timer?: ReturnType; - timedOutMessage?: string; - update: ClaimedTelegramSpooledUpdate; - updateId: number; - stopClaimRefresh: () => void; -}; - -const deferredSpooledUpdateClaimsByKey = new Map(); - -function buildDeferredSpooledUpdateClaimKey(update: ClaimedTelegramSpooledUpdate): string { - return `${update.pendingPath}:${update.claim?.claimToken ?? update.claim?.processId ?? "claimed"}`; -} - -type SpooledUpdateDrainResult = { - blockedByLane: Set; - started: number; -}; - -// Account health restarts create a new session in the same process while an old -// spooled handler may still be running after shutdown grace. -const TELEGRAM_POLLING_SESSION_STATE_KEY = Symbol.for("openclaw.telegram.pollingSessionState"); -type SpooledUpdateDrainHealth = { - lastCompletedAt: number; -}; - -function getTelegramPollingSessionState(): { - activeHandlersByLane: Map; - drainHealthBySpool: Map; -} { - const globalRecord = globalThis as Record; - const existing = globalRecord[TELEGRAM_POLLING_SESSION_STATE_KEY] as - | { - activeHandlersByLane: Map; - drainHealthBySpool: Map; - } - | undefined; - if (existing) { - return existing; - } - const created = { - activeHandlersByLane: new Map(), - drainHealthBySpool: new Map(), - }; - globalRecord[TELEGRAM_POLLING_SESSION_STATE_KEY] = created; - return created; -} - -const { - activeHandlersByLane: activeSpooledUpdateHandlersByLane, - drainHealthBySpool: spooledUpdateDrainHealthBySpool, -} = getTelegramPollingSessionState(); - -function getSpooledUpdateDrainHealth(spoolDir: string): SpooledUpdateDrainHealth { - const existing = spooledUpdateDrainHealthBySpool.get(spoolDir); - if (existing) { - return existing; - } - const created = { lastCompletedAt: Date.now() }; - spooledUpdateDrainHealthBySpool.set(spoolDir, created); - return created; -} - -function resolveSpooledUpdateHandlerTimeoutMs(params: { - configured?: number; - env?: NodeJS.ProcessEnv; -}): number { - const candidates = [ - params.configured, - Number(params.env?.[TELEGRAM_SPOOLED_HANDLER_TIMEOUT_ENV]), - ]; - for (const candidate of candidates) { - const timeoutMs = clampPositiveTimerTimeoutMs(candidate); - if (timeoutMs !== undefined) { - return timeoutMs; - } - } - return ISOLATED_INGRESS_ADOPTION_STALL_MS; -} - -function buildSpooledUpdateHandlerKey(params: { spoolDir: string; laneKey: string }): string { - return `${params.spoolDir}\0${params.laneKey}`; -} - -function isSpooledUpdateHandlerKeyForSpool(handlerKey: string, spoolDir: string): boolean { - return handlerKey.startsWith(`${spoolDir}\0`); -} - export class TelegramPollingSession { #restartBackoffState = createTelegramRestartBackoffState(); #webhookCleared = false; #forceRestarted = false; #activeRunner: ReturnType | undefined; #activeCycleAbort: AbortController | undefined; - #spooledUpdateHandlerKeys = new Set(); - #deferredSpooledUpdateClaimKeys = new Set(); #transportState: TelegramPollingTransportState; #status: ReturnType; #stallThresholdMs: number; #spooledUpdateHandlerTimeoutMs: number; - #spooledUpdateHandlerAbortGraceMs: number; #deliveryDrainInFlight = false; #nextDeliveryDrainAt = 0; @@ -315,16 +134,12 @@ export class TelegramPollingSession { }); this.#status = createTelegramPollingStatusPublisher(opts.setStatus); this.#stallThresholdMs = resolvePollingStallThresholdMs(opts.stallThresholdMs); - this.#spooledUpdateHandlerTimeoutMs = resolveSpooledUpdateHandlerTimeoutMs({ + this.#spooledUpdateHandlerTimeoutMs = resolveTelegramAdoptionStallTimeoutMs({ ...(opts.isolatedIngress?.spooledUpdateHandlerTimeoutMs !== undefined ? { configured: opts.isolatedIngress.spooledUpdateHandlerTimeoutMs } : {}), env: process.env, }); - this.#spooledUpdateHandlerAbortGraceMs = resolvePositiveTimerTimeoutMs( - opts.isolatedIngress?.spooledUpdateHandlerAbortGraceMs, - TELEGRAM_SPOOLED_HANDLER_ABORT_GRACE_MS, - ); } get activeRunner() { @@ -529,632 +344,40 @@ export class TelegramPollingSession { } } - async #claimNextSpooledUpdate(params: { - blockedLaneKeys: Set; - candidateUpdateIds: readonly number[]; - spoolDir: string; - }): Promise { - try { - return await claimNextTelegramSpooledUpdate({ - spoolDir: params.spoolDir, - blockedLaneKeys: params.blockedLaneKeys, - botInfo: this.opts.botInfo, - candidateUpdateIds: params.candidateUpdateIds, - scanLimit: TELEGRAM_SPOOLED_DRAIN_SCAN_LIMIT, - }); - } catch (err) { - this.opts.log( - `[telegram][diag] spooled update claim failed; keeping pending updates for retry: ${formatErrorMessage(err)}`, - ); - return null; - } - } + #ingressDrain: ReturnType | undefined; - #startSpooledUpdateClaimRefresh( - update: ClaimedTelegramSpooledUpdate, - isDrainHealthy: () => boolean, - onDrainUnhealthy: () => void, - ): () => void { - // Refresh only while this process owns useful work and its drain loop is making progress. - // Stopping the lease on a stalled drain lets another process recover the lane. - let stopped = false; - let refreshing = false; - const refresh = async (): Promise => { - if (stopped || refreshing) { - return; - } - if (!isDrainHealthy()) { - onDrainUnhealthy(); - stopped = true; - clearInterval(timer); - return; - } - refreshing = true; - try { - const refreshed = await refreshTelegramSpooledUpdateClaim(update); - if (!refreshed && !stopped) { - onDrainUnhealthy(); - stopped = true; - clearInterval(timer); - } - } catch (err) { - this.opts.log( - `[telegram][diag] spooled update ${update.updateId} claim refresh failed: ${formatErrorMessage(err)}`, - ); - if (!stopped) { - onDrainUnhealthy(); - stopped = true; - clearInterval(timer); - } - } finally { - refreshing = false; - } - }; - const timer = setInterval(() => { - void refresh(); - }, TELEGRAM_SPOOLED_CLAIM_REFRESH_INTERVAL_MS); - timer.unref?.(); - return () => { - if (stopped) { - return; - } - stopped = true; - clearInterval(timer); - }; - } - - async #handleClaimedSpooledUpdate(params: { + /** Long-lived drain for this session; dispose only when the cycle ends. */ + #getOrCreateSpooledDrain(params: { bot: TelegramBot; - onTurnAdopted: () => void; - stopClaimRefresh: () => void; - update: ClaimedTelegramSpooledUpdate; - }): Promise { - let replay: { deferredWork?: TelegramSpooledReplayDeferredParticipant }; - try { - const update = params.update.update as Parameters[0]; - replay = await runWithTelegramSpooledReplayUpdate(update, async () => { - await params.bot.handleUpdate(update); - }); - } catch (err) { - params.stopClaimRefresh(); - await this.#releaseFailedSpooledUpdate({ - err, - update: params.update, - }); - return false; - } - if (replay.deferredWork) { - this.#registerDeferredSpooledUpdate({ - deferredWork: replay.deferredWork, - laneKey: this.#spooledUpdateLaneKey(params.update), - onTurnAdopted: params.onTurnAdopted, - stopClaimRefresh: params.stopClaimRefresh, - update: params.update, - }); - return true; - } - try { - await completeTelegramSpooledUpdateWithRetry({ - update: params.update, - abortSignal: this.opts.abortSignal, - onRetry: ({ attempt, delayMs, error }) => { - this.opts.log( - `[telegram][diag] spooled update ${params.update.updateId} completion retry ${attempt} scheduled in ${formatDurationPrecise(delayMs)}: ${formatErrorMessage(error)}`, - ); - }, - }); - return true; - } catch (err) { - this.opts.log( - `[telegram][diag] spooled update ${params.update.updateId} completed but could not tombstone its claimed spool row: ${formatErrorMessage(err)}`, - ); - return false; - } - } - - #registerDeferredSpooledUpdate(params: { - deferredWork: TelegramSpooledReplayDeferredParticipant; - laneKey: string; - onTurnAdopted: () => void; - stopClaimRefresh: () => void; - update: ClaimedTelegramSpooledUpdate; - }): void { - const claimKey = buildDeferredSpooledUpdateClaimKey(params.update); - const previous = deferredSpooledUpdateClaimsByKey.get(claimKey); - if (previous) { - if (previous.timer) { - clearTimeout(previous.timer); - } - previous.stopClaimRefresh(); - deferredSpooledUpdateClaimsByKey.delete(claimKey); - } - let settled = false; - const releaseState = (): void => { - state.stopClaimRefresh(); - if (deferredSpooledUpdateClaimsByKey.get(claimKey) === state) { - deferredSpooledUpdateClaimsByKey.delete(claimKey); - } - this.#deferredSpooledUpdateClaimKeys.delete(claimKey); - }; - const finish = async (result: TelegramMessageProcessingResult): Promise => { - if (settled) { - return; - } - settled = true; - if (state.timer) { - clearTimeout(state.timer); - } - if (result.kind === "completed") { - // Claim refresh must continue through tombstone retry, but durable - // adoption transfers cancellation ownership away from ingress. - params.onTurnAdopted(); - } - if (result.kind === "failed-retryable") { - releaseState(); - if (state.timedOutMessage) { - await this.#failTimedOutDeferredSpooledUpdate(state); - return; - } - await this.#releaseFailedSpooledUpdate({ - err: result.error, - update: params.update, - }); - return; - } - try { - await completeTelegramSpooledUpdateWithRetry({ - update: params.update, - abortSignal: this.opts.abortSignal, - onRetry: ({ attempt, delayMs, error }) => { - this.opts.log( - `[telegram][diag] spooled update ${params.update.updateId} buffered completion retry ${attempt} scheduled in ${formatDurationPrecise(delayMs)}: ${formatErrorMessage(error)}`, - ); - }, - }); - } catch (err) { - this.opts.log( - `[telegram][diag] spooled update ${params.update.updateId} completed after buffered processing but could not tombstone its claimed spool row: ${formatErrorMessage(err)}`, - ); - } finally { - releaseState(); - } - }; - const state: DeferredSpooledUpdateClaimState = { - claimKey, - laneKey: params.laneKey, - task: params.deferredWork.task.then(finish, async (err: unknown) => { - await finish({ kind: "failed-retryable", error: err }); - }), - update: params.update, - updateId: params.update.updateId, - stopClaimRefresh: params.stopClaimRefresh, - }; - state.timer = setTimeout(() => { - const age = formatDurationPrecise(this.#spooledUpdateHandlerTimeoutMs); - // Pre-adoption only: once the deferred participant settles at adoption, - // this timer is cleared. A fire means ingress never adopted the turn. - state.timedOutMessage = `Telegram isolated polling spool pre-adoption timed out behind update ${params.update.updateId} on lane ${params.laneKey} after ${age}; marking the update failed (handler-timeout) and keeping the claim out of retry.`; - params.deferredWork.settle({ - kind: "failed-retryable", - error: new Error(state.timedOutMessage), - }); - }, this.#spooledUpdateHandlerTimeoutMs); - state.timer.unref?.(); - deferredSpooledUpdateClaimsByKey.set(claimKey, state); - this.#deferredSpooledUpdateClaimKeys.add(claimKey); - } - - #isDeferredSpooledUpdateClaim(update: ClaimedTelegramSpooledUpdate): boolean { - return deferredSpooledUpdateClaimsByKey.has(buildDeferredSpooledUpdateClaimKey(update)); - } - - async #failTimedOutDeferredSpooledUpdate(state: DeferredSpooledUpdateClaimState): Promise { - const message = - state.timedOutMessage ?? - `Telegram isolated polling spool pre-adoption timed out behind update ${state.updateId} on lane ${state.laneKey}; marking the update failed.`; - try { - const failed = await failTelegramSpooledUpdateClaim({ - update: state.update, - reason: "handler-timeout", - message, - }); - if (!failed) { - this.opts.log( - `[telegram][diag] timed out pre-adoption spooled update ${state.updateId} no longer had a processing marker to fail.`, - ); - this.#status.notePollingError(message); - return; - } - } catch (err) { - this.opts.log( - `[telegram][diag] timed out pre-adoption spooled update ${state.updateId} could not be marked failed: ${formatErrorMessage(err)}`, - ); - this.#status.notePollingError(message); - return; - } - // Pre-adoption only: if a reply fence opened before adoption, release it. - const scopedReplyFenceLaneKey = buildTelegramReplyFenceLaneKey({ - accountId: this.opts.accountId, - sequentialKey: state.laneKey, - }); - const abortedReplyWork = supersedeTelegramReplyFenceLane(scopedReplyFenceLaneKey); - if (!abortedReplyWork) { - this.opts.log( - `[telegram][diag] timed out pre-adoption spooled update ${state.updateId} had no active reply fence on lane ${state.laneKey}.`, - ); - } - this.opts.log(`[telegram] ${message}`); - this.#status.notePollingError(message); - } - - async #releaseFailedSpooledUpdate(params: { - err: unknown; - update: ClaimedTelegramSpooledUpdate; - }): Promise { - const laneKey = this.#spooledUpdateLaneKey(params.update); - const nonRetryable = resolveNonRetryableSpooledUpdateFailure(params.err); - if (nonRetryable) { - try { - const failed = await failTelegramSpooledUpdateClaim({ - update: params.update, - reason: nonRetryable.reason, - message: nonRetryable.message, - }); - if (!failed) { - this.opts.log( - `[telegram][diag] spooled update ${params.update.updateId} failed with non-retryable ${nonRetryable.reason}, but no processing marker remained to dead-letter.`, - ); - return; - } - this.opts.log( - `[telegram][diag] spooled update ${params.update.updateId} failed with non-retryable ${nonRetryable.reason}; dead-lettered: ${nonRetryable.message}`, - ); - return; - } catch (failErr) { - this.opts.log( - `[telegram][diag] spooled update ${params.update.updateId} failed with non-retryable ${nonRetryable.reason}, but could not be dead-lettered: ${formatErrorMessage(failErr)}`, - ); - } - } - const attempt = resolveSpooledUpdateAttemptNumber(params.update); - if (shouldDeadLetterRetryableSpooledUpdate(params.update, attempt)) { - const message = formatErrorMessage(params.err); - try { - const failed = await failTelegramSpooledUpdateClaim({ - update: params.update, - reason: "retry-limit-exceeded", - message, - }); - if (!failed) { - this.opts.log( - `[telegram][diag] spooled update ${params.update.updateId} on lane ${laneKey} reached retry limit, but no processing marker remained to dead-letter.`, - ); - return; - } - // Retryable poison updates must eventually become tombstones, but not - // during ordinary transient provider or state-store outages. - this.opts.log( - `[telegram][warn] spooled update ${params.update.updateId} on lane ${laneKey} reached retry limit after ${attempt} attempts; dead-lettered: ${message}`, - ); - return; - } catch (failErr) { - this.opts.log( - `[telegram][diag] spooled update ${params.update.updateId} on lane ${laneKey} reached retry limit, but could not be dead-lettered: ${formatErrorMessage(failErr)}`, - ); - } - } - try { - await releaseTelegramSpooledUpdateClaim(params.update, { - lastError: formatErrorMessage(params.err), - }); - } catch (releaseErr) { - this.opts.log( - `[telegram][diag] spooled update ${params.update.updateId} failed and could not be requeued: ${formatErrorMessage(releaseErr)}`, - ); - return; - } - this.opts.log( - `[telegram][diag] spooled update ${params.update.updateId} failed; keeping for retry attempt ${attempt + 1}/${TELEGRAM_SPOOLED_RETRY_MAX_ATTEMPTS}: ${formatErrorMessage(params.err)}`, - ); - } - - async #waitForSpooledUpdateHandlers(): Promise { - await Promise.allSettled([ - ...[...this.#spooledUpdateHandlerKeys] - .map((handlerKey) => activeSpooledUpdateHandlersByLane.get(handlerKey)?.task) - .filter((task): task is Promise => Boolean(task)), - ...[...this.#deferredSpooledUpdateClaimKeys] - .map((claimKey) => deferredSpooledUpdateClaimsByKey.get(claimKey)?.task) - .filter((task): task is Promise => Boolean(task)), - ]); - } - - #spooledUpdateLaneKey(update: TelegramSpooledUpdate): string { - return this.#rawSpooledUpdateLaneKey(update.update); - } - - #rawSpooledUpdateLaneKey(update: unknown): string { - return getTelegramSequentialKey({ - update: update as Parameters[0]["update"], - ...(this.opts.botInfo ? { me: this.opts.botInfo } : {}), - }); - } - - #activeSpooledUpdateHandlerKeysForSpool(spoolDir: string): Set { - const handlerKeys = new Set(); - for (const handlerKey of activeSpooledUpdateHandlersByLane.keys()) { - if (isSpooledUpdateHandlerKeyForSpool(handlerKey, spoolDir)) { - handlerKeys.add(handlerKey); - } - } - return handlerKeys; - } - - #activeSpooledUpdateLaneKeysForSpool(spoolDir: string): Set { - const laneKeys = new Set(); - for (const handlerKey of this.#activeSpooledUpdateHandlerKeysForSpool(spoolDir)) { - const handler = activeSpooledUpdateHandlersByLane.get(handlerKey); - if (handler) { - laneKeys.add(handler.laneKey); - } - } - return laneKeys; - } - - async #drainSpooledUpdates(params: { - bot: TelegramBot; - isDrainHealthy: () => boolean; - shouldStop: () => boolean; spoolDir: string; - }): Promise { - const activeLaneKeys = this.#activeSpooledUpdateLaneKeysForSpool(params.spoolDir); - await recoverStaleTelegramSpooledUpdateClaims({ + }): ReturnType { + if (this.#ingressDrain) { + return this.#ingressDrain; + } + this.#ingressDrain = createTelegramTransportIngressDrain({ spoolDir: params.spoolDir, - staleMs: 0, - shouldRecover: (claim) => - !this.#isDeferredSpooledUpdateClaim(claim) && - !activeLaneKeys.has(this.#spooledUpdateLaneKey(claim)) && - !isTelegramSpooledUpdateClaimOwnedByOtherLiveProcess(claim), - shouldRecoverCorrupt: (claim) => - !(claim.laneKey && activeLaneKeys.has(claim.laneKey)) && - !isTelegramSpooledCorruptClaimOwnedByOtherLiveProcess(claim), - }); - const claimedLaneKeys = new Set( - ( - await listTelegramSpooledUpdateClaims({ - spoolDir: params.spoolDir, - }) - ) - .filter((claim) => !this.#isDeferredSpooledUpdateClaim(claim)) - .map((claim) => this.#spooledUpdateLaneKey(claim)), - ); - const updates = await listTelegramSpooledUpdates({ - spoolDir: params.spoolDir, - limit: TELEGRAM_SPOOLED_DRAIN_SCAN_LIMIT, - }); - const candidateUpdateIds = updates.map((update) => update.updateId); - const blockedByLane = new Set(); - const retryDelayedLaneKeys = new Set(); - for (const update of updates) { - const laneKey = this.#spooledUpdateLaneKey(update); - const handlerKey = buildSpooledUpdateHandlerKey({ spoolDir: params.spoolDir, laneKey }); - if (activeSpooledUpdateHandlersByLane.has(handlerKey)) { - blockedByLane.add(handlerKey); - } - // Release increments attempts and stamps lastAttemptAt. The drain blocks - // that lane until the retry window expires so poison rows cannot hot-loop. - if (resolveSpooledUpdateRetryDelayMs(update) > 0) { - retryDelayedLaneKeys.add(laneKey); - } - } - const blockedLaneKeys = new Set([ - ...activeLaneKeys, - ...claimedLaneKeys, - ...retryDelayedLaneKeys, - ]); - let started = 0; - while (started < TELEGRAM_SPOOLED_DRAIN_START_LIMIT) { - if (params.shouldStop() || this.opts.abortSignal?.aborted) { - break; - } - const claimedUpdate = await this.#claimNextSpooledUpdate({ - blockedLaneKeys, - candidateUpdateIds, - spoolDir: params.spoolDir, - }); - if (!claimedUpdate) { - break; - } - if (params.shouldStop() || this.opts.abortSignal?.aborted) { - try { - await abandonTelegramSpooledUpdateClaim(claimedUpdate); - } catch (err) { - this.opts.log( - `[telegram][diag] spooled update ${claimedUpdate.updateId} could not be requeued after its polling cycle ended: ${formatErrorMessage(err)}`, - ); - } - break; - } - const laneKey = this.#spooledUpdateLaneKey(claimedUpdate); - const handlerKey = buildSpooledUpdateHandlerKey({ spoolDir: params.spoolDir, laneKey }); - if (activeSpooledUpdateHandlersByLane.has(handlerKey)) { - blockedByLane.add(handlerKey); - await abandonTelegramSpooledUpdateClaim(claimedUpdate); - blockedLaneKeys.add(laneKey); - continue; - } - let abortReplyWorkOnClaimRefreshFailure = true; - const stopClaimRefresh = this.#startSpooledUpdateClaimRefresh( - claimedUpdate, - params.isDrainHealthy, - () => { - if (!abortReplyWorkOnClaimRefreshFailure) { - return; - } - const scopedReplyFenceLaneKey = buildTelegramReplyFenceLaneKey({ - accountId: this.opts.accountId, - sequentialKey: laneKey, - }); - const abortedReplyWork = supersedeTelegramReplyFenceLane(scopedReplyFenceLaneKey); - if (!abortedReplyWork) { - this.opts.log( - `[telegram][diag] spooled update ${claimedUpdate.updateId} drain heartbeat expired without an active reply fence on lane ${laneKey}; stopping claim refresh.`, - ); - } - }, - ); - const handler = this.#handleClaimedSpooledUpdate({ - bot: params.bot, - onTurnAdopted: () => { - abortReplyWorkOnClaimRefreshFailure = false; - }, - stopClaimRefresh, - update: claimedUpdate, - }); - const state: SpooledUpdateHandlerState = { - handlerKey, - laneKey, - task: handler, - update: claimedUpdate, - updateId: claimedUpdate.updateId, - startedAt: Date.now(), - stopClaimRefresh, - }; - activeSpooledUpdateHandlersByLane.set(handlerKey, state); - this.#spooledUpdateHandlerKeys.add(handlerKey); - blockedLaneKeys.add(laneKey); - void handler.finally(() => { - if ( - !deferredSpooledUpdateClaimsByKey.has(buildDeferredSpooledUpdateClaimKey(claimedUpdate)) - ) { - state.stopClaimRefresh(); - } - if (activeSpooledUpdateHandlersByLane.get(handlerKey) === state) { - activeSpooledUpdateHandlersByLane.delete(handlerKey); - } - this.#spooledUpdateHandlerKeys.delete(handlerKey); - }); - started += 1; - } - return { blockedByLane, started }; - } - - #detectTimedOutSpooledHandler( - blockedHandlerKeys: Set, - ): { handler: SpooledUpdateHandlerState; ageMs: number } | null { - const now = Date.now(); - let timedOut: { handler: SpooledUpdateHandlerState; ageMs: number } | null = null; - for (const handlerKey of blockedHandlerKeys) { - const handler = activeSpooledUpdateHandlersByLane.get(handlerKey); - if (!handler || handler.timedOutAt !== undefined) { - continue; - } - const ageMs = now - handler.startedAt; - if (ageMs < this.#spooledUpdateHandlerTimeoutMs) { - continue; - } - if (!timedOut || ageMs > timedOut.ageMs) { - timedOut = { handler, ageMs }; - } - } - return timedOut; - } - - async #recoverTimedOutSpooledHandler( - blockedHandlerKeys: Set, - ): Promise<{ handlerKey: string; restart: boolean } | null> { - const timedOutHandler = this.#detectTimedOutSpooledHandler(blockedHandlerKeys); - if (!timedOutHandler) { - return null; - } - const handler = timedOutHandler.handler; - const activeHandler = activeSpooledUpdateHandlersByLane.get(handler.handlerKey); - if (!activeHandler || activeHandler !== handler) { - return null; - } - const age = formatDurationPrecise(timedOutHandler.ageMs); - activeHandler.timedOutAt = Date.now(); - activeHandler.stopClaimRefresh(); - // Pre-adoption stall: the active handler should return once deferred work - // is registered. A timeout here means ingress never reached adoption. - const message = `Telegram isolated polling spool handler timed out behind update ${handler.updateId} on lane ${handler.laneKey} after ${age}; marking the update failed (handler-timeout / pre-adoption) and restarting isolated ingress so later updates can drain.`; - activeHandler.timeoutMessage = message; - try { - const failed = await failTelegramSpooledUpdateClaim({ - update: handler.update, - reason: "handler-timeout", - message, - }); - if (!failed) { - this.opts.log( - `[telegram][diag] timed out spooled update ${handler.updateId} no longer had a processing marker to fail.`, - ); - this.#status.notePollingError(message); - return { handlerKey: handler.handlerKey, restart: false }; - } - } catch (err) { - this.opts.log( - `[telegram][diag] timed out spooled update ${handler.updateId} could not be marked failed: ${formatErrorMessage(err)}`, - ); - this.#status.notePollingError(message); - return { handlerKey: handler.handlerKey, restart: false }; - } - // Best-effort: supersede any reply fence already opened during pre-adoption - // setup so a wedged handleUpdate can return. After adoption the spool no - // longer owns the turn, so this path should not see a settled agent run. - const scopedReplyFenceLaneKey = buildTelegramReplyFenceLaneKey({ + bot: params.bot, + cfg: this.opts.config, accountId: this.opts.accountId, - sequentialKey: handler.laneKey, - }); - const abortedReplyWork = supersedeTelegramReplyFenceLane(scopedReplyFenceLaneKey); - if (!abortedReplyWork) { - this.opts.log( - `[telegram][diag] timed out spooled update ${handler.updateId} had no active reply fence on lane ${handler.laneKey}; keeping the lane guarded until the handler stops.`, - ); - } - const handlerStopped = await waitForSpooledHandlerTaskSettlement({ - task: handler.task, - timeoutMs: this.#spooledUpdateHandlerAbortGraceMs, + botInfo: this.opts.botInfo, + adoptionStallTimeoutMs: this.#spooledUpdateHandlerTimeoutMs, abortSignal: this.opts.abortSignal, + onLog: (message) => this.opts.log(message), }); - if ( - !handlerStopped && - activeSpooledUpdateHandlersByLane.get(handler.handlerKey) === activeHandler - ) { - this.opts.log( - `[telegram][diag] timed out spooled update ${handler.updateId} did not stop within ${formatDurationPrecise(this.#spooledUpdateHandlerAbortGraceMs)} after reply abort; keeping lane ${handler.laneKey} guarded.`, - ); - this.#status.notePollingError(message); - return { handlerKey: handler.handlerKey, restart: false }; - } - if (activeSpooledUpdateHandlersByLane.get(handler.handlerKey) === activeHandler) { - activeSpooledUpdateHandlersByLane.delete(handler.handlerKey); - } - this.#spooledUpdateHandlerKeys.delete(handler.handlerKey); - this.opts.log(`[telegram] ${message}`); - this.#status.notePollingError(message); - return { handlerKey: handler.handlerKey, restart: true }; + return this.#ingressDrain; } - #noteSpooledBacklogStalls(blockedHandlerKeys: Set): Set { - const stalled = new Set(); - const now = Date.now(); - for (const handlerKey of blockedHandlerKeys) { - const handler = activeSpooledUpdateHandlersByLane.get(handlerKey); - if (!handler || handler.timedOutAt !== undefined) { - continue; - } - const ageMs = now - handler.startedAt; - if (ageMs < ISOLATED_INGRESS_BACKLOG_STALL_MS) { - continue; - } - stalled.add(handlerKey); - if (!handler.backlogStatusMessage) { - handler.backlogStatusMessage = `Telegram isolated polling spool backlog stalled behind update ${handler.updateId} on lane ${handler.laneKey} for ${formatDurationPrecise(ageMs)}; marking polling unhealthy until the backlog drains.`; - this.#status.notePollingError(handler.backlogStatusMessage); - } + /** Pump the core-owned durable ingress drain for this session's spool. */ + async #pumpSpooledDrain(params: { + bot: TelegramBot; + spoolDir: string; + shouldStop: () => boolean; + }): Promise<{ started: number }> { + if (params.shouldStop()) { + return { started: 0 }; } - return stalled; + const drain = this.#getOrCreateSpooledDrain(params); + return await drain.drainOnce({ shouldStop: params.shouldStop }); } async #runIsolatedIngressCycle(bot: TelegramBot): Promise<"continue" | "exit"> { @@ -1224,7 +447,6 @@ export class TelegramPollingSession { const forceCyclePromise = new Promise((resolve) => { forceCycleResolve = resolve; }); - const stalledBacklogKeys = new Set(); let requestImmediateDrain: () => void = () => undefined; let drainRequested = false; let cycleEnding = false; @@ -1263,7 +485,7 @@ export class TelegramPollingSession { liveness.noteGetUpdatesSuccessCount(message.count, message.finishedAt); liveness.noteGetUpdatesFinished(); this.#noteHealthyPollingCycle(); - if (!restartRequested && stalledBacklogKeys.size === 0) { + if (!restartRequested) { this.#status.notePollSuccess(message.finishedAt); } this.#maybeDrainPendingDeliveries(message.finishedAt); @@ -1283,7 +505,7 @@ export class TelegramPollingSession { void writeTelegramSpooledUpdate({ spoolDir, update: message.update, - laneKey: this.#rawSpooledUpdateLaneKey(message.update), + laneKey: telegramSpooledUpdateLaneKey(message.update, this.opts.botInfo), }).then( (updateId) => { ackSpooledUpdate(message.requestId, { ok: true, updateId }); @@ -1310,11 +532,9 @@ export class TelegramPollingSession { this.opts.abortSignal?.addEventListener("abort", stopOnAbort, { once: true }); const drainIntervalMs = Math.max(100, Math.floor(ingress.drainIntervalMs ?? 500)); let drainActive = false; - const drainHealth = getSpooledUpdateDrainHealth(spoolDir); + const drainHealth = { lastCompletedAt: Date.now() }; // Fail closed when the spool stops making progress: keeping any claim live would // prevent a healthy process from recovering a wedged drain. - const isDrainHealthy = () => - Date.now() - drainHealth.lastCompletedAt <= TELEGRAM_SPOOLED_CLAIM_HEALTH_GRACE_MS; const stopBot = () => { return Promise.resolve(bot.stop()) .then(() => undefined) @@ -1359,48 +579,12 @@ export class TelegramPollingSession { drainRequested = false; let drainCompleted = false; try { - const drain = await this.#drainSpooledUpdates({ + await this.#pumpSpooledDrain({ bot, - isDrainHealthy, - shouldStop: () => cycleEnding, spoolDir, + shouldStop: () => cycleEnding, }); consecutiveDrainFailures = 0; - for (const handlerKey of stalledBacklogKeys) { - if ( - !activeSpooledUpdateHandlersByLane.has(handlerKey) || - !drain.blockedByLane.has(handlerKey) - ) { - stalledBacklogKeys.delete(handlerKey); - } - } - for (const handlerKey of drain.blockedByLane) { - const handler = activeSpooledUpdateHandlersByLane.get(handlerKey); - if (handler?.timedOutAt === undefined) { - continue; - } - stalledBacklogKeys.add(handlerKey); - if (handler.timeoutMessage) { - this.#status.notePollingError(handler.timeoutMessage); - } - } - for (const handlerKey of this.#noteSpooledBacklogStalls(drain.blockedByLane)) { - stalledBacklogKeys.add(handlerKey); - } - // Active handlers can outlive their owning session after shutdown grace. - // Recover every handler for this spool, including lone handlers with no backlog. - const timeoutCandidateHandlerKeys = this.#activeSpooledUpdateHandlerKeysForSpool(spoolDir); - for (const handlerKey of drain.blockedByLane) { - timeoutCandidateHandlerKeys.add(handlerKey); - } - const timedOutRecovery = await this.#recoverTimedOutSpooledHandler( - timeoutCandidateHandlerKeys, - ); - if (timedOutRecovery?.restart) { - requestStopForRestart(); - } else if (timedOutRecovery) { - stalledBacklogKeys.add(timedOutRecovery.handlerKey); - } drainCompleted = true; } catch (err) { consecutiveDrainFailures += 1; @@ -1520,8 +704,9 @@ export class TelegramPollingSession { await stopWorker(); if (!restartRequested) { await drainOnce(); - await waitForGracefulStop(() => this.#waitForSpooledUpdateHandlers()); } + this.#ingressDrain?.dispose(); + this.#ingressDrain = undefined; await waitForGracefulStop(stopBot); if (this.#activeCycleAbort === cycleAbortController) { this.#activeCycleAbort = undefined; diff --git a/extensions/telegram/src/spooled-update-retry-policy.ts b/extensions/telegram/src/spooled-update-retry-policy.ts deleted file mode 100644 index 6410cc44312d..000000000000 --- a/extensions/telegram/src/spooled-update-retry-policy.ts +++ /dev/null @@ -1,87 +0,0 @@ -// Telegram plugin module shares spooled update retry policy. -import { - collectErrorGraphCandidates, - formatErrorMessage, - readErrorName, -} from "openclaw/plugin-sdk/error-runtime"; -import type { BackoffPolicy } from "openclaw/plugin-sdk/runtime-env"; -import { computeBackoff } from "openclaw/plugin-sdk/runtime-env"; -import { isTelegramMessageDispatchReplayForgetError } from "./message-dispatch-dedupe.js"; -import type { TelegramSpooledUpdate } from "./telegram-ingress-spool.types.js"; - -export const TELEGRAM_SPOOLED_RETRY_MAX_ATTEMPTS = 8; -const TELEGRAM_SPOOLED_RETRY_DEAD_LETTER_MIN_AGE_MS = 24 * 60 * 60 * 1000; -const TELEGRAM_SPOOLED_RETRY_BASE_MS = 1_000; -const TELEGRAM_SPOOLED_RETRY_MAX_MS = 3 * 60_000; -const TELEGRAM_SPOOLED_COMPLETION_RETRY_POLICY: BackoffPolicy = { - initialMs: 250, - maxMs: 5_000, - factor: 2, - jitter: 0.2, -}; - -const MISSING_AGENT_HARNESS_ERROR_NAME = "MissingAgentHarnessError"; -const MISSING_AGENT_HARNESS_MESSAGE_RE = /Requested agent harness "[^"]+" is not registered\./u; - -type NonRetryableSpooledUpdateFailure = { - reason: "missing-agent-harness" | "dispatch-dedupe-rollback-failed"; - message: string; -}; - -export function resolveNonRetryableSpooledUpdateFailure( - err: unknown, -): NonRetryableSpooledUpdateFailure | null { - for (const candidate of collectErrorGraphCandidates(err, (current) => [ - current.cause, - current.error, - ])) { - const message = formatErrorMessage(candidate); - if (isTelegramMessageDispatchReplayForgetError(candidate)) { - // A committed dispatch key that cannot be rolled back makes retry unsafe: - // the next replay can be duplicate-suppressed and then deleted. - return { reason: "dispatch-dedupe-rollback-failed", message }; - } - if ( - readErrorName(candidate) === MISSING_AGENT_HARNESS_ERROR_NAME || - MISSING_AGENT_HARNESS_MESSAGE_RE.test(message) - ) { - return { reason: "missing-agent-harness", message }; - } - } - return null; -} - -export function resolveSpooledUpdateRetryDelayMs( - update: TelegramSpooledUpdate, - now = Date.now(), -): number { - const attempts = update.attempts ?? 0; - if (!update.lastError || update.lastAttemptAt === undefined || attempts <= 0) { - return 0; - } - const exponent = Math.min(attempts - 1, 8); - const delayMs = Math.min( - TELEGRAM_SPOOLED_RETRY_MAX_MS, - TELEGRAM_SPOOLED_RETRY_BASE_MS * 2 ** exponent, - ); - return Math.max(0, update.lastAttemptAt + delayMs - now); -} - -export function resolveSpooledUpdateAttemptNumber(update: TelegramSpooledUpdate): number { - return (update.attempts ?? 0) + 1; -} - -export function resolveSpooledUpdatePersistenceRetryDelayMs(attempt: number): number { - return computeBackoff(TELEGRAM_SPOOLED_COMPLETION_RETRY_POLICY, attempt); -} - -export function shouldDeadLetterRetryableSpooledUpdate( - update: TelegramSpooledUpdate, - attempt: number, - now = Date.now(), -): boolean { - return ( - attempt >= TELEGRAM_SPOOLED_RETRY_MAX_ATTEMPTS && - now - update.receivedAt >= TELEGRAM_SPOOLED_RETRY_DEAD_LETTER_MIN_AGE_MS - ); -} diff --git a/extensions/telegram/src/telegram-ingress-claim-owner.ts b/extensions/telegram/src/telegram-ingress-claim-owner.ts deleted file mode 100644 index c0cede2ca144..000000000000 --- a/extensions/telegram/src/telegram-ingress-claim-owner.ts +++ /dev/null @@ -1,185 +0,0 @@ -// Telegram plugin module implements telegram ingress claim-owner identity. -import childProcess from "node:child_process"; -import { randomUUID } from "node:crypto"; -import fsSync from "node:fs"; -import type { ChannelIngressQueueCorruptClaim } from "openclaw/plugin-sdk/channel-outbound"; -import type { - ClaimedTelegramSpooledUpdate, - TelegramSpooledUpdateClaimOwner, -} from "./telegram-ingress-spool.types.js"; - -// Liveness default: a claim older than its lease is never live-owner protected, -// so recovery can reclaim it even when the owner process still exists. -const TELEGRAM_SPOOLED_UPDATE_CLAIM_LEASE_MS = 30 * 60 * 1000; - -type TelegramSpooledClaimLivenessOptions = { - maxAgeMs?: number; - now?: number; - /** Test seam for PID existence (including Linux TID impersonation). */ - processExists?: (pid: number) => boolean; - /** Test seam for process start-time identity. */ - readProcessStartTime?: (pid: number) => number | null; -}; - -function readProcessStartTime(pid: number): number | null { - if (!Number.isSafeInteger(pid) || pid <= 0) { - return null; - } - if (process.platform === "darwin") { - try { - const startedAt = childProcess - .execFileSync("/bin/ps", ["-o", "lstart=", "-p", String(pid)], { - encoding: "utf8", - env: { ...process.env, LC_ALL: "C", TZ: "UTC" }, - stdio: ["ignore", "pipe", "ignore"], - }) - .trim(); - const startedAtMs = Date.parse(`${startedAt} UTC`); - return Number.isFinite(startedAtMs) ? Math.floor(startedAtMs / 1000) : null; - } catch { - return null; - } - } - if (process.platform !== "linux") { - return null; - } - try { - const stat = fsSync.readFileSync(`/proc/${pid}/stat`, "utf8"); - const commEndIndex = stat.lastIndexOf(")"); - if (commEndIndex < 0) { - return null; - } - const afterComm = stat.slice(commEndIndex + 1).trimStart(); - const fields = afterComm.split(/\s+/); - // field 22 (starttime) = index 19 after the comm-split (field 3 is index 0). - const starttime = Number(fields[19]); - return Number.isInteger(starttime) && starttime >= 0 ? starttime : null; - } catch { - return null; - } -} - -const TELEGRAM_SPOOLED_UPDATE_PROCESS_START_TIME = readProcessStartTime(process.pid); -// ownerId = pid:startToken:uuid. Starttime binds the PID to one process instance so -// Linux TIDs and recycled PIDs cannot impersonate a dead claim owner. -export const TELEGRAM_SPOOLED_UPDATE_PROCESS_ID = [ - process.pid, - TELEGRAM_SPOOLED_UPDATE_PROCESS_START_TIME ?? "x", - randomUUID(), -].join(":"); - -export function processPidFromOwnerId(ownerId: string): number { - const pid = Number.parseInt(ownerId.split(":", 1)[0] ?? "", 10); - return Number.isSafeInteger(pid) && pid > 0 ? pid : -1; -} - -// Canonical ownerId: pid:startToken:uuid. startToken is a numeric starttime, or -// the explicit "x" sentinel when the writer cannot supply one (win32). -type OwnerStartToken = - | { kind: "numeric"; value: number } - | { kind: "existence-only" } - | { kind: "missing" }; - -function parseOwnerStartToken(ownerId: string): OwnerStartToken { - const parts = ownerId.split(":"); - // Legacy pid:uuid owners (pre start-token releases) carry no instance binding. - // Keep existence-based liveness for them: reclaiming a fresh claim from a live - // old-version worker during a rolling upgrade would double-dispatch its update. - if (parts.length === 2) { - return { kind: "existence-only" }; - } - if (parts.length < 2) { - return { kind: "missing" }; - } - const startField = parts[1] ?? ""; - // Explicit "x": writer ran on a platform with no readable starttime (win32). - if (startField === "x") { - return { kind: "existence-only" }; - } - const starttime = Number(startField); - if (Number.isSafeInteger(starttime) && starttime >= 0) { - return { kind: "numeric", value: starttime }; - } - return { kind: "missing" }; -} - -function processExists(pid: number): boolean { - if (!Number.isSafeInteger(pid) || pid <= 0) { - return false; - } - try { - process.kill(pid, 0); - return true; - } catch (err) { - const code = (err as { code?: string }).code; - return code !== "ESRCH" && code !== "EINVAL"; - } -} - -function isFreshClaimOwner( - claim: TelegramSpooledUpdateClaimOwner, - options?: { maxAgeMs?: number; now?: number }, -): boolean { - const now = options?.now ?? Date.now(); - const maxAgeMs = options?.maxAgeMs ?? TELEGRAM_SPOOLED_UPDATE_CLAIM_LEASE_MS; - return now - claim.claimedAt < maxAgeMs; -} - -function isClaimOwnerProcessInstanceLive( - claim: Pick, - options?: TelegramSpooledClaimLivenessOptions, -): boolean { - const exists = options?.processExists ?? processExists; - const readStart = options?.readProcessStartTime ?? readProcessStartTime; - if (!exists(claim.processPid)) { - return false; - } - const startToken = parseOwnerStartToken(claim.processId); - if (startToken.kind === "missing") { - // Legacy/malformed owner ids have no process-instance binding; reclaim. - return false; - } - if (startToken.kind === "existence-only") { - // Legacy or `x` owners cannot prove instance identity. Fall back to - // processExists-only liveness — the pre-starttime lease contract — instead - // of stealing a fresh claim from a possibly live worker. - return true; - } - const actualStart = readStart(claim.processPid); - if (actualStart === null) { - // Starttime unreadable while the PID appears live. Keep lease protection - // via process existence so a readable-starttime peer is not stolen mid-run. - return true; - } - return actualStart === startToken.value; -} - -export function isTelegramSpooledUpdateClaimOwnedByOtherLiveProcess( - claim: ClaimedTelegramSpooledUpdate, - options?: TelegramSpooledClaimLivenessOptions, -): boolean { - return Boolean( - claim.claim && - claim.claim.processId !== TELEGRAM_SPOOLED_UPDATE_PROCESS_ID && - claim.claim.processPid !== process.pid && - isFreshClaimOwner(claim.claim, options) && - isClaimOwnerProcessInstanceLive(claim.claim, options), - ); -} - -export function isTelegramSpooledCorruptClaimOwnedByOtherLiveProcess( - claim: ChannelIngressQueueCorruptClaim, - options?: TelegramSpooledClaimLivenessOptions, -): boolean { - const processId = claim.claim.ownerId; - const processPid = processPidFromOwnerId(processId); - const owner = { processId, processPid, claimedAt: claim.claim.claimedAt }; - if (processId === TELEGRAM_SPOOLED_UPDATE_PROCESS_ID) { - return isFreshClaimOwner(owner, options); - } - return ( - processPid !== process.pid && - isFreshClaimOwner(owner, options) && - isClaimOwnerProcessInstanceLive(owner, options) - ); -} diff --git a/extensions/telegram/src/telegram-ingress-drain-factory.ts b/extensions/telegram/src/telegram-ingress-drain-factory.ts new file mode 100644 index 000000000000..d95ed753cb14 --- /dev/null +++ b/extensions/telegram/src/telegram-ingress-drain-factory.ts @@ -0,0 +1,65 @@ +// Telegram plugin module builds transport-shared durable ingress drains. +import type { ChannelIngressDrain } from "openclaw/plugin-sdk/channel-outbound"; +import type { OpenClawConfig } from "openclaw/plugin-sdk/config-contracts"; +import type { TelegramBotInfo } from "./bot-info.js"; +import type { TelegramMessageProcessingResult } from "./bot-processing-outcome.js"; +import { + createTelegramIngressDrain, + resolveTelegramAdoptionStallTimeoutMs, + type TelegramIngressDrainLifecycle, +} from "./telegram-ingress-drain.js"; +import { openTelegramIngressQueue } from "./telegram-ingress-spool.js"; + +export type TelegramSpooledBot = { + handleUpdate: (update: never) => Promise; +}; + +export type CreateTelegramTransportIngressDrainParams = { + spoolDir: string; + bot: TelegramSpooledBot; + cfg: OpenClawConfig; + accountId: string; + botInfo?: TelegramBotInfo; + adoptionStallTimeoutMs?: number; + onLog?: (message: string) => void; + abortSignal?: AbortSignal; + /** + * Optional override for full dispatch (tests). Default: bot.handleUpdate under + * the drain lifecycle via bot-message spooled replay path. + */ + dispatchUpdate?: ( + update: unknown, + lifecycle: TelegramIngressDrainLifecycle, + ) => Promise; +}; + +/** + * One drain for polling + webhook: claim → dispatch with turnAdoptionLifecycle → + * complete at adoption. Transport code only enqueues then pumps drainOnce(). + */ +export function createTelegramTransportIngressDrain( + params: CreateTelegramTransportIngressDrainParams, +): ChannelIngressDrain { + const queue = openTelegramIngressQueue(params.spoolDir); + const adoptionStallTimeoutMs = resolveTelegramAdoptionStallTimeoutMs({ + configured: params.adoptionStallTimeoutMs, + env: process.env, + }); + return createTelegramIngressDrain({ + queue, + cfg: params.cfg, + accountId: params.accountId, + botInfo: params.botInfo, + adoptionStallTimeoutMs, + ...(params.onLog ? { onLog: params.onLog } : {}), + ...(params.abortSignal ? { abortSignal: params.abortSignal } : {}), + dispatch: async (update, lifecycle) => { + if (params.dispatchUpdate) { + return await params.dispatchUpdate(update, lifecycle); + } + // Lifecycle is also on the spooled ALS frame (runWithTelegramSpooledReplayUpdate). + // bot-message merges it into turnAdoptionLifecycle for complete-at-adoption. + await params.bot.handleUpdate(update as never); + }, + }); +} diff --git a/extensions/telegram/src/telegram-ingress-drain.ts b/extensions/telegram/src/telegram-ingress-drain.ts new file mode 100644 index 000000000000..ad4fef94fae7 --- /dev/null +++ b/extensions/telegram/src/telegram-ingress-drain.ts @@ -0,0 +1,176 @@ +// Telegram plugin module owns the channel-side durable ingress drain adapter. +import { + bindIngressLifecycleToReplyOptions, + createChannelIngressDrain, + DEFAULT_INGRESS_ADOPTION_STALL_MS, + DEFAULT_INGRESS_RETRY_DEAD_LETTER_MIN_AGE_MS, + DEFAULT_INGRESS_RETRY_MAX_ATTEMPTS, + type ChannelIngressDrain, + type ChannelIngressQueue, +} from "openclaw/plugin-sdk/channel-outbound"; +import type { OpenClawConfig } from "openclaw/plugin-sdk/config-contracts"; +import { clampPositiveTimerTimeoutMs } from "openclaw/plugin-sdk/number-runtime"; +import type { TelegramBotInfo } from "./bot-info.js"; +import { + runWithTelegramSpooledReplayUpdate, + type TelegramMessageProcessingResult, +} from "./bot-processing-outcome.js"; +import { getTelegramSequentialKey } from "./sequential-key.js"; +import { resolveTelegramIngressNonRetryableFailure } from "./telegram-ingress-non-retryable.js"; +import type { TelegramSpooledUpdatePayload } from "./telegram-ingress-spool.payload.js"; +import { createShouldSupersedeTelegramSpooledPending } from "./telegram-ingress-supersede.js"; + +export const TELEGRAM_SPOOLED_HANDLER_TIMEOUT_ENV = "OPENCLAW_TELEGRAM_SPOOLED_HANDLER_TIMEOUT_MS"; +const TELEGRAM_SPOOLED_DRAIN_START_LIMIT = 100; +const TELEGRAM_SPOOLED_DRAIN_SCAN_LIMIT = TELEGRAM_SPOOLED_DRAIN_START_LIMIT * 10; + +export function resolveTelegramAdoptionStallTimeoutMs(params: { + configured?: number; + env?: NodeJS.ProcessEnv; +}): number { + const candidates = [ + params.configured, + Number(params.env?.[TELEGRAM_SPOOLED_HANDLER_TIMEOUT_ENV]), + ]; + for (const candidate of candidates) { + const timeoutMs = clampPositiveTimerTimeoutMs(candidate); + if (timeoutMs !== undefined) { + return timeoutMs; + } + } + return DEFAULT_INGRESS_ADOPTION_STALL_MS; +} + +export function telegramSpooledLaneKey(update: unknown, botInfo?: TelegramBotInfo): string { + return getTelegramSequentialKey({ + update: update as Parameters[0]["update"], + ...(botInfo ? { me: botInfo } : {}), + }); +} + +export type TelegramIngressDrainLifecycle = { + abortSignal: AbortSignal; + onAdopted: () => void | Promise; + onDeferred: () => void; + onAdoptionFinalizing: () => void; + onAbandoned: () => void; +}; + +export type TelegramIngressDrainDispatch = ( + update: unknown, + lifecycle: TelegramIngressDrainLifecycle, +) => Promise | TelegramMessageProcessingResult | void; + +export type CreateTelegramIngressDrainParams = { + queue: ChannelIngressQueue; + /** Required for authorization-gated supersede (numeric allowlist). */ + cfg: OpenClawConfig; + accountId: string; + botInfo?: TelegramBotInfo; + adoptionStallTimeoutMs?: number; + dispatch: TelegramIngressDrainDispatch; + onLog?: (message: string) => void; + abortSignal?: AbortSignal; +}; + +/** + * Shared polling/webhook drain over the core channel-ingress worker. + * + * room_event ambient work shares the sequential lane with the parent chat so a + * later user turn can supersede it pre-adoption; adopted user turns are never + * touched (core drain supersede is pre-adoption only). + */ +export function createTelegramIngressDrain( + params: CreateTelegramIngressDrainParams, +): ChannelIngressDrain { + return createChannelIngressDrain({ + queue: params.queue, + adoptionStallTimeoutMs: params.adoptionStallTimeoutMs ?? DEFAULT_INGRESS_ADOPTION_STALL_MS, + orderBy: "id", + scanLimit: TELEGRAM_SPOOLED_DRAIN_SCAN_LIMIT, + startLimit: TELEGRAM_SPOOLED_DRAIN_START_LIMIT, + retryPolicy: { + maxAttempts: DEFAULT_INGRESS_RETRY_MAX_ATTEMPTS, + deadLetterMinAgeMs: DEFAULT_INGRESS_RETRY_DEAD_LETTER_MIN_AGE_MS, + }, + resolveNonRetryableFailure: resolveTelegramIngressNonRetryableFailure, + shouldSupersedePending: createShouldSupersedeTelegramSpooledPending({ + cfg: params.cfg, + accountId: params.accountId, + ...(params.botInfo?.username ? { botUsername: params.botInfo.username } : {}), + }), + deriveLaneKey: (record) => telegramSpooledLaneKey(record.payload.update, params.botInfo), + ...(params.onLog ? { onLog: params.onLog } : {}), + ...(params.abortSignal ? { abortSignal: params.abortSignal } : {}), + dispatchClaimedEvent: async (event, lifecycle) => { + const bound = bindIngressLifecycleToReplyOptions(lifecycle); + const drainLifecycle: TelegramIngressDrainLifecycle = { + abortSignal: bound.turnAdoptionLifecycle.abortSignal, + onAdopted: bound.turnAdoptionLifecycle.onAdopted, + onDeferred: bound.turnAdoptionLifecycle.onDeferred, + onAdoptionFinalizing: lifecycle.onAdoptionFinalizing, + onAbandoned: bound.turnAdoptionLifecycle.onAbandoned, + }; + try { + const result = await runWithTelegramSpooledReplayUpdate( + event.payload.update as object, + async () => await params.dispatch(event.payload.update, drainLifecycle), + drainLifecycle, + ); + // Propagate explicit dispatch outcomes first. + const outcome = result.value; + if (outcome && typeof outcome === "object" && "kind" in outcome) { + if (outcome.kind === "failed-retryable") { + return { kind: "failed-retryable", error: outcome.error }; + } + if (outcome.kind === "completed" || outcome.kind === "skipped") { + return { kind: "completed" }; + } + } + // deferredWork exists for every spooled participant, not only genuine + // queued followups. Await the terminal participant outcome (or drain + // abort from guillotine/supersede) so failed-retryable releases and + // stalls do not hang the dispatch task forever. + const participant = result.deferredWork; + if (participant) { + const terminal = await new Promise((resolve, reject) => { + if (drainLifecycle.abortSignal.aborted) { + reject(drainLifecycle.abortSignal.reason ?? new Error("ingress-aborted")); + return; + } + const onAbort = () => { + reject(drainLifecycle.abortSignal.reason ?? new Error("ingress-aborted")); + }; + drainLifecycle.abortSignal.addEventListener("abort", onAbort, { once: true }); + void participant.task.then( + (value) => { + drainLifecycle.abortSignal.removeEventListener("abort", onAbort); + resolve(value); + }, + (error: unknown) => { + drainLifecycle.abortSignal.removeEventListener("abort", onAbort); + reject(error); + }, + ); + }).then( + (value) => value, + (error: unknown) => { + // Guillotine/supersede already own settleOnce — do not re-fail. + if (drainLifecycle.abortSignal.aborted) { + return { kind: "skipped" as const }; + } + throw error; + }, + ); + if (terminal.kind === "failed-retryable") { + return { kind: "failed-retryable", error: terminal.error }; + } + return { kind: "completed" }; + } + return { kind: "completed" }; + } catch (error) { + return { kind: "failed-retryable", error }; + } + }, + }); +} diff --git a/extensions/telegram/src/telegram-ingress-non-retryable.ts b/extensions/telegram/src/telegram-ingress-non-retryable.ts new file mode 100644 index 000000000000..933e4eccac47 --- /dev/null +++ b/extensions/telegram/src/telegram-ingress-non-retryable.ts @@ -0,0 +1,39 @@ +// Telegram plugin module classifies non-retryable spooled dispatch failures. +import { + collectErrorGraphCandidates, + formatErrorMessage, + readErrorName, +} from "openclaw/plugin-sdk/error-runtime"; +import { isTelegramMessageDispatchReplayForgetError } from "./message-dispatch-dedupe.js"; + +const MISSING_AGENT_HARNESS_ERROR_NAME = "MissingAgentHarnessError"; +const MISSING_AGENT_HARNESS_MESSAGE_RE = /Requested agent harness "[^"]+" is not registered\./u; + +export type TelegramIngressNonRetryableFailure = { + reason: "missing-agent-harness" | "dispatch-dedupe-rollback-failed"; + message: string; +}; + +/** Channel-owned non-retryable predicate for the core ingress drain. */ +export function resolveTelegramIngressNonRetryableFailure( + err: unknown, +): TelegramIngressNonRetryableFailure | null { + for (const candidate of collectErrorGraphCandidates(err, (current) => [ + current.cause, + current.error, + ])) { + const message = formatErrorMessage(candidate); + if (isTelegramMessageDispatchReplayForgetError(candidate)) { + // A committed dispatch key that cannot be rolled back makes retry unsafe: + // the next replay can be duplicate-suppressed and then deleted. + return { reason: "dispatch-dedupe-rollback-failed", message }; + } + if ( + readErrorName(candidate) === MISSING_AGENT_HARNESS_ERROR_NAME || + MISSING_AGENT_HARNESS_MESSAGE_RE.test(message) + ) { + return { reason: "missing-agent-harness", message }; + } + } + return null; +} diff --git a/extensions/telegram/src/telegram-ingress-spool.payload.ts b/extensions/telegram/src/telegram-ingress-spool.payload.ts new file mode 100644 index 000000000000..f5aa9b3c07fe --- /dev/null +++ b/extensions/telegram/src/telegram-ingress-spool.payload.ts @@ -0,0 +1,9 @@ +// Telegram plugin module defines durable ingress queue payload shape. +export type TelegramSpooledUpdatePayload = { + version: number; + updateId: number; + receivedAt: number; + update: unknown; +}; + +export const TELEGRAM_SPOOLED_UPDATE_PAYLOAD_VERSION = 1; diff --git a/extensions/telegram/src/telegram-ingress-spool.ts b/extensions/telegram/src/telegram-ingress-spool.ts index e45c4e0dbd2b..0c03bf582e9f 100644 --- a/extensions/telegram/src/telegram-ingress-spool.ts +++ b/extensions/telegram/src/telegram-ingress-spool.ts @@ -1,24 +1,25 @@ -// Telegram plugin module implements telegram ingress spool behavior. +// Telegram plugin module implements durable ingress enqueue + update_id mapping. import os from "node:os"; import path from "node:path"; -import type { - ChannelIngressQueue, - ChannelIngressQueueClaim, - ChannelIngressQueueClaimRef, - ChannelIngressQueueCorruptClaim, - ChannelIngressQueueRecord, +import { + INGRESS_CLAIM_PROCESS_ID, + processPidFromOwnerId, + type ChannelIngressQueue, + type ChannelIngressQueueClaim, + type ChannelIngressQueueClaimRef, + type ChannelIngressQueueCorruptClaim, + type ChannelIngressQueueRecord, } from "openclaw/plugin-sdk/channel-outbound"; -import { sleepWithAbort } from "openclaw/plugin-sdk/runtime-env"; +import { computeBackoff, type BackoffPolicy } from "openclaw/plugin-sdk/runtime-env"; import { resolveStateDir } from "openclaw/plugin-sdk/state-paths"; import type { TelegramBotInfo } from "./bot-info.js"; import { getTelegramRuntime } from "./runtime.js"; import { getTelegramSequentialKey } from "./sequential-key.js"; -import { resolveSpooledUpdatePersistenceRetryDelayMs } from "./spooled-update-retry-policy.js"; import { normalizeTelegramStateAccountId } from "./state-account-id.js"; import { - processPidFromOwnerId, - TELEGRAM_SPOOLED_UPDATE_PROCESS_ID, -} from "./telegram-ingress-claim-owner.js"; + TELEGRAM_SPOOLED_UPDATE_PAYLOAD_VERSION, + type TelegramSpooledUpdatePayload, +} from "./telegram-ingress-spool.payload.js"; import type { ClaimedTelegramSpooledUpdate, TelegramSpooledUpdate, @@ -28,29 +29,20 @@ export type { ClaimedTelegramSpooledUpdate, TelegramSpooledUpdate, } from "./telegram-ingress-spool.types.js"; +export type { TelegramSpooledUpdatePayload } from "./telegram-ingress-spool.payload.js"; -const SPOOL_VERSION = 1; const TELEGRAM_INGRESS_SPOOL_PREFIX = "ingress-spool-"; -const TELEGRAM_SPOOLED_UPDATE_PROCESSING_STALE_MS = 6 * 60 * 60 * 1000; const TELEGRAM_SPOOLED_UPDATE_FAILED_TTL_MS = 30 * 24 * 60 * 60 * 1000; const TELEGRAM_SPOOLED_UPDATE_FAILED_MAX_ENTRIES = 1000; const TELEGRAM_SPOOLED_UPDATE_COMPLETED_TTL_MS = 30 * 24 * 60 * 60 * 1000; const TELEGRAM_SPOOLED_UPDATE_COMPLETED_MAX_ENTRIES = 1000; - -type TelegramSpooledUpdatePayload = { - version: number; - updateId: number; - receivedAt: number; - update: unknown; +const TELEGRAM_SPOOLED_COMPLETION_RETRY_POLICY: BackoffPolicy = { + initialMs: 250, + maxMs: 5_000, + factor: 2, + jitter: 0.2, }; -class TelegramSpooledUpdateCompletionOwnershipError extends Error { - constructor(updateId: number) { - super(`Telegram spooled update ${updateId} lost claim ownership before completion.`); - this.name = "TelegramSpooledUpdateCompletionOwnershipError"; - } -} - function isValidUpdateId(value: unknown): value is number { return typeof value === "number" && Number.isSafeInteger(value) && value >= 0; } @@ -75,24 +67,12 @@ function resolveTelegramUpdateId(update: unknown): number | null { return isValidUpdateId(value) ? value : null; } -function spoolFileName(updateId: number): string { - return `${String(updateId).padStart(16, "0")}.json`; -} - -function processingFileName(updateId: number): string { - return `${spoolFileName(updateId)}.processing`; -} - -function queueEventId(updateId: number): string { +export function telegramQueueEventId(updateId: number): string { return String(updateId).padStart(16, "0"); } -function pendingPath(spoolDir: string, updateId: number): string { - return path.join(spoolDir, spoolFileName(updateId)); -} - -function processingPath(spoolDir: string, updateId: number): string { - return path.join(spoolDir, processingFileName(updateId)); +function spoolFileName(updateId: number): string { + return `${telegramQueueEventId(updateId)}.json`; } function resolveQueueParts(spoolDir: string): { @@ -110,13 +90,11 @@ function resolveQueueParts(spoolDir: string): { path.basename(path.dirname(spoolDir)) === "telegram" ? path.dirname(path.dirname(spoolDir)) : spoolDir; - return { - accountId, - stateDir, - }; + return { accountId, stateDir }; } -function createTelegramIngressQueue( +/** Open the account-scoped durable ingress queue for this spool directory. */ +export function openTelegramIngressQueue( spoolDir: string, ): ChannelIngressQueue { const parts = resolveQueueParts(spoolDir); @@ -126,6 +104,13 @@ function createTelegramIngressQueue( }); } +export function telegramSpooledUpdateLaneKey(update: unknown, botInfo?: TelegramBotInfo): string { + return getTelegramSequentialKey({ + update: update as Parameters[0]["update"], + ...(botInfo ? { me: botInfo } : {}), + }); +} + async function pruneTelegramIngressQueue( queue: ChannelIngressQueue, now?: number, @@ -139,62 +124,10 @@ async function pruneTelegramIngressQueue( }); } -function parseQueueRecord( - spoolDir: string, - record: ChannelIngressQueueRecord, -): TelegramSpooledUpdate | null { - const payload = record.payload; - if (payload.version !== SPOOL_VERSION || !isValidUpdateId(payload.updateId)) { - return null; - } - return { - updateId: payload.updateId, - path: pendingPath(spoolDir, payload.updateId), - update: payload.update, - receivedAt: payload.receivedAt, - attempts: record.attempts, - ...(record.lastAttemptAt === undefined ? {} : { lastAttemptAt: record.lastAttemptAt }), - ...(record.lastError === undefined ? {} : { lastError: record.lastError }), - }; -} - -function parseQueueClaim( - spoolDir: string, - record: ChannelIngressQueueClaim, -): ClaimedTelegramSpooledUpdate | null { - const update = parseQueueRecord(spoolDir, record); - if (!update) { - return null; - } - return { - ...update, - path: processingPath(spoolDir, update.updateId), - pendingPath: pendingPath(spoolDir, update.updateId), - claim: { - processId: record.claim.ownerId, - processPid: processPidFromOwnerId(record.claim.ownerId), - claimedAt: record.claim.claimedAt, - claimToken: record.claim.token, - }, - }; -} - -function spooledUpdateLaneKey(update: unknown, botInfo?: TelegramBotInfo): string { - return getTelegramSequentialKey({ - update: update as Parameters[0]["update"], - ...(botInfo ? { me: botInfo } : {}), - }); -} - -function sortTelegramUpdates(updates: T[]): T[] { - return updates.toSorted((a, b) => a.updateId - b.updateId); -} - -function queueMutationTarget(update: TelegramSpooledUpdate): string | ChannelIngressQueueClaimRef { - const id = queueEventId(update.updateId); - return update.claim?.claimToken ? { id, claim: { token: update.claim.claimToken } } : id; -} - +/** + * Durable-before-ack accept path: commit the update to the ingress queue. + * Polling advances offset only after this returns; webhook returns 200 only after. + */ export async function writeTelegramSpooledUpdate(params: { spoolDir: string; update: unknown; @@ -206,19 +139,19 @@ export async function writeTelegramSpooledUpdate(params: { throw new Error("Telegram update missing numeric update_id."); } const receivedAt = params.now ?? Date.now(); - const queue = createTelegramIngressQueue(params.spoolDir); + const queue = openTelegramIngressQueue(params.spoolDir); await pruneTelegramIngressQueue(queue, params.now); await queue.enqueue( - queueEventId(updateId), + telegramQueueEventId(updateId), { - version: SPOOL_VERSION, + version: TELEGRAM_SPOOLED_UPDATE_PAYLOAD_VERSION, updateId, receivedAt, update: params.update, }, { receivedAt, - laneKey: params.laneKey ?? spooledUpdateLaneKey(params.update), + laneKey: params.laneKey ?? telegramSpooledUpdateLaneKey(params.update), }, ); return updateId; @@ -228,54 +161,78 @@ export async function listTelegramSpooledUpdates(params: { spoolDir: string; limit?: number | "all"; }): Promise { - const records = await createTelegramIngressQueue(params.spoolDir).listPending({ + const records = await openTelegramIngressQueue(params.spoolDir).listPending({ limit: params.limit ?? 100, orderBy: "id", }); - return sortTelegramUpdates( - records.flatMap((record) => { - const update = parseQueueRecord(params.spoolDir, record); + return records + .flatMap((record) => { + const update = parsePendingRecord(params.spoolDir, record); return update ? [update] : []; - }), - ); + }) + .toSorted((a, b) => a.updateId - b.updateId); } -async function completeTelegramSpooledUpdate(update: TelegramSpooledUpdate): Promise { - const queue = createTelegramIngressQueue(path.dirname(update.path)); - // Successful rows stay as bounded tombstones: Telegram can refetch an update - // after dispatch, and callbacks have side effects that plain delete would rerun. - return await queue.complete(queueMutationTarget(update)); +function parsePendingRecord( + spoolDir: string, + record: ChannelIngressQueueRecord, +): TelegramSpooledUpdate | null { + const payload = record.payload; + if ( + payload.version !== TELEGRAM_SPOOLED_UPDATE_PAYLOAD_VERSION || + !isValidUpdateId(payload.updateId) + ) { + return null; + } + return { + updateId: payload.updateId, + path: path.join(spoolDir, spoolFileName(payload.updateId)), + update: payload.update, + receivedAt: payload.receivedAt, + attempts: record.attempts, + ...(record.lastAttemptAt === undefined ? {} : { lastAttemptAt: record.lastAttemptAt }), + ...(record.lastError === undefined ? {} : { lastError: record.lastError }), + }; } -export async function completeTelegramSpooledUpdateWithRetry(params: { - update: ClaimedTelegramSpooledUpdate; - abortSignal?: AbortSignal; - onRetry?: (retry: { attempt: number; delayMs: number; error: unknown }) => void; -}): Promise { - if (!params.update.claim?.claimToken) { - throw new TelegramSpooledUpdateCompletionOwnershipError(params.update.updateId); - } - let attempt = 0; - while (true) { - try { - const completed = await completeTelegramSpooledUpdate(params.update); - if (!completed) { - throw new TelegramSpooledUpdateCompletionOwnershipError(params.update.updateId); - } - return; - } catch (err) { - if ( - err instanceof TelegramSpooledUpdateCompletionOwnershipError || - params.abortSignal?.aborted - ) { - throw err; - } - attempt += 1; - const delayMs = resolveSpooledUpdatePersistenceRetryDelayMs(attempt); - params.onRetry?.({ attempt, delayMs, error: err }); - await sleepWithAbort(delayMs, params.abortSignal); - } +/** Backoff for irrevocable-adoption completion retries (bot-message only). */ +export function resolveSpooledUpdatePersistenceRetryDelayMs(attempt: number): number { + return computeBackoff(TELEGRAM_SPOOLED_COMPLETION_RETRY_POLICY, attempt); +} + +// --- Thin queue claim helpers (transport tests + recovery tools) --- +// Drain loops live in core; these wrap openTelegramIngressQueue only. + +function processingFileName(updateId: number): string { + return `${spoolFileName(updateId)}.processing`; +} + +function parseQueueClaim( + spoolDir: string, + record: ChannelIngressQueueClaim, +): ClaimedTelegramSpooledUpdate | null { + const update = parsePendingRecord(spoolDir, record); + if (!update) { + return null; } + const claimRef = record.claim.token; + return { + ...update, + path: path.join(spoolDir, processingFileName(update.updateId)), + pendingPath: path.join(spoolDir, spoolFileName(update.updateId)), + claim: { + processId: record.claim.ownerId, + processPid: processPidFromOwnerId(record.claim.ownerId), + claimedAt: record.claim.claimedAt, + claimToken: claimRef, + }, + }; +} + +function queueMutationTarget(update: TelegramSpooledUpdate): string | ChannelIngressQueueClaimRef { + const id = telegramQueueEventId(update.updateId); + const claimRef = update.claim?.claimToken; + return claimRef ? { id, claim: { token: claimRef } } : id; } export async function claimNextTelegramSpooledUpdate(params: { @@ -285,16 +242,16 @@ export async function claimNextTelegramSpooledUpdate(params: { candidateUpdateIds?: Iterable; scanLimit?: number; }): Promise { - const queue = createTelegramIngressQueue(params.spoolDir); + const queue = openTelegramIngressQueue(params.spoolDir); const claimed = await queue.claimNext({ - ownerId: TELEGRAM_SPOOLED_UPDATE_PROCESS_ID, + ownerId: INGRESS_CLAIM_PROCESS_ID, blockedLaneKeys: params.blockedLaneKeys, ...(params.candidateUpdateIds === undefined ? {} - : { candidateIds: [...params.candidateUpdateIds].map(queueEventId) }), + : { candidateIds: [...params.candidateUpdateIds].map(telegramQueueEventId) }), orderBy: "id", scanLimit: params.scanLimit, - deriveLaneKey: (record) => spooledUpdateLaneKey(record.payload.update, params.botInfo), + deriveLaneKey: (record) => telegramSpooledUpdateLaneKey(record.payload.update, params.botInfo), }); if (!claimed) { return null; @@ -310,68 +267,16 @@ export async function claimNextTelegramSpooledUpdate(params: { return null; } -export async function releaseTelegramSpooledUpdateClaim( - update: ClaimedTelegramSpooledUpdate, - options?: { lastError?: string; releasedAt?: number }, -): Promise { - await createTelegramIngressQueue(path.dirname(update.pendingPath)).release( - queueMutationTarget(update), - options, - ); -} - -export async function abandonTelegramSpooledUpdateClaim( - update: ClaimedTelegramSpooledUpdate, -): Promise { - await createTelegramIngressQueue(path.dirname(update.pendingPath)).release( - queueMutationTarget(update), - { recordAttempt: false }, - ); -} - -export async function refreshTelegramSpooledUpdateClaim( - update: ClaimedTelegramSpooledUpdate, - options?: { refreshedAt?: number }, -): Promise { - const claimToken = update.claim?.claimToken; - if (!claimToken) { - return false; - } - const queue = createTelegramIngressQueue(path.dirname(update.pendingPath)); - return ( - (await queue.refreshClaim?.( - { id: queueEventId(update.updateId), claim: { token: claimToken } }, - options, - )) ?? false - ); -} - -export async function failTelegramSpooledUpdateClaim(params: { - update: ClaimedTelegramSpooledUpdate; - reason: string; - message: string; - now?: number; -}): Promise { - const queue = createTelegramIngressQueue(path.dirname(params.update.pendingPath)); - const failed = await queue.fail(queueMutationTarget(params.update), { - reason: params.reason, - message: params.message, - ...(params.now === undefined ? {} : { failedAt: params.now }), - }); - await pruneTelegramIngressQueue(queue, params.now); - return failed; -} - export async function listTelegramSpooledUpdateClaims(params: { spoolDir: string; }): Promise { - const claims = await createTelegramIngressQueue(params.spoolDir).listClaims(); - return sortTelegramUpdates( - claims.flatMap((claim) => { + const claims = await openTelegramIngressQueue(params.spoolDir).listClaims(); + return claims + .flatMap((claim) => { const update = parseQueueClaim(params.spoolDir, claim); return update ? [update] : []; - }), - ); + }) + .toSorted((a, b) => a.updateId - b.updateId); } export async function recoverStaleTelegramSpooledUpdateClaims(params: { @@ -383,12 +288,12 @@ export async function recoverStaleTelegramSpooledUpdateClaims(params: { }): Promise { const shouldRecover = params.shouldRecover; const shouldRecoverCorrupt = params.shouldRecoverCorrupt; - return await createTelegramIngressQueue(params.spoolDir).recoverStaleClaims({ - staleMs: params.staleMs ?? TELEGRAM_SPOOLED_UPDATE_PROCESSING_STALE_MS, + return await openTelegramIngressQueue(params.spoolDir).recoverStaleClaims({ + staleMs: params.staleMs ?? 0, ...(params.now === undefined ? {} : { now: params.now }), ...(shouldRecover ? { - shouldRecover: async (claim: ChannelIngressQueueClaim) => { + shouldRecover: async (claim) => { const update = parseQueueClaim(params.spoolDir, claim); return update ? await shouldRecover(update) : false; }, @@ -397,3 +302,29 @@ export async function recoverStaleTelegramSpooledUpdateClaims(params: { ...(shouldRecoverCorrupt ? { shouldRecoverCorrupt } : {}), }); } + +export async function releaseTelegramSpooledUpdateClaim( + update: ClaimedTelegramSpooledUpdate, + options?: { lastError?: string; releasedAt?: number }, +): Promise { + await openTelegramIngressQueue(path.dirname(update.pendingPath)).release( + queueMutationTarget(update), + options, + ); +} + +export async function failTelegramSpooledUpdateClaim(params: { + update: ClaimedTelegramSpooledUpdate; + reason: string; + message: string; + now?: number; +}): Promise { + return await openTelegramIngressQueue(path.dirname(params.update.pendingPath)).fail( + queueMutationTarget(params.update), + { + reason: params.reason, + message: params.message, + ...(params.now === undefined ? {} : { failedAt: params.now }), + }, + ); +} diff --git a/extensions/telegram/src/telegram-ingress-supersede.ts b/extensions/telegram/src/telegram-ingress-supersede.ts new file mode 100644 index 000000000000..4c0f03b0fd7a --- /dev/null +++ b/extensions/telegram/src/telegram-ingress-supersede.ts @@ -0,0 +1,374 @@ +import type { + ChannelIngressQueueClaim, + ChannelIngressQueueRecord, +} from "openclaw/plugin-sdk/channel-outbound"; +import { + maybeResolveTextAlias, + normalizeCommandBody, +} from "openclaw/plugin-sdk/command-auth-native"; +import { + isAbortRequestText, + isBtwRequestText, +} from "openclaw/plugin-sdk/command-primitives-runtime"; +// Telegram plugin module owns pre-adoption supersede policy for durable ingress. +import type { OpenClawConfig } from "openclaw/plugin-sdk/config-contracts"; +import { resolveTelegramDmAllow } from "./access-groups.js"; +import { mergeTelegramAccountConfig } from "./account-config.js"; +import { + resolveTelegramCommandAuthorization, + resolveTelegramGroupAllowFromContext, + resolveTelegramMessageForumFlagHint, +} from "./bot/helpers.js"; +import { resolveTelegramScopedGroupConfig } from "./group-config-helpers.js"; +import { resolveTelegramCommandIngressAuthorization } from "./ingress.js"; +import { isTelegramReadOnlyControlLaneText } from "./sequential-key.js"; +import type { TelegramSpooledUpdatePayload } from "./telegram-ingress-spool.payload.js"; + +function isRecognizedTelegramTextCommand(rawText: string, botUsername?: string): boolean { + return ( + maybeResolveTextAlias( + normalizeCommandBody(rawText, botUsername ? { botUsername } : undefined), + ) != null + ); +} + +/** + * Whether a bot_command entity (or slash token) targets this bot. + * Same target rule as normalizeCommandBody: untargeted commands match any bot; + * @OtherBot is ignored when our identity is known. + */ +function isTelegramCommandTargetedAtBot(commandText: string, botUsername?: string): boolean { + const trimmed = commandText.trim(); + if (!trimmed.startsWith("/")) { + return false; + } + // normalizeCommandBody only strips @bot when the target equals botUsername. + // A non-matching @target leaves the body as `/cmd@other`, which is not ours. + const normalized = normalizeCommandBody( + trimmed, + botUsername ? { botUsername } : undefined, + ).trim(); + if (!normalized.startsWith("/")) { + return false; + } + // Untargeted, or successfully stripped for this bot. + if (!/^\/[^\s@]+@/u.test(normalized)) { + return true; + } + // Identity unknown: keep untargeted-permissive behavior for pre-getMe drains. + return !botUsername?.trim(); +} + +/** True when the update carries a bot_command entity addressed to this bot. */ +function updateHasBotCommandEntityForBot(update: unknown, botUsername?: string): boolean { + if (!update || typeof update !== "object") { + return false; + } + const root = update as Record; + for (const key of ["message", "edited_message", "channel_post", "edited_channel_post"] as const) { + const msg = root[key]; + if (!msg || typeof msg !== "object") { + continue; + } + const message = msg as { + text?: unknown; + caption?: unknown; + entities?: unknown; + caption_entities?: unknown; + }; + const body = + typeof message.text === "string" + ? message.text + : typeof message.caption === "string" + ? message.caption + : ""; + for (const entities of [message.entities, message.caption_entities]) { + if (!Array.isArray(entities)) { + continue; + } + for (const entity of entities) { + if (!entity || typeof entity !== "object") { + continue; + } + const ent = entity as { type?: unknown; offset?: unknown; length?: unknown }; + if (ent.type !== "bot_command") { + continue; + } + if (typeof ent.offset !== "number" || typeof ent.length !== "number") { + continue; + } + const commandText = body.slice(ent.offset, ent.offset + ent.length); + if (isTelegramCommandTargetedAtBot(commandText, botUsername)) { + return true; + } + } + } + } + return false; +} + +function extractUpdateText(update: unknown): string { + if (!update || typeof update !== "object") { + return ""; + } + const root = update as Record; + for (const key of ["message", "edited_message", "channel_post", "edited_channel_post"] as const) { + const msg = root[key]; + if (msg && typeof msg === "object") { + const text = (msg as { text?: unknown; caption?: unknown }).text; + if (typeof text === "string") { + return text; + } + const caption = (msg as { caption?: unknown }).caption; + if (typeof caption === "string") { + return caption; + } + } + } + const callback = root.callback_query; + if (callback && typeof callback === "object") { + const data = (callback as { data?: unknown }).data; + if (typeof data === "string") { + return data; + } + } + return ""; +} + +type UpdateSenderFacts = { + senderId: string; + senderUsername?: string; + chatId: number; + chatType?: string; + isGroup: boolean; + isTopicMessage?: boolean; + isForum?: boolean; + messageThreadId?: number; +}; + +function extractUpdateSenderFacts(update: unknown): UpdateSenderFacts | null { + if (!update || typeof update !== "object") { + return null; + } + const root = update as Record; + let message: Record | undefined; + for (const key of ["message", "edited_message", "channel_post", "edited_channel_post"] as const) { + const candidate = root[key]; + if (candidate && typeof candidate === "object") { + message = candidate as Record; + break; + } + } + if (!message) { + const callback = root.callback_query; + if (callback && typeof callback === "object") { + const cb = callback as Record; + const from = cb.from; + const msg = cb.message; + if (from && typeof from === "object" && msg && typeof msg === "object") { + message = msg as Record; + const chat = (msg as { chat?: { id?: unknown; type?: unknown; is_forum?: unknown } }).chat; + const fromObj = from as { id?: unknown; username?: unknown }; + if (typeof chat?.id === "number" && typeof fromObj.id === "number") { + const chatType = typeof chat.type === "string" ? chat.type : "private"; + return { + senderId: String(fromObj.id), + ...(typeof fromObj.username === "string" ? { senderUsername: fromObj.username } : {}), + chatId: chat.id, + chatType, + isGroup: chatType !== "private", + ...(typeof chat.is_forum === "boolean" ? { isForum: chat.is_forum } : {}), + ...(typeof (msg as { is_topic_message?: unknown }).is_topic_message === "boolean" + ? { + isTopicMessage: (msg as { is_topic_message: boolean }).is_topic_message, + } + : {}), + ...(typeof (msg as { message_thread_id?: unknown }).message_thread_id === "number" + ? { + messageThreadId: (msg as { message_thread_id: number }).message_thread_id, + } + : {}), + }; + } + } + } + return null; + } + const chat = message.chat as { id?: unknown; type?: unknown; is_forum?: unknown } | undefined; + const from = message.from as { id?: unknown; username?: unknown } | undefined; + if (typeof chat?.id !== "number" || typeof from?.id !== "number") { + return null; + } + const chatType = typeof chat.type === "string" ? chat.type : "private"; + return { + senderId: String(from.id), + ...(typeof from.username === "string" ? { senderUsername: from.username } : {}), + chatId: chat.id, + chatType, + isGroup: chatType !== "private", + ...(typeof chat.is_forum === "boolean" ? { isForum: chat.is_forum } : {}), + ...(typeof message.is_topic_message === "boolean" + ? { isTopicMessage: message.is_topic_message as boolean } + : {}), + ...(typeof message.message_thread_id === "number" + ? { messageThreadId: message.message_thread_id as number } + : {}), + }; +} + +/** Ambient room_event-shaped updates (no user text body) stay supersedable. */ +export function isTelegramAmbientSpooledUpdate(update: unknown): boolean { + if (!update || typeof update !== "object") { + return false; + } + const root = update as Record; + return ( + root.message_reaction != null || + root.message_reaction_count != null || + root.chat_member != null || + root.my_chat_member != null || + root.chat_join_request != null || + root.chat_boost != null || + root.removed_chat_boost != null + ); +} + +export type TelegramSupersedeAuthContext = { + cfg: OpenClawConfig; + accountId: string; + /** Bot username for @bot command targeting (from getMe / botInfo). */ + botUsername?: string; + /** Test seam / preloaded pairing-store ids; defaults to live pairing store. */ +}; + +/** + * Whether the raw update's sender is command-authorized. + * Reuses resolveTelegramGroupAllowFromContext — same group/topic allowFrom + * overrides and access-group expansion as normal message ingress. + */ +export async function isTelegramSpooledUpdateSenderAuthorized( + update: unknown, + auth: TelegramSupersedeAuthContext, +): Promise { + const facts = extractUpdateSenderFacts(update); + if (!facts) { + return false; + } + const accountCfg = mergeTelegramAccountConfig(auth.cfg, auth.accountId); + const dmPolicy = accountCfg.dmPolicy ?? "pairing"; + const allowFrom = accountCfg.allowFrom; + const groupAllowFrom = accountCfg.groupAllowFrom ?? accountCfg.allowFrom; + const isForum = + resolveTelegramMessageForumFlagHint({ + chatType: facts.chatType as "private" | "group" | "supergroup" | "channel" | undefined, + isForum: facts.isForum, + isTopicMessage: facts.isTopicMessage, + }) ?? false; + + const groupAllowContext = await resolveTelegramGroupAllowFromContext({ + cfg: auth.cfg, + chatId: facts.chatId, + accountId: auth.accountId, + dmPolicy, + allowFrom, + senderId: facts.senderId, + isGroup: facts.isGroup, + isForum, + messageThreadId: facts.messageThreadId, + groupAllowFrom, + resolveTelegramGroupConfig: (chatId, messageThreadId, cfg) => { + const telegramCfg = mergeTelegramAccountConfig(cfg, auth.accountId); + return resolveTelegramScopedGroupConfig(telegramCfg, chatId, messageThreadId); + }, + }); + + const { resolvedThreadId, storeAllowFrom, groupAllowOverride, effectiveGroupAllow } = + groupAllowContext; + + const dmAllow = await resolveTelegramDmAllow({ + cfg: auth.cfg, + groupAllowOverride, + allowFrom, + accountId: auth.accountId, + senderId: facts.senderId, + storeAllowFrom: facts.isGroup ? [] : storeAllowFrom, + dmPolicy, + }); + + const ownerAccess = resolveTelegramCommandAuthorization({ + cfg: auth.cfg, + accountId: auth.accountId, + chatId: facts.chatId, + isGroup: facts.isGroup, + ...(resolvedThreadId !== undefined ? { resolvedThreadId } : {}), + senderId: facts.senderId, + ...(facts.senderUsername !== undefined ? { senderUsername: facts.senderUsername } : {}), + }); + const gate = await resolveTelegramCommandIngressAuthorization({ + accountId: auth.accountId, + cfg: auth.cfg, + dmPolicy, + isGroup: facts.isGroup, + chatId: facts.chatId, + ...(resolvedThreadId !== undefined ? { resolvedThreadId } : {}), + senderId: facts.senderId, + effectiveDmAllow: dmAllow.effectiveAllow, + effectiveGroupAllow, + ownerAccess, + eventKind: "message", + allowTextCommands: true, + hasControlCommand: true, + modeWhenAccessGroupsOff: "allow", + includeDmAllowForGroupCommands: false, + }); + return gate.authorized; +} + +/** + * Drain-level supersede predicate over raw spooled payloads. + * Authorization is resolved from the new event's numeric sender via the same + * ingress command gate as the old fence (CommandAuthorized). + */ +export function createShouldSupersedeTelegramSpooledPending( + auth: TelegramSupersedeAuthContext, +): ( + newEvent: ChannelIngressQueueRecord, + pendingEvent: ChannelIngressQueueClaim, +) => boolean | Promise { + return async (newEvent, pendingEvent) => { + const pendingUpdate = pendingEvent.payload.update; + const newUpdate = newEvent.payload.update; + // Ambient pending supersede still requires an authorized sender — same as the + // old fence (post-auth). Unauthorized strangers cannot cancel pre-adoption work. + if ( + isTelegramAmbientSpooledUpdate(pendingUpdate) && + !isTelegramAmbientSpooledUpdate(newUpdate) + ) { + return await isTelegramSpooledUpdateSenderAuthorized(newUpdate, auth); + } + const text = extractUpdateText(newUpdate); + if (!text) { + return false; + } + const commandOptions = auth.botUsername ? { botUsername: auth.botUsername } : undefined; + if ( + isBtwRequestText(text, commandOptions) || + isTelegramReadOnlyControlLaneText({ + rawText: text, + ...(auth.botUsername ? { botUsername: auth.botUsername } : {}), + }) + ) { + return false; + } + // Abort, static text alias, or native bot_command entity (incl. skill commands) + // addressed to this bot. Never bare `/` prefixes without a bot_command entity. + const isAbort = isAbortRequestText(text, commandOptions); + const isCommand = + isRecognizedTelegramTextCommand(text, auth.botUsername) || + updateHasBotCommandEntityForBot(newUpdate, auth.botUsername); + if (!isAbort && !isCommand) { + return false; + } + return await isTelegramSpooledUpdateSenderAuthorized(newUpdate, auth); + }; +} diff --git a/extensions/telegram/src/telegram-reply-fence.ts b/extensions/telegram/src/telegram-reply-fence.ts deleted file mode 100644 index c84048754b84..000000000000 --- a/extensions/telegram/src/telegram-reply-fence.ts +++ /dev/null @@ -1,255 +0,0 @@ -// Telegram plugin module implements telegram reply fence behavior. -import { - isExplicitCommandTurn, - type CommandTurnContext, -} from "openclaw/plugin-sdk/channel-inbound"; -import { - maybeResolveTextAlias, - normalizeCommandBody, -} from "openclaw/plugin-sdk/command-auth-native"; -import { - isAbortRequestText, - isBtwRequestText, -} from "openclaw/plugin-sdk/command-primitives-runtime"; -import { isTelegramReadOnlyControlLaneText } from "./sequential-key.js"; - -type TelegramReplyFenceState = { - generation: number; - activeDispatches: number; - abortControllers?: Set; - laneKeys?: Set; -}; - -type TelegramReplyFenceKey = { - activeKey: string; - roomEventKey: string; -}; - -// Newer accepted turns and authorized aborts can arrive ahead of older same-session reply work. -const TELEGRAM_REPLY_FENCE_STATE_KEY = Symbol.for("openclaw.telegram.replyFenceState"); - -function getTelegramReplyFenceState(): { - byKey: Map; - keysByLane: Map>; -} { - const globalRecord = globalThis as Record; - const existing = globalRecord[TELEGRAM_REPLY_FENCE_STATE_KEY] as - | { - byKey: Map; - keysByLane: Map>; - } - | undefined; - if (existing) { - return existing; - } - const created = { - byKey: new Map(), - keysByLane: new Map>(), - }; - globalRecord[TELEGRAM_REPLY_FENCE_STATE_KEY] = created; - return created; -} - -const { byKey: telegramReplyFenceByKey, keysByLane: telegramReplyFenceKeysByLane } = - getTelegramReplyFenceState(); - -export function buildTelegramReplyFenceLaneKey(params: { - accountId: string; - sequentialKey: string; -}): string { - return `${params.accountId}\0${params.sequentialKey}`; -} - -export function buildTelegramNonInterruptingReplyFenceKey(params: { - activeKey: string; - laneKey: string; -}): string { - return `${buildTelegramNonInterruptingReplyFenceKeyPrefix(params.activeKey)}${params.laneKey}`; -} - -function buildTelegramNonInterruptingReplyFenceKeyPrefix(activeKey: string): string { - return `${activeKey}\0non-interrupting\0`; -} - -function normalizeTelegramFenceKey(value: unknown): string | undefined { - if (typeof value !== "string") { - return undefined; - } - const trimmed = value.trim(); - return trimmed.length > 0 ? trimmed : undefined; -} - -export function resolveTelegramReplyFenceKey(params: { - ctxPayload: { SessionKey?: string; CommandTargetSessionKey?: string; InboundEventKind?: string }; - chatId: number | string; - threadSpec: { id?: number | string | null; scope?: string }; -}): TelegramReplyFenceKey { - const baseKey = - normalizeTelegramFenceKey(params.ctxPayload.CommandTargetSessionKey) ?? - normalizeTelegramFenceKey(params.ctxPayload.SessionKey) ?? - `telegram:${String(params.chatId)}:${params.threadSpec.scope ?? "default"}:${params.threadSpec.id ?? "root"}`; - const roomEventKey = `${baseKey}:room_event`; - return { - activeKey: params.ctxPayload.InboundEventKind === "room_event" ? roomEventKey : baseKey, - roomEventKey, - }; -} - -function abortTelegramReplyFenceControllers(state: TelegramReplyFenceState): void { - for (const controller of state.abortControllers ?? []) { - controller.abort(); - } - state.abortControllers?.clear(); -} - -function deleteTelegramReplyFenceState(key: string, state: TelegramReplyFenceState): void { - telegramReplyFenceByKey.delete(key); - for (const laneKey of state.laneKeys ?? []) { - const keys = telegramReplyFenceKeysByLane.get(laneKey); - keys?.delete(key); - if (keys?.size === 0) { - telegramReplyFenceKeysByLane.delete(laneKey); - } - } -} - -function maybeDeleteTelegramReplyFenceState(key: string, state: TelegramReplyFenceState): void { - if (state.activeDispatches <= 0 && (state.abortControllers?.size ?? 0) === 0) { - deleteTelegramReplyFenceState(key, state); - } else { - telegramReplyFenceByKey.set(key, state); - } -} - -export function beginTelegramReplyFence(params: { - key: string; - supersede: boolean; - abortController?: AbortController; - laneKey?: string; -}): number { - const existing = telegramReplyFenceByKey.get(params.key); - const state: TelegramReplyFenceState = existing ?? { - generation: 0, - activeDispatches: 0, - }; - if (params.supersede) { - state.generation += 1; - abortTelegramReplyFenceControllers(state); - supersedeTelegramNonInterruptingReplyFenceChildren(params.key); - } - if (params.abortController) { - (state.abortControllers ??= new Set()).add(params.abortController); - } - const laneKey = normalizeTelegramFenceKey(params.laneKey); - if (laneKey) { - (state.laneKeys ??= new Set()).add(laneKey); - const keys = telegramReplyFenceKeysByLane.get(laneKey) ?? new Set(); - keys.add(params.key); - telegramReplyFenceKeysByLane.set(laneKey, keys); - } - state.activeDispatches += 1; - telegramReplyFenceByKey.set(params.key, state); - return state.generation; -} - -function supersedeTelegramReplyFenceState(key: string): boolean { - const state = telegramReplyFenceByKey.get(key); - if (!state) { - return false; - } - state.generation += 1; - abortTelegramReplyFenceControllers(state); - maybeDeleteTelegramReplyFenceState(key, state); - return true; -} - -function supersedeTelegramNonInterruptingReplyFenceChildren(key: string): boolean { - let superseded = false; - const childPrefix = buildTelegramNonInterruptingReplyFenceKeyPrefix(key); - for (const childKey of telegramReplyFenceByKey.keys()) { - if (childKey.startsWith(childPrefix)) { - superseded = supersedeTelegramReplyFenceState(childKey) || superseded; - } - } - return superseded; -} - -export function supersedeTelegramReplyFence(key: string): boolean { - let superseded = supersedeTelegramReplyFenceState(key); - superseded = supersedeTelegramNonInterruptingReplyFenceChildren(key) || superseded; - return superseded; -} - -export function supersedeTelegramReplyFenceLane(laneKey: string): boolean { - const keys = [...(telegramReplyFenceKeysByLane.get(laneKey) ?? [])]; - let superseded = false; - for (const key of keys) { - superseded = supersedeTelegramReplyFence(key) || superseded; - } - return superseded; -} - -export function isTelegramReplyFenceSuperseded(params: { - key: string; - generation: number; -}): boolean { - return (telegramReplyFenceByKey.get(params.key)?.generation ?? 0) !== params.generation; -} - -export function endTelegramReplyFence(key: string, abortController?: AbortController): void { - const state = telegramReplyFenceByKey.get(key); - if (!state) { - return; - } - if (abortController) { - state.abortControllers?.delete(abortController); - } - state.activeDispatches = Math.max(0, state.activeDispatches - 1); - maybeDeleteTelegramReplyFenceState(key, state); -} - -export function releaseTelegramReplyFenceAbortController( - key: string, - abortController?: AbortController, -): void { - if (!abortController) { - return; - } - const state = telegramReplyFenceByKey.get(key); - if (!state) { - return; - } - state.abortControllers?.delete(abortController); - maybeDeleteTelegramReplyFenceState(key, state); -} - -function isRecognizedTelegramTextCommand(rawText: string): boolean { - return maybeResolveTextAlias(normalizeCommandBody(rawText)) != null; -} - -export function shouldSupersedeTelegramReplyFence(ctxPayload: { - Body?: string; - ChatType?: string; - RawBody?: string; - CommandBody?: string; - CommandAuthorized: boolean; - CommandTurn?: CommandTurnContext; -}): boolean { - const dispatchText = ctxPayload.CommandBody ?? ctxPayload.RawBody ?? ctxPayload.Body ?? ""; - if (isAbortRequestText(dispatchText)) { - return ctxPayload.CommandAuthorized; - } - if ( - isBtwRequestText(dispatchText) || - isTelegramReadOnlyControlLaneText({ rawText: dispatchText }) - ) { - return false; - } - // One rule for all chat types: only authorized explicit/native commands - // supersede. Normal messages never abort an active turn at the transport - // fence; core queue policy owns steer/followup/interrupt after adoption. - return ( - ctxPayload.CommandAuthorized && - (isExplicitCommandTurn(ctxPayload.CommandTurn) || isRecognizedTelegramTextCommand(dispatchText)) - ); -} diff --git a/extensions/telegram/src/webhook.ts b/extensions/telegram/src/webhook.ts index 8b12ccfbf805..e852ace383b3 100644 --- a/extensions/telegram/src/webhook.ts +++ b/extensions/telegram/src/webhook.ts @@ -33,43 +33,15 @@ import { readJsonBodyWithLimit } from "openclaw/plugin-sdk/webhook-request-guard import { mergeTelegramAccountConfig } from "./account-config.js"; import { resolveTelegramAllowedUpdates } from "./allowed-updates.js"; import { withTelegramApiErrorLogging } from "./api-logging.js"; -import { - runWithTelegramSpooledReplayUpdate, - type TelegramMessageProcessingResult, - type TelegramSpooledReplayDeferredParticipant, -} from "./bot-processing-outcome.js"; import { createTelegramBot } from "./bot.js"; import { resolveTelegramTransport } from "./fetch.js"; import { isRetryableTelegramApiError } from "./network-errors.js"; import { getTelegramSequentialKey } from "./sequential-key.js"; +import { createTelegramTransportIngressDrain } from "./telegram-ingress-drain-factory.js"; import { - resolveNonRetryableSpooledUpdateFailure, - resolveSpooledUpdateAttemptNumber, - resolveSpooledUpdateRetryDelayMs, - shouldDeadLetterRetryableSpooledUpdate, - TELEGRAM_SPOOLED_RETRY_MAX_ATTEMPTS, -} from "./spooled-update-retry-policy.js"; -import { - isTelegramSpooledCorruptClaimOwnedByOtherLiveProcess, - isTelegramSpooledUpdateClaimOwnedByOtherLiveProcess, -} from "./telegram-ingress-claim-owner.js"; -import { - claimNextTelegramSpooledUpdate, - completeTelegramSpooledUpdateWithRetry, - failTelegramSpooledUpdateClaim, - listTelegramSpooledUpdateClaims, - listTelegramSpooledUpdates, - recoverStaleTelegramSpooledUpdateClaims, - refreshTelegramSpooledUpdateClaim, - releaseTelegramSpooledUpdateClaim, resolveTelegramIngressSpoolDir, writeTelegramSpooledUpdate, - type ClaimedTelegramSpooledUpdate, } from "./telegram-ingress-spool.js"; -import { - buildTelegramReplyFenceLaneKey, - supersedeTelegramReplyFenceLane, -} from "./telegram-reply-fence.js"; import { createTelegramWebhookStatusPublisher } from "./webhook-status.js"; const TELEGRAM_WEBHOOK_MAX_BODY_BYTES = 1024 * 1024; @@ -77,38 +49,12 @@ const TELEGRAM_WEBHOOK_BODY_TIMEOUT_MS = 30_000; const TELEGRAM_WEBHOOK_ACCEPTED_HEADER = "x-openclaw-delivery-accepted"; const TELEGRAM_WEBHOOK_ACCEPTED_VALUE = "durable"; const TELEGRAM_WEBHOOK_SPOOLED_DRAIN_INTERVAL_MS = 500; -const TELEGRAM_WEBHOOK_SPOOLED_CLAIM_REFRESH_INTERVAL_MS = 5 * 60 * 1000; -const TELEGRAM_WEBHOOK_SPOOLED_HANDLER_TIMEOUT_MS = 25 * 60_000; -const TELEGRAM_WEBHOOK_SPOOLED_HANDLER_ABORT_GRACE_MS = 5_000; -const TELEGRAM_WEBHOOK_SPOOLED_DRAIN_START_LIMIT = 100; -const TELEGRAM_WEBHOOK_SPOOLED_DRAIN_SCAN_LIMIT = TELEGRAM_WEBHOOK_SPOOLED_DRAIN_START_LIMIT * 10; const TELEGRAM_WEBHOOK_REGISTRATION_RETRY_POLICY: BackoffPolicy = { initialMs: 5_000, maxMs: 60_000, factor: 2, jitter: 0.2, }; -type ActiveWebhookSpooledHandler = { - laneKey: string; -}; - -const activeWebhookSpooledHandlersByLane = new Map(); - -function buildWebhookSpooledHandlerKey(params: { laneKey: string; spoolDir: string }): string { - return `${params.spoolDir}\0${params.laneKey}`; -} - -function resolveActiveWebhookSpooledLaneKeys(spoolDir: string): Set { - const laneKeys = new Set(); - const prefix = `${spoolDir}\0`; - for (const [handlerKey, handler] of activeWebhookSpooledHandlersByLane) { - if (handlerKey.startsWith(prefix)) { - laneKeys.add(handler.laneKey); - } - } - return laneKeys; -} - async function listenHttpServer(params: { server: ReturnType; port: number; @@ -336,335 +282,6 @@ function resolveWebhookSpooledUpdateLaneKey(update: unknown): string { }); } -async function releaseFailedWebhookSpooledUpdate(params: { - err: unknown; - log: (line: string) => void; - update: ClaimedTelegramSpooledUpdate; -}): Promise { - const laneKey = resolveWebhookSpooledUpdateLaneKey(params.update.update); - const nonRetryable = resolveNonRetryableSpooledUpdateFailure(params.err); - if (nonRetryable) { - const failed = await failTelegramSpooledUpdateClaim({ - update: params.update, - reason: nonRetryable.reason, - message: nonRetryable.message, - }); - if (failed) { - params.log( - `[telegram][diag] webhook spooled update ${params.update.updateId} failed with non-retryable ${nonRetryable.reason}; dead-lettered: ${nonRetryable.message}`, - ); - } - return; - } - - const attempt = resolveSpooledUpdateAttemptNumber(params.update); - if (shouldDeadLetterRetryableSpooledUpdate(params.update, attempt)) { - const message = formatErrorMessage(params.err); - const failed = await failTelegramSpooledUpdateClaim({ - update: params.update, - reason: "retry-limit-exceeded", - message, - }); - if (failed) { - // Retryable poison updates must eventually become tombstones, but not - // during ordinary transient provider or state-store outages. - params.log( - `[telegram][warn] webhook spooled update ${params.update.updateId} on lane ${laneKey} reached retry limit after ${attempt} attempts; dead-lettered: ${message}`, - ); - } - return; - } - - await releaseTelegramSpooledUpdateClaim(params.update, { - lastError: formatErrorMessage(params.err), - }); - params.log( - `[telegram][diag] webhook spooled update ${params.update.updateId} failed; keeping for retry attempt ${attempt + 1}/${TELEGRAM_SPOOLED_RETRY_MAX_ATTEMPTS}: ${formatErrorMessage(params.err)}`, - ); -} - -function startWebhookSpooledUpdateClaimRefresh(params: { - log: (line: string) => void; - update: ClaimedTelegramSpooledUpdate; -}): () => void { - let stopped = false; - let refreshing = false; - const refresh = async (): Promise => { - if (stopped || refreshing) { - return; - } - refreshing = true; - try { - const refreshed = await refreshTelegramSpooledUpdateClaim(params.update); - if (!refreshed && !stopped) { - params.log( - `[telegram][diag] webhook spooled update ${params.update.updateId} claim refresh lost ownership`, - ); - } - } catch (err) { - params.log( - `[telegram][diag] webhook spooled update ${params.update.updateId} claim refresh failed: ${formatErrorMessage(err)}`, - ); - } finally { - refreshing = false; - } - }; - const timer = setInterval(() => { - void refresh(); - }, TELEGRAM_WEBHOOK_SPOOLED_CLAIM_REFRESH_INTERVAL_MS); - timer.unref?.(); - return () => { - if (stopped) { - return; - } - stopped = true; - clearInterval(timer); - }; -} - -type WebhookSpooledDeferredWorkResult = TelegramMessageProcessingResult & { - timedOut?: boolean; -}; - -class WebhookSpooledHandlerTimeoutError extends Error { - constructor( - message: string, - readonly replayTask: Promise<{ deferredWork?: TelegramSpooledReplayDeferredParticipant }>, - ) { - super(message); - this.name = "WebhookSpooledHandlerTimeoutError"; - } -} - -function formatWebhookSpooledHandlerTimeoutMessage(params: { - laneKey: string; - updateId: number; -}): string { - const age = formatDurationPrecise(TELEGRAM_WEBHOOK_SPOOLED_HANDLER_TIMEOUT_MS); - return `Telegram webhook spool processing timed out behind update ${params.updateId} on lane ${params.laneKey} after ${age}; marking the update failed.`; -} - -async function failTimedOutWebhookSpooledUpdate(params: { - log: (line: string) => void; - message: string; - update: ClaimedTelegramSpooledUpdate; -}): Promise { - const failed = await failTelegramSpooledUpdateClaim({ - update: params.update, - reason: "handler-timeout", - message: params.message, - }); - if (!failed) { - params.log( - `[telegram][diag] timed out webhook spooled update ${params.update.updateId} no longer had a processing marker to fail.`, - ); - } -} - -async function waitForTimedOutWebhookReplayGrace(params: { - log: (line: string) => void; - replayTask: Promise<{ deferredWork?: TelegramSpooledReplayDeferredParticipant }>; - updateId: number; -}): Promise { - let timer: ReturnType | undefined; - try { - return await Promise.race([ - params.replayTask.then( - () => true, - (replayErr: unknown) => { - params.log( - `[telegram][diag] timed out webhook spooled update ${params.updateId} replay later failed: ${formatErrorMessage(replayErr)}`, - ); - return true; - }, - ), - new Promise((resolve) => { - timer = setTimeout(() => resolve(false), TELEGRAM_WEBHOOK_SPOOLED_HANDLER_ABORT_GRACE_MS); - timer.unref?.(); - }), - ]); - } finally { - if (timer) { - clearTimeout(timer); - } - } -} - -type WebhookSpooledUpdateHandlerResult = { - retainLaneGuardTask?: Promise; -}; - -async function runWebhookSpooledReplayWithTimeout(params: { - bot: ReturnType; - laneKey: string; - rawUpdate: object; - update: Parameters["handleUpdate"]>[0]; - updateId: number; -}): Promise<{ deferredWork?: TelegramSpooledReplayDeferredParticipant }> { - let timer: ReturnType | undefined; - const replayTask = runWithTelegramSpooledReplayUpdate(params.rawUpdate, async () => { - await params.bot.handleUpdate(params.update); - }); - replayTask.catch(() => undefined); - const timeout = new Promise((_resolve, reject) => { - timer = setTimeout(() => { - reject( - new WebhookSpooledHandlerTimeoutError( - formatWebhookSpooledHandlerTimeoutMessage({ - laneKey: params.laneKey, - updateId: params.updateId, - }), - replayTask, - ), - ); - }, TELEGRAM_WEBHOOK_SPOOLED_HANDLER_TIMEOUT_MS); - timer.unref?.(); - }); - try { - return await Promise.race([replayTask, timeout]); - } finally { - if (timer) { - clearTimeout(timer); - } - } -} - -async function waitForWebhookSpooledDeferredWork(params: { - deferredWork: TelegramSpooledReplayDeferredParticipant; - laneKey: string; - log: (line: string) => void; - update: ClaimedTelegramSpooledUpdate; -}): Promise { - let timeoutError: Error | undefined; - const timer = setTimeout(() => { - const age = formatDurationPrecise(TELEGRAM_WEBHOOK_SPOOLED_HANDLER_TIMEOUT_MS); - const message = `Telegram webhook spool buffered processing timed out behind update ${params.update.updateId} on lane ${params.laneKey} after ${age}; marking the update failed.`; - params.log(`[telegram] ${message}`); - timeoutError = new Error(message); - params.deferredWork.settle({ - kind: "failed-retryable", - error: timeoutError, - }); - }, TELEGRAM_WEBHOOK_SPOOLED_HANDLER_TIMEOUT_MS); - timer.unref?.(); - try { - const result = await params.deferredWork.task.catch( - (err: unknown): TelegramMessageProcessingResult => ({ - kind: "failed-retryable", - error: err, - }), - ); - // A durable-adoption hold can discard the timeout and settle completed. - // Only the exact timeout result owns the handler-timeout failure path. - return timeoutError !== undefined && - result.kind === "failed-retryable" && - result.error === timeoutError - ? { ...result, timedOut: true } - : result; - } finally { - clearTimeout(timer); - } -} - -async function handleWebhookSpooledUpdate(params: { - accountId: string; - abortSignal?: AbortSignal; - bot: ReturnType; - log: (line: string) => void; - update: ClaimedTelegramSpooledUpdate; -}): Promise { - let replay: { deferredWork?: TelegramSpooledReplayDeferredParticipant }; - try { - const rawUpdate = params.update.update; - if (!rawUpdate || typeof rawUpdate !== "object") { - throw new Error("Telegram spooled webhook update payload was invalid."); - } - const laneKey = resolveWebhookSpooledUpdateLaneKey(rawUpdate); - const update = rawUpdate as Parameters[0]; - replay = await runWebhookSpooledReplayWithTimeout({ - bot: params.bot, - laneKey, - rawUpdate, - update, - updateId: params.update.updateId, - }); - } catch (err) { - if (err instanceof WebhookSpooledHandlerTimeoutError) { - params.log(`[telegram] ${err.message}`); - const scopedReplyFenceLaneKey = buildTelegramReplyFenceLaneKey({ - accountId: params.accountId, - sequentialKey: resolveWebhookSpooledUpdateLaneKey(params.update.update), - }); - const abortedReplyWork = supersedeTelegramReplyFenceLane(scopedReplyFenceLaneKey); - if (!abortedReplyWork) { - params.log( - `[telegram][diag] timed out webhook spooled update ${params.update.updateId} had no active reply fence on lane ${scopedReplyFenceLaneKey}.`, - ); - } - await failTimedOutWebhookSpooledUpdate({ - log: params.log, - message: err.message, - update: params.update, - }); - const replaySettled = await waitForTimedOutWebhookReplayGrace({ - log: params.log, - replayTask: err.replayTask, - updateId: params.update.updateId, - }); - if (replaySettled) { - return {}; - } - return { - retainLaneGuardTask: err.replayTask.catch((replayErr: unknown) => { - params.log( - `[telegram][diag] timed out webhook spooled update ${params.update.updateId} replay later failed: ${formatErrorMessage(replayErr)}`, - ); - }), - }; - } - await releaseFailedWebhookSpooledUpdate({ - err, - log: params.log, - update: params.update, - }); - return {}; - } - if (replay.deferredWork) { - const result = await waitForWebhookSpooledDeferredWork({ - deferredWork: replay.deferredWork, - laneKey: resolveWebhookSpooledUpdateLaneKey(params.update.update), - log: params.log, - update: params.update, - }); - if (result.kind === "failed-retryable") { - if (result.timedOut) { - await failTimedOutWebhookSpooledUpdate({ - log: params.log, - message: formatErrorMessage(result.error), - update: params.update, - }); - return {}; - } - await releaseFailedWebhookSpooledUpdate({ - err: result.error, - log: params.log, - update: params.update, - }); - return {}; - } - } - await completeTelegramSpooledUpdateWithRetry({ - update: params.update, - abortSignal: params.abortSignal, - onRetry: ({ attempt, delayMs, error }) => { - params.log( - `[telegram][diag] webhook spooled update ${params.update.updateId} completion retry ${attempt} scheduled in ${formatDurationPrecise(delayMs)}: ${formatErrorMessage(error)}`, - ); - }, - }); - return {}; -} - export async function startTelegramWebhook(opts: { token: string; accountId?: string; @@ -703,9 +320,6 @@ export async function startTelegramWebhook(opts: { const spoolDir = opts.spoolDir ?? resolveTelegramIngressSpoolDir({ accountId: opts.accountId }); let shutDown = false; const shutdownAbortController = new AbortController(); - const webhookAbortSignal = opts.abortSignal - ? AbortSignal.any([shutdownAbortController.signal, opts.abortSignal]) - : shutdownAbortController.signal; const telegramAccountConfig = opts.config ? mergeTelegramAccountConfig(opts.config, opts.accountId ?? "default") : undefined; @@ -755,6 +369,7 @@ export async function startTelegramWebhook(opts: { const log = (line: string) => runtime.log?.(line); let drainActive = false; let drainRequested = false; + let webhookIngressDrain: ReturnType | undefined; const drainWebhookSpool = async (): Promise => { if (shutDown || opts.abortSignal?.aborted) { return; @@ -766,105 +381,30 @@ export async function startTelegramWebhook(opts: { drainActive = true; drainRequested = false; try { - const activeWebhookSpooledLaneKeys = resolveActiveWebhookSpooledLaneKeys(spoolDir); - await recoverStaleTelegramSpooledUpdateClaims({ + // Shutdown must abort in-flight drain work (tombstone retries), not just + // stop the next claim; the composed signal carries webhook stop + caller abort. + const webhookAbortSignal = opts.abortSignal + ? AbortSignal.any([shutdownAbortController.signal, opts.abortSignal]) + : shutdownAbortController.signal; + webhookIngressDrain ??= createTelegramTransportIngressDrain({ spoolDir, - staleMs: 0, - shouldRecover: (claim) => - !activeWebhookSpooledLaneKeys.has(resolveWebhookSpooledUpdateLaneKey(claim.update)) && - !isTelegramSpooledUpdateClaimOwnedByOtherLiveProcess(claim), - shouldRecoverCorrupt: (claim) => - !(claim.laneKey && activeWebhookSpooledLaneKeys.has(claim.laneKey)) && - !isTelegramSpooledCorruptClaimOwnedByOtherLiveProcess(claim), + bot, + cfg: opts.config ?? {}, + accountId: opts.accountId ?? "default", + // Pre-migration product default: 25m claim→adoption stall for webhook. + adoptionStallTimeoutMs: 25 * 60_000, + abortSignal: webhookAbortSignal, + onLog: (message) => log(`webhook ${message}`), }); - const claimedLaneKeys = new Set( - ( - await listTelegramSpooledUpdateClaims({ - spoolDir, - }) - ).map((claim) => resolveWebhookSpooledUpdateLaneKey(claim.update)), - ); - const updates = await listTelegramSpooledUpdates({ - spoolDir, - limit: TELEGRAM_WEBHOOK_SPOOLED_DRAIN_SCAN_LIMIT, + await webhookIngressDrain.drainOnce({ + shouldStop: () => shutDown || webhookAbortSignal.aborted, }); - const candidateUpdateIds = updates.map((update) => update.updateId); - const blockedLaneKeys = new Set([...activeWebhookSpooledLaneKeys, ...claimedLaneKeys]); - for (const update of updates) { - // Release stamps lastAttemptAt; block the lane until backoff expires so - // webhook replay cannot hot-loop a retryable poison update. - if (resolveSpooledUpdateRetryDelayMs(update) > 0) { - blockedLaneKeys.add(resolveWebhookSpooledUpdateLaneKey(update.update)); - } - } - let started = 0; - while (started < TELEGRAM_WEBHOOK_SPOOLED_DRAIN_START_LIMIT) { - if (shutDown || opts.abortSignal?.aborted) { - break; - } - const claimedUpdate = await claimNextTelegramSpooledUpdate({ - spoolDir, - blockedLaneKeys, - candidateUpdateIds, - scanLimit: TELEGRAM_WEBHOOK_SPOOLED_DRAIN_SCAN_LIMIT, - }); - if (!claimedUpdate) { - break; - } - const laneKey = resolveWebhookSpooledUpdateLaneKey(claimedUpdate.update); - const handlerKey = buildWebhookSpooledHandlerKey({ spoolDir, laneKey }); - // Webhook HTTP requests and same-process restarts can overlap; keep - // one process-global active claim per spool lane to preserve ordering. - const handlerState: ActiveWebhookSpooledHandler = { laneKey }; - activeWebhookSpooledHandlersByLane.set(handlerKey, handlerState); - blockedLaneKeys.add(laneKey); - // Claim ownership has a finite lease; refresh while the handler runs so - // another process cannot recover and replay this update concurrently. - const stopClaimRefresh = startWebhookSpooledUpdateClaimRefresh({ - log, - update: claimedUpdate, - }); - let retainLaneGuardTask: Promise | undefined; - void handleWebhookSpooledUpdate({ - accountId: opts.accountId ?? "default", - abortSignal: webhookAbortSignal, - bot, - log, - update: claimedUpdate, - }) - .then((result) => { - retainLaneGuardTask = result.retainLaneGuardTask; - if (retainLaneGuardTask) { - void retainLaneGuardTask.finally(() => { - if (activeWebhookSpooledHandlersByLane.get(handlerKey) === handlerState) { - activeWebhookSpooledHandlersByLane.delete(handlerKey); - } - void Promise.resolve().then(drainWebhookSpool); - }); - } - }) - .catch((err: unknown) => { - runtime.log?.( - `[telegram][diag] webhook spooled update ${claimedUpdate.updateId} handler failed after claim: ${formatErrorMessage(err)}`, - ); - }) - .finally(() => { - stopClaimRefresh(); - if ( - !retainLaneGuardTask && - activeWebhookSpooledHandlersByLane.get(handlerKey) === handlerState - ) { - activeWebhookSpooledHandlersByLane.delete(handlerKey); - } - void Promise.resolve().then(drainWebhookSpool); - }); - started += 1; - } } catch (err) { - runtime.log?.(`[telegram][diag] webhook spool drain failed: ${formatErrorMessage(err)}`); + log(`[telegram][diag] webhook spool drain failed: ${formatErrorMessage(err)}`); } finally { drainActive = false; if (drainRequested && !shutDown && !opts.abortSignal?.aborted) { + drainRequested = false; void Promise.resolve().then(drainWebhookSpool); } } @@ -1008,6 +548,8 @@ export async function startTelegramWebhook(opts: { if (drainTimer) { clearInterval(drainTimer); } + webhookIngressDrain?.dispose(); + webhookIngressDrain = undefined; server.close(); await bot.stop(); // The webhook owns this transport because it resolved and injected it into