dandri/ukmesh
1
0
Fork 0
mirror of https://github.com/gadgethd/ukmesh.git synced 2026-09-09 08:43:45 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(ws): replace per-row canonical_node_id calls with alias joins in initial-state feed queries

Browse Source 
The cold WS initial-state fetch (fetchInitialState -> getRecentPackets +
getRecentMessages) called meshcore_canonical_node_id() ~100k times per
fetch (per-row subquery against node_identity_aliases). 6.97s cold on the
2-vCPU box; under load it exceeded the 8s WS_INITIAL_STATE_DB_TIMEOUT_MS
abort, so the map never received an initial snapshot (empty map +
synthetic_check_failed). Replace the per-row function calls with
node_identity_aliases LEFT JOINs (COALESCE(alias.canonical_node_id,
upper(btrim(node_id))) - exact function semantics, table ~20 rows).
Measured: full getRecentMessages 6968ms -> 1772ms (4x); getRecentPackets
same pattern. Also raise the WS_INITIAL_STATE_DB_TIMEOUT_MS env cap
9s -> 60s (default stays 8s; deploy sets 30s) so slow-but-working fetches
complete and warm the cache instead of aborting forever.
281/281 backend tests pass; tsc clean. Live: initial_state 502ms warm,
all synthetic checks green.
This commit is contained in:
gadgethd
2026-08-09 02:43:44 +00:00
parent 8981979e22
commit 119eaf529d
2 changed files with 17 additions and 11 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/src/db/index.ts
+16 -10
View File
@@ -788,14 +788,16 @@ export async function getRecentPackets(
`WITH recent_packets AS (
SELECT DISTINCT ON (p.packet_hash)
p.time, p.packet_hash,
meshcore_canonical_node_id(p.rx_node_id) AS rx_node_id,
meshcore_canonical_node_id(p.src_node_id) AS src_node_id, p.topic,
COALESCE(pa_rx.canonical_node_id, upper(btrim(p.rx_node_id))) AS rx_node_id,
COALESCE(pa_src.canonical_node_id, upper(btrim(p.src_node_id))) AS src_node_id, p.topic,
p.topic_prefix, p.iata, p.packet_type, p.route_type, p.hop_count, p.rssi, p.snr,
COALESCE(p.payload->>'_summary', pd.summary) AS summary,
p.advert_count, p.path_hashes, p.path_hash_size_bytes,
p.network, p.transport_codes, p.region_scope
FROM packets p
LEFT JOIN packet_decryptions pd ON pd.packet_hash = p.packet_hash
LEFT JOIN node_identity_aliases pa_rx ON pa_rx.source_node_id = upper(btrim(p.rx_node_id))
LEFT JOIN node_identity_aliases pa_src ON pa_src.source_node_id = upper(btrim(p.src_node_id))
WHERE p.time > ${fiveMinAgo}
${buildPacketScopeClause(scope, 'p', network)}
${buildPublicPacketPrivacyClause('p')}
@@ -809,13 +811,14 @@ export async function getRecentPackets(
packet_stats AS (
SELECT
packet_hash,
ARRAY_AGG(DISTINCT meshcore_canonical_node_id(rx_node_id)
ORDER BY meshcore_canonical_node_id(rx_node_id))
FILTER (WHERE rx_node_id IS NOT NULL) AS observer_node_ids,
ARRAY_AGG(DISTINCT COALESCE(pa2.canonical_node_id, upper(btrim(packets.rx_node_id)))
ORDER BY COALESCE(pa2.canonical_node_id, upper(btrim(packets.rx_node_id))))
FILTER (WHERE packets.rx_node_id IS NOT NULL) AS observer_node_ids,
ARRAY_AGG(DISTINCT iata ORDER BY iata) FILTER (WHERE NULLIF(TRIM(iata), '') IS NOT NULL) AS observer_iatas,
COUNT(*) FILTER (WHERE COALESCE(payload->>'direction', 'rx') <> 'tx')::int AS rx_count,
COUNT(*) FILTER (WHERE COALESCE(payload->>'direction', 'rx') = 'tx')::int AS tx_count
FROM packets
LEFT JOIN node_identity_aliases pa2 ON pa2.source_node_id = upper(btrim(packets.rx_node_id))
WHERE packet_hash = ANY(SELECT packet_hash FROM recent_packets)
AND time > ${fiveMinAgo}
${buildPacketScopeClause(scope, '', network)}
@@ -856,8 +859,8 @@ export async function getRecentMessages(
`WITH recent_msgs AS (
SELECT DISTINCT ON (p.packet_hash)
p.time, p.packet_hash,
meshcore_canonical_node_id(p.rx_node_id) AS rx_node_id,
meshcore_canonical_node_id(p.src_node_id) AS src_node_id, p.topic,
COALESCE(pa_rx.canonical_node_id, upper(btrim(p.rx_node_id))) AS rx_node_id,
COALESCE(pa_src.canonical_node_id, upper(btrim(p.src_node_id))) AS src_node_id, p.topic,
p.iata,
p.packet_type, p.hop_count, p.rssi, p.snr, p.payload,
COALESCE(p.payload->>'_summary', pd.summary) AS summary,
@@ -865,6 +868,8 @@ export async function getRecentMessages(
p.network
FROM packets p
LEFT JOIN packet_decryptions pd ON pd.packet_hash = p.packet_hash
LEFT JOIN node_identity_aliases pa_rx ON pa_rx.source_node_id = upper(btrim(p.rx_node_id))
LEFT JOIN node_identity_aliases pa_src ON pa_src.source_node_id = upper(btrim(p.src_node_id))
WHERE p.packet_type = 5
AND p.time > NOW() - INTERVAL '24 hours'
${buildPacketScopeClause(scope, 'p', network)}
@@ -876,13 +881,14 @@ export async function getRecentMessages(
msg_stats AS (
SELECT
packet_hash,
ARRAY_AGG(DISTINCT meshcore_canonical_node_id(rx_node_id)
ORDER BY meshcore_canonical_node_id(rx_node_id))
FILTER (WHERE rx_node_id IS NOT NULL) AS observer_node_ids,
ARRAY_AGG(DISTINCT COALESCE(pa2.canonical_node_id, upper(btrim(packets.rx_node_id)))
ORDER BY COALESCE(pa2.canonical_node_id, upper(btrim(packets.rx_node_id))))
FILTER (WHERE packets.rx_node_id IS NOT NULL) AS observer_node_ids,
ARRAY_AGG(DISTINCT iata ORDER BY iata) FILTER (WHERE NULLIF(TRIM(iata), '') IS NOT NULL) AS observer_iatas,
COUNT(*) FILTER (WHERE COALESCE(payload->>'direction', 'rx') <> 'tx')::int AS rx_count,
COUNT(*) FILTER (WHERE COALESCE(payload->>'direction', 'rx') = 'tx')::int AS tx_count
FROM packets
LEFT JOIN node_identity_aliases pa2 ON pa2.source_node_id = upper(btrim(packets.rx_node_id))
WHERE packet_hash = ANY(SELECT packet_hash FROM recent_msgs)
AND time > NOW() - INTERVAL '24 hours'
${buildPacketScopeClause(scope, '', network)}
backend/src/ws/server.ts
+1 -1
View File
@@ -44,7 +44,7 @@ const WS_MAX_PENDING_HANDSHAKES = boundedEnvInteger('WS_MAX_PENDING_HANDSHAKES',
const WS_INITIAL_STATE_CONCURRENCY = boundedEnvInteger('WS_INITIAL_STATE_CONCURRENCY', 8, 1, 128);
const WS_INITIAL_STATE_QUEUE_MAX = boundedEnvInteger('WS_INITIAL_STATE_QUEUE_MAX', 64, 0, 1_000);
const WS_INITIAL_STATE_MAX_BYTES = boundedEnvInteger('WS_INITIAL_STATE_MAX_BYTES', 4 * 1_048_576, 16 * 1024, 32 * 1_048_576);
const WS_INITIAL_STATE_DB_TIMEOUT_MS = boundedEnvInteger('WS_INITIAL_STATE_DB_TIMEOUT_MS', 8_000, 1_000, 9_000);
const WS_INITIAL_STATE_DB_TIMEOUT_MS = boundedEnvInteger('WS_INITIAL_STATE_DB_TIMEOUT_MS', 8_000, 1_000, 60_000);
const WS_HEARTBEAT_INTERVAL_MS = boundedEnvInteger('WS_HEARTBEAT_INTERVAL_MS', 30_000, 5_000, 120_000);
type InitialStateRow = Record<string, unknown>;