daemon answers the dashboard again (85 GB events, duplicate tails); click to copy errors (0.13.22)

apps/cli/package.json

@@ -1,6 +1,6 @@
 {
   "name": "@profullstack/threatcrush",
-  "version": "0.13.21",
+  "version": "0.13.22",
   "description": "All-in-one security agent daemon — monitor, detect, scan, and protect servers in real-time",
   "bin": {
     "threatcrush": "./dist/index.js"

apps/cli/src/commands/monitor.ts

@@ -1,6 +1,6 @@
-import { existsSync, createReadStream, statSync, accessSync, constants } from 'node:fs';
-import { createInterface } from 'node:readline';
+import { existsSync, accessSync, constants } from 'node:fs';
 import { watch } from 'node:fs';
+import { LogTail } from '../core/log-tail.js';
 import chalk from 'chalk';
 import { formatEvent, logger, banner } from '../core/logger.js';
 import { assessNginxRequest, attackSeverity, autoDetectParser, configureAttackDetection, parseAuthLog, parseNginxLog } from '../core/log-parser.js';
@@ -110,37 +110,15 @@ export async function monitorCommand(options: MonitorOptions): Promise<void> {
 function tailLog(source: LogWatcher): void {
   const { path, name, category } = source;
 
-  // Start reading from end of file
-  const stat = statSync(path);
-  let position = stat.size;
-
+  // From the end of the file; each new line exactly once (see LogTail).
+  const tail = new LogTail(path);
+  let warned = false;
   const checkForNewData = () => {
     try {
-      const currentStat = statSync(path);
-      if (currentStat.size <= position) {
-        if (currentStat.size < position) position = 0; // file rotated
-        return;
-      }
-
-      const stream = createReadStream(path, { start: position, encoding: 'utf-8' });
-      stream.on('error', (err) => {
-        // Permission or IO error mid-read; log once and stop polling this source.
-        logger.warn(`stopped tailing ${path}: ${(err as NodeJS.ErrnoException).code || err.message}`);
-        position = currentStat.size;
-      });
-      const rl = createInterface({ input: stream });
-      rl.on('error', () => { /* surfaced via stream 'error' above */ });
-
-      rl.on('line', (line) => {
-        if (!line.trim()) return;
-        processLine(line, name, category);
-      });
-
-      rl.on('close', () => {
-        position = currentStat.size;
-      });
-    } catch {
-      // file may not be readable
+      tail.poll((line) => processLine(line, name, category));
+    } catch (err) {
+      if (!warned) logger.warn(`cannot read ${path}: ${(err as NodeJS.ErrnoException).code || (err as Error).message}`);
+      warned = true;
     }
   };
 

apps/cli/src/core/__tests__/log-tail.test.ts

@@ -0,0 +1,70 @@
+import { describe, it, expect } from 'vitest';
+import { appendFileSync, mkdtempSync, renameSync, writeFileSync } from 'node:fs';
+import { tmpdir } from 'node:os';
+import { join } from 'node:path';
+import { LogTail } from '../log-tail.js';
+
+const fresh = () => {
+  const file = join(mkdtempSync(join(tmpdir(), 'tc-tail-')), 'access.log');
+  writeFileSync(file, 'old line that predates the tail\n');
+  return file;
+};
+
+const collect = (tail: LogTail, maxBytes?: number) => {
+  const lines: string[] = [];
+  tail.poll((l) => lines.push(l), maxBytes);
+  return lines;
+};
+
+describe('LogTail', () => {
+  it('starts at the end and hands over each new line exactly once', () => {
+    const file = fresh();
+    const tail = new LogTail(file);
+    expect(collect(tail)).toEqual([]);
+    appendFileSync(file, 'a\nb\n');
+    expect(collect(tail)).toEqual(['a', 'b']);
+    // Polling again with nothing new must not replay anything: the old reader
+    // re-read from a stale offset whenever a read outlived the poll interval.
+    expect(collect(tail)).toEqual([]);
+    expect(collect(tail)).toEqual([]);
+  });
+
+  it('holds a line until its newline arrives', () => {
+    const file = fresh();
+    const tail = new LogTail(file);
+    appendFileSync(file, 'half a li');
+    expect(collect(tail)).toEqual([]);
+    appendFileSync(file, 'ne\nnext\n');
+    expect(collect(tail)).toEqual(['half a line', 'next']);
+  });
+
+  it('follows rotation: a new file (new inode) or a truncated one starts from the top', () => {
+    const file = fresh();
+    const tail = new LogTail(file);
+    appendFileSync(file, 'before\n');
+    expect(collect(tail)).toEqual(['before']);
+    renameSync(file, `${file}.1`);
+    writeFileSync(file, 'first in the new file\n');
+    expect(collect(tail)).toEqual(['first in the new file']);
+    writeFileSync(file, 'x\n'); // truncated in place
+    expect(collect(tail)).toEqual(['x']);
+  });
+
+  it('spreads a burst over polls instead of reading it all at once', () => {
+    const file = fresh();
+    const tail = new LogTail(file);
+    appendFileSync(file, Array.from({ length: 100 }, (_, i) => `line ${String(i).padStart(3, '0')}`).join('\n') + '\n');
+    const first = collect(tail, 300); // ~33 lines' worth
+    expect(first.length).toBeGreaterThan(0);
+    expect(first.length).toBeLessThan(100);
+    const all = [...first];
+    for (let i = 0; i < 10; i++) all.push(...collect(tail, 300));
+    expect(all).toHaveLength(100);
+    expect(new Set(all).size).toBe(100);
+  });
+
+  it('a missing file is not an error', () => {
+    const tail = new LogTail('/nonexistent/threatcrush/test.log');
+    expect(collect(tail)).toEqual([]);
+  });
+});

apps/cli/src/core/__tests__/state-retention.test.ts

@@ -0,0 +1,56 @@
+import { describe, it, expect, beforeAll } from 'vitest';
+import {
+  initStateDB,
+  insertEvent,
+  pruneEvents,
+  getEventCount,
+  getThreatCount,
+  getTopSources,
+} from '../state.js';
+
+const DAY = 86_400_000;
+const NOW = Date.parse('2026-10-06T12:00:00Z');
+
+const ev = (ageMs: number, ip: string, severity: 'info' | 'high' = 'info') =>
+  insertEvent({
+    timestamp: new Date(NOW - ageMs),
+    module: 'log-watcher',
+    category: 'web',
+    severity,
+    message: 'GET /',
+    source_ip: ip,
+  } as any);
+
+describe('events retention (the 85 GB state.db that starved the daemon)', () => {
+  beforeAll(() => {
+    initStateDB(':memory:');
+    for (let i = 0; i < 30; i++) ev(20 * DAY, '10.0.0.1'); // past retention
+    for (let i = 0; i < 20; i++) ev(1 * DAY, '10.0.0.2', i < 5 ? 'high' : 'info');
+    for (let i = 0; i < 10; i++) ev(60_000, '10.0.0.3');
+  });
+
+  it('counts every event ever recorded without scanning the table', () => {
+    expect(getEventCount()).toBe(60);
+  });
+
+  it('deletes events past the retention, oldest first, in batches', () => {
+    const deleted = pruneEvents({ retentionSeconds: 14 * 86_400, now: NOW, batch: 7, maxBatches: 100 });
+    expect(deleted).toBe(30);
+    expect(getTopSources(10).map((s) => s.ip)).not.toContain('10.0.0.1');
+    // "ever recorded" keeps counting what was pruned
+    expect(getEventCount()).toBe(60);
+  });
+
+  it('caps the number of rows, dropping the oldest', () => {
+    const deleted = pruneEvents({ retentionSeconds: 365 * 86_400, maxEvents: 15, now: NOW, batch: 4, maxBatches: 100 });
+    expect(deleted).toBe(15);
+    const top = getTopSources(10);
+    expect(top.find((s) => s.ip === '10.0.0.3')?.count).toBe(10);
+    expect(top.find((s) => s.ip === '10.0.0.2')?.count).toBe(5);
+  });
+
+  it('threat and recent counts look at recent events only', () => {
+    expect(getThreatCount()).toBe(0); // the 5 high events were the oldest 15 dropped
+    expect(getEventCount(new Date(NOW - 3_600_000))).toBe(10);
+  });
+});

apps/cli/src/core/config.ts

@@ -45,6 +45,7 @@ export function loadConfig(configPath?: string): ThreatCrushConfig {
       detection: parsed.detection,
       cloud: parsed.cloud,
       watchdog: parsed.watchdog,
+      storage: parsed.storage,
       license: parsed.license,
     };
   } catch {

apps/cli/src/core/log-tail.ts

@@ -0,0 +1,82 @@
+import { closeSync, fstatSync, openSync, readSync, statSync } from 'node:fs';
+
+/**
+ * Reads what was appended to a log since the last poll, exactly once.
+ *
+ * The watchers used to start a new read stream on every poll and record the
+ * new position only when that stream closed. When the daemon was busy (a slow
+ * state database), a read outlived the poll interval and the next poll started
+ * another from the same offset: the same lines were processed again and again,
+ * each copy stored as a new event, which made the database slower still, and
+ * every overlapping stream held a file descriptor (dev2: 3,259 on syslog,
+ * 1,549 on rotated-away logs). Here a poll reads synchronously, bounded to the
+ * bytes that are there now, in fixed chunks, and never overlaps itself.
+ */
+export class LogTail {
+  private pos: number;
+  private partial = '';
+  private inode: number | undefined;
+
+  /** Starts at the end of the file: history is not replayed. */
+  constructor(readonly path: string, fromStart = false) {
+    try {
+      const st = statSync(path);
+      this.pos = fromStart ? 0 : st.size;
+      this.inode = st.ino;
+    } catch {
+      this.pos = 0;
+    }
+  }
+
+  get position(): number {
+    return this.pos;
+  }
+
+  /**
+   * Hands each complete new line to onLine. A line still being written (no
+   * newline yet) waits for the next poll. Rotation (a new inode, or a file
+   * shorter than our position) restarts at the top of the new file.
+   * At most maxBytes are read per poll, so a burst is spread over polls
+   * instead of blocking the daemon.
+   */
+  poll(onLine: (line: string) => void, maxBytes = 4 << 20): number {
+    let fd: number;
+    try {
+      fd = openSync(this.path, 'r');
+    } catch {
+      return 0;
+    }
+    try {
+      const st = fstatSync(fd);
+      if ((this.inode !== undefined && st.ino !== this.inode) || st.size < this.pos) {
+        this.pos = 0;
+        this.partial = '';
+      }
+      this.inode = st.ino;
+      const end = Math.min(st.size, this.pos + maxBytes);
+      if (end <= this.pos) return 0;
+      const buf = Buffer.allocUnsafe(Math.min(end - this.pos, 1 << 20));
+      let lines = 0;
+      while (this.pos < end) {
+        const want = Math.min(buf.length, end - this.pos);
+        const n = readSync(fd, buf, 0, want, this.pos);
+        if (n <= 0) break;
+        this.pos += n;
+        const text = this.partial + buf.toString('utf-8', 0, n);
+        const parts = text.split('\n');
+        this.partial = parts.pop() ?? '';
+        for (const line of parts) {
+          if (line.trim()) {
+            lines++;
+            onLine(line);
+          }
+        }
+      }
+      // A runaway "line" (binary junk, no newline) must not grow forever.
+      if (this.partial.length > 1 << 20) this.partial = '';
+      return lines;
+    } finally {
+      closeSync(fd);
+    }
+  }
+}

apps/cli/src/core/state.ts

@@ -102,38 +102,92 @@ export function getRecentEvents(limit: number = 50): ThreatEvent[] {
   return rows.map(rowToEvent);
 }
 
+// Every query the daemon answers over IPC must cost the same on a 50 MB
+// database as on an 85 GB one: better-sqlite3 is synchronous, so a scan of the
+// whole events table freezes the daemon, and a ban sent from the dashboard
+// times out behind it. Counts and top sources therefore look at the newest
+// RECENT_WINDOW events only (newest by id, which is the primary key).
+export const RECENT_WINDOW = 200_000;
+
+/** Events ever recorded (the AUTOINCREMENT sequence): O(1), never a scan. */
+function eventsEverRecorded(database: Database.Database): number {
+  const row = database.prepare(`SELECT seq FROM sqlite_sequence WHERE name = 'events'`).get() as { seq?: number } | undefined;
+  return row?.seq ?? 0;
+}
+
 export function getEventCount(since?: Date): number {
   const database = tryDb();
   if (!database) return 0;
   if (since) {
-    return (database.prepare(`SELECT COUNT(*) as count FROM events WHERE timestamp >= ?`)
-      .get(since.toISOString()) as any).count;
+    return (database.prepare(
+      `SELECT COUNT(*) as count FROM (SELECT timestamp FROM events ORDER BY id DESC LIMIT ?) WHERE timestamp >= ?`,
+    ).get(RECENT_WINDOW, since.toISOString()) as any).count;
   }
-  return (database.prepare(`SELECT COUNT(*) as count FROM events`).get() as any).count;
+  return eventsEverRecorded(database);
 }
 
 export function getThreatCount(since?: Date): number {
   const database = tryDb();
   if (!database) return 0;
   const severities = "('medium','high','critical')";
-  if (since) {
-    return (database.prepare(
-      `SELECT COUNT(*) as count FROM events WHERE severity IN ${severities} AND timestamp >= ?`
-    ).get(since.toISOString()) as any).count;
-  }
+  const cutoff = since ? since.toISOString() : '';
   return (database.prepare(
-    `SELECT COUNT(*) as count FROM events WHERE severity IN ${severities}`
-  ).get() as any).count;
+    `SELECT COUNT(*) as count FROM (SELECT severity, timestamp FROM events ORDER BY id DESC LIMIT ?)
+     WHERE severity IN ${severities} AND timestamp >= ?`,
+  ).get(RECENT_WINDOW, cutoff) as any).count;
 }
 
 export function getTopSources(limit: number = 10): Array<{ ip: string; count: number }> {
   const database = tryDb();
   if (!database) return [];
   return database.prepare(`
-    SELECT source_ip as ip, COUNT(*) as count FROM events
+    SELECT source_ip as ip, COUNT(*) as count
+    FROM (SELECT source_ip FROM events ORDER BY id DESC LIMIT ?)
     WHERE source_ip IS NOT NULL
     GROUP BY source_ip ORDER BY count DESC LIMIT ?
-  `).all(limit) as any[];
+  `).all(RECENT_WINDOW, limit) as any[];
+}
+
+export const DEFAULT_EVENT_RETENTION_SECONDS = 14 * 86_400;
+export const DEFAULT_MAX_EVENTS = 2_000_000;
+
+/**
+ * Deletes events older than the retention, and beyond the row cap, in small
+ * batches so no single call holds the daemon for long. Returns how many went.
+ * The file does not shrink (SQLite reuses the space), which is what we want:
+ * it stops growing.
+ */
+export function pruneEvents(
+  opts: { retentionSeconds?: number; maxEvents?: number; batch?: number; maxBatches?: number; now?: number } = {},
+): number {
+  const database = tryDb();
+  if (!database) return 0;
+  const retention = opts.retentionSeconds ?? DEFAULT_EVENT_RETENTION_SECONDS;
+  const maxEvents = opts.maxEvents ?? DEFAULT_MAX_EVENTS;
+  const batch = opts.batch ?? 5_000;
+  const maxBatches = opts.maxBatches ?? 20;
+  const cutoff = new Date((opts.now ?? Date.now()) - retention * 1000).toISOString();
+  const byAge = database.prepare(
+    `DELETE FROM events WHERE id IN (SELECT id FROM events WHERE timestamp < ? ORDER BY id LIMIT ?)`,
+  );
+  const lowest = database.prepare(`SELECT MIN(id) as lo, MAX(id) as hi FROM events`);
+  const byCount = database.prepare(`DELETE FROM events WHERE id < ?`);
+  let deleted = 0;
+  for (let i = 0; i < maxBatches; i++) {
+    const n = byAge.run(cutoff, batch).changes;
+    deleted += n;
+    if (n < batch) break;
+  }
+  // Ids are dense apart from what was deleted, so hi - cap is where to cut;
+  // one batch at a time from the bottom.
+  for (let i = 0; i < maxBatches; i++) {
+    const { lo, hi } = lowest.get() as { lo: number | null; hi: number | null };
+    if (lo === null || hi === null || hi - lo + 1 <= maxEvents) break;
+    const n = byCount.run(Math.min(lo + batch, hi - maxEvents + 1)).changes;
+    deleted += n;
+    if (n === 0) break;
+  }
+  return deleted;
 }
 
 export function getModuleState(module: string, key: string): unknown {