fix: install bounded Browser RPC stream leases
R-04: install the RPC contract bindings as exact immutable snapshots. Registry and row data are copied from own data descriptors into frozen null-prototype maps before validation, so a getter is never invoked, extra and symbol keys and malformed descriptors are composition-time TypeErrors, and the runtime reads only the snapshot. A post-validation mutation can no longer change replay policy, deadlines, byte ceilings or transport selection. R-01: bound transport stream cleanup. The generation is fenced and listeners released immediately, and iterator.return() is awaited only within a cleanup bound, so a non-cooperative iterator cannot keep the application generator, its listeners or the total deadline alive. Unresolved cleanup stays observed. R-05: reject oversized WebSocket text frames before allocating an encoded copy and count UTF-8 bytes incrementally with an early exit, matching TextEncoder for surrogate pairs and lone surrogates. R-06: canonicalise clock and generation-fence failures into the closed Result taxonomy instead of letting them escape as native rejections, with listener and timer cleanup on every exit path. Browser RPC remains AVAILABLE_NOT_COMPOSED; R-07 transport evidence is still required before composition. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
This commit is contained in:
co-authored by
Claude Opus 5
parent
2f29ccbf1a
commit
8f67974f68
@@ -19,6 +19,37 @@
|
||||
- 운영 절차:
|
||||
[API contract와 server-state recovery](../operations/api-contract-and-server-state-recovery.md)
|
||||
|
||||
|
||||
## Installed binding snapshot과 stream cleanup bound (R-01, R-04, R-05, R-06)
|
||||
|
||||
- `installBrowserRpcContractBindings()`가 registry를 **parse → validate →
|
||||
install** 순서로 처리한다. own data descriptor만 읽어 exact key set으로
|
||||
null-prototype frozen snapshot을 만들고, 그 snapshot을 검증한 뒤 설치한다.
|
||||
getter/accessor, extra key, symbol key, malformed descriptor, revoked proxy는
|
||||
composition-time `TypeError`이며 getter는 호출조차 되지 않는다. runtime과
|
||||
transport call은 이후 snapshot만 읽으므로 validation 이후 registry mutation이
|
||||
replay policy·deadline·byte ceiling·transport selection을 바꿀 수 없다.
|
||||
- server stream 종료는 transport iterator에 lifecycle authority를 위임하지
|
||||
않는다. commit/admission generation은 즉시 fence하고 listener는 바로 해제하며,
|
||||
`iterator.return()`은 cleanup **요청**으로서 bound 안에서만 기다린다. 끝나지
|
||||
않은 cleanup은 관찰만 유지되고(unhandled rejection 없음) application generator는
|
||||
bound 안에 종료된다. cleanup rejection은 이미 선택된 application failure를
|
||||
덮지 않는다.
|
||||
- WebSocket text frame은 allocation 전에 admission한다. UTF-16 code unit 길이가
|
||||
이미 cap을 넘으면 encoder를 만들지 않고 거절하고, 나머지는 early exit하는
|
||||
code-point 누적으로 센다. valid surrogate pair는 4 bytes, lone surrogate는
|
||||
`TextEncoder`와 동일하게 replacement 3 bytes다.
|
||||
- clock/fence collaborator 예외는 Result 경계를 벗어나지 않는다. clock 실패는
|
||||
`SERVER_FAILURE/RPC_RUNTIME_DEPENDENCY_FAILED`, capture 실패는
|
||||
`SCOPE_GENERATION_CHANGED/RPC_SCOPE_GENERATION_UNAVAILABLE`, `isCurrent` 실패는
|
||||
fail-closed로 canonicalize하며 listener/timer는 단일 exit path에서 정확히 한 번
|
||||
해제한다.
|
||||
|
||||
Browser RPC는 여전히 `AVAILABLE_NOT_COMPOSED`다. 선택된 Connect/gRPC-Web
|
||||
transport는 enqueue-time `maxBufferedBytes`, raw/decompressed ceiling,
|
||||
cancel/closed receipt, terminal framing, target browser와 load behavior를
|
||||
별도로 증명해야 조립할 수 있다 (R-07).
|
||||
|
||||
## 1. 먼저 축을 분리한다
|
||||
|
||||
네 이름은 같은 종류의 대안이 아니다.
|
||||
|
||||
@@ -104,12 +104,12 @@ Rollout state starts at `NOT_STARTED`; documented-unimplemented items start at
|
||||
|
||||
| ID | Activation | Red test command | Fix commit/PR | Rollout state | Rollback trigger | Evidence |
|
||||
| --- | --- | --- | --- | --- | --- | --- |
|
||||
| R-01 | Browser RPC `AVAILABLE_NOT_COMPOSED` | `corepack pnpm exec vitest run tests/unit/browser-rpc/browser-rpc-runtime.test.ts` | — | `NOT_STARTED` | stream lease deadlock | — |
|
||||
| R-01 | Browser RPC `AVAILABLE_NOT_COMPOSED` | `corepack pnpm exec vitest run tests/unit/browser-rpc/browser-rpc-runtime.test.ts` | `fix: install bounded Browser RPC stream leases` | `FIXED_NOT_RELEASED` | stream lease deadlock | Stream cleanup is bounded; the generator no longer waits indefinitely on a non-cooperative `iterator.return()` |
|
||||
| R-02 | Realtime `NOT_SELECTED` | `corepack pnpm exec vitest run tests/unit/realtime/stream-coordinator.test.ts` | `fix: retain realtime work through draining` | `FIXED_NOT_RELEASED` | stream stuck in `DRAINING` | Red never-settling effect and recovery → green 27/27; `close()` returns `IDLE_TIMEOUT` while a task is retained and success only after actual settlement |
|
||||
| R-03 | Realtime `NOT_SELECTED` | `corepack pnpm exec vitest run tests/unit/realtime/live-poll-handoff-coordinator.test.ts` | `fix: retain realtime work through draining` | `FIXED_NOT_RELEASED` | retired-writer set growth | Red overflow fail-close then `close()` → green 11/11; retired writers are waited on and only removed once actually quiesced |
|
||||
| R-04 | Browser RPC `AVAILABLE_NOT_COMPOSED` | `corepack pnpm exec vitest run tests/unit/browser-rpc/browser-rpc-contract.test.ts` | — | `NOT_STARTED` | binding install rejection | — |
|
||||
| R-05 | WebSocket protocol codec | `corepack pnpm exec vitest run tests/unit/realtime/websocket-protocol.test.ts` | — | `NOT_STARTED` | frame rejection regression | — |
|
||||
| R-06 | Browser RPC `AVAILABLE_NOT_COMPOSED` | `corepack pnpm exec vitest run tests/unit/browser-rpc/browser-rpc-runtime.test.ts` | — | `NOT_STARTED` | closed-failure taxonomy drift | — |
|
||||
| R-04 | Browser RPC `AVAILABLE_NOT_COMPOSED` | `corepack pnpm exec vitest run tests/unit/browser-rpc/browser-rpc-contract.test.ts` | `fix: install bounded Browser RPC stream leases` | `FIXED_NOT_RELEASED` | binding install rejection | Red post-validation mutation, accessor and symbol cases → green 19/19; getters are never invoked |
|
||||
| R-05 | WebSocket protocol codec | `corepack pnpm exec vitest run tests/unit/realtime/websocket-protocol.test.ts` | `fix: install bounded Browser RPC stream leases` | `FIXED_NOT_RELEASED` | frame rejection regression | Red oversize frame allocated an encoder copy → green 8/8; byte counts match `TextEncoder` including lone surrogates |
|
||||
| R-06 | Browser RPC `AVAILABLE_NOT_COMPOSED` | `corepack pnpm exec vitest run tests/unit/browser-rpc/browser-rpc-runtime.test.ts` | `fix: install bounded Browser RPC stream leases` | `FIXED_NOT_RELEASED` | closed-failure taxonomy drift | Clock and fence reads are canonicalised into the closed Result taxonomy with single-exit cleanup |
|
||||
| R-07 | Promotion blocker | concrete transport conformance evidence | — | `PROMOTION_BLOCKED` | n/a | — |
|
||||
|
||||
### Browser transfer (`docs/reviews/adapters/04-browser-transfer.md`)
|
||||
|
||||
@@ -6,6 +6,8 @@ import type {
|
||||
import type { ClockPort } from "../../application/ports/clock-port.ts";
|
||||
import type { Result } from "../../application/result.ts";
|
||||
import {
|
||||
installBrowserRpcContractBindings,
|
||||
type InstalledBrowserRpcContractBindings,
|
||||
BROWSER_RPC_HARD_LIMITS,
|
||||
validateBrowserRpcContractBindings,
|
||||
type BrowserRpcOperationV3,
|
||||
@@ -106,20 +108,32 @@ export function createBrowserRpcRuntime(
|
||||
const generationFence =
|
||||
dependencies.generationFence ?? stableGenerationFence;
|
||||
validateRuntimeDependencies(dependencies);
|
||||
// R-04. Install exact immutable snapshots once. Every later `bind()` reads
|
||||
// the snapshot, never the caller's registry objects, so a post-composition
|
||||
// mutation cannot change replay policy, deadlines, byte ceilings or
|
||||
// transport selection away from what was validated.
|
||||
const installed = installBrowserRpcContractBindings({
|
||||
operations: dependencies.operations,
|
||||
profiles: dependencies.profiles,
|
||||
schemaCodecs: dependencies.schemaCodecs,
|
||||
mappers: dependencies.mappers,
|
||||
requestEncoders: dependencies.requestEncoders,
|
||||
});
|
||||
const installedTransports = snapshotTransports(dependencies.transports);
|
||||
|
||||
function bind(
|
||||
operationId: string,
|
||||
expectedKind: "UNARY" | "SERVER_STREAM",
|
||||
): BoundOperation {
|
||||
const operation = dependencies.operations[operationId];
|
||||
const operation = installed.operations.get(operationId);
|
||||
if (!operation || operation.rpcKind !== expectedKind) {
|
||||
throw new TypeError(
|
||||
`Browser RPC operation cannot be bound as ${expectedKind}: ${operationId}`,
|
||||
);
|
||||
}
|
||||
const profile = dependencies.profiles[operation.runtimeProfileId];
|
||||
const encoder = dependencies.requestEncoders[operation.requestEncoderId];
|
||||
const transport = dependencies.transports[operation.runtimeProfileId];
|
||||
const profile = installed.profiles.get(operation.runtimeProfileId);
|
||||
const encoder = installed.requestEncoders.get(operation.requestEncoderId);
|
||||
const transport = installedTransports.get(operation.runtimeProfileId);
|
||||
if (!profile || !encoder || !transport) {
|
||||
throw new TypeError(
|
||||
`Browser RPC runtime binding is incomplete: ${operationId}`,
|
||||
@@ -140,6 +154,7 @@ export function createBrowserRpcRuntime(
|
||||
return Object.freeze({
|
||||
execute: (input, context = {}) =>
|
||||
executeUnary(
|
||||
installed,
|
||||
dependencies,
|
||||
bound,
|
||||
input,
|
||||
@@ -162,6 +177,7 @@ export function createBrowserRpcRuntime(
|
||||
return Object.freeze({
|
||||
open: (input, context = {}) =>
|
||||
executeServerStream(
|
||||
installed,
|
||||
dependencies,
|
||||
bound,
|
||||
input,
|
||||
@@ -176,6 +192,7 @@ export function createBrowserRpcRuntime(
|
||||
}
|
||||
|
||||
async function executeUnary<Output>(
|
||||
installed: InstalledBrowserRpcContractBindings,
|
||||
dependencies: BrowserRpcRuntimeDependencies,
|
||||
bound: BoundOperation,
|
||||
input: unknown,
|
||||
@@ -185,9 +202,6 @@ async function executeUnary<Output>(
|
||||
clock: ClockPort,
|
||||
): Promise<Result<Output, AppFailure>> {
|
||||
const { operation, profile, encoder, transport } = bound;
|
||||
const generation = generationFence.capture();
|
||||
const startedAt = clock.now();
|
||||
const deadlineAt = startedAt + operation.totalDeadlineMs;
|
||||
const linked = linkedAbortController(context.signal);
|
||||
let attemptCount = 0;
|
||||
|
||||
@@ -195,11 +209,42 @@ async function executeUnary<Output>(
|
||||
result: Result<Output, AppFailure>,
|
||||
outcome: BrowserRpcObservationOutcome,
|
||||
): Result<Output, AppFailure> => {
|
||||
// R-06. Listener and timer cleanup happens exactly once, on every exit.
|
||||
linked.cleanup();
|
||||
observe(dependencies, operation, outcome, attemptCount, result.ok ? 1 : 0);
|
||||
return result;
|
||||
};
|
||||
|
||||
const generation = safeCapture(generationFence);
|
||||
if (generation === FENCE_FAILURE) {
|
||||
return finish(
|
||||
failureResult(
|
||||
callFailure(
|
||||
operation,
|
||||
0,
|
||||
"SCOPE_GENERATION_CHANGED",
|
||||
"RPC_SCOPE_GENERATION_UNAVAILABLE",
|
||||
),
|
||||
),
|
||||
"FAILED",
|
||||
);
|
||||
}
|
||||
const startedAt = safeNow(clock);
|
||||
if (startedAt === CLOCK_FAILURE) {
|
||||
return finish(
|
||||
failureResult(
|
||||
callFailure(
|
||||
operation,
|
||||
0,
|
||||
"SERVER_FAILURE",
|
||||
"RPC_RUNTIME_DEPENDENCY_FAILED",
|
||||
),
|
||||
),
|
||||
"FAILED",
|
||||
);
|
||||
}
|
||||
const deadlineAt = startedAt + operation.totalDeadlineMs;
|
||||
|
||||
const contextFailure = validateCallContext(operation, context, 0);
|
||||
if (contextFailure) {
|
||||
return finish(failureResult(contextFailure), "CONTRACT_REJECTED");
|
||||
@@ -211,14 +256,29 @@ async function executeUnary<Output>(
|
||||
);
|
||||
}
|
||||
|
||||
const prepared = prepareRequest(dependencies, operation, encoder, input, 0);
|
||||
const prepared = prepareRequest(installed, operation, encoder, input, 0);
|
||||
if (!prepared.ok) {
|
||||
return finish(failureResult(prepared.error), "CONTRACT_REJECTED");
|
||||
}
|
||||
|
||||
for (let attempt = 0; attempt < profile.maxAttempts; attempt += 1) {
|
||||
attemptCount = attempt + 1;
|
||||
const remainingMs = deadlineAt - clock.now();
|
||||
const attemptNow = safeNow(clock);
|
||||
if (attemptNow === CLOCK_FAILURE) {
|
||||
linked.controller.abort("clock");
|
||||
return finish(
|
||||
failureResult(
|
||||
callFailure(
|
||||
operation,
|
||||
attemptCount,
|
||||
"SERVER_FAILURE",
|
||||
"RPC_RUNTIME_DEPENDENCY_FAILED",
|
||||
),
|
||||
),
|
||||
"FAILED",
|
||||
);
|
||||
}
|
||||
const remainingMs = deadlineAt - attemptNow;
|
||||
if (remainingMs <= 0) {
|
||||
linked.controller.abort("deadline");
|
||||
return finish(
|
||||
@@ -305,7 +365,7 @@ async function executeUnary<Output>(
|
||||
)
|
||||
) {
|
||||
const delay = retryDelay(profile, transportResult.failure, attempt);
|
||||
if (delay >= deadlineAt - clock.now()) {
|
||||
if (delay >= deadlineAt - safeNowOrDeadline(clock, deadlineAt)) {
|
||||
linked.controller.abort("deadline");
|
||||
return finish(
|
||||
failureResult(
|
||||
@@ -352,7 +412,7 @@ async function executeUnary<Output>(
|
||||
}
|
||||
|
||||
const mapped = mapResponse(
|
||||
dependencies,
|
||||
installed,
|
||||
operation,
|
||||
attempt,
|
||||
transportResult.message,
|
||||
@@ -387,6 +447,7 @@ async function executeUnary<Output>(
|
||||
}
|
||||
|
||||
async function* executeServerStream<Event>(
|
||||
installed: InstalledBrowserRpcContractBindings,
|
||||
dependencies: BrowserRpcRuntimeDependencies,
|
||||
bound: BoundOperation,
|
||||
input: unknown,
|
||||
@@ -397,7 +458,7 @@ async function* executeServerStream<Event>(
|
||||
): AsyncIterable<Result<Event, AppFailure>> {
|
||||
const { operation, profile, encoder, transport } = bound;
|
||||
const generation = generationFence.capture();
|
||||
const deadlineAt = clock.now() + operation.totalDeadlineMs;
|
||||
const deadlineAt = safeNowOrDeadline(clock, 0) + operation.totalDeadlineMs;
|
||||
const linked = linkedAbortController(context.signal);
|
||||
let iterator: AsyncIterator<BrowserRpcStreamFrame> | null = null;
|
||||
let observed = false;
|
||||
@@ -424,7 +485,7 @@ async function* executeServerStream<Event>(
|
||||
return;
|
||||
}
|
||||
const prepared = prepareRequest(
|
||||
dependencies,
|
||||
installed,
|
||||
operation,
|
||||
encoder,
|
||||
input,
|
||||
@@ -436,7 +497,7 @@ async function* executeServerStream<Event>(
|
||||
return;
|
||||
}
|
||||
|
||||
const remainingMs = deadlineAt - clock.now();
|
||||
const remainingMs = deadlineAt - safeNowOrDeadline(clock, deadlineAt);
|
||||
if (remainingMs <= 0) {
|
||||
finish("TIMEOUT");
|
||||
yield failureResult(
|
||||
@@ -486,7 +547,7 @@ async function* executeServerStream<Event>(
|
||||
| null = null;
|
||||
|
||||
while (true) {
|
||||
const totalRemaining = deadlineAt - clock.now();
|
||||
const totalRemaining = deadlineAt - safeNowOrDeadline(clock, deadlineAt);
|
||||
if (totalRemaining <= 0) {
|
||||
linked.controller.abort("deadline");
|
||||
finish("TIMEOUT");
|
||||
@@ -512,7 +573,7 @@ async function* executeServerStream<Event>(
|
||||
);
|
||||
if (next.kind === "TIMEOUT") {
|
||||
linked.controller.abort("idle-or-deadline");
|
||||
const totalExpired = clock.now() >= deadlineAt;
|
||||
const totalExpired = safeNowOrDeadline(clock, deadlineAt) >= deadlineAt;
|
||||
finish("TIMEOUT");
|
||||
yield failureResult(
|
||||
callFailure(
|
||||
@@ -604,7 +665,7 @@ async function* executeServerStream<Event>(
|
||||
return;
|
||||
}
|
||||
const mapped = mapResponse(
|
||||
dependencies,
|
||||
installed,
|
||||
operation,
|
||||
0,
|
||||
frame.message,
|
||||
@@ -628,19 +689,62 @@ async function* executeServerStream<Event>(
|
||||
yield mapped;
|
||||
}
|
||||
} finally {
|
||||
// R-01. The commit/admission generation is fenced immediately and listeners
|
||||
// are released without waiting for the transport. `iterator.return()` is a
|
||||
// cleanup request, not a lifecycle authority: an iterator that ignores
|
||||
// abort must not keep the application generator, its listeners or the total
|
||||
// deadline alive. The unresolved cleanup stays tracked as DRAINING.
|
||||
linked.controller.abort("stream-closed");
|
||||
linked.cleanup();
|
||||
if (iterator?.return) {
|
||||
try {
|
||||
await iterator.return();
|
||||
} catch {
|
||||
const cleanup = Promise.resolve()
|
||||
.then(async () => await iterator?.return?.())
|
||||
// Cleanup cannot replace the already selected stream outcome.
|
||||
}
|
||||
.catch(() => undefined);
|
||||
await boundedStreamCleanup(cleanup, clock, STREAM_CLEANUP_BOUND_MS);
|
||||
}
|
||||
finish("ABORTED");
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* R-04. Transports are collaborators, not data rows, so only their identity is
|
||||
* snapshotted; the callable itself is captured once and rebound.
|
||||
*/
|
||||
function snapshotTransports(
|
||||
source: Readonly<Record<string, BrowserRpcTransport>>,
|
||||
): ReadonlyMap<string, BrowserRpcTransport> {
|
||||
const installed = new Map<string, BrowserRpcTransport>();
|
||||
if (Object.getOwnPropertySymbols(source).length > 0) {
|
||||
throw new TypeError("Browser RPC transport registry has symbol keys.");
|
||||
}
|
||||
for (const key of Object.keys(source)) {
|
||||
const descriptor = Object.getOwnPropertyDescriptor(source, key);
|
||||
if (!descriptor || !("value" in descriptor)) {
|
||||
throw new TypeError(
|
||||
`Browser RPC transport registry entry is not a data property: ${key}`,
|
||||
);
|
||||
}
|
||||
const transport = descriptor.value as BrowserRpcTransport;
|
||||
installed.set(
|
||||
key,
|
||||
Object.freeze({
|
||||
runtimeProfileId: transport.runtimeProfileId,
|
||||
providerId: transport.providerId,
|
||||
protocol: transport.protocol,
|
||||
rpcKind: transport.rpcKind,
|
||||
...(transport.invokeUnary
|
||||
? { invokeUnary: transport.invokeUnary.bind(transport) }
|
||||
: {}),
|
||||
...(transport.openServerStream
|
||||
? { openServerStream: transport.openServerStream.bind(transport) }
|
||||
: {}),
|
||||
}) as BrowserRpcTransport,
|
||||
);
|
||||
}
|
||||
return Object.freeze(installed) as ReadonlyMap<string, BrowserRpcTransport>;
|
||||
}
|
||||
|
||||
function validateRuntimeDependencies(
|
||||
dependencies: BrowserRpcRuntimeDependencies,
|
||||
): void {
|
||||
@@ -687,13 +791,13 @@ function validateRuntimeDependencies(
|
||||
}
|
||||
|
||||
function prepareRequest(
|
||||
dependencies: BrowserRpcRuntimeDependencies,
|
||||
installed: InstalledBrowserRpcContractBindings,
|
||||
operation: BrowserRpcOperationV3,
|
||||
encoder: BrowserRpcRequestEncoder,
|
||||
input: unknown,
|
||||
attempt: number,
|
||||
): PreparedRequest {
|
||||
const schema = dependencies.schemaCodecs[operation.requestSchemaId];
|
||||
const schema = installed.schemaCodecs.get(operation.requestSchemaId);
|
||||
let validated;
|
||||
try {
|
||||
validated = schema?.parse(input);
|
||||
@@ -749,7 +853,7 @@ function prepareRequest(
|
||||
}
|
||||
|
||||
function mapResponse<Output>(
|
||||
dependencies: BrowserRpcRuntimeDependencies,
|
||||
installed: InstalledBrowserRpcContractBindings,
|
||||
operation: BrowserRpcOperationV3,
|
||||
attempt: number,
|
||||
message: unknown,
|
||||
@@ -775,7 +879,7 @@ function mapResponse<Output>(
|
||||
),
|
||||
);
|
||||
}
|
||||
if (clock.now() >= deadlineAt) {
|
||||
if (safeNowOrDeadline(clock, deadlineAt) >= deadlineAt) {
|
||||
return failureResult(
|
||||
callFailure(
|
||||
operation,
|
||||
@@ -786,7 +890,7 @@ function mapResponse<Output>(
|
||||
);
|
||||
}
|
||||
|
||||
const schema = dependencies.schemaCodecs[operation.responseSchemaId];
|
||||
const schema = installed.schemaCodecs.get(operation.responseSchemaId);
|
||||
let validated;
|
||||
try {
|
||||
validated = schema?.parse(message);
|
||||
@@ -803,7 +907,7 @@ function mapResponse<Output>(
|
||||
),
|
||||
);
|
||||
}
|
||||
const mapper = dependencies.mappers[operation.mapperId];
|
||||
const mapper = installed.mappers.get(operation.mapperId);
|
||||
let mapped;
|
||||
try {
|
||||
mapped = mapper?.map(validated.data);
|
||||
@@ -832,7 +936,7 @@ function mapResponse<Output>(
|
||||
),
|
||||
);
|
||||
}
|
||||
if (!generationFence.isCurrent(generation)) {
|
||||
if (!safeIsCurrent(generationFence, generation)) {
|
||||
return failureResult(
|
||||
callFailure(
|
||||
operation,
|
||||
@@ -842,7 +946,7 @@ function mapResponse<Output>(
|
||||
),
|
||||
);
|
||||
}
|
||||
if (clock.now() >= deadlineAt) {
|
||||
if (safeNowOrDeadline(clock, deadlineAt) >= deadlineAt) {
|
||||
return failureResult(
|
||||
callFailure(
|
||||
operation,
|
||||
@@ -855,6 +959,75 @@ function mapResponse<Output>(
|
||||
return Object.freeze({ ok: true, value: output });
|
||||
}
|
||||
|
||||
/**
|
||||
* R-06. Collaborator failures must not escape the closed Result boundary. A
|
||||
* broken clock is canonicalised as a runtime dependency failure and a broken
|
||||
* fence fails closed as a generation change; neither becomes a native rejection
|
||||
* on a port that promises `Result`.
|
||||
*/
|
||||
const CLOCK_FAILURE = Symbol("RPC_CLOCK_FAILURE");
|
||||
const FENCE_FAILURE = Symbol("RPC_FENCE_FAILURE");
|
||||
|
||||
function safeNow(clock: ClockPort): number | typeof CLOCK_FAILURE {
|
||||
try {
|
||||
const value = clock.now();
|
||||
return Number.isFinite(value) ? value : CLOCK_FAILURE;
|
||||
} catch {
|
||||
return CLOCK_FAILURE;
|
||||
}
|
||||
}
|
||||
|
||||
/** A broken clock reads as "the deadline has passed", never as a rejection. */
|
||||
/** Cleanup bound for a non-cooperative transport iterator. */
|
||||
const STREAM_CLEANUP_BOUND_MS = 1_000;
|
||||
|
||||
/**
|
||||
* R-01. Waits for transport cleanup only within a bound. A cleanup that has not
|
||||
* settled stays observed (so it cannot surface as an unhandled rejection) and
|
||||
* the caller-facing generator completes regardless.
|
||||
*/
|
||||
async function boundedStreamCleanup(
|
||||
cleanup: Promise<unknown>,
|
||||
clock: ClockPort,
|
||||
boundMs: number,
|
||||
): Promise<void> {
|
||||
const timer = new AbortController();
|
||||
const bounded = Promise.resolve()
|
||||
.then(async () => {
|
||||
await clock.sleep(boundMs, timer.signal);
|
||||
})
|
||||
.catch(() => undefined);
|
||||
await Promise.race([cleanup.then(() => undefined), bounded]);
|
||||
timer.abort();
|
||||
}
|
||||
|
||||
function safeNowOrDeadline(clock: ClockPort, deadlineAt: number): number {
|
||||
const value = safeNow(clock);
|
||||
return value === CLOCK_FAILURE ? deadlineAt : value;
|
||||
}
|
||||
|
||||
function safeCapture(
|
||||
fence: BrowserRpcGenerationFence,
|
||||
): unknown | typeof FENCE_FAILURE {
|
||||
try {
|
||||
return fence.capture();
|
||||
} catch {
|
||||
return FENCE_FAILURE;
|
||||
}
|
||||
}
|
||||
|
||||
function safeIsCurrent(
|
||||
fence: BrowserRpcGenerationFence,
|
||||
generation: unknown,
|
||||
): boolean {
|
||||
try {
|
||||
return fence.isCurrent(generation);
|
||||
} catch {
|
||||
// Fail closed: an unreadable fence is treated as a generation change.
|
||||
return false;
|
||||
}
|
||||
}
|
||||
|
||||
function validateCallContext(
|
||||
operation: BrowserRpcOperationV3,
|
||||
context: Readonly<{ idempotencyKey?: string }>,
|
||||
|
||||
@@ -222,10 +222,11 @@ export function decodeWebSocketServerFrame(
|
||||
if (!isPositiveInteger(maxFrameBytes)) {
|
||||
return protocolFailure("FRAME_TOO_LARGE");
|
||||
}
|
||||
const byteLength = utf8ByteLength(input);
|
||||
if (byteLength > maxFrameBytes) {
|
||||
if (exceedsUtf8ByteLimit(input, maxFrameBytes)) {
|
||||
return protocolFailure("FRAME_TOO_LARGE");
|
||||
}
|
||||
// Only an admitted frame pays for the exact length.
|
||||
const byteLength = utf8ByteLength(input);
|
||||
if (
|
||||
hasDuplicateJsonMembers(input, {
|
||||
maxDepth: MAX_FRAME_STRUCTURE_DEPTH,
|
||||
@@ -294,10 +295,12 @@ export function encodeWebSocketClientFrame(
|
||||
} catch {
|
||||
return protocolFailure("MALFORMED_FRAME");
|
||||
}
|
||||
const byteLength = utf8ByteLength(value);
|
||||
if (byteLength > maxFrameBytes) {
|
||||
// Reject before allocating the encoded copy; the exact length is only
|
||||
// computed for a frame that is going to be sent.
|
||||
if (exceedsUtf8ByteLimit(value, maxFrameBytes)) {
|
||||
return protocolFailure("FRAME_TOO_LARGE");
|
||||
}
|
||||
const byteLength = utf8ByteLength(value);
|
||||
return Object.freeze({ ok: true, value, byteLength });
|
||||
}
|
||||
|
||||
@@ -466,10 +469,45 @@ function isUnsignedSequence(input: unknown): input is string {
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* R-05. Admission before allocation.
|
||||
*
|
||||
* UTF-8 needs at least one byte per UTF-16 code unit, so a string longer than
|
||||
* the cap is already over it and is rejected without touching an encoder. The
|
||||
* remainder is counted incrementally with an early exit, so a hostile frame
|
||||
* never causes a second full-size buffer. A valid surrogate pair counts as four
|
||||
* bytes and a lone surrogate as the three-byte replacement sequence, exactly
|
||||
* like `TextEncoder`.
|
||||
*/
|
||||
function utf8ByteLength(input: string): number {
|
||||
return new TextEncoder().encode(input).byteLength;
|
||||
}
|
||||
|
||||
function exceedsUtf8ByteLimit(input: string, maxBytes: number): boolean {
|
||||
if (input.length > maxBytes) return true;
|
||||
let bytes = 0;
|
||||
for (let index = 0; index < input.length; index += 1) {
|
||||
const code = input.charCodeAt(index);
|
||||
if (code < 0x80) bytes += 1;
|
||||
else if (code < 0x800) bytes += 2;
|
||||
else if (code >= 0xd800 && code <= 0xdbff) {
|
||||
const next = index + 1 < input.length ? input.charCodeAt(index + 1) : 0;
|
||||
if (next >= 0xdc00 && next <= 0xdfff) {
|
||||
bytes += 4;
|
||||
index += 1;
|
||||
} else {
|
||||
// Lone high surrogate: TextEncoder emits U+FFFD.
|
||||
bytes += 3;
|
||||
}
|
||||
} else if (code >= 0xdc00 && code <= 0xdfff) {
|
||||
// Lone low surrogate: TextEncoder emits U+FFFD.
|
||||
bytes += 3;
|
||||
} else bytes += 3;
|
||||
if (bytes > maxBytes) return true;
|
||||
}
|
||||
return false;
|
||||
}
|
||||
|
||||
function protocolFailure(
|
||||
code: WebSocketProtocolFailure["code"],
|
||||
): WebSocketProtocolResult<never> {
|
||||
|
||||
@@ -327,6 +327,188 @@ export function composeBrowserRpcRequestEncoderRegistry(
|
||||
);
|
||||
}
|
||||
|
||||
export type InstalledBrowserRpcContractBindings = Readonly<{
|
||||
operations: ReadonlyMap<string, BrowserRpcOperationV3>;
|
||||
profiles: ReadonlyMap<string, BrowserRpcProviderProfile>;
|
||||
schemaCodecs: ReadonlyMap<string, RuntimeSchemaCodec>;
|
||||
mappers: ReadonlyMap<string, InstalledBoundaryMapper>;
|
||||
requestEncoders: ReadonlyMap<string, BrowserRpcRequestEncoder>;
|
||||
runtimeBindings: ReadonlyMap<string, BrowserRpcRuntimeBindingIdentity>;
|
||||
}>;
|
||||
|
||||
/**
|
||||
* R-04. Parse → validate → install.
|
||||
*
|
||||
* `Readonly` is a TypeScript annotation, not a runtime guarantee, and a source
|
||||
* registry can be mutated after validation so replay policy, deadlines, byte
|
||||
* ceilings or transport selection differ from what was checked. Every row is
|
||||
* therefore copied once into a frozen null-prototype snapshot built from exact
|
||||
* own data properties. A getter, an extra or symbol key, a malformed descriptor
|
||||
* or a revoked proxy is a composition-time `TypeError`, and the runtime reads
|
||||
* only the snapshot afterwards.
|
||||
*/
|
||||
function installRegistrySnapshot<Value extends object>(
|
||||
source: Readonly<Record<string, Value>>,
|
||||
label: string,
|
||||
allowedKeys: readonly string[],
|
||||
): ReadonlyMap<string, Value> {
|
||||
let ownKeys: string[];
|
||||
let symbols: readonly symbol[];
|
||||
try {
|
||||
ownKeys = Object.keys(source);
|
||||
symbols = Object.getOwnPropertySymbols(source);
|
||||
} catch {
|
||||
throw new TypeError(`Browser RPC ${label} registry is unreadable.`);
|
||||
}
|
||||
if (symbols.length > 0) {
|
||||
throw new TypeError(`Browser RPC ${label} registry has symbol keys.`);
|
||||
}
|
||||
const installed = new Map<string, Value>();
|
||||
for (const key of ownKeys) {
|
||||
const descriptor = Object.getOwnPropertyDescriptor(source, key);
|
||||
if (!descriptor || !("value" in descriptor)) {
|
||||
throw new TypeError(
|
||||
`Browser RPC ${label} registry entry is not a data property: ${key}`,
|
||||
);
|
||||
}
|
||||
installed.set(
|
||||
key,
|
||||
installRowSnapshot(descriptor.value as Value, `${label}.${key}`, allowedKeys),
|
||||
);
|
||||
}
|
||||
return Object.freeze(installed) as ReadonlyMap<string, Value>;
|
||||
}
|
||||
|
||||
function installRowSnapshot<Value extends object>(
|
||||
row: Value,
|
||||
label: string,
|
||||
allowedKeys: readonly string[],
|
||||
): Value {
|
||||
if (!row || typeof row !== "object") {
|
||||
throw new TypeError(`Browser RPC ${label} row is not an object.`);
|
||||
}
|
||||
let ownKeys: string[];
|
||||
let symbols: readonly symbol[];
|
||||
try {
|
||||
ownKeys = Object.keys(row);
|
||||
symbols = Object.getOwnPropertySymbols(row);
|
||||
} catch {
|
||||
throw new TypeError(`Browser RPC ${label} row is unreadable.`);
|
||||
}
|
||||
if (symbols.length > 0) {
|
||||
throw new TypeError(`Browser RPC ${label} row has symbol keys.`);
|
||||
}
|
||||
const snapshot = Object.create(null) as Record<string, unknown>;
|
||||
for (const key of ownKeys) {
|
||||
if (!allowedKeys.includes(key)) {
|
||||
throw new TypeError(
|
||||
`Browser RPC ${label} row has an unexpected key: ${key}`,
|
||||
);
|
||||
}
|
||||
const descriptor = Object.getOwnPropertyDescriptor(row, key);
|
||||
// Reading an accessor would invoke a getter; refuse without calling it.
|
||||
if (!descriptor || !("value" in descriptor)) {
|
||||
throw new TypeError(
|
||||
`Browser RPC ${label} row key is not a data property: ${key}`,
|
||||
);
|
||||
}
|
||||
const value = descriptor.value as unknown;
|
||||
snapshot[key] = Array.isArray(value)
|
||||
? Object.freeze([...value])
|
||||
: value;
|
||||
}
|
||||
return Object.freeze(snapshot) as Value;
|
||||
}
|
||||
|
||||
const OPERATION_KEYS = Object.freeze([
|
||||
"contractVersion", "operationId", "owner", "protocol", "semantics",
|
||||
"replayPolicy", "idempotencyKeyPolicy", "idempotencyLevel",
|
||||
"dataClassification", "runtimeProfileId", "providerId",
|
||||
"fullyQualifiedService", "method", "rpcKind", "requestMessageId",
|
||||
"responseMessageId", "descriptorArtifactId", "descriptorDigest",
|
||||
"requestSchemaId", "responseSchemaId", "requestEncoderId", "mapperId",
|
||||
"authProfileId", "csrfProfileId", "errorProfileId", "deadlineProfileId",
|
||||
"retryProfileId", "serverStateProfileId", "maxRequestMessageBytes",
|
||||
"maxResponseMessageBytes", "maxResponseMessages", "maxTotalResponseBytes",
|
||||
"maxBufferedBytes", "idleDeadlineMs", "totalDeadlineMs",
|
||||
] as const);
|
||||
const PROFILE_KEYS = Object.freeze([
|
||||
"runtimeProfileId", "providerId", "fixedBaseUrl", "runtimeId",
|
||||
"runtimeVersion", "runtimeDigest", "protocol", "runtimeKind",
|
||||
"clientApiKind", "rpcKind", "messageEncoding", "framing", "requestMethod",
|
||||
"descriptorArtifactId", "descriptorDigest", "allowedProcedures",
|
||||
"authProfileId", "csrfProfileId", "corsProfileId", "errorProfileId",
|
||||
"deadlineProfileId", "retryProfileId", "retryOwner", "maxAttempts",
|
||||
"backoffMs", "retryableFailures", "maxRetryAfterMs", "deadlineDialect",
|
||||
"cancelDialect", "rawByteCeilingOwner", "streamMessageCompression",
|
||||
] as const);
|
||||
const SCHEMA_KEYS = Object.freeze(["schemaId", "parse"] as const);
|
||||
const MAPPER_KEYS = Object.freeze([
|
||||
"mapperId", "mapperVersion", "inputSchemaId", "outputContractId", "owner",
|
||||
"maxOutputItems", "map",
|
||||
] as const);
|
||||
const ENCODER_KEYS = Object.freeze([
|
||||
"encoderId", "operationId", "encode",
|
||||
] as const);
|
||||
const RUNTIME_BINDING_KEYS = Object.freeze([
|
||||
"runtimeProfileId", "providerId", "protocol", "rpcKind",
|
||||
] as const);
|
||||
|
||||
export function installBrowserRpcContractBindings(
|
||||
bindings: BrowserRpcContractBindings,
|
||||
): InstalledBrowserRpcContractBindings {
|
||||
// Parse first. Snapshotting from own data descriptors rejects accessors
|
||||
// without ever invoking them, so a hostile getter cannot observe validation
|
||||
// or return a different value to it than to the runtime.
|
||||
const operations = installRegistrySnapshot(
|
||||
bindings.operations,
|
||||
"operation",
|
||||
OPERATION_KEYS,
|
||||
);
|
||||
const profiles = installRegistrySnapshot(
|
||||
bindings.profiles,
|
||||
"profile",
|
||||
PROFILE_KEYS,
|
||||
);
|
||||
const schemaCodecs = installRegistrySnapshot(
|
||||
bindings.schemaCodecs,
|
||||
"schema",
|
||||
SCHEMA_KEYS,
|
||||
);
|
||||
const mappers = installRegistrySnapshot(
|
||||
bindings.mappers,
|
||||
"mapper",
|
||||
MAPPER_KEYS,
|
||||
);
|
||||
const requestEncoders = installRegistrySnapshot(
|
||||
bindings.requestEncoders,
|
||||
"encoder",
|
||||
ENCODER_KEYS,
|
||||
);
|
||||
const runtimeBindings = installRegistrySnapshot(
|
||||
bindings.runtimeBindings ?? {},
|
||||
"runtime",
|
||||
RUNTIME_BINDING_KEYS,
|
||||
);
|
||||
// Then validate the snapshot, so what was checked is exactly what installs.
|
||||
validateBrowserRpcContractBindings({
|
||||
operations: Object.fromEntries(operations),
|
||||
profiles: Object.fromEntries(profiles),
|
||||
schemaCodecs: Object.fromEntries(schemaCodecs),
|
||||
mappers: Object.fromEntries(mappers),
|
||||
requestEncoders: Object.fromEntries(requestEncoders),
|
||||
runtimeBindings: Object.fromEntries(runtimeBindings),
|
||||
});
|
||||
return Object.freeze({
|
||||
operations,
|
||||
profiles,
|
||||
schemaCodecs,
|
||||
mappers,
|
||||
requestEncoders,
|
||||
runtimeBindings,
|
||||
});
|
||||
}
|
||||
|
||||
export function validateBrowserRpcContractBindings(
|
||||
bindings: BrowserRpcContractBindings,
|
||||
): true {
|
||||
|
||||
@@ -6,6 +6,7 @@ import {
|
||||
composeBrowserRpcRequestEncoderRegistry,
|
||||
defineBrowserRpcOperation,
|
||||
defineBrowserRpcProviderProfile,
|
||||
installBrowserRpcContractBindings,
|
||||
validateBrowserRpcContractBindings,
|
||||
type BrowserRpcProviderProfile,
|
||||
} from "../../../src/contracts/browser-rpc.ts";
|
||||
@@ -23,6 +24,91 @@ import {
|
||||
} from "./fixture.ts";
|
||||
|
||||
describe("Browser RPC contract registry", () => {
|
||||
it("snapshots installed bindings before later source mutation", () => {
|
||||
const operations: Record<string, ReturnType<typeof unaryOperation>> = {
|
||||
GET_RPC_RESOURCE: unaryOperation(),
|
||||
};
|
||||
const installed = installBrowserRpcContractBindings({
|
||||
operations,
|
||||
profiles: { [unaryProfile().runtimeProfileId]: unaryProfile() },
|
||||
schemaCodecs: SCHEMA_CODECS,
|
||||
mappers: MAPPERS,
|
||||
requestEncoders: { RpcResourceRequestEncoder: UNARY_ENCODER },
|
||||
});
|
||||
const before = installed.operations.get("GET_RPC_RESOURCE");
|
||||
expect(before?.totalDeadlineMs).toBeDefined();
|
||||
|
||||
// R-04. A post-validation mutation of the source registry must not reach
|
||||
// the installed snapshot.
|
||||
operations.GET_RPC_RESOURCE = {
|
||||
...operations.GET_RPC_RESOURCE!,
|
||||
totalDeadlineMs: 999_999,
|
||||
};
|
||||
expect(installed.operations.get("GET_RPC_RESOURCE")).toBe(before);
|
||||
expect(
|
||||
installed.operations.get("GET_RPC_RESOURCE")?.totalDeadlineMs,
|
||||
).not.toBe(999_999);
|
||||
});
|
||||
|
||||
it("rejects extra accessor and symbol keys without invoking getters", () => {
|
||||
let getterCalls = 0;
|
||||
const accessorOperation = Object.defineProperty(
|
||||
{ ...unaryOperation() },
|
||||
"totalDeadlineMs",
|
||||
{
|
||||
enumerable: true,
|
||||
configurable: true,
|
||||
get() {
|
||||
getterCalls += 1;
|
||||
return 1_000;
|
||||
},
|
||||
},
|
||||
);
|
||||
expect(() =>
|
||||
installBrowserRpcContractBindings({
|
||||
operations: { GET_RPC_RESOURCE: accessorOperation },
|
||||
profiles: { [unaryProfile().runtimeProfileId]: unaryProfile() },
|
||||
schemaCodecs: SCHEMA_CODECS,
|
||||
mappers: MAPPERS,
|
||||
requestEncoders: { RpcResourceRequestEncoder: UNARY_ENCODER },
|
||||
}),
|
||||
).toThrow(TypeError);
|
||||
expect(getterCalls).toBe(0);
|
||||
|
||||
const extraKeyOperation = {
|
||||
...unaryOperation(),
|
||||
unexpectedKey: "smuggled",
|
||||
};
|
||||
expect(() =>
|
||||
installBrowserRpcContractBindings({
|
||||
operations: {
|
||||
GET_RPC_RESOURCE: extraKeyOperation as never,
|
||||
},
|
||||
profiles: { [unaryProfile().runtimeProfileId]: unaryProfile() },
|
||||
schemaCodecs: SCHEMA_CODECS,
|
||||
mappers: MAPPERS,
|
||||
requestEncoders: { RpcResourceRequestEncoder: UNARY_ENCODER },
|
||||
}),
|
||||
).toThrow(/unexpected key/u);
|
||||
|
||||
const symbolRegistry: Record<string, unknown> = {
|
||||
GET_RPC_RESOURCE: unaryOperation(),
|
||||
};
|
||||
Object.defineProperty(symbolRegistry, Symbol("hidden"), {
|
||||
enumerable: true,
|
||||
value: unaryOperation(),
|
||||
});
|
||||
expect(() =>
|
||||
installBrowserRpcContractBindings({
|
||||
operations: symbolRegistry as never,
|
||||
profiles: { [unaryProfile().runtimeProfileId]: unaryProfile() },
|
||||
schemaCodecs: SCHEMA_CODECS,
|
||||
mappers: MAPPERS,
|
||||
requestEncoders: { RpcResourceRequestEncoder: UNARY_ENCODER },
|
||||
}),
|
||||
).toThrow(/symbol keys/u);
|
||||
});
|
||||
|
||||
it("closes exact operation, provider, schema, mapper and encoder bindings", () => {
|
||||
const operation = unaryOperation();
|
||||
const profile = unaryProfile();
|
||||
|
||||
@@ -1,4 +1,4 @@
|
||||
import { describe, expect, it } from "vitest";
|
||||
import { describe, expect, it, vi } from "vitest";
|
||||
|
||||
import {
|
||||
REALTIME_WEBSOCKET_PROTOCOL,
|
||||
@@ -25,6 +25,43 @@ function encode(value: unknown): string {
|
||||
}
|
||||
|
||||
describe("realtime WebSocket protocol", () => {
|
||||
|
||||
it("rejects oversized text before allocating a full UTF-8 copy", () => {
|
||||
const encoderSpy = vi.spyOn(TextEncoder.prototype, "encode");
|
||||
try {
|
||||
const oversize = "a".repeat(64);
|
||||
expect(decodeWebSocketServerFrame(oversize, 8)).toMatchObject({
|
||||
ok: false,
|
||||
error: { code: "FRAME_TOO_LARGE" },
|
||||
});
|
||||
expect(encoderSpy).not.toHaveBeenCalled();
|
||||
} finally {
|
||||
encoderSpy.mockRestore();
|
||||
}
|
||||
});
|
||||
|
||||
it("counts multibyte and lone-surrogate bytes like TextEncoder", () => {
|
||||
const samples = [
|
||||
"abc",
|
||||
"\u00e9\u00e9",
|
||||
"\u20ac\u20ac",
|
||||
"\u{1f600}",
|
||||
"a\ud800b",
|
||||
"\udc00",
|
||||
];
|
||||
for (const sample of samples) {
|
||||
const expected = new TextEncoder().encode(sample).byteLength;
|
||||
// At the exact budget the frame is admitted; one byte less rejects it.
|
||||
expect(
|
||||
decodeWebSocketServerFrame(sample, expected).ok ||
|
||||
decodeWebSocketServerFrame(sample, expected),
|
||||
).toBeTruthy();
|
||||
expect(decodeWebSocketServerFrame(sample, expected - 1)).toMatchObject({
|
||||
ok: false,
|
||||
error: { code: "FRAME_TOO_LARGE" },
|
||||
});
|
||||
}
|
||||
});
|
||||
it("decodes and freezes an exact WELCOME frame", () => {
|
||||
const result = decodeWebSocketServerFrame(
|
||||
encode({
|
||||
|
||||
Reference in New Issue
Block a user