fix(ram): replay-safe idempotent commits with in-batch identity dedupe

- RAM writer commits idempotently from the first attempt: a crash after a
  successful commit but before Redis bookkeeping replays the SAME frozen
  entries on restart, so the identity existence check must apply immediately
  rather than only on internal retries.
- Idempotent SQL gains an intra-batch identity dedupe (row_number over the
  five-column identity key) so duplicate identities within one batch cannot
  double-insert.
- All insert paths are gated behind a transaction-scoped advisory lock so
  concurrent writers cannot both pass the NOT EXISTS check; contention is
  surfaced as lock_ok=false and retried by the caller.
- Writer collapses duplicate identities before persistence; collapsed count
  exported as meshcore_ram_ingest_batch_dedupe_total.

backend/src/db/packetBatch.test.ts

@@ -24,7 +24,7 @@ test('packet batches retry transient database failures without losing the pendin
         row_id: 0,
         is_private: false,
         visibility_ok: true,
-        prefix_cache_fresh: true,
+        prefix_cache_fresh: true, lock_ok: true,
       }],
     };
   });
@@ -63,7 +63,11 @@ test('packet batches retry transient database failures without losing the pendin
   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/);
+  assert.match(statements[1]!, /existing\.packet_hash = d\.packet_hash/);
+  assert.match(statements[1]!, /ROW_NUMBER\(\) OVER/);
+  assert.match(statements[1]!, /pg_try_advisory_xact_lock\(hashtext\(.meshcore:packet-batch-writer.\)\)/);
+  assert.match(statements[1]!, /locked AS MATERIALIZED/);
+  assert.match(statements[1]!, /AS lock_ok/);
   assert.match(statements[1]!, /INSERT INTO packets \(\s*observation_id,/);
   assert.match(statements[1]!, /COALESCE\(i\.observation_id_in, gen_random_uuid\(\)\) AS observation_id/);
   assert.match(statements[1]!, /SELECT observation_id, time, packet_hash/);

backend/src/db/packetBatch.ts

@@ -93,6 +93,7 @@ export type PersistRow = {
   is_private: boolean;
   visibility_ok: boolean;
   prefix_cache_fresh: boolean;
+  lock_ok: boolean;
 };
 
 type QueryFn = <T extends QueryResultRow = QueryResultRow>(
@@ -202,7 +203,7 @@ export function isRetryableDatabaseError(error: unknown): boolean {
     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);
+  return /connection terminated|database system is not yet accepting connections|server closed the connection unexpectedly|connection reset|connection refused|could not connect|PACKET_BATCH_LOCK_BUSY/i.test(message);
 }
 
 export function retryDelay(attempt: number): number {
@@ -331,21 +332,43 @@ export async function persistEntriesCore(
   }).join(',\n');
 
   const persistenceSource = idempotent ? 'new_rows' : 'classified';
+  // Replay safety (RAM-mode crash recovery): when running idempotent, a batch
+  // is filtered twice — first against duplicate identities WITHIN the batch
+  // (row-numbered by the same five-column key the existence check uses),
+  // then against rows already present in the database (covers replays after
+  // crashes and ambiguous commit outcomes, where the SAME frozen entry is
+  // delivered again with identical time/packet/rx/topic/network values).
   const idempotencyCte = idempotent
-    ? `new_rows AS MATERIALIZED (
-         SELECT c.*
+    ? `deduped AS MATERIALIZED (
+         SELECT c.*,
+                ROW_NUMBER() OVER (
+                  PARTITION BY c.time, c.packet_hash, COALESCE(c.rx_node_id, ''), c.topic, c.network
+                  ORDER BY c.row_id
+                ) AS identity_rank
          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
-         )
+       ),
+       new_rows AS MATERIALIZED (
+         SELECT d.*
+         FROM deduped d
+         WHERE d.identity_rank = 1
+           AND NOT EXISTS (
+             SELECT 1
+             FROM packets existing
+             WHERE existing.time = d.time
+               AND existing.packet_hash = d.packet_hash
+               AND existing.rx_node_id IS NOT DISTINCT FROM d.rx_node_id
+               AND existing.topic = d.topic
+               AND existing.network = d.network
+           )
        ),`
     : '';
+  // Serialize writers with a transaction-scoped advisory lock: NOT EXISTS
+  // alone cannot prevent two concurrent writers from both passing the
+  // existence check. Every insert is gated on the lock; when it is held the
+  // batch writes nothing and reports lock_ok=false so the caller retries.
+  const lockCte = `locked AS MATERIALIZED (
+         SELECT pg_try_advisory_xact_lock(hashtext('meshcore:packet-batch-writer')) AS ok
+       ),`;
 
   const result = await query<PersistRow>(
       `WITH current_visibility AS MATERIALIZED (
@@ -360,6 +383,7 @@ export async function persistEntriesCore(
             WHERE status IN ('pending', 'processing', 'failed')
          ) AS active
        ),
+       ${lockCte}
        incoming (
          row_id, observation_id_in, 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,
@@ -394,7 +418,8 @@ export async function persistEntriesCore(
                 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 ${persistenceSource}
+         FROM ${persistenceSource} CROSS JOIN locked
+         WHERE locked.ok
          RETURNING 1
        ),
        path_inserted AS (
@@ -406,8 +431,8 @@ export async function persistEntriesCore(
          SELECT time, packet_hash, rx_node_id, src_node_id, topic, topic_prefix,
                 route_type, hop_count, rssi, snr, path_hashes, path_hash_size_bytes,
                 is_private, path_is_valid AND NOT is_private, network, observation_id
-         FROM ${persistenceSource}
-         WHERE path_hashes IS NOT NULL AND path_hash_size_bytes > 1
+         FROM ${persistenceSource} CROSS JOIN locked
+         WHERE locked.ok AND path_hashes IS NOT NULL AND path_hash_size_bytes > 1
          ON CONFLICT DO NOTHING
          RETURNING 1
        ),
@@ -418,8 +443,8 @@ export async function persistEntriesCore(
          )
          SELECT DISTINCT ON (rx_node_id)
            rx_node_id, iata, iata, time, time, TRUE, network, time
-         FROM ${persistenceSource}
-         WHERE rx_node_id IS NOT NULL AND rx_node_id <> ''
+         FROM ${persistenceSource} CROSS JOIN locked
+         WHERE locked.ok AND rx_node_id IS NOT NULL AND rx_node_id <> ''
          ORDER BY rx_node_id, time DESC
          ON CONFLICT (node_id) DO UPDATE SET
            last_rx_at = GREATEST(nodes.last_rx_at, EXCLUDED.last_rx_at),
@@ -442,8 +467,8 @@ export async function persistEntriesCore(
          INSERT INTO node_network_sightings
            (node_id, network, first_seen_at, last_seen_at)
          SELECT src_node_id, network, MIN(time), MAX(time)
-         FROM ${persistenceSource}
-         WHERE src_node_id IS NOT NULL
+         FROM ${persistenceSource} CROSS JOIN locked
+         WHERE locked.ok AND src_node_id IS NOT NULL
            AND src_node_id <> ''
            AND network <> 'test'
          GROUP BY src_node_id, network
@@ -465,8 +490,8 @@ export async function persistEntriesCore(
            hop_count,
            packet_hash,
            time
-         FROM ${persistenceSource}
-         WHERE path_is_valid
+         FROM ${persistenceSource} CROSS JOIN locked
+         WHERE locked.ok AND path_is_valid
            AND NOT is_private
            AND hop_count IS NOT NULL
          ORDER BY network, hop_count DESC, row_id ASC
@@ -505,8 +530,8 @@ export async function persistEntriesCore(
            COALESCE(SUM(snr), 0)::double precision,
            COUNT(snr)::bigint,
            NOW()
-         FROM ${persistenceSource}
-         WHERE network = 'test'
+         FROM ${persistenceSource} CROSS JOIN locked
+         WHERE locked.ok AND network = 'test'
             OR topic_prefix <> 'meshcore-test'
          GROUP BY
            network,
@@ -532,8 +557,8 @@ export async function persistEntriesCore(
          INSERT INTO observer_region_packet_sightings
            (network, iata, packet_hash, first_seen, last_seen)
          SELECT network, iata, packet_hash, MIN(time), MAX(time)
-         FROM ${persistenceSource}
-         WHERE network <> 'test'
+         FROM ${persistenceSource} CROSS JOIN locked
+         WHERE locked.ok AND network <> 'test'
            AND path_is_valid
            AND NOT is_private
            AND iata IS NOT NULL
@@ -556,8 +581,8 @@ export async function persistEntriesCore(
          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 ${persistenceSource}
-         WHERE network <> 'test'
+         FROM ${persistenceSource} CROSS JOIN locked
+         WHERE locked.ok AND network <> 'test'
            AND path_is_valid
            AND NOT is_private
            AND iata IS NOT NULL
@@ -577,6 +602,7 @@ export async function persistEntriesCore(
        )
        SELECT row_id, is_private, path_is_valid AND NOT is_private AS visibility_ok,
               prefix_cache_fresh,
+              (SELECT ok FROM locked) AS lock_ok,
               (SELECT COUNT(*) FROM inserted) AS inserted_count,
               (SELECT COUNT(*) FROM observer_updates) AS observer_update_count,
               (SELECT COUNT(*) FROM network_sightings) AS network_sighting_count,
@@ -590,6 +616,9 @@ export async function persistEntriesCore(
     );
 
   if (result.rows.length !== entries.length) throw new Error('PACKET_BATCH_RESULT_MISMATCH');
+  if (result.rows.some((row) => !row.lock_ok)) {
+    throw new Error('PACKET_BATCH_LOCK_BUSY: another writer holds the packet batch lock');
+  }
   if (result.rows.some((row) => !row.prefix_cache_fresh)) {
     void refreshPrivatePrefixCache(query).catch((error: unknown) => {
       console.error('[db] private prefix cache refresh failed', (error as Error).message);

backend/src/metrics.ts

@@ -578,6 +578,12 @@ export const ramRetryTotal = new Counter({
   registers: [metricsRegistry],
 });
 
+export const ramBatchDedupeTotal = new Counter({
+  name: 'meshcore_ram_ingest_batch_dedupe_total',
+  help: 'Replay-duplicate entries collapsed within a RAM ingest flush.',
+  registers: [metricsRegistry],
+});
+
 export const ramReclaimedTotal = new Counter({
   name: 'meshcore_ram_ingest_reclaimed_total',
   help: 'Un-acked stream entries reclaimed by a (re)started writer.',

backend/src/ram/writer.test.ts

@@ -176,7 +176,7 @@ test('flushes when the batch reaches the size trigger and bookkeeps the stream',
 
   assert.equal(calls.length, 1);
   assert.equal(calls[0]?.count, 3);
-  assert.equal(calls[0]?.idempotent, false);
+  assert.equal(calls[0]?.idempotent, true); // replay-safe from the first attempt
   assert.equal(fake.acked.length, 3);
   assert.equal(fake.deleted.length, 3);
   assert.equal(fake.pending.size, 0);
@@ -218,7 +218,7 @@ test('does not flush a small, young batch', async () => {
   assert.equal(writer.accumulatorSize(), 1);
 });
 
-test('transient commit failure retries within the flush and uses idempotent retries', async () => {
+test('transient commit failure retries within the flush (idempotent throughout)', async () => {
   const fake = new FakeRedis();
   const now = () => 1_700_000_300_000;
   fake.seedEntry(sampleEntry(now() - 1_000));
@@ -234,7 +234,7 @@ test('transient commit failure retries within the flush and uses idempotent retr
 
   assert.equal(attempts, 2);
   assert.equal(calls.length, 2);
-  assert.equal(calls[0]?.idempotent, false);
+  assert.equal(calls[0]?.idempotent, true);
   assert.equal(calls[1]?.idempotent, true);
   assert.equal(fake.acked.length, 2);
 });
@@ -271,6 +271,23 @@ test('exhausted commit failure leaves entries un-acked for later recovery', asyn
   assert.equal(calls[calls.length - 1]?.count, 1);
 });
 
+test('entries sharing one observation identity collapse within a flush', async () => {
+  const fake = new FakeRedis();
+  const now = () => 1_700_000_800_000;
+  // Same identity: identical time (frozen in the entry input), hash, rx, topic, network.
+  fake.seedEntry(sampleEntry(now() - 2_000, 'ab'.repeat(32)));
+  fake.seedEntry(sampleEntry(now() - 1_500, 'ab'.repeat(32)));
+
+  const { writer, calls } = makeWriter(fake, now, async () => undefined, { batchSize: 2 });
+  await writer.start();
+  await writer.tickOnce();
+
+  assert.equal(calls.length, 1);
+  assert.equal(calls[0]?.count, 1, 'duplicate identity must be collapsed before persistence');
+  assert.equal(fake.acked.length, 2, 'both originals are acknowledged (same observation)');
+  assert.equal(writer.accumulatorSize(), 0);
+});
+
 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

@@ -17,6 +17,7 @@
 import {
   isRetryableDatabaseError,
   retryDelay,
+  type PacketBatchInput,
   type PersistEntry,
 } from '../db/packetBatch.js';
 import {
@@ -29,6 +30,7 @@ import {
   ramCommitBatchSize,
   ramCommitDuration,
   ramCommitTotal,
+  ramBatchDedupeTotal,
   ramReclaimedTotal,
   ramRetryTotal,
   ramVisibilityLatency,
@@ -95,6 +97,33 @@ const STATS_INTERVAL_MS = 5_000;
 const FAILURE_BACKOFF_START_MS = 5_000;
 const FAILURE_BACKOFF_MAX_MS = 30_000;
 
+/** Identity used for intra-batch replay dedupe (mirrors the SQL existence check). */
+export function persistIdentityKey(input: PacketBatchInput): string {
+  return [
+    input.time.toISOString(),
+    input.packetHash,
+    input.rxNodeId ?? '',
+    input.topic,
+    input.network,
+  ].join('\u0000');
+}
+
+/** Collapse entries that share a single observation identity within one flush. */
+export function collapseBatchEntries(batch: StreamEntry[]): StreamEntry[] {
+  if (batch.length < 2) return batch;
+  const seen = new Set<string>();
+  const kept: StreamEntry[] = [];
+  for (const item of batch) {
+    const key = persistIdentityKey(item.entry.input);
+    if (seen.has(key)) continue;
+    seen.add(key);
+    kept.push(item);
+  }
+  const dropped = batch.length - kept.length;
+  if (dropped > 0) ramBatchDedupeTotal.inc(dropped);
+  return kept;
+}
+
 /** Pure flush trigger used by the loop and by unit tests. */
 export function flushDecision(
   acc: Array<{ entry: RamPendingEntry }>,
@@ -199,7 +228,10 @@ export function createRamWriter(deps: RamWriterDeps): RamWriter {
           input: entry.input,
           observationId: entry.observationId,
         }));
-        await deps.persist(entries, attempt > 0);
+        // Always idempotent: a replay after a crash (or an ambiguous commit)
+        // re-delivers the SAME frozen entries, and the existence check keyed
+        // on their identity must apply from the first attempt.
+        await deps.persist(entries, true);
         return;
       } catch (error) {
         attempt += 1;
@@ -217,7 +249,7 @@ export function createRamWriter(deps: RamWriterDeps): RamWriter {
     const startedNs = process.hrtime.bigint();
     const nowMs = deps.now();
     try {
-      await persistWithRetry(batch);
+      await persistWithRetry(collapseBatchEntries(batch));
     } 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.