Skip to content
11 changes: 6 additions & 5 deletions nodes/config/ui_base.js
Original file line number Diff line number Diff line change
Expand Up @@ -34,6 +34,8 @@ module.exports = function (RED) {
datastore.setConfig(RED)
statestore.setConfig(RED)

let warnedStoreUnavailable = false

/**
* @typedef {import('socket.io').Socket} Socket
* @typedef {import('socket.io').Server} Server
Expand Down Expand Up @@ -395,13 +397,12 @@ module.exports = function (RED) {
onUnavailable: (err) => { unavailable = err },
onReplaced: () => node.warn('global.dashboardStore was replaced by a flow since the last deploy, discarding the stored value of every widget. The store has been re-created. To write your own data, set global.dashboardStore.<key> rather than replacing global.dashboardStore itself.')
})
if (!store) {
datastore.disableStore()
node.warn('Dashboard data store disabled: the global context store doesn\'t hold objects by reference (e.g. cache: false), so live state can\'t work. Use the memory store or localfilesystem with cache: true.' + (unavailable ? ' Reading it failed with: ' + unavailable.message : ''))
if (!store && !warnedStoreUnavailable) {
warnedStoreUnavailable = true
node.warn('Dashboard data store: flows can\'t use global.dashboardStore, because the default context store doesn\'t keep values in memory (e.g. localfilesystem with cache: false). Dashboards still work normally. To use global.dashboardStore from flows, set contextStorage.default to memory, or localfilesystem with cache: true.' + (unavailable ? ' (Reason: ' + unavailable.message + ')' : ''))

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Works nicely. I tried cache: false and the dashboard kept its data through a full redeploy. Minor: the warning shows on every full deploy, which might get a bit noisy for those users.

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

ah good call fixed

}
} catch (err) {
datastore.disableStore()
node.warn('Dashboard data store disabled: could not initialise it in global context (' + err.message + ').')
node.warn('Dashboard data store: flows can\'t use global.dashboardStore, because saving it to global context failed (' + err.message + '). Dashboards still work normally.')
}

n.root = RED.settings.httpNodeRoot || '/'
Expand Down
142 changes: 38 additions & 104 deletions nodes/store/data.js
Original file line number Diff line number Diff line change
@@ -1,23 +1,21 @@
const { isClientScoped } = require('../utils/index.js')

const { attachToContext, MERGE_ENTRY, APPEND_ENTRY, SET_SERIES, TRIM_SERIES } = require('./reactive.js')

const data = {}
const { createDataStore, attachToContext, MERGE_ENTRY, APPEND_ENTRY, SET_SERIES, TRIM_SERIES, ENTRY_MSG } = require('./reactive.js')

const config = {
RED: null
}

let storeOptions = {}
let storeEnabled = false
let fallbackStore = null
let storeContext = null
let storeBase = null
let warned = false

function warnOnce (node, problem) {
if (warned || !node) return
function warnOnce (problem) {
if (warned || !storeBase) return
warned = true
node.warn(`Dashboard data store: ${problem}. Further store failures are not reported until the next full deploy or restart.`)
storeBase.warn(`Dashboard data store: ${problem}. Further store failures are not reported until the next full deploy or restart.`)
}

const attachOptions = () => ({ clone: config.RED.util.cloneMessage, ...storeOptions })
Expand All @@ -29,48 +27,28 @@ function getOrCreateStore (globalContext) {
return store
}

function getStoreOrWarn (action, id) {
try {
return fallbackStore || getOrCreateStore(storeContext)
} catch (err) { warnOnce(`${action} ${id} failed (${err.message})`) }
}

function initStore (globalContext, { node, ...opts } = {}) {
storeOptions = opts
storeEnabled = true
storeContext = globalContext
storeBase = node
warned = false
return attachToContext(globalContext, attachOptions())
}

function disableStore () {
storeEnabled = false
}

function writeToStore (node, msg) {
if (!storeEnabled) return
if (!msg || typeof msg !== 'object') return
try {
getOrCreateStore(node.context().global)[MERGE_ENTRY](node.id, msg)
} catch (err) { warnOnce(node, `a store write failed (${err.message}), so global.dashboardStore may be out of date`) }
}

function appendToStore (node, msg) {
if (!storeEnabled) return
if (!msg || typeof msg !== 'object') return
try {
getOrCreateStore(node.context().global)[APPEND_ENTRY](node.id, msg)
} catch (err) { warnOnce(node, `a store write failed (${err.message}), so global.dashboardStore may be out of date`) }
}

function replaceInStore (node, msgs) {
if (!storeEnabled) return
if (!Array.isArray(msgs) || msgs.some((m) => !m || typeof m !== 'object')) return
try {
getOrCreateStore(node.context().global)[SET_SERIES](node.id, msgs)
} catch (err) { warnOnce(node, `a store write failed (${err.message}), so global.dashboardStore may be out of date`) }
}

function trimInStore (node, trim) {
if (!storeEnabled) return
let store = null
try {
getOrCreateStore(node.context().global)[TRIM_SERIES](node.id, trim)
} catch (err) { warnOnce(node, `a store write failed (${err.message}), so global.dashboardStore may be out of date`) }
store = attachToContext(globalContext, attachOptions())

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Noticed something with a persistent context store (localfilesystem, cache: true). The store only gets written to disk when something else in global context is set, and then only the values go. After a restart my text input came back with its old value, which main never did, and the chart read back as { payload: [points] } instead of a list of msgs (20 points for a "last 5 per topic" chart). The chart showed empty and recovered on the next message, but a widget calling get() would see the odd shape. So whether state survives a restart ends up depending on an unrelated global.set. Might be worth picking one behaviour, either always starting fresh like main or persisting properly.

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Yahh makes sense, went with starting fresh like main for now, which also fixes the chart coming back as { payload: [points] } here.

#2322 also keeps a chart's messages in step with its points for flow writes. I do want state to eventually survive restarts, but it can be settled in the persistence work #2237

} finally {
if (store) {
fallbackStore = null
} else if (!fallbackStore) {
fallbackStore = createDataStore(attachOptions())
}
}
return store
}

function latestPerTopic (msgs, maxPoints) {
Expand Down Expand Up @@ -120,26 +98,17 @@ function canSaveInStore (base, node, msg) {
return checks.length === 0 || !checks.includes(false)
}

// Strip msg of properties that are not needed for storage
function stripMsg (msg) {
const newMsg = config.RED.util.cloneMessage(msg)

// don't need to store ui_updates in the datastore, as this is handled in statestore
delete newMsg.ui_update

return newMsg
}

const getters = {
RED () {
return config.RED
},
// given a widget id, return the latest msg received
msg (id) {
return config.RED.util.cloneMessage(data[id])
return config.RED.util.cloneMessage(getStoreOrWarn('reading', id)?.[ENTRY_MSG](id))
},
has (id) {
return !!data[id]
const store = getStoreOrWarn('reading', id)
return !!store && id in store
}
}

Expand All @@ -150,13 +119,10 @@ const setters = {
},
// remove data associated to a given widget
clear (id) {
delete data[id]
if (!storeEnabled) {
return
const store = getStoreOrWarn('clearing', id)
if (store) {
delete store[id]
}
try {
delete getOrCreateStore(storeContext)[id]
} catch (err) { warnOnce(storeBase, `clearing the store entry for ${id} failed (${err.message}), so global.dashboardStore may still hold its old value`) }
},
/**
*
Expand All @@ -166,35 +132,20 @@ const setters = {
*/
save (base, node, msg) {
if (Array.isArray(msg)) {
/// need to check msg by msg
const filtered = []
for (const m of msg) {
if (canSaveInStore(base, node, m)) {
filtered.push(config.RED.util.cloneMessage(m))
}
}
data[node.id] = filtered
replaceInStore(node, filtered)
} else {
if (canSaveInStore(base, node, msg)) {
const newMsg = stripMsg(msg)
data[node.id] = {
...data[node.id],
...newMsg
}
writeToStore(node, newMsg)
const msgs = msg.filter((m) => canSaveInStore(base, node, m))
if (msgs.every((m) => m && typeof m === 'object')) {
getStoreOrWarn('saving', node.id)?.[SET_SERIES](node.id, msgs)
}
} else if (msg && typeof msg === 'object' && canSaveInStore(base, node, msg)) {
const { ui_update: uiUpdate, ...rest } = msg
getStoreOrWarn('saving', node.id)?.[MERGE_ENTRY](node.id, rest)
}
},
// given a widget id, and msg, store in an array of history of values
// useful for charting widgets
append (base, node, msg) {
if (canSaveInStore(base, node, msg)) {
if (!data[node.id]) {
data[node.id] = []
}
data[node.id].push(config.RED.util.cloneMessage(msg))
appendToStore(node, msg)
if (msg && typeof msg === 'object' && canSaveInStore(base, node, msg)) {
getStoreOrWarn('saving', node.id)?.[APPEND_ENTRY](node.id, msg)
}
},
/**
Expand All @@ -204,28 +155,12 @@ const setters = {
* @param {(msg) => Boolean} filterFunction
*/
filter (base, node, filterFunction) {
const currentData = data[node.id]
if (filterFunction && Array.isArray(currentData) && currentData.length) {
const filteredMessages = currentData.filter(filterFunction)
if (filteredMessages.length !== currentData.length) {
// no need for save operation to process messages - just apply them
data[node.id] = filteredMessages
}
}
if (filterFunction) {
trimInStore(node, (msgs) => msgs.filter(filterFunction))
getStoreOrWarn('trimming', node.id)?.[TRIM_SERIES](node.id, (msgs) => msgs.filter(filterFunction))
}
},
keepLatestPerTopic (base, node, maxPoints) {
trimInStore(node, (msgs) => latestPerTopic(msgs, maxPoints))
const currentData = data[node.id]
if (!Array.isArray(currentData)) {
return
}
const kept = latestPerTopic(currentData, maxPoints)
if (kept.length !== currentData.length) {
data[node.id] = kept
}
getStoreOrWarn('trimming', node.id)?.[TRIM_SERIES](node.id, (msgs) => latestPerTopic(msgs, maxPoints))
}
}

Expand All @@ -239,6 +174,5 @@ module.exports = {
filter: setters.filter,
keepLatestPerTopic: setters.keepLatestPerTopic,
clear: setters.clear,
initStore,
disableStore
initStore
}
28 changes: 18 additions & 10 deletions nodes/store/reactive.js
Original file line number Diff line number Diff line change
Expand Up @@ -3,6 +3,7 @@ const APPEND_ENTRY = Symbol.for('@flowfuse/node-red-dashboard/appendEntry')
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'

Expand Down Expand Up @@ -41,7 +42,6 @@ function deepReactive (value, notify, path, clone, seen = new WeakMap()) {
if (!isReactable(value)) return value
if (seen.has(value)) return seen.get(value)
const proxy = new Proxy(value, {
get (t, p, r) { return Reflect.get(t, p, r) },
set (t, p, v) {
Comment thread
n-lark marked this conversation as resolved.
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
Expand Down Expand Up @@ -112,15 +112,23 @@ function createDataStore ({ maxHistory = 5, onChange, clone = deepClone, now = D
}

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

records[ENTRY_MSG] = (prop) => {
const rec = records[prop]
return rec ? msgOf(rec) : undefined
}

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 { payload: storedPayload, ...stored } = previous
const keepValue = current && !Array.isArray(current.msg) && !('payload' in msg)
const rec = keepValue ? current : writeValue(prop, cloneValue(payload))
const fields = { ...stored, ...freezeOwned(cloneValue(incoming)) }
const owned = freezeOwned(cloneValue(incoming))
const value = keepValue ? undefined : cloneValue(payload)
const rec = keepValue ? current : writeValue(prop, value)
const fields = { ...stored, ...owned }
const hasPayload = 'payload' in msg || (keepValue && 'payload' in current.msg)
rec.msg = Object.freeze(hasPayload ? { ...fields, payload: rec.value } : fields)
rec.timestamp = now()
Expand Down Expand Up @@ -163,13 +171,13 @@ function createDataStore ({ maxHistory = 5, onChange, clone = deepClone, now = D
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, [])
rec.msg = []
}
// the proxy clones on push, so points must not be cloned again here
const { _datapoint, ...rest } = msg
const owned = cloneValue(rest)
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)
Expand Down Expand Up @@ -208,7 +216,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) ? Object.freeze(trimmedSeries(rec)) : rec.msg
const msg = msgOf(rec)
return Object.freeze({ value: rec.value, msg, quality: rec.quality, timestamp: rec.timestamp, history: Object.freeze(rec.history.slice()) })
}
const rec = target[prop]
Expand Down Expand Up @@ -241,7 +249,7 @@ function createDataStore ({ maxHistory = 5, onChange, clone = deepClone, now = D
const injected = new WeakMap()

// Inject the store into global context once. On redeploy the existing store is reused;
// after a restart with a persistent context store, existing plain data is rehydrated.
// after a restart it starts fresh and plain data a flow put in its place is copied in.
// Returns null when the context store can't hold a live proxy: a cache-off store serves
// only async access so a sync get throws, and a serialising store hands back a copy.
function attachToContext (globalContext, opts = {}) {
Expand All @@ -261,7 +269,7 @@ function attachToContext (globalContext, opts = {}) {
const previous = injected.get(globalContext)

const store = createDataStore(opts)
if (isReactable(existing) && !Array.isArray(existing)) {
if (previous && isReactable(existing) && !Array.isArray(existing)) {
for (const [k, v] of Object.entries(existing)) store[k] = v
}
globalContext.set(namespace, store)
Expand All @@ -271,4 +279,4 @@ function attachToContext (globalContext, opts = {}) {
return store
}

module.exports = { createDataStore, attachToContext, NAMESPACE, STORE, APPEND_ENTRY, SET_SERIES, MERGE_ENTRY, TRIM_SERIES }
module.exports = { createDataStore, attachToContext, NAMESPACE, STORE, APPEND_ENTRY, SET_SERIES, MERGE_ENTRY, TRIM_SERIES, ENTRY_MSG }
13 changes: 12 additions & 1 deletion test/nodes/config-nodes/ui_base.spec.js
Original file line number Diff line number Diff line change
Expand Up @@ -153,12 +153,23 @@ describe('ui-base config node: cache-off context store', function () {
await helper.load(nodeImports, flow)

const warned = helper.log().args.filter((args) =>
args[0].level === helper.log().WARN && /hold objects by reference/.test(args[0].msg)
args[0].level === helper.log().WARN && /doesn't keep values in memory/.test(args[0].msg)
)
warned.should.have.length(1)
touched.set.should.equal(0)
})

it('warns once per start, not on every full deploy', async function () {
await helper.load(nodeImports, flow)

await helper.setFlows(flow, 'full')

const warned = helper.log().args.filter((args) =>
args[0].level === helper.log().WARN && /doesn't keep values in memory/.test(args[0].msg)
)
warned.should.have.length(1)
})

it('stops the write path instead of throwing on every message', async function () {
await helper.load(nodeImports, flow)
const text = helper.getNode('node-ui-text')
Expand Down
Loading
Loading