From 08c8d30f105f96093ca547d36764e4aa266b92cf Mon Sep 17 00:00:00 2001 From: Jinwoo Hong <73622457+Jinwoo-H@users.noreply.github.com> Date: Sun, 11 Oct 2026 02:54:27 -0400 Subject: [PATCH] fix(e2ee): stop dropping terminal output on a backed-up legacy connection (#26528) * fix(browser): retry a screencast frame a viewer's socket refused An idle page sends no next frame, so a refused last frame left the viewer on a stale picture. The desktop's relay leg to a server uses legacy E2EE, which drops on a full socket where a direct v2 phone queues; retrying the newest pending frame restores parity. Retries never count toward ghost eviction. Claude-Session: https://claude.ai/code/session_01JNnE9qzUZMMnqpZWCqM3nb * fix(e2ee): queue legacy binary replies with text instead of dropping them Legacy E2EE (every direct phone, the desktop's server link) sent binary straight to the socket and silently dropped it once the outbound budget was full, so terminals lost output and binary could overtake queued text. Binary now shares the ordered text queue and closes on overflow like v2; the lossy screencast opts into dropWhenBacklogged and its retry delivers the newest frame after a drain. Claude-Session: https://claude.ai/code/session_01JNnE9qzUZMMnqpZWCqM3nb * fix(e2ee): keep a backlogged screencast viewer instead of evicting it as a ghost A lossy frame skipped on a busy socket returned false like a dead socket, so a long backlog evicted a live viewer. Skips now return 'backlogged', which retries without counting toward eviction; legacy judges idleness on the sealed size before encrypting, so skipped frames cost no encryption. Claude-Session: https://claude.ai/code/session_01JNnE9qzUZMMnqpZWCqM3nb * test(browser): expect the ready gate to forward screencast send options The pre-ready gate now passes the per-frame send options through to the socket sender, so the replayed startup frame arrives with an options slot. Claude-Session: https://claude.ai/code/session_01JNnE9qzUZMMnqpZWCqM3nb * test(browser): share the streamed-guest double across screencast tests Main's renderer throttle lease now calls setBackgroundThrottling on the guest, so the bare { isDestroyed } guest in two branch tests threw. One shared double keeps all three screencast suites on the shape the lease needs. Claude-Session: https://claude.ai/code/session_01JNnE9qzUZMMnqpZWCqM3nb --- ...er-screencast-subscriber-frame-delivery.ts | 57 ++++++++ ...me-browser-screencast-e2ee-backlog.test.ts | 126 ++++++++++++++++ ...-runtime-browser-screencast-fanout.test.ts | 24 ++-- ...ser-screencast-pending-frame-retry.test.ts | 136 ++++++++++++++++++ src/main/runtime/orca-runtime-browser.test.ts | 4 +- .../mobile-creation-and-orchestration.spec.ts | 2 +- ...browser-screencast-frame-admission.test.ts | 20 ++- ...mote-browser-screencast-frame-admission.ts | 15 +- src/main/runtime/rpc/core.ts | 3 +- .../e2ee-channel-text-backpressure.test.ts | 112 ++++++++++++++- src/main/runtime/rpc/e2ee-channel-v2.test.ts | 22 +++ src/main/runtime/rpc/e2ee-channel.test.ts | 3 +- src/main/runtime/rpc/e2ee-channel.ts | 40 +++--- .../terminal-multiplex-frame-delivery.ts | 3 +- .../rpc/mobile-e2ee-desktop-outbound-owner.ts | 76 ++++++---- .../rpc/mobile-e2ee-outbound-admission.ts | 13 +- .../rpc/mobile-e2ee-v2-desktop-outbound.ts | 18 ++- src/main/runtime/rpc/mobile-socket-wiring.ts | 3 +- src/main/runtime/rpc/rpc-binary-sender.ts | 12 ++ .../runtime-browser-commands-browser-click.ts | 2 + ...-commands-browser-command-target-params.ts | 4 +- ...ime-browser-commands-browser-screencast.ts | 23 +-- .../runtime-browser-screencast-controller.ts | 7 +- .../runtime-edge-command-controller.ts | 3 +- .../runtime-rpc-websocket-dispatch.ts | 3 +- ...-page-browser-commands-host-test-double.ts | 9 ++ src/shared/ws-outbound-backpressure-queue.ts | 16 ++- 27 files changed, 639 insertions(+), 117 deletions(-) create mode 100644 src/main/runtime/browser-screencast-subscriber-frame-delivery.ts create mode 100644 src/main/runtime/orca-runtime-browser-screencast-e2ee-backlog.test.ts create mode 100644 src/main/runtime/orca-runtime-browser-screencast-pending-frame-retry.test.ts create mode 100644 src/main/runtime/rpc/rpc-binary-sender.ts diff --git a/src/main/runtime/browser-screencast-subscriber-frame-delivery.ts b/src/main/runtime/browser-screencast-subscriber-frame-delivery.ts new file mode 100644 index 00000000000..1e6621acc20 --- /dev/null +++ b/src/main/runtime/browser-screencast-subscriber-frame-delivery.ts @@ -0,0 +1,57 @@ +import type { ActiveBrowserScreencastSubscriber } from './runtime-browser-commands-browser-command-target-params' +import { recordScreencastSubscriberSend } from './browser-screencast-ghost-subscriber-eviction' +import { sendRemoteBrowserScreencastFrame } from './remote-browser-screencast-frame-admission' + +// Why 50: the frame pacer's own backpressure retry interval. +export const SCREENCAST_PENDING_FRAME_RETRY_MS = 50 + +/** + * Sends one frame to one viewer. A frame its socket did not take is kept as the viewer's newest + * pending frame and retried until taken: an idle page produces no next frame, so without the retry + * a viewer behind a briefly full socket keeps a stale picture until the page changes. + */ +export function deliverScreencastSubscriberFrame( + subscriber: ActiveBrowserScreencastSubscriber, + bytes: Uint8Array +): void { + const outcome = sendRemoteBrowserScreencastFrame(subscriber.sendBinary, bytes) + subscriber.pendingFrame = outcome === 'handled' ? null : bytes + // Why: a backlogged socket is alive and busy, so only a refusal is evidence of a ghost viewer. + if (outcome !== 'backlogged') { + subscriber.delivery = recordScreencastSubscriberSend(subscriber.delivery, outcome === 'handled') + } + if (outcome !== 'handled') { + schedulePendingFrameRetry(subscriber) + } +} + +/** Paths that remove a viewer that may have received frames call this, so no retry outlives it. */ +export function cancelScreencastSubscriberFrameRetry( + subscriber: ActiveBrowserScreencastSubscriber +): void { + if (subscriber.pendingFrameRetry) { + clearTimeout(subscriber.pendingFrameRetry) + subscriber.pendingFrameRetry = null + } + subscriber.pendingFrame = null +} + +function schedulePendingFrameRetry(subscriber: ActiveBrowserScreencastSubscriber): void { + if (subscriber.pendingFrameRetry) { + return + } + subscriber.pendingFrameRetry = setTimeout(() => { + subscriber.pendingFrameRetry = null + const bytes = subscriber.pendingFrame + if (!bytes) { + return + } + // Why a refused retry is not recorded: ghost eviction must advance only on produced frames. + if (sendRemoteBrowserScreencastFrame(subscriber.sendBinary, bytes) === 'handled') { + subscriber.pendingFrame = null + subscriber.delivery = recordScreencastSubscriberSend(subscriber.delivery, true) + } else { + schedulePendingFrameRetry(subscriber) + } + }, SCREENCAST_PENDING_FRAME_RETRY_MS) +} diff --git a/src/main/runtime/orca-runtime-browser-screencast-e2ee-backlog.test.ts b/src/main/runtime/orca-runtime-browser-screencast-e2ee-backlog.test.ts new file mode 100644 index 00000000000..b8df6851dae --- /dev/null +++ b/src/main/runtime/orca-runtime-browser-screencast-e2ee-backlog.test.ts @@ -0,0 +1,126 @@ +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' +import type { WebSocket } from 'ws' +import { BROWSER_SCREENCAST_GHOST_SUBSCRIBER_REFUSAL_LIMIT } from './browser-screencast-ghost-subscriber-eviction' +import { SCREENCAST_PENDING_FRAME_RETRY_MS } from './browser-screencast-subscriber-frame-delivery' +import { + createSinglePageBrowserCommandsHost, + createStreamedGuest +} from './single-page-browser-commands-host-test-double' +import { E2EEChannel } from './rpc/e2ee-channel' +import { decrypt, decryptBytes, deriveSharedKey, encrypt, generateKeyPair } from './rpc/e2ee-crypto' +import type { RpcBinarySender } from './rpc/rpc-binary-sender' + +const { webContentsFromId, startBrowserScreencast } = vi.hoisted(() => ({ + webContentsFromId: vi.fn(), + startBrowserScreencast: vi.fn() +})) + +vi.mock('electron', () => ({ + ipcMain: { on: vi.fn(), removeListener: vi.fn() }, + webContents: { fromId: webContentsFromId } +})) +vi.mock('../browser/browser-screencast-stream', () => ({ startBrowserScreencast })) +vi.mock('../telemetry/client', () => ({ track: vi.fn() })) + +/** A real legacy E2EE channel, as every directly connected phone uses. */ +function connectLegacyChannel() { + const serverKeys = generateKeyPair() + const clientKeys = generateKeyPair() + const sent: (string | Buffer)[] = [] + const ws = { + OPEN: 1 as const, + readyState: 1, + bufferedAmount: 0, + send: vi.fn((data: string | Buffer) => sent.push(data)), + close: vi.fn() + } + const onError = vi.fn() + const channel = new E2EEChannel(ws as unknown as WebSocket, { + serverSecretKey: serverKeys.secretKey, + resolveAuthenticatedDevice: (token) => + token === 'valid-token' + ? { deviceId: 'device-1', deviceToken: token, scope: 'mobile' } + : null, + onReady: vi.fn(), + onError + }) + const sharedKey = deriveSharedKey(clientKeys.secretKey, serverKeys.publicKey) + channel.handleRawMessage( + JSON.stringify({ + type: 'e2ee_hello', + publicKeyB64: Buffer.from(clientKeys.publicKey).toString('base64') + }) + ) + channel.handleRawMessage( + encrypt(JSON.stringify({ type: 'e2ee_auth', deviceToken: 'valid-token' }), sharedKey) + ) + let reply!: (response: string) => void + let sendBinary!: RpcBinarySender + channel.onMessage((_plaintext, textReply, binaryReply) => { + reply = textReply + sendBinary = binaryReply + }) + channel.handleRawMessage(encrypt('{"id":"s","method":"browser.screencast"}', sharedKey)) + const decode = (frame: string | Buffer): string => + typeof frame === 'string' + ? decrypt(frame, sharedKey)! + : `binary:${Buffer.from(decryptBytes(new Uint8Array(frame), sharedKey)!).join(',')}` + return { ws, sent, onError, reply, sendBinary, decode, channel } +} + +async function subscribe(sendBinary: RpcBinarySender) { + const { RuntimeBrowserCommands } = await import('./orca-runtime-browser') + let resolveDone!: () => void + const done = new Promise((resolve) => { + resolveDone = resolve + }) + const stop = vi.fn(() => resolveDone()) + startBrowserScreencast.mockResolvedValue({ stop, done, updateViewport: vi.fn(async () => {}) }) + const commands = new RuntimeBrowserCommands(createSinglePageBrowserCommandsHost()) + const subscription = await commands.browserScreencast( + { worktree: 'id:wt-1', page: 'page-1', format: 'jpeg' }, + { sendBinary } + ) + const onFrame: (bytes: Uint8Array) => unknown = + startBrowserScreencast.mock.calls[0][1].onFrame + return { subscription, onFrame, stop } +} + +describe('screencast over a backlogged legacy E2EE socket', () => { + beforeEach(() => { + vi.useFakeTimers() + webContentsFromId.mockReset() + webContentsFromId.mockReturnValue(createStreamedGuest()) + startBrowserScreencast.mockReset() + }) + afterEach(() => { + vi.useRealTimers() + }) + + // Why: a busy socket is alive; evicting its viewer as a ghost would blank a working phone. + it('keeps the viewer through a long backlog and then delivers only the newest frame', async () => { + const transport = connectLegacyChannel() + const { subscription, onFrame, stop } = await subscribe(transport.sendBinary) + onFrame(new Uint8Array([0])) + const baseline = transport.sent.length + + transport.ws.bufferedAmount = 9 * 1024 * 1024 + transport.reply('{"parked":true}') + const frameCount = BROWSER_SCREENCAST_GHOST_SUBSCRIBER_REFUSAL_LIMIT + 10 + for (let frame = 1; frame <= frameCount; frame++) { + onFrame(new Uint8Array([frame])) + } + expect(stop).not.toHaveBeenCalled() + expect(transport.sent.length).toBe(baseline) + + transport.ws.bufferedAmount = 0 + await vi.advanceTimersByTimeAsync(SCREENCAST_PENDING_FRAME_RETRY_MS * 2) + expect(transport.sent.slice(baseline).map(transport.decode)).toEqual([ + '{"parked":true}', + `binary:${frameCount}` + ]) + expect(transport.onError).not.toHaveBeenCalled() + subscription.session.stop() + transport.channel.destroy() + }) +}) diff --git a/src/main/runtime/orca-runtime-browser-screencast-fanout.test.ts b/src/main/runtime/orca-runtime-browser-screencast-fanout.test.ts index 863f7aee302..abdfacac33c 100644 --- a/src/main/runtime/orca-runtime-browser-screencast-fanout.test.ts +++ b/src/main/runtime/orca-runtime-browser-screencast-fanout.test.ts @@ -2,7 +2,10 @@ import { beforeEach, describe, expect, it, vi } from 'vitest' import { REMOTE_RUNTIME_MAX_OUTBOUND_BINARY_FRAME_BYTES } from '../../shared/remote-runtime-memory-limits' import { BROWSER_SCREENCAST_GHOST_SUBSCRIBER_REFUSAL_LIMIT } from './browser-screencast-ghost-subscriber-eviction' import type { RuntimeBrowserCommands } from './orca-runtime-browser' -import { createSinglePageBrowserCommandsHost } from './single-page-browser-commands-host-test-double' +import { + createSinglePageBrowserCommandsHost, + createStreamedGuest +} from './single-page-browser-commands-host-test-double' const { webContentsFromId, startBrowserScreencast } = vi.hoisted(() => ({ webContentsFromId: vi.fn(), @@ -23,15 +26,6 @@ function deferred() { return { promise, resolve } } -// A guest the page stream keeps painting while it streams. -function createStreamedGuest() { - return { - isDestroyed: () => false, - setBackgroundThrottling: vi.fn(), - capturePage: vi.fn(async () => null) - } -} - describe('RuntimeBrowserCommands screencast fanout', () => { beforeEach(() => { webContentsFromId.mockReset() @@ -72,8 +66,8 @@ describe('RuntimeBrowserCommands screencast fanout', () => { const frame = new Uint8Array([1, 2, 3]) expect(startBrowserScreencast).toHaveBeenCalledOnce() expect(startBrowserScreencast.mock.calls[0][1].onFrame(frame)).toBe(true) - expect(firstSend).toHaveBeenCalledWith(frame) - expect(secondSend).toHaveBeenCalledWith(frame) + expect(firstSend).toHaveBeenCalledWith(frame, { dropWhenBacklogged: true }) + expect(secondSend).toHaveBeenCalledWith(frame, { dropWhenBacklogged: true }) expect(updateViewport).toHaveBeenLastCalledWith( expect.objectContaining({ viewportWidth: 800, viewportHeight: 600 }) ) @@ -167,12 +161,12 @@ describe('RuntimeBrowserCommands screencast fanout', () => { { sendBinary: joinerSend } ) - expect(creatorSend).toHaveBeenCalledWith(snapshot) - expect(joinerSend).toHaveBeenCalledWith(snapshot) + expect(creatorSend).toHaveBeenCalledWith(snapshot, { dropWhenBacklogged: true }) + expect(joinerSend).toHaveBeenCalledWith(snapshot, { dropWhenBacklogged: true }) joinerSend.mockClear() gateOpen = true joiner.flushPendingFrame() - expect(joinerSend).toHaveBeenCalledExactlyOnceWith(snapshot) + expect(joinerSend).toHaveBeenCalledExactlyOnceWith(snapshot, { dropWhenBacklogged: true }) // The accepted replay is not retained, so a second flush cannot duplicate it. joinerSend.mockClear() diff --git a/src/main/runtime/orca-runtime-browser-screencast-pending-frame-retry.test.ts b/src/main/runtime/orca-runtime-browser-screencast-pending-frame-retry.test.ts new file mode 100644 index 00000000000..2fbf9927e69 --- /dev/null +++ b/src/main/runtime/orca-runtime-browser-screencast-pending-frame-retry.test.ts @@ -0,0 +1,136 @@ +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' +import { BROWSER_SCREENCAST_GHOST_SUBSCRIBER_REFUSAL_LIMIT } from './browser-screencast-ghost-subscriber-eviction' +import { SCREENCAST_PENDING_FRAME_RETRY_MS } from './browser-screencast-subscriber-frame-delivery' +import { + createSinglePageBrowserCommandsHost, + createStreamedGuest +} from './single-page-browser-commands-host-test-double' +import type { RpcBinarySender } from './rpc/rpc-binary-sender' + +const { webContentsFromId, startBrowserScreencast } = vi.hoisted(() => ({ + webContentsFromId: vi.fn(), + startBrowserScreencast: vi.fn() +})) + +vi.mock('electron', () => ({ + ipcMain: { on: vi.fn(), removeListener: vi.fn() }, + webContents: { fromId: webContentsFromId } +})) +vi.mock('../browser/browser-screencast-stream', () => ({ startBrowserScreencast })) + +async function subscribe(sendBinary: RpcBinarySender) { + const { RuntimeBrowserCommands } = await import('./orca-runtime-browser') + let resolveDone!: () => void + const done = new Promise((resolve) => { + resolveDone = resolve + }) + const stop = vi.fn(() => resolveDone()) + startBrowserScreencast.mockResolvedValue({ + stop, + done, + updateViewport: vi.fn(async () => {}) + }) + const commands = new RuntimeBrowserCommands(createSinglePageBrowserCommandsHost()) + const subscription = await commands.browserScreencast( + { worktree: 'id:wt-1', page: 'page-1', format: 'jpeg' }, + { sendBinary } + ) + const onFrame: (bytes: Uint8Array) => unknown = + startBrowserScreencast.mock.calls[0][1].onFrame + return { subscription, onFrame, stop } +} + +describe('screencast frames a viewer refused', () => { + beforeEach(() => { + vi.useFakeTimers() + webContentsFromId.mockReset() + webContentsFromId.mockReturnValue(createStreamedGuest()) + startBrowserScreencast.mockReset() + }) + afterEach(() => { + vi.useRealTimers() + }) + + // Why: an idle page sends no next frame, so a refused last frame must reach the viewer once its + // socket drains, or a relayed phone (whose server leg drops on a full socket) keeps a stale page. + it('delivers the last refused frame once the socket drains, with no new frame produced', async () => { + let socketFull = true + const received: Uint8Array[] = [] + const { subscription, onFrame } = await subscribe((bytes) => { + if (socketFull) { + return false + } + received.push(bytes) + return true + }) + const first = new Uint8Array([1]) + const last = new Uint8Array([2]) + onFrame(first) + onFrame(last) + await vi.advanceTimersByTimeAsync(SCREENCAST_PENDING_FRAME_RETRY_MS * 4) + expect(received).toEqual([]) + + socketFull = false + await vi.advanceTimersByTimeAsync(SCREENCAST_PENDING_FRAME_RETRY_MS) + expect(received).toEqual([last]) + await vi.advanceTimersByTimeAsync(SCREENCAST_PENDING_FRAME_RETRY_MS * 4) + expect(received).toEqual([last]) + subscription.session.stop() + }) + + it('retries the newest refused frame when a newer one is refused mid-retry', async () => { + let socketFull = true + const received: Uint8Array[] = [] + const { subscription, onFrame } = await subscribe((bytes) => { + if (socketFull) { + return false + } + received.push(bytes) + return true + }) + const stale = new Uint8Array([1]) + const newest = new Uint8Array([2]) + onFrame(stale) + await vi.advanceTimersByTimeAsync(SCREENCAST_PENDING_FRAME_RETRY_MS * 2) + onFrame(newest) + socketFull = false + await vi.advanceTimersByTimeAsync(SCREENCAST_PENDING_FRAME_RETRY_MS * 4) + expect(received).toEqual([newest]) + subscription.session.stop() + }) + + // Why: a lossy frame must not queue behind a backlog; the retry above delivers the newest instead. + it('asks the transport to drop frames while the socket is backlogged', async () => { + const sendBinary = vi.fn(() => true) + const { subscription, onFrame } = await subscribe(sendBinary) + onFrame(new Uint8Array([1])) + expect(sendBinary).toHaveBeenCalledWith(new Uint8Array([1]), { dropWhenBacklogged: true }) + subscription.session.stop() + }) + + it('stops retrying once the viewer leaves', async () => { + const sendBinary = vi.fn(() => false) + const { subscription, onFrame } = await subscribe(sendBinary) + onFrame(new Uint8Array([1])) + subscription.session.stop() + await subscription.session.done + sendBinary.mockClear() + await vi.advanceTimersByTimeAsync(SCREENCAST_PENDING_FRAME_RETRY_MS * 4) + expect(sendBinary).not.toHaveBeenCalled() + }) + + it('does not count refused retries toward ghost eviction', async () => { + let refusing = false + const { subscription, onFrame, stop } = await subscribe(() => !refusing) + onFrame(new Uint8Array([0])) + refusing = true + onFrame(new Uint8Array([1])) + await vi.advanceTimersByTimeAsync( + SCREENCAST_PENDING_FRAME_RETRY_MS * BROWSER_SCREENCAST_GHOST_SUBSCRIBER_REFUSAL_LIMIT * 2 + ) + // Eviction is judged when a frame is produced. + onFrame(new Uint8Array([2])) + expect(stop).not.toHaveBeenCalled() + subscription.session.stop() + }) +}) diff --git a/src/main/runtime/orca-runtime-browser.test.ts b/src/main/runtime/orca-runtime-browser.test.ts index 27accca72bf..a2d9569650d 100644 --- a/src/main/runtime/orca-runtime-browser.test.ts +++ b/src/main/runtime/orca-runtime-browser.test.ts @@ -669,8 +669,8 @@ describe('RuntimeBrowserCommands browser screencast', () => { expect(first.subscriptionId).not.toBe(second.subscriptionId) const frame = new Uint8Array([1, 2, 3]) expect(startBrowserScreencastMock.mock.calls[0][1].onFrame(frame)).toBe(true) - expect(firstSend).toHaveBeenCalledWith(frame) - expect(secondSend).toHaveBeenCalledWith(frame) + expect(firstSend).toHaveBeenCalledWith(frame, { dropWhenBacklogged: true }) + expect(secondSend).toHaveBeenCalledWith(frame, { dropWhenBacklogged: true }) expect(updateViewport).toHaveBeenLastCalledWith( expect.objectContaining({ viewportWidth: 800, viewportHeight: 600 }) ) diff --git a/src/main/runtime/orca-runtime-tests/mobile-creation-and-orchestration.spec.ts b/src/main/runtime/orca-runtime-tests/mobile-creation-and-orchestration.spec.ts index 05101384754..9357f7248f7 100644 --- a/src/main/runtime/orca-runtime-tests/mobile-creation-and-orchestration.spec.ts +++ b/src/main/runtime/orca-runtime-tests/mobile-creation-and-orchestration.spec.ts @@ -672,7 +672,7 @@ describe('OrcaRuntimeService', () => { ) // The gate still refuses the frame on arrival; it reaches the client only via the // post-ready replay, so a static page does not leave the subscriber frameless. - expect(sendBinary).toHaveBeenCalledExactlyOnceWith(startupFrame) + expect(sendBinary).toHaveBeenCalledExactlyOnceWith(startupFrame, undefined) runtime.cleanupSubscription('browser-screencast:page-1:first') await task diff --git a/src/main/runtime/remote-browser-screencast-frame-admission.test.ts b/src/main/runtime/remote-browser-screencast-frame-admission.test.ts index 8f0f1e83081..e29d58d045a 100644 --- a/src/main/runtime/remote-browser-screencast-frame-admission.test.ts +++ b/src/main/runtime/remote-browser-screencast-frame-admission.test.ts @@ -7,6 +7,7 @@ import { REMOTE_RUNTIME_MAX_OUTBOUND_BINARY_FRAME_BYTES } from '../../shared/rem import { startBrowserScreencast } from '../browser/browser-screencast-stream' import { E2EEChannel } from './rpc/e2ee-channel' import { sendRemoteBrowserScreencastFrame } from './remote-browser-screencast-frame-admission' +import type { RpcBinarySender } from './rpc/rpc-binary-sender' function createMockWebContents() { let attached = false @@ -58,7 +59,7 @@ function createRuntimeBinarySender() { channel.handleRawMessage( encrypt(JSON.stringify({ type: 'e2ee_auth', deviceToken: 'valid-token' }), sharedKey) ) - let sendBinary: ((bytes: Uint8Array) => boolean | void) | undefined + let sendBinary: RpcBinarySender | undefined channel.onMessage((_plaintext, _sendText, sendBinaryReply) => { sendBinary = sendBinaryReply }) @@ -74,7 +75,7 @@ describe('sendRemoteBrowserScreencastFrame', () => { const sendBinary = vi.fn(() => true) const oversized = new Uint8Array(REMOTE_RUNTIME_MAX_OUTBOUND_BINARY_FRAME_BYTES + 1) - expect(sendRemoteBrowserScreencastFrame(sendBinary, oversized)).toBe(true) + expect(sendRemoteBrowserScreencastFrame(sendBinary, oversized)).toBe('handled') expect(sendBinary).not.toHaveBeenCalled() }) @@ -86,14 +87,20 @@ describe('sendRemoteBrowserScreencastFrame', () => { vi.fn(() => false), withinLimit ) - ).toBe(false) + ).toBe('refused') expect( sendRemoteBrowserScreencastFrame( vi.fn(() => true), withinLimit ) - ).toBe(true) - expect(sendRemoteBrowserScreencastFrame(vi.fn(), withinLimit)).toBe(true) + ).toBe('handled') + expect(sendRemoteBrowserScreencastFrame(vi.fn(), withinLimit)).toBe('handled') + expect( + sendRemoteBrowserScreencastFrame( + vi.fn(() => 'backlogged' as const), + withinLimit + ) + ).toBe('backlogged') }) it('drops an oversized encoded frame without closing or stalling the paired stream', async () => { @@ -107,7 +114,8 @@ describe('sendRemoteBrowserScreencastFrame', () => { maxHeight: 2160, everyNthFrame: 1, minFrameIntervalMs: 0, - onFrame: (bytes) => sendRemoteBrowserScreencastFrame(transport.sendBinary, bytes) + onFrame: (bytes) => + sendRemoteBrowserScreencastFrame(transport.sendBinary, bytes) !== 'refused' }) try { diff --git a/src/main/runtime/remote-browser-screencast-frame-admission.ts b/src/main/runtime/remote-browser-screencast-frame-admission.ts index 6e4d1255b2a..395c1bd0163 100644 --- a/src/main/runtime/remote-browser-screencast-frame-admission.ts +++ b/src/main/runtime/remote-browser-screencast-frame-admission.ts @@ -1,14 +1,19 @@ import { isRemoteRuntimeBinaryFrameWithinLimit } from '../../shared/remote-runtime-memory-limits' +import type { RpcBinarySender } from './rpc/rpc-binary-sender' + +/** 'backlogged': a live socket skipped the frame; 'refused': the transport would not take it. */ +export type ScreencastFrameSendOutcome = 'handled' | 'backlogged' | 'refused' // Why: the E2EE channel closes the socket (1013) on an over-limit binary frame, and the -// screencast producer reads `false` as backpressure and retries the identical frame. Reporting +// screencast producer reads a refusal as backpressure and retries the identical frame. Reporting // an over-limit frame as handled drops it so the stream advances instead of retrying forever. export function sendRemoteBrowserScreencastFrame( - sendBinary: (bytes: Uint8Array) => boolean | void, + sendBinary: RpcBinarySender, bytes: Uint8Array -): boolean { +): ScreencastFrameSendOutcome { if (!isRemoteRuntimeBinaryFrameWithinLimit(bytes)) { - return true + return 'handled' } - return sendBinary(bytes) !== false + const sent = sendBinary(bytes, { dropWhenBacklogged: true }) + return sent === 'backlogged' ? 'backlogged' : sent === false ? 'refused' : 'handled' } diff --git a/src/main/runtime/rpc/core.ts b/src/main/runtime/rpc/core.ts index ac9756c63bd..86226f83522 100644 --- a/src/main/runtime/rpc/core.ts +++ b/src/main/runtime/rpc/core.ts @@ -16,6 +16,7 @@ import type { import type { OrchestrationCompatibilityEvidence } from '../../../shared/orchestration-compatibility-evidence' import type { OrchestrationSessionCaller } from '../orchestration/orchestration-caller-identity' import type { RpcCallerIdentity } from './rpc-caller-identity' +import type { RpcBinarySender } from './rpc-binary-sender' import type { RpcMethodPermission } from './rpc-method-permission' export type PairingRpcContext = { @@ -123,7 +124,7 @@ export type RpcContext = { sync(phones: readonly DelegatedPhone[]): DelegatedMobileDeviceSyncResult } // Why: mobile terminal traffic bypasses JSON streaming; undefined on Unix/socket and non-E2EE WebSocket paths. - sendBinary?: (bytes: Uint8Array) => boolean | void + sendBinary?: RpcBinarySender // Why: binary terminal frames arrive outside JSON-RPC once a stream is established; handlers register only the stream IDs they created. registerBinaryStreamHandler?: ( streamId: number, diff --git a/src/main/runtime/rpc/e2ee-channel-text-backpressure.test.ts b/src/main/runtime/rpc/e2ee-channel-text-backpressure.test.ts index 5761ac504aa..c88c430ecd8 100644 --- a/src/main/runtime/rpc/e2ee-channel-text-backpressure.test.ts +++ b/src/main/runtime/rpc/e2ee-channel-text-backpressure.test.ts @@ -1,7 +1,8 @@ import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' import type { WebSocket } from 'ws' +import type { RpcBinarySendResult } from './rpc-binary-sender' import { E2EEChannel, type E2EEChannelOptions } from './e2ee-channel' -import { deriveSharedKey, decrypt, encrypt, generateKeyPair } from './e2ee-crypto' +import { deriveSharedKey, decrypt, decryptBytes, encrypt, generateKeyPair } from './e2ee-crypto' import { createMobileE2EEOutboundMemoryBudget } from './mobile-e2ee-outbound-memory-budget' import { REMOTE_RUNTIME_MAX_OUTBOUND_JSON_BYTES } from '../../../shared/remote-runtime-memory-limits' @@ -20,12 +21,12 @@ function publicKeyToBase64(key: Uint8Array): string { } function createMockWs() { - const sent: string[] = [] + const sent: (string | Buffer)[] = [] return { OPEN: 1 as const, readyState: 1, bufferedAmount: 0, - send: vi.fn((data: string) => { + send: vi.fn((data: string | Buffer) => { sent.push(data) }), close: vi.fn(), @@ -93,7 +94,9 @@ describe('E2EE text reply backpressure', () => { ctx.ws.bufferedAmount = 0 vi.runOnlyPendingTimers() - const replies = ctx.ws.sent.slice(baseline).map((frame) => decrypt(frame, ctx.sharedKey)) + const replies = ctx.ws.sent + .slice(baseline) + .map((frame) => decrypt(String(frame), ctx.sharedKey)) expect(replies).toEqual(['{"seq":1}', '{"seq":2}', '{"seq":3}']) expect(ctx.onError).not.toHaveBeenCalled() }) @@ -103,7 +106,7 @@ describe('E2EE text reply backpressure', () => { const baseline = ctx.ws.sent.length emitReply(ctx, '{"ok":true}') expect(ctx.ws.sent.length).toBe(baseline + 1) - expect(decrypt(ctx.ws.sent[baseline]!, ctx.sharedKey)).toBe('{"ok":true}') + expect(decrypt(String(ctx.ws.sent[baseline]), ctx.sharedKey)).toBe('{"ok":true}') }) it('still closes an oversized reply when telemetry throws', () => { @@ -143,3 +146,102 @@ describe('E2EE text reply backpressure', () => { expect(outboundMemoryBudget.evidence().queuedBytes).toBe(0) }) }) + +describe('E2EE legacy binary reply backpressure', () => { + beforeEach(() => { + vi.useFakeTimers() + trackMock.mockReset() + }) + afterEach(() => { + vi.useRealTimers() + }) + + function decodeSent(ctx: ReturnType, frame: string | Buffer): string { + return typeof frame === 'string' + ? decrypt(frame, ctx.sharedKey)! + : `binary:${Buffer.from(decryptBytes(new Uint8Array(frame), ctx.sharedKey)!).toString()}` + } + + // Why: legacy terminal.subscribe has no seq-gap check, so a dropped or reordered binary frame corrupts the phone's screen. + it('holds binary replies in one order with text while the outbound budget is full', () => { + const outboundMemoryBudget = createMobileE2EEOutboundMemoryBudget({ maxBufferedBytes: 1_000 }) + const ctx = setup({ outboundMemoryBudget }) + const baseline = ctx.ws.sent.length + ctx.ws.bufferedAmount = 1_001 + const accepted: RpcBinarySendResult[] = [] + + ctx.channel.onMessage((_plaintext, encryptedReply, encryptedBinaryReply) => { + encryptedReply('{"seq":1}') + accepted.push(encryptedBinaryReply(Buffer.from('2'))) + encryptedReply('{"seq":3}') + accepted.push(encryptedBinaryReply(Buffer.from('4'))) + }) + ctx.channel.handleRawMessage(encrypt('{"id":"x","method":"terminal.subscribe"}', ctx.sharedKey)) + expect(ctx.ws.sent.length).toBe(baseline) + + ctx.ws.bufferedAmount = 0 + vi.runOnlyPendingTimers() + + expect(accepted).toEqual([true, true]) + expect(ctx.ws.sent.slice(baseline).map((frame) => decodeSent(ctx, frame))).toEqual([ + '{"seq":1}', + 'binary:2', + '{"seq":3}', + 'binary:4' + ]) + expect(ctx.onError).not.toHaveBeenCalled() + }) + + it('closes for budget instead of dropping when queued binary overflows', () => { + const outboundMemoryBudget = createMobileE2EEOutboundMemoryBudget({ + maxBufferedBytes: 1_000, + maxQueuedBytes: 150 + }) + const ctx = setup({ outboundMemoryBudget }) + ctx.ws.bufferedAmount = 1_001 + + ctx.channel.onMessage((_plaintext, _encryptedReply, encryptedBinaryReply) => { + encryptedBinaryReply(new Uint8Array(100)) + encryptedBinaryReply(new Uint8Array(100)) + }) + ctx.channel.handleRawMessage(encrypt('{"id":"x","method":"terminal.subscribe"}', ctx.sharedKey)) + + expect(ctx.onError).toHaveBeenCalledWith(1013, 'Outbound reply buffer overflow') + expect(trackMock).toHaveBeenCalledWith('remote_outbound_budget_close', { emitter: 'queue' }) + }) + + it('refuses binary once the socket has left OPEN', () => { + const ctx = setup() + ctx.ws.readyState = 3 + let accepted: RpcBinarySendResult = undefined + + ctx.channel.onMessage((_plaintext, _encryptedReply, encryptedBinaryReply) => { + accepted = encryptedBinaryReply(new Uint8Array([1])) + }) + ctx.channel.handleRawMessage(encrypt('{"id":"x","method":"terminal.subscribe"}', ctx.sharedKey)) + + expect(accepted).toBe(false) + }) + + it('drops a backlogged lossy binary reply instead of queueing it, then sends once drained', () => { + const ctx = setup() + const baseline = ctx.ws.sent.length + ctx.ws.bufferedAmount = 9 * 1024 * 1024 + const accepted: RpcBinarySendResult[] = [] + ctx.channel.onMessage((_plaintext, _encryptedReply, encryptedBinaryReply) => { + accepted.push(encryptedBinaryReply(Buffer.from('frame'), { dropWhenBacklogged: true })) + }) + const request = encrypt('{"id":"x","method":"browser.screencast"}', ctx.sharedKey) + ctx.channel.handleRawMessage(request) + ctx.ws.bufferedAmount = 0 + vi.runOnlyPendingTimers() + expect(ctx.ws.sent.length).toBe(baseline) + + ctx.channel.handleRawMessage(request) + expect(accepted).toEqual(['backlogged', true]) + expect(ctx.ws.sent.slice(baseline).map((frame) => decodeSent(ctx, frame))).toEqual([ + 'binary:frame' + ]) + expect(ctx.onError).not.toHaveBeenCalled() + }) +}) diff --git a/src/main/runtime/rpc/e2ee-channel-v2.test.ts b/src/main/runtime/rpc/e2ee-channel-v2.test.ts index b26b057abaf..ddb4859a2d7 100644 --- a/src/main/runtime/rpc/e2ee-channel-v2.test.ts +++ b/src/main/runtime/rpc/e2ee-channel-v2.test.ts @@ -12,6 +12,7 @@ import { sealMobileE2EEV2Frame } from '../../../shared/mobile-e2ee-v2-framing' import { deriveSharedKey } from './e2ee-crypto' +import type { RpcBinarySendResult } from './rpc-binary-sender' import { E2EEChannel } from './e2ee-channel' import { deriveMobileE2EEV2KeySchedule } from './mobile-e2ee-v2-key-schedule' @@ -276,4 +277,25 @@ describe('E2EEChannel v2', () => { ) expect(ctx.ws.sent[3]!.options).toEqual({ binary: true }) }) + + it('drops a backlogged lossy binary reply without spending a counter, then sends once drained', () => { + const ctx = setup() + const { schedule } = startV2(ctx) + authenticate(ctx, schedule) + ctx.ws.bufferedAmount = 9 * 1024 * 1024 + const accepted: RpcBinarySendResult[] = [] + ctx.channel.onMessage((_request, _textReply, binaryReply) => { + accepted.push(binaryReply(new Uint8Array([1]), { dropWhenBacklogged: true })) + }) + ctx.channel.handleRawMessage(clientText('{"method":"browser.screencast"}', schedule, 1n)) + ctx.ws.bufferedAmount = 0 + vi.runOnlyPendingTimers() + expect(ctx.ws.sent).toHaveLength(2) + + ctx.channel.handleRawMessage(clientText('{"method":"browser.screencast"}', schedule, 2n)) + expect(accepted).toEqual(['backlogged', true]) + expect(openServerFrame(ctx.ws.sent[2]!.data, 'binary', schedule, 1n)).toEqual( + new Uint8Array([1]) + ) + }) }) diff --git a/src/main/runtime/rpc/e2ee-channel.test.ts b/src/main/runtime/rpc/e2ee-channel.test.ts index f4a09397d9a..5f61c4085a9 100644 --- a/src/main/runtime/rpc/e2ee-channel.test.ts +++ b/src/main/runtime/rpc/e2ee-channel.test.ts @@ -1,5 +1,6 @@ import { describe, expect, it, vi, beforeEach, afterEach } from 'vitest' import type { WebSocket } from 'ws' +import type { RpcBinarySendResult } from './rpc-binary-sender' import { E2EEChannel, type E2EEChannelOptions } from './e2ee-channel' import { generateKeyPair, deriveSharedKey, encrypt, decrypt, encryptBytes } from './e2ee-crypto' import { @@ -308,7 +309,7 @@ describe('E2EEChannel', () => { const ctx = setup() const sharedKey = doHandshake(ctx) const sentBefore = ctx.ws.sent.length - let accepted: boolean | void = undefined + let accepted: RpcBinarySendResult = undefined ctx.channel.onMessage((_plaintext, _encryptedReply, encryptedBinaryReply) => { accepted = encryptedBinaryReply( diff --git a/src/main/runtime/rpc/e2ee-channel.ts b/src/main/runtime/rpc/e2ee-channel.ts index a5ecbbdde33..9d70eeae5ae 100644 --- a/src/main/runtime/rpc/e2ee-channel.ts +++ b/src/main/runtime/rpc/e2ee-channel.ts @@ -1,6 +1,6 @@ // Why: this channel keeps E2EE framing out of RPC handlers, which consume plaintext across transports. import type { WebSocket } from 'ws' -import { deriveSharedKey, encrypt, decrypt, encryptBytes, decryptBytes } from './e2ee-crypto' +import { deriveSharedKey, encrypt, decrypt, decryptBytes } from './e2ee-crypto' import { DesktopMobileE2EEV2Session, type DesktopMobileE2EEV2Context @@ -17,6 +17,11 @@ import { parseRemoteRuntimeJsonText } from '../../../shared/remote-runtime-reque import type { MobileE2EEOutboundMemoryBudget } from './mobile-e2ee-outbound-memory-budget' import { MobileE2EEDesktopOutboundOwner } from './mobile-e2ee-desktop-outbound-owner' import { parseRuntimeClientCapabilities } from './runtime-client-capabilities' +import type { + RpcBinarySendOptions, + RpcBinarySendResult, + RpcBinarySender +} from './rpc-binary-sender' import type { RuntimeCapability } from '../../../shared/protocol-version' import type { EventProps } from '../../../shared/telemetry-events' import { track } from '../../telemetry/client' @@ -61,7 +66,7 @@ export class E2EEChannel { | (( plaintext: string, encryptedReply: (response: string) => void, - encryptedBinaryReply: (response: Uint8Array) => boolean | void + encryptedBinaryReply: RpcBinarySender ) => void) | null = null private binaryMessageHandler: ((plaintext: Uint8Array) => void) | null = null @@ -78,7 +83,12 @@ export class E2EEChannel { this.onError = options.onError this.transportContext = options.transportContext ?? { transport: 'direct' } this.requireV2 = options.requireV2 ?? false - this.outbound = new MobileE2EEDesktopOutboundOwner(ws, options.outboundMemoryBudget) + this.outbound = new MobileE2EEDesktopOutboundOwner( + ws, + () => Boolean(this.sharedKey), + () => this.closeForOutboundBudget('queue'), + options.outboundMemoryBudget + ) this.handshakeTimer = setTimeout(() => { this.onError(4002, 'E2EE handshake timeout') @@ -89,7 +99,7 @@ export class E2EEChannel { handler: ( plaintext: string, encryptedReply: (response: string) => void, - encryptedBinaryReply: (response: Uint8Array) => boolean | void + encryptedBinaryReply: RpcBinarySender ) => void ): void { this.messageHandler = handler @@ -154,13 +164,9 @@ export class E2EEChannel { this.closeForOutboundBudget('size') return } - this.outbound.enqueueLegacyText( - encrypt(response, this.sharedKey), - () => Boolean(this.sharedKey), - () => this.closeForOutboundBudget('queue') - ) + this.outbound.enqueueLegacyText(encrypt(response, this.sharedKey)) } - const encryptedBinaryReply = (response: Uint8Array): boolean => { + const encryptedBinaryReply: RpcBinarySender = (response, options) => { if (!this.sharedKey || this.ws.readyState !== this.ws.OPEN) { return false } @@ -168,11 +174,7 @@ export class E2EEChannel { this.closeForOutboundBudget('size') return false } - if (!this.outbound.canSend(response.byteLength + 40)) { - return false - } - this.ws.send(Buffer.from(encryptBytes(response, this.sharedKey)), { binary: true }) - return true + return this.outbound.enqueueLegacyBinary(response, this.sharedKey, options) } this.messageHandler?.(plaintext, encryptedReply, encryptedBinaryReply) } @@ -291,13 +293,13 @@ export class E2EEChannel { this.messageHandler?.( plaintext, (response) => this.enqueueV2({ kind: 'text', plaintext: response }), - (response) => this.enqueueV2({ kind: 'binary', plaintext: response }) + (response, options) => this.enqueueV2({ kind: 'binary', plaintext: response }, options) ), onProtocolError: () => this.onError(4001, 'Invalid binary message before authentication') }) } - private enqueueV2(item: V2OutboundItem): boolean { + private enqueueV2(item: V2OutboundItem, options?: RpcBinarySendOptions): RpcBinarySendResult { if (!this.v2Session || this.ws.readyState !== this.ws.OPEN) { return false } @@ -305,7 +307,7 @@ export class E2EEChannel { this.closeForOutboundBudget('size') return false } - return this.outbound.enqueueV2(item, this.v2Session, () => this.closeForOutboundBudget('queue')) + return this.outbound.enqueueV2(item, this.v2Session, options) } // Why: this close kills the whole remote session. `size` means a producer emitted something @@ -324,7 +326,7 @@ export class E2EEChannel { this.enqueueV2({ kind: 'text', plaintext: JSON.stringify(message) }) } else if (this.ws.readyState === this.ws.OPEN && this.sharedKey) { const frame = encrypt(JSON.stringify(message), this.sharedKey) - this.outbound.sendLegacyFrame(frame, () => this.closeForOutboundBudget('queue')) + this.outbound.sendLegacyFrame(frame) } } diff --git a/src/main/runtime/rpc/methods/terminal/terminal-multiplex-frame-delivery.ts b/src/main/runtime/rpc/methods/terminal/terminal-multiplex-frame-delivery.ts index 09682e10c08..dab3f8beaab 100644 --- a/src/main/runtime/rpc/methods/terminal/terminal-multiplex-frame-delivery.ts +++ b/src/main/runtime/rpc/methods/terminal/terminal-multiplex-frame-delivery.ts @@ -13,6 +13,7 @@ import { import type { TerminalOutputFrameChunk } from '../../terminal-output-frame-chunks' import type { TerminalStreamInputOutcome } from './terminal-input-delivery' import type { TerminalMultiplexStream } from './terminal-stream-types' +import type { RpcBinarySendResult } from '../../rpc-binary-sender' export function installMultiplexFrameDelivery(state: TerminalMultiplexConnection): void { const { sendBinary, emit } = state @@ -30,7 +31,7 @@ export function installMultiplexFrameDelivery(state: TerminalMultiplexConnection // Why: a seq-less Output chunk must carry sentinel 0, not the control-frame state.cursor, or it poisons the client's frame-drop tracker. const resolvedSeq = typeof seq === 'number' ? seq : opcode === TerminalStreamOpcode.Output ? 0 : state.cursor++ - let sent: boolean | void + let sent: RpcBinarySendResult try { sent = sendBinary(encodeTerminalStreamFrame({ opcode, streamId, seq: resolvedSeq, payload })) } catch { diff --git a/src/main/runtime/rpc/mobile-e2ee-desktop-outbound-owner.ts b/src/main/runtime/rpc/mobile-e2ee-desktop-outbound-owner.ts index 4eb6157bbd9..ee076381f43 100644 --- a/src/main/runtime/rpc/mobile-e2ee-desktop-outbound-owner.ts +++ b/src/main/runtime/rpc/mobile-e2ee-desktop-outbound-owner.ts @@ -1,8 +1,10 @@ import type { WebSocket } from 'ws' import type { WsOutboundBackpressureQueue } from '../../../shared/ws-outbound-backpressure-queue' -import { createLegacyMobileE2EETextReplyQueue } from './mobile-e2ee-outbound-admission' +import { createLegacyMobileE2EEReplyQueue } from './mobile-e2ee-outbound-admission' +import { encryptBytes } from './e2ee-crypto' import { createDesktopMobileE2EEV2OutboundQueue, + desktopMobileE2EEV2OutboundItemBytes, type DesktopMobileE2EEV2OutboundItem } from './mobile-e2ee-v2-desktop-outbound' import type { DesktopMobileE2EEV2Session } from './mobile-e2ee-v2-desktop-session' @@ -11,56 +13,61 @@ import { type MobileE2EEOutboundMemoryBudget, type MobileE2EEOutboundSocketMemory } from './mobile-e2ee-outbound-memory-budget' +import type { RpcBinarySendOptions } from './rpc-binary-sender' export class MobileE2EEDesktopOutboundOwner { private readonly memoryBudget: MobileE2EEOutboundMemoryBudget private readonly socketMemory: MobileE2EEOutboundSocketMemory | null - private legacyQueue: WsOutboundBackpressureQueue | null = null + // Why: one queue for text and binary, so a binary frame can neither overtake parked text nor drop silently. + private legacyQueue: WsOutboundBackpressureQueue | null = null private v2Queue: WsOutboundBackpressureQueue | null = null constructor( private readonly ws: WebSocket, + private readonly isLegacyKeyed: () => boolean, + private readonly onOverflow: () => void, memoryBudget: MobileE2EEOutboundMemoryBudget = createMobileE2EEOutboundMemoryBudget() ) { this.memoryBudget = memoryBudget this.socketMemory = memoryBudget.registerBufferedAmount(() => ws.bufferedAmount) } - canSend(bytes: number): boolean { - return this.socketMemory?.canSend(bytes) === true - } - - sendLegacyFrame(frame: string, onOverflow: () => void): boolean { - if (!this.canSend(frame.length) || this.ws.readyState !== this.ws.OPEN) { - onOverflow() + sendLegacyFrame(frame: string): boolean { + if (this.socketMemory?.canSend(frame.length) !== true || this.ws.readyState !== this.ws.OPEN) { + this.onOverflow() return false } this.ws.send(frame) return true } - enqueueLegacyText(frame: string, isKeyed: () => boolean, onOverflow: () => void): boolean { - if (!this.socketMemory) { - onOverflow() + enqueueLegacyText(frame: string): boolean { + return this.legacy()?.enqueue(frame) ?? false + } + + enqueueLegacyBinary( + bytes: Uint8Array, + sharedKey: Uint8Array, + options?: RpcBinarySendOptions + ): boolean | 'backlogged' { + const queue = this.legacy() + if (!queue) { return false } - this.legacyQueue ??= createLegacyMobileE2EETextReplyQueue({ - ws: this.ws, - isKeyed, - memoryBudget: this.memoryBudget, - socketMemory: this.socketMemory, - onOverflow - }) - return this.legacyQueue.enqueue(frame) + // Why: idleness is judged on the sealed size (+40) before sealing, so a skipped frame costs no encryption. + if (options?.dropWhenBacklogged && !queue.isIdle(bytes.byteLength + 40)) { + return 'backlogged' + } + return queue.enqueue(Buffer.from(encryptBytes(bytes, sharedKey))) } enqueueV2( item: DesktopMobileE2EEV2OutboundItem, session: DesktopMobileE2EEV2Session, - onOverflow: () => void - ): boolean { + options?: RpcBinarySendOptions + ): boolean | 'backlogged' { if (!this.socketMemory) { - onOverflow() + this.onOverflow() return false } this.v2Queue ??= createDesktopMobileE2EEV2OutboundQueue({ @@ -68,11 +75,32 @@ export class MobileE2EEDesktopOutboundOwner { session, memoryBudget: this.memoryBudget, socketMemory: this.socketMemory, - onOverflow + onOverflow: this.onOverflow }) + if ( + options?.dropWhenBacklogged && + !this.v2Queue.isIdle(desktopMobileE2EEV2OutboundItemBytes(item)) + ) { + return 'backlogged' + } return this.v2Queue.enqueue(item) } + private legacy(): WsOutboundBackpressureQueue | null { + if (!this.socketMemory) { + this.onOverflow() + return null + } + this.legacyQueue ??= createLegacyMobileE2EEReplyQueue({ + ws: this.ws, + isKeyed: this.isLegacyKeyed, + memoryBudget: this.memoryBudget, + socketMemory: this.socketMemory, + onOverflow: this.onOverflow + }) + return this.legacyQueue + } + dispose(): void { this.legacyQueue?.dispose() this.legacyQueue = null diff --git a/src/main/runtime/rpc/mobile-e2ee-outbound-admission.ts b/src/main/runtime/rpc/mobile-e2ee-outbound-admission.ts index f759f6f5c47..23bc58e2b4d 100644 --- a/src/main/runtime/rpc/mobile-e2ee-outbound-admission.ts +++ b/src/main/runtime/rpc/mobile-e2ee-outbound-admission.ts @@ -41,17 +41,18 @@ export function isMobileE2EEOutboundItemWithinLimit( : isMobileE2EEBinaryPayloadWithinLimit(item.plaintext) } -export function createLegacyMobileE2EETextReplyQueue(args: { +export function createLegacyMobileE2EEReplyQueue(args: { ws: WebSocket isKeyed: () => boolean onOverflow: () => void memoryBudget: MobileE2EEOutboundMemoryBudget socketMemory: MobileE2EEOutboundSocketMemory -}): WsOutboundBackpressureQueue { - return createWsOutboundBackpressureQueue({ - send: (frame) => args.ws.send(frame), - // Encrypted replies are base64 ASCII strings, so length === byte count. - byteLengthOf: (frame) => frame.length, +}): WsOutboundBackpressureQueue { + return createWsOutboundBackpressureQueue({ + send: (frame) => + typeof frame === 'string' ? args.ws.send(frame) : args.ws.send(frame, { binary: true }), + // Encrypted text replies are base64 ASCII strings, so length === byte count. + byteLengthOf: (frame) => (typeof frame === 'string' ? frame.length : frame.byteLength), getBufferedAmount: () => args.ws.bufferedAmount, isWritable: () => args.isKeyed() && args.ws.readyState === args.ws.OPEN, canSend: (bytes) => args.socketMemory.canSend(bytes), diff --git a/src/main/runtime/rpc/mobile-e2ee-v2-desktop-outbound.ts b/src/main/runtime/rpc/mobile-e2ee-v2-desktop-outbound.ts index 9bf581037f2..e0c3068e3f8 100644 --- a/src/main/runtime/rpc/mobile-e2ee-v2-desktop-outbound.ts +++ b/src/main/runtime/rpc/mobile-e2ee-v2-desktop-outbound.ts @@ -17,6 +17,16 @@ export type DesktopMobileE2EEV2OutboundItem = | { kind: 'text'; plaintext: string } | { kind: 'binary'; plaintext: Uint8Array } +export function desktopMobileE2EEV2OutboundItemBytes( + item: DesktopMobileE2EEV2OutboundItem +): number { + const bytes = + item.kind === 'text' + ? mobileE2EETextPayloadAdmissionBytes(item.plaintext) + : mobileE2EEBinaryPayloadAdmissionBytes(item.plaintext) + return Number.isFinite(bytes) ? bytes + 82 : bytes +} + export function createDesktopMobileE2EEV2OutboundQueue(args: { ws: WebSocket session: DesktopMobileE2EEV2Session @@ -34,13 +44,7 @@ export function createDesktopMobileE2EEV2OutboundQueue(args: { args.ws.send(Buffer.from(args.session.sealBinary(item.plaintext)), { binary: true }) } }, - byteLengthOf: (item) => { - const bytes = - item.kind === 'text' - ? mobileE2EETextPayloadAdmissionBytes(item.plaintext) - : mobileE2EEBinaryPayloadAdmissionBytes(item.plaintext) - return Number.isFinite(bytes) ? bytes + 82 : bytes - }, + byteLengthOf: desktopMobileE2EEV2OutboundItemBytes, getBufferedAmount: () => args.ws.bufferedAmount, isWritable: () => args.ws.readyState === args.ws.OPEN, canSend: (bytes) => args.socketMemory.canSend(bytes), diff --git a/src/main/runtime/rpc/mobile-socket-wiring.ts b/src/main/runtime/rpc/mobile-socket-wiring.ts index 2d536b15038..66cda2b0c35 100644 --- a/src/main/runtime/rpc/mobile-socket-wiring.ts +++ b/src/main/runtime/rpc/mobile-socket-wiring.ts @@ -5,6 +5,7 @@ import type { E2EEKeypair } from '../e2ee-keypair' import { E2EEChannel, type E2EEAuthenticatedDevice } from './e2ee-channel' import { createMobileE2EEOutboundMemoryBudget } from './mobile-e2ee-outbound-memory-budget' import type { RuntimeCapability } from '../../../shared/protocol-version' +import type { RpcBinarySender } from './rpc-binary-sender' type MobileSocketPayload = string | Uint8Array @@ -48,7 +49,7 @@ type MobileSocketWiringOptions = { socket: AuthenticatedMobileSocket, plaintext: string, reply: (response: string) => void, - sendBinary: (response: Uint8Array) => boolean | void + sendBinary: RpcBinarySender ) => void onBinary: (socket: AuthenticatedMobileSocket, bytes: Uint8Array) => void onClose: (socket: AuthenticatedMobileSocket | null, hasOtherConnections: boolean) => void diff --git a/src/main/runtime/rpc/rpc-binary-sender.ts b/src/main/runtime/rpc/rpc-binary-sender.ts new file mode 100644 index 00000000000..2a56ac5f28a --- /dev/null +++ b/src/main/runtime/rpc/rpc-binary-sender.ts @@ -0,0 +1,12 @@ +export type RpcBinarySendOptions = { + // Why: a lossy stream resends its newest frame later, so queueing stale frames behind a backlog only delays it. + dropWhenBacklogged?: boolean +} + +/** False: not sent or queued. 'backlogged': skipped under dropWhenBacklogged; the socket is alive. */ +export type RpcBinarySendResult = boolean | void | 'backlogged' + +export type RpcBinarySender = ( + bytes: Uint8Array, + options?: RpcBinarySendOptions +) => RpcBinarySendResult diff --git a/src/main/runtime/runtime-browser-commands-browser-click.ts b/src/main/runtime/runtime-browser-commands-browser-click.ts index 49fe9b7900c..619fce3aed0 100644 --- a/src/main/runtime/runtime-browser-commands-browser-click.ts +++ b/src/main/runtime/runtime-browser-commands-browser-click.ts @@ -20,6 +20,7 @@ import type { BrowserTypeResult } from '../../shared/runtime-types' import type { BrowserScreencastSession } from '../browser/browser-screencast-stream-types' +import { cancelScreencastSubscriberFrameRetry } from './browser-screencast-subscriber-frame-delivery' export class RuntimeBrowserCommandsWithBrowserClick extends RuntimeBrowserCommandsWithActiveScreencastsByPageId { async browserClick( @@ -152,6 +153,7 @@ export class RuntimeBrowserCommandsWithBrowserClick extends RuntimeBrowserComman return } active.subscribers.delete(subscriptionId) + cancelScreencastSubscriberFrameRetry(subscriber) subscriber.resolveDone() if (active.viewportOwnerSubscriptionId === subscriptionId) { const fallback = Array.from(active.subscribers.entries()).findLast(([, candidate]) => diff --git a/src/main/runtime/runtime-browser-commands-browser-command-target-params.ts b/src/main/runtime/runtime-browser-commands-browser-command-target-params.ts index 2e6a447b4f7..c85f6ddc176 100644 --- a/src/main/runtime/runtime-browser-commands-browser-command-target-params.ts +++ b/src/main/runtime/runtime-browser-commands-browser-command-target-params.ts @@ -18,6 +18,7 @@ import type { RuntimeBrowserPageRegistry } from './runtime-browser-page-registry import type { BrowserWindow } from 'electron' import type { BrowserBackend } from '../browser/browser-backend' import type { BrowserSessionTabSelectionOptions } from './browser-tab-create-publication' +import type { RpcBinarySender } from './rpc/rpc-binary-sender' export type BrowserCommandTargetParams = { worktree?: string @@ -58,13 +59,14 @@ export type BrowserScreencastStartResult = { } export type ActiveBrowserScreencastSubscriber = { - sendBinary: (bytes: Uint8Array) => boolean | void + sendBinary: RpcBinarySender emit?: (event: BrowserScreencastResult) => void done: Promise resolveDone: () => void viewport: BrowserScreencastViewport budget: BrowserScreencastFrameBudget pendingFrame: Uint8Array | null + pendingFrameRetry: ReturnType | null // Why: identifies the viewer across reconnects, which the RPC connectionId cannot — a new // socket never reuses the old id, so a reconnecting device would stack a second subscription. pairedDeviceId?: string diff --git a/src/main/runtime/runtime-browser-commands-browser-screencast.ts b/src/main/runtime/runtime-browser-commands-browser-screencast.ts index 2103f37c326..48a3ba23496 100644 --- a/src/main/runtime/runtime-browser-commands-browser-screencast.ts +++ b/src/main/runtime/runtime-browser-commands-browser-screencast.ts @@ -18,19 +18,22 @@ import { BrowserError } from '../browser/browser-error' import { randomUUID } from 'node:crypto' import { startBrowserScreencast } from '../browser/browser-screencast-stream' import { rendererPublicationThrottle } from '../window/renderer-publication-throttle' -import { sendRemoteBrowserScreencastFrame } from './remote-browser-screencast-frame-admission' +import { + cancelScreencastSubscriberFrameRetry, + deliverScreencastSubscriberFrame +} from './browser-screencast-subscriber-frame-delivery' import { INITIAL_SCREENCAST_SUBSCRIBER_DELIVERY, - recordScreencastSubscriberSend, screencastSubscriberIsGhost } from './browser-screencast-ghost-subscriber-eviction' import type { BrowserScreencastSession } from '../browser/browser-screencast-stream-types' +import type { RpcBinarySender } from './rpc/rpc-binary-sender' export class RuntimeBrowserCommandsWithBrowserScreencast extends RuntimeBrowserCommandsWithBrowserClick { async browserScreencast( params: BrowserScreencastParams, stream: { - sendBinary: (bytes: Uint8Array) => boolean | void + sendBinary: RpcBinarySender emit?: (event: BrowserScreencastResult) => void pairedDeviceId?: string } @@ -82,11 +85,8 @@ export class RuntimeBrowserCommandsWithBrowserScreencast extends RuntimeBrowserC onFrame: (bytes) => { const ghosts: string[] = [] for (const [subscriptionId, subscriber] of record.subscribers) { - // A slow viewer drops this frame without stalling every other viewer, but the - // newest refusal is retained so a gate that opens later can still be filled. - const delivered = sendRemoteBrowserScreencastFrame(subscriber.sendBinary, bytes) - subscriber.pendingFrame = delivered ? null : bytes - subscriber.delivery = recordScreencastSubscriberSend(subscriber.delivery, delivered) + // A slow viewer drops this frame without stalling every other viewer. + deliverScreencastSubscriberFrame(subscriber, bytes) if (screencastSubscriberIsGhost(subscriber.delivery)) { ghosts.push(subscriptionId) } @@ -131,6 +131,7 @@ export class RuntimeBrowserCommandsWithBrowserScreencast extends RuntimeBrowserC this.activeScreencastsByPageId.delete(browserPageId) } for (const subscriber of record.subscribers.values()) { + cancelScreencastSubscriberFrameRetry(subscriber) subscriber.resolveDone() } record.subscribers.clear() @@ -145,6 +146,7 @@ export class RuntimeBrowserCommandsWithBrowserScreencast extends RuntimeBrowserC viewport, budget, pendingFrame: null, + pendingFrameRetry: null, pairedDeviceId: stream.pairedDeviceId, delivery: INITIAL_SCREENCAST_SUBSCRIBER_DELIVERY }) @@ -180,17 +182,16 @@ export class RuntimeBrowserCommandsWithBrowserScreencast extends RuntimeBrowserC } return { subscriptionId, + // Why: the frame the pre-ready gate refused goes out with ready, not a retry tick later. flushPendingFrame: () => { const subscriber = active.subscribers.get(subscriptionId) const bytes = subscriber?.pendingFrame if (!subscriber || !bytes) { return } - const delivered = sendRemoteBrowserScreencastFrame(subscriber.sendBinary, bytes) - subscriber.pendingFrame = delivered ? null : bytes // The replay is this subscriber's first chance to reach its socket, so it is also where // an eviction-eligible delivery history starts. - subscriber.delivery = recordScreencastSubscriberSend(subscriber.delivery, delivered) + deliverScreencastSubscriberFrame(subscriber, bytes) }, session: { done: subscriberDone, diff --git a/src/main/runtime/runtime-browser-screencast-controller.ts b/src/main/runtime/runtime-browser-screencast-controller.ts index 7e3eec68f8f..05c265cf9dd 100644 --- a/src/main/runtime/runtime-browser-screencast-controller.ts +++ b/src/main/runtime/runtime-browser-screencast-controller.ts @@ -6,6 +6,7 @@ import { type BrowserScreencastSubscriber } from './browser-screencast-driver-scope' import type { RuntimeBrowserCommands } from './orca-runtime-browser' +import type { RpcBinarySender } from './rpc/rpc-binary-sender' type RuntimeBrowserScreencastControllerDeps = { getCommands: () => RuntimeBrowserCommands @@ -61,7 +62,7 @@ export class RuntimeBrowserScreencastController { connectionId?: string pairedDeviceId?: string clientKind?: 'mobile' | 'runtime' - sendBinary?: (bytes: Uint8Array) => boolean | void + sendBinary?: RpcBinarySender signal?: AbortSignal emit: (result: BrowserScreencastResult) => void } @@ -114,11 +115,11 @@ export class RuntimeBrowserScreencastController { end(emitEnd) } const abortScreencast = (): void => cancel() - const sendBinaryAfterReady = (bytes: Uint8Array): boolean | void => { + const sendBinaryAfterReady: RpcBinarySender = (bytes, sendOptions) => { if (!readyEmitted) { return false } - return options.sendBinary?.(bytes) + return options.sendBinary?.(bytes, sendOptions) } // Why: rotation can happen before ready, so replacements are connection-scoped immediately. diff --git a/src/main/runtime/runtime-edge-command-controller.ts b/src/main/runtime/runtime-edge-command-controller.ts index 12b5cbbec20..031af2452af 100644 --- a/src/main/runtime/runtime-edge-command-controller.ts +++ b/src/main/runtime/runtime-edge-command-controller.ts @@ -4,6 +4,7 @@ import { RuntimeEmulatorCommands } from './orca-runtime-emulator' import { RuntimeBrowserScreencastController } from './runtime-browser-screencast-controller' import { createRuntimeBrowserCommands } from './runtime-browser-commands-factory' import { RuntimeJiraCommands } from './runtime-jira-commands' +import type { RpcBinarySender } from './rpc/rpc-binary-sender' type PublicMethods = Pick type BrowserSurface = Omit, 'browserScreencast'> & { @@ -13,7 +14,7 @@ type BrowserSurface = Omit, 'browserScreen connectionId?: string pairedDeviceId?: string clientKind?: 'mobile' | 'runtime' - sendBinary?: (bytes: Uint8Array) => boolean | void + sendBinary?: RpcBinarySender signal?: AbortSignal emit: (result: BrowserScreencastResult) => void } diff --git a/src/main/runtime/runtime-rpc/runtime-rpc-websocket-dispatch.ts b/src/main/runtime/runtime-rpc/runtime-rpc-websocket-dispatch.ts index e761255222c..c3809ab190a 100644 --- a/src/main/runtime/runtime-rpc/runtime-rpc-websocket-dispatch.ts +++ b/src/main/runtime/runtime-rpc/runtime-rpc-websocket-dispatch.ts @@ -10,6 +10,7 @@ import type { WebSocketTransport } from '../rpc/ws-transport' import type { DeviceScope } from '../device-registry' import { RuntimeRpcMobileDevices } from './runtime-rpc-mobile-devices' import { classifyRuntimeLongPoll } from './runtime-rpc-long-poll' +import type { RpcBinarySender } from '../rpc/rpc-binary-sender' import type { RpcCallerScope } from '../rpc/rpc-caller-scope' import { limitRuntimeRpcReplySize } from './runtime-rpc-reply-size-limit' @@ -32,7 +33,7 @@ export class RuntimeRpcWebSocketDispatch extends RuntimeRpcMobileDevices { protected async handleWebSocketMessage( rawMessage: string, reply: (response: string) => void, - sendBinary: (response: Uint8Array) => boolean | void, + sendBinary: RpcBinarySender, wsTransport?: WebSocketTransport, ws?: WebSocket, authenticatedDeviceToken?: string | null, diff --git a/src/main/runtime/single-page-browser-commands-host-test-double.ts b/src/main/runtime/single-page-browser-commands-host-test-double.ts index 44153c2ad46..b9a63cf7f77 100644 --- a/src/main/runtime/single-page-browser-commands-host-test-double.ts +++ b/src/main/runtime/single-page-browser-commands-host-test-double.ts @@ -9,6 +9,15 @@ export type AuthoritativeWindowDouble = { webContents: RendererPublicationThrottleTarget } +// A guest the page stream keeps painting while it streams. +export function createStreamedGuest(): RendererPublicationThrottleTarget { + return { + isDestroyed: () => false, + setBackgroundThrottling: vi.fn(), + capturePage: vi.fn(async () => null) + } +} + /** A browser command host whose worktree `wt-1` has one registered page, `page-1`. */ export function createSinglePageBrowserCommandsHost( window: AuthoritativeWindowDouble | null = null diff --git a/src/shared/ws-outbound-backpressure-queue.ts b/src/shared/ws-outbound-backpressure-queue.ts index 0c0667e307f..059ce507861 100644 --- a/src/shared/ws-outbound-backpressure-queue.ts +++ b/src/shared/ws-outbound-backpressure-queue.ts @@ -62,6 +62,8 @@ export type WsOutboundBackpressureQueue = { enqueue: (frame: TFrame) => boolean /** Queue-or-send a frame and allow its owner to cancel it before wire delivery. */ enqueueCancelable: (frame: TFrame) => WsOutboundEnqueueResult + /** True when nothing is parked and the wire has room, so enqueue would send this size directly. */ + isIdle: (frameBytes: number) => boolean /** Bytes currently held (not yet handed to the wire). */ queuedBytes: () => number evidence: () => { queuedBytes: number; queuedFrames: number; storageSlots: number } @@ -257,6 +259,12 @@ export function createWsOutboundBackpressureQueue( } } + const isIdle = (bytes: number): boolean => + queuedFrames === 0 && + options.isWritable() && + bufferedAmount() <= softCapBytes && + (options.canSend?.(bytes, false) ?? true) + const enqueueCancelable = (frame: TFrame): WsOutboundEnqueueResult => { if (disposed || overflowed) { return { accepted: false, queued: false, cancel: () => false } @@ -267,12 +275,7 @@ export function createWsOutboundBackpressureQueue( return { accepted: false, queued: false, cancel: () => false } } // Fast path: nothing parked and the wire is under the cap — send directly. - if ( - queuedFrames === 0 && - options.isWritable() && - bufferedAmount() <= softCapBytes && - (options.canSend?.(bytes, false) ?? true) - ) { + if (isIdle(bytes)) { return { accepted: sendFrame(frame), queued: false, @@ -312,6 +315,7 @@ export function createWsOutboundBackpressureQueue( return enqueueCancelable(frame).accepted }, enqueueCancelable, + isIdle, queuedBytes: () => queued, evidence: () => ({ queuedBytes: queued,