Merge pull request #134 from souloss/feat/proxy-retry-stream-idle

fix(proxy-retry): streamIdleTimeoutMs hang guard (retry modes only) + reportSwallowed + stats cleanup

Author: weiesky

server/i18n.js

@@ -2975,11 +2975,6 @@ const i18nData = {
     "tr": "IM botu '{id}' zaten çalışıyor (pid {pid}); ikinci bir örnek başlatılması reddedildi.",
     "uk": "IM-бот '{id}' вже запущено (pid {pid}); запуск другого екземпляра відхилено."
   },
-  "server.proxyStats.starting": { "zh": "代理重试统计已启动", "en": "Proxy retry stats started", "zh-TW": "代理重試統計已啟動", "ko": "프록시 재시도 통계가 시작됨", "ja": "プロキシリトライ統計を開始しました", "de": "Proxy-Wiederholungsstatistiken gestartet", "es": "Estadísticas de reintentos de proxy iniciadas", "fr": "Statistiques de relances du proxy démarrées", "it": "Statistiche retry del proxy avviate", "da": "Proxy-genforsøgsstatistik startet", "pl": "Statystyki ponownych prób proxy uruchomione", "ru": "Статистика повторов прокси запущена", "ar": "بدأت إحصائيات إعادة محاولات الوكيل", "no": "Proxy-retry-statistikk startet", "pt-BR": "Estatísticas de retry do proxy iniciadas", "th": "เริ่มสถิติการลองใหม่ของพร็อกซีแล้ว", "tr": "Proxy yeniden deneme istatistikleri başladı", "uk": "Статистику повторів проксі запущено" },
-  "server.proxyStats.modeLabel": { "zh": "重试模式", "en": "Retry mode", "zh-TW": "重試模式", "ko": "재시도 모드", "ja": "リトライモード", "de": "Wiederholungsmodus", "es": "Modo de reintento", "fr": "Mode de relance", "it": "Modalità retry", "da": "Genforsøgstilstand", "pl": "Tryb ponownych prób", "ru": "Режим повторов", "ar": "وضع إعادة المحاولة", "no": "Retry-modus", "pt-BR": "Modo de retry", "th": "โหมดการลองใหม่", "tr": "Yeniden deneme modu", "uk": "Режим повторів" },
-  "server.proxyStats.disabled": { "zh": "代理重试已禁用", "en": "Proxy retry disabled", "zh-TW": "代理重試已停用", "ko": "프록시 재시도 비활성화됨", "ja": "プロキシリトライは無効です", "de": "Proxy-Wiederholung deaktiviert", "es": "Reintento de proxy deshabilitado", "fr": "Relance du proxy désactivée", "it": "Retry del proxy disattivato", "da": "Proxy-genforsøg deaktiveret", "pl": "Ponowne próby proxy wyłączone", "ru": "Повторы прокси отключены", "ar": "تم تعطيل إعادة محاولات الوكيل", "no": "Proxy-retry deaktivert", "pt-BR": "Retry do proxy desativado", "th": "ปิดการลองใหม่ของพร็อกซีแล้ว", "tr": "Proxy yeniden deneme devre dışı", "uk": "Повтори проксі вимкнено" },
-  "server.proxyStats.configLoaded": { "zh": "代理重试配置已加载", "en": "Proxy retry config loaded", "zh-TW": "代理重試設定已載入", "ko": "프록시 재시도 설정이 로드됨", "ja": "プロキシリトライ設定を読み込みました", "de": "Proxy-Wiederholungskonfiguration geladen", "es": "Configuración de reintento de proxy cargada", "fr": "Configuration de relance du proxy chargée", "it": "Configurazione retry del proxy caricata", "da": "Proxy-genforsøgskonfiguration indlæst", "pl": "Konfiguracja ponownych prób proxy załadowana", "ru": "Конфигурация повторов прокси загружена", "ar": "تم تحميل تكوين إعادة محاولات الوكيل", "no": "Proxy-retry-konfigurasjon lastet", "pt-BR": "Configuração de retry do proxy carregada", "th": "โหลดการตั้งค่าการลองใหม่ของพร็อกซีแล้ว", "tr": "Proxy yeniden deneme yapılandırması yüklendi", "uk": "Конфігурацію повторів проксі завантажено" },
-  "server.proxyStats.refreshed": { "zh": "代理统计已刷新", "en": "Proxy stats refreshed", "zh-TW": "代理統計已重新整理", "ko": "프록시 통계가 새로고침됨", "ja": "プロキシ統計を更新しました", "de": "Proxy-Statistiken aktualisiert", "es": "Estadísticas de proxy actualizadas", "fr": "Statistiques du proxy actualisées", "it": "Statistiche del proxy aggiornate", "da": "Proxy-statistik opdateret", "pl": "Statystyki proxy odświeżone", "ru": "Статистика прокси обновлена", "ar": "تم تحديث إحصائيات الوكيل", "no": "Proxy-statistikk oppdatert", "pt-BR": "Estatísticas do proxy atualizadas", "th": "รีเฟรชสถิติพร็อกซีแล้ว", "tr": "Proxy istatistikleri yenilendi", "uk": "Статистику проксі оновлено" }
 };
 
 // 将 { key: { lang: text } } 转换为 { lang: { key: text } }

server/lib/proxy-retry.js

@@ -17,6 +17,7 @@
 //   - race/stagger use AbortController; cancelled requests must be released correctly.
 import { resolveProfileModel } from './interceptor-core.js';
 import { readFileSync, existsSync } from 'node:fs';
+import { reportSwallowed } from './error-report.js';
 
 // ── Configuration ─────────────────────────────────────────────────
 
@@ -162,7 +163,11 @@ export function resolveRetryConfig(env = process.env, options = {}) {
         const fileRaw = JSON.parse(readFileSync(_retryConfigPath, 'utf-8'));
         Object.assign(cfg, validateRetryConfig(fileRaw));
       }
-    } catch { /* file missing/corrupt → use env only, don't block */ }
+    } catch (err) {
+      // retry-config.json written by the UI is corrupt/unreadable → fall back to env.
+      // Not fatal, but a silent swallow would hide a config the user believes is active.
+      reportSwallowed('proxyRetry.load-config-file', err);
+    }
   }
 
   return cfg;
@@ -219,7 +224,10 @@ export function isStreamResponse(response) {
   try {
     const ct = response?.headers?.get?.('content-type') || '';
     return typeof ct === 'string' && ct.toLowerCase().includes('text/event-stream');
-  } catch {
+  } catch (err) {
+    // A misbehaving header object would make us misclassify the response, which
+    // changes retry behavior (streaming 200 is never retried). Surface it.
+    reportSwallowed('proxyRetry.is-stream-response', err);
     return false;
   }
 }
@@ -233,7 +241,10 @@ export function extractModel(body) {
     const s = typeof body === 'string' ? body : body.toString('utf-8');
     const obj = JSON.parse(s);
     return typeof obj.model === 'string' ? obj.model : '';
-  } catch {
+  } catch (err) {
+    // Unparseable body → stats detail record loses the model field. Surface it
+    // so a regression in request-body handling isn't hidden behind empty models.
+    reportSwallowed('proxyRetry.extract-model', err);
     return '';
   }
 }
@@ -253,7 +264,11 @@ export function applyModelReplacement(body, profile) {
     if (!target) return body;
     obj.model = target;
     return JSON.stringify(obj);
-  } catch {
+  } catch (err) {
+    // JSON.parse failure means model replacement silently no-ops; the request
+    // still goes out with the original (un-replaced) model. Surface it so the
+    // mismatch between configured replacement and actual body isn't silent.
+    reportSwallowed('proxyRetry.apply-model-replacement', err);
     return body;
   }
 }
@@ -299,13 +314,67 @@ function computeWaitMs(status, retryAfterHeader, cfg) {
 
 // ── Single fetch wrapper ─────────────────────────────────────────
 
+/**
+ * Attaches a streaming-idle watchdog to a streaming response body.
+ *
+ * Why: connectTimeoutMs only bounds time-to-HEADERS; once headers arrive the
+ * connect timer is cleared and the retry loop breaks (streaming 200 is never
+ * retried — retry-before-first-byte strategy). If the upstream then stalls
+ * (200 headers but no body chunk ever arrives — a hung upstream), the piped
+ * response would hang indefinitely, pinning the client socket and the upstream
+ * socket until the client gives up. streamIdleTimeoutMs bounds the max gap
+ * between two chunks; exceeding it errors the body so proxy.js's pipeline
+ * surfaces the stall instead of hanging.
+ *
+ * Implemented as a TransformStream pass-through so response.body stays a valid
+ * ReadableStream (Readable.fromWeb in proxy.js keeps working): each enqueued
+ * chunk resets the timer; a stalled stream fires the timer, which calls
+ * controller.error(), aborting the fetch's underlying body and breaking the
+ * pipeline.
+ *
+ * @param {ReadableStream} body original streaming body
+ * @param {number} idleMs max gap between chunks (0 = disabled)
+ * @param {AbortSignal} signal external signal (race/stagger loser cancel + client disconnect)
+ * @returns {ReadableStream} watched body (same chunks, bounded idle)
+ */
+function applyStreamIdleWatchdog(body, idleMs, signal) {
+  if (!body || typeof body?.pipeThrough !== 'function') return body;
+  if (!idleMs || idleMs <= 0) return body;
+  let timer = null;
+  let aborted = false;
+  const arm = () => {
+    if (timer) clearTimeout(timer);
+    timer = setTimeout(() => {
+      aborted = true;
+      controller.error(new Error(`proxy stream idle timeout (${idleMs}ms)`));
+    }, idleMs);
+  };
+  let controller;
+  const transform = new TransformStream({
+    start(ctl) { controller = ctl; arm(); if (signal) signal.addEventListener('abort', disarm, { once: true }); },
+    transform(chunk, ctl) {
+      if (aborted) return; // already errored — drop late chunks
+      ctl.enqueue(chunk);
+      arm(); // reset on each chunk
+    },
+    flush() { disarm(); },
+    cancel() { disarm(); },
+  });
+  function disarm() { if (timer) { clearTimeout(timer); timer = null; } }
+  // teeThrough keeps our transform in the path; pipeThrough returns the readable end.
+  // Only pass signal when present — pipeThrough rejects a null/undefined signal.
+  return signal
+    ? body.pipeThrough(transform, { signal })
+    : body.pipeThrough(transform);
+}
+
 /**
  * Executes a single fetch request with the x-cc-viewer-trace header + network proxy dispatcher.
  * Returns the raw Response. Does not throw (on network errors returns { __networkError: true, status: 0 }).
  *
  * @param {string} url full URL
  * @param {object} fetchOptions method/headers/body
- * @param {object} ctx { dispatcher, connectTimeoutMs, signal }
+ * @param {object} ctx { dispatcher, connectTimeoutMs, streamIdleTimeoutMs, signal }
  */
 async function singleFetch(url, fetchOptions, ctx) {
   const opts = {
@@ -342,10 +411,25 @@ async function singleFetch(url, fetchOptions, ctx) {
 
   try {
     const response = await fetch(url, opts);
+    // Streaming responses: attach the idle watchdog so a hung body (headers in,
+    // no chunks) breaks within streamIdleTimeoutMs instead of pinning sockets.
+    // connectTimeoutMs already cleared below can't help — it only bound headers.
+    if (response?.body && ctx.streamIdleTimeoutMs > 0 && isStreamResponse(response)) {
+      const watched = applyStreamIdleWatchdog(response.body, ctx.streamIdleTimeoutMs, ctx.signal);
+      return new Response(watched, {
+        status: response.status,
+        statusText: response.statusText,
+        headers: response.headers,
+      });
+    }
     return response;
   } catch (err) {
     // Network error/timeout/cancellation → return a pseudo response; status=0 indicates an error
     const aborted = ctx.signal?.aborted || timeoutCtl?.signal.aborted;
+    // Aborts are expected (race loser cancellation, client disconnect, connect
+    // timeout) — not diagnostic. Only surface genuine network errors so a
+    // failing upstream isn't hidden behind status=0 pseudo-responses.
+    if (err && !aborted) reportSwallowed('proxyRetry.single-fetch', err);
     return {
       __networkError: true,
       __aborted: !!aborted,
@@ -416,7 +500,17 @@ export async function executeRequest({ url, fetchOptions, retryConfig, ctx }) {
   // headers well past 10s — so with retry disabled we must not introduce a new
   // failure mode. The timeout applies only when a retry mode is active.
   const effectiveConnectTimeoutMs = cfg.mode === 'off' ? 0 : cfg.connectTimeoutMs;
-  const commonCtx = { dispatcher, connectTimeoutMs: effectiveConnectTimeoutMs };
+  // streamIdleTimeoutMs is gated to retry modes only (NOT off), mirroring the
+  // connectTimeoutMs off-exclusion above. The watchdog wraps response.body in a
+  // TransformStream, but the interceptor (server/interceptor.js) already
+  // reconstructs response.body via getReader() + a new ReadableStream for
+  // logging/live-streaming; under concurrent load the watchdog's pipeThrough
+  // races that reconstruction and surfaces as a spurious `fetch failed` →
+  // status 0 → 502. off mode is the legacy pass-through path (no retry), so the
+  // watchdog's value (bound idle on a hung stream) is marginal here and the
+  // interceptor already observes the stream — serial/race/stagger keep the guard.
+  const effectiveStreamIdleMs = cfg.mode === 'off' ? 0 : (cfg.streamIdleTimeoutMs > 0 ? cfg.streamIdleTimeoutMs : 0);
+  const commonCtx = { dispatcher, connectTimeoutMs: effectiveConnectTimeoutMs, streamIdleTimeoutMs: effectiveStreamIdleMs };
 
   if (cfg.mode === 'off' || cfg.mode === 'serial') {
     // off / serial: serial retry. off = no retry (break on any status); serial = controlled by maxRetries (0=infinite, capped by deadline)

server/lib/stats-worker.js

@@ -177,11 +177,16 @@ function generateProjectStats(projectDir, projectName, onlyFile) {
   // Proxy retry shards (proxy_YYYY-MM-DD.jsonl) live at the project top level
   // and can exist without any v2 session (proxy-only usage) — only bail out
   // when BOTH are absent so aggregateProxyStats still runs for proxy-only dirs.
-  let proxyFiles = [];
+  // Existence-only check: the full sorted file list is read once inside
+  // aggregateProxyStats (called below), so here we just need to know whether
+  // ANY proxy shard exists — short-circuit avoids a redundant full readdir+filter.
+  let hasProxyFiles = false;
   try {
-    proxyFiles = readdirSync(projectDir).filter(f => f.startsWith('proxy_') && f.endsWith('.jsonl'));
+    for (const f of readdirSync(projectDir)) {
+      if (f.startsWith('proxy_') && f.endsWith('.jsonl')) { hasProxyFiles = true; break; }
+    }
   } catch { /* unreadable project dir → nothing to aggregate from it either */ }
-  if (sessionIds.length === 0 && proxyFiles.length === 0) return;
+  if (sessionIds.length === 0 && !hasProxyFiles) return;
 
   const filesStats = {};
   const topModels = {};
@@ -249,7 +254,7 @@ function generateProjectStats(projectDir, projectName, onlyFile) {
 
   // No parsable session yet (dirs without journals) — keep whatever exists,
   // unless proxy shards are present (they alone justify a stats write).
-  if (Object.keys(filesStats).length === 0 && proxyFiles.length === 0) return;
+  if (Object.keys(filesStats).length === 0 && !hasProxyFiles) return;
 
   // 计算全局汇总
   let totalRequests = 0;

test/proxy-retry-stream-idle.test.js

@@ -0,0 +1,111 @@
+/**
+ * server/lib/proxy-retry.js — streaming idle timeout (streamIdleTimeoutMs).
+ *
+ * Covers the hang point: when an upstream returns 200 + text/event-stream headers
+ * but the body never produces a data chunk (hung upstream), the connect timeout
+ * (cleared once headers arrive) cannot rescue the in-flight pipeline. The
+ * streamIdleTimeoutMs watchdog must abort such a stalled stream so the client
+ * socket + upstream socket are not pinned indefinitely.
+ *
+ * Mock strategy: globalThis.fetch returns a Response-like object whose body is a
+ * real ReadableStream we control. We either stall it (no chunk) to trigger the
+ * watchdog, or emit chunks on schedule to verify the watchdog resets per chunk
+ * and does NOT fire on a healthy stream.
+ */
+import { describe, it, beforeEach, afterEach } from 'node:test';
+import assert from 'node:assert/strict';
+import { executeRequest, DEFAULT_RETRY_CONFIG } from '../server/lib/proxy-retry.js';
+
+/**
+ * Build a fake streaming Response whose body is a real ReadableStream.
+ * @param {(controller: ReadableStreamDefaultController) => void} emit controls chunk emission.
+ */
+function makeStreamResponse({ status = 200, contentType = 'text/event-stream', emit }) {
+  const stream = new ReadableStream({
+    start(controller) {
+      if (emit) emit(controller);
+    },
+  });
+  return new Response(stream, {
+    status,
+    statusText: status === 200 ? 'OK' : 'Error',
+    headers: { 'content-type': contentType },
+  });
+}
+
+let _originalFetch;
+beforeEach(() => { _originalFetch = globalThis.fetch; });
+afterEach(() => { globalThis.fetch = _originalFetch; });
+
+describe('proxy-retry streamIdleTimeoutMs hang guard', () => {
+  it('hung stream (200 + no chunk) → watchdog aborts, response surfaces as stalled', async () => {
+    // fetch resolves with 200 headers; body never emits. Without the watchdog
+    // the response would sit in pipeline forever. The watchdog must abort it.
+    globalThis.fetch = async () => makeStreamResponse({
+      emit: () => { /* never enqueue — hung upstream */ },
+    });
+
+    const r = await executeRequest({
+      url: 'https://example.com/v1/messages',
+      fetchOptions: { method: 'POST', headers: {} },
+      retryConfig: { ...DEFAULT_RETRY_CONFIG, mode: 'serial', streamIdleTimeoutMs: 50, maxRetries: 5 },
+      ctx: {},
+    });
+
+    assert.equal(r.finalStatus, 200, 'headers already arrived → status stays 200');
+    assert.equal(r.attempts, 1, 'streaming 200 is never retried');
+
+    // The body must error within a bounded time (the watchdog), so proxy.js's
+    // pipeline surfaces the stall instead of hanging forever. We read the body
+    // via getReader: a healthy open stream would never reject; an aborted
+    // (watchdog-fired) stream rejects with the idle-timeout error.
+    const reader = r.response.body.getReader();
+    let errored = false;
+    let errName = '';
+    const settled = await Promise.race([
+      reader.read().then(
+        () => reader.read(), // a chunk arrived — keep draining to catch the eventual error
+      ).catch((e) => { errored = true; errName = e?.message || String(e); }),
+      new Promise((resolve) => setTimeout(() => resolve('timeout'), 500)),
+    ]);
+    if (errored) {
+      assert.match(errName, /idle timeout/i, `body should error with idle-timeout, got: ${errName}`);
+    } else {
+      // If no error within 500ms (well past the 50ms budget), the watchdog did NOT fire → hang regression.
+      assert.fail('hung stream was not aborted by streamIdleTimeoutMs within 500ms (hang regression)');
+    }
+  });
+
+  it('healthy stream (chunks arriving) → watchdog resets, does NOT abort', async () => {
+    // Keep the total stream open longer than the idle budget while every gap
+    // stays below it. A watchdog that fails to reset will abort this stream.
+    globalThis.fetch = async () => {
+      return makeStreamResponse({
+        emit: (controller) => {
+          let chunks = 0;
+          const iv = setInterval(() => {
+            controller.enqueue(new TextEncoder().encode('data: hi\n\n'));
+            chunks++;
+            if (chunks === 8) {
+              clearInterval(iv);
+              controller.close();
+            }
+          }, 10);
+        },
+      });
+    };
+
+    const r = await executeRequest({
+      url: 'https://example.com/v1/messages',
+      fetchOptions: { method: 'POST', headers: {} },
+      retryConfig: { ...DEFAULT_RETRY_CONFIG, mode: 'serial', streamIdleTimeoutMs: 40, maxRetries: 5 },
+      ctx: {},
+    });
+
+    assert.equal(r.finalStatus, 200);
+    assert.equal(r.attempts, 1);
+    assert.equal(r.response.headers.get('content-type'), 'text/event-stream');
+    const body = await r.response.text();
+    assert.equal(body, 'data: hi\n\n'.repeat(8));
+  });
+});