268 lines
10 KiB
TypeScript
268 lines
10 KiB
TypeScript
/**
|
|
* ConnectionController: stream pumping into sinks, the strict readiness
|
|
* handshake (describe + both streams' onOpen, timeout-guarded), generation
|
|
* abort on loss, backoff reconnection, state transitions, and sink-exception
|
|
* isolation. Real (short) timers — the timeout and backoff are configurable,
|
|
* so tests run them at millisecond scale.
|
|
*/
|
|
|
|
import { describe, expect, it, vi } from 'vitest'
|
|
import type { SessionId } from '../src/client/api.ts'
|
|
import type { ConnectionState } from '../src/client/connection.ts'
|
|
import { ConnectionController } from '../src/client/connection.ts'
|
|
import { FakeApiClient, deferred, ok } from './fake-api.ts'
|
|
|
|
const SID = 'fk-c1' as SessionId
|
|
const FAST = { backoffBaseMs: 10, backoffFactor: 1, backoffMaxMs: 10, streamOpenTimeoutMs: 500 }
|
|
|
|
function subscribedFrame(lastSeq = 0) {
|
|
return { type: 'session/subscribed', sessionId: SID, lastSeq } as const
|
|
}
|
|
|
|
describe('connection lifecycle', () => {
|
|
it('announces connected after describe + both streams open, then pumps frames to sinks', async () => {
|
|
const api = new FakeApiClient()
|
|
const muxSeen: string[] = []
|
|
let connected = 0
|
|
const controller = new ConnectionController(api, {
|
|
onMuxEnvelope: envelope => muxSeen.push(envelope.payload.type),
|
|
onConnected: () => { connected++ },
|
|
}, FAST)
|
|
controller.start()
|
|
try {
|
|
await vi.waitFor(() => { expect(connected).toBe(1) })
|
|
api.pushMux(subscribedFrame())
|
|
await vi.waitFor(() => { expect(muxSeen).toEqual(['session/subscribed']) })
|
|
expect(api.callsOf('host.describe')).toHaveLength(1)
|
|
} finally {
|
|
controller.stop()
|
|
}
|
|
})
|
|
|
|
it('reconnects with a fresh generation when a stream fails, and stop() ends the loop', async () => {
|
|
const api = new FakeApiClient()
|
|
let connected = 0
|
|
const warnSpy = vi.spyOn(console, 'warn').mockImplementation(() => undefined)
|
|
const controller = new ConnectionController(api, { onConnected: () => { connected++ } }, FAST)
|
|
controller.start()
|
|
try {
|
|
await vi.waitFor(() => { expect(connected).toBe(1) })
|
|
api.failStreams(new Error('stream torn'))
|
|
await vi.waitFor(() => { expect(connected).toBe(2) }) // new generation after backoff
|
|
expect(api.openMuxCount).toBe(1) // the dead generation's stream is gone, exactly one live
|
|
} finally {
|
|
controller.stop()
|
|
warnSpy.mockRestore()
|
|
}
|
|
// stop() aborts the live generation (streams tear down) and no reconnect follows.
|
|
await vi.waitFor(() => { expect(api.openMuxCount).toBe(0) })
|
|
await new Promise(resolve => setTimeout(resolve, 40))
|
|
expect(api.openMuxCount).toBe(0)
|
|
})
|
|
|
|
it('treats describe failure as generation failure and retries', async () => {
|
|
const api = new FakeApiClient()
|
|
const gate = deferred<Awaited<ReturnType<FakeApiClient['onDescribe']>>>()
|
|
let describeCalls = 0
|
|
api.onDescribe = () => {
|
|
describeCalls++
|
|
return describeCalls === 1 ? Promise.reject(new Error('host down')) : gate.promise
|
|
}
|
|
let connected = 0
|
|
const warnSpy = vi.spyOn(console, 'warn').mockImplementation(() => undefined)
|
|
const controller = new ConnectionController(api, { onConnected: () => { connected++ } }, FAST)
|
|
controller.start()
|
|
try {
|
|
await vi.waitFor(() => { expect(describeCalls).toBe(2) }) // retried after backoff
|
|
expect(connected).toBe(0) // never announced during the failed generation
|
|
gate.resolve(ok({ version: '0', cwd: '/f', attachedSessions: 0 }))
|
|
await vi.waitFor(() => { expect(connected).toBe(1) })
|
|
} finally {
|
|
controller.stop()
|
|
warnSpy.mockRestore()
|
|
}
|
|
})
|
|
|
|
it('treats a host.describe business error as generation failure', async () => {
|
|
const api = new FakeApiClient()
|
|
let describeCalls = 0
|
|
api.onDescribe = () => {
|
|
describeCalls += 1
|
|
if (describeCalls === 1) {
|
|
return Promise.resolve({
|
|
rpcId: 'bad-describe' as never,
|
|
result: {
|
|
ok: false as const,
|
|
error: { code: 'internal' as const, message: 'not ready', details: {} },
|
|
},
|
|
})
|
|
}
|
|
return Promise.resolve(ok({ version: '0', cwd: '/f', attachedSessions: 0 }))
|
|
}
|
|
let connected = 0
|
|
const warnSpy = vi.spyOn(console, 'warn').mockImplementation(() => undefined)
|
|
const controller = new ConnectionController(api, { onConnected: () => { connected++ } }, FAST)
|
|
controller.start()
|
|
try {
|
|
await vi.waitFor(() => { expect(describeCalls).toBe(2) })
|
|
await vi.waitFor(() => { expect(connected).toBe(1) })
|
|
} finally {
|
|
controller.stop()
|
|
warnSpy.mockRestore()
|
|
}
|
|
})
|
|
|
|
it('converges stream/error frames into reconnect instead of dispatching them', async () => {
|
|
const api = new FakeApiClient()
|
|
const muxSeen: string[] = []
|
|
let connected = 0
|
|
const warnSpy = vi.spyOn(console, 'warn').mockImplementation(() => undefined)
|
|
const controller = new ConnectionController(api, {
|
|
onMuxEnvelope: envelope => muxSeen.push(envelope.payload.type),
|
|
onConnected: () => { connected++ },
|
|
}, FAST)
|
|
controller.start()
|
|
try {
|
|
await vi.waitFor(() => { expect(connected).toBe(1) })
|
|
api.pushMux({ type: 'stream/error', error: { code: 'internal', message: 'impl broke', details: {} } })
|
|
await vi.waitFor(() => { expect(connected).toBe(2) }) // treated as loss → reconnect
|
|
expect(muxSeen).toEqual([]) // never forwarded to the business sink
|
|
} finally {
|
|
controller.stop()
|
|
warnSpy.mockRestore()
|
|
}
|
|
})
|
|
|
|
it('isolates sink exceptions from the pump', async () => {
|
|
const api = new FakeApiClient()
|
|
const seen: string[] = []
|
|
let connected = 0
|
|
const errorSpy = vi.spyOn(console, 'error').mockImplementation(() => undefined)
|
|
const controller = new ConnectionController(api, {
|
|
onMuxEnvelope: (envelope) => {
|
|
seen.push(envelope.payload.type)
|
|
throw new Error('business layer bug')
|
|
},
|
|
onConnected: () => { connected++ },
|
|
}, FAST)
|
|
controller.start()
|
|
try {
|
|
await vi.waitFor(() => { expect(connected).toBe(1) })
|
|
api.pushMux(subscribedFrame(1))
|
|
api.pushMux(subscribedFrame(2))
|
|
await vi.waitFor(() => { expect(seen).toHaveLength(2) }) // second frame still pumped
|
|
expect(connected).toBe(1) // no reconnect triggered by the sink throw
|
|
} finally {
|
|
controller.stop()
|
|
errorSpy.mockRestore()
|
|
}
|
|
})
|
|
|
|
it('holds onConnected until both streams establish even after describe succeeds', async () => {
|
|
const api = new FakeApiClient()
|
|
api.holdStreamOpen = true // describe resolves immediately; stream establishment is in the case's hand
|
|
let connected = 0
|
|
const controller = new ConnectionController(api, { onConnected: () => { connected++ } }, FAST)
|
|
controller.start()
|
|
try {
|
|
await vi.waitFor(() => { expect(api.callsOf('host.describe')).toHaveLength(1) })
|
|
await new Promise(resolve => setTimeout(resolve, 30))
|
|
expect(connected).toBe(0) // describe alone must not announce
|
|
api.releaseStreamOpens()
|
|
await vi.waitFor(() => { expect(connected).toBe(1) })
|
|
} finally {
|
|
controller.stop()
|
|
}
|
|
})
|
|
|
|
it('proceeds as connected via the timeout guard when a carrier never fires onOpen', async () => {
|
|
const api = new FakeApiClient()
|
|
api.suppressStreamOpen = true // misbehaving carrier: streams open but onOpen never fires
|
|
let connected = 0
|
|
const controller = new ConnectionController(api, { onConnected: () => { connected++ } }, { ...FAST, streamOpenTimeoutMs: 20 })
|
|
controller.start()
|
|
try {
|
|
await vi.waitFor(() => { expect(connected).toBe(1) }) // handshake resolved by the guard, not wedged
|
|
} finally {
|
|
controller.stop()
|
|
}
|
|
})
|
|
|
|
it('emits deduplicated connected/reconnecting state transitions', async () => {
|
|
const api = new FakeApiClient()
|
|
const states: ConnectionState[] = []
|
|
let connected = 0
|
|
const warnSpy = vi.spyOn(console, 'warn').mockImplementation(() => undefined)
|
|
const controller = new ConnectionController(api, {
|
|
onConnected: () => { connected++ },
|
|
onStateChange: state => states.push(state),
|
|
}, FAST)
|
|
controller.start()
|
|
try {
|
|
await vi.waitFor(() => { expect(connected).toBe(1) })
|
|
expect(states).toEqual(['connected'])
|
|
api.failStreams(new Error('torn'))
|
|
await vi.waitFor(() => { expect(connected).toBe(2) })
|
|
expect(states).toEqual(['connected', 'reconnecting', 'connected'])
|
|
} finally {
|
|
controller.stop()
|
|
warnSpy.mockRestore()
|
|
}
|
|
})
|
|
|
|
it('deduplicates consecutive reconnecting emissions across two straight failures', async () => {
|
|
const api = new FakeApiClient()
|
|
const gate = deferred<Awaited<ReturnType<FakeApiClient['onDescribe']>>>()
|
|
let describeCalls = 0
|
|
api.onDescribe = () => {
|
|
describeCalls++
|
|
return describeCalls <= 2 ? Promise.reject(new Error('down')) : gate.promise
|
|
}
|
|
const states: ConnectionState[] = []
|
|
let connected = 0
|
|
const warnSpy = vi.spyOn(console, 'warn').mockImplementation(() => undefined)
|
|
const controller = new ConnectionController(api, {
|
|
onConnected: () => { connected++ },
|
|
onStateChange: state => states.push(state),
|
|
}, FAST)
|
|
controller.start()
|
|
try {
|
|
await vi.waitFor(() => { expect(describeCalls).toBe(3) })
|
|
gate.resolve(ok({ version: '0', cwd: '/f', attachedSessions: 0 }))
|
|
await vi.waitFor(() => { expect(connected).toBe(1) })
|
|
expect(states).toEqual(['reconnecting', 'connected']) // two failures, one reconnecting emission
|
|
} finally {
|
|
controller.stop()
|
|
warnSpy.mockRestore()
|
|
}
|
|
})
|
|
|
|
it('runs with no sinks at all (every callback slot optional)', async () => {
|
|
const api = new FakeApiClient()
|
|
const controller = new ConnectionController(api, {}, FAST)
|
|
controller.start()
|
|
try {
|
|
await vi.waitFor(() => { expect(api.callsOf('host.describe')).toHaveLength(1) })
|
|
api.pushMux(subscribedFrame()) // pumped with sink undefined: dropped silently
|
|
await new Promise(resolve => setTimeout(resolve, 20))
|
|
} finally {
|
|
controller.stop()
|
|
}
|
|
})
|
|
|
|
it('start() is idempotent (one loop, one stream set)', async () => {
|
|
const api = new FakeApiClient()
|
|
let connected = 0
|
|
const controller = new ConnectionController(api, { onConnected: () => { connected++ } }, FAST)
|
|
controller.start()
|
|
controller.start()
|
|
try {
|
|
await vi.waitFor(() => { expect(connected).toBe(1) })
|
|
expect(api.openMuxCount).toBe(1)
|
|
expect(api.callsOf('host.describe')).toHaveLength(1)
|
|
} finally {
|
|
controller.stop()
|
|
}
|
|
})
|
|
})
|