fix(mongodb-runner): DRIVERS-3673 keep server logs after mongodb-runner start exits

packages/mongodb-runner/src/cli.spec.ts

@@ -70,6 +70,63 @@ describe('cli', function () {
     await runCli(['prune']);
   });
 
+  it('keeps writing server logs after `start` has exited', async function () {
+    const logDir = path.join(tmpDir, 'logs-after-start');
+    const runnerDir = path.join(tmpDir, 'runner-logs-after-start');
+    const stdout = await runCli([
+      'start',
+      '--topology',
+      'standalone',
+      '--runnerDir',
+      runnerDir,
+      '--logDir',
+      logDir,
+    ]);
+    try {
+      const [logFile] = await fs.readdir(logDir);
+      const logPath = path.join(logDir, logFile);
+      const sizeAtExit = (await fs.stat(logPath)).size;
+      expect(sizeAtExit, 'startup output is in the log').to.be.greaterThan(0);
+
+      // Accepting a connection is logged, and the runner is no longer around
+      // to pipe it into the file.
+      const client = new MongoClient(stdout.trim());
+      await client.db('admin').command({ ping: 1 });
+      await client.close();
+
+      let sizeNow = sizeAtExit;
+      for (let i = 0; i < 50 && sizeNow === sizeAtExit; i++) {
+        await new Promise((resolve) => setTimeout(resolve, 100));
+        sizeNow = (await fs.stat(logPath)).size;
+      }
+      expect(sizeNow, 'log grew after start exited').to.be.greaterThan(
+        sizeAtExit,
+      );
+    } finally {
+      await runCli(['stop', '--all', '--runnerDir', runnerDir]);
+    }
+  });
+
+  it('reports a failed server startup when --logDir is set', async function () {
+    // The failure is timing-dependent (the process used to exit 0 silently
+    // when the server died between log polls), so repeat it.
+    for (let i = 0; i < 5; i++) {
+      const stderr = await runCliExpectingFailure([
+        'start',
+        '--topology',
+        'standalone',
+        '--runnerDir',
+        path.join(tmpDir, `runner-failed-start-${i}`),
+        '--logDir',
+        path.join(tmpDir, `logs-failed-start-${i}`),
+        '--',
+        '--setParameter',
+        'nonexistentParam=1',
+      ]);
+      expect(stderr, `attempt ${i}`).to.include('Server failed to start');
+    }
+  });
+
   it('can execute against a cluster', async function () {
     const stdout = await runCli([
       'exec',

packages/mongodb-runner/src/mongologreader.spec.ts

@@ -0,0 +1,25 @@
+import { expect } from 'chai';
+import { Readable } from 'stream';
+import { createLogEntryIterator } from './mongologreader';
+
+describe('createLogEntryIterator', function () {
+  it('skips lines that are not log entries', async function () {
+    const json = JSON.stringify({
+      t: { $date: '2024-01-01T00:00:00.000Z' },
+      s: 'I',
+      c: 'NETWORK',
+      id: 23016,
+      ctx: 'listener',
+      msg: 'Waiting for connections',
+      attr: { port: 1234 },
+    });
+    const input = Readable.from([
+      `Unable to find image 'mongo' locally\n${json}\n`,
+    ]);
+    const entries = [];
+    for await (const entry of createLogEntryIterator(input)) {
+      entries.push(entry);
+    }
+    expect(entries.map((e) => e.id)).to.deep.equal([23016]);
+  });
+});

packages/mongodb-runner/src/mongologreader.ts

@@ -13,7 +13,10 @@ export async function* createLogEntryIterator(
       const logEntry = parseAnyLogEntry(line);
       yield logEntry;
     } catch (error: any) {
-      break;
+      // The server's stderr shares the log file when `logDir` is set (as does
+      // docker's, in docker mode), and it can contain non-log lines. Skipping
+      // them is safe: the stream still ends when the process exits.
+      continue;
     }
   }
 }

packages/mongodb-runner/src/mongoserver.ts

@@ -1,6 +1,6 @@
 import type { ChildProcess } from 'child_process';
 import { spawn } from 'child_process';
-import { promises as fs, createWriteStream } from 'fs';
+import { promises as fs } from 'fs';
 import type { LogEntry } from './mongologreader';
 import {
   createLogEntryIterator,
@@ -9,6 +9,8 @@ import {
   isFailureToSetupListener,
 } from './mongologreader';
 import { Readable } from 'stream';
+import type { FileTailStream } from './tail-file';
+import { tailFileUntilExit } from './tail-file';
 import type { Document, MongoClientOptions } from 'mongodb';
 import { MongoClient, MongoServerError } from 'mongodb';
 import path from 'path';
@@ -35,7 +37,7 @@ export interface MongoServerOptions {
   binary: string;
   /** Directory for temporary files, e.g., database contents */
   tmpDir: string;
-  /** If set, log file output will be piped through here. */
+  /** If set, each server writes its log to a file in this directory. */
   logDir?: string;
   /** Arguments to pass to the MongoDB binary. May or may not contain --port */
   args?: string[];
@@ -112,6 +114,7 @@ export class MongoServer extends EventEmitter<MongoServerEvents> {
   public uuid: string = uuid();
   private buildInfo?: Document;
   private childProcess?: ChildProcess;
+  private logTail?: FileTailStream;
   private pid?: number;
   private host = '127.0.0.1';
   private port?: number;
@@ -314,40 +317,48 @@ export class MongoServer extends EventEmitter<MongoServerEvents> {
     debug('starting server', commandline);
     const [executable, ...args] = commandline;
     srv.commandline = commandline;
-    const proc = spawn(executable, args, {
-      stdio: ['inherit', 'pipe', 'pipe'],
-      cwd: options.tmpDir,
-      detached: options.detached ?? true,
-    });
-    await once(proc, 'spawn');
-    srv.childProcess = proc;
-    srv.pid = proc.pid;
-
-    const { stdout, stderr } = proc;
-    stdout.setEncoding('utf8');
-    stderr.setEncoding('utf8');
-
+    let stdout: Readable;
     if (options.logDir) {
+      // The server writes straight to the file rather than through a pipe to
+      // this process, so it keeps logging after we exit (`start` exits once the
+      // cluster is up). We follow the file to see the startup output instead.
+      await fs.mkdir(options.logDir, { recursive: true });
       const outFile = path.join(
         options.logDir,
-        `${options.binary}-${String(proc.pid)}-${new Date()
+        `${options.binary}-${srv.uuid}-${new Date()
           .toISOString()
           .replace(/[^-_a-zA-Z0-9.]/g, '')}.log`,
       );
-      await fs.mkdir(options.logDir, { recursive: true });
-      const outStream = createWriteStream(outFile);
-      await once(outStream, 'open');
-      stdout.pipe(outStream, { end: false });
-      stderr.pipe(outStream, { end: false });
-      Promise.allSettled([once(stdout, 'end'), once(stderr, 'end')]).then(
-        () => outStream.end(),
-        () => {
-          /* cannot throw */
-        },
-      );
+      const handle = await fs.open(outFile, 'a');
+      let proc: ChildProcess;
+      try {
+        proc = spawn(executable, args, {
+          stdio: ['inherit', handle.fd, handle.fd],
+          cwd: options.tmpDir,
+          detached: options.detached ?? true,
+        });
+        await once(proc, 'spawn');
+      } finally {
+        await handle.close(); // the child process holds its own copy
+      }
+      srv.childProcess = proc;
+      srv.pid = proc.pid;
+      srv.logTail = tailFileUntilExit(outFile, proc);
+      stdout = srv.logTail;
     } else {
-      stderr.on('data', (chunk) => debug('server stderr', chunk));
-      stderr.resume();
+      const proc = spawn(executable, args, {
+        stdio: ['inherit', 'pipe', 'pipe'],
+        cwd: options.tmpDir,
+        detached: options.detached ?? true,
+      });
+      await once(proc, 'spawn');
+      srv.childProcess = proc;
+      srv.pid = proc.pid;
+      stdout = proc.stdout;
+      stdout.setEncoding('utf8');
+      proc.stderr.setEncoding('utf8');
+      proc.stderr.on('data', (chunk) => debug('server stderr', chunk));
+      proc.stderr.resume();
       stdout.resume();
     }
 
@@ -643,12 +654,14 @@ export class MongoServer extends EventEmitter<MongoServerEvents> {
 
   ref(): void {
     this.childProcess?.ref();
+    this.logTail?.ref();
     (this.childProcess?.stdout as any)?.ref();
     (this.childProcess?.stderr as any)?.ref();
   }
 
   unref(): void {
     this.childProcess?.unref();
+    this.logTail?.unref();
     (this.childProcess?.stdout as any)?.unref();
     (this.childProcess?.stderr as any)?.unref();
   }

packages/mongodb-runner/src/tail-file.spec.ts

@@ -0,0 +1,64 @@
+import { expect } from 'chai';
+import { promises as fs } from 'fs';
+import os from 'os';
+import path from 'path';
+import { EventEmitter } from 'events';
+import type { ChildProcess } from 'child_process';
+import { tailFileUntilExit } from './tail-file';
+
+describe('tailFileUntilExit', function () {
+  let tmpDir: string;
+  let file: string;
+
+  beforeEach(async function () {
+    tmpDir = await fs.mkdtemp(path.join(os.tmpdir(), 'tail-file-'));
+    file = path.join(tmpDir, 'out.log');
+    await fs.writeFile(file, '');
+  });
+
+  afterEach(async function () {
+    await fs.rm(tmpDir, { recursive: true, force: true });
+  });
+
+  function fakeProc(): ChildProcess & EventEmitter {
+    return Object.assign(new EventEmitter(), {
+      exitCode: null,
+      signalCode: null,
+    }) as unknown as ChildProcess & EventEmitter;
+  }
+
+  it('yields existing and later-appended data, and ends after exit', async function () {
+    const proc = fakeProc();
+    await fs.writeFile(file, 'one\n');
+    const stream = tailFileUntilExit(file, proc);
+
+    const chunks: string[] = [];
+    const done = (async () => {
+      for await (const chunk of stream) chunks.push(chunk as string);
+    })();
+
+    await new Promise((resolve) => setTimeout(resolve, 100));
+    await fs.appendFile(file, 'two\n');
+    await new Promise((resolve) => setTimeout(resolve, 100));
+    // Written right before exit: must still be delivered.
+    await fs.appendFile(file, 'three\n');
+    proc.emit('exit', 0, null);
+    await done;
+
+    expect(chunks.join(''), 'all data is read before ending').to.equal(
+      'one\ntwo\nthree\n',
+    );
+  });
+
+  it('ends immediately for an already-exited process once drained', async function () {
+    const proc = fakeProc();
+    (proc as any).exitCode = 1;
+    await fs.writeFile(file, 'bye\n');
+
+    let data = '';
+    for await (const chunk of tailFileUntilExit(file, proc)) {
+      data += chunk as string;
+    }
+    expect(data, 'data is still delivered').to.equal('bye\n');
+  });
+});

packages/mongodb-runner/src/tail-file.ts

@@ -0,0 +1,104 @@
+import type { ChildProcess } from 'child_process';
+import fs from 'fs';
+import { Readable } from 'stream';
+
+const POLL_INTERVAL_MS = 25;
+const READ_SIZE = 64 * 1024;
+
+/**
+ * A readable stream of the (utf8) contents of a file that another process is
+ * appending to. It ends once `isDone()` returns true and the file has been
+ * read to its end.
+ *
+ * Like a child process's stdio pipes, the stream keeps the event loop alive
+ * until `unref()` is called. That is needed while waiting for a server to
+ * start: if the server exits, nothing else would keep us around to notice.
+ */
+export class FileTailStream extends Readable {
+  private fd?: number;
+  private timer?: NodeJS.Timeout;
+  private polling = false;
+  private refed = true;
+
+  constructor(
+    private readonly filePath: string,
+    private readonly isDone: () => boolean,
+  ) {
+    super({ encoding: 'utf8' });
+  }
+
+  override _construct(callback: (err?: Error | null) => void): void {
+    fs.open(this.filePath, 'r', (err, fd) => {
+      this.fd = fd;
+      callback(err);
+    });
+  }
+
+  override _read(): void {
+    if (this.polling) return;
+    this.polling = true;
+    this.poll();
+  }
+
+  ref(): void {
+    this.refed = true;
+    this.timer?.ref();
+  }
+
+  unref(): void {
+    this.refed = false;
+    this.timer?.unref();
+  }
+
+  override _destroy(
+    err: Error | null,
+    callback: (err?: Error | null) => void,
+  ): void {
+    clearTimeout(this.timer);
+    if (this.fd === undefined) return callback(err);
+    fs.close(this.fd, () => callback(err));
+  }
+
+  private poll(): void {
+    // Sample this before reading: anything written before the process exited
+    // is then guaranteed to be visible to the read below.
+    const done = this.isDone();
+    fs.read(
+      this.fd as number,
+      Buffer.allocUnsafe(READ_SIZE),
+      0,
+      READ_SIZE,
+      null,
+      (err, bytesRead, buf) => {
+        if (this.destroyed) return;
+        if (err) return void this.destroy(err);
+        if (bytesRead > 0) {
+          if (this.push(buf.subarray(0, bytesRead))) return this.poll();
+          this.polling = false;
+        } else if (done) {
+          this.push(null);
+        } else {
+          this.timer = setTimeout(() => this.poll(), POLL_INTERVAL_MS);
+          if (!this.refed) this.timer.unref();
+        }
+      },
+    );
+  }
+}
+
+/**
+ * Follow `filePath` until `proc` has exited and all of its output has been read.
+ *
+ * This lets the process write straight to a file (so it keeps logging after the
+ * current process is gone) while we still observe its output while we are alive.
+ */
+export function tailFileUntilExit(
+  filePath: string,
+  proc: ChildProcess,
+): FileTailStream {
+  let exited = proc.exitCode !== null || proc.signalCode !== null;
+  proc.once('exit', () => {
+    exited = true;
+  });
+  return new FileTailStream(filePath, () => exited);
+}