mirror of
https://github.com/stablyai/orca.git
synced 2026-09-27 08:02:35 +00:00
fix(mobile): preserve delivery ambiguity across transport cutover
Let physical close settle requests and retain its error as the cutover cause, copying only an existing delivery-unknown mark. Pin sent and unsent caller outcomes and both cutover predicate carriers. Claude-Session: https://claude.ai/code/session_01JNnE9qzUZMMnqpZWCqM3nb
This commit is contained in:
@@ -1,6 +1,11 @@
|
||||
import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'
|
||||
import { connect } from './rpc-client'
|
||||
import { isRpcDeliveryUnknown } from './rpc-delivery-ambiguity'
|
||||
import {
|
||||
createStableLogicalRpcClient,
|
||||
isLogicalClientCutoverError,
|
||||
LogicalClientCutoverError
|
||||
} from './stable-logical-rpc-client'
|
||||
|
||||
vi.mock('./e2ee', () => ({
|
||||
generateKeyPair: () => ({
|
||||
@@ -72,7 +77,7 @@ function hasSentRequest(socket: MockWebSocket, method: string): boolean {
|
||||
|
||||
function connectAuthenticated(): { client: ReturnType<typeof connect>; socket: MockWebSocket } {
|
||||
const client = connect('ws://desktop.invalid', 'token', 'server-key')
|
||||
const socket = mockSockets[0]!
|
||||
const socket = mockSockets[mockSockets.length - 1]!
|
||||
socket.open()
|
||||
socket.receive(JSON.stringify({ type: 'e2ee_ready' }))
|
||||
socket.receive('encrypted:{"type":"e2ee_authenticated"}')
|
||||
@@ -94,6 +99,52 @@ describe('mobile rpc-client delivery ambiguity marking', () => {
|
||||
globalThis.WebSocket = originalWebSocket
|
||||
})
|
||||
|
||||
it.each([true, false])(
|
||||
'preserves physical delivery evidence at the cutover caller (sent=%s)',
|
||||
async (sent) => {
|
||||
const physical = sent
|
||||
? connectAuthenticated()
|
||||
: {
|
||||
client: connect('ws://desktop.invalid', 'token', 'server-key'),
|
||||
socket: mockSockets[0]!
|
||||
}
|
||||
const client = createStableLogicalRpcClient(physical.client, 'lan')
|
||||
const replacement = connectAuthenticated()
|
||||
const requestError = client
|
||||
.sendRequest('worktree.create', { name: 'new' })
|
||||
.catch((error: unknown) => error)
|
||||
await Promise.resolve()
|
||||
expect(hasSentRequest(physical.socket, 'worktree.create')).toBe(sent)
|
||||
|
||||
await client.migrateTo(replacement.client, 'relay')
|
||||
|
||||
const error = await requestError
|
||||
expect(isLogicalClientCutoverError(error)).toBe(true)
|
||||
expect(isRpcDeliveryUnknown(error)).toBe(sent)
|
||||
expect(error).toBeInstanceOf(LogicalClientCutoverError)
|
||||
expect(isRpcDeliveryUnknown((error as Error).cause)).toBe(sent)
|
||||
expect(hasSentRequest(replacement.socket, 'worktree.create')).toBe(false)
|
||||
expect(
|
||||
physical.socket.sent.filter((payload) => payload.includes('worktree.create'))
|
||||
).toHaveLength(sent ? 1 : 0)
|
||||
client.close()
|
||||
}
|
||||
)
|
||||
|
||||
it('recognizes a cutover by class even when its message changes', () => {
|
||||
const error = new LogicalClientCutoverError()
|
||||
error.message = 'wrapped migration'
|
||||
expect(isLogicalClientCutoverError(error)).toBe(true)
|
||||
})
|
||||
|
||||
it('recognizes a cutover message from another bundle copy', () => {
|
||||
expect(isLogicalClientCutoverError(new Error('RPC interrupted by connection migration'))).toBe(
|
||||
true
|
||||
)
|
||||
expect(isLogicalClientCutoverError(new Error('Client closed'))).toBe(false)
|
||||
expect(isLogicalClientCutoverError('RPC interrupted by connection migration')).toBe(false)
|
||||
})
|
||||
|
||||
it('marks in-flight requests as delivery-unknown when the socket drops', async () => {
|
||||
const { client, socket } = connectAuthenticated()
|
||||
const requestError = client.sendRequest('terminal.send', { terminal: 't' }).then(
|
||||
|
||||
@@ -90,6 +90,7 @@ describe('stable logical RPC client', () => {
|
||||
const nextSession = new FakeSession('connecting')
|
||||
const pending = deferred<RpcResponse>()
|
||||
oldSession.sendRequest.mockReturnValue(pending.promise)
|
||||
oldSession.close.mockImplementation(() => pending.reject(new Error('Client closed')))
|
||||
nextSession.sendRequest.mockResolvedValue(success('next'))
|
||||
const client = createStableLogicalRpcClient(oldSession, 'lan')
|
||||
const stream = vi.fn()
|
||||
|
||||
@@ -7,12 +7,16 @@ import {
|
||||
import { waitForAuthenticated } from './replacement-session-authentication'
|
||||
import { projectMobileRpcRequestParams } from './mobile-rpc-request-projection'
|
||||
import { LogicalClientConnectionPath } from './logical-client-connection-path'
|
||||
import { isRpcDeliveryUnknown, markRpcDeliveryUnknown } from './rpc-delivery-ambiguity'
|
||||
|
||||
export type MobileConnectionPath = 'lan' | 'tailscale' | 'relay'
|
||||
|
||||
export class LogicalClientCutoverError extends Error {
|
||||
constructor() {
|
||||
super('RPC interrupted by connection migration')
|
||||
constructor(cause?: unknown) {
|
||||
super('RPC interrupted by connection migration', { cause })
|
||||
if (isRpcDeliveryUnknown(cause)) {
|
||||
markRpcDeliveryUnknown(this)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -33,10 +37,6 @@ type SubscriptionRecord = {
|
||||
cancelled: boolean
|
||||
}
|
||||
|
||||
type PendingRequest = {
|
||||
reject: (error: Error) => void
|
||||
}
|
||||
|
||||
export type StableLogicalRpcClient = RpcClient & {
|
||||
migrateTo(
|
||||
session: RpcClient,
|
||||
@@ -75,7 +75,6 @@ export function createStableLogicalRpcClient(
|
||||
let nextSubscriptionId = 0
|
||||
let activeStateUnsubscribe: (() => void) | null = null
|
||||
const subscriptions = new Map<number, SubscriptionRecord>()
|
||||
const pendingRequests = new Set<PendingRequest>()
|
||||
const stateListeners = new Set<(state: ConnectionState) => void>()
|
||||
let state = initialSession.getState()
|
||||
const connectionPath = new LogicalClientConnectionPath(() => state === 'connected')
|
||||
@@ -93,13 +92,10 @@ export function createStableLogicalRpcClient(
|
||||
const requestGeneration = generation
|
||||
const session = activeSession
|
||||
return new Promise<RpcResponse>((resolve, reject) => {
|
||||
const pending = { reject }
|
||||
pendingRequests.add(pending)
|
||||
void session
|
||||
.sendRequest(method, projectMobileRpcRequestParams(method, params), options)
|
||||
.then(
|
||||
(response) => {
|
||||
pendingRequests.delete(pending)
|
||||
if (closed) {
|
||||
reject(new Error('Client closed'))
|
||||
} else if (requestGeneration !== generation) {
|
||||
@@ -109,8 +105,9 @@ export function createStableLogicalRpcClient(
|
||||
}
|
||||
},
|
||||
(error: unknown) => {
|
||||
pendingRequests.delete(pending)
|
||||
reject(error)
|
||||
reject(
|
||||
requestGeneration !== generation ? new LogicalClientCutoverError(error) : error
|
||||
)
|
||||
}
|
||||
)
|
||||
})
|
||||
@@ -265,15 +262,12 @@ export function createStableLogicalRpcClient(
|
||||
suspended = false
|
||||
previousStateUnsubscribe?.()
|
||||
bindActiveState(nextSession, nextGeneration)
|
||||
for (const pending of pendingRequests) {
|
||||
pending.reject(new LogicalClientCutoverError())
|
||||
}
|
||||
pendingRequests.clear()
|
||||
state = nextSession.getState()
|
||||
connectionPath.clearAfterConnected()
|
||||
for (const listener of stateListeners) {
|
||||
listener(state)
|
||||
}
|
||||
// Only the physical sender knows whether a pending request reached the wire.
|
||||
previous.close()
|
||||
},
|
||||
|
||||
|
||||
@@ -39,7 +39,11 @@ function fakeSession(): FakeSession {
|
||||
getLastConnectedAt: () => null,
|
||||
onStateChange: () => () => {},
|
||||
notifyForeground: () => {},
|
||||
close: () => {}
|
||||
close: () => {
|
||||
for (const settle of pending.splice(0)) {
|
||||
settle(new Error('Client closed'))
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
return fake
|
||||
|
||||
Reference in New Issue
Block a user