Skip to content

Commit 31aa63e

Browse files
committed
fix(proxy): streamIdleTimeoutMs hang guard + reportSwallowed + stats cleanup
proxy-retry engine: - Implement the streamIdleTimeoutMs watchdog (previously a dead, validated-but- unconsumed field). A hung upstream that returns 200 + text/event-stream headers but never produces a body chunk would pin both the client and upstream sockets indefinitely — connectTimeoutMs is cleared once headers arrive and streaming 200 is never retried (retry-before-first-byte). applyStreamIdleWatchdog wraps the streaming body in a TransformStream pass-through: each chunk resets the idle timer; exceeding the budget errors the body so proxy.js's pipeline surfaces the stall instead of hanging. Applies in all modes (including off). - Replace 5 silent swallowed catches with reportSwallowed('proxyRetry.*', err): load-config-file, is-stream-response, extract-model, apply-model-replacement, single-fetch (non-abort network errors only, to avoid flooding on expected race-loser/client-disconnect cancellations). Per CLAUDE.md, fetch-handling catches with diagnostic value must not swallow silently. stats-worker: - Merge the double readdirSync on the project dir: the early-return gate in generateProjectStats now uses an existence-only short-circuit (hasProxyFiles) instead of a full filter; the single full sorted read remains in aggregateProxyStats. Fixes a dangling proxyFiles reference that the rename would have introduced. i18n: - Remove 5 dead server.proxyStats.* keys (starting/modeLabel/disabled/ configLoaded/refreshed) with zero consumers across server/src/test. Tests: - New test/proxy-retry-stream-idle.test.js: hung stream aborts within the idle budget; healthy stream (chunks every 5ms) is not falsely aborted.
1 parent dd2fa26 commit 31aa63e

4 files changed

Lines changed: 215 additions & 15 deletions

File tree

server/i18n.js

Lines changed: 0 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -2975,11 +2975,6 @@ const i18nData = {
29752975
"tr": "IM botu '{id}' zaten çalışıyor (pid {pid}); ikinci bir örnek başlatılması reddedildi.",
29762976
"uk": "IM-бот '{id}' вже запущено (pid {pid}); запуск другого екземпляра відхилено."
29772977
},
2978-
"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": "Статистику повторів проксі запущено" },
2979-
"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": "Режим повторів" },
2980-
"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": "Повтори проксі вимкнено" },
2981-
"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": "Конфігурацію повторів проксі завантажено" },
2982-
"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": "Статистику проксі оновлено" }
29832978
};
29842979

29852980
// 将 { key: { lang: text } } 转换为 { lang: { key: text } }

server/lib/proxy-retry.js

Lines changed: 89 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -17,6 +17,7 @@
1717
// - race/stagger use AbortController; cancelled requests must be released correctly.
1818
import { resolveProfileModel } from './interceptor-core.js';
1919
import { readFileSync, existsSync } from 'node:fs';
20+
import { reportSwallowed } from './error-report.js';
2021

2122
// ── Configuration ─────────────────────────────────────────────────
2223

@@ -162,7 +163,11 @@ export function resolveRetryConfig(env = process.env, options = {}) {
162163
const fileRaw = JSON.parse(readFileSync(_retryConfigPath, 'utf-8'));
163164
Object.assign(cfg, validateRetryConfig(fileRaw));
164165
}
165-
} catch { /* file missing/corrupt → use env only, don't block */ }
166+
} catch (err) {
167+
// retry-config.json written by the UI is corrupt/unreadable → fall back to env.
168+
// Not fatal, but a silent swallow would hide a config the user believes is active.
169+
reportSwallowed('proxyRetry.load-config-file', err);
170+
}
166171
}
167172

168173
return cfg;
@@ -219,7 +224,10 @@ export function isStreamResponse(response) {
219224
try {
220225
const ct = response?.headers?.get?.('content-type') || '';
221226
return typeof ct === 'string' && ct.toLowerCase().includes('text/event-stream');
222-
} catch {
227+
} catch (err) {
228+
// A misbehaving header object would make us misclassify the response, which
229+
// changes retry behavior (streaming 200 is never retried). Surface it.
230+
reportSwallowed('proxyRetry.is-stream-response', err);
223231
return false;
224232
}
225233
}
@@ -233,7 +241,10 @@ export function extractModel(body) {
233241
const s = typeof body === 'string' ? body : body.toString('utf-8');
234242
const obj = JSON.parse(s);
235243
return typeof obj.model === 'string' ? obj.model : '';
236-
} catch {
244+
} catch (err) {
245+
// Unparseable body → stats detail record loses the model field. Surface it
246+
// so a regression in request-body handling isn't hidden behind empty models.
247+
reportSwallowed('proxyRetry.extract-model', err);
237248
return '';
238249
}
239250
}
@@ -253,7 +264,11 @@ export function applyModelReplacement(body, profile) {
253264
if (!target) return body;
254265
obj.model = target;
255266
return JSON.stringify(obj);
256-
} catch {
267+
} catch (err) {
268+
// JSON.parse failure means model replacement silently no-ops; the request
269+
// still goes out with the original (un-replaced) model. Surface it so the
270+
// mismatch between configured replacement and actual body isn't silent.
271+
reportSwallowed('proxyRetry.apply-model-replacement', err);
257272
return body;
258273
}
259274
}
@@ -299,13 +314,67 @@ function computeWaitMs(status, retryAfterHeader, cfg) {
299314

300315
// ── Single fetch wrapper ─────────────────────────────────────────
301316

317+
/**
318+
* Attaches a streaming-idle watchdog to a streaming response body.
319+
*
320+
* Why: connectTimeoutMs only bounds time-to-HEADERS; once headers arrive the
321+
* connect timer is cleared and the retry loop breaks (streaming 200 is never
322+
* retried — retry-before-first-byte strategy). If the upstream then stalls
323+
* (200 headers but no body chunk ever arrives — a hung upstream), the piped
324+
* response would hang indefinitely, pinning the client socket and the upstream
325+
* socket until the client gives up. streamIdleTimeoutMs bounds the max gap
326+
* between two chunks; exceeding it errors the body so proxy.js's pipeline
327+
* surfaces the stall instead of hanging.
328+
*
329+
* Implemented as a TransformStream pass-through so response.body stays a valid
330+
* ReadableStream (Readable.fromWeb in proxy.js keeps working): each enqueued
331+
* chunk resets the timer; a stalled stream fires the timer, which calls
332+
* controller.error(), aborting the fetch's underlying body and breaking the
333+
* pipeline.
334+
*
335+
* @param {ReadableStream} body original streaming body
336+
* @param {number} idleMs max gap between chunks (0 = disabled)
337+
* @param {AbortSignal} signal external signal (race/stagger loser cancel + client disconnect)
338+
* @returns {ReadableStream} watched body (same chunks, bounded idle)
339+
*/
340+
function applyStreamIdleWatchdog(body, idleMs, signal) {
341+
if (!body || typeof body?.pipeThrough !== 'function') return body;
342+
if (!idleMs || idleMs <= 0) return body;
343+
let timer = null;
344+
let aborted = false;
345+
const arm = () => {
346+
if (timer) clearTimeout(timer);
347+
timer = setTimeout(() => {
348+
aborted = true;
349+
controller.error(new Error(`proxy stream idle timeout (${idleMs}ms)`));
350+
}, idleMs);
351+
};
352+
let controller;
353+
const transform = new TransformStream({
354+
start(ctl) { controller = ctl; arm(); if (signal) signal.addEventListener('abort', disarm, { once: true }); },
355+
transform(chunk, ctl) {
356+
if (aborted) return; // already errored — drop late chunks
357+
ctl.enqueue(chunk);
358+
arm(); // reset on each chunk
359+
},
360+
flush() { disarm(); },
361+
cancel() { disarm(); },
362+
});
363+
function disarm() { if (timer) { clearTimeout(timer); timer = null; } }
364+
// teeThrough keeps our transform in the path; pipeThrough returns the readable end.
365+
// Only pass signal when present — pipeThrough rejects a null/undefined signal.
366+
return signal
367+
? body.pipeThrough(transform, { signal })
368+
: body.pipeThrough(transform);
369+
}
370+
302371
/**
303372
* Executes a single fetch request with the x-cc-viewer-trace header + network proxy dispatcher.
304373
* Returns the raw Response. Does not throw (on network errors returns { __networkError: true, status: 0 }).
305374
*
306375
* @param {string} url full URL
307376
* @param {object} fetchOptions method/headers/body
308-
* @param {object} ctx { dispatcher, connectTimeoutMs, signal }
377+
* @param {object} ctx { dispatcher, connectTimeoutMs, streamIdleTimeoutMs, signal }
309378
*/
310379
async function singleFetch(url, fetchOptions, ctx) {
311380
const opts = {
@@ -342,10 +411,21 @@ async function singleFetch(url, fetchOptions, ctx) {
342411

343412
try {
344413
const response = await fetch(url, opts);
414+
// Streaming responses: attach the idle watchdog so a hung body (headers in,
415+
// no chunks) breaks within streamIdleTimeoutMs instead of pinning sockets.
416+
// connectTimeoutMs already cleared below can't help — it only bound headers.
417+
if (response?.body && ctx.streamIdleTimeoutMs > 0 && isStreamResponse(response)) {
418+
const watched = applyStreamIdleWatchdog(response.body, ctx.streamIdleTimeoutMs, ctx.signal);
419+
return { ...response, body: watched, __streamWatched: true };
420+
}
345421
return response;
346422
} catch (err) {
347423
// Network error/timeout/cancellation → return a pseudo response; status=0 indicates an error
348424
const aborted = ctx.signal?.aborted || timeoutCtl?.signal.aborted;
425+
// Aborts are expected (race loser cancellation, client disconnect, connect
426+
// timeout) — not diagnostic. Only surface genuine network errors so a
427+
// failing upstream isn't hidden behind status=0 pseudo-responses.
428+
if (err && !aborted) reportSwallowed('proxyRetry.single-fetch', err);
349429
return {
350430
__networkError: true,
351431
__aborted: !!aborted,
@@ -416,7 +496,10 @@ export async function executeRequest({ url, fetchOptions, retryConfig, ctx }) {
416496
// headers well past 10s — so with retry disabled we must not introduce a new
417497
// failure mode. The timeout applies only when a retry mode is active.
418498
const effectiveConnectTimeoutMs = cfg.mode === 'off' ? 0 : cfg.connectTimeoutMs;
419-
const commonCtx = { dispatcher, connectTimeoutMs: effectiveConnectTimeoutMs };
499+
// streamIdleTimeoutMs applies to streaming responses in ALL modes (including off):
500+
// even with retry disabled, a hung streaming body must not pin sockets forever.
501+
const effectiveStreamIdleMs = cfg.streamIdleTimeoutMs > 0 ? cfg.streamIdleTimeoutMs : 0;
502+
const commonCtx = { dispatcher, connectTimeoutMs: effectiveConnectTimeoutMs, streamIdleTimeoutMs: effectiveStreamIdleMs };
420503

421504
if (cfg.mode === 'off' || cfg.mode === 'serial') {
422505
// 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

Lines changed: 9 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -177,11 +177,16 @@ function generateProjectStats(projectDir, projectName, onlyFile) {
177177
// Proxy retry shards (proxy_YYYY-MM-DD.jsonl) live at the project top level
178178
// and can exist without any v2 session (proxy-only usage) — only bail out
179179
// when BOTH are absent so aggregateProxyStats still runs for proxy-only dirs.
180-
let proxyFiles = [];
180+
// Existence-only check: the full sorted file list is read once inside
181+
// aggregateProxyStats (called below), so here we just need to know whether
182+
// ANY proxy shard exists — short-circuit avoids a redundant full readdir+filter.
183+
let hasProxyFiles = false;
181184
try {
182-
proxyFiles = readdirSync(projectDir).filter(f => f.startsWith('proxy_') && f.endsWith('.jsonl'));
185+
for (const f of readdirSync(projectDir)) {
186+
if (f.startsWith('proxy_') && f.endsWith('.jsonl')) { hasProxyFiles = true; break; }
187+
}
183188
} catch { /* unreadable project dir → nothing to aggregate from it either */ }
184-
if (sessionIds.length === 0 && proxyFiles.length === 0) return;
189+
if (sessionIds.length === 0 && !hasProxyFiles) return;
185190

186191
const filesStats = {};
187192
const topModels = {};
@@ -249,7 +254,7 @@ function generateProjectStats(projectDir, projectName, onlyFile) {
249254

250255
// No parsable session yet (dirs without journals) — keep whatever exists,
251256
// unless proxy shards are present (they alone justify a stats write).
252-
if (Object.keys(filesStats).length === 0 && proxyFiles.length === 0) return;
257+
if (Object.keys(filesStats).length === 0 && !hasProxyFiles) return;
253258

254259
// 计算全局汇总
255260
let totalRequests = 0;
Lines changed: 117 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,117 @@
1+
/**
2+
* server/lib/proxy-retry.js — streaming idle timeout (streamIdleTimeoutMs).
3+
*
4+
* Covers the hang point: when an upstream returns 200 + text/event-stream headers
5+
* but the body never produces a data chunk (hung upstream), the connect timeout
6+
* (cleared once headers arrive) cannot rescue the in-flight pipeline. The
7+
* streamIdleTimeoutMs watchdog must abort such a stalled stream so the client
8+
* socket + upstream socket are not pinned indefinitely.
9+
*
10+
* Mock strategy: globalThis.fetch returns a Response-like object whose body is a
11+
* real ReadableStream we control. We either stall it (no chunk) to trigger the
12+
* watchdog, or emit chunks on schedule to verify the watchdog resets per chunk
13+
* and does NOT fire on a healthy stream.
14+
*/
15+
import { describe, it, beforeEach, afterEach } from 'node:test';
16+
import assert from 'node:assert/strict';
17+
import { executeRequest, DEFAULT_RETRY_CONFIG } from '../server/lib/proxy-retry.js';
18+
19+
/**
20+
* Build a fake streaming Response whose body is a real ReadableStream.
21+
* @param {(controller: ReadableStreamDefaultController) => void} emit controls chunk emission.
22+
*/
23+
function makeStreamResponse({ status = 200, contentType = 'text/event-stream', emit }) {
24+
const stream = new ReadableStream({
25+
start(controller) {
26+
if (emit) emit(controller);
27+
},
28+
});
29+
const headers = new Map();
30+
headers.set('content-type', contentType);
31+
return {
32+
status,
33+
statusText: status === 200 ? 'OK' : 'Error',
34+
ok: status > 0 && status < 400,
35+
headers: {
36+
get(k) { return headers.get(k.toLowerCase()) || null; },
37+
entries() { return headers.entries(); },
38+
},
39+
body: stream,
40+
text: async () => '',
41+
json: async () => ({}),
42+
};
43+
}
44+
45+
let _originalFetch;
46+
beforeEach(() => { _originalFetch = globalThis.fetch; });
47+
afterEach(() => { globalThis.fetch = _originalFetch; });
48+
49+
describe('proxy-retry streamIdleTimeoutMs hang guard', () => {
50+
it('hung stream (200 + no chunk) → watchdog aborts, response surfaces as stalled', async () => {
51+
// fetch resolves with 200 headers; body never emits. Without the watchdog
52+
// the response would sit in pipeline forever. The watchdog must abort it.
53+
globalThis.fetch = async () => makeStreamResponse({
54+
emit: () => { /* never enqueue — hung upstream */ },
55+
});
56+
57+
const r = await executeRequest({
58+
url: 'https://example.com/v1/messages',
59+
fetchOptions: { method: 'POST', headers: {} },
60+
retryConfig: { ...DEFAULT_RETRY_CONFIG, mode: 'serial', streamIdleTimeoutMs: 50, maxRetries: 5 },
61+
ctx: {},
62+
});
63+
64+
assert.equal(r.finalStatus, 200, 'headers already arrived → status stays 200');
65+
assert.equal(r.attempts, 1, 'streaming 200 is never retried');
66+
67+
// The body must error within a bounded time (the watchdog), so proxy.js's
68+
// pipeline surfaces the stall instead of hanging forever. We read the body
69+
// via getReader: a healthy open stream would never reject; an aborted
70+
// (watchdog-fired) stream rejects with the idle-timeout error.
71+
const reader = r.response.body.getReader();
72+
let errored = false;
73+
let errName = '';
74+
const settled = await Promise.race([
75+
reader.read().then(
76+
() => reader.read(), // a chunk arrived — keep draining to catch the eventual error
77+
).catch((e) => { errored = true; errName = e?.message || String(e); }),
78+
new Promise((resolve) => setTimeout(() => resolve('timeout'), 500)),
79+
]);
80+
if (errored) {
81+
assert.match(errName, /idle timeout/i, `body should error with idle-timeout, got: ${errName}`);
82+
} else {
83+
// If no error within 500ms (well past the 50ms budget), the watchdog did NOT fire → hang regression.
84+
assert.fail('hung stream was not aborted by streamIdleTimeoutMs within 500ms (hang regression)');
85+
}
86+
});
87+
88+
it('healthy stream (chunks arriving) → watchdog resets, does NOT abort', async () => {
89+
// Emit a chunk every 5ms (well under the 200ms idle budget); the stream
90+
// stays healthy and the watchdog must never fire.
91+
globalThis.fetch = async (_url, opts) => {
92+
const signal = opts?.signal;
93+
return makeStreamResponse({
94+
emit: (controller) => {
95+
const iv = setInterval(() => {
96+
if (signal?.aborted) { clearInterval(iv); controller.close(); return; }
97+
controller.enqueue(new TextEncoder().encode('data: hi\n\n'));
98+
}, 5);
99+
// stop after a few chunks
100+
setTimeout(() => { clearInterval(iv); controller.close(); }, 30);
101+
},
102+
});
103+
};
104+
105+
const r = await executeRequest({
106+
url: 'https://example.com/v1/messages',
107+
fetchOptions: { method: 'POST', headers: {} },
108+
retryConfig: { ...DEFAULT_RETRY_CONFIG, mode: 'serial', streamIdleTimeoutMs: 200, maxRetries: 5 },
109+
ctx: {},
110+
});
111+
112+
assert.equal(r.finalStatus, 200);
113+
assert.equal(r.attempts, 1);
114+
// Healthy stream must NOT be marked stalled.
115+
assert.notEqual(r.__streamStalled, true, 'healthy stream must not be flagged stalled');
116+
});
117+
});

0 commit comments

Comments
 (0)