diff --git a/changelog.d/fixes/11873-streamtiming-monotonic-clock.md b/changelog.d/fixes/11873-streamtiming-monotonic-clock.md new file mode 100644 index 0000000000..9b6de75ac9 --- /dev/null +++ b/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 diff --git a/open-sse/utils/streamTiming.ts b/open-sse/utils/streamTiming.ts index 9f26f5d5b3..ee5385ebb3 100644 --- a/open-sse/utils/streamTiming.ts +++ b/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; diff --git a/tests/unit/stream-timing.test.ts b/tests/unit/stream-timing.test.ts index b6c5782579..fed4f52cf4 100644 --- a/tests/unit/stream-timing.test.ts +++ b/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}`); + }); +});