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 <Galin.Iliev@microsoft.com>
Co-authored-by: Ayaan Zaidi <hi@obviy.us>

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",
+    );
+  });
 });

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 {

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();
     },
   };
 }

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<typeof runTelegramIngressWorkerRuntime>[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",
+      }),
+    );
+  });
+});