diff --git a/src/checks.ts b/src/checks.ts index 0e7be08..4977407 100644 --- a/src/checks.ts +++ b/src/checks.ts @@ -475,15 +475,28 @@ export function runChecks(bundle: EvidenceBundle): CheckResult[] { } // ── AX 7: keyless flow ─────────────────────────────────────────────────── + // PASSES on one existence proof: any sampled door answering 2xx with no + // key. Own-scope keyed doors (a /keys/me, a /usage) CORRECTLY 401 a + // keyless probe — their refusals are the declared contract, not evidence + // against keyless-first — so the observe side walks candidates until a + // keyless door is found (hinted-keyless first, escalation past the first + // wave, src/discovery.ts). The check therefore fails ONLY when NO walked + // door grants keyless access at all. { const succeeded = probes.filter((p) => p.status !== null && p.status >= 200 && p.status < 300) + const allAuthRefused = probes.length > 0 && probes.every((p) => p.status === 401 || p.status === 403) checks.push(check('keyless-flow', 'at least one declared endpoint answers 2xx with no key', 7, probes.map((p) => p.role), probes.length === 0 ? { verdict: 'fail', detail: 'no keyless GET candidates discoverable from agents.json/OpenAPI — nothing an agent can try without an account' } : succeeded.length > 0 ? pass(`${succeeded.length}/${probes.length} sampled endpoint(s) answered 2xx keyless (seed ${bundle.seed})`) - : { verdict: 'fail', detail: `all ${probes.length} sampled keyless candidates failed (statuses: ${probes.map((p) => p.status ?? 'ERR').join(', ')})` })) + : { + verdict: 'fail', + detail: allAuthRefused + ? `no keyless access found: every sampled door (${probes.length}) demands a key (statuses: ${probes.map((p) => p.status ?? 'ERR').join(', ')}) — keyed doors answering 401 are legitimate, but at least one declared door must answer 2xx with no key` + : `all ${probes.length} sampled keyless candidates failed (statuses: ${probes.map((p) => p.status ?? 'ERR').join(', ')})`, + })) } // ── AX 8: 402 offers ───────────────────────────────────────────────────── diff --git a/src/discovery.ts b/src/discovery.ts index 507322a..7371058 100644 --- a/src/discovery.ts +++ b/src/discovery.ts @@ -1018,6 +1018,33 @@ function scrubAsMetadataEvidence(ev: Evidence | undefined): void { export const MAX_KEYLESS_PROBES = 3 +/** + * Escalation ceiling for the keyless sample (ax: keyless-flow robustness). + * The first wave is MAX_KEYLESS_PROBES doors; when NONE of them answers 2xx + * the sampler keeps walking the remaining candidates — a property with many + * own-scope KEYED doors (/keys/me, /usage — which CORRECTLY 401 keyless) must + * not fail keyless-flow just because the random three all landed on keyed + * doors while /listings answers 200 keyless. The walk stops at the FIRST + * keyless success (the check needs one existence proof, not a census) or at + * this ceiling (politeness: the escalation may spend at most + * MAX_KEYLESS_PROBE_TOTAL requests of the shared budget, and it only fires on + * surfaces where the cheap wave found no keyless door at all). A surface with + * genuinely NO keyless access probes up to the ceiling, every door refuses, + * and keyless-flow still FAILS — escalation widens the search, never the + * acceptance. + */ +export const MAX_KEYLESS_PROBE_TOTAL = 12 + +/** + * Card-declared per-endpoint auth hints (agents.json interfaces.http[].auth) + * that mark a door EXPLICITLY keyless. Honored twice: a hinted-keyless door + * sorts to the FRONT of the probe order (the strongest candidates are tried + * first), and a door hinted as anything else (apiKey, bearer, oauth…) is + * excluded from the keyless candidate pool entirely (its 401 is the declared + * contract, not evidence against keyless-first). + */ +export const KEYLESS_AUTH_HINT_RE = /none|keyless|public/i + /** * Upper bound on requests the contract-diff pass (step 5, below) will itself * fire, independent of however much of the shared politeness budget happens @@ -1830,16 +1857,42 @@ export async function observeTarget( // 2. Seeded endpoint sampling — which endpoints get probed is not // predictable before the run (the seed is fresh), but fully replayable // after it (the seed is in the report). + const templated = (p: string) => p.includes('{') || p.includes('%7B') // URL templates aren't probeable const candidatePaths = dedupe([ ...openapi.probeCandidates.map((c) => c.path), ...agents.endpoints - .filter((e) => e.method === 'GET' && (!e.auth || /none|keyless|public/i.test(e.auth))) + .filter((e) => e.method === 'GET' && (!e.auth || KEYLESS_AUTH_HINT_RE.test(e.auth))) .map((e) => pathOf(e.url, origin)) .filter((p): p is string => p !== undefined), ]) - .filter((p) => !p.includes('{') && !p.includes('%7B')) // URL templates aren't probeable + .filter((p) => !templated(p)) .sort() - for (const path of sampleSeeded(candidatePaths, MAX_KEYLESS_PROBES, seed)) { + // Doors the card EXPLICITLY hints keyless (interfaces.http[].auth "none"/ + // "keyless"/"public") are the strongest candidates: they go FIRST in the + // probe order, so a card that hints its doors resolves keyless-flow inside + // the cheap first wave. Unhinted candidates (openapi GETs with no security, + // interfaces.http entries with no auth member) follow in seeded order. + const hintedKeyless = new Set( + agents.endpoints + .filter((e) => e.method === 'GET' && typeof e.auth === 'string' && KEYLESS_AUTH_HINT_RE.test(e.auth)) + .map((e) => pathOf(e.url, origin)) + .filter((p): p is string => p !== undefined && !templated(p) && candidatePaths.includes(p)), + ) + const probeOrder = [ + ...sampleSeeded([...hintedKeyless].sort(), hintedKeyless.size, seed), + ...sampleSeeded(candidatePaths.filter((p) => !hintedKeyless.has(p)), candidatePaths.length, seed), + ] + // First wave: MAX_KEYLESS_PROBES doors (unchanged evidence volume for the + // sibling checks that read the sample). Escalation: while NO door has + // answered 2xx keyless yet, keep walking the remaining candidates up to + // MAX_KEYLESS_PROBE_TOTAL — so keyless-flow can only fail a surface where + // every walked door refused, never one where the random wave happened to + // land on own-scope keyed doors. Stops at the first keyless success. + let keylessProbed = 0 + let keylessFound = false + for (const path of probeOrder) { + if (keylessProbed >= MAX_KEYLESS_PROBE_TOTAL) break + if (keylessProbed >= MAX_KEYLESS_PROBES && keylessFound) break // Candidate paths are CARD-DERIVED: openapi path keys are raw attacker // input, and a key that does NOT begin with "/" (e.g. "@evil.example/x" or // ".evil.example/x") makes `${origin}${path}` resolve OFF-ORIGIN. The @@ -1849,7 +1902,9 @@ export async function observeTarget( // origin and pass unchanged. const url = `${origin}${path}` if (!isPubliclyRoutableSameOrigin(url, origin)) continue - await observer.observe(ROLE.keyless('GET', path), url, { accept: 'application/json' }) + const ev = await observer.observe(ROLE.keyless('GET', path), url, { accept: 'application/json' }) + keylessProbed += 1 + if (ev.status !== null && ev.status >= 200 && ev.status < 300) keylessFound = true } // 2c. Clause-3 typed-body legibility beyond the root (ax-fsg): a target that diff --git a/src/exec/runner.ts b/src/exec/runner.ts index cd48d42..770b933 100644 --- a/src/exec/runner.ts +++ b/src/exec/runner.ts @@ -124,6 +124,86 @@ export function unavailableExecRunner(reason: string = RUNNER_UNAVAILABLE_NO_BIN // The parent-side egress gateway — the floor, applied where the parent owns it // --------------------------------------------------------------------------- +/** + * THE CONNECTION BUDGET (ax: burst-vs-sibling isolation). A workerd isolate + * may hold only ~6 simultaneous open connections; past the budget the runtime + * force-closes the least-recently-used open response body, which surfaces to + * whoever was reading it as "Response closed due to connection limit". A suite + * test that BURSTS concurrent fetches (a 429 rate-limit test firing dozens of + * requests, bodies never read) would otherwise exhaust the budget and corrupt + * a SIBLING test's plain GET-and-parse mid-read — the verdict would then + * reflect the runner's plumbing, not the target's behavior. Two guards, both + * applied per isolate AND at the parent gateway: + * + * 1. A SEMAPHORE bounds in-flight fetches to EXEC_MAX_CONCURRENT_FETCHES + * (headroom under the budget); excess fetches QUEUE, they are never + * refused — the suite's assertions see the same statuses/bodies, only + * the wire-level concurrency is shaped. + * 2. Every non-stream response body is FULLY BUFFERED while the slot is + * held (`bufferResponse`), so an unread body can never pin a connection + * after its fetch resolves. `text/event-stream` responses pass through + * as live streams (buffering one would hang until the wall breaker) and + * release their slot on arrival. + * + * Neither guard changes what a suite test can assert: status, statusText, + * headers, url, and body bytes are preserved verbatim. + */ +export const EXEC_MAX_CONCURRENT_FETCHES = 5 + +/** A tiny FIFO semaphore bounding concurrent in-flight fetches. */ +export interface FetchLimiter { + acquire(): Promise + release(): void +} + +export function createFetchLimiter(max: number = EXEC_MAX_CONCURRENT_FETCHES): FetchLimiter { + let inFlight = 0 + const waiters: Array<() => void> = [] + return { + acquire(): Promise { + if (inFlight < max) { + inFlight += 1 + return Promise.resolve() + } + return new Promise((resolve) => waiters.push(resolve)) + }, + release(): void { + const next = waiters.shift() + if (next !== undefined) next() // the slot transfers; inFlight is unchanged + else inFlight -= 1 + }, + } +} + +/** Statuses the Response constructor refuses a body for (RFC 9110 null-body). */ +const NULL_BODY_STATUSES = new Set([101, 204, 205, 304]) + +/** + * Read a response FULLY into memory and hand back a memory-backed equivalent + * — same status, statusText, headers, url, and body bytes — so the underlying + * connection is consumed and closed the moment the fetch resolves, whether or + * not the suite ever reads the body. `text/event-stream` responses are + * returned untouched (a live stream must stay live). This is the half of the + * connection-budget guard that makes an UNREAD body harmless. + */ +export async function bufferResponse(res: Response): Promise { + const ctype = (res.headers.get('content-type') ?? '').toLowerCase() + if (ctype.includes('text/event-stream')) return res + if (res.body === null) return res + const buf = await res.arrayBuffer() + const out = new Response(NULL_BODY_STATUSES.has(res.status) ? null : buf, { + status: res.status, + statusText: res.statusText, + headers: res.headers, + }) + try { + Object.defineProperty(out, 'url', { value: res.url }) + } catch { + // A runtime with a non-configurable url getter keeps its own value. + } + return out +} + /** * Marker header stamped on every gateway-refused response: `"violation"` when * the floor/verb gate refused (a `GateViolation` was recorded), `"error"` when @@ -158,21 +238,27 @@ export const GATEWAY_MARKER_HEADER = 'x-apiqa-gateway' export async function gatewayFetch( request: Request, realFetch: (url: string, init?: RequestInit) => Promise = (url, init) => fetch(url, init), - opts: { sandbox?: boolean; violations?: GateViolation[] } = {}, + opts: { sandbox?: boolean; violations?: GateViolation[]; limiter?: FetchLimiter } = {}, ): Promise { const violations = opts.violations ?? [] const before = violations.length const gated = createGatedFetch({ realFetch, sandbox: opts.sandbox ?? true, violations }) + // The connection-budget guard, parent side: bound in-flight upstream + // fetches and consume every non-stream body while the slot is held — a + // suite burst can queue here, it can never pin unread upstream connections + // against the parent isolate's budget (see EXEC_MAX_CONCURRENT_FETCHES). + if (opts.limiter !== undefined) await opts.limiter.acquire() try { const headers: Record = {} request.headers.forEach((v, k) => { headers[k] = v }) - return await gated(request.url, { + const res = await gated(request.url, { method: request.method, headers, body: request.method === 'GET' || request.method === 'HEAD' ? undefined : await request.text(), }) + return await bufferResponse(res) } catch (err) { return new Response( JSON.stringify({ type: 'BLOCKED', reason: err instanceof Error ? err.message : String(err) }), @@ -184,6 +270,8 @@ export async function gatewayFetch( }, }, ) + } finally { + opts.limiter?.release() } } @@ -206,8 +294,12 @@ export function createOutboundGateway( opts: { sandbox?: boolean } = {}, ): OutboundGatewayLike { const violations: GateViolation[] = [] + // ONE limiter per gateway instance: the module-level deploy gateway is + // shared across concurrent runs, so this bound is what actually protects + // the parent isolate's connection budget from ANY combination of suites. + const limiter = createFetchLimiter() return { - fetch: (request: Request) => gatewayFetch(request, realFetch, { ...opts, violations }), + fetch: (request: Request) => gatewayFetch(request, realFetch, { ...opts, violations, limiter }), drainViolations: () => violations.splice(0, violations.length), } } @@ -242,6 +334,7 @@ const DOCUMENT = ${JSON.stringify(req.artifactKind === 'document')} const HAS_MODULE = ${JSON.stringify(hasModule)} const EXPORT_NAME = ${JSON.stringify(req.exportName ?? null)} const MARKER = ${JSON.stringify(GATEWAY_MARKER_HEADER)} +const MAX_CONCURRENT_FETCHES = ${JSON.stringify(EXEC_MAX_CONCURRENT_FETCHES)} // Captured at entry evaluation — BEFORE any suite byte runs — so suite code // patching Response/JSON cannot forge the body the parent folds. (The @@ -256,6 +349,29 @@ export default { globalThis[${JSON.stringify('__APIQA_VITEST_RUNS__')}] = { [${JSON.stringify(HOSTED_RUN_ID)}]: { api: harness.api } } Math.random = seededRandom(SEED) const realFetch = globalThis.fetch.bind(globalThis) + // THE CONNECTION-BUDGET GUARD, isolate side: workerd holds ~6 simultaneous + // connections and force-closes the least-recently-used open body past the + // budget — so a test bursting concurrent fetches (bodies never read) must + // not be able to truncate a SIBLING test's response mid-parse. In-flight + // fetches are bounded by a FIFO semaphore (excess QUEUES, nothing is + // refused), and every non-stream body is fully buffered while the slot is + // held, so an unread response can never pin a connection. Event streams + // pass through live. Status/headers/url/bytes are preserved verbatim — + // the suite's assertions are untouched, only wire concurrency is shaped. + let inFlightFetches = 0 + const fetchWaiters = [] + const acquireFetchSlot = () => { + if (inFlightFetches < MAX_CONCURRENT_FETCHES) { + inFlightFetches += 1 + return Promise.resolve() + } + return new Promise((resolve) => fetchWaiters.push(resolve)) + } + const releaseFetchSlot = () => { + const next = fetchWaiters.shift() + if (next !== undefined) next() + else inFlightFetches -= 1 + } globalThis.fetch = async (input, init) => { const url = typeof input === 'string' ? input : String(input && input.url ? input.url : input) const method = ((init && init.method) || (input && input.method) || 'GET').toUpperCase() @@ -264,23 +380,36 @@ export default { violations.push({ url, reason }) throw new Error(reason) } - const res = await realFetch(input, init) - // Every egress rides globalOutbound = the parent gateway; a marked 403 - // is the gateway's refusal. Record it (violation ⇒ fails the run even - // if caught) and THROW — the same shape the local gated fetch gives. - if (res.status === 403) { - const marker = res.headers.get(MARKER) - if (marker !== null) { - let reason = 'refused by the egress gateway' - try { - const body = await res.clone().json() - if (body && typeof body.reason === 'string') reason = body.reason - } catch {} - if (marker === 'violation') violations.push({ url, reason }) - throw new Error(reason) + await acquireFetchSlot() + try { + const res = await realFetch(input, init) + const ctype = (res.headers.get('content-type') || '').toLowerCase() + // A live event stream stays live (buffering would hang to the wall + // breaker); its slot frees on arrival, the stream itself rides on. + if (ctype.indexOf('text/event-stream') !== -1) return res + const buf = res.body === null ? null : await res.arrayBuffer() + // Every egress rides globalOutbound = the parent gateway; a marked 403 + // is the gateway's refusal. Record it (violation ⇒ fails the run even + // if caught) and THROW — the same shape the local gated fetch gives. + if (res.status === 403) { + const marker = res.headers.get(MARKER) + if (marker !== null) { + let reason = 'refused by the egress gateway' + try { + const body = JSON.parse(new TextDecoder().decode(buf)) + if (body && typeof body.reason === 'string') reason = body.reason + } catch {} + if (marker === 'violation') violations.push({ url, reason }) + throw new Error(reason) + } } + const nullBody = res.status === 101 || res.status === 204 || res.status === 205 || res.status === 304 + const out = new Response(nullBody ? null : buf, { status: res.status, statusText: res.statusText, headers: res.headers }) + try { Object.defineProperty(out, 'url', { value: res.url }) } catch {} + return out + } finally { + releaseFetchSlot() } - return res } try { if (DOCUMENT) for (const n of SUBSET_GLOBALS) globalThis[n] = harness.api[n] diff --git a/test/checks.test.ts b/test/checks.test.ts index 6674ee8..f75160a 100644 --- a/test/checks.test.ts +++ b/test/checks.test.ts @@ -212,3 +212,97 @@ describe('content-negotiation grades Accept: application/json (ax-c7m)', () => { expect(grade).toBe('A+') }) }) + +// --------------------------------------------------------------------------- +// keyless-flow sampler robustness (ax: keyed-door-heavy properties). +// +// Modeled on apis.vin: a card declaring MANY own-scope KEYED doors (/keys/me, +// /usage, /dealer/leads — all CORRECTLY 401 a keyless probe) beside a few +// genuinely keyless doors (/listings, /pricing → 200). The old sampler drew 3 +// random candidates and failed when all 3 landed on keyed doors — keyless- +// first held on the surface, the verdict was sampling noise. The fix walks +// candidates (card-hinted keyless doors first, then escalation past the first +// wave up to MAX_KEYLESS_PROBE_TOTAL) until a keyless door is FOUND, and +// fails ONLY when no walked door grants keyless access — escalation widens +// the search, never the acceptance. +// --------------------------------------------------------------------------- + +import { MAX_KEYLESS_PROBES, MAX_KEYLESS_PROBE_TOTAL } from '../src/discovery.js' +import type { Evidence } from '../src/types.js' + +function keyedHeavyRoutes(opts: { keyed: number; keyless: string[]; hintKeyless: boolean }): Routes { + const base = goodTargetRoutes() + const agents = JSON.parse(base['GET /.well-known/agents.json']!({ method: 'GET', accept: '*/*' }).body!) + const paths: Record = {} + const over: Routes = {} + agents.interfaces.http = {} + for (let i = 1; i <= opts.keyed; i++) { + const p = `/api/keyed${String(i).padStart(2, '0')}` + // NO auth hint on the card, NO security in the openapi — exactly the + // under-hinted card shape that made keyed doors look like candidates. + agents.interfaces.http[`keyed${i}`] = { method: 'GET', url: `${GOOD}${p}` } + paths[p] = { get: { responses: { '200': { description: 'ok' } } } } + over[`GET ${p}`] = () => ({ status: 401, contentType: 'application/json', body: '{"error":"key required"}' }) + } + for (const p of opts.keyless) { + agents.interfaces.http[p] = opts.hintKeyless + ? { method: 'GET', url: `${GOOD}${p}`, auth: 'none' } + : { method: 'GET', url: `${GOOD}${p}` } + paths[p] = { get: { responses: { '200': { description: 'ok' } } } } + over[`GET ${p}`] = () => ({ status: 200, contentType: 'application/json', body: '{"items":[{"id":1}]}' }) + } + const openapi = { openapi: '3.1.0', info: { title: 'keyed-heavy', version: '1.0.0' }, paths } + return withOverrides(withoutRoutes(base, 'GET /api/status', 'GET /api/widgets'), { + 'GET /.well-known/agents.json': () => ({ status: 200, contentType: 'application/json', body: JSON.stringify(agents) }), + 'GET /openapi.json': () => ({ status: 200, contentType: 'application/json', body: JSON.stringify(openapi) }), + ...over, + }) +} + +const keylessProbesOf = (bundle: { items: Evidence[] }) => + bundle.items.filter((e) => e.role.startsWith('probe:endpoint:')) + +describe('keyless-flow sampler — robust against keyed-door-heavy cards', () => { + it('(a) many keyed doors + ONE keyless door → PASSES on every seed (escalation finds the keyless door)', async () => { + // 9 keyed + 1 keyless = 10 candidates ≤ MAX_KEYLESS_PROBE_TOTAL: the walk + // is GUARANTEED to reach /api/listings whatever the seed draws first. + for (const seed of [1, 7, 42, 1337]) { + const { bundle, checks } = await judge(keyedHeavyRoutes({ keyed: 9, keyless: ['/api/listings'], hintKeyless: false }), seed) + const c = checks.find((x) => x.id === 'keyless-flow') + expect(c?.verdict, `seed ${seed}: ${c?.detail}`).toBe('pass') + const probes = keylessProbesOf(bundle) + expect(probes.length).toBeGreaterThanOrEqual(MAX_KEYLESS_PROBES) + expect(probes.length).toBeLessThanOrEqual(MAX_KEYLESS_PROBE_TOTAL) + expect(probes.some((p) => p.status === 200)).toBe(true) + } + }) + + it('a card-hinted keyless door (interfaces.http auth:"none") is walked FIRST — resolved inside the cheap wave', async () => { + const { bundle, checks } = await judge(keyedHeavyRoutes({ keyed: 9, keyless: ['/api/listings'], hintKeyless: true })) + expect(checks.find((x) => x.id === 'keyless-flow')?.verdict).toBe('pass') + const probes = keylessProbesOf(bundle) + // Hinted-first ordering: the keyless door succeeded at probe #1, so the + // walk stopped at the ordinary first wave — no escalation spend. + expect(probes.length).toBe(MAX_KEYLESS_PROBES) + expect(probes[0]!.role).toBe('probe:endpoint:GET /api/listings') + expect(probes[0]!.status).toBe(200) + }) + + it('(b) ZERO keyless access still FAILS — escalation walks to the cap, every door refuses, no leniency', async () => { + const { bundle, checks } = await judge(keyedHeavyRoutes({ keyed: 14, keyless: [], hintKeyless: false })) + const c = checks.find((x) => x.id === 'keyless-flow') + expect(c?.verdict).toBe('fail') + expect(c?.detail).toMatch(/no keyless access found/) + expect(c?.detail).toMatch(/at least one declared door must answer 2xx with no key/) + const probes = keylessProbesOf(bundle) + // 14 candidates, but the politeness ceiling bounds the walk. + expect(probes.length).toBe(MAX_KEYLESS_PROBE_TOTAL) + for (const p of probes) expect(p.status).toBe(401) + }) + + it('(b2) a SMALL all-keyed card also fails (both waves exhausted below the caps)', async () => { + const { bundle, checks } = await judge(keyedHeavyRoutes({ keyed: 4, keyless: [], hintKeyless: false })) + expect(checks.find((x) => x.id === 'keyless-flow')?.verdict).toBe('fail') + expect(keylessProbesOf(bundle)).toHaveLength(4) // every candidate walked, all refused + }) +}) diff --git a/test/vitest-subset.test.ts b/test/vitest-subset.test.ts index b602db5..cc0209b 100644 --- a/test/vitest-subset.test.ts +++ b/test/vitest-subset.test.ts @@ -38,10 +38,12 @@ import { type GateViolation, } from '../src/exec/dialect.js' import { + EXEC_MAX_CONCURRENT_FETCHES, GATEWAY_MARKER_HEADER, GATEWAY_RECORD_UNREADABLE, RUNNER_UNAVAILABLE_NO_BINDING, RUNNER_UNAVAILABLE_NO_OUTBOUND, + bufferResponse, buildWorkerCode, createOutboundGateway, gatewayFetch, @@ -868,3 +870,209 @@ it('deterministic', () => { expect(Math.random()).toBeLessThan(1) }) expect(outcome).toEqual({ status: 'failed', reason: GATEWAY_RECORD_UNREADABLE }) }) }) + +// --------------------------------------------------------------------------- +// The CONNECTION BUDGET — one suite test bursting concurrent fetches must +// never corrupt a sibling test's response (ax: burst-vs-sibling isolation). +// +// workerd grants an isolate ~6 simultaneous connections and force-closes the +// least-recently-used OPEN response body past the budget; the reader then sees +// "Response closed due to connection limit". The transport below models +// exactly that failure mode: a body COUNTS AS OPEN until fully read (or +// canceled), and any fetch arriving while the budget is exhausted gets a body +// that truncates mid-stream. Unfixed, a 429 rate-limit burst whose bodies are +// never read leaves the budget pinned and the NEXT test's plain GET-and-parse +// reads truncated non-JSON — the exact apis.vin corruption. +// --------------------------------------------------------------------------- + +interface BudgetedTransport { + fetchImpl: (url: string, init?: RequestInit) => Promise + stats: { readonly open: number; readonly peakOpen: number; readonly truncated: number } +} + +function budgetedTransport( + routes: Record { status: number; contentType?: string; body: string }>, + budget = 6, +): BudgetedTransport { + let open = 0 + let peakOpen = 0 + let truncated = 0 + const fetchImpl = async (url: string): Promise => { + const route = + routes[new URL(url).pathname] ?? + ((): { status: number; contentType?: string; body: string } => ({ status: 404, body: '{"error":"not found"}' })) + const { status, contentType, body } = route() + const bytes = new TextEncoder().encode(body) + const headers = { 'content-type': contentType ?? 'application/json' } + if (open >= budget) { + // Budget exhausted: the runtime closes this response's body mid-flight. + truncated += 1 + const stream = new ReadableStream({ + start(c) { + c.enqueue(bytes.slice(0, Math.max(1, Math.floor(bytes.length / 2)))) + c.error(new Error('Response closed due to connection limit')) + }, + }) + return new Response(stream, { status, headers }) + } + open += 1 + peakOpen = Math.max(peakOpen, open) + let settled = false + const settle = () => { + if (!settled) { + settled = true + open -= 1 + } + } + // The body counts as an OPEN connection until fully read or canceled. + const stream = new ReadableStream({ + start(c) { + c.enqueue(bytes) + }, + pull(c) { + c.close() + settle() + }, + cancel() { + settle() + }, + }) + return new Response(stream, { status, headers }) + } + return { + fetchImpl, + stats: { + get open() { + return open + }, + get peakOpen() { + return peakOpen + }, + get truncated() { + return truncated + }, + }, + } +} + +describe('connection budget — a bursting test cannot corrupt a sibling test (hosted path)', () => { + const BURST_SUITE = ` +it('rate limit burst', async () => { + const responses = await Promise.all(Array.from({ length: 20 }, () => fetch('${ORIGIN}/burst'))) + expect(responses).toHaveLength(20) + expect(responses.every((r) => r.status === 429)).toBeTruthy() +}) +it('listings keyless ok', async () => { + const r = await fetch('${ORIGIN}/listings') + expect(r.status).toBe(200) + const body = await r.json() + expect(body.items).toHaveLength(2) +}) +` + + const LISTINGS_OK = () => ({ + status: 200, + body: JSON.stringify({ items: [{ vin: '1HGCM82633A004352' }, { vin: '1HGCM82633A004353' }] }), + }) + + it('test A bursts 20 concurrent fetches (bodies never read), test B still reads VALID JSON — both pass', async () => { + const transport = budgetedTransport({ + '/burst': () => ({ status: 429, body: '{"type":"RATE_LIMIT"}' }), + '/listings': LISTINGS_OK, + }) + const gateway = createOutboundGateway(transport.fetchImpl) + const outcome = await workerLoaderExecRunner(simulatedLoader(gateway), { outbound: gateway }).run( + req({ testsSource: BURST_SUITE }), + ) + expect(outcome.status).toBe('ran') + if (outcome.status === 'ran') { + for (const r of outcome.results) expect(r.status, `${r.name}: ${r.reason ?? ''}`).toBe('pass') + } + // The runner never exceeded the isolate budget and never pinned an unread + // body: the burst was shaped + drained, not passed through as 20 open wires. + expect(transport.stats.peakOpen).toBeLessThanOrEqual(EXEC_MAX_CONCURRENT_FETCHES) + expect(transport.stats.truncated).toBe(0) + expect(transport.stats.open).toBe(0) + }) + + it('the guard is NOT lenient: a listings body that is GENUINELY non-JSON still fails test B by parse error', async () => { + const transport = budgetedTransport({ + '/burst': () => ({ status: 429, body: '{"type":"RATE_LIMIT"}' }), + '/listings': () => ({ status: 200, contentType: 'text/html', body: 'not json' }), + }) + const gateway = createOutboundGateway(transport.fetchImpl) + const outcome = await workerLoaderExecRunner(simulatedLoader(gateway), { outbound: gateway }).run( + req({ testsSource: BURST_SUITE }), + ) + expect(outcome.status).toBe('ran') + if (outcome.status === 'ran') { + const burst = outcome.results.find((r) => r.name === 'rate limit burst') + const listings = outcome.results.find((r) => r.name === 'listings keyless ok') + expect(burst?.status).toBe('pass') + expect(listings?.status).toBe('fail') // the target's own defect, reported truthfully + } + }) + + it('REGRESSION SHAPE: an unshaped passthrough of the same burst DOES corrupt the sibling under the budget model', async () => { + // Pin that the transport model actually reproduces the failure the guard + // exists for: without shaping/buffering, 20 unread burst bodies pin the + // budget and the sibling's read truncates. (Raw transport, no runner.) + const transport = budgetedTransport({ + '/burst': () => ({ status: 429, body: '{"type":"RATE_LIMIT"}' }), + '/listings': LISTINGS_OK, + }) + const burst = await Promise.all(Array.from({ length: 20 }, () => transport.fetchImpl(`${ORIGIN}/burst`))) + expect(burst.every((r) => r.status === 429)).toBe(true) + const listings = await transport.fetchImpl(`${ORIGIN}/listings`) + await expect(listings.json()).rejects.toThrow(/connection limit/) + }) + + it('gateway bounds in-flight upstream fetches to EXEC_MAX_CONCURRENT_FETCHES (excess queues, none refused)', async () => { + let inFlight = 0 + let peak = 0 + const slowFetch = async (url: string): Promise => { + inFlight += 1 + peak = Math.max(peak, inFlight) + await new Promise((r) => setTimeout(r, 5)) + inFlight -= 1 + return new Response('{"ok":true}', { status: 200, headers: { 'content-type': 'application/json' } }) + } + const gateway = createOutboundGateway(slowFetch) + const responses = await Promise.all( + Array.from({ length: 20 }, () => gateway.fetch(new Request('https://public.example/x'))), + ) + expect(responses).toHaveLength(20) + for (const r of responses) expect(r.status).toBe(200) + expect(peak).toBeLessThanOrEqual(EXEC_MAX_CONCURRENT_FETCHES) + }) + + it('bufferResponse consumes the wire immediately and preserves status/statusText/headers/url/bytes', async () => { + const transport = budgetedTransport({ '/listings': LISTINGS_OK }) + const raw = await transport.fetchImpl(`${ORIGIN}/listings`) + expect(transport.stats.open).toBe(1) + const buffered = await bufferResponse(raw) + expect(transport.stats.open).toBe(0) // the connection freed BEFORE anyone reads the body + expect(buffered.status).toBe(200) + expect(buffered.headers.get('content-type')).toBe('application/json') + const body = (await buffered.json()) as { items: unknown[] } + expect(body.items).toHaveLength(2) + }) + + it('bufferResponse leaves a live text/event-stream STREAMING (no hang, no buffering)', async () => { + // An SSE body never ends; buffering it would hang to the wall breaker. + let controller!: ReadableStreamDefaultController + const live = new ReadableStream({ + start(c) { + controller = c + c.enqueue(new TextEncoder().encode('data: {"tick":1}\n\n')) + }, + }) + const res = new Response(live, { status: 200, headers: { 'content-type': 'text/event-stream' } }) + const out = await bufferResponse(res) // must resolve promptly — the identity, not a buffer + expect(out).toBe(res) + const reader = out.body!.getReader() + const first = await reader.read() + expect(new TextDecoder().decode(first.value)).toContain('"tick":1') + controller.close() + }) +})