Measure live telemetry freshness at frame receipt instead of the producer clock

Pecan showed STOPPED on every data row whenever the car/ECU clock was skewed
from the viewing browser. getFrequency anchored its 2-second liveness window to
the browser's Date.now() while frames carry the car's absolute timestamps, so a
clock skew greater than the window dropped every frame outside it and reported 0
Hz even while data streamed.

Track each message buffer's wall-clock receipt time (lastReceivedAt) separately
from the producer timestamp, and anchor the live window to newest-sample
timestamp plus the time elapsed since it arrived. Steady skew now cancels out so
a streaming feed reads live, while the window still empties to 0 Hz once frames
genuinely stop. Replay path is unchanged. Adds regression tests for both cases.

Author: haoruizhou

pecan/src/lib/DataStore.test.ts

@@ -71,6 +71,51 @@ describe('DataStore', () => {
     expect(dataStore.getFrequency(msgID, FREQUENCY_WINDOW_MS)).toBe(0.0);
   });
 
+  it('reports live frequency when producer timestamps lag the browser clock (clock skew)', () => {
+    // Regression: the car/ECU stamps frames with its own absolute epoch-ms clock.
+    // If that clock runs behind this browser by more than the freshness window,
+    // every frame is >2s "old" by wall clock even though data is streaming live.
+    const browserNow = Date.now();
+    const SKEW = 30_000; // car clock is 30s behind the browser
+    const msgID = '0xSKEW';
+
+    // 5 frames arriving in real time, each carrying a producer timestamp 30s in the past.
+    for (let i = 0; i < 5; i++) {
+      vi.setSystemTime(browserNow + i * 200);        // wall-clock receipt time
+      dataStore.ingestMessage({
+        msgID,
+        messageName: 'Skewed',
+        data: {},
+        rawData: '00',
+        timestamp: browserNow + i * 200 - SKEW,      // producer clock, lagging
+      });
+    }
+
+    // Frames are flowing → must read as live, not STOPPED (hz 0).
+    expect(dataStore.getFrequency(msgID, FREQUENCY_WINDOW_MS)).toBeGreaterThan(0);
+  });
+
+  it('still reports STOPPED (0 Hz) when a skewed feed genuinely stops', () => {
+    const browserNow = Date.now();
+    const SKEW = 30_000;
+    const msgID = '0xSKEWSTOP';
+
+    for (let i = 0; i < 5; i++) {
+      vi.setSystemTime(browserNow + i * 200);
+      dataStore.ingestMessage({
+        msgID,
+        messageName: 'Skewed',
+        data: {},
+        rawData: '00',
+        timestamp: browserNow + i * 200 - SKEW,
+      });
+    }
+
+    // No new frames for 3 wall-clock seconds — the feed has truly stopped.
+    vi.setSystemTime(browserNow + 800 + 3000);
+    expect(dataStore.getFrequency(msgID, FREQUENCY_WINDOW_MS)).toBe(0);
+  });
+
   it('should handle multiple message IDs independently', () => {
     const now = Date.now();
     vi.setSystemTime(now);

pecan/src/lib/DataStore.ts

@@ -57,6 +57,10 @@ export interface TelemetrySample {
 interface MessageBuffer {
   samples: TelemetrySample[];
   lastUpdated: number;
+  // Wall-clock time (Date.now()) when the newest sample was ingested. Tracked
+  // separately from sample.timestamp (the producer's clock) so that liveness is
+  // measured at receipt — immune to clock skew between the car/ECU and this browser.
+  lastReceivedAt: number;
 }
 
 export type TelemetrySource = "live" | "replay";
@@ -551,6 +555,9 @@ class DataStore {
     preserveTimestamp?: boolean;
     source?: TelemetrySource;
   }): void {
+    // Wall-clock receipt time, captured before any timestamp rewriting below.
+    const receivedAt = Date.now();
+
     // Fix for old timestamps from recorded data
     // If timestamp is more than 1 hour old, use current time
     let timestamp = message.timestamp || Date.now();
@@ -597,6 +604,7 @@ class DataStore {
       buffers.byMsgId.set(msgID, {
         samples: [],
         lastUpdated: timestamp,
+        lastReceivedAt: receivedAt,
       });
     }
 
@@ -605,6 +613,7 @@ class DataStore {
     // Add new sample
     messageBuffer.samples.push(sample);
     messageBuffer.lastUpdated = timestamp;
+    messageBuffer.lastReceivedAt = receivedAt;
 
     // Prune old samples (rolling window)
     this.pruneOldSamples(msgID, source);
@@ -644,6 +653,9 @@ class DataStore {
 
     let newestTimestampBySource: Partial<Record<TelemetrySource, number>> = {};
 
+    // Wall-clock receipt time for this batch, captured before timestamp rewriting.
+    const receivedAt = Date.now();
+
     for (const message of messages) {
       // Fix for old timestamps from recorded data unless explicitly preserved.
       let timestamp = message.timestamp || Date.now();
@@ -684,12 +696,14 @@ class DataStore {
         buffers.byMsgId.set(sample.msgID, {
           samples: [],
           lastUpdated: timestamp,
+          lastReceivedAt: receivedAt,
         });
       }
 
       const messageBuffer = buffers.byMsgId.get(sample.msgID)!;
       messageBuffer.samples.push(sample);
       messageBuffer.lastUpdated = timestamp;
+      messageBuffer.lastReceivedAt = receivedAt;
       this.pruneOldSamples(sample.msgID, source);
 
       buffers.trace.push(sample);
@@ -928,9 +942,15 @@ class DataStore {
     const messageBuffer = this.getSourceBuffers(source).byMsgId.get(msgID);
     if (!messageBuffer || messageBuffer.samples.length === 0) return 0;
 
+    const newestTimestamp = messageBuffer.samples[messageBuffer.samples.length - 1].timestamp;
+    // Live "now" is anchored to the newest sample's producer timestamp plus the
+    // wall-clock time elapsed since it arrived — i.e. liveness is measured at
+    // receipt, not against the car/ECU clock. This keeps a streaming feed from
+    // reading STOPPED when the producer's clock is skewed from the browser, while
+    // still letting the window empty (→ 0 Hz) once frames genuinely stop arriving.
     const now = source === "replay"
-      ? messageBuffer.samples[messageBuffer.samples.length - 1].timestamp
-      : Date.now();
+      ? newestTimestamp
+      : newestTimestamp + (Date.now() - messageBuffer.lastReceivedAt);
     const cutoffTime = now - windowMs;
     const startIdx   = binarySearchFirstGte(messageBuffer.samples, cutoffTime);
     const count      = messageBuffer.samples.length - startIdx;