feat: [2322] Dashboard data store - Chart performance

nodes/store/data.js

@@ -104,7 +104,7 @@ const getters = {
     },
     // given a widget id, return the latest msg received
     msg (id) {
-        return config.RED.util.cloneMessage(getStoreOrWarn('reading', id)?.[ENTRY_MSG](id))
+        return getStoreOrWarn('reading', id)?.[ENTRY_MSG](id)
     },
     has (id) {
         const store = getStoreOrWarn('reading', id)

nodes/store/reactive.js

@@ -4,7 +4,6 @@ const SET_SERIES = Symbol.for('@flowfuse/node-red-dashboard/setSeries')
 const MERGE_ENTRY = Symbol.for('@flowfuse/node-red-dashboard/mergeEntry')
 const TRIM_SERIES = Symbol.for('@flowfuse/node-red-dashboard/trimSeries')
 const ENTRY_MSG = Symbol.for('@flowfuse/node-red-dashboard/entryMsg')
-const TRIM_BATCH = 60
 const NAMESPACE = 'dashboardStore'
 
 class Entry {
@@ -14,7 +13,7 @@ class Entry {
         this.quality = 'GOOD'
         this.timestamp = 0
         this.history = []
-        this.trims = null
+        this.plainPoints = undefined
     }
 }
 
@@ -38,15 +37,15 @@ function deepClone (v, seen = new WeakMap()) {
     return out
 }
 
-function deepReactive (value, notify, path, clone, seen = new WeakMap()) {
+function deepReactive (value, notify, path, clone, seen = new WeakMap(), wrapOnRead = false) {
     if (!isReactable(value)) return value
     if (seen.has(value)) return seen.get(value)
-    const proxy = new Proxy(value, {
+    const traps = {
         set (t, p, v) {
             if (typeof p === 'symbol') return Reflect.set(t, p, v)
             // a push writes the new index then grows length; only a shrink is a change worth reporting
             const shrank = p === 'length' && Array.isArray(t) && v < t.length
-            t[p] = deepReactive(clone(v), notify, `${path}.${String(p)}`, clone)
+            t[p] = wrapOnRead ? clone(v) : deepReactive(clone(v), notify, `${path}.${String(p)}`, clone)
             if (p !== 'length') notify(`${path}.${String(p)}`)
             else if (shrank) notify(path)
             return true
@@ -57,10 +56,23 @@ function deepReactive (value, notify, path, clone, seen = new WeakMap()) {
             if (existed) notify(`${path}.${String(p)}`)
             return ok
         }
-    })
+    }
+    if (wrapOnRead) {
+        traps.get = (t, p, r) => {
+            const v = Reflect.get(t, p, r)
+            const descriptor = typeof p === 'symbol' ? undefined : Object.getOwnPropertyDescriptor(t, p)
+            if (!descriptor || (!descriptor.writable && !descriptor.configurable)) {
+                return v
+            }
+            return deepReactive(v, notify, `${path}.${p}`, clone, seen, true)
+        }
+    }
+    const proxy = new Proxy(value, traps)
     // registered before walking children so a cycle resolves to this proxy instead of recursing
     seen.set(value, proxy)
-    for (const k of Object.keys(value)) value[k] = deepReactive(value[k], notify, `${path}.${k}`, clone, seen)
+    if (!wrapOnRead) {
+        for (const k of Object.keys(value)) value[k] = deepReactive(value[k], notify, `${path}.${k}`, clone, seen)
+    }
     return proxy
 }
 
@@ -87,7 +99,7 @@ function createDataStore ({ maxHistory = 5, onChange, clone = deepClone, now = D
     records[STORE] = true
 
     // returns the record without notifying, so a caller can finish writing it before onChange fires
-    const writeValue = (prop, value) => {
+    const writeValue = (prop, value, wrapOnRead = false) => {
         let rec = records[prop]
         if (!rec) {
             rec = new Entry()
@@ -99,30 +111,45 @@ function createDataStore ({ maxHistory = 5, onChange, clone = deepClone, now = D
         }
         const notify = (path) => {
             // a replaced value keeps its proxies alive; they must not report into the record any more
-            if (rec.value !== wrapped) return
+            if (wrapOnRead ? rec.plainPoints !== value : rec.value !== wrapped) return
             rec.timestamp = now()
             emit(prop, rec, path)
         }
-        const wrapped = deepReactive(value, notify, prop, cloneValue)
+        const wrapped = deepReactive(value, notify, prop, cloneValue, new WeakMap(), wrapOnRead)
         rec.value = wrapped
-        rec.trims = null
+        rec.plainPoints = wrapOnRead ? value : undefined
         rec.msg = Object.freeze({ payload: wrapped })
         rec.timestamp = now()
         return rec
     }
 
-    const trimmedSeries = (rec) => rec.trims ? rec.trims.kept.concat(rec.msg.slice(rec.trims.seen)) : rec.msg.slice()
-    const msgOf = (rec) => Array.isArray(rec.msg) ? Object.freeze(trimmedSeries(rec)) : rec.msg
+    const pointsOf = (d) => (d === undefined || d === null) ? [] : (Array.isArray(d) ? d : [d])
+    const pointCount = (d) => (d === undefined || d === null) ? 0 : (Array.isArray(d) ? d.length : 1)
+
+    const messagesWithLivePoints = (rec) => {
+        let i = 0
+        return Object.freeze(rec.msg.map((m) => {
+            const n = pointCount(m._datapoint)
+            if (n === 0) {
+                return m
+            }
+            const livePoints = Array.from({ length: n }, () => rec.value[i++])
+            return Object.freeze({ ...m, _datapoint: Array.isArray(m._datapoint) ? Object.freeze(livePoints) : livePoints[0] })
+        }))
+    }
 
     records[ENTRY_MSG] = (prop) => {
         const rec = records[prop]
-        return rec ? msgOf(rec) : undefined
+        if (!rec) {
+            return undefined
+        }
+        return clone(rec.msg)
     }
 
     records[MERGE_ENTRY] = (prop, msg) => {
         const current = records[prop]
         const { payload, ...incoming } = msg
-        const previous = Array.isArray(current?.msg) ? freezeOwned(cloneValue({ ...trimmedSeries(current) })) : current?.msg || {}
+        const previous = Array.isArray(current?.msg) ? freezeOwned(cloneValue({ ...current.msg })) : current?.msg || {}
         const { payload: storedPayload, ...stored } = previous
         const keepValue = current && !Array.isArray(current.msg) && !('payload' in msg)
         const owned = freezeOwned(cloneValue(incoming))
@@ -135,8 +162,6 @@ function createDataStore ({ maxHistory = 5, onChange, clone = deepClone, now = D
         emit(prop, rec, prop)
     }
 
-    const pointsOf = (d) => (d === undefined || d === null) ? [] : (Array.isArray(d) ? d : [d])
-
     const toStoredDatapoint = (series, datapoint, start, end) => {
         if (end === start) return undefined
         return Array.isArray(datapoint) ? Object.freeze(series.slice(start, end)) : series[start]
@@ -159,9 +184,9 @@ function createDataStore ({ maxHistory = 5, onChange, clone = deepClone, now = D
             for (const p of pointsOf(datapoint)) points.push(p)
             return { msg, owned, datapoint, start, end: points.length }
         })
-        const rec = writeValue(prop, points)
+        const rec = writeValue(prop, points, true)
         rec.msg = parts.map(({ msg, owned, datapoint, start, end }) => {
-            const stored = toStoredDatapoint(rec.value, datapoint, start, end)
+            const stored = toStoredDatapoint(points, datapoint, start, end)
             if (stored === undefined && '_datapoint' in msg) owned._datapoint = datapoint
             return toStoredMessage(owned, stored)
         })
@@ -175,12 +200,12 @@ function createDataStore ({ maxHistory = 5, onChange, clone = deepClone, now = D
         const { _datapoint, ...rest } = msg
         const owned = cloneValue(rest)
         if (!rec || replaced) {
-            rec = writeValue(prop, [])
+            rec = writeValue(prop, [], true)
             rec.msg = []
         }
         const start = rec.value.length
         for (const p of pointsOf(_datapoint)) rec.value.push(p)
-        const stored = toStoredDatapoint(rec.value, _datapoint, start, rec.value.length)
+        const stored = toStoredDatapoint(rec.plainPoints, _datapoint, start, rec.value.length)
         if (stored === undefined && '_datapoint' in msg) owned._datapoint = cloneValue(_datapoint)
         rec.msg.push(toStoredMessage(owned, stored))
         if (replaced) emit(prop, rec, prop)
@@ -191,20 +216,37 @@ function createDataStore ({ maxHistory = 5, onChange, clone = deepClone, now = D
         if (!rec || !Array.isArray(rec.msg)) {
             return
         }
-        const pending = rec.trims || (rec.trims = { calls: 0, kept: [], seen: 0 })
-        for (let i = pending.seen; i < rec.msg.length; i++) {
-            pending.kept.push(rec.msg[i])
-        }
-        pending.seen = rec.msg.length
-        const before = pending.kept.length
-        pending.kept = trim(pending.kept)
-        if (pending.kept.length === before) {
+        const msgs = trim(rec.msg)
+        if (msgs.length === rec.msg.length) {
             return
         }
-        pending.calls++
-        if (pending.calls >= TRIM_BATCH) {
-            records[SET_SERIES](prop, trimmedSeries(rec))
+        const points = rec.plainPoints
+        const removedMessages = rec.msg.length - msgs.length
+        if (msgs.length === 0 || msgs[0] === rec.msg[removedMessages]) {
+            let removedPoints = 0
+            for (let i = 0; i < removedMessages; i++) {
+                removedPoints += pointCount(rec.msg[i]._datapoint)
+            }
+            points.splice(0, removedPoints)
+        } else {
+            let readAt = 0
+            let writeAt = 0
+            let nextKept = 0
+            for (const m of rec.msg) {
+                const n = pointCount(m._datapoint)
+                if (m === msgs[nextKept]) {
+                    nextKept++
+                    for (let k = 0; k < n; k++) {
+                        points[writeAt++] = points[readAt + k]
+                    }
+                }
+                readAt += n
+            }
+            points.length = writeAt
         }
+        writeValue(prop, points, true)
+        rec.msg = msgs
+        emit(prop, rec, prop)
     }
 
     const handler = {
@@ -216,7 +258,7 @@ function createDataStore ({ maxHistory = 5, onChange, clone = deepClone, now = D
             if (prop.startsWith('$')) {
                 const rec = target[prop.slice(1)]
                 if (!rec) return undefined
-                const msg = msgOf(rec)
+                const msg = Array.isArray(rec.msg) ? messagesWithLivePoints(rec) : rec.msg
                 return Object.freeze({ value: rec.value, msg, quality: rec.quality, timestamp: rec.timestamp, history: Object.freeze(rec.history.slice()) })
             }
             const rec = target[prop]
@@ -238,7 +280,10 @@ function createDataStore ({ maxHistory = 5, onChange, clone = deepClone, now = D
             const rec = target[prop]
             const existed = prop in target
             const ok = delete target[prop]
-            if (rec) rec.value = undefined
+            if (rec) {
+                rec.value = undefined
+                rec.plainPoints = undefined
+            }
             if (existed) emit(prop, undefined, prop)
             return ok
         }

test/store/data.spec.js

@@ -361,7 +361,7 @@ describe('store: data.js chart writes', function () {
         store['$chart-clear'].msg.should.eql([])
     })
 
-    it('does not rebuild the stored series until the batch is reached', function () {
+    it('trims the stored series on every filter', function () {
         const global = storeGlobal()
         const node = fakeNode('chart-3')
         for (let i = 0; i < 100; i++) {
@@ -373,11 +373,11 @@ describe('store: data.js chart writes', function () {
         }
 
         liveStore(global, 'canary-3').should.equal('live')
-        global.get('dashboardStore')['chart-3'].should.have.length(100)
+        global.get('dashboardStore')['chart-3'].should.have.length(41)
         datastore.get('chart-3').should.have.length(41)
     })
 
-    it('rebuilds the stored series on the batch boundary', function () {
+    it('trims value and messages together', function () {
         const global = storeGlobal()
         const node = fakeNode('chart-batch')
         for (let i = 0; i < 100; i++) {

test/store/differential.spec.js

@@ -54,6 +54,16 @@ function storeHolds (global, id) {
     return store['$' + id]?.msg
 }
 
+function seriesInStep (global, id) {
+    const store = global.get('dashboardStore')
+    const msgs = store['$' + id]?.msg
+    if (!Array.isArray(msgs)) {
+        return
+    }
+    const points = msgs.flatMap((m) => (m._datapoint === undefined || m._datapoint === null) ? [] : [].concat(m._datapoint))
+    should(store[id]).eql(points, `value of ${id} matches its messages' points`)
+}
+
 function compareAfterEachStep (steps, ids, global = fakeGlobal()) {
     const warnings = []
     const baseNode = { warn: (m) => warnings.push(m) }
@@ -76,6 +86,7 @@ function compareAfterEachStep (steps, ids, global = fakeGlobal()) {
         datastore.has(step.id).should.equal(legacy.has(step.id), `${step.id} after ${step.op}`)
         if (shared) {
             should(storeHolds(global, step.id)).eql(legacy.get(step.id), `store for ${step.id} after ${step.op}`)
+            seriesInStep(global, step.id)
         }
         if (legacy.has(step.id)) {
             comparedWithData++
@@ -85,6 +96,7 @@ function compareAfterEachStep (steps, ids, global = fakeGlobal()) {
         should(datastore.get(id)).eql(legacy.get(id), id)
         if (shared) {
             should(storeHolds(global, id)).eql(legacy.get(id), `store for ${id}`)
+            seriesInStep(global, id)
         }
     }
     return { comparedWithData, warnings, shared }
@@ -103,6 +115,7 @@ function compareAtEnd (steps, ids) {
     const withData = ids.filter((id) => legacy.has(id))
     for (const id of ids) {
         should(storeHolds(global, id)).eql(legacy.get(id), `store for ${id}`)
+        seriesInStep(global, id)
     }
     return { withData, warnings }
 }