From 1b58ecb5af0a7ff5b366fdaae88d43bd7a1e8a71 Mon Sep 17 00:00:00 2001 From: Galin Iliev Date: Tue, 4 Aug 2026 10:19:13 +0300 Subject: [PATCH] fix(telegram): use monotonic polling liveness Measure Telegram polling deadlines with a monotonic clock so wall-clock corrections cannot trigger false restarts or negative durations. Preserve the worker deadline and continued-stall recovery. Co-authored-by: Galin Iliev Co-authored-by: Ayaan Zaidi --- .../telegram/src/polling-liveness.test.ts | 89 +++++++++++++++++-- extensions/telegram/src/polling-liveness.ts | 82 ++++++++++------- .../telegram/src/polling-session.test.ts | 4 + .../telegram-ingress-worker.deadline.test.ts | 82 +++++++++++++++++ 4 files changed, 217 insertions(+), 40 deletions(-) create mode 100644 extensions/telegram/src/telegram-ingress-worker.deadline.test.ts diff --git a/extensions/telegram/src/polling-liveness.test.ts b/extensions/telegram/src/polling-liveness.test.ts index 3b56b7ed1e87..a27bd9170bc1 100644 --- a/extensions/telegram/src/polling-liveness.test.ts +++ b/extensions/telegram/src/polling-liveness.test.ts @@ -6,12 +6,17 @@ const POLL_STALL_THRESHOLD_MS = 90_000; describe("TelegramPollingLivenessTracker", () => { it("records successful getUpdates calls and publishes poll success time", () => { - const nowValues = [0, 10, 25]; - const now = vi.fn(() => nowValues.shift() ?? 25); + let now = 0; const onPollSuccess = vi.fn(); - const tracker = new TelegramPollingLivenessTracker({ now, onPollSuccess }); + const tracker = new TelegramPollingLivenessTracker({ + now: () => now, + monotonicNow: () => now, + onPollSuccess, + }); + now = 10; tracker.noteGetUpdatesStarted({ offset: 42 }); + now = 25; tracker.noteGetUpdatesSuccess([{ update_id: 1 }, { update_id: 2 }]); tracker.noteGetUpdatesFinished(); @@ -23,8 +28,10 @@ describe("TelegramPollingLivenessTracker", () => { it("detects stale polling without considering unrelated API activity", () => { let now = 0; - const tracker = new TelegramPollingLivenessTracker({ now: () => now }); + const tracker = new TelegramPollingLivenessTracker({ monotonicNow: () => now }); + now = 45_000; + expect(tracker.detectStall({ thresholdMs: POLL_STALL_THRESHOLD_MS })).toBeNull(); now = 120_001; expect( tracker.detectStall({ @@ -35,8 +42,10 @@ describe("TelegramPollingLivenessTracker", () => { it("detects and throttles stale polling diagnostics", () => { let now = 0; - const tracker = new TelegramPollingLivenessTracker({ now: () => now }); + const tracker = new TelegramPollingLivenessTracker({ monotonicNow: () => now }); + now = 45_000; + expect(tracker.detectStall({ thresholdMs: POLL_STALL_THRESHOLD_MS })).toBeNull(); now = 120_001; const stall = tracker.detectStall({ thresholdMs: POLL_STALL_THRESHOLD_MS, @@ -54,11 +63,16 @@ describe("TelegramPollingLivenessTracker", () => { it("reports active stuck getUpdates calls", () => { let now = 0; - const tracker = new TelegramPollingLivenessTracker({ now: () => now }); + const tracker = new TelegramPollingLivenessTracker({ + now: () => now, + monotonicNow: () => now, + }); now = 1; tracker.noteGetUpdatesStarted({ offset: 7 }); + now = 45_000; + expect(tracker.detectStall({ thresholdMs: POLL_STALL_THRESHOLD_MS })).toBeNull(); now = 120_001; const stall = tracker.detectStall({ thresholdMs: POLL_STALL_THRESHOLD_MS, @@ -71,4 +85,67 @@ describe("TelegramPollingLivenessTracker", () => { tracker.noteGetUpdatesSuccess([]); tracker.noteGetUpdatesFinished(); }); + + it("does not treat a wall-clock correction as an active getUpdates stall", () => { + let wallNow = 1_000; + let monotonicNow = 0; + const tracker = new TelegramPollingLivenessTracker({ + now: () => wallNow, + monotonicNow: () => monotonicNow, + }); + + tracker.noteGetUpdatesStarted({ offset: 7 }); + + wallNow += 154_000; + monotonicNow += 30_000; + + expect(tracker.detectStall({ thresholdMs: POLL_STALL_THRESHOLD_MS })).toBeNull(); + + monotonicNow += POLL_STALL_THRESHOLD_MS - 30_000 + 1; + const stall = tracker.detectStall({ + thresholdMs: POLL_STALL_THRESHOLD_MS, + }); + + expect(stall?.message).toContain("active getUpdates stuck"); + }); + + it("measures completed polls across backward wall-clock corrections", () => { + let wallNow = 200_000; + let monotonicNow = 1_000; + const tracker = new TelegramPollingLivenessTracker({ + now: () => wallNow, + monotonicNow: () => monotonicNow, + }); + + tracker.noteGetUpdatesStarted({ offset: 7 }); + wallNow -= 124_193; + monotonicNow += 30_000; + tracker.noteGetUpdatesSuccess([]); + tracker.noteGetUpdatesFinished(); + + expect(tracker.formatDiagnosticFields()).toContain( + "startedAt=200000 finishedAt=75807 durationMs=30000", + ); + }); + + it("rebases liveness after the watchdog itself was paused", () => { + let now = 1_000; + const tracker = new TelegramPollingLivenessTracker({ + now: () => now, + monotonicNow: () => now, + }); + tracker.noteGetUpdatesStarted({ offset: 7 }); + + now += 10 * 60 * 60 * 1_000; + expect(tracker.detectStall({ thresholdMs: POLL_STALL_THRESHOLD_MS })).toBeNull(); + + now += 30_000; + expect(tracker.detectStall({ thresholdMs: POLL_STALL_THRESHOLD_MS })).toBeNull(); + now += 30_000; + expect(tracker.detectStall({ thresholdMs: POLL_STALL_THRESHOLD_MS })).toBeNull(); + now += 30_001; + expect(tracker.detectStall({ thresholdMs: POLL_STALL_THRESHOLD_MS })?.message).toContain( + "active getUpdates stuck", + ); + }); }); diff --git a/extensions/telegram/src/polling-liveness.ts b/extensions/telegram/src/polling-liveness.ts index 1298e18f3340..92857a020cd9 100644 --- a/extensions/telegram/src/polling-liveness.ts +++ b/extensions/telegram/src/polling-liveness.ts @@ -4,6 +4,7 @@ import { formatErrorMessage } from "openclaw/plugin-sdk/ssrf-runtime"; type TelegramPollingLivenessTrackerOptions = { now?: () => number; + monotonicNow?: () => number; onPollSuccess?: (finishedAt: number) => void; }; @@ -12,20 +13,22 @@ type TelegramPollingStall = { }; export class TelegramPollingLivenessTracker { - #lastGetUpdatesAt: number; - #lastGetUpdatesActivityAt: number; + #lastGetUpdatesActivityMonotonicAt: number; #lastGetUpdatesStartedAt: number | null = null; + #lastGetUpdatesStartedMonotonicAt: number | null = null; #lastGetUpdatesFinishedAt: number | null = null; #lastGetUpdatesDurationMs: number | null = null; #lastGetUpdatesOutcome = "not-started"; #lastGetUpdatesError: string | null = null; #lastGetUpdatesOffset: number | null = null; #inFlightGetUpdates = 0; - #stallDiagLoggedAt = 0; + #stallDiagLoggedMonotonicAt = 0; + #lastStallCheckMonotonicAt: number; constructor(private readonly options: TelegramPollingLivenessTrackerOptions = {}) { - this.#lastGetUpdatesAt = this.#now(); - this.#lastGetUpdatesActivityAt = this.#lastGetUpdatesAt; + const monotonicNow = this.#monotonicNow(); + this.#lastGetUpdatesActivityMonotonicAt = monotonicNow; + this.#lastStallCheckMonotonicAt = monotonicNow; } get inFlightGetUpdates() { @@ -33,9 +36,12 @@ export class TelegramPollingLivenessTracker { } noteGetUpdatesStarted(payload: unknown, at = this.#now()) { - this.#lastGetUpdatesAt = at; - this.#lastGetUpdatesActivityAt = at; + const startedMonotonicAt = this.#monotonicNow(); + this.#lastGetUpdatesActivityMonotonicAt = startedMonotonicAt; this.#lastGetUpdatesStartedAt = at; + this.#lastGetUpdatesStartedMonotonicAt = startedMonotonicAt; + this.#lastGetUpdatesFinishedAt = null; + this.#lastGetUpdatesDurationMs = null; this.#lastGetUpdatesOffset = resolveGetUpdatesOffset(payload); this.#inFlightGetUpdates += 1; this.#lastGetUpdatesOutcome = "started"; @@ -43,29 +49,20 @@ export class TelegramPollingLivenessTracker { } noteGetUpdatesSuccess(result: unknown, at = this.#now()) { - this.#lastGetUpdatesActivityAt = at; - this.#lastGetUpdatesFinishedAt = at; - this.#lastGetUpdatesDurationMs = - this.#lastGetUpdatesStartedAt == null ? null : at - this.#lastGetUpdatesStartedAt; + this.#noteGetUpdatesCompleted(at); this.#lastGetUpdatesOutcome = Array.isArray(result) ? `ok:${result.length}` : "ok"; this.options.onPollSuccess?.(at); } noteGetUpdatesSuccessCount(count: number, at = this.#now()) { - this.#lastGetUpdatesActivityAt = at; - this.#lastGetUpdatesFinishedAt = at; - this.#lastGetUpdatesDurationMs = - this.#lastGetUpdatesStartedAt == null ? null : at - this.#lastGetUpdatesStartedAt; + this.#noteGetUpdatesCompleted(at); const normalizedCount = Number.isFinite(count) ? Math.max(0, Math.floor(count)) : 0; this.#lastGetUpdatesOutcome = `ok:${normalizedCount}`; this.options.onPollSuccess?.(at); } noteGetUpdatesError(err: unknown, at = this.#now()) { - this.#lastGetUpdatesActivityAt = at; - this.#lastGetUpdatesFinishedAt = at; - this.#lastGetUpdatesDurationMs = - this.#lastGetUpdatesStartedAt == null ? null : at - this.#lastGetUpdatesStartedAt; + this.#noteGetUpdatesCompleted(at); this.#lastGetUpdatesOutcome = "error"; this.#lastGetUpdatesError = formatErrorMessage(err); } @@ -74,28 +71,31 @@ export class TelegramPollingLivenessTracker { this.#inFlightGetUpdates = Math.max(0, this.#inFlightGetUpdates - 1); } - noteGetUpdatesActivity(at = this.#now()) { - this.#lastGetUpdatesActivityAt = at; + noteGetUpdatesActivity() { + this.#lastGetUpdatesActivityMonotonicAt = this.#monotonicNow(); } - detectStall(params: { thresholdMs: number; now?: number }): TelegramPollingStall | null { - const now = params.now ?? this.#now(); - const activeElapsed = - this.#inFlightGetUpdates > 0 && this.#lastGetUpdatesStartedAt != null - ? now - this.#lastGetUpdatesActivityAt - : 0; - const idleElapsed = - this.#inFlightGetUpdates > 0 - ? 0 - : now - (this.#lastGetUpdatesFinishedAt ?? this.#lastGetUpdatesAt); - const elapsed = this.#inFlightGetUpdates > 0 ? activeElapsed : idleElapsed; + detectStall(params: { thresholdMs: number }): TelegramPollingStall | null { + const monotonicNow = this.#monotonicNow(); + const checkGap = monotonicNow - this.#lastStallCheckMonotonicAt; + this.#lastStallCheckMonotonicAt = monotonicNow; + // The watchdog cannot distinguish a stalled poll from delayed callbacks after + // missing two full detection windows. Rebase once, then observe normally. + if (checkGap > params.thresholdMs * 2) { + this.#lastGetUpdatesActivityMonotonicAt = monotonicNow; + return null; + } + const elapsed = monotonicNow - this.#lastGetUpdatesActivityMonotonicAt; if (elapsed <= params.thresholdMs) { return null; } - if (this.#stallDiagLoggedAt && now - this.#stallDiagLoggedAt < params.thresholdMs / 2) { + if ( + this.#stallDiagLoggedMonotonicAt && + monotonicNow - this.#stallDiagLoggedMonotonicAt < params.thresholdMs / 2 + ) { return null; } - this.#stallDiagLoggedAt = now; + this.#stallDiagLoggedMonotonicAt = monotonicNow; const elapsedLabel = this.#inFlightGetUpdates > 0 @@ -115,6 +115,20 @@ export class TelegramPollingLivenessTracker { #now(): number { return this.options.now?.() ?? Date.now(); } + + #monotonicNow(): number { + return this.options.monotonicNow?.() ?? performance.now(); + } + + #noteGetUpdatesCompleted(finishedAt: number): void { + const finishedMonotonicAt = this.#monotonicNow(); + this.#lastGetUpdatesActivityMonotonicAt = finishedMonotonicAt; + this.#lastGetUpdatesFinishedAt = finishedAt; + this.#lastGetUpdatesDurationMs = + this.#lastGetUpdatesStartedMonotonicAt == null + ? null + : finishedMonotonicAt - this.#lastGetUpdatesStartedMonotonicAt; + } } function resolveGetUpdatesOffset(payload: unknown): number | null { diff --git a/extensions/telegram/src/polling-session.test.ts b/extensions/telegram/src/polling-session.test.ts index cff580f8bc07..f23080ed926e 100644 --- a/extensions/telegram/src/polling-session.test.ts +++ b/extensions/telegram/src/polling-session.test.ts @@ -323,6 +323,7 @@ function makeIsolatedBot(params?: { } function installPollingStallWatchdogHarness(dateNowSequence: readonly number[] = [0, 0]) { + let monotonicNow = dateNowSequence[0] ?? 0; let watchdog: (() => void) | undefined; let resolveWatchdog: ((fn: () => void) => void) | undefined; const watchdogReady = new Promise<() => void>((resolve) => { @@ -364,6 +365,7 @@ function installPollingStallWatchdogHarness(dateNowSequence: readonly number[] = realClearTimeout(timeoutId); }); const dateNowSpy = vi.spyOn(Date, "now"); + const performanceNowSpy = vi.spyOn(performance, "now").mockImplementation(() => monotonicNow); for (const value of dateNowSequence) { dateNowSpy.mockImplementationOnce(() => value); } @@ -403,6 +405,7 @@ function installPollingStallWatchdogHarness(dateNowSequence: readonly number[] = }); }, setNow(now: number) { + monotonicNow = now; dateNowSpy.mockReset(); dateNowSpy.mockImplementation(() => now); }, @@ -412,6 +415,7 @@ function installPollingStallWatchdogHarness(dateNowSequence: readonly number[] = setTimeoutSpy.mockRestore(); clearTimeoutSpy.mockRestore(); dateNowSpy.mockRestore(); + performanceNowSpy.mockRestore(); }, }; } diff --git a/extensions/telegram/src/telegram-ingress-worker.deadline.test.ts b/extensions/telegram/src/telegram-ingress-worker.deadline.test.ts new file mode 100644 index 000000000000..188408cdce62 --- /dev/null +++ b/extensions/telegram/src/telegram-ingress-worker.deadline.test.ts @@ -0,0 +1,82 @@ +// Telegram tests cover the isolated ingress request deadline boundary. +import { afterEach, describe, expect, it, vi } from "vitest"; +import type { + TelegramIngressWorkerCommand, + TelegramIngressWorkerMessage, +} from "./telegram-ingress-worker.js"; +import { runTelegramIngressWorkerRuntime } from "./telegram-ingress-worker.runtime.js"; + +type RuntimePort = Parameters[0]["port"]; + +afterEach(() => { + vi.useRealTimers(); +}); + +describe("telegram ingress worker request deadline", () => { + it("aborts a stalled response body at the getUpdates deadline", async () => { + vi.useFakeTimers(); + const messages: TelegramIngressWorkerMessage[] = []; + const listeners = new Set<(message: TelegramIngressWorkerCommand) => void>(); + let requestSignal: AbortSignal | undefined; + const sendCommand = (message: TelegramIngressWorkerCommand) => { + for (const listener of listeners) { + listener(message); + } + }; + const port: RuntimePort = { + postMessage(message) { + messages.push(message); + if (message.type === "poll-error") { + sendCommand({ type: "stop" }); + } + }, + onMessage(listener) { + listeners.add(listener); + }, + close() {}, + }; + const fetchImpl: typeof fetch = async (_url, init) => { + requestSignal = init?.signal ?? undefined; + const signal = requestSignal; + return new Response( + new ReadableStream({ + start(controller) { + signal?.addEventListener("abort", () => controller.error(signal.reason), { + once: true, + }); + }, + }), + { status: 200 }, + ); + }; + const done = runTelegramIngressWorkerRuntime({ + options: { + token: "test-auth-token", + accountId: "acct", + initialUpdateId: null, + spoolDir: "/tmp/openclaw-telegram-ingress-worker-deadline-test", + apiRoot: "https://api.telegram.test", + }, + port, + deps: { + fetch: fetchImpl, + closeTransport: async () => {}, + }, + }); + + await vi.advanceTimersByTimeAsync(44_999); + expect(requestSignal?.aborted).toBe(false); + expect(messages.some((message) => message.type === "poll-error")).toBe(false); + + await vi.advanceTimersByTimeAsync(1); + await done; + + expect(requestSignal?.aborted).toBe(true); + expect(messages).toContainEqual( + expect.objectContaining({ + type: "poll-error", + message: "Telegram getUpdates timed out", + }), + ); + }); +});