From f6fd15068050736fc23a2999a71850adf9593877 Mon Sep 17 00:00:00 2001 From: rukh-debug Date: Thu, 16 Apr 2026 20:31:09 -0400 Subject: [PATCH 1/5] docs: add CPU optimization design for log ingestion pipeline --- .../2026-04-16-cpu-optimization-design.md | 198 ++++++++++++++++++ 1 file changed, 198 insertions(+) create mode 100644 thoughts/shared/designs/2026-04-16-cpu-optimization-design.md diff --git a/thoughts/shared/designs/2026-04-16-cpu-optimization-design.md b/thoughts/shared/designs/2026-04-16-cpu-optimization-design.md new file mode 100644 index 0000000..30a48e0 --- /dev/null +++ b/thoughts/shared/designs/2026-04-16-cpu-optimization-design.md @@ -0,0 +1,198 @@ +--- +date: 2026-04-16 +topic: "CPU Optimization for Log Ingestion Pipeline" +status: validated +--- + +# CPU Optimization Design + +## Problem Statement + +The NDNS Analytics dashboard experiences high CPU usage from the log ingestion pipeline. The root cause is two-fold: + +1. **SSE stream processes logs one-by-one** through the full processing pipeline (hash, dedup, device upsert, profile state update) — for high-traffic profiles this means hundreds of DB operations per minute +2. **Poller + SSE run simultaneously**, duplicating work for each profile +3. **Per-log overhead is excessive**: JSON.stringify + SHA256 for every single log, individual device upserts, profile state read-then-write on every event + +## Constraints + +- Must not break existing log ingestion correctness (no lost or duplicate logs) +- Must not require database schema changes +- Must maintain real-time feel for the dashboard (sub-5s log appearance is acceptable) +- Must work with both SSE and Poller ingestion paths +- Existing webhook and notification system must continue to function + +## Approach + +Three-phase progressive optimization. Phase 1 has the highest impact and lowest risk. + +### Phase 1: SSE Batching (Highest Impact) + +Buffer SSE messages and flush as batches instead of processing one-by-one. + +- SSE messages accumulate in a time-bounded buffer +- Flush conditions: 2 seconds elapsed OR 50 logs accumulated +- Flush calls the existing `processLogBatch()` function (already designed for batch processing) +- Expected reduction: 10-50x fewer DB operations for SSE-sourced logs + +### Phase 2: Hashing & Dedup Optimization + +Replace expensive hash computation and add in-memory dedup cache. + +- **Hash optimization**: Replace `JSON.stringify + SHA256` with concatenation of stable fields (timestamp + domain + device + action + profileId) + lighter hash (or even just the concatenated string as key) +- **In-memory LRU cache**: Maintain a Set of recently-seen hashes per profile (last 10,000). Skip DB dedup query for cache hits. Only query DB on cache miss. +- Expected reduction: Near-zero DB dedup queries for active streams, 5-10x faster hash computation + +### Phase 3: Device Upsert Batching & Profile State Throttling + +- **Device upsert batching**: Accumulate device upsert operations and flush as a single batch DB query per batch cycle +- **Profile state throttling**: Update profile state (last seen, counts) on a timer (every 10s) instead of per-log +- **Poller role reduction**: Reduce poller to fallback role (5-minute interval) since SSE provides real-time data +- Expected reduction: Device upserts go from N-per-batch to 1-per-batch; profile updates go from N-per-minute to 6-per-minute + +## Architecture + +### Before (Current) + +``` +SSE Stream ──► processLogBatch([singleLog]) ──► per-log: + ├── JSON.stringify + SHA256 + ├── DB dedup query + ├── Individual device upsert + ├── Profile state update + └── Webhook evaluation + +Poller (30s) ──► processLogBatch(logs[]) ──► (same pipeline but batched) +``` + +### After (Optimized) + +``` +SSE Stream ──► LogBuffer ──► flush (2s or 50) ──► processLogBatch(batch) + ├── Field-concat hash + ├── In-memory dedup check → DB only on miss + ├── Batch device upsert + └── Throttled profile state + +Poller (5min) ──► processLogBatch(logs[]) ──► (same optimized pipeline) +``` + +## Components + +### LogBuffer (New) + +- Manages per-profile log accumulation +- Two flush triggers: time-based (2s) and size-based (50 logs) +- Flushes call `processLogBatch()` with accumulated logs +- Thread-safe via profile lock (reuses existing locking pattern) + +### DedupCache (New) + +- In-memory LRU cache storing recent event hashes per profile +- Default capacity: 10,000 entries per profile +- Lookup: O(1) Set membership check +- Eviction: simple size-based (drop oldest when full) +- On profile change/restart: cache is empty, falls back to DB (safe) + +### Modified: SSEStreamer + +- Replace immediate `processLogBatch([log])` with `logBuffer.add(log)` +- Buffer handles batching and flush + +### Modified: buildEventHash() + +- Replace `JSON.stringify(fullLog) + createHash('sha256')` with concatenation of stable fields +- Fields: `${profileId}|${timestamp}|${domain}|${device.id}|${action}|${server}` +- Hash with a lighter algorithm or use the concatenated string directly as key + +### Modified: processLogBatch() dedup section + +- Check in-memory cache first +- Only query DB for hashes not in cache +- Add all new hashes to cache after processing + +### Modified: Device upsert logic + +- Collect all unique devices from batch +- Single batched upsert (INSERT ... ON CONFLICT UPDATE) instead of individual queries + +### Modified: Profile state updates + +- Move to a periodic timer (every 10s per profile) +- Batch writes aggregated stats instead of per-log updates + +### Modified: LogPoller + +- Increase interval from 30s to 5 minutes (300s) +- Role changes from primary to fallback/safety-net +- Still processes through same optimized batch pipeline + +## Data Flow + +### Optimized SSE Flow + +1. SSE EventSource receives message → parse to log object +2. Add to LogBuffer for this profile +3. On flush trigger (time or size): + a. Build hashes for all logs using stable-field concatenation + b. Check DedupCache for all hashes → split into cache-hits (skip) and cache-misses + c. For cache-misses: query DB for existing hashes + d. Filter truly new logs + e. Batch insert new logs to DB + f. Batch upsert devices + g. Add new hashes to DedupCache + h. Update profile state (throttled, may not fire every batch) + i. Evaluate and fire webhooks for new logs + +### Poller Flow (Unchanged Pattern, Optimized Internally) + +1. Timer fires every 5 minutes +2. Fetch logs from NextDNS API +3. Process through same optimized batch pipeline +4. Serves as safety net for any logs missed by SSE + +## Error Handling + +- **SSE buffer overflow**: If buffer grows beyond 500 logs without flushing, force-flush immediately +- **DedupCache miss on restart**: Empty cache means DB queries on first batch after restart — same as today, no degradation +- **Hash algorithm change**: New hash format will not match old hashes. After deployment, first batch per profile will not dedup against pre-deployment logs. Acceptable: means ~1 batch of potential duplicates, which retention cleanup handles +- **DB errors during batch flush**: Existing error handling applies — log error, retry on next cycle +- **Profile lock contention**: Buffer flush and poller both acquire profile lock. Lock is already promise-chained, so they serialize naturally + +## Testing Strategy + +### Unit Tests + +- LogBuffer: flush triggers (time, size, overflow), empty buffer handling +- DedupCache: hit/miss/eviction behavior, per-profile isolation +- buildEventHash: verify new hash produces consistent results for same input + +### Integration Tests + +- SSE → Buffer → processLogBatch → DB: verify no log loss +- Simultaneous SSE + Poller: verify no duplicates +- High-volume test: simulate 1000 logs/min through SSE, verify batch processing + +### Performance Validation + +- CPU profiling before/after during sustained SSE traffic +- DB query count comparison (should drop 10-50x) +- Memory usage monitoring for DedupCache (should be bounded) + +## Open Questions + +1. **Profile count**: How many profiles are typically active? Affects DedupCache memory sizing. +2. **SSE vs Poller usage**: Is SSE actually enabled in production? If only poller is used, Phase 1 priority shifts. +3. **Webhook latency tolerance**: The 2s buffer window adds delay to webhook triggers. Is this acceptable? +4. **Hash backward compatibility**: Should we maintain old hash format alongside new, or accept the one-time dedup gap? + +## Estimated Impact + +| Metric | Before | After (Phase 1) | After (All Phases) | +|--------|--------|-----------------|-------------------| +| DB ops per log (SSE) | ~4-5 | ~0.1-0.2 | ~0.05-0.1 | +| Hash computations/sec | 1 per log | 1 per log | 1 per log (cheaper) | +| Dedup DB queries | 1 per batch | 1 per batch | ~0.01 per batch (cache hits) | +| Device upserts | 1 per log | 1 per batch | 1 per batch | +| Profile state updates | 1 per log | 1 per batch | ~0.17 per second (6/min) | +| CPU from ingestion | HIGH | MEDIUM | LOW | From 7507e402b03de5e2529d3c4a3be5d9d4fa642e21 Mon Sep 17 00:00:00 2001 From: rukh-debug Date: Thu, 16 Apr 2026 20:35:08 -0400 Subject: [PATCH 2/5] =?UTF-8?q?perf:=20SSE=20log=20batching=20=E2=80=94=20?= =?UTF-8?q?buffer=20logs=20and=20flush=20in=20batches?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Phase 1 of CPU optimization. SSE stream previously called processLogBatch for every single incoming message, causing massive DB overhead. Now: - New LogBuffer class accumulates SSE messages - Flushes every 2 seconds or at 50 logs (whichever first) - Force-flush at 500 logs (overflow protection) - Profile state updates throttled to 10s intervals - Removes per-log DB read-then-write for profile state - Expected 10-50x reduction in DB ops for SSE-sourced logs --- src/lib/ingestion/log-buffer.ts | 79 +++++++++++++++++++++++++++++++ src/lib/ingestion/sse-streamer.ts | 77 ++++++++++++++++++++++++------ 2 files changed, 141 insertions(+), 15 deletions(-) create mode 100644 src/lib/ingestion/log-buffer.ts diff --git a/src/lib/ingestion/log-buffer.ts b/src/lib/ingestion/log-buffer.ts new file mode 100644 index 0000000..fae1dcb --- /dev/null +++ b/src/lib/ingestion/log-buffer.ts @@ -0,0 +1,79 @@ +import { processLogBatch } from "./log-processor"; +import type { NextDNSLog } from "@/types/nextdns"; +import { createLogger } from "@/lib/logger"; + +const log = createLogger("log-buffer"); + +const DEFAULT_FLUSH_INTERVAL_MS = 2000; +const DEFAULT_MAX_BUFFER_SIZE = 50; +const OVERFLOW_LIMIT = 500; + +export class LogBuffer { + private profileId: string; + private buffer: NextDNSLog[] = []; + private flushTimer: ReturnType | null = null; + private flushing = false; + private flushIntervalMs: number; + private maxBufferSize: number; + + constructor( + profileId: string, + flushIntervalMs = DEFAULT_FLUSH_INTERVAL_MS, + maxBufferSize = DEFAULT_MAX_BUFFER_SIZE + ) { + this.profileId = profileId; + this.flushIntervalMs = flushIntervalMs; + this.maxBufferSize = maxBufferSize; + } + + add(entry: NextDNSLog) { + this.buffer.push(entry); + + if (this.buffer.length >= this.maxBufferSize || this.buffer.length >= OVERFLOW_LIMIT) { + this.flush(); + return; + } + + if (!this.flushTimer) { + this.flushTimer = setTimeout(() => this.flush(), this.flushIntervalMs); + } + } + + flush() { + if (this.flushTimer) { + clearTimeout(this.flushTimer); + this.flushTimer = null; + } + + if (this.buffer.length === 0 || this.flushing) { + return; + } + + const batch = this.buffer.splice(0); + this.flushing = true; + + processLogBatch(this.profileId, batch) + .then((result) => { + if (result.inserted > 0) { + log.debug( + { profileId: this.profileId, attempted: result.attempted, inserted: result.inserted }, + "Buffer flush complete" + ); + } + }) + .catch((error) => { + log.error({ err: error, profileId: this.profileId, batchSize: batch.length }, "Buffer flush error"); + }) + .finally(() => { + this.flushing = false; + }); + } + + destroy() { + if (this.flushTimer) { + clearTimeout(this.flushTimer); + this.flushTimer = null; + } + this.flush(); + } +} diff --git a/src/lib/ingestion/sse-streamer.ts b/src/lib/ingestion/sse-streamer.ts index b430b1f..18112f4 100644 --- a/src/lib/ingestion/sse-streamer.ts +++ b/src/lib/ingestion/sse-streamer.ts @@ -1,10 +1,10 @@ import { getDb } from "@/lib/db"; import { profiles } from "@/lib/db/schema"; import { eq } from "drizzle-orm"; -import { processLogBatch } from "./log-processor"; import type { NextDNSLog } from "@/types/nextdns"; import { EventSource } from "eventsource"; import { createLogger } from "@/lib/logger"; +import { LogBuffer } from "./log-buffer"; const log = createLogger("sse"); @@ -16,6 +16,8 @@ function maxIso(left: string | null | undefined, right: string) { return left >= right ? left : right; } +const PROFILE_STATE_THROTTLE_MS = 10_000; + export class SSEStreamer { private profileId: string; private apiKey: string; @@ -23,10 +25,15 @@ export class SSEStreamer { private backoffMs = 5000; private maxBackoff = 120000; private shouldReconnect = true; + private logBuffer: LogBuffer; + private lastProfileUpdateAt = 0; + private latestStreamId: string | null = null; + private latestTimestamp: string | null = null; constructor(profileId: string, apiKey: string) { this.profileId = profileId; this.apiKey = apiKey; + this.logBuffer = new LogBuffer(profileId); } start() { @@ -41,9 +48,45 @@ export class SSEStreamer { this.eventSource.close(); this.eventSource = null; } + this.logBuffer.destroy(); log.info({ profileId: this.profileId }, "Stopped SSE stream"); } + private async maybeUpdateProfileState() { + const now = Date.now(); + if (now - this.lastProfileUpdateAt < PROFILE_STATE_THROTTLE_MS) { + return; + } + this.lastProfileUpdateAt = now; + + if (!this.latestTimestamp && !this.latestStreamId) { + return; + } + + try { + const db = getDb(); + const currentRows = await db.select().from(profiles).where(eq(profiles.id, this.profileId)); + const current = currentRows[0] ?? null; + if (!current) return; + + await db.update(profiles) + .set({ + lastIngestedAt: this.latestTimestamp + ? maxIso(current.lastIngestedAt, this.latestTimestamp) + : current.lastIngestedAt, + lastStreamId: this.latestStreamId || current.lastStreamId, + lastSuccessfulStreamAt: new Date().toISOString(), + updatedAt: new Date().toISOString(), + }) + .where(eq(profiles.id, this.profileId)); + + this.latestTimestamp = null; + this.latestStreamId = null; + } catch (error) { + log.error({ err: error, profileId: this.profileId }, "Profile state update error"); + } + } + private connect() { if (!this.shouldReconnect) return; @@ -76,23 +119,16 @@ export class SSEStreamer { this.eventSource.addEventListener("message", async (event: MessageEvent) => { try { - const log: NextDNSLog = JSON.parse(event.data); - await processLogBatch(this.profileId, [log]); + const parsed: NextDNSLog = JSON.parse(event.data); - const currentRows = await db.select().from(profiles).where(eq(profiles.id, this.profileId)); - const current = currentRows[0] ?? null; - if (!current) { - return; + // Track latest timestamp and stream ID for throttled profile updates + if (event.lastEventId) { + this.latestStreamId = event.lastEventId; } + this.latestTimestamp = parsed.timestamp; - await db.update(profiles) - .set({ - lastIngestedAt: maxIso(current.lastIngestedAt, log.timestamp), - lastStreamId: event.lastEventId || current.lastStreamId, - lastSuccessfulStreamAt: new Date().toISOString(), - updatedAt: new Date().toISOString(), - }) - .where(eq(profiles.id, this.profileId)); + // Buffer the log instead of processing one-by-one + this.logBuffer.add(parsed); this.backoffMs = 5000; } catch (error) { @@ -100,7 +136,18 @@ export class SSEStreamer { } }); + // Periodic profile state flush + const profileStateTimer = setInterval(() => { + if (!this.shouldReconnect) { + clearInterval(profileStateTimer); + return; + } + this.maybeUpdateProfileState(); + }, PROFILE_STATE_THROTTLE_MS); + this.eventSource.addEventListener("error", () => { + clearInterval(profileStateTimer); + this.maybeUpdateProfileState(); this.eventSource?.close(); this.eventSource = null; From f830bfc4f5c35930b783530663a8a1d935c098fe Mon Sep 17 00:00:00 2001 From: rukh-debug Date: Thu, 16 Apr 2026 20:37:08 -0400 Subject: [PATCH 3/5] perf: optimize hash computation and add in-memory dedup cache MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Phase 2 of CPU optimization: - buildEventHash: replace JSON.stringify(fullObject) with fast field concatenation using pipe-delimited stable fields — avoids serializing the entire log object structure - New DedupCache: in-memory LRU Set per profile (10K entries) - O(1) Set membership check before hitting DB - Only queries DB for hashes not in cache (near-zero DB dedup queries for active streams) - Auto-evicts oldest entries when over capacity - Safe: empty cache on restart falls back to DB queries --- src/lib/ingestion/dedup-cache.ts | 77 ++++++++++++++++++++++++++++++ src/lib/ingestion/log-processor.ts | 59 +++++++++++++---------- 2 files changed, 111 insertions(+), 25 deletions(-) create mode 100644 src/lib/ingestion/dedup-cache.ts diff --git a/src/lib/ingestion/dedup-cache.ts b/src/lib/ingestion/dedup-cache.ts new file mode 100644 index 0000000..f893582 --- /dev/null +++ b/src/lib/ingestion/dedup-cache.ts @@ -0,0 +1,77 @@ +import { createLogger } from "@/lib/logger"; + +const log = createLogger("dedup-cache"); + +const DEFAULT_MAX_ENTRIES = 10_000; + +export class DedupCache { + private profileCaches = new Map>(); + private maxEntries: number; + + constructor(maxEntries = DEFAULT_MAX_ENTRIES) { + this.maxEntries = maxEntries; + } + + /** Check which hashes are NOT in cache (i.e., need DB lookup) */ + filterMisses(profileId: string, hashes: string[]): string[] { + let cache = this.profileCaches.get(profileId); + if (!cache) { + cache = new Set(); + this.profileCaches.set(profileId, cache); + } + + const misses: string[] = []; + for (const hash of hashes) { + if (!cache.has(hash)) { + misses.push(hash); + } + } + return misses; + } + + /** Add hashes to the cache for a profile */ + addHashes(profileId: string, hashes: string[]) { + let cache = this.profileCaches.get(profileId); + if (!cache) { + cache = new Set(); + this.profileCaches.set(profileId, cache); + } + + for (const hash of hashes) { + cache.add(hash); + } + + // Evict oldest entries if over capacity + if (cache.size > this.maxEntries) { + const excess = cache.size - this.maxEntries; + const iterator = cache.values(); + for (let i = 0; i < excess; i++) { + const entry = iterator.next(); + if (!entry.done) { + cache.delete(entry.value); + } + } + } + } + + /** Remove a profile's cache when profile is stopped */ + evictProfile(profileId: string) { + this.profileCaches.delete(profileId); + } + + getStats() { + return { + profiles: this.profileCaches.size, + totalEntries: [...this.profileCaches.values()].reduce((sum, s) => sum + s.size, 0), + }; + } +} + +let _cache: DedupCache | null = null; + +export function getDedupCache(): DedupCache { + if (!_cache) { + _cache = new DedupCache(); + } + return _cache; +} diff --git a/src/lib/ingestion/log-processor.ts b/src/lib/ingestion/log-processor.ts index dcd48a8..7d4b2c5 100644 --- a/src/lib/ingestion/log-processor.ts +++ b/src/lib/ingestion/log-processor.ts @@ -11,6 +11,7 @@ import { summarizeTagMatches, } from "@/lib/alerts/tagging"; import { createLogger } from "@/lib/logger"; +import { getDedupCache } from "./dedup-cache"; const log = createLogger("log-processor"); @@ -47,31 +48,26 @@ function normalizeReasons(reasons: NextDNSBlockReason[] | undefined) { } function buildEventHash(profileId: string, log: NextDNSLog) { - const payload = JSON.stringify({ + // Fast field concatenation instead of JSON.stringify of the full object + const reasons = normalizeReasons(log.reasons); + const parts = [ profileId, - timestamp: log.timestamp, - domain: log.domain, - root: log.root ?? null, - tracker: log.tracker ?? null, - type: log.type ?? null, - dnssec: log.dnssec ?? null, - encrypted: log.encrypted, - protocol: log.protocol, - clientIp: log.clientIp, - client: log.client ?? null, - device: log.device - ? { - id: log.device.id, - name: log.device.name, - model: log.device.model ?? null, - localIp: log.device.localIp ?? null, - } - : null, - status: log.status, - reasons: normalizeReasons(log.reasons), - }); - - return createHash("sha256").update(payload).digest("hex"); + log.timestamp, + log.domain, + log.root ?? "", + log.tracker ?? "", + log.type ?? "", + log.dnssec ?? "", + String(log.encrypted), + log.protocol, + log.clientIp, + log.client ?? "", + log.device ? `${log.device.id}|${log.device.name}|${log.device.model ?? ""}|${log.device.localIp ?? ""}` : "", + log.status, + reasons.map((r) => `${r.id}:${r.name}`).join(","), + ]; + + return createHash("sha256").update(parts.join("|")).digest("hex"); } function chunk(items: T[], size: number) { @@ -165,7 +161,17 @@ async function processLogBatchUnsafe( const existingHashes = new Set(); const hashes = preparedLogs.map((entry) => entry.row.eventHash); - for (const hashBatch of chunk(hashes, HASH_QUERY_BATCH_SIZE)) { + // Check in-memory dedup cache first — skip DB for known hashes + const dedupCache = getDedupCache(); + const cacheMisses = dedupCache.filterMisses(profileId, hashes); + + // All hashes found in memory cache — nothing to do + if (cacheMisses.length === 0) { + return { attempted: logs.length, inserted: 0 }; + } + + // Only query DB for hashes not found in memory cache + for (const hashBatch of chunk(cacheMisses, HASH_QUERY_BATCH_SIZE)) { const rows = await db .select({ eventHash: dnsLogs.eventHash }) .from(dnsLogs) @@ -181,6 +187,9 @@ async function processLogBatchUnsafe( } } + // Add cache-miss hashes to the memory cache (they've now been checked against DB) + dedupCache.addHashes(profileId, cacheMisses); + const freshLogs = preparedLogs.filter((entry) => !existingHashes.has(entry.row.eventHash)); if (freshLogs.length === 0) { return { attempted: logs.length, inserted: 0 }; From 7982887f16a03961484f75c09b5d240e7358db62 Mon Sep 17 00:00:00 2001 From: rukh-debug Date: Thu, 16 Apr 2026 20:39:11 -0400 Subject: [PATCH 4/5] perf: batch device upserts and increase poller fallback interval Phase 3 of CPU optimization: - Device upserts: replaced per-device individual UPDATE/INSERT loop with single batch INSERT ... ON CONFLICT UPDATE using SQL expressions. Uses COALESCE(NULLIF(...)) to preserve existing non-null values and GREATEST() for lastSeenAt. Reduces N DB queries per batch to 1. - Poller interval: increased default from 30s to 300s (5 minutes). SSE stream now provides real-time data; poller serves as fallback safety-net only. Users can override via POLL_INTERVAL_SECONDS env var. --- src/lib/ingestion/ingestion-manager.ts | 4 +- src/lib/ingestion/log-processor.ts | 139 ++++++++++++++----------- 2 files changed, 79 insertions(+), 64 deletions(-) diff --git a/src/lib/ingestion/ingestion-manager.ts b/src/lib/ingestion/ingestion-manager.ts index cb2c194..1ca227b 100644 --- a/src/lib/ingestion/ingestion-manager.ts +++ b/src/lib/ingestion/ingestion-manager.ts @@ -34,7 +34,7 @@ export class IngestionManager { const db = getDb(); const allProfiles = await db.select().from(profiles); - const pollInterval = parseInt(process.env.POLL_INTERVAL_SECONDS || "30") * 1000; + const pollInterval = parseInt(process.env.POLL_INTERVAL_SECONDS || "300") * 1000; log.info({ profileCount: allProfiles.length }, "Found profiles"); @@ -69,7 +69,7 @@ export class IngestionManager { return; } - const pollInterval = parseInt(process.env.POLL_INTERVAL_SECONDS || "30") * 1000; + const pollInterval = parseInt(process.env.POLL_INTERVAL_SECONDS || "300") * 1000; await this.startProfile(profileId, pollInterval); } diff --git a/src/lib/ingestion/log-processor.ts b/src/lib/ingestion/log-processor.ts index 7d4b2c5..25fd3b0 100644 --- a/src/lib/ingestion/log-processor.ts +++ b/src/lib/ingestion/log-processor.ts @@ -1,7 +1,7 @@ import { createHash } from "node:crypto"; import { getDb } from "@/lib/db"; import { devices, dnsLogs, groups, profiles } from "@/lib/db/schema"; -import { and, eq, inArray } from "drizzle-orm"; +import { and, eq, inArray, sql } from "drizzle-orm"; import type { NextDNSBlockReason, NextDNSLog } from "@/types/nextdns"; import { fireWebhooks } from "@/lib/webhooks/trigger"; import { getNumericSetting } from "@/lib/settings"; @@ -195,17 +195,40 @@ async function processLogBatchUnsafe( return { attempted: logs.length, inserted: 0 }; } - const freshDeviceIds = [ - ...new Set(freshLogs.map((entry) => entry.log.device?.id).filter(Boolean)), - ] as string[]; - const existingDevices = freshDeviceIds.length > 0 - ? await db - .select() - .from(devices) - .where(inArray(devices.id, freshDeviceIds as [string, ...string[]])) + // Collect unique devices from fresh logs, tracking latest timestamp per device + const deviceUpdates = new Map(); + + for (const { log } of freshLogs) { + if (!log.device?.id) continue; + + const existing = deviceUpdates.get(log.device.id); + const timestamp = existing + ? maxTimestamp(existing.lastSeenAt, log.timestamp) + : log.timestamp; + + deviceUpdates.set(log.device.id, { + id: log.device.id, + name: log.device.name, + model: log.device.model || existing?.model || null, + localIp: log.device.localIp || existing?.localIp || null, + lastSeenAt: timestamp, + }); + } + + // Fetch existing devices to distinguish new vs updated + const deviceIds = [...deviceUpdates.keys()] as [string, ...string[]]; + const existingDevices = deviceIds.length > 0 + ? await db.select().from(devices).where(inArray(devices.id, deviceIds)) : []; - const existingDeviceMap = new Map(existingDevices.map((device) => [device.id, device])); - const devicePersonMap = new Map(existingDevices.map((device) => [device.id, device.groupId])); + const existingDeviceMap = new Map(existingDevices.map((d) => [d.id, d])); + const devicePersonMap = new Map(existingDevices.map((d) => [d.id, d.groupId])); + const newDevices: Array<{ id: string; name: string; @@ -213,61 +236,53 @@ async function processLogBatchUnsafe( localIp?: string | null; }> = []; - for (const { log } of freshLogs) { - if (!log.device?.id) { - continue; - } + // Batch upsert all devices in one query + if (deviceUpdates.size > 0) { + const allDeviceValues = [...deviceUpdates.values()].map((d) => ({ + id: d.id, + profileId, + name: d.name, + model: d.model, + localIp: d.localIp, + lastSeenAt: d.lastSeenAt, + })); - const existing = existingDeviceMap.get(log.device.id); - if (existing) { - await db.update(devices) - .set({ - name: log.device.name, - model: log.device.model || existing.model, - localIp: log.device.localIp || existing.localIp, - lastSeenAt: maxTimestamp(existing.lastSeenAt, log.timestamp), + // Use ON CONFLICT to batch insert new + update existing in one query + await db.insert(devices) + .values(allDeviceValues) + .onConflictDoUpdate({ + target: devices.id, + set: { + name: sql`EXCLUDED.name`, + model: sql`COALESCE(NULLIF(EXCLUDED.model, ''), devices.model)`, + localIp: sql`COALESCE(NULLIF(EXCLUDED.local_ip, ''), devices.local_ip)`, + lastSeenAt: sql`GREATEST(devices.last_seen_at, EXCLUDED.last_seen_at)`, updatedAt: new Date().toISOString(), - }) - .where(eq(devices.id, log.device.id)); - existingDeviceMap.set(log.device.id, { - ...existing, - name: log.device.name, - model: log.device.model || existing.model, - localIp: log.device.localIp || existing.localIp, - lastSeenAt: maxTimestamp(existing.lastSeenAt, log.timestamp), + }, }); - continue; - } - await db.insert(devices) - .values({ - id: log.device.id, - profileId, - name: log.device.name, - model: log.device.model, - localIp: log.device.localIp, - lastSeenAt: log.timestamp, - }); - const deviceRecord = { - id: log.device.id, - profileId, - name: log.device.name, - model: log.device.model ?? null, - localIp: log.device.localIp ?? null, - groupId: null, - personId: null, - lastSeenAt: log.timestamp, - createdAt: null, - updatedAt: null, - }; - existingDeviceMap.set(log.device.id, deviceRecord); - newDevices.push({ - id: log.device.id, - name: log.device.name, - model: log.device.model, - localIp: log.device.localIp, - }); - devicePersonMap.set(log.device.id, null); + // Identify new devices (not in existing map) for webhooks + for (const [deviceId, update] of deviceUpdates) { + if (!existingDeviceMap.has(deviceId)) { + newDevices.push({ + id: deviceId, + name: update.name, + model: update.model, + localIp: update.localIp, + }); + devicePersonMap.set(deviceId, null); + } else { + // Update local map with latest data for volume spike webhooks + const existing = existingDeviceMap.get(deviceId)!; + existingDeviceMap.set(deviceId, { + ...existing, + name: update.name, + model: update.model || existing.model, + localIp: update.localIp || existing.localIp, + lastSeenAt: update.lastSeenAt, + }); + } + } } // Batch-resolve group names for webhook enrichment From dfd0b02f69a9ad06db8bbb11b71e405c4864b080 Mon Sep 17 00:00:00 2001 From: rukh-debug Date: Thu, 16 Apr 2026 20:41:40 -0400 Subject: [PATCH 5/5] docs: add changelog for CPU optimization release --- CHANGELOG.md | 11 +++++++++++ 1 file changed, 11 insertions(+) create mode 100644 CHANGELOG.md diff --git a/CHANGELOG.md b/CHANGELOG.md new file mode 100644 index 0000000..77906c8 --- /dev/null +++ b/CHANGELOG.md @@ -0,0 +1,11 @@ +# Changelog + +## 2026-04-16 — CPU Optimization for Log Ingestion + +**SSE Batching** — SSE stream now buffers logs and flushes in batches (2s / 50 logs) instead of processing one-by-one. Profile state updates throttled to 10s intervals. + +**Hash & Dedup Cache** — Event hashing uses faster field concatenation. New in-memory dedup cache (10K entries/profile) skips DB queries for known hashes. + +**Batch Device Upserts** — Device updates consolidated into a single `INSERT ... ON CONFLICT UPDATE` per batch instead of individual queries per device. + +**Poller Interval** — Default poller interval increased from 30s to 300s. SSE is the primary source; poller is now a fallback safety-net. Override with `POLL_INTERVAL_SECONDS`.