From e77c82aa3440e93e140675c5cd202c2e334e90ca Mon Sep 17 00:00:00 2001 From: gadgethd <111318106+gadgethd@users.noreply.github.com> Date: Fri, 4 Sep 2026 19:37:15 +0000 Subject: [PATCH] perf(health): combine repeated packet diagnostic scans --- .../packetDiagnostics.integration.test.ts | 58 ++++++++++ backend/src/health/packetDiagnostics.ts | 51 +++++++++ backend/src/health/status.ts | 100 ++---------------- backend/src/tools/benchmarkHealthQueries.ts | 63 +++++++++++ 4 files changed, 183 insertions(+), 89 deletions(-) create mode 100644 backend/src/health/packetDiagnostics.integration.test.ts create mode 100644 backend/src/health/packetDiagnostics.ts create mode 100644 backend/src/tools/benchmarkHealthQueries.ts diff --git a/backend/src/health/packetDiagnostics.integration.test.ts b/backend/src/health/packetDiagnostics.integration.test.ts new file mode 100644 index 0000000..90b45a1 --- /dev/null +++ b/backend/src/health/packetDiagnostics.integration.test.ts @@ -0,0 +1,58 @@ +import assert from 'node:assert/strict'; +import test from 'node:test'; +import pg from 'pg'; +import { INGEST_HEALTH_SQL, PATH_HASH_HEALTH_SQL } from './packetDiagnostics.js'; + +// Always shadow packets with a session-local table, and roll it back on exit. +const connectionString = process.env['TEST_DATABASE_URL']; +test('health packet diagnostics preserve mixed paths, test isolation, ties, and empty data', { + skip: !connectionString, +}, async () => { + const client = new pg.Client({ connectionString }); + await client.connect(); + try { + await client.query('BEGIN'); + await client.query(`CREATE TEMP TABLE packets ( + time timestamptz, rx_node_id text, network text, topic text, path_hashes text[] + ) ON COMMIT DROP`); + const paths = async () => (await client.query(PATH_HASH_HEALTH_SQL)).rows[0]; + const ingest = async () => (await client.query(INGEST_HEALTH_SQL)).rows[0]; + assert.deepEqual(await paths(), { + one_byte: '0', two_byte: '0', three_byte: '0', + latest_multibyte_at: null, multibyte_packets_24h: '0', + }); + assert.deepEqual(await ingest(), { + stale_nodes: '0', active_nodes: '0', max_stale_minutes: null, + stale_threshold_minutes: '15', global_last_packet_at: null, + }); + + await client.query(`INSERT INTO packets VALUES + (NOW() - INTERVAL '1 minute', 'active', 'ukmesh', 'meshcore/x', ARRAY['AA', 'BBBB', 'CCCCCC', NULL, '', 'DDD']), + (NOW() - INTERVAL '20 minutes', 'stale', NULL, 'meshcore/x', ARRAY['BB', 'BB']), + (NOW() - INTERVAL '25 hours', 'no-path', 'ukmesh', 'meshcore/x', NULL), + (NOW() - INTERVAL '24 hours', 'boundary', 'ukmesh', 'meshcore/x', ARRAY['CCCC']), + (NOW() - INTERVAL '4 days', 'old', 'ukmesh', 'meshcore/x', ARRAY['EEEE']), + (NOW() - INTERVAL '2 minutes', 'test-latest', 'ukmesh', 'meshcore/x', ARRAY[]::text[]), + (NOW() - INTERVAL '1 minute', 'test-latest', 'test', 'meshcore-test/x', ARRAY['FFFF']), + (NOW() - INTERVAL '2 minutes', 'tied', 'ukmesh', 'meshcore/x', NULL), + (NOW() - INTERVAL '2 minutes', 'tied', 'test', 'meshcore-test/x', NULL), + (NOW() - INTERVAL '3 minutes', 'public-latest', 'test', 'meshcore-test/x', NULL), + (NOW() - INTERVAL '2 minutes', 'public-latest', 'ukmesh', 'meshcore/x', ARRAY[NULL]::text[]), + (NOW() - INTERVAL '1 minute', 'test-topic', 'ukmesh', 'meshcore-test/x', ARRAY['12345678']), + (NOW() - INTERVAL '1 minute', 'null-topic', 'ukmesh', NULL, NULL), + (NOW() - INTERVAL '1 minute', '', 'ukmesh', 'meshcore/x', ARRAY[]::text[]), + (NOW() - INTERVAL '1 minute', NULL, 'ukmesh', 'meshcore/x', ARRAY[]::text[])`); + const expectedTime = (await client.query(`SELECT (NOW() - INTERVAL '1 minute')::text AS time`)).rows[0].time; + assert.deepEqual(await paths(), { + one_byte: '3', two_byte: '1', three_byte: '1', + latest_multibyte_at: expectedTime, multibyte_packets_24h: '2', + }); + assert.deepEqual(await ingest(), { + stale_nodes: '3', active_nodes: '5', max_stale_minutes: '1500', + stale_threshold_minutes: '15', global_last_packet_at: expectedTime, + }); + } finally { + await client.query('ROLLBACK').catch(() => {}); + await client.end(); + } +}); diff --git a/backend/src/health/packetDiagnostics.ts b/backend/src/health/packetDiagnostics.ts new file mode 100644 index 0000000..5b047b7 --- /dev/null +++ b/backend/src/health/packetDiagnostics.ts @@ -0,0 +1,51 @@ +/** Keep the existing three-day activity and test-observer exclusion semantics + * while deriving all per-observer timestamps in one packets scan. */ +export const INGEST_HEALTH_SQL = `WITH latest_rx AS ( + SELECT rx_node_id, + MAX(time) FILTER ( + WHERE network IS DISTINCT FROM 'test' + AND split_part(topic, '/', 1) <> 'meshcore-test' + ) AS last_packet_at, + MAX(time) AS latest_at, + MAX(time) FILTER (WHERE network = 'test') AS latest_test_at + FROM packets + WHERE time > NOW() - INTERVAL '3 days' + AND rx_node_id IS NOT NULL AND rx_node_id <> '' + GROUP BY rx_node_id +), active_rx AS ( + SELECT rx_node_id, last_packet_at + FROM latest_rx + WHERE last_packet_at > NOW() - INTERVAL '3 days' + AND latest_at IS DISTINCT FROM latest_test_at +) +SELECT + COUNT(*) FILTER (WHERE last_packet_at < NOW() - INTERVAL '15 minutes')::text AS stale_nodes, + COUNT(*)::text AS active_nodes, + MAX(CASE WHEN last_packet_at < NOW() - INTERVAL '15 minutes' + THEN FLOOR(EXTRACT(EPOCH FROM (NOW() - last_packet_at)) / 60) + ELSE NULL END)::text AS max_stale_minutes, + '15'::text AS stale_threshold_minutes, + (SELECT MAX(time)::text FROM packets + WHERE network IS DISTINCT FROM 'test' + AND split_part(topic, '/', 1) <> 'meshcore-test') AS global_last_packet_at +FROM active_rx`; + +/** Count actual hash widths, including malformed/mixed-width historical paths. + * One array traversal supplies both hop counts and packet-level diagnostics; + * do not infer these diagnostics from the advertised path_hash_size_bytes. */ +export const PATH_HASH_HEALTH_SQL = `SELECT + COALESCE(SUM(hops.one_byte), 0)::text AS one_byte, + COALESCE(SUM(hops.two_byte), 0)::text AS two_byte, + COALESCE(SUM(hops.three_byte), 0)::text AS three_byte, + MAX(p.time) FILTER (WHERE hops.multibyte)::text AS latest_multibyte_at, + COUNT(*) FILTER (WHERE hops.multibyte)::text AS multibyte_packets_24h +FROM packets p +CROSS JOIN LATERAL ( + SELECT COUNT(*) FILTER (WHERE length(h) = 2) AS one_byte, + COUNT(*) FILTER (WHERE length(h) = 4) AS two_byte, + COUNT(*) FILTER (WHERE length(h) = 6) AS three_byte, + BOOL_OR(length(h) > 2) AS multibyte + FROM unnest(p.path_hashes) AS h +) hops +WHERE p.time > NOW() - INTERVAL '24 hours' + AND p.network IS DISTINCT FROM 'test'`; diff --git a/backend/src/health/status.ts b/backend/src/health/status.ts index 9c75717..f689ab0 100644 --- a/backend/src/health/status.ts +++ b/backend/src/health/status.ts @@ -3,6 +3,7 @@ import { Redis } from 'ioredis'; import { query } from '../db/index.js'; import { getRedisConnectionOptions, getRedisUrl } from '../platform/config/redis.js'; import { configuredLifecycleTargets } from '../db/dataLifecycle.js'; +import { INGEST_HEALTH_SQL, PATH_HASH_HEALTH_SQL } from './packetDiagnostics.js'; import { loadVerifiedRestoreReceipt, REQUIRED_BACKUP_DATASETS, @@ -632,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, pathHashWidths, multibyteSummary, operationalChecks, databaseMaintenance, databaseRuntime, redisDurability] = await Promise.all([ + const [workers, history, errors1h, ingest, pathDiagnostics, operationalChecks, databaseMaintenance, databaseRuntime, redisDurability] = await Promise.all([ currentWorkerStatuses(sysStats), query<{ ts: string; @@ -657,82 +658,16 @@ export async function getWorkerHealthOverview() { stale_threshold_minutes: string; global_last_packet_at: string | null; }>( - `WITH latest_rx AS ( - -- Bounded to the active_rx window below; an unbounded scan of the - -- packets hypertable here takes 20s+ and gives identical results. - SELECT rx_node_id, MAX(time) AS last_packet_at - FROM packets - WHERE time > NOW() - INTERVAL '3 days' - AND rx_node_id IS NOT NULL - AND rx_node_id <> '' - AND network IS DISTINCT FROM 'test' - AND split_part(topic, '/', 1) <> 'meshcore-test' - GROUP BY rx_node_id - ), - test_active AS ( - SELECT rx_node_id - FROM packets - WHERE time > NOW() - INTERVAL '3 days' - AND rx_node_id IS NOT NULL AND rx_node_id <> '' - GROUP BY rx_node_id - HAVING MAX(time) = MAX(time) FILTER (WHERE network = 'test') - ), - active_rx AS ( - SELECT rx_node_id, last_packet_at - FROM latest_rx - WHERE last_packet_at > NOW() - INTERVAL '3 days' - AND rx_node_id NOT IN (SELECT rx_node_id FROM test_active) - ) - SELECT - COUNT(*) FILTER (WHERE last_packet_at < NOW() - INTERVAL '15 minutes')::text AS stale_nodes, - COUNT(*)::text AS active_nodes, - MAX( - CASE - WHEN last_packet_at < NOW() - INTERVAL '15 minutes' - THEN FLOOR(EXTRACT(EPOCH FROM (NOW() - last_packet_at)) / 60) - ELSE NULL - END - )::text AS max_stale_minutes, - '15'::text AS stale_threshold_minutes, - (SELECT MAX(time)::text - FROM packets - WHERE network IS DISTINCT FROM 'test' - AND split_part(topic, '/', 1) <> 'meshcore-test') AS global_last_packet_at - FROM active_rx`, - ), - query<{ - hash_hex_len: string; - hop_count: string; - }>( - `SELECT length(h)::text AS hash_hex_len, COUNT(*)::text AS hop_count - FROM packets p - CROSS JOIN LATERAL unnest(p.path_hashes) AS h - WHERE p.time > NOW() - INTERVAL '24 hours' - AND p.network IS DISTINCT FROM 'test' - GROUP BY 1`, + INGEST_HEALTH_SQL, ), query<{ + one_byte: string; + two_byte: string; + three_byte: string; latest_multibyte_at: string | null; multibyte_packets_24h: string; }>( - `SELECT - MAX(time) FILTER ( - WHERE EXISTS ( - SELECT 1 - FROM unnest(path_hashes) AS h - WHERE length(h) > 2 - ) - )::text AS latest_multibyte_at, - COUNT(*) FILTER ( - WHERE EXISTS ( - SELECT 1 - FROM unnest(path_hashes) AS h - WHERE length(h) > 2 - ) - )::text AS multibyte_packets_24h - FROM packets - WHERE time > NOW() - INTERVAL '24 hours' - AND network IS DISTINCT FROM 'test'`, + PATH_HASH_HEALTH_SQL, ), getOperationalChecksCached(), query<{ @@ -774,25 +709,12 @@ export async function getWorkerHealthOverview() { const activeNodes = Number(ingestRow?.active_nodes ?? 0); const maxStaleMinutes = Number(ingestRow?.max_stale_minutes ?? 0); const staleThresholdMinutes = Number(ingestRow?.stale_threshold_minutes ?? 15); - const widthToBucket: Record = { - 2: 'one_byte', - 4: 'two_byte', - 6: 'three_byte', - }; + const multibyteRow = pathDiagnostics.rows[0]; const pathHashStats = { - one_byte: 0, - two_byte: 0, - three_byte: 0, + one_byte: Number(multibyteRow?.one_byte ?? 0), + two_byte: Number(multibyteRow?.two_byte ?? 0), + three_byte: Number(multibyteRow?.three_byte ?? 0), }; - - for (const row of pathHashWidths.rows) { - const width = Number(row.hash_hex_len ?? 0); - const bucket = widthToBucket[width]; - if (!bucket) continue; - pathHashStats[bucket] += Number(row.hop_count ?? 0); - } - - const multibyteRow = multibyteSummary.rows[0]; const problems: HealthProblem[] = []; const lastPacketAt = ingestRow?.global_last_packet_at ? Date.parse(ingestRow.global_last_packet_at) : Number.NaN; const packetAgeMinutes = Number.isFinite(lastPacketAt) ? Math.floor((Date.now() - lastPacketAt) / 60_000) : null; diff --git a/backend/src/tools/benchmarkHealthQueries.ts b/backend/src/tools/benchmarkHealthQueries.ts new file mode 100644 index 0000000..286b0e3 --- /dev/null +++ b/backend/src/tools/benchmarkHealthQueries.ts @@ -0,0 +1,63 @@ +/** Read-only parity/timing comparison against a pre-refactor Git revision. + * Usage: npx tsx src/tools/benchmarkHealthQueries.ts CONTAINER BASELINE_COMMIT + * Uses the container's existing psql credentials without printing them. */ +import assert from 'node:assert/strict'; +import { execFileSync } from 'node:child_process'; +import { INGEST_HEALTH_SQL, PATH_HASH_HEALTH_SQL } from '../health/packetDiagnostics.js'; + +const [container, revision] = process.argv.slice(2); +if (!container || !revision || !/^[a-zA-Z0-9_.-]+$/.test(container) || !/^[a-f0-9]{7,40}$/i.test(revision)) { + throw new Error('Usage: benchmarkHealthQueries.ts CONTAINER BASELINE_COMMIT'); +} +const baseline = execFileSync('git', ['show', `${revision}:backend/src/health/status.ts`], { encoding: 'utf8' }); +function extract(pattern: RegExp): string { + const match = baseline.match(pattern)?.[1]; + if (!match) throw new Error('Baseline health query not found'); + return match; +} +const oldIngest = extract(/`(WITH latest_rx AS \([\s\S]*?FROM active_rx)`/); +const oldWidths = extract(/`(SELECT length\(h\)::text AS hash_hex_len,[\s\S]*?GROUP BY 1)`/); +const oldMultibyte = extract(/`(SELECT\s+MAX\(time\) FILTER \([\s\S]*?AND network IS DISTINCT FROM 'test')`/); +const oldPaths = `WITH widths AS (${oldWidths}), multibyte AS (${oldMultibyte}) + SELECT COALESCE((SELECT hop_count FROM widths WHERE hash_hex_len = '2'), '0') AS one_byte, + COALESCE((SELECT hop_count FROM widths WHERE hash_hex_len = '4'), '0') AS two_byte, + COALESCE((SELECT hop_count FROM widths WHERE hash_hex_len = '6'), '0') AS three_byte, + latest_multibyte_at, multibyte_packets_24h FROM multibyte`; +const queries = { oldIngest, newIngest: INGEST_HEALTH_SQL, oldPaths, newPaths: PATH_HASH_HEALTH_SQL }; +const sql = [ + 'BEGIN ISOLATION LEVEL REPEATABLE READ READ ONLY;', + "SET LOCAL statement_timeout = '30s';", + "SET LOCAL idle_in_transaction_session_timeout = '30s';", + ...(['ingest', 'paths'] as const).map((name) => { + const oldSql = name === 'ingest' ? oldIngest : oldPaths; + const newSql = name === 'ingest' ? INGEST_HEALTH_SQL : PATH_HASH_HEALTH_SQL; + return `WITH before AS (${oldSql}), after AS (${newSql}) + SELECT json_build_object('parity', '${name}', 'different_rows', COUNT(*)) + FROM ((SELECT * FROM before EXCEPT ALL SELECT * FROM after) + UNION ALL (SELECT * FROM after EXCEPT ALL SELECT * FROM before)) diff;`; + }), + // Alternate ordering to reduce systematic warm-cache bias. + ...Array.from({ length: 3 }, (_, run) => { + const entries = Object.entries(queries); + if (run % 2) entries.reverse(); + return entries.flatMap(([name, query]) => [ + `SELECT json_build_object('query', '${name}', 'run', ${run + 1});`, + `EXPLAIN (ANALYZE, TIMING OFF, SUMMARY ON, FORMAT JSON) ${query};`, + ]).join('\n'); + }), + 'ROLLBACK;', +].join('\n'); +const output = execFileSync('docker', [ + 'exec', '-i', container, 'sh', '-c', + 'psql -X -qAt -v ON_ERROR_STOP=1 -U "$POSTGRES_USER" -d "$POSTGRES_DB"', +], { input: sql, encoding: 'utf8', timeout: 180_000, maxBuffer: 16 * 1024 * 1024 }); +// Emit only aggregate parity and timing, never packet data or query plans. +for (const line of output.split('\n')) { + if (line.startsWith('{"parity"')) { + const result = JSON.parse(line) as { different_rows: number }; + assert.equal(result.different_rows, 0, line); + console.log(line); + } else if (line.startsWith('{"query"') || /"(?:Planning|Execution) Time":/.test(line)) { + console.log(line.trim()); + } +}