vultr/openclaw
1
0
Fork 0
mirror of https://github.com/openclaw/openclaw.git synced 2026-07-25 15:31:16 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(telegram): polling worker pins a CPU core when the Bot API answers empty getUpdates instantly (#111063)

Browse Source 
* fix(telegram): pace fast empty ingress polls

Some Bot API deployments (notably self-hosted telegram-bot-api servers and
intermediary proxies) can answer getUpdates immediately instead of holding
the connection for the requested long-poll timeout. With no pending updates
that turns the polling ingress worker loop into a busy spin: each empty
poll completes in milliseconds and the next one starts right away, pinning
a CPU core.

Enforce a 1s floor between consecutive empty getUpdates cycles, measured
from poll start so responses that honor the long-poll timeout are never
delayed. The wait listens to the worker stop signal, keeping shutdown
immediate. Non-empty polls are not paced, so pending traffic still drains
at full speed.

* fix(telegram): adapt empty poll backoff

---------

Co-authored-by: Arseniy Palagin <valeradzigurda3@gmail.com>
Co-authored-by: Peter Steinberger <steipete@gmail.com>
This commit is contained in:
Arseniy Palagin
2026-07-19 05:54:03 +03:00
committed by GitHub
parent 491b87c936
commit bdfe2a4e60
2 changed files with 112 additions and 28 deletions
Download Patch File Download Diff File Expand all files Collapse all files

64
extensions/telegram/src/telegram-ingress-worker.runtime.test.ts
View File

@@ -23,7 +23,10 @@ function htmlResponse(status: number, body: string): Response {
});
}
function createRuntime(responses: Response[]): {
function createRuntime(
responses: Response[],
options: { stopAfterPollSuccesses?: number } = {},
): {
calls: number[];
messages: TelegramIngressWorkerMessage[];
done: Promise<void>;
@@ -31,6 +34,7 @@ function createRuntime(responses: Response[]): {
const calls: number[] = [];
const messages: TelegramIngressWorkerMessage[] = [];
const listeners = new Set<(message: TelegramIngressWorkerCommand) => void>();
let pollSuccesses = 0;
const sendCommand = (message: TelegramIngressWorkerCommand) => {
for (const listener of listeners) {
listener(message);
@@ -39,8 +43,18 @@ function createRuntime(responses: Response[]): {
const port: RuntimePort = {
postMessage(message) {
messages.push(message);
if (message.type === "update") {
sendCommand({
type: "spool-ack",
requestId: message.requestId,
result: { ok: true, updateId: 42 },
});
}
if (message.type === "poll-success") {
sendCommand({ type: "stop" });
pollSuccesses += 1;
if (pollSuccesses >= (options.stopAfterPollSuccesses ?? 1)) {
sendCommand({ type: "stop" });
}
}
},
onMessage(listener) {
@@ -79,6 +93,52 @@ afterEach(() => {
vi.useRealTimers();
});
describe("telegram ingress worker poll cadence", () => {
it("backs off consecutive empty polls without hot spinning", async () => {
vi.useFakeTimers();
vi.setSystemTime(new Date("2026-07-01T12:00:00.000Z"));
const runtime = createRuntime(
Array.from({ length: 9 }, () => jsonResponse(200, { ok: true, result: [] })),
{ stopAfterPollSuccesses: 9 },
);
expect(runtime.calls).toHaveLength(1);
await flushRuntime();
expect(runtime.calls).toHaveLength(2);
await vi.advanceTimersByTimeAsync(3_550);
await runtime.done;
const firstCall = expectDefined(runtime.calls[0], "first Telegram poll call");
expect(runtime.calls.map((calledAt) => calledAt - firstCall)).toEqual([
0, 0, 50, 150, 350, 750, 1_550, 2_550, 3_550,
]);
});
it("resets to immediate polling when activity resumes", async () => {
vi.useFakeTimers();
vi.setSystemTime(new Date("2026-07-01T12:00:00.000Z"));
const runtime = createRuntime(
[
jsonResponse(200, { ok: true, result: [] }),
jsonResponse(200, { ok: true, result: [] }),
jsonResponse(200, { ok: true, result: [{ update_id: 42 }] }),
jsonResponse(200, { ok: true, result: [] }),
],
{ stopAfterPollSuccesses: 4 },
);
await flushRuntime();
await vi.advanceTimersByTimeAsync(50);
await runtime.done;
const firstCall = expectDefined(runtime.calls[0], "first Telegram poll call");
expect(runtime.calls.map((calledAt) => calledAt - firstCall)).toEqual([0, 0, 50, 50]);
expect(runtime.messages).toContainEqual(
expect.objectContaining({ type: "spooled", updateId: 42 }),
);
});
});
describe("telegram ingress worker durable-before-offset", () => {
it("advances getUpdates offset only after parent spool-ack commits", async () => {
vi.useFakeTimers();

76
extensions/telegram/src/telegram-ingress-worker.runtime.ts
View File

@@ -1,6 +1,11 @@
// Telegram plugin module implements telegram ingress worker behavior.
import { parentPort, workerData } from "node:worker_threads";
import { readResponseWithLimit } from "openclaw/plugin-sdk/response-limit-runtime";
import {
computeBackoff,
sleepWithAbort,
type BackoffPolicy,
} from "openclaw/plugin-sdk/runtime-env";
import { resolveTelegramAllowedUpdates } from "./allowed-updates.js";
import { normalizeTelegramApiRoot } from "./api-root.js";
import { resolveTelegramTransport } from "./fetch.js";
@@ -21,8 +26,18 @@ const pollLimit = 100;
// getUpdates can return up to 100 updates; 4 MiB is a generous bound that no legitimate
// Telegram Bot API response will reach, guarding against misbehaving/hostile endpoints.
const TELEGRAM_GET_UPDATES_MAX_RESPONSE_BYTES = 4 * 1024 * 1024;
const retryInitialMs = 1000;
const retryMaxMs = 30_000;
const TELEGRAM_EMPTY_POLL_BACKOFF_POLICY: BackoffPolicy = {
initialMs: 50,
maxMs: 1_000,
factor: 2,
jitter: 0,
};
const TELEGRAM_RETRY_BACKOFF_POLICY: BackoffPolicy = {
initialMs: 1_000,
maxMs: 30_000,
factor: 2,
jitter: 0,
};
type TelegramGetUpdatesJson = {
ok?: unknown;
@@ -55,22 +70,6 @@ type TelegramIngressWorkerRuntimeData = TelegramIngressWorkerOptions & {
runtime: typeof TELEGRAM_INGRESS_WORKER_RUNTIME_MARKER;
};
function sleep(ms: number, signal: AbortSignal): Promise<void> {
if (signal.aborted) {
return Promise.resolve();
}
return new Promise((resolve) => {
const done = () => {
clearTimeout(timeout);
signal.removeEventListener("abort", done);
resolve();
};
const timeout = setTimeout(done, ms);
timeout.unref?.();
signal.addEventListener("abort", done, { once: true });
});
}
function formatErrorMessage(err: unknown): string {
if (err instanceof Error) {
return err.message || err.name;
@@ -98,10 +97,6 @@ function postPollError(port: TelegramIngressRuntimePort, err: unknown): void {
});
}
function resolveBackoff(attempt: number): number {
return Math.min(retryMaxMs, retryInitialMs * 2 ** Math.max(0, attempt - 1));
}
function createTelegramGetUpdatesError(params: {
message: string;
errorCode?: number;
@@ -200,6 +195,7 @@ export async function runTelegramIngressWorkerRuntime(params: {
const pollTimeoutSeconds = resolveTelegramLongPollTimeoutSeconds(options.timeoutSeconds);
let lastUpdateId = options.initialUpdateId;
let failures = 0;
let consecutiveEmptyPolls = 0;
port.onMessage((message) => {
if (message?.type === "stop") {
@@ -284,10 +280,30 @@ export async function runTelegramIngressWorkerRuntime(params: {
count: result.length,
finishedAt: Date.now(),
});
if (result.length > 0) {
consecutiveEmptyPolls = 0;
continue;
}
consecutiveEmptyPolls += 1;
if (consecutiveEmptyPolls > 1) {
// Some Bot API endpoints return empty long polls immediately. Escalate only
// while idle, then reset above so active chats keep draining without delay.
const minIntervalMs = computeBackoff(
TELEGRAM_EMPTY_POLL_BACKOFF_POLICY,
consecutiveEmptyPolls - 1,
);
const elapsedMs = Math.max(0, Date.now() - startedAt);
if (elapsedMs < minIntervalMs) {
await sleepWithAbort(minIntervalMs - elapsedMs, stopController.signal, {
ref: false,
});
}
}
} catch (err) {
if (stopped) {
break;
}
consecutiveEmptyPolls = 0;
failures += 1;
postPollError(port, err);
// 409 must propagate to the parent: it owns duplicate-poller/webhook
@@ -295,10 +311,18 @@ export async function runTelegramIngressWorkerRuntime(params: {
if (!isRetryableTelegramApiError(err, { context: "polling" })) {
throw err;
}
await sleep(
readTelegramRetryAfterMs(err) ?? resolveBackoff(failures),
stopController.signal,
);
try {
await sleepWithAbort(
readTelegramRetryAfterMs(err) ??
computeBackoff(TELEGRAM_RETRY_BACKOFF_POLICY, failures),
stopController.signal,
{ ref: false },
);
} catch (sleepErr) {
if (!stopped) {
throw sleepErr;
}
}
}
}
} finally {