Fix MHI room-temp race: serialize per-unit DB writes
handleMhiStatus() rewrites the whole cached row from an in-memory cumulative state object on every single signal message, via an unawaited pool.query() per message. When several retained signals for the same unit arrive in a tight burst (e.g. right after a broker reconnect), their DB writes can complete out of dispatch order — an older write, captured before a later tempAmbientSts update was known, could land in Postgres AFTER a newer write that already had the correct room temp, silently reverting it to stale/null. User reported AC room temp "randomly" missing from the dashboard while the correct value was visible in the broker's retained topic — exactly this symptom, not an MQTT delivery gap. Fixed by chaining each unit's writes through a per-externalRef promise queue (mhiWriteChain) so completion order can never diverge from dispatch order. state is read at actual write time, not snapshotted early, so a queued write still picks up whatever's freshest by the time its turn comes up.
This commit is contained in:
parent
d818bd1e7a
commit
59a3e091d8
1 changed files with 48 additions and 25 deletions
|
|
@ -55,6 +55,19 @@ const MHI_CMD_SPEC = {
|
||||||
// single-signal message can still write a complete-as-known row. Repopulated from
|
// single-signal message can still write a complete-as-known row. Repopulated from
|
||||||
// the broker's retained messages on every (re)connect, so a backend restart self-heals.
|
// the broker's retained messages on every (re)connect, so a backend restart self-heals.
|
||||||
const mhiState = new Map()
|
const mhiState = new Map()
|
||||||
|
// externalRef -> promise chain, serializes this unit's DB writes. Every signal message
|
||||||
|
// rewrites the WHOLE cached row from the in-memory state (see handleMhiStatus below);
|
||||||
|
// without this, several retained messages arriving in a tight burst (e.g. right after a
|
||||||
|
// broker reconnect) can complete their unawaited pool.query() calls out of dispatch
|
||||||
|
// order — an older write, captured before a later signal was known, can then land AFTER
|
||||||
|
// the newer one and silently revert a field (room temp especially) back to stale/null,
|
||||||
|
// even though the in-memory `state` object was already correct. Chaining guarantees
|
||||||
|
// writes for one unit land in the order they were queued, and since `state` is read at
|
||||||
|
// actual execution time (not snapshotted early), a queued write always picks up
|
||||||
|
// whatever's freshest by the time its turn comes up. Found 2026-08-25: user saw AC room
|
||||||
|
// temp "randomly" missing in the UI while the correct value was sitting right there in
|
||||||
|
// the broker's retained topic — this race is the actual cause, not an MQTT delivery gap.
|
||||||
|
const mhiWriteChain = new Map()
|
||||||
|
|
||||||
function mqttBool(v) {
|
function mqttBool(v) {
|
||||||
return v === true || v === 1 || v === '1' || v === 'true'
|
return v === true || v === 1 || v === '1' || v === 'true'
|
||||||
|
|
@ -324,7 +337,20 @@ async function handleMhiStatus(externalRef, iuLabel, signal, payload) {
|
||||||
mhiState.set(externalRef, state)
|
mhiState.set(externalRef, state)
|
||||||
|
|
||||||
const iuIndex = parseInt(String(iuLabel).replace(/\D/g, ''), 10) || null
|
const iuIndex = parseInt(String(iuLabel).replace(/\D/g, ''), 10) || null
|
||||||
try {
|
|
||||||
|
// Chain onto this unit's write queue rather than firing the query directly — see
|
||||||
|
// mhiWriteChain's comment above. `state` is read inside writeMhiState() at whatever
|
||||||
|
// point its turn actually comes up, not snapshotted now, so a queued write is never
|
||||||
|
// stale even if more signals land while it's waiting.
|
||||||
|
const prev = mhiWriteChain.get(externalRef) || Promise.resolve()
|
||||||
|
const next = prev
|
||||||
|
.then(() => writeMhiState(externalRef, iuLabel, iuIndex, state))
|
||||||
|
.catch(err => console.error('[mqtt] failed to record MHI status update:', err.message))
|
||||||
|
mhiWriteChain.set(externalRef, next)
|
||||||
|
await next
|
||||||
|
}
|
||||||
|
|
||||||
|
async function writeMhiState(externalRef, iuLabel, iuIndex, state) {
|
||||||
await pool.query(
|
await pool.query(
|
||||||
`INSERT INTO zone_devices
|
`INSERT INTO zone_devices
|
||||||
(device_type, external_ref, discovered_name, mhi_unit_index,
|
(device_type, external_ref, discovered_name, mhi_unit_index,
|
||||||
|
|
@ -346,9 +372,6 @@ async function handleMhiStatus(externalRef, iuLabel, signal, payload) {
|
||||||
state.fan_speed ?? null, state.health_state,
|
state.fan_speed ?? null, state.health_state,
|
||||||
]
|
]
|
||||||
)
|
)
|
||||||
} catch (err) {
|
|
||||||
console.error('[mqtt] failed to record MHI status update:', err.message)
|
|
||||||
}
|
|
||||||
}
|
}
|
||||||
|
|
||||||
// Publish command(s) to one MHI unit. externalRef is the unit's topic base
|
// Publish command(s) to one MHI unit. externalRef is the unit's topic base
|
||||||
|
|
|
||||||
Loading…
Add table
Add a link
Reference in a new issue