From a7390e3b3a6f8cf0dab2787a470982924be34c69 Mon Sep 17 00:00:00 2001 From: DongHyeonka Date: Fri, 14 Aug 2026 16:52:57 +0900 Subject: [PATCH] fix: give Browser RPC server streams a cancellable lease and a DRAINING fence MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit RPC-RR-01. openServerStream returned a bare AsyncIterable, which gave the runtime no way to stop the physical stream or to learn when it actually closed: iterator.return() is a request a non-cooperative implementation may ignore. The runtime could therefore time out, report the call finished, and admit a second stream for the same operation while the first was still running against the server. The transport now returns a lease — streamId, frames, cancel(reason) and waitClosed() — decoded from own data descriptors before the runtime registers it, so an accessor cannot hand the registry one object and the cancellation path another. The runtime registers the lease the moment the physical stream exists, cancels exactly once on exit, and keeps the entry until waitClosed() settles. A second stream for the same operation is refused as CONFLICT / RPC_STREAM_DRAINING while that entry stands, and the refusal never reaches the transport. While updating the suites this also corrected two RPC-RR-02/RPC-RR-04 stream tests that were passing for the wrong reason: they sent an input the request schema rejects, so they never reached the fence or the frame decoder. With the correct input the frame test fails on a permissive decoder, as it should. Co-Authored-By: Claude Opus 5 (1M context) --- .../browser-rpc/browser-rpc-runtime.ts | 132 +++++++++++--- src/adapters/browser-rpc/index.ts | 2 + src/adapters/browser-rpc/transport.ts | 81 ++++++++- .../unavailable-browser-rpc-transport.ts | 21 ++- .../browser-rpc-remediation.test.ts | 169 +++++++++++++++++- .../browser-rpc/browser-rpc-runtime.test.ts | 41 ++++- 6 files changed, 412 insertions(+), 34 deletions(-) diff --git a/src/adapters/browser-rpc/browser-rpc-runtime.ts b/src/adapters/browser-rpc/browser-rpc-runtime.ts index a5d6d86..e546ced 100644 --- a/src/adapters/browser-rpc/browser-rpc-runtime.ts +++ b/src/adapters/browser-rpc/browser-rpc-runtime.ts @@ -27,7 +27,9 @@ import { } from "../../contracts/errors.ts"; import type { RuntimeSchemaCodec } from "../../contracts/schema-registry.ts"; import { systemClock } from "../platform/system-clock.ts"; +import { decodeServerStreamLease } from "./transport.ts"; import type { + BrowserRpcServerStreamLease, BrowserRpcStreamFrame, BrowserRpcTransport, BrowserRpcTransportFailure, @@ -128,6 +130,12 @@ export function createBrowserRpcRuntime( requestEncoders: dependencies.requestEncoders, }); + // 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, + // before `waitClosed()` settles — so a second stream for the same operation + // cannot be admitted while the first is still running against the server. + const activeStreams = new Map(); + function bind( operationId: string, expectedKind: "UNARY" | "SERVER_STREAM", @@ -192,12 +200,24 @@ export function createBrowserRpcRuntime( isEvent, generationFence, clock, + activeStreams, ), }); }, }); } +/** + * RPC-RR-01. A physical stream the runtime has opened and not yet confirmed + * closed. `draining` is true from the moment cancellation is requested until + * `waitClosed()` settles; while it is true the operation admits no new stream. + */ +type ActiveStreamLease = Readonly<{ + streamId: string; + cancel(reason: string): void; + closed: Promise; +}>; + async function executeUnary( installed: InstalledBrowserRpcContractBindings, dependencies: BrowserRpcRuntimeDependencies, @@ -462,11 +482,13 @@ async function* executeServerStream( isEvent: (value: unknown) => value is Event, generationFence: BrowserRpcGenerationFence, clock: ClockPort, + activeStreams: Map, ): AsyncIterable> { const { operation, profile, encoder, transport } = bound; const deadlineAt = safeNowOrDeadline(clock, 0) + operation.totalDeadlineMs; const linked = linkedAbortController(context.signal); let iterator: AsyncIterator | null = null; + let lease: BrowserRpcServerStreamLease | null = null; let observed = false; let messageCount = 0; @@ -533,23 +555,66 @@ async function* executeServerStream( ); return; } - let stream: AsyncIterable; - try { - stream = transport.openServerStream!( - Object.freeze({ - operation, - profile, - request: prepared.value, - encodedRequestBytes: prepared.encodedBytes, - attempt: 1, - timeoutMs: Math.max(1, remainingMs), - signal: linked.controller.signal, - ...(context.idempotencyKey - ? { idempotencyKey: context.idempotencyKey } - : {}), - }), + // RPC-RR-01. Admission fence. A previous physical stream for this operation + // that has not confirmed closure still owns the server-side resource, so a + // second one is refused rather than opened alongside it. + if (activeStreams.has(operation.operationId)) { + finish("CONTRACT_REJECTED"); + yield failureResult( + callFailure(operation, 0, "CONFLICT", "RPC_STREAM_DRAINING"), ); - iterator = stream[Symbol.asyncIterator](); + return; + } + + let decodedLease: BrowserRpcServerStreamLease | null; + try { + decodedLease = decodeServerStreamLease( + transport.openServerStream!( + Object.freeze({ + operation, + profile, + request: prepared.value, + encodedRequestBytes: prepared.encodedBytes, + attempt: 1, + timeoutMs: Math.max(1, remainingMs), + signal: linked.controller.signal, + ...(context.idempotencyKey + ? { idempotencyKey: context.idempotencyKey } + : {}), + }), + ), + ); + } catch { + finish("FAILED"); + yield failureResult( + callFailure( + operation, + 0, + "SERVER_FAILURE", + "RPC_TRANSPORT_EXECUTION_FAILED", + ), + ); + return; + } + if (!decodedLease) { + finish("CONTRACT_REJECTED"); + yield failureResult(protocolFailure(operation, 0)); + return; + } + lease = decodedLease; + // Registered at the moment the physical stream exists, not after a timeout. + activeStreams.set( + operation.operationId, + Object.freeze({ + streamId: lease.streamId, + cancel: lease.cancel, + closed: Promise.resolve() + .then(() => lease!.waitClosed()) + .catch(() => undefined) as Promise, + }), + ); + try { + iterator = lease.frames[Symbol.asyncIterator](); } catch { finish("FAILED"); yield failureResult( @@ -712,13 +777,22 @@ async function* executeServerStream( 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. + // R-01 / RPC-RR-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, so + // the lease's own `cancel` is what actually stops the physical stream. linked.controller.abort("stream-closed"); linked.cleanup(); + const registered = lease + ? activeStreams.get(operation.operationId) + : undefined; + if (registered && registered.streamId === lease?.streamId) { + try { + registered.cancel("stream-closed"); + } catch { + // A transport that refuses to cancel stays DRAINING below. + } + } if (iterator?.return) { const cleanup = Promise.resolve() .then(async () => await iterator?.return?.()) @@ -726,6 +800,22 @@ async function* executeServerStream( .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, + clock, + STREAM_CLEANUP_BOUND_MS, + ); + } finish("ABORTED"); } } diff --git a/src/adapters/browser-rpc/index.ts b/src/adapters/browser-rpc/index.ts index 282b398..ffa11a9 100644 --- a/src/adapters/browser-rpc/index.ts +++ b/src/adapters/browser-rpc/index.ts @@ -7,7 +7,9 @@ export { type BrowserRpcRuntimeDependencies, } from "./browser-rpc-runtime.ts"; export { + decodeServerStreamLease, defineBrowserRpcTransport, + type BrowserRpcServerStreamLease, type BrowserRpcStreamFrame, type BrowserRpcTransport, type BrowserRpcTransportCall, diff --git a/src/adapters/browser-rpc/transport.ts b/src/adapters/browser-rpc/transport.ts index a1130bd..004d3a1 100644 --- a/src/adapters/browser-rpc/transport.ts +++ b/src/adapters/browser-rpc/transport.ts @@ -50,6 +50,29 @@ export type BrowserRpcStreamFrame = failure: BrowserRpcTransportFailure; }>; +/** + * RPC-RR-01. A server stream is a physical resource, not just a sequence. + * + * A bare `AsyncIterable` gives the runtime no way to cancel the underlying + * stream or to learn when it actually closed: `iterator.return()` is a request + * a non-cooperative implementation may ignore. The runtime could then time out, + * report the call finished, and admit a second stream for the same operation + * while the first was still running against the server. + * + * The lease separates the three concerns the runtime needs: + * + * - `frames` is the sequence, + * - `cancel(reason)` is a synchronous request to stop the physical stream, + * - `waitClosed()` settles only once that stream is really closed, + * - `streamId` names the physical stream so two leases are never confused. + */ +export type BrowserRpcServerStreamLease = Readonly<{ + streamId: string; + frames: AsyncIterable; + cancel(reason: string): void; + waitClosed(): Promise; +}>; + export type BrowserRpcTransport = BrowserRpcRuntimeBindingIdentity & Readonly<{ invokeUnary?( @@ -57,9 +80,65 @@ export type BrowserRpcTransport = BrowserRpcRuntimeBindingIdentity & ): Promise; openServerStream?( call: BrowserRpcTransportCall, - ): AsyncIterable; + ): BrowserRpcServerStreamLease; }>; +const STREAM_ID = /^[A-Za-z0-9][A-Za-z0-9._:-]{0,127}$/u; + +/** + * RPC-RR-01. Decodes a lease from own data descriptors before the runtime + * registers it, so an accessor cannot hand the registry one object and the + * cancellation path another. + */ +export function decodeServerStreamLease( + value: unknown, +): BrowserRpcServerStreamLease | null { + if (value === null || typeof value !== "object") return null; + let streamId: unknown; + let frames: unknown; + let cancel: unknown; + let waitClosed: unknown; + try { + if (Object.getOwnPropertySymbols(value).length > 0) return null; + const names = Object.getOwnPropertyNames(value).sort(); + const expected = ["cancel", "frames", "streamId", "waitClosed"]; + if ( + names.length !== expected.length || + names.some((name, index) => name !== expected[index]) + ) { + return null; + } + for (const name of names) { + const descriptor = Object.getOwnPropertyDescriptor(value, name); + if (!descriptor || !("value" in descriptor)) return null; + } + streamId = Object.getOwnPropertyDescriptor(value, "streamId")?.value; + frames = Object.getOwnPropertyDescriptor(value, "frames")?.value; + cancel = Object.getOwnPropertyDescriptor(value, "cancel")?.value; + waitClosed = Object.getOwnPropertyDescriptor(value, "waitClosed")?.value; + } 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" + ) { + return null; + } + return Object.freeze({ + streamId, + frames: frames as AsyncIterable, + cancel: (cancel as (reason: string) => void).bind(value), + waitClosed: (waitClosed as () => Promise).bind(value), + }); +} + export function defineBrowserRpcTransport( transport: BrowserRpcTransport, ): BrowserRpcTransport { diff --git a/src/adapters/browser-rpc/unavailable-browser-rpc-transport.ts b/src/adapters/browser-rpc/unavailable-browser-rpc-transport.ts index cff96d8..f33d32b 100644 --- a/src/adapters/browser-rpc/unavailable-browser-rpc-transport.ts +++ b/src/adapters/browser-rpc/unavailable-browser-rpc-transport.ts @@ -31,11 +31,22 @@ export function createUnavailableBrowserRpcTransport(input: Readonly<{ } return defineBrowserRpcTransport({ ...input, - async *openServerStream() { - yield Object.freeze({ - kind: "TERMINAL", - ok: false, - failure: Object.freeze({ code: "UNAVAILABLE" }), + // RPC-RR-01. Even a stream that never opens hands back a lease, so the + // runtime's registry and cancellation path have one shape to work with. + openServerStream() { + return Object.freeze({ + streamId: `unavailable-${input.runtimeProfileId}`, + frames: Object.freeze({ + async *[Symbol.asyncIterator]() { + yield Object.freeze({ + kind: "TERMINAL" as const, + ok: false as const, + failure: Object.freeze({ code: "UNAVAILABLE" as const }), + }); + }, + }), + cancel() {}, + async waitClosed() {}, }); }, }); diff --git a/tests/unit/browser-rpc/browser-rpc-remediation.test.ts b/tests/unit/browser-rpc/browser-rpc-remediation.test.ts index e295139..b75404c 100644 --- a/tests/unit/browser-rpc/browser-rpc-remediation.test.ts +++ b/tests/unit/browser-rpc/browser-rpc-remediation.test.ts @@ -3,6 +3,7 @@ import { describe, expect, it } from "vitest"; import { createBrowserRpcRuntime, defineBrowserRpcTransport, + type BrowserRpcServerStreamLease, type BrowserRpcStreamFrame, type BrowserRpcTransport, } from "../../../src/adapters/browser-rpc/index.ts"; @@ -61,6 +62,60 @@ function unaryTransport( }); } +type LeaseProbe = Readonly<{ + transport: BrowserRpcTransport; + cancels: string[]; + readonly opened: number; + close(): void; +}>; + +/** + * RPC-RR-01. A transport whose physical stream does not close on its own. The + * runtime must cancel it exactly once and must not admit a second stream for + * the same operation until `waitClosed()` settles. + */ +function nonCooperativeStreamTransport(): LeaseProbe { + const cancels: string[] = []; + const counter = { opened: 0 }; + let release: (() => void) | undefined; + const closed = new Promise((resolve) => { + release = resolve; + }); + const openServerStream = (): BrowserRpcServerStreamLease => { + counter.opened += 1; + return { + streamId: `physical-${counter.opened}`, + frames: { + [Symbol.asyncIterator]: () => + ({ + // Never yields and never settles: the runtime's own deadline is the + // only thing that can end the call. + next: () => new Promise(() => {}), + }) as AsyncIterator, + }, + cancel(reason: string) { + cancels.push(reason); + }, + waitClosed: () => closed, + }; + }; + const transport = defineBrowserRpcTransport({ + runtimeProfileId: "CONNECT_REFERENCE_STREAM", + providerId: "REFERENCE_RPC", + protocol: "CONNECT_HTTP", + rpcKind: "SERVER_STREAM", + openServerStream, + }); + return Object.freeze({ + transport, + cancels, + get opened() { + return counter.opened; + }, + close: () => release?.(), + }) as LeaseProbe; +} + function streamTransport( frames: readonly BrowserRpcStreamFrame[], ): BrowserRpcTransport { @@ -70,9 +125,14 @@ function streamTransport( protocol: "CONNECT_HTTP", rpcKind: "SERVER_STREAM", openServerStream: () => ({ - async *[Symbol.asyncIterator]() { - for (const frame of frames) yield frame; + streamId: "remediation-stream", + frames: { + async *[Symbol.asyncIterator]() { + for (const frame of frames) yield frame; + }, }, + cancel() {}, + async waitClosed() {}, }), }); } @@ -124,7 +184,7 @@ describe("RPC-RR-02 synchronous collaborator throws stay inside Result", () => { const results = await collect( runtime .bindServerStream("WATCH_RPC_RESOURCES", isResourceView) - .open({ topic: "resources" }), + .open({ resourceId: "scope-1" }), ); expect(results.length).toBeGreaterThan(0); expect(results.every((value) => (value as { ok: boolean }).ok === false)).toBe( @@ -306,10 +366,111 @@ describe("RPC-RR-04 transport results and frames are exactly decoded", () => { const results = await collect( runtime .bindServerStream("WATCH_RPC_RESOURCES", isResourceView) - .open({ topic: "resources" }), + .open({ resourceId: "scope-1" }), ); expect(results.every((value) => (value as { ok: boolean }).ok === false)).toBe( true, ); }); }); + +/** + * RPC-RR-01. A bare `AsyncIterable` gave the runtime no way to cancel the + * physical stream or to learn when it actually closed, so a timed-out call left + * the first stream running against the server while a second was admitted. + */ +describe("RPC-RR-01 server stream leases and the DRAINING fence", () => { + const shortDeadline = () => + streamOperation({ totalDeadlineMs: 25, idleDeadlineMs: 25 }); + + it("cancels the physical stream exactly once after a timeout", async () => { + const probe = nonCooperativeStreamTransport(); + 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 results = await collect( + runtime + .bindServerStream("WATCH_RPC_RESOURCES", isResourceView) + .open({ resourceId: "scope-1" }), + ); + + expect(results.every((value) => (value as { ok: boolean }).ok === false)).toBe( + true, + ); + expect(probe.cancels.length).toBe(1); + }); + + it("refuses a second stream while the first has not confirmed closure", async () => { + const probe = nonCooperativeStreamTransport(); + 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); + + const second = await collect(stream.open({ resourceId: "scope-1" })); + expect(second).toHaveLength(1); + expect(second[0]).toMatchObject({ + ok: false, + error: { code: "RPC_STREAM_DRAINING" }, + }); + // The refused call never reached the transport. + expect(probe.opened).toBe(1); + + // Once the transport confirms closure the operation admits work again. + probe.close(); + await new Promise((resolve) => setTimeout(resolve, 5)); + await collect(stream.open({ resourceId: "scope-1" })); + expect(probe.opened).toBe(2); + }); + + it("refuses a malformed lease as a protocol failure", async () => { + for (const malformed of [ + { frames: { [Symbol.asyncIterator]: () => ({ next: async () => ({ done: true, value: undefined }) }) }, cancel() {}, async waitClosed() {} }, + { streamId: "", frames: { [Symbol.asyncIterator]: () => ({ next: async () => ({ done: true, value: undefined }) }) }, cancel() {}, async waitClosed() {} }, + { streamId: "s1", frames: {}, cancel() {}, async waitClosed() {} }, + { streamId: "s1", frames: { [Symbol.asyncIterator]: () => ({ next: async () => ({ done: true, value: undefined }) }) }, cancel: 1, async waitClosed() {} }, + ]) { + const runtime = createBrowserRpcRuntime({ + operations: { WATCH_RPC_RESOURCES: streamOperation() }, + profiles: { CONNECT_REFERENCE_STREAM: streamProfile() }, + schemaCodecs: SCHEMA_CODECS, + mappers: MAPPERS, + requestEncoders: { RpcResourceStreamRequestEncoder: STREAM_ENCODER }, + transports: { + CONNECT_REFERENCE_STREAM: defineBrowserRpcTransport({ + runtimeProfileId: "CONNECT_REFERENCE_STREAM", + providerId: "REFERENCE_RPC", + protocol: "CONNECT_HTTP", + rpcKind: "SERVER_STREAM", + openServerStream: () => malformed as never, + }), + }, + }); + const results = await collect( + runtime + .bindServerStream("WATCH_RPC_RESOURCES", isResourceView) + .open({ resourceId: "scope-1" }), + ); + expect( + results.every((value) => (value as { ok: boolean }).ok === false), + ).toBe(true); + } + }); +}); diff --git a/tests/unit/browser-rpc/browser-rpc-runtime.test.ts b/tests/unit/browser-rpc/browser-rpc-runtime.test.ts index d193364..0027308 100644 --- a/tests/unit/browser-rpc/browser-rpc-runtime.test.ts +++ b/tests/unit/browser-rpc/browser-rpc-runtime.test.ts @@ -5,6 +5,7 @@ import { createUnavailableBrowserRpcTransport, defineBrowserRpcTransport, type BrowserRpcObservation, + type BrowserRpcServerStreamLease, type BrowserRpcStreamFrame, type BrowserRpcTransport, } from "../../../src/adapters/browser-rpc/index.ts"; @@ -257,8 +258,12 @@ describe("Browser RPC provider-neutral runtime", () => { failure: { code: "UNAVAILABLE" }, }; }, - async *openServerStream() { - yield Object.freeze({ kind: "TERMINAL", ok: true }); + openServerStream() { + return serverStreamLease( + (async function* () { + yield Object.freeze({ kind: "TERMINAL" as const, ok: true as const }); + })(), + ); }, }), ).toThrow("transport is invalid"); @@ -425,11 +430,41 @@ function streamTransport( protocol: "CONNECT_HTTP", rpcKind: "SERVER_STREAM", openServerStream(call) { - return source(call.signal); + return serverStreamLease(source(call.signal)); }, }); } +/** + * RPC-RR-01. Wraps a plain frame sequence in the lease the transport contract + * requires. `cancel` resolves `waitClosed`, which is what a cooperative + * transport does. + */ +let leaseSequence = 0; + +function serverStreamLease( + frames: AsyncIterable, + options: Readonly<{ + onCancel?: (reason: string) => void; + closeOnCancel?: boolean; + }> = {}, +): BrowserRpcServerStreamLease { + leaseSequence += 1; + let release: (() => void) | undefined; + const closed = new Promise((resolve) => { + release = resolve; + }); + return Object.freeze({ + streamId: `test-stream-${leaseSequence}`, + frames, + cancel(reason: string) { + options.onCancel?.(reason); + if (options.closeOnCancel !== false) release?.(); + }, + waitClosed: () => closed, + }); +} + function message( id: string, name: string,