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,