christian/epaper-display
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

perf: throttle MQTT DB writes with 2s batched flush

Browse Source 
Collect payloads in memory Map, flush to SQLite every 2 seconds
instead of on every message. Prevents DB lock and slowness with
high-volume brokers like Venus OS.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
This commit is contained in:
Christian Mueller 2026-04-06 21:03:36 +02:00
parent 9fbc5a1b45
commit 25d07a6bc4
2 changed files with 53 additions and 33 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

51
display/middleware/src/modules/mqtt/broker-manager.ts
View File

@ -162,33 +162,40 @@ export function connectBroker(id: number): Promise<mqtt.MqttClient> {
resolve(client);
});
client.on('message', (topic: string, payloadBuf: Buffer) => {
// Throttle DB writes: cache payloads in memory, flush periodically
const payloadCache = new Map<string, string>();
const flushPayloads = (): void => {
const db = getDb();
const payload = payloadBuf.toString();
for (const [t, p] of payloadCache.entries()) {
const topicRecord = db
.prepare('SELECT * FROM mqtt_topics WHERE broker_id = ? AND topic = ?')
.get(id, t) as MqttTopic | undefined;
if (!topicRecord) continue;
// Find the topic record
const topicRecord = db
.prepare(
'SELECT * FROM mqtt_topics WHERE broker_id = ? AND topic = ?'
)
.get(id, topic) as MqttTopic | undefined;
db.run('UPDATE mqtt_topics SET last_payload = ? WHERE id = ?', [p, topicRecord.id]);
if (!topicRecord) return;
const extracted = extractValue(p, topicRecord.json_path);
if (extracted === null) continue;
// Update last_payload
db.run(
'UPDATE mqtt_topics SET last_payload = ? WHERE id = ?',
[payload, topicRecord.id]
);
// Extract value and update datapoint
const extracted = extractValue(payload, topicRecord.json_path);
if (extracted === null) return;
const dp = getDatapointByTopicId(topicRecord.id);
if (dp) {
updateDatapointValue(dp.id, extracted);
const dp = getDatapointByTopicId(topicRecord.id);
if (dp) {
updateDatapointValue(dp.id, extracted);
}
}
payloadCache.clear();
};
// Flush every 2 seconds
const flushTimer = setInterval(flushPayloads, 2000);
client.on('message', (topic: string, payloadBuf: Buffer) => {
payloadCache.set(topic, payloadBuf.toString());
});
// Clean up flush timer on disconnect
client.on('close', () => {
clearInterval(flushTimer);
flushPayloads();
});
client.on('error', (err: Error) => {

35
display/middleware/src/modules/mqtt/topic-discovery.ts
View File

@ -87,24 +87,37 @@ export function startDiscovery(brokerId: number, prefix?: string): Promise<MqttT
const discoveredTopics: Set<string> = new Set();
const finalize = (): void => {
clearInterval(flushInterval);
flushToDb(); // Final flush
if (tempClient) {
tempClient.end(true);
}
resolve(getTopicsForBroker(brokerId));
};
const onMessage = (topic: string, payload: Buffer): void => {
const payloadStr = payload.toString();
if (!discoveredTopics.has(topic)) {
discoveredTopics.add(topic);
ensureTopicExists(brokerId, topic);
}
// Always update last_payload so the user sees the latest value
// Collect payloads in memory, batch-write to DB periodically
const payloadCache: Map<string, string> = new Map();
const flushToDb = (): void => {
const db = getDb();
db.run(
'UPDATE mqtt_topics SET last_payload = ? WHERE broker_id = ? AND topic = ?',
[payloadStr, brokerId, topic]
);
for (const [t, p] of payloadCache.entries()) {
if (!discoveredTopics.has(t)) {
discoveredTopics.add(t);
ensureTopicExists(brokerId, t);
}
db.run(
'UPDATE mqtt_topics SET last_payload = ? WHERE broker_id = ? AND topic = ?',
[p, brokerId, t]
);
}
payloadCache.clear();
};
// Flush every 2 seconds instead of on every message
const flushInterval = setInterval(flushToDb, 2000);
const onMessage = (topic: string, payload: Buffer): void => {
payloadCache.set(topic, payload.toString());
};
if (existingClient && existingClient.connected) {