fix(open-sse): sample stream TTFT/ITL from a monotonic clock (#11873)

Switches stream TTFT/ITL sampling from Date.now() (wall clock) to performance.now() (monotonic) — an NTP correction or manual clock adjustment mid-stream was poisoning routing metrics (inflated TTFT on forward steps, silently-dropped negative TTFT on backward steps). Matches the existing earlyStreamKeepalive.ts precedent on the same streaming path. Thanks!

changelog.d/fixes/11873-streamtiming-monotonic-clock.md

@@ -0,0 +1 @@
+- **fix(open-sse):** stream TTFT/ITL are sampled from a monotonic clock (`performance.now`) instead of `Date.now`, so an NTP correction or wall-clock jump can no longer inflate a genuine TTFT or produce a negative one that the `ttft >= 0` guard silently drops — protecting the router's OTel/EWMA/`usage_history`/speed-ranking signals ([#11873](https://github.com/diegosouzapw/OmniRoute/pull/11873)) — thanks @pacocartones

open-sse/utils/streamTiming.ts

@@ -19,6 +19,16 @@
  *
  * The object is cheap to construct, plain mutable state, and safe under the
  * event loop's single thread (each stream owns its own instance).
+ *
+ * All timestamps are sampled from `performance.now()` (a monotonic clock,
+ * milliseconds since an arbitrary process-relative origin), NOT `Date.now()`
+ * (wall clock). Every field here is consumed only as an intra-instance delta
+ * (`ttftMs`, `avgItlMs`, `totalMs`), so a monotonic source keeps TTFT/ITL
+ * immune to NTP steps and wall-clock jumps that would otherwise poison the
+ * router's quality signals. Consequently these values are NOT epoch timestamps
+ * and must never be serialized, persisted, or compared across StreamTiming
+ * instances as absolute times — the same convention `earlyStreamKeepalive.ts`
+ * already follows on this streaming path.
  */
 export interface StreamTiming {
   startedAt: number;
@@ -45,7 +55,7 @@ const MAX_INTER_CHUNK_GAPS = 32;
 
 export function createStreamTiming(): StreamTiming {
   const timing: StreamTiming = {
-    startedAt: Date.now(),
+    startedAt: performance.now(),
     firstByteAt: null,
     firstForwardAt: null,
     lastForwardAt: null,
@@ -53,10 +63,10 @@ export function createStreamTiming(): StreamTiming {
     forwardedChunks: 0,
     interrupted: false,
     markByte() {
-      if (this.firstByteAt === null) this.firstByteAt = Date.now();
+      if (this.firstByteAt === null) this.firstByteAt = performance.now();
     },
     markForward() {
-      const now = Date.now();
+      const now = performance.now();
       if (this.firstForwardAt === null) this.firstForwardAt = now;
       if (this.lastForwardAt !== null && this.interChunkGaps.length < MAX_INTER_CHUNK_GAPS) {
         this.interChunkGaps.push(now - this.lastForwardAt);
@@ -76,7 +86,7 @@ export function createStreamTiming(): StreamTiming {
       return sum / this.interChunkGaps.length;
     },
     totalMs() {
-      return Date.now() - this.startedAt;
+      return performance.now() - this.startedAt;
     },
   };
   return timing;

tests/unit/stream-timing.test.ts

@@ -12,6 +12,16 @@ import test from "node:test";
 import assert from "node:assert/strict";
 import { createStreamTiming, type StreamTiming } from "../../open-sse/utils/streamTiming.ts";
 
+// streamTiming samples a high-resolution monotonic clock (performance.now).
+// `setTimeout(N)` does NOT guarantee that clock advances by a full N ms before
+// the callback runs: libuv schedules timers against its own cached loop clock,
+// which can trail performance.now() by a fraction of a millisecond, so a freshly
+// sampled performance.now() delta occasionally lands just under the nominal
+// sleep. Lower-bound timing assertions allow this scheduling slack. Event-loop
+// load only makes timers fire LATE (larger delta), never earlier, so the bound
+// stays safe on slow CI while remaining tight enough to prove a real delay.
+const TIMER_SLACK_MS = 5;
+
 test("ttft() is null when nothing was forwarded", () => {
   const t = createStreamTiming();
   t.markByte();
@@ -25,7 +35,7 @@ test("ttft() measures first-forwarded-chunk latency (byte vs forward distinguish
   await new Promise((r) => setTimeout(r, 20));
   t.markForward(); // first chunk forwarded 20ms later
   const ttft = t.ttftMs();
-  assert.ok(ttft !== null && ttft >= 20 && ttft < 5000, `ttft=${ttft}`);
+  assert.ok(ttft !== null && ttft >= 20 - TIMER_SLACK_MS && ttft < 5000, `ttft=${ttft}`);
   assert.ok(t.firstByteAt !== null);
   assert.ok(t.firstByteAt! < t.firstForwardAt!, "first byte precedes first forward");
 });
@@ -37,7 +47,7 @@ test("avgItlMs() measures mean inter-chunk gap across multiple chunks", async ()
     await new Promise((r) => setTimeout(r, 10));
   }
   const itl = t.avgItlMs();
-  assert.ok(itl !== null && itl >= 8 && itl < 5000, `itl=${itl}`);
+  assert.ok(itl !== null && itl >= 10 - TIMER_SLACK_MS && itl < 5000, `itl=${itl}`);
   assert.equal(t.forwardedChunks, 4);
 });
 
@@ -75,7 +85,7 @@ test("normal completion: totalMs() is monotonic and >= first-forward latency", a
   t.markForward();
   const total = t.totalMs();
   const ttft = t.ttftMs();
-  assert.ok(total >= 15);
+  assert.ok(total >= 15 - TIMER_SLACK_MS, `total=${total}`);
   assert.ok(ttft !== null && ttft <= total, "ttft must be <= total duration");
 });
 
@@ -84,3 +94,64 @@ test("max inter-chunk samples are bounded (memory bound)", async () => {
   for (let i = 0; i < 200; i++) t.markForward();
   assert.ok(t.interChunkGaps.length <= 32, `bounded to 32 samples, got ${t.interChunkGaps.length}`);
 });
+
+/**
+ * Wall-clock robustness: TTFT/ITL must be sampled from a MONOTONIC clock, so a
+ * mid-stream NTP correction or manual wall-clock adjustment cannot poison them.
+ * These metrics feed the router (OTel, quality EWMA, usage_history.ttft_ms,
+ * speedRanking), so a corrupted value degrades a healthy provider.
+ *
+ * We simulate a wall-clock jump by stubbing `Date.now` between marks. With the
+ * fix (`performance.now`, monotonic) the injected jump has no effect. If the
+ * seam ever regresses to `Date.now`, the injected jump leaks straight into the
+ * reported metric and these assertions fail.
+ */
+function withStubbedDateNow(run: (jumpMs: (delta: number) => void) => void): void {
+  const realNow = Date.now;
+  let fake = realNow.call(Date);
+  Date.now = () => fake;
+  try {
+    run((delta) => {
+      fake += delta;
+    });
+  } finally {
+    Date.now = realNow;
+  }
+}
+
+test("forward wall-clock jump does not inflate TTFT (monotonic clock)", () => {
+  withStubbedDateNow((jumpMs) => {
+    const t = createStreamTiming();
+    t.markByte();
+    jumpMs(5_000); // +5s NTP step between first byte and first forward
+    t.markForward();
+    const ttft = t.ttftMs();
+    // Real elapsed is sub-millisecond; a Date.now-based seam would report ~5000.
+    assert.ok(ttft !== null && ttft >= 0 && ttft < 1_000, `ttft must ignore +5s wall jump, got ${ttft}`);
+  });
+});
+
+test("backward wall-clock jump does not yield negative TTFT (monotonic clock)", () => {
+  withStubbedDateNow((jumpMs) => {
+    const t = createStreamTiming();
+    t.markByte();
+    jumpMs(-2_000); // clock stepped backwards between byte and forward
+    t.markForward();
+    const ttft = t.ttftMs();
+    // A Date.now-based seam would report ~-2000, silently discarded downstream
+    // by the `ttft >= 0` guard (invisible data loss).
+    assert.ok(ttft !== null && ttft >= 0 && ttft < 1_000, `ttft must never go negative, got ${ttft}`);
+  });
+});
+
+test("wall-clock jump does not corrupt inter-chunk ITL (monotonic clock)", () => {
+  withStubbedDateNow((jumpMs) => {
+    const t = createStreamTiming();
+    t.markForward();
+    jumpMs(3_000); // +3s NTP step between two forwarded chunks
+    t.markForward();
+    const itl = t.avgItlMs();
+    // A Date.now-based seam would record a 3000ms gap; monotonic stays near 0.
+    assert.ok(itl !== null && itl >= 0 && itl < 1_000, `itl must ignore +3s wall jump, got ${itl}`);
+  });
+});