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,