[PR] fix: first-byte stall warn per (origin,session) with cooldown instead of once-per-origin-per-process (#5164)

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<string>();
-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<string, number>();
+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<typeof setTimeout> | 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();

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 <private-host>, so the warn line must never leak the real upstream host.
     assert.ok(calls[0].msg.includes(": POST https://<private-host>/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);