fix(adapter-utils): order stdin file writes in the sandbox process-session bridge (#11406)
## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work > - Agent adapters can run through a sandbox process-session bridge > - The bridge writes streamed standard input to files before the remote process reads them > - Concurrent file writes can make a later chunk visible before an earlier chunk > - The remote process can then parse a tail fragment and wait forever for the missing head > - This pull request serializes host writes and makes an unexpected file gap a loud error > - The benefit is ordered input with a bounded failure path for sandbox ACP sessions ## Linked Issues or Issue Description No public GitHub issue exists for this change. The description below follows `.github/ISSUE_TEMPLATE/bug_report.yml`. **What happened?** A sandbox ACP process-session bridge could stall after its handshake when a run sent a large prompt. The host sent one un-awaited file write for each standard input chunk. A small later chunk could finish before a large earlier chunk. The remote poller then sent the tail bytes first. The agent parser raised an error on the tail fragment, and the head bytes stayed buffered without a newline. **Expected behavior** The bridge must expose standard input files in sequence. The remote poller must report a clear error when an earlier file remains missing beyond the retry budget. **Steps to reproduce** 1. Start an ACP session through a sandbox process-session bridge. 2. Send a prompt that produces multiple standard input file chunks. 3. Delay finalization of an earlier chunk while a later chunk completes. 4. Observe that the remote parser can receive the later chunk first and the session can stop without a clear error. **Paperclip version or commit** The change targets the current `master` branch at the submitted commit. **Deployment mode** The bug affects sandbox execution. **Agent adapter(s) involved** The failure affects the ACP process-session bridge. **Database mode** Not database-related. **Additional context** The fix keeps the existing per-file atomic write behavior. It adds ordering at the host write boundary and a bounded ordering check in the shared wrapper poll tail. ## What Changed - Add a per-session promise chain for host standard input file writes. - Keep a failed write from blocking later chain entries. - Track the next expected sequence number in the shared wrapper poll tail. - Hold later files while an earlier file is missing within the existing retry budget. - Emit a loud error and advance after the retry budget expires. - Add regression tests for host ordering, gap holding, and the loud error path. ## Verification - `npx vitest run packages/adapter-utils/src/execution-target-stdin-race.test.ts` — 9 tests passed. - `npx vitest run packages/adapter-utils/src/execution-target-sandbox.test.ts` — 43 tests passed. - `pnpm --filter @paperclipai/adapter-utils typecheck` — clean. - With the source fix reverted, the 3 new tests fail and the 6 original tests pass. - CI must pass on the pull request before merge. ## Risks Low risk. - The host now serializes writes for each session, which can reduce write parallelism. - A failed write still emits one error and destroys the socket, as before. - The wrapper can emit a loud error after the existing retry budget when a file gap persists. - The change does not alter the atomic per-file write behavior. ## Model Used OpenAI GPT-5, model ID `gpt-5`, with tool use and code execution. The context window and internal reasoning details are not disclosed. ## Checklist - [x] I have included a thinking path that traces from project context to this change - [x] I have specified the model used (with version and capability details) - [x] I have checked ROADMAP.md and confirmed this PR does not duplicate planned core work - [x] I have searched GitHub for duplicate or related PRs and linked them above - [x] I have either (a) linked existing issues with `Fixes: #` / `Closes #` / `Refs #` OR (b) described the issue in-PR following the relevant issue template - [x] I have not referenced internal/instance-local Paperclip issues or links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip` URLs) - [x] My branch name describes the change (e.g. `docs/...`, `fix/...`) and contains no internal Paperclip ticket id or instance-derived details - [x] I have run tests locally and they pass - [x] I have added or updated tests where applicable - [x] I have updated relevant documentation to reflect my changes - [x] I have considered and documented any risks above - [x] All Paperclip CI gates are green - [x] Greptile is 5/5 with no open P2s, recommendations, or follow-ups - [x] I will address all Greptile and reviewer comments before requesting merge --------- Co-authored-by: Paperclip <noreply@paperclip.ing>
This commit is contained in:
Nicky Leach
GitHub
parent
6f26f2a450
commit
ed8075b535
|
|
@ -1,13 +1,18 @@
|
|||
import { execFile as execFileCallback, spawn } from "node:child_process";
|
||||
import { mkdir, mkdtemp, readFile, readdir, rename, rm, writeFile } from "node:fs/promises";
|
||||
import net from "node:net";
|
||||
import os from "node:os";
|
||||
import path from "node:path";
|
||||
import { promisify } from "node:util";
|
||||
import { afterEach, describe, expect, it } from "vitest";
|
||||
|
||||
import { getProcessSessionRemoteSource } from "./execution-target.js";
|
||||
import {
|
||||
getProcessSessionRemoteSource,
|
||||
startAdapterExecutionTargetProcessSessionBridge,
|
||||
type AdapterSandboxExecutionTarget,
|
||||
} from "./execution-target.js";
|
||||
import { createCommandManagedSandboxCallbackBridgeQueueClient } from "./sandbox-callback-bridge.js";
|
||||
import type { RunProcessResult } from "./server-utils.js";
|
||||
import { runChildProcess, type RunProcessResult } from "./server-utils.js";
|
||||
|
||||
const execFile = promisify(execFileCallback);
|
||||
|
||||
|
|
@ -222,6 +227,270 @@ describe("stdin file race (parent PAP-4037)", () => {
|
|||
expect(poller.frames.some((frame) => frame.type === "exit")).toBe(true);
|
||||
});
|
||||
|
||||
it("holds a later stdin file until the missing earlier file appears", async () => {
|
||||
const poller = await startPollerWrapper();
|
||||
|
||||
// File 2 is complete, but file 1 has not appeared yet (a host reordering).
|
||||
// The poller must not deliver file 2 ahead of the missing file 1. It holds
|
||||
// the send order and waits for the earlier file.
|
||||
await poller.writeFileAtomic("000000000002.json", stdinMessage("second-payload"));
|
||||
await delay(300);
|
||||
// File 2 is still on disk and nothing was delivered: the poller holds it.
|
||||
const afterHold = await readdir(poller.stdinDir);
|
||||
expect(afterHold).toContain("000000000002.json");
|
||||
expect(collectDelivered(poller.frames)).toBe("");
|
||||
|
||||
// File 1 arrives. The poller now delivers file 1 then file 2, in send order.
|
||||
await poller.writeFileAtomic("000000000001.json", stdinMessage("first-payload"));
|
||||
await waitFor(() => collectDelivered(poller.frames).includes("second-payload"));
|
||||
expect(collectDelivered(poller.frames)).toBe("first-payloadsecond-payload");
|
||||
|
||||
await poller.writeFileAtomic("000000000003.json", stdinEndMessage);
|
||||
await poller.exited;
|
||||
expect(poller.frames.some((frame) => frame.type === "exit")).toBe(true);
|
||||
});
|
||||
|
||||
it("fails loud and advances past a missing stdin file after the retry limit", async () => {
|
||||
const poller = await startPollerWrapper({ maxRetries: 3 });
|
||||
|
||||
// File 1 never appears. File 2 is complete. After the retry limit the poller
|
||||
// writes a loud error event and advances past the gap, then delivers file 2.
|
||||
// So a permanent reordering fails loud, never silently.
|
||||
await poller.writeFileAtomic("000000000002.json", stdinMessage("after-gap"));
|
||||
|
||||
await waitFor(() =>
|
||||
poller.frames.some(
|
||||
(frame) =>
|
||||
frame.type === "error" &&
|
||||
typeof frame.message === "string" &&
|
||||
frame.message.includes("Advanced past missing stdin files"),
|
||||
),
|
||||
);
|
||||
await waitFor(() => collectDelivered(poller.frames).includes("after-gap"));
|
||||
|
||||
await poller.writeFileAtomic("000000000003.json", stdinEndMessage);
|
||||
await poller.exited;
|
||||
expect(collectDelivered(poller.frames)).toBe("after-gap");
|
||||
expect(poller.frames.some((frame) => frame.type === "exit")).toBe(true);
|
||||
});
|
||||
|
||||
// ---- Host serialization test (drives the real bridge) -----------------
|
||||
|
||||
// A runner that runs each bridge shell script as a real child process, so the
|
||||
// test drives the whole legacy-poll bridge: the socket handler, the command-
|
||||
// managed `writeTextFile` script, the nohup wrapper, and the output poll.
|
||||
function createLocalSandboxRunner(
|
||||
onExecute?: (script: string) => Promise<void>,
|
||||
) {
|
||||
let counter = 0;
|
||||
return {
|
||||
execute: async (input: {
|
||||
command: string;
|
||||
args?: string[];
|
||||
cwd?: string;
|
||||
env?: Record<string, string>;
|
||||
stdin?: string;
|
||||
timeoutMs?: number;
|
||||
onLog?: (stream: "stdout" | "stderr", chunk: string) => Promise<void>;
|
||||
}): Promise<RunProcessResult> => {
|
||||
counter += 1;
|
||||
const script = input.args?.[1] ?? "";
|
||||
if (onExecute) await onExecute(script);
|
||||
const command =
|
||||
input.command === "bash" ? "/bin/bash" : input.command === "sh" ? "/bin/sh" : input.command;
|
||||
return runChildProcess(`stdin-order-run-${counter}`, command, input.args ?? [], {
|
||||
cwd: input.cwd ?? process.cwd(),
|
||||
env: input.env ?? {},
|
||||
stdin: input.stdin,
|
||||
timeoutSec: Math.max(1, Math.ceil((input.timeoutMs ?? 30_000) / 1000)),
|
||||
graceSec: 5,
|
||||
onLog: input.onLog ?? (async () => {}),
|
||||
});
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
it("serializes host stdin writes so a slow earlier write still lands first", async () => {
|
||||
const rootDir = await mkdtemp(path.join(os.tmpdir(), "paperclip-stdin-host-order-"));
|
||||
cleanupDirs.push(rootDir);
|
||||
// The child echoes every stdin byte to stdout, so the wrapper reports the
|
||||
// exact bytes and order the child received on its stdin.
|
||||
const childPath = path.join(rootDir, "echo-child.mjs");
|
||||
await writeFile(childPath, "process.stdin.on('data', (c) => process.stdout.write(c));\n", "utf8");
|
||||
|
||||
// Record the send-order-relevant event: the completion of each stdin file's
|
||||
// finalize (atomic rename). Delay the finalize of the FIRST file, so its
|
||||
// write resolves slower than the second. Without serialization the second
|
||||
// rename would land first; the per-session chain must keep the send order.
|
||||
const finalizeOrder: string[] = [];
|
||||
const runner = createLocalSandboxRunner(async (script) => {
|
||||
const finalizeMatch = /base64 -d[\s\S]*mv '[^']*\.decoded' '([^']+\.json)'/.exec(script);
|
||||
if (finalizeMatch) {
|
||||
const remotePath = finalizeMatch[1];
|
||||
if (remotePath.endsWith("000000000001.json")) await delay(300);
|
||||
finalizeOrder.push(path.posix.basename(remotePath));
|
||||
}
|
||||
});
|
||||
const target: AdapterSandboxExecutionTarget = {
|
||||
kind: "remote",
|
||||
transport: "sandbox",
|
||||
providerKey: "local-test",
|
||||
remoteCwd: rootDir,
|
||||
timeoutMs: 30_000,
|
||||
runner,
|
||||
};
|
||||
|
||||
const bridge = await startAdapterExecutionTargetProcessSessionBridge({
|
||||
runId: "run-stdin-host-order",
|
||||
target,
|
||||
runtimeRootDir: path.posix.join(rootDir, ".paperclip-runtime", "acpx"),
|
||||
adapterKey: "acpx",
|
||||
command: process.execPath,
|
||||
args: [childPath],
|
||||
cwd: rootDir,
|
||||
env: {},
|
||||
timeoutSec: 5,
|
||||
onLog: async () => {},
|
||||
});
|
||||
expect(bridge).not.toBeNull();
|
||||
|
||||
let peer: net.Socket | null = null;
|
||||
try {
|
||||
const proxySource = await readFile(bridge!.agentCommand, "utf8");
|
||||
const port = Number(/port: (\d+)/.exec(proxySource)?.[1] ?? Number.NaN);
|
||||
const tokenLiteral = /const token = (".*?");/.exec(proxySource)?.[1];
|
||||
expect(Number.isFinite(port)).toBe(true);
|
||||
const token = JSON.parse(tokenLiteral as string) as string;
|
||||
|
||||
const peerSocket = net.createConnection({ host: "127.0.0.1", port });
|
||||
peer = peerSocket;
|
||||
peerSocket.setEncoding("utf8");
|
||||
peerSocket.on("error", () => undefined);
|
||||
const delivered: string[] = [];
|
||||
let peerBuffer = "";
|
||||
peerSocket.on("data", (chunk: string) => {
|
||||
peerBuffer += chunk;
|
||||
const lines = peerBuffer.split("\n");
|
||||
peerBuffer = lines.pop() || "";
|
||||
for (const line of lines) {
|
||||
if (!line.trim()) continue;
|
||||
const frame = JSON.parse(line) as { type?: string; stream?: string; data?: string };
|
||||
if (frame.type === "data" && frame.stream === "stdout" && typeof frame.data === "string") {
|
||||
delivered.push(Buffer.from(frame.data, "base64").toString("utf8"));
|
||||
}
|
||||
}
|
||||
});
|
||||
await new Promise<void>((resolve, reject) => {
|
||||
peerSocket.once("connect", () => resolve());
|
||||
peerSocket.once("error", reject);
|
||||
});
|
||||
|
||||
// Send two stdin messages back to back. The first authenticates and writes
|
||||
// file 1; the second writes file 2. Both are scheduled before file 1's
|
||||
// delayed finalize resolves, so an un-chained handler would race them.
|
||||
const head = `${JSON.stringify({ token, type: "stdin", data: Buffer.from("HEAD_ONE_", "utf8").toString("base64") })}\n`;
|
||||
const tail = `${JSON.stringify({ token, type: "stdin", data: Buffer.from("TAIL_TWO", "utf8").toString("base64") })}\n`;
|
||||
peerSocket.write(head);
|
||||
peerSocket.write(tail);
|
||||
|
||||
// The two finalize renames complete in send order, not in the order the
|
||||
// delayed and fast writes would otherwise finish.
|
||||
await waitFor(() => finalizeOrder.length >= 2, 8_000);
|
||||
expect(finalizeOrder.slice(0, 2)).toEqual(["000000000001.json", "000000000002.json"]);
|
||||
|
||||
// End to end: the child receives the two payloads intact and in send
|
||||
// order, so the prompt is byte-identical on the child stdin.
|
||||
await waitFor(() => delivered.join("").includes("TAIL_TWO"), 8_000);
|
||||
expect(delivered.join("")).toBe("HEAD_ONE_TAIL_TWO");
|
||||
} finally {
|
||||
peer?.destroy();
|
||||
await bridge?.stop();
|
||||
}
|
||||
});
|
||||
|
||||
it("holds stdinEnd on stop until an earlier pending stdin write lands first", async () => {
|
||||
const rootDir = await mkdtemp(path.join(os.tmpdir(), "paperclip-stdin-stop-order-"));
|
||||
cleanupDirs.push(rootDir);
|
||||
const childPath = path.join(rootDir, "echo-child.mjs");
|
||||
await writeFile(childPath, "process.stdin.on('data', (c) => process.stdout.write(c));\n", "utf8");
|
||||
|
||||
// Record each stdin file finalize (atomic rename). `finalizeStarted` marks
|
||||
// the start; `finalizeOrder` marks the completion. Delay the FIRST chunk's
|
||||
// finalize, so its write is still pending when `stop()` runs. `stop()` must
|
||||
// chain the `stdinEnd` write after the pending chunk, so file 2 (stdinEnd)
|
||||
// never finishes its rename before file 1.
|
||||
const finalizeStarted: string[] = [];
|
||||
const finalizeOrder: string[] = [];
|
||||
const runner = createLocalSandboxRunner(async (script) => {
|
||||
const finalizeMatch = /base64 -d[\s\S]*mv '[^']*\.decoded' '([^']+\.json)'/.exec(script);
|
||||
if (finalizeMatch) {
|
||||
const name = path.posix.basename(finalizeMatch[1]);
|
||||
finalizeStarted.push(name);
|
||||
if (name === "000000000001.json") await delay(300);
|
||||
finalizeOrder.push(name);
|
||||
}
|
||||
});
|
||||
const target: AdapterSandboxExecutionTarget = {
|
||||
kind: "remote",
|
||||
transport: "sandbox",
|
||||
providerKey: "local-test",
|
||||
remoteCwd: rootDir,
|
||||
timeoutMs: 30_000,
|
||||
runner,
|
||||
};
|
||||
|
||||
const bridge = await startAdapterExecutionTargetProcessSessionBridge({
|
||||
runId: "run-stdin-stop-order",
|
||||
target,
|
||||
runtimeRootDir: path.posix.join(rootDir, ".paperclip-runtime", "acpx"),
|
||||
adapterKey: "acpx",
|
||||
command: process.execPath,
|
||||
args: [childPath],
|
||||
cwd: rootDir,
|
||||
env: {},
|
||||
timeoutSec: 5,
|
||||
onLog: async () => {},
|
||||
});
|
||||
expect(bridge).not.toBeNull();
|
||||
|
||||
let peer: net.Socket | null = null;
|
||||
let stopped = false;
|
||||
try {
|
||||
const proxySource = await readFile(bridge!.agentCommand, "utf8");
|
||||
const port = Number(/port: (\d+)/.exec(proxySource)?.[1] ?? Number.NaN);
|
||||
const tokenLiteral = /const token = (".*?");/.exec(proxySource)?.[1];
|
||||
expect(Number.isFinite(port)).toBe(true);
|
||||
const token = JSON.parse(tokenLiteral as string) as string;
|
||||
|
||||
const peerSocket = net.createConnection({ host: "127.0.0.1", port });
|
||||
peer = peerSocket;
|
||||
peerSocket.setEncoding("utf8");
|
||||
peerSocket.on("error", () => undefined);
|
||||
peerSocket.on("data", () => undefined);
|
||||
await new Promise<void>((resolve, reject) => {
|
||||
peerSocket.once("connect", () => resolve());
|
||||
peerSocket.once("error", reject);
|
||||
});
|
||||
|
||||
// Send one stdin message. It authenticates and writes file 1, whose
|
||||
// finalize the runner delays. Wait until that finalize has started, so the
|
||||
// write is in flight when `stop()` runs.
|
||||
const head = `${JSON.stringify({ token, type: "stdin", data: Buffer.from("HEAD_ONE_", "utf8").toString("base64") })}\n`;
|
||||
peerSocket.write(head);
|
||||
await waitFor(() => finalizeStarted.includes("000000000001.json"), 8_000);
|
||||
|
||||
// Stop the bridge while file 1's write is still pending. `stop()` awaits
|
||||
// the chained `stdinEnd` write, so both finalizes are complete when it
|
||||
// returns, in send order.
|
||||
await bridge!.stop();
|
||||
stopped = true;
|
||||
expect(finalizeOrder).toEqual(["000000000001.json", "000000000002.json"]);
|
||||
} finally {
|
||||
peer?.destroy();
|
||||
if (!stopped) await bridge?.stop();
|
||||
}
|
||||
});
|
||||
|
||||
// ---- Host atomic-write tests ------------------------------------------
|
||||
|
||||
// A runner that executes each bridge shell script on the local filesystem,
|
||||
|
|
|
|||
|
|
@ -1532,6 +1532,14 @@ export async function startAdapterExecutionTargetProcessSessionBridge(input: {
|
|||
signalStopped = resolve;
|
||||
});
|
||||
let stdinSeq = 0;
|
||||
// One promise chain per session that serializes the stdin file writes. Each
|
||||
// write is multi-exec on the command-managed client: prepare, append per 32
|
||||
// KiB, then an atomic rename. The chain makes the rename for file N finish
|
||||
// before the write for file N+1 starts, so the files land in send order.
|
||||
// Without it the writes overlap. A small later chunk can then rename ahead of
|
||||
// a big earlier chunk, so the wrapper reads the stdin bytes out of order and
|
||||
// corrupts a large prompt on the stdin path.
|
||||
let stdinWriteChain: Promise<void> = Promise.resolve();
|
||||
let pollTimer: NodeJS.Timeout | null = null;
|
||||
const pendingRemoteEvents: Array<{
|
||||
type?: string;
|
||||
|
|
@ -1642,9 +1650,21 @@ export async function startAdapterExecutionTargetProcessSessionBridge(input: {
|
|||
if (stdinPayload) {
|
||||
stdinSeq += 1;
|
||||
const name = `${String(stdinSeq).padStart(12, "0")}.json`;
|
||||
void runRuntimeWork(AGENT_SESSION_SEND_INPUT_SPAN, () =>
|
||||
client.writeTextFile(path.posix.join(stdinDir, name), jsonLine(stdinPayload)),
|
||||
).catch((error) => {
|
||||
const filePath = path.posix.join(stdinDir, name);
|
||||
// Chain this write after the previous one, so the atomic rename for
|
||||
// file N finishes before the write for file N+1 starts. Keep the
|
||||
// per-message `sandbox.agentSession.sendInput` span inside the chain.
|
||||
const write = stdinWriteChain.then(() =>
|
||||
runRuntimeWork(AGENT_SESSION_SEND_INPUT_SPAN, () =>
|
||||
client.writeTextFile(filePath, jsonLine(stdinPayload)),
|
||||
),
|
||||
);
|
||||
// The next message chains after this write on success or failure, so a
|
||||
// failed write never blocks the chain. This mirrors the wrapper
|
||||
// `writeChain` pattern for its event files.
|
||||
stdinWriteChain = write.then(() => undefined, () => undefined);
|
||||
// Keep the failure behavior: send one error line, then destroy the socket.
|
||||
write.catch((error) => {
|
||||
nextSocket.write(jsonLine({ type: "error", message: error instanceof Error ? error.message : String(error) }));
|
||||
nextSocket.destroy();
|
||||
});
|
||||
|
|
@ -1830,10 +1850,21 @@ export async function startAdapterExecutionTargetProcessSessionBridge(input: {
|
|||
if (pollTimer) clearTimeout(pollTimer);
|
||||
for (const liveSocket of liveSockets) liveSocket.destroy();
|
||||
await new Promise<void>((resolve) => server.close(() => resolve())).catch(() => undefined);
|
||||
await client.writeTextFile(
|
||||
path.posix.join(stdinDir, `${String(stdinSeq + 1).padStart(12, "0")}.json`),
|
||||
jsonLine({ type: "stdinEnd" }),
|
||||
).catch(() => undefined);
|
||||
// Wait for every accepted stdin write before `stdinEnd`. The socket handler
|
||||
// fires each chunk write un-awaited through `stdinWriteChain`, so an earlier
|
||||
// chunk can still be pending here. Chain the `stdinEnd` write onto the same
|
||||
// per-session chain, so its file rename never finishes before an earlier
|
||||
// chunk. `stdinSeq` is stable now, because the sockets are destroyed and the
|
||||
// server is closed, so no new message can increment it.
|
||||
const stdinEndPath = path.posix.join(
|
||||
stdinDir,
|
||||
`${String(stdinSeq + 1).padStart(12, "0")}.json`,
|
||||
);
|
||||
const stdinEndWrite = stdinWriteChain.then(() =>
|
||||
client.writeTextFile(stdinEndPath, jsonLine({ type: "stdinEnd" })),
|
||||
);
|
||||
stdinWriteChain = stdinEndWrite.then(() => undefined, () => undefined);
|
||||
await stdinEndWrite.catch(() => undefined);
|
||||
await client.remove(sessionDir).catch(() => undefined);
|
||||
await fs.rm(proxyDir, { recursive: true, force: true }).catch(() => undefined);
|
||||
},
|
||||
|
|
@ -1913,12 +1944,40 @@ const stdinMaxParseRetries = (() => {
|
|||
return Number.isFinite(raw) && raw > 0 ? raw : 100;
|
||||
})();
|
||||
const stdinParseRetries = new Map();
|
||||
// Track the next expected sequence number. The host writes the stdin files in
|
||||
// send order and pads the number to 12 digits, starting at 1. The files sort in
|
||||
// send order. When the smallest present number is greater than expected, an
|
||||
// earlier file has not appeared yet: a missing file, not an unreadable one. Hold
|
||||
// the send order and wait for it, bounded by the same retry budget as the
|
||||
// unreadable-file path. This turns a reordering into a loud error, never silent
|
||||
// corruption.
|
||||
let stdinExpectedSeq = 1;
|
||||
let stdinGapRetries = 0;
|
||||
|
||||
async function pollStdin() {
|
||||
while (!stdinClosed) {
|
||||
const entries = (await fs.readdir(stdinDir).catch(() => [])).filter((name) => name.endsWith(".json")).sort();
|
||||
for (const name of entries) {
|
||||
if (stdinClosed) break;
|
||||
const entrySeq = Number.parseInt(name, 10);
|
||||
// Hold the send order when an earlier file has not appeared. Do not consume
|
||||
// this later file: wait for the missing file on a later cycle, bounded by
|
||||
// the retry budget. After the budget, fail loud and advance past the gap,
|
||||
// so the present file can run.
|
||||
if (Number.isFinite(entrySeq) && entrySeq > stdinExpectedSeq) {
|
||||
stdinGapRetries += 1;
|
||||
if (stdinGapRetries < stdinMaxParseRetries) {
|
||||
break;
|
||||
}
|
||||
await writeEvent({
|
||||
type: "error",
|
||||
message:
|
||||
"Advanced past missing stdin files " + stdinExpectedSeq + " to " + (entrySeq - 1) +
|
||||
" after " + stdinMaxParseRetries + " retries.",
|
||||
});
|
||||
stdinGapRetries = 0;
|
||||
stdinExpectedSeq = entrySeq;
|
||||
}
|
||||
const file = path.posix.join(stdinDir, name);
|
||||
let message;
|
||||
try {
|
||||
|
|
@ -1941,6 +2000,10 @@ async function pollStdin() {
|
|||
"Dropped unreadable stdin file after " + stdinMaxParseRetries + " retries: " + name + ": " +
|
||||
(error instanceof Error ? error.message : String(error)),
|
||||
});
|
||||
// The file is resolved (dropped). Advance the expected number and reset
|
||||
// the gap budget, then let the loop go on to the next entry.
|
||||
if (Number.isFinite(entrySeq)) stdinExpectedSeq = entrySeq + 1;
|
||||
stdinGapRetries = 0;
|
||||
continue;
|
||||
}
|
||||
// The file is not readable yet and is not past the retry limit. Keep it
|
||||
|
|
@ -1954,6 +2017,10 @@ async function pollStdin() {
|
|||
// then act on the message. A later cycle never re-reads a handled file.
|
||||
stdinParseRetries.delete(name);
|
||||
await fs.rm(file, { force: true }).catch(() => undefined);
|
||||
// The file is handled. Advance the expected number and reset the gap
|
||||
// budget, so the next expected file starts fresh.
|
||||
if (Number.isFinite(entrySeq)) stdinExpectedSeq = entrySeq + 1;
|
||||
stdinGapRetries = 0;
|
||||
if (message.type === "stdin" && typeof message.data === "string") {
|
||||
if (!stdinClosed) child.stdin.write(Buffer.from(message.data, "base64"));
|
||||
} else if (message.type === "stdinEnd") {
|
||||
|
|
|
|||
Loading…
Reference in New Issue