refactor(adapter-utils): delete the dead duplex body-chunk protocol code (#12186)
## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work > - The adapter utilities package provides transport code for sandbox agents > - The retired `duplex_v1` broker no longer produces or consumes body-chunk frames > - Dead protocol code remains in the host codec, gateway copy, bridge options, and tests > - This pull request removes that dead code and keeps the READY handshake unchanged > - The benefit is a smaller transport surface with fewer unused paths to maintain ## Linked Issues or Issue Description **What existing behavior does this improve?** This change improves the adapter utilities code that supports sandbox duplex readiness and frame handling. **Subsystem affected** `packages/adapter-utils/` — sandbox transport codecs, execution targets, and callback bridge tests. **Current behavior** The repository keeps body-chunk frame types, validators, a body spool, decoder limits, and tests after the `duplex_v1` broker removal. No live producer or consumer uses this code. **Proposed behavior** Remove the unused body-chunk protocol code and retain the READY handshake, its strict checks, and its size limits. **Reason and benefit** The removal reduces dead code and keeps the host and embedded gateway paths easier to inspect. It adds no new behavior. **Breaking changes** The removed frame types now decode as `unknown_type`. The live readiness gate already ignores those frames. The READY handshake stays byte-for-byte compatible. **Additional context** This cleanup follows [PR #12171](https://github.com/paperclipai/paperclip/pull/12171), which removed the duplex broker. ## What Changed - Remove `duplex-body-spool.ts` and its test. - Remove unused body-chunk frame types, validators, decoder code, vectors, and limits. - Remove the unused `reassembledBody` option and decoder limit environment entry. - Remove the embedded gateway decoder copy and the unused frame type map. - Keep the READY handshake and its existing boundary tests unchanged in behavior. ## Verification - `pnpm -F @paperclip/adapter-utils typecheck` passes. - The duplex frame codec test passes with 30 tests. - The sandbox execution-target test passes with 136 tests. - The sandbox callback bridge test passes with 46 tests. - CI must confirm all required checks after it starts. ## Risks Low risk. The change removes code only. The READY handshake, HTTP/2 body path, and byte-ledger path remain unchanged. ## Model Used OpenAI Codex, GPT-5, tool use and code execution. ## 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
822e0aed93
commit
802f2af154
|
|
@ -1,306 +0,0 @@
|
|||
import { promises as fs } from "node:fs";
|
||||
import os from "node:os";
|
||||
import path from "node:path";
|
||||
import { afterEach, beforeEach, describe, expect, it } from "vitest";
|
||||
|
||||
import { DUPLEX_FRAME_VERSION, type DuplexBodyChunkFrame } from "./duplex-frame-codec.js";
|
||||
import {
|
||||
DuplexBodyReceiver,
|
||||
type DuplexBodyReceiverConfig,
|
||||
splitBodyIntoChunkFrames,
|
||||
sweepStaleSpillDirectories,
|
||||
} from "./duplex-body-spool.js";
|
||||
|
||||
// Read the whole reassembled body back, whatever path it took. A memory body
|
||||
// returns its buffer; a spilled body streams from disk. The test uses this only
|
||||
// to assert the round-trip content, never on a hot path.
|
||||
async function readBodyBytes(body: {
|
||||
spilled: boolean;
|
||||
createReadStream: () => NodeJS.ReadableStream;
|
||||
}): Promise<Buffer> {
|
||||
const chunks: Buffer[] = [];
|
||||
for await (const chunk of body.createReadStream()) {
|
||||
chunks.push(Buffer.isBuffer(chunk) ? chunk : Buffer.from(chunk));
|
||||
}
|
||||
return Buffer.concat(chunks);
|
||||
}
|
||||
|
||||
let spillRoot: string;
|
||||
|
||||
beforeEach(async () => {
|
||||
spillRoot = await fs.mkdtemp(path.join(os.tmpdir(), "duplex-spool-test-"));
|
||||
});
|
||||
|
||||
afterEach(async () => {
|
||||
await fs.rm(spillRoot, { recursive: true, force: true }).catch(() => undefined);
|
||||
});
|
||||
|
||||
function makeReceiver(overrides: DuplexBodyReceiverConfig = {}): Promise<DuplexBodyReceiver> {
|
||||
return DuplexBodyReceiver.create({ spillRootDir: spillRoot, ...overrides });
|
||||
}
|
||||
|
||||
const RAW = 8;
|
||||
|
||||
// Feed a whole body through a receiver as an envelope plus chunk frames. Return
|
||||
// the settled body handle. The caller pushes the chunks the split produces.
|
||||
async function transfer(
|
||||
receiver: DuplexBodyReceiver,
|
||||
id: string,
|
||||
body: Buffer,
|
||||
): ReturnType<DuplexBodyReceiver["begin"]> {
|
||||
const promise = receiver.begin(id, body.length);
|
||||
const frames = splitBodyIntoChunkFrames(id, body, DUPLEX_FRAME_VERSION, RAW);
|
||||
for (const frame of frames) {
|
||||
const result = receiver.pushChunk(frame);
|
||||
expect(result.ok).toBe(true);
|
||||
}
|
||||
return promise;
|
||||
}
|
||||
|
||||
describe("duplex body reassembler — round trip", () => {
|
||||
it("reassembles a small body on the memory path", async () => {
|
||||
const receiver = await makeReceiver({ spillThresholdBytes: 1024, rawChunkBytes: RAW });
|
||||
const source = Buffer.from("hello duplex body");
|
||||
const body = await transfer(receiver, "req-mem", source);
|
||||
expect(body.spilled).toBe(false);
|
||||
expect(body.byteCount).toBe(source.length);
|
||||
expect(body.readMemory().equals(source)).toBe(true);
|
||||
expect((await readBodyBytes(body)).equals(source)).toBe(true);
|
||||
await body.dispose();
|
||||
});
|
||||
|
||||
it("reassembles a large body on the spill path without holding it in memory", async () => {
|
||||
const receiver = await makeReceiver({ spillThresholdBytes: 16, rawChunkBytes: RAW });
|
||||
const source = Buffer.from("abcdefghijklmnopqrstuvwxyz0123456789ABCD"); // 40 bytes
|
||||
const body = await transfer(receiver, "req-spill", source);
|
||||
expect(body.spilled).toBe(true);
|
||||
expect(body.byteCount).toBe(source.length);
|
||||
// A spilled body never loads the whole body into memory on the receive side.
|
||||
expect(() => body.readMemory()).toThrow();
|
||||
expect(receiver.spillUsedBytes).toBe(source.length);
|
||||
expect(receiver.spillOpenFiles).toBe(1);
|
||||
expect((await readBodyBytes(body)).equals(source)).toBe(true);
|
||||
await body.dispose();
|
||||
// The dispose path unlinks the file and releases the reservation.
|
||||
expect(receiver.spillUsedBytes).toBe(0);
|
||||
expect(receiver.spillOpenFiles).toBe(0);
|
||||
const entries = await fs.readdir(receiver.spillDir);
|
||||
expect(entries).toHaveLength(0);
|
||||
await receiver.destroy();
|
||||
});
|
||||
|
||||
it("completes a zero-length body at once and accepts no chunk", async () => {
|
||||
const receiver = await makeReceiver({ spillThresholdBytes: 16, rawChunkBytes: RAW });
|
||||
const body = await receiver.begin("req-zero", 0);
|
||||
expect(body.byteCount).toBe(0);
|
||||
expect(body.readMemory().length).toBe(0);
|
||||
const stray = receiver.pushChunk({
|
||||
version: DUPLEX_FRAME_VERSION,
|
||||
type: "body_chunk",
|
||||
id: "req-zero",
|
||||
seq: 0,
|
||||
data: Buffer.from("x").toString("base64"),
|
||||
});
|
||||
expect(stray.ok).toBe(false);
|
||||
if (!stray.ok) expect(stray.error.code).toBe("zero_body_has_chunk");
|
||||
await body.dispose();
|
||||
});
|
||||
});
|
||||
|
||||
describe("duplex body reassembler — rejection matrix", () => {
|
||||
function chunk(id: string, seq: number, data: string): DuplexBodyChunkFrame {
|
||||
return { version: DUPLEX_FRAME_VERSION, type: "body_chunk", id, seq, data };
|
||||
}
|
||||
|
||||
it("rejects a duplicate seq", async () => {
|
||||
const receiver = await makeReceiver({ spillThresholdBytes: 1024, rawChunkBytes: RAW });
|
||||
void receiver.begin("d", 16);
|
||||
expect(receiver.pushChunk(chunk("d", 0, Buffer.alloc(RAW, 1).toString("base64"))).ok).toBe(true);
|
||||
const dup = receiver.pushChunk(chunk("d", 0, Buffer.alloc(RAW, 2).toString("base64")));
|
||||
expect(dup.ok).toBe(false);
|
||||
if (!dup.ok) expect(dup.error.code).toBe("seq_out_of_order");
|
||||
});
|
||||
|
||||
it("rejects a gap in the seq order", async () => {
|
||||
const receiver = await makeReceiver({ spillThresholdBytes: 1024, rawChunkBytes: RAW });
|
||||
void receiver.begin("g", 24);
|
||||
expect(receiver.pushChunk(chunk("g", 0, Buffer.alloc(RAW, 1).toString("base64"))).ok).toBe(true);
|
||||
const gap = receiver.pushChunk(chunk("g", 2, Buffer.alloc(RAW, 2).toString("base64")));
|
||||
expect(gap.ok).toBe(false);
|
||||
if (!gap.ok) expect(gap.error.code).toBe("seq_out_of_order");
|
||||
});
|
||||
|
||||
it("rejects a reordered seq", async () => {
|
||||
const receiver = await makeReceiver({ spillThresholdBytes: 1024, rawChunkBytes: RAW });
|
||||
void receiver.begin("r", 24);
|
||||
const reorder = receiver.pushChunk(chunk("r", 1, Buffer.alloc(RAW, 1).toString("base64")));
|
||||
expect(reorder.ok).toBe(false);
|
||||
if (!reorder.ok) expect(reorder.error.code).toBe("seq_out_of_order");
|
||||
});
|
||||
|
||||
it("rejects non-canonical base64", async () => {
|
||||
const receiver = await makeReceiver({ spillThresholdBytes: 1024, rawChunkBytes: RAW });
|
||||
const begun = receiver.begin("b", 8);
|
||||
begun.catch(() => undefined);
|
||||
// "AB==" carries non-zero bits in the final sextet, so it decodes to one
|
||||
// byte and re-encodes to "AA==". The round-trip mismatch marks it not
|
||||
// canonical.
|
||||
const bad = receiver.pushChunk(chunk("b", 0, "AB=="));
|
||||
expect(bad.ok).toBe(false);
|
||||
if (!bad.ok) expect(bad.error.code).toBe("chunk_not_canonical_base64");
|
||||
});
|
||||
|
||||
it("rejects a short non-final chunk (truncation)", async () => {
|
||||
const receiver = await makeReceiver({ spillThresholdBytes: 1024, rawChunkBytes: RAW });
|
||||
void receiver.begin("t", 24);
|
||||
// A non-final chunk must be exactly RAW bytes. A 4-byte chunk that does not
|
||||
// complete the 24-byte body is a truncation.
|
||||
const truncation = receiver.pushChunk(chunk("t", 0, Buffer.alloc(4, 1).toString("base64")));
|
||||
expect(truncation.ok).toBe(false);
|
||||
if (!truncation.ok) expect(truncation.error.code).toBe("chunk_wrong_size");
|
||||
});
|
||||
|
||||
it("rejects an empty non-final chunk", async () => {
|
||||
const receiver = await makeReceiver({ spillThresholdBytes: 1024, rawChunkBytes: RAW });
|
||||
void receiver.begin("e", 16);
|
||||
const empty = receiver.pushChunk(chunk("e", 0, ""));
|
||||
expect(empty.ok).toBe(false);
|
||||
if (!empty.ok) expect(empty.error.code).toBe("empty_chunk");
|
||||
});
|
||||
|
||||
it("rejects an overrun past the declared body size", async () => {
|
||||
const receiver = await makeReceiver({ spillThresholdBytes: 1024, rawChunkBytes: RAW });
|
||||
// A 12-byte body takes one full RAW chunk, then a final chunk of 4 bytes. A
|
||||
// second full RAW chunk overruns the declared size.
|
||||
void receiver.begin("o", 12);
|
||||
expect(receiver.pushChunk(chunk("o", 0, Buffer.alloc(RAW, 1).toString("base64"))).ok).toBe(true);
|
||||
const overrun = receiver.pushChunk(chunk("o", 1, Buffer.alloc(RAW, 2).toString("base64")));
|
||||
expect(overrun.ok).toBe(false);
|
||||
if (!overrun.ok) expect(overrun.error.code).toBe("body_overrun");
|
||||
});
|
||||
|
||||
it("rejects an over-size final chunk", async () => {
|
||||
const receiver = await makeReceiver({ spillThresholdBytes: 1024, rawChunkBytes: RAW });
|
||||
void receiver.begin("os", 40);
|
||||
const over = receiver.pushChunk(chunk("os", 0, Buffer.alloc(RAW + 4, 1).toString("base64")));
|
||||
expect(over.ok).toBe(false);
|
||||
if (!over.ok) expect(over.error.code).toBe("chunk_wrong_size");
|
||||
});
|
||||
|
||||
it("rejects a chunk that arrives with no matching envelope", async () => {
|
||||
const receiver = await makeReceiver({ spillThresholdBytes: 1024, rawChunkBytes: RAW });
|
||||
const orphan = receiver.pushChunk(chunk("missing", 0, Buffer.alloc(RAW, 1).toString("base64")));
|
||||
expect(orphan.ok).toBe(false);
|
||||
if (!orphan.ok) expect(orphan.error.code).toBe("chunk_without_envelope");
|
||||
});
|
||||
|
||||
it("rejects a chunk that arrives after completion", async () => {
|
||||
const receiver = await makeReceiver({ spillThresholdBytes: 1024, rawChunkBytes: RAW });
|
||||
const promise = receiver.begin("c", RAW);
|
||||
expect(receiver.pushChunk(chunk("c", 0, Buffer.alloc(RAW, 1).toString("base64"))).ok).toBe(true);
|
||||
await promise;
|
||||
const late = receiver.pushChunk(chunk("c", 1, Buffer.alloc(RAW, 2).toString("base64")));
|
||||
expect(late.ok).toBe(false);
|
||||
if (!late.ok) expect(late.error.code).toBe("chunk_after_completion");
|
||||
});
|
||||
});
|
||||
|
||||
describe("duplex body reassembler — spill caps fail closed", () => {
|
||||
it("fails closed when a write exceeds the per-channel byte cap and unlinks the partial file", async () => {
|
||||
// A 5-byte cap admits the first RAW=8 chunk write? No: reserve 8 > cap 5, so
|
||||
// the first write fails closed. Use a cap that admits one chunk but not two.
|
||||
const receiver = await makeReceiver({
|
||||
spillThresholdBytes: 4,
|
||||
rawChunkBytes: RAW,
|
||||
channelSpillByteCap: RAW + 2,
|
||||
});
|
||||
const promise = receiver.begin("cap", 24);
|
||||
expect(receiver.pushChunk({ version: DUPLEX_FRAME_VERSION, type: "body_chunk", id: "cap", seq: 0, data: Buffer.alloc(RAW, 1).toString("base64") }).ok).toBe(true);
|
||||
// The second chunk write reserves past the byte cap.
|
||||
receiver.pushChunk({ version: DUPLEX_FRAME_VERSION, type: "body_chunk", id: "cap", seq: 1, data: Buffer.alloc(RAW, 2).toString("base64") });
|
||||
await expect(promise).rejects.toMatchObject({ code: "spill_cap_exceeded" });
|
||||
// The partial file is gone and the reservation released.
|
||||
expect(receiver.spillUsedBytes).toBe(0);
|
||||
expect(receiver.spillOpenFiles).toBe(0);
|
||||
const entries = await fs.readdir(receiver.spillDir);
|
||||
expect(entries).toHaveLength(0);
|
||||
await receiver.destroy();
|
||||
});
|
||||
|
||||
it("fails closed when the open-file cap is reached", async () => {
|
||||
const receiver = await makeReceiver({
|
||||
spillThresholdBytes: 4,
|
||||
rawChunkBytes: RAW,
|
||||
channelSpillFileCap: 1,
|
||||
});
|
||||
// The first spill body claims the one file slot and stays open. The channel
|
||||
// dispose rejects this in-flight body, so attach the catch at once.
|
||||
const first = receiver.begin("f1", 24);
|
||||
first.catch(() => undefined);
|
||||
expect(receiver.pushChunk({ version: DUPLEX_FRAME_VERSION, type: "body_chunk", id: "f1", seq: 0, data: Buffer.alloc(RAW, 1).toString("base64") }).ok).toBe(true);
|
||||
// The second spill body cannot claim a file slot.
|
||||
await expect(receiver.begin("f2", 24)).rejects.toMatchObject({ code: "spill_cap_exceeded" });
|
||||
expect(receiver.spillOpenFiles).toBe(1);
|
||||
// The dispose of the in-flight first body rejects its begin promise with a
|
||||
// terminal channel-closed error and frees the file slot.
|
||||
await receiver.disposeBody("f1");
|
||||
await expect(first).rejects.toMatchObject({ code: "channel_closed" });
|
||||
expect(receiver.spillOpenFiles).toBe(0);
|
||||
await receiver.destroy();
|
||||
});
|
||||
});
|
||||
|
||||
describe("duplex body reassembler — lifecycle cleanup", () => {
|
||||
it("releases the reservation and unlinks the file on channel death", async () => {
|
||||
const receiver = await makeReceiver({ spillThresholdBytes: 4, rawChunkBytes: RAW });
|
||||
// The channel death rejects this in-flight body, so attach the catch at once.
|
||||
const dying = receiver.begin("death", 24);
|
||||
dying.catch(() => undefined);
|
||||
expect(receiver.pushChunk({ version: DUPLEX_FRAME_VERSION, type: "body_chunk", id: "death", seq: 0, data: Buffer.alloc(RAW, 1).toString("base64") }).ok).toBe(true);
|
||||
// Wait for the write to land, so the file exists before the channel dies.
|
||||
await new Promise((resolve) => setTimeout(resolve, 20));
|
||||
expect(receiver.spillOpenFiles).toBe(1);
|
||||
await receiver.disposeAll();
|
||||
await expect(dying).rejects.toMatchObject({ code: "channel_closed" });
|
||||
expect(receiver.spillUsedBytes).toBe(0);
|
||||
expect(receiver.spillOpenFiles).toBe(0);
|
||||
const entries = await fs.readdir(receiver.spillDir);
|
||||
expect(entries).toHaveLength(0);
|
||||
await receiver.destroy();
|
||||
});
|
||||
});
|
||||
|
||||
describe("duplex body reassembler — spill file safety", () => {
|
||||
it("creates the spill directory with mode 0700 and each spill file with mode 0600", async () => {
|
||||
const receiver = await makeReceiver({ spillThresholdBytes: 4, rawChunkBytes: RAW });
|
||||
const dirStat = await fs.stat(receiver.spillDir);
|
||||
expect(dirStat.mode & 0o777).toBe(0o700);
|
||||
// The final destroy rejects this in-flight body, so attach the catch at once.
|
||||
const pending = receiver.begin("mode", 8);
|
||||
pending.catch(() => undefined);
|
||||
receiver.pushChunk({ version: DUPLEX_FRAME_VERSION, type: "body_chunk", id: "mode", seq: 0, data: Buffer.alloc(RAW, 1).toString("base64") });
|
||||
await new Promise((resolve) => setTimeout(resolve, 20));
|
||||
const entries = await fs.readdir(receiver.spillDir);
|
||||
expect(entries).toHaveLength(1);
|
||||
// The file name is random, so no frame id rides the path.
|
||||
expect(entries[0]).not.toContain("mode");
|
||||
const fileStat = await fs.stat(path.join(receiver.spillDir, entries[0]));
|
||||
expect(fileStat.mode & 0o777).toBe(0o600);
|
||||
await receiver.destroy();
|
||||
});
|
||||
});
|
||||
|
||||
describe("duplex body reassembler — startup sweep", () => {
|
||||
it("removes a stale spill directory from a dead process but keeps a live one", async () => {
|
||||
const deadPid = 999999; // A pid that is almost never alive.
|
||||
const staleDir = path.join(spillRoot, `paperclip-duplex-spill-${deadPid}-stale`);
|
||||
await fs.mkdir(staleDir);
|
||||
await fs.writeFile(path.join(staleDir, "leftover.body"), "x");
|
||||
const liveDir = path.join(spillRoot, `paperclip-duplex-spill-${process.pid}-live`);
|
||||
await fs.mkdir(liveDir);
|
||||
await sweepStaleSpillDirectories(spillRoot);
|
||||
await expect(fs.stat(staleDir)).rejects.toThrow();
|
||||
// The sweep keeps a directory that belongs to a live process.
|
||||
await expect(fs.stat(liveDir)).resolves.toBeDefined();
|
||||
});
|
||||
});
|
||||
|
|
@ -1,596 +0,0 @@
|
|||
/**
|
||||
* Receive-side body reassembly and spill for the sandbox duplex channel.
|
||||
*
|
||||
* The duplex codec splits one request body or one response body into an envelope
|
||||
* frame plus a sequence of `body_chunk` frames. This module reassembles those
|
||||
* chunks back into one body. The receiver reads `bodyByteCount` from the envelope
|
||||
* first, then chooses one of two paths before it reads the first chunk:
|
||||
* - memory path: a body at or below the spill threshold stays in a memory
|
||||
* buffer.
|
||||
* - spill path: a body above the spill threshold streams to a temporary file,
|
||||
* so the receiver never holds a whole large body in memory.
|
||||
*
|
||||
* The spill path bounds the disk a single channel can consume. A per-channel
|
||||
* arena caps the bytes on disk and the number of open spill files. The receiver
|
||||
* reserves a file slot and reserves bytes before it opens a file and before each
|
||||
* write. A reserve that exceeds a cap fails closed: it rejects the body as a
|
||||
* resource error and unlinks any partial file.
|
||||
*
|
||||
* The reassembler treats every frame as untrusted. It rejects a bad chunk as a
|
||||
* loud, terminal protocol error: a non-canonical base64 payload, a wrong chunk
|
||||
* size, an out-of-order `seq`, a total that does not equal `bodyByteCount`, or a
|
||||
* chunk that arrives for a completed or unknown body.
|
||||
*/
|
||||
|
||||
import { randomBytes } from "node:crypto";
|
||||
import { createReadStream } from "node:fs";
|
||||
import { promises as fs } from "node:fs";
|
||||
import type { FileHandle } from "node:fs/promises";
|
||||
import os from "node:os";
|
||||
import path from "node:path";
|
||||
import { Readable } from "node:stream";
|
||||
|
||||
import {
|
||||
DUPLEX_BODY_CHUNK_RAW_BYTES,
|
||||
type DuplexBodyChunkFrame,
|
||||
} from "./duplex-frame-codec.js";
|
||||
|
||||
/**
|
||||
* The default spill threshold, in bytes. A body at or below this size stays in
|
||||
* memory. A body above it spills to disk. The value is 1 MiB. It stays
|
||||
* injectable so a test can lower it and exercise the spill path.
|
||||
*/
|
||||
export const DEFAULT_DUPLEX_BODY_SPILL_THRESHOLD_BYTES = 1024 * 1024;
|
||||
|
||||
/**
|
||||
* The default per-channel spill byte cap. The receiver keeps at most this many
|
||||
* bytes on disk at one time for one channel. The value is 64 MiB. It is about
|
||||
* 0.64% of the default 10 GiB sandbox disk and stays far below the security
|
||||
* ceiling of 25% of the disk quota. It stays injectable so a test can lower it.
|
||||
*/
|
||||
export const DEFAULT_DUPLEX_CHANNEL_SPILL_BYTE_CAP = 64 * 1024 * 1024;
|
||||
|
||||
/**
|
||||
* The default per-channel open-file cap. The receiver keeps at most this many
|
||||
* spill files open at one time for one channel. The value derives from the byte
|
||||
* cap and the spill threshold: 64 MiB divided by the 1 MiB spill floor gives 64.
|
||||
* It stays injectable so a test can lower it.
|
||||
*/
|
||||
export const DEFAULT_DUPLEX_CHANNEL_SPILL_FILE_CAP = 64;
|
||||
|
||||
/** The prefix of every spill directory name. The startup sweep matches it. */
|
||||
const SPILL_DIR_PREFIX = "paperclip-duplex-spill-";
|
||||
|
||||
/** The config a caller passes to build a {@link DuplexBodyReceiver}. */
|
||||
export interface DuplexBodyReceiverConfig {
|
||||
/** The spill threshold in bytes. Defaults to {@link DEFAULT_DUPLEX_BODY_SPILL_THRESHOLD_BYTES}. */
|
||||
spillThresholdBytes?: number;
|
||||
/** The fixed raw chunk size in bytes. Defaults to {@link DUPLEX_BODY_CHUNK_RAW_BYTES}. */
|
||||
rawChunkBytes?: number;
|
||||
/** The per-channel spill byte cap. Defaults to {@link DEFAULT_DUPLEX_CHANNEL_SPILL_BYTE_CAP}. */
|
||||
channelSpillByteCap?: number;
|
||||
/** The per-channel open-file cap. Defaults to {@link DEFAULT_DUPLEX_CHANNEL_SPILL_FILE_CAP}. */
|
||||
channelSpillFileCap?: number;
|
||||
/** The root directory that holds the spill directory. Defaults to the OS temp dir. */
|
||||
spillRootDir?: string;
|
||||
/** A diagnostic sink for a non-fatal cleanup note. The receiver names no body id here. */
|
||||
logger?: (message: string) => void;
|
||||
}
|
||||
|
||||
/** The reason the reassembler rejected a body. Each reason is a terminal protocol error. */
|
||||
export type DuplexBodyErrorCode =
|
||||
| "chunk_without_envelope"
|
||||
| "chunk_after_completion"
|
||||
| "seq_out_of_order"
|
||||
| "chunk_not_canonical_base64"
|
||||
| "chunk_wrong_size"
|
||||
| "empty_chunk"
|
||||
| "body_overrun"
|
||||
| "zero_body_has_chunk"
|
||||
| "spill_cap_exceeded"
|
||||
| "spill_write_failed"
|
||||
| "path_escape"
|
||||
| "channel_closed";
|
||||
|
||||
/** A reassembly error. The receiver returns it; it never leaks a raw path or a raw provider string. */
|
||||
export class DuplexBodyError extends Error {
|
||||
readonly code: DuplexBodyErrorCode;
|
||||
constructor(code: DuplexBodyErrorCode, message: string) {
|
||||
super(message);
|
||||
this.name = "DuplexBodyError";
|
||||
this.code = code;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* One reassembled body. The memory path holds the whole body in a buffer. The
|
||||
* spill path holds a temporary file. A caller reads the memory buffer, or opens
|
||||
* a fresh read stream, then disposes the body once.
|
||||
*/
|
||||
export interface ReassembledBody {
|
||||
/** The raw byte length of the body. */
|
||||
readonly byteCount: number;
|
||||
/** True when the body spilled to disk. */
|
||||
readonly spilled: boolean;
|
||||
/**
|
||||
* The whole body as a buffer. The memory path returns it. The spill path
|
||||
* throws, because the spill path must never load a whole large body into
|
||||
* memory on the receive side. A caller streams a spilled body instead.
|
||||
*/
|
||||
readMemory(): Buffer;
|
||||
/** A fresh read stream over the body. The memory path streams one buffer; the spill path streams the file. */
|
||||
createReadStream(): Readable;
|
||||
/** Release the body. The spill path unlinks the file and releases the reservation exactly once. */
|
||||
dispose(): Promise<void>;
|
||||
}
|
||||
|
||||
/**
|
||||
* The per-channel arena that bounds the spill disk use. It tracks the bytes on
|
||||
* disk and the open files, and it owns the spill directory. It hands out
|
||||
* reservations and takes them back. The reservation math is synchronous, so a
|
||||
* concurrent reserve never races past a cap.
|
||||
*/
|
||||
class SpillArena {
|
||||
usedBytes = 0;
|
||||
openFiles = 0;
|
||||
constructor(
|
||||
readonly dir: string,
|
||||
private readonly byteCap: number,
|
||||
private readonly fileCap: number,
|
||||
) {}
|
||||
|
||||
/** Reserve one open-file slot. Return false when the file cap is already at the limit. */
|
||||
reserveFile(): boolean {
|
||||
if (this.openFiles >= this.fileCap) return false;
|
||||
this.openFiles += 1;
|
||||
return true;
|
||||
}
|
||||
|
||||
/** Release one open-file slot. */
|
||||
releaseFile(): void {
|
||||
if (this.openFiles > 0) this.openFiles -= 1;
|
||||
}
|
||||
|
||||
/** Reserve `n` bytes of disk. Return false when the byte cap would be exceeded. */
|
||||
reserveBytes(n: number): boolean {
|
||||
if (this.usedBytes + n > this.byteCap) return false;
|
||||
this.usedBytes += n;
|
||||
return true;
|
||||
}
|
||||
|
||||
/** Release `n` bytes of disk. */
|
||||
releaseBytes(n: number): void {
|
||||
this.usedBytes -= n;
|
||||
if (this.usedBytes < 0) this.usedBytes = 0;
|
||||
}
|
||||
}
|
||||
|
||||
/** The internal state of one in-flight body. */
|
||||
interface ActiveBody {
|
||||
bodyByteCount: number;
|
||||
nextSeq: number;
|
||||
receivedBytes: number;
|
||||
spilled: boolean;
|
||||
/** Memory-path accumulation. Empty for the spill path. */
|
||||
memoryChunks: Buffer[];
|
||||
/** The spill file path. Null for the memory path. */
|
||||
filePath: string | null;
|
||||
/** The open spill file handle. Null for the memory path or before the first write. */
|
||||
handle: FileHandle | null;
|
||||
/** The bytes this body reserved in the arena. The dispose path releases exactly this many. */
|
||||
reservedBytes: number;
|
||||
/** True when the body holds an arena file slot that the dispose path must release. */
|
||||
holdsFileSlot: boolean;
|
||||
/** True once the dispose path ran, so it runs its release once. */
|
||||
disposed: boolean;
|
||||
/** Resolve the completion promise the caller awaits. */
|
||||
resolve: (body: ReassembledBody) => void;
|
||||
/** Reject the completion promise on a terminal error. */
|
||||
reject: (error: DuplexBodyError) => void;
|
||||
/** Serialize the spill writes, so the chunks land on disk in `seq` order. */
|
||||
writeChain: Promise<void>;
|
||||
/** True once the body settled (resolved or rejected), so a late chunk never double-settles. */
|
||||
settled: boolean;
|
||||
}
|
||||
|
||||
/** The synchronous result of one `pushChunk` call. */
|
||||
export type DuplexBodyPushResult =
|
||||
| { ok: true }
|
||||
| { ok: false; error: DuplexBodyError };
|
||||
|
||||
/**
|
||||
* A per-channel receive-side body reassembler. One instance serves one channel.
|
||||
* The caller feeds it one envelope per body, then the `body_chunk` frames for
|
||||
* that body. The reassembler reassembles the body on the memory path or the
|
||||
* spill path and settles one completion promise per body.
|
||||
*/
|
||||
export class DuplexBodyReceiver {
|
||||
private readonly spillThresholdBytes: number;
|
||||
private readonly rawChunkBytes: number;
|
||||
private readonly arena: SpillArena;
|
||||
private readonly logger?: (message: string) => void;
|
||||
private readonly active = new Map<string, ActiveBody>();
|
||||
|
||||
private constructor(arena: SpillArena, config: DuplexBodyReceiverConfig) {
|
||||
this.arena = arena;
|
||||
this.spillThresholdBytes =
|
||||
config.spillThresholdBytes ?? DEFAULT_DUPLEX_BODY_SPILL_THRESHOLD_BYTES;
|
||||
this.rawChunkBytes = config.rawChunkBytes ?? DUPLEX_BODY_CHUNK_RAW_BYTES;
|
||||
this.logger = config.logger;
|
||||
}
|
||||
|
||||
/**
|
||||
* Build a receiver. The receiver owns the spill directory. It creates the
|
||||
* directory with `mkdtemp` and mode 0700, then sweeps stale spill directories
|
||||
* from a prior process. The sweep runs and awaits before the receiver serves a
|
||||
* body, so no caller races a half-swept directory.
|
||||
*/
|
||||
static async create(config: DuplexBodyReceiverConfig = {}): Promise<DuplexBodyReceiver> {
|
||||
const root = config.spillRootDir ?? os.tmpdir();
|
||||
await sweepStaleSpillDirectories(root, config.logger);
|
||||
const dir = await fs.mkdtemp(path.join(root, `${SPILL_DIR_PREFIX}${process.pid}-`));
|
||||
// `mkdtemp` creates the directory with mode 0700 on a POSIX host by default,
|
||||
// but a permissive umask can widen it. Set the mode so the owner alone reads,
|
||||
// writes, and lists the spill directory.
|
||||
await fs.chmod(dir, 0o700);
|
||||
const arena = new SpillArena(
|
||||
dir,
|
||||
config.channelSpillByteCap ?? DEFAULT_DUPLEX_CHANNEL_SPILL_BYTE_CAP,
|
||||
config.channelSpillFileCap ?? DEFAULT_DUPLEX_CHANNEL_SPILL_FILE_CAP,
|
||||
);
|
||||
return new DuplexBodyReceiver(arena, config);
|
||||
}
|
||||
|
||||
/** The spill directory this receiver owns. A test reads it to assert containment and cleanup. */
|
||||
get spillDir(): string {
|
||||
return this.arena.dir;
|
||||
}
|
||||
|
||||
/** The bytes on disk across all in-flight bodies for this channel. A test asserts the reservation. */
|
||||
get spillUsedBytes(): number {
|
||||
return this.arena.usedBytes;
|
||||
}
|
||||
|
||||
/** The open spill files for this channel. A test asserts the file-slot reservation. */
|
||||
get spillOpenFiles(): number {
|
||||
return this.arena.openFiles;
|
||||
}
|
||||
|
||||
/**
|
||||
* Begin one body. The caller passes the envelope `id` and `bodyByteCount`. The
|
||||
* receiver chooses the memory path or the spill path from `bodyByteCount`
|
||||
* before the first chunk arrives. A `bodyByteCount` of 0 completes at once with
|
||||
* an empty body and accepts no chunk. The returned promise resolves with the
|
||||
* reassembled body, or rejects with a {@link DuplexBodyError} on a terminal
|
||||
* protocol or resource error.
|
||||
*/
|
||||
begin(id: string, bodyByteCount: number): Promise<ReassembledBody> {
|
||||
return new Promise<ReassembledBody>((resolve, reject) => {
|
||||
// A repeated envelope id for an in-flight body is a protocol violation. The
|
||||
// caller must not begin the same id twice. Reject the new begin and leave
|
||||
// the in-flight body untouched.
|
||||
if (this.active.has(id)) {
|
||||
reject(new DuplexBodyError("chunk_after_completion", "duplicate body envelope id"));
|
||||
return;
|
||||
}
|
||||
const body: ActiveBody = {
|
||||
bodyByteCount,
|
||||
nextSeq: 0,
|
||||
receivedBytes: 0,
|
||||
spilled: bodyByteCount > this.spillThresholdBytes,
|
||||
memoryChunks: [],
|
||||
filePath: null,
|
||||
handle: null,
|
||||
reservedBytes: 0,
|
||||
holdsFileSlot: false,
|
||||
disposed: false,
|
||||
resolve,
|
||||
reject,
|
||||
writeChain: Promise.resolve(),
|
||||
settled: false,
|
||||
};
|
||||
this.active.set(id, body);
|
||||
if (bodyByteCount === 0) {
|
||||
// The zero-length body completes at once. It accepts no chunk, so the
|
||||
// receiver keeps the id in a completed state to reject a stray chunk.
|
||||
this.complete(id, body);
|
||||
return;
|
||||
}
|
||||
if (body.spilled) {
|
||||
// Reserve one file slot before the receiver opens the spill file. A
|
||||
// reserve past the file cap fails closed at once.
|
||||
if (!this.arena.reserveFile()) {
|
||||
this.settleReject(id, body, new DuplexBodyError("spill_cap_exceeded", "spill file cap reached"));
|
||||
return;
|
||||
}
|
||||
body.holdsFileSlot = true;
|
||||
// Open the spill file with a random name and mode 0600. The name never
|
||||
// derives from the frame id, so no id rides a path. Verify the resolved
|
||||
// path stays inside the spill directory before the receiver writes.
|
||||
this.openSpillFile(id, body);
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
/**
|
||||
* Push one `body_chunk` frame. The receiver validates the chunk against the
|
||||
* in-flight body, then appends it on the memory path or the spill path. The
|
||||
* return value reports a synchronous validation outcome. A spill write failure
|
||||
* or a completion rejects the begin promise instead.
|
||||
*/
|
||||
pushChunk(frame: DuplexBodyChunkFrame): DuplexBodyPushResult {
|
||||
const body = this.active.get(frame.id);
|
||||
if (!body) {
|
||||
return fail("chunk_without_envelope", "body_chunk arrived with no matching envelope");
|
||||
}
|
||||
if (body.bodyByteCount === 0) {
|
||||
return fail("zero_body_has_chunk", "a zero-length body accepts no body_chunk");
|
||||
}
|
||||
if (body.settled) {
|
||||
return fail("chunk_after_completion", "body_chunk arrived after the body completed");
|
||||
}
|
||||
if (frame.seq !== body.nextSeq) {
|
||||
return fail("seq_out_of_order", "body_chunk seq is a duplicate, a gap, or a reorder");
|
||||
}
|
||||
// Decode the base64 payload and confirm it is canonical. A non-canonical
|
||||
// payload re-encodes to a different string, so the round-trip catches it.
|
||||
const decoded = Buffer.from(frame.data, "base64");
|
||||
if (decoded.toString("base64") !== frame.data) {
|
||||
return fail("chunk_not_canonical_base64", "body_chunk data is not canonical base64");
|
||||
}
|
||||
if (decoded.length === 0) {
|
||||
return fail("empty_chunk", "body_chunk is empty");
|
||||
}
|
||||
const nextReceived = body.receivedBytes + decoded.length;
|
||||
if (nextReceived > body.bodyByteCount) {
|
||||
return fail("body_overrun", "body_chunk overruns the declared body size");
|
||||
}
|
||||
const isFinal = nextReceived === body.bodyByteCount;
|
||||
if (!isFinal && decoded.length !== this.rawChunkBytes) {
|
||||
// A non-final chunk must be exactly the fixed raw slice size. A short
|
||||
// non-final chunk is a truncation; an over-size chunk is a framing fault.
|
||||
return fail("chunk_wrong_size", "a non-final body_chunk is not the fixed slice size");
|
||||
}
|
||||
if (decoded.length > this.rawChunkBytes) {
|
||||
// The final chunk must also stay within the fixed raw slice size.
|
||||
return fail("chunk_wrong_size", "a body_chunk exceeds the fixed slice size");
|
||||
}
|
||||
body.nextSeq += 1;
|
||||
body.receivedBytes = nextReceived;
|
||||
if (body.spilled) {
|
||||
this.appendSpill(frame.id, body, decoded, isFinal);
|
||||
} else {
|
||||
body.memoryChunks.push(decoded);
|
||||
if (isFinal) this.complete(frame.id, body);
|
||||
}
|
||||
return { ok: true };
|
||||
}
|
||||
|
||||
/**
|
||||
* Dispose one body by id, whatever its state. The dispose path unlinks the
|
||||
* spill file and releases the arena reservation exactly once. The caller runs
|
||||
* it on every settle path and on channel death, so a spill file never lingers.
|
||||
*/
|
||||
async disposeBody(id: string): Promise<void> {
|
||||
const body = this.active.get(id);
|
||||
if (!body) return;
|
||||
this.active.delete(id);
|
||||
await this.release(body);
|
||||
}
|
||||
|
||||
/**
|
||||
* Dispose every in-flight body. The caller runs it on channel death, so a
|
||||
* channel that dies mid-body leaves no spill file behind.
|
||||
*/
|
||||
async disposeAll(): Promise<void> {
|
||||
const ids = [...this.active.keys()];
|
||||
await Promise.all(ids.map((id) => this.disposeBody(id)));
|
||||
}
|
||||
|
||||
/**
|
||||
* Remove the spill directory this receiver owns. The caller runs it after it
|
||||
* disposes every body, on channel shutdown.
|
||||
*/
|
||||
async destroy(): Promise<void> {
|
||||
await this.disposeAll();
|
||||
await fs.rm(this.arena.dir, { recursive: true, force: true }).catch(() => undefined);
|
||||
}
|
||||
|
||||
private openSpillFile(id: string, body: ActiveBody): void {
|
||||
const name = `${randomBytes(24).toString("hex")}.body`;
|
||||
const filePath = path.join(this.arena.dir, name);
|
||||
// Path containment. The resolved file path must stay inside the spill
|
||||
// directory. A generated random name never escapes, but the check is defense
|
||||
// in depth against a future name scheme.
|
||||
const resolvedDir = path.resolve(this.arena.dir);
|
||||
const resolvedFile = path.resolve(filePath);
|
||||
if (resolvedFile !== path.join(resolvedDir, name) || !resolvedFile.startsWith(resolvedDir + path.sep)) {
|
||||
this.settleReject(id, body, new DuplexBodyError("path_escape", "spill path escapes the spill directory"));
|
||||
return;
|
||||
}
|
||||
body.filePath = filePath;
|
||||
// Open the file once, exclusive, mode 0600. The write chain appends each
|
||||
// chunk in order to this one handle.
|
||||
body.writeChain = body.writeChain.then(async () => {
|
||||
body.handle = await fs.open(filePath, "wx", 0o600);
|
||||
});
|
||||
body.writeChain.catch((error) => {
|
||||
this.settleReject(id, body, new DuplexBodyError("spill_write_failed", errorText(error)));
|
||||
});
|
||||
}
|
||||
|
||||
private appendSpill(id: string, body: ActiveBody, decoded: Buffer, isFinal: boolean): void {
|
||||
// Reserve the bytes before the write. A reserve past the byte cap fails
|
||||
// closed: reject the body and unlink the partial file.
|
||||
if (!this.arena.reserveBytes(decoded.length)) {
|
||||
this.settleReject(id, body, new DuplexBodyError("spill_cap_exceeded", "spill byte cap reached"));
|
||||
return;
|
||||
}
|
||||
body.reservedBytes += decoded.length;
|
||||
body.writeChain = body.writeChain.then(async () => {
|
||||
if (body.settled) return;
|
||||
if (!body.handle) throw new Error("spill file handle is not open");
|
||||
await body.handle.write(decoded);
|
||||
if (isFinal) {
|
||||
await body.handle.close();
|
||||
body.handle = null;
|
||||
this.complete(id, body);
|
||||
}
|
||||
});
|
||||
body.writeChain.catch((error) => {
|
||||
if (error instanceof DuplexBodyError) return;
|
||||
this.settleReject(id, body, new DuplexBodyError("spill_write_failed", errorText(error)));
|
||||
});
|
||||
}
|
||||
|
||||
private complete(id: string, body: ActiveBody): void {
|
||||
if (body.settled) return;
|
||||
body.settled = true;
|
||||
const receiver = this;
|
||||
const handle: ReassembledBody = {
|
||||
byteCount: body.bodyByteCount,
|
||||
spilled: body.spilled,
|
||||
readMemory(): Buffer {
|
||||
if (body.spilled) {
|
||||
throw new DuplexBodyError("spill_write_failed", "a spilled body has no in-memory buffer");
|
||||
}
|
||||
return Buffer.concat(body.memoryChunks);
|
||||
},
|
||||
createReadStream(): Readable {
|
||||
if (body.spilled && body.filePath) return createReadStream(body.filePath);
|
||||
return Readable.from(Buffer.concat(body.memoryChunks));
|
||||
},
|
||||
async dispose(): Promise<void> {
|
||||
await receiver.disposeBody(id);
|
||||
},
|
||||
};
|
||||
body.resolve(handle);
|
||||
}
|
||||
|
||||
private settleReject(id: string, body: ActiveBody, error: DuplexBodyError): void {
|
||||
if (body.settled) return;
|
||||
body.settled = true;
|
||||
// Release the reservation and unlink the partial file before the reject, so
|
||||
// a rejected body never keeps a spill file or a reservation. The reject fires
|
||||
// after the release, so a caller that awaits the reject sees a clean arena.
|
||||
void this.disposeBody(id).finally(() => body.reject(error));
|
||||
}
|
||||
|
||||
private async release(body: ActiveBody): Promise<void> {
|
||||
if (body.disposed) return;
|
||||
body.disposed = true;
|
||||
// Wait for the pending writes to settle, so the unlink never races an open
|
||||
// handle. Swallow a write error here; the reject path already reported it.
|
||||
await body.writeChain.catch(() => undefined);
|
||||
if (body.handle) {
|
||||
await body.handle.close().catch(() => undefined);
|
||||
body.handle = null;
|
||||
}
|
||||
if (body.filePath) {
|
||||
await fs.rm(body.filePath, { force: true }).catch((error) => {
|
||||
this.logger?.(`Duplex spill unlink failed: ${errorText(error)}`);
|
||||
});
|
||||
}
|
||||
if (body.reservedBytes > 0) {
|
||||
this.arena.releaseBytes(body.reservedBytes);
|
||||
body.reservedBytes = 0;
|
||||
}
|
||||
if (body.holdsFileSlot) {
|
||||
this.arena.releaseFile();
|
||||
body.holdsFileSlot = false;
|
||||
}
|
||||
// A body that the channel disposes before it settles must not leave its
|
||||
// begin promise pending. Reject it with a terminal channel-closed error, so
|
||||
// a caller that awaits the body sees the channel death. A body that already
|
||||
// settled keeps its outcome: `complete` resolves it, and `settleReject` sets
|
||||
// `settled` before it calls dispose, so this guard never double-settles.
|
||||
if (!body.settled) {
|
||||
body.settled = true;
|
||||
body.reject(
|
||||
new DuplexBodyError("channel_closed", "the channel closed before the body completed"),
|
||||
);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
function fail(code: DuplexBodyErrorCode, message: string): DuplexBodyPushResult {
|
||||
return { ok: false, error: new DuplexBodyError(code, message) };
|
||||
}
|
||||
|
||||
function errorText(error: unknown): string {
|
||||
return error instanceof Error ? error.message : String(error);
|
||||
}
|
||||
|
||||
/**
|
||||
* Split a whole body buffer into `body_chunk` frames. The send side reads the
|
||||
* source body once, then calls this to build the frames. Each frame carries one
|
||||
* raw slice of `rawChunkBytes` as base64 text, except the final frame, which
|
||||
* carries the remaining bytes. A zero-length body yields no frame.
|
||||
*/
|
||||
export function splitBodyIntoChunkFrames(
|
||||
id: string,
|
||||
body: Buffer,
|
||||
version: number,
|
||||
rawChunkBytes: number = DUPLEX_BODY_CHUNK_RAW_BYTES,
|
||||
): DuplexBodyChunkFrame[] {
|
||||
const frames: DuplexBodyChunkFrame[] = [];
|
||||
let seq = 0;
|
||||
for (let offset = 0; offset < body.length; offset += rawChunkBytes) {
|
||||
const slice = body.subarray(offset, offset + rawChunkBytes);
|
||||
frames.push({
|
||||
version,
|
||||
type: "body_chunk",
|
||||
id,
|
||||
seq,
|
||||
data: slice.toString("base64"),
|
||||
});
|
||||
seq += 1;
|
||||
}
|
||||
return frames;
|
||||
}
|
||||
|
||||
/**
|
||||
* Sweep stale spill directories from a prior process. The receiver runs it at
|
||||
* startup, before it serves a body. It removes a spill directory whose embedded
|
||||
* pid is not this process and is not a live process, so a crash that left a
|
||||
* directory behind does not leak disk. It never removes a directory of a live
|
||||
* process, so a concurrent channel keeps its own spill directory.
|
||||
*/
|
||||
export async function sweepStaleSpillDirectories(
|
||||
root: string,
|
||||
logger?: (message: string) => void,
|
||||
): Promise<void> {
|
||||
const entries = await fs.readdir(root, { withFileTypes: true }).catch(() => []);
|
||||
await Promise.all(
|
||||
entries.map(async (entry) => {
|
||||
if (!entry.isDirectory() || !entry.name.startsWith(SPILL_DIR_PREFIX)) return;
|
||||
const pid = parsePidFromSpillDir(entry.name);
|
||||
if (pid !== null && pid !== process.pid && isProcessAlive(pid)) return;
|
||||
if (pid === process.pid) return;
|
||||
await fs.rm(path.join(root, entry.name), { recursive: true, force: true }).catch((error) => {
|
||||
logger?.(`Duplex stale spill sweep failed: ${errorText(error)}`);
|
||||
});
|
||||
}),
|
||||
);
|
||||
}
|
||||
|
||||
/** Read the pid segment from a spill directory name, or null when it is absent. */
|
||||
function parsePidFromSpillDir(name: string): number | null {
|
||||
const rest = name.slice(SPILL_DIR_PREFIX.length);
|
||||
const dash = rest.indexOf("-");
|
||||
const pidText = dash === -1 ? rest : rest.slice(0, dash);
|
||||
const pid = Number(pidText);
|
||||
return Number.isInteger(pid) && pid > 0 ? pid : null;
|
||||
}
|
||||
|
||||
/** Return true when a process with the pid is alive. A signal 0 probes without a real signal. */
|
||||
function isProcessAlive(pid: number): boolean {
|
||||
try {
|
||||
process.kill(pid, 0);
|
||||
return true;
|
||||
} catch (error) {
|
||||
// EPERM means the process exists but this process cannot signal it. ESRCH
|
||||
// means no such process. Treat only ESRCH as dead.
|
||||
return (error as NodeJS.ErrnoException).code === "EPERM";
|
||||
}
|
||||
}
|
||||
|
|
@ -3,16 +3,12 @@ import { fileURLToPath } from "node:url";
|
|||
import { describe, expect, it } from "vitest";
|
||||
import {
|
||||
DEFAULT_MAX_DUPLEX_FRAME_BYTES,
|
||||
DEFAULT_MAX_DUPLEX_REQUEST_ID_BYTES,
|
||||
DUPLEX_FRAME_VERSION,
|
||||
DuplexFrameDecoder,
|
||||
decodeDuplexLine,
|
||||
encodeDuplexFrame,
|
||||
encodeDuplexFrameChecked,
|
||||
type DuplexDecodeResult,
|
||||
type DuplexFrame,
|
||||
} from "./duplex-frame-codec.js";
|
||||
import { DuplexAggregateByteLedger } from "./duplex-aggregate-byte-ledger.js";
|
||||
|
||||
// One expected decode result in the fixture: either a decoded frame or the code
|
||||
// of a protocol error. The codec must never throw on the read path.
|
||||
|
|
@ -20,17 +16,8 @@ type ExpectedResult = { frame: DuplexFrame } | { error: string };
|
|||
|
||||
interface Vector {
|
||||
name: string;
|
||||
category:
|
||||
| "valid"
|
||||
| "invalid"
|
||||
| "partial"
|
||||
| "oversized"
|
||||
| "versionMismatch"
|
||||
| "idBoundValid"
|
||||
| "idBoundInvalid";
|
||||
category: "valid" | "invalid" | "versionMismatch";
|
||||
bytes: string;
|
||||
splitByteOffsets?: number[];
|
||||
maxFrameBytes?: number;
|
||||
roundTrip?: boolean;
|
||||
expected: ExpectedResult[];
|
||||
}
|
||||
|
|
@ -56,40 +43,6 @@ const fixturePath = fileURLToPath(
|
|||
);
|
||||
const fixture = JSON.parse(readFileSync(fixturePath, "utf8")) as Fixture;
|
||||
|
||||
// Cut one UTF-8 byte stream into chunks at the byte offsets. The decoder must
|
||||
// keep partial bytes between chunks, so the split can fall inside a multi-byte
|
||||
// character.
|
||||
function toChunks(bytes: string, offsets: number[] | undefined): Buffer[] {
|
||||
const buffer = Buffer.from(bytes, "utf8");
|
||||
if (!offsets || offsets.length === 0) return [buffer];
|
||||
const bounds = [0, ...offsets, buffer.length];
|
||||
const chunks: Buffer[] = [];
|
||||
for (let i = 0; i < bounds.length - 1; i += 1) {
|
||||
chunks.push(buffer.subarray(bounds[i], bounds[i + 1]));
|
||||
}
|
||||
return chunks;
|
||||
}
|
||||
|
||||
function pushAll(decoder: DuplexFrameDecoder, chunks: Buffer[]): DuplexDecodeResult[] {
|
||||
const results: DuplexDecodeResult[] = [];
|
||||
for (const chunk of chunks) results.push(...decoder.push(chunk));
|
||||
return results;
|
||||
}
|
||||
|
||||
function assertMatches(results: DuplexDecodeResult[], expected: ExpectedResult[]): void {
|
||||
expect(results).toHaveLength(expected.length);
|
||||
results.forEach((result, index) => {
|
||||
const want = expected[index];
|
||||
if ("frame" in want) {
|
||||
expect(result.ok).toBe(true);
|
||||
if (result.ok) expect(result.frame).toEqual(want.frame);
|
||||
} else {
|
||||
expect(result.ok).toBe(false);
|
||||
if (!result.ok) expect(result.error.code).toBe(want.error);
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
describe("duplex frame codec fixture", () => {
|
||||
it("the fixture version matches the codec version", () => {
|
||||
expect(fixture.frameVersion).toBe(DUPLEX_FRAME_VERSION);
|
||||
|
|
@ -98,19 +51,26 @@ describe("duplex frame codec fixture", () => {
|
|||
|
||||
it("every category has at least one vector", () => {
|
||||
const categories = new Set(fixture.vectors.map((vector) => vector.category));
|
||||
for (const category of ["valid", "invalid", "partial", "oversized", "versionMismatch"]) {
|
||||
for (const category of ["valid", "invalid", "versionMismatch"]) {
|
||||
expect(categories).toContain(category);
|
||||
}
|
||||
});
|
||||
|
||||
for (const vector of fixture.vectors) {
|
||||
it(`decodes the ${vector.name} vector`, () => {
|
||||
const decoder = new DuplexFrameDecoder(
|
||||
vector.maxFrameBytes ? { maxFrameBytes: vector.maxFrameBytes } : undefined,
|
||||
);
|
||||
const chunks = toChunks(vector.bytes, vector.splitByteOffsets);
|
||||
const results = pushAll(decoder, chunks);
|
||||
assertMatches(results, vector.expected);
|
||||
// Every remaining vector is one complete line, so `decodeDuplexLine`
|
||||
// reads it directly. The trailing newline in the fixture bytes is not
|
||||
// part of the line the decoder reads.
|
||||
const line = vector.bytes.endsWith("\n") ? vector.bytes.slice(0, -1) : vector.bytes;
|
||||
const result = decodeDuplexLine(line);
|
||||
const want = vector.expected[0];
|
||||
if ("frame" in want) {
|
||||
expect(result.ok).toBe(true);
|
||||
if (result.ok) expect(result.frame).toEqual(want.frame);
|
||||
} else {
|
||||
expect(result.ok).toBe(false);
|
||||
if (!result.ok) expect(result.error.code).toBe(want.error);
|
||||
}
|
||||
});
|
||||
}
|
||||
});
|
||||
|
|
@ -126,7 +86,7 @@ describe("round-trip", () => {
|
|||
),
|
||||
),
|
||||
);
|
||||
for (const type of ["request", "response", "body_chunk", "ready", "heartbeat", "close", "error"]) {
|
||||
for (const type of ["ready", "heartbeat", "close", "error"]) {
|
||||
expect(types).toContain(type);
|
||||
}
|
||||
});
|
||||
|
|
@ -191,249 +151,6 @@ describe("ready frame schema", () => {
|
|||
});
|
||||
});
|
||||
|
||||
describe("streaming decoder behavior", () => {
|
||||
it("emits nothing until a full line arrives, then the complete frame", () => {
|
||||
const decoder = new DuplexFrameDecoder();
|
||||
const line = encodeDuplexFrame({ version: DUPLEX_FRAME_VERSION, type: "heartbeat" });
|
||||
const bytes = Buffer.from(line, "utf8");
|
||||
const first = decoder.push(bytes.subarray(0, 3));
|
||||
expect(first).toHaveLength(0);
|
||||
const second = decoder.push(bytes.subarray(3));
|
||||
expect(second).toHaveLength(1);
|
||||
expect(second[0].ok).toBe(true);
|
||||
});
|
||||
|
||||
it("keeps a multi-byte UTF-8 sequence valid across a chunk boundary", () => {
|
||||
const decoder = new DuplexFrameDecoder();
|
||||
const frame: DuplexFrame = {
|
||||
version: DUPLEX_FRAME_VERSION,
|
||||
type: "request",
|
||||
id: "req-emoji",
|
||||
method: "POST",
|
||||
path: "/x",
|
||||
query: "",
|
||||
headers: { "x-note": "😀" },
|
||||
bodyByteCount: 0,
|
||||
};
|
||||
const bytes = Buffer.from(encodeDuplexFrame(frame), "utf8");
|
||||
// Split inside the four-byte emoji, right before a continuation byte.
|
||||
let cut = -1;
|
||||
for (let i = 1; i < bytes.length; i += 1) {
|
||||
if ((bytes[i] & 0xc0) === 0x80) {
|
||||
cut = i;
|
||||
break;
|
||||
}
|
||||
}
|
||||
expect(cut).toBeGreaterThan(0);
|
||||
const results = [
|
||||
...decoder.push(bytes.subarray(0, cut)),
|
||||
...decoder.push(bytes.subarray(cut)),
|
||||
];
|
||||
expect(results).toHaveLength(1);
|
||||
expect(results[0].ok).toBe(true);
|
||||
if (results[0].ok) expect(results[0].frame).toEqual(frame);
|
||||
});
|
||||
|
||||
it("rejects an oversized frame with a protocol error, then resynchronizes", () => {
|
||||
const decoder = new DuplexFrameDecoder({ maxFrameBytes: 32 });
|
||||
const flood = `${"z".repeat(100)}\n`;
|
||||
const good = encodeDuplexFrame({ version: DUPLEX_FRAME_VERSION, type: "heartbeat" });
|
||||
const results = decoder.push(Buffer.from(flood + good, "utf8"));
|
||||
expect(results).toHaveLength(2);
|
||||
expect(results[0].ok).toBe(false);
|
||||
if (!results[0].ok) expect(results[0].error.code).toBe("frame_too_large");
|
||||
expect(results[1].ok).toBe(true);
|
||||
});
|
||||
|
||||
it("rejects a version-mismatch frame with a protocol error", () => {
|
||||
const decoder = new DuplexFrameDecoder();
|
||||
const results = decoder.push(
|
||||
Buffer.from(`${JSON.stringify({ version: 999, type: "heartbeat" })}\n`, "utf8"),
|
||||
);
|
||||
expect(results).toHaveLength(1);
|
||||
expect(results[0].ok).toBe(false);
|
||||
if (!results[0].ok) expect(results[0].error.code).toBe("version_mismatch");
|
||||
});
|
||||
|
||||
it("never throws on a malformed read; it returns a protocol error", () => {
|
||||
const decoder = new DuplexFrameDecoder();
|
||||
expect(() => decoder.push(Buffer.from("this is not json\n", "utf8"))).not.toThrow();
|
||||
const results = decoder.push(Buffer.from("still not json\n", "utf8"));
|
||||
expect(results[0].ok).toBe(false);
|
||||
});
|
||||
});
|
||||
|
||||
describe("request id byte bound", () => {
|
||||
// The id byte bound caps the memory the host broker retains per distinct
|
||||
// request. The bound is 256 bytes; a UTF-8 id byte can differ from a character.
|
||||
function requestWithId(id: string): string {
|
||||
return JSON.stringify({
|
||||
version: DUPLEX_FRAME_VERSION,
|
||||
type: "request",
|
||||
id,
|
||||
method: "GET",
|
||||
path: "/",
|
||||
query: "",
|
||||
headers: {},
|
||||
bodyByteCount: 0,
|
||||
});
|
||||
}
|
||||
|
||||
it("accepts an id at the maximum byte size", () => {
|
||||
const id = "a".repeat(DEFAULT_MAX_DUPLEX_REQUEST_ID_BYTES);
|
||||
const result = decodeDuplexLine(requestWithId(id));
|
||||
expect(result.ok).toBe(true);
|
||||
if (result.ok && result.frame.type === "request") expect(result.frame.id).toBe(id);
|
||||
});
|
||||
|
||||
it("rejects an over-limit id with an id_too_large protocol error and never throws", () => {
|
||||
const id = "a".repeat(DEFAULT_MAX_DUPLEX_REQUEST_ID_BYTES + 1);
|
||||
expect(() => decodeDuplexLine(requestWithId(id))).not.toThrow();
|
||||
const result = decodeDuplexLine(requestWithId(id));
|
||||
expect(result.ok).toBe(false);
|
||||
if (!result.ok) expect(result.error.code).toBe("id_too_large");
|
||||
});
|
||||
|
||||
it("measures the id in bytes, not characters, so a multi-byte id over the byte bound is rejected", () => {
|
||||
// Each "😀" is four UTF-8 bytes. 65 code points make 260 bytes, one code point
|
||||
// makes 4 bytes, so 65 stays over the 256-byte bound while the character count
|
||||
// stays under it.
|
||||
const id = "😀".repeat(65);
|
||||
expect(id.length).toBeLessThan(DEFAULT_MAX_DUPLEX_REQUEST_ID_BYTES);
|
||||
const result = decodeDuplexLine(requestWithId(id));
|
||||
expect(result.ok).toBe(false);
|
||||
if (!result.ok) expect(result.error.code).toBe("id_too_large");
|
||||
});
|
||||
|
||||
it("applies the same bound to the response id", () => {
|
||||
const id = "b".repeat(DEFAULT_MAX_DUPLEX_REQUEST_ID_BYTES + 1);
|
||||
const result = decodeDuplexLine(
|
||||
JSON.stringify({
|
||||
version: DUPLEX_FRAME_VERSION,
|
||||
type: "response",
|
||||
id,
|
||||
status: 200,
|
||||
headers: {},
|
||||
bodyByteCount: 0,
|
||||
outcome: "completed",
|
||||
}),
|
||||
);
|
||||
expect(result.ok).toBe(false);
|
||||
if (!result.ok) expect(result.error.code).toBe("id_too_large");
|
||||
});
|
||||
});
|
||||
|
||||
describe("decoder aggregate byte ledger charging", () => {
|
||||
// The host injects the process-owned ledger into the decoder. The decoder
|
||||
// charges the raw partial-frame bytes it retains between chunks, plus the peak
|
||||
// replacement buffer it allocates on concat. It never retains an uncharged
|
||||
// buffer, and it fails closed when a reservation would pass the ceiling.
|
||||
function makeLedger(ceilingBytes = 1024): DuplexAggregateByteLedger {
|
||||
return new DuplexAggregateByteLedger({ ceilingBytes });
|
||||
}
|
||||
|
||||
it("charges the retained partial frame, then releases it when the frame completes", () => {
|
||||
const ledger = makeLedger();
|
||||
const decoder = new DuplexFrameDecoder({ aggregateByteLedger: ledger });
|
||||
const line = encodeDuplexFrame({ version: DUPLEX_FRAME_VERSION, type: "heartbeat" });
|
||||
const bytes = Buffer.from(line, "utf8");
|
||||
|
||||
// A partial frame with no newline stays retained and stays charged.
|
||||
const first = decoder.push(bytes.subarray(0, 3));
|
||||
expect(first).toHaveLength(0);
|
||||
expect(ledger.bytesInUse).toBe(3);
|
||||
expect(ledger.liveTokenCount).toBe(1);
|
||||
|
||||
// The completing chunk delivers the frame and releases the retained bytes.
|
||||
const second = decoder.push(bytes.subarray(3));
|
||||
expect(second).toHaveLength(1);
|
||||
expect(second[0].ok).toBe(true);
|
||||
expect(ledger.bytesInUse).toBe(0);
|
||||
expect(ledger.liveTokenCount).toBe(0);
|
||||
});
|
||||
|
||||
it("covers the peak concat buffer and settles on the remaining bytes", () => {
|
||||
const ledger = makeLedger();
|
||||
const decoder = new DuplexFrameDecoder({ aggregateByteLedger: ledger });
|
||||
const first = encodeDuplexFrame({ version: DUPLEX_FRAME_VERSION, type: "heartbeat" });
|
||||
const partial = '{"version":1,"type":"hea';
|
||||
// One full frame plus a partial second frame arrive in one chunk. The full
|
||||
// frame decodes and releases; the partial tail stays charged at its exact size.
|
||||
const results = decoder.push(Buffer.from(first + partial, "utf8"));
|
||||
expect(results).toHaveLength(1);
|
||||
expect(results[0].ok).toBe(true);
|
||||
expect(ledger.bytesInUse).toBe(Buffer.byteLength(partial, "utf8"));
|
||||
expect(ledger.liveTokenCount).toBe(1);
|
||||
});
|
||||
|
||||
it("fails closed when a chunk would pass the ceiling and retains nothing", () => {
|
||||
const ledger = makeLedger(16);
|
||||
const decoder = new DuplexFrameDecoder({ aggregateByteLedger: ledger });
|
||||
const results = decoder.push(Buffer.from("z".repeat(64), "utf8"));
|
||||
expect(results).toHaveLength(1);
|
||||
expect(results[0].ok).toBe(false);
|
||||
if (!results[0].ok) {
|
||||
expect(results[0].error.code).toBe("aggregate_bytes_exceeded");
|
||||
}
|
||||
// The decoder retained nothing, so the ledger stays at zero.
|
||||
expect(ledger.bytesInUse).toBe(0);
|
||||
expect(ledger.liveTokenCount).toBe(0);
|
||||
});
|
||||
|
||||
it("resynchronizes on the next newline after an aggregate rejection", () => {
|
||||
const ledger = makeLedger(64);
|
||||
const decoder = new DuplexFrameDecoder({ aggregateByteLedger: ledger });
|
||||
// A first chunk over the ceiling fails closed.
|
||||
const rejected = decoder.push(Buffer.from("z".repeat(128), "utf8"));
|
||||
expect(rejected[0].ok).toBe(false);
|
||||
|
||||
// The next chunk starts with a newline, so the decoder drops the stale tail
|
||||
// and decodes the good frame that follows.
|
||||
const good = encodeDuplexFrame({ version: DUPLEX_FRAME_VERSION, type: "heartbeat" });
|
||||
const results = decoder.push(Buffer.from(`\n${good}`, "utf8"));
|
||||
expect(results).toHaveLength(1);
|
||||
expect(results[0].ok).toBe(true);
|
||||
expect(ledger.bytesInUse).toBe(0);
|
||||
expect(ledger.liveTokenCount).toBe(0);
|
||||
});
|
||||
|
||||
it("releases the retained token when an oversized frame is discarded", () => {
|
||||
const ledger = makeLedger();
|
||||
const decoder = new DuplexFrameDecoder({ maxFrameBytes: 16, aggregateByteLedger: ledger });
|
||||
// An oversized frame with no newline is discarded and its bytes are released.
|
||||
const results = decoder.push(Buffer.from("z".repeat(64), "utf8"));
|
||||
expect(results).toHaveLength(1);
|
||||
expect(results[0].ok).toBe(false);
|
||||
if (!results[0].ok) expect(results[0].error.code).toBe("frame_too_large");
|
||||
expect(ledger.bytesInUse).toBe(0);
|
||||
expect(ledger.liveTokenCount).toBe(0);
|
||||
});
|
||||
|
||||
it("releases the retained partial frame on dispose", () => {
|
||||
const ledger = makeLedger();
|
||||
const decoder = new DuplexFrameDecoder({ aggregateByteLedger: ledger });
|
||||
decoder.push(Buffer.from('{"version":1', "utf8"));
|
||||
expect(ledger.bytesInUse).toBeGreaterThan(0);
|
||||
decoder.dispose();
|
||||
expect(ledger.bytesInUse).toBe(0);
|
||||
expect(ledger.liveTokenCount).toBe(0);
|
||||
// A second dispose is a no-op and records no accounting defect.
|
||||
decoder.dispose();
|
||||
expect(ledger.bytesInUse).toBe(0);
|
||||
});
|
||||
|
||||
it("charges nothing when the caller injects no ledger", () => {
|
||||
const decoder = new DuplexFrameDecoder();
|
||||
const results = decoder.push(Buffer.from('{"version":1', "utf8"));
|
||||
expect(results).toHaveLength(0);
|
||||
// No ledger means no charge; the decoder still buffers and decodes as before.
|
||||
const line = encodeDuplexFrame({ version: DUPLEX_FRAME_VERSION, type: "heartbeat" });
|
||||
const done = decoder.push(Buffer.from(`}\n${line}`, "utf8"));
|
||||
// The stitched first frame is malformed JSON; the second is a valid heartbeat.
|
||||
expect(done.some((result) => result.ok)).toBe(true);
|
||||
});
|
||||
});
|
||||
|
||||
describe("size-checked encode", () => {
|
||||
it("runs every shared encode vector and matches the expected result", () => {
|
||||
// Every codec copy runs the same encode vectors. This copy proves it enforces
|
||||
|
|
@ -458,12 +175,9 @@ describe("size-checked encode", () => {
|
|||
it("rejects an over-bound frame with a typed outcome and never throws", () => {
|
||||
const frame: DuplexFrame = {
|
||||
version: DUPLEX_FRAME_VERSION,
|
||||
type: "response",
|
||||
id: "big",
|
||||
status: 200,
|
||||
headers: { "x-pad": "x".repeat(2_000) },
|
||||
bodyByteCount: 0,
|
||||
outcome: "completed",
|
||||
type: "error",
|
||||
code: "read_timeout",
|
||||
message: "x".repeat(2_000),
|
||||
};
|
||||
expect(() => encodeDuplexFrameChecked(frame, 1_000)).not.toThrow();
|
||||
const result = encodeDuplexFrameChecked(frame, 1_000);
|
||||
|
|
@ -475,17 +189,14 @@ describe("size-checked encode", () => {
|
|||
// Build a frame whose encoded JSON is exactly N bytes. The guard measures the
|
||||
// JSON without the newline, so N bytes encodes and N+1 bytes fails. This is the
|
||||
// same boundary the decoder applies, so encode and decode agree exactly.
|
||||
// Pad a header value to grow the encoded JSON by an exact byte count. The
|
||||
// "x-pad" value adds one byte per character, so the frame lands on an exact
|
||||
// size the guard measures without the trailing newline.
|
||||
// Pad the message to grow the encoded JSON by an exact byte count. Each
|
||||
// padding character adds one byte, so the frame lands on an exact size the
|
||||
// guard measures without the trailing newline.
|
||||
const pad = (n: number): DuplexFrame => ({
|
||||
version: DUPLEX_FRAME_VERSION,
|
||||
type: "response",
|
||||
id: "boundary",
|
||||
status: 200,
|
||||
headers: { "x-pad": "x".repeat(n) },
|
||||
bodyByteCount: 0,
|
||||
outcome: "completed",
|
||||
type: "error",
|
||||
code: "boundary",
|
||||
message: "x".repeat(n),
|
||||
});
|
||||
const baseBytes = Buffer.byteLength(JSON.stringify(pad(0)), "utf8");
|
||||
const limit = baseBytes + 10;
|
||||
|
|
|
|||
|
|
@ -6,41 +6,19 @@
|
|||
* fixture file (`duplex-frame-vectors.json`) proves the two copies stay wire
|
||||
* compatible: both copies decode the same bytes to the same frames.
|
||||
*
|
||||
* The codec has two sides:
|
||||
* - encode: turn a frame object into one line of JSON with a trailing newline.
|
||||
* - decode: turn a byte stream into frames. The streaming decoder keeps partial
|
||||
* bytes between chunks, so a frame split across chunks and a multi-byte UTF-8
|
||||
* sequence split across chunks both decode correctly.
|
||||
* The decoder never throws on the read path. A malformed or version-mismatch
|
||||
* frame becomes a protocol-error result, not an exception. This keeps one bad
|
||||
* frame from crashing the read loop.
|
||||
*
|
||||
* The decoder never throws on the read path. A malformed, oversized, or
|
||||
* version-mismatch frame becomes a protocol-error result, not an exception. This
|
||||
* keeps one bad frame from crashing the read loop.
|
||||
*
|
||||
* HTTP/2 is the preferred transport. `queue_v1` is the soft-deprecated fallback.
|
||||
* The `http2_v1` host readiness gate imports {@link decodeDuplexLine} from this
|
||||
* file to read the one READY line every gateway sends, so this file's READY
|
||||
* frame path stays live for both transports. `duplex-body-spool.ts` still
|
||||
* imports the request, response, and body-chunk frame types from this file, so
|
||||
* this file keeps every frame type here.
|
||||
* file to read the one READY line every gateway sends. That is the only frame
|
||||
* this module's decode side still reads in production; the gateway itself
|
||||
* never decodes, it only writes one READY line with {@link encodeDuplexFrame}.
|
||||
*/
|
||||
|
||||
import {
|
||||
DUPLEX_CHANNEL_AGGREGATE_BYTES_EXCEEDED,
|
||||
type DuplexAggregateByteLedger,
|
||||
type ReservationToken,
|
||||
} from "./duplex-aggregate-byte-ledger.js";
|
||||
|
||||
/** The wire version this codec reads and writes. */
|
||||
export const DUPLEX_FRAME_VERSION = 2;
|
||||
|
||||
/**
|
||||
* The fixed raw byte size of one body slice. The sender splits a body into
|
||||
* `body_chunk` frames of this raw size, except the final chunk, which holds the
|
||||
* remaining bytes. The base64 text of one full slice is about 349 KiB, so each
|
||||
* encoded `body_chunk` frame stays well under {@link DEFAULT_MAX_DUPLEX_FRAME_BYTES}.
|
||||
*/
|
||||
export const DUPLEX_BODY_CHUNK_RAW_BYTES = 256 * 1024;
|
||||
|
||||
/**
|
||||
* The default maximum size of one frame, in bytes. The decoder rejects a longer
|
||||
* frame with a `frame_too_large` protocol error. The value matches the per-chunk
|
||||
|
|
@ -48,88 +26,6 @@ export const DUPLEX_BODY_CHUNK_RAW_BYTES = 256 * 1024;
|
|||
*/
|
||||
export const DEFAULT_MAX_DUPLEX_FRAME_BYTES = 1_000_000;
|
||||
|
||||
/**
|
||||
* The maximum size of the frame `id` field, in bytes. The decoder rejects a
|
||||
* request or a response frame that carries a longer id with an `id_too_large`
|
||||
* protocol error. The read path returns the error; it never throws.
|
||||
*
|
||||
* The generated gateway builds each request id with `randomUUID()`, so a real id
|
||||
* is 36 ASCII bytes. This bound of 256 bytes gives large headroom for that id and
|
||||
* for any short future id scheme. The bound also caps the bytes the host broker
|
||||
* retains per distinct request. The broker keeps one id per distinct dispatched
|
||||
* request for the no-replay guarantee, so the id bound sets the per-id ceiling of
|
||||
* that retained memory.
|
||||
*/
|
||||
export const DEFAULT_MAX_DUPLEX_REQUEST_ID_BYTES = 256;
|
||||
|
||||
const NEWLINE_BYTE = 0x0a;
|
||||
const EMPTY = Buffer.alloc(0);
|
||||
|
||||
/** The frame type strings. One value goes in the `type` field of every frame. */
|
||||
export const DUPLEX_FRAME_TYPES = {
|
||||
request: "request",
|
||||
response: "response",
|
||||
body_chunk: "body_chunk",
|
||||
ready: "ready",
|
||||
heartbeat: "heartbeat",
|
||||
close: "close",
|
||||
error: "error",
|
||||
} as const;
|
||||
|
||||
/** The outcome of a response frame. A loss response carries a non-completed outcome. */
|
||||
export type DuplexResponseOutcome = "completed" | "indeterminate" | "unavailable";
|
||||
|
||||
/**
|
||||
* A request envelope frame. The gateway forwards it to the host API path. The
|
||||
* envelope carries `bodyByteCount`, the raw byte length of the request body. The
|
||||
* body itself rides one or more `body_chunk` frames that share the same `id`. A
|
||||
* `bodyByteCount` of 0 means an empty body and no `body_chunk` frame follows.
|
||||
*/
|
||||
export interface DuplexRequestFrame {
|
||||
version: number;
|
||||
type: "request";
|
||||
id: string;
|
||||
method: string;
|
||||
path: string;
|
||||
query: string;
|
||||
headers: Record<string, string>;
|
||||
bodyByteCount: number;
|
||||
}
|
||||
|
||||
/**
|
||||
* A response envelope frame. The host returns it for one request id. The
|
||||
* envelope carries `bodyByteCount`, the raw byte length of the response body.
|
||||
* The body itself rides one or more `body_chunk` frames that share the same
|
||||
* `id`. A `bodyByteCount` of 0 means an empty body and no `body_chunk` frame
|
||||
* follows.
|
||||
*/
|
||||
export interface DuplexResponseFrame {
|
||||
version: number;
|
||||
type: "response";
|
||||
id: string;
|
||||
status: number;
|
||||
headers: Record<string, string>;
|
||||
bodyByteCount: number;
|
||||
outcome: DuplexResponseOutcome;
|
||||
}
|
||||
|
||||
/**
|
||||
* A body-chunk frame. One frame carries one raw body slice as base64 text in
|
||||
* `data`. The frames for one body share the `id` of the envelope. `seq` starts
|
||||
* at 0 and increases by one for each next chunk, with no gap and no repeat. Each
|
||||
* raw slice is {@link DUPLEX_BODY_CHUNK_RAW_BYTES} bytes, except the final slice,
|
||||
* which holds the remaining bytes. The receiver reassembles the slices in `seq`
|
||||
* order and stops when the total raw byte count equals the envelope
|
||||
* `bodyByteCount`.
|
||||
*/
|
||||
export interface DuplexBodyChunkFrame {
|
||||
version: number;
|
||||
type: "body_chunk";
|
||||
id: string;
|
||||
seq: number;
|
||||
data: string;
|
||||
}
|
||||
|
||||
/**
|
||||
* The READY control frame. The gateway sends it one time after it binds the
|
||||
* host-assigned listener port. READY is a liveness signal, not an address
|
||||
|
|
@ -170,14 +66,7 @@ export interface DuplexErrorFrame {
|
|||
}
|
||||
|
||||
/** Any frame the codec reads or writes. */
|
||||
export type DuplexFrame =
|
||||
| DuplexRequestFrame
|
||||
| DuplexResponseFrame
|
||||
| DuplexBodyChunkFrame
|
||||
| DuplexReadyFrame
|
||||
| DuplexHeartbeatFrame
|
||||
| DuplexCloseFrame
|
||||
| DuplexErrorFrame;
|
||||
export type DuplexFrame = DuplexReadyFrame | DuplexHeartbeatFrame | DuplexCloseFrame | DuplexErrorFrame;
|
||||
|
||||
/** The reason the decoder rejected one line. */
|
||||
export type DuplexProtocolErrorCode =
|
||||
|
|
@ -208,12 +97,6 @@ export type DuplexEncodeResult =
|
|||
| { ok: true; line: string }
|
||||
| { ok: false; error: { code: "frame_too_large"; message: string } };
|
||||
|
||||
const RESPONSE_OUTCOMES: ReadonlySet<string> = new Set<DuplexResponseOutcome>([
|
||||
"completed",
|
||||
"indeterminate",
|
||||
"unavailable",
|
||||
]);
|
||||
|
||||
function ok(frame: DuplexFrame): DuplexDecodeResult {
|
||||
return { ok: true, frame };
|
||||
}
|
||||
|
|
@ -226,24 +109,6 @@ function isPlainObject(value: unknown): value is Record<string, unknown> {
|
|||
return typeof value === "object" && value !== null && !Array.isArray(value);
|
||||
}
|
||||
|
||||
function isStringRecord(value: unknown): value is Record<string, string> {
|
||||
if (!isPlainObject(value)) return false;
|
||||
for (const entry of Object.values(value)) {
|
||||
if (typeof entry !== "string") return false;
|
||||
}
|
||||
return true;
|
||||
}
|
||||
|
||||
/** Return true when the id byte size is within {@link DEFAULT_MAX_DUPLEX_REQUEST_ID_BYTES}. */
|
||||
function idWithinLimit(id: string): boolean {
|
||||
return Buffer.byteLength(id, "utf8") <= DEFAULT_MAX_DUPLEX_REQUEST_ID_BYTES;
|
||||
}
|
||||
|
||||
/** Return true when the value is a safe integer that is zero or positive. */
|
||||
function isSafeNonNegativeInteger(value: unknown): value is number {
|
||||
return typeof value === "number" && Number.isSafeInteger(value) && value >= 0;
|
||||
}
|
||||
|
||||
/**
|
||||
* Encode one frame to a single line of JSON with a trailing newline. `JSON.stringify`
|
||||
* escapes any newline inside a string value, so the returned line holds no
|
||||
|
|
@ -307,12 +172,6 @@ export function decodeDuplexLine(line: string | Buffer): DuplexDecodeResult {
|
|||
|
||||
function validateFrame(frame: Record<string, unknown>): DuplexDecodeResult {
|
||||
switch (frame.type) {
|
||||
case "request":
|
||||
return validateRequest(frame);
|
||||
case "response":
|
||||
return validateResponse(frame);
|
||||
case "body_chunk":
|
||||
return validateBodyChunk(frame);
|
||||
case "ready":
|
||||
return validateReady(frame);
|
||||
case "heartbeat":
|
||||
|
|
@ -326,68 +185,6 @@ function validateFrame(frame: Record<string, unknown>): DuplexDecodeResult {
|
|||
}
|
||||
}
|
||||
|
||||
function validateRequest(frame: Record<string, unknown>): DuplexDecodeResult {
|
||||
if (
|
||||
typeof frame.id !== "string" ||
|
||||
typeof frame.method !== "string" ||
|
||||
typeof frame.path !== "string" ||
|
||||
typeof frame.query !== "string" ||
|
||||
!isSafeNonNegativeInteger(frame.bodyByteCount) ||
|
||||
!isStringRecord(frame.headers)
|
||||
) {
|
||||
return fail("malformed_frame", "request frame has a missing or wrong-typed field");
|
||||
}
|
||||
// Bound the id byte size at the protocol level. The host broker retains one id
|
||||
// per distinct dispatched request, so an unbounded id is a memory-exhaustion
|
||||
// path. Reject an over-limit id as a protocol error; the read path never throws.
|
||||
if (!idWithinLimit(frame.id)) {
|
||||
return fail("id_too_large", "request frame id exceeds the maximum size");
|
||||
}
|
||||
return ok(frame as unknown as DuplexRequestFrame);
|
||||
}
|
||||
|
||||
function validateResponse(frame: Record<string, unknown>): DuplexDecodeResult {
|
||||
if (
|
||||
typeof frame.id !== "string" ||
|
||||
typeof frame.status !== "number" ||
|
||||
!isSafeNonNegativeInteger(frame.bodyByteCount) ||
|
||||
!isStringRecord(frame.headers) ||
|
||||
typeof frame.outcome !== "string" ||
|
||||
!RESPONSE_OUTCOMES.has(frame.outcome)
|
||||
) {
|
||||
return fail("malformed_frame", "response frame has a missing or wrong-typed field");
|
||||
}
|
||||
// Apply the same id byte bound to the response id. The host echoes the request
|
||||
// id on the response, so the same rule keeps both frame ids under one ceiling.
|
||||
if (!idWithinLimit(frame.id)) {
|
||||
return fail("id_too_large", "response frame id exceeds the maximum size");
|
||||
}
|
||||
return ok(frame as unknown as DuplexResponseFrame);
|
||||
}
|
||||
|
||||
/**
|
||||
* Validate a `body_chunk` frame at the structural level. The codec is stateless,
|
||||
* so it checks only the per-frame shape here:
|
||||
* - `id` is a string within {@link DEFAULT_MAX_DUPLEX_REQUEST_ID_BYTES}.
|
||||
* - `seq` is a safe integer that is zero or positive.
|
||||
* - `data` is a string.
|
||||
* The receive-side reassembler owns the stateful checks: the base64 canonical
|
||||
* form, the fixed slice size, the `seq` order, and the total against
|
||||
* `bodyByteCount`. The codec never decodes `data` here.
|
||||
*/
|
||||
function validateBodyChunk(frame: Record<string, unknown>): DuplexDecodeResult {
|
||||
if (typeof frame.id !== "string" || typeof frame.data !== "string") {
|
||||
return fail("malformed_frame", "body_chunk frame has a missing or wrong-typed field");
|
||||
}
|
||||
if (!isSafeNonNegativeInteger(frame.seq)) {
|
||||
return fail("malformed_frame", "body_chunk frame has a missing or wrong-typed seq");
|
||||
}
|
||||
if (!idWithinLimit(frame.id)) {
|
||||
return fail("id_too_large", "body_chunk frame id exceeds the maximum size");
|
||||
}
|
||||
return ok(frame as unknown as DuplexBodyChunkFrame);
|
||||
}
|
||||
|
||||
function validateReady(frame: Record<string, unknown>): DuplexDecodeResult {
|
||||
// READY carries a liveness nonce, not an address. The schema is strict: a valid
|
||||
// READY frame holds exactly `version`, `type`, and `nonce`. The decoder rejects
|
||||
|
|
@ -421,164 +218,3 @@ function validateError(frame: Record<string, unknown>): DuplexDecodeResult {
|
|||
}
|
||||
return ok(frame as unknown as DuplexErrorFrame);
|
||||
}
|
||||
|
||||
/** Options for a {@link DuplexFrameDecoder}. */
|
||||
export interface DuplexFrameDecoderOptions {
|
||||
/** The maximum size of one frame, in bytes. Defaults to {@link DEFAULT_MAX_DUPLEX_FRAME_BYTES}. */
|
||||
maxFrameBytes?: number;
|
||||
/**
|
||||
* The process-owned aggregate byte ledger. When present, the decoder reserves
|
||||
* the exact bytes of the raw partial frame it retains between chunks, and the
|
||||
* peak replacement buffer it allocates on concat, against this ledger under the
|
||||
* `decoder_buffer` owner. The host injects the same ledger object it injects at
|
||||
* every other retention site, so one gauge bounds the aggregate retained bytes.
|
||||
* When absent the decoder charges nothing and behaves as before.
|
||||
*
|
||||
* The generated sandbox decoder runs in a separate operating-system process. It
|
||||
* cannot share this host ledger. It receives its own separate cap instead.
|
||||
*/
|
||||
aggregateByteLedger?: DuplexAggregateByteLedger;
|
||||
}
|
||||
|
||||
/**
|
||||
* A streaming decoder for a byte stream of newline-delimited JSON frames.
|
||||
*
|
||||
* Call `push` with each chunk. The decoder keeps the bytes of an incomplete
|
||||
* frame between calls, so a frame that spans two chunks decodes on the chunk
|
||||
* that completes it. The decoder buffers raw bytes and decodes UTF-8 only on a
|
||||
* complete line, so a multi-byte sequence split across chunks stays valid. The
|
||||
* newline byte `0x0A` never appears inside a multi-byte UTF-8 sequence, so a
|
||||
* split on that byte is always safe.
|
||||
*
|
||||
* The decoder enforces the maximum frame size. It rejects an oversized frame
|
||||
* with a `frame_too_large` protocol error, then discards bytes up to the next
|
||||
* newline to resynchronize. It never throws on the read path.
|
||||
*/
|
||||
export class DuplexFrameDecoder {
|
||||
private buffer: Buffer = EMPTY;
|
||||
private discarding = false;
|
||||
private readonly maxFrameBytes: number;
|
||||
private readonly ledger: DuplexAggregateByteLedger | null;
|
||||
/**
|
||||
* The token for the bytes currently retained in `this.buffer`. Its byte amount
|
||||
* equals `this.buffer.length` at the end of every `push`. It is `null` when the
|
||||
* buffer is empty or the decoder has no ledger.
|
||||
*/
|
||||
private retainedToken: ReservationToken | null = null;
|
||||
|
||||
constructor(options: DuplexFrameDecoderOptions = {}) {
|
||||
this.maxFrameBytes = options.maxFrameBytes ?? DEFAULT_MAX_DUPLEX_FRAME_BYTES;
|
||||
this.ledger = options.aggregateByteLedger ?? null;
|
||||
}
|
||||
|
||||
/** Feed one chunk. Return the frames and protocol errors that complete on it. */
|
||||
push(chunk: Buffer | Uint8Array | string): DuplexDecodeResult[] {
|
||||
const incoming =
|
||||
typeof chunk === "string" ? Buffer.from(chunk, "utf8") : Buffer.from(chunk);
|
||||
|
||||
// Reserve the incoming bytes before the concat allocates the replacement
|
||||
// buffer. The retained token already covers the current buffer, so the two
|
||||
// tokens together cover the peak `old + incoming` allocation. A rejected
|
||||
// reservation fails closed: the decoder drops the incoming chunk, releases
|
||||
// the retained buffer, resynchronizes at the next newline, and reports the
|
||||
// fixed aggregate rejection marker. It retains nothing uncharged.
|
||||
let incomingToken: ReservationToken | null = null;
|
||||
if (this.ledger && incoming.length > 0) {
|
||||
incomingToken = this.ledger.reserve("decoder_buffer", incoming.length);
|
||||
if (incomingToken === null) {
|
||||
this.releaseRetained();
|
||||
this.buffer = EMPTY;
|
||||
this.discarding = true;
|
||||
return [fail("aggregate_bytes_exceeded", DUPLEX_CHANNEL_AGGREGATE_BYTES_EXCEEDED)];
|
||||
}
|
||||
}
|
||||
|
||||
this.buffer =
|
||||
this.buffer.length === 0 ? incoming : Buffer.concat([this.buffer, incoming]);
|
||||
|
||||
const results: DuplexDecodeResult[] = [];
|
||||
for (;;) {
|
||||
if (this.discarding) {
|
||||
// Drop the tail of an oversized frame until the next newline.
|
||||
const newlineIndex = this.buffer.indexOf(NEWLINE_BYTE);
|
||||
if (newlineIndex === -1) {
|
||||
this.buffer = EMPTY;
|
||||
break;
|
||||
}
|
||||
this.buffer = this.buffer.subarray(newlineIndex + 1);
|
||||
this.discarding = false;
|
||||
continue;
|
||||
}
|
||||
|
||||
const newlineIndex = this.buffer.indexOf(NEWLINE_BYTE);
|
||||
if (newlineIndex === -1) {
|
||||
// No complete line yet. Reject an incomplete frame that already passed
|
||||
// the size bound, then resynchronize at the next newline.
|
||||
if (this.buffer.length > this.maxFrameBytes) {
|
||||
results.push(fail("frame_too_large", "frame exceeds the maximum size"));
|
||||
this.discarding = true;
|
||||
this.buffer = EMPTY;
|
||||
}
|
||||
break;
|
||||
}
|
||||
|
||||
const line = this.buffer.subarray(0, newlineIndex);
|
||||
this.buffer = this.buffer.subarray(newlineIndex + 1);
|
||||
if (line.length === 0) continue; // Skip a blank line.
|
||||
if (line.length > this.maxFrameBytes) {
|
||||
results.push(fail("frame_too_large", "frame exceeds the maximum size"));
|
||||
continue;
|
||||
}
|
||||
results.push(decodeDuplexLine(line));
|
||||
}
|
||||
|
||||
// Reconcile the ledger to the bytes still retained in `this.buffer`. Release
|
||||
// the old retained token and the incoming token, then reserve one token for
|
||||
// the remainder. The remainder is a subset of the just-released
|
||||
// `old + incoming` bytes, so this reserve always fits.
|
||||
this.reconcileRetained(incomingToken);
|
||||
return results;
|
||||
}
|
||||
|
||||
/**
|
||||
* Release the retained-buffer token and drop the buffer. The host calls this at
|
||||
* channel teardown, so the decoder never leaks a `decoder_buffer` token after
|
||||
* the channel ends. A second call is a no-op, because the field is already
|
||||
* `null`.
|
||||
*/
|
||||
dispose(): void {
|
||||
this.releaseRetained();
|
||||
this.buffer = EMPTY;
|
||||
this.discarding = false;
|
||||
}
|
||||
|
||||
/** Release the retained-buffer token one time and clear the field. */
|
||||
private releaseRetained(): void {
|
||||
if (this.ledger && this.retainedToken) {
|
||||
this.ledger.release(this.retainedToken);
|
||||
}
|
||||
this.retainedToken = null;
|
||||
}
|
||||
|
||||
/**
|
||||
* Move the ledger charge to the bytes still in `this.buffer`. Release the old
|
||||
* retained token and the `incoming` token, then reserve one token for the
|
||||
* remaining bytes. A `null` reserve here is an accounting defect, because the
|
||||
* remainder never passes the just-released bytes; the decoder fails closed and
|
||||
* drops the buffer so it retains nothing uncharged.
|
||||
*/
|
||||
private reconcileRetained(incomingToken: ReservationToken | null): void {
|
||||
if (!this.ledger) return;
|
||||
this.releaseRetained();
|
||||
if (incomingToken) {
|
||||
this.ledger.release(incomingToken);
|
||||
}
|
||||
if (this.buffer.length > 0) {
|
||||
this.retainedToken = this.ledger.reserve("decoder_buffer", this.buffer.length);
|
||||
if (this.retainedToken === null) {
|
||||
this.buffer = EMPTY;
|
||||
this.discarding = true;
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -1,151 +1,8 @@
|
|||
{
|
||||
"frameVersion": 2,
|
||||
"defaultMaxFrameBytes": 1000000,
|
||||
"description": "Shared wire-compatibility vectors for the duplex frame codec. Every codec copy decodes the same bytes. bytes is a UTF-8 byte stream; splitByteOffsets cuts it into chunks; expected lists the ordered decode results (frame or protocol-error code).",
|
||||
"description": "Shared wire-compatibility vectors for the duplex frame codec. Every codec copy decodes the same bytes. bytes is a UTF-8 byte stream; expected lists the decode result (frame or protocol-error code).",
|
||||
"vectors": [
|
||||
{
|
||||
"name": "valid-request",
|
||||
"category": "valid",
|
||||
"bytes": "{\"version\":2,\"type\":\"request\",\"id\":\"req-1\",\"method\":\"GET\",\"path\":\"/api/issues/PAP-1\",\"query\":\"expand=comments\",\"headers\":{\"accept\":\"application/json\",\"content-type\":\"application/json\"},\"bodyByteCount\":0}\n",
|
||||
"roundTrip": true,
|
||||
"expected": [
|
||||
{
|
||||
"frame": {
|
||||
"version": 2,
|
||||
"type": "request",
|
||||
"id": "req-1",
|
||||
"method": "GET",
|
||||
"path": "/api/issues/PAP-1",
|
||||
"query": "expand=comments",
|
||||
"headers": {
|
||||
"accept": "application/json",
|
||||
"content-type": "application/json"
|
||||
},
|
||||
"bodyByteCount": 0
|
||||
}
|
||||
}
|
||||
]
|
||||
},
|
||||
{
|
||||
"name": "valid-request-with-body-count",
|
||||
"category": "valid",
|
||||
"bytes": "{\"version\":2,\"type\":\"request\",\"id\":\"req-body\",\"method\":\"GET\",\"path\":\"/api/issues/PAP-1\",\"query\":\"expand=comments\",\"headers\":{\"accept\":\"application/json\",\"content-type\":\"application/json\"},\"bodyByteCount\":1048576}\n",
|
||||
"roundTrip": true,
|
||||
"expected": [
|
||||
{
|
||||
"frame": {
|
||||
"version": 2,
|
||||
"type": "request",
|
||||
"id": "req-body",
|
||||
"method": "GET",
|
||||
"path": "/api/issues/PAP-1",
|
||||
"query": "expand=comments",
|
||||
"headers": {
|
||||
"accept": "application/json",
|
||||
"content-type": "application/json"
|
||||
},
|
||||
"bodyByteCount": 1048576
|
||||
}
|
||||
}
|
||||
]
|
||||
},
|
||||
{
|
||||
"name": "valid-response-completed",
|
||||
"category": "valid",
|
||||
"bytes": "{\"version\":2,\"type\":\"response\",\"id\":\"req-1\",\"status\":200,\"headers\":{\"content-type\":\"application/json\"},\"bodyByteCount\":11,\"outcome\":\"completed\"}\n",
|
||||
"roundTrip": true,
|
||||
"expected": [
|
||||
{
|
||||
"frame": {
|
||||
"version": 2,
|
||||
"type": "response",
|
||||
"id": "req-1",
|
||||
"status": 200,
|
||||
"headers": {
|
||||
"content-type": "application/json"
|
||||
},
|
||||
"bodyByteCount": 11,
|
||||
"outcome": "completed"
|
||||
}
|
||||
}
|
||||
]
|
||||
},
|
||||
{
|
||||
"name": "valid-response-indeterminate",
|
||||
"category": "valid",
|
||||
"bytes": "{\"version\":2,\"type\":\"response\",\"id\":\"req-2\",\"status\":409,\"headers\":{\"content-type\":\"application/json\"},\"bodyByteCount\":33,\"outcome\":\"indeterminate\"}\n",
|
||||
"roundTrip": true,
|
||||
"expected": [
|
||||
{
|
||||
"frame": {
|
||||
"version": 2,
|
||||
"type": "response",
|
||||
"id": "req-2",
|
||||
"status": 409,
|
||||
"headers": {
|
||||
"content-type": "application/json"
|
||||
},
|
||||
"bodyByteCount": 33,
|
||||
"outcome": "indeterminate"
|
||||
}
|
||||
}
|
||||
]
|
||||
},
|
||||
{
|
||||
"name": "valid-response-unavailable",
|
||||
"category": "valid",
|
||||
"bytes": "{\"version\":2,\"type\":\"response\",\"id\":\"req-3\",\"status\":503,\"headers\":{\"content-type\":\"application/json\"},\"bodyByteCount\":30,\"outcome\":\"unavailable\"}\n",
|
||||
"roundTrip": true,
|
||||
"expected": [
|
||||
{
|
||||
"frame": {
|
||||
"version": 2,
|
||||
"type": "response",
|
||||
"id": "req-3",
|
||||
"status": 503,
|
||||
"headers": {
|
||||
"content-type": "application/json"
|
||||
},
|
||||
"bodyByteCount": 30,
|
||||
"outcome": "unavailable"
|
||||
}
|
||||
}
|
||||
]
|
||||
},
|
||||
{
|
||||
"name": "valid-body-chunk",
|
||||
"category": "valid",
|
||||
"bytes": "{\"version\":2,\"type\":\"body_chunk\",\"id\":\"req-1\",\"seq\":0,\"data\":\"aGVsbG8gYm9keSBjaHVuaw==\"}\n",
|
||||
"roundTrip": true,
|
||||
"expected": [
|
||||
{
|
||||
"frame": {
|
||||
"version": 2,
|
||||
"type": "body_chunk",
|
||||
"id": "req-1",
|
||||
"seq": 0,
|
||||
"data": "aGVsbG8gYm9keSBjaHVuaw=="
|
||||
}
|
||||
}
|
||||
]
|
||||
},
|
||||
{
|
||||
"name": "valid-body-chunk-final",
|
||||
"category": "valid",
|
||||
"bytes": "{\"version\":2,\"type\":\"body_chunk\",\"id\":\"req-1\",\"seq\":7,\"data\":\"dGFpbA==\"}\n",
|
||||
"roundTrip": true,
|
||||
"expected": [
|
||||
{
|
||||
"frame": {
|
||||
"version": 2,
|
||||
"type": "body_chunk",
|
||||
"id": "req-1",
|
||||
"seq": 7,
|
||||
"data": "dGFpbA=="
|
||||
}
|
||||
}
|
||||
]
|
||||
},
|
||||
{
|
||||
"name": "valid-ready",
|
||||
"category": "valid",
|
||||
|
|
@ -275,223 +132,6 @@
|
|||
}
|
||||
]
|
||||
},
|
||||
{
|
||||
"name": "invalid-missing-id",
|
||||
"category": "invalid",
|
||||
"bytes": "{\"version\":2,\"type\":\"request\",\"method\":\"GET\",\"path\":\"/\",\"query\":\"\",\"headers\":{},\"bodyByteCount\":0}\n",
|
||||
"expected": [
|
||||
{
|
||||
"error": "malformed_frame"
|
||||
}
|
||||
]
|
||||
},
|
||||
{
|
||||
"name": "invalid-request-body-count-string",
|
||||
"category": "invalid",
|
||||
"bytes": "{\"version\":2,\"type\":\"request\",\"id\":\"req-1\",\"method\":\"GET\",\"path\":\"/\",\"query\":\"\",\"headers\":{},\"bodyByteCount\":\"10\"}\n",
|
||||
"expected": [
|
||||
{
|
||||
"error": "malformed_frame"
|
||||
}
|
||||
]
|
||||
},
|
||||
{
|
||||
"name": "invalid-request-body-count-negative",
|
||||
"category": "invalid",
|
||||
"bytes": "{\"version\":2,\"type\":\"request\",\"id\":\"req-1\",\"method\":\"GET\",\"path\":\"/\",\"query\":\"\",\"headers\":{},\"bodyByteCount\":-1}\n",
|
||||
"expected": [
|
||||
{
|
||||
"error": "malformed_frame"
|
||||
}
|
||||
]
|
||||
},
|
||||
{
|
||||
"name": "invalid-request-body-count-fractional",
|
||||
"category": "invalid",
|
||||
"bytes": "{\"version\":2,\"type\":\"request\",\"id\":\"req-1\",\"method\":\"GET\",\"path\":\"/\",\"query\":\"\",\"headers\":{},\"bodyByteCount\":1.5}\n",
|
||||
"expected": [
|
||||
{
|
||||
"error": "malformed_frame"
|
||||
}
|
||||
]
|
||||
},
|
||||
{
|
||||
"name": "invalid-bad-headers",
|
||||
"category": "invalid",
|
||||
"bytes": "{\"version\":2,\"type\":\"response\",\"id\":\"req-1\",\"status\":200,\"headers\":{\"accept\":7},\"bodyByteCount\":10,\"outcome\":\"completed\"}\n",
|
||||
"expected": [
|
||||
{
|
||||
"error": "malformed_frame"
|
||||
}
|
||||
]
|
||||
},
|
||||
{
|
||||
"name": "invalid-body-chunk-negative-seq",
|
||||
"category": "invalid",
|
||||
"bytes": "{\"version\":2,\"type\":\"body_chunk\",\"id\":\"req-1\",\"seq\":-1,\"data\":\"eA==\"}\n",
|
||||
"expected": [
|
||||
{
|
||||
"error": "malformed_frame"
|
||||
}
|
||||
]
|
||||
},
|
||||
{
|
||||
"name": "invalid-body-chunk-fractional-seq",
|
||||
"category": "invalid",
|
||||
"bytes": "{\"version\":2,\"type\":\"body_chunk\",\"id\":\"req-1\",\"seq\":2.5,\"data\":\"eA==\"}\n",
|
||||
"expected": [
|
||||
{
|
||||
"error": "malformed_frame"
|
||||
}
|
||||
]
|
||||
},
|
||||
{
|
||||
"name": "invalid-body-chunk-missing-data",
|
||||
"category": "invalid",
|
||||
"bytes": "{\"version\":2,\"type\":\"body_chunk\",\"id\":\"req-1\",\"seq\":0}\n",
|
||||
"expected": [
|
||||
{
|
||||
"error": "malformed_frame"
|
||||
}
|
||||
]
|
||||
},
|
||||
{
|
||||
"name": "invalid-body-chunk-nonstring-data",
|
||||
"category": "invalid",
|
||||
"bytes": "{\"version\":2,\"type\":\"body_chunk\",\"id\":\"req-1\",\"seq\":0,\"data\":42}\n",
|
||||
"expected": [
|
||||
{
|
||||
"error": "malformed_frame"
|
||||
}
|
||||
]
|
||||
},
|
||||
{
|
||||
"name": "partial-request-split",
|
||||
"category": "partial",
|
||||
"bytes": "{\"version\":2,\"type\":\"request\",\"id\":\"req-1\",\"method\":\"GET\",\"path\":\"/api/issues/PAP-1\",\"query\":\"expand=comments\",\"headers\":{\"accept\":\"application/json\",\"content-type\":\"application/json\"},\"bodyByteCount\":0}\n",
|
||||
"splitByteOffsets": [
|
||||
102
|
||||
],
|
||||
"expected": [
|
||||
{
|
||||
"frame": {
|
||||
"version": 2,
|
||||
"type": "request",
|
||||
"id": "req-1",
|
||||
"method": "GET",
|
||||
"path": "/api/issues/PAP-1",
|
||||
"query": "expand=comments",
|
||||
"headers": {
|
||||
"accept": "application/json",
|
||||
"content-type": "application/json"
|
||||
},
|
||||
"bodyByteCount": 0
|
||||
}
|
||||
}
|
||||
]
|
||||
},
|
||||
{
|
||||
"name": "partial-utf8-split",
|
||||
"category": "partial",
|
||||
"bytes": "{\"version\":2,\"type\":\"request\",\"id\":\"req-utf8\",\"method\":\"POST\",\"path\":\"/api/issues/PAP-1\",\"query\":\"expand=comments\",\"headers\":{\"content-type\":\"application/json\",\"x-note\":\"héllo 😀 wörld\"},\"bodyByteCount\":0}\n",
|
||||
"splitByteOffsets": [
|
||||
172
|
||||
],
|
||||
"expected": [
|
||||
{
|
||||
"frame": {
|
||||
"version": 2,
|
||||
"type": "request",
|
||||
"id": "req-utf8",
|
||||
"method": "POST",
|
||||
"path": "/api/issues/PAP-1",
|
||||
"query": "expand=comments",
|
||||
"headers": {
|
||||
"content-type": "application/json",
|
||||
"x-note": "héllo 😀 wörld"
|
||||
},
|
||||
"bodyByteCount": 0
|
||||
}
|
||||
}
|
||||
]
|
||||
},
|
||||
{
|
||||
"name": "partial-two-frames",
|
||||
"category": "partial",
|
||||
"bytes": "{\"version\":2,\"type\":\"request\",\"id\":\"req-1\",\"method\":\"GET\",\"path\":\"/api/issues/PAP-1\",\"query\":\"expand=comments\",\"headers\":{\"accept\":\"application/json\",\"content-type\":\"application/json\"},\"bodyByteCount\":0}\n{\"version\":2,\"type\":\"response\",\"id\":\"req-1\",\"status\":200,\"headers\":{\"content-type\":\"application/json\"},\"bodyByteCount\":11,\"outcome\":\"completed\"}\n",
|
||||
"splitByteOffsets": [
|
||||
7,
|
||||
174,
|
||||
344
|
||||
],
|
||||
"expected": [
|
||||
{
|
||||
"frame": {
|
||||
"version": 2,
|
||||
"type": "request",
|
||||
"id": "req-1",
|
||||
"method": "GET",
|
||||
"path": "/api/issues/PAP-1",
|
||||
"query": "expand=comments",
|
||||
"headers": {
|
||||
"accept": "application/json",
|
||||
"content-type": "application/json"
|
||||
},
|
||||
"bodyByteCount": 0
|
||||
}
|
||||
},
|
||||
{
|
||||
"frame": {
|
||||
"version": 2,
|
||||
"type": "response",
|
||||
"id": "req-1",
|
||||
"status": 200,
|
||||
"headers": {
|
||||
"content-type": "application/json"
|
||||
},
|
||||
"bodyByteCount": 11,
|
||||
"outcome": "completed"
|
||||
}
|
||||
}
|
||||
]
|
||||
},
|
||||
{
|
||||
"name": "partial-no-trailing-newline",
|
||||
"category": "partial",
|
||||
"bytes": "{\"version\":2,\"type\":\"heartbeat\"}",
|
||||
"expected": []
|
||||
},
|
||||
{
|
||||
"name": "oversized-complete-line",
|
||||
"category": "oversized",
|
||||
"maxFrameBytes": 64,
|
||||
"bytes": "{\"version\":2,\"type\":\"request\",\"id\":\"req-1\",\"method\":\"GET\",\"path\":\"/api/issues/PAP-1\",\"query\":\"q=xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx\",\"headers\":{\"accept\":\"application/json\",\"content-type\":\"application/json\"},\"bodyByteCount\":0}\n",
|
||||
"expected": [
|
||||
{
|
||||
"error": "frame_too_large"
|
||||
}
|
||||
]
|
||||
},
|
||||
{
|
||||
"name": "oversized-flood-then-resync",
|
||||
"category": "oversized",
|
||||
"maxFrameBytes": 64,
|
||||
"splitByteOffsets": [
|
||||
40
|
||||
],
|
||||
"bytes": "yyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyy\n{\"version\":2,\"type\":\"heartbeat\"}\n",
|
||||
"expected": [
|
||||
{
|
||||
"error": "frame_too_large"
|
||||
},
|
||||
{
|
||||
"frame": {
|
||||
"version": 2,
|
||||
"type": "heartbeat"
|
||||
}
|
||||
}
|
||||
]
|
||||
},
|
||||
{
|
||||
"name": "version-mismatch-lower",
|
||||
"category": "versionMismatch",
|
||||
|
|
@ -521,46 +161,6 @@
|
|||
"error": "version_mismatch"
|
||||
}
|
||||
]
|
||||
},
|
||||
{
|
||||
"name": "request-id-at-max-bytes",
|
||||
"category": "idBoundValid",
|
||||
"bytes": "{\"version\":2,\"type\":\"request\",\"id\":\"aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa\",\"method\":\"GET\",\"path\":\"/\",\"query\":\"\",\"headers\":{},\"bodyByteCount\":0}\n",
|
||||
"roundTrip": true,
|
||||
"expected": [
|
||||
{
|
||||
"frame": {
|
||||
"version": 2,
|
||||
"type": "request",
|
||||
"id": "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa",
|
||||
"method": "GET",
|
||||
"path": "/",
|
||||
"query": "",
|
||||
"headers": {},
|
||||
"bodyByteCount": 0
|
||||
}
|
||||
}
|
||||
]
|
||||
},
|
||||
{
|
||||
"name": "request-id-over-max-bytes",
|
||||
"category": "idBoundInvalid",
|
||||
"bytes": "{\"version\":2,\"type\":\"request\",\"id\":\"aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa\",\"method\":\"GET\",\"path\":\"/\",\"query\":\"\",\"headers\":{},\"bodyByteCount\":0}\n",
|
||||
"expected": [
|
||||
{
|
||||
"error": "id_too_large"
|
||||
}
|
||||
]
|
||||
},
|
||||
{
|
||||
"name": "body-chunk-id-over-max-bytes",
|
||||
"category": "idBoundInvalid",
|
||||
"bytes": "{\"version\":2,\"type\":\"body_chunk\",\"id\":\"aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa\",\"seq\":0,\"data\":\"eA==\"}\n",
|
||||
"expected": [
|
||||
{
|
||||
"error": "id_too_large"
|
||||
}
|
||||
]
|
||||
}
|
||||
],
|
||||
"encodeDescription": "Encode-bound vectors. Each codec copy runs the size-checked encode against the frame with the maxFrameBytes limit, then asserts the same result. An ok result must decode back to the same frame. This proves both copies enforce the same bound on encode.",
|
||||
|
|
@ -570,27 +170,9 @@
|
|||
"maxFrameBytes": 200,
|
||||
"frame": {
|
||||
"version": 2,
|
||||
"type": "request",
|
||||
"id": "req-enc-1",
|
||||
"method": "GET",
|
||||
"path": "/",
|
||||
"query": "",
|
||||
"headers": {},
|
||||
"bodyByteCount": 5
|
||||
},
|
||||
"expected": {
|
||||
"ok": true
|
||||
}
|
||||
},
|
||||
{
|
||||
"name": "encode-body-chunk-within-bound",
|
||||
"maxFrameBytes": 200,
|
||||
"frame": {
|
||||
"version": 2,
|
||||
"type": "body_chunk",
|
||||
"id": "req-enc-3",
|
||||
"seq": 0,
|
||||
"data": "Y2h1bmsgcGF5bG9hZA=="
|
||||
"type": "error",
|
||||
"code": "read_timeout",
|
||||
"message": "upstream did not respond"
|
||||
},
|
||||
"expected": {
|
||||
"ok": true
|
||||
|
|
@ -612,15 +194,9 @@
|
|||
"maxFrameBytes": 200,
|
||||
"frame": {
|
||||
"version": 2,
|
||||
"type": "response",
|
||||
"id": "req-enc-2",
|
||||
"status": 200,
|
||||
"headers": {
|
||||
"content-type": "application/json",
|
||||
"x-pad": "xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx"
|
||||
},
|
||||
"bodyByteCount": 0,
|
||||
"outcome": "completed"
|
||||
"type": "error",
|
||||
"code": "read_timeout",
|
||||
"message": "xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx"
|
||||
},
|
||||
"expected": {
|
||||
"ok": false,
|
||||
|
|
|
|||
|
|
@ -6,7 +6,6 @@ import { execFile, spawn } from "node:child_process";
|
|||
import { mkdir, mkdtemp, readdir, readFile, rm, writeFile } from "node:fs/promises";
|
||||
import os from "node:os";
|
||||
import path from "node:path";
|
||||
import { fileURLToPath } from "node:url";
|
||||
import { promisify } from "node:util";
|
||||
import { afterEach, describe, expect, it, vi } from "vitest";
|
||||
|
||||
|
|
@ -45,7 +44,6 @@ import {
|
|||
} from "./acpx-engine/startup-timing.js";
|
||||
import {
|
||||
DuplexAggregateByteLedger,
|
||||
DUPLEX_AGGREGATE_TOKEN_OWNERS,
|
||||
DUPLEX_CHANNEL_AGGREGATE_BYTES_EXCEEDED,
|
||||
} from "./duplex-aggregate-byte-ledger.js";
|
||||
import { createSandboxRunLogTailFactory, type SandboxRunLogTailFactory } from "./sandbox-run-log-stream.js";
|
||||
|
|
@ -54,11 +52,9 @@ import { shellQuote } from "./ssh.js";
|
|||
import type { CommandManagedDuplexChannel } from "./command-managed-runtime.js";
|
||||
import {
|
||||
DEFAULT_MAX_DUPLEX_FRAME_BYTES,
|
||||
DuplexFrameDecoder,
|
||||
DUPLEX_FRAME_VERSION,
|
||||
decodeDuplexLine,
|
||||
encodeDuplexFrame,
|
||||
type DuplexResponseFrame,
|
||||
} from "./duplex-frame-codec.js";
|
||||
import { DUPLEX_CHANNEL_LOST_ERROR_CODE } from "./bridge-transport-contract.js";
|
||||
import {
|
||||
|
|
@ -3064,7 +3060,6 @@ describe("sandbox adapter execution targets", () => {
|
|||
// what the broker wrote back through it.
|
||||
interface DuplexSelectionControl {
|
||||
openCount: number;
|
||||
written: DuplexResponseFrame[];
|
||||
writtenTypes: string[];
|
||||
stopCount: number;
|
||||
closeCount: number;
|
||||
|
|
@ -3095,7 +3090,6 @@ describe("sandbox adapter execution targets", () => {
|
|||
const base = createLocalSandboxRunner();
|
||||
const control: DuplexSelectionControl = {
|
||||
openCount: 0,
|
||||
written: [],
|
||||
writtenTypes: [],
|
||||
stopCount: 0,
|
||||
closeCount: 0,
|
||||
|
|
@ -3115,7 +3109,6 @@ describe("sandbox adapter execution targets", () => {
|
|||
const decoded = decodeDuplexLine(Buffer.from(data).toString("utf8").replace(/\n$/, ""));
|
||||
if (decoded.ok) {
|
||||
control.writtenTypes.push(decoded.frame.type);
|
||||
if (decoded.frame.type === "response") control.written.push(decoded.frame);
|
||||
}
|
||||
},
|
||||
onData(listener: (chunk: Uint8Array) => void): void {
|
||||
|
|
@ -5797,6 +5790,46 @@ describe("sandbox adapter execution targets", () => {
|
|||
expect(mode).toBe("http2_v1");
|
||||
}, 20000);
|
||||
|
||||
// Case 5: a version-2 line that once decoded as a `request` envelope frame
|
||||
// now decodes as `unknown_type`, because the codec no longer validates that
|
||||
// frame type. The gate treats it the same as any other non-READY line: skip
|
||||
// it and keep scanning. This holds for both a bare line (the frame starts at
|
||||
// offset 0, so the leading-prefix retry never runs) and a line where noise
|
||||
// comes before the frame on the same line (the retry runs, decodes the frame
|
||||
// part, and still does not find a READY type). Either way the valid READY
|
||||
// line that follows still authenticates.
|
||||
it("PTY replay: accepts READY after a bare and a prefixed former-frame line", async () => {
|
||||
const { mode } = await runReadinessReplay((ctx) => {
|
||||
const formerFrame =
|
||||
'{"version":2,"type":"request","id":"r-1","method":"GET","path":"/","query":"","headers":{},"bodyByteCount":0}';
|
||||
ctx.emitRaw(`${formerFrame}\n`);
|
||||
// No newline between the prompt prefix and the former-frame line: same line.
|
||||
ctx.emitRaw(`prompt$ ${formerFrame}\n`);
|
||||
ctx.emitRaw('{"version":2,"type":"ready","nonce":"' + ctx.nonce + '"}\n');
|
||||
ctx.connectHttp2();
|
||||
});
|
||||
expect(mode).toBe("http2_v1");
|
||||
}, 20000);
|
||||
|
||||
// Case 6: three lines the gate must reject without ending the handshake — a
|
||||
// wrong-version READY-shaped line, a READY line with a smuggled extra field,
|
||||
// and a same-line second-frame smuggling attempt (no newline between the two
|
||||
// JSON values, so the whole line fails to parse as one JSON value). Each one
|
||||
// fails the strict decode and the gate skips it, the same as any other
|
||||
// pre-READY noise; the valid READY line that follows still authenticates.
|
||||
it("PTY replay: accepts READY after a wrong-version, an extra-field, and a same-line smuggling attempt", async () => {
|
||||
const { mode } = await runReadinessReplay((ctx) => {
|
||||
ctx.emitRaw(`{"version":1,"type":"ready","nonce":"${ctx.nonce}"}\n`);
|
||||
ctx.emitRaw(`{"version":2,"type":"ready","nonce":"${ctx.nonce}","address":"http://127.0.0.1:1"}\n`);
|
||||
ctx.emitRaw(
|
||||
`{"version":2,"type":"ready","nonce":"${ctx.nonce}"}{"version":2,"type":"ready","nonce":"${ctx.nonce}"}\n`,
|
||||
);
|
||||
ctx.emitRaw(`{"version":2,"type":"ready","nonce":"${ctx.nonce}"}\n`);
|
||||
ctx.connectHttp2();
|
||||
});
|
||||
expect(mode).toBe("http2_v1");
|
||||
}, 20000);
|
||||
|
||||
it("test_the_session_starts_at_the_client_preface_after_the_ready_line", async () => {
|
||||
// The gate retains every byte after the accepted READY line, and the host
|
||||
// starts the HTTP/2 session at the client preface offset inside that
|
||||
|
|
@ -5922,223 +5955,37 @@ describe("sandbox adapter execution targets", () => {
|
|||
|
||||
});
|
||||
|
||||
// One decode result from the embedded codec. The shape mirrors the host codec:
|
||||
// a valid frame or a protocol error with a code.
|
||||
interface EmbeddedDecodeResult {
|
||||
ok: boolean;
|
||||
frame?: unknown;
|
||||
error?: { code: string; message: string };
|
||||
}
|
||||
|
||||
// The names the embedded codec source declares. A test wraps the source and
|
||||
// reads these names back.
|
||||
type EmbeddedEncodeResult =
|
||||
| { ok: true; line: string }
|
||||
| { ok: false; error: { code: string; message: string } };
|
||||
|
||||
// reads these names back. The generated gateway never decodes, so the embedded
|
||||
// copy declares only the frame version and the encode function.
|
||||
interface EmbeddedCodec {
|
||||
encodeDuplexFrame: (frame: unknown) => string;
|
||||
encodeDuplexFrameChecked: (frame: unknown, maxFrameBytes?: number) => EmbeddedEncodeResult;
|
||||
decodeDuplexLine: (line: string | Buffer) => EmbeddedDecodeResult;
|
||||
DuplexFrameDecoder: new (options?: { maxFrameBytes?: number; maxAggregateBytes?: number }) => {
|
||||
push: (chunk: Buffer) => EmbeddedDecodeResult[];
|
||||
scope: string;
|
||||
bytesInUse: number;
|
||||
aggregateRejections: number;
|
||||
};
|
||||
DUPLEX_FRAME_VERSION: number;
|
||||
DEFAULT_MAX_DUPLEX_FRAME_BYTES: number;
|
||||
DUPLEX_DECODER_SCOPE: string;
|
||||
DEFAULT_MAX_DUPLEX_DECODER_BYTES: number;
|
||||
}
|
||||
|
||||
type ExpectedVectorResult = { frame: unknown } | { error: string };
|
||||
|
||||
interface DuplexFrameVector {
|
||||
name: string;
|
||||
category: string;
|
||||
bytes: string;
|
||||
splitByteOffsets?: number[];
|
||||
maxFrameBytes?: number;
|
||||
roundTrip?: boolean;
|
||||
expected: ExpectedVectorResult[];
|
||||
}
|
||||
|
||||
interface DuplexEncodeVector {
|
||||
name: string;
|
||||
maxFrameBytes: number;
|
||||
frame: unknown;
|
||||
expected: { ok: true } | { ok: false; error: string };
|
||||
}
|
||||
|
||||
interface DuplexFrameFixture {
|
||||
frameVersion: number;
|
||||
defaultMaxFrameBytes: number;
|
||||
vectors: DuplexFrameVector[];
|
||||
encodeVectors: DuplexEncodeVector[];
|
||||
}
|
||||
|
||||
// This describe block covers the zero-dependency codec every generated
|
||||
// gateway embeds (`DUPLEX_GATEWAY_CODEC_SOURCE`), and the sandbox-process
|
||||
// decoder byte cap. The `http2_v1` gateway embeds the same codec source to
|
||||
// send its one READY line, so this coverage stays live for the active
|
||||
// transport.
|
||||
// gateway embeds (`DUPLEX_GATEWAY_CODEC_SOURCE`). The `http2_v1` gateway
|
||||
// embeds this same codec source to send its one READY line, so this coverage
|
||||
// stays live for the active transport.
|
||||
describe("embedded sandbox gateway codec", () => {
|
||||
it("embedded gateway codec passes every vector in the shared fixture", async () => {
|
||||
it("encodes the READY frame to the exact byte-for-byte wire format", () => {
|
||||
// `JSON.stringify` writes object keys in insertion order, so this pins the
|
||||
// exact key order the gateway writes: version, then type, then nonce. A
|
||||
// reordered or reformatted call site would change the bytes on the wire
|
||||
// without failing a looser, parse-then-compare assertion.
|
||||
const codecFactory = new Function(
|
||||
`${getSandboxDuplexGatewayCodecSource()}\nreturn { encodeDuplexFrame, encodeDuplexFrameChecked, decodeDuplexLine, DuplexFrameDecoder, DUPLEX_FRAME_VERSION, DEFAULT_MAX_DUPLEX_FRAME_BYTES, DUPLEX_DECODER_SCOPE, DEFAULT_MAX_DUPLEX_DECODER_BYTES };`,
|
||||
`${getSandboxDuplexGatewayCodecSource()}\nreturn { encodeDuplexFrame, DUPLEX_FRAME_VERSION };`,
|
||||
) as unknown as () => EmbeddedCodec;
|
||||
const codec = codecFactory();
|
||||
|
||||
const fixturePath = fileURLToPath(new URL("./duplex-frame-vectors.json", import.meta.url));
|
||||
const fixture = JSON.parse(await readFile(fixturePath, "utf8")) as DuplexFrameFixture;
|
||||
expect(codec.DUPLEX_FRAME_VERSION).toBe(DUPLEX_FRAME_VERSION);
|
||||
const nonce = "fixed-test-nonce";
|
||||
const encoded = codec.encodeDuplexFrame({ version: codec.DUPLEX_FRAME_VERSION, type: "ready", nonce });
|
||||
expect(encoded).toBe(`{"version":2,"type":"ready","nonce":"${nonce}"}\n`);
|
||||
|
||||
expect(fixture.frameVersion).toBe(codec.DUPLEX_FRAME_VERSION);
|
||||
expect(fixture.defaultMaxFrameBytes).toBe(codec.DEFAULT_MAX_DUPLEX_FRAME_BYTES);
|
||||
expect(fixture.vectors.length).toBeGreaterThanOrEqual(22);
|
||||
|
||||
const failures: string[] = [];
|
||||
for (const vector of fixture.vectors) {
|
||||
const decoder = new codec.DuplexFrameDecoder(
|
||||
vector.maxFrameBytes ? { maxFrameBytes: vector.maxFrameBytes } : undefined,
|
||||
);
|
||||
const buffer = Buffer.from(vector.bytes, "utf8");
|
||||
const offsets = vector.splitByteOffsets;
|
||||
const bounds =
|
||||
offsets && offsets.length > 0 ? [0, ...offsets, buffer.length] : [0, buffer.length];
|
||||
const results: EmbeddedDecodeResult[] = [];
|
||||
for (let index = 0; index < bounds.length - 1; index += 1) {
|
||||
results.push(...decoder.push(buffer.subarray(bounds[index], bounds[index + 1])));
|
||||
}
|
||||
|
||||
if (results.length !== vector.expected.length) {
|
||||
failures.push(`${vector.name}: got ${results.length} results, want ${vector.expected.length}`);
|
||||
continue;
|
||||
}
|
||||
vector.expected.forEach((want, index) => {
|
||||
const got = results[index];
|
||||
if ("frame" in want) {
|
||||
if (!got.ok) {
|
||||
failures.push(`${vector.name}[${index}]: expected a frame, got an error`);
|
||||
return;
|
||||
}
|
||||
try {
|
||||
expect(got.frame).toEqual(want.frame);
|
||||
} catch {
|
||||
failures.push(`${vector.name}[${index}]: frame does not match`);
|
||||
}
|
||||
} else {
|
||||
if (got.ok) {
|
||||
failures.push(`${vector.name}[${index}]: expected an error, got a frame`);
|
||||
return;
|
||||
}
|
||||
if (got.error?.code !== want.error) {
|
||||
failures.push(`${vector.name}[${index}]: error ${got.error?.code} != ${want.error}`);
|
||||
}
|
||||
}
|
||||
});
|
||||
}
|
||||
expect(failures).toEqual([]);
|
||||
|
||||
// The encode side stays wire compatible too: one line, one newline, and the
|
||||
// same frame after a decode round trip.
|
||||
for (const vector of fixture.vectors.filter((entry) => entry.roundTrip)) {
|
||||
const want = vector.expected[0];
|
||||
if (!("frame" in want)) continue;
|
||||
const encoded = codec.encodeDuplexFrame(want.frame);
|
||||
expect(encoded.endsWith("\n")).toBe(true);
|
||||
expect(encoded.slice(0, -1)).not.toContain("\n");
|
||||
const decoded = codec.decodeDuplexLine(encoded.slice(0, -1));
|
||||
expect(decoded.ok).toBe(true);
|
||||
expect(decoded.frame).toEqual(want.frame);
|
||||
}
|
||||
|
||||
// The embedded copy enforces the same encode bound as the host copy. It runs
|
||||
// every shared encode vector and matches the expected result, so both copies
|
||||
// reject the same oversized frame with the same `frame_too_large` code.
|
||||
expect(fixture.encodeVectors.length).toBeGreaterThanOrEqual(3);
|
||||
const encodeFailures: string[] = [];
|
||||
for (const vector of fixture.encodeVectors) {
|
||||
const result = codec.encodeDuplexFrameChecked(vector.frame, vector.maxFrameBytes);
|
||||
if (result.ok !== vector.expected.ok) {
|
||||
encodeFailures.push(`${vector.name}: ok=${result.ok}, want ${vector.expected.ok}`);
|
||||
continue;
|
||||
}
|
||||
if (result.ok) {
|
||||
if (!result.line.endsWith("\n") || result.line.slice(0, -1).includes("\n")) {
|
||||
encodeFailures.push(`${vector.name}: line is not exactly one frame line`);
|
||||
continue;
|
||||
}
|
||||
const decoded = codec.decodeDuplexLine(result.line.slice(0, -1));
|
||||
if (!decoded.ok) {
|
||||
encodeFailures.push(`${vector.name}: an ok encode did not decode back`);
|
||||
continue;
|
||||
}
|
||||
try {
|
||||
expect(decoded.frame).toEqual(vector.frame);
|
||||
} catch {
|
||||
encodeFailures.push(`${vector.name}: decoded frame does not match`);
|
||||
}
|
||||
} else if (!vector.expected.ok && result.error.code !== vector.expected.error) {
|
||||
encodeFailures.push(`${vector.name}: error ${result.error.code} != ${vector.expected.error}`);
|
||||
}
|
||||
}
|
||||
expect(encodeFailures).toEqual([]);
|
||||
});
|
||||
|
||||
it("bounds the sandbox decoder with a separate sandbox_process scope, distinct from the host ledger scope", () => {
|
||||
// The generated gateway runs in a separate operating-system process, so its
|
||||
// decoder cannot share the host aggregate byte ledger. It enforces a separate
|
||||
// local cap under the `sandbox_process` scope. This test wraps the embedded
|
||||
// source and the host decoder, then proves the two scopes never overlap and the
|
||||
// sandbox cap fails closed.
|
||||
const codecFactory = new Function(
|
||||
`${getSandboxDuplexGatewayCodecSource()}\nreturn { encodeDuplexFrame, decodeDuplexLine, DuplexFrameDecoder, DUPLEX_FRAME_VERSION, DEFAULT_MAX_DUPLEX_FRAME_BYTES, DUPLEX_DECODER_SCOPE, DEFAULT_MAX_DUPLEX_DECODER_BYTES };`,
|
||||
) as unknown as () => EmbeddedCodec;
|
||||
const codec = codecFactory();
|
||||
|
||||
// The sandbox scope is the fixed `sandbox_process` label.
|
||||
expect(codec.DUPLEX_DECODER_SCOPE).toBe("sandbox_process");
|
||||
expect(codec.DEFAULT_MAX_DUPLEX_DECODER_BYTES).toBeGreaterThan(codec.DEFAULT_MAX_DUPLEX_FRAME_BYTES);
|
||||
// The two scopes are distinct: the host owner set never carries the sandbox
|
||||
// scope, so the sandbox counter can never map to a host aggregate token.
|
||||
expect((DUPLEX_AGGREGATE_TOKEN_OWNERS as readonly string[]).includes("sandbox_process")).toBe(false);
|
||||
|
||||
// The sandbox decoder tracks its own `sandbox_process` counter. A frame under
|
||||
// the cap charges the local counter, and the counter returns to zero once the
|
||||
// frame drains.
|
||||
const sandboxDecoder = new codec.DuplexFrameDecoder({ maxAggregateBytes: 64 });
|
||||
expect(sandboxDecoder.scope).toBe("sandbox_process");
|
||||
const partial = sandboxDecoder.push(Buffer.from('{"version":1,', "utf8"));
|
||||
expect(partial).toEqual([]);
|
||||
expect(sandboxDecoder.bytesInUse).toBe(Buffer.byteLength('{"version":1,', "utf8"));
|
||||
sandboxDecoder.push(Buffer.from('"type":"heartbeat"}\n', "utf8"));
|
||||
expect(sandboxDecoder.bytesInUse).toBe(0);
|
||||
|
||||
// A chunk over the local cap fails closed. The decoder retains nothing, reports
|
||||
// the aggregate rejection, and increments only its local `sandbox_process`
|
||||
// counter.
|
||||
const cappedDecoder = new codec.DuplexFrameDecoder({ maxAggregateBytes: 8 });
|
||||
const overCap = cappedDecoder.push(Buffer.from("x".repeat(64), "utf8"));
|
||||
expect(overCap.length).toBe(1);
|
||||
const rejection = overCap[0];
|
||||
expect(rejection.ok).toBe(false);
|
||||
expect(rejection.error?.code).toBe("aggregate_bytes_exceeded");
|
||||
expect(cappedDecoder.bytesInUse).toBe(0);
|
||||
expect(cappedDecoder.aggregateRejections).toBe(1);
|
||||
|
||||
// The host decoder charges the host aggregate byte ledger under the
|
||||
// `decoder_buffer` owner. It never uses the sandbox scope. This proves the two
|
||||
// implementations use distinct scopes: the host uses the injected ledger; the
|
||||
// sandbox uses its local counter.
|
||||
const ledger = new DuplexAggregateByteLedger({ ceilingBytes: 1024 });
|
||||
const hostDecoder = new DuplexFrameDecoder({ aggregateByteLedger: ledger });
|
||||
hostDecoder.push(Buffer.from('{"version":1,', "utf8"));
|
||||
expect(ledger.bytesInUse).toBe(Buffer.byteLength('{"version":1,', "utf8"));
|
||||
expect(ledger.liveTokenCount).toBeGreaterThan(0);
|
||||
hostDecoder.dispose();
|
||||
expect(ledger.bytesInUse).toBe(0);
|
||||
expect(ledger.liveTokenCount).toBe(0);
|
||||
// The host encode side produces the same bytes for the same frame, so the
|
||||
// two copies stay wire compatible on the one frame the gateway still sends.
|
||||
expect(encoded).toBe(encodeDuplexFrame({ version: DUPLEX_FRAME_VERSION, type: "ready", nonce }));
|
||||
});
|
||||
});
|
||||
|
||||
|
|
|
|||
|
|
@ -2,7 +2,6 @@ import fs from "node:fs/promises";
|
|||
import net from "node:net";
|
||||
import os from "node:os";
|
||||
import path from "node:path";
|
||||
import { Readable } from "node:stream";
|
||||
import { randomBytes, randomUUID } from "node:crypto";
|
||||
import type { SshRemoteExecutionSpec } from "./ssh.js";
|
||||
import {
|
||||
|
|
@ -33,7 +32,6 @@ import {
|
|||
createSandboxCallbackBridgeAsset,
|
||||
createSandboxCallbackBridgeToken,
|
||||
DEFAULT_SANDBOX_CALLBACK_BRIDGE_MAX_BODY_BYTES,
|
||||
DEFAULT_SANDBOX_DUPLEX_DECODER_MAX_BYTES,
|
||||
SANDBOX_CALLBACK_BRIDGE_ENTRYPOINT,
|
||||
SANDBOX_CALLBACK_BRIDGE_HTTP2_MODE,
|
||||
sandboxCallbackBridgeDirectories,
|
||||
|
|
@ -57,7 +55,6 @@ import {
|
|||
type DuplexBrokerRunDisposition,
|
||||
} from "./bridge-transport-contract.js";
|
||||
import { decodeDuplexLine, DEFAULT_MAX_DUPLEX_FRAME_BYTES } from "./duplex-frame-codec.js";
|
||||
import type { ReassembledBody } from "./duplex-body-spool.js";
|
||||
import {
|
||||
createDuplexObservability,
|
||||
mapHttp2EventToDuplexLossReason,
|
||||
|
|
@ -3691,12 +3688,6 @@ export async function startAdapterExecutionTargetPaperclipBridge(input: {
|
|||
signal?: AbortSignal,
|
||||
options?: {
|
||||
suppressDebugLog?: boolean;
|
||||
/**
|
||||
* The duplex broker passes the reassembled request body here. The forward
|
||||
* streams it to the host, so a spilled body never loads into memory on the
|
||||
* receive side. The forward never disposes it; the broker owns its lifecycle.
|
||||
*/
|
||||
reassembledBody?: ReassembledBody;
|
||||
},
|
||||
): Promise<{ status: number; headers: Record<string, string>; body: string }> => {
|
||||
const method = request.method.trim().toUpperCase() || "GET";
|
||||
|
|
@ -3722,24 +3713,15 @@ export async function startAdapterExecutionTargetPaperclipBridge(input: {
|
|||
// the forward budget here, whichever comes first.
|
||||
const timeoutSignal = AbortSignal.timeout(forwardTimeoutMs);
|
||||
const forwardSignal = signal ? AbortSignal.any([signal, timeoutSignal]) : timeoutSignal;
|
||||
// Build the request-body init. A GET or a HEAD carries no body. The duplex
|
||||
// broker passes the reassembled body: stream it, so a body that spilled to
|
||||
// disk never loads into memory here. A streamed body needs `duplex: "half"`.
|
||||
// The file bridge passes the whole body as one string.
|
||||
const forwardInit: RequestInit & { duplex?: "half" } = {
|
||||
// Build the request-body init. A GET or a HEAD carries no body. The file
|
||||
// bridge passes the whole body as one string.
|
||||
const forwardInit: RequestInit = {
|
||||
method,
|
||||
headers,
|
||||
signal: forwardSignal,
|
||||
};
|
||||
if (method !== "GET" && method !== "HEAD") {
|
||||
if (options?.reassembledBody) {
|
||||
forwardInit.body = Readable.toWeb(
|
||||
options.reassembledBody.createReadStream(),
|
||||
) as unknown as ReadableStream<Uint8Array>;
|
||||
forwardInit.duplex = "half";
|
||||
} else if (typeof request.body === "string") {
|
||||
forwardInit.body = request.body;
|
||||
}
|
||||
if (method !== "GET" && method !== "HEAD" && typeof request.body === "string") {
|
||||
forwardInit.body = request.body;
|
||||
}
|
||||
const response = await fetch(buildBridgeForwardUrl(hostApiUrl, request), forwardInit);
|
||||
if (emitDebugLog) {
|
||||
|
|
@ -3873,11 +3855,6 @@ export async function startAdapterExecutionTargetPaperclipBridge(input: {
|
|||
PAPERCLIP_BRIDGE_PORT: String(assignedPort),
|
||||
PAPERCLIP_BRIDGE_NONCE: nonce,
|
||||
PAPERCLIP_BRIDGE_MAX_BODY_BYTES: String(maxBodyBytes),
|
||||
// The separate sandbox-process raw-decoder cap. The generated gateway runs
|
||||
// in a different operating-system process, so it cannot share the host
|
||||
// aggregate byte ledger. It enforces this cap locally under the
|
||||
// `sandbox_process` scope, and the provider memory allocation bounds it.
|
||||
PAPERCLIP_BRIDGE_MAX_DUPLEX_DECODER_BYTES: String(DEFAULT_SANDBOX_DUPLEX_DECODER_MAX_BYTES),
|
||||
};
|
||||
const command = buildDuplexGatewayLaunchArgv({
|
||||
shellCommand,
|
||||
|
|
|
|||
|
|
@ -3075,8 +3075,12 @@ describe("sandbox callback bridge", () => {
|
|||
|
||||
const newlineIndex = firstBytes.indexOf(0x0a);
|
||||
expect(newlineIndex).toBeGreaterThan(0);
|
||||
// Assert the exact UTF-8 bytes, not a parsed-and-matched object.
|
||||
// `JSON.stringify` writes keys in the object-literal insertion order, so a
|
||||
// reordered or reformatted call site at the gateway's one `writeFrame` call
|
||||
// would change the bytes on the wire without failing a looser assertion.
|
||||
const readyLine = firstBytes.subarray(0, newlineIndex).toString("utf8");
|
||||
expect(JSON.parse(readyLine)).toMatchObject({ type: "ready", nonce });
|
||||
expect(readyLine).toBe(`{"version":2,"type":"ready","nonce":"${nonce}"}`);
|
||||
const afterReady = firstBytes.subarray(newlineIndex + 1, newlineIndex + 1 + preface.length);
|
||||
expect(afterReady).toEqual(preface);
|
||||
}, 15_000);
|
||||
|
|
|
|||
|
|
@ -21,14 +21,6 @@ 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;
|
||||
// The default cap on the aggregate raw bytes the in-sandbox duplex frame decoder
|
||||
// retains between chunks. The generated gateway runs in a separate operating-system
|
||||
// process, so it cannot share the host aggregate byte ledger. It enforces this
|
||||
// separate cap locally under the `sandbox_process` scope, and the provider memory
|
||||
// allocation bounds it. The host passes the value through
|
||||
// PAPERCLIP_BRIDGE_MAX_DUPLEX_DECODER_BYTES. The default is well above one maximum
|
||||
// frame, so a legitimate single frame never trips it.
|
||||
const DEFAULT_BRIDGE_MAX_DUPLEX_DECODER_BYTES = 8 * 1024 * 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
|
||||
|
|
@ -95,18 +87,6 @@ const CALLBACK_BRIDGE_WORKER_FAILED_SPAN = "sandbox.callbackBridge.workerFailed"
|
|||
|
||||
export const DEFAULT_SANDBOX_CALLBACK_BRIDGE_MAX_BODY_BYTES = DEFAULT_BRIDGE_MAX_BODY_BYTES;
|
||||
|
||||
/**
|
||||
* The default cap on the aggregate raw bytes the generated in-sandbox duplex frame
|
||||
* decoder retains. The host passes it through
|
||||
* `PAPERCLIP_BRIDGE_MAX_DUPLEX_DECODER_BYTES`. The in-sandbox decoder enforces the
|
||||
* cap locally under the `sandbox_process` scope; it never touches the host
|
||||
* aggregate byte ledger.
|
||||
*/
|
||||
export const DEFAULT_SANDBOX_DUPLEX_DECODER_MAX_BYTES = DEFAULT_BRIDGE_MAX_DUPLEX_DECODER_BYTES;
|
||||
|
||||
/** The scope label the in-sandbox duplex decoder reports for its local byte counter. */
|
||||
export const SANDBOX_DUPLEX_DECODER_SCOPE = "sandbox_process";
|
||||
|
||||
export interface SandboxCallbackBridgeRouteRule {
|
||||
method: string;
|
||||
path: RegExp;
|
||||
|
|
@ -2027,246 +2007,30 @@ export function createSandboxHttp2BridgeGateway(
|
|||
|
||||
/**
|
||||
* The zero-dependency codec the generated duplex gateway embeds. It is a plain
|
||||
* JavaScript copy of the host codec in `duplex-frame-codec.ts`. It uses only the
|
||||
* `Buffer`, `JSON`, `Object`, `Set`, and `Array` globals, so the generated
|
||||
* `.mjs` needs no workspace import. It carries no template literal and no `${`
|
||||
* sequence, so it embeds inside the gateway template literal with no escape.
|
||||
* JavaScript copy of the encode side of the host codec in `duplex-frame-codec.ts`.
|
||||
* The generated gateway never decodes: it writes exactly one READY line, then
|
||||
* hands stdout to its transport. So this copy carries only the frame version and
|
||||
* the encode function, not a decoder. It uses only the `Buffer` and `JSON`
|
||||
* globals, so the generated `.mjs` needs no workspace import. It carries no
|
||||
* template literal and no `${` sequence, so it embeds inside the gateway
|
||||
* template literal with no escape.
|
||||
*
|
||||
* A shared fixture file (`duplex-frame-vectors.json`) proves this copy stays wire
|
||||
* compatible with the host codec. {@link getSandboxDuplexGatewayCodecSource}
|
||||
* returns this exact source so a test can run every fixture vector against it.
|
||||
* A shared fixture file (`duplex-frame-vectors.json`) proves the host encode side
|
||||
* stays wire compatible with this copy. {@link getSandboxDuplexGatewayCodecSource}
|
||||
* returns this exact source so a test can run the READY encode vector against it.
|
||||
*/
|
||||
const DUPLEX_GATEWAY_CODEC_SOURCE = `const DUPLEX_FRAME_VERSION = 2;
|
||||
const DUPLEX_BODY_CHUNK_RAW_BYTES = 256 * 1024;
|
||||
const DEFAULT_MAX_DUPLEX_FRAME_BYTES = 1000000;
|
||||
const DEFAULT_MAX_DUPLEX_REQUEST_ID_BYTES = 256;
|
||||
const DEFAULT_MAX_DUPLEX_DECODER_BYTES = ${DEFAULT_BRIDGE_MAX_DUPLEX_DECODER_BYTES};
|
||||
const DUPLEX_DECODER_SCOPE = "${SANDBOX_DUPLEX_DECODER_SCOPE}";
|
||||
const DUPLEX_NEWLINE_BYTE = 0x0a;
|
||||
const DUPLEX_EMPTY = Buffer.alloc(0);
|
||||
const DUPLEX_RESPONSE_OUTCOMES = new Set(["completed", "indeterminate", "unavailable"]);
|
||||
|
||||
function duplexOk(frame) {
|
||||
return { ok: true, frame: frame };
|
||||
}
|
||||
|
||||
function duplexFail(code, message) {
|
||||
return { ok: false, error: { code: code, message: message } };
|
||||
}
|
||||
|
||||
function duplexIsPlainObject(value) {
|
||||
return typeof value === "object" && value !== null && !Array.isArray(value);
|
||||
}
|
||||
|
||||
function duplexIsStringRecord(value) {
|
||||
if (!duplexIsPlainObject(value)) return false;
|
||||
for (const entry of Object.values(value)) {
|
||||
if (typeof entry !== "string") return false;
|
||||
}
|
||||
return true;
|
||||
}
|
||||
|
||||
function duplexIsSafeNonNegativeInteger(value) {
|
||||
return typeof value === "number" && Number.isSafeInteger(value) && value >= 0;
|
||||
}
|
||||
|
||||
function encodeDuplexFrame(frame) {
|
||||
return JSON.stringify(frame) + "\\n";
|
||||
}
|
||||
|
||||
function encodeDuplexFrameChecked(frame, maxFrameBytes) {
|
||||
const limit = maxFrameBytes != null ? maxFrameBytes : DEFAULT_MAX_DUPLEX_FRAME_BYTES;
|
||||
const json = JSON.stringify(frame);
|
||||
if (Buffer.byteLength(json, "utf8") > limit) {
|
||||
return { ok: false, error: { code: "frame_too_large", message: "frame exceeds the maximum size" } };
|
||||
}
|
||||
return { ok: true, line: json + "\\n" };
|
||||
}
|
||||
|
||||
function decodeDuplexLine(line) {
|
||||
const text = typeof line === "string" ? line : line.toString("utf8");
|
||||
let parsed;
|
||||
try {
|
||||
parsed = JSON.parse(text);
|
||||
} catch (error) {
|
||||
return duplexFail("malformed_frame", "frame is not valid JSON");
|
||||
}
|
||||
if (!duplexIsPlainObject(parsed)) {
|
||||
return duplexFail("malformed_frame", "frame is not a JSON object");
|
||||
}
|
||||
if (parsed.version !== DUPLEX_FRAME_VERSION) {
|
||||
return duplexFail("version_mismatch", "frame version " + String(parsed.version) + " is not " + DUPLEX_FRAME_VERSION);
|
||||
}
|
||||
return duplexValidateFrame(parsed);
|
||||
}
|
||||
|
||||
function duplexValidateFrame(frame) {
|
||||
switch (frame.type) {
|
||||
case "request":
|
||||
return duplexValidateRequest(frame);
|
||||
case "response":
|
||||
return duplexValidateResponse(frame);
|
||||
case "body_chunk":
|
||||
return duplexValidateBodyChunk(frame);
|
||||
case "ready":
|
||||
return duplexValidateReady(frame);
|
||||
case "heartbeat":
|
||||
return duplexOk(frame);
|
||||
case "close":
|
||||
return duplexOk(frame);
|
||||
case "error":
|
||||
return duplexValidateError(frame);
|
||||
default:
|
||||
return duplexFail("unknown_type", "unknown frame type " + JSON.stringify(frame.type));
|
||||
}
|
||||
}
|
||||
|
||||
function duplexValidateRequest(frame) {
|
||||
if (
|
||||
typeof frame.id !== "string" ||
|
||||
typeof frame.method !== "string" ||
|
||||
typeof frame.path !== "string" ||
|
||||
typeof frame.query !== "string" ||
|
||||
!duplexIsSafeNonNegativeInteger(frame.bodyByteCount) ||
|
||||
!duplexIsStringRecord(frame.headers)
|
||||
) {
|
||||
return duplexFail("malformed_frame", "request frame has a missing or wrong-typed field");
|
||||
}
|
||||
if (Buffer.byteLength(frame.id, "utf8") > DEFAULT_MAX_DUPLEX_REQUEST_ID_BYTES) {
|
||||
return duplexFail("id_too_large", "request frame id exceeds the maximum size");
|
||||
}
|
||||
return duplexOk(frame);
|
||||
}
|
||||
|
||||
function duplexValidateResponse(frame) {
|
||||
if (
|
||||
typeof frame.id !== "string" ||
|
||||
typeof frame.status !== "number" ||
|
||||
!duplexIsSafeNonNegativeInteger(frame.bodyByteCount) ||
|
||||
!duplexIsStringRecord(frame.headers) ||
|
||||
typeof frame.outcome !== "string" ||
|
||||
!DUPLEX_RESPONSE_OUTCOMES.has(frame.outcome)
|
||||
) {
|
||||
return duplexFail("malformed_frame", "response frame has a missing or wrong-typed field");
|
||||
}
|
||||
if (Buffer.byteLength(frame.id, "utf8") > DEFAULT_MAX_DUPLEX_REQUEST_ID_BYTES) {
|
||||
return duplexFail("id_too_large", "response frame id exceeds the maximum size");
|
||||
}
|
||||
return duplexOk(frame);
|
||||
}
|
||||
|
||||
function duplexValidateBodyChunk(frame) {
|
||||
if (typeof frame.id !== "string" || typeof frame.data !== "string") {
|
||||
return duplexFail("malformed_frame", "body_chunk frame has a missing or wrong-typed field");
|
||||
}
|
||||
if (!duplexIsSafeNonNegativeInteger(frame.seq)) {
|
||||
return duplexFail("malformed_frame", "body_chunk frame has a missing or wrong-typed seq");
|
||||
}
|
||||
if (Buffer.byteLength(frame.id, "utf8") > DEFAULT_MAX_DUPLEX_REQUEST_ID_BYTES) {
|
||||
return duplexFail("id_too_large", "body_chunk frame id exceeds the maximum size");
|
||||
}
|
||||
return duplexOk(frame);
|
||||
}
|
||||
|
||||
function duplexValidateReady(frame) {
|
||||
if (typeof frame.nonce !== "string") {
|
||||
return duplexFail("malformed_frame", "ready frame has a missing or wrong-typed nonce");
|
||||
}
|
||||
for (const key of Object.keys(frame)) {
|
||||
if (key !== "version" && key !== "type" && key !== "nonce") {
|
||||
return duplexFail("malformed_frame", "ready frame has an unexpected field");
|
||||
}
|
||||
}
|
||||
return duplexOk(frame);
|
||||
}
|
||||
|
||||
function duplexValidateError(frame) {
|
||||
if (typeof frame.code !== "string") {
|
||||
return duplexFail("malformed_frame", "error frame has a missing or wrong-typed code");
|
||||
}
|
||||
if (frame.message !== undefined && typeof frame.message !== "string") {
|
||||
return duplexFail("malformed_frame", "error frame has a wrong-typed message");
|
||||
}
|
||||
return duplexOk(frame);
|
||||
}
|
||||
|
||||
class DuplexFrameDecoder {
|
||||
constructor(options) {
|
||||
this.buffer = DUPLEX_EMPTY;
|
||||
this.discarding = false;
|
||||
this.maxFrameBytes =
|
||||
options && options.maxFrameBytes != null ? options.maxFrameBytes : DEFAULT_MAX_DUPLEX_FRAME_BYTES;
|
||||
// The in-sandbox decoder runs in a separate operating-system process, so it
|
||||
// cannot share the host aggregate byte ledger. It bounds its own retained
|
||||
// bytes with a local cap under the "sandbox_process" scope. The counters here
|
||||
// never increment or release a host aggregate token.
|
||||
this.scope = DUPLEX_DECODER_SCOPE;
|
||||
this.maxAggregateBytes =
|
||||
options && options.maxAggregateBytes != null
|
||||
? options.maxAggregateBytes
|
||||
: DEFAULT_MAX_DUPLEX_DECODER_BYTES;
|
||||
this.bytesInUse = 0;
|
||||
this.aggregateRejections = 0;
|
||||
}
|
||||
|
||||
push(chunk) {
|
||||
const incoming = typeof chunk === "string" ? Buffer.from(chunk, "utf8") : Buffer.from(chunk);
|
||||
// Enforce the local sandbox-process cap before the concat allocates the peak
|
||||
// "old + incoming" buffer. A rejection fails closed: the decoder drops the
|
||||
// retained buffer and the incoming chunk, resynchronizes at the next newline,
|
||||
// and reports the aggregate rejection. It retains nothing over the cap.
|
||||
if (this.buffer.length + incoming.length > this.maxAggregateBytes) {
|
||||
this.buffer = DUPLEX_EMPTY;
|
||||
this.discarding = true;
|
||||
this.bytesInUse = 0;
|
||||
this.aggregateRejections += 1;
|
||||
return [duplexFail("aggregate_bytes_exceeded", "aggregate retained bytes exceeded the sandbox-process cap")];
|
||||
}
|
||||
this.buffer = this.buffer.length === 0 ? incoming : Buffer.concat([this.buffer, incoming]);
|
||||
const results = [];
|
||||
for (;;) {
|
||||
if (this.discarding) {
|
||||
const newlineIndex = this.buffer.indexOf(DUPLEX_NEWLINE_BYTE);
|
||||
if (newlineIndex === -1) {
|
||||
this.buffer = DUPLEX_EMPTY;
|
||||
break;
|
||||
}
|
||||
this.buffer = this.buffer.subarray(newlineIndex + 1);
|
||||
this.discarding = false;
|
||||
continue;
|
||||
}
|
||||
const newlineIndex = this.buffer.indexOf(DUPLEX_NEWLINE_BYTE);
|
||||
if (newlineIndex === -1) {
|
||||
if (this.buffer.length > this.maxFrameBytes) {
|
||||
results.push(duplexFail("frame_too_large", "frame exceeds the maximum size"));
|
||||
this.discarding = true;
|
||||
this.buffer = DUPLEX_EMPTY;
|
||||
}
|
||||
break;
|
||||
}
|
||||
const line = this.buffer.subarray(0, newlineIndex);
|
||||
this.buffer = this.buffer.subarray(newlineIndex + 1);
|
||||
if (line.length === 0) continue;
|
||||
if (line.length > this.maxFrameBytes) {
|
||||
results.push(duplexFail("frame_too_large", "frame exceeds the maximum size"));
|
||||
continue;
|
||||
}
|
||||
results.push(decodeDuplexLine(line));
|
||||
}
|
||||
// Reconcile the local sandbox-process counter to the bytes still retained.
|
||||
this.bytesInUse = this.buffer.length;
|
||||
return results;
|
||||
}
|
||||
}`;
|
||||
|
||||
/**
|
||||
* Return the exact zero-dependency codec source the generated duplex gateway
|
||||
* embeds. A test runs every fixture vector against this source, so it proves the
|
||||
* embedded copy decodes the same bytes as the host codec. The source declares
|
||||
* `encodeDuplexFrame`, `encodeDuplexFrameChecked`, `decodeDuplexLine`,
|
||||
* `DuplexFrameDecoder`, `DUPLEX_FRAME_VERSION`, and
|
||||
* `DEFAULT_MAX_DUPLEX_FRAME_BYTES`, but exports none of them; a caller wraps it to
|
||||
* read those names.
|
||||
* embeds. A test wraps this source and calls `encodeDuplexFrame` to prove the
|
||||
* embedded copy encodes the READY frame the same way the host encode side does.
|
||||
* The source declares `encodeDuplexFrame` and `DUPLEX_FRAME_VERSION`, but exports
|
||||
* neither; a caller wraps it to read those names.
|
||||
*/
|
||||
export function getSandboxDuplexGatewayCodecSource(): string {
|
||||
return DUPLEX_GATEWAY_CODEC_SOURCE;
|
||||
|
|
|
|||
Loading…
Reference in New Issue