From adf303e757505438a3fc54d54a24c9e9679f654e Mon Sep 17 00:00:00 2001 From: Noley Holland Date: Thu, 22 Oct 2026 12:55:55 -0700 Subject: [PATCH 1/4] Move chart series helpers into series.js --- nodes/store/reactive.js | 55 +++--------------- nodes/store/series.js | 60 +++++++++++++++++++ test/store/series.spec.js | 118 ++++++++++++++++++++++++++++++++++++++ 3 files changed, 185 insertions(+), 48 deletions(-) create mode 100644 nodes/store/series.js create mode 100644 test/store/series.spec.js diff --git a/nodes/store/reactive.js b/nodes/store/reactive.js index 5d4591784..96b85df63 100644 --- a/nodes/store/reactive.js +++ b/nodes/store/reactive.js @@ -1,3 +1,5 @@ +const { pointsOf, storedDatapoint, compactTrimmed, linkLivePoints } = require('./series.js') + const STORE = Symbol.for('@flowfuse/node-red-dashboard/store') const APPEND_ENTRY = Symbol.for('@flowfuse/node-red-dashboard/appendEntry') const SET_SERIES = Symbol.for('@flowfuse/node-red-dashboard/setSeries') @@ -123,21 +125,6 @@ function createDataStore ({ maxHistory = 5, onChange, clone = deepClone, now = D return rec } - 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] if (!rec) { @@ -162,11 +149,6 @@ function createDataStore ({ maxHistory = 5, onChange, clone = deepClone, now = D emit(prop, rec, prop) } - const toStoredDatapoint = (series, datapoint, start, end) => { - if (end === start) return undefined - return Array.isArray(datapoint) ? Object.freeze(series.slice(start, end)) : series[start] - } - const toStoredMessage = (owned, datapoint) => { freezeOwned(owned, '_datapoint') if (datapoint !== undefined) owned._datapoint = datapoint @@ -186,7 +168,7 @@ function createDataStore ({ maxHistory = 5, onChange, clone = deepClone, now = D }) const rec = writeValue(prop, points, true) rec.msg = parts.map(({ msg, owned, datapoint, start, end }) => { - const stored = toStoredDatapoint(points, datapoint, start, end) + const stored = storedDatapoint(points, datapoint, start, end) if (stored === undefined && '_datapoint' in msg) owned._datapoint = datapoint return toStoredMessage(owned, stored) }) @@ -205,7 +187,7 @@ function createDataStore ({ maxHistory = 5, onChange, clone = deepClone, now = D } const start = rec.value.length for (const p of pointsOf(_datapoint)) rec.value.push(p) - const stored = toStoredDatapoint(rec.plainPoints, _datapoint, start, rec.value.length) + const stored = storedDatapoint(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) @@ -220,31 +202,8 @@ function createDataStore ({ maxHistory = 5, onChange, clone = deepClone, now = D if (msgs.length === rec.msg.length) { return } - 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) + compactTrimmed(rec.plainPoints, rec.msg, msgs) + writeValue(prop, rec.plainPoints, true) rec.msg = msgs emit(prop, rec, prop) } @@ -258,7 +217,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 = Array.isArray(rec.msg) ? messagesWithLivePoints(rec) : rec.msg + const msg = Array.isArray(rec.msg) ? linkLivePoints(rec.msg, (i) => rec.value[i]) : 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] diff --git a/nodes/store/series.js b/nodes/store/series.js new file mode 100644 index 000000000..76b33b1f9 --- /dev/null +++ b/nodes/store/series.js @@ -0,0 +1,60 @@ +function pointsOf (datapoint) { + if (datapoint === undefined || datapoint === null) { + return [] + } + return Array.isArray(datapoint) ? datapoint : [datapoint] +} + +function pointCount (datapoint) { + if (datapoint === undefined || datapoint === null) { + return 0 + } + return Array.isArray(datapoint) ? datapoint.length : 1 +} + +function storedDatapoint (points, datapoint, start, end) { + if (end === start) { + return undefined + } + return Array.isArray(datapoint) ? Object.freeze(points.slice(start, end)) : points[start] +} + +function compactTrimmed (points, msgs, kept) { + const removedMessages = msgs.length - kept.length + if (kept.length === 0 || kept[0] === msgs[removedMessages]) { + let removedPoints = 0 + for (let i = 0; i < removedMessages; i++) { + removedPoints += pointCount(msgs[i]._datapoint) + } + points.splice(0, removedPoints) + return + } + let readAt = 0 + let writeAt = 0 + let nextKept = 0 + for (const m of msgs) { + const n = pointCount(m._datapoint) + if (m === kept[nextKept]) { + nextKept++ + for (let k = 0; k < n; k++) { + points[writeAt++] = points[readAt + k] + } + } + readAt += n + } + points.length = writeAt +} + +function linkLivePoints (msgs, pointAt) { + let i = 0 + return Object.freeze(msgs.map((m) => { + const n = pointCount(m._datapoint) + if (n === 0) { + return m + } + const livePoints = Array.from({ length: n }, () => pointAt(i++)) + return Object.freeze({ ...m, _datapoint: Array.isArray(m._datapoint) ? Object.freeze(livePoints) : livePoints[0] }) + })) +} + +module.exports = { pointsOf, pointCount, storedDatapoint, compactTrimmed, linkLivePoints } diff --git a/test/store/series.spec.js b/test/store/series.spec.js new file mode 100644 index 000000000..d23ae9e93 --- /dev/null +++ b/test/store/series.spec.js @@ -0,0 +1,118 @@ +const should = require('should') // eslint-disable-line no-unused-vars + +const { pointsOf, pointCount, storedDatapoint, compactTrimmed, linkLivePoints } = require('../../nodes/store/series.js') + +const pt = (x, category = 'a') => ({ category, x, y: x }) +const msg = (payload, datapoint) => datapoint === undefined ? { payload } : { payload, _datapoint: datapoint } + +describe('store: series', function () { + describe('pointsOf and pointCount', function () { + it('treats a missing datapoint as no points', function () { + pointsOf(undefined).should.eql([]) + pointsOf(null).should.eql([]) + pointCount(undefined).should.equal(0) + pointCount(null).should.equal(0) + }) + + it('treats an object as one point', function () { + const p = pt(1) + + pointsOf(p).should.eql([p]) + pointCount(p).should.equal(1) + }) + + it('treats an array as its points', function () { + const points = [pt(1), pt(2, 'b')] + + pointsOf(points).should.equal(points) + pointCount(points).should.equal(2) + pointCount([]).should.equal(0) + }) + }) + + describe('storedDatapoint', function () { + const points = [pt(0), pt(1), pt(2)] + + it('stores nothing for an empty range', function () { + should(storedDatapoint(points, pt(9), 1, 1)).be.undefined() + }) + + it('links the point for a single datapoint', function () { + storedDatapoint(points, pt(9), 1, 2).should.equal(points[1]) + }) + + it('links a frozen slice for an array datapoint', function () { + const stored = storedDatapoint(points, [pt(8), pt(9)], 1, 3) + + stored.should.eql([points[1], points[2]]) + stored[0].should.equal(points[1]) + Object.isFrozen(stored).should.equal(true) + }) + }) + + describe('compactTrimmed', function () { + it('removes the oldest points when only the head of the messages went', function () { + const points = [pt(0), pt(1), pt(2), pt(3)] + const kept = points.slice(2) + const msgs = points.map((p, i) => msg(i, p)) + + compactTrimmed(points, msgs, msgs.slice(2)) + + points.should.eql(kept) + points[0].should.equal(kept[0]) + }) + + it('keeps exactly the points of the kept messages from the middle', function () { + const points = [pt(0), pt(1), pt(2), pt(3)] + const msgs = points.map((p, i) => msg(i, p)) + + compactTrimmed(points, msgs, [msgs[1], msgs[3]]) + + points.map((p) => p.x).should.eql([1, 3]) + }) + + it('counts every point of a multi-point message and none for a message without points', function () { + const points = [pt(0), pt(0, 'b'), pt(1), pt(2), pt(2, 'b')] + const msgs = [msg(0, [points[0], points[1]]), msg('none'), msg(1, points[2]), msg(2, [points[3], points[4]])] + + compactTrimmed(points, msgs, msgs.slice(1)) + points.map((p) => p.x).should.eql([1, 2, 2]) + + const after = msgs.slice(1) + compactTrimmed(points, after, [after[0], after[2]]) + points.map((p) => p.x).should.eql([2, 2]) + }) + }) + + describe('linkLivePoints', function () { + it('pairs each message with the points at its positions', function () { + const msgs = [msg(0, pt(0)), msg('none'), msg(1, [pt(1), pt(1, 'b')])] + const live = ['L0', 'L1', 'L2'] + + const linked = linkLivePoints(msgs, (i) => live[i]) + + linked[0]._datapoint.should.equal('L0') + linked[1].should.equal(msgs[1]) + linked[2]._datapoint.should.eql(['L1', 'L2']) + }) + + it('freezes the list, each linked message and a multi-point list', function () { + const msgs = [msg(0, pt(0)), msg(1, [pt(1), pt(1, 'b')])] + + const linked = linkLivePoints(msgs, (i) => ({ i })) + + Object.isFrozen(linked).should.equal(true) + Object.isFrozen(linked[0]).should.equal(true) + Object.isFrozen(linked[1]._datapoint).should.equal(true) + }) + + it('leaves the stored messages unchanged', function () { + const p = pt(0) + const msgs = [msg(0, p)] + + linkLivePoints(msgs, () => 'live') + + msgs[0]._datapoint.should.equal(p) + }) + }) +}) From 95bb632bce54c6e52ec28d2144ac4f7ea5a7bfa4 Mon Sep 17 00:00:00 2001 From: Noley Holland Date: Thu, 22 Oct 2026 13:01:38 -0700 Subject: [PATCH 2/4] Move the generic proxy and freeze helpers into utils.js --- nodes/store/reactive.js | 75 +---------------------------------------- nodes/store/utils.js | 75 +++++++++++++++++++++++++++++++++++++++++ 2 files changed, 76 insertions(+), 74 deletions(-) create mode 100644 nodes/store/utils.js diff --git a/nodes/store/reactive.js b/nodes/store/reactive.js index 96b85df63..1cff93f1c 100644 --- a/nodes/store/reactive.js +++ b/nodes/store/reactive.js @@ -1,4 +1,5 @@ const { pointsOf, storedDatapoint, compactTrimmed, linkLivePoints } = require('./series.js') +const { isReactable, deepClone, deepReactive, deepFreeze, freezeOwned } = require('./utils.js') const STORE = Symbol.for('@flowfuse/node-red-dashboard/store') const APPEND_ENTRY = Symbol.for('@flowfuse/node-red-dashboard/appendEntry') @@ -19,80 +20,6 @@ class Entry { } } -// only plain objects and arrays are deep-watched; Buffers, Dates, class instances etc. are opaque leaves -function isReactable (v) { - if (Array.isArray(v)) return true - if (v === null || typeof v !== 'object') return false - const proto = Object.getPrototypeOf(v) - return proto === null || Object.getPrototypeOf(proto) === null -} - -// reads through proxies (structuredClone rejects them); production injects RED.util.cloneMessage -function deepClone (v, seen = new WeakMap()) { - if (v === null || typeof v !== 'object') return v - if (Buffer.isBuffer(v)) return Buffer.from(v) - if (v instanceof Date) return new Date(v) - if (seen.has(v)) return seen.get(v) - const out = Array.isArray(v) ? [] : {} - seen.set(v, out) - for (const k of Object.keys(v)) out[k] = deepClone(v[k], seen) - return out -} - -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 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] = 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 - }, - deleteProperty (t, p) { - const existed = p in t - const ok = delete t[p] - 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) - if (!wrapOnRead) { - for (const k of Object.keys(value)) value[k] = deepReactive(value[k], notify, `${path}.${k}`, clone, seen) - } - return proxy -} - -function deepFreeze (v) { - if (!isReactable(v) || Object.isFrozen(v)) return v - Object.freeze(v) - for (const k of Object.keys(v)) deepFreeze(v[k]) - return v -} - -function freezeOwned (owned, except) { - for (const k of Object.keys(owned)) { - if (k === except || k === 'req' || k === 'res') continue - deepFreeze(owned[k]) - } - return owned -} - function createDataStore ({ maxHistory = 5, onChange, clone = deepClone, now = Date.now } = {}) { const cloneValue = (v) => (v && typeof v === 'object') ? clone(v) : v const emit = (prop, rec, path) => { try { onChange?.(prop, rec, path) } catch (err) {} } diff --git a/nodes/store/utils.js b/nodes/store/utils.js new file mode 100644 index 000000000..5ce25f876 --- /dev/null +++ b/nodes/store/utils.js @@ -0,0 +1,75 @@ +// only plain objects and arrays are deep-watched; Buffers, Dates, class instances etc. are opaque leaves +function isReactable (v) { + if (Array.isArray(v)) return true + if (v === null || typeof v !== 'object') return false + const proto = Object.getPrototypeOf(v) + return proto === null || Object.getPrototypeOf(proto) === null +} + +// reads through proxies (structuredClone rejects them); production injects RED.util.cloneMessage +function deepClone (v, seen = new WeakMap()) { + if (v === null || typeof v !== 'object') return v + if (Buffer.isBuffer(v)) return Buffer.from(v) + if (v instanceof Date) return new Date(v) + if (seen.has(v)) return seen.get(v) + const out = Array.isArray(v) ? [] : {} + seen.set(v, out) + for (const k of Object.keys(v)) out[k] = deepClone(v[k], seen) + return out +} + +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 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] = 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 + }, + deleteProperty (t, p) { + const existed = p in t + const ok = delete t[p] + 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) + if (!wrapOnRead) { + for (const k of Object.keys(value)) value[k] = deepReactive(value[k], notify, `${path}.${k}`, clone, seen) + } + return proxy +} + +function deepFreeze (v) { + if (!isReactable(v) || Object.isFrozen(v)) return v + Object.freeze(v) + for (const k of Object.keys(v)) deepFreeze(v[k]) + return v +} + +function freezeOwned (owned, except) { + for (const k of Object.keys(owned)) { + if (k === except || k === 'req' || k === 'res') continue + deepFreeze(owned[k]) + } + return owned +} + +module.exports = { isReactable, deepClone, deepReactive, deepFreeze, freezeOwned } From e14e9643fe4b5a02f49e85a1cc03c7035b994bf0 Mon Sep 17 00:00:00 2001 From: Noley Holland Date: Thu, 22 Oct 2026 13:18:08 -0700 Subject: [PATCH 3/4] Make the chart series a SeriesEntry class --- nodes/store/reactive.js | 111 +++++++++++++++++------------------- nodes/store/series.js | 103 ++++++++++++++++++++++++++++++++- test/store/reactive.spec.js | 58 +++++++++++++++++++ 3 files changed, 211 insertions(+), 61 deletions(-) diff --git a/nodes/store/reactive.js b/nodes/store/reactive.js index 1cff93f1c..b07a2c861 100644 --- a/nodes/store/reactive.js +++ b/nodes/store/reactive.js @@ -1,4 +1,4 @@ -const { pointsOf, storedDatapoint, compactTrimmed, linkLivePoints } = require('./series.js') +const { preparedSeries, SeriesEntry } = require('./series.js') const { isReactable, deepClone, deepReactive, deepFreeze, freezeOwned } = require('./utils.js') const STORE = Symbol.for('@flowfuse/node-red-dashboard/store') @@ -16,7 +16,6 @@ class Entry { this.quality = 'GOOD' this.timestamp = 0 this.history = [] - this.plainPoints = undefined } } @@ -27,31 +26,59 @@ function createDataStore ({ maxHistory = 5, onChange, clone = deepClone, now = D records[STORE] = true + const seriesStore = { clone: cloneValue, emit, now, entryFor: (prop) => records[prop] } + + const snapshot = (rec) => { + // snapshotting a whole array on every write (e.g. a table's rows) would blow up memory + if (Array.isArray(rec.value) || !('payload' in rec.msg)) { + return + } + rec.history.push(Object.freeze({ value: deepFreeze(cloneValue(rec.value)), timestamp: rec.timestamp })) + if (rec.history.length > maxHistory) { + rec.history.shift() + } + } + // returns the record without notifying, so a caller can finish writing it before onChange fires - const writeValue = (prop, value, wrapOnRead = false) => { + const writeValue = (prop, value) => { let rec = records[prop] - if (!rec) { + if (rec instanceof SeriesEntry) { + const history = rec.history + rec = new Entry() + rec.history = history + records[prop] = rec + } else if (rec) { + snapshot(rec) + } else { rec = new Entry() records[prop] = rec - } else if (!Array.isArray(rec.value) && 'payload' in rec.msg) { - // snapshotting a whole array on every write (e.g. a table's rows) would blow up memory - rec.history.push(Object.freeze({ value: deepFreeze(cloneValue(rec.value)), timestamp: rec.timestamp })) - if (rec.history.length > maxHistory) rec.history.shift() } const notify = (path) => { // a replaced value keeps its proxies alive; they must not report into the record any more - if (wrapOnRead ? rec.plainPoints !== value : rec.value !== wrapped) return + if (records[prop] !== rec || rec.value !== wrapped) return rec.timestamp = now() emit(prop, rec, path) } - const wrapped = deepReactive(value, notify, prop, cloneValue, new WeakMap(), wrapOnRead) + const wrapped = deepReactive(value, notify, prop, cloneValue) rec.value = wrapped - rec.plainPoints = wrapOnRead ? value : undefined rec.msg = Object.freeze({ payload: wrapped }) rec.timestamp = now() return rec } + const seriesAt = (prop) => { + const current = records[prop] + if (current instanceof SeriesEntry) { + return current + } + if (current) { + snapshot(current) + } + const entry = new SeriesEntry(prop, seriesStore, current ? current.history : []) + records[prop] = entry + return entry + } + records[ENTRY_MSG] = (prop) => { const rec = records[prop] if (!rec) { @@ -63,9 +90,10 @@ function createDataStore ({ maxHistory = 5, onChange, clone = deepClone, now = D records[MERGE_ENTRY] = (prop, msg) => { const current = records[prop] const { payload, ...incoming } = msg - const previous = Array.isArray(current?.msg) ? freezeOwned(cloneValue({ ...current.msg })) : current?.msg || {} + const isSeries = current instanceof SeriesEntry + const previous = isSeries ? freezeOwned(cloneValue({ ...current.msg })) : current?.msg || {} const { payload: storedPayload, ...stored } = previous - const keepValue = current && !Array.isArray(current.msg) && !('payload' in msg) + const keepValue = current && !isSeries && !('payload' in msg) const owned = freezeOwned(cloneValue(incoming)) const value = keepValue ? undefined : cloneValue(payload) const rec = keepValue ? current : writeValue(prop, value) @@ -76,63 +104,27 @@ function createDataStore ({ maxHistory = 5, onChange, clone = deepClone, now = D emit(prop, rec, prop) } - const toStoredMessage = (owned, datapoint) => { - freezeOwned(owned, '_datapoint') - if (datapoint !== undefined) owned._datapoint = datapoint - else deepFreeze(owned._datapoint) - return Object.freeze(owned) - } - records[SET_SERIES] = (prop, msgs) => { - const points = [] - const parts = msgs.map((msg) => { - const { _datapoint, ...rest } = msg - const owned = cloneValue(rest) - const datapoint = cloneValue(_datapoint) - const start = points.length - for (const p of pointsOf(datapoint)) points.push(p) - return { msg, owned, datapoint, start, end: points.length } - }) - const rec = writeValue(prop, points, true) - rec.msg = parts.map(({ msg, owned, datapoint, start, end }) => { - const stored = storedDatapoint(points, datapoint, start, end) - if (stored === undefined && '_datapoint' in msg) owned._datapoint = datapoint - return toStoredMessage(owned, stored) - }) - emit(prop, rec, prop) + const prepared = preparedSeries(msgs, cloneValue) + seriesAt(prop).replaceAll(prepared) } records[APPEND_ENTRY] = (prop, msg) => { - let rec = records[prop] - const replaced = !!rec && !(Array.isArray(rec.value) && Array.isArray(rec.msg)) - // the proxy clones on push, so points must not be cloned again here const { _datapoint, ...rest } = msg const owned = cloneValue(rest) - if (!rec || replaced) { - rec = writeValue(prop, [], true) - rec.msg = [] + const replaced = prop in records && !(records[prop] instanceof SeriesEntry) + const entry = seriesAt(prop) + entry.append(owned, _datapoint, '_datapoint' in msg) + if (replaced) { + emit(prop, entry, prop) } - const start = rec.value.length - for (const p of pointsOf(_datapoint)) rec.value.push(p) - const stored = storedDatapoint(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) } records[TRIM_SERIES] = (prop, trim) => { const rec = records[prop] - if (!rec || !Array.isArray(rec.msg)) { - return + if (rec instanceof SeriesEntry) { + rec.trim(trim) } - const msgs = trim(rec.msg) - if (msgs.length === rec.msg.length) { - return - } - compactTrimmed(rec.plainPoints, rec.msg, msgs) - writeValue(prop, rec.plainPoints, true) - rec.msg = msgs - emit(prop, rec, prop) } const handler = { @@ -144,7 +136,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 = Array.isArray(rec.msg) ? linkLivePoints(rec.msg, (i) => rec.value[i]) : rec.msg + const msg = rec instanceof SeriesEntry ? rec.linkedMessages() : 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] @@ -168,7 +160,6 @@ function createDataStore ({ maxHistory = 5, onChange, clone = deepClone, now = D const ok = delete target[prop] if (rec) { rec.value = undefined - rec.plainPoints = undefined } if (existed) emit(prop, undefined, prop) return ok diff --git a/nodes/store/series.js b/nodes/store/series.js index 76b33b1f9..2cfc34840 100644 --- a/nodes/store/series.js +++ b/nodes/store/series.js @@ -1,3 +1,5 @@ +const { deepReactive, deepFreeze, freezeOwned } = require('./utils.js') + function pointsOf (datapoint) { if (datapoint === undefined || datapoint === null) { return [] @@ -57,4 +59,103 @@ function linkLivePoints (msgs, pointAt) { })) } -module.exports = { pointsOf, pointCount, storedDatapoint, compactTrimmed, linkLivePoints } +function toStoredMessage (owned, datapoint) { + freezeOwned(owned, '_datapoint') + if (datapoint !== undefined) { + owned._datapoint = datapoint + } else { + deepFreeze(owned._datapoint) + } + return Object.freeze(owned) +} + +// copies every message before the entry changes, so one that can't be copied leaves the store as it was +function preparedSeries (msgs, clone) { + const points = [] + const parts = msgs.map((msg) => { + const { _datapoint, ...rest } = msg + const owned = clone(rest) + const datapoint = clone(_datapoint) + const start = points.length + points.push(...pointsOf(datapoint)) + return { owned, datapoint, hasDatapoint: '_datapoint' in msg, start, end: points.length } + }) + return { points, parts } +} + +class SeriesEntry { + constructor (key, store, history) { + this.key = key + this.store = store + this.points = [] + this.msg = [] + this.quality = 'GOOD' + this.history = history + this.wrap() + } + + wrap () { + const points = this.points + this.value = deepReactive(points, (path) => this.notify(points, path), this.key, this.store.clone, new WeakMap(), true) + this.timestamp = this.store.now() + } + + // a proxy taken before the key was replaced, deleted or re-seeded must not report + isCurrent (points) { + return this.store.entryFor(this.key) === this && this.points === points + } + + notify (points, path) { + if (!this.isCurrent(points)) { + return + } + this.timestamp = this.store.now() + this.store.emit(this.key, this, path) + } + + commit () { + this.wrap() + this.store.emit(this.key, this, this.key) + } + + replaceAll ({ points, parts }) { + this.points = points + this.msg = parts.map(({ owned, datapoint, hasDatapoint, start, end }) => { + const stored = storedDatapoint(points, datapoint, start, end) + if (stored === undefined && hasDatapoint) { + owned._datapoint = datapoint + } + return toStoredMessage(owned, stored) + }) + this.commit() + } + + append (owned, datapoint, hasDatapoint) { + const start = this.points.length + // the proxy clones on push, so points must not be cloned again here + for (const point of pointsOf(datapoint)) { + this.value.push(point) + } + const stored = storedDatapoint(this.points, datapoint, start, this.points.length) + if (stored === undefined && hasDatapoint) { + owned._datapoint = this.store.clone(datapoint) + } + this.msg.push(toStoredMessage(owned, stored)) + } + + trim (keep) { + const kept = keep(this.msg) + if (kept.length === this.msg.length) { + return + } + compactTrimmed(this.points, this.msg, kept) + this.msg = kept + this.commit() + } + + linkedMessages () { + return linkLivePoints(this.msg, (i) => this.value[i]) + } +} + +module.exports = { pointsOf, pointCount, storedDatapoint, compactTrimmed, linkLivePoints, toStoredMessage, preparedSeries, SeriesEntry } diff --git a/test/store/reactive.spec.js b/test/store/reactive.spec.js index 033590586..4b3188eae 100644 --- a/test/store/reactive.spec.js +++ b/test/store/reactive.spec.js @@ -1776,3 +1776,61 @@ describe('store: series points', function () { types.isProxy(seen).should.equal(false) }) }) + +describe('store: series entries', function () { + const pt = (x, category = 'a') => ({ category, x, y: x }) + const built = (p) => ({ payload: p.y, topic: p.category, _datapoint: p }) + const uncopiable = () => makeStore({ clone: (v) => { if (v && v.uncopiable) throw new Error('cannot copy'); return util.cloneMessage(v) } }) + + it('keeps history when a value becomes a series and back', function () { + const { store } = makeStore({ clone: util.cloneMessage }) + store.v = 1 + store[APPEND_ENTRY]('v', { payload: 2, _datapoint: pt(2) }) + + store.v = 3 + store.v = 4 + + store.$v.history.map((h) => h.value).should.eql([1, 3]) + }) + + it('keeps history when a series replaces a value', function () { + const { store } = makeStore({ clone: util.cloneMessage }) + store.v = 1 + + store[SET_SERIES]('v', [built(pt(2))]) + + store.$v.history.map((h) => h.value).should.eql([1]) + }) + + it('stops an old value proxy reporting once the key becomes a series', function () { + const { store, log } = makeStore({ clone: util.cloneMessage }) + store.v = { a: 1 } + const held = store.v + store[SET_SERIES]('v', [built(pt(2))]) + log.length = 0 + + held.a = 2 + + log.should.eql([]) + }) + + it('leaves the series unchanged when an appended message cannot be copied', function () { + const { store } = uncopiable() + store[APPEND_ENTRY]('c', { payload: 0, _datapoint: pt(0) }) + + should(() => store[APPEND_ENTRY]('c', { payload: 1, uncopiable: true, _datapoint: pt(1) })).throw('cannot copy') + + store.c.should.have.length(1) + store[ENTRY_MSG]('c').should.have.length(1) + }) + + it('leaves a value unchanged when a series for its key cannot be copied', function () { + const { store } = uncopiable() + store.v = 1 + + should(() => store[SET_SERIES]('v', [built(pt(2)), { payload: 3, uncopiable: true, _datapoint: pt(3) }])).throw('cannot copy') + + store[ENTRY_MSG]('v').should.eql({ payload: 1 }) + store.$v.history.should.eql([]) + }) +}) From 3b3f6171e56f4d11aabe43f9cb5a72fdd7d44aed Mon Sep 17 00:00:00 2001 From: Noley Holland Date: Thu, 22 Oct 2026 13:42:38 -0700 Subject: [PATCH 4/4] Fix copy order spec, large datapoints, clearer SeriesEntry storeProps --- nodes/store/reactive.js | 4 ++-- nodes/store/series.js | 30 ++++++++++++++++-------------- test/store/reactive.spec.js | 19 ++++++++++++++----- test/store/series.spec.js | 8 ++++---- 4 files changed, 36 insertions(+), 25 deletions(-) diff --git a/nodes/store/reactive.js b/nodes/store/reactive.js index b07a2c861..652f2db5f 100644 --- a/nodes/store/reactive.js +++ b/nodes/store/reactive.js @@ -26,7 +26,7 @@ function createDataStore ({ maxHistory = 5, onChange, clone = deepClone, now = D records[STORE] = true - const seriesStore = { clone: cloneValue, emit, now, entryFor: (prop) => records[prop] } + const storeProps = { cloneValue, emit, now, entryFor: (prop) => records[prop] } const snapshot = (rec) => { // snapshotting a whole array on every write (e.g. a table's rows) would blow up memory @@ -74,7 +74,7 @@ function createDataStore ({ maxHistory = 5, onChange, clone = deepClone, now = D if (current) { snapshot(current) } - const entry = new SeriesEntry(prop, seriesStore, current ? current.history : []) + const entry = new SeriesEntry(prop, storeProps, current ? current.history : []) records[prop] = entry return entry } diff --git a/nodes/store/series.js b/nodes/store/series.js index 2cfc34840..d8ab27a11 100644 --- a/nodes/store/series.js +++ b/nodes/store/series.js @@ -47,15 +47,15 @@ function compactTrimmed (points, msgs, kept) { points.length = writeAt } -function linkLivePoints (msgs, pointAt) { +function linkLivePoints (msgs, livePoints) { let i = 0 return Object.freeze(msgs.map((m) => { const n = pointCount(m._datapoint) if (n === 0) { return m } - const livePoints = Array.from({ length: n }, () => pointAt(i++)) - return Object.freeze({ ...m, _datapoint: Array.isArray(m._datapoint) ? Object.freeze(livePoints) : livePoints[0] }) + const linked = Array.from({ length: n }, () => livePoints[i++]) + return Object.freeze({ ...m, _datapoint: Array.isArray(m._datapoint) ? Object.freeze(linked) : linked[0] }) })) } @@ -77,16 +77,18 @@ function preparedSeries (msgs, clone) { const owned = clone(rest) const datapoint = clone(_datapoint) const start = points.length - points.push(...pointsOf(datapoint)) + for (const point of pointsOf(datapoint)) { + points.push(point) + } return { owned, datapoint, hasDatapoint: '_datapoint' in msg, start, end: points.length } }) return { points, parts } } class SeriesEntry { - constructor (key, store, history) { + constructor (key, storeProps, history) { this.key = key - this.store = store + this.storeProps = storeProps this.points = [] this.msg = [] this.quality = 'GOOD' @@ -96,26 +98,26 @@ class SeriesEntry { wrap () { const points = this.points - this.value = deepReactive(points, (path) => this.notify(points, path), this.key, this.store.clone, new WeakMap(), true) - this.timestamp = this.store.now() + this.value = deepReactive(points, (path) => this.notify(points, path), this.key, this.storeProps.cloneValue, new WeakMap(), true) + this.timestamp = this.storeProps.now() } // a proxy taken before the key was replaced, deleted or re-seeded must not report isCurrent (points) { - return this.store.entryFor(this.key) === this && this.points === points + return this.storeProps.entryFor(this.key) === this && this.points === points } notify (points, path) { if (!this.isCurrent(points)) { return } - this.timestamp = this.store.now() - this.store.emit(this.key, this, path) + this.timestamp = this.storeProps.now() + this.storeProps.emit(this.key, this, path) } commit () { this.wrap() - this.store.emit(this.key, this, this.key) + this.storeProps.emit(this.key, this, this.key) } replaceAll ({ points, parts }) { @@ -138,7 +140,7 @@ class SeriesEntry { } const stored = storedDatapoint(this.points, datapoint, start, this.points.length) if (stored === undefined && hasDatapoint) { - owned._datapoint = this.store.clone(datapoint) + owned._datapoint = this.storeProps.cloneValue(datapoint) } this.msg.push(toStoredMessage(owned, stored)) } @@ -154,7 +156,7 @@ class SeriesEntry { } linkedMessages () { - return linkLivePoints(this.msg, (i) => this.value[i]) + return linkLivePoints(this.msg, this.value) } } diff --git a/test/store/reactive.spec.js b/test/store/reactive.spec.js index 4b3188eae..bed97591a 100644 --- a/test/store/reactive.spec.js +++ b/test/store/reactive.spec.js @@ -1814,14 +1814,23 @@ describe('store: series entries', function () { log.should.eql([]) }) - it('leaves the series unchanged when an appended message cannot be copied', function () { + it('leaves a value unchanged when an appended message cannot be copied', function () { const { store } = uncopiable() - store[APPEND_ENTRY]('c', { payload: 0, _datapoint: pt(0) }) + store.c = 1 - should(() => store[APPEND_ENTRY]('c', { payload: 1, uncopiable: true, _datapoint: pt(1) })).throw('cannot copy') + should(() => store[APPEND_ENTRY]('c', { payload: 2, uncopiable: true, _datapoint: pt(2) })).throw('cannot copy') - store.c.should.have.length(1) - store[ENTRY_MSG]('c').should.have.length(1) + store[ENTRY_MSG]('c').should.eql({ payload: 1 }) + store.$c.history.should.eql([]) + }) + + it('stores a message with a very large number of points', function () { + const { store } = makeStore() + const points = Array.from({ length: 200000 }, (_, i) => pt(i)) + + store[SET_SERIES]('c', [{ payload: 'many', _datapoint: points }]) + + store.c.should.have.length(200000) }) it('leaves a value unchanged when a series for its key cannot be copied', function () { diff --git a/test/store/series.spec.js b/test/store/series.spec.js index d23ae9e93..bcacc2575 100644 --- a/test/store/series.spec.js +++ b/test/store/series.spec.js @@ -85,11 +85,11 @@ describe('store: series', function () { }) describe('linkLivePoints', function () { - it('pairs each message with the points at its positions', function () { + it('pairs each message with the live points at its positions', function () { const msgs = [msg(0, pt(0)), msg('none'), msg(1, [pt(1), pt(1, 'b')])] const live = ['L0', 'L1', 'L2'] - const linked = linkLivePoints(msgs, (i) => live[i]) + const linked = linkLivePoints(msgs, live) linked[0]._datapoint.should.equal('L0') linked[1].should.equal(msgs[1]) @@ -99,7 +99,7 @@ describe('store: series', function () { it('freezes the list, each linked message and a multi-point list', function () { const msgs = [msg(0, pt(0)), msg(1, [pt(1), pt(1, 'b')])] - const linked = linkLivePoints(msgs, (i) => ({ i })) + const linked = linkLivePoints(msgs, [{ i: 0 }, { i: 1 }, { i: 2 }]) Object.isFrozen(linked).should.equal(true) Object.isFrozen(linked[0]).should.equal(true) @@ -110,7 +110,7 @@ describe('store: series', function () { const p = pt(0) const msgs = [msg(0, p)] - linkLivePoints(msgs, () => 'live') + linkLivePoints(msgs, ['live']) msgs[0]._datapoint.should.equal(p) })