diff --git a/docs/content/docs/resource-limits.mdx b/docs/content/docs/resource-limits.mdx index c087b15d58..b8b2b6f6ad 100644 --- a/docs/content/docs/resource-limits.mdx +++ b/docs/content/docs/resource-limits.mdx @@ -71,6 +71,14 @@ Library VMs share a sidecar process by default. Limits that apply to the whole process, such as `runtime.executor.maxActiveVms`, are set when creating a dedicated sidecar with `AgentOs.createSidecar()`, as in the example above. +Each library request to the sidecar must fit in one protocol frame. The sidecar +reports its frame limit (16 MiB) when the client connects, and the limit is not +configurable. The client rejects a larger request with `SidecarFrameTooLarge` +before it sends it, so the sidecar and the other VMs on it keep running. Binary +file content is base64-encoded, so one `writeFile` or `writeFiles` call carries +at most about 12 MiB of binary data. To move larger files, split them into +several files or mount a host directory. + ### Sidecar memory On Linux with glibc, the sidecar returns memory freed by a disposed VM's V8 diff --git a/packages/core/src/index.ts b/packages/core/src/index.ts index 3af1b4104c..3a270b35a6 100644 --- a/packages/core/src/index.ts +++ b/packages/core/src/index.ts @@ -2,6 +2,7 @@ export { SidecarEventBufferOverflow } from "@rivet-dev/agentos-runtime-core/event-buffer"; export { + SidecarFrameTooLarge, SidecarProcessError, SidecarProcessExited, SidecarRejectedError, diff --git a/packages/core/tests/sidecar-exit.test.ts b/packages/core/tests/sidecar-exit.test.ts index 1e21745d46..2ecc3c23a4 100644 --- a/packages/core/tests/sidecar-exit.test.ts +++ b/packages/core/tests/sidecar-exit.test.ts @@ -3,7 +3,7 @@ import { join, resolve } from "node:path"; import { fileURLToPath } from "node:url"; import { SidecarProcessExited } from "@rivet-dev/agentos-runtime-core/sidecar-client"; import { expect, test } from "vitest"; -import { AgentOs } from "../src/index.js"; +import { AgentOs, SidecarFrameTooLarge } from "../src/index.js"; import { NativeSidecarProcessClient, serializeRootFilesystemForSidecar, @@ -143,3 +143,43 @@ test("a sidecar outlives more VM create and dispose cycles than its event buffer } await client.dispose(); }, 60_000); + +test("an oversized request rejects without stopping the sidecar its VMs share", async () => { + const uncaught: unknown[] = []; + const recordUncaught = (error: unknown) => uncaught.push(error); + process.on("uncaughtException", recordUncaught); + const sidecar = await AgentOs.createSidecar(); + const placement = { kind: "explicit", handle: sidecar } as const; + const writer = await AgentOs.create({ + defaultSoftware: false, + sidecar: placement, + }); + const bystander = await AgentOs.create({ + defaultSoftware: false, + sidecar: placement, + }); + try { + // Base64 content encoding puts this request over the sidecar frame limit. + const error = await writer.filesystem + .writeFile("/tmp/big.bin", new Uint8Array(13 * 1024 * 1024)) + .catch((caught: unknown) => caught); + expect(error).toMatchObject({ name: "SidecarFrameTooLarge" }); + expect(error).toBeInstanceOf(SidecarFrameTooLarge); + const { frameBytes, maxFrameBytes } = error as SidecarFrameTooLarge; + expect(frameBytes).toBeGreaterThan(maxFrameBytes); + expect((error as Error).message).toContain(`limit is ${maxFrameBytes}`); + + // Another VM on the same sidecar keeps working, and so does the writer. + expect(await bystander.javascript.evaluate("1 + 1")).toMatchObject({ + outcome: "succeeded", + value: 2, + }); + await writer.filesystem.writeFile("/tmp/small.txt", "ok"); + expect(uncaught).toEqual([]); + } finally { + process.off("uncaughtException", recordUncaught); + await writer.dispose().catch(() => {}); + await bystander.dispose().catch(() => {}); + await sidecar.dispose().catch(() => {}); + } +}, 60_000); diff --git a/packages/runtime-core/src/frame-stream.ts b/packages/runtime-core/src/frame-stream.ts index 5e2b95657e..5e9c68aec0 100644 --- a/packages/runtime-core/src/frame-stream.ts +++ b/packages/runtime-core/src/frame-stream.ts @@ -4,12 +4,18 @@ import { encodeLengthPrefixedPayload, tryDecodeLengthPrefixedPayload, } from "./framing.js"; +import { SidecarFrameTooLarge } from "./sidecar-errors.js"; export interface StdioFrameTransportOptions { stdin: Writable; stdout: Readable; encodeFrame: (frame: TWriteFrame) => Uint8Array; decodeFrame: (payload: Uint8Array) => TReadFrame; + /** + * Largest payload the peer accepts, or `undefined` while unknown. A larger + * frame is rejected before it is written. + */ + maxFrameBytes?: () => number | undefined; } export interface FrameTransport { @@ -208,17 +214,21 @@ export class StdioFrameTransport private readonly stdout: Readable; private readonly encodeFrame: (frame: TWriteFrame) => Uint8Array; private readonly decodeFrame: (payload: Uint8Array) => TReadFrame; + private readonly maxFrameBytes: (() => number | undefined) | undefined; private readonly sharedStdio: boolean; private readonly frameListeners = new Set<(frame: TReadFrame) => void>(); private readonly errorListeners = new Set<(error: Error) => void>(); private readonly endListeners = new Set<() => void>(); private stdoutBuffer: ByteArray = new Uint8Array(0); + private pendingWrites = 0; + private disposed = false; constructor(options: StdioFrameTransportOptions) { this.stdin = options.stdin; this.stdout = options.stdout; this.encodeFrame = options.encodeFrame; this.decodeFrame = options.decodeFrame; + this.maxFrameBytes = options.maxFrameBytes; this.sharedStdio = this.stdin === (this.stdout as unknown as Writable); if (!this.sharedStdio) { this.stdin.on("error", this.handleError); @@ -250,10 +260,24 @@ export class StdioFrameTransport } async writeFrame(frame: TWriteFrame): Promise { + if (this.disposed) { + throw new Error("stdio frame transport disposed"); + } const payload = this.encodeFrame(frame); + const maxFrameBytes = this.maxFrameBytes?.(); + // The peer cannot resynchronize after an oversized length prefix, so + // writing one would end the stream for every user of it. + if (maxFrameBytes !== undefined && payload.length > maxFrameBytes) { + throw new SidecarFrameTooLarge({ + frameBytes: payload.length, + maxFrameBytes, + }); + } const encoded = encodeLengthPrefixedPayload(payload); + this.pendingWrites += 1; await new Promise((resolve, reject) => { this.stdin.write(encoded, (error) => { + this.pendingWrites -= 1; if (error) { reject(error); return; @@ -264,6 +288,10 @@ export class StdioFrameTransport } dispose(): void { + if (this.disposed) { + return; + } + this.disposed = true; if (!this.sharedStdio) { this.stdin.off("error", this.handleError); } @@ -273,6 +301,14 @@ export class StdioFrameTransport this.frameListeners.clear(); this.errorListeners.clear(); this.endListeners.clear(); + if (this.pendingWrites > 0) { + // A write still in flight rejects its own promise, but Node also emits + // the failure (for example EPIPE from a dead sidecar) as a stream + // `error` event. Without a listener that event crashes the host. + const stdin = this.stdin; + stdin.on("error", logLateStreamError); + stdin.once("close", () => stdin.off("error", logLateStreamError)); + } } private readonly handleData = (chunk: ByteArray | string): void => { @@ -319,6 +355,13 @@ export class StdioFrameTransport } } +function logLateStreamError(error: Error): void { + console.error( + "[agentOS] sidecar stream failed after its frame transport was disposed:", + error, + ); +} + function concatBytes(left: ByteArray, right: ByteArray): ByteArray { if (left.length === 0) { return right; diff --git a/packages/runtime-core/src/protocol-client.ts b/packages/runtime-core/src/protocol-client.ts index 1c55778e9a..7a5e3c037e 100644 --- a/packages/runtime-core/src/protocol-client.ts +++ b/packages/runtime-core/src/protocol-client.ts @@ -91,6 +91,11 @@ export class SidecarProtocolClient { LiveSidecarRequestFrame >; private closedError: Error | null = null; + /** + * Frame limit from the sidecar's authenticate response. Only the small + * authenticate request is written before it is known. + */ + private maxFrameBytes: number | undefined; private readonly closeListeners = new Set<(error: Error) => void>(); private readonly eventWaiters = new Set<{ matches: (event: LiveEventFrame) => boolean; @@ -113,7 +118,11 @@ export class SidecarProtocolClient { LiveEventFrame, LiveSidecarRequestFrame >({ - frameTransport: resolveProtocolFrameTransport(options, this.payloadCodec), + frameTransport: resolveProtocolFrameTransport( + options, + this.payloadCodec, + () => this.maxFrameBytes, + ), encodeFrame: (frame) => encodeProtocolFramePayload(frame, this.payloadCodec), decodeFrame: (payload) => @@ -223,6 +232,9 @@ export class SidecarProtocolClient { if (response.payload.type === "rejected") { throw new SidecarRejectedError(request.request_id, response.payload); } + if (response.payload.type === "authenticated") { + this.maxFrameBytes = response.payload.max_frame_bytes; + } return response; } @@ -424,6 +436,7 @@ export class SidecarProtocolClient { function resolveProtocolFrameTransport( options: SidecarProtocolClientOptions, payloadCodec: ProtocolFramePayloadCodec, + maxFrameBytes: () => number | undefined, ): FrameTransport< LiveResponseFrame | LiveEventFrame | LiveSidecarRequestFrame, LiveRequestFrame | LiveSidecarResponseFrame | LiveControlFrame @@ -451,6 +464,7 @@ function resolveProtocolFrameTransport( stdin: options.stdin, stdout: options.stdout, encodeFrame: (frame) => encodeProtocolFramePayload(frame, payloadCodec), + maxFrameBytes, decodeFrame: (payload) => decodeProtocolFramePayload(payload, payloadCodec) as | LiveResponseFrame @@ -485,6 +499,7 @@ function resolveProtocolFrameTransport( stdin: options.stdin, stdout: options.stdout, encodeFrame: (frame) => encodeProtocolFramePayload(frame, payloadCodec), + maxFrameBytes, decodeFrame: (payload) => { const frame = decodeProtocolFramePayload(payload, payloadCodec); validateOrdinaryInboundFrame(frame); @@ -498,6 +513,7 @@ function resolveProtocolFrameTransport( stdin: options.control, stdout: options.control, encodeFrame: (frame) => encodeProtocolFramePayload(frame, payloadCodec), + maxFrameBytes, decodeFrame: (payload) => { const frame = decodeProtocolFramePayload(payload, payloadCodec); validateControlInboundFrame(frame); diff --git a/packages/runtime-core/src/sidecar-errors.ts b/packages/runtime-core/src/sidecar-errors.ts index c3c7e5d140..3e1a3bed17 100644 --- a/packages/runtime-core/src/sidecar-errors.ts +++ b/packages/runtime-core/src/sidecar-errors.ts @@ -34,6 +34,24 @@ export class SidecarRejectedError extends Error { } } +/** + * A host frame exceeded the sidecar's negotiated `max_frame_bytes`. The client + * rejects it before writing, so the sidecar and its other VMs keep running. + */ +export class SidecarFrameTooLarge extends Error { + readonly frameBytes: number; + readonly maxFrameBytes: number; + + constructor(options: { frameBytes: number; maxFrameBytes: number }) { + super( + `sidecar protocol frame is ${options.frameBytes} bytes, limit is ${options.maxFrameBytes} (sidecar max_frame_bytes); the request was not sent. Send less data per request, for example by writing large content in several smaller files or by mounting a host directory`, + ); + this.name = "SidecarFrameTooLarge"; + this.frameBytes = options.frameBytes; + this.maxFrameBytes = options.maxFrameBytes; + } +} + export class SidecarProcessExited extends Error { readonly exitCode: number | null; readonly signal: string | null; diff --git a/packages/runtime-core/src/sidecar-process.ts b/packages/runtime-core/src/sidecar-process.ts index 25b807ab01..4648204247 100644 --- a/packages/runtime-core/src/sidecar-process.ts +++ b/packages/runtime-core/src/sidecar-process.ts @@ -53,6 +53,7 @@ import type { export { SidecarEventBufferOverflow } from "./event-buffer.js"; export { + SidecarFrameTooLarge, SidecarProcessError, SidecarProcessExited, SidecarSilenceTimeout, diff --git a/packages/runtime-core/tests/frame-stream.test.ts b/packages/runtime-core/tests/frame-stream.test.ts index 76c5a6beef..5f3aaec10d 100644 --- a/packages/runtime-core/tests/frame-stream.test.ts +++ b/packages/runtime-core/tests/frame-stream.test.ts @@ -1,5 +1,5 @@ -import { PassThrough } from "node:stream"; -import { describe, expect, test } from "vitest"; +import { PassThrough, Writable } from "node:stream"; +import { describe, expect, test, vi } from "vitest"; import { StdioFrameTransport } from "../src/frame-stream.js"; import { encodeLengthPrefixedPayload } from "../src/framing.js"; @@ -111,6 +111,41 @@ describe("stdio frame transport", () => { expect(stdio.listenerCount("error")).toBe(initialErrorListeners); }); + test("a write that fails after dispose is logged instead of crashing the host", async () => { + let finishWrite: ((error?: Error | null) => void) | undefined; + const stdin = new Writable({ + write(_chunk, _encoding, callback) { + finishWrite = callback; + }, + }); + const transport = new StdioFrameTransport({ + stdin, + stdout: new PassThrough(), + encodeFrame: (frame) => Buffer.from(frame, "utf8"), + decodeFrame: (payload) => Buffer.from(payload).toString("utf8"), + }); + const uncaught: unknown[] = []; + const recordUncaught = (error: unknown) => uncaught.push(error); + process.on("uncaughtException", recordUncaught); + const logged = vi.spyOn(console, "error").mockImplementation(() => {}); + try { + const write = transport.writeFrame("in flight"); + transport.dispose(); + const brokenPipe = Object.assign(new Error("write EPIPE"), { + code: "EPIPE", + }); + finishWrite?.(brokenPipe); + + await expect(write).rejects.toBe(brokenPipe); + await new Promise((resolve) => setImmediate(resolve)); + expect(uncaught).toEqual([]); + expect(logged).toHaveBeenCalledWith(expect.any(String), brokenPipe); + } finally { + logged.mockRestore(); + process.off("uncaughtException", recordUncaught); + } + }); + test("reports decode errors", async () => { const stdin = new PassThrough(); const stdout = new PassThrough();