Merge pull request #354 from Lob26/fix/realtime-debug-log-hardening

fix(realtime): bound the debug-log sink and stop blocking on its writes

CHANGELOG.md

@@ -133,6 +133,8 @@
 
 - Expose portable radio, camera-type and regional source helpers; keep HTTP transport separate from record normalization.
 
+- The Realtime debug-log endpoint is bounded on every axis it was not: an always-on per-client rate limit, asynchronous appends through a serialized queue instead of a synchronous write on the request path, and rotation of the log file at 32 MB keeping one prior generation. The 8 MB cap applied to a single request body and never to the file those requests accumulated into, so any local page could grow it for as long as the dev server ran. A malformed record and a failed write are now told apart, 400 from 500, and neither answer carries the error text.
+
 - Three proxy paths no longer relay upstream or JS error text to the client. The HUD summary passed OpenAI's own `error.message` through whenever upstream was not ok, carrying request ids and quota wording; the Realtime token route passed through non-success response bodies and echoed JS errors, which can expose upstream details; and a failed CCTV media fetch stored the raw errno as the camera's health message, which reaches the screen through `GET /api/cctv/health` rather than through the sanitized response beside it. Logs now name the failure and the upstream status without the text.
 
 - Separate vessel records and feed acquisition from rendering while preserving selection, partial-feed retention, sea-surface placement and request cancellation.

docs/CURRENT-STATE.md

@@ -3583,6 +3583,7 @@ are omitted rather than framing the wrong part of the globe.
 - `/api/route` proxies bounded OSRM route requests for annotation routes, with profile allowlisting, distance caps, response caps, caching, and sanitized "no route found" errors.
 - Track endpoints: `/api/ais-live/track?mmsi=` (server-accumulated ring buffers; sub-route handled before the rows snapshot), `/api/opensky-track?icao24=` (OAuth, 60s cache, sanitized errors, independent OpenSky credit bucket), `/api/adsblol/trace?hex=` (60s cache, 5MB cap, ODbL attribution required in UI).
 - Realtime debug logs redact API keys, bearer tokens, client secrets, and image data URLs before writing to disk; request bodies are size-capped.
+- `/api/realtime/debug-log` enforces an always-on 120-per-client/400-global one-minute limiter (not the opt-in `GEV_RATELIMIT_OPENAI_PER_MIN` bucket the cost-bearing OpenAI routes share), appends asynchronously through a serialized queue, and rotates `realtime-conversations.jsonl` at 32 MB keeping one prior generation, so the sink is bounded at twice that regardless of session length. A malformed record answers 400 and a failed write 500, both with a fixed message.
 
 ## UI/UX Runtime Defaults
 

server/providers/common/request.js

@@ -17,17 +17,29 @@ async function readRequestBodyCapped(req, maxBytes) {
 function readRequestBody(req, maxBytes = 1024 * 1024) {
   return new Promise((resolve, reject) => {
     let total = 0;
+    let bodyTooLarge = false;
     const chunks = [];
     req.on('data', (chunk) => {
+      // Keep draining after the cap without retaining more bytes. Destroying
+      // the request also destroys its socket, so the handler cannot deliver
+      // the fixed error response it constructs for BODY_TOO_LARGE.
+      if (bodyTooLarge) return;
       total += chunk.length;
       if (total > maxBytes) {
-        reject(new Error(`Request body exceeds ${maxBytes} bytes`));
-        req.destroy();
+        // Same marker readRequestBodyCapped sets, so a caller can tell an
+        // oversized body from a genuine failure without matching on text.
+        bodyTooLarge = true;
+        chunks.length = 0;
+        const error = new Error(`Request body exceeds ${maxBytes} bytes`);
+        error.code = 'BODY_TOO_LARGE';
+        reject(error);
         return;
       }
       chunks.push(chunk);
     });
-    req.on('end', () => resolve(Buffer.concat(chunks).toString('utf8')));
+    req.on('end', () => {
+      if (!bodyTooLarge) resolve(Buffer.concat(chunks).toString('utf8'));
+    });
     req.on('error', reject);
   });
 }

server/providers/openai/debug-log.js

@@ -1,16 +1,65 @@
 import { defaultSourceRoot } from '../common/source-root.js';
 import path from 'node:path';
 import { readRequestBody } from '../common/request.js';
-import fs from 'node:fs';
+import { promises as fsp } from 'node:fs';
+import { makeRateLimiter, clientKey } from '../common/rate-limit.js';
 
+/** Cap on one request body. */
 const REALTIME_DEBUG_LOG_MAX_BYTES = 8 * 1024 * 1024;
 
+/**
+ * Ceiling for the live log before it rotates. One generation is kept
+ * (`…jsonl.1`), so the sink is bounded at twice this however long a session
+ * runs. REALTIME_DEBUG_LOG_MAX_BYTES caps a single request; it never bounded
+ * the file those requests accumulate into.
+ */
+const REALTIME_DEBUG_LOG_MAX_FILE_BYTES = 32 * 1024 * 1024;
+
+/**
+ * Per-IP write ceiling. Always on, unlike the opt-in limiter the cost-bearing
+ * OpenAI routes share: throttling those by default would change what a user
+ * spends, while this sink spends disk and event-loop time. A voice session
+ * writes far below this, so the cap needs no configuration to stay invisible.
+ */
+const REALTIME_DEBUG_LOG_MAX_PER_MIN = 120;
+
 function createDebugLogHandler({ sourceRoot = defaultSourceRoot } = {}) {
-  const REALTIME_DEBUG_LOG_DIR = path.join(sourceRoot, '.gev-logs');
-  const REALTIME_DEBUG_LOG_FILE = path.join(
-    REALTIME_DEBUG_LOG_DIR,
-    'realtime-conversations.jsonl',
-  );
+  const logDir = path.join(sourceRoot, '.gev-logs');
+  const logFile = path.join(logDir, 'realtime-conversations.jsonl');
+  const allow = makeRateLimiter({
+    windowMs: 60_000,
+    max: REALTIME_DEBUG_LOG_MAX_PER_MIN,
+    globalMax: 400,
+  });
+
+  // Appends are chained rather than fired concurrently because the rotation
+  // check is a read-then-write on file size: run in parallel, every caller
+  // stats a file still under the ceiling, none of them rotates, and the log
+  // grows past it. The records themselves are O_APPEND and never interleave.
+  let queue = Promise.resolve();
+
+  function append(line) {
+    const done = queue.then(async () => {
+      await fsp.mkdir(logDir, { recursive: true });
+      const size = await fsp.stat(logFile).then(
+        (stat) => stat.size,
+        () => 0,
+      );
+      if (size + Buffer.byteLength(line) > REALTIME_DEBUG_LOG_MAX_FILE_BYTES) {
+        // Node does not replace an existing destination on Windows. Remove the
+        // retained generation explicitly so every rotation behaves the same on
+        // supported platforms rather than failing after the first 64 MB.
+        await fsp.rm(`${logFile}.1`, { force: true });
+        await fsp.rename(logFile, `${logFile}.1`);
+      }
+      await fsp.appendFile(logFile, line);
+    });
+    // The caller still sees its own rejection, but one failed append must not
+    // poison the chain for every request queued behind it.
+    queue = done.catch(() => {});
+    return done;
+  }
+
   return async (req, res) => {
     if (req.method !== 'POST') {
       res.statusCode = 405;
@@ -19,12 +68,19 @@ function createDebugLogHandler({ sourceRoot = defaultSourceRoot } = {}) {
       return;
     }
 
+    if (!allow(clientKey(req))) {
+      res.writeHead(429, {
+        'Content-Type': 'application/json',
+        'Retry-After': '60',
+      });
+      res.end(JSON.stringify({ error: 'Rate limit exceeded' }));
+      return;
+    }
+
     try {
       const body = await readRequestBody(req, REALTIME_DEBUG_LOG_MAX_BYTES);
       const record = JSON.parse(body || '{}');
-      fs.mkdirSync(REALTIME_DEBUG_LOG_DIR, { recursive: true });
-      fs.appendFileSync(
-        REALTIME_DEBUG_LOG_FILE,
+      await append(
         `${JSON.stringify({
           loggedAt: new Date().toISOString(),
           ...record,
@@ -33,13 +89,16 @@ function createDebugLogHandler({ sourceRoot = defaultSourceRoot } = {}) {
       res.statusCode = 204;
       res.end();
     } catch (error) {
-      res.statusCode = 400;
+      // A malformed record or an oversized body is the caller's fault; a failed
+      // write is ours, and only the second is worth a log line. Neither answer
+      // carries the error text, which for an fs failure is an errno and an
+      // absolute path.
+      const callerFault =
+        error instanceof SyntaxError || error?.code === 'BODY_TOO_LARGE';
+      if (!callerFault) console.warn('[realtime-debug-log] append failed');
+      res.statusCode = callerFault ? 400 : 500;
       res.setHeader('Content-Type', 'application/json');
-      res.end(
-        JSON.stringify({
-          error: error?.message || 'Failed to write Realtime debug log',
-        }),
-      );
+      res.end(JSON.stringify({ error: 'Failed to write Realtime debug log' }));
     }
   };
 }

src/tooling/localServices.test.mjs

@@ -1,5 +1,6 @@
 import assert from 'node:assert/strict';
 import test from 'node:test';
+import { createServer } from 'node:http';
 import { Readable } from 'node:stream';
 import {
   mkdtempSync,
@@ -337,3 +338,99 @@ test('OpenAI routes answer generically when the upstream or the request fails',
   assert.equal(token.body.includes('api.openai.com'), false);
   assert.equal(token.body.includes('fixture-upstream-secret'), false);
 });
+
+test('the debug-log sink stays bounded, rate limited, and quiet about failures', async (t) => {
+  const sourceRoot = root(t);
+  const handler = install(openAiRealtimeProxy({ sourceRoot })).get(
+    '/api/realtime/debug-log',
+  );
+  const file = path.join(sourceRoot, '.gev-logs/realtime-conversations.jsonl');
+  const write = (record) =>
+    request(handler, { method: 'POST', body: JSON.stringify(record) });
+
+  // A malformed record is the caller's fault and a 400; neither answer carries
+  // the error text, which for a write failure is an errno and an absolute path.
+  const malformed = await request(handler, { method: 'POST', body: '{nope' });
+  assert.equal(malformed.status, 400);
+  assert.deepEqual(malformed.json(), {
+    error: 'Failed to write Realtime debug log',
+  });
+
+  // The limiter is always on, unlike the opt-in one the cost-bearing routes
+  // share: 120/min per IP, far above what a voice session writes.
+  let limited = null;
+  let accepted = 0;
+  for (let n = 0; n < 130 && !limited; n += 1) {
+    const response = await write({ n });
+    if (response.status === 429) limited = response;
+    else if (response.status === 204) accepted += 1;
+  }
+  assert.ok(limited, 'the sink refuses a caller past its per-minute ceiling');
+  assert.equal(limited.headers['retry-after'], '60');
+  assert.deepEqual(limited.json(), { error: 'Rate limit exceeded' });
+  // The limiter counts requests rather than successful writes, so the malformed
+  // record above already spent one of the 120 slots.
+  assert.equal(accepted, 119);
+
+  // Every accepted record is on disk and parses: the queue serializes appends,
+  // so none was lost or truncated by the ones beside it.
+  const lines = readFileSync(file, 'utf8').split('\n').filter(Boolean);
+  assert.equal(lines.length, accepted);
+  for (const line of lines) assert.doesNotThrow(() => JSON.parse(line));
+  assert.ok(lines.every((line) => JSON.parse(line).loggedAt));
+});
+
+test('an oversized debug-log request receives the fixed error response', async (t) => {
+  const handler = install(openAiRealtimeProxy({ sourceRoot: root(t) })).get(
+    '/api/realtime/debug-log',
+  );
+  const server = createServer(handler);
+  await new Promise((resolve) => server.listen(0, '127.0.0.1', resolve));
+  t.after(() => new Promise((resolve) => server.close(resolve)));
+
+  const response = await fetch(
+    `http://127.0.0.1:${server.address().port}/api/realtime/debug-log`,
+    { method: 'POST', body: 'x'.repeat(8 * 1024 * 1024 + 1) },
+  );
+  assert.equal(response.status, 400);
+  assert.deepEqual(await response.json(), {
+    error: 'Failed to write Realtime debug log',
+  });
+});
+
+test('the debug log rotates instead of growing without bound', async (t) => {
+  const sourceRoot = root(t);
+  const handler = install(openAiRealtimeProxy({ sourceRoot })).get(
+    '/api/realtime/debug-log',
+  );
+  const file = path.join(sourceRoot, '.gev-logs/realtime-conversations.jsonl');
+
+  // 8 MB bounds one request body; nothing bounded the file until now, so a
+  // single page could grow it for as long as the dev server ran. Each record
+  // here is ~1 MB, well inside the body cap.
+  const pad = 'x'.repeat(1024 * 1024);
+  // Seventy-two records force two rotations. The second one replaces an
+  // existing `.1`, which requires an explicit removal on Windows.
+  for (let n = 0; n < 72; n += 1) {
+    assert.equal(
+      (
+        await request(handler, {
+          method: 'POST',
+          body: JSON.stringify({ n, pad }),
+        })
+      ).status,
+      204,
+    );
+  }
+
+  const live = statSync(file).size;
+  const previous = statSync(`${file}.1`).size;
+  assert.ok(live <= 32 * 1024 * 1024, `live log under the ceiling (${live})`);
+  assert.ok(
+    live + previous <= 64 * 1024 * 1024,
+    'both generations together stay within twice the ceiling',
+  );
+  // Unrotated, these records would be ~75 MB in one file.
+  assert.ok(live + previous < 64 * 1024 * 1024);
+  assert.ok(!existsSync(`${file}.2`), 'exactly one generation is retained');
+});