mirror of
https://github.com/stablyai/orca.git
synced 2026-10-19 08:03:13 +00:00
fix(relay): fence detached socket callbacks
This commit is contained in:
@@ -109,7 +109,11 @@ describe('CloudRelayTransport', () => {
|
||||
)
|
||||
}
|
||||
const emit = (event: string, ...args: unknown[]): void => {
|
||||
for (const fn of listeners.get(event) ?? []) {
|
||||
const eventListeners = listeners.get(event) ?? []
|
||||
if (event === 'error' && eventListeners.length === 0) {
|
||||
throw args[0]
|
||||
}
|
||||
for (const fn of eventListeners) {
|
||||
fn(...args)
|
||||
}
|
||||
}
|
||||
@@ -122,7 +126,7 @@ describe('CloudRelayTransport', () => {
|
||||
on: addListener,
|
||||
once: addListener,
|
||||
off: removeListener,
|
||||
send: () => {},
|
||||
send: vi.fn(),
|
||||
terminate: () => {}
|
||||
}
|
||||
const onConnectionClosed = vi.fn()
|
||||
@@ -133,6 +137,13 @@ describe('CloudRelayTransport', () => {
|
||||
createSocket: () => fakeSocket as unknown as WebSocketClient,
|
||||
onConnectionClosed
|
||||
})
|
||||
let reply: ((response: string) => void) | null = null
|
||||
const onMessage = vi.fn(
|
||||
(_message: string | Uint8Array<ArrayBufferLike>, respond: (response: string) => void) => {
|
||||
reply = respond
|
||||
}
|
||||
)
|
||||
transport.onMessage(onMessage)
|
||||
const opening = transport.openConnection({
|
||||
connId: 'conn-1',
|
||||
connTicket: 'ticket-1',
|
||||
@@ -142,11 +153,16 @@ describe('CloudRelayTransport', () => {
|
||||
})
|
||||
emit('open')
|
||||
await opening
|
||||
vi.mocked(fakeSocket.send).mockClear()
|
||||
emit('message', 'before-stop', false)
|
||||
expect(onMessage).toHaveBeenCalledOnce()
|
||||
|
||||
let stopped = false
|
||||
const stopPromise = transport.stop().then(() => {
|
||||
stopped = true
|
||||
})
|
||||
emit('message', 'during-stop', false)
|
||||
expect(onMessage).toHaveBeenCalledOnce()
|
||||
await vi.advanceTimersByTimeAsync(4_999)
|
||||
expect(stopped).toBe(false)
|
||||
await vi.advanceTimersByTimeAsync(1)
|
||||
@@ -165,6 +181,14 @@ describe('CloudRelayTransport', () => {
|
||||
// late frames must not reach wiring after their metadata was released.
|
||||
expect(() => emit('message', 'late-after-stop', false)).not.toThrow()
|
||||
expect(onLateMessage).not.toHaveBeenCalled()
|
||||
expect(reply).not.toBeNull()
|
||||
await transport.start()
|
||||
reply!('late-reply')
|
||||
expect(fakeSocket.send).not.toHaveBeenCalled()
|
||||
expect(() => emit('error', new Error('late socket failure'))).not.toThrow()
|
||||
emit('close')
|
||||
expect(listeners.get('error')).toHaveLength(0)
|
||||
expect(listeners.get('close')).toHaveLength(0)
|
||||
expect(() => transport.setGeneration(2)).not.toThrow()
|
||||
} finally {
|
||||
vi.useRealTimers()
|
||||
|
||||
@@ -163,6 +163,9 @@ export class CloudRelayTransport implements RpcTransport, MobileSocketTransport
|
||||
this.finalizeConnection(connection.connId, socket)
|
||||
}
|
||||
const onMessage = (raw: RawData, isBinary: boolean): void => {
|
||||
if (this.stopped || finalized) {
|
||||
return
|
||||
}
|
||||
if (!attached) {
|
||||
attached = true
|
||||
clearTimeout(deadline)
|
||||
@@ -173,7 +176,7 @@ export class CloudRelayTransport implements RpcTransport, MobileSocketTransport
|
||||
this.messageHandler?.(
|
||||
message,
|
||||
(response) => {
|
||||
if (socket.readyState === socket.OPEN) {
|
||||
if (!this.stopped && !finalized && socket.readyState === socket.OPEN) {
|
||||
socket.send(response)
|
||||
}
|
||||
},
|
||||
@@ -203,6 +206,7 @@ export class CloudRelayTransport implements RpcTransport, MobileSocketTransport
|
||||
}
|
||||
}
|
||||
this.detachListenersBySocket.set(socket, () => {
|
||||
finalized = true
|
||||
socket.off('message', onMessage)
|
||||
socket.off('open', onOpen)
|
||||
socket.off('error', onError)
|
||||
@@ -236,6 +240,7 @@ export class CloudRelayTransport implements RpcTransport, MobileSocketTransport
|
||||
if (connectionId) {
|
||||
this.finalizeConnection(connectionId, socket)
|
||||
}
|
||||
this.quarantineDetachedSocket(socket)
|
||||
resolve()
|
||||
}, RELAY_SOCKET_CLOSE_TIMEOUT_MS)
|
||||
socket.once('close', onClose)
|
||||
@@ -250,6 +255,20 @@ export class CloudRelayTransport implements RpcTransport, MobileSocketTransport
|
||||
return metadata?.transport === 'relay' ? metadata.basisConnId : undefined
|
||||
}
|
||||
|
||||
private quarantineDetachedSocket(socket: WebSocket): void {
|
||||
if (socket.readyState === socket.CLOSED) {
|
||||
return
|
||||
}
|
||||
// Why: forced cleanup can precede ws's terminal error/close event.
|
||||
const swallowLateError = (): void => {}
|
||||
const clearQuarantine = (): void => {
|
||||
socket.off('error', swallowLateError)
|
||||
socket.off('close', clearQuarantine)
|
||||
}
|
||||
socket.on('error', swallowLateError)
|
||||
socket.once('close', clearQuarantine)
|
||||
}
|
||||
|
||||
private finalizeConnection(connectionId: string, socket: WebSocket): void {
|
||||
if (this.socketsByConnectionId.get(connectionId) !== socket) {
|
||||
return
|
||||
|
||||
Reference in New Issue
Block a user