fix(security): event-store retention + race + total-cap defects (DC-116) [glm-grade=B]
CI / Test & Lint (push) Canceled after 0s
CI / Security audit (push) Canceled after 0s

- query().total: full-scan true count (was capped at offset+limit by an
  early break — dashboard 24h stat read ≤1000 vs real ~46k)
- trim trigger/curer mismatch: byte trigger + line curer never converged
  when avg line > ~524B (live avg 355B); now keeps maxDisk lines OR
  ≤80% byte budget, whichever retains fewer (TRIM_TARGET_FACTOR);
  budget env-overridable via SECURITY_EVENT_TRIM_BYTES / opts.trimSizeLimit
- trim/append race: trim renamed over the file mid-append losing events
  to the unlinked inode; now single-flight, queue-empty gated, holds the
  write lock, error paths always release + lazily re-kick (no hot loop)

Judge: glm-4.6@zai-coding-paas adversarial cold-read, grade B clean
(0 blocking, 4 polish — 2 folded, 1 already-satisfied, 1 deferred as
judge-endorsed safer). 127 suites / 2849 tests green.
This commit is contained in:
Hermes
2026-08-23 15:02:00 -07:00
parent 65457ff8e0
commit 2dce6dca5e
2 changed files with 325 additions and 28 deletions
@@ -0,0 +1,231 @@
/**
* DC-116 regression pins — security event store retention + query.total.
*
* Background (2026-08-23, one day after DC-113 activated the caddy source):
* live store had 46,494 events (16.5MB) growing ~2MB/day. Cold review of
* src/security/event-store.js found three defects:
*
* 1. query().total lied: the scan broke at offset+limit, so `total` was
* capped at the page size (<=1000). LIVE user-facing impact — the
* dashboard "N events (24h)" stat (status/js/security-center.js reads
* data.total) and GET /hosts/:id/health events_24h showed 1000 when
* the real 24h count was tens of thousands.
* 2. Trim trigger/curer mismatch: trigger was byte-based (>50MB) but the
* curer was line-count-based (no-op unless >maxDisk=100k lines). If the
* average line ever exceeded ~524B (50MB/100k — 0.5% of live lines were
* already >524B, scanner bursts inflate metadata), trim fired on every
* append and rewrote nothing — unbounded file + full-file re-read on
* the write path.
* 3. Trim/append race: trim renamed over the file with appends in flight;
* events appended after trim's readFile landed on the unlinked inode
* and were silently lost.
*
* Tests use the REAL store with temp files. No mocks of the module under test.
*/
const path = require('path');
const fs = require('fs');
const os = require('os');
// Hermetic sinks (same pattern as caddy-worker-pipeline-dc113.test.js)
const TMP_DIR = fs.mkdtempSync(path.join(os.tmpdir(), 'dc116-store-'));
process.env.SECURITY_EVENT_LOG_FILE = path.join(TMP_DIR, 'security-events.jsonl');
const { SecurityEventStore } = require('../src/security/event-store');
const silence = { info: () => {}, warn: () => {}, error: () => {} };
function makeStore(opts = {}) {
return new SecurityEventStore({
log: silence,
filePath: path.join(TMP_DIR, `store-${Date.now()}-${Math.random().toString(36).slice(2)}.jsonl`),
...opts,
});
}
// Deterministic event factory. `target` carries the unique marker — it is
// never overridden by the fat-payload tests, which replace `message`.
function ev(n, over = {}) {
return {
source_type: 'api',
actor: `actor-${n % 5}`,
action: `action-${n % 3}`,
target: `t-${n}`,
outcome: 'success',
severity: 'info',
message: `event ${n}`,
...over,
};
}
// Wait until the write queue is fully drained and no trim is in flight
async function settle(store, ms = 50) {
if (store.writeQueue.length === 0 && !store.writing && !store._trimScheduled) return;
await new Promise((r) => setTimeout(r, ms));
return settle(store, ms);
}
describe('DC-116: query().total is the true match count, not the page size', () => {
test('total reflects all matching events beyond limit/offset', async () => {
const store = makeStore({ maxMemory: 10000 });
for (let i = 0; i < 250; i++) store.append(ev(i));
await settle(store);
// Page of 10 — total must be 250, not 10
const r1 = store.query({ limit: 10 });
expect(r1.events).toHaveLength(10);
expect(r1.total).toBe(250);
// Same through pagination
const r2 = store.query({ limit: 100, offset: 200 });
expect(r2.events).toHaveLength(50);
expect(r2.total).toBe(250);
// Filters count matches beyond the page too
const r3 = store.query({ limit: 5, actor: 'actor-1' });
expect(r3.total).toBe(50);
expect(r3.events.every((e) => e.actor === 'actor-1')).toBe(true);
});
test('pages are disjoint and newest-first across offsets (dashboard pagination)', async () => {
const store = makeStore({ maxMemory: 10000 });
for (let i = 0; i < 30; i++) store.append(ev(i));
await settle(store);
const p1 = store.query({ limit: 10, offset: 0 }).events;
const p2 = store.query({ limit: 10, offset: 10 }).events;
const p3 = store.query({ limit: 10, offset: 20 }).events;
const ids = [...p1, ...p2, ...p3].map((e) => e.id);
expect(ids).toHaveLength(30);
expect(new Set(ids).size).toBe(30); // no overlap, no loss
// Newest first: event 29 (appended last) leads page 1
expect(p1[0].message).toBe('event 29');
expect(p3[9].message).toBe('event 0');
});
});
describe('DC-116: byte-budget trim always converges below the trigger', () => {
test('trims when byte budget exceeded even under the line cap (old code no-oped)', async () => {
// Fat lines (~600B each): 40 lines = ~24KB > 16KB budget, but well under
// any line cap. Pre-DC-116, _trim() returned early (lines <= maxDisk)
// while _maybeTrim kept firing.
const store = makeStore({ maxDisk: 1000, trimSizeLimit: 16 * 1024 });
const fat = 'x'.repeat(600);
for (let i = 0; i < 40; i++) store.append(ev(i, { message: fat }));
await settle(store, 100);
const size = fs.statSync(store.filePath).size;
expect(size).toBeLessThan(16 * 1024); // under the trigger
// The most recent events survived the trim
const lines = fs.readFileSync(store.filePath, 'utf8').trim().split('\n');
expect(lines.length).toBeGreaterThan(0);
expect(lines.length).toBeLessThanOrEqual(40);
const last = JSON.parse(lines[lines.length - 1]);
expect(last.target).toBe('t-39');
});
test('respects the line cap when lines are thin (maxDisk still honored)', async () => {
// Thin lines (~120B): 300 lines = ~36KB > 16KB budget; maxDisk=100 must
// cap retained lines at 100 (~12KB) — under budget either way.
const store = makeStore({ maxDisk: 100, trimSizeLimit: 16 * 1024 });
for (let i = 0; i < 300; i++) store.append(ev(i));
await settle(store, 100);
const lines = fs.readFileSync(store.filePath, 'utf8').trim().split('\n');
expect(lines.length).toBeLessThanOrEqual(100);
expect(fs.statSync(store.filePath).size).toBeLessThan(16 * 1024);
const last = JSON.parse(lines[lines.length - 1]);
expect(last.target).toBe('t-299');
});
test('byte ceiling drops oldest lines even when under the line cap (both constraints reconcile)', async () => {
// maxDisk=1000 (no line pressure) but budget forces byte reduction:
// 40 fat lines ~24KB -> must fall under 80% of 16KB = 12.8KB (~21 lines)
const store = makeStore({ maxDisk: 1000, trimSizeLimit: 16 * 1024 });
const fat = 'x'.repeat(600);
for (let i = 0; i < 40; i++) store.append(ev(i, { message: fat }));
await settle(store, 100);
const size = fs.statSync(store.filePath).size;
expect(size).toBeLessThanOrEqual(Math.floor(16 * 1024 * 0.8) + 700); // ceiling + one fat line
expect(size).toBeLessThan(16 * 1024);
});
});
describe('DC-116: trim/append race — events appended around a trim are never lost', () => {
test('appends landing during trim survive (write lock serializes trim vs append)', async () => {
const store = makeStore({ maxDisk: 50, trimSizeLimit: 8 * 1024 });
const fat = 'x'.repeat(400);
// Push past the byte budget so the NEXT idle write path triggers a trim
for (let i = 0; i < 20; i++) store.append(ev(i, { message: fat }));
await settle(store, 100);
// Rapid-fire appends around trims: each burst re-crosses the 8KB budget,
// forcing multiple trims while appends keep flowing. Budget sized so the
// FINAL burst (~3.3KB) always fits under the post-trim ceiling — the
// retention contract guarantees the newest burst survives intact.
const ids = [];
for (let round = 0; round < 5; round++) {
for (let i = 0; i < 6; i++) {
const stored = store.append(ev(100 + round * 6 + i, { message: fat }));
ids.push(stored.id);
}
await settle(store, 100);
}
// Every appended event must be either on disk or accounted for by the
// explicit retention caps (maxDisk=50 lines / 8KB byte budget). The last
// burst MUST be fully on disk (it fits the budget; nothing newer exists).
const lines = fs.readFileSync(store.filePath, 'utf8').trim().split('\n');
const diskIds = new Set(lines.map((l) => JSON.parse(l).id));
const lastBurst = ids.slice(-6);
for (const id of lastBurst) {
expect(diskIds.has(id)).toBe(true);
}
// And the file is back under budget
expect(fs.statSync(store.filePath).size).toBeLessThan(8 * 1024);
});
test('in-memory index stays queryable and consistent right after a trim', async () => {
const store = makeStore({ maxDisk: 10, trimSizeLimit: 8 * 1024 });
for (let i = 0; i < 60; i++) store.append(ev(i, { message: 'y'.repeat(300) }));
await settle(store, 150);
// Disk kept <=10 lines; memory still serves the capped window
const lines = fs.readFileSync(store.filePath, 'utf8').trim().split('\n');
expect(lines.length).toBeLessThanOrEqual(10);
const q = store.query({ limit: 5 });
expect(q.total).toBe(store.size());
expect(q.events).toHaveLength(5);
});
});
describe('DC-116: trim error paths release the write lock (no wedged store)', () => {
test('rename failure resets _trimScheduled and writing so later appends flow', async () => {
const store = makeStore({ maxDisk: 5, trimSizeLimit: 2 * 1024 });
const fat = 'x'.repeat(500);
for (let i = 0; i < 10; i++) store.append(ev(i, { message: fat }));
await settle(store, 100);
// Sabotage: make the tmp path unwritable so writeFile inside _trim fails
const tmpPath = store.filePath + '.tmp';
fs.mkdirSync(tmpPath); // a DIRECTORY at the tmp path breaks writeFile
for (let i = 10; i < 16; i++) store.append(ev(i, { message: fat }));
await settle(store, 200);
// Lock must be released despite the failure
expect(store.writing).toBe(false);
expect(store._trimScheduled).toBe(false);
fs.rmSync(tmpPath, { recursive: true, force: true });
// Appends still land on disk after the sabotage is cleared (write path
// was never wedged). The post-append idle trim may legitimately SHRINK
// the file back under budget, so assert on content, not size.
const last = store.append(ev(99, { message: fat }));
await settle(store, 100);
const lines = fs.readFileSync(store.filePath, 'utf8').trim().split('\n');
const diskIds = new Set(lines.map((l) => JSON.parse(l).id));
expect(diskIds.has(last.id)).toBe(true);
});
});
+94 -28
View File
@@ -34,6 +34,12 @@ const EVENT_STORE_FILE = process.env.SECURITY_EVENT_LOG_FILE
const MAX_EVENTS_IN_MEMORY = parseInt(process.env.SECURITY_EVENT_MAX_MEMORY || '10000', 10); const MAX_EVENTS_IN_MEMORY = parseInt(process.env.SECURITY_EVENT_MAX_MEMORY || '10000', 10);
const MAX_EVENTS_ON_DISK = parseInt(process.env.SECURITY_EVENT_MAX_DISK || '100000', 10); const MAX_EVENTS_ON_DISK = parseInt(process.env.SECURITY_EVENT_MAX_DISK || '100000', 10);
// DC-116: size trigger for disk trim. Env-overridable so operators (and tests)
// can tighten it without rebuilding. Default unchanged: 50MB.
const TRIM_TARGET_FACTOR = 0.8; // post-trim target: ≤80% of the byte budget
const DEFAULT_TRIM_SIZE_LIMIT = parseInt(
process.env.SECURITY_EVENT_TRIM_BYTES || String(50 * 1024 * 1024), 10);
const VALID_SOURCE_TYPES = new Set(['api', 'caddy', 'fail2ban', 'shared-bans', 'syslog', 'agent']); const VALID_SOURCE_TYPES = new Set(['api', 'caddy', 'fail2ban', 'shared-bans', 'syslog', 'agent']);
const VALID_SEVERITIES = new Set(['info', 'notice', 'warn', 'error', 'critical']); const VALID_SEVERITIES = new Set(['info', 'notice', 'warn', 'error', 'critical']);
const VALID_OUTCOMES = new Set(['success', 'failure', 'denied', 'blocked', 'rate-limited', 'error', 'unknown']); const VALID_OUTCOMES = new Set(['success', 'failure', 'denied', 'blocked', 'rate-limited', 'error', 'unknown']);
@@ -44,12 +50,15 @@ class SecurityEventStore extends EventEmitter {
this.filePath = opts.filePath || EVENT_STORE_FILE; this.filePath = opts.filePath || EVENT_STORE_FILE;
this.maxMemory = opts.maxMemory || MAX_EVENTS_IN_MEMORY; this.maxMemory = opts.maxMemory || MAX_EVENTS_IN_MEMORY;
this.maxDisk = opts.maxDisk || MAX_EVENTS_ON_DISK; this.maxDisk = opts.maxDisk || MAX_EVENTS_ON_DISK;
// DC-116: byte budget for the on-disk file (trigger AND target of _trim)
this.trimSizeLimit = opts.trimSizeLimit != null ? opts.trimSizeLimit : DEFAULT_TRIM_SIZE_LIMIT;
this.log = opts.log || console; this.log = opts.log || console;
this.events = []; // newest first this.events = []; // newest first
this.byId = new Map(); this.byId = new Map();
this.lastWriteLine = 0; // byte offset of last successfully-written line this.lastWriteLine = 0; // byte offset of last successfully-written line
this.writeQueue = []; // serialized write buffer this.writeQueue = []; // serialized write buffer
this.writing = false; this.writing = false;
this._trimScheduled = false; // DC-116: one in-flight trim at a time
this._load(); this._load();
} }
@@ -170,59 +179,108 @@ class SecurityEventStore extends EventEmitter {
/** /**
* Serialize appends to disk. Writes one line at a time, doesn't truncate. * Serialize appends to disk. Writes one line at a time, doesn't truncate.
* Disk trimming happens separately via _trim(). * Disk trimming happens separately via _maybeTrim() — and only while the
* write queue is empty, so an append can never land on the unlinked
* pre-trim inode (DC-116).
*/ */
_flushQueue() { _flushQueue() {
if (this.writing) return; if (this.writing) return;
const next = this.writeQueue.shift(); const next = this.writeQueue.shift();
if (!next) return; if (!next) {
// Write path idle — safe point to run a pending trim (no in-flight
// append can race the rename; new appends queue behind this.writing).
this._maybeTrim();
return;
}
this.writing = true; this.writing = true;
const line = JSON.stringify(next) + '\n'; const line = JSON.stringify(next) + '\n';
fs.appendFile(this.filePath, line, 'utf8', (err) => { fs.appendFile(this.filePath, line, 'utf8', (err) => {
this.writing = false; this.writing = false;
if (err) { if (err) {
this.log.error?.('security', 'write failed', { error: err.message }); this.log.error?.('security', 'write failed', { error: err.message });
// Re-queue so we don't lose the event on transient errors // Re-queue so we don't lose the event on transient errors. Do NOT
// auto-retry here — a persistent failure (disk full, perms) would
// turn setImmediate into a hot loop. The next append() re-kicks
// the flush (same semantics as before DC-116).
this.writeQueue.unshift(next); this.writeQueue.unshift(next);
return;
}
// Drain the rest of the queue before considering a trim. (Note: this
// merely defers to the next tick — it batches, it does not throttle;
// with an instantly-draining queue each drain can still end in a trim.)
if (this.writeQueue.length > 0) {
setImmediate(() => this._flushQueue());
} else { } else {
// Try next
if (this.writeQueue.length > 0) setImmediate(() => this._flushQueue());
this._maybeTrim(); this._maybeTrim();
} }
}); });
} }
/** /**
* Trim disk log if it exceeds maxDisk lines. Done in the background — never * Trim disk log when it exceeds the byte budget. Runs on the write path
* blocks an append(). Strategy: rewrite the file keeping the most recent * only when the queue is empty (see _flushQueue), and at most one trim is
* in flight at a time. Strategy: rewrite the file keeping the most recent
* maxDisk lines, atomically (write tmp + rename). * maxDisk lines, atomically (write tmp + rename).
*
* DC-116 fix — two prior bugs:
* 1. Trigger/curer mismatch: the trigger was size-based (>50MB) but the
* curer was line-count-based (keep maxDisk=100k lines). If the average
* line exceeds SIZE_LIMIT/maxDisk (~524B) the trim no-ops forever while
* the size trigger keeps firing — unbounded file + a full-file stat
* (and potentially re-read) on every append. Now: trim fires on size
* and keeps the most recent maxDisk LINES OR enough BYTES to get under
* 80% of the budget, whichever retains fewer lines — the file always
* shrinks back below the trigger.
* 2. Trim/append race: trim renamed over the file while unrelated appends
* were in flight, silently losing them to the unlinked inode. Now trim
* runs only between writes (queue empty, this.writing false) and holds
* the write lock for its duration.
*/ */
_maybeTrim() { _maybeTrim() {
if (this._trimScheduled) return; // one at a time
fs.stat(this.filePath, (err, st) => { fs.stat(this.filePath, (err, st) => {
if (err || !st) return; if (err || !st) return;
// Cheap heuristic: if file is > 50MB we always trim. Otherwise count lines. if (st.size < this.trimSizeLimit) return;
const SIZE_LIMIT = 50 * 1024 * 1024; this._trimScheduled = true;
if (st.size < SIZE_LIMIT) return; this.writing = true; // hold the write lock for the whole trim
this._trim(); this._trim(st.size);
}); });
} }
_trim() { _trim(fileSizeBytes) {
this.log.info?.('security', 'trimming event store', { file: this.filePath }); this.log.info?.('security', 'trimming event store', {
file: this.filePath,
size_bytes: fileSizeBytes,
keep_lines: this.maxDisk,
budget_bytes: this.trimSizeLimit,
});
fs.readFile(this.filePath, 'utf8', (err, content) => { fs.readFile(this.filePath, 'utf8', (err, content) => {
if (err) return; const done = (e) => {
this._trimScheduled = false;
this.writing = false; // release the write lock
if (this.writeQueue.length > 0) setImmediate(() => this._flushQueue());
if (e) this.log.error?.('security', 'trim failed', { error: e.message });
};
if (err) return done(err);
const lines = content.split('\n').filter(l => l.trim()); const lines = content.split('\n').filter(l => l.trim());
if (lines.length <= this.maxDisk) return; if (lines.length === 0) return done();
const kept = lines.slice(-this.maxDisk).join('\n') + '\n';
// Byte-aware reconciliation: keep the most recent maxDisk lines, but if
// that slice alone still exceeds ~80% of the byte budget, drop further
// lines (oldest first) until it fits. Always retains at least one line.
const keep = lines.slice(-this.maxDisk);
let keepBytes = Buffer.byteLength(keep.join('\n') + '\n', 'utf8');
const byteCeiling = Math.floor(this.trimSizeLimit * TRIM_TARGET_FACTOR);
while (keep.length > 1 && keepBytes > byteCeiling) {
keepBytes -= Buffer.byteLength(keep.shift() + '\n', 'utf8');
}
const kept = keep.join('\n') + '\n';
const tmp = this.filePath + '.tmp'; const tmp = this.filePath + '.tmp';
fs.writeFile(tmp, kept, 'utf8', (e) => { fs.writeFile(tmp, kept, 'utf8', (e) => {
if (e) { if (e) return done(e);
this.log.error?.('security', 'trim write failed', { error: e.message }); fs.rename(tmp, this.filePath, done);
return;
}
fs.rename(tmp, this.filePath, (e2) => {
if (e2) this.log.error?.('security', 'trim rename failed', { error: e2.message });
});
}); });
}); });
} }
@@ -230,6 +288,13 @@ class SecurityEventStore extends EventEmitter {
/** /**
* Query events. All filters are AND-combined. Results are newest-first. * Query events. All filters are AND-combined. Results are newest-first.
* *
* `total` is the true count of ALL matching events in the memory window
* (DC-116 fix: the loop previously broke at offset+limit, so `total` was
* silently capped at the page size — the dashboard's "N events (24h)" stat
* and hosts/:id/health events_24h read 1000 when the real count was tens
* of thousands). The scan now always completes; per-page cost is bounded
* by the in-memory cap (maxMemory, default 10k).
*
* @param {object} q - query * @param {object} q - query
* limit : number, default 100, max 1000 * limit : number, default 100, max 1000
* offset : number, default 0 * offset : number, default 0
@@ -249,7 +314,8 @@ class SecurityEventStore extends EventEmitter {
const severities = this._toArr(q.severity); const severities = this._toArr(q.severity);
const outcomes = this._toArr(q.outcome); const outcomes = this._toArr(q.outcome);
const matches = []; let total = 0;
const page = [];
for (const ev of this.events) { for (const ev of this.events) {
if (sourceTypes.length && !sourceTypes.includes(ev.source_type)) continue; if (sourceTypes.length && !sourceTypes.includes(ev.source_type)) continue;
if (q.source_host && ev.source_host !== q.source_host) continue; if (q.source_host && ev.source_host !== q.source_host) continue;
@@ -261,13 +327,13 @@ class SecurityEventStore extends EventEmitter {
if (q.since && ev.ts < q.since) continue; if (q.since && ev.ts < q.since) continue;
if (q.until && ev.ts >= q.until) continue; if (q.until && ev.ts >= q.until) continue;
if (q.target && ev.target !== q.target) continue; if (q.target && ev.target !== q.target) continue;
matches.push(ev); total++;
if (matches.length >= offset + limit) break; // avoid scanning further if (total > offset && page.length < limit) page.push(ev);
} }
return { return {
total: matches.length, total,
events: matches.slice(offset, offset + limit), events: page,
}; };
} }