mirror of
https://github.com/Kpa-clawbot/meshcore-analyzer.git
synced 2026-08-27 22:54:05 +00:00
perf: in-memory packet store — all reads from RAM, SQLite write-only
- PacketStore loads all packets into memory on startup (~11MB for 27K packets) - Indexed by id, hash, observer, and node pubkey for fast lookups - /api/packets, /api/packets/timestamps, /api/packets/:id all served from RAM - MQTT ingest writes to both RAM + SQLite - Configurable maxMemoryMB (default 1024MB) in config.json packetStore section - groupByHash queries computed in-memory - Packet store stats exposed in /api/perf - Expected: /api/packets goes from 77ms to <1ms
This commit is contained in:
you
@@ -43,5 +43,10 @@
|
||||
"nodeSearch": 10,
|
||||
"invalidationDebounce": 30,
|
||||
"_comment": "All values in seconds. Server uses these directly. Client fetches via /api/config/cache."
|
||||
},
|
||||
"packetStore": {
|
||||
"maxMemoryMB": 1024,
|
||||
"estimatedPacketBytes": 450,
|
||||
"_comment": "In-memory packet store. maxMemoryMB caps RAM usage. All packets loaded on startup, served from RAM."
|
||||
}
|
||||
}
|
||||
+269
@@ -0,0 +1,269 @@
|
||||
'use strict';
|
||||
|
||||
/**
|
||||
* In-memory packet store — loads all packets from SQLite on startup,
|
||||
* serves reads from RAM, writes to both RAM + SQLite.
|
||||
* Caps memory at configurable limit (default 1GB).
|
||||
*/
|
||||
class PacketStore {
|
||||
constructor(dbModule, config = {}) {
|
||||
this.dbModule = dbModule; // The full db module (has .db, .insertPacket, .getPacket)
|
||||
this.db = dbModule.db; // Raw better-sqlite3 instance for queries
|
||||
this.maxBytes = (config.maxMemoryMB || 1024) * 1024 * 1024;
|
||||
this.estPacketBytes = config.estimatedPacketBytes || 450;
|
||||
this.maxPackets = Math.floor(this.maxBytes / this.estPacketBytes);
|
||||
|
||||
// Core storage: array sorted by timestamp DESC (newest first)
|
||||
this.packets = [];
|
||||
// Indexes
|
||||
this.byId = new Map();
|
||||
this.byHash = new Map(); // hash → [packet, ...]
|
||||
this.byObserver = new Map(); // observer_id → [packet, ...]
|
||||
this.byNode = new Map(); // pubkey → [packet, ...]
|
||||
|
||||
this.loaded = false;
|
||||
this.stats = { totalLoaded: 0, evicted: 0, inserts: 0, queries: 0 };
|
||||
}
|
||||
|
||||
/** Load all packets from SQLite into memory */
|
||||
load() {
|
||||
const t0 = Date.now();
|
||||
const rows = this.db.prepare(
|
||||
'SELECT * FROM packets ORDER BY timestamp DESC'
|
||||
).all();
|
||||
|
||||
for (const row of rows) {
|
||||
if (this.packets.length >= this.maxPackets) break;
|
||||
this._index(row);
|
||||
this.packets.push(row);
|
||||
}
|
||||
|
||||
this.stats.totalLoaded = this.packets.length;
|
||||
this.loaded = true;
|
||||
const elapsed = Date.now() - t0;
|
||||
console.log(`[PacketStore] Loaded ${this.packets.length} packets in ${elapsed}ms (${Math.round(this.packets.length * this.estPacketBytes / 1024 / 1024)}MB est)`);
|
||||
return this;
|
||||
}
|
||||
|
||||
/** Index a packet into all lookup maps */
|
||||
_index(pkt) {
|
||||
this.byId.set(pkt.id, pkt);
|
||||
|
||||
if (pkt.hash) {
|
||||
if (!this.byHash.has(pkt.hash)) this.byHash.set(pkt.hash, []);
|
||||
this.byHash.get(pkt.hash).push(pkt);
|
||||
}
|
||||
|
||||
if (pkt.observer_id) {
|
||||
if (!this.byObserver.has(pkt.observer_id)) this.byObserver.set(pkt.observer_id, []);
|
||||
this.byObserver.get(pkt.observer_id).push(pkt);
|
||||
}
|
||||
|
||||
// Index by node pubkeys mentioned in decoded_json
|
||||
this._indexByNode(pkt);
|
||||
}
|
||||
|
||||
/** Extract node pubkeys/names from decoded_json and index */
|
||||
_indexByNode(pkt) {
|
||||
if (!pkt.decoded_json) return;
|
||||
try {
|
||||
const decoded = JSON.parse(pkt.decoded_json);
|
||||
const keys = new Set();
|
||||
if (decoded.pubKey) keys.add(decoded.pubKey);
|
||||
if (decoded.destPubKey) keys.add(decoded.destPubKey);
|
||||
if (decoded.srcPubKey) keys.add(decoded.srcPubKey);
|
||||
for (const k of keys) {
|
||||
if (!this.byNode.has(k)) this.byNode.set(k, []);
|
||||
this.byNode.get(k).push(pkt);
|
||||
}
|
||||
} catch {}
|
||||
}
|
||||
|
||||
/** Remove oldest packets when over memory limit */
|
||||
_evict() {
|
||||
while (this.packets.length > this.maxPackets) {
|
||||
const old = this.packets.pop();
|
||||
this.byId.delete(old.id);
|
||||
// Remove from hash index
|
||||
if (old.hash && this.byHash.has(old.hash)) {
|
||||
const arr = this.byHash.get(old.hash).filter(p => p.id !== old.id);
|
||||
if (arr.length) this.byHash.set(old.hash, arr); else this.byHash.delete(old.hash);
|
||||
}
|
||||
// Remove from observer index
|
||||
if (old.observer_id && this.byObserver.has(old.observer_id)) {
|
||||
const arr = this.byObserver.get(old.observer_id).filter(p => p.id !== old.id);
|
||||
if (arr.length) this.byObserver.set(old.observer_id, arr); else this.byObserver.delete(old.observer_id);
|
||||
}
|
||||
// Skip node index cleanup for eviction (expensive, low value)
|
||||
this.stats.evicted++;
|
||||
}
|
||||
}
|
||||
|
||||
/** Insert a new packet (to both memory and SQLite) */
|
||||
insert(packetData) {
|
||||
const id = this.dbModule.insertPacket(packetData);
|
||||
const row = this.dbModule.getPacket(id);
|
||||
if (row) {
|
||||
this.packets.unshift(row); // newest first
|
||||
this._index(row);
|
||||
this._evict();
|
||||
this.stats.inserts++;
|
||||
}
|
||||
return id;
|
||||
}
|
||||
|
||||
/** Query packets with filters — all from memory */
|
||||
query({ limit = 50, offset = 0, type, route, region, observer, hash, since, until, node, order = 'DESC' } = {}) {
|
||||
this.stats.queries++;
|
||||
let results = this.packets;
|
||||
|
||||
// Use indexes for single-key filters when possible
|
||||
if (hash && !type && !route && !region && !observer && !since && !until && !node) {
|
||||
results = this.byHash.get(hash) || [];
|
||||
} else if (observer && !type && !route && !region && !hash && !since && !until && !node) {
|
||||
results = this.byObserver.get(observer) || [];
|
||||
} else if (node && !type && !route && !region && !observer && !hash && !since && !until) {
|
||||
results = this.byNode.get(node) || [];
|
||||
} else {
|
||||
// Apply filters sequentially
|
||||
if (type !== undefined) {
|
||||
const t = Number(type);
|
||||
results = results.filter(p => p.payload_type === t);
|
||||
}
|
||||
if (route !== undefined) {
|
||||
const r = Number(route);
|
||||
results = results.filter(p => p.route_type === r);
|
||||
}
|
||||
if (observer) results = results.filter(p => p.observer_id === observer);
|
||||
if (hash) results = results.filter(p => p.hash === hash);
|
||||
if (since) results = results.filter(p => p.timestamp > since);
|
||||
if (until) results = results.filter(p => p.timestamp < until);
|
||||
if (region) {
|
||||
// Need to look up observers for this region
|
||||
const regionObservers = new Set();
|
||||
try {
|
||||
const obs = this.db.prepare('SELECT id FROM observers WHERE iata = ?').all(region);
|
||||
obs.forEach(o => regionObservers.add(o.id));
|
||||
} catch {}
|
||||
results = results.filter(p => regionObservers.has(p.observer_id));
|
||||
}
|
||||
if (node) {
|
||||
// Check indexed results first, fall back to text search
|
||||
const indexed = this.byNode.get(node);
|
||||
if (indexed) {
|
||||
const idSet = new Set(indexed.map(p => p.id));
|
||||
results = results.filter(p => idSet.has(p.id));
|
||||
} else {
|
||||
// Text search fallback (node name)
|
||||
results = results.filter(p =>
|
||||
p.decoded_json && p.decoded_json.includes(node)
|
||||
);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
const total = results.length;
|
||||
|
||||
// Sort
|
||||
if (order === 'ASC') {
|
||||
results = results.slice().sort((a, b) => {
|
||||
if (a.timestamp < b.timestamp) return -1;
|
||||
if (a.timestamp > b.timestamp) return 1;
|
||||
return 0;
|
||||
});
|
||||
}
|
||||
// Default DESC — packets array is already sorted newest-first
|
||||
|
||||
// Paginate
|
||||
const paginated = results.slice(Number(offset), Number(offset) + Number(limit));
|
||||
return { packets: paginated, total };
|
||||
}
|
||||
|
||||
/** Query with groupByHash — aggregate packets by content hash */
|
||||
queryGrouped({ limit = 50, offset = 0, type, route, region, observer, hash, since, until, node } = {}) {
|
||||
this.stats.queries++;
|
||||
|
||||
// Get filtered results first
|
||||
const { packets: filtered, total: filteredTotal } = this.query({
|
||||
limit: 999999, offset: 0, type, route, region, observer, hash, since, until, node
|
||||
});
|
||||
|
||||
// Group by hash
|
||||
const groups = new Map();
|
||||
for (const p of filtered) {
|
||||
const h = p.hash || p.id.toString();
|
||||
if (!groups.has(h)) {
|
||||
groups.set(h, {
|
||||
hash: p.hash,
|
||||
observer_count: new Set(),
|
||||
count: 0,
|
||||
latest: p.timestamp,
|
||||
observer_id: p.observer_id,
|
||||
observer_name: p.observer_name,
|
||||
path_json: p.path_json,
|
||||
payload_type: p.payload_type,
|
||||
raw_hex: p.raw_hex,
|
||||
decoded_json: p.decoded_json,
|
||||
});
|
||||
}
|
||||
const g = groups.get(h);
|
||||
g.count++;
|
||||
if (p.observer_id) g.observer_count.add(p.observer_id);
|
||||
if (p.timestamp > g.latest) {
|
||||
g.latest = p.timestamp;
|
||||
}
|
||||
// Keep longest path
|
||||
if (p.path_json && (!g.path_json || p.path_json.length > g.path_json.length)) {
|
||||
g.path_json = p.path_json;
|
||||
g.raw_hex = p.raw_hex;
|
||||
}
|
||||
}
|
||||
|
||||
// Sort by latest DESC, paginate
|
||||
const sorted = [...groups.values()]
|
||||
.map(g => ({ ...g, observer_count: g.observer_count.size }))
|
||||
.sort((a, b) => b.latest.localeCompare(a.latest));
|
||||
|
||||
const total = sorted.length;
|
||||
const paginated = sorted.slice(Number(offset), Number(offset) + Number(limit));
|
||||
return { packets: paginated, total };
|
||||
}
|
||||
|
||||
/** Get timestamps for sparkline */
|
||||
getTimestamps(since) {
|
||||
const results = [];
|
||||
for (const p of this.packets) {
|
||||
if (p.timestamp <= since) break; // sorted DESC, so we can stop early
|
||||
results.push(p.timestamp);
|
||||
}
|
||||
return results.reverse(); // return ASC
|
||||
}
|
||||
|
||||
/** Get a single packet by ID */
|
||||
getById(id) {
|
||||
return this.byId.get(id) || null;
|
||||
}
|
||||
|
||||
/** Get all siblings of a packet (same hash) */
|
||||
getSiblings(hash) {
|
||||
return this.byHash.get(hash) || [];
|
||||
}
|
||||
|
||||
/** Memory stats */
|
||||
getStats() {
|
||||
return {
|
||||
...this.stats,
|
||||
inMemory: this.packets.length,
|
||||
maxPackets: this.maxPackets,
|
||||
estimatedMB: Math.round(this.packets.length * this.estPacketBytes / 1024 / 1024),
|
||||
maxMB: Math.round(this.maxBytes / 1024 / 1024),
|
||||
indexes: {
|
||||
byHash: this.byHash.size,
|
||||
byObserver: this.byObserver.size,
|
||||
byNode: this.byNode.size,
|
||||
}
|
||||
};
|
||||
}
|
||||
}
|
||||
|
||||
module.exports = PacketStore;
|
||||
@@ -8,6 +8,7 @@ const path = require('path');
|
||||
const config = require('./config.json');
|
||||
const decoder = require('./decoder');
|
||||
const crypto = require('crypto');
|
||||
const PacketStore = require('./packet-store');
|
||||
|
||||
// Compute a content hash from raw hex: header byte + payload (skipping path hops)
|
||||
// This correctly groups retransmissions of the same packet (same content, different paths)
|
||||
@@ -26,6 +27,7 @@ function computeContentHash(rawHex) {
|
||||
} catch { return rawHex.slice(0, 16); }
|
||||
}
|
||||
const db = require('./db');
|
||||
const pktStore = new PacketStore(db, config.packetStore || {}).load();
|
||||
const channelKeys = require("./config.json").channelKeys || {};
|
||||
|
||||
// --- Cache TTL config (seconds → ms) ---
|
||||
@@ -159,6 +161,7 @@ app.get('/api/perf', (req, res) => {
|
||||
endpoints: Object.fromEntries(sorted),
|
||||
slowQueries: perfStats.slowQueries.slice(-20),
|
||||
cache: { size: cache.size, hits: cache.hits, misses: cache.misses, hitRate: cache.hits + cache.misses > 0 ? Math.round(cache.hits / (cache.hits + cache.misses) * 1000) / 10 : 0 },
|
||||
packetStore: pktStore.getStats(),
|
||||
});
|
||||
});
|
||||
|
||||
@@ -298,7 +301,7 @@ try {
|
||||
const observerId = parts[2] || null;
|
||||
const region = parts[1] || null;
|
||||
|
||||
const packetId = db.insertPacket({
|
||||
const packetId = pktStore.insert({
|
||||
raw_hex: msg.raw,
|
||||
timestamp: now,
|
||||
observer_id: observerId,
|
||||
@@ -370,7 +373,7 @@ try {
|
||||
const role = advert.role || (advert.flags?.repeater ? 'repeater' : advert.flags?.room ? 'room' : 'companion');
|
||||
db.upsertNode({ public_key: pubKey, name, role, lat, lon, last_seen: now });
|
||||
|
||||
const packetId = db.insertPacket({
|
||||
const packetId = pktStore.insert({
|
||||
raw_hex: null,
|
||||
timestamp: now,
|
||||
observer_id: 'companion',
|
||||
@@ -401,7 +404,7 @@ try {
|
||||
const senderKey = `sender-${senderName.toLowerCase().replace(/[^a-z0-9]/g, '')}`;
|
||||
db.upsertNode({ public_key: senderKey, name: senderName, role: 'companion', lat: null, lon: null, last_seen: now });
|
||||
}
|
||||
const packetId = db.insertPacket({
|
||||
const packetId = pktStore.insert({
|
||||
raw_hex: null,
|
||||
timestamp: now,
|
||||
observer_id: 'companion',
|
||||
@@ -423,7 +426,7 @@ try {
|
||||
// Handle direct messages
|
||||
if (topic.startsWith('meshcore/message/direct/')) {
|
||||
const dm = msg.payload || msg;
|
||||
const packetId = db.insertPacket({
|
||||
const packetId = pktStore.insert({
|
||||
raw_hex: null,
|
||||
timestamp: dm.timestamp || now,
|
||||
observer_id: 'companion',
|
||||
@@ -443,7 +446,7 @@ try {
|
||||
// Handle traceroute
|
||||
if (topic.startsWith('meshcore/traceroute/')) {
|
||||
const trace = msg.payload || msg;
|
||||
const packetId = db.insertPacket({
|
||||
const packetId = pktStore.insert({
|
||||
raw_hex: null,
|
||||
timestamp: now,
|
||||
observer_id: 'companion',
|
||||
@@ -489,57 +492,31 @@ app.get('/api/stats', (req, res) => {
|
||||
|
||||
app.get('/api/packets', (req, res) => {
|
||||
const { limit = 50, offset = 0, type, route, region, observer, hash, since, until, groupByHash, node } = req.query;
|
||||
|
||||
const order = req.query.order === 'asc' ? 'ASC' : 'DESC';
|
||||
|
||||
if (groupByHash === 'true') {
|
||||
let where = [];
|
||||
let params = {};
|
||||
if (type !== undefined) { where.push('payload_type = @type'); params.type = Number(type); }
|
||||
if (route !== undefined) { where.push('route_type = @route'); params.route = Number(route); }
|
||||
if (region) { where.push('observer_id IN (SELECT id FROM observers WHERE iata = @region)'); params.region = region; }
|
||||
if (observer) { where.push('observer_id = @observer'); params.observer = observer; }
|
||||
if (hash) { where.push('hash = @hash'); params.hash = hash; }
|
||||
if (since) { where.push('timestamp > @since'); params.since = since; }
|
||||
if (until) { where.push('timestamp < @until'); params.until = until; }
|
||||
if (node) { where.push("(decoded_json LIKE @nodePattern OR decoded_json LIKE @nodeNamePattern)"); params.nodePattern = `%${node}%`; const n = db.db.prepare('SELECT name FROM nodes WHERE public_key = ?').get(node); params.nodeNamePattern = n ? `%${n.name}%` : `%${node}%`; }
|
||||
const clause = where.length ? 'WHERE ' + where.join(' AND ') : '';
|
||||
const packets = db.db.prepare(`SELECT hash, COUNT(DISTINCT observer_id) as observer_count, COUNT(*) as count, MAX(timestamp) as latest, (SELECT observer_id FROM packets pObs WHERE pObs.hash = packets.hash ORDER BY pObs.timestamp ASC LIMIT 1) as observer_id, (SELECT observer_name FROM packets pOn WHERE pOn.hash = packets.hash ORDER BY pOn.timestamp ASC LIMIT 1) as observer_name, (SELECT path_json FROM packets p2 WHERE p2.hash = packets.hash ORDER BY LENGTH(path_json) DESC LIMIT 1) as path_json, (SELECT payload_type FROM packets p3 WHERE p3.hash = packets.hash ORDER BY p3.timestamp DESC LIMIT 1) as payload_type, (SELECT raw_hex FROM packets p4 WHERE p4.hash = packets.hash ORDER BY LENGTH(raw_hex) DESC LIMIT 1) as raw_hex, (SELECT decoded_json FROM packets p5 WHERE p5.hash = packets.hash ORDER BY p5.timestamp DESC LIMIT 1) as decoded_json FROM packets ${clause} GROUP BY hash ORDER BY latest DESC LIMIT @limit OFFSET @offset`).all({ ...params, limit: Number(limit), offset: Number(offset) });
|
||||
const total = db.db.prepare(`SELECT COUNT(DISTINCT hash) as count FROM packets ${clause}`).get(params).count;
|
||||
return res.json({ packets, total });
|
||||
return res.json(pktStore.queryGrouped({ limit, offset, type, route, region, observer, hash, since, until, node }));
|
||||
}
|
||||
|
||||
let where = [];
|
||||
let params = {};
|
||||
if (type !== undefined) { where.push('payload_type = @type'); params.type = Number(type); }
|
||||
if (route !== undefined) { where.push('route_type = @route'); params.route = Number(route); }
|
||||
if (region) { where.push('observer_id IN (SELECT id FROM observers WHERE iata = @region)'); params.region = region; }
|
||||
if (observer) { where.push('observer_id = @observer'); params.observer = observer; }
|
||||
if (hash) { where.push('hash = @hash'); params.hash = hash; }
|
||||
if (since) { where.push('timestamp > @since'); params.since = since; }
|
||||
if (until) { where.push('timestamp < @until'); params.until = until; }
|
||||
if (node) { where.push("(decoded_json LIKE @nodePattern OR decoded_json LIKE @nodeNamePattern)"); params.nodePattern = `%${node}%`; const nn = db.db.prepare('SELECT name FROM nodes WHERE public_key = ?').get(node); params.nodeNamePattern = nn ? `%${nn.name}%` : `%${node}%`; }
|
||||
const clause = where.length ? 'WHERE ' + where.join(' AND ') : '';
|
||||
const orderDir = req.query.order === 'asc' ? 'ASC' : 'DESC';
|
||||
const packets = db.db.prepare(`SELECT * FROM packets ${clause} ORDER BY timestamp ${orderDir} LIMIT @limit OFFSET @offset`).all({ ...params, limit: Number(limit), offset: Number(offset) });
|
||||
const total = db.db.prepare(`SELECT COUNT(*) as count FROM packets ${clause}`).get(params).count;
|
||||
res.json({ packets, total });
|
||||
res.json(pktStore.query({ limit, offset, type, route, region, observer, hash, since, until, node, order }));
|
||||
});
|
||||
|
||||
// Lightweight endpoint: just timestamps for timeline sparkline
|
||||
app.get('/api/packets/timestamps', (req, res) => {
|
||||
const { since } = req.query;
|
||||
if (!since) return res.status(400).json({ error: 'since required' });
|
||||
const rows = db.db.prepare('SELECT timestamp FROM packets WHERE timestamp > ? ORDER BY timestamp ASC').all(since);
|
||||
res.json(rows.map(r => r.timestamp));
|
||||
res.json(pktStore.getTimestamps(since));
|
||||
});
|
||||
|
||||
app.get('/api/packets/:id', (req, res) => {
|
||||
const packet = db.getPacket(Number(req.params.id));
|
||||
const packet = pktStore.getById(Number(req.params.id)) || db.getPacket(Number(req.params.id));
|
||||
if (!packet) return res.status(404).json({ error: 'Not found' });
|
||||
|
||||
// Use the sibling with the longest path (most hops) for display
|
||||
if (packet.hash) {
|
||||
const best = db.db.prepare('SELECT id, path_json, raw_hex FROM packets WHERE hash = ? ORDER BY LENGTH(path_json) DESC LIMIT 1').get(packet.hash);
|
||||
if (best && best.path_json && best.path_json.length > (packet.path_json || '').length) {
|
||||
const siblings = pktStore.getSiblings(packet.hash);
|
||||
const best = siblings.reduce((a, b) => (b.path_json || '').length > (a.path_json || '').length ? b : a, packet);
|
||||
if (best.path_json && best.path_json.length > (packet.path_json || '').length) {
|
||||
packet.path_json = best.path_json;
|
||||
packet.raw_hex = best.raw_hex;
|
||||
}
|
||||
@@ -638,7 +615,7 @@ app.post('/api/packets', (req, res) => {
|
||||
const decoded = decoder.decodePacket(hex, channelKeys);
|
||||
const now = new Date().toISOString();
|
||||
|
||||
const packetId = db.insertPacket({
|
||||
const packetId = pktStore.insert({
|
||||
raw_hex: hex.toUpperCase(),
|
||||
timestamp: now,
|
||||
observer_id: observer || null,
|
||||
|
||||
Reference in New Issue
Block a user