mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-10-26 18:31:34 +01:00
feat(adapter-utils): bound the http2 bridge live session count before the bind
Reserve one wrapper budget from the http2 bridge admission gate before the host binds the sandbox channel to an HTTP/2 session. A run that finds no free session slot disposes the pending replay, closes the partial channel, reports the new host_session_capacity fallback reason, and uses the file bridge instead. The host releases the reserved slot exactly one time, on the bound channel's close event and on a bind failure. Co-authored-by: Paperclip <noreply@paperclip.ing>
This commit is contained in:
1 parent
4f7c423cd3
commit
be1a212c30
4 files changed
+486
-55
No files matched your search
@@ -108,7 +108,11 @@ export type DuplexOutcomeValue = "ok" | "error";
|
||||
* names a missing or an invalid HTTP/2 client connection preface inside the
|
||||
* bounded readiness buffer: the host found no valid preface after the
|
||||
* accepted READY line, aborted the HTTP/2 open, and moved the run to
|
||||
* `queue_v1` one time.
|
||||
* `queue_v1` one time. The `host_session_capacity` reason names a full live
|
||||
* HTTP/2 bridge session cap: every host-wide session slot was in use, so the
|
||||
* host never bound the channel to an HTTP/2 session. This differs from
|
||||
* `route_busy`, which names a full worker-manager route ceiling, a separate
|
||||
* cap on a separate resource.
|
||||
*/
|
||||
export type DuplexFallbackReason =
|
||||
| "gate_off"
|
||||
@@ -122,7 +126,8 @@ export type DuplexFallbackReason =
|
||||
| "ready_timeout"
|
||||
| "contaminated"
|
||||
| "aggregate_bytes_exceeded"
|
||||
| "preface_missing";
|
||||
| "preface_missing"
|
||||
| "host_session_capacity";
|
||||
|
||||
/** The class of a terminal loss, relative to the first request dispatch. */
|
||||
export type DuplexLossClass = "pre_dispatch" | "post_dispatch";
|
||||
|
||||
@@ -1,7 +1,7 @@
|
||||
import { createServer } from "node:http";
|
||||
import http2 from "node:http2";
|
||||
import net from "node:net";
|
||||
import { duplexPair, type Duplex } from "node:stream";
|
||||
import { duplexPair, Duplex } from "node:stream";
|
||||
import { execFile, spawn } from "node:child_process";
|
||||
import { mkdir, mkdtemp, readdir, readFile, rm, writeFile } from "node:fs/promises";
|
||||
import os from "node:os";
|
||||
@@ -13,6 +13,7 @@ import { getSandboxDuplexGatewayCodecSource } from "./sandbox-callback-bridge.js
|
||||
|
||||
import {
|
||||
__duplexReadinessTesting,
|
||||
__http2BridgeSessionAdmissionTesting,
|
||||
__http2PrefaceScanTesting,
|
||||
DEFAULT_REMOTE_SANDBOX_ADAPTER_TIMEOUT_SEC,
|
||||
adapterExecutionTargetDuplexObservabilityRecorder,
|
||||
@@ -58,6 +59,12 @@ import {
|
||||
encodeDuplexFrame,
|
||||
} from "./duplex-frame-codec.js";
|
||||
import { DUPLEX_CHANNEL_LOST_ERROR_CODE } from "./bridge-transport-contract.js";
|
||||
import {
|
||||
configureHttp2BridgeAdmissionGate,
|
||||
getHttp2BridgeAdmissionGate,
|
||||
Http2BridgeAdmissionGate,
|
||||
} from "./http2-bridge-admission.js";
|
||||
import type { Http2BridgeServerHandle } from "./http2-bridge-server.js";
|
||||
import {
|
||||
DUPLEX_AGGREGATE_BYTE_LEDGER_METRIC_NAMES,
|
||||
DUPLEX_COUNTER_AGGREGATE_BYTE_ACCOUNTING_UNDERFLOW_TOTAL,
|
||||
@@ -171,6 +178,14 @@ describe("sandbox adapter execution targets", () => {
|
||||
}
|
||||
});
|
||||
|
||||
// The host HTTP/2 bridge admission gate is one process-wide singleton
|
||||
// (`getHttp2BridgeAdmissionGate`). A test that configures a small session
|
||||
// cap, or reserves a session slot directly, must not leak that state into
|
||||
// a later test in this file.
|
||||
afterEach(() => {
|
||||
configureHttp2BridgeAdmissionGate({});
|
||||
});
|
||||
|
||||
function createLocalSandboxRunner() {
|
||||
let counter = 0;
|
||||
return {
|
||||
@@ -5543,6 +5558,291 @@ describe("sandbox adapter execution targets", () => {
|
||||
}
|
||||
}, 20000);
|
||||
|
||||
it("a run past the live session cap binds no HTTP/2 session and uses the file bridge", async () => {
|
||||
const rootDir = await mkdtemp(path.join(os.tmpdir(), "paperclip-http2-session-cap-mode-"));
|
||||
cleanupDirs.push(rootDir);
|
||||
const remoteCwd = path.join(rootDir, "workspace");
|
||||
await mkdir(remoteCwd, { recursive: true });
|
||||
const api = await startRecordingApiServer();
|
||||
const { runner, control } = makeHttp2SelectionRunner((ctx) => {
|
||||
ctx.emitReady();
|
||||
// A valid client preface still arrives — the session cap, not a
|
||||
// preface failure, is what must turn this run away.
|
||||
ctx.connectHttp2();
|
||||
});
|
||||
const target: AdapterSandboxExecutionTarget = {
|
||||
kind: "remote",
|
||||
transport: "sandbox",
|
||||
providerKey: "daytona",
|
||||
remoteCwd,
|
||||
timeoutMs: 30_000,
|
||||
runner,
|
||||
effectiveCapabilities: duplexCapabilities(true),
|
||||
};
|
||||
|
||||
// Fill the one live-session slot the cap allows, before the run even
|
||||
// starts, so the run below finds no free slot.
|
||||
configureHttp2BridgeAdmissionGate({ maxSessions: 1 });
|
||||
const releasePreexistingSession = getHttp2BridgeAdmissionGate().tryAcquireSession();
|
||||
expect(releasePreexistingSession).not.toBeNull();
|
||||
|
||||
const bridge = await startAdapterExecutionTargetPaperclipBridge({
|
||||
runId: "run-session-cap-mode",
|
||||
target,
|
||||
runtimeRootDir: path.join(remoteCwd, ".paperclip-runtime", "codex"),
|
||||
adapterKey: "codex",
|
||||
hostApiToken: "real-run-jwt",
|
||||
hostApiUrl: api.origin,
|
||||
enableSandboxDuplexBridge: true,
|
||||
});
|
||||
try {
|
||||
expect(bridge).not.toBeNull();
|
||||
expect(control.openCount).toBe(1);
|
||||
// The live-session cap was already full: the host never bound this
|
||||
// channel to an HTTP/2 session, so the run serves over the file bridge.
|
||||
expect(bridge?.env.PAPERCLIP_API_BRIDGE_MODE).toBe("queue_v1");
|
||||
} finally {
|
||||
releasePreexistingSession?.();
|
||||
await bridge?.stop();
|
||||
await api.close();
|
||||
}
|
||||
}, 20000);
|
||||
|
||||
it("a refused session disposes the pending replay and closes the partial channel inside the cleanup budget", async () => {
|
||||
const rootDir = await mkdtemp(path.join(os.tmpdir(), "paperclip-http2-session-cap-cleanup-"));
|
||||
cleanupDirs.push(rootDir);
|
||||
const remoteCwd = path.join(rootDir, "workspace");
|
||||
await mkdir(remoteCwd, { recursive: true });
|
||||
const api = await startRecordingApiServer();
|
||||
const { runner, control } = makeHttp2SelectionRunner((ctx) => {
|
||||
ctx.emitReady();
|
||||
ctx.connectHttp2();
|
||||
});
|
||||
const target: AdapterSandboxExecutionTarget = {
|
||||
kind: "remote",
|
||||
transport: "sandbox",
|
||||
providerKey: "daytona",
|
||||
remoteCwd,
|
||||
timeoutMs: 30_000,
|
||||
runner,
|
||||
effectiveCapabilities: duplexCapabilities(true),
|
||||
};
|
||||
|
||||
configureHttp2BridgeAdmissionGate({ maxSessions: 1 });
|
||||
const releasePreexistingSession = getHttp2BridgeAdmissionGate().tryAcquireSession();
|
||||
expect(releasePreexistingSession).not.toBeNull();
|
||||
|
||||
const bridge = await startAdapterExecutionTargetPaperclipBridge({
|
||||
runId: "run-session-cap-cleanup",
|
||||
target,
|
||||
runtimeRootDir: path.join(remoteCwd, ".paperclip-runtime", "codex"),
|
||||
adapterKey: "codex",
|
||||
hostApiToken: "real-run-jwt",
|
||||
hostApiUrl: api.origin,
|
||||
enableSandboxDuplexBridge: true,
|
||||
});
|
||||
try {
|
||||
expect(bridge).not.toBeNull();
|
||||
// The refused session closed the partial channel inside the bounded
|
||||
// cleanup budget: no live provider session remains.
|
||||
expect(control.closeCount + control.stopCount).toBeGreaterThanOrEqual(1);
|
||||
} finally {
|
||||
releasePreexistingSession?.();
|
||||
await bridge?.stop();
|
||||
await api.close();
|
||||
}
|
||||
}, 20000);
|
||||
|
||||
it("the fallback for a full session cap reports host_session_capacity", async () => {
|
||||
const rootDir = await mkdtemp(path.join(os.tmpdir(), "paperclip-http2-session-cap-reason-"));
|
||||
cleanupDirs.push(rootDir);
|
||||
const remoteCwd = path.join(rootDir, "workspace");
|
||||
await mkdir(remoteCwd, { recursive: true });
|
||||
const api = await startRecordingApiServer();
|
||||
const { runner } = makeHttp2SelectionRunner((ctx) => {
|
||||
ctx.emitReady();
|
||||
ctx.connectHttp2();
|
||||
});
|
||||
const { recorder, counters } = createRecordingDuplexRecorder();
|
||||
const target: AdapterSandboxExecutionTarget = {
|
||||
kind: "remote",
|
||||
transport: "sandbox",
|
||||
providerKey: "daytona",
|
||||
remoteCwd,
|
||||
timeoutMs: 30_000,
|
||||
runner,
|
||||
effectiveCapabilities: duplexCapabilities(true),
|
||||
};
|
||||
|
||||
configureHttp2BridgeAdmissionGate({ maxSessions: 1 });
|
||||
const releasePreexistingSession = getHttp2BridgeAdmissionGate().tryAcquireSession();
|
||||
expect(releasePreexistingSession).not.toBeNull();
|
||||
|
||||
const bridge = await startAdapterExecutionTargetPaperclipBridge({
|
||||
runId: "run-session-cap-reason",
|
||||
target,
|
||||
runtimeRootDir: path.join(remoteCwd, ".paperclip-runtime", "codex"),
|
||||
adapterKey: "codex",
|
||||
hostApiToken: "real-run-jwt",
|
||||
hostApiUrl: api.origin,
|
||||
enableSandboxDuplexBridge: true,
|
||||
duplexObservabilityRecorder: recorder,
|
||||
});
|
||||
try {
|
||||
expect(bridge).not.toBeNull();
|
||||
const fallback = counters.find((c) => c.metric === DUPLEX_COUNTER_FALLBACK_TOTAL);
|
||||
expect(fallback?.dimensions.fallback_reason).toBe("host_session_capacity");
|
||||
} finally {
|
||||
releasePreexistingSession?.();
|
||||
await bridge?.stop();
|
||||
await api.close();
|
||||
}
|
||||
}, 20000);
|
||||
|
||||
it("a closed session frees its slot for the next run", async () => {
|
||||
configureHttp2BridgeAdmissionGate({ maxSessions: 1 });
|
||||
|
||||
const rootDir1 = await mkdtemp(path.join(os.tmpdir(), "paperclip-http2-session-reuse-1-"));
|
||||
cleanupDirs.push(rootDir1);
|
||||
const remoteCwd1 = path.join(rootDir1, "workspace");
|
||||
await mkdir(remoteCwd1, { recursive: true });
|
||||
const api1 = await startRecordingApiServer();
|
||||
const sessionRef1: { current: http2.ClientHttp2Session | null } = { current: null };
|
||||
const { runner: runner1 } = makeHttp2SelectionRunner((ctx) => {
|
||||
ctx.emitReady();
|
||||
sessionRef1.current = ctx.connectHttp2();
|
||||
});
|
||||
const target1: AdapterSandboxExecutionTarget = {
|
||||
kind: "remote",
|
||||
transport: "sandbox",
|
||||
providerKey: "daytona",
|
||||
remoteCwd: remoteCwd1,
|
||||
timeoutMs: 30_000,
|
||||
runner: runner1,
|
||||
effectiveCapabilities: duplexCapabilities(true),
|
||||
};
|
||||
const bridge1 = await startAdapterExecutionTargetPaperclipBridge({
|
||||
runId: "run-session-reuse-1",
|
||||
target: target1,
|
||||
runtimeRootDir: path.join(remoteCwd1, ".paperclip-runtime", "codex"),
|
||||
adapterKey: "codex",
|
||||
hostApiToken: "real-run-jwt",
|
||||
hostApiUrl: api1.origin,
|
||||
enableSandboxDuplexBridge: true,
|
||||
});
|
||||
expect(bridge1?.env.PAPERCLIP_API_BRIDGE_MODE).toBe("http2_v1");
|
||||
// The first run holds the one session slot the cap allows.
|
||||
expect(getHttp2BridgeAdmissionGate().activeSessionCount).toBe(1);
|
||||
|
||||
await bridge1?.stop();
|
||||
sessionRef1.current?.close();
|
||||
await api1.close();
|
||||
// `stop()` released the slot, either through the bound `Duplex`'s
|
||||
// `close` event or the idempotent safety-net release inside `stop()`.
|
||||
expect(getHttp2BridgeAdmissionGate().activeSessionCount).toBe(0);
|
||||
|
||||
const rootDir2 = await mkdtemp(path.join(os.tmpdir(), "paperclip-http2-session-reuse-2-"));
|
||||
cleanupDirs.push(rootDir2);
|
||||
const remoteCwd2 = path.join(rootDir2, "workspace");
|
||||
await mkdir(remoteCwd2, { recursive: true });
|
||||
const api2 = await startRecordingApiServer();
|
||||
const sessionRef2: { current: http2.ClientHttp2Session | null } = { current: null };
|
||||
const { runner: runner2 } = makeHttp2SelectionRunner((ctx) => {
|
||||
ctx.emitReady();
|
||||
sessionRef2.current = ctx.connectHttp2();
|
||||
});
|
||||
const target2: AdapterSandboxExecutionTarget = {
|
||||
kind: "remote",
|
||||
transport: "sandbox",
|
||||
providerKey: "daytona",
|
||||
remoteCwd: remoteCwd2,
|
||||
timeoutMs: 30_000,
|
||||
runner: runner2,
|
||||
effectiveCapabilities: duplexCapabilities(true),
|
||||
};
|
||||
const bridge2 = await startAdapterExecutionTargetPaperclipBridge({
|
||||
runId: "run-session-reuse-2",
|
||||
target: target2,
|
||||
runtimeRootDir: path.join(remoteCwd2, ".paperclip-runtime", "codex"),
|
||||
adapterKey: "codex",
|
||||
hostApiToken: "real-run-jwt",
|
||||
hostApiUrl: api2.origin,
|
||||
enableSandboxDuplexBridge: true,
|
||||
});
|
||||
try {
|
||||
// The freed slot let a second run select the HTTP/2 transport again.
|
||||
expect(bridge2?.env.PAPERCLIP_API_BRIDGE_MODE).toBe("http2_v1");
|
||||
} finally {
|
||||
sessionRef2.current?.close();
|
||||
await bridge2?.stop();
|
||||
await api2.close();
|
||||
}
|
||||
}, 20000);
|
||||
|
||||
describe("bindHttp2BridgeSessionWithAdmission", () => {
|
||||
function fakeChannelForAdmissionTest(): CommandManagedDuplexChannel {
|
||||
return {
|
||||
write: () => {},
|
||||
onData: () => {},
|
||||
onExit: () => {},
|
||||
stop: () => {},
|
||||
close: async () => {},
|
||||
};
|
||||
}
|
||||
|
||||
it("a bind failure frees the session slot exactly one time", () => {
|
||||
const gate = new Http2BridgeAdmissionGate({ maxParallel: 4, maxSessions: 4 });
|
||||
const bindError = new Error("simulated http2 bindChannel failure");
|
||||
const fakeHandle: Http2BridgeServerHandle = {
|
||||
server: {} as unknown as http2.Http2Server,
|
||||
bindChannel: () => {
|
||||
throw bindError;
|
||||
},
|
||||
close: async () => {},
|
||||
};
|
||||
expect(() =>
|
||||
__http2BridgeSessionAdmissionTesting.bindHttp2BridgeSessionWithAdmission(
|
||||
gate,
|
||||
fakeHandle,
|
||||
fakeChannelForAdmissionTest(),
|
||||
),
|
||||
).toThrow(bindError);
|
||||
// The reservation never outlives a channel that never bound.
|
||||
expect(gate.activeSessionCount).toBe(0);
|
||||
});
|
||||
|
||||
it("the host reserves the session slot before bindChannel, not after", () => {
|
||||
const gate = new Http2BridgeAdmissionGate({ maxParallel: 4, maxSessions: 4 });
|
||||
const order: string[] = [];
|
||||
const originalTryAcquireSession = gate.tryAcquireSession.bind(gate);
|
||||
gate.tryAcquireSession = () => {
|
||||
order.push("reserve");
|
||||
return originalTryAcquireSession();
|
||||
};
|
||||
const fakeDuplex = new Duplex({
|
||||
read() {},
|
||||
write(_chunk, _encoding, callback) {
|
||||
callback();
|
||||
},
|
||||
});
|
||||
const fakeHandle: Http2BridgeServerHandle = {
|
||||
server: {} as unknown as http2.Http2Server,
|
||||
bindChannel: () => {
|
||||
order.push("bind");
|
||||
return fakeDuplex;
|
||||
},
|
||||
close: async () => {},
|
||||
};
|
||||
const result = __http2BridgeSessionAdmissionTesting.bindHttp2BridgeSessionWithAdmission(
|
||||
gate,
|
||||
fakeHandle,
|
||||
fakeChannelForAdmissionTest(),
|
||||
);
|
||||
expect(result).not.toBeNull();
|
||||
expect(order).toEqual(["reserve", "bind"]);
|
||||
});
|
||||
});
|
||||
|
||||
it("test_the_host_writes_no_byte_before_the_ready_line_is_accepted", async () => {
|
||||
// The launch wrapper gives the gateway no start acknowledgment (accepted
|
||||
// fact from PAP-5065): a host byte written before the READY line is
|
||||
|
||||
@@ -2,6 +2,7 @@ import fs from "node:fs/promises";
|
||||
import net from "node:net";
|
||||
import os from "node:os";
|
||||
import path from "node:path";
|
||||
import type { Duplex } from "node:stream";
|
||||
import { randomBytes, randomUUID } from "node:crypto";
|
||||
import type { SshRemoteExecutionSpec } from "./ssh.js";
|
||||
import {
|
||||
@@ -43,7 +44,9 @@ import {
|
||||
import {
|
||||
createHttp2BridgeServer,
|
||||
type Http2BridgeForwardHandler,
|
||||
type Http2BridgeServerHandle,
|
||||
} from "./http2-bridge-server.js";
|
||||
import { getHttp2BridgeAdmissionGate, type Http2BridgeAdmissionGate } from "./http2-bridge-admission.js";
|
||||
import {
|
||||
createSandboxRunLogTailFactory,
|
||||
type SandboxRunLogTailFactory,
|
||||
@@ -3540,6 +3543,45 @@ export const __http2PrefaceScanTesting = {
|
||||
},
|
||||
};
|
||||
|
||||
/**
|
||||
* Reserve one live-session admission slot, then bind the channel to the
|
||||
* HTTP/2 server. Returns `null` at once when the cap is already full; the
|
||||
* caller must never bind the channel in that case. `bindChannel` is
|
||||
* synchronous, so a bind failure releases the reserved slot before it
|
||||
* re-throws the original error — the reservation never outlives a channel
|
||||
* that never bound. On success, the returned `boundDuplex` carries a
|
||||
* `close` listener that releases the same slot; `release` is idempotent,
|
||||
* so a caller may also call it again on a later teardown path with no
|
||||
* second slot freed.
|
||||
*/
|
||||
function bindHttp2BridgeSessionWithAdmission(
|
||||
gate: Http2BridgeAdmissionGate,
|
||||
http2Server: Http2BridgeServerHandle,
|
||||
channel: CommandManagedDuplexChannel,
|
||||
): { boundDuplex: Duplex; release: () => void } | null {
|
||||
const release = gate.tryAcquireSession();
|
||||
if (!release) return null;
|
||||
let boundDuplex: Duplex;
|
||||
try {
|
||||
boundDuplex = http2Server.bindChannel(channel);
|
||||
} catch (error) {
|
||||
release();
|
||||
throw error;
|
||||
}
|
||||
boundDuplex.on("close", release);
|
||||
return { boundDuplex, release };
|
||||
}
|
||||
|
||||
/**
|
||||
* Test-only surface for {@link bindHttp2BridgeSessionWithAdmission}. A test
|
||||
* drives the reserve-then-bind-then-release cycle, including a forced bind
|
||||
* failure, with a fake gate and a fake server handle, without the whole
|
||||
* bridge. Production code never reads this export.
|
||||
*/
|
||||
export const __http2BridgeSessionAdmissionTesting = {
|
||||
bindHttp2BridgeSessionWithAdmission,
|
||||
};
|
||||
|
||||
/**
|
||||
* The terminal run disposition for the `http2_v1` path, in the same shape as
|
||||
* {@link DuplexBrokerRunDisposition}. A `failed` disposition means a terminal
|
||||
@@ -4629,60 +4671,89 @@ export async function startAdapterExecutionTargetPaperclipBridge(input: {
|
||||
stop: () => prefaceScan.scanned.stop(),
|
||||
close: () => prefaceScan.scanned.close(),
|
||||
};
|
||||
const boundDuplex = http2Server.bindChannel(channelForHttp2Server);
|
||||
// Also catches the `Duplex` this wrapper destroys when the bounded
|
||||
// read backpressure queue overflows post-bind, so that loss still
|
||||
// reaches `recordHttp2Loss` the same way any other write fault does.
|
||||
boundDuplex.on("error", () => recordHttp2Loss("write_error"));
|
||||
|
||||
duplexChannelOpen.ready();
|
||||
await onLog(
|
||||
"stdout",
|
||||
"[paperclip] Sandbox HTTP/2 transport ready; serving the host-assigned origin.\n",
|
||||
// Reserve one live-session slot before the bind, not after: the
|
||||
// host must hold the full wrapper budget for the whole session
|
||||
// before the wrapper exists. `tryAcquireSession` never waits; a
|
||||
// caller that waited here would hold a partly-open channel for an
|
||||
// unbounded time.
|
||||
const admittedSession = bindHttp2BridgeSessionWithAdmission(
|
||||
getHttp2BridgeAdmissionGate(),
|
||||
http2Server,
|
||||
channelForHttp2Server,
|
||||
);
|
||||
// Stream run logs on the http2 path with the same gate and the same
|
||||
// log line as the file path. The http2 path starts no file-bridge
|
||||
// worker, so create the log directory before the tail starts.
|
||||
let duplexRunLogTail: SandboxRunLogTailFactory | null = null;
|
||||
if (target.transport === "sandbox" && target.streamRunLogs !== false) {
|
||||
const duplexLogsDir = sandboxCallbackBridgeDirectories(queueDir).logsDir;
|
||||
await ensureSandboxRunLogDirectory({
|
||||
runner,
|
||||
remoteCwd: target.remoteCwd,
|
||||
logsDir: duplexLogsDir,
|
||||
shellCommand,
|
||||
timeoutMs: bridgeTimeoutMs,
|
||||
});
|
||||
duplexRunLogTail = createSandboxRunLogTailFactory({
|
||||
runner,
|
||||
remoteCwd: target.remoteCwd,
|
||||
logsDir: duplexLogsDir,
|
||||
shellCommand,
|
||||
});
|
||||
await onLog("stdout", "[paperclip] Sandbox run log streaming enabled for this run.\n");
|
||||
if (!admittedSession) {
|
||||
// Fail closed, the same shape as a missing preface: close the
|
||||
// partial channel inside the cleanup budget, then select the
|
||||
// file bridge. The host never bound this channel to an HTTP/2
|
||||
// session, so no request reached it or any endpoint.
|
||||
gate.disposePendingReplay();
|
||||
await closeDuplexChannelWithinBudget(openedChannel, DEFAULT_DUPLEX_CLEANUP_BUDGET_MS);
|
||||
duplexChannelOpen.fallback("host_session_capacity");
|
||||
await onLog(
|
||||
"stderr",
|
||||
"[paperclip] The host HTTP/2 bridge live session cap is full (host_session_capacity). Using the file bridge.\n",
|
||||
);
|
||||
} else {
|
||||
const { boundDuplex, release: releaseSessionSlot } = admittedSession;
|
||||
// Also catches the `Duplex` this wrapper destroys when the bounded
|
||||
// read backpressure queue overflows post-bind, so that loss still
|
||||
// reaches `recordHttp2Loss` the same way any other write fault does.
|
||||
boundDuplex.on("error", () => recordHttp2Loss("write_error"));
|
||||
|
||||
duplexChannelOpen.ready();
|
||||
await onLog(
|
||||
"stdout",
|
||||
"[paperclip] Sandbox HTTP/2 transport ready; serving the host-assigned origin.\n",
|
||||
);
|
||||
// Stream run logs on the http2 path with the same gate and the same
|
||||
// log line as the file path. The http2 path starts no file-bridge
|
||||
// worker, so create the log directory before the tail starts.
|
||||
let duplexRunLogTail: SandboxRunLogTailFactory | null = null;
|
||||
if (target.transport === "sandbox" && target.streamRunLogs !== false) {
|
||||
const duplexLogsDir = sandboxCallbackBridgeDirectories(queueDir).logsDir;
|
||||
await ensureSandboxRunLogDirectory({
|
||||
runner,
|
||||
remoteCwd: target.remoteCwd,
|
||||
logsDir: duplexLogsDir,
|
||||
shellCommand,
|
||||
timeoutMs: bridgeTimeoutMs,
|
||||
});
|
||||
duplexRunLogTail = createSandboxRunLogTailFactory({
|
||||
runner,
|
||||
remoteCwd: target.remoteCwd,
|
||||
logsDir: duplexLogsDir,
|
||||
shellCommand,
|
||||
});
|
||||
await onLog("stdout", "[paperclip] Sandbox run log streaming enabled for this run.\n");
|
||||
}
|
||||
return {
|
||||
env: {
|
||||
PAPERCLIP_API_URL: sandboxOrigin,
|
||||
PAPERCLIP_API_KEY: bridgeToken,
|
||||
PAPERCLIP_API_BRIDGE_MODE: SANDBOX_CALLBACK_BRIDGE_HTTP2_MODE,
|
||||
},
|
||||
runLogTail: duplexRunLogTail,
|
||||
readRunDisposition: (): DuplexBrokerRunDisposition => dispositionLatch.disposition,
|
||||
// Atomically read the latch and mark the orderly completion for the
|
||||
// ACP success-eligible terminal, so no await separates the read from
|
||||
// the mark and a teardown loss cannot slip in between.
|
||||
settleRunDisposition: (): DuplexBrokerRunDisposition => dispositionLatch.settleRunDisposition(),
|
||||
markOrderlyCompletion: (): void => dispositionLatch.markOrderlyCompletion(),
|
||||
stop: async () => {
|
||||
// Close the HTTP/2 server's sessions, then the channel, before
|
||||
// lease release, so no live provider session remains when the
|
||||
// caller releases the lease.
|
||||
await http2Server.close();
|
||||
await closeDuplexChannelWithinBudget(openedChannel, DEFAULT_DUPLEX_CLEANUP_BUDGET_MS);
|
||||
// Idempotent: releases the session slot here too, in case the
|
||||
// bound `Duplex` never emits `close` on this exact teardown
|
||||
// path. A second call from the `close` listener above frees
|
||||
// no second slot.
|
||||
releaseSessionSlot();
|
||||
await bridgeAsset.cleanup();
|
||||
},
|
||||
};
|
||||
}
|
||||
return {
|
||||
env: {
|
||||
PAPERCLIP_API_URL: sandboxOrigin,
|
||||
PAPERCLIP_API_KEY: bridgeToken,
|
||||
PAPERCLIP_API_BRIDGE_MODE: SANDBOX_CALLBACK_BRIDGE_HTTP2_MODE,
|
||||
},
|
||||
runLogTail: duplexRunLogTail,
|
||||
readRunDisposition: (): DuplexBrokerRunDisposition => dispositionLatch.disposition,
|
||||
// Atomically read the latch and mark the orderly completion for the
|
||||
// ACP success-eligible terminal, so no await separates the read from
|
||||
// the mark and a teardown loss cannot slip in between.
|
||||
settleRunDisposition: (): DuplexBrokerRunDisposition => dispositionLatch.settleRunDisposition(),
|
||||
markOrderlyCompletion: (): void => dispositionLatch.markOrderlyCompletion(),
|
||||
stop: async () => {
|
||||
// Close the HTTP/2 server's sessions, then the channel, before
|
||||
// lease release, so no live provider session remains when the
|
||||
// caller releases the lease.
|
||||
await http2Server.close();
|
||||
await closeDuplexChannelWithinBudget(openedChannel, DEFAULT_DUPLEX_CLEANUP_BUDGET_MS);
|
||||
await bridgeAsset.cleanup();
|
||||
},
|
||||
};
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -33,7 +33,12 @@ import {
|
||||
import { createSandboxHttp2BridgeGateway } from "./sandbox-callback-bridge.js";
|
||||
import type { CommandManagedDuplexChannel } from "./command-managed-runtime.js";
|
||||
import {
|
||||
DEFAULT_MAX_LIVE_HTTP2_BRIDGE_SESSIONS,
|
||||
DEFAULT_MAX_PARALLEL_HTTP2_BRIDGE_REQUESTS,
|
||||
http2BridgeAdmissionBudgetBytes,
|
||||
Http2BridgeAdmissionGate,
|
||||
HTTP2_BRIDGE_ADMISSION_MEMORY_TARGET_BYTES,
|
||||
HTTP2_BRIDGE_ADMISSION_RESERVED_BUDGET_BYTES,
|
||||
HTTP2_BRIDGE_LIVE_SESSION_BUDGET_BYTES,
|
||||
} from "./http2-bridge-admission.js";
|
||||
|
||||
@@ -1730,5 +1735,55 @@ describe("createHttp2BridgeServer + createSandboxHttp2BridgeGateway", () => {
|
||||
rawClient.close();
|
||||
}
|
||||
});
|
||||
|
||||
it("the default caps keep the named byte total inside the reserved budget", () => {
|
||||
// 64 admitted streams x 868,351 bytes, plus 8 live sessions x
|
||||
// 16,777,216 bytes: 189,792,192 named bytes in total. This stays
|
||||
// inside the 201,326,592-byte reserved budget and below the
|
||||
// 268,435,456-byte target.
|
||||
const total = http2BridgeAdmissionBudgetBytes(
|
||||
DEFAULT_MAX_PARALLEL_HTTP2_BRIDGE_REQUESTS,
|
||||
DEFAULT_MAX_LIVE_HTTP2_BRIDGE_SESSIONS,
|
||||
);
|
||||
expect(total).toBe(189_792_192);
|
||||
expect(total).toBeLessThanOrEqual(HTTP2_BRIDGE_ADMISSION_RESERVED_BUDGET_BYTES);
|
||||
expect(total).toBeLessThan(HTTP2_BRIDGE_ADMISSION_MEMORY_TARGET_BYTES);
|
||||
});
|
||||
|
||||
it("one more live session than the default cap would pass the reserved budget", () => {
|
||||
// 64 admitted streams x 868,351 bytes, plus 9 live sessions x
|
||||
// 16,777,216 bytes: 206,569,408 named bytes. This passes the
|
||||
// 201,326,592-byte reserved budget.
|
||||
const total = http2BridgeAdmissionBudgetBytes(
|
||||
DEFAULT_MAX_PARALLEL_HTTP2_BRIDGE_REQUESTS,
|
||||
DEFAULT_MAX_LIVE_HTTP2_BRIDGE_SESSIONS + 1,
|
||||
);
|
||||
expect(total).toBe(206_569_408);
|
||||
expect(total).toBeGreaterThan(HTTP2_BRIDGE_ADMISSION_RESERVED_BUDGET_BYTES);
|
||||
});
|
||||
|
||||
it("bindChannel's Duplex emits close once the handle closes the session it belongs to", async () => {
|
||||
const admissionGate = new Http2BridgeAdmissionGate({ maxParallel: 4, maxSessions: 4 });
|
||||
const [serverSide, clientSide] = duplexPair();
|
||||
const channel = fakeChannelFromDuplex(serverSide);
|
||||
const handle = createHttp2BridgeServer({
|
||||
bridgeToken: BRIDGE_TOKEN,
|
||||
forwardRequest: async () => ({ status: 200 }),
|
||||
admissionGate,
|
||||
});
|
||||
const boundDuplex = handle.bindChannel(channel);
|
||||
const rawClient = connectRawClient(clientSide);
|
||||
try {
|
||||
// Wait for the session to actually establish, the same way a real
|
||||
// sandbox client would, before the close-teardown path runs.
|
||||
await new Promise<void>((resolve) => rawClient.once("connect", resolve));
|
||||
const closed = new Promise<void>((resolve) => boundDuplex.once("close", resolve));
|
||||
await handle.close();
|
||||
await closed;
|
||||
expect(boundDuplex.destroyed).toBe(true);
|
||||
} finally {
|
||||
rawClient.close();
|
||||
}
|
||||
});
|
||||
});
|
||||
});
|
||||
Reference in new issue
Block a user