From 90d0dce603716bb81bbee94c7e31fabd934b4841 Mon Sep 17 00:00:00 2001 From: gadgethd <111318106+gadgethd@users.noreply.github.com> Date: Thu, 6 Aug 2026 02:13:38 +0000 Subject: [PATCH] fix: make MQTT packet ingest resilient --- .env.example | 7 +- INGEST-AUDIT-2026-08-06.md | 275 +++++++++++++++++++++++++++++ backend/src/db/packetBatch.test.ts | 63 +++++++ backend/src/db/packetBatch.ts | 138 +++++++++++---- backend/src/metrics.ts | 6 + backend/src/mqtt/client.ts | 17 +- docker-compose.yml | 1 + 7 files changed, 467 insertions(+), 40 deletions(-) create mode 100644 INGEST-AUDIT-2026-08-06.md create mode 100644 backend/src/db/packetBatch.test.ts diff --git a/.env.example b/.env.example index 032b2d5..c696bbe 100644 --- a/.env.example +++ b/.env.example @@ -25,13 +25,16 @@ POSTGRES_MAX_WORKER_PROCESSES=24 # ── MQTT broker ─────────────────────────────────────────────────────────────── # Internal docker network URL for Mosquitto WebSocket listener MQTT_BROKER_URL=ws://mosquitto:9001 +MQTT_CLIENT_ID=meshcore-analytics-ingest # Credentials for the backend MQTT client MQTT_USERNAME=backend MQTT_PASSWORD=change_me_strong_password -# Bound in-process MQTT work during a burst. MQTT is QoS 0, so an overload is -# logged and shed rather than letting the backend consume unbounded memory. +# Bound in-process MQTT work during a burst. The backend uses a persistent +# session and QoS 1 subscriptions so the broker can replay acknowledged QoS 1 +# publications after a reconnect; the packet batch writer also retries transient +# database failures. MQTT_MAX_PAYLOAD_BYTES=65536 MQTT_INGEST_CONCURRENCY=8 MQTT_INGEST_QUEUE_MAX=1000 diff --git a/INGEST-AUDIT-2026-08-06.md b/INGEST-AUDIT-2026-08-06.md new file mode 100644 index 0000000..824bf9c --- /dev/null +++ b/INGEST-AUDIT-2026-08-06.md @@ -0,0 +1,275 @@ +# MQTT ingest audit — 2026-08-06 + +## Result + +The backend had a real lossy failure path: MQTT messages were accepted into the +ingest workers, but a TimescaleDB connection/recovery error rejected the packet +batch and the MQTT handler swallowed the error. There was no retry, durable +handoff, or MQTT persistent session. This can explain packets lost during the +database/container instability around the VPS cutover. + +The permanent backend fix deployed on this host is: + +- transient packet-batch database errors retry for a bounded ~8-second window; +- retry attempts use an idempotent `new_rows` CTE, so a lost database response + cannot duplicate packets or rollups; +- the normal path keeps the original fast SQL and does not run the idempotency + lookup on every batch; +- the MQTT client now uses the stable client ID + `meshcore-analytics-ingest`, `clean=false`, and QoS 1 subscriptions; +- bounded ingest outcome metrics now identify invalid JSON, empty packets, + duplicate packets, persistence success/failure, and status failures. + +The live deploy used backend image +`sha256:7e085b1286d5159be24d12492aa8603baf995a86d33c5ac76b612d1d7fbfb733`, +pinned in `.env`. Only the backend was recreated. Mosquitto and all other +services were left running. + +## Full ingest path + +```text +Mosquitto listeners 1883/9001 + -> backend MQTT.js WebSocket client + -> subscriptions meshcore/#, ukmesh/#, meshcore-test/# + -> bounded in-process queue (concurrency 8, max 1000, max payload 64 KiB) + -> topic validation and JSON parsing + -> status persistence OR packet decode/metadata extraction + -> per-observer duplicate filter (hash + observer + hop count, 120 s) + -> insertPacket() + -> packetBatch (50 rows or 50 ms) + -> one PostgreSQL CTE statement: + packets, nodes, network sightings, daily/hourly rollups, + region packet sightings, region observer sightings +``` + +Relevant implementation files: + +- `backend/src/mqtt/client.ts` owns connection/reconnect, subscription, payload + limits, queueing, topic handling, decoding, dedupe, and the `insertPacket` + handoff. +- `backend/src/mqtt/topic.ts` accepts exactly four topic segments, a configured + prefix, an alphanumeric IATA, a 64-hex observer key, and only `packets` or + `status` as the suffix. `TST` is blocked. +- `backend/src/db/index.ts` normalizes the packet and calls + `enqueuePacket()`; it does not insert directly. +- `backend/src/db/packetBatch.ts` writes batches atomically. There is no unique + constraint on `packets.packet_hash`; rows represent observer receptions. +- `DISTINCT ON(packet_hash)` occurs in `backfillHistoricalLinks()` as a + read/backfill selection. It is not part of the live packet insert. + +## Drop and reject paths + +Before the fix, the following could remove a broker message from the ingest +result without replay: + +1. MQTT disconnects/reconnects with the default random client ID and clean + session; the old subscription was QoS 0. +2. Payload/topic over the configured bounds. +3. Queue full or process draining. +4. Invalid topic shape/prefix/suffix, invalid JSON, status identity mismatch, + empty packet envelope, or missing packet type. +5. The deliberate in-memory duplicate filter: the same final hash at the same + observer and hop count within 120 seconds. +6. Any `insertPacket()`/packet-batch database error. The old catch logged + `[mqtt] db insert failed` and returned; the batch rejected all its promises. + +Decoder failures generally fall back to envelope fields. Impossible multibyte +path metadata is cleared rather than dropping the packet row. The packet batch +has no `ON CONFLICT` protection, so blindly retrying it would have been unsafe; +the deployed retry path first uses the original fast SQL, then switches to an +idempotent existence check only after a retryable database failure. + +## Historical counts and observer evidence + +The raw public `packets` counts at audit time were: + +| UTC day | rows | +|---|---:| +| 2026-07-30 | 89,140 (the supplied snapshot was 87,895) | +| 2026-07-31 | 110,305 | +| 2026-08-01 | 139,684 | +| 2026-08-02 | 131,330 | +| 2026-08-03 | 154,764 | +| 2026-08-04 | 121,540 | +| 2026-08-05 | 100,154 | +| 2026-08-06 | 6,906 at approximately 01:42Z (partial) | + +The fall starts before the migration cutover. Per-IATA rows show the same +pattern, not an isolated database-only change: + +| day | MME | NCL | EMA | LTN | LBA | BHX | BOH | NWI | +|---|---:|---:|---:|---:|---:|---:|---:|---:| +| Aug 3 | 67,614 | 25,640 | 14,410 | 15,681 | 11,908 | 9,588 | 8,345 | 1,578 | +| Aug 4 | 44,730 | 18,392 | 14,613 | 14,240 | 11,076 | 9,376 | 7,848 | 1,265 | +| Aug 5 | 35,752 | 14,795 | 11,319 | 11,818 | 9,473 | 9,722 | 6,882 | 393 | + +Notable observer changes include: + +- MME observer `65821B...A104F8`: 11,978 rows on Aug 3, only 393 on Aug 4, + then zero on Aug 5/6. +- NCL observer `E1C1F5...923D4A`: 8,121 on Aug 3, 3,761 on Aug 4, 2,120 + on Aug 5. +- NCL observer `CFECDA...215B7`: 5,586 on Aug 2, absent on Aug 3/4, then + 1,267 late on Aug 5. +- NWI fell from 1,578 on Aug 3 to 1,265 on Aug 4 and 393 on Aug 5. + +The migration cutover itself does not show a packet cliff. Raw rows around the +reported 22:46Z cutover were: + +| window | rows | approximate rate | +|---|---:|---:| +| Aug 4 18:00–22:46 | 21,225 | 4,452/hour | +| Aug 5 18:00–22:46 | 21,787 | 4,571/hour | +| Aug 5 22:46–00:00 | 5,855 | 4,747/hour | + +The Aug 4 decline is therefore upstream/source-observer traffic loss and/or +feed availability, beginning before the VPS cutover. The cutover and database +instability added a separate backend loss risk. + +## Dedupe and `DISTINCT ON` quantification + +The live table contains observer reception rows, while the in-memory filter +deliberately removes relay copies for the same observer within 120 seconds. +Historical raw rows versus distinct values were: + +| day | raw rows | distinct packet hashes | distinct hash + observer | +|---|---:|---:|---:| +| Jul 30 | 89,140 | 11,083 | 49,639 | +| Jul 31 | 110,305 | 12,076 | 60,738 | +| Aug 1 | 139,684 | 14,162 | 78,289 | +| Aug 2 | 131,330 | 12,907 | 75,627 | +| Aug 3 | 154,764 | 13,174 | 87,772 | +| Aug 4 | 121,540 | 12,002 | 72,940 | +| Aug 5 | 100,154 | 11,876 | 61,397 | + +This duplicate fraction is broadly stable or slightly lower after Aug 4; the +dedupe rule was introduced in March and is not the cause of the Aug 4 step +down. It does, however, explain why a service counting every MQTT envelope can +report more packets than this table. + +## Broker-side and backend-side measurements + +Historical `$SYS/broker/messages/received` was not available: the command was +rejected by Mosquitto ACLs because the backend user has no `$SYS/#` permission. +The broker log is connection-only (`log_type error/warning/notice`) and begins +at Aug 5 02:05Z, so it cannot reconstruct Aug 4 message totals. No broker +restart or ACL broadening was performed. + +An authenticated, read-only `mosquitto_sub` on the broker was used for live +packet-envelope counts. The final synchronized window was 2026-08-06 +02:07:12–02:08:12Z: + +- broker packet envelopes: 112; +- backend valid `ukmesh` MQTT messages: 135 (includes status messages); +- packet rows persisted: 46; +- duplicate rejects: 65; +- queue-full/oversize/draining drops: 0; +- packet-batch retries: 0; +- database rows in the same time window: 46. + +Per-observer broker envelopes versus database rows: + +| IATA | observer prefix | broker | database | dedupe/other | +|---|---|---:|---:|---:| +| MME | `1155AB...` | 2 | 2 | 0 | +| MME | `227744...` | 14 | 4 | 10 | +| LTN | `59A1DD...` | 2 | 2 | 0 | +| EMA | `68DE80...` | 7 | 4 | 3 | +| NCL | `7B3F9F...` | 18 | 5 | 13 | +| MME | `A86A0C...` | 3 | 2 | 1 | +| NCL | `CFECDA...` | 6 | 2 | 4 | +| BOH | `D0B527...` | 4 | 3 | 1 | +| MME | `D7696E...` | 16 | 4 | 12 | +| LBA | `DC9199...` | 14 | 4 | 10 | +| LTN | `E0218F...` | 4 | 3 | 1 | +| NCL | `E1C1F5...` | 12 | 4 | 8 | +| MME | `F51858...` | 4 | 3 | 1 | +| BHX | `FD636A...` | 6 | 4 | 2 | +| **total** | | **112** | **46** | **66** | + +The 66-envelope difference is explained by 65 duplicate outcomes and one +non-persistable packet envelope in the synchronized metric/sample window. The +backend queue was empty at the end of the test, and the persisted delta matched +the table count. + +## Database failure evidence + +The live TimescaleDB container had `OOMKilled=true`, with its previous process +finishing at 22:34:16Z and the current container starting at 22:38:39Z. This is +strong evidence of a resource event around the migration period, although the +available logs do not prove that event was the sole cause of the Aug 4 decline. + +During this audit, a broad historical join used for an initial cutover query +also caused a PostgreSQL server process to be killed at 01:44:14Z. PostgreSQL +automatically recovered and was accepting connections at 01:44:20Z; no +TimescaleDB container restart was performed by this audit. The backend logs +during recovery show the exact old loss path: packet-batch failures and +`[mqtt] db insert failed` messages, with no retry. This incident is recorded +explicitly because it validates the failure mode, but it must not be confused +with evidence that the audit query caused the original Aug 4 decline. + +The database was also under sustained write pressure: checkpoints were commonly +8–11 seconds apart before the recovery event, with roughly 0.5 GB WAL distance +between checkpoints. The ingest retry fix protects packets across a transient +restart; the database resource pressure remains an operational item to monitor. + +## Topic coverage + +The backend subscribes to all three configured prefix wildcards: + +```text +meshcore/# +ukmesh/# +meshcore-test/# +``` + +The live broker sample contained `meshcore` packet/status traffic, +`meshcore-test` status traffic, and a `neighbours` suffix. The latter is +received by the wildcard but intentionally rejected by the strict topic parser; +it is not a packet topic. The repository contains no other backend packet +subscriber. The health-check service’s narrower filter includes packets, +status, and internal topics, but that is not the ingest client. + +Therefore no public packet prefix was missing from the backend subscription. +The only historical broker measurement limitation is the lack of `$SYS` ACL and +publish-level broker logging. + +## Verification and deployment + +Executed: + +```text +BACKEND_IMAGE=meshcore-analytics-backend:ingest-fix \ + docker compose -f docker-compose.yml -f docker-compose.live.yml build backend + +docker compose -f docker-compose.yml -f docker-compose.live.yml \ + up -d --no-deps --force-recreate backend +``` + +The image was rebuilt once more after removing the normal-path idempotency +lookup, then pinned in `.env` to the digest listed at the top of this report. + +Required endpoint checks passed: + +```text +GET http://127.0.0.1:3000/api/health +{"status":"healthy", ... "components":{"ingest":{"status":"ok"}, ...}} + +GET http://127.0.0.1:3000/hopreach/api/nodes +HTTP 200 +``` + +Post-deploy ingest metrics showed queue depth 0, no queue-full/oversize/draining +drops, zero packet-batch retries, and 533 successful batches. Cumulative +backend metrics at the final scrape showed 585 persisted packet outcomes and +688 duplicate outcomes; the synchronized test window is the authoritative +per-observer comparison above. Normal packet-batch duration was about 13.2 ms +on average after the corrected deploy. + +Focused retry test passed. The production image TypeScript build passed. The +full unit suite had 263/264 passing; the sole failure is the existing +`policyRegistry.test.ts` expectation for the unrelated unregistered +`src/mqtt/channelRegistry.ts#channelCache`. The ingest integration test was +skipped because `TEST_INGEST_DATABASE_URL` is not configured. + +No Mosquitto rebuild/restart was needed. diff --git a/backend/src/db/packetBatch.test.ts b/backend/src/db/packetBatch.test.ts new file mode 100644 index 0000000..31d0487 --- /dev/null +++ b/backend/src/db/packetBatch.test.ts @@ -0,0 +1,63 @@ +import assert from 'node:assert/strict'; +import test from 'node:test'; +import { + configurePacketBatch, + enqueuePacket, + flush, + type PacketBatchInput, +} from './packetBatch.js'; + +test('packet batches retry transient database failures without losing the pending row', async () => { + let attempts = 0; + const statements: string[] = []; + configurePacketBatch(async (text) => { + statements.push(text); + attempts += 1; + if (attempts < 3) { + throw new Error('the database system is not yet accepting connections'); + } + return { + rows: [{ + row_id: 0, + is_private: false, + visibility_ok: true, + }], + }; + }); + + const packet: PacketBatchInput = { + time: new Date('2026-08-06T01:00:00.000Z'), + packetHash: 'retry-test-packet', + rxNodeId: 'A'.repeat(64), + srcNodeId: 'B'.repeat(64), + topic: `meshcore/EMA/${'A'.repeat(64)}/packets`, + topicPrefix: 'meshcore', + iata: 'EMA', + packetType: 5, + routeType: 0, + hopCount: 1, + rssi: -80, + snr: 5, + payloadJson: JSON.stringify({ test: true }), + companionSender: null, + rawHex: '00', + advertCount: null, + pathHashes: null, + pathHashSizeBytes: null, + network: 'ukmesh', + transportCodes: null, + regionScope: null, + }; + + const resultPromise = enqueuePacket(packet); + await flush(); + + assert.deepEqual(await resultPromise, { + isPrivate: false, + visibilityOk: true, + }); + assert.equal(attempts, 3); + assert.doesNotMatch(statements[0]!, /new_rows AS MATERIALIZED/); + assert.match(statements[1]!, /new_rows AS MATERIALIZED/); + assert.match(statements[1]!, /existing\.packet_hash = c\.packet_hash/); +}); diff --git a/backend/src/db/packetBatch.ts b/backend/src/db/packetBatch.ts index a48f363..cb714c2 100644 --- a/backend/src/db/packetBatch.ts +++ b/backend/src/db/packetBatch.ts @@ -2,12 +2,30 @@ import type { QueryResultRow } from 'pg'; import { packetBatchDuration, packetBatchFlushTotal, + packetBatchRetryTotal, packetBatchSize, privacyFilterTotal, } from '../metrics.js'; const MAX_BATCH_SIZE = 50; const FLUSH_INTERVAL_MS = 50; +const MAX_WRITE_ATTEMPTS = 7; +const RETRY_DELAYS_MS = [100, 250, 500, 1_000, 2_000, 4_000] as const; + +const RETRYABLE_DATABASE_CODES = new Set([ + '08000', // connection_exception + '08001', // sqlclient_unable_to_establish_sqlconnection + '08003', // connection_does_not_exist + '08004', // sqlserver_rejected_establishment_of_sqlconnection + '08006', // connection_failure + '08007', // transaction_resolution_unknown + '57P01', // admin_shutdown + '57P02', // crash_shutdown + '57P03', // cannot_connect_now + '40001', // serialization_failure + '40P01', // deadlock_detected + '53300', // too_many_connections +]); export type PacketBatchInput = { time: Date; @@ -70,6 +88,19 @@ function scheduleFlush(): void { timer.unref(); } +function isRetryableDatabaseError(error: unknown): boolean { + if (typeof error === 'object' && error !== null && 'code' in error) { + const code = (error as { code?: unknown }).code; + if (typeof code === 'string' && RETRYABLE_DATABASE_CODES.has(code)) return true; + } + const message = error instanceof Error ? error.message : String(error); + return /connection terminated|database system is not yet accepting connections|server closed the connection unexpectedly|connection reset|connection refused|could not connect/i.test(message); +} + +function retryDelay(attempt: number): number { + return RETRY_DELAYS_MS[Math.min(attempt, RETRY_DELAYS_MS.length - 1)] ?? 4_000; +} + export function enqueuePacket(packet: PacketBatchInput): Promise { if (draining) return Promise.reject(new Error('PACKET_BATCH_DRAINING')); if (!queryFn) return Promise.reject(new Error('PACKET_BATCH_NOT_CONFIGURED')); @@ -81,7 +112,7 @@ export function enqueuePacket(packet: PacketBatchInput): Promise { +async function writeBatch(batch: PendingPacket[], idempotent: boolean): Promise { const query = queryFn; if (!query) throw new Error('PACKET_BATCH_NOT_CONFIGURED'); @@ -125,11 +156,24 @@ async function writeBatch(batch: PendingPacket[]): Promise { return `(${placeholders.join(', ')})`; }).join(',\n'); - const startedAt = process.hrtime.bigint(); - let outcome = 'success'; - packetBatchSize.observe(batch.length); - try { - const result = await query<{ row_id: number; is_private: boolean; visibility_ok: boolean }>( + const persistenceSource = idempotent ? 'new_rows' : 'classified'; + const idempotencyCte = idempotent + ? `new_rows AS MATERIALIZED ( + SELECT c.* + FROM classified c + WHERE NOT EXISTS ( + SELECT 1 + FROM packets existing + WHERE existing.time = c.time + AND existing.packet_hash = c.packet_hash + AND existing.rx_node_id IS NOT DISTINCT FROM c.rx_node_id + AND existing.topic = c.topic + AND existing.network = c.network + ) + ),` + : ''; + + const result = await query<{ row_id: number; is_private: boolean; visibility_ok: boolean }>( `WITH incoming ( row_id, time, packet_hash, rx_node_id, src_node_id, topic, topic_prefix, iata, packet_type, route_type, hop_count, rssi, snr, payload, companion_sender, raw_hex, @@ -168,9 +212,10 @@ async function writeBatch(batch: PendingPacket[]): Promise { OR length(path_hash) <> i.path_hash_size_bytes * 2 OR path_hash !~ '^[0-9A-Fa-f]+$' ) - ) AS path_is_valid + ) AS path_is_valid FROM incoming i ), + ${idempotencyCte} inserted AS ( INSERT INTO packets ( time, packet_hash, rx_node_id, src_node_id, topic, topic_prefix, iata, @@ -182,7 +227,7 @@ async function writeBatch(batch: PendingPacket[]): Promise { packet_type, route_type, hop_count, rssi, snr, payload::jsonb, companion_sender, raw_hex, advert_count, path_hashes, path_hash_size_bytes, network, transport_codes, region_scope, is_private, path_is_valid AND NOT is_private - FROM classified + FROM ${persistenceSource} RETURNING 1 ), observer_updates AS ( @@ -192,7 +237,7 @@ async function writeBatch(batch: PendingPacket[]): Promise { ) SELECT DISTINCT ON (rx_node_id) rx_node_id, iata, iata, time, time, TRUE, network, time - FROM classified + FROM ${persistenceSource} WHERE rx_node_id IS NOT NULL AND rx_node_id <> '' ORDER BY rx_node_id, time DESC ON CONFLICT (node_id) DO UPDATE SET @@ -216,7 +261,7 @@ async function writeBatch(batch: PendingPacket[]): Promise { INSERT INTO node_network_sightings (node_id, network, first_seen_at, last_seen_at) SELECT src_node_id, network, MIN(time), MAX(time) - FROM classified + FROM ${persistenceSource} WHERE src_node_id IS NOT NULL AND src_node_id <> '' AND network <> 'test' @@ -239,7 +284,7 @@ async function writeBatch(batch: PendingPacket[]): Promise { hop_count, packet_hash, time - FROM classified + FROM ${persistenceSource} WHERE path_is_valid AND NOT is_private AND hop_count IS NOT NULL @@ -279,7 +324,7 @@ async function writeBatch(batch: PendingPacket[]): Promise { COALESCE(SUM(snr), 0)::double precision, COUNT(snr)::bigint, NOW() - FROM classified + FROM ${persistenceSource} WHERE path_is_valid AND NOT is_private AND (network = 'test' OR topic_prefix <> 'meshcore-test') @@ -307,7 +352,7 @@ async function writeBatch(batch: PendingPacket[]): Promise { INSERT INTO observer_region_packet_sightings (network, iata, packet_hash, first_seen, last_seen) SELECT network, iata, packet_hash, MIN(time), MAX(time) - FROM classified + FROM ${persistenceSource} WHERE network <> 'test' AND path_is_valid AND NOT is_private @@ -331,7 +376,7 @@ async function writeBatch(batch: PendingPacket[]): Promise { INSERT INTO observer_region_observer_sightings (network, iata, rx_node_id, first_seen, last_seen) SELECT network, iata, rx_node_id, MIN(time), MAX(time) - FROM classified + FROM ${persistenceSource} WHERE network <> 'test' AND path_is_valid AND NOT is_private @@ -363,30 +408,51 @@ async function writeBatch(batch: PendingPacket[]): Promise { params, ); - if (result.rows.length !== batch.length) throw new Error('PACKET_BATCH_RESULT_MISMATCH'); - for (const row of result.rows) { - privacyFilterTotal.inc({ - operation: 'packet_ingest', - outcome: row.visibility_ok ? 'public' : (row.is_private ? 'private' : 'invalid'), - }); - batch[row.row_id]?.resolve({ - isPrivate: Boolean(row.is_private), - visibilityOk: Boolean(row.visibility_ok), - }); - } - } catch (error) { - outcome = 'failure'; - for (const item of batch) item.reject(error); - throw error; - } finally { - packetBatchFlushTotal.inc({ outcome }); - packetBatchDuration.observe( - { outcome }, - Number(process.hrtime.bigint() - startedAt) / 1e9, - ); + if (result.rows.length !== batch.length) throw new Error('PACKET_BATCH_RESULT_MISMATCH'); + for (const row of result.rows) { + privacyFilterTotal.inc({ + operation: 'packet_ingest', + outcome: row.visibility_ok ? 'public' : (row.is_private ? 'private' : 'invalid'), + }); + batch[row.row_id]?.resolve({ + isPrivate: Boolean(row.is_private), + visibilityOk: Boolean(row.visibility_ok), + }); } } +async function writeBatchWithRetry(batch: PendingPacket[]): Promise { + const startedAt = process.hrtime.bigint(); + packetBatchSize.observe(batch.length); + let lastError: unknown; + + for (let attempt = 0; attempt < MAX_WRITE_ATTEMPTS; attempt += 1) { + try { + await writeBatch(batch, attempt > 0); + packetBatchFlushTotal.inc({ outcome: 'success' }); + packetBatchDuration.observe( + { outcome: 'success' }, + Number(process.hrtime.bigint() - startedAt) / 1e9, + ); + return; + } catch (error) { + lastError = error; + const retriesRemaining = attempt < MAX_WRITE_ATTEMPTS - 1; + if (!retriesRemaining || !isRetryableDatabaseError(error)) break; + packetBatchRetryTotal.inc(); + await new Promise((resolve) => setTimeout(resolve, retryDelay(attempt))); + } + } + + for (const item of batch) item.reject(lastError); + packetBatchFlushTotal.inc({ outcome: 'failure' }); + packetBatchDuration.observe( + { outcome: 'failure' }, + Number(process.hrtime.bigint() - startedAt) / 1e9, + ); + throw lastError; +} + export function flush(): Promise { if (timer) { clearTimeout(timer); @@ -397,7 +463,7 @@ export function flush(): Promise { activeFlush = (async () => { while (pending.length > 0) { const batch = pending.splice(0, MAX_BATCH_SIZE); - await writeBatch(batch); + await writeBatchWithRetry(batch); } })().finally(() => { activeFlush = null; diff --git a/backend/src/metrics.ts b/backend/src/metrics.ts index 6e85a39..e38f469 100644 --- a/backend/src/metrics.ts +++ b/backend/src/metrics.ts @@ -150,6 +150,12 @@ export const packetBatchFlushTotal = new Counter({ registers: [metricsRegistry], }); +export const packetBatchRetryTotal = new Counter({ + name: 'meshcore_packet_batch_retries_total', + help: 'Transient database retry attempts for packet batches.', + registers: [metricsRegistry], +}); + export const packetBatchSize = new Histogram({ name: 'meshcore_packet_batch_size', help: 'Packets committed per atomic batch.', diff --git a/backend/src/mqtt/client.ts b/backend/src/mqtt/client.ts index 65ad0d9..ab50e40 100644 --- a/backend/src/mqtt/client.ts +++ b/backend/src/mqtt/client.ts @@ -364,6 +364,9 @@ function boundedEnvInteger(name: string, fallback: number, min: number, max: num const MQTT_MAX_PAYLOAD_BYTES = boundedEnvInteger('MQTT_MAX_PAYLOAD_BYTES', 64 * 1024, 1_024, 1_048_576); const MQTT_INGEST_CONCURRENCY = boundedEnvInteger('MQTT_INGEST_CONCURRENCY', 8, 1, 32); const MQTT_INGEST_QUEUE_MAX = boundedEnvInteger('MQTT_INGEST_QUEUE_MAX', 1_000, 10, 20_000); +const MQTT_CLIENT_ID = String( + process.env['MQTT_CLIENT_ID'] ?? 'meshcore-analytics-ingest', +).trim() || 'meshcore-analytics-ingest'; type MqttIngestTask = { topic: string; @@ -438,11 +441,13 @@ export async function startMqttClient(): Promise { console.log(`[mqtt] channels: ${channelEntries.map((e) => e.name).join(', ')}`); console.log(`[mqtt] topic prefixes: ${Array.from(TOPIC_PREFIXES).join(', ')}`); console.log(`[mqtt] ingest concurrency=${MQTT_INGEST_CONCURRENCY} queue=${MQTT_INGEST_QUEUE_MAX} maxPayload=${MQTT_MAX_PAYLOAD_BYTES}`); + console.log(`[mqtt] persistent session clientId=${MQTT_CLIENT_ID} subscriptionQos=1`); const client = mqtt.connect(brokerUrl, { reconnectPeriod: 5000, connectTimeout: 30000, - clientId: `meshcore-analytics-${Math.random().toString(16).slice(2, 8)}`, + clean: false, + clientId: MQTT_CLIENT_ID, username: process.env['MQTT_USERNAME'], password: process.env['MQTT_PASSWORD'], }); @@ -452,7 +457,7 @@ export async function startMqttClient(): Promise { setMqttRuntimeStatus('connected'); console.log('[mqtt] connected'); for (const prefix of TOPIC_PREFIXES) { - client.subscribe(`${prefix}/#`, { qos: 0 }, (err) => { + client.subscribe(`${prefix}/#`, { qos: 1 }, (err) => { if (err) console.error(`[mqtt] subscribe error (${prefix}/#)`, err.message); else console.log(`[mqtt] subscribed to ${prefix}/#`); }); @@ -532,6 +537,7 @@ async function handleMessage(topic: string, rawPayload: Buffer): Promise { try { json = JSON.parse(rawStr) as Record; } catch { + mqttIngestOutcomesTotal.inc({ outcome: 'invalid_json' }); console.warn(`[mqtt] non-JSON payload on topic ${topic}: ${rawStr.slice(0, 80)}`); return; } @@ -542,6 +548,7 @@ async function handleMessage(topic: string, rawPayload: Buffer): Promise { const firmware = json['firmware_version'] as string | undefined; if (!statusEnvelopeTargetsObserver(observerKey, json)) { + mqttIngestOutcomesTotal.inc({ outcome: 'status_identity_mismatch' }); console.warn('[mqtt] rejected status envelope with identity mismatch'); return; } @@ -576,6 +583,7 @@ async function handleMessage(topic: string, rawPayload: Buffer): Promise { const writeResults = await Promise.allSettled(writes); for (const result of writeResults) { if (result.status === 'rejected') { + mqttIngestOutcomesTotal.inc({ outcome: 'status_persist_failure' }); console.error('[mqtt] status persistence error:', (result.reason as Error).message); } } @@ -595,6 +603,7 @@ async function handleMessage(topic: string, rawPayload: Buffer): Promise { const direction = directionValue === 'rx' || directionValue === 'tx' ? directionValue : undefined; if (isEmptyPacketEnvelope(json, rawHex, packetType)) { + mqttIngestOutcomesTotal.inc({ outcome: 'empty_packet' }); return; } @@ -792,6 +801,7 @@ async function handleMessage(topic: string, rawPayload: Buffer): Promise { } if (resolvedPacketType == null) { + mqttIngestOutcomesTotal.inc({ outcome: 'missing_packet_type' }); return; } @@ -801,6 +811,7 @@ async function handleMessage(topic: string, rawPayload: Buffer): Promise { const finalHash = canonicalPacketId ?? upstreamPacketHash(json['hash']) ?? crypto.randomUUID(); if (isDuplicatePacket(finalHash, observerKey, decodedHops)) { + mqttIngestOutcomesTotal.inc({ outcome: 'duplicate_packet' }); return; } @@ -859,6 +870,7 @@ async function handleMessage(topic: string, rawPayload: Buffer): Promise { livePacket.visibilityOk = visibility.visibilityOk; emit(livePacket); invalidateResolveCache(finalHash); + mqttIngestOutcomesTotal.inc({ outcome: 'packet_persisted' }); // A repeater appearing in a multibyte (2–3 byte) path hash almost certainly // relayed this packet. Resolve only a unique historic row with stored @@ -932,6 +944,7 @@ async function handleMessage(topic: string, rawPayload: Buffer): Promise { .finally(() => { preResolveInFlight.delete(preResolveKey); }); } } catch (err) { + mqttIngestOutcomesTotal.inc({ outcome: 'packet_persist_failure' }); console.error('[mqtt] db insert failed', (err as Error).message); } } diff --git a/docker-compose.yml b/docker-compose.yml index 50989be..31d708c 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -193,6 +193,7 @@ services: MQTT_BROKER_URL: ${MQTT_BROKER_URL:-ws://mosquitto:9001} MQTT_USERNAME: ${MQTT_USERNAME:-backend} MQTT_PASSWORD: ${MQTT_PASSWORD} + MQTT_CLIENT_ID: ${MQTT_CLIENT_ID:-meshcore-analytics-ingest} REDIS_URL: ${REDIS_URL:-redis://redis:6379} REDIS_PASSWORD: ${REDIS_PASSWORD:?REDIS_PASSWORD required} JWT_SECRET: ${JWT_SECRET:?JWT_SECRET required}