From c683acdec8572e6b6dbc5ba07ccdfcbb062f10ce Mon Sep 17 00:00:00 2001 From: gadgethd <111318106+gadgethd@users.noreply.github.com> Date: Sun, 11 Oct 2026 22:26:54 +0000 Subject: [PATCH] fix(ingest): enforce F01 public batches and F05/F12/F13 read parity Expose the F16 credentialed public source for revocable CoreScope projections; preserve stable identities, generation fences and bounded overlap paging. --- FIX-PLAN.md | 28 +++ FIX-REPORT.md | 64 +++++++ backend/package.json | 2 +- backend/src/api/routes/misc.ts | 6 +- backend/src/db/index.ts | 46 +++-- .../src/db/packetBatch.integration.test.ts | 15 +- backend/src/db/packetBatch.ts | 15 +- backend/src/db/recentRead.integration.test.ts | 133 ++++++++++++++ backend/src/index.ts | 2 + backend/src/metrics.ts | 7 + backend/src/mqtt/client.ts | 34 ++-- backend/src/privacy/publicBatch.test.ts | 41 +++++ backend/src/privacy/publicBatch.ts | 63 +++++++ backend/src/ram/liveRead.ts | 69 +++++--- backend/src/ram/liveStore.test.ts | 68 +++++++- backend/src/ram/liveStore.ts | 3 +- backend/src/ram/publicSource.test.ts | 41 +++++ backend/src/ram/publicSource.ts | 59 +++++++ backend/src/ws/privacy.test.ts | 19 +- backend/src/ws/privacy.ts | 28 +++ .../publicAuthorization.integration.test.ts | 101 +++++++++++ backend/src/ws/server.ts | 163 +++++++----------- docker-compose.yml | 1 + docs/ram-first-ingestion.md | 39 ++++- 24 files changed, 871 insertions(+), 176 deletions(-) create mode 100644 backend/src/db/recentRead.integration.test.ts create mode 100644 backend/src/privacy/publicBatch.test.ts create mode 100644 backend/src/privacy/publicBatch.ts create mode 100644 backend/src/ram/publicSource.test.ts create mode 100644 backend/src/ram/publicSource.ts create mode 100644 backend/src/ws/publicAuthorization.integration.test.ts diff --git a/FIX-PLAN.md b/FIX-PLAN.md index 5f3d557..2811961 100644 --- a/FIX-PLAN.md +++ b/FIX-PLAN.md @@ -1,5 +1,33 @@ # UKMesh ingestion repair plan — 2026-10-11 +## Current authorization/read wave + +The working tree adds F01 generation fences to SQL aggregates, WS initial +cache delivery and bounded live batches. F12/F13 parity now includes the +requested SQL page plus aggregates for captured overlap hashes, final paging +after merge, canonical aliases, stable-ID counts, representative ranking and +millisecond/tie ordering. The real fixture also found an F05 interaction: +natural-identity replay suppression must apply only to entries without a +stable observation ID; distinct stable IDs are arbitrated by the ledger. + +CoreScope public mode uses a dedicated revocable SQLite projection and private +ingestor replay journal. The read-only server checks upstream authority before +and after API reads and before WS fanout. Generation changes rebuild copied +measurement/derived data and replace caches. Existing untracked/mixed data is +preserved and initialization refused. The filtered credentialed HTTP live +source removes the public companion's raw Redis credential; PostgreSQL's +committed-order ledger remains required. This stays within the review's +authorization, identity and handoff design. + +Focused and real regressions pass, including the ready WS 800-message gate +and the paired crash gate through the projection. RAM typecheck and full +backend tests pass 476/476 (202.440s); integration passes 7 PG + 2 Redis + +1 WS test, with zero skips. CoreScope server passes (103.965s); its final +ingestor suite remains running. After committing/pushing and green CI, continue +F09/F14 loaded headroom/accounting, F15 held runtime shutdown and F16 actual +principal isolation/drain, then final source-labelled images and whole-stack +performance. The overall mission remains active. + Specification: `/home/ben/ukmesh-fix-context-20261011/SOL-REVIEW.md`, especially findings F01–F17, §§4, 7–9. This plan records implementation order and will be updated as evidence changes. The independent review takes precedence over the earlier fix checklist. ## Boundaries and sequence diff --git a/FIX-REPORT.md b/FIX-REPORT.md index 4df17be..8d78bbc 100644 --- a/FIX-REPORT.md +++ b/FIX-REPORT.md @@ -2,6 +2,70 @@ Specification: `/home/ben/ukmesh-fix-context-20261011/SOL-REVIEW.md`. This is a working record; **the full repair objective is not yet achieved**. Status below reflects inspected code and completed verification only. +## Current F01/F05/F12/F13/F16 working-tree wave + +The RAM documentation checkpoint `c98c2f53e2921bed017fa1d0ecc2016f20a5c374` +is pushed; its workflows +[38177795874](https://github.com/gadgethd/ukmesh/actions/runs/38177795874) and +[38177798421](https://github.com/gadgethd/ukmesh/actions/runs/38177798421) +both concluded success. The runtime changes described next are newer and +await their own commit/CI; earlier historical evidence below is unchanged. + +F01 SQL aggregates now carry the snapshot generation and fail closed when +current authority differs. WS initial caches recheck generation before delivery. +MQTT and WS live events use bounded batches with one current authority query, +including active-payload byte reservations and unavailable/overflow metrics. +The actual ready WebSocket test reproduced the old privacy bypass, then passed +observer/path revocation, authority failure, cached reconnect refusal and +**800/800** allowed deliveries. This proves that isolated ready-reader gate, +not whole-stack MQTT performance. + +F12/F13 real SQL/RAM comparisons cover family and unrelated network scopes, +alias/canonical/other observers, full/slim fields, RX/TX and repeated sightings, +representative replacement, five-minute filtering, message rows and page limits. +SQL includes aggregates for captured overlap hashes so a pending representative +moving onto page one retains older committed observations. Dates retain +milliseconds; ties use observation IDs and packet hashes. The fixture exposed +and repaired F05's stable-ID/natural-dedupe interaction: two distinct IDs with +identical fields remain separate observations. Legacy entries without IDs +retain the natural-identity fallback. A genuine older Repeatable Read snapshot +is rejected after the authority generation changes. + +CoreScope's uncommitted F01/F16 implementation requires a dedicated SQLite +public projection and private ingestor provenance journal. On a policy change +it fences readiness, removes revoked journal records and reconstructs copied +and derived measurements. The server stays read-only and checks authority +before and after buffered API reads and before bounded WS fanout; cache/index +generations are replaced. Old untracked data is preserved and fenced instead +of silently purged. Duplicate-source enrichment preserves participant evidence. +Its actual PostgreSQL-authority/SQLite regression covers observer, source and +all one/two/three-byte prefix revocation across the UK family while preserving +an unrelated network; the final full-suite result is pending. + +F16 adds `/internal/ingest/public/{policy,entries}`, disabled unless a bearer +secret of at least 32 characters is configured. Every captured read filters +against current authority; private entries advance only the opaque cursor. +CoreScope public mode accepts this HTTP source and rejects raw Redis URL/DSN +configuration, requiring the paired PG ledger. Tests cover credentials, limits, +cursor ordering, unavailable authority and projection crash/replay barriers. +Operational principal isolation, loaded headroom, final drain/rollback and +whole-stack performance remain required. + +Owned evidence is under +`verification-sol-20261011/evidence/`: `ukmesh-ws-authority-baseline-ready.log` +and `ukmesh-ws-authority-after.log`, `ukmesh-read-paging-before.log` / +`ukmesh-read-paging-stable-id.log`, `ukmesh-f13-ties-before.log`, +`core-f01-copied-before.log`, `core-f01-untracked-node-before-valid.log`, +`core-f01-enrichment-before.log`, and `core-public-paired-final2.log`. +The final full backend run passes **476/476** in **202.440s**, including the +alert receiver; typecheck passes. The complete integration command passes +**7 PostgreSQL + 2 Redis + 1 WS tests**, zero skips, after giving the legacy +empty-schema fixture its own database. Evidence is +`ukmesh-public-final-full-tests.log` / `ukmesh-public-final-full-integration.log`. +CoreScope's server suite passes in **103.965s**; its final ingestor gate is +still running. The expanded image harness requires exactly ten passing tests. +Production and the original bench services remain unchanged. + ## Branches and gates **Successor progress, 2026-10-11.** The authoritative remote checkpoints were diff --git a/backend/package.json b/backend/package.json index edfa3cb..d04bfe0 100644 --- a/backend/package.json +++ b/backend/package.json @@ -21,7 +21,7 @@ "load:realtime": "tsx src/tools/loadRealtime.ts", "test": "node --import tsx --test $(find src -name '*.test.ts' ! -name '*.integration.test.ts' -print) && bash ../scripts/test-alert-receiver.sh", "test:analysis-integration": "node --import tsx --test src/analysis/runState.integration.test.ts", - "test:ingest-integration": "timeout 300s node --import tsx --test src/db/packetBatch.integration.test.ts && timeout 90s node --import tsx --test src/ram/ram.integration.test.ts", + "test:ingest-integration": "timeout 300s node --import tsx --test src/db/packetBatch.integration.test.ts src/db/recentRead.integration.test.ts && timeout 90s node --import tsx --test src/ram/ram.integration.test.ts && timeout 60s node --import tsx --test src/ws/publicAuthorization.integration.test.ts", "typecheck": "tsc --noEmit", "contract:generate": "tsx src/tools/generateOpenApi.ts", "contract:check": "tsx src/tools/generateOpenApi.ts --check", diff --git a/backend/src/api/routes/misc.ts b/backend/src/api/routes/misc.ts index 1811ae9..6e13b0d 100644 --- a/backend/src/api/routes/misc.ts +++ b/backend/src/api/routes/misc.ts @@ -30,6 +30,8 @@ type GetRecentPacketsFn = ( observer?: string, fields?: 'full' | 'slim', signal?: AbortSignal, + readQuery?: QueryFn, + overlapHashes?: string[], ) => Promise; type GetRecentPacketEventsFn = (limit: number, network?: string, observer?: string) => Promise; type GetPacketDetailFn = (hash: string, network?: string) => Promise; @@ -92,8 +94,8 @@ export function registerMiscRoutes(router: Router, deps: MiscRouteDeps): void { const packets = raw ? await getRecentPacketEvents(limit, network, observer) : await readRecentWithPending( - () => getRecentPackets( - limit, network, observer, fields, AbortSignal.timeout(30_000), + (overlapHashes) => getRecentPackets( + limit, network, observer, fields, AbortSignal.timeout(30_000), undefined, overlapHashes, ) as Promise[]>, { limit, network, observer, fields }, ); diff --git a/backend/src/db/index.ts b/backend/src/db/index.ts index 9d1d05d..9a51126 100644 --- a/backend/src/db/index.ts +++ b/backend/src/db/index.ts @@ -11,6 +11,7 @@ import { configurePacketBatch, enqueuePacket, type PacketBatchResult, + type QueryFn, } from './packetBatch.js'; import { nodeAliasArraySql, publicPacketPrivacySql } from '../api/utils/networkFilters.js'; import { @@ -814,13 +815,21 @@ export async function getRecentPackets( observer?: string, fields: 'full' | 'slim' = 'full', signal?: AbortSignal, + readQuery: QueryFn = query, + overlapHashes: string[] = [], ) { const includeObservationIds = getRamIngestMode() === 'ram'; const scope = buildScopePlaceholders(2, network, observer); + const overlapParam = includeObservationIds && overlapHashes.length > 0 ? `$${2 + scope.params.length}` : null; const fiveMinAgo = 'NOW() - INTERVAL \'5 minutes\''; - const res = await query( + const res = await readQuery( `WITH recent_packets AS ( SELECT DISTINCT ON (p.packet_hash) + ${includeObservationIds ? `(CASE WHEN p.payload ? 'appData' THEN 16 ELSE 0 END + + CASE WHEN p.src_node_id IS NOT NULL THEN 8 ELSE 0 END + + CASE WHEN p.advert_count IS NOT NULL THEN 4 ELSE 0 END + + CASE WHEN p.packet_type = 4 THEN 2 ELSE 0 END) AS _representative_rank,` : ''} + ${includeObservationIds ? 'p.observation_id::text AS _representative_id,' : ''} p.time, p.packet_hash, COALESCE(pa_rx.canonical_node_id, upper(btrim(p.rx_node_id))) AS rx_node_id, COALESCE(pa_src.canonical_node_id, upper(btrim(p.src_node_id))) AS src_node_id, p.topic, @@ -840,7 +849,7 @@ export async function getRecentPackets( CASE WHEN p.src_node_id IS NOT NULL THEN 1 ELSE 0 END DESC, CASE WHEN p.advert_count IS NOT NULL THEN 1 ELSE 0 END DESC, CASE WHEN p.packet_type = 4 THEN 1 ELSE 0 END DESC, - p.time DESC + p.time DESC, p.observation_id DESC NULLS LAST ), packet_stats AS ( SELECT @@ -868,13 +877,17 @@ export async function getRecentPackets( ${fields === 'full' ? 'rp.topic, rp.topic_prefix, rp.iata, rp.route_type, rp.network, rp.transport_codes, rp.region_scope,' : ''} - ${includeObservationIds ? 'ps._observation_ids, rp.network AS _policy_network,' : ''} + ${includeObservationIds ? `ps._observation_ids, rp.network AS _policy_network, + (SELECT generation::text FROM public_visibility_state WHERE singleton) AS _policy_generation, + rp._representative_rank, rp._representative_id,` : ''} ps.observer_node_ids, ps.observer_iatas, ps.rx_count, ps.tx_count FROM recent_packets rp LEFT JOIN packet_stats ps ON ps.packet_hash = rp.packet_hash - ORDER BY rp.time DESC - LIMIT $1`, - [limit, ...scope.params], + ${overlapParam ? `WHERE rp.packet_hash = ANY(${overlapParam}::text[]) + OR rp.packet_hash IN (SELECT packet_hash FROM recent_packets ORDER BY time DESC, packet_hash LIMIT $1)` : ''} + ORDER BY rp.time DESC, rp.packet_hash + LIMIT $1${overlapParam ? ` + CARDINALITY(${overlapParam}::text[])` : ''}`, + [limit, ...scope.params, ...(overlapParam ? [overlapHashes] : [])], signal, ); return res.rows; @@ -890,12 +903,17 @@ export async function getRecentMessages( network?: string, observer?: string, signal?: AbortSignal, + readQuery: QueryFn = query, + overlapHashes: string[] = [], ) { const includeObservationIds = getRamIngestMode() === 'ram'; const scope = buildScopePlaceholders(2, network, observer); - const res = await query( + const overlapParam = includeObservationIds && overlapHashes.length > 0 ? `$${2 + scope.params.length}` : null; + const res = await readQuery( `WITH recent_msgs AS ( SELECT DISTINCT ON (p.packet_hash) + ${includeObservationIds ? 'CASE WHEN p.src_node_id IS NOT NULL THEN 1 ELSE 0 END AS _representative_rank,' : ''} + ${includeObservationIds ? 'p.observation_id::text AS _representative_id,' : ''} p.time, p.packet_hash, COALESCE(pa_rx.canonical_node_id, upper(btrim(p.rx_node_id))) AS rx_node_id, COALESCE(pa_src.canonical_node_id, upper(btrim(p.src_node_id))) AS src_node_id, p.topic, @@ -914,7 +932,7 @@ export async function getRecentMessages( ${buildPublicPacketPrivacyClause('p')} ORDER BY p.packet_hash, CASE WHEN p.src_node_id IS NOT NULL THEN 1 ELSE 0 END DESC, - p.time DESC + p.time DESC, p.observation_id DESC NULLS LAST ), msg_stats AS ( SELECT @@ -939,13 +957,17 @@ export async function getRecentMessages( m.time, m.packet_hash, m.rx_node_id, m.src_node_id, m.topic, m.iata, m.packet_type, m.hop_count, m.rssi, m.snr, m.payload, m.summary, m.advert_count, m.path_hashes, m.path_hash_size_bytes, - ${includeObservationIds ? 'ms._observation_ids, m.network AS _policy_network,' : ''} + ${includeObservationIds ? `ms._observation_ids, m.network AS _policy_network, + (SELECT generation::text FROM public_visibility_state WHERE singleton) AS _policy_generation, + m._representative_rank, m._representative_id,` : ''} ms.observer_node_ids, ms.observer_iatas, ms.rx_count, ms.tx_count FROM recent_msgs m LEFT JOIN msg_stats ms ON ms.packet_hash = m.packet_hash - ORDER BY m.time DESC - LIMIT $1`, - [limit, ...scope.params], + ${overlapParam ? `WHERE m.packet_hash = ANY(${overlapParam}::text[]) + OR m.packet_hash IN (SELECT packet_hash FROM recent_msgs ORDER BY time DESC, packet_hash LIMIT $1)` : ''} + ORDER BY m.time DESC, m.packet_hash + LIMIT $1${overlapParam ? ` + CARDINALITY(${overlapParam}::text[])` : ''}`, + [limit, ...scope.params, ...(overlapParam ? [overlapHashes] : [])], signal, ); return res.rows; diff --git a/backend/src/db/packetBatch.integration.test.ts b/backend/src/db/packetBatch.integration.test.ts index f85cda0..50fff54 100644 --- a/backend/src/db/packetBatch.integration.test.ts +++ b/backend/src/db/packetBatch.integration.test.ts @@ -360,6 +360,8 @@ test('F06 real PG snapshot and Redis expiry preserve the captured handoff', { assert.deepEqual(merged.map((row) => row.packet_hash), [input.packetHash]); const later = await readRecentWithPending(async () => (await pool.query(` SELECT packet_hash, time, network, ARRAY[observation_id::text] AS _observation_ids, + (SELECT generation::text FROM public_visibility_state WHERE singleton) AS _policy_generation, + 0 AS _representative_rank, 1::int AS rx_count, 0::int AS tx_count FROM packets`)).rows, { limit: 10, network: 'ukmesh', resolveAliases: async () => new Map() }); assert.deepEqual(new Set(later.map((row) => row.packet_hash)), new Set([input.packetHash, '61'.repeat(32)])); @@ -726,12 +728,21 @@ test('privacy rename commits and rematerializes a compressed chunk above the old test('packet batch atomically coalesces observer, sighting, and stats writes', { skip: databaseUrl ? false : 'TEST_INGEST_DATABASE_URL is not configured', }, async (t) => { + const admin = new Pool({ connectionString: databaseUrl, max: 1 }); + const name = `meshcore_coalesce_${process.pid}_${Date.now()}`; + await admin.query(`CREATE DATABASE "${name}"`); + const ownedUrl = new URL(databaseUrl!); + ownedUrl.pathname = `/${name}`; const pool = new Pool({ - connectionString: databaseUrl, + connectionString: ownedUrl.toString(), application_name: 'meshcore-ingest-integration-test', max: 2, }); - t.after(() => pool.end()); + t.after(async () => { + await pool.end(); + await admin.query(`DROP DATABASE IF EXISTS "${name}" WITH (FORCE)`); + await admin.end(); + }); const baseSql = fs.readFileSync(new URL('./schema/base.sql', import.meta.url), 'utf8'); await pool.query(baseSql); diff --git a/backend/src/db/packetBatch.ts b/backend/src/db/packetBatch.ts index fc2b635..93adbc9 100644 --- a/backend/src/db/packetBatch.ts +++ b/backend/src/db/packetBatch.ts @@ -395,17 +395,14 @@ export async function persistEntriesCore( }).join(',\n'); const candidateSource = idempotent ? 'novel_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 sighting identity 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 observation fields). + // Stable upstream IDs are arbitrated by the ledger, preserving distinct + // sightings even when all their observation fields match. The legacy + // natural-identity fallback applies only to entries without an upstream ID. const idempotencyCte = idempotent ? `deduped AS MATERIALIZED ( SELECT c.*, ROW_NUMBER() OVER ( - PARTITION BY c.time, c.packet_hash, c.rx_node_id, c.topic, c.network, + PARTITION BY c.observation_id_in, c.time, c.packet_hash, c.rx_node_id, c.topic, c.network, c.hop_count, c.route_type, c.path_hashes, c.path_hash_size_bytes, (c.payload::jsonb->>'direction') ORDER BY c.row_id @@ -416,7 +413,7 @@ export async function persistEntriesCore( SELECT d.* FROM deduped d WHERE d.identity_rank = 1 - AND NOT EXISTS ( + AND (d.observation_id_in IS NOT NULL OR NOT EXISTS ( SELECT 1 FROM packets existing WHERE existing.time = d.time @@ -429,7 +426,7 @@ export async function persistEntriesCore( AND existing.path_hashes IS NOT DISTINCT FROM d.path_hashes AND existing.path_hash_size_bytes IS NOT DISTINCT FROM d.path_hash_size_bytes AND (existing.payload->>'direction') IS NOT DISTINCT FROM (d.payload::jsonb->>'direction') - ) + )) ),` : ''; // The lock must be a separate command on the same transaction/client. In diff --git a/backend/src/db/recentRead.integration.test.ts b/backend/src/db/recentRead.integration.test.ts new file mode 100644 index 0000000..abb7aef --- /dev/null +++ b/backend/src/db/recentRead.integration.test.ts @@ -0,0 +1,133 @@ +import assert from 'node:assert/strict'; +import fs from 'node:fs'; +import test from 'node:test'; +import pg from 'pg'; +import { Redis } from 'ioredis'; +import { closeDb, getRecentMessages, getRecentPackets } from './index.js'; +import { runMigrations } from './migrations.js'; +import { configurePacketBatch, persistEntriesCore, type QueryFn } from './packetBatch.js'; +import { admitPacketToRam, configureRamStore, readPendingEntries } from '../ram/liveStore.js'; +import type { InsertPacketInput } from './packetInput.js'; +import { readRecentWithPending, mergePendingPacketsIntoRecentRows } from '../ram/liveRead.js'; +import { getRamStreamKey } from '../ram/protocol.js'; +import { processNextPrivacyRematerialization } from '../workers/privacy-rematerialization.js'; + +const url = process.env['TEST_INGEST_DATABASE_URL']; +const redisURL = process.env['TEST_REDIS_URL']; +const local = (value?: string) => value && ['localhost', '127.0.0.1'].includes(new URL(value).hostname); +const json = (value: unknown) => JSON.parse(JSON.stringify(value)); + +test('F01/F05/F12/F13 real SQL and RAM contracts agree across scoped paging, stable IDs and privacy generations', { + skip: local(url) && local(redisURL) ? false : 'loopback PostgreSQL and Redis fixtures required', +}, async (t) => { + const admin = new pg.Pool({ connectionString: url, max: 1 }); + const name = `meshcore_read_${process.pid}_${Date.now()}`; + await admin.query(`CREATE DATABASE "${name}"`); + const ownedURL = new URL(url!); + ownedURL.pathname = `/${name}`; + const pool = new pg.Pool({ connectionString: ownedURL.toString(), max: 3 }); + const redis = new Redis(redisURL!, { lazyConnect: true, enableOfflineQueue: false }); + const oldMode = process.env['MQTT_INGEST_MODE']; + const oldKey = process.env['MQTT_RAM_STREAM_KEY']; + process.env['MQTT_INGEST_MODE'] = 'ram'; + process.env['MQTT_RAM_STREAM_KEY'] = `meshcore:read:${process.pid}:${Date.now()}`; + const key = getRamStreamKey(); + t.after(async () => { + configureRamStore(null); + configurePacketBatch(null); + const keys = await redis.keys(`${key}*`); + if (keys.length) await redis.del(...keys); + await redis.quit(); + await closeDb(); + await pool.end(); + await admin.query(`DROP DATABASE IF EXISTS "${name}" WITH (FORCE)`); + await admin.end(); + if (oldMode === undefined) delete process.env['MQTT_INGEST_MODE']; else process.env['MQTT_INGEST_MODE'] = oldMode; + if (oldKey === undefined) delete process.env['MQTT_RAM_STREAM_KEY']; else process.env['MQTT_RAM_STREAM_KEY'] = oldKey; + }); + await redis.connect(); + await pool.query(fs.readFileSync(new URL('./schema/base.sql', import.meta.url), 'utf8')); + await runMigrations(pool); + const query: QueryFn = (sql, params) => pool.query(sql, params); + const transaction = async (work: (query: QueryFn) => Promise) => { + const client = await pool.connect(); + try { + await client.query('BEGIN'); + const result = await work((sql, params) => client.query(sql, params)); + await client.query('COMMIT'); + return result; + } catch (error) { await client.query('ROLLBACK'); throw error; } + finally { client.release(); } + }; + configureRamStore(redis); + configurePacketBatch(query, transaction); + const admit = async (input: InsertPacketInput & { time: Date }) => { + t.mock.timers.enable({ apis: ['Date'], now: input.time.getTime() }); + try { + const accepted = await admitPacketToRam(input); + const stored = (await readPendingEntries(10)).find(({ entry }) => entry.observationId === accepted.observationId)!.entry; + return { input: stored.input, observationId: stored.observationId }; + } finally { t.mock.timers.reset(); } + }; + const alias = 'AA'.repeat(32), canonical = 'BB'.repeat(32), other = 'CD'.repeat(32); + await pool.query(`INSERT INTO node_identity_aliases + (source_node_id, canonical_node_id, confidence, reason) VALUES ($1, $2, 'high', 'fixture')`, [alias, canonical]); + while (await processNextPrivacyRematerialization(pool, { chunkPauseMs: 0 })) { /* finish fixture policy */ } + const aliases = async () => new Map([[alias, canonical]]); + const observed = Date.now(); + const base = { packetHash: '80'.repeat(32), rxNodeId: alias, topic: `meshcore/ZZZ/${alias}/packets`, + network: 'northeast', rawHex: '00', packetType: 5, pathHashes: [] as string[], + time: new Date(observed - 4_000), payload: { direction: 'rx', _summary: 'encrypted' } }; + const committed = await admit(base); + await persistEntriesCore([committed], query, true, transaction); + const older = { input: { ...committed.input, time: new Date(observed - 120_000) }, + observationId: '00000000-0000-4000-8000-000000000042' }; + const pageLeader = await admit({ ...base, packetHash: '81'.repeat(32), time: new Date(observed - 3_500) }); + await persistEntriesCore([older, pageLeader], query, true, transaction); + const pending = [ + await admit({ ...base, time: new Date(observed - 2_000), payload: { direction: 'tx', _summary: 'newer' } }), + await admit({ ...base, rxNodeId: other, srcNodeId: other, + time: new Date(observed - 3_000), payload: { appData: { text: 'decoded' }, direction: 'rx', _summary: 'decoded' } }), + await admit({ ...base, packetHash: '81'.repeat(32), time: new Date(observed - 3_500) }), + await admit({ ...base, packetHash: '82'.repeat(32), network: 'othernet' }), + await admit({ ...base, packetHash: '83'.repeat(32), time: new Date(observed - 301_000) }), + ]; + const merged = new Map(); + for (const scope of [undefined, 'ukmesh', 'northeast', 'teesside', 'othernet']) { + for (const observer of [undefined, alias, canonical, other]) { + for (const fields of ['full', 'slim'] as const) { + for (const limit of [1, 5]) { + const opts = { limit, network: scope, observer, fields, resolveAliases: aliases, nowMs: observed }; + merged.set(JSON.stringify(opts), json(await readRecentWithPending( + (hashes) => getRecentPackets(limit, scope, observer, fields, undefined, query, hashes), opts))); + } + } + } + } + const messages = json(await readRecentWithPending((hashes) => getRecentMessages(5, 'ukmesh', undefined, undefined, query, hashes), + { limit: 5, network: 'ukmesh', rowKind: 'messages', packetTypes: [5], resolveAliases: aliases, nowMs: observed })); + await persistEntriesCore(pending, query, true, transaction); + await redis.del(key); // compare against the complete SQL projection, without overlap + for (const [serialized, expected] of merged) { + const opts = JSON.parse(serialized); + const actual = await readRecentWithPending( + () => getRecentPackets(opts.limit, opts.network, opts.observer, opts.fields, undefined, query), + { ...opts, resolveAliases: aliases, nowMs: observed }); + assert.deepEqual(json(actual), expected, serialized); + } + assert.deepEqual(json(await readRecentWithPending(() => getRecentMessages(5, 'ukmesh', undefined, undefined, query), + { limit: 5, network: 'ukmesh', rowKind: 'messages', packetTypes: [5], resolveAliases: aliases, nowMs: observed })), messages); + const snapshot = await pool.connect(); + try { + await snapshot.query('BEGIN ISOLATION LEVEL REPEATABLE READ'); + const before = await getRecentPackets(5, 'ukmesh', undefined, 'full', undefined, + (sql, params) => snapshot.query(sql, params)); + assert.equal(before[0]?.rx_count, 3); + await pool.query('UPDATE public_visibility_state SET generation = generation + 1 WHERE singleton'); + await pool.query(`UPDATE packet_visibility_materialization_state SET visibility_generation = + (SELECT generation FROM public_visibility_state WHERE singleton) WHERE singleton`); + assert.deepEqual(await mergePendingPacketsIntoRecentRows(before, { limit: 5, network: 'ukmesh' }), [], + 'fresh authority rejects a genuine older SQL aggregate snapshot'); + await snapshot.query('COMMIT'); + } finally { await snapshot.query('ROLLBACK'); snapshot.release(); } +}); diff --git a/backend/src/index.ts b/backend/src/index.ts index 4f8c655..e0f24a0 100644 --- a/backend/src/index.ts +++ b/backend/src/index.ts @@ -19,6 +19,7 @@ import { stopMqttConnectionMonitor, } from './mqtt/connectionMonitor.js'; import { initWebSocketServer, closeWebSocketServer, broadcastPacket, broadcastNodeUpdate, broadcastNodeUpsert } from './ws/server.js'; +import { createPublicRamSource } from './ram/publicSource.js'; import apiRoutes, { startOwnerLastHopPrewarm, stopOwnerLastHopPrewarm } from './api/routes.js'; import { startRegisteredStatsWarmup } from './api/routes/stats.js'; import { initSpamMessageAnalyzer, stopSpamMessageAnalyzer } from './spam/analyzer.js'; @@ -316,6 +317,7 @@ async function main() { // API routes app.use('/api', apiRoutes); + app.use('/internal/ingest/public', createPublicRamSource()); // Health check app.get('/healthz', (_req, res) => res.json({ status: 'ok', ts: Date.now() })); diff --git a/backend/src/metrics.ts b/backend/src/metrics.ts index 105bafe..406a34e 100644 --- a/backend/src/metrics.ts +++ b/backend/src/metrics.ts @@ -30,6 +30,13 @@ export const websocketClients = new Gauge({ registers: [metricsRegistry], }); +export const publicLiveOutcomesTotal = new Counter({ + name: 'meshcore_public_live_outcomes_total', + help: 'Public live events authorized or suppressed at a bounded delivery batch.', + labelNames: ['phase', 'outcome'] as const, + registers: [metricsRegistry], +}); + export const statsRecomputeTotal = new Counter({ name: 'meshcore_stats_recompute_total', help: 'Cold statistics recomputations.', diff --git a/backend/src/mqtt/client.ts b/backend/src/mqtt/client.ts index 42881e4..bc47640 100644 --- a/backend/src/mqtt/client.ts +++ b/backend/src/mqtt/client.ts @@ -22,6 +22,7 @@ import { persistStatusTelemetrySample } from './statusTelemetry.js'; import { observationIdentityKey, packetAlreadyPersisted, type ObservationIdentity } from './observationIdentity.js'; import { assertSupportedIngestConnections } from './ingestConfig.js'; import { waitForShutdownStep } from './shutdownDeadline.js'; +import { createPublicBatch } from '../privacy/publicBatch.js'; import { boundedNetworkMetricLabel, mqttIngestAckDuration, @@ -30,6 +31,7 @@ import { mqttMessagesTotal, postPersistDroppedTotal, postPersistInFlightGauge, + publicLiveOutcomesTotal, } from '../metrics.js'; type PacketCallback = (packet: LivePacket, upstreamEnvelope?: Record) => void; @@ -50,6 +52,20 @@ let mqttRuntimeStatus: MqttRuntimeStatus = { state: 'starting', changedAt: new D const mqttClients: MqttClient[] = []; let mqttStopping = false; const MQTT_INGEST_MODE = getRamIngestMode(); +const publicPackets = createPublicBatch<{ packet: LivePacket; publish: () => void }>({ + maxItems: 10_000, maxBytes: 8 * 1_048_576, + loadPolicy: loadCurrentPublicPrivacyPolicy, + refused: (outcome, count) => publicLiveOutcomesTotal.inc({ phase: 'mqtt', outcome }, count), + deliver: (items, policy) => { + for (const { packet, publish } of items) { + const current = policy.classify({ network: packet.network ?? '', + rxNodeId: packet.rxNodeId ?? null, srcNodeId: packet.srcNodeId ?? null, + pathHashes: packet.path ?? null, pathHashSizeBytes: packet.pathHashSizeBytes ?? null }); + publicLiveOutcomesTotal.inc({ phase: 'mqtt', outcome: current.visibilityOk ? 'authorized' : 'private' }); + if (current.visibilityOk) { packet.isPrivate = false; packet.visibilityOk = true; publish(); } + } + }, +}); let mqttDurabilityFailure: string | undefined; function setMqttRuntimeStatus(state: MqttRuntimeStatus['state'], lastError?: string): void { @@ -454,6 +470,7 @@ export async function stopMqttClient(timeoutMs = 20_000): Promise { flushPendingNodeEmits(); } await waitForShutdownStep(closePacketBatch(), deadline, 'batch'); + await waitForShutdownStep(publicPackets.close(), deadline, 'public-live'); setMqttRuntimeStatus('disconnected'); } @@ -893,21 +910,8 @@ async function handleMessage( } }; if (MQTT_INGEST_MODE === 'ram') { - // Public emission is independent of broker acknowledgement. It uses a - // fresh authority snapshot; a PG outage suppresses the public event. - void loadCurrentPublicPrivacyPolicy().then((policy) => { - if (!policy) return; - const current = policy.classify({ - network, rxNodeId: observerKey, srcNodeId: srcNodeId ?? null, - pathHashes: path ?? null, pathHashSizeBytes: decodedPathHashSizeBytes ?? null, - }); - if (!current.visibilityOk) return; - livePacket.isPrivate = false; - livePacket.visibilityOk = true; - publishLive(); - }).catch((error: unknown) => { - console.error('[mqtt] public live authority unavailable:', error instanceof Error ? error.message : error); - }); + publicPackets.enqueue({ packet: livePacket, publish: publishLive }, + Buffer.byteLength(JSON.stringify([livePacket, json]))); } else publishLive(); invalidateResolveCache(finalHash); mqttIngestOutcomesTotal.inc({ diff --git a/backend/src/privacy/publicBatch.test.ts b/backend/src/privacy/publicBatch.test.ts new file mode 100644 index 0000000..0d8ba9b --- /dev/null +++ b/backend/src/privacy/publicBatch.test.ts @@ -0,0 +1,41 @@ +import assert from 'node:assert/strict'; +import test from 'node:test'; +import { createPublicBatch } from './publicBatch.js'; +import { PrivatePrefixCache } from './privatePrefixCache.js'; + +test('F01 public batches query authority once, preserve FIFO and reserve active payload bytes', async () => { + const policy = new PrivatePrefixCache(); policy.replace(2, []); + let release!: (value: PrivatePrefixCache) => void; + let queries = 0; + const delivered: number[] = [], refused: string[] = []; + const batch = createPublicBatch({ maxItems: 3, maxBytes: 30, delayMs: 60_000, + loadPolicy: () => { queries++; return new Promise((resolve) => { release = resolve; }); }, + deliver: (items) => { delivered.push(...items); }, refused: (reason) => { refused.push(reason); } }); + batch.enqueue(1, 10); batch.enqueue(2, 10); + const active = batch.flush(); + batch.enqueue(3, 10); + assert.equal(batch.enqueue(4, 10), false); + assert.deepEqual(batch.size(), { count: 3, bytes: 30 }); + assert.equal(queries, 1); + release(policy); await active; + assert.deepEqual(delivered, [1, 2]); + const next = batch.flush(); release(policy); await next; + assert.deepEqual(delivered, [1, 2, 3]); + assert.deepEqual(refused, ['overflow']); + assert.deepEqual(batch.size(), { count: 0, bytes: 0 }); + await batch.close(); +}); + +test('F01 unavailable authority and shutdown suppress captured public batches', async () => { + let resolve!: (value: null) => void; + let deliveries = 0, refusals = 0; + const batch = createPublicBatch({ maxItems: 2, maxBytes: 20, delayMs: 60_000, + loadPolicy: () => new Promise((r) => { resolve = r; }), + deliver: () => { deliveries++; }, refused: () => { refusals++; } }); + batch.enqueue(1, 10); const active = batch.flush(); resolve(null); await active; + assert.equal(refusals, 1); assert.equal(deliveries, 0); + batch.enqueue(2, 10); const held = batch.flush(); const closed = batch.close(); + resolve(null); await held; await closed; + assert.equal(deliveries, 0); assert.equal(refusals, 1); + assert.deepEqual(batch.size(), { count: 0, bytes: 0 }); +}); diff --git a/backend/src/privacy/publicBatch.ts b/backend/src/privacy/publicBatch.ts new file mode 100644 index 0000000..0a42d09 --- /dev/null +++ b/backend/src/privacy/publicBatch.ts @@ -0,0 +1,63 @@ +import type { PrivatePrefixCache } from './privatePrefixCache.js'; + +/** A bounded FIFO with one current-authority query per delivery batch. */ +export function createPublicBatch(options: { + loadPolicy: () => Promise; + deliver: (items: T[], policy: PrivatePrefixCache) => void; + refused: (reason: 'overflow' | 'unavailable', count: number) => void; + maxItems: number; + maxBytes: number; + delayMs?: number; +}) { + let pending: Array<{ item: T; bytes: number }> = []; + let bytes = 0, count = 0; + let timer: ReturnType | null = null; + let active: Promise | null = null; + let closed = false; + const schedule = () => { + if (!closed && !timer && !active && pending.length) { + timer = setTimeout(() => { timer = null; void flush(); }, options.delayMs ?? 16); + } + }; + const flush = (): Promise => { + if (active) return active; + if (!pending.length || closed) return Promise.resolve(); + if (timer) { clearTimeout(timer); timer = null; } + const batch = pending; pending = []; + const batchBytes = batch.reduce((sum, value) => sum + value.bytes, 0); + const work = (async () => { + try { + const policy = await options.loadPolicy(); + if (!closed) { + if (policy) options.deliver(batch.map(({ item }) => item), policy); + else options.refused('unavailable', batch.length); + } + } catch { + if (!closed) options.refused('unavailable', batch.length); + } finally { + bytes -= batchBytes; count -= batch.length; + } + })().finally(() => { active = null; schedule(); }); + active = work; + return work; + }; + return { + enqueue(item: T, itemBytes: number): boolean { + if (closed || !Number.isSafeInteger(itemBytes) || itemBytes < 0 + || count >= options.maxItems || bytes + itemBytes > options.maxBytes) { + options.refused('overflow', 1); return false; + } + pending.push({ item, bytes: itemBytes }); bytes += itemBytes; count++; + schedule(); return true; + }, + flush, + close(): Promise { + closed = true; + if (timer) { clearTimeout(timer); timer = null; } + bytes -= pending.reduce((sum, value) => sum + value.bytes, 0); + count -= pending.length; pending = []; + return active ?? Promise.resolve(); + }, + size: () => ({ count, bytes }), + }; +} diff --git a/backend/src/ram/liveRead.ts b/backend/src/ram/liveRead.ts index 3ce88cb..e95b437 100644 --- a/backend/src/ram/liveRead.ts +++ b/backend/src/ram/liveRead.ts @@ -40,6 +40,7 @@ export type MergeOptions = { resolveAliases?: (nodeIds: string[]) => Promise>; /** Restrict pending contributions to these packet types (e.g. [5] for message history). */ packetTypes?: number[]; + rowKind?: 'packets' | 'messages'; /** Capture before starting the SQL read to survive overlap expiry. */ pendingSnapshot?: PendingStreamEntry[]; }; @@ -50,6 +51,7 @@ function upper(value: unknown): string | null { } function timeMs(value: unknown): number { + if (value instanceof Date) return value.getTime(); const parsed = Date.parse(String(value ?? '')); return Number.isFinite(parsed) ? parsed : 0; } @@ -97,19 +99,21 @@ function canonical(value: string | null, aliases: Map): string | return value ? aliases.get(value) ?? value : null; } -function representativeRank(item: EntryWithTx): number { +function representativeRank(item: EntryWithTx, rowKind: 'packets' | 'messages'): number { const { entry } = item; const { payload } = entryPayload(entry); + if (rowKind === 'messages') return entry.input.srcNodeId !== null ? 1 : 0; return (payload && Object.hasOwn(payload, 'appData') ? 16 : 0) - + (upper(entry.input.srcNodeId) ? 8 : 0) + + (entry.input.srcNodeId !== null ? 8 : 0) + (entry.input.advertCount !== null && entry.input.advertCount !== undefined ? 4 : 0) + (entry.input.packetType === 4 ? 2 : 0); } /** Aggregate pending observations into a single row (mirrors the SQL observer aggregation). */ -function buildMergedRow(group: EntryWithTx[], includePayload: boolean, fields: 'full' | 'slim', aliases: Map): RecentRow { - const newestFirst = [...group].sort((a, b) => representativeRank(b) - representativeRank(a) - || timeMs(b.entry.input.time) - timeMs(a.entry.input.time)); +function buildMergedRow(group: EntryWithTx[], includePayload: boolean, fields: 'full' | 'slim', aliases: Map, rowKind: 'packets' | 'messages'): RecentRow { + const newestFirst = [...group].sort((a, b) => representativeRank(b, rowKind) - representativeRank(a, rowKind) + || timeMs(b.entry.input.time) - timeMs(a.entry.input.time) + || b.entry.observationId.localeCompare(a.entry.observationId)); const representative = newestFirst[0]!; const observers = new Set(); const iatas = new Set(); @@ -118,7 +122,7 @@ function buildMergedRow(group: EntryWithTx[], includePayload: boolean, fields: ' for (const { entry, isTx } of group) { const rx = canonical(upper(entry.input.rxNodeId), aliases); if (rx) observers.add(rx); - if (entry.input.iata) iatas.add(entry.input.iata); + if (entry.input.iata?.trim()) iatas.add(entry.input.iata); if (isTx) txCount += 1; else rxCount += 1; } @@ -137,21 +141,23 @@ function buildMergedRow(group: EntryWithTx[], includePayload: boolean, fields: ' advert_count: input.advertCount, path_hashes: input.pathHashes, path_hash_size_bytes: input.pathHashSizeBytes, - ...(fields === 'full' ? { + ...(rowKind === 'messages' ? { topic: input.topic, iata: input.iata } : fields === 'full' ? { topic: input.topic, topic_prefix: input.topicPrefix, iata: input.iata, route_type: input.routeType, network: input.network, transport_codes: input.transportCodes, region_scope: input.regionScope, } : {}), - observer_node_ids: [...observers], - observer_iatas: [...iatas], + observer_node_ids: observers.size ? [...observers].sort() : null, + observer_iatas: iatas.size ? [...iatas].sort() : null, rx_count: rxCount, tx_count: txCount, - ...(includePayload ? { payload } : {}), + ...(includePayload || rowKind === 'messages' ? { payload } : {}), + _representative_rank: representativeRank(representative, rowKind), + _representative_id: representative.entry.observationId, }; } /** Add genuinely new pending observations to an existing database row. */ -function augmentRow(row: RecentRow, group: EntryWithTx[], aliases: Map): void { +function augmentRow(row: RecentRow, group: EntryWithTx[], aliases: Map, opts: MergeOptions): void { const committedIds = new Set( Array.isArray(row['_observation_ids']) ? (row['_observation_ids'] as unknown[]).map(String) @@ -171,22 +177,36 @@ function augmentRow(row: RecentRow, group: EntryWithTx[], aliases: Map 0) { - row['observer_node_ids'] = [...existing]; - row['observer_iatas'] = [...iatas]; + row['observer_node_ids'] = existing.size ? [...existing].sort() : null; + row['observer_iatas'] = iatas.size ? [...iatas].sort() : null; row['rx_count'] = Number(row['rx_count'] ?? 0) + rxAdd; row['tx_count'] = Number(row['tx_count'] ?? 0) + txAdd; } + const candidate = buildMergedRow(group, opts.includePayload === true, opts.fields ?? 'full', aliases, opts.rowKind ?? 'packets'); + const rank = Number(candidate['_representative_rank']); + const oldRank = Number(row['_representative_rank']); + const candidateTime = timeMs(candidate['time']), rowTime = timeMs(row['time']); + if (rank > oldRank || (rank === oldRank && (candidateTime > rowTime + || (candidateTime === rowTime && String(candidate['_representative_id']) > String(row['_representative_id'] ?? ''))))) { + // Counts describe all observations; representative fields follow the same + // SQL preference across both sources, including retained overlap. + const { observer_node_ids: _nodes, observer_iatas: _iatas, rx_count: _rx, tx_count: _tx, ...fields } = candidate; + Object.assign(row, fields); + } } function stripInternalIds(rows: T[]): T[] { for (const row of rows) { if (row && typeof row === 'object') delete (row as RecentRow)['_observation_ids']; if (row && typeof row === 'object') delete (row as RecentRow)['_policy_network']; + if (row && typeof row === 'object') delete (row as RecentRow)['_policy_generation']; + if (row && typeof row === 'object') delete (row as RecentRow)['_representative_rank']; + if (row && typeof row === 'object') delete (row as RecentRow)['_representative_id']; } return rows; } @@ -199,12 +219,13 @@ function stripInternalIds(rows: T[]): T[] { * Permission is checked after SQL, never frozen with the captured payload. */ export async function readRecentWithPending( - loadRows: () => Promise, + loadRows: (overlapHashes: string[]) => Promise, opts: MergeOptions, ): Promise { const pendingSnapshot = getRamIngestMode() === 'ram' ? await readPendingEntries(opts.limit) : []; - return mergePendingPacketsIntoRecentRows(await loadRows(), { ...opts, pendingSnapshot }); + const overlapHashes = [...new Set(pendingSnapshot.map(({ entry }) => entry.input.packetHash))]; + return mergePendingPacketsIntoRecentRows(await loadRows(overlapHashes), { ...opts, pendingSnapshot }); } export async function mergePendingPacketsIntoRecentRows( @@ -215,7 +236,8 @@ export async function mergePendingPacketsIntoRecentRows( try { const policy = await loadCurrentPublicPrivacyPolicy(); if (!policy) return []; - const base = (rows as unknown as RecentRow[]).filter((row) => policy.classify({ + const base = (rows as unknown as RecentRow[]).filter((row) => + Number(row['_policy_generation']) === policy.currentGeneration && policy.classify({ network: String(row['network'] ?? row['_policy_network'] ?? opts.network ?? ''), rxNodeId: typeof row['rx_node_id'] === 'string' ? row['rx_node_id'] : null, srcNodeId: typeof row['src_node_id'] === 'string' ? row['src_node_id'] : null, @@ -226,7 +248,7 @@ export async function mergePendingPacketsIntoRecentRows( // distinct hashes. Sampling only the newest N observations would hide a // matching hash behind a large observer fanout or other packet types. const pending = opts.pendingSnapshot ?? await readPendingEntries(opts.limit); - if (pending.length === 0) return stripInternalIds(base as unknown as T[]); + if (pending.length === 0) return stripInternalIds(base.slice(0, opts.limit) as unknown as T[]); const cutoff = (opts.nowMs ?? Date.now()) - (opts.windowMs ?? (opts.packetTypes ? 86_400_000 : 300_000)); const eligible = pending.filter(({ entry }) => { const input = entry.input; @@ -243,7 +265,7 @@ export async function mergePendingPacketsIntoRecentRows( pathHashSizeBytes: input.pathHashSizeBytes, }).visibilityOk; }); - if (eligible.length === 0) return stripInternalIds(base as unknown as T[]); + if (eligible.length === 0) return stripInternalIds(base.slice(0, opts.limit) as unknown as T[]); const nodeIds = new Set(); if (opts.observer) nodeIds.add(opts.observer.toUpperCase()); for (const { entry } of eligible) { @@ -280,15 +302,14 @@ export async function mergePendingPacketsIntoRecentRows( if (dbRow) { // The database already carries a row (possibly with its own observer // set): augment it with genuinely new pending observations only. - augmentRow(dbRow, group, aliases); + augmentRow(dbRow, group, aliases, opts); continue; } - merged.push(buildMergedRow(group, opts.includePayload === true, opts.fields ?? 'full', aliases)); + merged.push(buildMergedRow(group, opts.includePayload === true, opts.fields ?? 'full', aliases, opts.rowKind ?? 'packets')); } - if (merged.length === 0) return stripInternalIds(base as unknown as T[]); - const combined = [...base, ...merged]; - combined.sort((left, right) => timeMs(right['time']) - timeMs(left['time'])); + combined.sort((left, right) => timeMs(right['time']) - timeMs(left['time']) + || String(left['packet_hash']).localeCompare(String(right['packet_hash']))); return stripInternalIds(combined.slice(0, opts.limit) as unknown as T[]); } catch (error) { console.error('[ram] live merge failed (returning database rows only)', (error as Error).message); diff --git a/backend/src/ram/liveStore.test.ts b/backend/src/ram/liveStore.test.ts index 38cfc57..cd71d05 100644 --- a/backend/src/ram/liveStore.test.ts +++ b/backend/src/ram/liveStore.test.ts @@ -333,7 +333,8 @@ test('merged reads append only public, scoped, non-duplicate pending rows', asyn ['100-0', ['e', encodePendingEntry(entryFor('34'.repeat(32), { time: '2026-10-11T12:10:00.000Z' }))]], // public, new (newer than the DB row) ]; configureRamStore(fake); - const rows = [{ packet_hash: 'ab'.repeat(32), time: new Date('2026-10-11T12:04:00.000Z') }]; + const rows = [{ packet_hash: 'ab'.repeat(32), time: new Date('2026-10-11T12:04:00.000Z'), + _policy_generation: '7', _representative_rank: 2 }]; const merged = await mergePendingPacketsIntoRecentRows(rows, { limit: 10, network: 'ukmesh' }); assert.equal(merged.length, 3, 'formerly private admission follows current public policy'); assert.equal(merged[0]?.['packet_hash'], '34'.repeat(32)); // newer pending first @@ -395,6 +396,7 @@ test('F12 merge deduplicates committed overlap by ID and counts a new same-obser rx_count: 1, tx_count: 0, _observation_ids: [entryFor(h).observationId], + _policy_generation: '7', _representative_rank: 2, }, ]; const merged = await mergePendingPacketsIntoRecentRows(rows, { limit: 10, network: 'ukmesh' }); @@ -478,6 +480,52 @@ test('F01 a privacy flip overrides the loaded public cache for RAM and PG rows', assert.deepEqual(await mergePendingPacketsIntoRecentRows(oldPgRow, { limit: 10, network: 'ukmesh' }), []); }); +test('F01 old SQL aggregate contributions fail closed across a policy generation change', async () => { + const privateObserver = 'aa'.repeat(32).toUpperCase(); + configurePacketBatch(async () => ({ rows: [{ + generation: '10', materialized_generation: '10', remat_active: false, + prefixes: [{ node_id: privateObserver, network: 'teesside', prefix_size_bytes: 1, prefix: 'AA' }], + }] })); + configureRamStore(new FakeStore()); + const snapshot = [{ packet_hash: 'fe'.repeat(32), time: '2026-10-11T12:00:00Z', + rx_node_id: 'cd'.repeat(32), network: 'northeast', + observer_node_ids: ['CD'.repeat(32), privateObserver], rx_count: 2, + _policy_generation: '9' }]; + const merged = await mergeRecentRows(snapshot, { limit: 10, network: 'ukmesh' }); + assert.deepEqual(merged, [], 'a public representative cannot authorize stale aggregate participants'); + assert.deepEqual(await mergeRecentRows([{ ...snapshot[0], _policy_generation: undefined }], + { limit: 10 }), [], 'missing generation cannot authorize a SQL aggregate'); +}); + +test('F13 a better pending representative replaces SQL fields and reorders the whole page', async () => { + await loadPublicPrivacyCache(); + const fake = new FakeStore(); + const hash = '83'.repeat(32); + const entry = entryFor(hash, { rx: 'aa'.repeat(32), time: '2026-10-11T13:01:00.000Z' }); + entry.input.payloadJson = '{"appData":{},"_summary":"decoded","direction":"tx"}'; + entry.input.iata = 'AAA'; + fake.rev = [['400-0', ['e', encodePendingEntry(entry)]]]; + configureRamStore(fake); + const rows = [ + { packet_hash: '84'.repeat(32), time: '2026-10-11T13:00:30.000Z', _policy_generation: '7' }, + { packet_hash: hash, time: '2026-10-11T13:00:00.000Z', summary: 'encrypted', + rx_node_id: 'cd'.repeat(32), observer_node_ids: ['CD'.repeat(32)], observer_iatas: ['ZZZ'], + rx_count: 1, tx_count: 0, _observation_ids: ['11111111-1111-1111-1111-111111111111'], + _policy_generation: '7', _representative_rank: 2 }, + ]; + const merged = await mergeRecentRows(rows, { limit: 2, network: 'ukmesh', + nowMs: Date.parse('2026-10-11T13:01:01Z'), resolveAliases: async () => new Map() }); + assert.equal(merged[0]?.packet_hash, hash, 'augmentation alone must also sort/page'); + assert.equal(merged[0]?.summary, 'decoded'); + assert.equal(merged[0]?.rx_node_id, 'AA'.repeat(32)); + assert.deepEqual(merged[0]?.observer_node_ids, ['AA'.repeat(32), 'CD'.repeat(32)]); + assert.deepEqual(merged[0]?.observer_iatas, ['AAA', 'ZZZ']); + assert.equal(merged[0]?.rx_count, 1); + assert.equal(merged[0]?.tx_count, 1); + assert.equal('_representative_rank' in merged[0]!, false); + assert.equal('_policy_generation' in merged[0]!, false); +}); + test('F13 RAM recent rows honor family scope, aliases, window, and slim DTO', async () => { await loadPublicPrivacyCache(); const fake = new FakeStore(); @@ -506,3 +554,21 @@ test('F13 RAM recent rows honor family scope, aliases, window, and slim DTO', as assert.equal('topic' in rows[0]!, false); assert.equal('network' in rows[0]!, false); }); + +test('F13 representative ties use stable identity and paging preserves milliseconds', async () => { + await loadPublicPrivacyCache(); + configureRamStore(new FakeStore()); + const hash = '85'.repeat(32); + const olderID = '11111111-1111-4111-8111-111111111111'; + const entry = entryFor(hash, { time: '2026-10-11T13:00:00.100Z' }); + entry.observationId = 'ffffffff-ffff-4fff-8fff-ffffffffffff'; + entry.input.rssi = -25; + const row = { packet_hash:hash, time:new Date('2026-10-11T13:00:00.100Z'), rssi:-90, + _policy_generation:'7', _representative_rank:2, _representative_id:olderID, + _observation_ids:[olderID], network:'ukmesh' }; + const merged = await mergeRecentRows([row], {limit:1, network:'ukmesh', + pendingSnapshot:[{id:'400-0',entry}], nowMs:Date.parse('2026-10-11T13:00:01Z'), + resolveAliases:async()=>new Map()}); + assert.equal(merged[0]?.rssi,-25,'equal-rank, equal-time representative chooses greatest stable ID'); + assert.equal('_representative_id' in merged[0]!,false); +}); diff --git a/backend/src/ram/liveStore.ts b/backend/src/ram/liveStore.ts index 8d54927..42cc764 100644 --- a/backend/src/ram/liveStore.ts +++ b/backend/src/ram/liveStore.ts @@ -342,7 +342,7 @@ export type PendingStreamEntry = { id: string; entry: RamPendingEntry }; * The caller deduplicates by observation ID against the same PG read snapshot. * A cumulative marker cannot classify entries admitted during a commit. */ -export async function readPendingEntries(limit: number): Promise { +export async function readPendingEntries(limit: number, strict = false): Promise { if (getRamIngestMode() !== 'ram' || limit <= 0) return []; try { const client = getClient(); @@ -362,6 +362,7 @@ export async function readPendingEntries(limit: number): Promise { + process.env['MQTT_INGEST_MODE'] = 'ram'; + const policy = new PrivatePrefixCache(); + const privateId = 'AA'.repeat(32); + policy.replace(2, [{ node_id: privateId, network: 'teesside', prefix_size_bytes: 1, prefix: 'AA' }]); + let unavailable = false; + const input = buildPacketBatchInput({ topic: 'meshcore/ZZZ/observer/packets', packetHash: 'allowed', rawHex: '00', network: 'northeast' }); + const entries = [{ id: '10-0', entry: { observationId: '11111111-1111-4111-8111-111111111111', admittedAtMs: 10, + isPrivate: false, visibilityOk: true, input: { ...input, packetHash: 'private', rxNodeId: privateId } } }, + { id: '11-0', entry: { observationId: '22222222-2222-4222-8222-222222222222', admittedAtMs: 11, + isPrivate: true, visibilityOk: false, input } }]; + const app = express(); + const token = 'fixture'.repeat(8); + app.use('/public', createPublicRamSource({ token: () => token, read: async () => entries, + loadPolicy: async () => unavailable ? null : policy })); + const server = http.createServer(app); + await new Promise((resolve) => server.listen(0, '127.0.0.1', resolve)); + const address = server.address(); assert.ok(address && typeof address !== 'string'); + const root = `http://127.0.0.1:${address.port}/public`; + const get = (path: string, authorized = true) => fetch(root + path, { headers: authorized ? { authorization: `Bearer ${token}` } : {} }); + try { + assert.equal((await get('/entries', false)).status, 401); + assert.deepEqual(await (await get('/entries?limit=1')).json(), { generation: 2, entries: [], nextCursor: '10-0' }); + const page = await (await get('/entries?after=10-0')).json(); + assert.equal(page.entries.length, 1); assert.equal(page.nextCursor, '11-0'); + assert.equal(JSON.parse(page.entries[0].value).vo, 1); + assert.equal(JSON.parse(page.entries[0].value).d.packetHash, 'allowed'); + assert.equal((await get('/entries?limit=501')).status, 400); + unavailable = true; + assert.equal((await get('/policy')).status, 503); + assert.equal((await get('/entries')).status, 503); + } finally { await new Promise((resolve) => server.close(() => resolve())); } +}); diff --git a/backend/src/ram/publicSource.ts b/backend/src/ram/publicSource.ts new file mode 100644 index 0000000..55b4ae2 --- /dev/null +++ b/backend/src/ram/publicSource.ts @@ -0,0 +1,59 @@ +import { timingSafeEqual } from 'node:crypto'; +import { Router } from 'express'; +import { loadCurrentPublicPrivacyPolicy } from '../db/packetBatch.js'; +import { readPendingEntries, type PendingStreamEntry } from './liveStore.js'; +import { encodePendingEntry, getRamIngestMode } from './protocol.js'; +import type { PrivatePrefixCache } from '../privacy/privatePrefixCache.js'; + +function compareIds(left: string, right: string): number { + const [lm, ls] = left.split('-').map(BigInt), [rm, rs] = right.split('-').map(BigInt); + return lm! < rm! ? -1 : lm! > rm! ? 1 : ls! < rs! ? -1 : ls! > rs! ? 1 : 0; +} + +/** Credentialed public source: consumers never receive a raw Redis principal. */ +export function createPublicRamSource(deps: { + token?: () => string | undefined; + read?: () => Promise; + loadPolicy?: () => Promise; +} = {}): Router { + const router = Router(); + const token = deps.token ?? (() => process.env['MQTT_RAM_PUBLIC_READ_TOKEN']); + const read = deps.read ?? (() => readPendingEntries(10_000, true)); + const loadPolicy = deps.loadPolicy ?? loadCurrentPublicPrivacyPolicy; + router.use((req, res, next) => { + const expected = token(); + if (!expected || expected.length < 32) { res.sendStatus(404); return; } + const actual = req.header('authorization') ?? ''; + const required = `Bearer ${expected}`; + if (Buffer.byteLength(actual) !== Buffer.byteLength(required) + || !timingSafeEqual(Buffer.from(actual), Buffer.from(required))) { res.sendStatus(401); return; } + res.set('Cache-Control', 'no-store'); next(); + }); + router.get('/policy', async (_req, res) => { + try { + const policy = await loadPolicy(); + if (!policy) { res.sendStatus(503); return; } + res.json({ generation: policy.currentGeneration }); + } catch { res.sendStatus(503); } + }); + router.get('/entries', async (req, res) => { + const after = String(req.query['after'] ?? '0-0'); + const limit = Number(req.query['limit'] ?? 500); + if (!/^\d{1,20}-\d{1,20}$/.test(after) || !Number.isSafeInteger(limit) || limit < 1 || limit > 500) { + res.sendStatus(400); return; + } + if (getRamIngestMode() !== 'ram') { res.sendStatus(503); return; } + try { + const batch = (await read()).filter(({ id }) => compareIds(id, after) > 0).slice(0, limit); + const policy = await loadPolicy(); + if (!policy) { res.sendStatus(503); return; } + const entries = batch.filter(({ entry }) => policy.classify({ + network: entry.input.network, rxNodeId: entry.input.rxNodeId, srcNodeId: entry.input.srcNodeId, + pathHashes: entry.input.pathHashes, pathHashSizeBytes: entry.input.pathHashSizeBytes, + }).visibilityOk).map(({ id, entry }) => ({ id, + value: encodePendingEntry({ ...entry, visibilityOk: true, isPrivate: false }) })); + res.json({ generation: policy.currentGeneration, entries, nextCursor: batch.at(-1)?.id ?? after }); + } catch { res.sendStatus(503); } + }); + return router; +} diff --git a/backend/src/ws/privacy.test.ts b/backend/src/ws/privacy.test.ts index fe51fae..b0bd8a0 100644 --- a/backend/src/ws/privacy.test.ts +++ b/backend/src/ws/privacy.test.ts @@ -1,10 +1,27 @@ import assert from 'node:assert/strict'; import test from 'node:test'; import type { WSMessage } from '../types/index.js'; -import { PublicWsPrivacyIndex } from './privacy.js'; +import { authorizePublicWsMessage, PublicWsPrivacyIndex } from './privacy.js'; +import { PrivatePrefixCache } from '../privacy/privatePrefixCache.js'; const privateId = 'ab'.repeat(32); +test('F01 broadcast authority checks observers, sources, all path widths and network families', () => { + const policy = new PrivatePrefixCache(); + policy.replace(7, [1, 2, 3].map((size) => ({ node_id: privateId, network: 'teesside', + prefix_size_bytes: size, prefix: privateId.slice(0, size * 2) }))); + const message = (data: unknown) => ({ type: 'packet', data, ts: 1 } as WSMessage); + for (const network of ['ukmesh', 'northeast', 'teesside']) { + assert.equal(authorizePublicWsMessage(message({ network, rxNodeId: privateId, visibilityOk: true }), policy), null); + assert.equal(authorizePublicWsMessage(message({ network, srcNodeId: privateId, visibilityOk: true }), policy), null); + for (const size of [1, 2, 3]) assert.equal(authorizePublicWsMessage(message({ network, + path: [privateId.slice(0, size * 2)], pathHashSizeBytes: size, visibilityOk: true }), policy), null); + } + assert.ok(authorizePublicWsMessage(message({ network: 'othernet', rxNodeId: privateId, visibilityOk: false }), policy)); + assert.equal(authorizePublicWsMessage(message({ rxNodeId: privateId, visibilityOk: true }), policy), null); + assert.equal(authorizePublicWsMessage(message({ network: 'ukmesh', path: ['GG'], pathHashSizeBytes: 1 }), policy), null); +}); + test('live packets fail closed until the privacy index is ready', () => { const privacy = new PublicWsPrivacyIndex(); const message = { type: 'packet', data: { packetHash: 'hash' }, ts: 1 } as WSMessage; diff --git a/backend/src/ws/privacy.ts b/backend/src/ws/privacy.ts index 2074ef9..0c91fcb 100644 --- a/backend/src/ws/privacy.ts +++ b/backend/src/ws/privacy.ts @@ -1,5 +1,33 @@ import type { LivePacket, WSMessage } from '../types/index.js'; import { isPrivateNode } from '../api/utils/privateNode.js'; +import type { PrivatePrefixCache } from '../privacy/privatePrefixCache.js'; + +/** Current authority for all public message types; frozen flags are metadata. */ +export function authorizePublicWsMessage(message: WSMessage, policy: PrivatePrefixCache): WSMessage | null { + if (!message.data || typeof message.data !== 'object') return null; + const data = message.data as Record; + const network = String(data['network'] ?? ''); + if (!network.trim()) return null; + const classify = (rx: unknown, src: unknown, path: unknown = null, size: unknown = null) => policy.classify({ + network, rxNodeId: typeof rx === 'string' ? rx : null, srcNodeId: typeof src === 'string' ? src : null, + pathHashes: Array.isArray(path) ? path as string[] : null, pathHashSizeBytes: typeof size === 'number' ? size : null, + }).visibilityOk; + if (message.type === 'packet') { + if (!classify(data['rxNodeId'], data['srcNodeId'], data['path'], data['pathHashSizeBytes'])) return null; + const packet = data as unknown as LivePacket; + const appData = packet.payload?.['appData']; + const name = appData && typeof appData === 'object' ? (appData as Record)['name'] : packet.payload?.['name']; + if (packet.packetType === 4 && isPrivateNode(typeof name === 'string' ? name : null)) return null; + return { ...message, data: { ...data, visibilityOk: true, isPrivate: false } }; + } + if (message.type === 'node_upsert') { + if (isPrivateNode(typeof data['name'] === 'string' ? data['name'] : null)) return null; + return classify(data['observer_id'], data['node_id']) ? message : null; + } + if (message.type === 'node_update') return classify(data['observerId'], data['nodeId']) ? message : null; + if (message.type === 'link_update') return classify(data['node_a_id'], data['node_b_id']) ? message : null; + return null; // Initial state has a separate generation-fenced query contract. +} export class PublicWsPrivacyIndex { private readonly nodeIds = new Set(); diff --git a/backend/src/ws/publicAuthorization.integration.test.ts b/backend/src/ws/publicAuthorization.integration.test.ts new file mode 100644 index 0000000..5881ea7 --- /dev/null +++ b/backend/src/ws/publicAuthorization.integration.test.ts @@ -0,0 +1,101 @@ +import assert from 'node:assert/strict'; +import fs from 'node:fs'; +import http from 'node:http'; +import { once } from 'node:events'; +import test from 'node:test'; +import pg from 'pg'; +import { Redis } from 'ioredis'; +import { WebSocket } from 'ws'; +import { runMigrations } from '../db/migrations.js'; + +const databaseURL = process.env['TEST_INGEST_DATABASE_URL']; +const redisURL = process.env['TEST_REDIS_URL']; +const local = (url?: string) => url && ['127.0.0.1', 'localhost'].includes(new URL(url).hostname); +const delay = (ms: number) => new Promise((resolve) => setTimeout(resolve, ms)); + +test('F01 ready WebSocket batches and cached initial state obey current authority', { + skip: local(databaseURL) && local(redisURL) ? false : 'loopback PG/Redis fixtures required', timeout: 40_000, +}, async () => { + const admin = new pg.Pool({ connectionString: databaseURL, max: 1 }); + const name = `meshcore_ws_${process.pid}_${Date.now()}`; + await admin.query(`CREATE DATABASE "${name}"`); + const url = new URL(databaseURL!); url.pathname = `/${name}`; + const pool = new pg.Pool({ connectionString: url.toString(), max: 2 }); + await pool.query(fs.readFileSync(new URL('../db/schema/base.sql', import.meta.url), 'utf8')); + await runMigrations(pool); + const observer = 'AA'.repeat(32), source = 'CC'.repeat(32); + await pool.query(`INSERT INTO nodes(node_id, public_key, network, name) VALUES + ($1, $1, 'teesside', 'Observer'), ($2, $2, 'teesside', 'Relay')`, [observer, 'BB'.repeat(32)]); + process.env['DATABASE_URL'] = url.toString(); + process.env['REDIS_URL'] = redisURL; + delete process.env['REDIS_PASSWORD']; + process.env['WS_INITIAL_STATE_ENABLED'] = '1'; + process.env['WARMUP_NETWORKS'] = ''; + process.env['MQTT_INGEST_MODE'] = 'ram'; + const { initWebSocketServer, closeWebSocketServer } = await import('./server.js'); + const { closeDb } = await import('../db/index.js'); + const { configureRamStore } = await import('../ram/liveStore.js'); + const pub = new Redis(redisURL!); + await pub.ping(); + configureRamStore(pub); + const server = http.createServer(); + const wss = initWebSocketServer(server); + const clients: WebSocket[] = []; + try { + server.listen(0, '127.0.0.1'); await once(server, 'listening'); + const address = server.address(); assert.ok(address && typeof address !== 'string'); + const socketURL = `ws://127.0.0.1:${address.port}/ws?network=ukmesh`; + const client = new WebSocket(socketURL); clients.push(client); + const seen: string[] = []; + client.on('message', (raw) => { + for (const line of raw.toString().split('\n')) { + const message = JSON.parse(line); + if (message.type === 'packet') seen.push(message.data.packetHash); + } + }); + const [initial] = await once(client, 'message'); + assert.equal(JSON.parse(initial.toString()).type, 'initial_state'); + for (let attempt = 0; attempt < 100; attempt++) { + if (Number((await pub.call('PUBSUB', 'NUMSUB', 'meshcore:live') as string[])[1]) === 1) break; + await delay(10); + } + const packet = (hash: string, extra = {}) => JSON.stringify({ type: 'packet', ts: Date.now(), + data: { id: hash, packetHash: hash, topic: `meshcore/ZZZ/${observer}/packets`, + packetType: 5, network: 'northeast', rxNodeId: observer, srcNodeId: source, + visibilityOk: true, isPrivate: false, ...extra } }); + for (let attempt = 0; attempt < 100 && !seen.includes('ready'); attempt++) { + await pub.publish('meshcore:live', packet('ready')); await delay(50); + } + assert.ok(seen.includes('ready'), 'prove the subscribed reader can receive before changing policy'); + // Frozen vo=1 and a stale WS private-name index cannot grant permission. + await pool.query(`INSERT INTO private_node_prefixes(node_id, network, prefix_size_bytes, prefix) + VALUES ($1, 'teesside', 1, 'AA'), ($2, 'teesside', 2, 'BBBB')`, [observer, 'BB'.repeat(32)]); + await pool.query('UPDATE public_visibility_state SET generation = generation + 1 WHERE singleton'); + await pool.query(`UPDATE packet_visibility_materialization_state SET visibility_generation = + (SELECT generation FROM public_visibility_state WHERE singleton) WHERE singleton`); + await pub.publish('meshcore:live', packet('revoked-observer')); + await pub.publish('meshcore:live', packet('revoked-path', { rxNodeId: source, path: ['BBBB'], pathHashSizeBytes: 2 })); + await delay(150); + assert.equal(seen.some((hash) => hash.startsWith('revoked-')), false, 'privacy changes must fence queued/live data'); + const burst = Array.from({ length: 800 }, (_, i) => `burst-${i}`); + await Promise.all(burst.map((hash) => pub.publish('meshcore:live', packet(hash, { rxNodeId: source })))); + for (let attempt = 0; attempt < 200 && seen.filter((h) => h.startsWith('burst-')).length < 800; attempt++) await delay(10); + assert.deepEqual(new Set(seen.filter((h) => h.startsWith('burst-'))), new Set(burst), 'a ready, bounded reader receives the full batch'); + // This DB is owned by the test. Missing authority suppresses broadcasts + // and cached initial data, even though the cached epoch was once public. + await pool.query('ALTER TABLE public_visibility_state RENAME TO public_visibility_unavailable'); + await pub.publish('meshcore:live', packet('unavailable', { rxNodeId: source })); + await delay(150); + assert.equal(seen.includes('unavailable'), false); + const cachedClient = new WebSocket(socketURL); clients.push(cachedClient); + const [code] = await once(cachedClient, 'close'); + assert.equal(code, 1013, 'unavailable authority cannot reuse cached initial state'); + } finally { + clients.forEach((client) => client.terminate()); + await closeWebSocketServer(wss); + configureRamStore(null); + await new Promise((resolve) => server.close(() => resolve())); + await pub.quit(); await closeDb(); await pool.end(); + await admin.query(`DROP DATABASE "${name}" WITH (FORCE)`); await admin.end(); + } +}); diff --git a/backend/src/ws/server.ts b/backend/src/ws/server.ts index 9a290b5..264f292 100644 --- a/backend/src/ws/server.ts +++ b/backend/src/ws/server.ts @@ -17,11 +17,14 @@ import { websocketAdmissionDecision, } from './limits.js'; import { isPrivateNode } from '../api/utils/privateNode.js'; -import { PublicWsPrivacyIndex } from './privacy.js'; +import { authorizePublicWsMessage, PublicWsPrivacyIndex } from './privacy.js'; +import { loadCurrentPublicPrivacyPolicy } from '../db/packetBatch.js'; +import { createPublicBatch } from '../privacy/publicBatch.js'; import { trustedClientIp } from '../http/trustedProxy.js'; import { websocketAdmissionsTotal, websocketClients, + publicLiveOutcomesTotal, } from '../metrics.js'; import { selectInitialViableLinks } from './initialStateLinks.js'; import { readRecentWithPending } from '../ram/liveRead.js'; @@ -179,6 +182,7 @@ const viableLinksCache = new BoundedTtlMap { } async function fetchInitialState(network: string | undefined, observer: string | undefined): Promise { + const policy = await loadCurrentPublicPrivacyPolicy(); + if (!policy) throw new Error('WS_PUBLIC_AUTHORITY_UNAVAILABLE'); const key = `${network ?? ''}:${observer ?? ''}`; const cached = initialStateCache.get(key); - if (cached && (Date.now() - cached.ts) < INITIAL_STATE_TTL_MS) return cached; + if (cached && cached.generation === policy.currentGeneration && (Date.now() - cached.ts) < INITIAL_STATE_TTL_MS) return cached; // If a fetch is already in flight for this key, share it — don't pile on the DB. const existing = initialStateInflight.get(key); @@ -234,16 +240,16 @@ async function fetchInitialState(network: string | undefined, observer: string | const [rawNodes, packets, messages, rawViableLinks] = await Promise.all([ getNodes(network, observer, 'slim', signal), readRecentWithPending( - () => getRecentPackets(7, network, observer, 'full', signal), + (hashes) => getRecentPackets(7, network, observer, 'full', signal, undefined, hashes), { limit: 7, network, observer }, ), readRecentWithPending( - () => getRecentMessages(200, network, observer, signal), + (hashes) => getRecentMessages(200, network, observer, signal, undefined, hashes), // Message history is group-text only: keep the pending merge in the // same contract as getRecentMessages (packet_type = 5). - { limit: 200, network, observer, includePayload: true, packetTypes: [5] }, + { limit: 200, network, observer, includePayload: true, packetTypes: [5], rowKind: 'messages' }, ), - getCachedViableLinks(network, observer, signal), + getCachedViableLinks(network, observer, signal, policy.currentGeneration), ]); const nodes = rawNodes .filter((node) => !isPrivateNode(node.name)) @@ -259,11 +265,14 @@ async function fetchInitialState(network: string | undefined, observer: string | ); const entry: InitialStateEntry = { ts: Date.now(), + generation: policy.currentGeneration, nodes, packets: slimPackets, messages: slimMessages, viableLinks, }; + const current = await loadCurrentPublicPrivacyPolicy(); + if (!current || current.currentGeneration !== entry.generation) throw new Error('WS_INITIAL_STATE_POLICY_CHANGED'); initialStateCache.set(key, entry); return entry; } finally { @@ -294,8 +303,9 @@ async function getCachedViableLinks( network?: string, observer?: string, signal?: AbortSignal, + generation = 0, ) { - const key = cacheKey(network, observer); + const key = `${cacheKey(network, observer)}:${generation}`; const cached = viableLinksCache.get(key); if (cached && (Date.now() - cached.ts) < VIABLE_LINK_CACHE_TTL_MS) return cached.data; const data = await getViableLinks(network, observer, signal); @@ -495,15 +505,7 @@ export function initWebSocketServer(httpServer: Server): WebSocketServer { }); const clientScopes = new Map(); - type QueuedClientMessages = { - messages: string[]; - byteLength: number; - }; - const messageQueue = new Map(); - let flushTimeout: ReturnType | null = null; - const disconnectSlowClient = (client: WebSocket, reason: string) => { - messageQueue.delete(client); console.warn(`[ws] disconnecting slow client: ${reason}`); client.terminate(); }; @@ -537,7 +539,6 @@ export function initWebSocketServer(httpServer: Server): WebSocketServer { // or outlive a client that disconnects during the database read. ws.on('close', () => { clientScopes.delete(ws); - messageQueue.delete(ws); websocketClients.set(wss.clients.size); console.log('[ws] client disconnected, total:', wss.clients.size); }); @@ -590,6 +591,8 @@ export function initWebSocketServer(httpServer: Server): WebSocketServer { ws.close(1013, 'initial state is temporarily unavailable'); return; } + const current = await loadCurrentPublicPrivacyPolicy(); + if (!current || current.currentGeneration !== entry.generation) throw new Error('WS_INITIAL_STATE_POLICY_CHANGED'); if (ws.readyState === WebSocket.OPEN) ws.send(serialized); } catch (err) { console.error('[ws] initial state error', (err as Error).message); @@ -629,101 +632,51 @@ export function initWebSocketServer(httpServer: Server): WebSocketServer { clearInterval(privacyRefreshTimer); }); - const flushMessageQueue = () => { - if (messageQueue.size === 0) { - flushTimeout = null; - return; - } - - for (const [client, queued] of messageQueue) { - if (client.readyState !== WebSocket.OPEN || queued.messages.length === 0) continue; - if (client.bufferedAmount > WS_MAX_BUFFERED_BYTES) { - disconnectSlowClient(client, `socket buffer exceeded ${WS_MAX_BUFFERED_BYTES} bytes`); - continue; + const publicMessages = createPublicBatch({ + maxItems: 10_000, maxBytes: WS_MAX_QUEUE_BYTES, + loadPolicy: loadCurrentPublicPrivacyPolicy, + refused: (outcome, count) => publicLiveOutcomesTotal.inc({ phase: 'ws', outcome }, count), + deliver: (items, policy) => { + // Query once for the captured broadcast batch, classify each event once, + // then fan out synchronously. No stale per-client queues survive a fence. + const authorized: Array<{ message: WSMessage; serialized: string; bytes: number }> = []; + for (const item of items) { + const message = authorizePublicWsMessage(item, policy); + publicLiveOutcomesTotal.inc({ phase: 'ws', outcome: message ? 'authorized' : 'private' }); + if (message) { + const serialized = JSON.stringify(message); + authorized.push({ message, serialized, bytes: Buffer.byteLength(serialized) }); + } } - - // Send all queued messages - join with newlines for efficiency - client.send(queued.messages.join('\n')); - } - - messageQueue.clear(); - flushTimeout = null; - }; - - const scheduleFlush = () => { - if (flushTimeout !== null) return; - flushTimeout = setTimeout(flushMessageQueue, 16); - }; - - // Fan-out Redis messages to all connected WS clients - now batched - sub.on('message', (_channel: string, messageStr: string) => { - if (wss.clients.size === 0) return; - let parsed: WSMessage | null = null; - try { - parsed = JSON.parse(messageStr) as WSMessage; - } catch { - return; - } - parsed = publicPrivacy.filterMessage(parsed); - if (!parsed) return; - messageStr = JSON.stringify(parsed); - - if (LOG_WS_PACKETS && parsed?.type === 'packet') { - console.log('[ws-sub] received packet:', (parsed.data as LivePacket)?.packetHash); - } - - const messageByteLength = Buffer.byteLength(messageStr); - if (messageByteLength > WS_MAX_QUEUE_BYTES) { - console.warn(`[ws] skipping oversized live message (${messageByteLength} bytes)`); - return; - } - - const isLiveFeedPacket = parsed?.type === 'packet' - && Number((parsed.data as { packetType?: number } | null)?.packetType) === 5; - - for (const client of wss.clients) { - if (client.readyState !== WebSocket.OPEN) continue; - const scope = clientScopes.get(client); - if (parsed && scope && !shouldSendMessage(parsed, scope)) continue; - if (parsed && scope) trackScopedNodes(parsed, scope); - - // GRP chat packets power the live feed — send immediately so the UI - // never waits on the 16ms batch timer. Other events stay batched. - if (isLiveFeedPacket) { - if (client.bufferedAmount > WS_MAX_BUFFERED_BYTES) { - disconnectSlowClient(client, `socket buffer exceeded ${WS_MAX_BUFFERED_BYTES} bytes`); + for (const client of wss.clients) { + if (client.readyState !== WebSocket.OPEN) continue; + const scope = clientScopes.get(client); + const lines: string[] = []; + let bytes = 0; + for (const item of authorized) { + if (scope && !shouldSendMessage(item.message, scope)) continue; + if (scope) trackScopedNodes(item.message, scope); + lines.push(item.serialized); bytes += item.bytes + 1; + } + if (!lines.length) continue; + if (client.bufferedAmount > WS_MAX_BUFFERED_BYTES || bytes > WS_MAX_QUEUE_BYTES) { + disconnectSlowClient(client, 'public live delivery exceeded the bounded socket budget'); continue; } - // Flush any already-queued events first so ordering stays intact. - const existing = messageQueue.get(client); - if (existing && existing.messages.length > 0) { - client.send(existing.messages.join('\n') + '\n' + messageStr); - messageQueue.delete(client); - } else { - client.send(messageStr); - } - continue; + client.send(lines.join('\n')); } + }, + }); + wss.once('close', () => { void publicMessages.close(); }); - // Queue messages briefly for batched fan-out, with per-client bounds so - // a stalled browser cannot retain an unbounded live-event backlog. - const existing = messageQueue.get(client); - if ( - client.bufferedAmount > WS_MAX_BUFFERED_BYTES - || (existing?.byteLength ?? 0) + messageByteLength > WS_MAX_QUEUE_BYTES - ) { - disconnectSlowClient(client, `live queue exceeded ${WS_MAX_QUEUE_BYTES} bytes`); - continue; - } - if (existing) { - existing.messages.push(messageStr); - existing.byteLength += messageByteLength; - } else { - messageQueue.set(client, { messages: [messageStr], byteLength: messageByteLength }); - } + sub.on('message', (_channel: string, messageStr: string) => { + if (wss.clients.size === 0) return; + let message: WSMessage; + try { message = JSON.parse(messageStr) as WSMessage; } catch { return; } + if (LOG_WS_PACKETS && message.type === 'packet') { + console.log('[ws-sub] received packet:', (message.data as LivePacket)?.packetHash); } - - if (!isLiveFeedPacket) scheduleFlush(); + publicMessages.enqueue(message, Buffer.byteLength(messageStr)); }); return wss; diff --git a/docker-compose.yml b/docker-compose.yml index 3e06f1c..0fd9f4d 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -130,6 +130,7 @@ services: MQTT_RAM_CONSUMER_GROUP: ${MQTT_RAM_CONSUMER_GROUP:-ingest-writers} MQTT_RAM_CONSUMER_NAME: ${MQTT_RAM_CONSUMER_NAME:-backend} MQTT_RAM_DRAIN_DEADLINE_MS: ${MQTT_RAM_DRAIN_DEADLINE_MS:-600000} + MQTT_RAM_PUBLIC_READ_TOKEN: ${MQTT_RAM_PUBLIC_READ_TOKEN:-} PATH_RESOLVE_WORKERS: ${PATH_RESOLVE_WORKERS:-2} PATH_RESOLVE_BACKGROUND_QUEUE_MAX: ${PATH_RESOLVE_BACKGROUND_QUEUE_MAX:-128} PATH_RESOLVE_INTERACTIVE_QUEUE_MAX: ${PATH_RESOLVE_INTERACTIVE_QUEUE_MAX:-32} diff --git a/docs/ram-first-ingestion.md b/docs/ram-first-ingestion.md index 1bd9a1d..b33da82 100644 --- a/docs/ram-first-ingestion.md +++ b/docs/ram-first-ingestion.md @@ -6,17 +6,50 @@ Redis Stream admission; the writer commits each observation later. A Redis loss before PostgreSQL commit can therefore lose already acknowledged telemetry. Keep the Redis instance dedicated to this queue, persistent (AOF), and on a `noeviction` policy. Do not expose the raw stream to public API credentials; -only backend and the explicitly configured private companion reader should -have stream access. +only the backend and a separate quarantine operator should have raw stream +access. Public companions use the credentialed filtered HTTP source below. The pending live overlay applies to recent packet lists and the initial WebSocket packet/message lists. Packet detail and long channel history remain PostgreSQL views and become available after commit; they do not advertise pending entries. Recent packets use a five-minute window, while the initial message list uses 24 hours. The overlay reads at most the configured 10,000 -stream entries per request; a larger legacy stream must be drained before +stream entries per request. SQL also returns aggregates for captured RAM +hashes, and the merged distinct-hash page is limited after representative +selection. A larger legacy stream must be drained before turning on RAM mode. +## Public companion authorization + +Set `MQTT_RAM_PUBLIC_READ_TOKEN` to a random secret of at least 32 characters; +Compose passes it to the backend. The endpoint is disabled when unset. Give +the companion this token, not a Redis credential. Restrict the endpoint's +network reachability to the companion: + +- `GET /internal/ingest/public/policy` returns the current completed public + policy generation, or 503 while authority is unavailable or rematerializing. +- `GET /internal/ingest/public/entries?after=0-0&limit=500` returns currently + public observations with stable IDs and a `nextCursor`. The limit is 1–500. + Send `Authorization: Bearer ` to both endpoints. Private observations + advance the opaque stream cursor without exposing their payloads. + +Authorization uses a fresh policy snapshot after each captured read. REST +merges reject SQL aggregates from an older generation. Initial WebSocket +caches carry a generation and recheck it before delivery. Live MQTT and +WebSocket delivery use one authority query per bounded batch, with explicit +overflow/unavailable counters in `meshcore_public_live_outcomes_total`. +This contract authorizes a captured delivery batch at that final check; +already delivered data and network bytes in flight cannot be recalled. + +The stream is a bounded live overlap. Pair it with PostgreSQL's committed-order +ledger and coverage fence for completeness across retention or cursor loss. +The repaired CoreScope public mode additionally requires an empty dedicated +SQLite measurement database, an ingestor-owned provenance journal and a server +revocation fence. Its ingestor rebuilds copied and derived measurements on a +generation change; the read-only server returns 503 until that generation is +complete and matches current authority. Preserve an old untracked database +and backfill a new dedicated database rather than purging mixed-source data. + ## Enable and monitor Set `MQTT_INGEST_MODE=ram` in the Compose `.env`. The standard