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

@@ -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,

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<void> {
+};
+
+/**
+ * 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<void> {
   const { finalHash, decodedPathHashSizeBytes, path, decodedRouteType, network, resolvedPacketType } = args;
 
   // A repeater appearing in a multibyte (2–3 byte) path hash almost certainly

backend/src/ram/writer.test.ts

@@ -151,7 +151,7 @@ function makeWriter(
   fake: FakeRedis,
   now: () => number,
   persistImpl: (entries: unknown[], idempotent: boolean) => Promise<void>,
-  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;

backend/src/ram/writer.ts

@@ -102,6 +102,8 @@ export type RamWriterDeps = {
   consumer: () => string;
   statsTick: (() => Promise<void>) | 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<void> {

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<void> {
         });
       }
     },
+    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 ?? ''));