🪦 feat: Request Tombstones, Memory Watermarks & Server Error Logging

.env.example

@@ -55,3 +55,7 @@ SESSION_SECRET=
 # ADMIN_PANEL_INDEX_CACHE_CONTROL=   # overrides INDEX_CACHE_CONTROL for admin panel
 # ADMIN_PANEL_INDEX_PRAGMA=          # overrides INDEX_PRAGMA for admin panel
 # ADMIN_PANEL_INDEX_EXPIRES=         # overrides INDEX_EXPIRES for admin panel
+
+# Request and memory observability
+# ADMIN_PANEL_REQUEST_LOG=false      # disables request arrival/completion logging (default: enabled)
+# ADMIN_PANEL_MEMORY_LOG_THRESHOLDS_MB=256,384,448 # RSS thresholds in MiB, each logged once per upward crossing

server.ts

@@ -1,5 +1,13 @@
 import { Glob } from 'bun';
 import { join } from 'node:path';
+import {
+  isProbeRequest,
+  reportOnBodyComplete,
+  createFloodGuard,
+  formatLoggedPath,
+  createMemoryWatermark,
+  parseMemoryThresholdsMb,
+} from './src/server/logging';
 import {
   metricsResponse,
   httpRequestsTotal,
@@ -97,18 +105,71 @@ type Handler = { default: { fetch: (req: Request) => Promise<Response> } };
 
 const { default: handler } = (await import(SERVER_ENTRY.href)) as Handler;
 
-async function withHttpMetrics(
+const REQUEST_LOG = process.env.ADMIN_PANEL_REQUEST_LOG !== 'false';
+const floodGuard = createFloodGuard();
+
+async function withHttpObservability(
   req: Request,
   pathname: string,
   getResponse: () => Response | Promise<Response>,
+  options?: { monitorBody?: boolean },
 ): Promise<Response> {
   const path = normalizeMetricsPath(pathname);
   const end = httpRequestDurationSeconds.startTimer({ method: req.method, path });
-  const res = await getResponse();
+
+  // The arrival line is logged before the handler runs so that a request that
+  // kills the process still leaves its method and path as the last log line.
+  let logCompletion = false;
+  let startedAt = 0;
+  let loggedPath = '';
+  if (REQUEST_LOG && !isProbeRequest(req.headers.get('user-agent'), new URL(req.url).pathname)) {
+    const { admitted, suppressedInPriorWindow } = floodGuard.admit(Date.now());
+    if (suppressedInPriorWindow > 0) {
+      console.log(`[req] suppressed ${suppressedInPriorWindow} requests in prior window`);
+    }
+    if (admitted) {
+      loggedPath = formatLoggedPath(new URL(req.url).pathname);
+      startedAt = performance.now();
+      console.log(`[req] ${req.method} ${loggedPath}`);
+      logCompletion = true;
+    }
+  }
+
+  const logDone = (status: string, note = ''): void => {
+    const elapsed = Math.round(performance.now() - startedAt);
+    console.log(`[req] ${req.method} ${loggedPath} ${status} ${elapsed}ms${note}`);
+  };
+
+  let res: Response;
+  try {
+    res = await getResponse();
+  } catch (err) {
+    // Bun's error callback turns this into a 500, but that line carries no method or
+    // path, so the arrival tombstone would otherwise never be closed out.
+    httpRequestsTotal.inc({ method: req.method, path, status_code: '500' });
+    end({ status_code: '500' });
+    if (logCompletion) logDone('500', ' handler-error');
+    throw err;
+  }
+
   const statusCode = String(res.status);
   httpRequestsTotal.inc({ method: req.method, path, status_code: statusCode });
   end({ status_code: statusCode });
-  return res;
+  if (!logCompletion) return res;
+
+  // Bun discards a HEAD body without reading or cancelling it, so a monitored
+  // stream would never settle and the completion line would never be emitted.
+  // A native file body must also be left intact: replacing it with a JavaScript
+  // stream costs Bun's sendfile path and the Content-Length it derives from the
+  // file, turning every static asset into a chunked transfer.
+  if (req.method === 'HEAD' || options?.monitorBody === false) {
+    logDone(statusCode);
+    return res;
+  }
+
+  return reportOnBodyComplete(res, (outcome) =>
+    logDone(statusCode, outcome === 'stream-error' ? ' stream-error' : ''),
+  );
 }
 
 async function buildStaticRoutes(): Promise<Record<string, (req: Request) => Promise<Response>>> {
@@ -118,11 +179,16 @@ async function buildStaticRoutes(): Promise<Record<string, (req: Request) => Pro
     const cache = getCacheHeaders(path);
     const routePath = `${BASE_PATH}/${path}`;
     routes[routePath] = (req) =>
-      withHttpMetrics(req, routePath, () => {
-        const res = new Response(file, { headers: { 'Content-Type': file.type, ...cache } });
-        applySecurityHeaders(res.headers);
-        return res;
-      });
+      withHttpObservability(
+        req,
+        routePath,
+        () => {
+          const res = new Response(file, { headers: { 'Content-Type': file.type, ...cache } });
+          applySecurityHeaders(res.headers);
+          return res;
+        },
+        { monitorBody: false },
+      );
   }
   return routes;
 }
@@ -139,7 +205,7 @@ const server = Bun.serve({
       const metricsPath = BASE_PATH && url.pathname.startsWith(BASE_PATH)
         ? url.pathname.slice(BASE_PATH.length) || '/'
         : url.pathname;
-      const res = await withHttpMetrics(req, metricsPath, () => handler.fetch(req));
+      const res = await withHttpObservability(req, metricsPath, () => handler.fetch(req));
       const patched = new Response(res.body, res);
       for (const [k, v] of Object.entries(NO_CACHE)) {
         patched.headers.set(k, v);
@@ -148,8 +214,27 @@ const server = Bun.serve({
       return patched;
     },
   },
+  error(error: Error): Response {
+    console.error(`[error] unhandled server error: ${error.message}`, error.stack ?? '');
+    return new Response('Internal Server Error', { status: 500 });
+  },
 });
 
+const memoryWatermark = createMemoryWatermark(
+  parseMemoryThresholdsMb(env.ADMIN_PANEL_MEMORY_LOG_THRESHOLDS_MB),
+);
+const MEMORY_CHECK_INTERVAL_MS = 10_000;
+setInterval(() => {
+  const { rss, heapUsed } = process.memoryUsage();
+  const crossedMb = memoryWatermark.check(rss);
+  if (crossedMb !== null) {
+    const mib = 1024 * 1024;
+    console.log(
+      `[mem] rss=${Math.round(rss / mib)}Mi heapUsed=${Math.round(heapUsed / mib)}Mi (crossed ${crossedMb}Mi)`,
+    );
+  }
+}, MEMORY_CHECK_INTERVAL_MS);
+
 console.log(`Admin panel listening on http://localhost:${server.port}${BASE_PATH}/`);
 
 if (!process.env.ADMIN_PANEL_METRICS_SECRET) {

src/server/logging.test.ts

@@ -0,0 +1,189 @@
+import { describe, it, expect } from 'vitest';
+import {
+  isProbeRequest,
+  createFloodGuard,
+  formatLoggedPath,
+  createMemoryWatermark,
+  reportOnBodyComplete,
+  parseMemoryThresholdsMb,
+} from './logging';
+
+const MIB = 1024 * 1024;
+
+describe('isProbeRequest', () => {
+  it.each([
+    ['kube-probe/1.29', true],
+    ['kube-probe/1.31+', true],
+    ['Mozilla/5.0 (Macintosh)', false],
+    ['curl/8.7.1', false],
+    ['', false],
+    [null, false],
+  ])('user-agent %s on the probe path -> %s', (userAgent, expected) => {
+    expect(isProbeRequest(userAgent, '/health')).toBe(expected);
+  });
+
+  it.each(['/', '/api/config', '/admin/users'])(
+    'refuses to suppress %s even for a probe user-agent',
+    (pathname) => {
+      expect(isProbeRequest('kube-probe/1.29', pathname)).toBe(false);
+    },
+  );
+});
+
+describe('formatLoggedPath', () => {
+  it('passes short paths through unchanged', () => {
+    expect(formatLoggedPath('/auth/openid/callback')).toBe('/auth/openid/callback');
+  });
+
+  it('truncates paths beyond 200 characters', () => {
+    const long = `/${'a'.repeat(500)}`;
+    const formatted = formatLoggedPath(long);
+    expect(formatted).toBe(`${long.slice(0, 200)}...(truncated)`);
+  });
+
+  it('keeps a path of exactly 200 characters intact', () => {
+    const exact = `/${'a'.repeat(199)}`;
+    expect(formatLoggedPath(exact)).toBe(exact);
+  });
+});
+
+describe('createFloodGuard', () => {
+  it('admits requests up to the cap within one window', () => {
+    const guard = createFloodGuard(3, 10_000);
+    expect(guard.admit(1_000).admitted).toBe(true);
+    expect(guard.admit(2_000).admitted).toBe(true);
+    expect(guard.admit(3_000).admitted).toBe(true);
+    expect(guard.admit(4_000).admitted).toBe(false);
+    expect(guard.admit(5_000).admitted).toBe(false);
+  });
+
+  it('reports the suppressed count once when a new window opens', () => {
+    const guard = createFloodGuard(2, 10_000);
+    guard.admit(1_000);
+    guard.admit(2_000);
+    guard.admit(3_000);
+    guard.admit(4_000);
+    const next = guard.admit(12_000);
+    expect(next.admitted).toBe(true);
+    expect(next.suppressedInPriorWindow).toBe(2);
+    expect(guard.admit(13_000).suppressedInPriorWindow).toBe(0);
+  });
+
+  it('resets the admission budget each window', () => {
+    const guard = createFloodGuard(1, 10_000);
+    expect(guard.admit(0).admitted).toBe(true);
+    expect(guard.admit(1).admitted).toBe(false);
+    expect(guard.admit(10_000).admitted).toBe(true);
+    expect(guard.admit(10_001).admitted).toBe(false);
+  });
+});
+
+describe('parseMemoryThresholdsMb', () => {
+  it.each([
+    [undefined, [256, 384, 448]],
+    ['', [256, 384, 448]],
+    ['100,200,300', [100, 200, 300]],
+    ['300, 100, 200', [100, 200, 300]],
+    ['512', [512]],
+    ['abc,-5,0', [256, 384, 448]],
+    ['abc,128', [128]],
+  ])('parses %s -> %s', (raw, expected) => {
+    expect(parseMemoryThresholdsMb(raw)).toEqual(expected);
+  });
+});
+
+describe('createMemoryWatermark', () => {
+  it('reports the highest crossed threshold regardless of caller ordering', () => {
+    const watermark = createMemoryWatermark([448, 256, 384]);
+
+    expect(watermark.check(500 * MIB)).toBe(448);
+  });
+
+  it('fires once per upward crossing and reports the highest threshold crossed', () => {
+    const watermark = createMemoryWatermark([256, 384, 448]);
+    expect(watermark.check(100 * MIB)).toBeNull();
+    expect(watermark.check(260 * MIB)).toBe(256);
+    expect(watermark.check(270 * MIB)).toBeNull();
+    expect(watermark.check(460 * MIB)).toBe(448);
+  });
+
+  it('re-arms a threshold after memory drops back below it', () => {
+    const watermark = createMemoryWatermark([256]);
+    expect(watermark.check(300 * MIB)).toBe(256);
+    expect(watermark.check(310 * MIB)).toBeNull();
+    expect(watermark.check(200 * MIB)).toBeNull();
+    expect(watermark.check(300 * MIB)).toBe(256);
+  });
+
+  it('stays silent while memory remains flat below every threshold', () => {
+    const watermark = createMemoryWatermark([256, 384, 448]);
+    expect(watermark.check(101 * MIB)).toBeNull();
+    expect(watermark.check(102 * MIB)).toBeNull();
+    expect(watermark.check(101 * MIB)).toBeNull();
+  });
+});
+
+describe('reportOnBodyComplete', () => {
+  it('reports immediately for a bodyless response', () => {
+    const outcomes: string[] = [];
+    const res = reportOnBodyComplete(new Response(null, { status: 204 }), (o) => outcomes.push(o));
+
+    expect(outcomes).toEqual(['ok']);
+    expect(res.status).toBe(204);
+  });
+
+  it('waits for the stream to drain before reporting success', async () => {
+    const outcomes: string[] = [];
+    const body = new ReadableStream<Uint8Array>({
+      start(controller) {
+        controller.enqueue(new TextEncoder().encode('chunk'));
+        controller.close();
+      },
+    });
+
+    const res = reportOnBodyComplete(new Response(body, { status: 200 }), (o) => outcomes.push(o));
+    expect(outcomes).toEqual([]);
+
+    await res.text();
+    expect(outcomes).toEqual(['ok']);
+  });
+
+  it('reports a stream error when the upstream body fails mid-transfer', async () => {
+    const outcomes: string[] = [];
+    const body = new ReadableStream<Uint8Array>({
+      start(controller) {
+        controller.enqueue(new TextEncoder().encode('partial'));
+        controller.error(new Error('upstream died'));
+      },
+    });
+
+    const res = reportOnBodyComplete(new Response(body, { status: 200 }), (o) => outcomes.push(o));
+    await expect(res.text()).rejects.toThrow();
+    expect(outcomes).toEqual(['stream-error']);
+  });
+
+  it('reports once when a cancel races a pending read', async () => {
+    const outcomes: string[] = [];
+    /** Never enqueues, so the wrapper's read is still pending when the cancel lands. */
+    const body = new ReadableStream<Uint8Array>({ start() {} });
+
+    const res = reportOnBodyComplete(new Response(body, { status: 200 }), (o) => outcomes.push(o));
+    const reader = res.body!.getReader();
+    const pending = reader.read();
+    await reader.cancel('client gone');
+    await pending.catch(() => {});
+    await new Promise((resolve) => setTimeout(resolve, 0));
+
+    expect(outcomes).toEqual(['stream-error']);
+  });
+
+  it('preserves status and headers', () => {
+    const res = reportOnBodyComplete(
+      new Response('csv', { status: 200, headers: { 'content-type': 'text/csv' } }),
+      () => {},
+    );
+
+    expect(res.status).toBe(200);
+    expect(res.headers.get('content-type')).toBe('text/csv');
+  });
+});