From 8d9da1b7c7d35a5e303966ce992d94b05fbcc2f0 Mon Sep 17 00:00:00 2001 From: gadgethd <111318106+gadgethd@users.noreply.github.com> Date: Sun, 11 Oct 2026 14:56:22 +0000 Subject: [PATCH] fix(ram): defer commit-dependent follow-ups to the writer post-commit hook In RAM mode persistOnce returns at admission, so the post-ingest callbacks that previously ran after durable persistence would fire against rows that are not in PostgreSQL yet: - the link queue (link worker reads packet rows) now enqueues from the RAM writer after the batch commit, and the live packet subscriber skips it in RAM mode; - the live-map follow-ups (multibyte path evidence, slow-mode scheduling, pre-resolution) move to the same post-commit hook, sharing the durable path in-flight bound and drop counters via schedulePostPersistFollowUps; - changes that are safe at admission stay there: WS broadcasts, node emits, resolve-cache invalidation and owner forwarding (self-contained envelope). Regression test: the hook fires with the persisted batch after bookkeeping, and never on failed commits. --- backend/src/index.ts | 4 ++- backend/src/mqtt/client.ts | 54 ++++++++++++++++++++++---------- backend/src/ram/writer.test.ts | 37 +++++++++++++++++++++- backend/src/ram/writer.ts | 12 ++++++- backend/src/ram/writerRuntime.ts | 33 ++++++++++++++++++- 5 files changed, 120 insertions(+), 20 deletions(-) diff --git a/backend/src/index.ts b/backend/src/index.ts index 2f85d94..230bcd3 100644 --- a/backend/src/index.ts +++ b/backend/src/index.ts @@ -233,7 +233,9 @@ async function main() { console.error('[packet-share] packet handling failed:', error instanceof Error ? error.message : error); }); } - if (packet.path?.length && packet.rxNodeId) { + // RAM mode defers commit-dependent link admission to the writer's + // post-commit hook (the link worker reads packet rows from PostgreSQL). + if (getRamIngestMode() !== 'ram' && packet.path?.length && packet.rxNodeId) { void queueLinkJob( packet.packetHash, packet.rxNodeId, diff --git a/backend/src/mqtt/client.ts b/backend/src/mqtt/client.ts index b274cf2..8e99911 100644 --- a/backend/src/mqtt/client.ts +++ b/backend/src/mqtt/client.ts @@ -922,26 +922,19 @@ async function handleMessage(topic: string, rawPayload: Buffer, redelivered = fa // 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({ + // RAM mode releases the MQTT acknowledgement at admission, far earlier + // than the batch commit. Commit-dependent live-map work must not run + // against rows that are not in PostgreSQL yet, so it is deferred to the + // RAM writer's post-commit hook (see ram/writerRuntime.ts). The durable + // path keeps running it right here, after the committed write. + if (MQTT_INGEST_MODE !== 'ram') { + schedulePostPersistFollowUps({ 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); }); } })(); @@ -963,14 +956,43 @@ async function handleMessage(topic: string, rawPayload: Buffer, redelivered = fa * warm-up all resolve after the durable write and run concurrently with the * next message's persistence. */ -async function runPostPersistFollowUps(args: { +export type PostPersistFollowUpArgs = { finalHash: string; decodedPathHashSizeBytes: number | undefined; path: string[] | undefined; decodedRouteType: number | undefined; network: string; resolvedPacketType: number | undefined; -}): Promise { +}; + +/** + * Run the best-effort live-map follow-ups under the global in-flight bound. + * Shared by the durable ingest path (after the committed write) and by the + * RAM writer's post-commit hook, so a stall cannot grow the in-flight set + * without limit in either mode; over the bound they are dropped (best-effort) + * and counted. + */ +export function schedulePostPersistFollowUps(args: PostPersistFollowUpArgs): void { + if (postPersistInFlight >= POST_PERSIST_MAX_INFLIGHT) { + postPersistDroppedTotal.inc(); + return; + } + postPersistInFlight++; + postPersistInFlightGauge.set(postPersistInFlight); + void runPostPersistFollowUps(args) + .catch((err: unknown) => { + console.error( + '[mqtt] post-persist follow-up failed:', + err instanceof Error ? err.message : String(err), + ); + }) + .finally(() => { + postPersistInFlight--; + postPersistInFlightGauge.set(postPersistInFlight); + }); +} + +async function runPostPersistFollowUps(args: PostPersistFollowUpArgs): Promise { const { finalHash, decodedPathHashSizeBytes, path, decodedRouteType, network, resolvedPacketType } = args; // A repeater appearing in a multibyte (2–3 byte) path hash almost certainly diff --git a/backend/src/ram/writer.test.ts b/backend/src/ram/writer.test.ts index 9ab3902..67531cc 100644 --- a/backend/src/ram/writer.test.ts +++ b/backend/src/ram/writer.test.ts @@ -151,7 +151,7 @@ function makeWriter( fake: FakeRedis, now: () => number, persistImpl: (entries: unknown[], idempotent: boolean) => Promise, - opts: { batchSize?: number; maxAgeMs?: number } = {}, + opts: { batchSize?: number; maxAgeMs?: number; onCommitted?: ((entries: unknown[]) => void) | null } = {}, ) { const calls: PersistCall[] = []; let healthy = 0; @@ -169,6 +169,7 @@ function makeWriter( group: () => 'ingest-writers', consumer: () => 'backend', statsTick: null, + onCommitted: opts.onCommitted ?? null, markHealthy: () => { healthy += 1; }, @@ -336,6 +337,40 @@ test('entries sharing one observation identity collapse within a flush', async ( assert.equal(writer.accumulatorSize(), 0); }); +test('post-commit hook fires for the committed batch only after success', async () => { + const fake = new FakeRedis(); + const now = () => 1_700_000_700_000; + fake.seedEntry(sampleEntry(now() - 1_000)); + fake.seedEntry(sampleEntry(now() - 900, 'cd'.repeat(32))); + + const seen: unknown[][] = []; + const { writer } = makeWriter(fake, now, async () => undefined, { + batchSize: 2, + onCommitted: (entries) => seen.push(entries), + }); + await writer.start(); + await writer.tickOnce(); + + assert.equal(seen.length, 1); + assert.equal(seen[0]?.length, 2, 'hook receives the persisted batch'); + assert.equal(fake.acked.length, 2, 'hook runs after the acknowledgement bookkeeping'); + + // A failing commit must not fire the hook. + fake.seedEntry(sampleEntry(now() - 800, 'ef'.repeat(32))); + fake.seedEntry(sampleEntry(now() - 700, '12'.repeat(32))); + const failing = makeWriter( + fake, + now, + async () => { + throw new Error('connection terminated unexpectedly'); + }, + { batchSize: 2, onCommitted: (entries) => seen.push(entries) }, + ); + await failing.writer.start(); + await failing.writer.tickOnce(); + assert.equal(seen.length, 1, 'no hook on failed commits'); +}); + test('markers are consumed and acknowledged without persistence', async () => { const fake = new FakeRedis(); const now = () => 1_700_000_500_000; diff --git a/backend/src/ram/writer.ts b/backend/src/ram/writer.ts index d6906fd..2a6ba39 100644 --- a/backend/src/ram/writer.ts +++ b/backend/src/ram/writer.ts @@ -102,6 +102,8 @@ export type RamWriterDeps = { consumer: () => string; statsTick: (() => Promise) | null; markHealthy: () => void; + /** Runs after a batch commits (bookkeeping included); receives the persisted batch. */ + onCommitted: ((entries: StreamEntry[]) => void) | null; log: (message: string, error?: unknown) => void; }; @@ -298,8 +300,9 @@ export function createRamWriter(deps: RamWriterDeps): RamWriter { const ids = batch.map((item) => item.id); const startedNs = process.hrtime.bigint(); const nowMs = deps.now(); + const persistBatch = collapseBatchEntries(batch); try { - await persistWithRetry(collapseBatchEntries(batch)); + await persistWithRetry(persistBatch); } catch (error) { // Leave the entries un-acked so they stay recoverable in the PEL; drop // them from memory and back off so a database outage cannot spin. @@ -357,6 +360,13 @@ export function createRamWriter(deps: RamWriterDeps): RamWriter { ramVisibilityLatency.observe(Math.max(0, (finishedMs - item.entry.admittedAtMs) / 1000)); } deps.markHealthy(); + if (deps.onCommitted) { + try { + deps.onCommitted(persistBatch); + } catch (error) { + deps.log('[ram] post-commit hook failed', error); + } + } } async function pruneMarkers(): Promise { diff --git a/backend/src/ram/writerRuntime.ts b/backend/src/ram/writerRuntime.ts index 6ebb5ab..9cb6baf 100644 --- a/backend/src/ram/writerRuntime.ts +++ b/backend/src/ram/writerRuntime.ts @@ -9,7 +9,8 @@ import { Redis } from 'ioredis'; import { getRedisConnectionOptions, getRedisUrl } from '../platform/config/redis.js'; import { query } from '../db/index.js'; import { persistEntriesCore, refreshAdmissionPrivacyCache } from '../db/packetBatch.js'; -import { markMqttPersistenceHealthy } from '../mqtt/client.js'; +import { markMqttPersistenceHealthy, schedulePostPersistFollowUps } from '../mqtt/client.js'; +import { queueLinkJob } from '../queue/publisher.js'; import { refreshPendingStats } from './liveStore.js'; import { getRamBatchSize, @@ -64,6 +65,36 @@ export async function startRamWriter(): Promise { }); } }, + onCommitted: (entries) => { + // Commit-dependent live-map work, mirroring the durable post-persist + // path: the link queue and the live-map follow-ups both read packet + // rows from PostgreSQL, so they run here — after the batch commit. + for (const { entry } of entries) { + const input = entry.input; + if (input.pathHashes && input.pathHashes.length > 0 && input.rxNodeId) { + void queueLinkJob( + input.packetHash, + input.rxNodeId, + input.srcNodeId ?? undefined, + input.pathHashes, + input.hopCount ?? undefined, + input.pathHashSizeBytes ?? undefined, + ).then((admission) => { + if (admission && ['full', 'oversized', 'worker_unavailable'].includes(admission.status)) { + console.warn('[link-queue] live observation not admitted', admission.status, input.packetHash); + } + }).catch((error: Error) => console.error('[link-queue] live admission failed', error.message)); + } + schedulePostPersistFollowUps({ + finalHash: input.packetHash, + decodedPathHashSizeBytes: input.pathHashSizeBytes ?? undefined, + path: input.pathHashes ?? undefined, + decodedRouteType: input.routeType ?? undefined, + network: input.network, + resolvedPacketType: input.packetType ?? undefined, + }); + } + }, markHealthy: markMqttPersistenceHealthy, log: (message, error) => { console.error(message, error instanceof Error ? error.message : (error ?? ''));