diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index a52a740..de28629 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -198,7 +198,7 @@ jobs: done < <(sed -n 's/^[[:space:]]*file: //p' .github/workflows/release.yml) compose_services="$(docker compose -f docker-compose.yml -f docker-compose.ci.yml --profile dev config --services)" for service in \ - db-migrate backend app-ukmesh website-ukmesh website-dev mesh-health-check \ + db-migrate backend privacy-remat-worker app-ukmesh website-ukmesh website-dev mesh-health-check \ mosquitto-reloader link-worker link-backfill-worker hopreach \ timescaledb mosquitto; do grep -Fxq "$service" <<<"$compose_services" diff --git a/backend/src/api/utils/networkFilters.test.ts b/backend/src/api/utils/networkFilters.test.ts index 078f825..cae73a9 100644 --- a/backend/src/api/utils/networkFilters.test.ts +++ b/backend/src/api/utils/networkFilters.test.ts @@ -33,7 +33,9 @@ test('public packet scopes keep legacy test-topic rows out', () => { assert.match(filters.packets, /is_private IS NOT TRUE/); assert.match(filters.packets, /packet_visibility_materialization_state/); assert.match(filters.packets, /cached_visibility\.visibility_generation = current_visibility\.generation/); - assert.doesNotMatch(filters.packets, /private_node_prefixes|unnest\(/); + assert.match(filters.packets, /privacy_rematerialization_queue remat/); + assert.match(filters.packets, /remat\.status IN \('pending', 'processing', 'failed'\)/); + assert.doesNotMatch(filters.packets, /private_node_prefixes/); assert.doesNotMatch(filters.packets, /private_node\.name LIKE '%🚫%'/); }); diff --git a/backend/src/api/utils/networkFilters.ts b/backend/src/api/utils/networkFilters.ts index e312b58..d767c27 100644 --- a/backend/src/api/utils/networkFilters.ts +++ b/backend/src/api/utils/networkFilters.ts @@ -44,6 +44,30 @@ export function publicPacketPrivacySql(alias?: string): string { WHERE cached_visibility.singleton = TRUE AND cached_visibility.visibility_generation = current_visibility.generation ) + AND NOT EXISTS ( + SELECT 1 + FROM privacy_rematerialization_queue remat + WHERE remat.target = 'packets' + AND remat.status IN ('pending', 'processing', 'failed') + AND ( + ${prefix}rx_node_id = remat.node_id + OR ${prefix}src_node_id = remat.node_id + OR ( + (${prefix}network = remat.network + OR (${prefix}network IN ('ukmesh', 'northeast', 'teesside') + AND remat.network IN ('ukmesh', 'northeast', 'teesside'))) + AND EXISTS ( + SELECT 1 + FROM unnest(COALESCE(${prefix}path_hashes, ARRAY[]::text[])) AS packet_prefix + WHERE UPPER(packet_prefix) IN ( + UPPER(LEFT(remat.node_id, 2)), + UPPER(LEFT(remat.node_id, 4)), + UPPER(LEFT(remat.node_id, 6)) + ) + ) + ) + ) + ) )`; } diff --git a/backend/src/db/migrations/056_async_privacy_rematerialization.sql b/backend/src/db/migrations/056_async_privacy_rematerialization.sql new file mode 100644 index 0000000..c5e6062 --- /dev/null +++ b/backend/src/db/migrations/056_async_privacy_rematerialization.sql @@ -0,0 +1,140 @@ +SET LOCAL lock_timeout = '5s'; +SET LOCAL statement_timeout = '30s'; + +-- Node privacy transitions now enqueue stored-row rematerialization. The +-- worker advances public_visibility_state.generation and +-- packet_visibility_materialization_state.visibility_generation together only +-- after a pass completes. While work is pending, both remain equal at the +-- previous generation: public read gates remain open and node-name filters +-- reflect the new state. Neither generation may advance independently here. +-- The BEFORE nodes lock trigger and identity-table generation trigger retain +-- their existing behavior. +CREATE TABLE IF NOT EXISTS privacy_rematerialization_queue ( + id BIGSERIAL PRIMARY KEY, + node_id TEXT NOT NULL, + network TEXT, + target TEXT NOT NULL DEFAULT 'packets' CHECK (target IN ('packets','packet_paths')), + status TEXT NOT NULL DEFAULT 'pending' CHECK (status IN ('pending','processing','done','failed')), + attempts INTEGER NOT NULL DEFAULT 0, + last_error TEXT, + requested_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), + started_at TIMESTAMPTZ, + finished_at TIMESTAMPTZ, + UNIQUE (target, node_id) +); +CREATE INDEX IF NOT EXISTS privacy_remat_queue_due_idx + ON privacy_rematerialization_queue (status, requested_at); + +CREATE OR REPLACE FUNCTION sync_private_node_prefixes() +RETURNS TRIGGER +LANGUAGE plpgsql +AS $$ +DECLARE + changed_node_id TEXT := COALESCE(NEW.node_id, OLD.node_id); + changed_network TEXT := COALESCE(NEW.network, OLD.network, 'ukmesh'); + old_private BOOLEAN := FALSE; + new_private BOOLEAN := FALSE; +BEGIN + IF TG_OP <> 'INSERT' THEN + old_private := COALESCE(OLD.name, '') LIKE '%🚫%'; + END IF; + IF TG_OP <> 'DELETE' THEN + new_private := COALESCE(NEW.name, '') LIKE '%🚫%'; + END IF; + + IF TG_OP = 'UPDATE' + AND OLD.name IS NOT DISTINCT FROM NEW.name + AND OLD.network IS NOT DISTINCT FROM NEW.network THEN + RETURN NEW; + END IF; + + DELETE FROM private_node_prefixes WHERE node_id = changed_node_id; + + IF TG_OP <> 'DELETE' AND COALESCE(NEW.name, '') LIKE '%🚫%' THEN + INSERT INTO private_node_prefixes (node_id, network, prefix_size_bytes, prefix) + VALUES + (NEW.node_id, COALESCE(NEW.network, 'ukmesh'), 1, UPPER(LEFT(NEW.node_id, 2))), + (NEW.node_id, COALESCE(NEW.network, 'ukmesh'), 2, UPPER(LEFT(NEW.node_id, 4))), + (NEW.node_id, COALESCE(NEW.network, 'ukmesh'), 3, UPPER(LEFT(NEW.node_id, 6))) + ON CONFLICT (node_id, network, prefix_size_bytes) DO UPDATE SET + prefix = EXCLUDED.prefix, + updated_at = NOW(); + END IF; + + IF old_private IS DISTINCT FROM new_private THEN + INSERT INTO privacy_rematerialization_queue (node_id, network, target) + VALUES (changed_node_id, changed_network, 'packets') + ON CONFLICT (target, node_id) DO UPDATE SET + requested_at = NOW(), + network = EXCLUDED.network, + status = CASE WHEN privacy_rematerialization_queue.status = 'processing' + THEN 'processing' ELSE 'pending' END, + attempts = CASE WHEN privacy_rematerialization_queue.status = 'processing' + THEN privacy_rematerialization_queue.attempts ELSE 0 END, + finished_at = CASE WHEN privacy_rematerialization_queue.status = 'processing' + THEN privacy_rematerialization_queue.finished_at ELSE NULL END; + END IF; + + IF TG_OP = 'DELETE' THEN RETURN OLD; END IF; + RETURN NEW; +END; +$$; + +DROP TRIGGER IF EXISTS nodes_private_prefix_materialization ON nodes; +CREATE TRIGGER nodes_private_prefix_materialization +AFTER INSERT OR DELETE OR UPDATE ON nodes +FOR EACH ROW +EXECUTE FUNCTION sync_private_node_prefixes(); + +CREATE OR REPLACE FUNCTION rematerialize_packet_path_privacy() +RETURNS TRIGGER +LANGUAGE plpgsql +AS $$ +DECLARE + changed_node_id TEXT := COALESCE(NEW.node_id, OLD.node_id); + changed_network TEXT := COALESCE(NEW.network, OLD.network, 'ukmesh'); + old_private BOOLEAN := TG_OP <> 'INSERT' AND COALESCE(OLD.name, '') LIKE '%🚫%'; + new_private BOOLEAN := TG_OP <> 'DELETE' AND COALESCE(NEW.name, '') LIKE '%🚫%'; +BEGIN + IF old_private IS DISTINCT FROM new_private THEN + INSERT INTO privacy_rematerialization_queue (node_id, network, target) + VALUES (changed_node_id, changed_network, 'packet_paths') + ON CONFLICT (target, node_id) DO UPDATE SET + requested_at = NOW(), + network = EXCLUDED.network, + status = CASE WHEN privacy_rematerialization_queue.status = 'processing' + THEN 'processing' ELSE 'pending' END, + attempts = CASE WHEN privacy_rematerialization_queue.status = 'processing' + THEN privacy_rematerialization_queue.attempts ELSE 0 END, + finished_at = CASE WHEN privacy_rematerialization_queue.status = 'processing' + THEN privacy_rematerialization_queue.finished_at ELSE NULL END; + END IF; + IF TG_OP = 'DELETE' THEN RETURN OLD; END IF; + RETURN NEW; +END; +$$; + +DROP TRIGGER IF EXISTS nodes_private_zz_packet_path_materialization ON nodes; +CREATE TRIGGER nodes_private_zz_packet_path_materialization +AFTER INSERT OR UPDATE OR DELETE ON nodes +FOR EACH ROW +EXECUTE FUNCTION rematerialize_packet_path_privacy(); + +-- The worker owns the generation pair. Nested prefix maintenance is already +-- ignored by bump_visibility_for_identity_table(); keep that function and its +-- direct identity-mutation behavior unchanged. +CREATE OR REPLACE FUNCTION bump_public_visibility_generation() +RETURNS TRIGGER +LANGUAGE plpgsql +AS $$ +BEGIN + IF TG_OP = 'DELETE' THEN RETURN OLD; END IF; + RETURN NEW; +END; +$$; + +DROP TRIGGER IF EXISTS nodes_public_visibility_generation ON nodes; +CREATE TRIGGER nodes_public_visibility_generation +AFTER INSERT OR UPDATE OR DELETE ON nodes +FOR EACH ROW +EXECUTE FUNCTION bump_public_visibility_generation(); diff --git a/backend/src/db/packetBatch.integration.test.ts b/backend/src/db/packetBatch.integration.test.ts index 010a32f..37431a6 100644 --- a/backend/src/db/packetBatch.integration.test.ts +++ b/backend/src/db/packetBatch.integration.test.ts @@ -20,6 +20,7 @@ import { flush, type PacketBatchInput, } from './packetBatch.js'; +import { processNextPrivacyRematerialization } from '../workers/privacy-rematerialization.js'; const { Pool } = pg; const databaseUrl = process.env['TEST_INGEST_DATABASE_URL']; @@ -102,6 +103,7 @@ test('base schema and every migration apply to a brand-new database', { assert.deepEqual(executed, [ '050_packet_paths_and_retention.sql', '051_reset_readiness.sql', + '056_async_privacy_rematerialization.sql', ]); assert.deepEqual(await runMigrations(freshPool), []); @@ -249,6 +251,169 @@ test('base schema and every migration apply to a brand-new database', { ]); }); +test('privacy rename commits and rematerializes a compressed chunk above the old DML budget', { + skip: databaseUrl ? false : 'TEST_INGEST_DATABASE_URL is not configured', +}, async (t) => { + const adminPool = new Pool({ connectionString: databaseUrl, max: 1 }); + const databaseName = `meshcore_remat_${process.pid}_${Date.now()}`; + const databaseIdentifier = `"${databaseName}"`; + const url = new URL(databaseUrl as string); + url.pathname = `/${databaseName}`; + let pool: pg.Pool | null = null; + t.after(async () => { + await pool?.end(); + await adminPool.query(`DROP DATABASE IF EXISTS ${databaseIdentifier} WITH (FORCE)`); + await adminPool.end(); + }); + + await adminPool.query(`CREATE DATABASE ${databaseIdentifier}`); + pool = new Pool({ connectionString: url.toString(), max: 2 }); + await pool.query(fs.readFileSync(new URL('./schema/base.sql', import.meta.url), 'utf8')); + await runMigrations(pool); + + const nodeId = 'F'.repeat(64); + await pool.query(`INSERT INTO nodes (node_id, name, network) VALUES ($1, 'incident public', 'ukmesh') + ON CONFLICT (node_id) DO UPDATE SET name = EXCLUDED.name`, [nodeId]); + configurePacketBatch((text, params) => pool!.query(text, params)); + const now = new Date(); + const smallPackets: PacketBatchInput[] = [0, 1, 2].map((index) => ({ + time: new Date(now.getTime() + index), + packetHash: `incident-small-${index}`, + rxNodeId: nodeId, + srcNodeId: nodeId, + topic: 'meshcore/ZZZ/incident', + topicPrefix: 'meshcore', + iata: 'ZZZ', + packetType: 4, + routeType: 0, + hopCount: 1, + rssi: -80, + snr: 5, + payloadJson: '{}', + companionSender: null, + rawHex: '00', + advertCount: null, + pathHashes: [nodeId.slice(0, 4)], + pathHashSizeBytes: 2, + network: 'ukmesh', + transportCodes: null, + regionScope: null, + })); + const writes = smallPackets.map((packet) => enqueuePacket(packet)); + await flush(); + assert.equal((await Promise.all(writes)).length, 3); + await pool.query(` + INSERT INTO packets (time, packet_hash, topic, network) + VALUES ($1, 'incident-unrelated', 'meshcore/ZZZ/unrelated', 'ukmesh') + `, [new Date(now.getTime() + 10_000)]); + + const fixtureTime = new Date(now); + fixtureTime.setUTCHours(12, 0, 0, 0); + fixtureTime.setUTCDate(fixtureTime.getUTCDate() - 2); + const fixtureRows = 100_001; + const inserted = await pool.query(` + INSERT INTO packets (time, packet_hash, rx_node_id, src_node_id, topic, network) + SELECT $1::timestamptz, 'incident-compressed-' || n::text, $2, $2, + 'meshcore/ZZZ/incident-compressed', 'ukmesh' + FROM generate_series(1, $3::int) AS n + `, [fixtureTime, nodeId, fixtureRows]); + assert.equal(inserted.rowCount, fixtureRows); + const chunk = await pool.query<{ chunk_schema: string; chunk_name: string }>(` + SELECT chunk_schema, chunk_name FROM timescaledb_information.chunks + WHERE hypertable_name = 'packets' AND range_start <= $1 AND range_end > $1 + `, [fixtureTime]); + assert.equal(chunk.rows.length, 1); + const chunkName = `"${chunk.rows[0]!.chunk_schema}"."${chunk.rows[0]!.chunk_name}"`; + await pool.query('SELECT compress_chunk($1::regclass)', [chunkName]); + const compressed = await pool.query<{ is_compressed: boolean }>(` + SELECT is_compressed FROM timescaledb_information.chunks + WHERE hypertable_name = 'packets' AND chunk_schema = $1 AND chunk_name = $2 + `, [chunk.rows[0]!.chunk_schema, chunk.rows[0]!.chunk_name]); + assert.equal(compressed.rows[0]?.is_compressed, true); + + const pair = async () => { + const result = await pool!.query<{ public_generation: string; materialized_generation: string }>(` + SELECT p.generation::text AS public_generation, + m.visibility_generation::text AS materialized_generation + FROM public_visibility_state p + JOIN packet_visibility_materialization_state m ON p.singleton = m.singleton + `); + assert.equal(result.rows[0]?.public_generation, result.rows[0]?.materialized_generation); + return Number(result.rows[0]?.public_generation); + }; + const initialGeneration = await pair(); + await pool.query("UPDATE nodes SET name = 'incident private 🚫' WHERE node_id = $1", [nodeId]); + assert.equal((await pool.query<{ name: string }>('SELECT name FROM nodes WHERE node_id = $1', [nodeId])).rows[0]?.name, 'incident private 🚫'); + assert.equal(await pair(), initialGeneration); + const queued = await pool.query<{ target: string; status: string }>(` + SELECT target, status FROM privacy_rematerialization_queue WHERE node_id = $1 ORDER BY target + `, [nodeId]); + assert.deepEqual(queued.rows, [ + { target: 'packet_paths', status: 'pending' }, + { target: 'packets', status: 'pending' }, + ]); + const duringPendingPacket = { + ...smallPackets[0]!, + time: new Date(now.getTime() + 5_000), + packetHash: 'incident-during-pending', + }; + const pendingWrite = enqueuePacket(duringPendingPacket); + await flush(); + assert.deepEqual(await pendingWrite, { isPrivate: true, visibilityOk: false }); + const persistedPending = await pool.query<{ is_private: boolean; visibility_ok: boolean }>( + 'SELECT is_private, visibility_ok FROM packets WHERE packet_hash = $1', + [duringPendingPacket.packetHash], + ); + assert.deepEqual(persistedPending.rows, [{ is_private: true, visibility_ok: false }]); + const publicScope = networkFilters('ukmesh'); + const visibleDuringPending = await pool.query<{ packet_hash: string }>(` + SELECT packet_hash FROM packets + WHERE packet_hash = ANY($2::text[]) ${publicScope.packets} + ORDER BY packet_hash + `, [publicScope.params[0], [smallPackets[0]!.packetHash, 'incident-unrelated']]); + assert.deepEqual(visibleDuringPending.rows, [{ packet_hash: 'incident-unrelated' }]); + + // The old trigger's default 100k decompression budget cannot update this + // chunk; the worker transaction explicitly removes that per-DML limit. + const limitedClient = await pool.connect(); + try { + await limitedClient.query('BEGIN'); + await limitedClient.query('SET LOCAL timescaledb.max_tuples_decompressed_per_dml_transaction = 100000'); + await assert.rejects(limitedClient.query(` + UPDATE packets SET is_private = is_private WHERE time = $1 AND rx_node_id = $2 + `, [fixtureTime, nodeId]), /decompress|tuple|limit/i); + } finally { + await limitedClient.query('ROLLBACK'); + limitedClient.release(); + } + + assert.equal(await processNextPrivacyRematerialization(pool, { chunkPauseMs: 0 }), true); + assert.ok(await pair() > initialGeneration); + assert.equal(await processNextPrivacyRematerialization(pool, { chunkPauseMs: 0 }), true); + const privateRows = await pool.query<{ count: string }>(` + SELECT COUNT(*)::text AS count FROM packets + WHERE time = $1 AND rx_node_id = $2 AND is_private AND NOT visibility_ok + `, [fixtureTime, nodeId]); + assert.equal(privateRows.rows[0]?.count, String(fixtureRows)); + const privatePaths = await pool.query<{ count: string }>(` + SELECT COUNT(*)::text AS count FROM packet_paths + WHERE rx_node_id = $1 AND is_private AND NOT visibility_ok + `, [nodeId]); + assert.equal(privatePaths.rows[0]?.count, '4'); + + const beforePublicGeneration = await pair(); + await pool.query("UPDATE nodes SET name = 'incident public' WHERE node_id = $1", [nodeId]); + assert.equal(await pair(), beforePublicGeneration); + assert.equal(await processNextPrivacyRematerialization(pool, { chunkPauseMs: 0 }), true); + assert.equal(await processNextPrivacyRematerialization(pool, { chunkPauseMs: 0 }), true); + assert.ok(await pair() > beforePublicGeneration); + const publicRows = await pool.query<{ count: string }>(` + SELECT COUNT(*)::text AS count FROM packets + WHERE time = $1 AND rx_node_id = $2 AND NOT is_private AND visibility_ok + `, [fixtureTime, nodeId]); + assert.equal(publicRows.rows[0]?.count, String(fixtureRows)); +}); + test('packet batch atomically coalesces observer, sighting, and stats writes', { skip: databaseUrl ? false : 'TEST_INGEST_DATABASE_URL is not configured', }, async (t) => { @@ -296,7 +461,8 @@ test('packet batch atomically coalesces observer, sighting, and stats writes', { process.env['MIGRATION_016_PRIVATE_PREFIXES_APPROVAL'] = previousApproval; } } - await pool.query('DELETE FROM nodes WHERE node_id = $1', [compatibilityNode]); + // Retain the public fixture: deleting a node cascades a direct prefix-table + // statement and deliberately invalidates the existing identity fence. const migrationRunLengths = migrationRuns.map((run) => run.length).sort((a, b) => a - b); assert.equal(migrationRunLengths[0], 0); assert.ok((migrationRunLengths[1] ?? 0) > 0); @@ -379,10 +545,38 @@ test('packet batch atomically coalesces observer, sighting, and stats writes', { 'DELETE FROM packets WHERE packet_hash = ANY($1::text[])', [Object.values(legacyHashes)], ); - await pool.query( - 'DELETE FROM nodes WHERE node_id = ANY($1::text[])', - [[legacyPrivateNode, legacyPublicNode]], - ); + // Leave these isolated test identities in the scratch database. Their + // deletion would independently invalidate the legacy identity fence. + await pool.query('DELETE FROM privacy_rematerialization_queue'); + + const visibilityPair = async () => { + const result = await pool.query<{ public_generation: string; materialized_generation: string }>(` + SELECT public.generation::text AS public_generation, + materialized.visibility_generation::text AS materialized_generation + FROM public_visibility_state public + JOIN packet_visibility_materialization_state materialized ON materialized.singleton = public.singleton + WHERE public.singleton = TRUE + `); + assert.equal(result.rows[0]?.public_generation, result.rows[0]?.materialized_generation); + return Number(result.rows[0]?.public_generation); + }; + const runQueuedRemat = async (nodeId: string) => { + const queue = await pool.query<{ target: string; status: string }>(` + SELECT target, status FROM privacy_rematerialization_queue + WHERE node_id = $1 ORDER BY target + `, [nodeId]); + assert.deepEqual(queue.rows, [ + { target: 'packet_paths', status: 'pending' }, + { target: 'packets', status: 'pending' }, + ]); + assert.equal(await processNextPrivacyRematerialization(pool, { chunkPauseMs: 0 }), true); + await visibilityPair(); + assert.equal(await processNextPrivacyRematerialization(pool, { chunkPauseMs: 0 }), true); + const completed = await pool.query<{ status: string }>( + 'SELECT status FROM privacy_rematerialization_queue WHERE node_id = $1', [nodeId], + ); + assert.deepEqual(completed.rows.map((row) => row.status), ['done', 'done']); + }; const network = `ingest_${suffix}`; const observerId = `A${suffix.padStart(63, 'A')}`.slice(0, 64); @@ -574,10 +768,17 @@ test('packet batch atomically coalesces observer, sighting, and stats writes', { } assert.deepEqual(shadowWarnings, []); + const beforePrivateGeneration = await visibilityPair(); await pool.query( `UPDATE nodes SET name = 'integration observer 🚫' WHERE node_id = $1`, [observerId], ); + assert.equal((await pool.query<{ name: string }>( + 'SELECT name FROM nodes WHERE node_id = $1', [observerId], + )).rows[0]?.name, 'integration observer 🚫'); + assert.equal(await visibilityPair(), beforePrivateGeneration); + await runQueuedRemat(observerId); + assert.ok(await visibilityPair() > beforePrivateGeneration); const privateBeforeRetention = await pool.query<{ private_packets: string; private_paths: string; @@ -593,10 +794,13 @@ test('packet batch atomically coalesces observer, sighting, and stats writes', { private_paths: '3', }); + const beforePublicGeneration = await visibilityPair(); await pool.query( `UPDATE nodes SET name = 'integration observer' WHERE node_id = $1`, [observerId], ); + assert.equal(await visibilityPair(), beforePublicGeneration); + await runQueuedRemat(observerId); const publicBeforeRetention = await pool.query<{ public_packets: string; public_paths: string; @@ -624,6 +828,7 @@ test('packet batch atomically coalesces observer, sighting, and stats writes', { `UPDATE nodes SET name = 'integration observer 🚫' WHERE node_id = $1`, [observerId], ); + await runQueuedRemat(observerId); const privateAfterRetention = await pool.query<{ packet_count: string; private_paths: string; @@ -641,6 +846,7 @@ test('packet batch atomically coalesces observer, sighting, and stats writes', { `UPDATE nodes SET name = 'integration observer' WHERE node_id = $1`, [observerId], ); + await runQueuedRemat(observerId); const visibility = await pool.query<{ generation: string }>( `SELECT generation::text FROM public_visibility_state WHERE singleton = TRUE`, diff --git a/backend/src/db/packetBatch.ts b/backend/src/db/packetBatch.ts index 62b51eb..5878372 100644 --- a/backend/src/db/packetBatch.ts +++ b/backend/src/db/packetBatch.ts @@ -246,6 +246,12 @@ async function writeBatch(batch: PendingPacket[], idempotent: boolean): Promise< WHERE singleton = TRUE FOR KEY SHARE ), + privacy_remat_state AS MATERIALIZED ( + SELECT EXISTS ( + SELECT 1 FROM privacy_rematerialization_queue + WHERE status IN ('pending', 'processing', 'failed') + ) AS active + ), incoming ( row_id, 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, @@ -256,12 +262,17 @@ async function writeBatch(batch: PendingPacket[], idempotent: boolean): Promise< SELECT i.*, gen_random_uuid() AS observation_id, CASE + WHEN remat.active THEN meshcore_path_matches_private( + i.network, i.rx_node_id, i.src_node_id, + i.path_hashes, i.path_hash_size_bytes + ) WHEN i.prefix_cache_generation = visibility.generation THEN i.classified_private ELSE TRUE END AS is_private, i.prefix_cache_generation = visibility.generation AS prefix_cache_fresh FROM incoming i CROSS JOIN current_visibility visibility + CROSS JOIN privacy_remat_state remat ), ${idempotencyCte} inserted AS ( diff --git a/backend/src/db/packetVisibilityFence.test.ts b/backend/src/db/packetVisibilityFence.test.ts index f847c39..abf33ab 100644 --- a/backend/src/db/packetVisibilityFence.test.ts +++ b/backend/src/db/packetVisibilityFence.test.ts @@ -6,6 +6,10 @@ const migration = readFileSync( new URL('./migrations/042_packet_visibility_fence.sql', import.meta.url), 'utf8', ); +const asyncMigration = readFileSync( + new URL('./migrations/056_async_privacy_rematerialization.sql', import.meta.url), + 'utf8', +); test('packet visibility materialization is fenced across privacy identity changes', () => { assert.match(migration, /packet_visibility_materialization_state/); @@ -16,3 +20,12 @@ test('packet visibility materialization is fenced across privacy identity change assert.match(migration, /TG_TABLE_NAME = 'private_node_prefixes' AND pg_trigger_depth\(\) > 1/); assert.match(migration, /deliberately leave the materialization\s+-- generation stale/); }); + +test('node triggers enqueue rematerialization and leave paired generation advancement to worker', () => { + assert.match(asyncMigration, /CREATE TABLE IF NOT EXISTS privacy_rematerialization_queue/); + assert.match(asyncMigration, /VALUES \(changed_node_id, changed_network, 'packets'\)/); + assert.match(asyncMigration, /VALUES \(changed_node_id, changed_network, 'packet_paths'\)/); + assert.doesNotMatch(asyncMigration, /UPDATE packets p|UPDATE packet_paths path/); + assert.match(asyncMigration, /CREATE TRIGGER nodes_public_visibility_generation/); + assert.doesNotMatch(asyncMigration, /SET generation = generation \+ 1/); +}); diff --git a/backend/src/db/publicVisibilityMigration.test.ts b/backend/src/db/publicVisibilityMigration.test.ts index fbab53b..3fa933c 100644 --- a/backend/src/db/publicVisibilityMigration.test.ts +++ b/backend/src/db/publicVisibilityMigration.test.ts @@ -6,9 +6,20 @@ const migration = readFileSync( new URL('./migrations/041_stable_public_visibility_generation.sql', import.meta.url), 'utf8', ); +const asyncMigration = readFileSync( + new URL('./migrations/056_async_privacy_rematerialization.sql', import.meta.url), + 'utf8', +); test('public visibility generation changes only when node privacy changes', () => { assert.match(migration, /old_private IS DISTINCT FROM new_private/); assert.doesNotMatch(migration, /decode_inputs_changed|OLD\.lat|OLD\.lon|OLD\.role/); assert.match(migration, /IF TG_OP = 'DELETE' THEN RETURN OLD/); }); + +test('current node trigger defers visibility generation changes to rematerialization', () => { + const currentFunction = asyncMigration.split('CREATE OR REPLACE FUNCTION bump_public_visibility_generation()', 2)[1]; + assert.ok(currentFunction); + assert.doesNotMatch(currentFunction, /UPDATE public_visibility_state/); + assert.match(currentFunction, /IF TG_OP = 'DELETE' THEN RETURN OLD/); +}); diff --git a/backend/src/health/status.ts b/backend/src/health/status.ts index f689ab0..c3802e0 100644 --- a/backend/src/health/status.ts +++ b/backend/src/health/status.ts @@ -633,7 +633,7 @@ export async function getWorkerHealthOverview() { // Compute system stats once — cpuUsagePct() diffs against lastCpuSample, // so calling it twice in one request gives a garbage near-zero second reading. const sysStats = systemStats(); - const [workers, history, errors1h, ingest, pathDiagnostics, operationalChecks, databaseMaintenance, databaseRuntime, redisDurability] = await Promise.all([ + const [workers, history, errors1h, ingest, pathDiagnostics, operationalChecks, databaseMaintenance, databaseRuntime, redisDurability, privacyRemat] = await Promise.all([ currentWorkerStatuses(sysStats), query<{ ts: string; @@ -702,6 +702,23 @@ export async function getWorkerHealthOverview() { WHERE datname = current_database()`, ), redisDurabilityState(), + query<{ + pending: string; + processing: string; + failed: string; + failed_retryable: string; + failed_exhausted: string; + oldest_pending_age_seconds: string | null; + }>(` + SELECT COUNT(*) FILTER (WHERE status = 'pending')::text AS pending, + COUNT(*) FILTER (WHERE status = 'processing')::text AS processing, + COUNT(*) FILTER (WHERE status = 'failed')::text AS failed, + COUNT(*) FILTER (WHERE status = 'failed' AND attempts < $1)::text AS failed_retryable, + COUNT(*) FILTER (WHERE status = 'failed' AND attempts >= $1)::text AS failed_exhausted, + MAX(EXTRACT(EPOCH FROM (NOW() - requested_at))) + FILTER (WHERE status = 'pending')::text AS oldest_pending_age_seconds + FROM privacy_rematerialization_queue + `, [Math.min(100, Math.max(1, Number(process.env['PRIVACY_REMAT_MAX_ATTEMPTS'] ?? 8) || 8))]), ]); const ingestRow = ingest.rows[0]; @@ -777,6 +794,17 @@ export async function getWorkerHealthOverview() { message: `${tablesNeedingVacuum} database table(s) exceed the dead-row vacuum threshold`, }); } + const privacyRematRow = privacyRemat.rows[0]; + const privacyRematProblem = privacyRematerializationBacklogProblem({ + pending: Number(privacyRematRow?.pending ?? 0), + processing: Number(privacyRematRow?.processing ?? 0), + failed: Number(privacyRematRow?.failed ?? 0), + failedRetryable: Number(privacyRematRow?.failed_retryable ?? 0), + failedExhausted: Number(privacyRematRow?.failed_exhausted ?? 0), + oldestPendingAgeSeconds: privacyRematRow?.oldest_pending_age_seconds == null + ? null : Number(privacyRematRow.oldest_pending_age_seconds), + }); + if (privacyRematProblem) problems.push(privacyRematProblem); if (redisDurability.maxmemory_policy !== 'noeviction') { problems.push({ code: 'redis_eviction_policy_unsafe', @@ -837,6 +865,27 @@ export type HealthProblem = { message: string; }; +export function privacyRematerializationBacklogProblem(state: { + pending: number; + processing: number; + failed: number; + failedRetryable: number; + failedExhausted: number; + oldestPendingAgeSeconds: number | null; +}): HealthProblem | null { + const age = state.oldestPendingAgeSeconds ?? 0; + const severity = age > 6 * 60 * 60 || state.failedExhausted > 0 + ? 'critical' + : age > 30 * 60 || state.failedRetryable > 0 ? 'warning' : null; + if (!severity) return null; + return { + code: 'privacy_rematerialization_backlog', + severity, + message: `Privacy rematerialization pending=${state.pending} processing=${state.processing} ` + + `failed=${state.failed} oldest_pending_age_minutes=${Math.floor(age / 60)}`, + }; +} + /** * Anonymous browser diagnostics are reported to operators but are deliberately * absent from the authoritative status calculation. diff --git a/backend/src/health/statusIntegrity.test.ts b/backend/src/health/statusIntegrity.test.ts index 34a0032..6a8d546 100644 --- a/backend/src/health/statusIntegrity.test.ts +++ b/backend/src/health/statusIntegrity.test.ts @@ -1,6 +1,6 @@ import assert from 'node:assert/strict'; import test from 'node:test'; -import { summarizeAuthoritativeHealth } from './status.js'; +import { privacyRematerializationBacklogProblem, summarizeAuthoritativeHealth } from './status.js'; test('anonymous frontend diagnostics cannot forge health severity', () => { assert.deepEqual(summarizeAuthoritativeHealth([], 1_000_000), { @@ -20,3 +20,17 @@ test('anonymous frontend diagnostics cannot forge health severity', () => { frontendErrors: 0, }); }); + +test('privacy rematerialization backlog ignores active work and classifies stale or failed work', () => { + const state = { + pending: 0, processing: 1, failed: 0, failedRetryable: 0, + failedExhausted: 0, oldestPendingAgeSeconds: null, + }; + assert.equal(privacyRematerializationBacklogProblem(state), null); + assert.equal(privacyRematerializationBacklogProblem({ ...state, pending: 1, oldestPendingAgeSeconds: 1_801 })?.severity, 'warning'); + assert.equal(privacyRematerializationBacklogProblem({ ...state, failed: 1, failedRetryable: 1 })?.severity, 'warning'); + const exhausted = privacyRematerializationBacklogProblem({ ...state, failed: 1, failedExhausted: 1 }); + assert.equal(exhausted?.severity, 'critical'); + assert.match(exhausted?.message ?? '', /pending=0 processing=1 failed=1 oldest_pending_age_minutes=0/); + assert.equal(privacyRematerializationBacklogProblem({ ...state, pending: 1, oldestPendingAgeSeconds: 21_601 })?.severity, 'critical'); +}); diff --git a/backend/src/metrics.ts b/backend/src/metrics.ts index 1abb8ff..159b64b 100644 --- a/backend/src/metrics.ts +++ b/backend/src/metrics.ts @@ -299,6 +299,7 @@ const METRIC_NETWORKS = new Set(['ukmesh', 'teesside', 'test']); const METRIC_ANALYSIS_WORKLOADS = new Set(['spam-analysis', 'path-learning']); const METRIC_WORKERS = new Set([ 'health', + 'privacy_remat', 'path_learning', 'synthetic', 'link_backfill', @@ -306,6 +307,8 @@ const METRIC_WORKERS = new Set([ ]); const METRIC_WORKER_PHASES = new Set([ 'snapshot', + 'pass', + 'chunk', 'cleanup', 'rebuild', 'refresh', diff --git a/backend/src/mqtt/client.ts b/backend/src/mqtt/client.ts index e925781..32e31b6 100644 --- a/backend/src/mqtt/client.ts +++ b/backend/src/mqtt/client.ts @@ -738,33 +738,38 @@ async function handleMessage(topic: string, rawPayload: Buffer): Promise { } if (!isSpam) { - const nodeUpdate = await upsertNode(nodeId, { - name: advertName, - lat: advertLat, - lon: advertLon, - role: appData?.['deviceRole'] as number | undefined, - iata: nodeIata, - publicKey: senderKey, - network, - advertHash: canonicalPacketId, - }); - advertCount = nodeUpdate.advertCount; + try { + const nodeUpdate = await upsertNode(nodeId, { + name: advertName, + lat: advertLat, + lon: advertLon, + role: appData?.['deviceRole'] as number | undefined, + iata: nodeIata, + publicKey: senderKey, + network, + advertHash: canonicalPacketId, + }); + advertCount = nodeUpdate.advertCount; - emitNodeUpsert({ - node_id: nodeId, - name: advertName, - lat: advertLat, - lon: advertLon, - role: appData?.['deviceRole'] as number | undefined, - iata, - network, - observer_id: observerKey, - public_key: senderKey, - last_seen: new Date().toISOString(), - is_online: true, - advert_count: advertCount, - coordinates_changed: nodeUpdate.coordinatesChanged, - }); + emitNodeUpsert({ + node_id: nodeId, + name: advertName, + lat: advertLat, + lon: advertLon, + role: appData?.['deviceRole'] as number | undefined, + iata, + network, + observer_id: observerKey, + public_key: senderKey, + last_seen: new Date().toISOString(), + is_online: true, + advert_count: advertCount, + coordinates_changed: nodeUpdate.coordinatesChanged, + }); + } catch (err) { + mqttIngestOutcomesTotal.inc({ outcome: 'advert_persist_failure' }); + console.error('[mqtt] advert node upsert failed:', (err as Error).message); + } } } diff --git a/backend/src/workers/privacy-rematerialization.ts b/backend/src/workers/privacy-rematerialization.ts new file mode 100644 index 0000000..c6c7bc6 --- /dev/null +++ b/backend/src/workers/privacy-rematerialization.ts @@ -0,0 +1,268 @@ +import 'node:process'; +import { setTimeout as delay } from 'node:timers/promises'; +import { pathToFileURL } from 'node:url'; +import pg from 'pg'; +import { observeWorkerOutcome } from '../metrics.js'; +import { startWorkerMetrics } from './workerMetrics.js'; + +const { Pool } = pg; + +function boundedEnvNumber(name: string, fallback: number, min: number, max: number): number { + const value = Number(process.env[name]); + return Number.isFinite(value) ? Math.min(max, Math.max(min, Math.trunc(value))) : fallback; +} + +const INTERVAL_MS = boundedEnvNumber('PRIVACY_REMAT_INTERVAL_MS', 120_000, 1_000, 3_600_000); +export const MAX_ATTEMPTS = boundedEnvNumber('PRIVACY_REMAT_MAX_ATTEMPTS', 8, 1, 100); +const CHUNK_PAUSE_MS = boundedEnvNumber('PRIVACY_REMAT_CHUNK_PAUSE_MS', 500, 0, 60_000); +const CHUNK_STATEMENT_TIMEOUT_MS = boundedEnvNumber( + 'PRIVACY_REMAT_CHUNK_STATEMENT_TIMEOUT_MS', 900_000, 1_000, 3_600_000, +); + +type Target = 'packets' | 'packet_paths'; +type QueueRow = { + id: string; + node_id: string; + network: string | null; + target: Target; + attempts: number; + requested_at: Date; + started_at: Date; +}; +type Chunk = { range_start: Date; range_end: Date }; + +// The two assignment expressions below are copied verbatim from migration 049. +// Keep packet privacy semantics here aligned with the deployed DB function. +const PACKET_UPDATE_SQL = `UPDATE packets p + SET is_private = EXISTS ( + SELECT 1 + FROM private_node_prefixes pp + WHERE ( + pp.network = p.network + OR ( + pp.network IN ('ukmesh', 'northeast', 'teesside') + AND p.network IN ('ukmesh', 'northeast', 'teesside') + ) + ) + AND ( + pp.node_id IN (p.rx_node_id, p.src_node_id) + OR ( + p.path_hash_size_bytes = pp.prefix_size_bytes + AND EXISTS ( + SELECT 1 + FROM unnest(COALESCE(p.path_hashes, ARRAY[]::text[])) AS packet_prefix + WHERE UPPER(packet_prefix) = pp.prefix + ) + ) + ) + ), + visibility_ok = ( + (COALESCE(cardinality(p.path_hashes), 0) = 0 OR p.path_hash_size_bytes BETWEEN 1 AND 3) + AND NOT EXISTS ( + SELECT 1 + FROM private_node_prefixes pp + WHERE ( + pp.network = p.network + OR ( + pp.network IN ('ukmesh', 'northeast', 'teesside') + AND p.network IN ('ukmesh', 'northeast', 'teesside') + ) + ) + AND ( + pp.node_id IN (p.rx_node_id, p.src_node_id) + OR ( + p.path_hash_size_bytes = pp.prefix_size_bytes + AND EXISTS ( + SELECT 1 + FROM unnest(COALESCE(p.path_hashes, ARRAY[]::text[])) AS packet_prefix + WHERE UPPER(packet_prefix) = pp.prefix + ) + ) + ) + ) + ) + WHERE p.time >= $1 AND p.time < $2 + AND ( + p.rx_node_id = $3 OR p.src_node_id = $3 + OR ((p.network = $4 OR (p.network IN ('ukmesh', 'northeast', 'teesside') + AND $4 IN ('ukmesh', 'northeast', 'teesside'))) + AND EXISTS ( + SELECT 1 FROM unnest(COALESCE(p.path_hashes, ARRAY[]::text[])) AS pp + WHERE UPPER(pp) IN (UPPER(LEFT($3, 2)), UPPER(LEFT($3, 4)), UPPER(LEFT($3, 6))) + )) + )`; + +const PATH_UPDATE_SQL = `UPDATE packet_paths path + SET is_private = meshcore_path_matches_private( + path.network, path.rx_node_id, path.src_node_id, + path.path_hashes, path.path_hash_size_bytes + ), + visibility_ok = meshcore_path_is_valid( + path.path_hashes, path.path_hash_size_bytes + ) AND NOT meshcore_path_matches_private( + path.network, path.rx_node_id, path.src_node_id, + path.path_hashes, path.path_hash_size_bytes + ) + WHERE path.time >= $1 AND path.time < $2 + AND (path.rx_node_id = $3 OR path.src_node_id = $3 + OR ((path.network = $4 OR (path.network IN ('ukmesh', 'northeast', 'teesside') + AND $4 IN ('ukmesh', 'northeast', 'teesside'))) + AND path.path_hashes && ARRAY[ + UPPER(LEFT($3, 2)), UPPER(LEFT($3, 4)), UPPER(LEFT($3, 6)) + ]::text[]))`; + +async function claim(pool: pg.Pool, maxAttempts: number): Promise { + const result = await pool.query(` + UPDATE privacy_rematerialization_queue + SET status = 'processing', started_at = NOW(), attempts = attempts + 1, + finished_at = NULL, last_error = NULL + WHERE id = ( + SELECT id FROM privacy_rematerialization_queue + WHERE status = 'pending' + OR (status = 'failed' AND attempts < $1 + AND finished_at < NOW() - (attempts * attempts || ' minutes')::interval) + ORDER BY requested_at LIMIT 1 FOR UPDATE SKIP LOCKED + ) + RETURNING id::text, node_id, network, target, attempts, requested_at, started_at + `, [maxAttempts]); + return result.rows[0] ?? null; +} + +async function processChunk(pool: pg.Pool, row: QueueRow, chunk: Chunk, timeoutMs: number): Promise { + for (let attempt = 1; attempt <= 3; attempt += 1) { + const client = await pool.connect(); + try { + await client.query('BEGIN'); + await client.query("SET LOCAL lock_timeout = '5s'"); + await client.query(`SET LOCAL statement_timeout = '${timeoutMs}ms'`); + await client.query('SET LOCAL timescaledb.max_tuples_decompressed_per_dml_transaction = 0'); + await client.query(row.target === 'packets' ? PACKET_UPDATE_SQL : PATH_UPDATE_SQL, [ + chunk.range_start, chunk.range_end, row.node_id, row.network, + ]); + await client.query('COMMIT'); + console.log(`[privacy-remat] chunk complete id=${row.id} target=${row.target} start=${new Date(chunk.range_start).toISOString()}`); + observeWorkerOutcome('privacy_remat', 'chunk', 'success'); + return; + } catch (error) { + await client.query('ROLLBACK').catch(() => {}); + if (attempt === 3) throw error; + observeWorkerOutcome('privacy_remat', 'chunk', 'retry'); + await delay(attempt * 1_000); + } finally { + client.release(); + } + } +} + +async function complete(pool: pg.Pool, row: QueueRow): Promise { + const client = await pool.connect(); + try { + await client.query('BEGIN'); + const completion = await client.query<{ status: string }>(` + UPDATE privacy_rematerialization_queue + SET status = CASE WHEN requested_at <= started_at THEN 'done' ELSE 'pending' END, + finished_at = CASE WHEN requested_at <= started_at THEN NOW() ELSE NULL END, + last_error = NULL + WHERE id = $1 AND status = 'processing' + RETURNING status + `, [row.id]); + if (completion.rowCount !== 1) throw new Error('PRIVACY_REMAT_CLAIM_LOST'); + + // A re-request that arrived during this pass still needs a complete pass; + // only a fully completed request may claim a new materialized generation. + if (completion.rows[0]?.status === 'done') await client.query(` + WITH next_generation AS ( + UPDATE public_visibility_state + SET generation = generation + 1, + updated_at = NOW() + WHERE singleton = TRUE + RETURNING generation + ) + INSERT INTO packet_visibility_materialization_state + (singleton, visibility_generation, updated_at) + SELECT TRUE, generation, NOW() FROM next_generation + ON CONFLICT (singleton) DO UPDATE SET + visibility_generation = EXCLUDED.visibility_generation, + updated_at = EXCLUDED.updated_at + `); + await client.query('COMMIT'); + console.log(`[privacy-remat] pass complete id=${row.id} status=${completion.rows[0]?.status}`); + observeWorkerOutcome('privacy_remat', 'pass', 'success'); + } catch (error) { + await client.query('ROLLBACK').catch(() => {}); + throw error; + } finally { + client.release(); + } +} + +export async function processNextPrivacyRematerialization( + pool: pg.Pool, + options: { maxAttempts?: number; chunkPauseMs?: number; chunkStatementTimeoutMs?: number } = {}, +): Promise { + const row = await claim(pool, options.maxAttempts ?? MAX_ATTEMPTS); + if (!row) return false; + console.log(`[privacy-remat] pass start id=${row.id} target=${row.target} attempt=${row.attempts}`); + try { + const chunks = await pool.query(` + SELECT range_start, range_end + FROM timescaledb_information.chunks + WHERE hypertable_schema = 'public' AND hypertable_name = $1 + ORDER BY range_start + `, [row.target]); + for (const [index, chunk] of chunks.rows.entries()) { + await processChunk(pool, row, chunk, options.chunkStatementTimeoutMs ?? CHUNK_STATEMENT_TIMEOUT_MS); + if (index < chunks.rows.length - 1) await delay(options.chunkPauseMs ?? CHUNK_PAUSE_MS); + } + await complete(pool, row); + } catch (error) { + const message = error instanceof Error ? error.message : String(error); + await pool.query(` + UPDATE privacy_rematerialization_queue + SET status = 'failed', last_error = $2, finished_at = NOW() + WHERE id = $1 AND status = 'processing' + `, [row.id, message.slice(0, 2_000)]); + observeWorkerOutcome('privacy_remat', 'pass', 'failure'); + console.error(`[privacy-remat] pass failed id=${row.id} target=${row.target}: ${message}`); + } + return true; +} + +async function main(): Promise { + startWorkerMetrics(); + const pool = new Pool({ + connectionString: process.env.DATABASE_URL, + application_name: process.env['DATABASE_APPLICATION_NAME'] ?? 'meshcore-privacy-remat', + max: 2, + connectionTimeoutMillis: 5_000, + idleTimeoutMillis: 30_000, + }); + pool.on('error', (error) => console.error('[privacy-remat] pool error:', error.message)); + const shutdown = new AbortController(); + process.once('SIGTERM', () => shutdown.abort()); + process.once('SIGINT', () => shutdown.abort()); + try { + while (!shutdown.signal.aborted) { + try { + await processNextPrivacyRematerialization(pool); + } catch (error) { + console.error('[privacy-remat] iteration failed:', error); + } + if (!shutdown.signal.aborted) { + try { await delay(INTERVAL_MS, undefined, { signal: shutdown.signal }); } + catch (error) { + if (!shutdown.signal.aborted) throw error; + } + } + } + } finally { + await pool.end(); + } +} + +if (process.argv[1] && import.meta.url === pathToFileURL(process.argv[1]).href) { + main().catch((error) => { + console.error('[privacy-remat] fatal startup error:', error); + process.exitCode = 1; + }); +} diff --git a/docker-compose.yml b/docker-compose.yml index 2b4ba60..999fb78 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -257,6 +257,7 @@ services: DATABASE_SKIP_SCHEMA_INIT: "true" STALE_MQTT_OBSERVER_CLEANUP_DAYS: ${STALE_MQTT_OBSERVER_CLEANUP_DAYS:-30} STALE_MQTT_OBSERVER_CLEANUP_INTERVAL_MS: ${STALE_MQTT_OBSERVER_CLEANUP_INTERVAL_MS:-21600000} + PRIVACY_REMAT_MAX_ATTEMPTS: ${PRIVACY_REMAT_MAX_ATTEMPTS:-8} HEALTH_VOLUME_PATHS: database=/health-volumes/database=${HEALTH_DATABASE_BUDGET_BYTES:-322122547200},redis=/health-volumes/redis=${HEALTH_REDIS_BUDGET_BYTES:-4294967296},terrain=/health-volumes/terrain=${HEALTH_TERRAIN_BUDGET_BYTES:-53687091200},logs=/health-volumes/logs=${HEALTH_LOG_BUDGET_BYTES:-21474836480} DATA_LIFECYCLE_RETENTION_ENABLED: "true" DATA_LIFECYCLE_RETENTION_TARGETS: ${DATA_LIFECYCLE_RETENTION_TARGETS:-packets,worker_health_snapshots,frontend_error_events,owner_alert_deliveries,operational_check_results,observer_region_packet_sightings,observer_region_observer_sightings,link_job_commits,ml_model_variant_packet_results,packet_decryptions,observer_registration_requests,operator_audit_events} @@ -270,6 +271,32 @@ services: - srtm_data:/health-volumes/terrain:ro - loki_data:/health-volumes/logs:ro + privacy-remat-worker: + <<: [*internal-hosts, *application-hardening] + image: ${BACKEND_IMAGE:-meshcore-analytics-backend:local} + build: + context: . + dockerfile: Dockerfile.backend + args: + SOURCE_REVISION: ${SOURCE_REVISION:-local} + restart: always + logging: *json-log-limits + mem_limit: "512m" + memswap_limit: "512m" + command: ["node", "dist/workers/privacy-rematerialization.js"] + environment: + DATABASE_URL: postgresql://${POSTGRES_USER:-meshcore}:${POSTGRES_PASSWORD}@timescaledb:5432/${POSTGRES_DB:-meshcore} + DATABASE_APPLICATION_NAME: meshcore-privacy-remat + DATABASE_SKIP_SCHEMA_INIT: "true" + DATABASE_STATEMENT_TIMEOUT_MS: 900000 + PRIVACY_REMAT_INTERVAL_MS: ${PRIVACY_REMAT_INTERVAL_MS:-120000} + PRIVACY_REMAT_MAX_ATTEMPTS: ${PRIVACY_REMAT_MAX_ATTEMPTS:-8} + PRIVACY_REMAT_CHUNK_PAUSE_MS: ${PRIVACY_REMAT_CHUNK_PAUSE_MS:-500} + NODE_ENV: production + depends_on: + db-migrate: + condition: service_completed_successfully + synthetic-monitor: <<: [*internal-hosts, *application-hardening] image: ${BACKEND_IMAGE:-meshcore-analytics-backend:local} diff --git a/scripts/replace-container.sh b/scripts/replace-container.sh index 2325278..de6586a 100755 --- a/scripts/replace-container.sh +++ b/scripts/replace-container.sh @@ -68,6 +68,7 @@ declare -A image_variables=( [db-migrate]=BACKEND_IMAGE [path-learning-worker]=BACKEND_IMAGE [health-worker]=BACKEND_IMAGE + [privacy-remat-worker]=BACKEND_IMAGE [synthetic-monitor]=BACKEND_IMAGE [link-backfill-worker]=BACKEND_IMAGE [alert-receiver]=BACKEND_IMAGE