donghyeon.kang/clean-architecture-frontend-template
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

fix: give Browser RPC server streams a cancellable lease and a DRAINING fence

Browse Source 
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) <noreply@anthropic.com>
This commit is contained in:
DongHyeonka
2026-08-14 16:52:57 +09:00
co-authored by Claude Opus 5
parent 250531aa43
commit a7390e3b3a
6 changed files with 412 additions and 34 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/adapters/browser-rpc/browser-rpc-runtime.ts
+111 -21
View File
@@ -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<string, ActiveStreamLease>();
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<void>;
}>;
async function executeUnary<Output>(
installed: InstalledBrowserRpcContractBindings,
dependencies: BrowserRpcRuntimeDependencies,
@@ -462,11 +482,13 @@ async function* executeServerStream<Event>(
isEvent: (value: unknown) => value is Event,
generationFence: BrowserRpcGenerationFence,
clock: ClockPort,
activeStreams: Map<string, ActiveStreamLease>,
): AsyncIterable<Result<Event, AppFailure>> {
const { operation, profile, encoder, transport } = bound;
const deadlineAt = safeNowOrDeadline(clock, 0) + operation.totalDeadlineMs;
const linked = linkedAbortController(context.signal);
let iterator: AsyncIterator<BrowserRpcStreamFrame> | null = null;
let lease: BrowserRpcServerStreamLease | null = null;
let observed = false;
let messageCount = 0;
@@ -533,23 +555,66 @@ async function* executeServerStream<Event>(
);
return;
}
let stream: AsyncIterable<BrowserRpcStreamFrame>;
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<void>,
}),
);
try {
iterator = lease.frames[Symbol.asyncIterator]();
} catch {
finish("FAILED");
yield failureResult(
@@ -712,13 +777,22 @@ 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.
// 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<Event>(
.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");
}
}