diff --git a/mobile/src/mobile-web-shell/bridge-host.ts b/mobile/src/mobile-web-shell/bridge-host.ts index 14410dcdfb3..4241e34a295 100644 --- a/mobile/src/mobile-web-shell/bridge-host.ts +++ b/mobile/src/mobile-web-shell/bridge-host.ts @@ -95,6 +95,8 @@ export function createBridgeHost(options: BridgeHostOptions): BridgeHost { let inFlight = 0 // One document's turn at the bridge. `close` ends it and the next `ready` begins the next one; // between the two the view belongs to no document, so nothing is served and nothing is posted. + // No epoch rides along: one native listener delivers page frames in order, so a straggler from + // the closed document is always behind it and ahead of the next document's `ready`. let serving = true let postFailureReported = false let notifyFailureReported = false diff --git a/mobile/src/mobile-web-shell/bridge/bridge-client-connection-cache.ts b/mobile/src/mobile-web-shell/bridge/bridge-client-connection-cache.ts new file mode 100644 index 00000000000..260d9bb044d --- /dev/null +++ b/mobile/src/mobile-web-shell/bridge/bridge-client-connection-cache.ts @@ -0,0 +1,85 @@ +import type { ConnectionState } from '../../transport/types' +import type { BridgeConnectionSnapshot } from './bridge-envelope' + +/** Why a `state` frame did not land. `unprimed` is a frame that beat `init`, `stale` one that lost to it. */ +export type BridgeSnapshotOutcome = 'applied' | 'stale' | 'unprimed' + +/** + * What the page's synchronous `RpcClient` getters read. + * + * Screens read `getState()` during render, so the answer has to already be here when the first one + * mounts: `init` primes it, every `state` refreshes it, and nothing is ever derived or guessed. A + * cache that answered `connecting` because it had not heard yet would move a golden. + */ +export class BridgeConnectionCache { + private held: BridgeConnectionSnapshot | null = null + private readonly listeners = new Set<(state: ConnectionState) => void>() + + read(): BridgeConnectionSnapshot | null { + return this.held + } + + /** From `init`. Re-priming with the same state is not a transition, so no listener hears one. */ + prime(snapshot: BridgeConnectionSnapshot): void { + const changed = this.held?.state !== snapshot.state + this.held = snapshot + if (changed) { + this.fanOut(snapshot.state) + } + } + + /** + * From `state`, one frame per transition on the shell's side, so every accepted one is fanned out. + * + * A snapshot whose generation went backwards is refused: the shell was rebuilt over a newer + * client and the page missed the `init` that would have said so, which makes what the page holds + * newer than what just arrived. Applying it would walk the cache backwards and leave every getter + * answering for a client that no longer exists. + */ + apply(snapshot: BridgeConnectionSnapshot): BridgeSnapshotOutcome { + const previous = this.held + if (previous === null) { + return 'unprimed' + } + if ( + previous.generation !== null && + snapshot.generation !== null && + snapshot.generation < previous.generation + ) { + return 'stale' + } + this.held = snapshot + this.fanOut(snapshot.state) + return 'applied' + } + + onStateChange(listener: (state: ConnectionState) => void): () => void { + this.listeners.add(listener) + return () => { + this.listeners.delete(listener) + } + } + + /** + * The page said goodbye. Every native client publishes `disconnected` when it closes and keeps + * answering its last snapshot afterwards, and the screens above this one are written to that: a + * getter that threw here, or a listener that never heard the transition, would leave a closing + * page rendering a dot that is still connected. + */ + close(): void { + const held = this.held + if (held !== null && held.state !== 'disconnected') { + this.held = { ...held, state: 'disconnected' } + this.fanOut('disconnected') + } + this.listeners.clear() + } + + // Walked in place: a listener that unsubscribes a sibling during the fan-out is what a `Set` + // iterator is specified to survive. + private fanOut(state: ConnectionState): void { + for (const listener of this.listeners) { + listener(state) + } + } +} diff --git a/mobile/src/mobile-web-shell/bridge/bridge-client-errors.ts b/mobile/src/mobile-web-shell/bridge/bridge-client-errors.ts new file mode 100644 index 00000000000..4445e5777f8 --- /dev/null +++ b/mobile/src/mobile-web-shell/bridge/bridge-client-errors.ts @@ -0,0 +1,50 @@ +import type { BridgeRefusal } from './bridge-caps' + +/** Everything the page's own client raises, as opposed to what it reconstructs from the shell. */ + +/** A call that needs a session the page is not in yet. Always a mount-order bug, never a retry. */ +export class BridgeClientNotReadyError extends Error { + constructor() { + super('the page bridge has no session yet; wait for init before calling the client') + this.name = 'BridgeClientNotReadyError' + } +} + +export class BridgeClientClosedError extends Error { + constructor() { + super('the page bridge was closed') + this.name = 'BridgeClientClosedError' + } +} + +/** A second `init` naming a different session: whatever the page still held belonged to the shell + * that is now gone, and the one that replaced it has never heard of any of it. */ +export class BridgeShellReplacedError extends Error { + constructor() { + super('the shell behind this page was replaced') + this.name = 'BridgeShellReplacedError' + } +} + +/** The page's copy of the shell's in-flight caps, refusing before the round trip rather than after. */ +export class BridgeClientCapExceededError extends Error { + constructor(message: string) { + super(message) + this.name = 'BridgeClientCapExceededError' + } +} + +export class BridgeReplyRefusedError extends Error { + constructor(refusal: BridgeRefusal) { + super(`the reply could not be read (${refusal})`) + this.name = 'BridgeReplyRefusedError' + } +} + +/** The frame never left the page, so this is a definite send failure and carries no delivery mark. */ +export class BridgeSendFailedError extends Error { + constructor() { + super('the request could not be posted to the shell') + this.name = 'BridgeSendFailedError' + } +} diff --git a/mobile/src/mobile-web-shell/bridge/bridge-client-init-handshake.ts b/mobile/src/mobile-web-shell/bridge/bridge-client-init-handshake.ts new file mode 100644 index 00000000000..60cb93aea40 --- /dev/null +++ b/mobile/src/mobile-web-shell/bridge/bridge-client-init-handshake.ts @@ -0,0 +1,48 @@ +/** The page asks again until the shell answers; a session has no other way to start. */ +export const BRIDGE_READY_RETRY_MIN_MS = 50 +export const BRIDGE_READY_RETRY_MAX_MS = 2000 + +export type BridgeInitHandshake = { + /** Posts `ready` now, and again on a widening backoff until `stop`. */ + start: () => void + stop: () => void + /** For a shell rebuilt under the page: the wait starts over from the floor. */ + restart: () => void +} + +/** + * How the page gets a session. + * + * The shell posts `init` when it is ready, but a page that loaded first, or reloaded after the shell + * had already sent one, would wait forever for a frame that has been and gone. Asking on a widening + * backoff costs one frame at a time and needs nothing remembered on the shell's side. + */ +export function createBridgeInitHandshake(ask: () => void): BridgeInitHandshake { + let timer: ReturnType | null = null + let delayMs = BRIDGE_READY_RETRY_MIN_MS + + function start(): void { + ask() + timer = setTimeout(() => { + delayMs = Math.min(delayMs * 2, BRIDGE_READY_RETRY_MAX_MS) + start() + }, delayMs) + } + + function stop(): void { + if (timer !== null) { + clearTimeout(timer) + timer = null + } + } + + return { + start, + stop, + restart: (): void => { + stop() + delayMs = BRIDGE_READY_RETRY_MIN_MS + start() + } + } +} diff --git a/mobile/src/mobile-web-shell/bridge/bridge-client-requests.ts b/mobile/src/mobile-web-shell/bridge/bridge-client-requests.ts new file mode 100644 index 00000000000..83c1880f80e --- /dev/null +++ b/mobile/src/mobile-web-shell/bridge/bridge-client-requests.ts @@ -0,0 +1,89 @@ +import { markRpcDeliveryUnknown } from '../../transport/rpc-delivery-ambiguity' +import type { RpcResponse } from '../../transport/types' +import { BridgeClientClosedError, BridgeReplyRefusedError } from './bridge-client-errors' +import type { BridgeReplyMessage } from './bridge-envelope' +import { BridgeReplyAssembler } from './bridge-reply-chunking' + +export type PendingRequest = { + resolve: (response: RpcResponse) => void + reject: (error: unknown) => void +} + +/** + * The page's in-flight requests, and the replies that settle them. + * + * Nothing here expires an id on its own, so every id this opens is discarded from the assembler the + * moment it settles or is abandoned: a reply whose last part never arrives would otherwise hold a + * slot until the page closes, and 64 of those are the whole in-flight budget. + */ +export class BridgeClientRequests { + private readonly pending = new Map() + private readonly assembler = new BridgeReplyAssembler() + + get size(): number { + return this.pending.size + } + + has(id: string): boolean { + return this.pending.has(id) + } + + open(id: string, request: PendingRequest): void { + this.pending.set(id, request) + } + + /** For a frame that never left the page: the caller settles it, and no part can have arrived for + * an id the shell was never told about, so there is no assembler slot to give back. */ + abandon(id: string): void { + this.pending.delete(id) + } + + acceptReply(message: BridgeReplyMessage): void { + const assembly = this.assembler.accept(message) + if (assembly.status === 'pending') { + // A part for an id nobody is waiting on still costs a slot until it is discarded. + if (!this.pending.has(message.id)) { + this.assembler.discard(message.id) + } + return + } + if (assembly.status === 'failed') { + this.fail(message.id, new BridgeReplyRefusedError(assembly.refusal)) + return + } + // A host `RpcFailure` resolves: it is data the caller reads, and the goldens record it. + this.settle(message.id, (request) => { + request.resolve(assembly.payload) + }) + } + + fail(id: string, error: unknown): void { + this.settle(id, (request) => { + request.reject(error) + }) + } + + /** + * Every pending request reaches its caller before this returns, and each one rejects + * delivery-unknown: the desktop may already have run it, and a caller told this was a definite + * send failure would offer to retry something that already happened. + */ + closeAll(reason: Error = new BridgeClientClosedError()): void { + const error = markRpcDeliveryUnknown(reason) + for (const request of this.pending.values()) { + request.reject(error) + } + this.pending.clear() + this.assembler.clear() + } + + private settle(id: string, settleWith: (request: PendingRequest) => void): void { + this.assembler.discard(id) + const request = this.pending.get(id) + if (request === undefined) { + return + } + this.pending.delete(id) + settleWith(request) + } +} diff --git a/mobile/src/mobile-web-shell/bridge/bridge-client-subscriptions.ts b/mobile/src/mobile-web-shell/bridge/bridge-client-subscriptions.ts new file mode 100644 index 00000000000..cbcdc9a1e03 --- /dev/null +++ b/mobile/src/mobile-web-shell/bridge/bridge-client-subscriptions.ts @@ -0,0 +1,183 @@ +import type { BrowserScreencastFrame } from '../../transport/browser-screencast-protocol' +import { + BRIDGE_PROTOCOL_VERSION, + type BridgeClientMessage, + type BridgeHostMessage +} from './bridge-envelope' +import { decodeBridgeScreencastFrame, type BridgeBinaryEvent } from './bridge-screencast-binary' + +type BridgeEventMessage = Extract + +/** Derived from the envelope's closed list, the same way the shell's ledger derives it: a reason + * added there is a compile error here rather than one this side silently never sees. */ +export type BridgeStreamEndReason = Extract['reason'] + +/** + * How far behind the page lets itself fall before it acks. + * + * The shell ends a stream at 256 unacked frames or 4 MiB. A quarter of each leaves room for the + * frames already in flight when an ack is posted, so a page that is keeping up never walks the + * shell's window down to the point where it ends a stream. `bridge-rpc-client-frames.test.ts` pins + * the ratio against the shell's own numbers. + */ +export const BRIDGE_ACK_INTERVAL_FRAMES = 64 +export const BRIDGE_ACK_INTERVAL_BYTES = 1024 * 1024 + +/** + * What a listener is handed when its stream dies under it, in the shape the native client's + * `emitError` uses. Consumers read `type` and act on it — `host-worktree-refresh.ts` clears the flag + * that says the event stream is live — so a stream that merely stops delivering leaves them waiting + * on a replay that is never coming. + */ +export type BridgeStreamErrorResult = { type: 'error'; message: string; error?: unknown } + +export function bridgeStreamError(message: string, error?: unknown): BridgeStreamErrorResult { + return error === undefined ? { type: 'error', message } : { type: 'error', message, error } +} + +type OpenStream = { + onData: (result: unknown) => void + onBinaryFrame?: (frame: BrowserScreencastFrame) => void + lastSeq: number + unackedFrames: number + unackedBytes: number +} + +type SubscriptionsOptions = { + /** False when the frame never left the page. */ + send: (frame: BridgeClientMessage) => boolean + /** A binary frame with no listener or no decodable image. Neither is recoverable in place. */ + onDroppedBinaryFrame: () => void +} + +/** Every stream the page opened, and the ack it owes the shell for each one. */ +export class BridgeClientSubscriptions { + private streams = new Map() + + constructor(private readonly options: SubscriptionsOptions) {} + + get size(): number { + return this.streams.size + } + + has(id: string): boolean { + return this.streams.has(id) + } + + /** False when the `subscribe` never left the page. The shell has not heard of the stream, so + * nothing will ever end it: the slot goes back here and the listener is told, which is what the + * native client does with a subscribe it could not send. */ + open( + id: string, + method: string, + params: unknown, + onData: (result: unknown) => void, + onBinaryFrame?: (frame: BrowserScreencastFrame) => void + ): boolean { + this.streams.set(id, { onData, onBinaryFrame, lastSeq: 0, unackedFrames: 0, unackedBytes: 0 }) + const sent = this.options.send({ + v: BRIDGE_PROTOCOL_VERSION, + type: 'subscribe', + id, + method, + params, + // Asked for only when there is something to hand the frames to, so a shell that pays to + // encode binary is one a listener is waiting on. + ...(onBinaryFrame === undefined ? {} : { wantsBinary: true }) + }) + if (sent) { + return true + } + this.streams.delete(id) + onData(bridgeStreamError('the subscribe could not be posted to the shell')) + return false + } + + /** `bytes` is the raw frame as the shell measured it, so both sides' windows agree exactly. */ + deliver(message: BridgeEventMessage, bytes: number): void { + const stream = this.streams.get(message.id) + if (stream === undefined) { + return + } + stream.lastSeq = message.seq + stream.unackedFrames += 1 + stream.unackedBytes += bytes + // Acked before the listener runs: the frame was received and read either way, and a listener + // that throws must not also wedge the stream by stranding the ack behind it. + this.ackIfDue(message.id, stream) + if ('binary' in message) { + this.deliverBinary(stream, message.binary) + return + } + stream.onData(message.payload) + } + + /** The shell already retired this stream, so nothing is posted back for it. The listener is told + * before the record goes: frames that merely stop arriving are indistinguishable from a quiet + * stream, and a consumer waiting on a replay would wait for the life of the document. */ + end(id: string, message: string, error?: unknown): void { + const stream = this.streams.get(id) + if (stream === undefined) { + return + } + this.streams.delete(id) + stream.onData(bridgeStreamError(message, error)) + } + + /** The page is done with the stream. Idempotent: a second dispose posts nothing. */ + cancel(id: string): void { + if (!this.streams.delete(id)) { + return + } + this.options.send({ + v: BRIDGE_PROTOCOL_VERSION, + type: 'cancel', + id, + target: 'subscription' + }) + } + + /** For `close`, which is the shell's authority to tear down both sides: a cancel per stream + * ahead of it would say the same thing twice. Silent, because the page asked for this one. */ + closeAll(): void { + this.streams.clear() + } + + /** For a shell replaced under the page: every stream it was serving died with it, and the + * listeners are the only ones in a position to do anything about that. */ + failAll(message: string): void { + // Out of the ledger before any listener runs: one that resubscribes on the way down is opening + // a stream against the shell that is arriving, and this loop must not take that one with it. + const ended = this.streams + this.streams = new Map() + for (const stream of ended.values()) { + stream.onData(bridgeStreamError(message)) + } + } + + private deliverBinary(stream: OpenStream, binary: BridgeBinaryEvent): void { + const onBinaryFrame = stream.onBinaryFrame + if (onBinaryFrame === undefined) { + this.options.onDroppedBinaryFrame() + return + } + const frame = decodeBridgeScreencastFrame(binary) + if (frame === null) { + this.options.onDroppedBinaryFrame() + return + } + onBinaryFrame(frame) + } + + private ackIfDue(id: string, stream: OpenStream): void { + if ( + stream.unackedFrames < BRIDGE_ACK_INTERVAL_FRAMES && + stream.unackedBytes < BRIDGE_ACK_INTERVAL_BYTES + ) { + return + } + stream.unackedFrames = 0 + stream.unackedBytes = 0 + this.options.send({ v: BRIDGE_PROTOCOL_VERSION, type: 'ack', id, seq: stream.lastSeq }) + } +} diff --git a/mobile/src/mobile-web-shell/bridge/bridge-port-pair-test-harness.ts b/mobile/src/mobile-web-shell/bridge/bridge-port-pair-test-harness.ts new file mode 100644 index 00000000000..0758a09e692 --- /dev/null +++ b/mobile/src/mobile-web-shell/bridge/bridge-port-pair-test-harness.ts @@ -0,0 +1,160 @@ +import { createBridgeHost, type BridgeHost, type BridgeHostDiagnostic } from '../bridge-host' +import { createFakeRpcClient, type FakeRpcClient } from '../bridge-host-test-fakes' +import { + readBridgeClientMessage, + readBridgeHostMessage, + type BridgeClientMessage, + type BridgeHostMessage +} from './bridge-envelope' +import { + createBridgeRpcClient, + type BridgeRpcClient, + type BridgeRpcClientDiagnostic +} from './bridge-rpc-client' + +/** + * The page and the shell wired to each other through the weakest transport that is still a + * transport, so a test of either one is a test of the pair. + * + * Two properties are the whole point. One FIFO per direction, because a `subscribe` that overtook a + * `sendRequest` would move the recorder's shared ordinal, which is what `write-ordinal.ts` exists to + * catch. And delivery on a microtask, the weakest async the golden runner's zero-time drains flush + * and the only one that moves no virtual millisecond. + */ +export type BridgePortPair = { + client: BridgeRpcClient + host: BridgeHost + rpc: FakeRpcClient + /** Everything each side posted, in the order it was posted, raw. */ + toShell: string[] + toPage: string[] + diagnostics: BridgeRpcClientDiagnostic[] + hostDiagnostics: BridgeHostDiagnostic[] + /** Runs both lanes until a full round moves nothing. */ + flush: () => Promise + /** Read back through the reader on the receiving side, so a frame this returns is one that lands. */ + readToShell: () => BridgeClientMessage[] + readToPage: () => BridgeHostMessage[] +} + +export type BridgePortPairOptions = { + rpc?: FakeRpcClient + sessionId?: string + buildId?: string +} + +type Lane = { + sent: string[] + push: (json: string) => void + readonly depth: number +} + +function createLane(deliver: (json: string) => void): Lane { + const sent: string[] = [] + const queue: string[] = [] + let scheduled = false + function drain(): void { + scheduled = false + const next = queue.shift() + if (next === undefined) { + return + } + deliver(next) + schedule() + } + function schedule(): void { + if (scheduled || queue.length === 0) { + return + } + scheduled = true + void Promise.resolve().then(drain) + } + return { + sent, + push(json: string): void { + sent.push(json) + queue.push(json) + schedule() + }, + get depth(): number { + return queue.length + } + } +} + +function readAll( + frames: readonly string[], + read: (json: string) => { ok: true; message: TMessage } | { ok: false; refusal: string } +): TMessage[] { + return frames.map((json) => { + const parsed = read(json) + if (!parsed.ok) { + throw new Error(`the other side would have refused this frame: ${parsed.refusal}`) + } + return parsed.message + }) +} + +export function createBridgePortPair(options: BridgePortPairOptions = {}): BridgePortPair { + const rpc = options.rpc ?? createFakeRpcClient() + const diagnostics: BridgeRpcClientDiagnostic[] = [] + const hostDiagnostics: BridgeHostDiagnostic[] = [] + let receiveOnPage: ((json: string) => void) | null = null + + const toPage = createLane((json) => { + receiveOnPage?.(json) + }) + const host = createBridgeHost({ + client: rpc, + post: (json) => { + toPage.push(json) + return Promise.resolve() + }, + buildId: options.buildId ?? 'build-a', + sessionId: options.sessionId ?? 'session-a', + onDiagnostic: (diagnostic) => hostDiagnostics.push(diagnostic) + }) + const toShell = createLane((json) => { + host.receive(json) + }) + const client = createBridgeRpcClient({ + send: (json) => { + toShell.push(json) + }, + onMessage: (handler) => { + receiveOnPage = handler + return () => { + receiveOnPage = null + } + }, + onDiagnostic: (diagnostic) => diagnostics.push(diagnostic) + }) + + return { + client, + host, + rpc, + toShell: toShell.sent, + toPage: toPage.sent, + diagnostics, + hostDiagnostics, + async flush(): Promise { + for (let round = 0; round < 64; round += 1) { + const moved = toShell.sent.length + toPage.sent.length + for (let turn = 0; turn < 8; turn += 1) { + await Promise.resolve() + } + const quiet = + toShell.depth === 0 && + toPage.depth === 0 && + moved === toShell.sent.length + toPage.sent.length + if (quiet) { + return + } + } + throw new Error('the port pair never went quiet') + }, + readToShell: () => readAll(toShell.sent, readBridgeClientMessage), + readToPage: () => readAll(toPage.sent, readBridgeHostMessage) + } +} diff --git a/mobile/src/mobile-web-shell/bridge/bridge-reply-chunking.ts b/mobile/src/mobile-web-shell/bridge/bridge-reply-chunking.ts index 5545030ad65..b05a0e2de85 100644 --- a/mobile/src/mobile-web-shell/bridge/bridge-reply-chunking.ts +++ b/mobile/src/mobile-web-shell/bridge/bridge-reply-chunking.ts @@ -127,10 +127,12 @@ const BRIDGE_MAX_ASSEMBLING_BYTES = BRIDGE_MAX_REPLY_BYTES * 4 /** * Parts may arrive in any order, so they are held by index rather than appended. * - * A failed id stays failed. Dropping it and starting over on the next part is what lets a sender - * walk past the ceiling one refusal at a time, so the refusal is remembered and every later part - * for that id gets the same answer. `discard` is how the page says the id is finished with, which - * is also how it becomes usable again. + * A failed id stays failed while the page still holds it. Dropping the refusal and starting over on + * the next part is what would let a sender walk past the ceiling one refusal at a time, so every + * later part for that id gets the same answer instead. Nothing is remembered for long: `discard` + * reopens the id, and the page's request ledger calls it as it settles the caller, so the tombstone + * normally lives no longer than the rest of the reply that raised it. The bound below is for the + * ids nothing settles. * * The number of ids held at once is bounded by the in-flight request cap, since a reply only exists * for a request the page made, and their bytes together by `BRIDGE_MAX_ASSEMBLING_BYTES`. Nothing diff --git a/mobile/src/mobile-web-shell/bridge/bridge-rpc-client-frames.test.ts b/mobile/src/mobile-web-shell/bridge/bridge-rpc-client-frames.test.ts new file mode 100644 index 00000000000..7c2ef099ce3 --- /dev/null +++ b/mobile/src/mobile-web-shell/bridge/bridge-rpc-client-frames.test.ts @@ -0,0 +1,759 @@ +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' +import { BrowserScreencastOpcode } from '../../transport/browser-screencast-protocol' +import { isRpcDeliveryUnknown } from '../../transport/rpc-delivery-ambiguity' +import { + BRIDGE_MAX_MESSAGE_BYTES, + BRIDGE_MAX_PENDING_REQUESTS, + BRIDGE_MAX_SUBSCRIPTIONS +} from './bridge-caps' +import { BRIDGE_MAX_UNACKED_BYTES, BRIDGE_MAX_UNACKED_FRAMES } from '../bridge-host-subscriptions' +import { + BRIDGE_ACK_INTERVAL_BYTES, + BRIDGE_ACK_INTERVAL_FRAMES +} from './bridge-client-subscriptions' +import { + BRIDGE_PROTOCOL_VERSION, + readBridgeClientMessage, + type BridgeClientMessage, + type BridgeHostMessage +} from './bridge-envelope' +import { + BRIDGE_READY_RETRY_MAX_MS, + BRIDGE_READY_RETRY_MIN_MS +} from './bridge-client-init-handshake' +import { + BridgeClientCapExceededError, + BridgeClientClosedError, + BridgeClientNotReadyError, + createBridgeRpcClient, + type BridgeRpcClientDiagnostic +} from './bridge-rpc-client' + +const CONNECTION = { + state: 'connected', + reconnectAttempt: 2, + lastConnectedAt: 1700, + lastInboundAt: 1800, + generation: 5 +} as const + +const INIT: BridgeHostMessage = { + v: BRIDGE_PROTOCOL_VERSION, + type: 'init', + sessionId: 'session-a', + buildId: 'build-a', + connection: CONNECTION, + grants: { rpc: { maxPendingRequests: 64, maxSubscriptions: 32 }, native: [] } +} + +type PageClientOptions = { + send?: (json: string) => void + /** A port that ignores its own unsubscribe, which is the only way to observe the read guard. */ + keepDeliveringAfterUnsubscribe?: boolean +} + +function createPageClient(options: PageClientOptions = {}) { + const sent: string[] = [] + const diagnostics: BridgeRpcClientDiagnostic[] = [] + let handler: ((json: string) => void) | null = null + const client = createBridgeRpcClient({ + send: (json) => { + sent.push(json) + options.send?.(json) + }, + onMessage: (received) => { + handler = received + return () => { + if (options.keepDeliveringAfterUnsubscribe !== true) { + handler = null + } + } + }, + onDiagnostic: (diagnostic) => { + diagnostics.push(diagnostic) + } + }) + return { + client, + sent, + diagnostics, + deliver(frame: unknown): void { + handler?.(JSON.stringify(frame)) + }, + deliverRaw(json: string): void { + handler?.(json) + }, + frames(): BridgeClientMessage[] { + return sent.map((json) => { + const read = readBridgeClientMessage(json) + if (!read.ok) { + throw new Error(`the shell would have refused this frame: ${read.refusal}`) + } + return read.message + }) + }, + start(): void { + this.deliver(INIT) + } + } +} + +/** Narrows what a rejection handed back, so a test reads an error rather than asserting one. */ +function readError(thrown: unknown): Error { + if (!(thrown instanceof Error)) { + throw new Error(`expected an Error, got ${typeof thrown}`) + } + return thrown +} + +function eventFrame(id: string, seq: number, payload: unknown): BridgeHostMessage { + return { v: BRIDGE_PROTOCOL_VERSION, type: 'event', id, seq, payload } +} + +/** The id the client minted for the nth exchange it opened, read back off its own frame. */ +function idOf(page: ReturnType, index: number): string { + const frame = page.frames().filter((message) => 'id' in message)[index] + if (frame === undefined || !('id' in frame)) { + throw new Error('the page opened no such exchange') + } + return frame.id +} + +beforeEach(() => { + vi.useFakeTimers() +}) + +afterEach(() => { + vi.useRealTimers() +}) + +describe('bridge client handshake', () => { + it('asks for a session as soon as it exists', () => { + const page = createPageClient() + expect(page.frames()).toEqual([{ v: BRIDGE_PROTOCOL_VERSION, type: 'ready' }]) + }) + + it('keeps asking on a widening backoff until init answers', () => { + const page = createPageClient() + vi.advanceTimersByTime(BRIDGE_READY_RETRY_MIN_MS) + expect(page.sent).toHaveLength(2) + vi.advanceTimersByTime(BRIDGE_READY_RETRY_MIN_MS) + expect(page.sent).toHaveLength(2) + vi.advanceTimersByTime(BRIDGE_READY_RETRY_MIN_MS) + expect(page.sent).toHaveLength(3) + vi.advanceTimersByTime(BRIDGE_READY_RETRY_MAX_MS * 4) + expect(page.sent.length).toBeGreaterThan(3) + }) + + it('asks no less often than the ceiling, however long the shell stays quiet', () => { + const page = createPageClient() + // Past the ceiling: doubling from the floor reaches it in six steps. An unclamped backoff is + // the same thing for a minute and then a page that gives up on a shell booting behind it. + vi.advanceTimersByTime(BRIDGE_READY_RETRY_MAX_MS * 4) + const asked = page.sent.length + vi.advanceTimersByTime(BRIDGE_READY_RETRY_MAX_MS) + expect(page.sent).toHaveLength(asked + 1) + vi.advanceTimersByTime(BRIDGE_READY_RETRY_MAX_MS * 10) + expect(page.sent).toHaveLength(asked + 11) + }) + + it('stops asking once init lands', () => { + const page = createPageClient() + page.start() + vi.advanceTimersByTime(BRIDGE_READY_RETRY_MAX_MS * 10) + expect(page.sent).toHaveLength(1) + }) + + it('keeps what it holds when the same shell answers a second time', async () => { + const page = createPageClient() + page.start() + const onData = vi.fn() + const answer = page.client.sendRequest('worktree.ps') + page.client.subscribe('terminal.stream', {}, onData) + const id = idOf(page, 1) + // Every `ready` is answered, so a page that re-asked before the first init landed hears two. + page.deliver(INIT) + page.deliver(eventFrame(id, 1, 'still live')) + expect(onData.mock.calls).toEqual([['still live']]) + page.deliver({ + v: BRIDGE_PROTOCOL_VERSION, + type: 'reply', + id: idOf(page, 0), + payload: { id: 'wire-1', ok: true, result: 'ok', _meta: { runtimeId: 'runtime-a' } } + }) + await expect(answer).resolves.toMatchObject({ ok: true }) + }) + + it('settles everything the shell it lost was holding before adopting the new one', async () => { + const page = createPageClient() + page.start() + const onData = vi.fn() + const answer = page.client.sendRequest('worktree.ps') + page.client.subscribe('terminal.stream', {}, onData) + // A rebuilt host under the same page: its tables are empty, so nothing the page still holds + // would ever be answered or ended from there. + page.deliver({ ...INIT, sessionId: 'session-b' }) + const error = await answer.catch((thrown: unknown) => thrown) + expect(readError(error).name).toBe('BridgeShellReplacedError') + expect(isRpcDeliveryUnknown(error)).toBe(true) + expect(onData.mock.calls).toEqual([[{ type: 'error', message: expect.any(String) }]]) + expect(page.client.getShellSession()?.sessionId).toBe('session-b') + }) + + it('reads the connection snapshot init primed it with', () => { + const page = createPageClient() + page.start() + expect(page.client.getState()).toBe('connected') + expect(page.client.getReconnectAttempt()).toBe(2) + expect(page.client.getLastConnectedAt()).toBe(1700) + expect(page.client.getLastInboundAt?.()).toBe(1800) + expect(page.client.getGeneration?.()).toBe(5) + expect(page.client.getShellSession()).toEqual({ + sessionId: 'session-a', + buildId: 'build-a', + grants: INIT.grants + }) + }) + + it('answers a generation the shell does not keep with a constant epoch', () => { + const page = createPageClient() + page.deliver({ ...INIT, connection: { ...CONNECTION, generation: null } }) + expect(page.client.getGeneration?.()).toBe(0) + }) + + it('tells a waiting listener once, and a late one immediately', () => { + const page = createPageClient() + const early = vi.fn() + const dropped = vi.fn() + const release = page.client.onReady(dropped) + page.client.onReady(early) + release() + page.start() + expect(early).toHaveBeenCalledTimes(1) + expect(dropped).not.toHaveBeenCalled() + const late = vi.fn() + page.client.onReady(late) + expect(late).toHaveBeenCalledTimes(1) + page.deliver(INIT) + expect(early).toHaveBeenCalledTimes(1) + }) +}) + +describe('bridge client before a session', () => { + it('refuses every member that would have to answer for one', () => { + const page = createPageClient() + expect(() => page.client.getState()).toThrow(BridgeClientNotReadyError) + expect(() => page.client.getReconnectAttempt()).toThrow(BridgeClientNotReadyError) + expect(() => page.client.getLastConnectedAt()).toThrow(BridgeClientNotReadyError) + expect(() => page.client.getLastInboundAt?.()).toThrow(BridgeClientNotReadyError) + expect(() => page.client.getGeneration?.()).toThrow(BridgeClientNotReadyError) + expect(() => page.client.sendRequest('worktree.ps')).toThrow(BridgeClientNotReadyError) + expect(() => page.client.subscribe('terminal.stream', {}, vi.fn())).toThrow( + BridgeClientNotReadyError + ) + expect(() => page.client.notifyForeground()).toThrow(BridgeClientNotReadyError) + expect(() => + page.client.updateTerminalSubscriptionViewport('t', { cols: 80, rows: 24 }) + ).toThrow(BridgeClientNotReadyError) + expect(page.sent).toHaveLength(1) + }) + + it('still registers a state listener and still closes', () => { + const page = createPageClient() + const listener = vi.fn() + expect(() => page.client.onStateChange(listener)()).not.toThrow() + expect(() => { + page.client.close() + }).not.toThrow() + }) + + it('drops a state frame that beat init rather than priming from it', () => { + const page = createPageClient() + page.deliver({ + v: BRIDGE_PROTOCOL_VERSION, + type: 'state', + connection: { ...CONNECTION, state: 'reconnecting' } + }) + expect(() => page.client.getState()).toThrow(BridgeClientNotReadyError) + expect(page.diagnostics).toEqual([]) + }) +}) + +describe('bridge client after close', () => { + it('goes inert instead of throwing into a teardown, and posts nothing more', async () => { + const page = createPageClient() + page.start() + page.client.close() + expect(page.frames().at(-1)).toEqual({ v: BRIDGE_PROTOCOL_VERSION, type: 'close' }) + const refused = page.client.sendRequest('worktree.ps') + await expect(refused).rejects.toThrow(BridgeClientClosedError) + expect(() => page.client.subscribe('terminal.stream', {}, vi.fn())()).not.toThrow() + expect(() => page.client.notifyForeground()).not.toThrow() + expect(() => { + page.client.updateTerminalSubscriptionViewport('t', { cols: 80, rows: 24 }) + }).not.toThrow() + page.client.close() + page.deliver(INIT) + expect(page.sent).toHaveLength(2) + }) + + it('publishes disconnected and keeps answering the snapshot it last held', () => { + const page = createPageClient() + page.start() + const listener = vi.fn() + page.client.onStateChange(listener) + page.client.close() + expect(listener).toHaveBeenCalledWith('disconnected') + expect(page.client.getState()).toBe('disconnected') + expect(page.client.getReconnectAttempt()).toBe(CONNECTION.reconnectAttempt) + expect(page.client.getLastConnectedAt()).toBe(CONNECTION.lastConnectedAt) + expect(page.client.getLastInboundAt?.()).toBe(CONNECTION.lastInboundAt) + expect(page.client.getGeneration?.()).toBe(CONNECTION.generation) + }) + + it('answers nothing it never heard: a close before init leaves the getters unready', () => { + const page = createPageClient() + const listener = vi.fn() + page.client.onStateChange(listener) + page.client.close() + expect(listener).not.toHaveBeenCalled() + expect(() => page.client.getState()).toThrow(BridgeClientNotReadyError) + }) + + it('reads nothing more, even from a port that kept delivering', () => { + const page = createPageClient({ keepDeliveringAfterUnsubscribe: true }) + page.start() + page.client.subscribe('terminal.stream', {}, vi.fn()) + const id = idOf(page, 0) + page.client.close() + page.deliver(INIT) + page.deliver(eventFrame(id, 1, 'late')) + page.deliverRaw('{ not json') + expect(page.diagnostics).toEqual([]) + expect(page.client.getState()).toBe('disconnected') + }) + + it('says goodbye once, without a cancel for each stream it owned', () => { + const page = createPageClient() + page.start() + page.client.subscribe('terminal.stream', {}, vi.fn()) + page.client.subscribe('terminal.stream', {}, vi.fn()) + page.client.close() + expect(page.frames().filter((frame) => frame.type === 'cancel')).toEqual([]) + expect(page.frames().filter((frame) => frame.type === 'close')).toHaveLength(1) + }) +}) + +describe('bridge client replies', () => { + it('rejects with the class and the delivery mark the shell captured', async () => { + const page = createPageClient() + page.start() + const answer = page.client.sendRequest('worktree.ps') + page.deliver({ + v: BRIDGE_PROTOCOL_VERSION, + type: 'error', + id: idOf(page, 0), + error: { + category: 'RpcTimeoutError', + message: 'timed out', + isRpcDeliveryUnknown: true, + code: 'ETIMEDOUT', + cause: { category: 'Error', message: 'socket closed', isRpcDeliveryUnknown: false } + } + }) + const error = await answer.catch((thrown: unknown) => thrown) + expect(error).toBeInstanceOf(Error) + expect(readError(error).name).toBe('RpcTimeoutError') + expect(isRpcDeliveryUnknown(error)).toBe(true) + expect(readError(readError(error).cause).message).toBe('socket closed') + }) + + it('rejects a reply the assembler refuses', async () => { + const page = createPageClient() + page.start() + const answer = page.client.sendRequest('worktree.ps') + const id = idOf(page, 0) + const part = { + v: BRIDGE_PROTOCOL_VERSION, + type: 'reply', + id, + part: { i: 0, of: 2 }, + chunk: '{' + } + page.deliver(part) + page.deliver(part) + await expect(answer).rejects.toThrow('duplicate-part') + }) + + it('drops a reply or an error for an id it never opened, and says so', () => { + const page = createPageClient() + page.start() + const stranger = 'z'.repeat(22) + page.deliver({ + v: BRIDGE_PROTOCOL_VERSION, + type: 'reply', + id: stranger, + payload: { id: stranger, ok: true, result: 1, _meta: { runtimeId: 'runtime-a' } } + }) + page.deliver({ + v: BRIDGE_PROTOCOL_VERSION, + type: 'error', + id: stranger, + error: { category: 'Error', message: 'gone', isRpcDeliveryUnknown: false } + }) + expect(page.diagnostics).toEqual([{ kind: 'unknown-id' }, { kind: 'unknown-id' }]) + }) + + it('frees the assembler slot of every id nobody is waiting on', async () => { + const page = createPageClient() + page.start() + const answer = page.client.sendRequest('worktree.ps') + const id = idOf(page, 0) + for (let index = 0; index < BRIDGE_MAX_PENDING_REQUESTS * 2; index += 1) { + page.deliver({ + v: BRIDGE_PROTOCOL_VERSION, + type: 'reply', + id: index.toString(36).padStart(22, 'z'), + part: { i: 0, of: 2 }, + chunk: '{"a":' + }) + } + const payload = { id, ok: true, result: 7, _meta: { runtimeId: 'runtime-a' } } + const serialized = JSON.stringify(payload) + const cut = Math.floor(serialized.length / 2) + page.deliver({ + v: BRIDGE_PROTOCOL_VERSION, + type: 'reply', + id, + part: { i: 0, of: 2 }, + chunk: serialized.slice(0, cut) + }) + page.deliver({ + v: BRIDGE_PROTOCOL_VERSION, + type: 'reply', + id, + part: { i: 1, of: 2 }, + chunk: serialized.slice(cut) + }) + await expect(answer).resolves.toEqual(payload) + }) + + it('gives back the assembler slot of every id it settles', async () => { + const page = createPageClient() + page.start() + const settled: Promise[] = [] + for (let index = 0; index < BRIDGE_MAX_PENDING_REQUESTS; index += 1) { + const abandoned = page.client.sendRequest('worktree.ps') + const id = idOf(page, index) + page.deliver({ + v: BRIDGE_PROTOCOL_VERSION, + type: 'reply', + id, + part: { i: 0, of: 2 }, + chunk: '{"a":' + }) + page.deliver({ + v: BRIDGE_PROTOCOL_VERSION, + type: 'error', + id, + error: { category: 'Error', message: 'gone', isRpcDeliveryUnknown: false } + }) + settled.push(abandoned.catch(() => undefined)) + } + const answer = page.client.sendRequest('worktree.ps') + const id = idOf(page, BRIDGE_MAX_PENDING_REQUESTS) + const payload = { id, ok: true, result: 'assembled', _meta: { runtimeId: 'runtime-a' } } + const serialized = JSON.stringify(payload) + const cut = Math.floor(serialized.length / 2) + page.deliver({ + v: BRIDGE_PROTOCOL_VERSION, + type: 'reply', + id, + part: { i: 0, of: 2 }, + chunk: serialized.slice(0, cut) + }) + page.deliver({ + v: BRIDGE_PROTOCOL_VERSION, + type: 'reply', + id, + part: { i: 1, of: 2 }, + chunk: serialized.slice(cut) + }) + await expect(answer).resolves.toEqual(payload) + await Promise.all(settled) + }) +}) + +describe('bridge client refusals and send failures', () => { + it('reports a frame its own reader will not take, and changes nothing', () => { + const page = createPageClient() + page.start() + page.deliverRaw('{ not json') + page.deliverRaw(JSON.stringify({ v: 99, type: 'state' })) + page.deliverRaw(`"${'z'.repeat(BRIDGE_MAX_MESSAGE_BYTES)}"`) + expect(page.diagnostics).toEqual([ + { kind: 'refused', refusal: 'malformed-json' }, + { kind: 'refused', refusal: 'unrecognised-message' }, + { kind: 'refused', refusal: 'oversized' } + ]) + expect(page.client.getState()).toBe('connected') + }) + + it('fails a request whose frame never left the page, without the delivery mark', async () => { + let live = true + const page = createPageClient({ + send: () => { + if (!live) { + throw new Error('the port is gone') + } + } + }) + page.start() + live = false + const answer = page.client.sendRequest('worktree.ps') + const error = await answer.catch((thrown: unknown) => thrown) + expect(readError(error).name).toBe('BridgeSendFailedError') + expect(isRpcDeliveryUnknown(error)).toBe(false) + expect(page.diagnostics.at(-1)).toEqual({ + kind: 'send-failed', + error: expect.any(Error) + }) + }) +}) + +describe('bridge client caps', () => { + it('refuses the request past the shell grant without a round trip', async () => { + const page = createPageClient() + page.start() + const answers: Promise[] = [] + for (let index = 0; index < BRIDGE_MAX_PENDING_REQUESTS; index += 1) { + answers.push(page.client.sendRequest('worktree.ps')) + } + const refused = page.client.sendRequest('worktree.ps') + await expect(refused).rejects.toThrow(BridgeClientCapExceededError) + expect(page.sent).toHaveLength(1 + BRIDGE_MAX_PENDING_REQUESTS) + page.client.close() + await Promise.allSettled(answers) + }) + + it('frees the page slot when the subscribe frame never left the page', () => { + let live = true + const page = createPageClient({ + send: () => { + if (!live) { + throw new Error('the port is gone') + } + } + }) + page.start() + live = false + const onData = vi.fn() + // Every one of these is a slot the shell was never told about, and nothing will ever end it. + for (let index = 0; index < BRIDGE_MAX_SUBSCRIPTIONS; index += 1) { + page.client.subscribe('terminal.stream', {}, onData) + } + expect(onData).toHaveBeenCalledTimes(BRIDGE_MAX_SUBSCRIPTIONS) + expect(onData.mock.calls.at(-1)?.[0]).toEqual({ type: 'error', message: expect.any(String) }) + expect(page.diagnostics).toHaveLength(BRIDGE_MAX_SUBSCRIPTIONS) + live = true + // Short of this, the page is at its cap for the life of the document: only a reload clears it. + const dispose = page.client.subscribe('terminal.stream', {}, vi.fn()) + dispose() + expect(page.frames().filter((frame) => frame.type === 'cancel')).toHaveLength(1) + }) + + it('refuses the subscription past the shell grant at the call site', () => { + const page = createPageClient() + page.start() + for (let index = 0; index < BRIDGE_MAX_SUBSCRIPTIONS; index += 1) { + page.client.subscribe('terminal.stream', {}, vi.fn()) + } + expect(() => page.client.subscribe('terminal.stream', {}, vi.fn())).toThrow( + BridgeClientCapExceededError + ) + expect(page.sent).toHaveLength(1 + BRIDGE_MAX_SUBSCRIPTIONS) + }) +}) + +describe('bridge client acks', () => { + it('stays well inside the window the shell ends a stream at', () => { + // The shell's own numbers, not a copy of them: a window narrowed there has to fail here. + expect(BRIDGE_ACK_INTERVAL_FRAMES * 4).toBeLessThanOrEqual(BRIDGE_MAX_UNACKED_FRAMES) + expect(BRIDGE_ACK_INTERVAL_BYTES * 4).toBeLessThanOrEqual(BRIDGE_MAX_UNACKED_BYTES) + }) + + it('acks the last seq it read once the frame interval is due', () => { + const page = createPageClient() + page.start() + page.client.subscribe('terminal.stream', {}, vi.fn()) + const id = idOf(page, 0) + for (let seq = 1; seq < BRIDGE_ACK_INTERVAL_FRAMES; seq += 1) { + page.deliver(eventFrame(id, seq, seq)) + } + expect(page.frames().filter((frame) => frame.type === 'ack')).toEqual([]) + page.deliver(eventFrame(id, BRIDGE_ACK_INTERVAL_FRAMES, 'last')) + expect(page.frames().filter((frame) => frame.type === 'ack')).toEqual([ + { v: BRIDGE_PROTOCOL_VERSION, type: 'ack', id, seq: BRIDGE_ACK_INTERVAL_FRAMES } + ]) + }) + + it('acks early when the bytes are due before the frames are', () => { + const page = createPageClient() + page.start() + page.client.subscribe('terminal.stream', {}, vi.fn()) + const id = idOf(page, 0) + const heavy = 'z'.repeat(BRIDGE_MAX_MESSAGE_BYTES - 1024) + page.deliver(eventFrame(id, 1, heavy)) + page.deliver(eventFrame(id, 2, heavy)) + expect(page.frames().filter((frame) => frame.type === 'ack')).toEqual([ + { v: BRIDGE_PROTOCOL_VERSION, type: 'ack', id, seq: 2 } + ]) + }) + + it('acks a frame whose listener throws, so a listener bug cannot wedge the stream', () => { + const page = createPageClient() + page.start() + page.client.subscribe('terminal.stream', {}, () => { + throw new Error('listener bug') + }) + const id = idOf(page, 0) + for (let seq = 1; seq <= BRIDGE_ACK_INTERVAL_FRAMES; seq += 1) { + expect(() => page.deliver(eventFrame(id, seq, seq))).toThrow('listener bug') + } + expect(page.frames().filter((frame) => frame.type === 'ack')).toHaveLength(1) + }) + + it('ignores an event for a stream it already disposed', () => { + const page = createPageClient() + page.start() + const onData = vi.fn() + const dispose = page.client.subscribe('terminal.stream', {}, onData) + const id = idOf(page, 0) + dispose() + dispose() + page.deliver(eventFrame(id, 1, 'late')) + expect(onData).not.toHaveBeenCalled() + expect(page.frames().filter((frame) => frame.type === 'cancel')).toHaveLength(1) + }) + + it('posts no cancel for a stream the shell ended before the page let go', () => { + const page = createPageClient() + page.start() + const dispose = page.client.subscribe('terminal.stream', {}, vi.fn()) + const id = idOf(page, 0) + page.deliver({ v: BRIDGE_PROTOCOL_VERSION, type: 'end', id, reason: 'closed' }) + // The screen unmounts on its own schedule, which is routinely after the shell gave up. + dispose() + expect(page.frames().filter((frame) => frame.type === 'cancel')).toEqual([]) + }) + + it('retires a stream the shell ended, tells the listener, and reports why', () => { + const page = createPageClient() + page.start() + const onData = vi.fn() + page.client.subscribe('terminal.stream', {}, onData) + const id = idOf(page, 0) + page.deliver({ v: BRIDGE_PROTOCOL_VERSION, type: 'end', id, reason: 'overflow' }) + page.deliver(eventFrame(id, 1, 'after the end')) + // The terminal result is the only thing a consumer hears. `host-worktree-refresh.ts` reads it + // to clear the flag that says the event stream is live; without it the list never refreshes + // again, because frames that stop arriving look exactly like a stream with nothing to say. + expect(onData.mock.calls).toEqual([[{ type: 'error', message: expect.any(String) }]]) + expect(page.diagnostics).toEqual([{ kind: 'stream-ended', reason: 'overflow' }]) + expect(page.frames().filter((frame) => frame.type === 'cancel')).toEqual([]) + }) + + it('tells the listener nothing when the page itself let the stream go', () => { + const page = createPageClient() + page.start() + const onData = vi.fn() + const dispose = page.client.subscribe('terminal.stream', {}, onData) + dispose() + // The caller that disposed is the one that would hear it, and it has already moved on. + expect(onData).not.toHaveBeenCalled() + expect(page.frames().filter((frame) => frame.type === 'cancel')).toHaveLength(1) + }) +}) + +describe('bridge client binary frames', () => { + const image = Uint8Array.from([137, 80, 78, 71, 13, 10, 26, 10]) + const b64 = btoa(String.fromCharCode(...image)) + + function binaryFrame(id: string, b64Image: string): BridgeHostMessage { + return { + v: BRIDGE_PROTOCOL_VERSION, + type: 'event', + id, + seq: 1, + binary: { b64: b64Image, format: 'png', frameSeq: 41, metadata: { imageWidth: 8 } } + } + } + + it('asks for binary only when a listener is there to read it', () => { + const page = createPageClient() + page.start() + page.client.subscribe('browser.screencast', {}, vi.fn()) + page.client.subscribe('browser.screencast', {}, vi.fn(), { onBinaryFrame: vi.fn() }) + const opened = page.frames().filter((frame) => frame.type === 'subscribe') + expect(opened[0]).not.toHaveProperty('wantsBinary') + expect(opened[1]).toHaveProperty('wantsBinary', true) + }) + + it('decodes to the frame a native listener would have been handed', () => { + const page = createPageClient() + page.start() + const onBinaryFrame = vi.fn() + page.client.subscribe('browser.screencast', {}, vi.fn(), { onBinaryFrame }) + page.deliver(binaryFrame(idOf(page, 0), b64)) + expect(onBinaryFrame).toHaveBeenCalledWith({ + opcode: BrowserScreencastOpcode.Frame, + seq: 41, + format: 'png', + metadata: { imageWidth: 8 }, + image + }) + }) + + it('carries every metadata field the shell measured', () => { + const page = createPageClient() + page.start() + const onBinaryFrame = vi.fn() + page.client.subscribe('browser.screencast', {}, vi.fn(), { onBinaryFrame }) + const metadata = { + offsetTop: 1, + pageScaleFactor: 2, + deviceWidth: 3, + deviceHeight: 4, + imageWidth: 5, + imageHeight: 6, + scrollOffsetX: 7, + scrollOffsetY: 8, + timestamp: 9 + } + page.deliver({ + v: BRIDGE_PROTOCOL_VERSION, + type: 'event', + id: idOf(page, 0), + seq: 1, + binary: { b64, format: 'jpeg', frameSeq: 0, metadata } + }) + expect(onBinaryFrame).toHaveBeenCalledWith( + expect.objectContaining({ format: 'jpeg', seq: 0, metadata }) + ) + }) + + it('drops a frame with no listener and one it cannot decode', () => { + const page = createPageClient() + page.start() + page.client.subscribe('browser.screencast', {}, vi.fn()) + page.deliver(binaryFrame(idOf(page, 0), b64)) + const onBinaryFrame = vi.fn() + page.client.subscribe('browser.screencast', {}, vi.fn(), { onBinaryFrame }) + page.deliver(binaryFrame(idOf(page, 1), '!!not base64!!')) + expect(onBinaryFrame).not.toHaveBeenCalled() + expect(page.diagnostics).toEqual([ + { kind: 'binary-frame-dropped' }, + { kind: 'binary-frame-dropped' } + ]) + }) +}) diff --git a/mobile/src/mobile-web-shell/bridge/bridge-rpc-client.test.ts b/mobile/src/mobile-web-shell/bridge/bridge-rpc-client.test.ts new file mode 100644 index 00000000000..1d31a4c1582 --- /dev/null +++ b/mobile/src/mobile-web-shell/bridge/bridge-rpc-client.test.ts @@ -0,0 +1,264 @@ +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' +import { + isRpcDeliveryUnknown, + markRpcDeliveryUnknown +} from '../../transport/rpc-delivery-ambiguity' +import type { RpcResponse } from '../../transport/types' +import { createFakeRpcClient } from '../bridge-host-test-fakes' +import { BRIDGE_MAX_MESSAGE_BYTES, BRIDGE_MAX_SUBSCRIPTIONS } from './bridge-caps' +import { createBridgePortPair, type BridgePortPair } from './bridge-port-pair-test-harness' + +/** + * The page's client and the shell's host, over one FIFO per direction. + * + * That `createBridgeRpcClient` returns an `RpcClient` is the type system's job and it is already + * done; what a test has to prove is that each member still means the same thing after a round trip, + * because the screens above it cannot tell which client they are holding. + */ + +function success(id: string, result: unknown): RpcResponse { + return { id, ok: true, result, _meta: { runtimeId: 'runtime-a' } } +} + +/** Narrows what a rejection handed back, so a test reads an error rather than asserting one. */ +function readError(thrown: unknown): Error { + if (!(thrown instanceof Error)) { + throw new Error(`expected an Error, got ${typeof thrown}`) + } + return thrown +} + +async function ready(pair: BridgePortPair): Promise { + await pair.flush() + return pair +} + +beforeEach(() => { + vi.useFakeTimers() +}) + +afterEach(() => { + vi.useRealTimers() +}) + +describe('bridge round trip: requests', () => { + it('reaches the shell with the arity the page called with', async () => { + const pair = await ready(createBridgePortPair()) + void pair.client.sendRequest('worktree.ps') + void pair.client.sendRequest('worktree.ps', { host: 'a' }) + void pair.client.sendRequest('worktree.ps', { host: 'a' }, { timeoutMs: 50 }) + await pair.flush() + expect(pair.rpc.requests.map((request) => request.args)).toEqual([ + ['worktree.ps'], + ['worktree.ps', { host: 'a' }], + ['worktree.ps', { host: 'a' }, { timeoutMs: 50 }] + ]) + }) + + it('resolves the response the shell answered with, field for field', async () => { + const pair = await ready(createBridgePortPair()) + const answer = pair.client.sendRequest('worktree.ps') + await pair.flush() + const response: RpcResponse = { + id: 'shell-side-id', + ok: true, + result: { rows: [1, 2, 3] }, + streaming: true, + _meta: { runtimeId: 'runtime-a' } + } + pair.rpc.requests[0]?.resolve(response) + await pair.flush() + await expect(answer).resolves.toEqual(response) + }) + + it('resolves a host failure, because a failure is data and not a rejection', async () => { + const pair = await ready(createBridgePortPair()) + const answer = pair.client.sendRequest('worktree.ps') + await pair.flush() + const failure: RpcResponse = { + id: 'shell-side-id', + ok: false, + error: { code: 'not_found', message: 'no such worktree', data: { host: 'a' } }, + _meta: { runtimeId: 'runtime-a' } + } + pair.rpc.requests[0]?.resolve(failure) + await pair.flush() + await expect(answer).resolves.toEqual(failure) + }) + + it('rejects with the class, the code and the delivery mark the shell captured', async () => { + const pair = await ready(createBridgePortPair()) + const answer = pair.client.sendRequest('worktree.ps') + await pair.flush() + class RpcTimeoutError extends Error { + code = 'ETIMEDOUT' + } + const thrown = markRpcDeliveryUnknown(new RpcTimeoutError('timed out after 50ms')) + thrown.cause = new Error('socket closed') + pair.rpc.requests[0]?.reject(thrown) + await pair.flush() + const error = await answer.catch((caught: unknown) => caught) + expect(error).toBeInstanceOf(Error) + expect(readError(error).name).toBe('RpcTimeoutError') + expect(readError(error).message).toBe('timed out after 50ms') + expect(isRpcDeliveryUnknown(error)).toBe(true) + expect(readError(readError(error).cause).message).toBe('socket closed') + }) + + it('reassembles a reply too big for one frame', async () => { + const pair = await ready(createBridgePortPair()) + const answer = pair.client.sendRequest('source-control.diff') + await pair.flush() + const result = { diff: 'z'.repeat(BRIDGE_MAX_MESSAGE_BYTES + 60_000) } + pair.rpc.requests[0]?.resolve(success('shell-side-id', result)) + await pair.flush() + await expect(answer).resolves.toEqual(success('shell-side-id', result)) + expect(pair.toPage.length).toBeGreaterThan(2) + }) +}) + +describe('bridge round trip: subscriptions', () => { + it('streams what the shell emits and stops when the page disposes', async () => { + const pair = await ready(createBridgePortPair()) + const onData = vi.fn() + const dispose = pair.client.subscribe('terminal.stream', { terminal: 't' }, onData) + await pair.flush() + expect(pair.rpc.streams[0]?.method).toBe('terminal.stream') + expect(pair.rpc.streams[0]?.params).toEqual({ terminal: 't' }) + pair.rpc.streams[0]?.emit({ type: 'data', chunk: 'hello' }) + await pair.flush() + expect(onData).toHaveBeenCalledWith({ type: 'data', chunk: 'hello' }) + dispose() + await pair.flush() + expect(pair.rpc.streams[0]?.unsubscribes).toBe(1) + pair.rpc.streams[0]?.emit({ type: 'data', chunk: 'after' }) + await pair.flush() + expect(onData).toHaveBeenCalledTimes(1) + }) + + it('frees the page slot when the shell refuses the subscribe', async () => { + const pair = await ready(createBridgePortPair()) + const onData = vi.fn() + const refuse = vi.spyOn(pair.rpc, 'subscribe').mockImplementation(() => { + throw new Error('the terminal is gone') + }) + pair.client.subscribe('terminal.stream', { terminal: 't' }, onData) + await pair.flush() + refuse.mockRestore() + expect(pair.diagnostics).toEqual([ + { kind: 'stream-failed', error: expect.objectContaining({ message: 'the terminal is gone' }) } + ]) + // The shell's own message reaches the listener, the way the native client passes one through. + expect(onData.mock.calls).toEqual([ + [{ type: 'error', message: 'the terminal is gone', error: expect.any(Error) }] + ]) + // A leaked slot is invisible until the page reaches its own cap, so that is where it is read. + for (let index = 0; index < BRIDGE_MAX_SUBSCRIPTIONS; index += 1) { + pair.client.subscribe('terminal.stream', {}, vi.fn()) + } + await pair.flush() + expect(pair.rpc.streams).toHaveLength(BRIDGE_MAX_SUBSCRIPTIONS) + }) + + it('keeps a long stream alive, because the acks free the shell window', async () => { + const pair = await ready(createBridgePortPair()) + const onData = vi.fn() + pair.client.subscribe('terminal.stream', {}, onData) + await pair.flush() + for (let batch = 0; batch < 8; batch += 1) { + for (let frame = 0; frame < 50; frame += 1) { + pair.rpc.streams[0]?.emit(`frame-${batch}-${frame}`) + } + await pair.flush() + } + expect(onData).toHaveBeenCalledTimes(400) + expect(pair.diagnostics).toEqual([]) + }) + + it('ends the stream when the page never gets a chance to ack', async () => { + const pair = await ready(createBridgePortPair()) + const onData = vi.fn() + pair.client.subscribe('terminal.stream', {}, onData) + await pair.flush() + for (let frame = 0; frame < 400; frame += 1) { + pair.rpc.streams[0]?.emit(`frame-${frame}`) + } + await pair.flush() + expect(pair.diagnostics).toEqual([{ kind: 'stream-ended', reason: 'overflow' }]) + expect(onData.mock.calls.length).toBeLessThan(400) + }) +}) + +describe('bridge round trip: notifications and state', () => { + it('carries both notifies to the shell client, with the arity each was called with', async () => { + const pair = await ready(createBridgePortPair()) + pair.client.notifyForeground() + pair.client.notifyForeground('app-resume') + pair.client.updateTerminalSubscriptionViewport('terminal-a', { cols: 120, rows: 40 }) + await pair.flush() + expect(pair.rpc.foregroundCalls).toEqual([[], ['app-resume']]) + expect(pair.rpc.viewports).toEqual([{ terminal: 'terminal-a', cols: 120, rows: 40 }]) + }) + + it('reads the shell client through init and fans out every change after it', async () => { + const rpc = createFakeRpcClient({ + getState: () => 'reconnecting', + getReconnectAttempt: () => 3, + getLastConnectedAt: () => 1234, + getLastInboundAt: () => 5678, + getGeneration: () => 9 + }) + const pair = await ready(createBridgePortPair({ rpc })) + expect(pair.client.getState()).toBe('reconnecting') + expect(pair.client.getReconnectAttempt()).toBe(3) + expect(pair.client.getLastConnectedAt()).toBe(1234) + expect(pair.client.getLastInboundAt?.()).toBe(5678) + expect(pair.client.getGeneration?.()).toBe(9) + const listener = vi.fn() + const release = pair.client.onStateChange(listener) + rpc.pushState('connected') + await pair.flush() + expect(listener).toHaveBeenCalledWith('connected') + expect(pair.client.getState()).toBe('connected') + release() + rpc.pushState('disconnected') + await pair.flush() + expect(listener).toHaveBeenCalledTimes(1) + }) + + it('refuses a snapshot from a shell that was rebuilt, and asks for a fresh init', async () => { + let generation = 5 + const rpc = createFakeRpcClient({ getGeneration: () => generation }) + const pair = await ready(createBridgePortPair({ rpc })) + const listener = vi.fn() + pair.client.onStateChange(listener) + const asked = pair.readToShell().filter((frame) => frame.type === 'ready').length + generation = 2 + rpc.pushState('reconnecting') + await pair.flush() + expect(pair.diagnostics).toEqual([{ kind: 'state-out-of-order' }]) + expect(listener).not.toHaveBeenCalled() + expect(pair.readToShell().filter((frame) => frame.type === 'ready').length).toBe(asked + 1) + // The fresh init is what re-primes the cache; the refused frame never touched it. + expect(pair.client.getState()).toBe('connected') + expect(pair.client.getGeneration?.()).toBe(2) + }) +}) + +describe('bridge round trip: close', () => { + it('settles pendings delivery-unknown, retires the streams, and leaves the shell client open', async () => { + const pair = await ready(createBridgePortPair()) + const closeShellClient = vi.spyOn(pair.rpc, 'close') + const answer = pair.client.sendRequest('worktree.ps') + pair.client.subscribe('terminal.stream', {}, vi.fn()) + await pair.flush() + pair.client.close() + const error = await answer.catch((caught: unknown) => caught) + expect(readError(error).name).toBe('BridgeClientClosedError') + expect(isRpcDeliveryUnknown(error)).toBe(true) + await pair.flush() + expect(pair.rpc.streams[0]?.unsubscribes).toBe(1) + expect(closeShellClient).not.toHaveBeenCalled() + expect(pair.readToShell().at(-1)).toEqual({ v: 1, type: 'close' }) + }) +}) diff --git a/mobile/src/mobile-web-shell/bridge/bridge-rpc-client.ts b/mobile/src/mobile-web-shell/bridge/bridge-rpc-client.ts new file mode 100644 index 00000000000..5329dc8175d --- /dev/null +++ b/mobile/src/mobile-web-shell/bridge/bridge-rpc-client.ts @@ -0,0 +1,382 @@ +import type { BrowserScreencastFrame } from '../../transport/browser-screencast-protocol' +import type { RpcClient, SendRequestOptions } from '../../transport/rpc-client' +import type { ConnectionState, ForegroundNudgeReason, RpcResponse } from '../../transport/types' +import { + BRIDGE_MAX_PENDING_REQUESTS, + BRIDGE_MAX_SUBSCRIPTIONS, + utf8ByteLength, + type BridgeRefusal +} from './bridge-caps' +import { BridgeConnectionCache } from './bridge-client-connection-cache' +import { createBridgeInitHandshake } from './bridge-client-init-handshake' +import { + BridgeClientCapExceededError, + BridgeClientClosedError, + BridgeClientNotReadyError, + BridgeSendFailedError, + BridgeShellReplacedError +} from './bridge-client-errors' +import { BridgeClientRequests } from './bridge-client-requests' +import { + BridgeClientSubscriptions, + type BridgeStreamEndReason +} from './bridge-client-subscriptions' +import { + BRIDGE_PROTOCOL_VERSION, + readBridgeHostMessage, + type BridgeClientMessage, + type BridgeConnectionSnapshot, + type BridgeGrants, + type BridgeHostMessage +} from './bridge-envelope' +import { reconstructBridgeError } from './bridge-error-capture' + +export { + BridgeClientCapExceededError, + BridgeClientClosedError, + BridgeClientNotReadyError, + BridgeReplyRefusedError, + BridgeSendFailedError, + BridgeShellReplacedError +} from './bridge-client-errors' + +/** Base64url, and the length the envelope's id pattern requires. Base36 digits are a subset of it. */ +const BRIDGE_ID_CHARS = 22 + +/** Nothing here is recoverable in place; each is worth a line in a log and none is retried. */ +export type BridgeRpcClientDiagnostic = + | { kind: 'refused'; refusal: BridgeRefusal } + | { kind: 'send-failed'; error: unknown } + | { kind: 'stream-ended'; reason: BridgeStreamEndReason } + | { kind: 'stream-failed'; error: unknown } + | { kind: 'state-out-of-order' } + | { kind: 'binary-frame-dropped' } + | { kind: 'unknown-id' } + +/** What `init` said this page is attached to. `grants` is what a call site checks before it posts. */ +export type BridgeShellSession = { + sessionId: string + buildId: string + grants: BridgeGrants +} + +export type BridgeRpcClientOptions = { + /** Posts one frame to the shell. May throw; nothing about returning proves delivery. */ + send: (json: string) => void + onMessage: (handler: (json: string) => void) => () => void + onDiagnostic?: (diagnostic: BridgeRpcClientDiagnostic) => void +} + +export type BridgeRpcClient = RpcClient & { + /** Fires once `init` has landed, immediately if it already has. Mount no screen before it. */ + onReady: (listener: () => void) => () => void + getShellSession: () => BridgeShellSession | null +} + +/** + * The page's `RpcClient`, which is a bridge and not a socket. + * + * Every member of the native contract is here, so `runRpcOperation` and the screens above it never + * learn which one they hold. Two properties make that honest. The getters are synchronous reads of a + * cache primed by `init`, because screens read them during render and an async read changes what the + * first render sees. And `close` never closes the shell's client: that one is shared with the native + * screens and the host catalog, so the page settles what it owns and says goodbye. + * + * Nothing may be called before `init`. The alternative is a stub answering `connecting` to a screen + * that then records the wrong first render, so a call arriving early throws instead. After `close` + * the opposite rule holds: every member goes inert and the getters keep answering the snapshot the + * page last held, marked `disconnected`, because an unmounting screen calls into a path with no + * catch on it. A stream the shell refuses or ends is not thrown anywhere either; it arrives as a + * diagnostic, which is the only channel `subscribe` leaves open once it has handed back a dispose. + */ +export function createBridgeRpcClient(options: BridgeRpcClientOptions): BridgeRpcClient { + const requests = new BridgeClientRequests() + const cache = new BridgeConnectionCache() + const readyListeners = new Set<() => void>() + let session: BridgeShellSession | null = null + let closed = false + let idCounter = 0 + + function report(diagnostic: BridgeRpcClientDiagnostic): void { + options.onDiagnostic?.(diagnostic) + } + + /** False when the frame never left. Every value in a page frame is one the caller handed in, so + * the throw this catches is the port's, never `JSON.stringify`'s. */ + function sendFrame(frame: BridgeClientMessage): boolean { + try { + options.send(JSON.stringify(frame)) + return true + } catch (error) { + report({ kind: 'send-failed', error }) + return false + } + } + + // Counted rather than random: a recorded run replays the same ids, and one page holds one client, + // so a counter is already unique across everything the shell is asked to keep in flight. + function nextId(): string { + idCounter += 1 + return idCounter.toString(36).padStart(BRIDGE_ID_CHARS, '0') + } + + const subscriptions = new BridgeClientSubscriptions({ + send: (frame) => sendFrame(frame), + onDroppedBinaryFrame: () => { + report({ kind: 'binary-frame-dropped' }) + } + }) + + const handshake = createBridgeInitHandshake(() => { + sendFrame({ v: BRIDGE_PROTOCOL_VERSION, type: 'ready' }) + }) + + /** + * A call before `init` is a mount-order bug and throws. A call after `close` is not: an unmounting + * screen posts one more nudge on its way out, and the native clients answer those inertly rather + * than throwing into a teardown path nobody wrote a catch for. Each member below says what inert + * means for its own return type. + */ + function requireSession(): void { + if (session === null && !closed) { + throw new BridgeClientNotReadyError() + } + } + + // Answers after `close` as well: what it holds is then the last snapshot, marked `disconnected`. + function snapshot(): BridgeConnectionSnapshot { + const held = cache.read() + if (held === null) { + throw new BridgeClientNotReadyError() + } + return held + } + + /** A second `init` is ordinary: the shell answers every `ready`, and a page that re-asked hears + * its own session again. A different id is not, and nothing the page held survives it. */ + function acceptInit(message: Extract): void { + handshake.stop() + if (session !== null && session.sessionId !== message.sessionId) { + const replaced = new BridgeShellReplacedError() + requests.closeAll(replaced) + subscriptions.failAll(replaced.message) + } + session = { sessionId: message.sessionId, buildId: message.buildId, grants: message.grants } + cache.prime(message.connection) + for (const listener of readyListeners) { + listener() + } + readyListeners.clear() + } + + /** A shell rebuilt under the page: what the cache holds is for a client that is already gone. */ + function acceptState(snapshotFromShell: BridgeConnectionSnapshot): void { + if (cache.apply(snapshotFromShell) !== 'stale') { + return + } + report({ kind: 'state-out-of-order' }) + handshake.restart() + } + + /** The shell's own words where it had any, the way the native client passes an RPC error message + * through to the listener it ends. */ + function describeStreamFailure(error: unknown): string { + return error instanceof Error ? error.message : 'the shell could not keep this stream open' + } + + /** The shell answers a refused `subscribe` with `error` on the stream's id. Nothing is pending to + * reject there, so routing it to the requests would drop it and hold the page's slot forever. */ + function failExchange(id: string, error: unknown): void { + if (subscriptions.has(id)) { + // Reported before the listener runs, so a listener that throws cannot swallow the diagnostic. + report({ kind: 'stream-failed', error }) + subscriptions.end(id, describeStreamFailure(error), error) + return + } + if (!requests.has(id)) { + report({ kind: 'unknown-id' }) + } + // Still routed: an id with a half-assembled reply behind it holds a slot until it is discarded. + requests.fail(id, error) + } + + function dispatch(message: BridgeHostMessage, json: string): void { + switch (message.type) { + case 'init': + acceptInit(message) + return + case 'state': + acceptState(message.connection) + return + case 'reply': + if (!requests.has(message.id)) { + report({ kind: 'unknown-id' }) + } + requests.acceptReply(message) + return + case 'error': + failExchange(message.id, reconstructBridgeError(message.error)) + return + case 'event': + subscriptions.deliver(message, utf8ByteLength(json)) + return + case 'end': + report({ kind: 'stream-ended', reason: message.reason }) + subscriptions.end(message.id, `the shell ended this stream (${message.reason})`) + return + } + } + + function receive(json: string): void { + if (closed) { + return + } + const read = readBridgeHostMessage(json) + if (!read.ok) { + report({ kind: 'refused', refusal: read.refusal }) + return + } + dispatch(read.message, json) + } + + function sendRequest(...args: [string, unknown?, SendRequestOptions?]): Promise { + // A call with no session is a page bug and throws; a call over the in-flight cap is the answer + // the shell would have posted back, so it arrives the way the shell's does, as a rejection. + requireSession() + if (closed) { + // Rejected, not thrown: `bindDeferredRpcOperation` hands this promise straight back, so a + // synchronous throw would escape past the caller's `catch` on the promise. + return Promise.reject(new BridgeClientClosedError()) + } + if (requests.size >= BRIDGE_MAX_PENDING_REQUESTS) { + return Promise.reject( + new BridgeClientCapExceededError(`over ${BRIDGE_MAX_PENDING_REQUESTS} requests in flight`) + ) + } + const [method, params, requestOptions] = args + const id = nextId() + return new Promise((resolve, reject) => { + requests.open(id, { resolve, reject }) + const sent = sendFrame({ + v: BRIDGE_PROTOCOL_VERSION, + type: 'request', + id, + method, + // Absent stays absent, because the shell replays whichever arity crossed. JSON drops an + // `undefined` value on its own, so an explicit `sendRequest(m, undefined)` reaches the shell + // as `sendRequest(m)`; no call site passes one, and no wire that carries `undefined` exists + // to carry it. The spread is what states the intent for a carrier that would. + ...(args.length > 1 ? { params } : {}), + ...(requestOptions === undefined ? {} : { options: requestOptions }) + }) + if (!sent) { + requests.abandon(id) + reject(new BridgeSendFailedError()) + } + }) + } + + function subscribe( + method: string, + params: unknown, + onData: (result: unknown) => void, + subscribeOptions?: { onBinaryFrame?: (frame: BrowserScreencastFrame) => void } + ): () => void { + requireSession() + if (closed) { + return () => undefined + } + // Thrown rather than reported: `subscribe` hands back an unsubscribe and nothing else, so a + // refusal the caller could read does not exist on this member. A refusal the shell posts back + // arrives too late to throw at all, and reaches the page as a `stream-failed` diagnostic. + if (subscriptions.size >= BRIDGE_MAX_SUBSCRIPTIONS) { + throw new BridgeClientCapExceededError(`over ${BRIDGE_MAX_SUBSCRIPTIONS} subscriptions`) + } + const id = nextId() + // A frame that never left already told the listener and gave the slot back; the caller still + // gets a dispose, because it has no way to know which of the two it is holding. + if (!subscriptions.open(id, method, params, onData, subscribeOptions?.onBinaryFrame)) { + return () => undefined + } + let disposed = false + return () => { + if (disposed) { + return + } + disposed = true + subscriptions.cancel(id) + } + } + + function close(): void { + if (closed) { + return + } + closed = true + handshake.stop() + subscriptions.closeAll() + sendFrame({ v: BRIDGE_PROTOCOL_VERSION, type: 'close' }) + requests.closeAll() + cache.close() + session = null + readyListeners.clear() + unsubscribeFromMessages() + } + + const unsubscribeFromMessages = options.onMessage(receive) + handshake.start() + + return { + sendRequest, + subscribe, + updateTerminalSubscriptionViewport: (terminal, viewport) => { + requireSession() + if (closed) { + return + } + sendFrame({ + v: BRIDGE_PROTOCOL_VERSION, + type: 'notify', + name: 'terminalViewport', + terminal, + cols: viewport.cols, + rows: viewport.rows + }) + }, + getState: (): ConnectionState => snapshot().state, + getReconnectAttempt: () => snapshot().reconnectAttempt, + getLastConnectedAt: () => snapshot().lastConnectedAt, + getLastInboundAt: () => snapshot().lastInboundAt, + // A shell client with no generation of its own never migrates, so its epoch is a constant and + // zero is as true as any other. The page still answers a number, because the member it stands in + // for is one the native screens read without asking whether it exists. + getGeneration: () => snapshot().generation ?? 0, + // Not gated on the session: it registers a listener and reads nothing, so it cannot answer + // wrongly, and a provider that subscribes before `init` is how a screen hears the first change. + onStateChange: (listener) => cache.onStateChange(listener), + notifyForeground: (reason?: ForegroundNudgeReason) => { + requireSession() + if (closed) { + return + } + sendFrame({ + v: BRIDGE_PROTOCOL_VERSION, + type: 'notify', + name: 'foreground', + ...(reason === undefined ? {} : { reason }) + }) + }, + close, + onReady: (listener) => { + if (session !== null) { + listener() + return () => undefined + } + readyListeners.add(listener) + return () => { + readyListeners.delete(listener) + } + }, + getShellSession: () => session + } +} diff --git a/mobile/src/mobile-web-shell/bridge/bridge-screencast-binary.ts b/mobile/src/mobile-web-shell/bridge/bridge-screencast-binary.ts new file mode 100644 index 00000000000..891c0e01238 --- /dev/null +++ b/mobile/src/mobile-web-shell/bridge/bridge-screencast-binary.ts @@ -0,0 +1,53 @@ +import { + BrowserScreencastOpcode, + type BrowserScreencastFrame +} from '../../transport/browser-screencast-protocol' +import type { BridgeHostMessage } from './bridge-envelope' + +/** + * The binary lane's page-side half: base64 in, the same `BrowserScreencastFrame` a native listener + * is handed out. + * + * There is no wire header to parse here. `decodeBrowserScreencastFrame` reads one because the + * socket carries a frame as a single buffer; the envelope already carries `format`, `frameSeq` and + * the metadata as JSON beside the image, so only the image is base64. C6 owns the encoder that + * produces this shape, and this is the inverse it has to satisfy. + */ +export type BridgeBinaryEvent = Extract< + Extract, + { binary: unknown } +>['binary'] + +/** `null` when the image is not base64: an undecodable frame is dropped, never guessed at. */ +export function decodeBridgeScreencastFrame( + event: BridgeBinaryEvent +): BrowserScreencastFrame | null { + const image = decodeBase64(event.b64) + if (image === null) { + return null + } + return { + opcode: BrowserScreencastOpcode.Frame, + // The screencast's own counter. The event frame's `seq` is the bridge's backpressure ordinal, + // and handing that one over would renumber every frame the page reports. + seq: event.frameSeq, + format: event.format, + metadata: event.metadata, + image + } +} + +/** Metro ships no `Buffer`; `atob` is what the pairing and E2EE paths already decode with. */ +function decodeBase64(value: string): Uint8Array | null { + let binary: string + try { + binary = atob(value) + } catch { + return null + } + const bytes = new Uint8Array(binary.length) + for (let index = 0; index < binary.length; index += 1) { + bytes[index] = binary.charCodeAt(index) + } + return bytes +} diff --git a/mobile/src/mobile-web-shell/bridge/orca-bridge-page-channel.test.ts b/mobile/src/mobile-web-shell/bridge/orca-bridge-page-channel.test.ts new file mode 100644 index 00000000000..518142a7796 --- /dev/null +++ b/mobile/src/mobile-web-shell/bridge/orca-bridge-page-channel.test.ts @@ -0,0 +1,53 @@ +import { afterEach, describe, expect, it, vi } from 'vitest' +import { + createOrcaBridgePageTransport, + readOrcaBridgePageChannel, + type OrcaBridgePageChannel +} from './orca-bridge-page-channel' + +function installChannel(channel: unknown): void { + Object.defineProperty(globalThis, 'orcaBridge', { value: channel, configurable: true }) +} + +function createChannel(): OrcaBridgePageChannel { + return { postMessage: vi.fn(), onmessage: null } +} + +afterEach(() => { + Reflect.deleteProperty(globalThis, 'orcaBridge') +}) + +describe('the page channel the shell installs', () => { + it('is absent in a browser, which is a page the bundle still has to open', () => { + expect(readOrcaBridgePageChannel()).toBeNull() + }) + + it('refuses a global of another shape rather than posting into it', () => { + installChannel({ postMessage: 'not a function', onmessage: null }) + expect(readOrcaBridgePageChannel()).toBeNull() + }) + + it('reads the installed object itself, so the page posts through the real sink', () => { + const channel = createChannel() + installChannel(channel) + expect(readOrcaBridgePageChannel()).toBe(channel) + }) +}) + +describe('the page channel as a client transport', () => { + it('posts what the client sends', () => { + const channel = createChannel() + createOrcaBridgePageTransport(channel).send('{"v":1}') + expect(channel.postMessage).toHaveBeenCalledWith('{"v":1}') + }) + + it('hands the client the frame off the event, and gives the slot back', () => { + const channel = createChannel() + const handler = vi.fn() + const release = createOrcaBridgePageTransport(channel).onMessage(handler) + channel.onmessage?.({ data: '{"v":1,"type":"init"}' }) + expect(handler).toHaveBeenCalledWith('{"v":1,"type":"init"}') + release() + expect(channel.onmessage).toBeNull() + }) +}) diff --git a/mobile/src/mobile-web-shell/bridge/orca-bridge-page-channel.ts b/mobile/src/mobile-web-shell/bridge/orca-bridge-page-channel.ts new file mode 100644 index 00000000000..57aa7f79fea --- /dev/null +++ b/mobile/src/mobile-web-shell/bridge/orca-bridge-page-channel.ts @@ -0,0 +1,51 @@ +import type { BridgeRpcClientOptions } from './bridge-rpc-client' + +/** + * The page's half of the native channel, as the document-start installer leaves it. + * + * `postMessage` and an `onmessage` assignment are the whole surface, and it is deliberately the + * intersection of the two platforms: Android's `addWebMessageListener` injects an object of this + * shape, and `MobileWebShellView.swift` installs one to match. Nothing else about the WebView is + * addressable from the page. + */ +export type OrcaBridgePageChannel = { + postMessage: (json: string) => void + onmessage: ((event: { data: string }) => void) | null +} + +/** + * `null` for a page that is not inside the shell — a browser, or a WebView mounted with the bridge + * off. That is a supported way to open the bundle, so the caller substitutes rather than throws. + */ +export function readOrcaBridgePageChannel(): OrcaBridgePageChannel | null { + const scope: typeof globalThis & { orcaBridge?: OrcaBridgePageChannel } = globalThis + const channel = scope.orcaBridge + if (channel === undefined || typeof channel.postMessage !== 'function') { + return null + } + return channel +} + +/** + * The channel as the page client's transport. + * + * One `onmessage` slot exists, so one client reads the channel; a second would silently take the + * first one's frames. The page holds exactly one client, which is what makes that safe. + */ +export function createOrcaBridgePageTransport( + channel: OrcaBridgePageChannel +): Pick { + return { + send: (json) => { + channel.postMessage(json) + }, + onMessage: (handler) => { + channel.onmessage = (event) => { + handler(event.data) + } + return () => { + channel.onmessage = null + } + } + } +} diff --git a/mobile/src/transport/client-context.web.test.tsx b/mobile/src/transport/client-context.web.test.tsx new file mode 100644 index 00000000000..b1164968e6a --- /dev/null +++ b/mobile/src/transport/client-context.web.test.tsx @@ -0,0 +1,161 @@ +import { createElement, type ReactElement } from 'react' +import { act, create } from 'react-test-renderer' +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' +import { BRIDGE_PROTOCOL_VERSION } from '../mobile-web-shell/bridge/bridge-envelope' +import type { RpcClientContextValue } from './rpc-client-context-contract' + +// The web file re-exports the screen hooks, and reaching the real ones imports the Expo runtime +// this test does not have. Nothing below calls one. +vi.mock('./host-client-hooks', () => ({ + useDisconnectHostClient: () => () => {}, + useForceReconnect: () => () => Promise.resolve(), + useForgetHostClient: () => () => {}, + useHostClient: () => ({ client: null, clientId: null, state: 'disconnected' }), + usePrimeHosts: () => () => {}, + useRefreshHostClient: () => () => {} +})) + +import { RpcClientProvider, useRpcClientContext } from './client-context.web' + +const INIT = { + v: BRIDGE_PROTOCOL_VERSION, + type: 'init', + sessionId: 'session-a', + buildId: 'build-a', + connection: { + state: 'connected', + reconnectAttempt: 2, + lastConnectedAt: 1700, + lastInboundAt: 1800, + generation: 5 + }, + grants: { rpc: { maxPendingRequests: 64, maxSubscriptions: 32 }, native: [] } +} + +/** What the page mounted, and what it holds — the two things the provider decides. */ +const screen: { mounts: number; context: RpcClientContextValue | null } = { + mounts: 0, + context: null +} + +function Screen(): null { + screen.context = useRpcClientContext() + screen.mounts += 1 + return null +} + +function render(): ReactElement { + return createElement(RpcClientProvider, null, createElement(Screen)) +} + +/** The channel the shell's document-start script installs, as a double. */ +function installChannel(): { posted: string[]; deliver: (frame: unknown) => void } { + const posted: string[] = [] + const channel: { + postMessage: (json: string) => void + onmessage: ((e: { data: string }) => void) | null + } = { + postMessage: (json) => { + posted.push(json) + }, + onmessage: null + } + Object.defineProperty(globalThis, 'orcaBridge', { value: channel, configurable: true }) + return { + posted, + deliver: (frame) => { + channel.onmessage?.({ data: JSON.stringify(frame) }) + } + } +} + +function readContext(): RpcClientContextValue { + const context = screen.context + if (context === null) { + throw new Error('no screen mounted') + } + return context +} + +beforeEach(() => { + vi.useFakeTimers() + screen.mounts = 0 + screen.context = null +}) + +afterEach(() => { + vi.useRealTimers() + Reflect.deleteProperty(globalThis, 'orcaBridge') +}) + +describe('the page provider inside the shell', () => { + it('mounts nothing until the shell answers with a session', () => { + const channel = installChannel() + act(() => { + create(render()) + }) + expect(screen.mounts).toBe(0) + expect(channel.posted.map((json: string) => JSON.parse(json).type)).toEqual(['ready']) + act(() => { + channel.deliver(INIT) + }) + expect(screen.mounts).toBe(1) + }) + + it('answers every screen with the one client the page has', () => { + const channel = installChannel() + act(() => { + create(render()) + }) + act(() => { + channel.deliver(INIT) + }) + const context = readContext() + const client = context.acquire('host-a', {}) + expect(client).not.toBeNull() + expect(context.getState('host-a')).toBe('connected') + expect(context.getReconnectAttempt('host-a')).toBe(2) + expect(context.getLastConnectedAt('host-a')).toBe(1700) + expect(context.getAllClients()).toEqual([{ hostId: 'host-a', client }]) + }) + + it('carries a state change from the shell to the screens watching it', () => { + const channel = installChannel() + act(() => { + create(render()) + }) + act(() => { + channel.deliver(INIT) + }) + const listener = vi.fn() + readContext().subscribeHostState('host-a', listener) + act(() => { + channel.deliver({ + v: BRIDGE_PROTOCOL_VERSION, + type: 'state', + connection: { ...INIT.connection, state: 'reconnecting' } + }) + }) + expect(listener).toHaveBeenCalledWith('reconnecting') + expect(readContext().getState('host-a')).toBe('reconnecting') + }) +}) + +describe('the page provider outside the shell', () => { + it('mounts the route tree at once, because no session is ever coming', () => { + act(() => { + create(render()) + }) + expect(screen.mounts).toBe(1) + expect(readContext().getState('host-a')).toBe('disconnected') + }) + + it('hands out a client that reaches nothing rather than none at all', async () => { + act(() => { + create(render()) + }) + const client = readContext().acquire('host-a', {}) + expect(client).not.toBeNull() + await expect(client?.sendRequest('worktree.ps')).rejects.toThrow('bridge transport unavailable') + }) +}) diff --git a/mobile/src/transport/client-context.web.tsx b/mobile/src/transport/client-context.web.tsx index 1776cbb859b..b070a7e0edd 100644 --- a/mobile/src/transport/client-context.web.tsx +++ b/mobile/src/transport/client-context.web.tsx @@ -1,6 +1,24 @@ -// Web sibling: RN Web has no pairing keychain and no websocket transport of its own, so the page -// gets a placeholder client until C0.4 lands BridgeRpcClient over the shell bridge. -import { createContext, useContext, useMemo, type ReactNode } from 'react' +// Web sibling: RN Web has no pairing keychain and no websocket transport of its own, so the page's +// client is the shell bridge. Nothing here dials, retries or pairs — the native client on the other +// side of the bridge already did, and this provider only carries what it holds across the boundary. +import { + createContext, + useContext, + useEffect, + useMemo, + useRef, + useState, + type ReactNode +} from 'react' +import { + createBridgeRpcClient, + type BridgeRpcClient, + type BridgeRpcClientDiagnostic +} from '../mobile-web-shell/bridge/bridge-rpc-client' +import { + createOrcaBridgePageTransport, + readOrcaBridgePageChannel +} from '../mobile-web-shell/bridge/orca-bridge-page-channel' import type { RpcClient } from './rpc-client' import type { ConnectionState, HostProfile } from './types' import type { RpcClientContextValue } from './rpc-client-context-contract' @@ -22,6 +40,12 @@ export class BridgeTransportUnavailableError extends Error { } } +/** + * For a page opened outside the shell: a browser, or a WebView mounted with the bridge off. + * + * It answers every member and reaches nothing, which is what lets the route tree mount and paint + * its empty states instead of crashing on a client that is not there. + */ function createPlaceholderClient(): RpcClient { return { sendRequest: (method) => Promise.reject(new BridgeTransportUnavailableError(method)), @@ -40,14 +64,64 @@ function createPlaceholderClient(): RpcClient { } } +/** One line per kind for the life of one page: a page that is failing frames fails all of them. */ +function createPageDiagnosticReporter(): (diagnostic: BridgeRpcClientDiagnostic) => void { + const reported = new Set() + return (diagnostic) => { + if (reported.has(diagnostic.kind)) { + return + } + reported.add(diagnostic.kind) + console.warn('[page-bridge]', diagnostic.kind, diagnostic) + } +} + const Ctx = createContext(null) export function RpcClientProvider({ children }: { children: ReactNode }) { + // Held in a ref as well as in state: the context value is built once, because `useHostClient` + // re-acquires whenever the value's identity changes. + const clientRef = useRef(null) + const acquiredRef = useRef>(new Set()) + const [ready, setReady] = useState(false) + + useEffect(() => { + const channel = readOrcaBridgePageChannel() + if (channel === null) { + // Nothing to wait for, so the tree mounts against the placeholder rather than never. + clientRef.current = createPlaceholderClient() + setReady(true) + return + } + const client: BridgeRpcClient = createBridgeRpcClient({ + ...createOrcaBridgePageTransport(channel), + onDiagnostic: createPageDiagnosticReporter() + }) + // Nothing mounts before `init`: every member of this client throws until the shell answers, + // and a screen that rendered first would record its first frame against a session-less client. + const release = client.onReady(() => { + clientRef.current = client + setReady(true) + }) + return () => { + release() + clientRef.current = null + setReady(false) + client.close() + } + }, []) + const value = useMemo(() => { - const client = createPlaceholderClient() - const disconnected: ConnectionState = 'disconnected' + const state = (): ConnectionState => clientRef.current?.getState() ?? 'connecting' return { - acquire: () => client, + // One client for one page: the shell opened this document for one host, so whichever host + // the route names is the host on the other side of the bridge. + acquire: (hostId: string) => { + acquiredRef.current.add(hostId) + return clientRef.current + }, + // The shell owns the connection, and a page client cannot be reopened once it says goodbye. + // Every member that would close, drop or re-dial one is inert here for that reason. release: () => {}, releaseAndCloseIfUnused: () => {}, closeIfUnused: () => {}, @@ -55,24 +129,33 @@ export function RpcClientProvider({ children }: { children: ReactNode }) { refreshHostClient: () => {}, forgetHostClient: () => {}, disconnectHostClient: () => {}, - getState: () => disconnected, - getKnownState: () => disconnected, + getState: state, + getKnownState: () => (clientRef.current === null ? null : state()), getClientId: () => null, - getReconnectAttempt: () => 0, - getLastConnectedAt: () => null, + getReconnectAttempt: () => clientRef.current?.getReconnectAttempt() ?? 0, + getLastConnectedAt: () => clientRef.current?.getLastConnectedAt() ?? null, // The page reaches its host through the shell bridge, which rides whatever path the RN // client already negotiated. 'relay' is the honest default until init carries the real one. getActivePath: () => 'relay', getPendingPath: () => null, + // Both are pairing verdicts, and pairing happened natively before this document existed. isPairingRejected: () => false, isHostSignedOut: () => false, - subscribeHostState: () => () => {}, - getAllClients: () => [], - subscribeAllHosts: () => () => {}, + subscribeHostState: (_hostId: string, listener: (next: ConnectionState) => void) => + clientRef.current?.onStateChange(listener) ?? (() => {}), + getAllClients: () => { + const client = clientRef.current + return client === null ? [] : [...acquiredRef.current].map((hostId) => ({ hostId, client })) + }, + subscribeAllHosts: (listener: () => void) => + clientRef.current?.onStateChange(() => { + listener() + }) ?? (() => {}), primeHosts: (_hosts: HostProfile[]) => {} } }, []) - return {children} + + return {ready ? children : null} } export function useRpcClientContext(): RpcClientContextValue { diff --git a/mobile/src/transport/unvalidated-rpc-request-port-inventory.ts b/mobile/src/transport/unvalidated-rpc-request-port-inventory.ts index d38baf9a76a..54d1f6a500d 100644 --- a/mobile/src/transport/unvalidated-rpc-request-port-inventory.ts +++ b/mobile/src/transport/unvalidated-rpc-request-port-inventory.ts @@ -25,13 +25,17 @@ export type UnvalidatedRpcRequestPortEntry = { /** Modules whose job is the port. These do not shrink to zero. */ export const UNVALIDATED_RPC_REQUEST_PORT_OWNERS: readonly UnvalidatedRpcRequestPortEntry[] = [ - // Carries the port across the page boundary for the hybrid shell. Not a call site: it picks no - // method, reads no reply and decides no acceptance — the page names the method and runs the - // typed operation over it, exactly as a native screen does over a socket client. + // Forwards raw requests as a transport, reads no reply. Not a call site: it picks no method and + // decides no acceptance — the page names the method and runs the typed operation over it, exactly + // as a native screen does over a socket client. { file: 'src/mobile-web-shell/bridge-host.ts', references: 3 }, + // The far end of that transport: it offers the port to the page and posts what it is handed, + // reading neither the method nor the reply. + { file: 'src/mobile-web-shell/bridge/bridge-rpc-client.ts', references: 1 }, // Fakes the port for the bridge host suites; a non-test file only because tsconfig excludes tests. { file: 'src/mobile-web-shell/bridge-host-test-fakes.ts', references: 1 }, - // Placeholder page transport until C0.4's BridgeRpcClient replaces it; rejects every call, reads no reply. + // The page's client is BridgeRpcClient over the shell bridge; this one reference is the + // placeholder it falls back to outside the shell, which rejects every call and reads no reply. { file: 'src/transport/client-context.web.tsx', references: 1 }, // Implements the port over the device-to-host websocket. { file: 'src/transport/direct-rpc-client.ts', references: 3 }, diff --git a/mobile/web-entry/web-overrides.json b/mobile/web-entry/web-overrides.json index d1a67d58ecb..0484ee95c52 100644 --- a/mobile/web-entry/web-overrides.json +++ b/mobile/web-entry/web-overrides.json @@ -3,7 +3,7 @@ "overrides": [ { "file": "src/transport/client-context.web.tsx", - "reason": "The page has no websocket transport and no pairing keychain. This is the single transport substitution point: a placeholder RpcClient until C0.4 lands BridgeRpcClient over the shell bridge." + "reason": "The page has no websocket transport and no pairing keychain. This is the single transport substitution point: BridgeRpcClient over the shell bridge, and a placeholder RpcClient for a page opened outside it, which is what lets the route tree mount in a plain browser." }, { "file": "packages/expo-two-way-audio/src/ExpoTwoWayAudioModule.web.ts",