fix: make MQTT packet ingest resilient

.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

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.

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/);
+});

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<PacketBatchResult> {
   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<PacketBatchResu
   return result;
 }
 
-async function writeBatch(batch: PendingPacket[]): Promise<void> {
+async function writeBatch(batch: PendingPacket[], idempotent: boolean): Promise<void> {
   const query = queryFn;
   if (!query) throw new Error('PACKET_BATCH_NOT_CONFIGURED');
 
@@ -125,11 +156,24 @@ async function writeBatch(batch: PendingPacket[]): Promise<void> {
     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<void> {
                   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<void> {
                 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<void> {
          )
          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<void> {
          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<void> {
            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<void> {
            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<void> {
          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<void> {
          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<void> {
       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<void> {
+  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<void> {
   if (timer) {
     clearTimeout(timer);
@@ -397,7 +463,7 @@ export function flush(): Promise<void> {
   activeFlush = (async () => {
     while (pending.length > 0) {
       const batch = pending.splice(0, MAX_BATCH_SIZE);
-      await writeBatch(batch);
+      await writeBatchWithRetry(batch);
     }
   })().finally(() => {
     activeFlush = null;

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.',

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<void> {
   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<void> {
     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<void> {
   try {
     json = JSON.parse(rawStr) as Record<string, unknown>;
   } 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<void> {
     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<void> {
     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<void> {
   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<void> {
   }
 
   if (resolvedPacketType == null) {
+    mqttIngestOutcomesTotal.inc({ outcome: 'missing_packet_type' });
     return;
   }
 
@@ -801,6 +811,7 @@ async function handleMessage(topic: string, rawPayload: Buffer): Promise<void> {
   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<void> {
     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<void> {
         .finally(() => { preResolveInFlight.delete(preResolveKey); });
     }
   } catch (err) {
+    mqttIngestOutcomesTotal.inc({ outcome: 'packet_persist_failure' });
     console.error('[mqtt] db insert failed', (err as Error).message);
   }
 }

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}