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 372025f2549 Send TS SDK logs to stderr on socket disconnect (#69302)
372025f2549 is described below
commit 372025f2549782f759c0eb8a1723bf732f70d1ba
Author: Guan-Ming Chiu <[email protected]>
AuthorDate: Sun Jul 19 15:30:12 2026 +0800
Send TS SDK logs to stderr on socket disconnect (#69302)
* Send TypeScript SDK logs to stderr when the log socket disconnects
Co-authored-by: Jason(Zhe-You) Liu
<[email protected]>
* Harden TS SDK log channel shutdown
* Unref log channel close flush timer so it cannot delay process exit
---------
Co-authored-by: Jason(Zhe-You) Liu
<[email protected]>
---
ts-sdk/src/coordinator/log-channel.ts | 61 +++++++++++---
ts-sdk/tests/coordinator/log-channel.test.ts | 121 ++++++++++++++++++++++++++-
2 files changed, 167 insertions(+), 15 deletions(-)
diff --git a/ts-sdk/src/coordinator/log-channel.ts
b/ts-sdk/src/coordinator/log-channel.ts
index e64737e3b67..dd3d78ea509 100644
--- a/ts-sdk/src/coordinator/log-channel.ts
+++ b/ts-sdk/src/coordinator/log-channel.ts
@@ -47,31 +47,48 @@ export interface LogRecord {
const DEFAULT_LOGGER_NAME = "ts-sdk";
+const CLOSE_FLUSH_TIMEOUT_MS = 3_000;
+
+interface LogChannelState {
+ sock: Socket;
+ connected: boolean;
+ closed: boolean;
+}
+
export class LogChannel {
- private readonly sock: Socket;
+ private readonly shared: LogChannelState;
private readonly name: string;
private readonly isRoot: boolean;
- private constructor(sock: Socket, name: string, isRoot: boolean) {
- this.sock = sock;
+ private constructor(shared: LogChannelState, name: string, isRoot: boolean) {
+ this.shared = shared;
this.name = name;
this.isRoot = isRoot;
- if (isRoot) {
- sock.on("error", (err) => {
- process.stderr.write(`[${this.name}] log socket error:
${err.message}\n`);
- });
- }
}
static async connect(addr: string, name: string = DEFAULT_LOGGER_NAME):
Promise<LogChannel> {
- return new LogChannel(await connectTcp(addr), name, true);
+ const shared: LogChannelState = {
+ sock: await connectTcp(addr),
+ connected: true,
+ closed: false,
+ };
+ shared.sock.on("error", (err) => {
+ shared.connected = false;
+ process.stderr.write(`[${name}] log socket error: ${err.message}\n`);
+ });
+ shared.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);
}
/** Create a sibling logger that shares the underlying socket but
* carries a hierarchical name (`parent.suffix`). Only the root
* owns the socket — children's `close()` is a no-op. */
child(suffix: string): LogChannel {
- return new LogChannel(this.sock, `${this.name}.${suffix}`, false);
+ return new LogChannel(this.shared, `${this.name}.${suffix}`, false);
}
/** Name reported in the `logger` field of every record this
@@ -87,7 +104,7 @@ export class LogChannel {
},
): void {
// Drop late records after the log socket has closed.
- if (this.sock.writableEnded) return;
+ if (this.shared.closed) return;
// Prepend the logger name to the event message so it surfaces in
// the Airflow UI task log view, which renders the message text but
// hides the `logger` JSON field. The field is still emitted for
@@ -99,7 +116,12 @@ export class LogChannel {
event: `[${this.name}] ${record.event}`,
timestamp: record.timestamp ?? new Date().toISOString(),
});
- this.sock.write(Buffer.from(line + "\n", "utf8"));
+ const payload = line + "\n";
+ if (!this.shared.connected || !this.shared.sock.writable) {
+ process.stderr.write(payload);
+ return;
+ }
+ this.shared.sock.write(Buffer.from(payload, "utf8"));
}
debug(event: string, args: Record<string, unknown> = {}): void {
@@ -120,8 +142,21 @@ export class LogChannel {
async close(): Promise<void> {
if (!this.isRoot) return;
+ this.shared.closed = true;
+ if (!this.shared.connected) {
+ this.shared.sock.destroy();
+ return;
+ }
return new Promise((resolve) => {
- this.sock.end(() => resolve());
+ const timer = setTimeout(() => {
+ this.shared.sock.destroy();
+ resolve();
+ }, CLOSE_FLUSH_TIMEOUT_MS);
+ timer.unref();
+ this.shared.sock.end(() => {
+ clearTimeout(timer);
+ resolve();
+ });
});
}
}
diff --git a/ts-sdk/tests/coordinator/log-channel.test.ts
b/ts-sdk/tests/coordinator/log-channel.test.ts
index 2b8a79579dc..ad85c487ce8 100644
--- a/ts-sdk/tests/coordinator/log-channel.test.ts
+++ b/ts-sdk/tests/coordinator/log-channel.test.ts
@@ -135,7 +135,7 @@ describe("LogChannel", () => {
const write = vi.spyOn(process.stderr, "write").mockImplementation(() =>
true);
const root = await LogChannel.connect(`127.0.0.1:${fx.port}`);
root.child("child");
- const sock = (root as unknown as { sock: net.Socket }).sock;
+ const sock = (root as unknown as { shared: { sock: net.Socket }
}).shared.sock;
try {
sock.emit("error", new Error("boom"));
@@ -163,10 +163,61 @@ describe("LogChannel", () => {
expect(records[0]).toMatchObject({ event: "[ts-sdk] before close" });
});
+ 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 sock = (root as unknown as { shared: { sock: net.Socket }
}).shared.sock;
+
+ try {
+ sock.emit("error", Object.assign(new Error("write EPIPE"), { code:
"EPIPE" }));
+ sock.destroy();
+ await fx.sockClosed;
+ expect(write).toHaveBeenCalledTimes(1);
+ expect(write.mock.calls[0]?.[0]).toBe("[ts-sdk] log socket error: write
EPIPE\n");
+ } finally {
+ write.mockRestore();
+ await root.close();
+ }
+ });
+
+ 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 sock = (root as unknown as { shared: { sock: net.Socket }
}).shared.sock;
+
+ try {
+ sock.end();
+ root.info("peer died");
+ const line = write.mock.calls.find((c) => String(c[0]).includes("peer
died"))?.[0];
+ expect(String(line)).toContain('"event":"[ts-sdk] peer died"');
+ } finally {
+ write.mockRestore();
+ await root.close();
+ await fx.sockClosed;
+ }
+ });
+
+ it("close() resolves via timeout when the flush never completes", async ()
=> {
+ const root = await LogChannel.connect(`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");
+
+ vi.useFakeTimers();
+ try {
+ const closed = root.close();
+ await vi.advanceTimersByTimeAsync(3_000);
+ await closed;
+ expect(destroy).toHaveBeenCalled();
+ } finally {
+ vi.useRealTimers();
+ }
+ });
+
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 sock = (root as unknown as { sock: net.Socket }).sock;
+ const sock = (root as unknown as { shared: { sock: net.Socket }
}).shared.sock;
try {
sock.emit("error", Object.assign(new Error("write EPIPE"), { code:
"EPIPE" }));
@@ -178,4 +229,70 @@ describe("LogChannel", () => {
await fx.sockClosed;
}
});
+
+ it("warns once and falls back to stderr after an unexpected disconnect",
async () => {
+ const serverSocks: net.Socket[] = [];
+ const received: Buffer[] = [];
+ const server = net.createServer((sock) => {
+ serverSocks.push(sock);
+ sock.on("data", (chunk) => received.push(chunk));
+ });
+ await new Promise<void>((resolve) => server.listen(0, "127.0.0.1",
resolve));
+ 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 child = root.child("comm");
+ const shared = (root as unknown as { shared: { connected: boolean }
}).shared;
+ try {
+ root.info("before disconnect");
+ await vi.waitFor(() => expect(readRecords(received)).toHaveLength(1));
+
+ serverSocks[0]!.destroy();
+ await vi.waitFor(() => expect(shared.connected).toBe(false));
+ expect(write).toHaveBeenCalledWith(
+ "[ts-sdk] log socket closed unexpectedly; further logs go to stderr\n",
+ );
+
+ root.info("while disconnected");
+ child.debug("child while disconnected");
+ const stderrRecords = write.mock.calls
+ .map((call) => String(call[0]))
+ .filter((line) => line.startsWith("{"))
+ .map((line) => JSON.parse(line) as Record<string, unknown>);
+ expect(stderrRecords.map((r) => r["event"])).toEqual([
+ "[ts-sdk] while disconnected",
+ "[ts-sdk.comm] child while disconnected",
+ ]);
+ expect(readRecords(received)).toHaveLength(1);
+ } finally {
+ write.mockRestore();
+ await root.close();
+ server.close();
+ }
+ });
+
+ it("close() while disconnected resolves and drops later records", async ()
=> {
+ const serverSocks: net.Socket[] = [];
+ const server = net.createServer((sock) => serverSocks.push(sock));
+ await new Promise<void>((resolve) => server.listen(0, "127.0.0.1",
resolve));
+ 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}`);
+ try {
+ await vi.waitFor(() => expect(serverSocks).toHaveLength(1));
+ serverSocks[0]!.destroy();
+ const shared = (root as unknown as { shared: { connected: boolean }
}).shared;
+ await vi.waitFor(() => expect(shared.connected).toBe(false));
+
+ await root.close();
+ write.mockClear();
+ root.info("dropped after close");
+ expect(write).not.toHaveBeenCalled();
+ } finally {
+ write.mockRestore();
+ server.close();
+ }
+ });
});
