@@ -17,6 +17,7 @@ import {
1717isRecoverableTelegramNetworkError,
1818isTelegramPollingNetworkError,
1919} from "./network-errors.js";
20+import { acquireTelegramPollingLease } from "./polling-lease.js";
2021import { makeProxyFetch } from "./proxy.js";
21222223export type { MonitorTelegramOpts } from "./monitor.types.js";
@@ -161,76 +162,96 @@ export async function monitorTelegramProvider(opts: MonitorTelegramOpts = {}) {
161162const { TelegramPollingSession, readTelegramUpdateOffset, writeTelegramUpdateOffset } =
162163await loadTelegramMonitorPollingRuntime();
163164164-if (isTelegramExecApprovalHandlerConfigured({ cfg, accountId: account.accountId })) {
165-registerChannelRuntimeContext({
166-channelRuntime: opts.channelRuntime,
167-channelId: "telegram",
168-accountId: account.accountId,
169-capability: CHANNEL_APPROVAL_NATIVE_RUNTIME_CONTEXT_CAPABILITY,
170-context: { token },
171-abortSignal: opts.abortSignal,
172-});
173-}
174-175-const persistedOffsetRaw = await readTelegramUpdateOffset({
165+const pollingLease = await acquireTelegramPollingLease({
166+ token,
176167accountId: account.accountId,
177-botToken: token,
168+abortSignal: opts.abortSignal,
178169});
179-let lastUpdateId = normalizePersistedUpdateId(persistedOffsetRaw);
180-if (persistedOffsetRaw !== null && lastUpdateId === null) {
170+if (pollingLease.waitedForPrevious) {
171+log(
172+`[telegram][diag] waited for previous polling session for bot token ${pollingLease.tokenFingerprint} before starting account "${account.accountId}".`,
173+);
174+}
175+if (pollingLease.replacedStoppingPrevious) {
181176log(
182-`[telegram] Ignoring invalid persisted update offset (${String(persistedOffsetRaw)}); starting without offset confirmation.`,
177+`[telegram][diag] previous polling session for bot token ${pollingLease.tokenFingerprint} did not stop within the lease wait; starting a replacement for account "${account.accountId}".`,
183178);
184179}
185180186-const persistUpdateId = async (updateId: number) => {
187-const normalizedUpdateId = normalizePersistedUpdateId(updateId);
188-if (normalizedUpdateId === null) {
189-log(`[telegram] Ignoring invalid update_id value: ${String(updateId)}`);
190-return;
191-}
192-if (lastUpdateId !== null && normalizedUpdateId <= lastUpdateId) {
193-return;
194-}
195-lastUpdateId = normalizedUpdateId;
196-try {
197-await writeTelegramUpdateOffset({
181+try {
182+if (isTelegramExecApprovalHandlerConfigured({ cfg, accountId: account.accountId })) {
183+registerChannelRuntimeContext({
184+channelRuntime: opts.channelRuntime,
185+channelId: "telegram",
198186accountId: account.accountId,
199-updateId: normalizedUpdateId,
200-botToken: token,
187+capability: CHANNEL_APPROVAL_NATIVE_RUNTIME_CONTEXT_CAPABILITY,
188+context: { token },
189+abortSignal: opts.abortSignal,
201190});
202-} catch (err) {
203-(opts.runtime?.error ?? console.error)(
204-`telegram: failed to persist update offset: ${String(err)}`,
205-);
206191}
207-};
208192209-// Preserve sticky IPv4 fallback state across clean/conflict restarts.
210-// Dirty polling cycles rebuild transport inside TelegramPollingSession.
211-const createTelegramTransportForPolling = () =>
212-resolveTelegramTransport(proxyFetch, {
213-network: account.config.network,
193+const persistedOffsetRaw = await readTelegramUpdateOffset({
194+accountId: account.accountId,
195+botToken: token,
214196});
215-const telegramTransport = createTelegramTransportForPolling();
197+let lastUpdateId = normalizePersistedUpdateId(persistedOffsetRaw);
198+if (persistedOffsetRaw !== null && lastUpdateId === null) {
199+log(
200+`[telegram] Ignoring invalid persisted update offset (${String(persistedOffsetRaw)}); starting without offset confirmation.`,
201+);
202+}
216203217-pollingSession = new TelegramPollingSession({
218- token,
219-config: cfg,
220-accountId: account.accountId,
221-runtime: opts.runtime,
222- proxyFetch,
223-abortSignal: opts.abortSignal,
224-runnerOptions: createTelegramRunnerOptions(cfg),
225-getLastUpdateId: () => lastUpdateId,
226- persistUpdateId,
227- log,
228- telegramTransport,
229-createTelegramTransport: createTelegramTransportForPolling,
230-stallThresholdMs: account.config.pollingStallThresholdMs,
231-setStatus: opts.setStatus,
232-});
233-await pollingSession.runUntilAbort();
204+const persistUpdateId = async (updateId: number) => {
205+const normalizedUpdateId = normalizePersistedUpdateId(updateId);
206+if (normalizedUpdateId === null) {
207+log(`[telegram] Ignoring invalid update_id value: ${String(updateId)}`);
208+return;
209+}
210+if (lastUpdateId !== null && normalizedUpdateId <= lastUpdateId) {
211+return;
212+}
213+lastUpdateId = normalizedUpdateId;
214+try {
215+await writeTelegramUpdateOffset({
216+accountId: account.accountId,
217+updateId: normalizedUpdateId,
218+botToken: token,
219+});
220+} catch (err) {
221+(opts.runtime?.error ?? console.error)(
222+`telegram: failed to persist update offset: ${String(err)}`,
223+);
224+}
225+};
226+227+// Preserve sticky IPv4 fallback state across clean/conflict restarts.
228+// Dirty polling cycles rebuild transport inside TelegramPollingSession.
229+const createTelegramTransportForPolling = () =>
230+resolveTelegramTransport(proxyFetch, {
231+network: account.config.network,
232+});
233+const telegramTransport = createTelegramTransportForPolling();
234+235+pollingSession = new TelegramPollingSession({
236+ token,
237+config: cfg,
238+accountId: account.accountId,
239+runtime: opts.runtime,
240+ proxyFetch,
241+abortSignal: opts.abortSignal,
242+runnerOptions: createTelegramRunnerOptions(cfg),
243+getLastUpdateId: () => lastUpdateId,
244+ persistUpdateId,
245+ log,
246+ telegramTransport,
247+createTelegramTransport: createTelegramTransportForPolling,
248+stallThresholdMs: account.config.pollingStallThresholdMs,
249+setStatus: opts.setStatus,
250+});
251+await pollingSession.runUntilAbort();
252+} finally {
253+pollingLease.release();
254+}
234255} finally {
235256unregisterHandler();
236257}
此内容由惯性聚合(RSS阅读器)自动聚合整理，仅供阅读参考。 原文来自 — 版权归原作者所有。