diff --git a/server/config.js b/server/config.js index 8bf7344..be1ae69 100644 --- a/server/config.js +++ b/server/config.js @@ -250,6 +250,23 @@ module.exports = { logCoalesceFlushMs: parseInt(process.env.LOG_COALESCE_FLUSH_MS) || 30000, lagFlushMs: parseInt(process.env.LAG_FLUSH_MS) || 10000, lagBufferMax: parseInt(process.env.LAG_BUFFER_MAX) || 2000, + + // Off-main-thread WAL checkpointer (db/wal-checkpointer). SQLite's default + // wal_autocheckpoint (1000 pages) runs a SYNCHRONOUS, fsync-heavy checkpoint inline + // on whichever write trips it — on slow storage that blocks the event loop ~600-750ms + // on a regular ~60s beat (the periodic p99 spike). We set wal_autocheckpoint=0 on the + // MAIN connection and checkpoint from a worker_threads worker instead. + // Interval: at a typical ~4MB/60s write rate the WAL grows ~1MB between runs — well under + // the old 4MB inline threshold — so each PASSIVE (and any rare escalation TRUNCATE) is + // cheap, while PASSIVE still reclaims frames promptly. 15s balances small-WAL vs worker load. + walCheckpointIntervalMs: parseInt(process.env.WAL_CHECKPOINT_INTERVAL_MS) || 15000, + // Starvation bound: PASSIVE skips frames held by active readers/writers, so under + // continuous writes it can perpetually under-checkpoint and the WAL grows unbounded. If the + // -wal file exceeds this high-water mark, the worker escalates to a (blocking) TRUNCATE. + walCheckpointHighWaterMB: parseInt(process.env.WAL_CHECKPOINT_HIGH_WATER_MB) || 16, + // ...or escalate if the WAL grew across this many consecutive PASSIVE runs (PASSIVE not + // keeping up even below the high-water). Belt-and-suspenders with the MB bound above. + walCheckpointStarvationRuns: parseInt(process.env.WAL_CHECKPOINT_STARVATION_RUNS) || 3, // #146 device_status_log write batching (lib/status-log-writer.js). Status // transitions are buffered and coalesced to the NET state per device per flush, // so a flapping device writes ~1 row/flush instead of a row per transition — diff --git a/server/db/wal-checkpointer-worker.js b/server/db/wal-checkpointer-worker.js new file mode 100644 index 0000000..a5aad68 --- /dev/null +++ b/server/db/wal-checkpointer-worker.js @@ -0,0 +1,66 @@ +// WAL checkpointer WORKER (worker_threads). Runs OFF the main event-loop thread so the +// synchronous, fsync-heavy checkpoint never blocks the loop (the ~60s p99 spike). +// +// CRITICAL: this worker opens its OWN better-sqlite3 Database() handle against the same +// file. better-sqlite3 handles are NOT thread-safe, so the main thread's handle is never +// shared into the worker — only the dbPath STRING is passed via workerData. SQLite WAL is +// designed for multiple connections to the same file, so a second connection checkpointing +// while the main connection writes is safe. +const { workerData, parentPort } = require('worker_threads'); +const fs = require('fs'); +const Database = require('better-sqlite3'); + +const { dbPath, intervalMs, highWaterBytes, starvationRuns } = workerData; + +// Fresh, worker-owned connection (NOT the main handle). +const db = new Database(dbPath); +db.pragma('busy_timeout = 5000'); // wait (on THIS worker thread) through the main writer's brief locks +db.pragma('wal_autocheckpoint = 0'); // this connection must never auto-checkpoint either + +const walFile = dbPath + '-wal'; +function walBytes() { try { return fs.statSync(walFile).size; } catch { return 0; } } + +let lastBytes = 0; +let growthRuns = 0; // consecutive PASSIVE runs where the WAL failed to shrink +let timer = null; + +function tick() { + try { + // PASSIVE never blocks writers, but skips frames pinned by active readers/writers — + // so on its own it can perpetually under-checkpoint. That's what the guard below bounds. + db.pragma('wal_checkpoint(PASSIVE)', { simple: false }); + const bytes = walBytes(); + + // --- STARVATION BOUND (this is where "WAL cannot grow forever" is enforced) --- + // Either signal forces a TRUNCATE, which BLOCKS until it has checkpointed everything + // and truncated the file to 0. Blocking is fatal on the loop but FINE here on the worker. + if (bytes > lastBytes) growthRuns++; else growthRuns = 0; + const overHighWater = bytes > highWaterBytes; + const starved = growthRuns >= starvationRuns; + + if (overHighWater || starved) { + db.pragma('wal_checkpoint(TRUNCATE)', { simple: false }); + const after = walBytes(); + post(`escalated TRUNCATE (${overHighWater ? 'high-water' : 'starvation'}): WAL ${(bytes / 1e6).toFixed(1)}MB -> ${(after / 1e6).toFixed(1)}MB`); + growthRuns = 0; + lastBytes = after; + } else { + lastBytes = bytes; + } + } catch (e) { + post('checkpoint error: ' + (e && e.message)); + } +} + +function post(log) { try { parentPort && parentPort.postMessage({ log }); } catch (_) {} } + +timer = setInterval(tick, intervalMs); + +// Clean shutdown: stop the timer, close our connection, exit THIS worker thread. +parentPort && parentPort.on('message', (m) => { + if (m && m.stop) { + if (timer) { clearInterval(timer); timer = null; } + try { db.close(); } catch (_) {} + process.exit(0); + } +}); diff --git a/server/db/wal-checkpointer.js b/server/db/wal-checkpointer.js new file mode 100644 index 0000000..7468ce0 --- /dev/null +++ b/server/db/wal-checkpointer.js @@ -0,0 +1,56 @@ +// Off-main-thread WAL checkpointer — main-thread controller. +// +// SQLite's default auto-checkpoint runs a synchronous, fsync-heavy checkpoint inline on the +// write that trips the 1000-page threshold; on slow storage that blocks the event loop for +// ~600-750ms on a ~60s beat (the periodic p99 spike). Here we disable inline auto-checkpoint +// on the MAIN connection and delegate checkpointing to a worker_threads worker that opens its +// OWN connection (see wal-checkpointer-worker.js) so the fsync blocks the worker, not the loop. +const path = require('path'); +const { Worker } = require('worker_threads'); +const config = require('../config'); + +let worker = null; + +// Call ONCE at boot, after the DB is open + migrated. `db` is the main connection (used only +// to flip the pragma + do a one-time handoff checkpoint on the main thread at boot). `dbPath` +// is the STRING the worker uses to open its own handle — the main handle is never shared. +function startWalCheckpointer(db, dbPath) { + if (worker) return worker; + + // From now on the main thread NEVER inline-checkpoints (removes the loop-blocking fsync). + db.pragma('wal_autocheckpoint = 0'); + // Hand the worker a clean WAL (one-time, at boot, on a small WAL — cheap). Explicit + // checkpoints are independent of wal_autocheckpoint, so this still works with it at 0. + try { db.pragma('wal_checkpoint(TRUNCATE)'); } catch (_) { /* best-effort */ } + + worker = new Worker(path.join(__dirname, 'wal-checkpointer-worker.js'), { + workerData: { + dbPath, // string only (thread-safe handoff) + intervalMs: config.walCheckpointIntervalMs, + highWaterBytes: config.walCheckpointHighWaterMB * 1024 * 1024, + starvationRuns: config.walCheckpointStarvationRuns, + }, + }); + worker.on('message', (m) => { if (m && m.log) console.log('[wal-checkpoint] ' + m.log); }); + worker.on('error', (e) => console.error('[wal-checkpoint] worker error:', e && e.message)); + worker.on('exit', (code) => { if (code !== 0) console.warn(`[wal-checkpoint] worker exited (code ${code})`); worker = null; }); + // A worker thread cannot outlive its process, but unref() also ensures it never KEEPS the + // process alive during shutdown — so there's no orphaned worker/connection either way. + worker.unref(); + + console.log(`[wal-checkpoint] off-thread checkpointer started (every ${config.walCheckpointIntervalMs}ms; escalate >${config.walCheckpointHighWaterMB}MB or ${config.walCheckpointStarvationRuns} growing runs)`); + return worker; +} + +// Graceful teardown: ask the worker to stop (clears its timer + closes its connection), then +// force-terminate as a backstop. Safe to call when not started. +async function stopWalCheckpointer() { + if (!worker) return; + const w = worker; + worker = null; + try { w.postMessage({ stop: true }); } catch (_) {} + await new Promise((r) => setTimeout(r, 150)); // let it close its handle cleanly + try { await w.terminate(); } catch (_) {} +} + +module.exports = { startWalCheckpointer, stopWalCheckpointer }; diff --git a/server/server.js b/server/server.js index 5a43764..d186afb 100644 --- a/server/server.js +++ b/server/server.js @@ -689,6 +689,26 @@ startActivationNudge(); const { startAgencyDigest } = require('./services/agency-digest'); startAgencyDigest(); +// Off-main-thread WAL checkpointer: disables inline auto-checkpoint on the main connection +// (the ~60s p99 spike = a synchronous fsync-heavy checkpoint on the loop) and runs PASSIVE +// (escalating to TRUNCATE if starved) from a worker thread. Started AFTER the DB is open+migrated. +const { startWalCheckpointer, stopWalCheckpointer } = require('./db/wal-checkpointer'); +startWalCheckpointer(require('./db/database').db, config.dbPath); + +// Graceful shutdown: stop the checkpointer worker (closes its own DB handle) + flush + close. +let _shuttingDown = false; +function gracefulShutdown(sig) { + if (_shuttingDown) return; _shuttingDown = true; + console.log(`[shutdown] ${sig} — stopping WAL checkpointer + closing DB`); + Promise.resolve(stopWalCheckpointer()).catch(() => {}).finally(() => { + try { require('./lib/status-log-writer').flush(); } catch (_) {} + try { require('./db/database').db.close(); } catch (_) {} + process.exit(0); + }); +} +process.on('SIGTERM', () => gracefulShutdown('SIGTERM')); +process.on('SIGINT', () => gracefulShutdown('SIGINT')); + // Handle provisioning via WebSocket notification const { db } = require('./db/database'); const originalProvisionRoute = require('./routes/provisioning');