fix(adapter-utils): add per-iteration timeout and watchdog to the sandbox callback bridge poll loop (#11341)
## Thinking Path > - Paperclip runs AI agents through adapters and sandboxed execution paths > - The sandbox callback bridge carries file requests between the host and a sandbox > - The poll loop waited forever when a sandbox call stopped responding > - A permanent wait stranded queued requests and hid the run failure > - This pull request adds bounded timeouts, abort handling, recovery backstops, and trace reporting > - The benefit is prompt request failure, safe mutation outcomes, run-level error reporting, and trace visibility ## Linked Issues or Issue Description **What happened?** The sandbox callback bridge could wait forever when a client call stopped responding without a rejection. **Expected behavior** The bridge should fail queued requests and report a run-level error when the sandbox channel stops responding. **Steps to reproduce** 1. Start a sandbox callback bridge. 2. Queue a request. 3. Make the sandbox call stop responding. 4. Observe that the request does not receive a failure response. **Paperclip version or commit** Commit `edc4f71b460c600f97cf44cb486d5cac72ca2db9`. **Deployment mode** Built from source. **Installation method** Built from source with pnpm. **Agent adapter(s) involved** Custom or external sandbox callback bridge. **Database mode** Not database-related. ## What Changed - Add a per-iteration timeout for `listJsonFiles` and `processRequestFile`. - Add a watchdog that fails pending requests when the loop makes no progress. - Abort a hung handler and use a non-retryable 504 backstop when its outcome can be indeterminate. - Retry recovery writes and keep queued requests when a recovery write fails. - Forward the indeterminate-outcome header through the execution target. - Record worker failures through the `sandbox.callbackBridge.workerFailed` trace span. - Add tests for timeout, watchdog, recovery, mutation safety, header forwarding, and fast-request behavior. ## Verification - Run `pnpm exec vitest run packages/adapter-utils/src/sandbox-callback-bridge.test.ts packages/adapter-utils/src/execution-target-sandbox.test.ts`. - Confirm that the PR test, typecheck, build, end-to-end, serialized test, and security checks pass. - Confirm that the current PR head is `edc4f71b460c600f97cf44cb486d5cac72ca2db9`. - Confirm that the PR changes four files: the callback bridge, its tests, the execution target, and its tests. ## Risks The default timeout can fail a slow but valid sandbox call. The defaults remain configurable, and the iteration timeout stays below the sandbox response deadline. A mutation that may have committed returns a non-retryable 504 outcome so the caller does not apply it twice. ## Model Used OpenAI Codex, GPT-5 current runtime, with extended reasoning and tool use. The exact context window is not exposed by the runtime. ## 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
8ee1fb21a6
commit
6f26f2a450
|
|
@ -2444,6 +2444,80 @@ describe("sandbox adapter execution targets", () => {
|
|||
}
|
||||
});
|
||||
|
||||
it("forwards the host indeterminate-outcome header so the sandbox server maps the 504 to a non-retryable 409", async () => {
|
||||
// The host marks a possibly-committed mutation with a 504 and the
|
||||
// `x-paperclip-bridge-outcome: indeterminate` header. The forward must keep
|
||||
// that header, so the in-sandbox server maps the 504 to a non-retryable 409.
|
||||
// If the forward drops the header, the client sees a retryable 504 and a
|
||||
// retry repeats a mutation that already committed.
|
||||
const rootDir = await mkdtemp(path.join(os.tmpdir(), "paperclip-execution-target-bridge-outcome-"));
|
||||
cleanupDirs.push(rootDir);
|
||||
const remoteCwd = path.join(rootDir, "workspace");
|
||||
const runtimeRootDir = path.join(remoteCwd, ".paperclip-runtime", "codex");
|
||||
await mkdir(runtimeRootDir, { recursive: true });
|
||||
|
||||
const responseBody = JSON.stringify({ error: "Mutation outcome is indeterminate.", outcome: "indeterminate", retryable: false });
|
||||
const apiServer = createServer((_req, res) => {
|
||||
res.writeHead(504, {
|
||||
"content-type": "application/json",
|
||||
"x-paperclip-bridge-outcome": "indeterminate",
|
||||
});
|
||||
res.end(responseBody);
|
||||
});
|
||||
await new Promise<void>((resolve, reject) => {
|
||||
apiServer.once("error", reject);
|
||||
apiServer.listen(0, "127.0.0.1", () => resolve());
|
||||
});
|
||||
const address = apiServer.address();
|
||||
if (!address || typeof address === "string") {
|
||||
throw new Error("Expected the bridge outcome test API server to listen on a TCP port.");
|
||||
}
|
||||
|
||||
const target: AdapterSandboxExecutionTarget = {
|
||||
kind: "remote",
|
||||
transport: "sandbox",
|
||||
providerKey: "e2b",
|
||||
environmentId: "env-1",
|
||||
leaseId: "lease-1",
|
||||
remoteCwd,
|
||||
runner: createLocalSandboxRunner(),
|
||||
timeoutMs: 30_000,
|
||||
};
|
||||
|
||||
const bridge = await startAdapterExecutionTargetPaperclipBridge({
|
||||
runId: "run-bridge-outcome",
|
||||
target,
|
||||
runtimeRootDir,
|
||||
adapterKey: "codex",
|
||||
hostApiToken: "real-run-jwt",
|
||||
hostApiUrl: `http://127.0.0.1:${address.port}`,
|
||||
});
|
||||
try {
|
||||
const response = await fetch(`${bridge!.env.PAPERCLIP_API_URL}/api/issues/issue-1/comments`, {
|
||||
method: "POST",
|
||||
headers: {
|
||||
authorization: `Bearer ${bridge!.env.PAPERCLIP_API_KEY}`,
|
||||
"content-type": "application/json",
|
||||
},
|
||||
body: JSON.stringify({ body: "Status update." }),
|
||||
});
|
||||
|
||||
// The sandbox server maps the indeterminate 504 to a non-retryable 409.
|
||||
expect(response.status).toBe(409);
|
||||
// The outcome header and body still reach the client, so a caller that
|
||||
// reads them still sees the indeterminate result.
|
||||
expect(response.headers.get("x-paperclip-bridge-outcome")).toBe("indeterminate");
|
||||
await expect(response.json()).resolves.toEqual({
|
||||
error: "Mutation outcome is indeterminate.",
|
||||
outcome: "indeterminate",
|
||||
retryable: false,
|
||||
});
|
||||
} finally {
|
||||
await bridge?.stop();
|
||||
await new Promise<void>((resolve) => apiServer.close(() => resolve()));
|
||||
}
|
||||
});
|
||||
|
||||
it("forwards bridge traffic to the local listen origin even when public API URLs are configured", async () => {
|
||||
const rootDir = await mkdtemp(path.join(os.tmpdir(), "paperclip-execution-target-bridge-local-"));
|
||||
cleanupDirs.push(rootDir);
|
||||
|
|
|
|||
|
|
@ -1246,7 +1246,12 @@ export function runtimeAssetDir(
|
|||
|
||||
function buildBridgeResponseHeaders(response: Response): Record<string, string> {
|
||||
const out: Record<string, string> = {};
|
||||
for (const key of ["content-type", "etag", "last-modified"]) {
|
||||
// Keep `x-paperclip-bridge-outcome` in this list. The host marks a
|
||||
// possibly-committed mutation with the `indeterminate` outcome. The in-sandbox
|
||||
// server reads that header to map the 504 to a terminal 409. If the forward
|
||||
// drops the header, the server keeps the retryable 504 and a caller that
|
||||
// retries 5xx can repeat a mutation that already committed.
|
||||
for (const key of ["content-type", "etag", "last-modified", "x-paperclip-bridge-outcome"]) {
|
||||
const value = response.headers.get(key);
|
||||
if (value && value.trim().length > 0) out[key] = value.trim();
|
||||
}
|
||||
|
|
@ -2162,7 +2167,7 @@ export async function startAdapterExecutionTargetPaperclipBridge(input: {
|
|||
maxBodyBytes,
|
||||
getRuntimeParentContext: input.getRuntimeParentContext,
|
||||
runtimeSpan: input.runtimeSpan,
|
||||
handleRequest: async (request) => {
|
||||
handleRequest: async (request, options) => {
|
||||
const method = request.method.trim().toUpperCase() || "GET";
|
||||
if (bridgeDebugEnabled) {
|
||||
await onLog(
|
||||
|
|
@ -2177,11 +2182,19 @@ export async function startAdapterExecutionTargetPaperclipBridge(input: {
|
|||
}
|
||||
headers.set("authorization", `Bearer ${hostApiToken}`);
|
||||
headers.set("x-paperclip-run-id", input.runId);
|
||||
// Abort the forward when the worker aborts the request (its per-iteration
|
||||
// timeout or watchdog fired), or after the 30s ceiling, whichever comes
|
||||
// first. The worker abort lets the bridge fail a hung forward fast
|
||||
// instead of stranding the request until the 30s ceiling.
|
||||
const timeoutSignal = AbortSignal.timeout(30_000);
|
||||
const forwardSignal = options?.signal
|
||||
? AbortSignal.any([options.signal, timeoutSignal])
|
||||
: timeoutSignal;
|
||||
const response = await fetch(buildBridgeForwardUrl(hostApiUrl, request), {
|
||||
method,
|
||||
headers,
|
||||
...(method === "GET" || method === "HEAD" ? {} : { body: request.body }),
|
||||
signal: AbortSignal.timeout(30_000),
|
||||
signal: forwardSignal,
|
||||
});
|
||||
if (bridgeDebugEnabled) {
|
||||
await onLog(
|
||||
|
|
|
|||
File diff suppressed because it is too large
Load Diff
|
|
@ -19,6 +19,37 @@ const DEFAULT_BRIDGE_RESPONSE_TIMEOUT_MS = 30_000;
|
|||
const DEFAULT_BRIDGE_STOP_TIMEOUT_MS = 2_000;
|
||||
const DEFAULT_BRIDGE_MAX_QUEUE_DEPTH = 64;
|
||||
const DEFAULT_BRIDGE_MAX_BODY_BYTES = 256 * 1024;
|
||||
// Per-iteration timeout for one poll-loop client call. A healthy control-plane
|
||||
// round trip finishes in well under one second, so 10s is far above a normal
|
||||
// iteration and never false-fires on a slow-but-live call. It is also well
|
||||
// under the in-sandbox 30s response deadline
|
||||
// (PAPERCLIP_BRIDGE_RESPONSE_TIMEOUT_MS), so the host loop fails fast and writes
|
||||
// 503 responses before the in-sandbox client gives up. A silently unresponsive
|
||||
// sandbox channel makes a client call hang with no reject; this timeout turns
|
||||
// that hang into a caught error, so the loop `catch` runs `failPendingRequests`.
|
||||
const DEFAULT_BRIDGE_ITERATION_TIMEOUT_MS = 10_000;
|
||||
// Watchdog backstop for a hang that the per-iteration timeout does not catch
|
||||
// (for example many slow-but-under-timeout calls, or a stall outside the awaited
|
||||
// calls). It is larger than one iteration timeout, so a single slow iteration
|
||||
// never trips it, and it stays under the in-sandbox 30s response deadline.
|
||||
const DEFAULT_BRIDGE_WATCHDOG_TIMEOUT_MS = 20_000;
|
||||
// Grace period the recovery path gives an aborted in-flight handler to finalize
|
||||
// its own response. The recovery path aborts the handler, then waits this long.
|
||||
// A cooperating handler threads the abort signal into its work, rejects, and
|
||||
// writes its own response inside the grace, so its accurate result wins. A
|
||||
// handler that ignores the signal and never settles does not write inside the
|
||||
// grace; the recovery path then writes a non-retryable 504 backstop, so the
|
||||
// request never strands with no response. The grace is well under the in-sandbox
|
||||
// 30s response deadline, so the backstop lands before the in-sandbox client
|
||||
// gives up.
|
||||
const DEFAULT_BRIDGE_ABORTED_HANDLER_GRACE_MS = 5_000;
|
||||
// The recovery path retries the 504 backstop write this many times before it
|
||||
// gives up. A single transient write failure, or one that exceeds the iteration
|
||||
// timeout, then does not strand the caller with no terminal response.
|
||||
const MAX_BACKSTOP_WRITE_ATTEMPTS = 3;
|
||||
// The delay between two 504 backstop write attempts. It is short, so all retries
|
||||
// finish well under the in-sandbox 30s response deadline.
|
||||
const BACKSTOP_WRITE_RETRY_MS = 50;
|
||||
const REMOTE_WRITE_BASE64_CHUNK_SIZE = 32 * 1024;
|
||||
const SANDBOX_CALLBACK_BRIDGE_ENTRYPOINT = "paperclip-bridge-server.mjs";
|
||||
const SANDBOX_EXEC_CHANNEL_ENV = "PAPERCLIP_SANDBOX_EXEC_CHANNEL";
|
||||
|
|
@ -28,6 +59,12 @@ const SANDBOX_EXEC_CHANNEL_BRIDGE = "bridge";
|
|||
* write the response, and remove the request file. */
|
||||
const CALLBACK_BRIDGE_RELAY_REQUEST_SPAN = "sandbox.callbackBridge.relayRequest";
|
||||
|
||||
/** Span name for a failed or hung bridge worker. The worker runs a throwing
|
||||
* function under this span through `input.runtimeSpan`, so the failure lands in
|
||||
* the run trace. The run and the orchestrator then see the hang, not only
|
||||
* stdout. */
|
||||
const CALLBACK_BRIDGE_WORKER_FAILED_SPAN = "sandbox.callbackBridge.workerFailed";
|
||||
|
||||
export const DEFAULT_SANDBOX_CALLBACK_BRIDGE_MAX_BODY_BYTES = DEFAULT_BRIDGE_MAX_BODY_BYTES;
|
||||
|
||||
export interface SandboxCallbackBridgeRouteRule {
|
||||
|
|
@ -203,6 +240,27 @@ function normalizeTimeoutMs(value: number | null | undefined, fallback: number):
|
|||
return typeof value === "number" && Number.isFinite(value) && value > 0 ? Math.trunc(value) : fallback;
|
||||
}
|
||||
|
||||
/**
|
||||
* Race a promise against a timeout. On timeout the returned promise rejects with
|
||||
* a clear error. The helper clears the timer on every settle path, so it leaks
|
||||
* no `setTimeout`. The wrapped promise is not cancelable; when it never settles,
|
||||
* it keeps running in the background, but the caller already moved on through
|
||||
* the reject.
|
||||
*/
|
||||
function withTimeout<T>(promise: Promise<T>, timeoutMs: number, label: string): Promise<T> {
|
||||
let timer: ReturnType<typeof setTimeout> | undefined;
|
||||
const timeout = new Promise<never>((_resolve, reject) => {
|
||||
timer = setTimeout(() => {
|
||||
reject(new Error(`${label} timed out after ${timeoutMs}ms.`));
|
||||
}, timeoutMs);
|
||||
});
|
||||
return Promise.race([promise, timeout]).finally(() => {
|
||||
if (timer !== undefined) {
|
||||
clearTimeout(timer);
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
function toBuffer(bytes: Buffer | Uint8Array | ArrayBuffer): Buffer {
|
||||
if (Buffer.isBuffer(bytes)) return bytes;
|
||||
if (bytes instanceof ArrayBuffer) return Buffer.from(bytes);
|
||||
|
|
@ -642,8 +700,30 @@ export async function startSandboxCallbackBridgeWorker(input: {
|
|||
client: SandboxCallbackBridgeQueueClient;
|
||||
queueDir: string;
|
||||
pollIntervalMs?: number | null;
|
||||
// Per-iteration timeout for one poll-loop client call (the `listJsonFiles`
|
||||
// poll and one `processRequestFile`). On timeout the loop `catch` runs
|
||||
// `failPendingRequests`. Defaults to DEFAULT_BRIDGE_ITERATION_TIMEOUT_MS.
|
||||
iterationTimeoutMs?: number | null;
|
||||
// Watchdog threshold. When the loop makes no successful iteration within this
|
||||
// time, the watchdog runs `failPendingRequests` and surfaces a run-level error
|
||||
// through `runtimeSpan`. Defaults to DEFAULT_BRIDGE_WATCHDOG_TIMEOUT_MS.
|
||||
watchdogTimeoutMs?: number | null;
|
||||
// Grace the recovery path gives an aborted in-flight handler to finalize its
|
||||
// own response before the recovery path writes a non-retryable 504 backstop.
|
||||
// Defaults to DEFAULT_BRIDGE_ABORTED_HANDLER_GRACE_MS.
|
||||
abortedHandlerGraceMs?: number | null;
|
||||
authorizeRequest?: (request: SandboxCallbackBridgeRequest) => string | null | Promise<string | null>;
|
||||
handleRequest: (request: SandboxCallbackBridgeRequest) => Promise<{
|
||||
// Handle one bridge request. The worker passes an `AbortSignal` through
|
||||
// `options.signal`. The per-iteration timeout, the watchdog, and worker
|
||||
// failure recovery abort it, so a handler that threads the signal into its
|
||||
// work (for example a `fetch`) stops and rejects instead of running forever.
|
||||
// The handler then finalizes with its own error response, so the request does
|
||||
// not strand with no response. A handler that ignores the signal keeps its
|
||||
// earlier behavior.
|
||||
handleRequest: (
|
||||
request: SandboxCallbackBridgeRequest,
|
||||
options?: { signal: AbortSignal },
|
||||
) => Promise<{
|
||||
status: number;
|
||||
headers?: Record<string, string>;
|
||||
body?: string;
|
||||
|
|
@ -663,6 +743,12 @@ export async function startSandboxCallbackBridgeWorker(input: {
|
|||
runtimeSpan?: RuntimeSpanRunner;
|
||||
}): Promise<SandboxCallbackBridgeWorkerHandle> {
|
||||
const pollIntervalMs = normalizeTimeoutMs(input.pollIntervalMs, DEFAULT_BRIDGE_POLL_INTERVAL_MS);
|
||||
const iterationTimeoutMs = normalizeTimeoutMs(input.iterationTimeoutMs, DEFAULT_BRIDGE_ITERATION_TIMEOUT_MS);
|
||||
const watchdogTimeoutMs = normalizeTimeoutMs(input.watchdogTimeoutMs, DEFAULT_BRIDGE_WATCHDOG_TIMEOUT_MS);
|
||||
const abortedHandlerGraceMs = normalizeTimeoutMs(
|
||||
input.abortedHandlerGraceMs,
|
||||
DEFAULT_BRIDGE_ABORTED_HANDLER_GRACE_MS,
|
||||
);
|
||||
const maxBodyBytes = normalizeTimeoutMs(input.maxBodyBytes, DEFAULT_BRIDGE_MAX_BODY_BYTES);
|
||||
const directories = sandboxCallbackBridgeDirectories(input.queueDir);
|
||||
const queueDirectories = [
|
||||
|
|
@ -694,96 +780,519 @@ export async function startSandboxCallbackBridgeWorker(input: {
|
|||
const buildWorkerFailureMessage = (error: unknown) =>
|
||||
`Sandbox callback bridge worker failed: ${error instanceof Error ? error.message : String(error)}`;
|
||||
|
||||
// Per-attempt finalization guard. Each `processRequestFile` call registers one,
|
||||
// keyed by the request file name. The guard is the completion fence between the
|
||||
// request handler and the per-iteration timeout or watchdog recovery. Node runs
|
||||
// one event loop, so a synchronous check-and-set of `claim` is atomic. The
|
||||
// first path to move `claim` off `unclaimed` wins.
|
||||
//
|
||||
// The `claim` value has three states:
|
||||
// - `unclaimed`: no path owns the request yet.
|
||||
// - `handler`: the request handler owns finalization. It set this before it
|
||||
// started the host operation, or when it wrote a 400/403/response. It will
|
||||
// write the real response.
|
||||
// - `abandon`: the recovery path owns the request. It writes a 503.
|
||||
//
|
||||
// The recovery path must never write a 503 for a request whose handler already
|
||||
// started. The per-iteration timeout and the watchdog cannot cancel a host
|
||||
// operation that is in flight. A 503 there makes the caller retry while the
|
||||
// original mutation still completes, so the mutation applies twice. So the
|
||||
// recovery path abandons only a request that the handler did not yet claim; the
|
||||
// handler, when it later reaches the host-operation claim, sees the abandon and
|
||||
// does not run the mutation. This keeps a retry after the 503 exactly-once.
|
||||
//
|
||||
// Each guard also holds an `AbortController`. The recovery path aborts it, so a
|
||||
// handler that already started (claim `handler`) stops its work and finalizes
|
||||
// with its own error response. The abort turns a stranded request into a prompt
|
||||
// error response for a handler that threads the signal into its work.
|
||||
//
|
||||
// A worker abort reaches the handler only after the host operation started, so
|
||||
// the mutation may have committed. The bridge cannot cancel a host operation
|
||||
// that is in flight. So the handler finalizes a worker-aborted request with a
|
||||
// non-retryable 504, not a retryable 502. The caller must not retry a 504, so
|
||||
// it never re-applies a mutation that already committed. A retry-safe 503 comes
|
||||
// only from the recovery path, and only before the host operation starts.
|
||||
//
|
||||
// A handler that ignores the abort signal and never settles would still keep
|
||||
// the request without a response, because the recovery path must not write a
|
||||
// competing 503 for an in-flight mutation. So the recovery path also arms a
|
||||
// backstop timer for each handler-owned request. It aborts the handler, then
|
||||
// waits `abortedHandlerGraceMs`. A cooperating handler finalizes inside the
|
||||
// grace, so its own response wins and `finalize` clears the timer. A handler
|
||||
// that never settles does not finalize inside the grace; the timer then writes
|
||||
// a non-retryable 504 backstop, so the request never strands. The backstop is
|
||||
// non-retryable for the same reason the handler's own 504 is: the recovery
|
||||
// cannot cancel a committed mutation, so the caller must not retry.
|
||||
//
|
||||
// The `finalized` flag is the single-writer fence between the handler's own
|
||||
// `finalize` and the backstop timer. Node runs one event loop, so the
|
||||
// synchronous check-and-set is atomic. The first path to set it writes the
|
||||
// terminal response; the other bails.
|
||||
type RequestFinalizeGuard = {
|
||||
claim: "unclaimed" | "handler" | "abandon";
|
||||
controller: AbortController;
|
||||
finalized: boolean;
|
||||
backstopTimer?: ReturnType<typeof setTimeout>;
|
||||
};
|
||||
const inFlightRequestGuards = new Map<string, RequestFinalizeGuard>();
|
||||
|
||||
const processRequestFile = async (fileName: string) => {
|
||||
// Skip a request that already has an active attempt. The guard map holds only
|
||||
// in-flight attempts; the attempt's finally removes its guard when it ends. A
|
||||
// file that still has a guard is in flight, or it waits for its aborted-handler
|
||||
// 504 backstop. A second attempt would register a new guard and re-run the host
|
||||
// mutation, so the mutation could apply twice.
|
||||
if (inFlightRequestGuards.has(fileName)) {
|
||||
return;
|
||||
}
|
||||
const requestPath = path.posix.join(directories.requestsDir, fileName);
|
||||
const responsePath = path.posix.join(directories.responsesDir, fileName);
|
||||
const raw = await input.client.readTextFile(requestPath);
|
||||
let request: SandboxCallbackBridgeRequest;
|
||||
try {
|
||||
request = JSON.parse(raw) as SandboxCallbackBridgeRequest;
|
||||
} catch {
|
||||
const requestId = fileName.replace(/\.json$/i, "") || randomUUID();
|
||||
await writeBridgeResponse(input.client, requestPath, responsePath, {
|
||||
id: requestId,
|
||||
status: 400,
|
||||
headers: { "content-type": "application/json" },
|
||||
body: JSON.stringify({ error: "Invalid bridge request payload." }),
|
||||
completedAt: new Date().toISOString(),
|
||||
});
|
||||
await input.client.remove(requestPath);
|
||||
return;
|
||||
}
|
||||
|
||||
const denialReason = await authorizeRequest(request);
|
||||
if (denialReason) {
|
||||
await writeBridgeResponse(input.client, requestPath, responsePath, {
|
||||
id: request.id,
|
||||
status: 403,
|
||||
headers: { "content-type": "application/json" },
|
||||
body: JSON.stringify({ error: denialReason }),
|
||||
completedAt: new Date().toISOString(),
|
||||
});
|
||||
await input.client.remove(requestPath);
|
||||
return;
|
||||
}
|
||||
|
||||
try {
|
||||
const result = await input.handleRequest(request);
|
||||
const responseBody = result.body ?? "";
|
||||
if (Buffer.byteLength(responseBody, "utf8") > maxBodyBytes) {
|
||||
throw new Error(`Bridge response body exceeded the configured size limit of ${maxBodyBytes} bytes.`);
|
||||
const guard: RequestFinalizeGuard = {
|
||||
claim: "unclaimed",
|
||||
controller: new AbortController(),
|
||||
finalized: false,
|
||||
};
|
||||
inFlightRequestGuards.set(fileName, guard);
|
||||
// Claim the request for the handler. Return `false` when the recovery path
|
||||
// already claimed it; the caller must then not run the mutation and must not
|
||||
// write a response, because the recovery path writes a 503 and the caller
|
||||
// may retry.
|
||||
const claimForHandler = (): boolean => {
|
||||
if (guard.claim === "abandon") {
|
||||
return false;
|
||||
}
|
||||
await writeBridgeResponse(input.client, requestPath, responsePath, {
|
||||
id: request.id,
|
||||
status: result.status,
|
||||
headers: result.headers ?? {},
|
||||
body: responseBody,
|
||||
completedAt: new Date().toISOString(),
|
||||
});
|
||||
} catch (error) {
|
||||
console.warn(
|
||||
`[paperclip] sandbox callback bridge handler failed for ${request.id}: ${error instanceof Error ? error.message : String(error)}`,
|
||||
);
|
||||
await writeBridgeResponse(input.client, requestPath, responsePath, {
|
||||
id: request.id,
|
||||
status: 502,
|
||||
headers: { "content-type": "application/json" },
|
||||
body: JSON.stringify({
|
||||
error: error instanceof Error ? error.message : String(error),
|
||||
}),
|
||||
completedAt: new Date().toISOString(),
|
||||
});
|
||||
guard.claim = "handler";
|
||||
return true;
|
||||
};
|
||||
// Finalize the request exactly once. Claim it for the handler first. When the
|
||||
// recovery path already won the claim, skip both the write and the remove.
|
||||
// The `finalized` fence stops a double write when the backstop timer already
|
||||
// wrote a 504 for a handler the recovery path aborted. The handler wins when
|
||||
// it settles inside the grace; the backstop wins when the handler never
|
||||
// settles.
|
||||
const finalize = async (response: SandboxCallbackBridgeResponse) => {
|
||||
if (!claimForHandler()) {
|
||||
return;
|
||||
}
|
||||
if (guard.finalized) {
|
||||
return;
|
||||
}
|
||||
guard.finalized = true;
|
||||
// This finalize now owns delivery for the request, so drop a pending
|
||||
// backstop timer. The handler settled inside the grace, so the backstop no
|
||||
// longer needs to wait; `finalize` delivers the terminal response itself,
|
||||
// inline, and never leaves the request file for a detached timer that the
|
||||
// busy poll loop could starve.
|
||||
if (guard.backstopTimer !== undefined) {
|
||||
clearTimeout(guard.backstopTimer);
|
||||
guard.backstopTimer = undefined;
|
||||
}
|
||||
// Write the handler response, bounded by the per-iteration timeout so a
|
||||
// hung sandbox channel never strands the caller until its own generic
|
||||
// deadline. Retry a transient write failure, exactly like the 504 backstop
|
||||
// write.
|
||||
let lastWriteError = "Sandbox callback bridge could not write the handler response.";
|
||||
for (let attempt = 1; attempt <= MAX_BACKSTOP_WRITE_ATTEMPTS; attempt += 1) {
|
||||
try {
|
||||
await withTimeout(
|
||||
writeBridgeResponse(input.client, requestPath, responsePath, response),
|
||||
iterationTimeoutMs,
|
||||
`Sandbox callback bridge write response for ${response.id}`,
|
||||
);
|
||||
await input.client.remove(requestPath).catch(() => undefined);
|
||||
return;
|
||||
} catch (error) {
|
||||
lastWriteError = error instanceof Error ? error.message : String(error);
|
||||
console.warn(
|
||||
`[paperclip] sandbox callback bridge failed to write response for ${response.id} (attempt ${attempt}/${MAX_BACKSTOP_WRITE_ATTEMPTS}): ${lastWriteError}`,
|
||||
);
|
||||
if (attempt < MAX_BACKSTOP_WRITE_ATTEMPTS) {
|
||||
await new Promise((resolve) => setTimeout(resolve, BACKSTOP_WRITE_RETRY_MS));
|
||||
}
|
||||
}
|
||||
}
|
||||
// Every handler-response write failed. Deliver a non-retryable 504 backstop
|
||||
// inline, so the caller gets a terminal response instead of a retryable 503
|
||||
// that repeats a possibly-committed mutation, or a strand until its own
|
||||
// deadline. Roll the `finalized` fence back so `writeAbortedHandlerBackstop`
|
||||
// can proceed; it re-fences, retries the 504 write, and removes the request
|
||||
// file. Await it, so the request file does not linger for the poll loop
|
||||
// while the loop still runs.
|
||||
guard.finalized = false;
|
||||
await writeAbortedHandlerBackstop(fileName, guard, lastWriteError);
|
||||
};
|
||||
try {
|
||||
const raw = await input.client.readTextFile(requestPath);
|
||||
let request: SandboxCallbackBridgeRequest;
|
||||
try {
|
||||
request = JSON.parse(raw) as SandboxCallbackBridgeRequest;
|
||||
} catch {
|
||||
const requestId = fileName.replace(/\.json$/i, "") || randomUUID();
|
||||
await finalize({
|
||||
id: requestId,
|
||||
status: 400,
|
||||
headers: { "content-type": "application/json" },
|
||||
body: JSON.stringify({ error: "Invalid bridge request payload." }),
|
||||
completedAt: new Date().toISOString(),
|
||||
});
|
||||
return;
|
||||
}
|
||||
|
||||
const denialReason = await authorizeRequest(request);
|
||||
if (denialReason) {
|
||||
await finalize({
|
||||
id: request.id,
|
||||
status: 403,
|
||||
headers: { "content-type": "application/json" },
|
||||
body: JSON.stringify({ error: denialReason }),
|
||||
completedAt: new Date().toISOString(),
|
||||
});
|
||||
return;
|
||||
}
|
||||
|
||||
// Claim the request for the handler before the host operation starts. When
|
||||
// the recovery path already claimed it, it writes a 503 and the caller may
|
||||
// retry, so do not run the mutation; the retry then applies it once. When
|
||||
// the handler claims first, the recovery path leaves the request alone and
|
||||
// the handler writes the real response.
|
||||
if (!claimForHandler()) {
|
||||
return;
|
||||
}
|
||||
|
||||
// Build the response, then finalize once. The handler already holds the
|
||||
// claim, so `finalize` writes the real response.
|
||||
let response: SandboxCallbackBridgeResponse;
|
||||
try {
|
||||
const result = await input.handleRequest(request, { signal: guard.controller.signal });
|
||||
const responseBody = result.body ?? "";
|
||||
if (Buffer.byteLength(responseBody, "utf8") > maxBodyBytes) {
|
||||
throw new Error(`Bridge response body exceeded the configured size limit of ${maxBodyBytes} bytes.`);
|
||||
}
|
||||
response = {
|
||||
id: request.id,
|
||||
status: result.status,
|
||||
headers: result.headers ?? {},
|
||||
body: responseBody,
|
||||
completedAt: new Date().toISOString(),
|
||||
};
|
||||
} catch (error) {
|
||||
console.warn(
|
||||
`[paperclip] sandbox callback bridge handler failed for ${request.id}: ${error instanceof Error ? error.message : String(error)}`,
|
||||
);
|
||||
// Tell a worker abort apart from a normal handler failure. The recovery
|
||||
// path aborts `guard.controller` when the per-iteration timeout or the
|
||||
// watchdog fires. The abort reaches this catch only after the handler
|
||||
// claimed the request and started the host operation. The bridge cannot
|
||||
// cancel a host operation that is in flight, so the mutation may have
|
||||
// committed. A 502 (or 503) is a retryable status: the caller retries it
|
||||
// and applies the mutation twice. So return a non-retryable 504 and mark
|
||||
// the outcome indeterminate. The caller must not retry a 504 from the
|
||||
// bridge, unlike the retry-safe 503 that the recovery path writes only
|
||||
// before the host operation starts.
|
||||
if (guard.controller.signal.aborted) {
|
||||
response = {
|
||||
id: request.id,
|
||||
status: 504,
|
||||
headers: {
|
||||
"content-type": "application/json",
|
||||
"x-paperclip-bridge-outcome": "indeterminate",
|
||||
},
|
||||
body: JSON.stringify({
|
||||
error: error instanceof Error ? error.message : String(error),
|
||||
outcome: "indeterminate",
|
||||
retryable: false,
|
||||
}),
|
||||
completedAt: new Date().toISOString(),
|
||||
};
|
||||
} else {
|
||||
response = {
|
||||
id: request.id,
|
||||
status: 502,
|
||||
headers: { "content-type": "application/json" },
|
||||
body: JSON.stringify({
|
||||
error: error instanceof Error ? error.message : String(error),
|
||||
}),
|
||||
completedAt: new Date().toISOString(),
|
||||
};
|
||||
}
|
||||
}
|
||||
await finalize(response);
|
||||
} finally {
|
||||
await input.client.remove(requestPath);
|
||||
// Drop the guard only when it still points to this attempt. A retry can
|
||||
// register a new attempt under the same file name; that new guard must
|
||||
// stay in the map. Keep the guard when a backstop is still pending: a
|
||||
// failed terminal write re-arms the backstop and keeps the request file,
|
||||
// so the guard must stay in the map. The poll loop then skips the file and
|
||||
// does not re-run a possibly-committed mutation before the backstop writes
|
||||
// its 504.
|
||||
if (guard.backstopTimer === undefined && inFlightRequestGuards.get(fileName) === guard) {
|
||||
inFlightRequestGuards.delete(fileName);
|
||||
}
|
||||
}
|
||||
};
|
||||
|
||||
const failPendingRequests = async (message: string) => {
|
||||
const fileNames = await input.client.listJsonFiles(directories.requestsDir).catch(() => []);
|
||||
// Write the non-retryable 504 backstop for an aborted handler that did not
|
||||
// finalize inside the grace. The `finalized` fence makes this a no-op when the
|
||||
// handler already wrote its own response. The request file name is the request
|
||||
// id plus `.json`, so derive the id from it without another client read that
|
||||
// could hang on the same dead channel.
|
||||
const writeAbortedHandlerBackstop = async (
|
||||
fileName: string,
|
||||
guard: RequestFinalizeGuard,
|
||||
message: string,
|
||||
) => {
|
||||
guard.backstopTimer = undefined;
|
||||
if (guard.finalized) {
|
||||
return;
|
||||
}
|
||||
guard.finalized = true;
|
||||
const requestPath = path.posix.join(directories.requestsDir, fileName);
|
||||
const responsePath = path.posix.join(directories.responsesDir, fileName);
|
||||
const requestId = fileName.replace(/\.json$/i, "") || randomUUID();
|
||||
for (let attempt = 1; attempt <= MAX_BACKSTOP_WRITE_ATTEMPTS; attempt += 1) {
|
||||
try {
|
||||
await withTimeout(
|
||||
writeBridgeResponse(input.client, requestPath, responsePath, {
|
||||
id: requestId,
|
||||
status: 504,
|
||||
headers: {
|
||||
"content-type": "application/json",
|
||||
"x-paperclip-bridge-outcome": "indeterminate",
|
||||
},
|
||||
body: JSON.stringify({ error: message, outcome: "indeterminate", retryable: false }),
|
||||
completedAt: new Date().toISOString(),
|
||||
}, {
|
||||
requireRequestPath: false,
|
||||
}),
|
||||
iterationTimeoutMs,
|
||||
`Sandbox callback bridge write 504 backstop for ${requestId}`,
|
||||
);
|
||||
// The 504 backstop reached the caller. Remove the request file, so the
|
||||
// poll loop does not list it again and re-run the mutation.
|
||||
await input.client.remove(requestPath).catch(() => undefined);
|
||||
return;
|
||||
} catch (error) {
|
||||
console.warn(
|
||||
`[paperclip] sandbox callback bridge failed to write 504 backstop for ${requestId} (attempt ${attempt}/${MAX_BACKSTOP_WRITE_ATTEMPTS}): ${error instanceof Error ? error.message : String(error)}`,
|
||||
);
|
||||
if (attempt < MAX_BACKSTOP_WRITE_ATTEMPTS) {
|
||||
await new Promise((resolve) => setTimeout(resolve, BACKSTOP_WRITE_RETRY_MS));
|
||||
}
|
||||
}
|
||||
}
|
||||
// Every backstop write failed. Keep the request file and clear the fence, so a
|
||||
// late handler can still finalize its own 504. The guard stays in the map, so
|
||||
// the poll loop skips the file and does not re-run the mutation. A removed
|
||||
// file plus a set fence would strand the caller until its own deadline and
|
||||
// give it a generic 502 instead of the terminal 504.
|
||||
//
|
||||
// Re-arm the backstop directly. A stuck handler never settles, so its
|
||||
// `processRequestFile` never runs the `finally` that drops the guard. The
|
||||
// poll loop then skips the file on each iteration, so every iteration
|
||||
// succeeds and the watchdog never trips again. No later recovery pass runs,
|
||||
// so a re-arm here is the only path that retries the 504 write. Clear the
|
||||
// fence before the re-arm, because `scheduleAbortedHandlerBackstop` bails on
|
||||
// a set fence. The re-arm only re-writes the 504 response; it never re-runs
|
||||
// the mutation, so a retry cannot apply the mutation twice.
|
||||
guard.finalized = false;
|
||||
scheduleAbortedHandlerBackstop(fileName, guard, message);
|
||||
};
|
||||
|
||||
// Arm the backstop timer for a handler the recovery path just aborted. It is
|
||||
// idempotent: a second recovery pass (the watchdog and the loop catch both run
|
||||
// `failPendingRequests`) does not re-arm a live timer or one that already
|
||||
// finalized.
|
||||
const scheduleAbortedHandlerBackstop = (
|
||||
fileName: string,
|
||||
guard: RequestFinalizeGuard,
|
||||
message: string,
|
||||
) => {
|
||||
if (guard.finalized || guard.backstopTimer !== undefined) {
|
||||
return;
|
||||
}
|
||||
guard.backstopTimer = setTimeout(() => {
|
||||
void writeAbortedHandlerBackstop(fileName, guard, message);
|
||||
}, abortedHandlerGraceMs);
|
||||
if (typeof guard.backstopTimer.unref === "function") {
|
||||
guard.backstopTimer.unref();
|
||||
}
|
||||
};
|
||||
|
||||
// Abort every queued request with a 503. The `abandonInFlight` option controls
|
||||
// the completion fence for a request a `processRequestFile` attempt still owns.
|
||||
// The timeout and watchdog recovery pass `true`: the loop already gave up on
|
||||
// the request. When the handler did not yet start the host operation, claim the
|
||||
// request so a later handler claim bails, then write the 503. When the handler
|
||||
// already started, skip the 503; the recovery cannot cancel an in-flight host
|
||||
// operation, and a 503 there would make the caller retry and apply the mutation
|
||||
// twice. The stop drain passes `false` (the default): a request the loop
|
||||
// already picked up keeps its normal completion, so a late handler result still
|
||||
// wins over the drain 503, exactly like the earlier stop behavior.
|
||||
const failPendingRequests = async (
|
||||
message: string,
|
||||
options: { abandonInFlight?: boolean } = {},
|
||||
) => {
|
||||
if (options.abandonInFlight) {
|
||||
// Abort every in-flight handler first, then arm its 504 backstop. The loop
|
||||
// already gave up on the request. A handler that threads the signal into
|
||||
// its work stops, rejects, and finalizes with its own error response inside
|
||||
// the grace, so the request does not strand. A handler that ignores the
|
||||
// signal and never settles does not finalize inside the grace; the backstop
|
||||
// timer then writes a non-retryable 504, so the request still never
|
||||
// strands. This reads the guard map directly, so it runs even when the
|
||||
// request listing below fails on the same dead channel. It never writes a
|
||||
// 503 for a handler-owned request: the recovery cannot cancel a committed
|
||||
// host mutation, so a retryable status there could apply the mutation twice.
|
||||
for (const [fileName, guard] of inFlightRequestGuards.entries()) {
|
||||
if (guard.claim === "handler") {
|
||||
guard.controller.abort(new Error(message));
|
||||
scheduleAbortedHandlerBackstop(fileName, guard, message);
|
||||
}
|
||||
}
|
||||
}
|
||||
// Wrap each client call in the per-iteration timeout. When the sandbox
|
||||
// channel is unresponsive, a client call hangs with no reject. The timeout
|
||||
// keeps this recovery path fail-fast, so it never re-hangs on the same dead
|
||||
// channel that triggered the recovery.
|
||||
const fileNames = await withTimeout(
|
||||
input.client.listJsonFiles(directories.requestsDir),
|
||||
iterationTimeoutMs,
|
||||
"Sandbox callback bridge list pending requests",
|
||||
).catch(() => [] as string[]);
|
||||
for (const fileName of fileNames) {
|
||||
const guard = inFlightRequestGuards.get(fileName);
|
||||
if (guard && guard.claim === "handler" && (options.abandonInFlight || guard.controller.signal.aborted)) {
|
||||
// The handler already started this request's host operation, or it already
|
||||
// finalized the request. The timeout and watchdog cannot cancel a host
|
||||
// operation that is in flight. A competing 503 here makes the caller retry
|
||||
// while the original mutation still completes, so the mutation applies
|
||||
// twice. Leave the request for the handler to finalize, or for the 504
|
||||
// backstop to finalize when the handler never settles. The recovery path
|
||||
// reaches this skip through `abandonInFlight`. The stop drain reaches it
|
||||
// only when the recovery path already aborted the handler, so a request
|
||||
// whose 504 backstop write failed and kept its file never gets a competing
|
||||
// 503 on stop. A normal in-flight handler at a graceful stop is not
|
||||
// aborted, so it still gets the stop drain 503.
|
||||
continue;
|
||||
}
|
||||
if (options.abandonInFlight && guard) {
|
||||
// The handler did not start the host operation yet. Claim the request, so
|
||||
// the handler bails at its host-operation claim instead of running the
|
||||
// mutation. A retry after the 503 then applies the mutation once.
|
||||
guard.claim = "abandon";
|
||||
}
|
||||
const requestPath = path.posix.join(directories.requestsDir, fileName);
|
||||
const responsePath = path.posix.join(directories.responsesDir, fileName);
|
||||
const requestId = fileName.replace(/\.json$/i, "") || randomUUID();
|
||||
let responseId = requestId;
|
||||
try {
|
||||
const raw = await input.client.readTextFile(requestPath);
|
||||
const parsed = JSON.parse(raw) as Partial<SandboxCallbackBridgeRequest>;
|
||||
await input.client.remove(requestPath).catch(() => undefined);
|
||||
await writeBridgeResponse(input.client, requestPath, responsePath, {
|
||||
id: typeof parsed.id === "string" && parsed.id.length > 0 ? parsed.id : requestId,
|
||||
status: 503,
|
||||
headers: { "content-type": "application/json" },
|
||||
body: JSON.stringify({ error: message }),
|
||||
completedAt: new Date().toISOString(),
|
||||
}, {
|
||||
requireRequestPath: false,
|
||||
});
|
||||
} catch (error) {
|
||||
console.warn(
|
||||
`[paperclip] sandbox callback bridge failed to abort pending request ${requestId}: ${error instanceof Error ? error.message : String(error)}`,
|
||||
const raw = await withTimeout(
|
||||
input.client.readTextFile(requestPath),
|
||||
iterationTimeoutMs,
|
||||
`Sandbox callback bridge read pending request ${requestId}`,
|
||||
);
|
||||
} finally {
|
||||
await input.client.remove(requestPath).catch(() => undefined);
|
||||
const parsed = JSON.parse(raw) as Partial<SandboxCallbackBridgeRequest>;
|
||||
if (typeof parsed.id === "string" && parsed.id.length > 0) {
|
||||
responseId = parsed.id;
|
||||
}
|
||||
} catch (error) {
|
||||
// The read or the parse failed, most likely on the same dead channel that
|
||||
// triggered this recovery. Keep the request file, so a later recovery pass
|
||||
// can still read it and deliver a terminal 503. A remove here drops the
|
||||
// request and strands the caller until its own deadline.
|
||||
console.warn(
|
||||
`[paperclip] sandbox callback bridge could not read pending request ${requestId}: ${error instanceof Error ? error.message : String(error)}`,
|
||||
);
|
||||
continue;
|
||||
}
|
||||
// Write the 503 first, then remove the request file only after the write
|
||||
// lands. Retry a transient failure, bounded by the per-iteration timeout,
|
||||
// exactly like the finalize and 504 backstop writes. When every attempt
|
||||
// fails, keep the request file. A later recovery pass, or the caller retry,
|
||||
// then still finds the queued request and delivers a terminal 503, instead
|
||||
// of a silent drop that strands the caller until its own deadline. The
|
||||
// request is unclaimed or abandoned, so its host mutation never ran; a later
|
||||
// 503 stays exactly-once.
|
||||
let wrote503 = false;
|
||||
let lastWriteError = "Sandbox callback bridge could not write the recovery 503.";
|
||||
for (let attempt = 1; attempt <= MAX_BACKSTOP_WRITE_ATTEMPTS; attempt += 1) {
|
||||
try {
|
||||
await withTimeout(
|
||||
writeBridgeResponse(input.client, requestPath, responsePath, {
|
||||
id: responseId,
|
||||
status: 503,
|
||||
headers: { "content-type": "application/json" },
|
||||
body: JSON.stringify({ error: message }),
|
||||
completedAt: new Date().toISOString(),
|
||||
}, {
|
||||
requireRequestPath: false,
|
||||
}),
|
||||
iterationTimeoutMs,
|
||||
`Sandbox callback bridge write 503 for ${requestId}`,
|
||||
);
|
||||
wrote503 = true;
|
||||
break;
|
||||
} catch (error) {
|
||||
lastWriteError = error instanceof Error ? error.message : String(error);
|
||||
console.warn(
|
||||
`[paperclip] sandbox callback bridge failed to write recovery 503 for ${requestId} (attempt ${attempt}/${MAX_BACKSTOP_WRITE_ATTEMPTS}): ${lastWriteError}`,
|
||||
);
|
||||
if (attempt < MAX_BACKSTOP_WRITE_ATTEMPTS) {
|
||||
await new Promise((resolve) => setTimeout(resolve, BACKSTOP_WRITE_RETRY_MS));
|
||||
}
|
||||
}
|
||||
}
|
||||
if (wrote503) {
|
||||
// The 503 landed. Remove the request file, so the poll loop does not
|
||||
// re-process it.
|
||||
await input.client.remove(requestPath).catch(() => undefined);
|
||||
} else {
|
||||
// Every 503 write failed. Keep the request file for a later recovery pass.
|
||||
console.warn(
|
||||
`[paperclip] sandbox callback bridge kept queued request ${requestId} after every recovery 503 write failed: ${lastWriteError}`,
|
||||
);
|
||||
}
|
||||
}
|
||||
};
|
||||
|
||||
// Surface a bridge-worker failure through the run trace, not only stdout. A
|
||||
// failed span under `input.runtimeSpan` records the error against the live run
|
||||
// span, so the run and the orchestrator see the hang. When no `runtimeSpan`
|
||||
// runner is wired (no injected tracer), the helper still writes a warn line,
|
||||
// so the failure is never silent.
|
||||
const surfaceRunError = async (error: Error) => {
|
||||
if (input.runtimeSpan) {
|
||||
try {
|
||||
await input.runtimeSpan(CALLBACK_BRIDGE_WORKER_FAILED_SPAN, async () => {
|
||||
throw error;
|
||||
});
|
||||
} catch {
|
||||
// `runtimeSpan` re-throws after it records the failed span. The error is
|
||||
// now on the trace; swallow it here so the worker recovery continues.
|
||||
}
|
||||
}
|
||||
console.warn(`[paperclip] ${error.message}`);
|
||||
};
|
||||
|
||||
// The timestamp of the last successful loop iteration. The watchdog compares
|
||||
// it to the current time. The idle branch and every processed request update
|
||||
// it, so steady progress keeps the watchdog re-armed.
|
||||
let lastSuccessfulIterationAt = Date.now();
|
||||
let watchdogTrippedAt: number | null = null;
|
||||
let watchdogTripInFlight = false;
|
||||
// Check often enough to fire soon after the threshold, but not so often that
|
||||
// the check adds load. One fifth of the threshold, with a 10ms floor.
|
||||
const watchdogCheckIntervalMs = Math.max(10, Math.floor(watchdogTimeoutMs / 5));
|
||||
|
||||
const handleWatchdogTrip = async (idleMs: number) => {
|
||||
const message = `Sandbox callback bridge made no successful poll iteration for ${idleMs}ms; the sandbox connection is unresponsive.`;
|
||||
await surfaceRunError(new Error(message));
|
||||
try {
|
||||
await failPendingRequests(message, { abandonInFlight: true });
|
||||
} catch (error) {
|
||||
console.warn(
|
||||
`[paperclip] sandbox callback bridge watchdog failed to abort queued requests: ${error instanceof Error ? error.message : String(error)}`,
|
||||
);
|
||||
}
|
||||
};
|
||||
|
||||
|
|
@ -794,10 +1303,37 @@ export async function startSandboxCallbackBridgeWorker(input: {
|
|||
// its `criticalPath` flag. `runWithoutActiveStep` empties the store for the
|
||||
// loop only; Node keeps the empty store on every later poll continuation.
|
||||
const loop = runWithoutActiveStep(() => (async () => {
|
||||
// The watchdog runs on its own timer, so it fires even while the loop is
|
||||
// stuck on an awaited client call. It is the backstop for a hang the
|
||||
// per-iteration timeout does not catch. `unref` keeps it from holding the
|
||||
// process open. The loop `finally` clears it on every exit.
|
||||
const watchdogTimer = setInterval(() => {
|
||||
if (settled || stopping) return;
|
||||
const idleMs = Date.now() - lastSuccessfulIterationAt;
|
||||
if (idleMs < watchdogTimeoutMs) return;
|
||||
// Fire once per hang period. Re-arm only after the loop advances
|
||||
// `lastSuccessfulIterationAt` past the last trip (a new successful
|
||||
// iteration), so a persistent hang never fires the watchdog repeatedly.
|
||||
if (watchdogTrippedAt !== null && watchdogTrippedAt >= lastSuccessfulIterationAt) return;
|
||||
if (watchdogTripInFlight) return;
|
||||
watchdogTrippedAt = Date.now();
|
||||
watchdogTripInFlight = true;
|
||||
void handleWatchdogTrip(idleMs).finally(() => {
|
||||
watchdogTripInFlight = false;
|
||||
});
|
||||
}, watchdogCheckIntervalMs);
|
||||
if (typeof watchdogTimer.unref === "function") {
|
||||
watchdogTimer.unref();
|
||||
}
|
||||
try {
|
||||
while (true) {
|
||||
const fileNames = await input.client.listJsonFiles(directories.requestsDir);
|
||||
const fileNames = await withTimeout(
|
||||
input.client.listJsonFiles(directories.requestsDir),
|
||||
iterationTimeoutMs,
|
||||
"Sandbox callback bridge list requests",
|
||||
);
|
||||
if (fileNames.length === 0) {
|
||||
lastSuccessfulIterationAt = Date.now();
|
||||
if (stopping) {
|
||||
break;
|
||||
}
|
||||
|
|
@ -815,32 +1351,41 @@ export async function startSandboxCallbackBridgeWorker(input: {
|
|||
// live parent switches to `agent.turn` during the turn and back to
|
||||
// `task.run` after it. Without a runner, the request runs under the
|
||||
// run parent with no wrapper span, exactly like the earlier behavior.
|
||||
await (input.runtimeSpan
|
||||
? input.runtimeSpan(CALLBACK_BRIDGE_RELAY_REQUEST_SPAN, () =>
|
||||
processRequestFile(fileName),
|
||||
)
|
||||
: runWithRuntimeParent(input.getRuntimeParentContext?.(), () =>
|
||||
processRequestFile(fileName),
|
||||
));
|
||||
// The per-iteration timeout wraps the whole request, so a hung
|
||||
// request rejects and the loop `catch` runs `failPendingRequests`.
|
||||
await withTimeout(
|
||||
input.runtimeSpan
|
||||
? input.runtimeSpan(CALLBACK_BRIDGE_RELAY_REQUEST_SPAN, () =>
|
||||
processRequestFile(fileName),
|
||||
)
|
||||
: runWithRuntimeParent(input.getRuntimeParentContext?.(), () =>
|
||||
processRequestFile(fileName),
|
||||
),
|
||||
iterationTimeoutMs,
|
||||
`Sandbox callback bridge process request ${fileName}`,
|
||||
);
|
||||
lastSuccessfulIterationAt = Date.now();
|
||||
} finally {
|
||||
inFlight -= 1;
|
||||
}
|
||||
}
|
||||
lastSuccessfulIterationAt = Date.now();
|
||||
if (stopping && Date.now() >= stopDeadline) {
|
||||
break;
|
||||
}
|
||||
}
|
||||
} catch (error) {
|
||||
const message = buildWorkerFailureMessage(error);
|
||||
console.warn(`[paperclip] ${message}`);
|
||||
await surfaceRunError(new Error(message));
|
||||
try {
|
||||
await failPendingRequests(message);
|
||||
await failPendingRequests(message, { abandonInFlight: true });
|
||||
} catch (failPendingError) {
|
||||
console.warn(
|
||||
`[paperclip] sandbox callback bridge failed to abort queued requests after worker failure: ${failPendingError instanceof Error ? failPendingError.message : String(failPendingError)}`,
|
||||
);
|
||||
}
|
||||
} finally {
|
||||
clearInterval(watchdogTimer);
|
||||
settled = true;
|
||||
if (settleResolve) {
|
||||
settleResolve();
|
||||
|
|
@ -1153,7 +1698,7 @@ export async function startSandboxCallbackBridgeServer(input: {
|
|||
};
|
||||
}
|
||||
|
||||
function getSandboxCallbackBridgeServerSource(): string {
|
||||
export function getSandboxCallbackBridgeServerSource(): string {
|
||||
return `import { randomUUID, timingSafeEqual } from "node:crypto";
|
||||
import { createServer } from "node:http";
|
||||
import { promises as fs } from "node:fs";
|
||||
|
|
@ -1278,8 +1823,21 @@ const server = createServer(async (req, res) => {
|
|||
await fs.rename(tempPath, requestPath);
|
||||
|
||||
const response = await waitForResponse(requestId);
|
||||
res.statusCode = typeof response.status === "number" ? response.status : 200;
|
||||
for (const [key, value] of Object.entries(response.headers || {})) {
|
||||
const responseHeaders = response.headers || {};
|
||||
// The host marks a possibly-committed mutation with an indeterminate outcome.
|
||||
// The host cannot cancel a host operation that is in flight, so the mutation
|
||||
// may have committed before the worker aborted the handler. A 5xx status is
|
||||
// retryable by convention, so a caller that retries 5xx would apply the
|
||||
// mutation twice. Map the indeterminate outcome to a non-retryable 409, so a
|
||||
// standard retry policy does not repeat the request. The outcome header and
|
||||
// body stay, so a caller that reads them still sees the indeterminate result.
|
||||
const bridgeOutcome = responseHeaders["x-paperclip-bridge-outcome"];
|
||||
if (bridgeOutcome === "indeterminate") {
|
||||
res.statusCode = 409;
|
||||
} else {
|
||||
res.statusCode = typeof response.status === "number" ? response.status : 200;
|
||||
}
|
||||
for (const [key, value] of Object.entries(responseHeaders)) {
|
||||
if (typeof value !== "string" || key.toLowerCase() === "content-length") continue;
|
||||
res.setHeader(key, value);
|
||||
}
|
||||
|
|
|
|||
Loading…
Reference in New Issue