mirror of
https://github.com/gadgethd/ukmesh.git
synced 2026-10-15 10:40:01 +00:00
fix(privacy): rematerialize packet privacy asynchronously (#107)
This commit is contained in:
gadgethd
@@ -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"
|
||||
|
||||
@@ -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 '%🚫%'/);
|
||||
});
|
||||
|
||||
|
||||
@@ -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))
|
||||
)
|
||||
)
|
||||
)
|
||||
)
|
||||
)
|
||||
)`;
|
||||
}
|
||||
|
||||
|
||||
@@ -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();
|
||||
@@ -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`,
|
||||
|
||||
@@ -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 (
|
||||
|
||||
@@ -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/);
|
||||
});
|
||||
|
||||
@@ -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/);
|
||||
});
|
||||
|
||||
@@ -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.
|
||||
|
||||
@@ -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');
|
||||
});
|
||||
|
||||
@@ -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',
|
||||
|
||||
+31
-26
@@ -738,33 +738,38 @@ async function handleMessage(topic: string, rawPayload: Buffer): Promise<void> {
|
||||
}
|
||||
|
||||
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);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -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<QueueRow | null> {
|
||||
const result = await pool.query<QueueRow>(`
|
||||
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<void> {
|
||||
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<void> {
|
||||
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<boolean> {
|
||||
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<Chunk>(`
|
||||
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<void> {
|
||||
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;
|
||||
});
|
||||
}
|
||||
@@ -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}
|
||||
|
||||
@@ -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
|
||||
|
||||
Reference in New Issue
Block a user