Buffer TS SDK logger output until the log socket connects

ts-sdk/src/coordinator/log-channel.ts

@@ -50,9 +50,10 @@ const DEFAULT_LOGGER_NAME = "ts-sdk";
 const CLOSE_FLUSH_TIMEOUT_MS = 3_000;
 
 interface LogChannelState {
-  sock: Socket;
+  sock: Socket | null;
   connected: boolean;
   closed: boolean;
+  buffer: string[];
 }
 
 export class LogChannel {
@@ -66,22 +67,39 @@ export class LogChannel {
     this.isRoot = isRoot;
   }
 
-  static async connect(addr: string, name: string = DEFAULT_LOGGER_NAME): Promise<LogChannel> {
-    const shared: LogChannelState = {
-      sock: await connectTcp(addr),
-      connected: true,
-      closed: false,
-    };
-    shared.sock.on("error", (err) => {
+  /** Create a root channel with no socket yet; records are buffered
+   *  until {@link LogChannel#connect} flushes them. */
+  static createBuffered(name: string = DEFAULT_LOGGER_NAME): LogChannel {
+    return new LogChannel({ sock: null, connected: false, closed: false, buffer: [] }, name, true);
+  }
+
+  /** Connect the socket and flush buffered records, in order. Root only,
+   *  and callable once: a second call would orphan the first socket. */
+  async connect(addr: string): Promise<void> {
+    if (!this.isRoot) {
+      throw new Error(`[${this.name}] connect() is root-only`);
+    }
+    if (this.shared.sock) {
+      throw new Error(`[${this.name}] connect() called more than once`);
+    }
+    const shared = this.shared;
+    const name = this.name;
+    const sock = await connectTcp(addr);
+    shared.sock = sock;
+    shared.connected = true;
+    sock.on("error", (err) => {
       shared.connected = false;
       process.stderr.write(`[${name}] log socket error: ${err.message}\n`);
     });
-    shared.sock.on("close", () => {
+    sock.on("close", () => {
       if (shared.closed || !shared.connected) return;
       shared.connected = false;
       process.stderr.write(`[${name}] log socket closed unexpectedly; further logs go to stderr\n`);
     });
-    return new LogChannel(shared, name, true);
+    for (const line of shared.buffer) {
+      sock.write(Buffer.from(line, "utf8"));
+    }
+    shared.buffer = [];
   }
 
   /** Create a sibling logger that shares the underlying socket but
@@ -117,11 +135,16 @@ export class LogChannel {
       timestamp: record.timestamp ?? new Date().toISOString(),
     });
     const payload = line + "\n";
-    if (!this.shared.connected || !this.shared.sock.writable) {
+    const sock = this.shared.sock;
+    if (!sock) {
+      this.shared.buffer.push(payload);
+      return;
+    }
+    if (!this.shared.connected || !sock.writable) {
       process.stderr.write(payload);
       return;
     }
-    this.shared.sock.write(Buffer.from(payload, "utf8"));
+    sock.write(Buffer.from(payload, "utf8"));
   }
 
   debug(event: string, args: Record<string, unknown> = {}): void {
@@ -143,17 +166,26 @@ export class LogChannel {
   async close(): Promise<void> {
     if (!this.isRoot) return;
     this.shared.closed = true;
+    const sock = this.shared.sock;
+    if (!sock) {
+      // never connected: surface buffered records instead of dropping them
+      for (const line of this.shared.buffer) {
+        process.stderr.write(line);
+      }
+      this.shared.buffer = [];
+      return;
+    }
     if (!this.shared.connected) {
-      this.shared.sock.destroy();
+      sock.destroy();
       return;
     }
     return new Promise((resolve) => {
       const timer = setTimeout(() => {
-        this.shared.sock.destroy();
+        sock.destroy();
         resolve();
       }, CLOSE_FLUSH_TIMEOUT_MS);
       timer.unref();
-      this.shared.sock.end(() => {
+      sock.end(() => {
         clearTimeout(timer);
         resolve();
       });

ts-sdk/src/coordinator/runtime.ts

@@ -126,12 +126,15 @@ export async function startCoordinator(opts: StartCoordinatorOptions = {}): Prom
   let runtimeAbort: RuntimeAbort | null = null;
 
   try {
-    // Connect log channel first so early failures are captured.
+    // Records emitted before the `--logs` socket connects are buffered and
+    // flushed on connect; if the connect fails, close() dumps them to stderr.
     // Root logger is `ts-sdk`; subsystems use child names (`ts-sdk.runtime`,
     // `ts-sdk.comm`, `ts-sdk.client`) so structlog's ConsoleRenderer prints
     // them as a distinct `[name]` column on the supervisor side.
-    logs = await LogChannel.connect(parsed.logsAddr);
+    logs = LogChannel.createBuffered();
     const runtimeLogs = logs.child("runtime");
+    runtimeLogs.debug("Connecting log socket", { logs_addr: parsed.logsAddr });
+    await logs.connect(parsed.logsAddr);
     const tasks = listRegisteredTasks();
     runtimeLogs.info("Coordinator runtime started", {
       registered_tasks: tasks,

ts-sdk/tests/coordinator/log-channel.test.ts

@@ -43,6 +43,13 @@ async function makeServer(): Promise<Fixture> {
   return { server, port, received, sockClosed };
 }
 
+async function connectChannel(addr: string, name?: string): Promise<LogChannel> {
+  const channel =
+    name === undefined ? LogChannel.createBuffered() : LogChannel.createBuffered(name);
+  await channel.connect(addr);
+  return channel;
+}
+
 function readRecords(received: Buffer[]): Record<string, unknown>[] {
   return Buffer.concat(received)
     .toString("utf8")
@@ -63,7 +70,7 @@ describe("LogChannel", () => {
   });
 
   it("defaults logger name to 'ts-sdk' and auto-stamps timestamp", async () => {
-    const ch = await LogChannel.connect(`127.0.0.1:${fx.port}`);
+    const ch = await connectChannel(`127.0.0.1:${fx.port}`);
     ch.info("hello");
     await ch.close();
     await fx.sockClosed;
@@ -82,7 +89,7 @@ describe("LogChannel", () => {
   });
 
   it("accepts a custom root name", async () => {
-    const ch = await LogChannel.connect(`127.0.0.1:${fx.port}`, "ts-sdk.runtime");
+    const ch = await connectChannel(`127.0.0.1:${fx.port}`, "ts-sdk.runtime");
     ch.warning("started");
     await ch.close();
     await fx.sockClosed;
@@ -97,7 +104,7 @@ describe("LogChannel", () => {
   });
 
   it("child() creates a hierarchical sibling sharing the socket", async () => {
-    const root = await LogChannel.connect(`127.0.0.1:${fx.port}`);
+    const root = await connectChannel(`127.0.0.1:${fx.port}`);
     const comm = root.child("comm");
     const client = root.child("client");
     expect(comm.loggerName).toBe("ts-sdk.comm");
@@ -115,7 +122,7 @@ describe("LogChannel", () => {
   });
 
   it("children must not close the shared socket", async () => {
-    const root = await LogChannel.connect(`127.0.0.1:${fx.port}`);
+    const root = await connectChannel(`127.0.0.1:${fx.port}`);
     const child = root.child("comm");
     // Child.close() is a no-op. Root.close() ends the socket.
     await child.close();
@@ -133,7 +140,7 @@ describe("LogChannel", () => {
 
   it("handles post-connect socket errors on the root channel", async () => {
     const write = vi.spyOn(process.stderr, "write").mockImplementation(() => true);
-    const root = await LogChannel.connect(`127.0.0.1:${fx.port}`);
+    const root = await connectChannel(`127.0.0.1:${fx.port}`);
     root.child("child");
     const sock = (root as unknown as { shared: { sock: net.Socket } }).shared.sock;
 
@@ -148,8 +155,54 @@ describe("LogChannel", () => {
     }
   });
 
+  it("buffers records emitted before connect and flushes them in order on connect", async () => {
+    const root = LogChannel.createBuffered();
+    const child = root.child("runtime");
+    child.debug("connecting", { attempt: 1 });
+    root.info("still starting");
+
+    await root.connect(`127.0.0.1:${fx.port}`);
+    root.info("connected");
+    await root.close();
+    await fx.sockClosed;
+
+    const records = readRecords(fx.received);
+    expect(records.map((r) => r["event"])).toEqual([
+      "[ts-sdk.runtime] connecting",
+      "[ts-sdk] still starting",
+      "[ts-sdk] connected",
+    ]);
+    expect(records[0]).toMatchObject({ attempt: 1, level: "debug" });
+  });
+
+  it("rejects connect() from a child and a second connect() on the root", async () => {
+    const root = await connectChannel(`127.0.0.1:${fx.port}`);
+    try {
+      await expect(root.child("runtime").connect(`127.0.0.1:${fx.port}`)).rejects.toThrow(
+        "root-only",
+      );
+      await expect(root.connect(`127.0.0.1:${fx.port}`)).rejects.toThrow("more than once");
+    } finally {
+      await root.close();
+      await fx.sockClosed;
+    }
+  });
+
+  it("close() before connect writes buffered records to stderr", async () => {
+    const write = vi.spyOn(process.stderr, "write").mockImplementation(() => true);
+    const root = LogChannel.createBuffered();
+    try {
+      root.info("never made it");
+      await root.close();
+      const line = write.mock.calls.find((c) => String(c[0]).includes("never made it"))?.[0];
+      expect(String(line)).toContain('"event":"[ts-sdk] never made it"');
+    } finally {
+      write.mockRestore();
+    }
+  });
+
   it("drops records sent after close instead of writing to the ended socket", async () => {
-    const root = await LogChannel.connect(`127.0.0.1:${fx.port}`);
+    const root = await connectChannel(`127.0.0.1:${fx.port}`);
     const child = root.child("comm");
     root.info("before close");
     await root.close();
@@ -165,7 +218,7 @@ describe("LogChannel", () => {
 
   it("writes only the error message when error and close fire together", async () => {
     const write = vi.spyOn(process.stderr, "write").mockImplementation(() => true);
-    const root = await LogChannel.connect(`127.0.0.1:${fx.port}`);
+    const root = await connectChannel(`127.0.0.1:${fx.port}`);
     const sock = (root as unknown as { shared: { sock: net.Socket } }).shared.sock;
 
     try {
@@ -182,7 +235,7 @@ describe("LogChannel", () => {
 
   it("falls back to stderr when the socket is no longer writable", async () => {
     const write = vi.spyOn(process.stderr, "write").mockImplementation(() => true);
-    const root = await LogChannel.connect(`127.0.0.1:${fx.port}`);
+    const root = await connectChannel(`127.0.0.1:${fx.port}`);
     const sock = (root as unknown as { shared: { sock: net.Socket } }).shared.sock;
 
     try {
@@ -198,7 +251,7 @@ describe("LogChannel", () => {
   });
 
   it("close() resolves via timeout when the flush never completes", async () => {
-    const root = await LogChannel.connect(`127.0.0.1:${fx.port}`);
+    const root = await connectChannel(`127.0.0.1:${fx.port}`);
     const sock = (root as unknown as { shared: { sock: net.Socket } }).shared.sock;
     vi.spyOn(sock, "end").mockImplementation(() => sock);
     const destroy = vi.spyOn(sock, "destroy");
@@ -216,7 +269,7 @@ describe("LogChannel", () => {
 
   it("reports close-time socket errors", async () => {
     const write = vi.spyOn(process.stderr, "write").mockImplementation(() => true);
-    const root = await LogChannel.connect(`127.0.0.1:${fx.port}`);
+    const root = await connectChannel(`127.0.0.1:${fx.port}`);
     const sock = (root as unknown as { shared: { sock: net.Socket } }).shared.sock;
 
     try {
@@ -241,7 +294,7 @@ describe("LogChannel", () => {
     const port = (server.address() as net.AddressInfo).port;
 
     const write = vi.spyOn(process.stderr, "write").mockImplementation(() => true);
-    const root = await LogChannel.connect(`127.0.0.1:${port}`);
+    const root = await connectChannel(`127.0.0.1:${port}`);
     const child = root.child("comm");
     const shared = (root as unknown as { shared: { connected: boolean } }).shared;
     try {
@@ -279,7 +332,7 @@ describe("LogChannel", () => {
     const port = (server.address() as net.AddressInfo).port;
 
     const write = vi.spyOn(process.stderr, "write").mockImplementation(() => true);
-    const root = await LogChannel.connect(`127.0.0.1:${port}`);
+    const root = await connectChannel(`127.0.0.1:${port}`);
     try {
       await vi.waitFor(() => expect(serverSocks).toHaveLength(1));
       serverSocks[0]!.destroy();