diff --git a/src/exchanges/backpack/gateway.ts b/src/exchanges/backpack/gateway.ts index da476ea..13e0d67 100644 --- a/src/exchanges/backpack/gateway.ts +++ b/src/exchanges/backpack/gateway.ts @@ -24,6 +24,7 @@ import type { TickerListener, KlineListener, } from "../adapter"; +import { ReconnectScheduler, fixedBackoff } from "../reconnect-scheduler"; const WebSocketCtor: typeof globalThis.WebSocket = typeof globalThis.WebSocket !== "undefined" @@ -99,7 +100,10 @@ export class BackpackGateway { private ws: WebSocket | null = null; private wsReady = false; private wsPingTimer: ReturnType | null = null; - private wsReconnectTimer: ReturnType | null = null; + private readonly wsReconnect = new ReconnectScheduler({ + connect: () => this.ensurePrivateSocket(), + backoff: fixedBackoff(WS_RECONNECT_DELAY), + }); private readonly wsTopics = new Set(); private wsConnecting = false; private readonly wsWindow: string; @@ -782,7 +786,7 @@ export class BackpackGateway { this.wsReady = false; this.ws = null; this.stopPing(); - this.scheduleReconnect(); + this.wsReconnect.schedule(); }; private handleWsError = (_event: any): void => { @@ -945,14 +949,6 @@ export class BackpackGateway { } } - private scheduleReconnect(): void { - if (this.wsReconnectTimer) return; - this.wsReconnectTimer = setTimeout(() => { - this.wsReconnectTimer = null; - this.ensurePrivateSocket(); - }, WS_RECONNECT_DELAY); - } - private detachWebSocket(): void { if (this.wsCleanup) { try { diff --git a/src/exchanges/lighter/gateway.ts b/src/exchanges/lighter/gateway.ts index 6324aea..76b6859 100644 --- a/src/exchanges/lighter/gateway.ts +++ b/src/exchanges/lighter/gateway.ts @@ -1,5 +1,6 @@ import { setInterval, clearInterval, setTimeout, clearTimeout } from "timers"; import WebSocket from "ws"; +import { ReconnectScheduler, linearBackoff } from "../reconnect-scheduler"; import type { AccountListener, DepthListener, @@ -150,6 +151,8 @@ const BALANCE_EPSILON = 1e-9; const KLINE_DEFAULT_COUNT = 120; const DEFAULT_TICKER_POLL_MS = 3000; const DEFAULT_KLINE_POLL_MS = 15000; +const WS_RECONNECT_BASE_MS = 2_000; +const WS_RECONNECT_MAX_MS = 30_000; const WS_HEARTBEAT_INTERVAL_MS = 5_000; const CLIENT_PING_INTERVAL_MS = 2_000; const WS_STALE_TIMEOUT_MS = 20_000; @@ -246,8 +249,14 @@ export class LighterGateway { private readonly orderIndexByClientId = new Map(); private ws: WebSocket | null = null; - private reconnectTimer: ReturnType | null = null; - private reconnectAttempts = 0; + private readonly reconnect: ReconnectScheduler = new ReconnectScheduler({ + connect: async () => { + await this.openWebSocket(); + this.reconnect.onConnected(); + }, + backoff: linearBackoff(WS_RECONNECT_BASE_MS, WS_RECONNECT_MAX_MS), + onError: (error) => this.logger("reconnect", error), + }); private readonly wsUrl: string; private connectPromise: Promise | null = null; private heartbeatTimer: ReturnType | null = null; @@ -759,7 +768,7 @@ export class LighterGateway { return; } this.stopStaleMonitor(); - this.scheduleReconnect(); + this.reconnect.schedule(); }); ws.on("error", (error) => { this.logger("ws:error", error); @@ -769,7 +778,7 @@ export class LighterGateway { return; } this.stopStaleMonitor(); - this.scheduleReconnect(); + this.reconnect.schedule(); }); }); } @@ -945,24 +954,6 @@ export class LighterGateway { return candidates[0] ?? null; } - private scheduleReconnect(): void { - if (this.reconnectTimer) return; - const attempt = this.reconnectAttempts + 1; - const delay = Math.min(2000 * attempt, 30_000); - this.reconnectTimer = setTimeout(() => { - this.reconnectTimer = null; - this.openWebSocket() - .then(() => { - this.reconnectAttempts = 0; - }) - .catch((error) => { - this.logger("reconnect", error); - this.reconnectAttempts = attempt; - this.scheduleReconnect(); - }); - }, delay); - } - private forceReconnect(reason: string): void { const now = Date.now(); if (this.staleReason && now - this.lastDepthUpdateAt < FEED_STALE_TIMEOUT_MS / 2) { @@ -978,7 +969,7 @@ export class LighterGateway { } this.stopHeartbeat(); this.stopClientPing(); - this.scheduleReconnect(); + this.reconnect.schedule(); } private startHeartbeat(): void { @@ -995,7 +986,7 @@ export class LighterGateway { } finally { this.stopHeartbeat(); this.stopClientPing(); - this.scheduleReconnect(); + this.reconnect.schedule(); } return; } diff --git a/src/exchanges/ondoperps/gateway.ts b/src/exchanges/ondoperps/gateway.ts index 3e2319d..cf02691 100644 --- a/src/exchanges/ondoperps/gateway.ts +++ b/src/exchanges/ondoperps/gateway.ts @@ -1,4 +1,5 @@ import { createHmac } from "node:crypto"; +import { ReconnectScheduler, exponentialBackoff } from "../reconnect-scheduler"; import type { AccountListener, ConnectionEventListener, @@ -42,10 +43,10 @@ const DEFAULT_WS_URL = "wss://api.ondoperps.xyz/ws"; const DEFAULT_SYMBOL = "BTC-USD.P"; const REQUEST_TIMEOUT_MS = 15_000; const WS_HEARTBEAT_MS = 30_000; +const WS_RECONNECT_BASE_MS = 1_000; const WS_RECONNECT_MAX_MS = 30_000; type Timer = ReturnType; -type Timeout = ReturnType; export interface OndoperpsGatewayOptions { apiKeyId: string; @@ -211,8 +212,11 @@ export class OndoperpsGateway { private wsLoginInFlight = false; private wsLoginFallbackAttempted = false; private wsEverOpened = false; - private wsReconnectDelayMs = 1_000; - private wsReconnectTimer: Timeout | null = null; + private readonly wsReconnect = new ReconnectScheduler({ + connect: () => this.connectWebSocket(), + backoff: exponentialBackoff(WS_RECONNECT_BASE_MS, WS_RECONNECT_MAX_MS), + onError: (error) => this.logger("reconnect", error), + }); private heartbeatTimer: Timer | null = null; private readonly sentSubscriptions = new Set(); @@ -590,10 +594,7 @@ export class OndoperpsGateway { private connectWebSocket(): void { if (this.ws && (this.ws.readyState === 0 || this.ws.readyState === 1)) return; - if (this.wsReconnectTimer) { - clearTimeout(this.wsReconnectTimer); - this.wsReconnectTimer = null; - } + this.wsReconnect.cancel(); try { const ws = this.webSocketFactory(this.wsUrl); this.ws = ws; @@ -605,7 +606,7 @@ export class OndoperpsGateway { ws.addEventListener("error", (event) => this.logger("websocket", event)); } catch (error) { this.logger("connectWebSocket", error); - this.scheduleReconnect(); + this.wsReconnect.schedule(); } } @@ -613,7 +614,7 @@ export class OndoperpsGateway { if (this.ws !== ws) return; const reconnected = this.wsEverOpened; this.wsEverOpened = true; - this.wsReconnectDelayMs = 1_000; + this.wsReconnect.onConnected(); this.wsAuthenticated = false; this.wsLoginInFlight = false; this.wsLoginFallbackAttempted = false; @@ -630,7 +631,7 @@ export class OndoperpsGateway { this.wsLoginInFlight = false; this.stopHeartbeat(); this.emitConnection("disconnected"); - this.scheduleReconnect(); + this.wsReconnect.schedule(); } private async handleWsMessage(raw: unknown): Promise { @@ -772,16 +773,6 @@ export class OndoperpsGateway { this.heartbeatTimer = null; } - private scheduleReconnect(): void { - if (this.wsReconnectTimer) return; - const delay = this.wsReconnectDelayMs; - this.wsReconnectDelayMs = Math.min(this.wsReconnectDelayMs * 2, WS_RECONNECT_MAX_MS); - this.wsReconnectTimer = setTimeout(() => { - this.wsReconnectTimer = null; - this.connectWebSocket(); - }, delay); - } - private startAccountPolling(): void { if (this.accountPollTimer) return; const poll = async () => { diff --git a/src/exchanges/reconnect-scheduler.ts b/src/exchanges/reconnect-scheduler.ts new file mode 100644 index 0000000..f3740bb --- /dev/null +++ b/src/exchanges/reconnect-scheduler.ts @@ -0,0 +1,104 @@ +/** + * How long to wait before the nth reconnect attempt (1-based). + * Return a fixed value for a constant delay, or grow it for backoff. + */ +export type BackoffPolicy = (attempt: number) => number; + +export const fixedBackoff = (delayMs: number): BackoffPolicy => () => delayMs; + +/** `base * 2^(attempt-1)`, capped at `maxMs`. */ +export const exponentialBackoff = (baseMs: number, maxMs: number): BackoffPolicy => (attempt) => + Math.min(baseMs * Math.pow(2, attempt - 1), maxMs); + +/** `base * attempt`, capped at `maxMs`. */ +export const linearBackoff = (baseMs: number, maxMs: number): BackoffPolicy => (attempt) => + Math.min(baseMs * attempt, maxMs); + +export interface ReconnectSchedulerOptions { + /** Reopens the socket. Rejections are reported and then retried. */ + connect: () => void | Promise; + backoff: BackoffPolicy; + /** Returns false to abandon reconnecting (e.g. the gateway was closed). */ + shouldReconnect?: () => boolean; + onError?: (error: unknown, attempt: number) => void; + onSchedule?: (delayMs: number, attempt: number) => void; +} + +/** + * Owns the reconnect timer for one socket. + * + * Every gateway hand-rolled the same three pieces — a "one pending attempt at a + * time" guard, an attempt counter feeding a backoff formula, and resetting that + * counter once the socket opens — each with its own field names and a slightly + * different formula. Forgetting the reset is the classic way backoff silently + * degrades into a 30-second stall after a transient blip, so the reset lives + * here next to the counter it guards. + * + * Deliberately narrow: connect/auth/subscribe/heartbeat differ per venue and + * stay in each gateway. + */ +export class ReconnectScheduler { + private timer: ReturnType | null = null; + private attempts = 0; + private stopped = false; + + constructor(private readonly options: ReconnectSchedulerOptions) {} + + /** Consecutive failed attempts since the last successful open. */ + get attemptCount(): number { + return this.attempts; + } + + get pending(): boolean { + return this.timer != null; + } + + /** Queues a reconnect. A no-op while one is already pending. */ + schedule(): void { + if (this.stopped || this.timer) return; + if (this.options.shouldReconnect && !this.options.shouldReconnect()) return; + + const attempt = this.attempts + 1; + const delay = this.options.backoff(attempt); + this.options.onSchedule?.(delay, attempt); + + this.timer = setTimeout(() => { + this.timer = null; + this.attempts = attempt; + if (this.stopped) return; + try { + const result = this.options.connect(); + if (result && typeof result.then === "function") { + result.catch((error) => this.handleFailure(error, attempt)); + } + } catch (error) { + this.handleFailure(error, attempt); + } + }, delay); + } + + /** Call once the socket is open: clears backoff so the next blip retries fast. */ + onConnected(): void { + this.attempts = 0; + this.cancel(); + } + + /** Cancels a pending attempt without ending the scheduler. */ + cancel(): void { + if (this.timer) { + clearTimeout(this.timer); + this.timer = null; + } + } + + /** Permanently stops reconnecting; use when the gateway shuts down. */ + stop(): void { + this.stopped = true; + this.cancel(); + } + + private handleFailure(error: unknown, attempt: number): void { + this.options.onError?.(error, attempt); + this.schedule(); + } +} diff --git a/src/exchanges/standx/gateway.ts b/src/exchanges/standx/gateway.ts index 8344f16..bfca6ae 100644 --- a/src/exchanges/standx/gateway.ts +++ b/src/exchanges/standx/gateway.ts @@ -1,4 +1,5 @@ import NodeWebSocket from "ws"; +import { ReconnectScheduler, exponentialBackoff } from "../reconnect-scheduler"; import crypto from "crypto"; import { sign, utils as edUtils, hashes as edHashes } from "@noble/ed25519"; import { sha512 } from "@noble/hashes/sha512"; @@ -434,7 +435,6 @@ export class StandxGateway { private marketWsReady = false; private marketWsAuthed = false; private marketWsAuthRequested = false; - private marketReconnectTimer: ReturnType | null = null; private readonly subscriptions = new Set(); // ========== 心跳与连接管理 ========== @@ -443,7 +443,15 @@ export class StandxGateway { // 心跳检查定时器 private heartbeatTimer: ReturnType | null = null; // 重连次数（用于指数退避） - private reconnectAttempts = 0; + private readonly marketReconnect = new ReconnectScheduler({ + connect: () => { + this.logDebug("attempting reconnect"); + this.connectMarketWs(); + }, + backoff: exponentialBackoff(WS_RECONNECT_DELAY_BASE, WS_RECONNECT_DELAY_MAX), + onSchedule: (delay, attempt) => + this.logDebug(`scheduling reconnect in ${delay}ms (attempt ${attempt})`), + }); // ========== 数据过时检测与 REST 备用 ========== // 上次收到行情数据（price/depth）的时间戳 @@ -893,7 +901,7 @@ export class StandxGateway { } private connectMarketWs(): void { - if (this.marketWs || this.marketReconnectTimer) return; + if (this.marketWs || this.marketReconnect.pending) return; this.marketWs = new WebSocketCtor(this.wsUrl); this.marketWsReady = false; this.marketWsAuthed = false; @@ -902,7 +910,7 @@ export class StandxGateway { this.marketWsReady = true; this.marketWsAuthed = false; // 重置重连计数和时间戳 - this.reconnectAttempts = 0; + this.marketReconnect.onConnected(); this.lastMessageTime = Date.now(); this.lastMarketDataTime = Date.now(); this.lastAccountDataTime = Date.now(); @@ -929,7 +937,7 @@ export class StandxGateway { if (wasReady) { this.onDisconnect(); } - this.scheduleReconnect(); + this.marketReconnect.schedule(); }; const handleError = (error: unknown) => { this.logger("marketWs", error); @@ -937,7 +945,7 @@ export class StandxGateway { // 因为某些 WebSocket 实现在握手失败时可能不触发 close 事件 if (this.marketWs && !this.marketWsReady) { this.marketWs = null; - this.scheduleReconnect(); + this.marketReconnect.schedule(); } }; @@ -960,22 +968,6 @@ export class StandxGateway { } } - private scheduleReconnect(): void { - if (this.marketReconnectTimer) return; - // 指数退避：delay = min(base * 2^attempts, max) - const delay = Math.min( - WS_RECONNECT_DELAY_BASE * Math.pow(2, this.reconnectAttempts), - WS_RECONNECT_DELAY_MAX - ); - this.reconnectAttempts += 1; - this.logDebug(`scheduling reconnect in ${delay}ms (attempt ${this.reconnectAttempts})`); - this.marketReconnectTimer = setTimeout(() => { - this.marketReconnectTimer = null; - this.logDebug("attempting reconnect"); - this.connectMarketWs(); - }, delay); - } - private handleMarketMessage(event: { data: any }): void { // 更新最后收到消息的时间（心跳监控） this.lastMessageTime = Date.now(); @@ -1322,9 +1314,9 @@ export class StandxGateway { if (wasReady) { this.onDisconnect(); } - // 立即重连（不使用指数退避，因为是主动行为） - this.reconnectAttempts = 0; - this.scheduleReconnect(); + // Deliberate teardown, so reset the backoff and retry at the base delay. + this.marketReconnect.onConnected(); + this.marketReconnect.schedule(); } private logDebug(context: string, detail?: unknown): void { diff --git a/tests/reconnect-scheduler.test.ts b/tests/reconnect-scheduler.test.ts new file mode 100644 index 0000000..742d180 --- /dev/null +++ b/tests/reconnect-scheduler.test.ts @@ -0,0 +1,169 @@ +import { describe, expect, it, vi, beforeEach, afterEach } from "vitest"; +import { + ReconnectScheduler, + exponentialBackoff, + fixedBackoff, + linearBackoff, +} from "../src/exchanges/reconnect-scheduler"; + +describe("backoff policies", () => { + it("fixed returns the same delay every attempt", () => { + const policy = fixedBackoff(3000); + expect([1, 2, 5].map(policy)).toEqual([3000, 3000, 3000]); + }); + + it("exponential doubles from the base and caps", () => { + const policy = exponentialBackoff(1000, 8000); + expect([1, 2, 3, 4, 5].map(policy)).toEqual([1000, 2000, 4000, 8000, 8000]); + }); + + it("linear grows by the base and caps", () => { + const policy = linearBackoff(2000, 30_000); + expect([1, 2, 3, 20].map(policy)).toEqual([2000, 4000, 6000, 30_000]); + }); +}); + +describe("ReconnectScheduler", () => { + beforeEach(() => vi.useFakeTimers()); + afterEach(() => vi.useRealTimers()); + + it("reconnects after the backoff delay", async () => { + const connect = vi.fn(); + const scheduler = new ReconnectScheduler({ connect, backoff: fixedBackoff(1000) }); + + scheduler.schedule(); + expect(connect).not.toHaveBeenCalled(); + await vi.advanceTimersByTimeAsync(1000); + expect(connect).toHaveBeenCalledTimes(1); + }); + + it("collapses repeated schedule() calls into one pending attempt", async () => { + const connect = vi.fn(); + const scheduler = new ReconnectScheduler({ connect, backoff: fixedBackoff(1000) }); + + scheduler.schedule(); + scheduler.schedule(); + scheduler.schedule(); + expect(scheduler.pending).toBe(true); + await vi.advanceTimersByTimeAsync(1000); + expect(connect).toHaveBeenCalledTimes(1); + }); + + it("grows the delay across consecutive failures", async () => { + const delays: number[] = []; + const scheduler = new ReconnectScheduler({ + connect: async () => { + throw new Error("refused"); + }, + backoff: exponentialBackoff(1000, 60_000), + onSchedule: (delay) => delays.push(delay), + }); + + scheduler.schedule(); + await vi.advanceTimersByTimeAsync(1000); + await vi.advanceTimersByTimeAsync(2000); + await vi.advanceTimersByTimeAsync(4000); + expect(delays.slice(0, 3)).toEqual([1000, 2000, 4000]); + }); + + it("resets the backoff once the socket opens", async () => { + const delays: number[] = []; + let failing = true; + const scheduler = new ReconnectScheduler({ + connect: async () => { + if (failing) throw new Error("refused"); + }, + backoff: exponentialBackoff(1000, 60_000), + onSchedule: (delay) => delays.push(delay), + }); + + scheduler.schedule(); + await vi.advanceTimersByTimeAsync(1000); + await vi.advanceTimersByTimeAsync(2000); + expect(scheduler.attemptCount).toBe(2); + + // A successful open must clear the counter, or the next transient blip + // would wait as long as the last outage did. + failing = false; + scheduler.onConnected(); + expect(scheduler.attemptCount).toBe(0); + + delays.length = 0; + scheduler.schedule(); + expect(delays[0]).toBe(1000); + }); + + it("reports a synchronous connect failure and retries", async () => { + const errors: unknown[] = []; + let calls = 0; + const scheduler = new ReconnectScheduler({ + connect: () => { + calls += 1; + if (calls === 1) throw new Error("boom"); + }, + backoff: fixedBackoff(500), + onError: (error) => errors.push(error), + }); + + scheduler.schedule(); + await vi.advanceTimersByTimeAsync(500); + expect(errors).toHaveLength(1); + await vi.advanceTimersByTimeAsync(500); + expect(calls).toBe(2); + }); + + it("honours shouldReconnect", async () => { + const connect = vi.fn(); + let running = false; + const scheduler = new ReconnectScheduler({ + connect, + backoff: fixedBackoff(100), + shouldReconnect: () => running, + }); + + scheduler.schedule(); + await vi.advanceTimersByTimeAsync(100); + expect(connect).not.toHaveBeenCalled(); + + running = true; + scheduler.schedule(); + await vi.advanceTimersByTimeAsync(100); + expect(connect).toHaveBeenCalledTimes(1); + }); + + it("cancel() drops the pending attempt but keeps the scheduler usable", async () => { + const connect = vi.fn(); + const scheduler = new ReconnectScheduler({ connect, backoff: fixedBackoff(100) }); + + scheduler.schedule(); + scheduler.cancel(); + await vi.advanceTimersByTimeAsync(1000); + expect(connect).not.toHaveBeenCalled(); + + scheduler.schedule(); + await vi.advanceTimersByTimeAsync(100); + expect(connect).toHaveBeenCalledTimes(1); + }); + + it("stop() is permanent", async () => { + const connect = vi.fn(); + const scheduler = new ReconnectScheduler({ connect, backoff: fixedBackoff(100) }); + + scheduler.schedule(); + scheduler.stop(); + await vi.advanceTimersByTimeAsync(1000); + scheduler.schedule(); + await vi.advanceTimersByTimeAsync(1000); + expect(connect).not.toHaveBeenCalled(); + }); + + it("does not reconnect when the timer fires after stop()", async () => { + const connect = vi.fn(); + const scheduler = new ReconnectScheduler({ connect, backoff: fixedBackoff(100) }); + + scheduler.schedule(); + scheduler.stop(); + await vi.advanceTimersByTimeAsync(500); + expect(connect).not.toHaveBeenCalled(); + }); +});