From 29f9633db7d18979b7fe48af200fd934ec73441b Mon Sep 17 00:00:00 2001 From: gadgethd <111318106+gadgethd@users.noreply.github.com> Date: Sun, 11 Oct 2026 14:57:14 +0000 Subject: [PATCH] fix(ram): bounded quarantine for undecodable entries (no silent loss) Undecodable stream values were acknowledged and deleted without recovery evidence. They now go to a bounded quarantine stream (approx MAXLEN 500, raw value size-capped, digest + byte length recorded) before the original is acknowledged and deleted, so a malformed entry can neither block the stream nor disappear without a trace. The structured log carries only id, length and digest, and the quarantine counter is exported as meshcore_ram_ingest_quarantined_total. --- backend/src/metrics.ts | 6 +++++ backend/src/ram/protocol.ts | 5 +++++ backend/src/ram/writer.test.ts | 15 ++++++++++++- backend/src/ram/writer.ts | 41 +++++++++++++++++++++++++++++++++- 4 files changed, 65 insertions(+), 2 deletions(-) diff --git a/backend/src/metrics.ts b/backend/src/metrics.ts index fa2850c..abe6673 100644 --- a/backend/src/metrics.ts +++ b/backend/src/metrics.ts @@ -584,6 +584,12 @@ export const ramBatchDedupeTotal = new Counter({ registers: [metricsRegistry], }); +export const ramQuarantinedTotal = new Counter({ + name: 'meshcore_ram_ingest_quarantined_total', + help: 'Undecodable stream entries moved to the bounded quarantine stream.', + registers: [metricsRegistry], +}); + export const ramSweepDeletedTotal = new Counter({ name: 'meshcore_ram_ingest_sweep_deleted_total', help: 'Committed, acknowledged stream entries reclaimed by the retention sweep.', diff --git a/backend/src/ram/protocol.ts b/backend/src/ram/protocol.ts index 9e8dfe1..d6f782f 100644 --- a/backend/src/ram/protocol.ts +++ b/backend/src/ram/protocol.ts @@ -47,6 +47,11 @@ export function getRamQueueBytesKey(): string { return `${getRamStreamKey()}:bytes`; } +/** Bounded dead-letter stream for undecodable entries (recovery evidence). */ +export function getRamQuarantineKey(): string { + return `${getRamStreamKey()}:quarantine`; +} + export function getRamConsumerName(): string { const value = process.env['MQTT_RAM_CONSUMER_NAME']?.trim(); return value && value.length > 0 ? value : 'backend'; diff --git a/backend/src/ram/writer.test.ts b/backend/src/ram/writer.test.ts index 67531cc..43adc0f 100644 --- a/backend/src/ram/writer.test.ts +++ b/backend/src/ram/writer.test.ts @@ -99,6 +99,12 @@ class FakeRedis { return ids.length; } + quarantined: string[] = []; + async xaddWithLimit(_key: string, _maxLen: number, _field: string, value: string): Promise { + this.quarantined.push(value); + return 'q-1'; + } + async xautoclaim(): Promise<[string, Array<[string, string[]]>, string[]]> { this.claimCalls += 1; return ['0-0', [], []]; @@ -384,7 +390,7 @@ test('markers are consumed and acknowledged without persistence', async () => { assert.equal(fake.acked.includes(markerId), true); }); -test('unparseable stream values are acknowledged and deleted', async () => { +test('unparseable stream values are quarantined, acknowledged and deleted', async () => { const fake = new FakeRedis(); const now = () => 1_700_000_600_000; const id = `${1_700_000_600_000}-7`; @@ -397,6 +403,13 @@ test('unparseable stream values are acknowledged and deleted', async () => { assert.equal(calls.length, 0); assert.equal(fake.acked.includes(id), true); assert.equal(fake.deleted.includes(id), true); + assert.equal(fake.quarantined.length, 1, 'recovery evidence is preserved'); + const record = JSON.parse(fake.quarantined[0] ?? '{}') as Record; + assert.equal(record['src'], id); + assert.equal(record['bytes'], Buffer.byteLength('not-json')); + assert.equal(typeof record['sha256'], 'string'); + assert.equal((record['sha256'] as string).length, 64); + assert.equal(record['raw'], 'not-json'); }); test('drain flushes everything that remains', async () => { diff --git a/backend/src/ram/writer.ts b/backend/src/ram/writer.ts index 2a6ba39..16596e8 100644 --- a/backend/src/ram/writer.ts +++ b/backend/src/ram/writer.ts @@ -19,6 +19,7 @@ * backoff; a crashed writer's in-flight entries are reclaimed from the PEL * (own '0' reads + XAUTOCLAIM). */ +import { createHash } from 'node:crypto'; import { isRetryableDatabaseError, retryDelay, @@ -29,6 +30,7 @@ import { decodeRamStreamValue, encodeCommitMarker, entryFieldName, + getRamQuarantineKey, getRamQueueBytesKey, RAM_ENTRY_OVERHEAD_BYTES, type RamPendingEntry, @@ -38,6 +40,7 @@ import { ramCommitDuration, ramCommitTotal, ramBatchDedupeTotal, + ramQuarantinedTotal, ramReclaimedTotal, ramRetryTotal, ramSweepDeletedTotal, @@ -60,6 +63,8 @@ export interface RamWriterRedis { xack(key: string, group: string, ...ids: string[]): Promise; xdel(key: string, ...ids: string[]): Promise; xadd(key: string, id: string, field: string, value: string): Promise; + /** XADD with an approximate MAXLEN cap; used for the bounded quarantine stream. */ + xaddWithLimit(key: string, maxLen: number, field: string, value: string): Promise; xautoclaim( key: string, group: string, @@ -123,6 +128,9 @@ const MIN_RETENTION_MS = 5_000; const SWEEP_INTERVAL_MS = 10_000; const SWEEP_SCAN_LIMIT = 2_000; const PEL_LIST_LIMIT = 5_000; +/** Bounded recovery evidence for undecodable entries. */ +const QUARANTINE_MAX_LEN = 500; +const QUARANTINE_VALUE_CAP = 65_536; const FAILURE_BACKOFF_START_MS = 5_000; const FAILURE_BACKOFF_MAX_MS = 30_000; @@ -204,6 +212,35 @@ export function createRamWriter(deps: RamWriterDeps): RamWriter { return Buffer.byteLength(value) + RAM_ENTRY_OVERHEAD_BYTES; } + /** + * Undecodable entries are acknowledged so they can never block the stream, + * but not discarded silently: a bounded quarantine stream keeps the raw + * value (size-capped) plus its digest for recovery, while the structured + * log line carries only the id, byte length and digest — never content. + */ + async function quarantineMalformed(id: string, value: string): Promise { + ramQuarantinedTotal.inc(); + const bytes = Buffer.byteLength(value); + const digest = createHash('sha256').update(value).digest('hex'); + deps.log(`[ram] quarantined undecodable stream entry id=${id} bytes=${bytes} sha256=${digest}`); + try { + await deps.redis.xaddWithLimit( + getRamQuarantineKey(), + QUARANTINE_MAX_LEN, + 'q', + JSON.stringify({ + src: id, + at: deps.now(), + sha256: digest, + bytes, + raw: value.length <= QUARANTINE_VALUE_CAP ? value : '', + }), + ); + } catch (error) { + deps.log('[ram] quarantine write failed', error); + } + } + function appendEntries(entries: StreamEntry[]): number { let added = 0; for (const item of entries) { @@ -233,7 +270,9 @@ export function createRamWriter(deps: RamWriterDeps): RamWriter { for (const [id, fields] of items) { const decoded = decodeRamStreamValue(fields[1]); if (!decoded) { - // Unparseable value: acknowledge + delete so it can never block the stream. + // Unparseable value: preserve bounded recovery evidence, then + // acknowledge + delete so it can never block the stream. + await quarantineMalformed(id, fields[1]); await deps.redis.xack(deps.streamKey(), deps.group(), id).catch(() => 0); await deps.redis.xdel(deps.streamKey(), id).catch(() => 0); continue;