fix(server): stop server crash from log-rotation rename race (#2837)

.changeset/fix-log-sink-rotation-race.md

@@ -0,0 +1,5 @@
+---
+"@inkeep/open-knowledge": patch
+---
+
+Fix a server crash where the app would report losing its connection and require a manual restart. Under normal logging load, multiple loggers writing to the shared server log file could race when the file rotated at its size cap: one writer renamed the active log out from under another, whose failed rename surfaced as an unhandled error that took the whole server process down. Log rotation is now serialized per file across every writer, and a log-sink write failure can no longer crash the server — a dropped log record is reported and the server keeps running.

packages/server/src/logger.test.ts

@@ -442,12 +442,12 @@ describe('Logger', () => {
       expect(existsSync(previousPath)).toBe(true);
     });
 
-    it('flushAllFileSinks drains every cached PinoLogger created via the factory', async () => {
-      // Two named loggers obtained through the factory — each PinoLogger
-      // instance owns its own PinoFileSink with a separate RotatingAppender
-      // promise chain, so draining one does not drain the other. Shutdown
-      // must walk every cached instance for log records written in the final
-      // moments before process.exit() to land on disk.
+    it('flushAllFileSinks drains records from every cached PinoLogger before exit', async () => {
+      // Two named loggers obtained through the factory write to the same log
+      // path, so their PinoFileSinks share one per-path rotation chain.
+      // flushAllFileSinks must drain that chain so records written in the final
+      // moments before process.exit() — from any logger — land on disk rather
+      // than being lost with the unflushed async write queue.
       loggerFactory.configure({
         pinoConfig: {
           options: { level: 'info' },

packages/server/src/logger.ts

@@ -411,9 +411,12 @@ class LoggerFactory {
    * RotatingAppender chain that must be awaited separately.
    *
    * Each PinoLogger constructed via the factory owns its own PinoFileSink
-   * instance (logger.ts's buildInstance() allocates one per logger), so the
-   * factory must walk every cached entry — draining any single logger does
-   * not drain the others. No-op when nothing has a sink wired.
+   * instance (logger.ts's buildInstance() allocates one per logger). Sinks
+   * targeting the same log path share one per-path rotation chain (see
+   * RotatingAppender), so draining any one of them also drains the others on
+   * that path; walking every cached entry stays correct and additionally
+   * covers sinks that target different paths. No-op when nothing has a sink
+   * wired.
    */
   async flushAllFileSinks(): Promise<void> {
     const drains: Promise<void>[] = [];

packages/server/src/telemetry-file-sink.test.ts

@@ -1,12 +1,23 @@
-import { existsSync, mkdtempSync, readFileSync, rmSync, statSync, writeFileSync } from 'node:fs';
+import {
+  existsSync,
+  mkdirSync,
+  mkdtempSync,
+  readFileSync,
+  rmSync,
+  statSync,
+  writeFileSync,
+} from 'node:fs';
 import { tmpdir } from 'node:os';
-import { join } from 'node:path';
+import { dirname, join } from 'node:path';
 import { context, trace } from '@opentelemetry/api';
 import { BasicTracerProvider, SimpleSpanProcessor } from '@opentelemetry/sdk-trace-base';
 import { afterEach, beforeEach, describe, expect, test } from 'vitest';
 import {
   DEFAULT_MAX_VALUE_BYTES,
   FileSpanExporter,
+  logsCurrentPath,
+  logsPreviousPath,
+  PinoFileSink,
   REDACTED_SENTINEL,
   RotatingAppender,
   ScrubbingSpanProcessor,
@@ -339,6 +350,119 @@ describe('RotatingAppender (shared rotation primitive)', () => {
     await appender.drain();
     expect(readFileSync(currentPath, 'utf-8')).toBe('1\n2\n3\n');
   });
+
+  // Invariant A: rotation is safe across EVERY writer to a path, not just
+  // within one appender instance. In production each getLogger(name) allocates
+  // its own PinoFileSink → its own RotatingAppender on the shared log path, so
+  // N loggers = N rotators on one file. The bug this guards against: if
+  // rotation were serialized only per instance, two appenders could both see
+  // "over cap", the first rename would win, and the second would get ENOENT
+  // because current no longer exists — that rejection is what killed the
+  // server. This test drives several appenders on one path across the cap and
+  // asserts NO append rejects (the shared per-path chain makes it atomic).
+  test('serializes rotation across independent appenders sharing one path (no ENOENT race)', async () => {
+    const currentPath = join(tmp, 'shared', 'shared-current.jsonl');
+    const previousPath = join(tmp, 'shared', 'shared-prev.jsonl');
+    const maxBytes = 512;
+    const appenders = Array.from(
+      { length: 6 },
+      () => new RotatingAppender({ currentPath, previousPath, maxBytes }),
+    );
+
+    // Each line is ~200 bytes so a handful of concurrent appends crosses the
+    // 512-byte cap and forces rotation while other appends are in flight.
+    const line = `${JSON.stringify({ level: 30, msg: 'x'.repeat(180) })}\n`;
+    const results = await Promise.allSettled(
+      appenders.flatMap((appender) => Array.from({ length: 40 }, () => appender.append(line))),
+    );
+
+    const rejected = results.filter((r) => r.status === 'rejected');
+    const reasons = rejected.map((r) => String((r as PromiseRejectedResult).reason));
+    expect({ rejectedCount: rejected.length, reasons }).toEqual({
+      rejectedCount: 0,
+      reasons: [],
+    });
+  });
+
+  // Invariant B: a log-sink write failure must NEVER become an unhandled
+  // 'error' on the PinoFileSink stream — pino.multistream attaches no 'error'
+  // listener, so any propagated error escalates to uncaughtException and kills
+  // the whole server. The sink must contain its own I/O failures: _write must
+  // resolve successfully (reporting the failure out-of-band), so the Writable
+  // never emits 'error'. Here we force the rotation rename to fail by planting
+  // a non-empty directory at the previous-generation path, then write past the
+  // cap. A contained sink drains cleanly with no error surfaced to either the
+  // per-write callback or an 'error' event.
+  test('contains sink write failures instead of crashing the process (no unhandled error)', async () => {
+    const projectDir = join(tmp, 'sink-fail');
+    const currentPath = logsCurrentPath(projectDir);
+    const previousPath = logsPreviousPath(projectDir);
+
+    // Plant a non-empty directory where rotation will try to rename current →
+    // prev, guaranteeing the rename rejects (ENOTEMPTY / EISDIR).
+    mkdirSync(dirname(previousPath), { recursive: true });
+    mkdirSync(previousPath);
+    writeFileSync(join(previousPath, 'occupied'), 'x');
+
+    const sink = new PinoFileSink({ projectDir, maxBytes: 64 });
+
+    let emittedError: unknown;
+    sink.on('error', (err) => {
+      emittedError = err;
+    });
+
+    const line = `${JSON.stringify({ level: 30, msg: 'y'.repeat(200) })}\n`;
+    const writeCallbackError = await new Promise<unknown>((resolve) => {
+      sink.write(line, (err) => resolve(err ?? null));
+    });
+    await sink.drain();
+
+    expect({
+      writeCallbackError: writeCallbackError ? String(writeCallbackError) : null,
+      emittedError: emittedError ? String(emittedError) : null,
+    }).toEqual({ writeCallbackError: null, emittedError: null });
+    // The write failed silently but the current file still exists — the sink
+    // must not have taken the process down or lost its file handle.
+    expect(existsSync(currentPath)).toBe(true);
+  });
+
+  // drain() (called from flushAllFileSinks at graceful shutdown) must flush any
+  // dropped-record count suppressed inside the final throttle window, so the
+  // "how bad was it" diagnostic isn't lost when the server exits mid-window.
+  test('drain flushes the suppressed-failure count at shutdown', async () => {
+    const projectDir = join(tmp, 'sink-drain-flush');
+
+    mkdirSync(dirname(logsPreviousPath(projectDir)), { recursive: true });
+    mkdirSync(logsPreviousPath(projectDir));
+    writeFileSync(join(logsPreviousPath(projectDir), 'occupied'), 'x');
+
+    const emissions: Array<{ event: string; suppressed: number }> = [];
+    const originalError = console.error;
+    console.error = (arg: unknown) => {
+      try {
+        emissions.push(JSON.parse(String(arg)));
+      } catch {
+        /* ignore non-JSON console.error */
+      }
+    };
+    try {
+      const sink = new PinoFileSink({ projectDir, maxBytes: 64 });
+      const line = `${JSON.stringify({ level: 30, msg: 'z'.repeat(200) })}\n`;
+      // Two failing writes in the same throttle window: the first emits
+      // (suppressed: 0), the second is suppressed (counter → 1).
+      await new Promise<void>((resolve) => sink.write(line, () => resolve()));
+      await new Promise<void>((resolve) => sink.write(line, () => resolve()));
+      await sink.drain();
+
+      const dropped = emissions.filter((e) => e.event === 'pino-file-sink-write-dropped');
+      // First window emission + the drain flush.
+      expect(dropped.length).toBe(2);
+      expect(dropped[0]?.suppressed).toBe(0);
+      expect(dropped[1]?.suppressed).toBe(1);
+    } finally {
+      console.error = originalError;
+    }
+  });
 });
 
 describe('ScrubbingSpanProcessor', () => {

packages/server/src/telemetry-file-sink.ts

@@ -41,10 +41,49 @@ export interface RotatingAppenderOpts {
   maxBytes: number;
 }
 
+/** Per-target-path serialization state, shared across every appender instance. */
+interface PathWriteState {
+  chain: Promise<unknown>;
+  parentDirEnsured: boolean;
+}
+
+/**
+ * Serialization state keyed by `currentPath`. Rotation must be serialized per
+ * TARGET FILE, not per appender instance: in production every `getLogger(name)`
+ * allocates its own `PinoFileSink` → its own `RotatingAppender` on the shared
+ * log path, so an instance-private chain let N loggers race the check-then-act
+ * (`statSync` size → `rename`) — the first rename wins and the rest get ENOENT
+ * because `currentPath` no longer exists. Keying the chain by path makes "the
+ * file is never touched from two contexts at once" structural regardless of how
+ * many instances point at it. Entries are process-lifetime but bounded — one
+ * per distinct sink path in the process (a log path, a spans path, and, when
+ * tolerance telemetry is enabled, a tolerance path per project); the map holds
+ * a handful of entries, never grows unboundedly.
+ *
+ * Invariant: all appenders sharing a `currentPath` must also agree on
+ * `previousPath` and `maxBytes` — the shared chain only serializes; each queued
+ * append still applies its own instance's fields, so divergent fields on one
+ * path would make rotation destination/threshold depend on which instance ran.
+ * Every real consumer holds this (one consumer per path, consistent opts). The
+ * key is the raw path string, so callers must also pass a consistent form
+ * (the `logs*Path`/`spans*Path` helpers already produce canonical absolute paths).
+ */
+const pathWriteState = new Map<string, PathWriteState>();
+
+function getPathWriteState(currentPath: string): PathWriteState {
+  let state = pathWriteState.get(currentPath);
+  if (!state) {
+    state = { chain: Promise.resolve(), parentDirEnsured: false };
+    pathWriteState.set(currentPath, state);
+  }
+  return state;
+}
+
 /**
- * Serialized append-with-rotation primitive. Two writers calling
- * `append()` concurrently are serialized through an internal promise
- * chain so the file is never touched from two contexts at once.
+ * Serialized append-with-rotation primitive. Every appender writing to a given
+ * `currentPath` — across all instances — is serialized through one shared
+ * promise chain (keyed by path, see {@link pathWriteState}) so the file is
+ * never touched from two contexts at once.
  *
  * SIGKILL between an append and its rotation can leave at most one
  * partial trailing line in `currentPath`; readers should tolerate it
@@ -55,13 +94,13 @@ export class RotatingAppender {
   readonly #currentPath: string;
   readonly #previousPath: string;
   readonly #maxBytes: number;
-  #writeChain: Promise<unknown> = Promise.resolve();
-  #parentDirEnsured = false;
+  readonly #state: PathWriteState;
 
   constructor(opts: RotatingAppenderOpts) {
     this.#currentPath = opts.currentPath;
     this.#previousPath = opts.previousPath;
     this.#maxBytes = opts.maxBytes;
+    this.#state = getPathWriteState(opts.currentPath);
   }
 
   /**
@@ -70,19 +109,19 @@ export class RotatingAppender {
    * underlying fs error if either step fails.
    */
   append(data: string | Uint8Array): Promise<void> {
-    const next = this.#writeChain
+    const next = this.#state.chain
       // Swallow prior errors so one bad write doesn't deadlock the chain;
       // the rejection still surfaced to that call's awaiter via its own
       // returned promise.
       .catch(() => undefined)
       .then(() => this.#doAppend(data));
-    this.#writeChain = next;
+    this.#state.chain = next;
     return next;
   }
 
   /** Resolve once the most recent enqueued append finishes (success or failure). */
   async drain(): Promise<void> {
-    await this.#writeChain.catch(() => undefined);
+    await this.#state.chain.catch(() => undefined);
   }
 
   async #doAppend(data: string | Uint8Array): Promise<void> {
@@ -96,9 +135,9 @@ export class RotatingAppender {
     // step/tick). Raw fs also spares the log sink a span per line. No
     // application disk writes flow through here, so the fs-traced STOP rule
     // (which governs application-level writes) is not in scope.
-    if (!this.#parentDirEnsured) {
+    if (!this.#state.parentDirEnsured) {
       await mkdir(dirname(this.#currentPath), { recursive: true });
-      this.#parentDirEnsured = true;
+      this.#state.parentDirEnsured = true;
     }
     await writeFile(this.#currentPath, data, { flag: 'a' });
     // Post-write rotation: if current now exceeds the cap, rename to prev.
@@ -110,11 +149,21 @@ export class RotatingAppender {
     } catch {
       // The file was removed externally (e.g., manual cleanup). Drop the
       // ensured flag so the next call recreates the dir tree.
-      this.#parentDirEnsured = false;
+      this.#state.parentDirEnsured = false;
       return;
     }
     if (size > this.#maxBytes) {
-      await rename(this.#currentPath, this.#previousPath);
+      try {
+        await rename(this.#currentPath, this.#previousPath);
+      } catch (err) {
+        // Benign lost-race: `currentPath` was already rotated or removed
+        // out from under us (ENOENT) — there is nothing left to move, so the
+        // rotation is effectively done. The per-path chain makes this
+        // unreachable between our own appenders; this guard covers only
+        // external removal. Re-throw anything else so real faults still
+        // surface to the caller (FileSpanExporter reports them as failures).
+        if ((err as NodeJS.ErrnoException)?.code !== 'ENOENT') throw err;
+      }
     }
   }
 }
@@ -411,6 +460,8 @@ export interface PinoFileSinkOpts {
  */
 export class PinoFileSink extends Writable {
   readonly #appender: RotatingAppender;
+  #lastFailureLogAt = 0;
+  #suppressedFailures = 0;
 
   constructor(opts: PinoFileSinkOpts) {
     // Pino chunks arrive as either Buffer or string depending on internal
@@ -432,12 +483,53 @@ export class PinoFileSink extends Writable {
   ): void {
     this.#appender.append(chunk).then(
       () => callback(),
-      (err: unknown) => callback(err instanceof Error ? err : new Error(String(err))),
+      (err: unknown) => {
+        // Contain the failure: a rejected append MUST NOT propagate to the
+        // Writable as an 'error' event. `pino.multistream` attaches no
+        // 'error' listener, so a surfaced error escalates to
+        // uncaughtException and takes the whole server down — the log sink
+        // must never be able to do that. Report out-of-band, then ack the
+        // write as handled (callback with no error).
+        this.#reportWriteFailure(err);
+        callback();
+      },
     );
   }
 
+  /**
+   * Report a dropped log record, throttled to at most one emission per 5s per
+   * sink INSTANCE (a persistently-failing rotation would otherwise emit one
+   * line per log record). The throttle state is per-instance, not per-path, so
+   * N sinks on the same path can each emit within a window — up to N lines per
+   * 5s per path under a shared failure. That is acceptable for a rare failure
+   * signal. Uses the structured telemetry-event console channel — NOT the pino
+   * logger: this sink IS a pino destination, so routing its own failure
+   * through pino would re-enter the broken sink.
+   */
+  #reportWriteFailure(err: unknown): void {
+    const now = Date.now();
+    if (now - this.#lastFailureLogAt < 5_000) {
+      this.#suppressedFailures++;
+      return;
+    }
+    this.#emitWriteFailure(err instanceof Error ? err.message : String(err), now);
+  }
+
+  #emitWriteFailure(error: string, now: number): void {
+    const suppressed = this.#suppressedFailures;
+    this.#suppressedFailures = 0;
+    this.#lastFailureLogAt = now;
+    console.error(JSON.stringify({ event: 'pino-file-sink-write-dropped', error, suppressed }));
+  }
+
   /** Resolve once any enqueued writes have settled — for tests + shutdown. */
   async drain(): Promise<void> {
     await this.#appender.drain();
+    // Flush any failures suppressed inside the last throttle window so a
+    // graceful shutdown (flushAllFileSinks) doesn't lose the final dropped
+    // count — an operator inspecting stderr should see how bad it got.
+    if (this.#suppressedFailures > 0) {
+      this.#emitWriteFailure('flushed at drain', Date.now());
+    }
   }
 }