From 9d2e52431a7e4506fdd16793e9a5f8105e0a816f Mon Sep 17 00:00:00 2001 From: "Guan-Ming (Wesley) Chiu" <105915352+guan404ming@users.noreply.github.com> Date: Sun, 19 Jul 2026 10:24:14 +0800 Subject: [PATCH 1/2] Buffer TS SDK logger output until the log socket connects Records emitted between subprocess start and the --logs socket connecting were silently dropped; the Go and Java SDKs already buffer this window and flush once connected. --- ts-sdk/src/coordinator/log-channel.ts | 60 +++++++++++++++----- ts-sdk/src/coordinator/runtime.ts | 7 ++- ts-sdk/tests/coordinator/log-channel.test.ts | 33 +++++++++++ 3 files changed, 84 insertions(+), 16 deletions(-) diff --git a/ts-sdk/src/coordinator/log-channel.ts b/ts-sdk/src/coordinator/log-channel.ts index dd3d78ea50969..59830a36ca995 100644 --- a/ts-sdk/src/coordinator/log-channel.ts +++ b/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; } + /** 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); + } + static async connect(addr: string, name: string = DEFAULT_LOGGER_NAME): Promise { - const shared: LogChannelState = { - sock: await connectTcp(addr), - connected: true, - closed: false, - }; - shared.sock.on("error", (err) => { + const channel = LogChannel.createBuffered(name); + await channel.connect(addr); + return channel; + } + + /** Connect the socket and flush buffered records, in order. Root only. */ + async connect(addr: string): Promise { + if (!this.isRoot) return; + 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 = {}): void { @@ -143,17 +166,26 @@ export class LogChannel { async close(): Promise { 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(); }); diff --git a/ts-sdk/src/coordinator/runtime.ts b/ts-sdk/src/coordinator/runtime.ts index 5626aae310619..a3a057b4c0176 100644 --- a/ts-sdk/src/coordinator/runtime.ts +++ b/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, diff --git a/ts-sdk/tests/coordinator/log-channel.test.ts b/ts-sdk/tests/coordinator/log-channel.test.ts index ad85c487ce8d1..85f6720c8ef4e 100644 --- a/ts-sdk/tests/coordinator/log-channel.test.ts +++ b/ts-sdk/tests/coordinator/log-channel.test.ts @@ -148,6 +148,39 @@ 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("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 child = root.child("comm"); From 5ed85dde655316ccb56309b57fe44175a134642e Mon Sep 17 00:00:00 2001 From: "Guan-Ming (Wesley) Chiu" <105915352+guan404ming@users.noreply.github.com> Date: Thu, 23 Jul 2026 20:20:42 +0800 Subject: [PATCH 2/2] Guard LogChannel.connect() against non-root and repeat calls Co-authored-by: Jason(Zhe-You) Liu <68415893+jason810496@users.noreply.github.com> --- ts-sdk/src/coordinator/log-channel.ts | 16 +++---- ts-sdk/tests/coordinator/log-channel.test.ts | 44 ++++++++++++++------ 2 files changed, 40 insertions(+), 20 deletions(-) diff --git a/ts-sdk/src/coordinator/log-channel.ts b/ts-sdk/src/coordinator/log-channel.ts index 59830a36ca995..9dd85482c01c9 100644 --- a/ts-sdk/src/coordinator/log-channel.ts +++ b/ts-sdk/src/coordinator/log-channel.ts @@ -73,15 +73,15 @@ export class LogChannel { return new LogChannel({ sock: null, connected: false, closed: false, buffer: [] }, name, true); } - static async connect(addr: string, name: string = DEFAULT_LOGGER_NAME): Promise { - const channel = LogChannel.createBuffered(name); - await channel.connect(addr); - return channel; - } - - /** Connect the socket and flush buffered records, in order. Root only. */ + /** 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 { - if (!this.isRoot) return; + 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); diff --git a/ts-sdk/tests/coordinator/log-channel.test.ts b/ts-sdk/tests/coordinator/log-channel.test.ts index 85f6720c8ef4e..592170f10aed8 100644 --- a/ts-sdk/tests/coordinator/log-channel.test.ts +++ b/ts-sdk/tests/coordinator/log-channel.test.ts @@ -43,6 +43,13 @@ async function makeServer(): Promise { return { server, port, received, sockClosed }; } +async function connectChannel(addr: string, name?: string): Promise { + const channel = + name === undefined ? LogChannel.createBuffered() : LogChannel.createBuffered(name); + await channel.connect(addr); + return channel; +} + function readRecords(received: Buffer[]): Record[] { 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; @@ -168,6 +175,19 @@ describe("LogChannel", () => { 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(); @@ -182,7 +202,7 @@ describe("LogChannel", () => { }); 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(); @@ -198,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 { @@ -215,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 { @@ -231,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"); @@ -249,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 { @@ -274,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 { @@ -312,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();