kevchess15/deepseek-harness
1
1
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions 1

refactor(connection): centralize websocket recovery

Browse source
This commit is contained in:
imccyu 2026-08-29 04:16:57 +08:00
parent f27021c9b6
commit ccfbbb443a
7 changed files with 247 additions and 95 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

4
packages/api/gateway/src/client/index.ts
View file

@ -164,8 +164,12 @@ class ClientRemoteService extends Service implements ClientRemote {
let loop: ReturnType<ConnectionHandle['start']> | undefined
const start = (): void => {
if (disposed) return
if (connection.rpc.open === undefined) this.streams.start()
loop = connection.start({
onConnected: () => { this.ownerCtx.emit('connection/reset') },
onReconnectRequested: () => {
if (connection.rpc.open === undefined) this.streams.reconnect()
},
})
}
const loader = ctx.get('loader') as LoaderReadiness | undefined

8
packages/api/gateway/src/client/remote-stream.ts
View file

@ -31,10 +31,10 @@ export interface RemoteStreamOptions<Item> {
/**
* Reopens one logical Remote stream across carrier generations.
*
* The Gateway owns physical retry timing, cancellation, and replacement. The
* domain consumer owns its opening item and every later item, and calls
* {@link RemoteStreamItem.accept} only after validating the opening
* baseline or cursor.
* Connection owns physical retry timing; Gateway performs each requested
* replacement. The domain consumer owns its opening item and every later
* item, and calls {@link RemoteStreamItem.accept} only after validating the
* opening baseline or cursor.
*/
export class RemoteStream<Item> implements AsyncIterable<RemoteStreamItem<Item>> {
private readonly lifetime = new AbortController()

111
packages/api/gateway/src/client/stream-client.ts
View file

@ -10,11 +10,7 @@ import {
import { randomUUID } from '@deepseek-ai/dsh-util-crypto'
const INTERNAL_BASE = 'http://dsh.internal'
const RECONNECT_BASE_MS = 500
const RECONNECT_FACTOR = 2
const RECONNECT_MAX_MS = 10_000
/** One Host-reported Remote stream failure. */
/** Physical Remote stream socket failure that may be retried by a domain transport. */
export class RemoteStreamCarrierError extends Error {
/**
@ -28,6 +24,7 @@ export class RemoteStreamCarrierError extends Error {
}
interface SocketWaiter {
readonly revision: number
resolve(socket: WebSocket): void
reject(error: unknown): void
}
@ -37,21 +34,43 @@ export class RemoteStreamMuxClient {
private socket: WebSocket | undefined
private cancelCandidate: ((error: Error) => void) | undefined
private keepAlive: Promise<void> | undefined
private keepAliveAbort: AbortController | undefined
private revision = 0
private readonly streams = new Map<string, StreamInbox>()
private readonly waiters = new Set<SocketWaiter>()
private running = false
private disposed = false
/** Start the persistent physical connection; repeated calls are inert. */
/** Ensure a physical attempt exists, following the current attempt once if needed. */
start(): void {
if (this.running || this.disposed) return
if (this.disposed) return
this.running = true
this.maintain()
if (this.socket?.readyState === WebSocket.OPEN) return
const pending = this.keepAlive
if (pending === undefined) this.maintain()
else void pending.then(() => { this.maintain() })
}
/** Cancel the current socket or retry wait and start a fresh attempt immediately. */
reconnect(): void {
if (!this.running || this.disposed) return
const failure = new RemoteStreamCarrierError('api gateway: Remote stream reconnect requested')
const pending = this.keepAlive
this.revision++
this.cancelCandidate?.(failure)
const socket = this.socket
if (socket !== undefined) {
this.socket = undefined
this.failAll(failure)
socket.close(4000, 'reconnect requested')
}
if (pending === undefined) this.maintain()
else void pending.then(() => { this.maintain() })
}
/**
* Open one logical stream on the persistent physical connection.
* If no physical attempt is active, opening waits for Connection to request
* one or for the signal to abort.
* @param endpoint - Typert Remote stream endpoint.
* @param payload - endpoint request encoded on the wire.
* @param signal - cancellation for this logical stream.
@ -62,7 +81,6 @@ export class RemoteStreamMuxClient {
payload: unknown,
signal: AbortSignal,
): AsyncGenerator {
this.start()
signal.throwIfAborted()
const streamId = randomUUID()
const inbox = new StreamInbox()
@ -101,16 +119,15 @@ export class RemoteStreamMuxClient {
}
/**
* Permanently stop reconnecting, close the physical socket, and fail every active logical stream.
* @returns once the background connection loop has stopped.
* Permanently stop the carrier, close the physical socket, and fail every
* active logical stream.
* @returns once the active connection attempt has stopped.
*/
async close(): Promise<void> {
if (!this.disposed) {
this.disposed = true
this.running = false
const error = new Error('api gateway: Remote stream client disposed')
this.keepAliveAbort?.abort(error)
this.keepAliveAbort = undefined
this.failAll(error)
for (const waiter of [...this.waiters]) waiter.reject(error)
this.cancelCandidate?.(error)
@ -176,7 +193,7 @@ export class RemoteStreamMuxClient {
signal.throwIfAborted()
if (this.socket?.readyState === WebSocket.OPEN) return Promise.resolve(this.socket)
if (this.disposed) return Promise.reject(new Error('api gateway: Remote stream client disposed'))
this.start()
if (!this.running) return Promise.reject(new Error('api gateway: Remote stream client not started'))
return new Promise((resolve, reject) => {
const aborted = (): void => { waiter.reject(signal.reason) }
const cleanup = (): void => {
@ -184,6 +201,7 @@ export class RemoteStreamMuxClient {
signal.removeEventListener('abort', aborted)
}
const waiter: SocketWaiter = {
revision: this.revision,
resolve: (socket) => {
cleanup()
resolve(socket)
@ -223,49 +241,27 @@ export class RemoteStreamMuxClient {
if (this.socket !== socket) return
this.socket = undefined
this.failAll(error)
this.maintain(error)
}
private maintain(previousFailure?: Error): void {
if (!this.running) return
if (this.keepAlive !== undefined) {
void this.keepAlive.then(() => { this.maintain(previousFailure) })
return
}
const abort = new AbortController()
this.keepAliveAbort = abort
const task = this.reconnect(abort.signal, previousFailure)
private maintain(): void {
if (!this.running || this.disposed) return
if (this.socket?.readyState === WebSocket.OPEN || this.keepAlive !== undefined) return
const revision = this.revision
const task = this.connect().then(
() => undefined,
(error: unknown) => {
if (!this.running) return
for (const waiter of [...this.waiters]) {
if (waiter.revision <= revision) waiter.reject(error)
}
},
)
this.keepAlive = task
void task.then(() => {
this.keepAlive = undefined
this.keepAliveAbort = undefined
})
}
private async reconnect(signal: AbortSignal, previousFailure?: Error): Promise<void> {
let attempt = 0
let failure = previousFailure
while (this.isRunning(signal) && this.socket?.readyState !== WebSocket.OPEN) {
if (failure !== undefined) {
attempt += 1
console.warn(`[api-gateway] Remote stream connection unavailable, retry #${String(attempt)}`, failure)
await sleep(backoffDelay(attempt), signal)
if (!this.isRunning(signal)) return
}
try {
await this.connect()
return
} catch (error) {
if (!this.isRunning(signal)) return
failure = error as Error
}
}
}
private isRunning(signal: AbortSignal): boolean {
return this.running && !signal.aborted
}
private failAll(error: unknown): void {
for (const stream of this.streams.values()) stream.fail(error)
}
@ -275,23 +271,6 @@ export class RemoteStreamMuxClient {
}
}
function backoffDelay(attempt: number): number {
const cap = Math.min(RECONNECT_MAX_MS, RECONNECT_BASE_MS * RECONNECT_FACTOR ** Math.max(0, attempt - 1))
return cap / 2 + Math.random() * (cap / 2)
}
function sleep(ms: number, signal: AbortSignal): Promise<void> {
return new Promise((resolve) => {
const timer = setTimeout(done, ms)
signal.addEventListener('abort', done, { once: true })
function done(): void {
clearTimeout(timer)
signal.removeEventListener('abort', done)
resolve()
}
})
}
class StreamInbox {
private readonly frames: RemoteStreamServerMessage[] = []
private wake: (() => void) | undefined

4
packages/api/gateway/src/index.ts
View file

@ -112,11 +112,11 @@ interface PendingRemoteEvent {
type ConnectionRpcResult = Awaited<ReturnType<ConnectionRpcHandler>>
type ConnectionRpcError = Extract<ConnectionRpcResult, { readonly ok: false }>['error']
const NEVER_ABORTED_SIGNAL = new AbortController().signal
const DEFAULT_WEBSOCKET_HEARTBEAT_INTERVAL_MS = 30_000
const DEFAULT_WEBSOCKET_HEARTBEAT_INTERVAL_MS = 2_000
/** Gateway transport configuration. */
export interface Config {
/** WebSocket Ping interval from 1 through 2,147,483,647 milliseconds. @default 30000 */
/** WebSocket Ping interval from 1 through 2,147,483,647 milliseconds. @default 2000 */
readonly websocketHeartbeatIntervalMs?: number
}

11
packages/api/gateway/src/stream-server.ts
View file

@ -23,6 +23,7 @@ export type RemoteStreamFailureMapper = (error: unknown) => RemoteStreamFailure
export class RemoteStreamMuxServer {
private readonly server = new WebSocketServer({ noServer: true })
private readonly connections = new Set<Promise<void>>()
private readonly heartbeatAlive = new WeakMap<WebSocket, boolean>()
private heartbeatTimer: NodeJS.Timeout | undefined
/**
@ -44,6 +45,8 @@ export class RemoteStreamMuxServer {
*/
handleUpgrade(req: IncomingMessage, socket: Duplex, head: Buffer): void {
this.server.handleUpgrade(req, socket, head, (websocket) => {
this.heartbeatAlive.set(websocket, true)
websocket.on('pong', () => { this.heartbeatAlive.set(websocket, true) })
this.startHeartbeat()
const connection = new RemoteStreamMuxConnection(websocket, this.open, this.failure)
const done = connection.run()
@ -71,7 +74,13 @@ export class RemoteStreamMuxServer {
if (this.heartbeatTimer !== undefined) return
this.heartbeatTimer = setInterval(() => {
for (const socket of this.server.clients) {
if (socket.readyState === WebSocket.OPEN) socket.ping()
if (socket.readyState !== WebSocket.OPEN) continue
if (this.heartbeatAlive.get(socket) === false) {
socket.terminate()
continue
}
this.heartbeatAlive.set(socket, false)
socket.ping()
}
}, this.heartbeatIntervalMs)
this.heartbeatTimer.unref()

127
packages/client/connection/src/client/connection.ts
View file

@ -12,12 +12,11 @@ export interface ConnectionGeneration {
readonly host: ConnectionHostInfo
}
/** Reconnect/backoff tunables (deployment-varying — no hardcoded tunables; these become the
* future `ctx.connection` plugin's Config). All fields optional; defaults below. */
/** Reconnect/backoff tunables. All fields are optional; defaults are below. */
export interface ConnectionConfig {
/** First-retry backoff cap in ms (jittered: actual delay is cap/2..cap). */
backoffBaseMs?: number
/** Exponential growth factor per consecutive failed attempt. */
/** Exponential growth factor per failed attempt; values at or below 1 make the base tier final. */
backoffFactor?: number
/** Upper bound for the backoff cap in ms. */
backoffMaxMs?: number
@ -32,6 +31,9 @@ const CONNECTION_DEFAULTS: Required<ConnectionConfig> = {
generationReadyTimeoutMs: 3_000,
}
const MANUAL_RECONNECT = new Error('connection: manual reconnect requested')
const NETWORK_STATE_CHANGED = new Error('connection: browser network state changed')
function sleep(ms: number, signal: AbortSignal): Promise<void> {
return new Promise((resolve) => {
const t = setTimeout(done, ms)
@ -44,17 +46,27 @@ function sleep(ms: number, signal: AbortSignal): Promise<void> {
})
}
/** Coarse connection state for the UI: 'connected' after each generation's handshake,
* 'reconnecting' the moment the generation fails (covers the whole backoff+retry span). */
export type ConnectionState = 'connected' | 'reconnecting'
function waitForAbort(signal: AbortSignal): Promise<void> {
if (signal.aborted) return Promise.resolve()
return new Promise((resolve) => {
signal.addEventListener('abort', () => { resolve() }, { once: true })
})
}
/** Connection lifecycle state published after the first attempt has an outcome. */
export type ConnectionState =
| 'connected'
| 'disconnected'
| 'connecting'
/** Connection-generation callbacks owned by API Gateway. */
export interface ConnectionSinks {
/** After the generation source reports ready, first connect included. */
onConnected?: (host: ConnectionHostInfo) => void
/** Coarse state transitions (deduplicated: fires only on change). The initial pre-connect
* span reports nothing — the UI treats "no state yet" as connecting, not as an outage. */
/** State transitions after the initial attempt has an outcome. Equivalent states are deduplicated. */
onStateChange?: (state: ConnectionState) => void
/** Start one fresh physical-carrier attempt before each logical retry. */
onReconnectRequested?: () => void
}
/**
@ -79,8 +91,11 @@ export class ConnectionController {
private generation = 0
private attempt = 0
private current: AbortController | null = null
private retryDelay: AbortController | null = null
private running = false
private lastState: ConnectionState | null = null
private immediateRetry = false
private networkAvailable = true
private lastState: ConnectionState | undefined
private readonly config: Required<ConnectionConfig>
constructor(
@ -103,14 +118,53 @@ export class ConnectionController {
this.running = false
this.current?.abort()
this.current = null
this.retryDelay?.abort()
this.retryDelay = null
}
/** Reset the retry sequence and replace the current generation or retry delay immediately. */
reconnect(): void {
if (!this.running) return
this.attempt = 0
this.immediateRetry = true
this.emitState('connecting')
if (!this.isRunning()) return
this.current?.abort(MANUAL_RECONNECT)
this.retryDelay?.abort(MANUAL_RECONNECT)
}
/**
* Suspend automatic retries while offline and restart backoff when the network returns.
* @param available - whether the browser reports network access.
*/
setNetworkAvailable(available: boolean): void {
if (this.networkAvailable === available) return
this.networkAvailable = available
this.attempt = 0
this.immediateRetry = false
if (!this.running) return
this.emitState(available ? 'connecting' : 'disconnected')
if (!this.isRunning()) return
this.current?.abort(NETWORK_STATE_CHANGED)
this.retryDelay?.abort(NETWORK_STATE_CHANGED)
}
private backoffCap(attempt: number): number {
const { backoffBaseMs, backoffFactor, backoffMaxMs } = this.config
return Math.min(backoffMaxMs, backoffBaseMs * backoffFactor ** Math.max(0, attempt - 1))
}
private backoffDelay(attempt: number): number {
const { backoffBaseMs, backoffFactor, backoffMaxMs } = this.config
const cap = Math.min(backoffMaxMs, backoffBaseMs * backoffFactor ** Math.max(0, attempt - 1))
const cap = this.backoffCap(attempt)
return cap / 2 + Math.random() * (cap / 2)
}
private isFinalBackoffTier(attempt: number): boolean {
const cap = this.backoffCap(attempt)
const nextCap = this.backoffCap(attempt + 1)
return cap >= this.config.backoffMaxMs || !Number.isFinite(nextCap) || nextCap <= cap
}
/** Read through a method: stop() flips the flag across awaits, so narrowing from the loop condition must not stick. */
private isRunning(): boolean {
return this.running
@ -122,7 +176,49 @@ export class ConnectionController {
}
private async loop(): Promise<void> {
let retry = false
while (this.running) {
if (!this.networkAvailable && !this.immediateRetry) {
const retryDelay = new AbortController()
this.retryDelay = retryDelay
this.emitState('disconnected')
await waitForAbort(retryDelay.signal)
if (this.retryDelay === retryDelay) this.retryDelay = null
if (!this.isRunning()) return
retry = true
continue
}
let manualAttempt = false
if (retry) {
const immediate = this.immediateRetry
this.immediateRetry = false
if (immediate) this.attempt = 0
manualAttempt = immediate
if (!immediate && this.attempt > 0 && this.isFinalBackoffTier(this.attempt)) {
const retryDelay = new AbortController()
this.retryDelay = retryDelay
this.emitState('disconnected')
await waitForAbort(retryDelay.signal)
if (this.retryDelay === retryDelay) this.retryDelay = null
continue
}
const attempt = ++this.attempt
this.emitState('connecting')
if (!this.isRunning()) return
if (!immediate) {
const retryDelay = new AbortController()
this.retryDelay = retryDelay
await sleep(this.backoffDelay(attempt), retryDelay.signal)
if (this.retryDelay === retryDelay) this.retryDelay = null
if (!this.isRunning()) return
if (retryDelay.signal.aborted) continue
}
console.warn(`[connection] connection lost, retry #${String(attempt)}`)
this.callSink(() => { this.sinks.onReconnectRequested?.() })
if (!this.isRunning()) return
}
const gen = ++this.generation
const ac = new AbortController()
this.current = ac
@ -182,17 +278,14 @@ export class ConnectionController {
this.callSink(() => { this.sinks.onConnected?.(host) })
}
} catch {
// Transport failure: treat as generation failure, fall through to the shared backoff.
// Transport failure: treat as generation failure, then enter the shared retry path.
if (!ac.signal.aborted) ac.abort()
}
await failed
if (!this.isRunning()) return
this.emitState('reconnecting')
this.attempt += 1
console.warn(`[connection] connection lost, retry #${this.attempt}`)
const idle = new AbortController()
await sleep(this.backoffDelay(this.attempt), idle.signal)
if (manualAttempt) this.attempt = 0
retry = true
}
}

77
packages/client/connection/src/client/index.ts
View file

@ -9,6 +9,7 @@ import {
type ConnectionGeneration,
type ConnectionGenerationSource,
type ConnectionSinks,
type ConnectionState,
} from './connection.ts'
import { createFixtureConnectionRpc } from './fixture.ts'
import { createWebConnectionRpc, type RpcFetch, type RpcStreamOpen } from './rpc.ts'
@ -61,6 +62,14 @@ export interface ConnectionGenerationState {
subscribe(listener: () => void): () => void
}
/** Observable recovery lifecycle of the owned Connection loop. */
export interface ConnectionStateSource {
/** Current state, or undefined before the first connection outcome. */
getSnapshot(): ConnectionState | undefined
/** Subscribe to state changes. */
subscribe(listener: () => void): () => void
}
/** Required services (none — this is the wire root). */
export const inject: string[] = []
@ -111,8 +120,12 @@ export interface ConnectionHandle {
readonly isLoopback: boolean
/** Current Remote event generation and the Host facts carried by its opening frame. */
readonly generation: ConnectionGenerationState
/** Current recovery lifecycle for connection-specific consumers. */
readonly state: ConnectionStateSource
/** Generic logical RPC channels over the same Connection transport. */
readonly rpc: ClientConnectionRpc
/** Reset retry progression and replace the current attempt immediately. */
reconnect(): void
/**
* Register the sole source defining Host generations. The source reports
* ready only after its incremental listeners are attached.
@ -124,16 +137,44 @@ export interface ConnectionHandle {
* Start the connect/reconnect loop with the consumer's state callbacks.
* API Gateway owns the loop; a second call throws.
* @param sinks - connection-state callbacks.
* @param config - reconnect/backoff tunables.
* @returns stop handle for the loop.
* @param config - reconnect timing tunables.
* @returns lifecycle controls for the loop.
*/
start(sinks: ConnectionSinks, config?: ConnectionConfig): { stop(): void }
start(sinks: ConnectionSinks, config?: ConnectionConfig): ConnectionLoop
}
/** Controls retained by the sole owner of a running connection loop. */
export interface ConnectionLoop {
/** Stop the loop and withdraw its active generation. */
stop(): void
}
interface ConnectionOwner {
readonly token: object
readonly source: ConnectionGenerationSource
readonly controller: ConnectionController
readonly stopNetworkWatch: () => void
}
interface BrowserNetworkTarget {
readonly navigator?: { readonly onLine?: boolean }
addEventListener(type: 'online' | 'offline', listener: () => void): void
removeEventListener(type: 'online' | 'offline', listener: () => void): void
}
function watchBrowserNetwork(controller: ConnectionController): () => void {
const browser = (globalThis as { readonly window?: BrowserNetworkTarget }).window
const initiallyAvailable = browser?.navigator?.onLine
if (browser === undefined || initiallyAvailable === undefined) return () => {}
const online = (): void => { controller.setNetworkAvailable(true) }
const offline = (): void => { controller.setNetworkAvailable(false) }
controller.setNetworkAvailable(initiallyAvailable)
browser.addEventListener('online', online)
browser.addEventListener('offline', offline)
return () => {
browser.removeEventListener('online', online)
browser.removeEventListener('offline', offline)
}
}
/**
@ -150,7 +191,9 @@ export function apply(ctx: Context): void {
let owner: ConnectionOwner | undefined
let generationId = 0
let generation: ConnectionGeneration | undefined
let state: ConnectionState | undefined
const generationListeners = new Set<() => void>()
const stateListeners = new Set<() => void>()
const publishGeneration = (next: ConnectionGeneration | undefined): void => {
if (Object.is(generation, next)) return
generation = next
@ -162,11 +205,24 @@ export function apply(ctx: Context): void {
}
}
}
const publishState = (next: ConnectionState | undefined): void => {
if (state === next) return
state = next
for (const listener of [...stateListeners]) {
try {
listener()
} catch (error) {
console.error('[connection] state listener threw:', error)
}
}
}
const releaseOwner = (current: ConnectionOwner): void => {
if (owner !== current) return
owner = undefined
current.stopNetworkWatch()
current.controller.stop()
publishGeneration(undefined)
publishState(undefined)
}
const handle: ConnectionHandle = {
isLoopback: transport?.ownsHost === true || pageLocation === undefined || isLoopbackHostname(pageLocation.hostname),
@ -177,7 +233,17 @@ export function apply(ctx: Context): void {
return () => { generationListeners.delete(listener) }
},
},
state: {
getSnapshot: () => state,
subscribe: (listener) => {
stateListeners.add(listener)
return () => { stateListeners.delete(listener) }
},
},
rpc,
reconnect() {
owner?.controller.reconnect()
},
registerGenerationSource(source) {
if (generationSource !== undefined) {
throw new Error('connection: a generation source is already registered')
@ -205,14 +271,15 @@ export function apply(ctx: Context): void {
sinks.onConnected?.(host)
},
onStateChange: (state) => {
if (state === 'reconnecting') {
if (state !== 'connected') {
publishGeneration(undefined)
}
if (!ownsGeneration()) return
publishState(state)
sinks.onStateChange?.(state)
},
}, config ?? {})
const current = { token, source, controller }
const current = { token, source, controller, stopNetworkWatch: watchBrowserNetwork(controller) }
owner = current
controller.start()
return {