From aa8ac356002a97d8c07090f923c6c487ae88a413 Mon Sep 17 00:00:00 2001 From: DongHyeonka Date: Sat, 15 Aug 2026 01:25:48 +0900 Subject: [PATCH] fix: make Browser RPC and Realtime own the physical work they report on A server stream's registration was pruned on any settled close receipt. `waitClosed()` rejecting, throwing synchronously, or not returning a promise at all was absorbed into a fulfilled `undefined`, so the runtime opened a second physical stream for the same operation while the first was still running against the server. Only a fulfilled, contract-shaped receipt confirms closure now; every negative receipt keeps the operation DRAINING. Cleanup also read foreign state outside the result boundary. A throwing iterator `return` accessor replaced the already selected timeout with a native `TypeError` and skipped the rest of the teardown, and the exported lease decoder threw on a hostile `Symbol.asyncIterator`. Both reads move inside their own boundaries, and the positive-close subscription is installed before any fallible cleanup. Composition validated the caller's registries before snapshotting them, so a hostile accessor ran twice during validation, and rows hiding fields behind a prototype or a non-enumerable key installed. Transport results were checked for allowed own keys only, so own `{ok,message,encodedBytes}` plus a prototype `injected` was a success and a missing `message` reached a permissive schema as `undefined`. In Realtime the tracked task was registered after the collaborator returned. An authority that re-entered `close()` from inside its own invocation saw an empty registry and got `{ok:true}` while its effect was pending. The task is now registered first and the collaborator is invoked a microtask later. A `scheduleTimeout` that threw was worse: the caller's own catch treated it as an apply failure and started a recovery beside the still-running effect, and `close()` rejected with a native `TypeError`. An uninstallable deadline now fails closed as an expired one. Co-Authored-By: Claude Opus 5 --- .../browser-rpc/browser-rpc-runtime.ts | 154 +++++++-- src/adapters/browser-rpc/transport.ts | 35 +- src/adapters/realtime/stream-coordinator.ts | 56 +++- src/contracts/browser-rpc.ts | 20 +- .../browser-rpc-remediation.test.ts | 314 ++++++++++++++++++ .../unit/realtime/stream-coordinator.test.ts | 173 ++++++++++ 6 files changed, 698 insertions(+), 54 deletions(-) diff --git a/src/adapters/browser-rpc/browser-rpc-runtime.ts b/src/adapters/browser-rpc/browser-rpc-runtime.ts index e546ced..21b6057 100644 --- a/src/adapters/browser-rpc/browser-rpc-runtime.ts +++ b/src/adapters/browser-rpc/browser-rpc-runtime.ts @@ -113,11 +113,11 @@ export function createBrowserRpcRuntime( const clock = dependencies.clock ?? systemClock; const generationFence = dependencies.generationFence ?? stableGenerationFence; - // RPC-RR-03. Snapshot before validating. Reading the caller's transport - // objects first would run their accessors, letting a hostile getter observe - // validation and then return something else to the runtime. + // RPC-RR-03 / RPC-03. Snapshot before validating — every registry, not just + // the transports. Handing the caller's own objects to the join validation + // first ran their accessors, so a hostile getter could observe validation and + // then answer the runtime differently. const installedTransports = snapshotTransports(dependencies.transports); - validateRuntimeDependencies(dependencies, installedTransports); // 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 @@ -129,6 +129,7 @@ export function createBrowserRpcRuntime( mappers: dependencies.mappers, requestEncoders: dependencies.requestEncoders, }); + validateRuntimeDependencies(installed, installedTransports); // RPC-RR-01. One entry per physical stream that has been opened and not yet // confirmed closed. A stream stays here through DRAINING — after cancel, @@ -215,9 +216,69 @@ export function createBrowserRpcRuntime( type ActiveStreamLease = Readonly<{ streamId: string; cancel(reason: string): void; - closed: Promise; + /** + * RPC-01. Fulfils with `true` only when the transport positively confirmed + * the physical stream closed, and with `false` for every negative receipt — + * a rejection, a synchronous throw, or a `waitClosed()` that did not even + * return a promise. A negative receipt keeps the operation DRAINING. + */ + closed: Promise; }>; +/** + * RPC-01. Turns a lease's close receipt into evidence the registry can trust. + * Only a fulfilled promise from a contract-shaped `waitClosed()` counts as + * confirmation that the physical stream ended. + */ +function confirmPhysicalClosure( + lease: Readonly<{ waitClosed(): Promise }>, +): Promise { + return Promise.resolve() + .then(() => { + const receipt: unknown = lease.waitClosed(); + if ( + receipt === null || + (typeof receipt !== "object" && typeof receipt !== "function") || + typeof (receipt as PromiseLike).then !== "function" + ) { + return false; + } + return Promise.resolve(receipt as PromiseLike).then( + () => true, + () => false, + ); + }) + .catch(() => false); +} + +/** + * RPC-02. Reads an iterator's optional `return` once, through its own data + * descriptor. Testing `iterator.return` directly ran a foreign accessor outside + * any boundary, so a throwing getter replaced the already selected stream + * outcome with a native rejection and skipped the rest of the cleanup. + */ +function safeIteratorReturn( + iterator: AsyncIterator | null | undefined, +): (() => unknown) | null { + if (!iterator) return null; + try { + let current: object | null = iterator; + while (current !== null) { + const descriptor = Object.getOwnPropertyDescriptor(current, "return"); + if (descriptor) { + if (!("value" in descriptor)) return null; + return typeof descriptor.value === "function" + ? (descriptor.value as () => unknown).bind(iterator) + : null; + } + current = Reflect.getPrototypeOf(current); + } + return null; + } catch { + return null; + } +} + async function executeUnary( installed: InstalledBrowserRpcContractBindings, dependencies: BrowserRpcRuntimeDependencies, @@ -608,9 +669,12 @@ async function* executeServerStream( Object.freeze({ streamId: lease.streamId, cancel: lease.cancel, - closed: Promise.resolve() - .then(() => lease!.waitClosed()) - .catch(() => undefined) as Promise, + // RPC-01. Only a fulfilled, contract-shaped receipt is evidence that the + // physical stream closed. Absorbing a rejection, a synchronous throw or + // a non-promise into `undefined` forged that evidence, and the registry + // then admitted a second stream for the same operation while the first + // was still running against the server. + closed: confirmPhysicalClosure(lease), }), ); try { @@ -793,25 +857,34 @@ async function* executeServerStream( // A transport that refuses to cancel stays DRAINING below. } } - if (iterator?.return) { + if (registered && registered.streamId === lease?.streamId) { + // RPC-02. The positive-close subscription is installed before any + // fallible cleanup. Running cleanup first meant a throwing `return` + // accessor could skip it entirely and strand the registry entry. + // The entry is removed only once the transport confirms the physical + // stream closed. Until then the operation stays DRAINING and admits + // nothing new — a bounded wait here would re-open the very hole this + // registry exists to close. + void registered.closed.then((confirmed) => { + if ( + confirmed && + activeStreams.get(operation.operationId) === registered + ) { + activeStreams.delete(operation.operationId); + } + }); + } + const returnIterator = safeIteratorReturn(iterator); + if (returnIterator) { const cleanup = Promise.resolve() - .then(async () => await iterator?.return?.()) + .then(async () => await returnIterator()) // Cleanup cannot replace the already selected stream outcome. .catch(() => undefined); await boundedStreamCleanup(cleanup, clock, STREAM_CLEANUP_BOUND_MS); } if (registered && registered.streamId === lease?.streamId) { - // The entry is removed only once the transport confirms the physical - // stream closed. Until then the operation stays DRAINING and admits - // nothing new — a bounded wait here would re-open the very hole this - // registry exists to close. - void registered.closed.then(() => { - if (activeStreams.get(operation.operationId) === registered) { - activeStreams.delete(operation.operationId); - } - }); await boundedStreamCleanup( - registered.closed, + registered.closed.then(() => undefined), clock, STREAM_CLEANUP_BOUND_MS, ); @@ -903,7 +976,7 @@ function snapshotTransports( } function validateRuntimeDependencies( - dependencies: BrowserRpcRuntimeDependencies, + installed: InstalledBrowserRpcContractBindings, transports: ReadOnlyRegistry, ): void { const runtimeBindings: Record< @@ -929,15 +1002,16 @@ function validateRuntimeDependencies( rpcKind: transport.rpcKind, }); } + // Only the installed snapshot reaches the join validation. validateBrowserRpcContractBindings({ - operations: dependencies.operations, - profiles: dependencies.profiles, - schemaCodecs: dependencies.schemaCodecs, - mappers: dependencies.mappers, - requestEncoders: dependencies.requestEncoders, + operations: Object.fromEntries(installed.operations.entries()), + profiles: Object.fromEntries(installed.profiles.entries()), + schemaCodecs: Object.fromEntries(installed.schemaCodecs.entries()), + mappers: Object.fromEntries(installed.mappers.entries()), + requestEncoders: Object.fromEntries(installed.requestEncoders.entries()), runtimeBindings: Object.freeze(runtimeBindings), }); - for (const operation of Object.values(dependencies.operations)) { + for (const operation of installed.operations.values()) { if (!transports.has(operation.runtimeProfileId)) { throw new TypeError( `Browser RPC transport is missing: ${operation.operationId}`, @@ -1248,17 +1322,32 @@ function ownDataValue(source: unknown, key: string): unknown { } } +/** + * RPC-04. An exact union means exactly these own data keys, all of them, on a + * plain object. Checking only that each own name was *allowed* let a result + * carry required fields it never declared and let a custom prototype smuggle + * metadata past the trust boundary while the shape still looked valid. + */ function exactOwnKeys( source: unknown, allowed: ReadonlySet, + required: ReadonlySet = allowed, ): boolean { if (source === null || typeof source !== "object") return false; try { if (Object.getOwnPropertySymbols(source).length > 0) return false; + const prototype = Reflect.getPrototypeOf(source); + if (prototype !== Object.prototype && prototype !== null) return false; + const present = new Set(); for (const key of Object.getOwnPropertyNames(source)) { if (!allowed.has(key)) return false; const descriptor = Object.getOwnPropertyDescriptor(source, key); if (!descriptor || !("value" in descriptor)) return false; + if (descriptor.enumerable !== true) return false; + present.add(key); + } + for (const key of required) { + if (!present.has(key)) return false; } return true; } catch { @@ -1271,6 +1360,9 @@ const UNARY_OK_KEYS: ReadonlySet = new Set([ "message", "encodedBytes", ]); +/** `message` is a wire field, so an absent one is a protocol breach even when + * a permissive schema would happily accept `undefined`. */ +const UNARY_OK_REQUIRED: ReadonlySet = UNARY_OK_KEYS; const UNARY_FAILED_KEYS: ReadonlySet = new Set(["ok", "failure"]); const FRAME_MESSAGE_KEYS: ReadonlySet = new Set([ "kind", @@ -1287,6 +1379,8 @@ const TRANSPORT_FAILURE_KEYS: ReadonlySet = new Set([ "code", "retryAfterMs", ]); +/** `retryAfterMs` is genuinely optional; `code` is not. */ +const TRANSPORT_FAILURE_REQUIRED: ReadonlySet = new Set(["code"]); function validateUnaryTransportResult( value: unknown, @@ -1294,7 +1388,7 @@ function validateUnaryTransportResult( const ok = ownDataValue(value, "ok"); if (typeof ok !== "boolean") return null; if (ok) { - if (!exactOwnKeys(value, UNARY_OK_KEYS)) return null; + if (!exactOwnKeys(value, UNARY_OK_KEYS, UNARY_OK_REQUIRED)) return null; const encodedBytes = ownDataValue(value, "encodedBytes"); if (!validEncodedByteCount(encodedBytes, Number.MAX_SAFE_INTEGER)) { return null; @@ -1347,7 +1441,9 @@ function validateStreamFrame(value: unknown): BrowserRpcStreamFrame | null { function decodeTransportFailure( value: unknown, ): BrowserRpcTransportFailure | null { - if (!exactOwnKeys(value, TRANSPORT_FAILURE_KEYS)) return null; + if (!exactOwnKeys(value, TRANSPORT_FAILURE_KEYS, TRANSPORT_FAILURE_REQUIRED)) { + return null; + } const code = ownDataValue(value, "code"); const retryAfterMs = ownDataValue(value, "retryAfterMs"); if (!TRANSPORT_FAILURE_CODES.has(code as BrowserRpcTransportFailureCode)) { diff --git a/src/adapters/browser-rpc/transport.ts b/src/adapters/browser-rpc/transport.ts index 004d3a1..2ee61f2 100644 --- a/src/adapters/browser-rpc/transport.ts +++ b/src/adapters/browser-rpc/transport.ts @@ -119,21 +119,34 @@ export function decodeServerStreamLease( } catch { return null; } - if ( - typeof streamId !== "string" || - !STREAM_ID.test(streamId) || - frames === null || - typeof frames !== "object" || - typeof (frames as AsyncIterable)[Symbol.asyncIterator] !== - "function" || - typeof cancel !== "function" || - typeof waitClosed !== "function" - ) { + // RPC-02. The async-iterator lookup is a read of foreign state like any + // other, so it happens inside the decoder's own boundary. Performing it after + // the `try` let a throwing `Symbol.asyncIterator` getter escape this + // function as a native `TypeError`, breaking the decoder's totality. + let openFrames: unknown; + try { + if ( + typeof streamId !== "string" || + !STREAM_ID.test(streamId) || + frames === null || + typeof frames !== "object" || + typeof cancel !== "function" || + typeof waitClosed !== "function" + ) { + return null; + } + openFrames = (frames as AsyncIterable)[Symbol.asyncIterator]; + if (typeof openFrames !== "function") return null; + } catch { return null; } + const iterate = (openFrames as () => AsyncIterator) + .bind(frames); return Object.freeze({ streamId, - frames: frames as AsyncIterable, + frames: Object.freeze({ + [Symbol.asyncIterator]: iterate, + }) as AsyncIterable, cancel: (cancel as (reason: string) => void).bind(value), waitClosed: (waitClosed as () => Promise).bind(value), }); diff --git a/src/adapters/realtime/stream-coordinator.ts b/src/adapters/realtime/stream-coordinator.ts index 4e159c2..f57cc96 100644 --- a/src/adapters/realtime/stream-coordinator.ts +++ b/src/adapters/realtime/stream-coordinator.ts @@ -185,6 +185,15 @@ export function createRealtimeStreamCoordinator( const states = new Map(); let closed = false; + /** RT-02. Releasing a timer is best effort and never a public failure. */ + const clearTimerSafely = (handle: unknown): void => { + try { + clearScheduledTimeout(handle); + } catch { + // A broken scheduler cannot change an already classified outcome. + } + }; + const TASK_TIMED_OUT = Symbol("REALTIME_TASK_TIMED_OUT"); /** @@ -194,10 +203,16 @@ export function createRealtimeStreamCoordinator( */ async function awaitTaskWithinDeadline( state: StreamState, - task: Promise, + invoke: () => Promise | Value, timeoutMs: number, revokeAndAbort: () => void, ): Promise { + // RT-01. The collaborator is invoked on the next microtask, after the task + // is already in the registry. Calling it first left a window in which an + // authority that re-entered `close()` from inside its own invocation saw an + // empty registry, so `close()` reported quiescence while its effect was + // still running. + const task = Promise.resolve().then(invoke); task.catch(() => {}); // RT-RR-01. The task is a physical effect the moment it is created, so it // is registered here rather than when its public wait happens to expire. @@ -212,12 +227,23 @@ export function createRealtimeStreamCoordinator( if (!state.closed) state.freshness = "STALE"; } }); + // RT-02. A scheduler that cannot install the deadline leaves the wait + // unbounded. Letting the exception escape turned a typed realtime result + // into a native rejection and — through the caller's own catch — started a + // recovery that overlapped the effect still running, so an install failure + // fails closed as an expired deadline instead. let handle: unknown; + let installed = false; const timeout = new Promise((resolve) => { - handle = scheduleTimeout(() => resolve(TASK_TIMED_OUT), timeoutMs); + try { + handle = scheduleTimeout(() => resolve(TASK_TIMED_OUT), timeoutMs); + installed = true; + } catch { + resolve(TASK_TIMED_OUT); + } }); const outcome = await Promise.race([task, timeout]); - clearScheduledTimeout(handle); + if (installed) clearTimerSafely(handle); if (outcome !== TASK_TIMED_OUT) return outcome; // The commit capability is revoked immediately; the work itself is not. @@ -514,7 +540,7 @@ export function createRealtimeStreamCoordinator( try { const applied = await awaitTaskWithinDeadline( state, - Promise.resolve( + () => dependencies.authority.effects.apply( eventType.effectProfileId, mapped.value, @@ -527,7 +553,6 @@ export function createRealtimeStreamCoordinator( }), effectAbort.signal, ), - ), limits.effectTimeoutMs, () => { // Commit capability is revoked permanently for this attempt. @@ -701,7 +726,7 @@ export function createRealtimeStreamCoordinator( try { const outcome = await awaitTaskWithinDeadline( state, - Promise.resolve( + () => dependencies.authority.recovery.recover( Object.freeze({ streamId: state.registration.id, @@ -711,7 +736,6 @@ export function createRealtimeStreamCoordinator( isCurrent: recoveryIsCurrent, }), ), - ), limits.recoveryTimeoutMs, () => { recoveryLeaseActive = false; @@ -1012,16 +1036,24 @@ export function createRealtimeStreamCoordinator( return realtimeSuccess(undefined); } let handle: unknown; + let installed = false; const drained = await Promise.race([ Promise.allSettled(retained).then(() => true), new Promise((resolve) => { - handle = scheduleTimeout( - () => resolve(false), - limits.drainTimeoutMs, - ); + try { + handle = scheduleTimeout( + () => resolve(false), + limits.drainTimeoutMs, + ); + installed = true; + } catch { + // RT-02. Without a drain bound this call cannot prove quiescence, so + // it reports the honest failure rather than rejecting natively. + resolve(false); + } }), ]); - clearScheduledTimeout(handle); + if (installed) clearTimerSafely(handle); if (!drained) { // Still DRAINING: the caller must not treat this as quiescence. return realtimeFailure("IDLE_TIMEOUT", "CLOSE", false); diff --git a/src/contracts/browser-rpc.ts b/src/contracts/browser-rpc.ts index 3e099f6..9188132 100644 --- a/src/contracts/browser-rpc.ts +++ b/src/contracts/browser-rpc.ts @@ -363,15 +363,23 @@ function installRegistrySnapshot( ): ReadOnlyRegistry { let ownKeys: string[]; let symbols: readonly symbol[]; + let prototype: object | null; try { - ownKeys = Object.keys(source); + // RPC-03. Own *names*, not just enumerable keys: a non-enumerable own entry + // is as much a smuggled row as an inherited one, and `Object.keys` never + // saw either. + ownKeys = Object.getOwnPropertyNames(source); symbols = Object.getOwnPropertySymbols(source); + prototype = Reflect.getPrototypeOf(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.`); } + if (prototype !== Object.prototype && prototype !== null) { + throw new TypeError(`Browser RPC ${label} registry has a custom prototype.`); + } const installed = new Map(); for (const key of ownKeys) { const descriptor = Object.getOwnPropertyDescriptor(source, key); @@ -398,15 +406,23 @@ function installRowSnapshot( } let ownKeys: string[]; let symbols: readonly symbol[]; + let prototype: object | null; try { - ownKeys = Object.keys(row); + ownKeys = Object.getOwnPropertyNames(row); symbols = Object.getOwnPropertySymbols(row); + prototype = Reflect.getPrototypeOf(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.`); } + // RPC-03. A custom prototype carries fields the name sweep never sees and + // stays live after installation, so the installed row would not be the row + // that was checked. + if (prototype !== Object.prototype && prototype !== null) { + throw new TypeError(`Browser RPC ${label} row has a custom prototype.`); + } const snapshot = Object.create(null) as Record; for (const key of ownKeys) { if (!allowedKeys.includes(key)) { diff --git a/tests/unit/browser-rpc/browser-rpc-remediation.test.ts b/tests/unit/browser-rpc/browser-rpc-remediation.test.ts index b75404c..dcd9ded 100644 --- a/tests/unit/browser-rpc/browser-rpc-remediation.test.ts +++ b/tests/unit/browser-rpc/browser-rpc-remediation.test.ts @@ -7,6 +7,7 @@ import { type BrowserRpcStreamFrame, type BrowserRpcTransport, } from "../../../src/adapters/browser-rpc/index.ts"; +import { decodeServerStreamLease } from "../../../src/adapters/browser-rpc/transport.ts"; import { installBrowserRpcContractBindings } from "../../../src/contracts/browser-rpc.ts"; import { MAPPERS, @@ -474,3 +475,316 @@ describe("RPC-RR-01 server stream leases and the DRAINING fence", () => { } }); }); + +/** + * RPC-01. A `waitClosed()` that rejects, throws, or does not even return a + * promise is not evidence that the physical stream closed. Absorbing all three + * into a fulfilled `undefined` let the registry prune a live stream and admit a + * second one for the same operation against the same server. + */ +describe("RPC-01 only a positive receipt confirms physical closure", () => { + const shortDeadline = () => + streamOperation({ totalDeadlineMs: 25, idleDeadlineMs: 25 }); + + function leaseProbeWith( + waitClosed: () => unknown, + ): Readonly<{ + transport: BrowserRpcTransport; + cancels: string[]; + opened(): number; + }> { + const cancels: string[] = []; + const counter = { opened: 0 }; + const transport = defineBrowserRpcTransport({ + runtimeProfileId: "CONNECT_REFERENCE_STREAM", + providerId: "REFERENCE_RPC", + protocol: "CONNECT_HTTP", + rpcKind: "SERVER_STREAM", + openServerStream: () => { + counter.opened += 1; + return { + streamId: `physical-${counter.opened}`, + frames: { + [Symbol.asyncIterator]: () => + ({ + next: () => new Promise(() => {}), + }) as AsyncIterator, + }, + cancel(reason: string) { + cancels.push(reason); + }, + waitClosed: waitClosed as () => Promise, + }; + }, + }); + return Object.freeze({ + transport, + cancels, + opened: () => counter.opened, + }); + } + + const negativeReceipts = [ + { + label: "rejects", + waitClosed: () => Promise.reject(new Error("never closed")), + }, + { + label: "throws synchronously", + waitClosed: () => { + throw new TypeError("waitClosed exploded"); + }, + }, + { label: "returns a non-promise", waitClosed: () => "closed" }, + { label: "never settles", waitClosed: () => new Promise(() => {}) }, + ]; + + for (const { label, waitClosed } of negativeReceipts) { + it(`keeps the operation DRAINING when waitClosed ${label}`, async () => { + const probe = leaseProbeWith(waitClosed); + const runtime = createBrowserRpcRuntime({ + operations: { WATCH_RPC_RESOURCES: shortDeadline() }, + profiles: { CONNECT_REFERENCE_STREAM: streamProfile() }, + schemaCodecs: SCHEMA_CODECS, + mappers: MAPPERS, + requestEncoders: { RpcResourceStreamRequestEncoder: STREAM_ENCODER }, + transports: { CONNECT_REFERENCE_STREAM: probe.transport }, + }); + const stream = runtime.bindServerStream( + "WATCH_RPC_RESOURCES", + isResourceView, + ); + + await collect(stream.open({ resourceId: "scope-1" })); + expect(probe.opened()).toBe(1); + expect(probe.cancels).toHaveLength(1); + + const second = await collect(stream.open({ resourceId: "scope-1" })); + expect(second).toHaveLength(1); + expect(second[0]).toMatchObject({ + ok: false, + error: { code: "RPC_STREAM_DRAINING" }, + }); + // No second physical stream was ever opened. + expect(probe.opened()).toBe(1); + }); + } + + it("admits a second stream once waitClosed fulfils", async () => { + let release: (() => void) | undefined; + const closed = new Promise((resolve) => { + release = resolve; + }); + const probe = leaseProbeWith(() => closed); + const runtime = createBrowserRpcRuntime({ + operations: { WATCH_RPC_RESOURCES: shortDeadline() }, + profiles: { CONNECT_REFERENCE_STREAM: streamProfile() }, + schemaCodecs: SCHEMA_CODECS, + mappers: MAPPERS, + requestEncoders: { RpcResourceStreamRequestEncoder: STREAM_ENCODER }, + transports: { CONNECT_REFERENCE_STREAM: probe.transport }, + }); + const stream = runtime.bindServerStream( + "WATCH_RPC_RESOURCES", + isResourceView, + ); + + await collect(stream.open({ resourceId: "scope-1" })); + release?.(); + await new Promise((resolve) => setTimeout(resolve, 5)); + await collect(stream.open({ resourceId: "scope-1" })); + expect(probe.opened()).toBe(2); + }); + + /** + * RPC-02. Cleanup reads foreign state, so it must stay inside the result + * boundary: a throwing `return` accessor replaced the already selected + * outcome with a native rejection and skipped the rest of the teardown. + */ + it("keeps the selected outcome when the iterator return accessor throws", async () => { + const cancels: string[] = []; + const transport = defineBrowserRpcTransport({ + runtimeProfileId: "CONNECT_REFERENCE_STREAM", + providerId: "REFERENCE_RPC", + protocol: "CONNECT_HTTP", + rpcKind: "SERVER_STREAM", + openServerStream: () => ({ + streamId: "physical-hostile", + frames: { + [Symbol.asyncIterator]: () => + Object.defineProperty( + { next: () => new Promise(() => {}) }, + "return", + { + enumerable: true, + get() { + throw new TypeError("return getter escaped"); + }, + }, + ) as AsyncIterator, + }, + cancel(reason: string) { + cancels.push(reason); + }, + waitClosed: async () => {}, + }), + }); + const runtime = createBrowserRpcRuntime({ + operations: { WATCH_RPC_RESOURCES: shortDeadline() }, + profiles: { CONNECT_REFERENCE_STREAM: streamProfile() }, + schemaCodecs: SCHEMA_CODECS, + mappers: MAPPERS, + requestEncoders: { RpcResourceStreamRequestEncoder: STREAM_ENCODER }, + transports: { CONNECT_REFERENCE_STREAM: transport }, + }); + + const results = await collect( + runtime + .bindServerStream("WATCH_RPC_RESOURCES", isResourceView) + .open({ resourceId: "scope-1" }), + ); + + expect(results).toHaveLength(1); + expect(results[0]).toMatchObject({ + ok: false, + error: { code: "RPC_TOTAL_DEADLINE_EXCEEDED" }, + }); + expect(cancels).toHaveLength(1); + }); + + it("decodes a hostile frames iterator to null instead of throwing", async () => { + const hostile = { + streamId: "physical-hostile", + frames: Object.defineProperty({}, Symbol.asyncIterator, { + enumerable: true, + get() { + throw new TypeError("async iterator getter escaped"); + }, + }), + cancel() {}, + async waitClosed() {}, + }; + expect(decodeServerStreamLease(hostile)).toBeNull(); + }); +}); + +/** + * RPC-04. An exact union is exactly these own data keys on a plain object. A + * custom prototype could otherwise carry metadata past the boundary while the + * own shape still looked valid, and a missing wire field could reach a + * permissive schema as `undefined`. + */ +describe("RPC-04 transport results are an exact union", () => { + const hostileResults = [ + { + label: "valid own fields with an inherited extra", + result: () => + Object.assign(Object.create({ injected: "prototype" }), { + ok: true, + message: { id: "a", name: "A" }, + encodedBytes: 8, + }), + }, + { + label: "a missing message", + result: () => ({ ok: true, encodedBytes: 8 }), + }, + { + label: "a non-enumerable own extra", + result: () => + Object.defineProperty( + { ok: true, message: { id: "a", name: "A" }, encodedBytes: 8 }, + "injected", + { enumerable: false, value: true }, + ), + }, + { + label: "a failure with an inherited code", + result: () => ({ + ok: false, + failure: Object.create({ code: "SERVER_FAILURE" }) as object, + }), + }, + ]; + + for (const { label, result } of hostileResults) { + it(`rejects ${label} as a protocol mismatch`, async () => { + const runtime = unaryRuntime( + unaryTransport(async () => result() as never), + ); + await expect( + runtime + .bindUnary("GET_RPC_RESOURCE", isResourceView) + .execute({ resourceId: "scope-1" }), + ).resolves.toMatchObject({ + ok: false, + error: { code: "RPC_PROTOCOL_MISMATCH" }, + }); + }); + } +}); + +/** + * RPC-03. Every registry is snapshotted before anything validates it, so a + * hostile accessor never runs, and a row that hides fields behind a prototype + * or a non-enumerable key is refused rather than installed. + */ +describe("RPC-03 registries are snapshotted before they are validated", () => { + const build = (operations: Record) => + createBrowserRpcRuntime({ + operations: operations as never, + profiles: { CONNECT_REFERENCE_UNARY: unaryProfile() }, + schemaCodecs: SCHEMA_CODECS, + mappers: MAPPERS, + requestEncoders: { RpcResourceRequestEncoder: UNARY_ENCODER }, + transports: { + CONNECT_REFERENCE_UNARY: unaryTransport(async () => ({ + ok: true, + message: { id: "a", name: "A" }, + encodedBytes: 8, + })), + }, + }); + + it("never invokes a registry accessor", () => { + let reads = 0; + const operations = Object.defineProperty({}, "GET_RPC_RESOURCE", { + enumerable: true, + get() { + reads += 1; + return unaryOperation(); + }, + }); + expect(() => build(operations)).toThrow(TypeError); + expect(reads).toBe(0); + }); + + const hostileRows = [ + { + label: "an inherited extra field", + row: () => + Object.assign(Object.create({ injected: true }), unaryOperation()), + }, + { + label: "a non-enumerable own extra field", + row: () => + Object.defineProperty({ ...unaryOperation() }, "injected", { + enumerable: false, + value: true, + }), + }, + { + label: "a symbol field", + row: () => ({ + ...unaryOperation(), + [Symbol.for("injected")]: true, + }), + }, + ]; + + for (const { label, row } of hostileRows) { + it(`refuses to install a row with ${label}`, () => { + expect(() => build({ GET_RPC_RESOURCE: row() })).toThrow(TypeError); + }); + } +}); diff --git a/tests/unit/realtime/stream-coordinator.test.ts b/tests/unit/realtime/stream-coordinator.test.ts index 27a02a3..4bf93e4 100644 --- a/tests/unit/realtime/stream-coordinator.test.ts +++ b/tests/unit/realtime/stream-coordinator.test.ts @@ -1068,6 +1068,8 @@ type HarnessOptions = Readonly<{ recoveryTimeoutMs?: number; drainTimeoutMs?: number; }>; + scheduleTimeout?: (callback: () => void, delayMs: number) => unknown; + clearScheduledTimeout?: (handle: unknown) => void; }>; function createHarness(options: HarnessOptions = {}) { @@ -1114,6 +1116,12 @@ function createHarness(options: HarnessOptions = {}) { now: () => 10_000, observe: options.observe, ...(options.taskLimits ? { taskLimits: options.taskLimits } : {}), + ...(options.scheduleTimeout + ? { scheduleTimeout: options.scheduleTimeout } + : {}), + ...(options.clearScheduledTimeout + ? { clearScheduledTimeout: options.clearScheduledTimeout } + : {}), }); return { @@ -1176,3 +1184,168 @@ function deferred() { }); return { promise, resolve }; } + +/** + * RT-01. The registry entry has to exist before the collaborator is called. + * Registering after the invocation returned left a window in which an authority + * that re-entered `close()` from inside its own callback saw an empty set, so + * `close()` reported quiescence while its effect was still running. + */ +describe("RT-01 reentrant close cannot pass the registration window", () => { + it("refuses quiescence when apply re-enters close during its invocation", async () => { + const applyGate = deferred>(); + let closeResult: RealtimeResult | undefined; + let harness: ReturnType | undefined; + harness = createHarness({ + apply: () => { + // Re-entered from inside the invocation itself. + void harness!.coordinator.close().then((result) => { + closeResult = result; + }); + return applyGate.promise; + }, + taskLimits: { effectTimeoutMs: 10_000, drainTimeoutMs: 20 }, + }); + await harness.initialize(); + + const applying = harness.coordinator.accept(harness.event()); + await new Promise((resolve) => setTimeout(resolve, 50)); + + expect(closeResult?.ok).toBe(false); + expect(closeResult?.ok ? null : closeResult?.error.kind).toBe( + "IDLE_TIMEOUT", + ); + + applyGate.resolve(realtimeSuccess(undefined)); + await applying; + // Once the raw task settled, a second close does converge. + await expect(harness.coordinator.close()).resolves.toMatchObject({ + ok: true, + }); + }); + + it("refuses quiescence when recovery re-enters close during its invocation", async () => { + const recoveryGate = deferred>(); + let closeResult: RealtimeResult | undefined; + let harness: ReturnType | undefined; + let recoveries = 0; + harness = createHarness({ + recover: () => { + recoveries += 1; + if (recoveries === 1) { + return Promise.resolve(realtimeSuccess(snapshotCommit("0"))); + } + void harness!.coordinator.close().then((result) => { + closeResult = result; + }); + return recoveryGate.promise; + }, + taskLimits: { recoveryTimeoutMs: 10_000, drainTimeoutMs: 20 }, + }); + await harness.initialize(); + + const recovering = harness.coordinator.recover(STREAM_ID, "SEQUENCE_GAP"); + await new Promise((resolve) => setTimeout(resolve, 50)); + + expect(closeResult?.ok).toBe(false); + expect(closeResult?.ok ? null : closeResult?.error.kind).toBe( + "IDLE_TIMEOUT", + ); + + recoveryGate.resolve(realtimeSuccess(snapshotCommit("0"))); + await recovering; + }); +}); + +/** + * RT-02. A scheduler that cannot install a deadline leaves the wait unbounded. + * Letting the exception escape turned a typed realtime result into a native + * rejection and, through the caller's own catch, started a recovery that + * overlapped the effect still running. + */ +describe("RT-02 a throwing scheduler stays inside the result contract", () => { + /** Installs deadlines normally until the stream is initialized. */ + function breakableScheduler() { + const state = { broken: false }; + return { + state, + scheduleTimeout: (callback: () => void, delayMs: number) => { + if (state.broken) throw new TypeError("scheduleTimeout exploded"); + return setTimeout(callback, delayMs); + }, + clearScheduledTimeout: (handle: unknown) => { + clearTimeout(handle as ReturnType); + }, + }; + } + + it("does not start a recovery that overlaps a pending apply", async () => { + const applyGate = deferred>(); + const scheduler = breakableScheduler(); + let recoveriesAfterApply = 0; + const harness = createHarness({ + apply: () => applyGate.promise, + recover: async () => { + recoveriesAfterApply += 1; + return realtimeSuccess(snapshotCommit("0")); + }, + taskLimits: { effectTimeoutMs: 10_000, drainTimeoutMs: 20 }, + scheduleTimeout: scheduler.scheduleTimeout, + clearScheduledTimeout: scheduler.clearScheduledTimeout, + }); + await harness.initialize(); + recoveriesAfterApply = 0; + scheduler.state.broken = true; + + const accepted = await harness.coordinator.accept(harness.event()); + + // The apply is still pending, so no recovery may have run beside it. + expect(recoveriesAfterApply).toBe(0); + expect(accepted.ok).toBe(false); + expect(harness.coordinator.lifecycle(STREAM_ID)).toBe("DRAINING"); + + applyGate.resolve(realtimeSuccess(undefined)); + }); + + it("returns a typed close failure instead of rejecting natively", async () => { + const applyGate = deferred>(); + const scheduler = breakableScheduler(); + const harness = createHarness({ + apply: () => applyGate.promise, + taskLimits: { effectTimeoutMs: 10_000, drainTimeoutMs: 20 }, + scheduleTimeout: scheduler.scheduleTimeout, + clearScheduledTimeout: scheduler.clearScheduledTimeout, + }); + await harness.initialize(); + scheduler.state.broken = true; + + const applying = harness.coordinator.accept(harness.event()); + // Let the apply reach the authority before the fence arrives. + await Promise.resolve(); + await Promise.resolve(); + const closed = await harness.coordinator.close(); + + // A drain that could not be bounded is not proof of quiescence. + expect(closed.ok).toBe(false); + expect(closed.ok ? null : closed.error.kind).toBe("IDLE_TIMEOUT"); + + applyGate.resolve(realtimeSuccess(undefined)); + await applying; + }); + + it("keeps the classified outcome when clearing a timer throws", async () => { + const harness = createHarness({ + clearScheduledTimeout: () => { + throw new TypeError("clearScheduledTimeout exploded"); + }, + }); + await harness.initialize(); + + await expect( + harness.coordinator.accept(harness.event()), + ).resolves.toMatchObject({ ok: true }); + await expect(harness.coordinator.close()).resolves.toMatchObject({ + ok: true, + }); + }); +});