fix(sandbox): reset step store for long-lived bridge work
A run-time `sandbox.exec` span attached to a dead startup step. Two bridge startup steps start long-lived work inside their measured step body: a poll timer, socket handlers, and a callback-bridge worker loop. Node snapshots the active-step store on each async resource at creation time, so this work kept the ended step store. Each later run-time exec then read that store and opened its span under the ended step, and it copied a wrong `criticalPath: false` flag. Add an exported `runWithoutActiveStep` helper in startup-timing and wrap the long-lived poll timer, the socket handlers, and the callback-bridge worker loop of both bridge lanes. Each continuation now reads an empty store, so each run-time exec opens an unparented span with no stale `criticalPath` flag. Control flow is unchanged and the change adds no new dependency. Co-authored-by: Paperclip <noreply@paperclip.ing>
This commit is contained in:
parent
bd7a13eb9c
commit
dd6742f7d0
|
|
@ -7,6 +7,7 @@ import {
|
|||
getActiveStepContext,
|
||||
measureStartupStep,
|
||||
normalizeProviderFamily,
|
||||
runWithoutActiveStep,
|
||||
SANDBOX_STARTUP_SPAN_ATTR_PREFIX,
|
||||
SANDBOX_STARTUP_SPAN_ATTRS,
|
||||
setSandboxRootSpanAttributes,
|
||||
|
|
@ -495,6 +496,86 @@ describe("getActiveStepContext", () => {
|
|||
}, { criticalPath: false });
|
||||
expect(seen!.criticalPath).toBe(false);
|
||||
});
|
||||
|
||||
it("keeps the ended step store in a timer scheduled inside the step body", async () => {
|
||||
// Node snapshots the active store on each async resource at creation time.
|
||||
// A timer scheduled inside a measured step body keeps that step store, even
|
||||
// after the step span ends. A later run-time exec then reads the ended step
|
||||
// and parents its `sandbox.exec` span to a dead startup step. This test locks
|
||||
// that leak, so the fix and its guard test stay honest.
|
||||
let storeInTimer: ReturnType<typeof getActiveStepContext> = null;
|
||||
let fireTimer!: () => void;
|
||||
const timerRan = new Promise<void>((resolve) => {
|
||||
fireTimer = resolve;
|
||||
});
|
||||
|
||||
await measureStartupStep(
|
||||
{ onEvent: vi.fn(async () => {}) },
|
||||
() => 0,
|
||||
"bridge.process-session",
|
||||
async () => {
|
||||
setTimeout(() => {
|
||||
storeInTimer = getActiveStepContext();
|
||||
fireTimer();
|
||||
}, 0);
|
||||
return "started";
|
||||
},
|
||||
{ criticalPath: false },
|
||||
);
|
||||
|
||||
// The step span already ended, so the main line reads no active step.
|
||||
expect(getActiveStepContext()).toBeNull();
|
||||
|
||||
await timerRan;
|
||||
|
||||
// Yet the timer callback still reads the ended step context. This is the
|
||||
// store leak the bridge boundary fix removes.
|
||||
expect(storeInTimer).not.toBeNull();
|
||||
expect(storeInTimer!.criticalPath).toBe(false);
|
||||
});
|
||||
|
||||
it("clears the active step for a continuation wrapped in runWithoutActiveStep", async () => {
|
||||
// The bridge boundary wraps its long-lived poll timer in `runWithoutActiveStep`.
|
||||
// A timer scheduled inside that empty store scope reads no active step, so a
|
||||
// later run-time exec opens an unparented span instead of one under the ended
|
||||
// startup step.
|
||||
let storeInTimer: ReturnType<typeof getActiveStepContext> = null;
|
||||
let sawTimer = false;
|
||||
let fireTimer!: () => void;
|
||||
const timerRan = new Promise<void>((resolve) => {
|
||||
fireTimer = resolve;
|
||||
});
|
||||
|
||||
await measureStartupStep(
|
||||
{ onEvent: vi.fn(async () => {}) },
|
||||
() => 0,
|
||||
"bridge.process-session",
|
||||
async () => {
|
||||
runWithoutActiveStep(() => {
|
||||
setTimeout(() => {
|
||||
storeInTimer = getActiveStepContext();
|
||||
sawTimer = true;
|
||||
fireTimer();
|
||||
}, 0);
|
||||
});
|
||||
return "started";
|
||||
},
|
||||
{ criticalPath: false },
|
||||
);
|
||||
|
||||
await timerRan;
|
||||
|
||||
expect(sawTimer).toBe(true);
|
||||
expect(storeInTimer).toBeNull();
|
||||
});
|
||||
|
||||
it("returns the work result and restores the previous active step", async () => {
|
||||
// Outside any measured step the previous store is empty, so the helper both
|
||||
// returns the work value and leaves the store empty afterward.
|
||||
const value = runWithoutActiveStep(() => "value");
|
||||
expect(value).toBe("value");
|
||||
expect(getActiveStepContext()).toBeNull();
|
||||
});
|
||||
});
|
||||
|
||||
describe("clampSpanLabel", () => {
|
||||
|
|
|
|||
|
|
@ -332,7 +332,7 @@ export interface ActiveStepContext {
|
|||
* a module-level singleton, so the value propagates across `await` boundaries
|
||||
* and across package boundaries that share this module.
|
||||
*/
|
||||
const activeStepContextStorage = new AsyncLocalStorage<ActiveStepContext>();
|
||||
const activeStepContextStorage = new AsyncLocalStorage<ActiveStepContext | undefined>();
|
||||
|
||||
/**
|
||||
* Return the active step context, or `null` when no measured step is running.
|
||||
|
|
@ -344,6 +344,26 @@ export function getActiveStepContext(): ActiveStepContext | null {
|
|||
return activeStepContextStorage.getStore() ?? null;
|
||||
}
|
||||
|
||||
/**
|
||||
* Run `work` with no active step context, then restore the previous store. A
|
||||
* bridge boundary uses this to start its long-lived poll timer and socket
|
||||
* handlers outside the measured step store.
|
||||
*
|
||||
* Node snapshots the active store on each async resource at creation time. So a
|
||||
* timer or a handler scheduled inside a measured step body keeps that step store
|
||||
* after the step span ends. A later run-time exec then reads the ended step and
|
||||
* parents its `sandbox.exec` span to a dead startup step, and it copies the
|
||||
* step's `criticalPath` flag. This helper resets the store for the wrapped work,
|
||||
* so each continuation reads an empty store. Each run-time exec then opens an
|
||||
* unparented span with no stale `criticalPath` flag.
|
||||
*
|
||||
* The helper forwards only the opaque store, so this package stays free of
|
||||
* `@opentelemetry/api`. It needs no Node version gate.
|
||||
*/
|
||||
export function runWithoutActiveStep<T>(work: () => T): T {
|
||||
return activeStepContextStorage.run(undefined, work);
|
||||
}
|
||||
|
||||
/**
|
||||
* Set a numeric span attribute only when the value is a finite number. A reader
|
||||
* that returns `undefined` (the counter is unavailable) yields no attribute,
|
||||
|
|
|
|||
|
|
@ -46,6 +46,7 @@ import {
|
|||
} from "./server-utils.js";
|
||||
import { sanitizeRemoteExecutionEnv } from "./remote-execution-env.js";
|
||||
import { preferredShellForSandbox, shellCommandArgs } from "./sandbox-shell.js";
|
||||
import { runWithoutActiveStep } from "./acpx-engine/startup-timing.js";
|
||||
import type { RuntimeProgressSink, RuntimeStatusSink } from "./runtime-progress.js";
|
||||
import type { LocalProcessSandboxOptions } from "./local-process-sandbox.js";
|
||||
|
||||
|
|
@ -1488,7 +1489,10 @@ export async function startAdapterExecutionTargetProcessSessionBridge(input: {
|
|||
};
|
||||
|
||||
const liveSockets = new Set<net.Socket>();
|
||||
const server = net.createServer((nextSocket) => {
|
||||
// Register the per-connection socket handlers outside the measured bridge step
|
||||
// store. A stdin write from a socket handler is a run-time exec, not startup
|
||||
// work, so its `sandbox.exec` span must not parent to the ended bridge step.
|
||||
const server = net.createServer((nextSocket) => runWithoutActiveStep(() => {
|
||||
liveSockets.add(nextSocket);
|
||||
nextSocket.setEncoding("utf8");
|
||||
nextSocket.on("error", () => undefined);
|
||||
|
|
@ -1547,7 +1551,7 @@ export async function startAdapterExecutionTargetProcessSessionBridge(input: {
|
|||
});
|
||||
}
|
||||
});
|
||||
});
|
||||
}));
|
||||
|
||||
const poll = async () => {
|
||||
if (stopping) return;
|
||||
|
|
@ -1572,16 +1576,24 @@ export async function startAdapterExecutionTargetProcessSessionBridge(input: {
|
|||
return;
|
||||
} finally {
|
||||
if (!stopping) {
|
||||
pollTimer = setTimeout(() => void poll(), 100);
|
||||
pollTimer.unref?.();
|
||||
schedulePoll();
|
||||
}
|
||||
}
|
||||
};
|
||||
|
||||
// Schedule the long-lived poll timer outside the measured bridge step store.
|
||||
// The poll loop reads remote event files with run-time execs, not startup
|
||||
// work, so a poll `sandbox.exec` span must not parent to the ended bridge step.
|
||||
// `runWithoutActiveStep` also empties the store for the re-arm timer that the
|
||||
// poll body schedules, so every later tick stays unparented too.
|
||||
const schedulePoll = () => {
|
||||
pollTimer = setTimeout(() => runWithoutActiveStep(() => void poll()), 100);
|
||||
pollTimer.unref?.();
|
||||
};
|
||||
|
||||
const port = await waitForLocalServerListen(server);
|
||||
const agentCommand = await writeProcessSessionProxyScript(proxyDir, port, token);
|
||||
pollTimer = setTimeout(() => void poll(), 100);
|
||||
pollTimer.unref?.();
|
||||
schedulePoll();
|
||||
|
||||
return {
|
||||
agentCommand,
|
||||
|
|
@ -1787,7 +1799,13 @@ export async function startAdapterExecutionTargetPaperclipBridge(input: {
|
|||
// this flag is enabled. Only intended for active debugging in trusted
|
||||
// environments.
|
||||
const bridgeDebugEnabled = isBridgeDebugEnabled(process.env);
|
||||
worker = await startSandboxCallbackBridgeWorker({
|
||||
// Start the long-lived callback-bridge worker loop outside the measured
|
||||
// bridge step store. The loop reads and writes queue files with run-time
|
||||
// execs for the whole run, not startup work, so a worker `sandbox.exec` span
|
||||
// must not parent to the ended `bridge.paperclip` step or copy its
|
||||
// `criticalPath` flag. `runWithoutActiveStep` empties the store for the loop
|
||||
// that the worker start schedules, so every later poll tick stays unparented.
|
||||
worker = await runWithoutActiveStep(() => startSandboxCallbackBridgeWorker({
|
||||
client,
|
||||
queueDir,
|
||||
maxBodyBytes,
|
||||
|
|
@ -1824,7 +1842,7 @@ export async function startAdapterExecutionTargetPaperclipBridge(input: {
|
|||
body: await readBridgeForwardResponseBody(response, maxBodyBytes),
|
||||
};
|
||||
},
|
||||
});
|
||||
}));
|
||||
server = await startSandboxCallbackBridgeServer({
|
||||
runner,
|
||||
remoteCwd: target.remoteCwd,
|
||||
|
|
|
|||
|
|
@ -10,6 +10,7 @@ vi.mock("../services/environment-config.js", () => ({
|
|||
|
||||
import {
|
||||
measureStartupStep,
|
||||
runWithoutActiveStep,
|
||||
SANDBOX_STARTUP_SPAN_ATTRS,
|
||||
} from "@paperclipai/adapter-utils/acpx-engine/startup-timing";
|
||||
import {
|
||||
|
|
@ -847,4 +848,116 @@ describe("resolveEnvironmentExecutionTarget", () => {
|
|||
// The seam reached the log callback exactly once (the stdout delivery).
|
||||
expect(onLog).toHaveBeenCalledTimes(1);
|
||||
});
|
||||
|
||||
// Fire one run-time exec from a bridge continuation that runs after the step
|
||||
// span ended. Each bridge step (`bridge.paperclip`, `bridge.process-session`)
|
||||
// starts long-lived work with `criticalPath: false`. The bridge boundary wraps
|
||||
// that long-lived work in `runWithoutActiveStep`, exactly as modeled here, so
|
||||
// the continuation reads an empty active step. Return the recorded exec span.
|
||||
async function runContinuationExec(step: string, options: { wrap: boolean }) {
|
||||
const { tracer, contextWithSpan, spans } = createRecordingTrace();
|
||||
const runner = await runnerFor({
|
||||
provider: "daytona",
|
||||
execResult: { exitCode: 0, signal: null, timedOut: false, stdout: "", stderr: "" },
|
||||
tracer,
|
||||
});
|
||||
|
||||
let resolveExec!: () => void;
|
||||
const execDone = new Promise<void>((resolve) => {
|
||||
resolveExec = resolve;
|
||||
});
|
||||
|
||||
// Schedule the exec from a timer inside the step body, so it fires after the
|
||||
// step span ends. The `wrap` flag models the fix: when true, the boundary
|
||||
// wraps the long-lived work in `runWithoutActiveStep`; when false, it models
|
||||
// the pre-fix leak.
|
||||
const scheduleContinuation = () => {
|
||||
setTimeout(() => {
|
||||
void runner.execute({ command: "echo" }).then(() => resolveExec());
|
||||
}, 0);
|
||||
};
|
||||
|
||||
await measureStartupStep(
|
||||
{},
|
||||
() => 0,
|
||||
step,
|
||||
async () => {
|
||||
if (options.wrap) {
|
||||
runWithoutActiveStep(scheduleContinuation);
|
||||
} else {
|
||||
scheduleContinuation();
|
||||
}
|
||||
return "started";
|
||||
},
|
||||
{ tracer, contextWithSpan, criticalPath: false },
|
||||
);
|
||||
|
||||
await execDone;
|
||||
return spans.find((span) => span.name === "sandbox.exec");
|
||||
}
|
||||
|
||||
it("opens an unparented exec span for a process-session bridge continuation", async () => {
|
||||
const execSpan = await runContinuationExec("bridge.process-session", { wrap: true });
|
||||
expect(execSpan).toBeTruthy();
|
||||
// The step span ended and the boundary emptied the store, so the continuation
|
||||
// exec opens a root span, not one under the dead bridge step.
|
||||
expect(execSpan!.parent).toBeNull();
|
||||
});
|
||||
|
||||
it("opens an unparented exec span for a paperclip bridge continuation", async () => {
|
||||
const execSpan = await runContinuationExec("bridge.paperclip", { wrap: true });
|
||||
expect(execSpan).toBeTruthy();
|
||||
expect(execSpan!.parent).toBeNull();
|
||||
});
|
||||
|
||||
it("does not copy the stale criticalPath = false flag onto a continuation exec", async () => {
|
||||
const execSpan = await runContinuationExec("bridge.process-session", { wrap: true });
|
||||
expect(execSpan).toBeTruthy();
|
||||
// The bridge step set `criticalPath: false`. The continuation reads an empty
|
||||
// store, so the exec span records the default `true`, never the stale `false`.
|
||||
expect(execSpan!.attributes[A.execCriticalPath]).toBe(true);
|
||||
expect(execSpan!.attributes[A.execCriticalPath]).not.toBe(false);
|
||||
});
|
||||
|
||||
it("leaks the ended step onto a continuation exec without the boundary wrap", async () => {
|
||||
// The mechanism guard: an unwrapped continuation keeps the ended bridge step
|
||||
// store, so the exec span parents to the dead step and copies its
|
||||
// `criticalPath: false`. The boundary wrap in the two tests above removes both
|
||||
// defects, so this suite fails if a future edit drops the wrap.
|
||||
const { tracer, contextWithSpan, spans } = createRecordingTrace();
|
||||
const runner = await runnerFor({
|
||||
provider: "daytona",
|
||||
execResult: { exitCode: 0, signal: null, timedOut: false, stdout: "", stderr: "" },
|
||||
tracer,
|
||||
});
|
||||
|
||||
let resolveExec!: () => void;
|
||||
const execDone = new Promise<void>((resolve) => {
|
||||
resolveExec = resolve;
|
||||
});
|
||||
|
||||
await measureStartupStep(
|
||||
{},
|
||||
() => 0,
|
||||
"bridge.process-session",
|
||||
async () => {
|
||||
setTimeout(() => {
|
||||
void runner.execute({ command: "echo" }).then(() => resolveExec());
|
||||
}, 0);
|
||||
return "started";
|
||||
},
|
||||
{ tracer, contextWithSpan, criticalPath: false },
|
||||
);
|
||||
|
||||
await execDone;
|
||||
|
||||
const stepSpan = spans.find((span) => span.name === "bridge.process-session");
|
||||
const execSpan = spans.find((span) => span.name === "sandbox.exec");
|
||||
expect(stepSpan).toBeTruthy();
|
||||
expect(execSpan).toBeTruthy();
|
||||
// The unwrapped continuation parents the exec span to the ended step and
|
||||
// copies the stale flag.
|
||||
expect(execSpan!.parent).toBe(stepSpan);
|
||||
expect(execSpan!.attributes[A.execCriticalPath]).toBe(false);
|
||||
});
|
||||
});
|
||||
|
|
|
|||
Loading…
Reference in New Issue