From d5223ff528e7162976a4471e3c55bbb3d69485cd Mon Sep 17 00:00:00 2001 From: gadgethd <111318106+gadgethd@users.noreply.github.com> Date: Tue, 22 Sep 2026 23:21:12 +0000 Subject: [PATCH] fix(mqtt): skip offline status telemetry samples --- backend/src/db/index.ts | 6 +- .../052_node_status_sample_status.sql | 3 + backend/src/db/schema/base.sql | 3 +- backend/src/mqtt/client.ts | 128 +------------- backend/src/mqtt/statusTelemetry.test.ts | 41 +++++ backend/src/mqtt/statusTelemetry.ts | 159 ++++++++++++++++++ 6 files changed, 215 insertions(+), 125 deletions(-) create mode 100644 backend/src/db/migrations/052_node_status_sample_status.sql create mode 100644 backend/src/mqtt/statusTelemetry.test.ts create mode 100644 backend/src/mqtt/statusTelemetry.ts diff --git a/backend/src/db/index.ts b/backend/src/db/index.ts index bdb5878..3402afb 100644 --- a/backend/src/db/index.ts +++ b/backend/src/db/index.ts @@ -709,12 +709,13 @@ export async function insertNodeStatusSample(sample: { channelUtilization?: number | null; airUtilTx?: number | null; stats?: Record | null; + status?: string | null; }): Promise { await pool.query( `WITH sample_insert AS ( INSERT INTO node_status_samples - (time, node_id, network, battery_mv, uptime_secs, tx_air_secs, rx_air_secs, channel_utilization, air_util_tx, stats) - VALUES (NOW(), $1, $2, $3, $4, $5, $6, $7, $8, $9) + (time, node_id, network, battery_mv, uptime_secs, tx_air_secs, rx_air_secs, channel_utilization, air_util_tx, stats, status) + VALUES (NOW(), $1, $2, $3, $4, $5, $6, $7, $8, $9, $10) RETURNING time ) UPDATE nodes @@ -732,6 +733,7 @@ export async function insertNodeStatusSample(sample: { sample.channelUtilization ?? null, sample.airUtilTx ?? null, sample.stats ? JSON.stringify(sample.stats) : null, + sample.status ?? null, ], ); } diff --git a/backend/src/db/migrations/052_node_status_sample_status.sql b/backend/src/db/migrations/052_node_status_sample_status.sql new file mode 100644 index 0000000..c534a9e --- /dev/null +++ b/backend/src/db/migrations/052_node_status_sample_status.sql @@ -0,0 +1,3 @@ +-- Preserve the MQTT status envelope alongside extracted node telemetry. +ALTER TABLE node_status_samples + ADD COLUMN IF NOT EXISTS status TEXT; diff --git a/backend/src/db/schema/base.sql b/backend/src/db/schema/base.sql index 18b8874..bceaa4f 100644 --- a/backend/src/db/schema/base.sql +++ b/backend/src/db/schema/base.sql @@ -229,7 +229,8 @@ CREATE TABLE IF NOT EXISTS node_status_samples ( rx_air_secs BIGINT, channel_utilization DOUBLE PRECISION, air_util_tx DOUBLE PRECISION, - stats JSONB + stats JSONB, + status TEXT ); SELECT create_hypertable('node_status_samples', 'time', if_not_exists => TRUE); diff --git a/backend/src/mqtt/client.ts b/backend/src/mqtt/client.ts index 7d9cfd1..b9b58e3 100644 --- a/backend/src/mqtt/client.ts +++ b/backend/src/mqtt/client.ts @@ -14,6 +14,7 @@ import { shouldDiscardUnverifiedTxAdvert, statusEnvelopeTargetsObserver } from ' import { extractNeighborNodes } from './neighborPayload.js'; import { parseMqttTopic } from './topic.js'; import { createDurableMqttHandleMessage } from './durableHandler.js'; +import { persistStatusTelemetrySample } from './statusTelemetry.js'; import { boundedNetworkMetricLabel, mqttIngestActive, @@ -194,112 +195,6 @@ function isEmptyPacketEnvelope(json: Record, rawHex: string, pa && (payloadLen ?? 0) <= 0; } -function toRecord(value: unknown): Record | undefined { - return value && typeof value === 'object' && !Array.isArray(value) - ? value as Record - : undefined; -} - -function readNum(obj: Record | undefined, ...keys: string[]): number | undefined { - if (!obj) return undefined; - for (const key of keys) { - if (!(key in obj)) continue; - const n = toNum(obj[key]); - if (n != null) return n; - } - return undefined; -} - -type StatusTelemetrySample = { - batteryMv?: number; - uptimeSecs?: number; - txAirSecs?: number; - rxAirSecs?: number; - channelUtilization?: number; - airUtilTx?: number; - stats?: Record; -}; - -const OWNER_STATUS_STATS_KEYS = new Set([ - 'battery_mv', 'solar_mv', 'board_temp_c', 'wifi_rssi', 'wifi_ssid', 'wifi_uptime_ms', - 'ntp_synced', 'ntp_sync_age_ms', 'boot_count', 'reset_reason', 'max_loop_ms', - 'max_loop_at_ms', 'nodes_heard_24h', 'channel_utilization', 'air_util_tx', 'air_util_rx', - 'last_rx_rssi', 'last_rx_snr', 'tx_power_dbm', 'config_version', 'config_crc32', - 'fs_free_bytes', 'fs_total_bytes', 'nvs_free_entries', 'channel_id', 'git_commit', 'boot_epoch', -]); - -function extractStatusTelemetry( - json: Record, - options?: { allowRawStatsOnly?: boolean }, -): StatusTelemetrySample | null { - const stats = toRecord(json['stats']); - const hasRawStats = Boolean(stats && Object.keys(stats).length > 0); - const hasOwnerStats = Boolean( - stats && ( - Object.keys(stats).some((key) => OWNER_STATUS_STATS_KEYS.has(key)) - || toRecord(stats['mqtt']) - ), - ); - const batteryMv = readNum(stats, 'battery_mv', 'batteryMv'); - const uptimeSecs = (() => { - const direct = readNum(stats, 'uptime_secs', 'uptimeSecs'); - if (direct != null) return direct; - const uptimeMs = readNum(stats, 'uptime_ms', 'uptimeMs'); - if (uptimeMs == null) return undefined; - return Math.floor(uptimeMs / 1000); - })(); - const txAirSecs = readNum(stats, 'tx_air_secs', 'txAirSecs'); - const rxAirSecs = readNum(stats, 'rx_air_secs', 'rxAirSecs'); - const channelUtilization = readNum( - stats, - 'channel_utilization', - 'channel_utilization_pct', - 'channel_util', - 'channelUtil', - 'channelUtilization', - ); - const airUtilTx = readNum( - stats, - 'air_util_tx', - 'air_util_tx_pct', - 'tx_air_util', - 'tx_air_utilization', - 'airUtilTx', - ); - - if ( - batteryMv == null - && uptimeSecs == null - && txAirSecs == null - && rxAirSecs == null - && channelUtilization == null - && airUtilTx == null - ) { - if ((options?.allowRawStatsOnly || hasOwnerStats) && hasRawStats) { - return { - batteryMv, - uptimeSecs, - txAirSecs, - rxAirSecs, - channelUtilization, - airUtilTx, - stats, - }; - } - return null; - } - - return { - batteryMv, - uptimeSecs, - txAirSecs, - rxAirSecs, - channelUtilization, - airUtilTx, - stats, - }; -} - /** In-flight pre-resolve tracking — prevents duplicate concurrent resolutions for the same hash/network. */ const preResolveInFlight = new Set(); const preResolveLastStarted = new Map(); @@ -552,22 +447,11 @@ async function handleMessage(topic: string, rawPayload: Buffer): Promise { allowTestOverride: network === 'test' && nodeId === observerKey, mqttObserver: true, })]; - const telemetry = extractStatusTelemetry(json, { - allowRawStatsOnly: network === 'test', - }); - if (telemetry) { - writes.push(insertNodeStatusSample({ - nodeId, - network, - batteryMv: telemetry.batteryMv, - uptimeSecs: telemetry.uptimeSecs, - txAirSecs: telemetry.txAirSecs, - rxAirSecs: telemetry.rxAirSecs, - channelUtilization: telemetry.channelUtilization, - airUtilTx: telemetry.airUtilTx, - stats: telemetry.stats, - })); - } + writes.push(persistStatusTelemetrySample( + json, + { nodeId, network, allowRawStatsOnly: network === 'test' }, + insertNodeStatusSample, + )); const writeResults = await Promise.allSettled(writes); const writeErrors: unknown[] = []; for (const result of writeResults) { diff --git a/backend/src/mqtt/statusTelemetry.test.ts b/backend/src/mqtt/statusTelemetry.test.ts new file mode 100644 index 0000000..a72ef5f --- /dev/null +++ b/backend/src/mqtt/statusTelemetry.test.ts @@ -0,0 +1,41 @@ +import assert from 'node:assert/strict'; +import test from 'node:test'; +import { persistStatusTelemetrySample, type NodeStatusSampleInput } from './statusTelemetry.js'; + +const NODE_ID = 'A'.repeat(64); + +test('offline Will telemetry is skipped, including a retained replay with boot-fresh stats', async () => { + const writes: NodeStatusSampleInput[] = []; + const written = await persistStatusTelemetrySample( + { + status: 'offline', + stats: { battery_mv: '4200', uptime_secs: '0', boot_count: '1' }, + }, + { nodeId: NODE_ID, network: 'ukmesh' }, + async (sample) => { writes.push(sample); }, + ); + + assert.equal(written, false); + assert.deepEqual(writes, []); +}); + +test('real telemetry is stored with its top-level status', async () => { + const writes: NodeStatusSampleInput[] = []; + const written = await persistStatusTelemetrySample( + { + status: 'online', + stats: { battery_mv: '4100', uptime_secs: '7200' }, + }, + { nodeId: NODE_ID, network: 'ukmesh' }, + async (sample) => { writes.push(sample); }, + ); + + assert.equal(written, true); + assert.equal(writes.length, 1); + assert.equal(writes[0]?.status, 'online'); + assert.equal(writes[0]?.nodeId, NODE_ID); + assert.equal(writes[0]?.network, 'ukmesh'); + assert.equal(writes[0]?.batteryMv, 4100); + assert.equal(writes[0]?.uptimeSecs, 7200); + assert.deepEqual(writes[0]?.stats, { battery_mv: '4100', uptime_secs: '7200' }); +}); diff --git a/backend/src/mqtt/statusTelemetry.ts b/backend/src/mqtt/statusTelemetry.ts new file mode 100644 index 0000000..0dfd9b4 --- /dev/null +++ b/backend/src/mqtt/statusTelemetry.ts @@ -0,0 +1,159 @@ +export type StatusTelemetrySample = { + status?: string; + batteryMv?: number; + uptimeSecs?: number; + txAirSecs?: number; + rxAirSecs?: number; + channelUtilization?: number; + airUtilTx?: number; + stats?: Record; +}; + +export type NodeStatusSampleInput = StatusTelemetrySample & { + nodeId: string; + network: string; +}; + +type StatusSampleOptions = { + nodeId: string; + network: string; + allowRawStatsOnly?: boolean; +}; + +type StatusSampleWriter = (sample: NodeStatusSampleInput) => Promise; + +const OWNER_STATUS_STATS_KEYS = new Set([ + 'battery_mv', 'solar_mv', 'board_temp_c', 'wifi_rssi', 'wifi_ssid', 'wifi_uptime_ms', + 'ntp_synced', 'ntp_sync_age_ms', 'boot_count', 'reset_reason', 'max_loop_ms', + 'max_loop_at_ms', 'nodes_heard_24h', 'channel_utilization', 'air_util_tx', 'air_util_rx', + 'last_rx_rssi', 'last_rx_snr', 'tx_power_dbm', 'config_version', 'config_crc32', + 'fs_free_bytes', 'fs_total_bytes', 'nvs_free_entries', 'channel_id', 'git_commit', 'boot_epoch', +]); + +function toRecord(value: unknown): Record | undefined { + return value && typeof value === 'object' && !Array.isArray(value) + ? value as Record + : undefined; +} + +function toNum(value: unknown): number | undefined { + if (typeof value === 'number') return Number.isFinite(value) ? value : undefined; + if (typeof value === 'string' && value.trim() !== '') { + const number = Number(value); + return Number.isFinite(number) ? number : undefined; + } + return undefined; +} + +function readNum(obj: Record | undefined, ...keys: string[]): number | undefined { + if (!obj) return undefined; + for (const key of keys) { + if (!(key in obj)) continue; + const number = toNum(obj[key]); + if (number != null) return number; + } + return undefined; +} + +function envelopeStatus(json: Record): string | undefined { + if (typeof json['status'] !== 'string') return undefined; + const status = json['status'].trim(); + return status || undefined; +} + +export function extractStatusTelemetry( + json: Record, + options?: { allowRawStatsOnly?: boolean }, +): StatusTelemetrySample | null { + const status = envelopeStatus(json); + // The LWT carries boot-fresh statistics. Ignore it on the same path whether + // it arrives live after a session drop or as a retained replay on subscribe. + if (status?.toLowerCase() === 'offline') return null; + + const stats = toRecord(json['stats']); + const hasRawStats = Boolean(stats && Object.keys(stats).length > 0); + const hasOwnerStats = Boolean( + stats && ( + Object.keys(stats).some((key) => OWNER_STATUS_STATS_KEYS.has(key)) + || toRecord(stats['mqtt']) + ), + ); + const batteryMv = readNum(stats, 'battery_mv', 'batteryMv'); + const uptimeSecs = (() => { + const direct = readNum(stats, 'uptime_secs', 'uptimeSecs'); + if (direct != null) return direct; + const uptimeMs = readNum(stats, 'uptime_ms', 'uptimeMs'); + if (uptimeMs == null) return undefined; + return Math.floor(uptimeMs / 1000); + })(); + const txAirSecs = readNum(stats, 'tx_air_secs', 'txAirSecs'); + const rxAirSecs = readNum(stats, 'rx_air_secs', 'rxAirSecs'); + const channelUtilization = readNum( + stats, + 'channel_utilization', + 'channel_utilization_pct', + 'channel_util', + 'channelUtil', + 'channelUtilization', + ); + const airUtilTx = readNum( + stats, + 'air_util_tx', + 'air_util_tx_pct', + 'tx_air_util', + 'tx_air_utilization', + 'airUtilTx', + ); + + if ( + batteryMv == null + && uptimeSecs == null + && txAirSecs == null + && rxAirSecs == null + && channelUtilization == null + && airUtilTx == null + ) { + if ((options?.allowRawStatsOnly || hasOwnerStats) && hasRawStats) { + return { + status, + batteryMv, + uptimeSecs, + txAirSecs, + rxAirSecs, + channelUtilization, + airUtilTx, + stats, + }; + } + return null; + } + + return { + status, + batteryMv, + uptimeSecs, + txAirSecs, + rxAirSecs, + channelUtilization, + airUtilTx, + stats, + }; +} + +export async function persistStatusTelemetrySample( + json: Record, + options: StatusSampleOptions, + writeSample: StatusSampleWriter, +): Promise { + const telemetry = extractStatusTelemetry(json, { + allowRawStatsOnly: options.allowRawStatsOnly, + }); + if (!telemetry) return false; + + await writeSample({ + nodeId: options.nodeId, + network: options.network, + ...telemetry, + }); + return true; +}