sse/timing: stop asserting wall-clock setTimeout against Date.now

CI Unit Tests (3/8) failed once npm ci started succeeding: setTimeout(15)
can fire before Date.now() has moved 15ms. Tests now drive createStreamTiming
with an injected clock; production still uses Date.now. The rewritten test
no longer trips unused-vars, so drop that file's leftover eslint suppression.

Signed-off-by: Minxi Hou <houminxi@gmail.com>

changelog.d/fixes/stream-timing-clock.md

@@ -0,0 +1 @@
+- **fix(sse):** `createStreamTiming` accepts an injectable clock so unit tests do not assert `Date.now()` against `setTimeout(15)`. CI runners can fire that timer before the wall clock has moved 15ms, which was failing `Unit Tests (3/8)` once `npm ci` started succeeding on Node 24.

config/quality/eslint-suppressions.json

@@ -6124,11 +6124,6 @@
       "count": 1
     }
   },
-  "tests/unit/stream-timing.test.ts": {
-    "@typescript-eslint/no-unused-vars": {
-      "count": 1
-    }
-  },
   "tests/unit/stream-utilities.test.ts": {
     "@typescript-eslint/no-unused-vars": {
       "count": 2

open-sse/utils/streamTiming.ts

@@ -19,7 +19,13 @@
  *
  * The object is cheap to construct, plain mutable state, and safe under the
  * event loop's single thread (each stream owns its own instance).
+ *
+ * Tests may pass a clock so assertions do not depend on wall-clock `setTimeout`
+ * (CI runners can fire a 15ms timer before `Date.now()` advances 15ms).
  */
+/** Milliseconds since epoch; same contract as Date.now. */
+export type StreamClock = () => number;
+
 export interface StreamTiming {
   startedAt: number;
   firstByteAt: number | null;
@@ -43,9 +49,9 @@ export interface StreamTiming {
 /** Max number of inter-chunk samples kept (bounds memory). */
 const MAX_INTER_CHUNK_GAPS = 32;
 
-export function createStreamTiming(): StreamTiming {
+export function createStreamTiming(now: StreamClock = Date.now): StreamTiming {
   const timing: StreamTiming = {
-    startedAt: Date.now(),
+    startedAt: now(),
     firstByteAt: null,
     firstForwardAt: null,
     lastForwardAt: null,
@@ -53,15 +59,15 @@ export function createStreamTiming(): StreamTiming {
     forwardedChunks: 0,
     interrupted: false,
     markByte() {
-      if (this.firstByteAt === null) this.firstByteAt = Date.now();
+      if (this.firstByteAt === null) this.firstByteAt = now();
     },
     markForward() {
-      const now = Date.now();
-      if (this.firstForwardAt === null) this.firstForwardAt = now;
+      const t = now();
+      if (this.firstForwardAt === null) this.firstForwardAt = t;
       if (this.lastForwardAt !== null && this.interChunkGaps.length < MAX_INTER_CHUNK_GAPS) {
-        this.interChunkGaps.push(now - this.lastForwardAt);
+        this.interChunkGaps.push(t - this.lastForwardAt);
       }
-      this.lastForwardAt = now;
+      this.lastForwardAt = t;
       this.forwardedChunks += 1;
     },
     markInterrupted() {
@@ -76,7 +82,7 @@ export function createStreamTiming(): StreamTiming {
       return sum / this.interChunkGaps.length;
     },
     totalMs() {
-      return Date.now() - this.startedAt;
+      return now() - this.startedAt;
     },
   };
   return timing;

tests/unit/stream-timing.test.ts

@@ -7,61 +7,77 @@
  *  - first-byte vs first-forward distinction
  *  - interruption marking
  *  - malformed/empty chunks do not corrupt timing
+ *
+ * Clock is injected so CI cannot fail because a 15ms timer fired before
+ * Date.now() advanced 15ms.
  */
 import test from "node:test";
 import assert from "node:assert/strict";
 import { createStreamTiming, type StreamTiming } from "../../open-sse/utils/streamTiming.ts";
 
+function fakeClock(start = 1_000): { now: () => number; advance: (ms: number) => void } {
+  let t = start;
+  return {
+    now: () => t,
+    advance: (ms: number) => {
+      t += ms;
+    },
+  };
+}
+
 test("ttft() is null when nothing was forwarded", () => {
-  const t = createStreamTiming();
+  const t = createStreamTiming(fakeClock().now);
   t.markByte();
   assert.equal(t.ttftMs(), null);
   assert.equal(t.avgItlMs(), null);
 });
 
-test("ttft() measures first-forwarded-chunk latency (byte vs forward distinguished)", async () => {
-  const t = createStreamTiming();
-  t.markByte(); // first upstream byte arrives immediately
-  await new Promise((r) => setTimeout(r, 20));
-  t.markForward(); // first chunk forwarded 20ms later
+test("ttft() measures first-forwarded-chunk latency (byte vs forward distinguished)", () => {
+  const clock = fakeClock();
+  const t = createStreamTiming(clock.now);
+  t.markByte();
+  clock.advance(20);
+  t.markForward();
   const ttft = t.ttftMs();
-  assert.ok(ttft !== null && ttft >= 20 && ttft < 5000, `ttft=${ttft}`);
+  assert.equal(ttft, 20);
   assert.ok(t.firstByteAt !== null);
   assert.ok(t.firstByteAt! < t.firstForwardAt!, "first byte precedes first forward");
 });
 
-test("avgItlMs() measures mean inter-chunk gap across multiple chunks", async () => {
-  const t = createStreamTiming();
+test("avgItlMs() measures mean inter-chunk gap across multiple chunks", () => {
+  const clock = fakeClock();
+  const t = createStreamTiming(clock.now);
   for (let i = 0; i < 4; i++) {
     t.markForward();
-    await new Promise((r) => setTimeout(r, 10));
+    clock.advance(10);
   }
   const itl = t.avgItlMs();
-  assert.ok(itl !== null && itl >= 8 && itl < 5000, `itl=${itl}`);
+  assert.equal(itl, 10);
   assert.equal(t.forwardedChunks, 4);
 });
 
 test("empty chunks do not corrupt timing (markByte without forward)", () => {
-  const t = createStreamTiming();
+  const t = createStreamTiming(fakeClock().now);
   t.markByte();
   t.markByte(); // duplicate bytes are idempotent for first-byte
+  t.markByte();
   assert.equal(t.ttftMs(), null, "no forward → no ttft");
   t.markForward();
   assert.ok(t.ttftMs() !== null);
 });
 
 test("malformed/keepalive-only traffic (no forward) yields no ttft", () => {
-  const t = createStreamTiming();
-  // Simulate a provider that only sends keepalives/blank lines, never data.
+  const t = createStreamTiming(fakeClock().now);
   for (let i = 0; i < 5; i++) t.markByte();
   assert.equal(t.ttftMs(), null);
   assert.equal(t.forwardedChunks, 0);
 });
 
-test("interruption is recorded and does not reset other timing", async () => {
-  const t = createStreamTiming();
+test("interruption is recorded and does not reset other timing", () => {
+  const clock = fakeClock();
+  const t = createStreamTiming(clock.now);
   t.markForward();
-  await new Promise((r) => setTimeout(r, 5));
+  clock.advance(5);
   t.markForward();
   t.markInterrupted();
   assert.equal(t.interrupted, true);
@@ -69,18 +85,33 @@ test("interruption is recorded and does not reset other timing", async () => {
   assert.ok(t.avgItlMs() !== null);
 });
 
-test("normal completion: totalMs() is monotonic and >= first-forward latency", async () => {
-  const t = createStreamTiming();
-  await new Promise((r) => setTimeout(r, 15));
+test("normal completion: totalMs() is monotonic and >= first-forward latency", () => {
+  const clock = fakeClock();
+  const t = createStreamTiming(clock.now);
+  clock.advance(15);
   t.markForward();
   const total = t.totalMs();
   const ttft = t.ttftMs();
-  assert.ok(total >= 15);
+  assert.equal(total, 15);
   assert.ok(ttft !== null && ttft <= total, "ttft must be <= total duration");
 });
 
-test("max inter-chunk samples are bounded (memory bound)", async () => {
-  const t = createStreamTiming();
+test("max inter-chunk samples are bounded (memory bound)", () => {
+  const t = createStreamTiming(fakeClock().now);
   for (let i = 0; i < 200; i++) t.markForward();
   assert.ok(t.interChunkGaps.length <= 32, `bounded to 32 samples, got ${t.interChunkGaps.length}`);
 });
+
+test("default clock is Date.now when no clock is passed", () => {
+  const realNow = Date.now;
+  let t0 = 5_000;
+  Date.now = () => t0;
+  try {
+    const t: StreamTiming = createStreamTiming();
+    t0 += 7;
+    t.markForward();
+    assert.equal(t.ttftMs(), 7);
+  } finally {
+    Date.now = realNow;
+  }
+});