dandri/ukmesh
1
0
Fork 0
mirror of https://github.com/gadgethd/ukmesh.git synced 2026-10-20 05:00:17 +00:00
Code Issues Packages Projects Releases Wiki Activity

perf(health): combine repeated packet diagnostic scans

Browse Source
This commit is contained in:
gadgethd
2026-09-04 19:56:07 +00:00
parent 2c2b9d9de4
commit e77c82aa34
4 changed files with 183 additions and 89 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/src/health/packetDiagnostics.integration.test.ts
+58
View File
@@ -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();
}
});
backend/src/health/packetDiagnostics.ts
+51
View File
@@ -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'`;
backend/src/health/status.ts
+11 -89
View File
@@ -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<number, 'one_byte' | 'two_byte' | 'three_byte'> = {
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;
backend/src/tools/benchmarkHealthQueries.ts
+63
View File
@@ -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());
}
}