From 0874aca749cd4b92834762dd0f1c34e6d54e114d Mon Sep 17 00:00:00 2001 From: Aikins Laryea Date: Tue, 27 Oct 2026 13:00:32 +0000 Subject: [PATCH] fix(sdk-pi): write the error log without blocking pi The default onError ran statSync, renameSync, and appendFileSync on pi's event loop, so a slow agent directory, such as a network mount set through PI_CODING_AGENT_DIR, held up pi's TUI on every reported error. Entries now queue on one promise chain that rotates and appends with fs/promises, in call order. New entries are dropped while 100 are pending. session_shutdown waits up to one second for the queue, so errors that the final flush reports still land before pi exits, and a stalled disk cannot keep pi from shutting down. Closes #90 --- packages/pi/README.md | 3 ++ packages/pi/src/error-log.ts | 61 +++++++++++++++++++++++++---- packages/pi/src/extension.ts | 5 ++- packages/pi/tests/error-log.test.ts | 35 +++++++++++++++++ packages/pi/tests/pi.test.ts | 33 +++++++++++++++- 5 files changed, 128 insertions(+), 9 deletions(-) create mode 100644 packages/pi/tests/error-log.test.ts diff --git a/packages/pi/README.md b/packages/pi/README.md index 1ce1f83..7eb144f 100644 --- a/packages/pi/README.md +++ b/packages/pi/README.md @@ -62,6 +62,9 @@ extension errors to `telemetry-dev.log` in the pi agent directory: `~/.pi/agent` `PI_CODING_AGENT_DIR` when you set it. The file has mode `0600`. When the file is more than 5 MB, the integration moves it to `telemetry-dev.log.1`. +Writes happen in the background, so a slow disk does not hold up pi. When 100 entries are already +waiting, later ones are dropped. `session_shutdown` waits up to one second for the queued entries. + A series of failed exports, for example during an ingest outage, is one entry. The next failure after an export reaches the ingest is a new entry. diff --git a/packages/pi/src/error-log.ts b/packages/pi/src/error-log.ts index 203e368..0142004 100644 --- a/packages/pi/src/error-log.ts +++ b/packages/pi/src/error-log.ts @@ -1,4 +1,4 @@ -import { appendFileSync, renameSync, statSync } from "node:fs"; +import { appendFile, rename, stat } from "node:fs/promises"; import { homedir } from "node:os"; import { join } from "node:path"; import { inspect } from "node:util"; @@ -6,20 +6,67 @@ import { inspect } from "node:util"; /** pi moves its own `mcp.log` to `mcp.log.1` when the file is more than 5 MB. */ const MAX_LOG_BYTES = 5 * 1024 * 1024; +/** Entries waiting for the disk; new entries are dropped while this many are pending. */ +const MAX_PENDING_ENTRIES = 100; + +/** How long `flushErrorLog` waits for a stalled disk before giving up. */ +const FLUSH_DEADLINE_MS = 1000; + +let pending = 0; +let queue: Promise = Promise.resolve(); + /** - * Appends an error to `telemetry-dev.log` in pi's agent directory. pi draws its TUI on the - * terminal, so an error written to stderr garbles the screen and is not kept. + * Queues an error for `telemetry-dev.log` in pi's agent directory and returns without waiting + * on the disk. pi draws its TUI on the terminal, so an error written to stderr garbles the + * screen and is not kept. */ export function writeErrorLog(error: unknown): void { + if (pending >= MAX_PENDING_ENTRIES) return; + const path = join(agentDir(), "telemetry-dev.log"); const entry = `${new Date().toISOString()} ${inspect(error).replaceAll("\n", "\n ")}\n`; + pending += 1; + queue = queue + .then(() => append(path, entry)) + .finally(() => { + pending -= 1; + }); +} + +/** Resolves once every queued entry has been written or has failed. Never rejects. */ +export function drainErrorLogForTesting(): Promise { + return queue; +} + +/** + * Resolves once every queued entry has been written or has failed, or after `FLUSH_DEADLINE_MS` + * if a write stalls. Never rejects. + */ +export async function flushErrorLog(): Promise { + let timer: ReturnType | undefined; + const deadline = new Promise((resolve) => { + timer = setTimeout(resolve, FLUSH_DEADLINE_MS); + timer.unref?.(); + }); + + try { + await Promise.race([queue, deadline]); + } finally { + clearTimeout(timer); + } +} + +async function append(path: string, entry: string): Promise { try { - if ((statSync(path, { throwIfNoEntry: false })?.size ?? 0) > MAX_LOG_BYTES) { - renameSync(path, `${path}.1`); - } + const size = await stat(path).then( + (stats) => stats.size, + () => 0, + ); + + if (size > MAX_LOG_BYTES) await rename(path, `${path}.1`); - appendFileSync(path, entry, { mode: 0o600 }); + await appendFile(path, entry, { mode: 0o600 }); } catch { // The log is best effort; telemetry must never fail a pi session. } diff --git a/packages/pi/src/extension.ts b/packages/pi/src/extension.ts index 23df768..9216863 100644 --- a/packages/pi/src/extension.ts +++ b/packages/pi/src/extension.ts @@ -10,7 +10,7 @@ import { } from "@telemetry-dev/sdk"; import { ensureInit, type ClientOverrides, type TelemetryDevPiOptions } from "./config.ts"; -import { writeErrorLog } from "./error-log.ts"; +import { flushErrorLog, writeErrorLog } from "./error-log.ts"; /** Options for {@link telemetryDevExtension}. */ export interface TelemetryDevExtensionOptions extends TelemetryDevPiOptions { @@ -454,6 +454,9 @@ export function telemetryDevExtension( } catch (error) { reportError(onError, error); } + + // pi can exit right after shutdown; land the errors that flush just reported. + await flushErrorLog(); }, ); }; diff --git a/packages/pi/tests/error-log.test.ts b/packages/pi/tests/error-log.test.ts new file mode 100644 index 0000000..97e5e6f --- /dev/null +++ b/packages/pi/tests/error-log.test.ts @@ -0,0 +1,35 @@ +import { mkdtemp, rm } from "node:fs/promises"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; + +import { expect, test, vi } from "vitest"; + +import { flushErrorLog, writeErrorLog } from "../src/error-log.ts"; + +const appendFile = vi.hoisted(() => vi.fn(() => new Promise(() => {}))); + +// A separate file, so the stalled write below cannot hold up the queue in other tests. +vi.mock("node:fs/promises", async (importOriginal) => ({ + ...(await importOriginal()), + appendFile, +})); + +test("flushErrorLog waits for a stalled write until its deadline, then gives up", async () => { + const directory = await mkdtemp(join(tmpdir(), "telemetry-pi-agent-")); + vi.stubEnv("PI_CODING_AGENT_DIR", directory); + + try { + writeErrorLog(new Error("stalled")); + await vi.waitFor(() => expect(appendFile).toHaveBeenCalledOnce()); + + const started = performance.now(); + await flushErrorLog(); + const elapsed = performance.now() - started; + + expect(elapsed).toBeGreaterThanOrEqual(900); + expect(elapsed).toBeLessThan(3000); + } finally { + vi.unstubAllEnvs(); + await rm(directory, { recursive: true, force: true }); + } +}); diff --git a/packages/pi/tests/pi.test.ts b/packages/pi/tests/pi.test.ts index c710e6f..b857bef 100644 --- a/packages/pi/tests/pi.test.ts +++ b/packages/pi/tests/pi.test.ts @@ -1,3 +1,4 @@ +import { existsSync } from "node:fs"; import { mkdtemp, readFile, rm, stat, writeFile } from "node:fs/promises"; import { tmpdir } from "node:os"; import { join } from "node:path"; @@ -21,6 +22,7 @@ import { import { afterEach, expect, test, vi } from "vitest"; import { resetForTesting } from "../src/config.ts"; +import { drainErrorLogForTesting, writeErrorLog } from "../src/error-log.ts"; import { telemetryDevExtension, type TelemetryDevExtensionOptions } from "../src/extension.ts"; interface Exporters extends ClientOverrides { @@ -753,10 +755,11 @@ test("errors go to a private, rotated telemetry-dev.log in the agent directory, // An earlier log of more than 5 MB moves to telemetry-dev.log.1. await writeFile(path, "x".repeat(5 * 1024 * 1024 + 1)); const handlers = new Map unknown>(); + let status = 200; telemetryDevExtension({ apiKey: "td_live_test", exportMode: "immediate", - fetch: async () => new Response(null, { status: 400 }), + fetch: async () => new Response(null, { status }), })({ on: (type: string, handler: never) => handlers.set(type, handler) }); await handlers.get("session_start")?.({ reason: "startup" }, makeContext()); @@ -769,6 +772,9 @@ test("errors go to a private, rotated telemetry-dev.log in the agent directory, }, makeContext(), ); + // The export that shutdown flushes fails; its report must still reach the log. + status = 400; + await handlers.get("session_shutdown")?.({ reason: "quit" }, makeContext()); await resetForTesting(); const log = await readFile(path, "utf8"); @@ -783,6 +789,31 @@ test("errors go to a private, rotated telemetry-dev.log in the agent directory, } }); +test("the error log keeps call order and drops entries past the pending limit", async () => { + const directory = await mkdtemp(join(tmpdir(), "telemetry-pi-agent-")); + vi.stubEnv("PI_CODING_AGENT_DIR", directory); + + try { + for (let index = 0; index < 150; index++) writeErrorLog(new Error(`entry-${index}`)); + + // writeErrorLog returns before touching the disk. + expect(existsSync(join(directory, "telemetry-dev.log"))).toBe(false); + await drainErrorLogForTesting(); + const entries = [ + ...(await readFile(join(directory, "telemetry-dev.log"), "utf8")).matchAll(/entry-(\d+)/g), + ].map((match) => Number(match[1])); + + expect(entries).toEqual(Array.from({ length: 100 }, (_, index) => index)); + + writeErrorLog(new Error("after drain")); + await drainErrorLogForTesting(); + expect(await readFile(join(directory, "telemetry-dev.log"), "utf8")).toContain("after drain"); + } finally { + vi.unstubAllEnvs(); + await rm(directory, { recursive: true, force: true }); + } +}); + test("each event reads the current session id", async () => { const harness = makeHarness(); await harness.emit({ type: "session_start", reason: "startup" }, makeContext("session-2"));