From 5b6d45e25ead489d82b42d057c946e3eefe56643 Mon Sep 17 00:00:00 2001 From: l5y <220195275+l5yth@users.noreply.github.com> Date: Wed, 22 Jul 2026 16:14:17 +0200 Subject: [PATCH] web: node table keeps latest telemetry of every type (per-field merge) (#847) --- ACCEPTANCE.md | 54 +++++++++ .../app/__tests__/snapshot-aggregator.test.js | 104 +++++++++++++++++- web/public/assets/js/app/main.js | 4 +- .../assets/js/app/snapshot-aggregator.js | 84 +++++++++++++- 4 files changed, 238 insertions(+), 8 deletions(-) diff --git a/ACCEPTANCE.md b/ACCEPTANCE.md index a4a2519..f6a21e2 100644 --- a/ACCEPTANCE.md +++ b/ACCEPTANCE.md @@ -2857,3 +2857,57 @@ differ). Still green unchanged: **HT-A4 / A5 / A6 / A7** (fallback ladder, one shared factory on both maps, no attribution, apex/contract untouched), **A1** (no broker — the basemap hosts are raster CDNs), **B1** (all suites), and **B4** (exact Apache header on the new `basemap-blend.test.js`). + +--- + +## Bugfix: Node-table telemetry hidden by newer packets of another type + +Meshtastic telemetry is a protobuf `oneof` — each packet carries exactly one +metric family (device / environment / power / air-quality; +`data/mesh_ingestor/handlers/telemetry.py`). The node table's environment +columns exist only through the client-side per-node telemetry merge +(`aggregateTelemetrySnapshots` → `mergeTelemetryIntoNodes`), which merged a +fixed `SNAPSHOT_WINDOW = 7` packet window: seven newer device/power packets +evicted the last environment packet wholesale, hiding temperature / humidity / +pressure (and, on the node detail page, IAQ etc.) although the rows were still +in the accumulator and the DB. Selection and precedence were also array-order +driven (first-7-encountered, position-0 wins), which is wrong for warm +IndexedDB cache seeds (key order) and incremental `mergeById` appends — stale +values could beat fresh ones. Fix: `aggregateTelemetrySnapshots` now performs a +**per-field latest-non-null merge** — each field takes the value from the +node's newest packet (by `rx_time`, falling back to `telemetry_time`) that +carries it non-null, order-independently, bounded by the caller's existing +7-day accumulator window instead of a packet count. A null/absent field never +clears an older valid value. Frontend read-side only — no API/DB/ingestor +change; apex (I) and privacy (II) untouched; protocol-neutral (IV). The raw +accumulators stay raw (CL-A1/bugfix A1 unchanged). + +### TM-A1 — per-field latest-non-null telemetry merge +```bash +( cd web && node --test public/assets/js/app/__tests__/snapshot-aggregator.test.js ) +``` +**Expected:** pass. With one environment packet followed by more than +`SNAPSHOT_WINDOW` newer device/power packets for the same node, the aggregate +retains the environment metrics (temperature / humidity / pressure) alongside +the newest device metrics; the newest non-null value per field wins regardless +of input array order (inputs that differ only in order produce identical +aggregates whenever timestamps differ; an equal-timestamp conflict resolves +deterministically to the row later in the input); a null/absent field never +overwrites an older valid value; the hidden `snapshots` history is +chronological and `latestSnapshot` is the newest packet by timestamp, not by +array position. + +### TM-R1 — Regression: prior acceptance still holds +```bash +( cd web && npm test ) && ( cd web && bundle exec rspec ) +``` +**Expected:** every prior check still passes. At risk and explicitly required +to remain green: **CL-A1** (the Log's raw-accumulator retention — +`main-log-snapshot-retention.test.js` — the fix changes only the aggregated +locals, never the accumulators), the node detail page and chart suites +(`node-details.test.js`, node-page chart tests — the aggregate keeps its +`snapshots` / `latestSnapshot` shape), and `data-merge.test.js` +(`mergeTelemetryIntoNodes` consumes one aggregate per node unchanged). Node / +position / neighbor aggregation keep their existing `SNAPSHOT_WINDOW` +semantics — only telemetry aggregation changes. No Ruby/Python surface is +touched (**C2** and the Python suite unaffected). diff --git a/web/public/assets/js/app/__tests__/snapshot-aggregator.test.js b/web/public/assets/js/app/__tests__/snapshot-aggregator.test.js index 4a603b2..0fcd113 100644 --- a/web/public/assets/js/app/__tests__/snapshot-aggregator.test.js +++ b/web/public/assets/js/app/__tests__/snapshot-aggregator.test.js @@ -80,7 +80,7 @@ test('aggregateNodeSnapshots reconciles identifiers and fills missing values', ( assert.equal(node.snapshots.length, 3); }); -test('aggregateTelemetrySnapshots and aggregatePositionSnapshots mirror node aggregation', () => { +test('aggregateTelemetrySnapshots and aggregatePositionSnapshots retain older non-null fields', () => { const telemetryEntries = [ { node_id: SAMPLE_NODE_ID, node_num: 5, temperature: null, rx_time: 20 }, { node_num: 5, temperature: 21.5, humidity: 52, rx_time: 10 }, @@ -89,7 +89,7 @@ test('aggregateTelemetrySnapshots and aggregatePositionSnapshots mirror node agg { node_id: SAMPLE_NODE_ID, node_num: 5, longitude: 13.4, rx_time: 25 }, { node_num: 5, latitude: 52.5, rx_time: 15 }, ]; - const telemetryAggregated = aggregateTelemetrySnapshots(telemetryEntries, { limit: 3 }); + const telemetryAggregated = aggregateTelemetrySnapshots(telemetryEntries); const positionAggregated = aggregatePositionSnapshots(positionEntries, { limit: 3 }); assert.equal(telemetryAggregated.length, 1); assert.equal(positionAggregated.length, 1); @@ -119,3 +119,103 @@ test('aggregateSnapshots returns an empty array when no entries are provided', ( assert.deepEqual(aggregateSnapshots(null, { keySelector: () => 'noop' }), []); assert.deepEqual(aggregateNodeSnapshots([], {}), []); }); + +// Regression: hidden telemetry in the node table. Meshtastic telemetry is a +// protobuf oneof — each packet carries exactly one metric family — so the +// aggregate must keep the newest non-null value per field across families +// instead of merging a fixed packet-count window (TM-A1 in ACCEPTANCE.md). + +/** + * Build a newest-first packet stream: one environment packet followed by + * ``newerCount`` newer device/power packets for the same node. + * + * @param {number} newerCount Number of non-environment packets newer than the + * environment packet. + * @returns {Array} Telemetry entries ordered newest-first. + */ +function environmentThenNewerPackets(newerCount) { + const entries = []; + for (let i = newerCount; i >= 1; i -= 1) { + entries.push({ + id: 9000 + i, + node_id: SAMPLE_NODE_ID, + rx_time: 1000 + i * 60, + telemetry_type: i % 2 ? 'device' : 'power', + // Power packets carry no extracted metric columns at all; device packets + // carry the device family only. + ...(i % 2 ? { battery_level: 80, voltage: 4.05 } : {}), + }); + } + entries.push({ + id: 9000, + node_id: SAMPLE_NODE_ID, + rx_time: 1000, + telemetry_type: 'environment', + temperature: 21.5, + relative_humidity: 40.2, + barometric_pressure: 1013.2, + }); + return entries; +} + +test('aggregateTelemetrySnapshots keeps environment metrics past SNAPSHOT_WINDOW newer packets of other types', () => { + const aggregated = aggregateTelemetrySnapshots(environmentThenNewerPackets(SNAPSHOT_WINDOW + 1)); + assert.equal(aggregated.length, 1); + const record = aggregated[0]; + // Latest known environment readings survive alongside the newest device values. + assert.equal(record.temperature, 21.5); + assert.equal(record.relative_humidity, 40.2); + assert.equal(record.barometric_pressure, 1013.2); + assert.equal(record.battery_level, 80); + assert.equal(record.voltage, 4.05); +}); + +test('aggregateTelemetrySnapshots prefers the newest non-null value per field regardless of input order', () => { + // Append/IndexedDB-seed order: the older packet sits first in the array, the + // way a warm cache seed (key order) or an incremental mergeById append holds it. + const older = { id: 3, node_id: SAMPLE_NODE_ID, rx_time: 1000, telemetry_type: 'device', battery_level: 20, voltage: 3.2 }; + const newer = { id: 9, node_id: SAMPLE_NODE_ID, rx_time: 2000, telemetry_type: 'device', battery_level: 90, voltage: 4.1 }; + for (const entries of [[older, newer], [newer, older]]) { + const aggregated = aggregateTelemetrySnapshots(entries); + assert.equal(aggregated.length, 1); + const record = aggregated[0]; + assert.equal(record.battery_level, 90); + assert.equal(record.voltage, 4.1); + // The hidden history stays chronological and the latest snapshot is the + // newest packet by timestamp, not by array position. + assert.deepEqual(record.snapshots.map(s => s.rx_time), [1000, 2000]); + assert.equal(record.latestSnapshot.rx_time, 2000); + } +}); + +test('aggregateTelemetrySnapshots resolves equal timestamps to the row later in the input', () => { + // Same-second packets are equally fresh; the deterministic tie rule is that + // the row appearing later in the input (a fresher append) wins the conflict. + const tied = aggregateTelemetrySnapshots([ + { id: 1, node_id: SAMPLE_NODE_ID, rx_time: 1000, battery_level: 20 }, + { id: 2, node_id: SAMPLE_NODE_ID, rx_time: 1000, battery_level: 90 }, + ]); + assert.equal(tied.length, 1); + assert.equal(tied[0].battery_level, 90); + assert.equal(tied[0].latestSnapshot.id, 2); + // Two untimestamped rows tie at -Infinity: the explicit comparator keeps the + // ordering deterministic (no NaN from subtracting infinities) and the later + // row still wins. + const untimed = aggregateTelemetrySnapshots([ + { id: 3, node_id: SAMPLE_NODE_ID, voltage: 3.2 }, + { id: 4, node_id: SAMPLE_NODE_ID, voltage: 4.1 }, + ]); + assert.equal(untimed.length, 1); + assert.equal(untimed[0].voltage, 4.1); + assert.deepEqual(untimed[0].snapshots.map(s => s.id), [3, 4]); +}); + +test('aggregateTelemetrySnapshots never overwrites a valid value with an empty field', () => { + const aggregated = aggregateTelemetrySnapshots([ + { id: 2, node_id: SAMPLE_NODE_ID, rx_time: 2000, telemetry_type: 'environment', temperature: 22.1, relative_humidity: null }, + { id: 1, node_id: SAMPLE_NODE_ID, rx_time: 1000, telemetry_type: 'environment', temperature: 21.5, relative_humidity: 40.2 }, + ]); + assert.equal(aggregated.length, 1); + assert.equal(aggregated[0].temperature, 22.1); + assert.equal(aggregated[0].relative_humidity, 40.2); +}); diff --git a/web/public/assets/js/app/main.js b/web/public/assets/js/app/main.js index 77c9cfc..80a618e 100644 --- a/web/public/assets/js/app/main.js +++ b/web/public/assets/js/app/main.js @@ -3526,7 +3526,9 @@ export function initializeApp(config) { // the non-enumerable ``snapshots`` history) and merges oldest-last (pinning to // the stalest reading), collapsing each node's history to {stale-first, newest} // so a telemetry/position Log entry flashes in and vanishes on the next tick. - // Keeping the accumulators raw gives every packet a stable, id-keyed Log entry + // (Telemetry now aggregates per-field by timestamp — TM-A1 — but re-storing + // its aggregate would still collapse the Log's per-packet history.) Keeping + // the accumulators raw gives every packet a stable, id-keyed Log entry // (bugfix A1). } diff --git a/web/public/assets/js/app/snapshot-aggregator.js b/web/public/assets/js/app/snapshot-aggregator.js index c7ec800..11e8c6d 100644 --- a/web/public/assets/js/app/snapshot-aggregator.js +++ b/web/public/assets/js/app/snapshot-aggregator.js @@ -210,15 +210,89 @@ export function aggregateNodeSnapshots(entries, { limit = SNAPSHOT_WINDOW } = {} } /** - * Aggregate telemetry packets for each node. + * Resolve the timestamp used to order a telemetry packet. + * + * ``rx_time`` is authoritative (stamped on every stored packet); the + * self-reported ``telemetry_time`` is the fallback. Entries carrying neither + * are treated as infinitely old so they only ever fill fields that no + * timestamped packet provides. + * + * @param {Object} entry Telemetry payload. + * @returns {number} Unix timestamp in seconds, or ``-Infinity`` when absent. + */ +function resolveTelemetryTimestamp(entry) { + const rxTime = normaliseNum(entry.rx_time ?? entry.rxTime); + if (rxTime != null) return rxTime; + const telemetryTime = normaliseNum(entry.telemetry_time ?? entry.telemetryTime); + if (telemetryTime != null) return telemetryTime; + return Number.NEGATIVE_INFINITY; +} + +/** + * Aggregate telemetry packets for each node with per-field latest-non-null + * semantics. + * + * Meshtastic telemetry is a protobuf ``oneof``: each packet carries exactly + * one metric family (device / environment / power / air-quality), so a + * packet-count window can evict one family wholesale when another family + * transmits more often. Instead of merging a bounded window, every field of + * the aggregate takes the value from the node's newest packet (per + * {@link resolveTelemetryTimestamp}) that carries it non-null. A null/absent + * field never overwrites an older valid value, and the result is independent + * of input order (warm cache seeds arrive in IndexedDB key order and + * incremental refreshes append, so the caller's array is not reliably + * newest-first). The retained set is bounded by the caller's accumulator + * window (7 days), not a packet count. + * + * Each aggregate exposes the same non-enumerable metadata as + * {@link aggregateSnapshots}: ``snapshots`` (the node's packets in + * chronological order, ties broken by input order; read-only references — do + * not mutate) and ``latestSnapshot`` (the newest packet by timestamp). * * @param {Array} entries Telemetry payloads. - * @param {{ limit?: number }} [options] Aggregation options. - * @returns {Array} Aggregated telemetry data. + * @returns {Array} Aggregated telemetry data, one entry per node. */ -export function aggregateTelemetrySnapshots(entries, { limit = SNAPSHOT_WINDOW } = {}) { +export function aggregateTelemetrySnapshots(entries) { + if (!Array.isArray(entries) || entries.length === 0) { + return []; + } const resolveKey = createNodeKeyResolver(); - return aggregateSnapshots(entries, { keySelector: resolveKey, limit }); + const groups = new Map(); + for (const entry of entries) { + if (!isObject(entry)) continue; + const key = resolveKey(entry); + if (!key) continue; + const group = groups.get(key); + if (group) { + group.push(entry); + } else { + groups.set(key, [entry]); + } + } + const aggregates = []; + for (const group of groups.values()) { + // Sort ascending by timestamp, preserving input order on ties so a row + // appended later (a fresher delta with the same rx second) wins. + const orderedSnapshots = group + .map((snapshot, index) => ({ snapshot, timestamp: resolveTelemetryTimestamp(snapshot), index })) + .sort((a, b) => { + // Explicit compare: subtraction yields NaN for two -Infinity stamps. + if (a.timestamp < b.timestamp) return -1; + if (a.timestamp > b.timestamp) return 1; + return a.index - b.index; + }) + .map(item => item.snapshot); + // Folding the null-skipping merge oldest→newest leaves each field holding + // the newest non-null value across all of the node's packets. + const target = {}; + for (const snapshot of orderedSnapshots) { + mergeSnapshotFields(target, snapshot); + } + defineHiddenProperty(target, 'snapshots', orderedSnapshots); + defineHiddenProperty(target, 'latestSnapshot', orderedSnapshots[orderedSnapshots.length - 1] ?? null); + aggregates.push(target); + } + return aggregates; } /**