mirror of
https://github.com/stablyai/orca.git
synced 2026-10-19 08:03:13 +00:00
fix(relay): detach retired mobile transports
This commit is contained in:
@@ -42,6 +42,7 @@ export class RelayControlOrigin {
|
||||
private leaseExpiresAt = 0
|
||||
private acceptingConnections = true
|
||||
private closed = false
|
||||
private readonly detachMobileSocketTransport: () => void
|
||||
|
||||
constructor(options: RelayControlOriginOptions) {
|
||||
this.options = options
|
||||
@@ -53,8 +54,9 @@ export class RelayControlOrigin {
|
||||
createSocket: options.createDataSocket,
|
||||
onConnectionClosed: (connectionId) => options.onConnectionReleased(connectionId, this)
|
||||
})
|
||||
options.mobileSocketWiring.attachTransport(this.transport, (ws) =>
|
||||
this.transport.metadataFor(ws)
|
||||
this.detachMobileSocketTransport = options.mobileSocketWiring.attachTransport(
|
||||
this.transport,
|
||||
(ws) => this.transport.metadataFor(ws)
|
||||
)
|
||||
}
|
||||
|
||||
@@ -152,7 +154,12 @@ export class RelayControlOrigin {
|
||||
}
|
||||
this.controls.clear()
|
||||
this.activeControl = null
|
||||
await this.transport.stop()
|
||||
try {
|
||||
await this.transport.stop()
|
||||
} finally {
|
||||
// Why: detaching earlier would skip socket-close cleanup in MobileSocketWiring.
|
||||
this.detachMobileSocketTransport()
|
||||
}
|
||||
}
|
||||
|
||||
closeNow(): void {
|
||||
|
||||
@@ -115,6 +115,7 @@ describe('RelaySessionBroker lifecycle ownership', () => {
|
||||
let current = true
|
||||
const statuses: string[] = []
|
||||
const keypair = nacl.box.keyPair()
|
||||
const detachTransport = vi.fn()
|
||||
const connecting = RelaySessionBroker.connect({
|
||||
authConfig: {
|
||||
relayTokenEndpoint: 'https://auth.example.test/v1/relay-token',
|
||||
@@ -127,12 +128,14 @@ describe('RelaySessionBroker lifecycle ownership', () => {
|
||||
publicKeyB64: Buffer.from(keypair.publicKey).toString('base64')
|
||||
},
|
||||
appVersion: '1.0.0',
|
||||
mobileSocketWiring: { attachTransport: vi.fn() } as never,
|
||||
mobileSocketWiring: { attachTransport: vi.fn(() => detachTransport) } as never,
|
||||
isCurrent: () => current,
|
||||
refreshAccessToken: async () => null,
|
||||
onStatus: (status) => statuses.push(status)
|
||||
})
|
||||
await vi.waitFor(() => expect(fakes.controls).toHaveLength(1))
|
||||
const transportStopped = deferred<void>()
|
||||
fakes.transports[0]!.stop.mockReturnValue(transportStopped.promise)
|
||||
current = false
|
||||
controlAck.resolve({
|
||||
type: 'host-hello-ack',
|
||||
@@ -147,6 +150,9 @@ describe('RelaySessionBroker lifecycle ownership', () => {
|
||||
await expect(connecting).rejects.toBeInstanceOf(StaleRelayBrokerError)
|
||||
expect(fakes.controls[0]!.closeNow).toHaveBeenCalledOnce()
|
||||
expect(fakes.transports[0]!.stop).toHaveBeenCalledOnce()
|
||||
expect(detachTransport).not.toHaveBeenCalled()
|
||||
transportStopped.resolve(undefined)
|
||||
await vi.waitFor(() => expect(detachTransport).toHaveBeenCalledOnce())
|
||||
expect(statuses).toEqual(['connecting'])
|
||||
})
|
||||
|
||||
@@ -316,7 +322,7 @@ function brokerOptions(
|
||||
publicKeyB64: Buffer.from(keypair.publicKey).toString('base64')
|
||||
},
|
||||
appVersion: '1.0.0',
|
||||
mobileSocketWiring: { attachTransport: vi.fn() } as never,
|
||||
mobileSocketWiring: { attachTransport: vi.fn(() => () => {}) } as never,
|
||||
isCurrent: () => true,
|
||||
refreshAccessToken: async () => null,
|
||||
onStatus: vi.fn(),
|
||||
|
||||
@@ -84,12 +84,19 @@ describe('MobileSocketWiring', () => {
|
||||
onBinary: vi.fn(),
|
||||
onClose: vi.fn()
|
||||
})
|
||||
wiring.attachTransport(direct)
|
||||
const detachDirect = wiring.attachTransport(direct)
|
||||
wiring.attachTransport(relay)
|
||||
|
||||
expect(wiring.terminateDeviceConnections('valid-token')).toBe(3)
|
||||
expect(direct.terminateClientConnections).toHaveBeenCalledWith('valid-token')
|
||||
expect(relay.terminateClientConnections).toHaveBeenCalledWith('valid-token')
|
||||
|
||||
detachDirect()
|
||||
direct.terminateClientConnections.mockClear()
|
||||
relay.terminateClientConnections.mockClear()
|
||||
expect(wiring.terminateDeviceConnections('valid-token')).toBe(2)
|
||||
expect(direct.terminateClientConnections).not.toHaveBeenCalled()
|
||||
expect(relay.terminateClientConnections).toHaveBeenCalledWith('valid-token')
|
||||
})
|
||||
|
||||
it('preserves the legacy direct handshake, identity, and close cleanup', () => {
|
||||
|
||||
@@ -90,12 +90,15 @@ export class MobileSocketWiring {
|
||||
getMetadata: (ws: WebSocket) => MobileSocketTransportMetadata = () => ({
|
||||
transport: 'direct'
|
||||
})
|
||||
): void {
|
||||
): () => void {
|
||||
this.transports.add(transport)
|
||||
transport.onMessage((message, _reply, ws) => {
|
||||
this.handleRawMessage(transport, ws, message, getMetadata(ws))
|
||||
})
|
||||
transport.onConnectionClose((_clientId, ws) => this.handleClose(ws))
|
||||
return () => {
|
||||
this.transports.delete(transport)
|
||||
}
|
||||
}
|
||||
|
||||
getConnectionId(ws: WebSocket): string | undefined {
|
||||
|
||||
Reference in New Issue
Block a user