@@ -42,6 +42,9 @@ const TELEGRAM_DISPATCHER_KEEP_ALIVE_MAX_TIMEOUT_MS = 600_000;
4242const TELEGRAM_DISPATCHER_CONNECTIONS_PER_ORIGIN = 10;
4343const TELEGRAM_DISPATCHER_PIPELINING = 1;
4444const TELEGRAM_STICKY_FALLBACK_PRIMARY_PROBE_SUCCESS_THRESHOLD = 5;
45+const TELEGRAM_TRANSPORT_ATTEMPT_FAILURE_THRESHOLD = 5;
46+const TELEGRAM_TRANSPORT_ATTEMPT_INITIAL_COOLDOWN_MS = 10_000;
47+const TELEGRAM_TRANSPORT_ATTEMPT_MAX_COOLDOWN_MS = 60_000;
45484649type TelegramAgentPoolOptions = {
4750allowH2: false;
@@ -80,6 +83,12 @@ type TelegramTransportAttempt = {
8083logMessage?: string;
8184};
828586+type TelegramTransportAttemptHealth = {
87+consecutiveFailures: number;
88+cooldownMs: number;
89+unhealthyUntilMs: number;
90+};
91+8392type TelegramDnsResultOrder = "ipv4first" | "verbatim";
84938594type LookupCallback =
@@ -110,22 +119,6 @@ type TelegramTransportFallbackContext = {
110119codes: Set<string>;
111120};
112121113-type TelegramTransportFallbackRule = {
114-name: string;
115-matches: (ctx: TelegramTransportFallbackContext) => boolean;
116-};
117-118-const TELEGRAM_TRANSPORT_FALLBACK_RULES: readonly TelegramTransportFallbackRule[] = [
119-{
120-name: "fetch-failed-envelope",
121-matches: ({ message }) => message.includes("fetch failed"),
122-},
123-{
124-name: "known-network-code",
125-matches: ({ codes }) => FALLBACK_RETRY_ERROR_CODES.some((code) => codes.has(code)),
126-},
127-];
128-129122function normalizeDnsResultOrder(value: string | null): TelegramDnsResultOrder | null {
130123if (value === "ipv4first" || value === "verbatim") {
131124return value;
@@ -446,20 +439,28 @@ function formatErrorCodes(err: unknown): string {
446439return codes.length > 0 ? codes.join(",") : "none";
447440}
448441442+class TelegramTransportAttemptUnhealthyError extends Error {
443+constructor(unhealthyUntilMs: number) {
444+const remainingMs = Math.max(0, unhealthyUntilMs - Date.now());
445+super(`telegram transport attempt temporarily unhealthy; retry after ${remainingMs}ms`);
446+this.name = "TelegramTransportAttemptUnhealthyError";
447+}
448+}
449+449450function shouldUseTelegramTransportFallback(err: unknown): boolean {
451+if (err instanceof TelegramTransportAttemptUnhealthyError) {
452+return true;
453+}
450454const ctx: TelegramTransportFallbackContext = {
451455message:
452456err && typeof err === "object" && "message" in err
453457 ? normalizeLowercaseStringOrEmpty(String(err.message))
454458 : "",
455459codes: collectErrorCodes(err),
456460};
457-for (const rule of TELEGRAM_TRANSPORT_FALLBACK_RULES) {
458-if (!rule.matches(ctx)) {
459-return false;
460-}
461-}
462-return true;
461+const hasFetchFailedEnvelope = ctx.message.includes("fetch failed");
462+const hasKnownNetworkCode = FALLBACK_RETRY_ERROR_CODES.some((code) => ctx.codes.has(code));
463+return hasKnownNetworkCode || (hasFetchFailedEnvelope && ctx.codes.size === 0);
463464}
464465465466export function shouldRetryTelegramTransportFallback(err: unknown): boolean {
@@ -643,12 +644,46 @@ export function resolveTelegramTransport(
643644let stickyAttemptIndex = 0;
644645let stickySuccessCount = 0;
645646let primaryProbeDue = false;
647+const attemptHealth = transportAttempts.map<TelegramTransportAttemptHealth>(() => ({
648+consecutiveFailures: 0,
649+cooldownMs: TELEGRAM_TRANSPORT_ATTEMPT_INITIAL_COOLDOWN_MS,
650+unhealthyUntilMs: 0,
651+}));
646652647653const resetStickyRecoveryProbe = (): void => {
648654stickySuccessCount = 0;
649655primaryProbeDue = false;
650656};
651657658+const getAttemptCooldownError = (attemptIndex: number): Error | null => {
659+const health = attemptHealth[attemptIndex];
660+if (health.unhealthyUntilMs <= Date.now()) {
661+return null;
662+}
663+return new TelegramTransportAttemptUnhealthyError(health.unhealthyUntilMs);
664+};
665+666+const recordAttemptFailure = (attemptIndex: number, err: unknown): void => {
667+if (!shouldUseTelegramTransportFallback(err)) {
668+return;
669+}
670+const health = attemptHealth[attemptIndex];
671+health.consecutiveFailures += 1;
672+if (health.consecutiveFailures < TELEGRAM_TRANSPORT_ATTEMPT_FAILURE_THRESHOLD) {
673+return;
674+}
675+const cooldownMs = Math.min(
676+TELEGRAM_TRANSPORT_ATTEMPT_MAX_COOLDOWN_MS,
677+Math.max(TELEGRAM_TRANSPORT_ATTEMPT_INITIAL_COOLDOWN_MS, health.cooldownMs),
678+);
679+health.consecutiveFailures = 0;
680+health.cooldownMs = Math.min(TELEGRAM_TRANSPORT_ATTEMPT_MAX_COOLDOWN_MS, cooldownMs * 2);
681+health.unhealthyUntilMs = Date.now() + cooldownMs;
682+log.warn(
683+`telegram transport attempt marked temporarily unhealthy for ${cooldownMs}ms (codes=${formatErrorCodes(err)})`,
684+);
685+};
686+652687const promoteStickyAttempt = (nextIndex: number, err: unknown, reason?: string): boolean => {
653688if (nextIndex <= stickyAttemptIndex || nextIndex >= transportAttempts.length) {
654689return false;
@@ -669,6 +704,11 @@ export function resolveTelegramTransport(
669704};
670705671706const recordSuccessfulAttempt = (attemptIndex: number): void => {
707+const health = attemptHealth[attemptIndex];
708+health.consecutiveFailures = 0;
709+health.cooldownMs = TELEGRAM_TRANSPORT_ATTEMPT_INITIAL_COOLDOWN_MS;
710+health.unhealthyUntilMs = 0;
711+672712if (stickyAttemptIndex === 0) {
673713resetStickyRecoveryProbe();
674714return;
@@ -700,50 +740,63 @@ export function resolveTelegramTransport(
700740(init as RequestInitWithDispatcher | undefined)?.dispatcher,
701741);
702742const stickyStartIndex = Math.min(stickyAttemptIndex, transportAttempts.length - 1);
703-const primaryProbe = !callerProvidedDispatcher && primaryProbeDue && stickyStartIndex > 0;
743+const stickyCooldownError = callerProvidedDispatcher
744+ ? null
745+ : getAttemptCooldownError(stickyStartIndex);
746+const primaryProbe =
747+!callerProvidedDispatcher &&
748+stickyStartIndex > 0 &&
749+(primaryProbeDue || stickyCooldownError !== null);
704750const startIndex = primaryProbe ? 0 : stickyStartIndex;
705751if (primaryProbe) {
706752primaryProbeDue = false;
707-log.debug("fetch fallback: re-probing primary dispatcher after sticky fallback successes");
708-}
709-let err: unknown;
710-711-try {
712-const response = await sourceFetch(
713-input,
714-withDispatcherIfMissing(init, transportAttempts[startIndex].createDispatcher()),
753+log.debug(
754+stickyCooldownError
755+ ? "fetch fallback: re-probing primary dispatcher while sticky fallback is cooling down"
756+ : "fetch fallback: re-probing primary dispatcher after sticky fallback successes",
715757);
716-captureHttpExchange({
717-url: resolveRequestUrl(input),
718-method: init?.method ?? "GET",
719-requestHeaders: init?.headers as Headers | Record<string, string> | undefined,
720-requestBody: (init as RequestInit & { body?: BodyInit | null })?.body ?? null,
721- response,
722-flowId: randomUUID(),
723-meta: { subsystem: "telegram-fetch" },
724-});
725-if (!callerProvidedDispatcher) {
726-recordSuccessfulAttempt(startIndex);
727-}
728-return response;
729-} catch (caught) {
730-err = caught;
731758}
759+let err: unknown;
732760733-if (!shouldUseTelegramTransportFallback(err)) {
734-throw err;
735-}
736761if (callerProvidedDispatcher) {
737-return sourceFetch(input, init ?? {});
762+try {
763+const response = await sourceFetch(input, init);
764+captureHttpExchange({
765+url: resolveRequestUrl(input),
766+method: init?.method ?? "GET",
767+requestHeaders: init?.headers as Headers | Record<string, string> | undefined,
768+requestBody: (init as RequestInit & { body?: BodyInit | null })?.body ?? null,
769+ response,
770+flowId: randomUUID(),
771+meta: { subsystem: "telegram-fetch" },
772+});
773+return response;
774+} catch (caught) {
775+if (!shouldUseTelegramTransportFallback(caught)) {
776+throw caught;
777+}
778+return sourceFetch(input, init ?? {});
779+}
738780}
739781740-for (let nextIndex = startIndex + 1; nextIndex < transportAttempts.length; nextIndex += 1) {
741-const nextAttempt = transportAttempts[nextIndex];
742-promoteStickyAttempt(nextIndex, err);
782+for (
783+let attemptIndex = startIndex;
784+attemptIndex < transportAttempts.length;
785+attemptIndex += 1
786+) {
787+const attempt = transportAttempts[attemptIndex];
788+if (attemptIndex > startIndex) {
789+promoteStickyAttempt(attemptIndex, err);
790+}
791+const cooldownError = getAttemptCooldownError(attemptIndex);
792+if (cooldownError) {
793+err = cooldownError;
794+continue;
795+}
743796try {
744797const response = await sourceFetch(
745798input,
746-withDispatcherIfMissing(init, nextAttempt.createDispatcher()),
799+withDispatcherIfMissing(init, attempt.createDispatcher()),
747800);
748801captureHttpExchange({
749802url: resolveRequestUrl(input),
@@ -752,15 +805,19 @@ export function resolveTelegramTransport(
752805requestBody: (init as RequestInit & { body?: BodyInit | null })?.body ?? null,
753806 response,
754807flowId: randomUUID(),
755-meta: { subsystem: "telegram-fetch", fallbackAttempt: nextIndex },
808+meta:
809+attemptIndex === startIndex
810+ ? { subsystem: "telegram-fetch" }
811+ : { subsystem: "telegram-fetch", fallbackAttempt: attemptIndex },
756812});
757-recordSuccessfulAttempt(nextIndex);
813+recordSuccessfulAttempt(attemptIndex);
758814return response;
759815} catch (caught) {
760816err = caught;
761817if (!shouldUseTelegramTransportFallback(err)) {
762818throw err;
763819}
820+recordAttemptFailure(attemptIndex, err);
764821}
765822}
766823此内容由惯性聚合(RSS阅读器)自动聚合整理，仅供阅读参考。 原文来自 — 版权归原作者所有。