fix(ingest): re-arbitrated dedupe takeover + bounded post-persist follow-ups

- Concurrent duplicates that arrive while the first persist is in flight now
  re-arbitrate on failure: the successor claim is registered inside
  waitForInFlightPacket in the same microtask as the emptiness check, so exactly
  one waiter takes over and the rest queue behind it (previously two waiters
  could both fall through after a failed attempt and persist concurrently,
  risking a double insert since packets carry no unique constraint).
  Regression tests cover the single- and chained-failure arbitrations.
- Post-persist follow-ups (multibyte evidence / slow-mode / pre-resolve) are now
  bounded at 64 in flight; over the bound they are dropped and counted
  (meshcore_mqtt_post_persist_dropped_total + _inflight gauge), so a database
  stall cannot grow them without limit.

backend/src/metrics.ts

@@ -499,3 +499,13 @@ export function updateDbPoolMetrics(
   dbPoolTotal.set({ state: `${poolName}_idle` }, snapshot.idleCount);
   dbPoolTotal.set({ state: `${poolName}_waiting` }, snapshot.waitingCount);
 }
+
+export const postPersistInFlightGauge = new Gauge({
+  name: 'meshcore_mqtt_post_persist_inflight',
+  help: 'Best-effort post-persist follow-ups currently in flight (bounded).',
+});
+
+export const postPersistDroppedTotal = new Counter({
+  name: 'meshcore_mqtt_post_persist_dropped_total',
+  help: 'Best-effort post-persist follow-ups dropped because the in-flight bound was reached.',
+});

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 { waitForInFlightPacket } from './dedupeRegistry.js';
 import { persistStatusTelemetrySample } from './statusTelemetry.js';
 import {
   boundedNetworkMetricLabel,
@@ -21,6 +22,8 @@ import {
   mqttIngestActive,
   mqttIngestOutcomesTotal,
   mqttMessagesTotal,
+  postPersistDroppedTotal,
+  postPersistInFlightGauge,
 } from '../metrics.js';
 
 type PacketCallback      = (packet: LivePacket, upstreamEnvelope?: Record<string, unknown>) => void;
@@ -253,6 +256,11 @@ function rememberPacket(packetHash: string, observerKey: string, hopCount: numbe
  * committed so this copy persists on its own.
  */
 const inFlightPackets = new Map<string, Promise<void>>();
+// Best-effort post-persist follow-ups are bounded: a stalled database must
+// not grow the in-flight set without limit. Over the bound they are dropped
+// (safe: everything here is explicitly best-effort) and counted.
+const POST_PERSIST_MAX_INFLIGHT = 64;
+let postPersistInFlight = 0;
 
 /**
  * Database-level duplicate confirmation for QoS 1 redeliveries — the in-memory
@@ -792,23 +800,21 @@ async function handleMessage(topic: string, rawPayload: Buffer, redelivered = fa
     return;
   }
 
-  const inFlight = inFlightPackets.get(dedupeKey);
-  if (inFlight) {
-    // The same packet is already being persisted by a concurrent ingest.
-    // Wait for its outcome: success makes this copy a confirmed duplicate,
-    // failure means nothing was committed and this copy persists below.
-    const persisted = await inFlight.then(() => true, () => false);
-    if (persisted) {
-      markMqttPersistenceHealthy();
-      mqttIngestOutcomesTotal.inc({ outcome: 'duplicate_packet' });
-      return;
-    }
-  }
+  // Arbitration against concurrent copies of this packet. The claim below is
+  // registered inside waitForInFlightPacket in the same microtask as its
+  // re-check, so after a failed predecessor exactly one waiter becomes the
+  // successor and the rest queue behind it (never two concurrent retries).
+  let claimResolve!: () => void;
+  let claimReject!: (error: unknown) => void;
+  const claim = new Promise<void>((resolve, reject) => {
+    claimResolve = resolve;
+    claimReject = reject;
+  });
+  claim.catch(() => { /* waiters observe failure via their own handlers */ });
 
-  // After a restart there is no in-memory dedup state to match against: a
-  // QoS 1 redelivery whose original commit survived the crash must not be
-  // persisted or counted a second time. Confirm against the retained window.
-  if (redelivered && await packetAlreadyPersisted(finalHash, observerKey, topic, network)) {
+  const claimState = await waitForInFlightPacket(inFlightPackets, dedupeKey, claim);
+  if (claimState === 'duplicate') {
+    claimResolve(); // never registered; settle for tidiness
     markMqttPersistenceHealthy();
     mqttIngestOutcomesTotal.inc({ outcome: 'duplicate_packet' });
     return;
@@ -881,6 +887,16 @@ async function handleMessage(topic: string, rawPayload: Buffer, redelivered = fa
   };
 
   const ingest = (async () => {
+    // Claimed above. After a restart there is no in-memory dedup state to
+    // match against: a QoS 1 redelivery whose original commit survived the
+    // crash must not be persisted or counted a second time. Confirm against
+    // the retained window before writing (inside the claim so the registered
+    // claim promise covers this path too).
+    if (redelivered && await packetAlreadyPersisted(finalHash, observerKey, topic, network)) {
+      markMqttPersistenceHealthy();
+      mqttIngestOutcomesTotal.inc({ outcome: 'duplicate_packet' });
+      return;
+    }
     try {
       await persistOnce();
     } catch (err) {
@@ -893,27 +909,41 @@ async function handleMessage(topic: string, rawPayload: Buffer, redelivered = fa
     }
     // Best-effort live-map follow-ups run once the durable write is done and
     // outside the acknowledgement path: PUBACK is released as soon as the
-    // write commits and these resolve asynchronously.
-    void runPostPersistFollowUps({
-      finalHash,
-      decodedPathHashSizeBytes,
-      path,
-      decodedRouteType,
-      network,
-      resolvedPacketType,
-    }).catch((err: unknown) => {
-      console.error(
-        '[mqtt] post-persist follow-up failed:',
-        err instanceof Error ? err.message : String(err),
-      );
-    });
+    // write commits and these resolve asynchronously. The in-flight count is
+    // bounded so a database stall cannot grow them without limit; over the
+    // bound they are dropped (best-effort) and counted.
+    if (postPersistInFlight >= POST_PERSIST_MAX_INFLIGHT) {
+      postPersistDroppedTotal.inc();
+    } else {
+      postPersistInFlight++;
+      postPersistInFlightGauge.set(postPersistInFlight);
+      void runPostPersistFollowUps({
+        finalHash,
+        decodedPathHashSizeBytes,
+        path,
+        decodedRouteType,
+        network,
+        resolvedPacketType,
+      }).catch((err: unknown) => {
+        console.error(
+          '[mqtt] post-persist follow-up failed:',
+          err instanceof Error ? err.message : String(err),
+        );
+      }).finally(() => {
+        postPersistInFlight--;
+        postPersistInFlightGauge.set(postPersistInFlight);
+      });
+    }
   })();
 
-  inFlightPackets.set(dedupeKey, ingest);
+  void ingest.then(
+    () => claimResolve(),
+    (err: unknown) => claimReject(err),
+  );
   try {
     await ingest;
   } finally {
-    if (inFlightPackets.get(dedupeKey) === ingest) inFlightPackets.delete(dedupeKey);
+    if (inFlightPackets.get(dedupeKey) === claim) inFlightPackets.delete(dedupeKey);
   }
 }
 

backend/src/mqtt/dedupeRegistry.test.ts

@@ -0,0 +1,82 @@
+import assert from 'node:assert/strict';
+import test from 'node:test';
+import { waitForInFlightPacket } from './dedupeRegistry.js';
+
+interface ClaimHandle {
+  claim: Promise<void>;
+  resolve: () => void;
+  reject: (error: unknown) => void;
+}
+
+function claimHandle(): ClaimHandle {
+  let resolve!: () => void;
+  let reject!: (error: unknown) => void;
+  const claim = new Promise<void>((res, rej) => {
+    resolve = res;
+    reject = rej;
+  });
+  claim.catch(() => { /* observed by waiters or the test */ });
+  return { claim, resolve, reject };
+}
+
+test('claims (and registers) when no attempt is in flight', async () => {
+  const registry = new Map<string, Promise<void>>();
+  const mine = claimHandle();
+  assert.equal(await waitForInFlightPacket(registry, 'k', mine.claim), 'claimed');
+  assert.equal(registry.get('k'), mine.claim);
+});
+
+test('resolves as duplicate when the in-flight attempt succeeds', async () => {
+  const registry = new Map<string, Promise<void>>();
+  const holder = claimHandle();
+  holder.resolve();
+  registry.set('k', holder.claim);
+  const mine = claimHandle();
+  assert.equal(await waitForInFlightPacket(registry, 'k', mine.claim), 'duplicate');
+  assert.equal(registry.get('k'), holder.claim); // untouched
+});
+
+test('re-arbitrates after a failed attempt: exactly one successor claims', async () => {
+  const registry = new Map<string, Promise<void>>();
+  const first = claimHandle();
+  registry.set('k', first.claim);
+
+  const b = claimHandle();
+  const c = claimHandle();
+  const bResult = waitForInFlightPacket(registry, 'k', b.claim);
+  const cResult = waitForInFlightPacket(registry, 'k', c.claim);
+
+  // The failed attempt rejects; its cleanup (handler finally) removes it
+  // before waiters re-check, exactly as in production.
+  first.reject(new Error('db down'));
+  registry.delete('k');
+
+  assert.equal(await bResult, 'claimed'); // the successor claims...
+  assert.equal(registry.get('k'), b.claim); // ...and registers its own attempt.
+  b.resolve(); // successor's persist succeeds
+  assert.equal(await cResult, 'duplicate'); // the other copy is a duplicate, not a second write.
+});
+
+test('successive failures chain without double-claiming or losing the packet', async () => {
+  const registry = new Map<string, Promise<void>>();
+  const first = claimHandle();
+  registry.set('k', first.claim);
+
+  const b = claimHandle();
+  const bResult = waitForInFlightPacket(registry, 'k', b.claim);
+
+  first.reject(new Error('first failure'));
+  registry.delete('k');
+  assert.equal(await bResult, 'claimed');
+
+  // c starts waiting behind b's claim; b's attempt also fails.
+  const c = claimHandle();
+  const cResult = waitForInFlightPacket(registry, 'k', c.claim);
+  b.reject(new Error('second failure'));
+  registry.delete('k');
+  assert.equal(await cResult, 'claimed');
+
+  // The third attempt succeeds; exactly one persist wins overall.
+  c.resolve();
+  await c.claim;
+});

backend/src/mqtt/dedupeRegistry.ts

@@ -0,0 +1,35 @@
+/**
+ * Claim arbitration for the in-flight packet dedupe registry.
+ *
+ * Multiple copies of the same packet can arrive concurrently (broker
+ * redeliveries, relay copies fanned across observers, several ingest
+ * connections). Only one copy may be mid-persist at a time; everyone else
+ * waits for its outcome:
+ *
+ *  - the attempt succeeds -> every waiter resolves as a duplicate (no write);
+ *  - the attempt fails    -> waiters re-arbitrate. The re-check and the claim
+ *    registration happen in the SAME microtask inside this function, so
+ *    exactly one successor claims; every later waiter queues behind that
+ *    successor rather than racing it. This closes the race where two waiters
+ *    both fall through after a failure and persist the packet concurrently
+ *    (double insert: packets carry no unique constraint).
+ *
+ * Contract: the caller passes the promise that will settle when its own
+ * attempt settles (`claim`). On 'claimed', the caller MUST settle that
+ * promise and remove it from the registry when done.
+ */
+export async function waitForInFlightPacket(
+  registry: Map<string, Promise<void>>,
+  dedupeKey: string,
+  claim: Promise<void>,
+): Promise<'claimed' | 'duplicate'> {
+  for (;;) {
+    const holder = registry.get(dedupeKey);
+    if (!holder) {
+      registry.set(dedupeKey, claim);
+      return 'claimed';
+    }
+    const persisted = await holder.then(() => true, () => false);
+    if (persisted) return 'duplicate';
+  }
+}