diff --git a/backend/src/metrics.ts b/backend/src/metrics.ts index 2d6066c..8866bb1 100644 --- a/backend/src/metrics.ts +++ b/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.', +}); diff --git a/backend/src/mqtt/client.ts b/backend/src/mqtt/client.ts index 528e462..e4d983b 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 { 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) => 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>(); +// 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((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); } } diff --git a/backend/src/mqtt/dedupeRegistry.test.ts b/backend/src/mqtt/dedupeRegistry.test.ts new file mode 100644 index 0000000..45cdacd --- /dev/null +++ b/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; + resolve: () => void; + reject: (error: unknown) => void; +} + +function claimHandle(): ClaimHandle { + let resolve!: () => void; + let reject!: (error: unknown) => void; + const claim = new Promise((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>(); + 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>(); + 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>(); + 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>(); + 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; +}); diff --git a/backend/src/mqtt/dedupeRegistry.ts b/backend/src/mqtt/dedupeRegistry.ts new file mode 100644 index 0000000..97ffc70 --- /dev/null +++ b/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>, + dedupeKey: string, + claim: Promise, +): 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'; + } +}