refactor(backpressure): reduce allocations in LatestValuesAccumulator

src/LatestValuesAccumulator.ts

@@ -1,16 +1,18 @@
 /**
- * LatestValuesAccumulator - Accumulates Signal K delta values during backpressure,
- * keeping only the latest value for each unique context:path:$source combination.
+ * Accumulates Signal K delta values during backpressure, keeping only the
+ * latest value for each unique context:path:$source combination.
  */
 
 import {
   Context,
   Delta,
   hasValues,
   Path,
+  PathValue,
   SourceRef,
   Timestamp,
-  Update
+  Update,
+  Value
 } from '@signalk/server-api'
 
 export interface AccumulatedItem {
@@ -28,6 +30,8 @@ export interface BackpressureDelta extends Delta {
   }
 }
 
+const UNKNOWN_SOURCE = 'unknown'
+
 /**
  * Accumulate latest value per context:path:$source during backpressure.
  * Only keeps the most recent value for each unique combination, dropping intermediate updates.
@@ -40,17 +44,28 @@ export function accumulateLatestValue(
   delta: Delta
 ): void {
   if (!delta.updates) return
+  const context = delta.context as Context
   for (const update of delta.updates) {
     if (!hasValues(update)) continue
+    const $source = update.$source
+    const sourceKey = $source || UNKNOWN_SOURCE
+    const timestamp = update.timestamp
     for (const pv of update.values) {
-      const key = `${delta.context}:${pv.path}:${update.$source || 'unknown'}`
-      accumulator.set(key, {
-        context: delta.context as Context,
-        path: pv.path,
-        value: pv.value,
-        $source: update.$source,
-        timestamp: update.timestamp
-      })
+      const key = `${context}:${pv.path}:${sourceKey}`
+      const existing = accumulator.get(key)
+      if (existing) {
+        existing.value = pv.value
+        existing.$source = $source
+        existing.timestamp = timestamp
+      } else {
+        accumulator.set(key, {
+          context,
+          path: pv.path,
+          value: pv.value,
+          $source,
+          timestamp
+        })
+      }
     }
   }
 }
@@ -71,45 +86,44 @@ export function buildFlushDeltas(
 
   const countBefore = accumulator.size
 
-  // Group by context
-  const byContext = new Map<Context, Map<string, Update>>()
-  for (const [, item] of accumulator) {
-    if (!byContext.has(item.context)) {
-      byContext.set(item.context, new Map())
+  const byContext = new Map<
+    Context,
+    Map<string, Update & { values: PathValue[] }>
+  >()
+  for (const item of accumulator.values()) {
+    let bySource = byContext.get(item.context)
+    if (!bySource) {
+      bySource = new Map()
+      byContext.set(item.context, bySource)
     }
-    // Group by $source within context
-    const bySource = byContext.get(item.context)!
-    const sourceKey = item.$source || 'unknown'
-    if (!bySource.has(sourceKey)) {
-      bySource.set(sourceKey, {
-        $source: item.$source as SourceRef,
-        timestamp: item.timestamp as Timestamp,
+    const sourceKey = item.$source || UNKNOWN_SOURCE
+    let update = bySource.get(sourceKey)
+    if (!update) {
+      update = {
+        $source: item.$source,
+        timestamp: item.timestamp,
         values: []
-      })
-    }
-    const update = bySource.get(sourceKey)!
-    if (hasValues(update)) {
-      update.values.push({
-        path: item.path as Path,
-        // eslint-disable-next-line @typescript-eslint/no-explicit-any
-        value: item.value as any
-      })
-      // Use the most recent timestamp for this source
-      if (
-        item.timestamp &&
-        (!update.timestamp || item.timestamp > update.timestamp)
-      ) {
-        update.timestamp = item.timestamp as Timestamp
       }
+      bySource.set(sourceKey, update)
+    }
+    update.values.push({
+      path: item.path,
+      value: item.value as Value
+    })
+    // Keep the newest timestamp seen for this source so the flushed delta reflects the latest update.
+    if (
+      item.timestamp &&
+      (!update.timestamp || item.timestamp > update.timestamp)
+    ) {
+      update.timestamp = item.timestamp
     }
   }
 
-  // Build one delta per context with backpressure indicator
   const deltas: BackpressureDelta[] = []
-  for (const [context, bySourceTime] of byContext) {
+  for (const [context, bySource] of byContext) {
     deltas.push({
       context,
-      updates: Array.from(bySourceTime.values()),
+      updates: Array.from(bySource.values()),
       $backpressure: {
         accumulated: countBefore,
         duration

test/LatestValuesAccumulator.js

@@ -326,5 +326,22 @@ describe('LatestValuesAccumulator', function () {
       expect(deltas[0].$backpressure.accumulated).to.equal(10)
       expect(deltas[0].$backpressure.duration).to.equal(2000)
     })
+
+    it('should propagate undefined timestamp when items have no timestamp', function () {
+      const accumulator = new Map()
+      accumulator.set('vessels.self:navigation.speedOverGround:gps', {
+        context: 'vessels.self',
+        path: 'navigation.speedOverGround',
+        value: 5.0,
+        $source: 'gps',
+        timestamp: undefined
+      })
+
+      const deltas = buildFlushDeltas(accumulator, 1000)
+
+      expect(deltas.length).to.equal(1)
+      expect(deltas[0].updates.length).to.equal(1)
+      expect(deltas[0].updates[0].timestamp).to.equal(undefined)
+    })
   })
 })