diff --git a/docs/project/2026-10-24-codeapi-sustained-load-plan.md b/docs/project/2026-10-24-codeapi-sustained-load-plan.md new file mode 100644 index 00000000..d0941238 --- /dev/null +++ b/docs/project/2026-10-24-codeapi-sustained-load-plan.md @@ -0,0 +1,89 @@ +# CodeAPI sustained STG load and caller fairness + +## Approval summary + +Approval mode: executable batch. The user agreed to the next sustained-load package, asked for a persisted plan and discussion with Opus 5.5, then authorized execution. This package extends the existing qualification harness and measures mixed work over a longer interval using the existing staging deployment and capacity. It changes qualification tools, tests and documentation; it changes no deployed service semantics. + +Run bounded STG verification after source checks and reviews. Begin with the existing direct API/artifact contract, then exercise three eight-minute plateaus with one, two and four requests offered every ten seconds. Use two fixed synthetic callers with equal offered work, deterministic Python CPU work and Python/Node memory-integrity work. Measure each plateau and caller separately, bind results to deployed images/configuration and fresh telemetry, and retain failed or incomplete coverage honestly. This measures the offered workload; it does not find maximum capacity or establish an availability objective. + +Authority includes source implementation, local checks/reviews/commits, ordinary task-branch push and draft PR delivery, bounded synthetic STG verification, the existing read-only signing scope, existing-context observations, owned Service forwards and owned synthetic artifact cleanup. Automatic unchanged retries, principal rotation, production traffic, deployment, capacity/rate-limit adjustment, fault injection and secret-policy changes are excluded. A diagnosed qualification-tool defect may receive an in-scope correction and necessary verification after focused tests, review and fresh preflight, retaining the same per-invocation workload, bounds and acceptance criteria. Preserve each invocation separately. Spot interruption and a new consumer trust integration remain later packages. A passing STG run does not authorize a production repetition. + +Done means the reviewed source package is delivered as a draft and STG verification either satisfies every finite acceptance criterion or preserves a precise blocker, with admitted work settled and owned cleanup accounted for. A failure holds later plateaus. Recovery here means stopping admission and cleaning only owned artifacts and local processes; no disruptive live repair follows. + +## Execution details + +### Working context and owners + +Target: fork main. Frozen source baseline: `8c7916ccd10699c7d18ff82a03996b452706ae8c`. Branch `rs/codeapi-sustained-load`, task worktree `trees/storage-reliability-investigation/trees/codeapi-sustained-load`. Artifact root remains `docs/project/`. All actual environment observations, identities, limits, measurements, approvals and operational receipts belong only in ignored `docs/project/_local/reviews/2026-10-24-sustained-load/`. Public source/plan/PR text describes intended behavior and source verification only. + +Execution mode: standard. Main owns authority, the shared-client authentication/budget seam, secrets, live execution, observer adaptation, integration and final proof. One bounded source executor implements the settled scheduler and behavioral tests after planning approval; it receives source-only scope. Native planner challenges the complete frozen plan; Opus 5.5 discusses methodology before execution and performs integrated final source review. Existing source and observer evidence is reused only where contracts remain unchanged. No new runtime dependency, service, toolchain or test framework is needed. No browser contract changes; screenshots do not apply. No ADR threshold is met. + +| Item | Exclusive execution owner | Dependency and acceptance | +| ------------------------------------- | ----------------------------------------------- | ---------------------------------------------------------------------------------------------- | +| Plan, contracts and methodology | Main with read-only planner and Opus discussion | Frozen draft, source-supported limits, explicit finite coverage and stop conditions | +| Existing client budget/auth extension | Main | Explicit sustained envelope, default envelope unchanged, one shared ledger across both callers | +| Paced scheduler and tests | Bounded source executor | Approved plan; only existing qualification module/test paths; deterministic offline fixtures | +| Observer, signing and STG invocation | Main | Source checks/reviews, exact scopes/current context, fresh preflight and scoped authority | +| Integrated review and draft delivery | Main with configured read-only roles | Immutable complete source range, terminal acceptance/cleanup evidence and private findings | + +### Source changes and invariant contracts + +Only modify `scripts/qualification/codeapi.mjs`, `scripts/qualification/codeapi.test.mjs`, and this plan. Existing CI already executes the qualification suite. Keep transport confinement, header/profile fencing, one-MiB bounds, no redirects/retries, receipt privacy, guest outcome checks, ownership tracking and cleanup unchanged. Existing contract/all/resources modes retain their original hard bounds and behavior. + +Add explicit `sustained` mode (`--sustained` for the operator CLI). Select a fixed sustained envelope inside the existing client: at most 400 HTTP requests, 360 execution-route calls, 1080000 requested guest milliseconds, sixteen owned files and four requests in flight. One client owns the entire ledger across both callers. Reserve pending signing/dispatch attempts in the same global in-flight count before awaiting credentials. Bound credential acquisition to at most one second and the remaining whole-request/slot deadline; a pending or unresolved supplier must never hold settlement indefinitely. Recheck stop state, monotonic invocation/slot/request deadlines and fresh guard immediately before transport dispatch. Race losers cannot dispatch later. Slot admission deadlines can only shorten the fixed envelope. All request time, including signing, fits the original sixty-second ordinary/fifteen-second cleanup bound. Do not create independent per-caller budgets or expose arbitrary raised limits. Add an optional per-execution auth supplier that reuses the existing call-level supplier and defaults to the original caller. Preserve owning suppliers in a private in-memory lookup for every returned object, including an unexpected output from caller B; cleanup uses the original owner. Supplier functions, tokens and auth values never enter receipts. Require both fixed suppliers in sustained mode; absence blocks before contract traffic. + +Whole invocation: 1800 seconds. Stop ordinary admission at 1620 seconds, reserve forty HTTP requests and the final 180 seconds for cleanup, and settle already admitted requests under the existing sixty-second response deadline. Cleanup requests retain fifteen-second deadlines and the invocation bound. Reserve before dispatch, fail closed on exhausted global budgets, and preserve fail/stop behavior across concurrent calls. A source regression must not widen the original modes. + +### Workload and timing + +Prepare three deterministic workload templates once before timed admission: the existing million-step Python SHA-256 chain, 32-MiB Python memory-integrity program and 32-MiB Node memory-integrity program. Separate a fixed workload seed from each request nonce. Extend the existing resource-workload helper with an optional seed defaulting to the nonce, preserving existing modes. Compute each template digest once from its fixed seed. Bind an inexpensive unique echoed nonce `runId:plateau:slot:caller:position` into each generated program and verify it exactly alongside the fixed digest. This detects swapped responses without expensive per-request oracle work. Preparation happens outside plateau timing and cannot silently extend the invocation deadline. Do not compute expensive per-request oracles on the timed dispatch path. + +The operator imports `runQualification` with fresh in-memory 240-second JWT supplier functions for each caller; it does not use constant CLI bearer values for this run, lengthen token lifetimes or save token files. The generic CLI still requires credentials valid for its full invocation and reports an authentication failure honestly if they expire. Run the existing full direct contract under caller A with caller B performing its ownership-denial check. Wait the existing full execution-rate window after the contract. Preflight refuses load unless the deployed ordinary per-principal execution window has an execution window of exactly 30000 ms and permits at least twelve requests in that window and the client can complete all fixed intervals within its admission bound. Do not change or evade those limits. Both principals are fixed for the complete invocation, not rotated. Baseline trust remains the already-approved issuer/profile. + +Each plateau lasts at least 480000 ms and has 48 slots spaced 10000 ms apart. Plateau one offers one request per slot (48 total), alternating caller A/B. Plateau two offers two requests per slot (96 total), one per caller. Plateau four offers four requests per slot (192 total), two per caller. The order of caller submission alternates by slot to reduce systematic first-submitter bias. Before reserving any wave, require current global in flight plus wave demand to be at most four and each caller in flight plus its wave demand to be at most two. Globally cap in flight at four and per caller at two. The offered plateau level is a burst ceiling, not a claim of continuously occupied execution slots. + +Choose workload kinds in a fixed cyclic mix `[cpu, memory, node_memory, cpu]` across the offered calls, balanced between callers within each plateau. For one-request slots, alternate callers and select kind at index `floor(slot / 2) % 4`, forming balanced eight-slot blocks. For two-request slots, each caller receives kind at index `slot % 4`. For four-request slots, each caller receives kinds at indices `(2 * slot) % 4` and `(2 * slot + 1) % 4`. Alternate caller submission order each slot. Each caller receives CPU/memory/Node-memory counts 12/6/6, 24/12/12 and 48/24/24 across the three plateaus. Receipt samples contain plateau, slot, caller label, kind, language, guest CPU/wall time and client elapsed time. Validate exact nonce/digest/byte count, successful guest status, empty stderr and no unexpected generated artifacts. + +Schedule from a monotonic clock. A slot more than 1000 ms late or still blocked by unsettled prior requests is incomplete coverage: stop admission, settle admitted work, preserve the cause and run cleanup. Do not catch up, bunch slots, retry calls or replace missing coverage with defaults. This is a finite offered-load coverage criterion, not a uniform latency SLO: higher offered waves can exhaust the shared in-flight budget sooner. Report the effective overlap and blocking condition per plateau. Every offered request must settle before that plateau's absolute end; the final slot has ten seconds of runway. A longer request can be an honest workload/transport limitation rather than proof of a service outage. The finite timing contract remains explicit. + +Wait through every complete plateau, including its final observation-only interval. Require a new guard observation timestamped at or after each plateau's end within ten seconds, then start the next plateau. Failed/missing interval proof holds dependent plateaus. Before any load admission, require actual elapsed time plus the 31000-ms rate wait plus three times 490000 ms to fit the 1620000-ms ordinary-admission bound. The clock for invocation and scheduler deadlines is monotonic; wall anchors are used only to match the same-host observer timestamps. + +Attach a rejection handler immediately to every dispatched request. Any failure stops new admission; retain all outstanding promises and await their settlement before cleanup begins. Failure receipts carry plateau, slot, caller and workload kind. Cleanup cannot race still-admitted work. + +Expected added load is 336 executions; the full existing contract contributes its unchanged ledger. Expected complete path: 348 execution-route calls, 362 HTTP requests, 1032500 requested guest milliseconds and three owned files. This stays below the sustained hard envelope. No file generation is added to the repeated workload; the existing contract owns all test objects. + +### Observations and acceptance + +Reuse the reviewed STG observer and in-memory JWK signer with only the longer duration, current baseline and output bindings adapted. Use the existing `codeapi-stg` profile and only `CODEAPI_JWT_PRIVATE_JWK_JSON`; no profile change or additional read is needed. Main verifies that exact current contract without printing/saving keys or tokens. Sign fresh short-lived tokens for stable caller A/B under the existing issuer/audience/profile; never inject a signing key into the source client, browser, repository, CI or receipts. + +Use existing cluster connectivity only; report an unavailable user tunnel. Owned Service forwards are temporary operator transport. Before admission, verify the actual pinned-source limiter key and whether rejected requests count; both fixed synthetic principals must select distinct principal-based execution buckets, with adequate shared-IP/global margin if a separate limiter exists. Any ordinary sustained-phase 429 stops admission without retry. Before admission, bind actual image digests, Argo producing revisions/UID/spec, effective nonsecret timeout/concurrency/rate/memory settings, replica owners and current KEDA bounds privately. Require the established configuration, original worker/runner ready, no restarts, healthy nodes, fresh CPU/memory observations and empty queues/sandbox activity. Freeze source/images/configuration through the invocation. Preserve one replica owner; do not scale, patch, restart, drain, evict or alter Argo to manufacture preconditions. + +Observe approximately every 2.5 seconds, batching the same independent reads with at most four workers, with the existing UUID-bound guard expiring after twenty seconds. Maintain original workload UID/readiness/restarts and image checks, node Ready/no MemoryPressure, resource freshness and fresh waiting/active queues. More than twenty-four waiting jobs immediately or more than eight for twenty seconds stops admission. Natural retirement of a previously observed additional runner is allowed only under the reviewed exact Deployment/ReplicaSet/template/HPA desired reduction, ready floor and multi-cycle revalidation; original-worker/runner turnover remains a stop, not an interruption-resilience test. Attribute sandbox activity samples to current serving runner identities. Require the original runner’s metric presence and freshness throughout. Exclude a retained sample only for a separately proven retired additional runner, with a frozen timestamp after absence and a bounded lookback. Unknown identities and missing or stale original/serving observations fail closed. Retain periodic health samples through the full final plateau. + +Pass requires all contract cases, exactly 48/96/192 successful on-time offered requests, equal complete caller allocation, all exact results, all three full intervals and boundary observations, global/per-caller concurrency and hard budgets honored, no unexpected 429/HTTP/guest failure or cleanup/receipt uncertainty. For each plateau and each original worker/runner container, require available CPU plus memory observations with underlying sample age at most sixty seconds in at least 87 of 96 fixed five-second bins; missing bins remain missing. These are freshness-availability bins, not independent source samples. Record observed source timestamps, distinct sample counts and effective resolution privately; do not describe the observer cadence as independent metric production. Record scale/topology changes per plateau and interpret that topology with the measurement. + +Record existing synthetic-canary overlap and possible unrelated activity using scoped aggregate queue/active measurements, explicitly allowing scrape lag and avoiding causal attribution or wider principal/content discovery. Report offered/admitted/completed/failed counts, client p50/p95/p99, achieved throughput, guest timings, per-caller results and latency distributions, resource ranges, queue observations and topology per plateau. Also report observed mean client occupancy as summed client durations divided by plateau duration; achieved throughput at this fixed arrival rate is not a capacity estimate. Equal offered/success counts establish allocation and completion, not scheduler fairness or an SLA. Do not infer per-job queue wait from client-minus-guest time, a speedup without a comparable baseline, leak freedom from ranges, or maximum capacity from the highest offered level. No threshold invented here is a production SLO. + +After the client has settled admitted work and completed its cleanup, allow at most 120 seconds for read-only final acceptance and owned-process/forward shutdown. No additional CodeAPI request or replay is permitted in that period. Begin the queue-drain deadline at client terminal settlement/cleanup completion and require fresh empty queues/sandbox activity within its first sixty seconds. Final acceptance also requires original workloads healthy without new restarts, unchanged image/Argo/KEDA bindings and fresh successful owner-bound producing storage canary Jobs/logs for both applicable STG canaries whose completion timestamp is at most twenty minutes old at the observation time; this is an evidence-age requirement, not permission to wait twenty minutes. A missing qualifying canary after the bounded 120-second post-run window blocks acceptance. Verify every owned object deletion and absence; stop/verify only owned forwards and child processes. Preserve historical resources, all prior evidence and the existing user connection. + +### Feature-wide test portfolio + +| Consequential risk | Test obligation and primary seam | Observable protection | +| -------------------------------------------------------------------- | ----------------------------------------------- | -------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | +| Long mode silently widens old safety limits or splits caller budgets | Extend existing client budget tests | Original envelopes stay exact; sustained global accounting and deadlines cannot be bypassed by caller choice; delayed/unresolved signing cannot dispatch late or after failure and cannot indefinitely block cleanup | +| Caller identity leaks, changes midway or contaminates metrics | Extend existing synthetic HTTP protocol fixture | Two fixed suppliers are used, default auth preserved, equal labeled offers and no credentials in receipts | +| Incomplete intervals, catch-up or missed slots appear successful | Extend existing fake-clock scheduler tests | Full plateau/boundary proof; late/blocked/overrun slots stop later admission without replay | +| Wrong work, errors or failed cleanup pass under sustained mode | Extend existing full-flow and failure controls | Exact workload validation, 429/error stop, owned cleanup and honest terminal partial/failure | +| Desired state is confused with measured runtime | No new maintained test: scoped live proof | Exact producing images/configuration, mandatory telemetry, every plateau and final health/canaries | + +### Delivery and later decisions + +Commit this reviewed plan before implementation. Reuse the existing resource-result validation through a module-level helper rather than duplicating it. Derive envelope selection from the validated mode after test-only options, preserve lowering-only caps, and test the exact expected full ledger. Main lands the client/auth/budget seam, then the assigned source executor owns the scheduler/test additions in the same two existing paths. Serialize writes; no concurrent writer touches those paths. Use existing repository toolchains and container-based dependency checks; host Git/gh remain outside containers. Run the fourteen original offline cases unchanged, extend only consequential behaviors above, and run focused formatting/syntax/source checks. Commit the coherent source, apply simplifier/risk review to the immutable slice, disposition findings and complete Opus 5.5 integrated final review before live admission and completed delivery. + +Run a finite STG invocation after all gates. A failed or incomplete invocation stops admission, settles work and accounts for owned cleanup. Preserve failed evidence; never join partial runs into acceptance. Diagnose qualification-tool defects before any corrected verification; reuse unaffected source evidence and review the changed observer predicates. An unchanged replay is excluded. Return to the user only for a material scope, behavior, authority, risk, cost or data-boundary change. Persist operational findings only privately, update generic source Progress, ordinary-push the task branch and create one draft PR with required CI/status recorded honestly. Marking ready, merge, image publication, production repetition/deployment, interruption drills and new consumer identities remain outside this package. + +### Progress + +Source package complete and reviewed in [draft PR #47](https://github.com/uzh-bf/code-interpreter/pull/47). The plan preceded implementation and native planning approval followed the Opus 5.5 methodology discussion. The extended pinned-Node24 suite passes nineteen cases; simplification and static-assertion corrections pass their affected checks. Scoped Opengrep has no findings or errors. The private observer cadence correction preserves the five-second freshness-availability bins and all acceptance thresholds; the same Opus 5.5 review session verified its bounded read concurrency and unchanged safety predicates. Dedicated risk review found no defects; integrated Opus 5.5 review passed for the complete package and operator code. Hosted CI passed all fourteen jobs at source-qualified head `132ec2f856a26371bdfbaf38a272977cd041cb6c`. This documentation-only progress update leaves the qualified source unchanged. + +The operational continuation, actual acceptance result, measurements and cleanup accounting belong exclusively in the existing ignored review directory. Consult those private receipts before any later live invocation or release decision; source/CI completion alone does not imply environment acceptance. Existing capacity-contract and production-promotion packages remain separate history. Merge, publication, deployments, production repetition and interruption drills retain their named authority boundaries. diff --git a/scripts/qualification/codeapi.mjs b/scripts/qualification/codeapi.mjs index ee860ea3..c64c9ce5 100644 --- a/scripts/qualification/codeapi.mjs +++ b/scripts/qualification/codeapi.mjs @@ -15,6 +15,12 @@ const CAPS = Object.freeze({ files: 16, guestMs: 120000, }); +const SUSTAINED_CAPS = Object.freeze({ + requests: 400, + executions: 360, + files: 16, + guestMs: 1080000, +}); export class QualificationError extends Error { constructor(reason, status) { @@ -212,18 +218,28 @@ export function createClient({ now = Date.now, request = exchange, receiptHealthy = () => true, + monotonic = now, + envelope = 'bounded', }) { const base = apiBase(baseUrl); check(UUID.test(runId), 'invalid_run_id'); check(typeof token === 'function', 'token_supplier_required'); + check(['bounded', 'sustained'].includes(envelope), 'invalid_envelope'); + const maximum = envelope === 'sustained' ? SUSTAINED_CAPS : CAPS; + const totalMs = envelope === 'sustained' ? 1800000 : 900000; + const admissionMs = envelope === 'sustained' ? 1620000 : 720000; const cap = Object.fromEntries( - Object.entries(CAPS).map(([k, v]) => [k, Math.min(v, limits[k] ?? v)]) + Object.entries(maximum).map(([k, v]) => [ + k, + Math.min(v, limits[k] ?? v), + ]) ); check( Object.values(cap).every(v => Number.isSafeInteger(v) && v >= 0), 'invalid_budget' ); - const start = now(); + const start = monotonic(); + const objectOwners = new Map(); const state = { requests: 0, executions: 0, @@ -234,6 +250,7 @@ export function createClient({ stopped: false, peak: 0, inFlight: 0, + dispatches: 0, }; const fail = error => { state.stopped = true; @@ -249,13 +266,33 @@ export function createClient({ }; async function call( path, - { cleanup = false, guestMs = 0, auth = token, ...options } = {} + { + cleanup = false, + guestMs = 0, + auth = token, + dispatchBy = Infinity, + ...options + } = {} ) { + const requestEnd = Math.min( + monotonic() + (cleanup ? 15000 : 60000), + start + totalMs + ); + const dispatchCheck = () => { + check(!state.stopped || cleanup, 'admission_stopped'); + check( + monotonic() - start < (cleanup ? totalMs : admissionMs), + 'run_deadline' + ); + check(monotonic() <= dispatchBy, 'dispatch_deadline'); + check(monotonic() < requestEnd, 'request_deadline'); + }; check(!state.stopped || cleanup, 'admission_stopped'); check(cleanup || receiptHealthy(), 'receipt_unavailable'); - check(now() - start < (cleanup ? 900000 : 720000), 'run_deadline'); + dispatchCheck(); if (!cleanup) await guardCheck(guardPath, runId, now()); - check(!state.stopped || cleanup, 'admission_stopped'); + dispatchCheck(); + check(state.inFlight < 4, 'concurrency_budget'); check( state.requests < (cleanup ? cap.requests : Math.max(0, cap.requests - 40)), @@ -273,23 +310,62 @@ export function createClient({ } state.requests++; const began = now(); - const credential = auth === null ? undefined : await auth(); - check( - auth === null || - (typeof credential === 'string' && credential.length > 0), - 'credential_unavailable' - ); - check(state.inFlight < 4, 'concurrency_budget'); state.inFlight++; state.peak = Math.max(state.peak, state.inFlight); + let dispatched = false; try { + const credential = + auth === null + ? undefined + : await new Promise((resolve, reject) => { + const timer = setTimeout( + () => + reject( + new QualificationError( + 'credential_deadline' + ) + ), + Math.max( + 0, + Math.min( + 1000, + requestEnd - monotonic(), + dispatchBy - monotonic() + ) + ) + ); + Promise.resolve() + .then(auth) + .then( + value => { + clearTimeout(timer); + resolve(value); + }, + () => { + clearTimeout(timer); + reject( + new QualificationError( + 'credential_unavailable' + ) + ); + } + ); + }); + check( + auth === null || + (typeof credential === 'string' && credential.length > 0), + 'credential_unavailable' + ); + dispatchCheck(); + if (!cleanup) await guardCheck(guardPath, runId, now()); + dispatchCheck(); + check(cleanup || receiptHealthy(), 'receipt_unavailable'); + dispatched = true; + state.dispatches++; const response = await request(base, path, { ...options, token: credential, - timeoutMs: Math.min( - cleanup ? 15000 : 60000, - 900000 - (now() - start) - ), + timeoutMs: requestEnd - monotonic(), headers: { 'X-CodeAPI-Expected-Profile': 'default', ...options.headers, @@ -315,13 +391,14 @@ export function createClient({ } return response; } catch (error) { - if (path === '/exec' || path === '/upload') state.uncertain = true; + if (dispatched && (path === '/exec' || path === '/upload')) + state.uncertain = true; throw error; } finally { state.inFlight--; } } - const own = (session, id, name) => { + const own = (session, id, name, auth = token) => { if (!ID.test(session ?? '') || !ID.test(id ?? '')) { state.uncertain = true; throw new QualificationError('unknown_output_reference'); @@ -341,6 +418,7 @@ export function createClient({ resource_id: runId, }; state.owned.push(ref); + objectOwners.set(ref, auth); receipt({ type: 'owned_object', id, storage_session_id: session }); check(typeof name === 'string', 'unusable_output_name'); check(state.owned.length <= cap.files, 'file_budget'); @@ -349,10 +427,19 @@ export function createClient({ async function execute( code, lang, - { files = [], timeout = 3000, expectTimeout = false } = {} + { + files = [], + timeout = 3000, + expectTimeout = false, + auth = token, + dispatchBy = Infinity, + } = {} ) { + check(typeof auth === 'function', 'token_supplier_required'); const began = now(); const response = await call('/exec', { + auth, + dispatchBy, method: 'POST', headers: { 'Content-Type': 'application/json' }, body: JSON.stringify({ code, lang, timeout, files }), @@ -381,7 +468,8 @@ export function createClient({ own( file?.storage_session_id ?? value.session_id, file?.id, - file?.name + file?.name, + auth ) ); } catch { @@ -418,18 +506,33 @@ export function createClient({ 'invalid_concurrency' ); let next = 0; + let admissionFailure; await Promise.all( Array.from({ length: concurrency }, async () => { - while (next < items.length && !state.stopped) { + while ( + next < items.length && + !state.stopped && + !admissionFailure + ) { const item = items[next++]; try { await fn(item); } catch (error) { - fail(error); + if ( + error instanceof QualificationError && + [ + 'concurrency_budget', + 'execution_budget', + 'request_budget', + ].includes(error.reason) + ) + admissionFailure ??= error; + else fail(error); } } }) ); + if (admissionFailure) fail(admissionFailure); } async function cleanup() { const objects = [...state.owned]; @@ -443,13 +546,17 @@ export function createClient({ const removed = await call('/files' + path, { method: 'DELETE', cleanup: true, + auth: objectOwners.get(object), }); check( removed.status === 200 || removed.status === 404, 'delete_failed' ); status( - await call('/download' + path, { cleanup: true }), + await call('/download' + path, { + cleanup: true, + auth: objectOwners.get(object), + }), 404 ); receipt({ @@ -705,19 +812,17 @@ async function load(client, waitMs) { const RESOURCE_BYTES = 32 * 1024 * 1024; const HASH_STEPS = 1000000; +const RESOURCE_KINDS = Object.freeze(['cpu', 'memory', 'node_memory']); -export function resourceWorkload(kind, nonce) { +function prepareResourceWorkload(kind, seed) { + check(RESOURCE_KINDS.includes(kind), 'invalid_resource_kind'); check( - ['cpu', 'memory', 'node_memory'].includes(kind), - 'invalid_resource_kind' - ); - check( - typeof nonce === 'string' && /^[a-zA-Z0-9:_-]{1,160}$/.test(nonce), + typeof seed === 'string' && /^[a-zA-Z0-9:_-]{1,160}$/.test(seed), 'invalid_resource_nonce' ); const memory = kind !== 'cpu'; - const pattern = createHash('sha256').update(nonce).digest()[0]; - let value = Buffer.from(nonce); + const pattern = createHash('sha256').update(seed).digest()[0]; + let value = Buffer.from(seed); if (memory) value = Buffer.alloc(RESOURCE_BYTES, pattern); else for (let i = 0; i < HASH_STEPS; i++) @@ -725,12 +830,29 @@ export function resourceWorkload(kind, nonce) { const digest = memory ? createHash('sha256').update(value).digest('hex') : value.toString('hex'); + return { + kind, + lang: kind === 'node_memory' ? 'node' : 'python', + pattern, + digest, + bytes: memory ? RESOURCE_BYTES : 0, + }; +} + +function renderResourceWorkload(template, nonce) { + check( + typeof nonce === 'string' && /^[a-zA-Z0-9:_-]{1,160}$/.test(nonce), + 'invalid_resource_nonce' + ); + const { kind, lang, pattern, digest, bytes } = template; + const memory = kind !== 'cpu'; const literal = JSON.stringify(nonce); const code = kind === 'node_memory' ? `// qualification-resource:node_memory import {createHash} from 'node:crypto'; const nonce=${literal}; +const seed=${JSON.stringify(template.seed)}; const cpu=process.cpuUsage(), wall=performance.now(); const data=Buffer.alloc(${RESOURCE_BYTES},${pattern}); for(let i=0;i 0 && + Number.isFinite(output.wallMs) && + output.wallMs > 0 && + output.wallMs <= 3000 && + result.created.length === 0, + 'resource_output_mismatch' + ); + return { + language: work.lang, + kind, + bytes: work.bytes, + cpuMs: output.cpuMs, + guestWallMs: output.wallMs, + durationMs: result.durationMs, + }; +} + async function resources( client, { waitMs, began, monotonic, pause, guardPath } @@ -775,33 +932,10 @@ async function resources( prepared ?? resourceWorkload(kind, `${client.runId}:${phase}:${index}`); const result = await client.execute(work.code, work.lang); - let output; - try { - output = JSON.parse(result.value.stdout); - } catch { - throw new QualificationError('resource_output_mismatch'); - } - check( - output.nonce === work.nonce && - output.digest === work.digest && - output.bytes === work.bytes && - Number.isFinite(output.cpuMs) && - output.cpuMs > 0 && - Number.isFinite(output.wallMs) && - output.wallMs > 0 && - output.wallMs <= 3000 && - result.created.length === 0, - 'resource_output_mismatch' - ); const item = { phase, index, - language: work.lang, - kind, - bytes: work.bytes, - cpuMs: output.cpuMs, - guestWallMs: output.wallMs, - durationMs: result.durationMs, + ...validateResourceResult(work, kind, result), }; client.receipt({ type: 'resource_sample', ...item }); return item; @@ -886,6 +1020,342 @@ async function resources( }); } +const SUSTAINED_SLOTS = 48; +const SUSTAINED_SPACING_MS = 10000; +const SUSTAINED_PLATEAU_MS = 480000; +const SUSTAINED_BOUNDARY_MS = 10000; +const SUSTAINED_KIND_CYCLE = Object.freeze([ + 'cpu', + 'memory', + 'node_memory', + 'cpu', +]); + +function sustainedWave(burst, slot) { + const callers = slot % 2 === 0 ? ['a', 'b'] : ['b', 'a']; + if (burst === 1) + return [ + { + caller: callers[0], + position: 0, + kind: SUSTAINED_KIND_CYCLE[Math.floor(slot / 2) % 4], + }, + ]; + if (burst === 2) + return callers.map(caller => ({ + caller, + position: 0, + kind: SUSTAINED_KIND_CYCLE[slot % 4], + })); + check(burst === 4, 'invalid_sustained_burst'); + return callers.flatMap(caller => + [0, 1].map(position => ({ + caller, + position, + kind: SUSTAINED_KIND_CYCLE[(2 * slot + position) % 4], + })) + ); +} + +function addAllocation(target, caller, kind) { + target[caller] ??= { cpu: 0, memory: 0, node_memory: 0 }; + target[caller][kind]++; +} + +async function sustained({ + client, + token, + comparisonToken, + waitMs, + began, + monotonic, + pause, + guardPath, +}) { + check( + monotonic() - began + 31000 + 3 * 490000 <= 1620000, + 'sustained_preflight_deadline' + ); + + const preparationBegan = monotonic(); + const templates = Object.fromEntries( + RESOURCE_KINDS.map(kind => { + const seed = `${client.runId}:sustained:template:${kind}`; + return [kind, { ...prepareResourceWorkload(kind, seed), seed }]; + }) + ); + await pause(Math.max(0, waitMs - (monotonic() - preparationBegan))); + + const suppliers = { a: token, b: comparisonToken }; + const reportFailure = (context, error) => { + const detail = + error instanceof QualificationError + ? { + reason: error.reason, + ...(error.status ? { httpStatus: error.status } : {}), + } + : { reason: 'unexpected_failure' }; + client.receipt({ + type: 'sustained_failure', + ...context, + ...detail, + }); + client.fail( + error instanceof QualificationError + ? error + : new QualificationError('unexpected_failure') + ); + }; + + for (const burst of [1, 2, 4]) { + if (client.state.stopped) break; + const phaseBegan = monotonic(), + wallBegan = client.now(), + end = phaseBegan + SUSTAINED_PLATEAU_MS; + const outstanding = new Set(); + const activeByCaller = { a: 0, b: 0 }; + const samples = []; + const scheduledAllocation = { + a: { cpu: 0, memory: 0, node_memory: 0 }, + b: { cpu: 0, memory: 0, node_memory: 0 }, + }; + const successfulAllocation = { + a: { cpu: 0, memory: 0, node_memory: 0 }, + b: { cpu: 0, memory: 0, node_memory: 0 }, + }; + const admittedByCaller = { a: 0, b: 0 }; + const successfulByCaller = { a: 0, b: 0 }; + const failedByCaller = { a: 0, b: 0 }; + let offered = 0, + admitted = 0, + successful = 0, + failed = 0, + attemptedMs = 0; + + client.receipt({ + type: 'sustained_start', + name: `${burst}x`, + plateau: burst, + offered: SUSTAINED_SLOTS * burst, + concurrency: burst, + startedAt: wallBegan, + durationMs: SUSTAINED_PLATEAU_MS, + }); + + const context = (slot, item) => ({ + plateau: burst, + slot, + caller: item.caller, + position: item.position, + order: item.order, + kind: item.kind, + }); + try { + for (let slot = 0; slot < SUSTAINED_SLOTS; slot++) { + if (client.state.stopped) break; + const due = phaseBegan + slot * SUSTAINED_SPACING_MS; + await pause(Math.max(0, due - monotonic())); + if (client.state.stopped) break; + const wave = sustainedWave(burst, slot); + wave.forEach((item, order) => { + item.order = order; + }); + offered += wave.length; + for (const item of wave) + addAllocation(scheduledAllocation, item.caller, item.kind); + if (monotonic() - due > 1000) { + for (const item of wave) + reportFailure( + context(slot, item), + new QualificationError('sustained_slot_late') + ); + break; + } + if ( + outstanding.size + wave.length > 4 || + wave.some( + item => + activeByCaller[item.caller] + + wave.filter(x => x.caller === item.caller) + .length > + 2 + ) + ) { + for (const item of wave) + reportFailure( + context(slot, item), + new QualificationError('sustained_slot_blocked') + ); + break; + } + + for (const item of wave) { + const nonce = `${client.runId}:${burst}:${slot}:${item.caller}:${item.position}`; + const offeredAt = monotonic(); + admitted++; + admittedByCaller[item.caller]++; + activeByCaller[item.caller]++; + const work = renderResourceWorkload( + templates[item.kind], + nonce + ); + const task = client + .execute(work.code, work.lang, { + auth: suppliers[item.caller], + dispatchBy: due + 1000, + }) + .then(async result => { + const measurement = validateResourceResult( + work, + item.kind, + result + ); + const sample = { + plateau: burst, + slot, + caller: item.caller, + position: item.position, + order: item.order, + ...measurement, + }; + samples.push(sample); + successful++; + successfulByCaller[item.caller]++; + addAllocation( + successfulAllocation, + item.caller, + item.kind + ); + client.receipt({ + type: 'sustained_sample', + ...sample, + }); + }) + .catch(error => { + failed++; + failedByCaller[item.caller]++; + reportFailure(context(slot, item), error); + }) + .finally(() => { + attemptedMs += monotonic() - offeredAt; + activeByCaller[item.caller]--; + outstanding.delete(task); + if (monotonic() > end) + reportFailure( + context(slot, item), + new QualificationError( + 'sustained_response_overrun' + ) + ); + }); + outstanding.add(task); + } + } + await Promise.all(outstanding); + } finally { + await Promise.all(outstanding); + } + + while (!client.state.stopped && monotonic() < end) + await pause(Math.max(1, Math.ceil(end - monotonic()))); + const durationMs = monotonic() - phaseBegan; + const perCaller = Object.fromEntries( + ['a', 'b'].map(caller => { + const callerSamples = samples.filter( + sample => sample.caller === caller + ); + return [ + caller, + { + offered: Object.values( + scheduledAllocation[caller] + ).reduce((sum, count) => sum + count, 0), + admitted: admittedByCaller[caller], + successful: successfulByCaller[caller], + failed: failedByCaller[caller], + allocation: scheduledAllocation[caller], + successfulAllocation: successfulAllocation[caller], + p50Ms: percentile( + callerSamples.map(x => x.durationMs), + 0.5 + ), + p95Ms: percentile( + callerSamples.map(x => x.durationMs), + 0.95 + ), + p99Ms: percentile( + callerSamples.map(x => x.durationMs), + 0.99 + ), + }, + ]; + }) + ); + client.receipt({ + type: 'sustained_phase', + name: `${burst}x`, + plateau: burst, + slots: SUSTAINED_SLOTS, + spacingMs: SUSTAINED_SPACING_MS, + burstCeiling: burst, + scheduledDurationMs: SUSTAINED_PLATEAU_MS, + offered, + admitted, + successful, + failed, + startedAt: wallBegan, + endedAt: client.now(), + durationMs, + achievedPerSecond: + durationMs > 0 ? successful / (durationMs / 1000) : 0, + occupancy: durationMs > 0 ? attemptedMs / durationMs : 0, + p50Ms: percentile( + samples.map(x => x.durationMs), + 0.5 + ), + p95Ms: percentile( + samples.map(x => x.durationMs), + 0.95 + ), + p99Ms: percentile( + samples.map(x => x.durationMs), + 0.99 + ), + allocation: scheduledAllocation, + successfulAllocation, + perCaller, + }); + + if (client.state.stopped) break; + const wallEnd = wallBegan + SUSTAINED_PLATEAU_MS; + const boundaryEnd = monotonic() + SUSTAINED_BOUNDARY_MS; + let boundary; + while (monotonic() <= boundaryEnd) { + try { + boundary = await guardCheck( + guardPath, + client.runId, + client.now() + ); + if (boundary.observedAt >= wallEnd) break; + } catch { + boundary = undefined; + } + const remaining = boundaryEnd - monotonic(); + if (remaining <= 0) break; + await pause(Math.min(1000, remaining)); + } + if (!boundary || boundary.observedAt < wallEnd) { + reportFailure( + { plateau: burst, boundary: true }, + new QualificationError('sustained_boundary_missing') + ); + break; + } + client.receipt({ type: 'sustained_boundary', plateau: burst }); + } +} + export async function runQualification({ baseUrl, token, @@ -897,7 +1367,16 @@ export async function runQualification({ waitMs = 31000, ...testing } = {}) { - check(['contract', 'all', 'resources'].includes(mode), 'invalid_mode'); + check( + ['contract', 'all', 'resources', 'sustained'].includes(mode), + 'invalid_mode' + ); + check( + mode !== 'sustained' || + (typeof token === 'function' && + typeof comparisonToken === 'function'), + 'sustained_suppliers_required' + ); check( (!testing.monotonic && !testing.pause) || testing.request, 'test_clock_requires_transport' @@ -949,6 +1428,8 @@ export async function runQualification({ receipt, receiptHealthy: () => !receiptFailed, ...testing, + envelope: mode === 'sustained' ? 'sustained' : 'bounded', + monotonic, }); receipt({ type: 'start', runId, mode }); try { @@ -962,6 +1443,17 @@ export async function runQualification({ pause, guardPath, }); + if (mode === 'sustained') + await sustained({ + client, + token, + comparisonToken, + waitMs, + began, + monotonic, + pause, + guardPath, + }); } catch (error) { client.fail(error); } @@ -1021,7 +1513,9 @@ if ( runId: process.env.CODEAPI_RUN_ID, guardPath: process.env.CODEAPI_GUARD, receiptPath: process.env.CODEAPI_RECEIPT, - mode: process.argv.includes('--resources') + mode: process.argv.includes('--sustained') + ? 'sustained' + : process.argv.includes('--resources') ? 'resources' : process.argv.includes('--load') ? 'all' diff --git a/scripts/qualification/codeapi.test.mjs b/scripts/qualification/codeapi.test.mjs index 59fce646..6d58482f 100644 --- a/scripts/qualification/codeapi.test.mjs +++ b/scripts/qualification/codeapi.test.mjs @@ -1,7 +1,14 @@ import test from 'node:test'; import assert from 'node:assert/strict'; import { createServer } from 'node:http'; -import { mkdtemp, writeFile, readFile, rm, stat } from 'node:fs/promises'; +import { + mkdtemp, + writeFile, + readFile, + rename, + rm, + stat, +} from 'node:fs/promises'; import { tmpdir } from 'node:os'; import { join } from 'node:path'; import { randomUUID } from 'node:crypto'; @@ -221,9 +228,13 @@ async function protocol( deleteFailure = false, rateLimit = false, corruptResource = false, + sustainedRateLimit = null, + sustainedDelay = null, } = {} ) { const objects = new Map(); + const resourceCache = new Map(); + const sustainedRequests = []; let counter = 0, executions = 0, peak = 0, @@ -252,30 +263,73 @@ async function protocol( res.setHeader('Retry-After', '30'); return reply(res, 429, { error: 'rate_limited' }); } + const resource = + /qualification-resource:(cpu|memory|node_memory)/.exec( + input.code + ); + let resourceRequest; + if (resource) { + const nonce = JSON.parse( + /nonce=("[^"\n]+")/.exec(input.code)[1] + ); + const seedMatch = /seed=("[^"\n]+")/.exec(input.code); + const seed = seedMatch ? JSON.parse(seedMatch[1]) : nonce; + const label = /^([^:]+):([124]):(\d+):([ab]):(\d+)$/.exec( + nonce + ); + resourceRequest = { + kind: resource[1], + nonce, + seed, + plateau: label ? Number(label[2]) : null, + slot: label ? Number(label[3]) : null, + caller: label ? label[4] : null, + position: label ? Number(label[5]) : null, + }; + if (label) sustainedRequests.push(resourceRequest); + if ( + sustainedRateLimit && + Object.entries(sustainedRateLimit).every( + ([key, value]) => resourceRequest[key] === value + ) + ) { + res.setHeader('Retry-After', '30'); + return reply(res, 429, { error: 'rate_limited' }); + } + } executions++; active++; peak = Math.max(peak, active); - await delay(2); + const delayed = + sustainedDelay && + resourceRequest && + Object.entries(sustainedDelay).every( + ([key, value]) => + key === 'ms' || resourceRequest[key] === value + ); + await delay(delayed ? sustainedDelay.ms : 2); active--; if (wrongRuntime) { const ref = put('unexpected.csv', Buffer.from('owned')); return reply(res, 200, success('wrong\n', { files: [ref] })); } - const resource = - /qualification-resource:(cpu|memory|node_memory)/.exec( - input.code - ); if (resource) { - const nonce = JSON.parse( - /nonce=("[^"\n]+")/.exec(input.code)[1] - ); - const work = resourceWorkload(resource[1], nonce); + const cacheKey = `${resourceRequest.kind}:${resourceRequest.seed}`; + let work = resourceCache.get(cacheKey); + if (!work) { + work = resourceWorkload( + resourceRequest.kind, + resourceRequest.nonce, + resourceRequest.seed + ); + resourceCache.set(cacheKey, work); + } return reply( res, 200, success( JSON.stringify({ - nonce, + nonce: resourceRequest.nonce, digest: corruptResource ? 'invalid' : work.digest, bytes: work.bytes, cpuMs: 25, @@ -343,12 +397,16 @@ async function protocol( return { baseUrl, objects, + sustainedRequests, get executions() { return executions; }, get peak() { return peak; }, + get active() { + return active; + }, }; } @@ -605,3 +663,477 @@ test('corrupt resource results stop admission and final-idle guard failure canno ); } }); + +test('sustained envelopes are global while legacy caps and deadlines stay fixed', async t => { + const c = await files(t); + let clock = 0; + const response = { + status: 200, + headers: {}, + bytes: Buffer.from(JSON.stringify(success())), + }; + const request = async () => response; + const common = { + ...c, + baseUrl: 'http://127.0.0.1:1/v1', + token: () => 'synthetic', + request, + now: () => Date.now() + clock, + monotonic: () => clock, + }; + + const legacy = createClient({ ...common, limits: { requests: 999 } }); + for (let i = 0; i < 120; i++) await legacy.call('/health'); + await assert.rejects(legacy.call('/health'), { + reason: 'request_budget', + }); + for (let i = 0; i < 40; i++) + await legacy.call('/health', { cleanup: true }); + assert.equal(legacy.state.requests, 160); + await assert.rejects(legacy.call('/health', { cleanup: true }), { + reason: 'request_budget', + }); + + clock = 0; + const legacyClock = createClient({ ...common, limits: { requests: 999 } }); + clock = 720000; + await assert.rejects(legacyClock.call('/health'), { + reason: 'run_deadline', + }); + await legacyClock.call('/health', { cleanup: true }); + clock = 900000; + await assert.rejects(legacyClock.call('/health', { cleanup: true }), { + reason: 'run_deadline', + }); + + clock = 0; + const legacyExec = createClient({ + ...common, + limits: { executions: 999, guestMs: 999999 }, + }); + for (let i = 0; i < 40; i++) + await legacyExec.call('/exec', { + method: 'POST', + body: '{}', + headers: { 'Content-Type': 'application/json' }, + }); + await assert.rejects( + legacyExec.call('/exec', { + method: 'POST', + body: '{}', + headers: { 'Content-Type': 'application/json' }, + }), + { reason: 'execution_budget' } + ); + + clock = 0; + const sustained = createClient({ + ...common, + envelope: 'sustained', + limits: { requests: 999, files: 999 }, + }); + for (let i = 0; i < 360; i++) await sustained.call('/health'); + assert.equal(sustained.state.requests, 360); + await assert.rejects(sustained.call('/health'), { + reason: 'request_budget', + }); + clock = 1620000; + await assert.rejects(sustained.call('/health'), { reason: 'run_deadline' }); + await sustained.call('/health', { cleanup: true }); + clock = 1800000; + await assert.rejects(sustained.call('/health', { cleanup: true }), { + reason: 'run_deadline', + }); + + clock = 0; + const caps = createClient({ + ...common, + envelope: 'sustained', + limits: { files: 999, guestMs: 9999999 }, + }); + await caps.execute('pass', 'python', { timeout: 1080000 }); + assert.equal(caps.state.guestMs, 1080000); + await assert.rejects(caps.execute('pass', 'python'), { + reason: 'execution_budget', + }); + for (let i = 0; i < 16; i++) + caps.own('s'.repeat(21), String(i).padStart(21, 'b'), `file-${i}`); + assert.throws(() => caps.own('s'.repeat(21), 'z'.repeat(21), 'overflow'), { + reason: 'file_budget', + }); + assert.equal(caps.state.owned.length, 17); +}); + +test('bounded suppliers, dispatch checks and caller-B owned cleanup stay explicit', async t => { + const c = await files(t); + let remote = 0; + const client = createClient({ + ...c, + baseUrl: 'http://127.0.0.1:1/v1', + token: () => 'synthetic', + envelope: 'sustained', + monotonic: () => 0, + request: async () => { + remote++; + return { + status: 200, + headers: {}, + bytes: Buffer.from(JSON.stringify(success())), + }; + }, + }); + + await assert.rejects( + client.execute('pass', 'python', { + auth: () => new Promise(() => {}), + dispatchBy: 80, + }), + { reason: 'credential_deadline' } + ); + assert.equal(remote, 0); + assert.equal(client.state.requests, 1); + assert.equal(client.state.executions, 1); + assert.equal(client.state.dispatches, 0); + assert.equal(client.state.inFlight, 0); + + await assert.rejects( + client.execute('pass', 'python', { + auth: async () => { + await delay(40); + return 'late'; + }, + dispatchBy: 10, + }), + { reason: 'credential_deadline' } + ); + assert.equal(remote, 0); + + let release; + let supplierStarted = false; + const racing = client.execute('pass', 'python', { + auth: () => { + supplierStarted = true; + return new Promise(resolve => { + release = resolve; + }); + }, + dispatchBy: 10000, + }); + while (!supplierStarted) await delay(1); + client.stop(); + release('too-late'); + await assert.rejects(racing, { reason: 'admission_stopped' }); + assert.equal(remote, 0); + + const receipts = []; + const ownerCalls = []; + const owner = createClient({ + ...c, + baseUrl: 'http://127.0.0.1:1/v1', + token: () => 'owner-a', + receipt: value => receipts.push(value), + request: async (_base, path, options) => { + if (path === '/exec') { + assert.equal(options.token, 'owner-b'); + return { + status: 200, + headers: {}, + bytes: Buffer.from( + JSON.stringify( + success('', { + files: [ + { + id: 'v'.repeat(21), + name: 'caller-b.csv', + storage_session_id: 's'.repeat(21), + }, + ], + }) + ) + ), + }; + } + ownerCalls.push(options.token); + return { + status: options.method === 'DELETE' ? 200 : 404, + headers: {}, + bytes: Buffer.from('{}'), + }; + }, + }); + const output = await owner.execute('pass', 'python', { + auth: () => 'owner-b', + }); + assert.equal(output.created.length, 1); + await owner.cleanup(); + assert.deepEqual(ownerCalls, ['owner-b', 'owner-b']); + assert.equal(JSON.stringify(receipts).includes('owner-b'), false); +}); + +async function sustainedRun( + t, + { + serverOptions = {}, + lateSlot = false, + holdFirst = false, + holdPlateau = 1, + holdCaller = null, + holdSlots = 1, + missingBoundary = false, + overrun = false, + earlyWake = false, + pauseMs = 20, + } = {} +) { + const c = await files(t), + server = await protocol(t, serverOptions), + epoch = Date.now(); + let elapsed = 0, + delayed = false, + zeroPauses = 0, + overrunAdded = false; + const writeGuard = async observedAt => { + const temporary = `${c.guardPath}.${randomUUID()}`; + await writeFile( + temporary, + JSON.stringify({ runId: c.runId, observedAt, ok: true }) + ); + await rename(temporary, c.guardPath); + }; + const result = await runQualification({ + ...c, + baseUrl: server.baseUrl, + token: () => 'synthetic-secret', + comparisonToken: () => 'other', + mode: 'sustained', + waitMs: 0, + now: () => epoch + elapsed, + monotonic: () => elapsed, + pause: async ms => { + zeroPauses += ms === 0 ? 1 : 0; + await delay(pauseMs); + if (lateSlot && !delayed && ms === 0 && zeroPauses === 2) { + elapsed += 1001; + delayed = true; + } + elapsed += earlyWake && ms > 1 ? ms - 1 : ms; + const missing = + missingBoundary && elapsed >= 480000 && elapsed <= 490000; + await writeGuard(epoch + (missing ? 479999 : elapsed)); + }, + request: async (...args) => { + let input; + try { + input = JSON.parse(args[2]?.body); + } catch {} + const label = + input?.code && + /nonce="[^"]*:(\d+):(\d+):([ab]):\d+"/.exec(input.code); + if ( + holdFirst && + label && + Number(label[1]) === holdPlateau && + (!holdCaller || label[3] === holdCaller) && + Number(label[2]) < holdSlots + ) { + return new Promise(resolve => { + setTimeout(async () => { + resolve(await exchange(...args)); + }, 500); + }); + } + if (overrun && !overrunAdded && input?.code?.includes(':1:47:')) { + elapsed += 11001; + overrunAdded = true; + } + await writeGuard(epoch + elapsed); + const response = await exchange(...args); + if (input?.code?.includes('qualification-resource')) elapsed += 1; + return response; + }, + }); + const events = (await readFile(c.receiptPath, 'utf8')) + .trim() + .split('\n') + .map(JSON.parse); + assert.equal(server.objects.size, 0); + assert.equal(server.active, 0); + assert.equal( + (await readFile(c.receiptPath, 'utf8')).includes('synthetic-secret'), + false + ); + return { result, events, server, elapsed, runId: c.runId }; +} + +test('sustained schedule binds nonces, kinds, callers and the full ledger', async t => { + const { result, events, server, elapsed, runId } = await sustainedRun(t); + assert.equal(result.status, 'passed'); + assert.equal(result.executions, 348); + assert.equal(result.requests, 362); + assert.equal(result.requestedGuestMs, 1032500); + assert.equal(result.ownedObjects, 3); + assert.ok(result.peakRequests <= 4); + assert.equal(elapsed, 1440000); + + const starts = events.filter(x => x.type === 'sustained_start'); + assert.deepEqual( + starts.map(x => [x.name, x.offered, x.concurrency, x.durationMs]), + [ + ['1x', 48, 1, 480000], + ['2x', 96, 2, 480000], + ['4x', 192, 4, 480000], + ] + ); + const phases = events.filter(x => x.type === 'sustained_phase'); + assert.deepEqual( + phases.map(x => [ + x.name, + x.offered, + x.admitted, + x.successful, + x.failed, + x.durationMs, + ]), + [ + ['1x', 48, 48, 48, 0, 480000], + ['2x', 96, 96, 96, 0, 480000], + ['4x', 192, 192, 192, 0, 480000], + ] + ); + for (const phase of phases) { + assert.ok(phase.occupancy > 0 && phase.occupancy <= 4); + for (const caller of ['a', 'b']) { + assert.deepEqual(phase.perCaller[caller].allocation, { + cpu: 12 * phase.plateau, + memory: 6 * phase.plateau, + node_memory: 6 * phase.plateau, + }); + assert.equal( + phase.perCaller[caller].successful, + 24 * phase.plateau + ); + } + } + + const samples = events.filter(x => x.type === 'sustained_sample'); + assert.equal(samples.length, 336); + assert.equal(server.sustainedRequests.length, 336); + assert.equal(new Set(server.sustainedRequests.map(x => x.nonce)).size, 336); + assert.equal(new Set(server.sustainedRequests.map(x => x.seed)).size, 3); + for (const request of server.sustainedRequests) { + assert.ok(request.nonce.startsWith(`${runId}:`)); + assert.match( + request.nonce.slice(runId.length + 1), + /^[124]:\d+:[ab]:\d+$/ + ); + } + const requestKinds = new Map( + server.sustainedRequests.map(x => [ + `${x.plateau}:${x.slot}:${x.caller}:${x.position}`, + x.kind, + ]) + ); + for (const sample of samples) { + assert.equal( + requestKinds.get( + `${sample.plateau}:${sample.slot}:${sample.caller}:${sample.position}` + ), + sample.kind + ); + assert.ok(sample.cpuMs > 0 && sample.guestWallMs > 0); + } + for (const phase of phases) { + const slots = Map.groupBy( + samples.filter(x => x.plateau === phase.plateau), + x => x.slot + ); + for (const [slot, wave] of slots) { + const ordered = wave.sort( + (left, right) => left.order - right.order + ); + assert.equal(ordered.length, phase.plateau); + assert.equal(ordered[0].caller, slot % 2 === 0 ? 'a' : 'b'); + } + } +}); + +test('a sustained failure stops admission but waits for concurrent settlement before cleanup', async t => { + const { result, events, server } = await sustainedRun(t, { + serverOptions: { + sustainedRateLimit: { plateau: 2, slot: 3, caller: 'a' }, + sustainedDelay: { + plateau: 2, + slot: 3, + caller: 'b', + ms: 80, + }, + }, + pauseMs: 30, + }); + assert.equal(result.status, 'failed'); + assert.ok(result.failures.some(x => x.reason === 'rate_limited')); + assert.equal(server.active, 0); + assert.equal(server.objects.size, 0); + const failure = events.find(x => x.type === 'sustained_failure'); + assert.deepEqual( + { + plateau: failure.plateau, + slot: failure.slot, + caller: failure.caller, + kind: failure.kind, + reason: failure.reason, + httpStatus: failure.httpStatus, + }, + { + plateau: 2, + slot: 3, + caller: 'a', + kind: 'cpu', + reason: 'rate_limited', + httpStatus: 429, + } + ); + const phases = events.filter(x => x.type === 'sustained_phase'); + assert.deepEqual( + phases.map(x => [x.plateau, x.successful, x.failed]), + [ + [1, 48, 0], + [2, 7, 1], + ] + ); +}); + +test('sustained intervals wait through an early timer wake before reporting duration', async t => { + const { result, events } = await sustainedRun(t, { earlyWake: true }); + assert.equal(result.status, 'passed'); + const phases = events.filter(x => x.type === 'sustained_phase'); + assert.equal(phases.length, 3); + assert.ok(phases.every(x => x.durationMs >= 480000)); +}); + +test('late, blocked, missing-boundary and overrun intervals stop without catch-up', async t => { + for (const [options, reason, phases] of [ + [{ lateSlot: true }, 'sustained_slot_late', 1], + [ + { + holdFirst: true, + holdPlateau: 2, + holdCaller: 'a', + holdSlots: 2, + }, + 'sustained_slot_blocked', + 2, + ], + [{ holdFirst: true, holdPlateau: 4 }, 'sustained_slot_blocked', 3], + [{ missingBoundary: true }, 'sustained_boundary_missing', 1], + [{ overrun: true }, 'sustained_response_overrun', 1], + ]) { + const { result, events } = await sustainedRun(t, options); + assert.equal(result.status, 'failed'); + assert.ok(result.failures.some(x => x.reason === reason)); + assert.equal( + events.filter(x => x.type === 'sustained_phase').length, + phases + ); + } +});