diff --git a/packages/adapter-utils/src/duplex-bridge-broker.test.ts b/packages/adapter-utils/src/duplex-bridge-broker.test.ts new file mode 100644 index 00000000000..c0d11e42d7d --- /dev/null +++ b/packages/adapter-utils/src/duplex-bridge-broker.test.ts @@ -0,0 +1,320 @@ +import { afterEach, describe, expect, it } from "vitest"; + +import { + assertDuplexBrokerLimits, + createDuplexBridgeBroker, + type DuplexBridgeBroker, + type DuplexBrokerForwardResult, +} from "./duplex-bridge-broker.js"; +import { + DUPLEX_FRAME_VERSION, + DuplexFrameDecoder, + encodeDuplexFrame, + type DuplexRequestFrame, + type DuplexResponseFrame, +} from "./duplex-frame-codec.js"; +import { + createDuplexTelemetry, + DUPLEX_SPAN_REQUEST, + type DuplexTelemetryCounterRecord, + type DuplexTelemetryEventRecord, + type DuplexTelemetrySpanRecord, +} from "./duplex-telemetry.js"; +import type { CommandManagedDuplexChannel } from "./command-managed-runtime.js"; + +/** + * Unit harness for the host broker limit gate. + * + * The harness drives frames straight into the broker through an in-memory + * channel. It does not spawn the gateway, so a test injects untrusted request + * frames directly, the same as a malicious provider that controls the transport. + * The channel records each frame the broker writes back, so a test asserts the + * exact response the broker returns for a refused request. + */ + +/** One pending forward the test controls. */ +interface PendingForward { + id: string; + resolve: (result: DuplexBrokerForwardResult) => void; + reject: (error: Error) => void; + settled: boolean; +} + +/** The in-memory channel plus the levers a test uses to drive the broker. */ +interface FakeChannelHarness { + channel: CommandManagedDuplexChannel; + /** Push one request frame into the broker read path. */ + feed: (frame: DuplexRequestFrame) => void; + /** The response frames the broker wrote back, in order. */ + responses: DuplexResponseFrame[]; + /** Every forward call the broker made, in order. */ + forwards: PendingForward[]; + /** Resolve the newest unsettled forward for one id with a 200 result. */ + resolveForward: (id: string, body?: string) => void; + /** Resolve every unsettled forward with a 200 result. */ + resolveAll: () => void; +} + +/** Build one valid request frame with distinctive, secret-looking fields. */ +function requestFrame(id: string): DuplexRequestFrame { + return { + version: DUPLEX_FRAME_VERSION, + type: "request", + id, + method: "POST", + path: `/api/issues/${id}`, + query: "?secret-query=leak", + headers: { authorization: "Bearer super-secret-provider-token" }, + body: JSON.stringify({ secret: "secret-request-body" }), + }; +} + +/** + * Build the in-memory channel harness. The channel keeps the broker read + * listener, so `feed` pushes an encoded frame into the broker. The channel + * decodes each written frame, so `responses` holds the response frames only. + */ +function createFakeChannelHarness(): FakeChannelHarness { + const responses: DuplexResponseFrame[] = []; + const forwards: PendingForward[] = []; + const writtenDecoder = new DuplexFrameDecoder(); + let dataListener: ((chunk: string) => void) | null = null; + + const channel: CommandManagedDuplexChannel = { + write: (data) => { + for (const result of writtenDecoder.push(data)) { + if (result.ok && result.frame.type === "response") { + responses.push(result.frame); + } + } + }, + onData: (listener) => { + dataListener = listener; + }, + onExit: () => undefined, + stop: () => undefined, + close: () => Promise.resolve(), + }; + + return { + channel, + feed: (frame) => { + if (!dataListener) throw new Error("The broker did not bind the data listener."); + dataListener(encodeDuplexFrame(frame)); + }, + responses, + forwards, + resolveForward: (id, body = JSON.stringify({ ok: true })) => { + const forward = [...forwards].reverse().find((entry) => entry.id === id && !entry.settled); + if (!forward) throw new Error(`No unsettled forward for id ${id}.`); + forward.settled = true; + forward.resolve({ status: 200, headers: { "content-type": "application/json" }, body }); + }, + resolveAll: () => { + for (const forward of forwards) { + if (forward.settled) continue; + forward.settled = true; + forward.resolve({ + status: 200, + headers: { "content-type": "application/json" }, + body: JSON.stringify({ ok: true }), + }); + } + }, + }; +} + +/** A forward handler that hands each call to the harness and never auto-resolves. */ +function controllableForward(harness: FakeChannelHarness) { + return (request: DuplexRequestFrame): Promise => + new Promise((resolve, reject) => { + harness.forwards.push({ id: request.id, resolve, reject, settled: false }); + }); +} + +/** The telemetry sink capture. It proves the broker records nothing for a refusal. */ +interface TelemetryCapture { + spans: DuplexTelemetrySpanRecord[]; + counters: DuplexTelemetryCounterRecord[]; + events: DuplexTelemetryEventRecord[]; +} + +/** Build the real telemetry facade over a capturing recorder. */ +function createTelemetryCapture(): { telemetry: ReturnType; capture: TelemetryCapture } { + const capture: TelemetryCapture = { spans: [], counters: [], events: [] }; + const telemetry = createDuplexTelemetry({ + providerKey: "daytona", + recorder: { + recordSpan: (record) => capture.spans.push(record), + incrementCounter: (record) => capture.counters.push(record), + emitEvent: (record) => capture.events.push(record), + }, + }); + return { telemetry, capture }; +} + +describe("duplex bridge broker request limits", () => { + const brokers: DuplexBridgeBroker[] = []; + + afterEach(async () => { + while (brokers.length > 0) { + const broker = brokers.pop(); + if (broker) await broker.close(); + } + }); + + it("rejects a non-positive or non-integer limit at construction", () => { + expect(() => assertDuplexBrokerLimits({ maxInFlightRequests: 0, maxLifetimeRequests: 10 })).toThrow(); + expect(() => assertDuplexBrokerLimits({ maxInFlightRequests: 5, maxLifetimeRequests: 0 })).toThrow(); + expect(() => assertDuplexBrokerLimits({ maxInFlightRequests: 1.5, maxLifetimeRequests: 10 })).toThrow(); + expect(() => + assertDuplexBrokerLimits({ maxInFlightRequests: Number.POSITIVE_INFINITY, maxLifetimeRequests: 10 }), + ).toThrow(); + expect(() => assertDuplexBrokerLimits({ maxInFlightRequests: 8, maxLifetimeRequests: 16 })).not.toThrow(); + }); + + it("bounds the in-flight forwards and refuses the excess with a retryable terminal response", () => { + const harness = createFakeChannelHarness(); + const { telemetry, capture } = createTelemetryCapture(); + const maxInFlightRequests = 4; + const broker = createDuplexBridgeBroker({ + channel: harness.channel, + forwardRequest: controllableForward(harness), + telemetry, + maxInFlightRequests, + }); + brokers.push(broker); + broker.start(); + + // Inject more than the limit of unique valid frames. Every forward hangs, so + // the broker holds the maximum number of in-flight forwards. + const total = 10; + const ids = Array.from({ length: total }, (_unused, index) => `flight-${index}`); + for (const id of ids) harness.feed(requestFrame(id)); + + // The broker forwarded only up to the limit. It refused the rest. + expect(harness.forwards).toHaveLength(maxInFlightRequests); + expect(harness.responses).toHaveLength(total - maxInFlightRequests); + for (const response of harness.responses) { + expect(response.status).toBe(503); + expect(response.outcome).toBe("unavailable"); + expect(response.headers["x-paperclip-bridge-outcome"]).toBe("unavailable"); + expect(JSON.parse(response.body)).toEqual({ + error: "Duplex broker capacity limit reached.", + outcome: "unavailable", + retryable: true, + }); + } + // The refusal leaks no route, query, body, or token. + const refusalText = harness.responses.map((response) => encodeDuplexFrame(response)).join(""); + expect(refusalText).not.toContain("secret"); + expect(refusalText).not.toContain("super-secret-provider-token"); + expect(refusalText).not.toContain("/api/issues/"); + + // The broker recorded no request span and no counter for a refusal. Only the + // delivered requests below produce a span. + expect(capture.spans).toHaveLength(0); + expect(capture.counters).toHaveLength(0); + + // Resolve the in-flight forwards. The broker delivers one response for each + // and drains the pending bookkeeping. + harness.resolveAll(); + }); + + it("frees capacity after a delivered request and forwards a resent refused id (no-replay preserved)", async () => { + const harness = createFakeChannelHarness(); + const maxInFlightRequests = 2; + const broker = createDuplexBridgeBroker({ + channel: harness.channel, + forwardRequest: controllableForward(harness), + maxInFlightRequests, + }); + brokers.push(broker); + broker.start(); + + // Saturate the in-flight limit with two hung forwards. + harness.feed(requestFrame("keep-0")); + harness.feed(requestFrame("keep-1")); + expect(harness.forwards).toHaveLength(2); + + // A third unique id is refused, retryable. The broker did not forward it and + // did not retain its id. + harness.feed(requestFrame("resend-me")); + expect(harness.forwards).toHaveLength(2); + expect(harness.responses).toHaveLength(1); + expect(JSON.parse(harness.responses[0].body).retryable).toBe(true); + + // Free one in-flight slot. The delivered request drains the pending record. + harness.resolveForward("keep-0"); + await Promise.resolve(); + + // The gateway resends the refused id. Capacity is free now, so the broker + // forwards it exactly one time. This proves the refusal did not poison the id. + harness.feed(requestFrame("resend-me")); + expect(harness.forwards.filter((forward) => forward.id === "resend-me")).toHaveLength(1); + + // A resend of an already-forwarded id never reaches the forward twice. + const forwardedKeep1 = harness.forwards.filter((forward) => forward.id === "keep-1").length; + harness.feed(requestFrame("keep-1")); + expect(harness.forwards.filter((forward) => forward.id === "keep-1")).toHaveLength(forwardedKeep1); + + harness.resolveAll(); + }); + + it("bounds the retained request ids over the channel lifetime and refuses each new id past the limit", async () => { + const harness = createFakeChannelHarness(); + const { telemetry, capture } = createTelemetryCapture(); + const maxLifetimeRequests = 3; + const broker = createDuplexBridgeBroker({ + channel: harness.channel, + // Resolve every forward at once, so each dispatched id leaves the in-flight + // count but stays in the retained-id set. + forwardRequest: async (request: DuplexRequestFrame): Promise => { + harness.forwards.push({ id: request.id, resolve: () => undefined, reject: () => undefined, settled: true }); + return { status: 200, headers: { "content-type": "application/json" }, body: JSON.stringify({ ok: true }) }; + }, + telemetry, + maxLifetimeRequests, + }); + brokers.push(broker); + broker.start(); + + // Inject more than the lifetime limit of unique valid frames. + const total = 9; + for (let index = 0; index < total; index += 1) { + harness.feed(requestFrame(`life-${index}`)); + await Promise.resolve(); + } + + // The broker forwarded only up to the lifetime limit. The retained-id set + // stays bounded, because a refused id never joins it. + expect(harness.forwards).toHaveLength(maxLifetimeRequests); + + // The broker delivered a real response for each forwarded request, then a + // bounded terminal refusal for each id past the limit. + const refusals = harness.responses.filter((response) => response.status === 503); + expect(refusals).toHaveLength(total - maxLifetimeRequests); + for (const refusal of refusals) { + expect(refusal.outcome).toBe("unavailable"); + // A lifetime refusal is terminal, so a resend does not help. + expect(JSON.parse(refusal.body).retryable).toBe(false); + } + + // A resend of a refused id stays refused, so the set never grows. + harness.feed(requestFrame("life-8")); + await Promise.resolve(); + expect(harness.forwards).toHaveLength(maxLifetimeRequests); + + // The telemetry surface stayed on the fixed request span for the delivered + // requests only. No refusal produced a span, a counter, or a loss record. + expect(capture.spans).toHaveLength(maxLifetimeRequests); + for (const span of capture.spans) { + expect(span.name).toBe(DUPLEX_SPAN_REQUEST); + expect(span.dimensions.outcome).toBe("ok"); + expect(Object.keys(span.dimensions).sort()).toEqual(["outcome", "provider", "transport"]); + } + expect(capture.counters).toHaveLength(0); + expect(broker.lossRecord).toBeNull(); + expect(broker.state).toBe("open"); + }); +});