refactor(exchanges): extract ReconnectScheduler from four gateways

standx, ondoperps, lighter and backpack each hand-rolled the same three pieces
around their socket: a 'one pending attempt at a time' timer guard, an attempt
counter feeding a backoff formula, and resetting that counter on open. Four
field-name conventions, four formulas, four places to forget the reset.

ReconnectScheduler owns the timer, the guard and the counter; the backoff stays
a caller-supplied policy (fixed / linear / exponential), since the venues
genuinely want different curves. Verified value-by-value that all four produce
the identical delay sequence they did before.

Scope was chosen, not maximal. The other five gateways' reconnect paths, and all
nine connect/auth/subscribe/heartbeat flows, stay where they are: those are
60-200 lines of per-venue protocol, and folding nine different auth and ping
semantics into one base class would be the wrong abstraction. I also checked the
reconnect paths for the classic defects — aster's timer that looked unclearable
is cleared at the top of connect(), and all three backoff counters do reset on
open — so there was no bug to fix here, only duplication.

12 new tests. 287 pass; tsc and oxlint clean.

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<typeof setInterval> | null = null;
-  private wsReconnectTimer: ReturnType<typeof setTimeout> | null = null;
+  private readonly wsReconnect = new ReconnectScheduler({
+    connect: () => this.ensurePrivateSocket(),
+    backoff: fixedBackoff(WS_RECONNECT_DELAY),
+  });
   private readonly wsTopics = new Set<string>();
   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 {

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<string, string>();
 
   private ws: WebSocket | null = null;
-  private reconnectTimer: ReturnType<typeof setTimeout> | 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<void> | null = null;
   private heartbeatTimer: ReturnType<typeof setInterval> | 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;
       }

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<typeof setInterval>;
-type Timeout = ReturnType<typeof setTimeout>;
 
 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<string>();
 
@@ -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<void> {
@@ -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 () => {

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<void>;
+  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<typeof setTimeout> | 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();
+  }
+}

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<typeof setTimeout> | null = null;
   private readonly subscriptions = new Set<string>();
 
   // ========== 心跳与连接管理 ==========
@@ -443,7 +443,15 @@ export class StandxGateway {
   // 心跳检查定时器
   private heartbeatTimer: ReturnType<typeof setInterval> | 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 {

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();
+  });
+});