diff --git a/HANDOFF.md b/HANDOFF.md index d7d30b9..d78ead4 100644 --- a/HANDOFF.md +++ b/HANDOFF.md @@ -1,29 +1,44 @@ -# B3 packet-sharing handoff +# Handoff — Mission B2 · MQTT Will fake telemetry (#332) ## Status -Backend packet-sharing routes, contracts, persistence, forwarding, and tests are committed locally in `fbf9803` (`feat(owner): add owner-controlled packet sharing`). No migration was applied, no service was restarted, and nothing was pushed. +Done. Offline Will payloads are skipped for status samples, ordinary telemetry stores its top-level status, and the schema migration is included but unapplied. -## WIP inventory +## Branch and commits -The packet-sharing slice rescued from the main checkout contains migrations 052–054 for destination/rule/delivery tables, destination metadata, and default-off rules; owner GET/POST routes and contracts; broker-secret encryption; destination configuration and delivery-queue services; MQTT forwarding and worker wiring; and route/config/forwarder tests. The main checkout also has mixed public-feed, tagger, owner-portal, and unrelated backend work. +- Branch: `fix/mqtt-will-telemetry` +- Base: `d44ef5e` (`origin/main`) +- Implementation: `ad473779ce0f1ff250f6a9494d8abe9abba7a9f1` — `fix(mqtt): skip offline status telemetry samples` +- Push state: not pushed; coordinator relays this branch. -The main checkout and its WIP remain untouched. The packet-sharing backend copy there becomes redundant after this change is merged and deployed. The mixed owner-portal UI edits were not included in the backend scope in this brief and need separate disposition if they are not handled elsewhere. +## Files changed -## Migration decision +- `backend/src/mqtt/client.ts` +- `backend/src/mqtt/statusTelemetry.ts` +- `backend/src/mqtt/statusTelemetry.test.ts` +- `backend/src/db/index.ts` +- `backend/src/db/schema/base.sql` +- `backend/src/db/migrations/057_node_status_sample_status.sql` -Kept the supplied `052_owner_packet_sharing.sql`, `053_owner_packet_share_destination_metadata.sql`, and `054_owner_packet_share_default_off.sql` names. No duplicate prefix exists for these files in the target worktree, so no renumbering was needed. Prefix 055 is reserved by `055_message_tags.sql` on `fix/health-latency-bundle`; the separate untracked `055_public_feed.sql` in the main checkout was not copied. +## Checks -## Tests +Commands were run from `backend/`: -- `cd backend && npm test` — passed, 341 tests. -- `cd backend && npm run typecheck` — passed. -- `git diff --check` — passed. +- `node --import tsx --test src/mqtt/statusTelemetry.test.ts` — passed, 2 tests. +- `npm test` — passed, 326 tests, 0 failures. The script excludes `*.integration.test.ts`. +- `npm run typecheck` — passed. -The worktree had no installed dependencies. Tests reused the existing main-checkout `backend/node_modules` through a temporary symlink in this worktree; the symlink was removed afterward. No tests wrote to a live database. +## Gated items for Ben -## Gated items and open questions +- Apply migration `057_node_status_sample_status.sql` through the normal deployment process, then deploy the ingest change. No live DB writes, migration application, or service restarts were performed in this worktree session. +- Historical rows and retained broker messages were not modified; clearing retained messages is out of scope for #332. -- Ben owns migration application and deployment. -- The target worktree does not contain the `OWNER_PACKET_SHARE_*` Compose passthroughs or `owner_packet_share_deliveries` in the retention target list described by the brief. No `.env` or config files were changed under the VPS constraint. Before deployment, Ben must confirm the runtime passes `OWNER_PACKET_SHARE_ENCRYPTION_KEY` to the backend and includes `owner_packet_share_deliveries` in lifecycle retention targets. Poll interval and batch size have code defaults. -- Confirm whether the packet-sharing owner-portal UI edits in the main checkout are tracked in another change; they were left untouched and are not in `fbf9803`. +## Open questions + +None. + +## Evidence paths + +- `BRIEF.md` — mission and constraints. +- `backend/src/mqtt/statusTelemetry.test.ts` — offline Will skip, ordinary sample storage, and top-level status coverage. +- `backend/src/db/migrations/057_node_status_sample_status.sql` — additive status column migration. 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/057_node_status_sample_status.sql b/backend/src/db/migrations/057_node_status_sample_status.sql new file mode 100644 index 0000000..c534a9e --- /dev/null +++ b/backend/src/db/migrations/057_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/packetBatch.integration.test.ts b/backend/src/db/packetBatch.integration.test.ts index ec49e6c..c443a46 100644 --- a/backend/src/db/packetBatch.integration.test.ts +++ b/backend/src/db/packetBatch.integration.test.ts @@ -107,6 +107,7 @@ test('base schema and every migration apply to a brand-new database', { '053_owner_packet_share_destination_metadata.sql', '054_owner_packet_share_default_off.sql', '056_async_privacy_rematerialization.sql', + '057_node_status_sample_status.sql', ]); assert.deepEqual(await runMigrations(freshPool), []); 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; +}