diff --git a/packages/api/gateway/src/client/index.ts b/packages/api/gateway/src/client/index.ts index accf7cafb0..853fc4242f 100644 --- a/packages/api/gateway/src/client/index.ts +++ b/packages/api/gateway/src/client/index.ts @@ -164,8 +164,12 @@ class ClientRemoteService extends Service implements ClientRemote { let loop: ReturnType | 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 diff --git a/packages/api/gateway/src/client/remote-stream.ts b/packages/api/gateway/src/client/remote-stream.ts index a2b018ae91..00b398afd7 100644 --- a/packages/api/gateway/src/client/remote-stream.ts +++ b/packages/api/gateway/src/client/remote-stream.ts @@ -31,10 +31,10 @@ export interface RemoteStreamOptions { /** * 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 implements AsyncIterable> { private readonly lifetime = new AbortController() diff --git a/packages/api/gateway/src/client/stream-client.ts b/packages/api/gateway/src/client/stream-client.ts index 9616102cce..79663d249b 100644 --- a/packages/api/gateway/src/client/stream-client.ts +++ b/packages/api/gateway/src/client/stream-client.ts @@ -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 | undefined - private keepAliveAbort: AbortController | undefined + private revision = 0 private readonly streams = new Map() private readonly waiters = new Set() 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 { 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 { - 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 { - 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 diff --git a/packages/api/gateway/src/index.ts b/packages/api/gateway/src/index.ts index 8610dc2881..25650ea2da 100644 --- a/packages/api/gateway/src/index.ts +++ b/packages/api/gateway/src/index.ts @@ -112,11 +112,11 @@ interface PendingRemoteEvent { type ConnectionRpcResult = Awaited> type ConnectionRpcError = Extract['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 } diff --git a/packages/api/gateway/src/stream-server.ts b/packages/api/gateway/src/stream-server.ts index 28a3589542..9f0d5d4cab 100644 --- a/packages/api/gateway/src/stream-server.ts +++ b/packages/api/gateway/src/stream-server.ts @@ -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>() + private readonly heartbeatAlive = new WeakMap() 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() diff --git a/packages/client/connection/src/client/connection.ts b/packages/client/connection/src/client/connection.ts index 17e946c80e..3020c80326 100644 --- a/packages/client/connection/src/client/connection.ts +++ b/packages/client/connection/src/client/connection.ts @@ -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 = { 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 { return new Promise((resolve) => { const t = setTimeout(done, ms) @@ -44,17 +46,27 @@ function sleep(ms: number, signal: AbortSignal): Promise { }) } -/** 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 { + 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 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 { + 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 } } diff --git a/packages/client/connection/src/client/index.ts b/packages/client/connection/src/client/index.ts index 325834b69e..482ef2db91 100644 --- a/packages/client/connection/src/client/index.ts +++ b/packages/client/connection/src/client/index.ts @@ -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 {