This is an automated email from the ASF dual-hosted git repository.
jason810496 pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/main by this push:
new b6cc81c2944 Buffer TS SDK logger output until the log socket connects
(#70085)
b6cc81c2944 is described below
commit b6cc81c2944ce87a56596e464868ed50e14ccb08
Author: Guan-Ming Chiu <[email protected]>
AuthorDate: Thu Jul 23 21:40:12 2026 +0800
Buffer TS SDK logger output until the log socket connects (#70085)
* 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.
* Guard LogChannel.connect() against non-root and repeat calls
Co-authored-by: Jason(Zhe-You) Liu
<[email protected]>
---------
Co-authored-by: Jason(Zhe-You) Liu
<[email protected]>
---
ts-sdk/src/coordinator/log-channel.ts | 62 ++++++++++++++++------
ts-sdk/src/coordinator/runtime.ts | 7 ++-
ts-sdk/tests/coordinator/log-channel.test.ts | 77 +++++++++++++++++++++++-----
3 files changed, 117 insertions(+), 29 deletions(-)
diff --git a/ts-sdk/src/coordinator/log-channel.ts
b/ts-sdk/src/coordinator/log-channel.ts
index dd3d78ea509..9dd85482c01 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;
}
- 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();
});
diff --git a/ts-sdk/src/coordinator/runtime.ts
b/ts-sdk/src/coordinator/runtime.ts
index 5626aae3106..a3a057b4c01 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 ad85c487ce8..592170f10ae 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<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();
