diff --git a/src/server/relay.ts b/src/server/relay.ts index 19173cedc4..b96e3c6e49 100644 --- a/src/server/relay.ts +++ b/src/server/relay.ts @@ -192,31 +192,49 @@ function wsOversizeDiagnosis(sentBytes: number, stats: ShedStats | undefined, bu return `WS upstream rejected the ${formatMiB(sentBytes)} MiB request with close 1009 (zero events received) and the payload cannot be reduced further (${why}). Trim older images/history so the frame fits under the wire limit (current pre-send budget: ${budget > 0 ? `${formatMiB(budget)} MiB` : "disabled"} — network.wsOutboundBytes / BILI_WS_OUTBOUND_BYTES).`; } -// #3434: upstream accepted the request then sent no bytes — the log shows +// #3434/#5164: upstream accepted the request then sent no bytes — the log shows // nothing between `forward …` and the outcome (client disconnect / idle-watchdog // abort), so an upstream stall reads as proxy-side. Arm a one-shot first-byte -// watchdog around the initial fetch: headers not arrived by the threshold → ONE -// warn naming target+method+lane. Additive logging only — deliberately NO -// fast-fail (would truncate long prefills, cf. #1706/#1714) and NO new config -// surface (code constant + test seam). +// watchdog around the initial fetch: headers not arrived by the threshold → a +// warn naming session+target+method+lane (+ how long it has actually been +// pending — under a post-wedge late fire that far exceeds the threshold and is +// exactly the number triage needs). Additive logging only — deliberately NO +// fast-fail (would truncate long prefills, cf. #1706/#1714; the fail-fast +// variant is an owner-gated decision, #5164/#5262) and NO new config surface +// (code constants + test seams). const UPSTREAM_FIRST_BYTE_WARN_MS = 30_000; let firstByteWarnMs = UPSTREAM_FIRST_BYTE_WARN_MS; export function _setFirstByteWarnMsForTest(ms?: number): void { firstByteWarnMs = ms ?? UPSTREAM_FIRST_BYTE_WARN_MS; } -// Dedupe once per origin per process (windowSourceLogged pattern) so a -// long-prefill local model logs at most one line, not one per slow request. -// Claimed at FIRE time (not arm time) so a fast request never suppresses a -// later genuine stall on the same origin. -const firstByteWarnedOrigins = new Set(); -export function _resetFirstByteWarnedOriginsForTest(): void { - firstByteWarnedOrigins.clear(); +// #5164: warn cadence is per (origin, session) with a COOLDOWN, replacing the +// old once-per-origin-per-process dedupe. The permanent Set burned its single +// warn on the first post-wedge fire and then silenced every later stalled turn +// for the rest of the process life — the diagnostic surface went to zero while +// the incident was still ongoing (#5139 Phase 2). The cooldown keeps #3434's +// anti-spam intent: a pathological fast stall loop on one origin+session logs +// at most one line per window instead of one per request. +const FIRST_BYTE_WARN_COOLDOWN_MS = 60_000; +let firstByteWarnCooldownMs = FIRST_BYTE_WARN_COOLDOWN_MS; +export function _setFirstByteWarnCooldownMsForTest(ms?: number): void { + firstByteWarnCooldownMs = ms ?? FIRST_BYTE_WARN_COOLDOWN_MS; +} +// Last-warn epoch per `${origin}\u0000${sessionId}`. Entries older than the +// cooldown are dead weight (they allow a warn anyway) and are swept at set +// time so a long-lived process cannot grow the map unboundedly. +const firstByteWarnLastAt = new Map(); +const FIRST_BYTE_WARN_MAP_CAP = 4096; +export function _resetFirstByteWarnStateForTest(): void { + firstByteWarnLastAt.clear(); } /** - * #3434: arm a one-shot watchdog logging ONE warn if the fetch behind `signal` - * is still pending after `firstByteWarnMs`. Returns the timer handle (the caller - * clears it when the fetch settles) or undefined when disarmed (threshold - * disabled, origin already warned, or signal already aborted). + * #3434/#5164: arm a one-shot watchdog logging a warn if the fetch behind + * `signal` is still pending after `firstByteWarnMs`. Returns the timer handle + * (the caller clears it when the fetch settles) or undefined when disarmed + * (threshold disabled or signal already aborted). The cooldown is checked at + * FIRE time, never at arm time — a request armed inside the cooldown window + * may settle before firing, and disarming it would suppress a genuine later + * stall (same claim-at-fire discipline as the original dedupe, #3434). */ export function armFirstByteWatchdog(o: { url: string; @@ -228,20 +246,29 @@ export function armFirstByteWatchdog(o: { }): ReturnType | undefined { const threshold = firstByteWarnMs; if (!Number.isFinite(threshold) || threshold <= 0) return undefined; + if (o.signal.aborted) return undefined; let origin: string; try { origin = new URL(o.url).origin; } catch { origin = o.url; } - if (firstByteWarnedOrigins.has(origin)) return undefined; - if (o.signal.aborted) return undefined; + const key = `${origin}\u0000${o.sessionId}`; const seconds = Math.round(threshold / 1000); + const armedAt = Date.now(); const fire = (): void => { - // First-to-fire claims the origin; concurrent same-origin timers stay silent. - if (firstByteWarnedOrigins.has(origin)) return; - firstByteWarnedOrigins.add(origin); - o.log("warn", `[${o.sessionId}] upstream awaiting first byte ≥${seconds}s: ${o.method} ${maskUrlForLog(o.url)} (lane=${o.lane})`); + const now = Date.now(); + const last = firstByteWarnLastAt.get(key); + if (last !== undefined && now - last < firstByteWarnCooldownMs) return; + if (firstByteWarnLastAt.size >= FIRST_BYTE_WARN_MAP_CAP) { + for (const [k, t] of firstByteWarnLastAt) { + if (now - t >= firstByteWarnCooldownMs) firstByteWarnLastAt.delete(k); + } + if (firstByteWarnLastAt.size >= FIRST_BYTE_WARN_MAP_CAP) firstByteWarnLastAt.clear(); + } + firstByteWarnLastAt.set(key, now); + const heldSeconds = Math.round((now - armedAt) / 1000); + o.log("warn", `[${o.sessionId}] upstream awaiting first byte ≥${seconds}s (pending ${heldSeconds}s): ${o.method} ${maskUrlForLog(o.url)} (lane=${o.lane})`); }; const timer = setTimeout(fire, threshold); timer.unref(); diff --git a/tests/issue3434-first-byte-watchdog.test.ts b/tests/issue3434-first-byte-watchdog.test.ts index 9dbe44e3ab..129beed495 100644 --- a/tests/issue3434-first-byte-watchdog.test.ts +++ b/tests/issue3434-first-byte-watchdog.test.ts @@ -1,7 +1,8 @@ import assert from "node:assert/strict"; import { beforeEach, test } from "node:test"; import { - _resetFirstByteWarnedOriginsForTest, + _resetFirstByteWarnStateForTest, + _setFirstByteWarnCooldownMsForTest, _setFirstByteWarnMsForTest, armFirstByteWatchdog, } from "../src/server/relay.ts"; @@ -19,11 +20,12 @@ function makeLog(): { log: (level: string, msg: string) => void; calls: LogCall[ } beforeEach(() => { - _resetFirstByteWarnedOriginsForTest(); + _resetFirstByteWarnStateForTest(); _setFirstByteWarnMsForTest(THRESHOLD_MS); + _setFirstByteWarnCooldownMsForTest(undefined); }); -test("warns exactly once when the fetch is still pending past the threshold", async () => { +test("warns when the fetch is still pending past the threshold", async () => { const { log, calls } = makeLog(); const ac = new AbortController(); const timer = armFirstByteWatchdog({ url: "https://api.example.com/v1/messages", method: "POST", lane: "side", sessionId: "sess-1", log, signal: ac.signal }); @@ -33,6 +35,10 @@ test("warns exactly once when the fetch is still pending past the threshold", as assert.equal(calls.length, 1); assert.equal(calls[0].level, "warn"); assert.ok(calls[0].msg.startsWith("[sess-1] upstream awaiting first byte ≥"), calls[0].msg); + // #5164: the line carries the ACTUAL pending duration — under a post-wedge + // late fire that far exceeds the nominal threshold and is the number triage + // needs (the incident turns were held ~300s while the line said "≥30s"). + assert.ok(calls[0].msg.includes("(pending "), calls[0].msg); // The target goes through maskUrlForLog — a non-allowlisted host is reduced // to , so the warn line must never leak the real upstream host. assert.ok(calls[0].msg.includes(": POST https:///v1/messages"), calls[0].msg); @@ -49,26 +55,67 @@ test("does not warn when the fetch settles before the threshold (caller clears)" assert.equal(calls.length, 0); }); -test("dedupes per origin: a later pending request to the same origin stays silent", async () => { +test("same origin+session warns again after the cooldown passes (no permanent quota, #5164)", async () => { + // Acceptance ②/④: a LATER stalled turn on the same origin+session — the + // host-timeout retry in the #5139 incident shape — must get its own line. + // The decision is fire-time vs last-warn-time (a decaying timestamp), never + // a permanent per-origin claim, so a post-wedge late fire after a quiet + // interval can no longer be silently swallowed by a burned quota. + _setFirstByteWarnCooldownMsForTest(80); const { log, calls } = makeLog(); const acA = new AbortController(); - const tA = armFirstByteWatchdog({ url: "https://dup.example.com/x", method: "POST", lane: "main", sessionId: "a", log, signal: acA.signal }); + const tA = armFirstByteWatchdog({ url: "https://dup.example.com/x", method: "POST", lane: "main", sessionId: "sess-1", log, signal: acA.signal }); await sleep(WAIT_MS); if (tA) clearTimeout(tA); assert.equal(calls.length, 1); - // Same origin (scheme+host+port), different path — disarmed at arm time. const acB = new AbortController(); - const tB = armFirstByteWatchdog({ url: "https://dup.example.com/y", method: "POST", lane: "main", sessionId: "b", log, signal: acB.signal }); - assert.equal(tB, undefined); + const tB = armFirstByteWatchdog({ url: "https://dup.example.com/y", method: "POST", lane: "main", sessionId: "sess-1", log, signal: acB.signal }); + await sleep(WAIT_MS + 100); + if (tB) clearTimeout(tB); + assert.equal(calls.length, 2); + assert.ok(calls[1].msg.startsWith("[sess-1] upstream awaiting first byte ≥"), calls[1].msg); +}); + +test("same origin+session within the cooldown stays silent (anti-spam negative direction, #3434 intent kept)", async () => { + // A pathological fast stall loop on one origin+session must not log one line + // per request: the second stall inside the cooldown window is suppressed. + _setFirstByteWarnCooldownMsForTest(2000); + const { log, calls } = makeLog(); + const acA = new AbortController(); + const tA = armFirstByteWatchdog({ url: "https://spam.example.com/x", method: "POST", lane: "main", sessionId: "sess-1", log, signal: acA.signal }); await sleep(WAIT_MS); + if (tA) clearTimeout(tA); + assert.equal(calls.length, 1); + const acB = new AbortController(); + const tB = armFirstByteWatchdog({ url: "https://spam.example.com/y", method: "POST", lane: "main", sessionId: "sess-1", log, signal: acB.signal }); + assert.notEqual(tB, undefined); // arming never consults warn history + await sleep(WAIT_MS); + if (tB) clearTimeout(tB); assert.equal(calls.length, 1); }); -test("different origins each warn once (dedupe is keyed by origin, not global)", async () => { +test("different sessions on the same origin each warn (no cross-session suppression, #5164)", async () => { + // One session's stall must not silence another session's stall on the same + // upstream (the old per-origin Set did exactly that at arm time). + _setFirstByteWarnCooldownMsForTest(2000); const { log, calls } = makeLog(); const acA = new AbortController(); - const tA = armFirstByteWatchdog({ url: "https://a.example.com/", method: "GET", lane: "main", sessionId: "a", log, signal: acA.signal }); const acB = new AbortController(); + const tA = armFirstByteWatchdog({ url: "https://x.example.com/", method: "GET", lane: "main", sessionId: "sess-a", log, signal: acA.signal }); + const tB = armFirstByteWatchdog({ url: "https://x.example.com/", method: "GET", lane: "main", sessionId: "sess-b", log, signal: acB.signal }); + assert.notEqual(tA, undefined); + assert.notEqual(tB, undefined); + await sleep(WAIT_MS); + if (tA) clearTimeout(tA); + if (tB) clearTimeout(tB); + assert.equal(calls.length, 2); +}); + +test("different origins each warn (key includes origin, not session only)", async () => { + const { log, calls } = makeLog(); + const acA = new AbortController(); + const acB = new AbortController(); + const tA = armFirstByteWatchdog({ url: "https://a.example.com/", method: "GET", lane: "main", sessionId: "a", log, signal: acA.signal }); const tB = armFirstByteWatchdog({ url: "https://b.example.com/", method: "GET", lane: "main", sessionId: "b", log, signal: acB.signal }); await sleep(WAIT_MS); if (tA) clearTimeout(tA); @@ -105,16 +152,16 @@ test("is disarmed when the threshold is disabled (<=0 or non-finite)", () => { assert.equal(calls.length, 0); }); -test("_resetFirstByteWarnedOriginsForTest lets a previously-warned origin warn again", async () => { +test("_resetFirstByteWarnStateForTest lets a previously-warned key warn again", async () => { const { log, calls } = makeLog(); const ac1 = new AbortController(); const t1 = armFirstByteWatchdog({ url: "https://rs.example.com/", method: "POST", lane: "main", sessionId: "a", log, signal: ac1.signal }); await sleep(WAIT_MS); if (t1) clearTimeout(t1); assert.equal(calls.length, 1); - _resetFirstByteWarnedOriginsForTest(); + _resetFirstByteWarnStateForTest(); const ac2 = new AbortController(); - const t2 = armFirstByteWatchdog({ url: "https://rs.example.com/", method: "POST", lane: "main", sessionId: "b", log, signal: ac2.signal }); + const t2 = armFirstByteWatchdog({ url: "https://rs.example.com/", method: "POST", lane: "main", sessionId: "a", log, signal: ac2.signal }); assert.notEqual(t2, undefined); await sleep(WAIT_MS); if (t2) clearTimeout(t2);