diff --git a/.changeset/periodic-stream-reconnect.md b/.changeset/periodic-stream-reconnect.md new file mode 100644 index 0000000..d8e7a7b --- /dev/null +++ b/.changeset/periodic-stream-reconnect.md @@ -0,0 +1,5 @@ +--- +"@imbingox/acex": patch +--- + +修复周期性行情流在业务 payload 停止推送但 WebSocket 仍保持 open 时的恢复行为。Binance funding / mark-price stream 超过配置的 stale 阈值后会主动重连并重放订阅；同一连接上的其它订阅仍会收到正常断线状态，不会被该业务 stale 恢复路径误吞。 diff --git a/.trellis/spec/sdk/adapters.md b/.trellis/spec/sdk/adapters.md index fbcb69d..ab694e3 100644 --- a/.trellis/spec/sdk/adapters.md +++ b/.trellis/spec/sdk/adapters.md @@ -323,12 +323,15 @@ adapter 不得自己构造 `AcexError` 或其他业务错误码——错误码 - **职责划分**： - 通用核（`SubscriptionMultiplexer`）：按 `connectionKey` 池化物理连接、引用计数、重连后重放订阅、控制帧批量+限速、per-subscription ready / freshness fan-out、连接级 stale watchdog、最后一个订阅关闭时拆连接、单连接订阅数达上限时同 `connectionKey` 开新连接（连接池）。 - - venue 策略（`VenueStreamProtocol`）：`subscriptionKey` / `connectionKey` / `connectionUrl` / `parseMessage` / `encodeSubscribe` / `encodeUnsubscribe` / `routeMessage(→ data|status|ack|ignore)`。所有交易所特定的 base URL、帧格式、消息路由判据都在这里，**不得**泄漏进通用核。 + - venue 策略（`VenueStreamProtocol`）：`subscriptionKey` / `connectionKey` / `connectionUrl` / `parseMessage` / `encodeSubscribe` / `encodeUnsubscribe` / `routeMessage(→ data|status|ack|ignore)` / 可选 `livenessPolicy(descriptor)`。所有交易所特定的 base URL、帧格式、消息路由判据和 logical stream 活性语义都在这里，**不得**泄漏进通用核。 - **logical stream 契约**： - `ready`：该订阅的 subscribe ACK / 等价订阅接受信号到达时 resolve；超 `initialMessageTimeoutMs` 未到则 reject 并清理该订阅。若首条 `data` 在 ACK 前到达，且 `routeMessage` 能给出匹配的 `subscriptionKey`，该 data 同时作为等价订阅接受信号、payload 和 freshness 来源。 - `status` route：只分发给 `callbacks.onStatus`，不清首包 timer、不 resolve `ready`、不把订阅 freshness 切成 `fresh`。只用于真正非数据型 stream 状态；Deribit quote 的 bid-only / ask-only / empty 必须使用 `data` route。 - `close()`：发出该订阅的 `UNSUBSCRIBE`；物理连接保持，直到其上最后一个订阅 close 才断开。幂等。 - - stale：不做 per-subscription 独立 stale。只要同一物理连接仍有其他订阅收到消息，静默订阅保持 `fresh`；整条连接超过 `staleAfterMs` 无任何有效消息时，由连接级 watchdog 对该连接所有订阅触发 `onFreshnessChange("stale","heartbeat_timeout")`。 + - stale 默认语义：未声明 business liveness policy 的订阅不做 per-subscription 独立 stale。只要同一物理连接仍有其他订阅收到消息，静默订阅保持 `fresh`；整条连接超过 `staleAfterMs` 无任何有效消息时，由连接级 watchdog 对该连接所有订阅触发 `onFreshnessChange("stale","heartbeat_timeout")`。 + - business liveness policy：周期性业务帧 stream（例如 Binance funding/mark price）可以由 `VenueStreamProtocol.livenessPolicy()` 返回 `{ kind: "periodic" | "low_volume"; staleAfterMs: number; onStale: "mark_stale" | "reconnect" }`。`staleAfterMs` 必须由 venue policy 显式给出，不能在复用器里回落到连接级 watchdog 阈值；timer 只能在订阅已接受后启动（subscribe ACK 成功、无 ACK 协议的 subscribe frame 已发送、或首条 `data` 作为等价接受信号），并且只由 routed `data` payload 刷新；ACK、status、ignore、pong 和其它连接活性都不算该 logical subscription 的业务 payload。 + - business stale 行为：policy timer 到期时只标记对应 logical subscription 的 subscriber stale，reason 固定为 `"heartbeat_timeout"`。`onStale:"reconnect"` 必须对当前 raw socket 触发 `ManagedWebSocketSession.restart()`，复用 close → reconnect → replay 路径；`restart()` 返回是否实际发起 raw close，只有返回 true 时才抑制触发订阅本身的 `onDisconnected()`，避免刚写入的 business `heartbeat_timeout` 立即被上层覆盖成 transport `ws_disconnected`。同一 socket 上的其它订阅仍应收到 `onDisconnected()`，因为它们经历的是底层 transport close。 + - Binance policy：`fundingRate` descriptor 使用 `periodic + reconnect`，且把 adapter stream options 里的 `staleAfterMs` 显式写入 policy；`l1book` 和未声明 policy 的其它 stream 保持默认 stale 语义。private user-data stream 不使用行情级 business liveness policy，继续走 private adapter/listenKey/reconcile 的长周期恢复契约。 - 断线：对该连接所有订阅 `onDisconnected()`（上层据此置 `ws_disconnected`），并**静默**置内部 freshness=stale（不再额外发 `heartbeat_timeout` freshness 事件）；重连 `open` 后自动重放全部活跃订阅，各订阅在各自首条消息回到 `fresh`。 - **限额（保守口径，依据 Binance 官方）**： - 单连接 stream 上限：Binance 现货与 USDⓈ-M 合约**均为 1024**（Options 才是 200，不适用）。本仓库 `maxSubscriptionsPerConnection` 取保守值 **200**，到上限即开新连接。200 不是交易所硬限制，而是运营上限：平衡控制帧开销、per-subscription heartbeat/control 消息、路由与 fan-out 的 CPU/内存成本、高订阅数下的延迟/丢包观测、连接稳定性与重连限额，以及突发订阅的安全余量。后续可依据 `tests/soak/`（尤其 market L1 continuity）与线上 telemetry/packet-loss/latency 指标重新调优。 @@ -378,6 +381,10 @@ interface VenueHeartbeat { | `close()` 被外部调用两次 | 第二次 no-op | 抛 `ClosedError` 或重复清 timer | | 稳态时 WS 被对端关闭 | 触发 `onDisconnected` + 内部触发 ManagedWebSocket 重连 | 让 adapter 自行 `setTimeout` 重连 | | 整条连接长时间不收消息 | 连接级 watchdog 对该连接所有订阅触发 `onFreshnessChange("stale", "heartbeat_timeout")` | 单个订阅静默时自行标 stale 或主动 close 重连 | +| periodic business stream 超过 `staleAfterMs` 没有 routed `data` | 只标记该 logical subscription `heartbeat_timeout`，按 policy 可 `restart()` raw socket 并 replay | 把 ACK/status/pong 当业务 payload，或用 `session.close()` 禁掉重连 | +| business stale 触发主动 restart | 对触发订阅保留 reason 为 `heartbeat_timeout`；同 socket 其它订阅仍收到 `onDisconnected()` | 用 connection 级 suppression 吞掉无关订阅的 transport disconnect | +| `ManagedWebSocketSession.restart()` 因 socket closing/closed no-op | 返回 false，复用器清掉本次 suppression 标记 | no-op 后留下 suppression，导致后续真实 close 被吞掉 | +| 未声明 liveness policy 的订阅静默但同连接仍活跃 | 不启动 per-sub timer、不主动重连；继续依赖连接级 watchdog | 默认给所有 logical subscription 加短周期 business timeout | | 交易所要求应用层客户端 ping | 在 `VenueStreamProtocol.heartbeat` 声明文本帧、pong 判定、调度模式；由 ManagedWebSocket 统一调度 | adapter/multiplexer 外层自行 `setInterval` 发 ping | | 收到应用层 pong raw 帧 | 在 `parseMessage()` 前消费，清 pending pong、刷新连接活性，不路由给订阅 | 让 pong 进入 `routeMessage()` 或误判为 data/ack | | 应用层 ping 超过 `pongTimeoutMs` 未收到 pong | 对底层 raw socket 调 `close()`，复用 close→reconnect→replay 路径 | 调 session `close()` 禁掉重连，或直接在 adapter 手写重连 | @@ -496,6 +503,7 @@ bun run test - 修改或新增 `fetchServerTime()` 时必须覆盖正常解析、HTTP 失败、无自动重试、缺失 / 非 number `serverTime`、限流顺序、单调时钟 RTT、manager 错误包装与不支持 venue。 - 修改或新增 public market REST query（如 raw trades / funding history）时必须覆盖：endpoint/参数映射、`receivedAt` 采样、raw clone、manager canonical 化、unsupported venue/market、adapter failure error wrapping、rate-limit semantic plan。 - 修改 Binance derivatives catalog 归一化时必须覆盖 `TRADIFI_PERPETUAL` fixture，断言其 symbol 为永续格式（如 `AAPL/USDT:USDT`）、`type:"swap"`、无 `expiry`，并能走 L1 book 与 funding/mark price 订阅。 +- 修改 `SubscriptionMultiplexer` business liveness policy 时，`tests/unit/subscription-multiplexer.test.ts` 必须覆盖：periodic subscription 业务帧 stale 后标记 `heartbeat_timeout` 并 raw socket restart/replay；未声明 policy 的订阅不因单订阅静默主动重连；无 ACK 协议在 subscribe frame 发送后启动 timer；业务 stale restart 不触发普通 `onDisconnected()` 覆盖上层 reason。`tests/unit/binance-stream-protocol.test.ts` 必须覆盖 Binance funding descriptor 返回 `periodic + reconnect`，L1 descriptor 不返回 policy；`tests/integration/market.test.ts` 必须覆盖 funding stale 后旧 socket 关闭、新 socket replay、收到新 mark price 后 snapshot 恢复 fresh。 - `tests/integration/market.test.ts`、`tests/integration/account.test.ts`、`tests/integration/order.test.ts` 针对各 manager 的集成测试仍然过——这些测试通过 fake REST / fake WebSocket 间接验证 adapter contract。 - 修改 `refreshAccount()` 或 account polling 时，必须覆盖 REST refresh 后 `risk.updated` 与 `position.updated` 都通过 public event/getter 可见。 - 新增随行情变化的账户字段时，测试必须覆盖“没有 WS 消息、只有 refresh/polling”时字段仍会更新；同时覆盖 WS 断线期间 refresh 成功不会把 stream status 改成 healthy。 diff --git a/src/adapters/binance/adapter.ts b/src/adapters/binance/adapter.ts index 5c34dff..d2d79ef 100644 --- a/src/adapters/binance/adapter.ts +++ b/src/adapters/binance/adapter.ts @@ -254,18 +254,24 @@ export class BinanceMarketAdapter implements MarketAdapter { BinanceStreamMessage, BinanceStreamDescriptor, BinanceStreamPayload - >(new BinanceStreamProtocol(), { - initialMessageTimeoutMs: config.initialMessageTimeoutMs, - staleAfterMs: config.staleAfterMs, - reconnectDelayMs: config.reconnectDelayMs, - reconnectMaxDelayMs: config.reconnectMaxDelayMs, - controlFrameMaxPerSec: BINANCE_CONTROL_FRAME_MAX_PER_SEC, - maxSubscriptionsPerConnection: BINANCE_MAX_SUBSCRIPTIONS_PER_CONNECTION, - now: config.now, - onReconnect: ({ descriptors }) => { - this.emitReconnectMetric(descriptors); + >( + new BinanceStreamProtocol({ + fundingRateStaleAfterMs: config.staleAfterMs, + }), + { + initialMessageTimeoutMs: config.initialMessageTimeoutMs, + staleAfterMs: config.staleAfterMs, + reconnectDelayMs: config.reconnectDelayMs, + reconnectMaxDelayMs: config.reconnectMaxDelayMs, + controlFrameMaxPerSec: BINANCE_CONTROL_FRAME_MAX_PER_SEC, + maxSubscriptionsPerConnection: + BINANCE_MAX_SUBSCRIPTIONS_PER_CONNECTION, + now: config.now, + onReconnect: ({ descriptors }) => { + this.emitReconnectMetric(descriptors); + }, }, - }); + ); this.multiplexerConfig = config; return this.multiplexer; } diff --git a/src/adapters/binance/stream-protocol.ts b/src/adapters/binance/stream-protocol.ts index 608d03e..d5961a0 100644 --- a/src/adapters/binance/stream-protocol.ts +++ b/src/adapters/binance/stream-protocol.ts @@ -1,5 +1,6 @@ import type { EncodedVenueControlFrame, + StreamLivenessPolicy, VenueControlAck, VenueStreamProtocol, } from "../../internal/subscription-multiplexer.ts"; @@ -48,6 +49,10 @@ export type BinanceStreamPayload = readonly exchangeTs?: number; }; +interface BinanceStreamProtocolOptions { + readonly fundingRateStaleAfterMs: number; +} + const BINANCE_SPOT_WS_BASE_URL = "wss://stream.binance.com:9443/ws"; const BINANCE_USDM_WS_BASE_URL = "wss://fstream.binance.com/ws"; const BINANCE_USDM_MARKET_WS_BASE_URL = "wss://fstream.binance.com/market/ws"; @@ -94,6 +99,8 @@ export class BinanceStreamProtocol { private nextControlFrameId = 1; + constructor(private readonly options: BinanceStreamProtocolOptions) {} + subscriptionKey(descriptor: BinanceStreamDescriptor): string { return `${descriptor.channel}:${descriptor.market.id}`; } @@ -131,6 +138,20 @@ export class BinanceStreamProtocol return this.encodeControlFrame("UNSUBSCRIBE", descriptors); } + livenessPolicy( + descriptor: BinanceStreamDescriptor, + ): StreamLivenessPolicy | undefined { + if (descriptor.channel !== "fundingRate") { + return undefined; + } + + return { + kind: "periodic", + staleAfterMs: this.options.fundingRateStaleAfterMs, + onStale: "reconnect", + }; + } + routeMessage(message: BinanceStreamMessage): | { kind: "data"; diff --git a/src/internal/managed-websocket.ts b/src/internal/managed-websocket.ts index 0644084..3423f08 100644 --- a/src/internal/managed-websocket.ts +++ b/src/internal/managed-websocket.ts @@ -62,6 +62,7 @@ export interface ManagedWebSocketOptions { export interface ManagedWebSocketSession { readonly ready: Promise; send(data: string): void; + restart(reason?: string): boolean; close(): void; } @@ -476,6 +477,19 @@ export function createManagedWebSocket( activeSocket.send(data); }, + restart(reason = "websocket restart requested"): boolean { + if ( + closed || + !activeSocket || + activeSocket.readyState === WebSocket.CLOSING || + activeSocket.readyState === WebSocket.CLOSED + ) { + return false; + } + + activeSocket.close(1000, reason); + return true; + }, close() { if (closed) { return; diff --git a/src/internal/subscription-multiplexer.ts b/src/internal/subscription-multiplexer.ts index c19763b..c21b3b9 100644 --- a/src/internal/subscription-multiplexer.ts +++ b/src/internal/subscription-multiplexer.ts @@ -9,6 +9,15 @@ type Freshness = "fresh" | "stale"; type StaleReason = "heartbeat_timeout"; type ControlFrameKind = "subscribe" | "unsubscribe"; type ControlFrameAckId = number | string; +type LivenessStaleAction = "mark_stale" | "reconnect"; + +export type StreamLivenessPolicy = + | { readonly kind: "event_driven" } + | { + readonly kind: "periodic" | "low_volume"; + readonly staleAfterMs: number; + readonly onStale: LivenessStaleAction; + }; export interface EncodedVenueControlFrame { readonly data: string; @@ -57,6 +66,7 @@ export interface VenueStreamProtocol< parseMessage(data: string): TMessage | undefined; encodeSubscribe(descriptors: TDescriptor[]): VenueControlFrameEncoding; encodeUnsubscribe(descriptors: TDescriptor[]): VenueControlFrameEncoding; + livenessPolicy?(descriptor: TDescriptor): StreamLivenessPolicy | undefined; routeMessage( message: TMessage, ): @@ -100,7 +110,10 @@ interface LocalSubscriber { interface SubState { readonly descriptor: TDescriptor; readonly subscribers: Set>; + readonly livenessPolicy: StreamLivenessPolicy | undefined; ready: boolean; + livenessTimer: TimerHandle | undefined; + livenessStaleNotified: boolean; } interface ControlFrame { @@ -128,6 +141,7 @@ interface ConnectionState { isOpen: boolean; hasOpened: boolean; closeAfterControlQueueDrained: boolean; + readonly suppressedDisconnectSubscriptionKeys: Set; controlTimer: TimerHandle | undefined; lastControlSentAt: number | undefined; } @@ -171,6 +185,15 @@ function controlAckKey(id: ControlFrameAckId): string { return String(id); } +function tracksBusinessLiveness( + policy: StreamLivenessPolicy | undefined, +): policy is Extract< + StreamLivenessPolicy, + { readonly kind: "periodic" | "low_volume" } +> { + return policy?.kind === "periodic" || policy?.kind === "low_volume"; +} + export class SubscriptionMultiplexer< TMessage, TDescriptor, @@ -253,7 +276,10 @@ export class SubscriptionMultiplexer< const sub: SubState = { descriptor, subscribers: new Set([localSubscriber]), + livenessPolicy: this.protocol.livenessPolicy?.(descriptor), ready: false, + livenessTimer: undefined, + livenessStaleNotified: false, }; connection.subs.set(subscriptionKey, sub); @@ -331,6 +357,7 @@ export class SubscriptionMultiplexer< isOpen: false, hasOpened: false, closeAfterControlQueueDrained: false, + suppressedDisconnectSubscriptionKeys: new Set(), controlTimer: undefined, lastControlSentAt: undefined, }; @@ -355,7 +382,7 @@ export class SubscriptionMultiplexer< messageWatchdog: { staleAfterMs: this.options.staleAfterMs, onStale: () => { - this.markAllStale(connection, "heartbeat_timeout"); + this.handleConnectionStale(connection); }, }, reconnect: { @@ -469,6 +496,10 @@ export class SubscriptionMultiplexer< private handleUnexpectedClose( connection: ConnectionState, ): void { + const suppressedKeys = new Set( + connection.suppressedDisconnectSubscriptionKeys, + ); + connection.suppressedDisconnectSubscriptionKeys.clear(); connection.isOpen = false; connection.lastControlSentAt = undefined; if (connection.controlTimer) { @@ -476,6 +507,7 @@ export class SubscriptionMultiplexer< connection.controlTimer = undefined; } this.clearPendingControlAcks(connection); + this.clearConnectionBusinessLivenessTimers(connection); if (connection.subs.size === 0) { this.closeConnection(connection); @@ -483,7 +515,11 @@ export class SubscriptionMultiplexer< } this.markAllStaleSilently(connection); - for (const sub of connection.subs.values()) { + for (const [subscriptionKey, sub] of connection.subs) { + if (suppressedKeys.has(subscriptionKey)) { + continue; + } + for (const localSubscriber of sub.subscribers) { localSubscriber.callbacks.onDisconnected(); } @@ -511,7 +547,13 @@ export class SubscriptionMultiplexer< } if (routed.kind === "data" && !sub.ready) { - this.markSubscriptionReady(sub); + this.markSubscriptionReady(connection, routed.subscriptionKey, sub, { + startLivenessTimer: false, + }); + } + + if (routed.kind === "data") { + this.noteSubscriptionData(connection, routed.subscriptionKey, sub); } if (sub.subscribers.size === 1) { @@ -616,6 +658,7 @@ export class SubscriptionMultiplexer< for (const localSubscriber of sub.subscribers) { this.clearInitialTimer(localSubscriber); } + this.clearBusinessLivenessTimer(sub); } private resolveSubReady( @@ -630,13 +673,22 @@ export class SubscriptionMultiplexer< } private markSubscriptionReady( + connection: ConnectionState, + subscriptionKey: string, sub: SubState, + options: { startLivenessTimer: boolean } = { + startLivenessTimer: true, + }, ): void { sub.ready = true; for (const localSubscriber of sub.subscribers) { this.clearInitialTimer(localSubscriber); this.resolveSubReady(localSubscriber); } + + if (options.startLivenessTimer) { + this.scheduleBusinessLivenessTimer(connection, subscriptionKey, sub); + } } private rejectSubscriptionReady( @@ -908,7 +960,7 @@ export class SubscriptionMultiplexer< for (const subscriptionKey of frame.descriptors.keys()) { const sub = connection.subs.get(subscriptionKey); if (sub) { - this.markSubscriptionReady(sub); + this.markSubscriptionReady(connection, subscriptionKey, sub); } } } @@ -961,7 +1013,7 @@ export class SubscriptionMultiplexer< for (const subscriptionKey of pending.subscriptionKeys) { const sub = connection.subs.get(subscriptionKey); if (sub) { - this.markSubscriptionReady(sub); + this.markSubscriptionReady(connection, subscriptionKey, sub); } } } @@ -1077,4 +1129,101 @@ export class SubscriptionMultiplexer< localSubscriber.callbacks.onDisconnected(); } } + + private handleConnectionStale( + connection: ConnectionState, + ): void { + this.markAllStale(connection, "heartbeat_timeout"); + } + + private noteSubscriptionData( + connection: ConnectionState, + subscriptionKey: string, + sub: SubState, + ): void { + sub.livenessStaleNotified = false; + this.scheduleBusinessLivenessTimer(connection, subscriptionKey, sub, { + resetExisting: true, + }); + } + + private scheduleBusinessLivenessTimer( + connection: ConnectionState, + subscriptionKey: string, + sub: SubState, + options: { resetExisting: boolean } = { resetExisting: false }, + ): void { + const policy = sub.livenessPolicy; + if (!tracksBusinessLiveness(policy)) { + return; + } + + if (sub.livenessTimer) { + if (!options.resetExisting) { + return; + } + + this.clearTimer(sub.livenessTimer); + sub.livenessTimer = undefined; + } + + sub.livenessStaleNotified = false; + sub.livenessTimer = this.setTimer(() => { + sub.livenessTimer = undefined; + if ( + connection.subs.get(subscriptionKey) !== sub || + !connection.isOpen || + sub.livenessStaleNotified + ) { + return; + } + + sub.livenessStaleNotified = true; + this.markSubscriptionStale(sub, "heartbeat_timeout"); + if (policy.onStale === "reconnect") { + connection.suppressedDisconnectSubscriptionKeys.add(subscriptionKey); + const restarted = connection.session.restart( + `business liveness timeout for ${subscriptionKey}`, + ); + if (!restarted) { + connection.suppressedDisconnectSubscriptionKeys.delete( + subscriptionKey, + ); + } + } + }, policy.staleAfterMs); + } + + private clearBusinessLivenessTimer( + sub: SubState, + ): void { + if (!sub.livenessTimer) { + return; + } + + this.clearTimer(sub.livenessTimer); + sub.livenessTimer = undefined; + } + + private clearConnectionBusinessLivenessTimers( + connection: ConnectionState, + ): void { + for (const sub of connection.subs.values()) { + this.clearBusinessLivenessTimer(sub); + } + } + + private markSubscriptionStale( + sub: SubState, + reason: StaleReason, + ): void { + for (const localSubscriber of sub.subscribers) { + if (localSubscriber.freshness === "stale") { + continue; + } + + localSubscriber.freshness = "stale"; + localSubscriber.callbacks.onFreshnessChange("stale", reason); + } + } } diff --git a/tests/integration/market.test.ts b/tests/integration/market.test.ts index aed37b1..4c07296 100644 --- a/tests/integration/market.test.ts +++ b/tests/integration/market.test.ts @@ -1159,19 +1159,7 @@ test("funding rate stream handles stale disconnect and reconnect", async () => { freshness: "stale", reason: "heartbeat_timeout", }); - - firstSocket.disconnect(); - await Bun.sleep(0); - - expect( - client.market.getMarketStatus({ - venue: "binance", - symbol: "BTC/USDT:USDT", - }), - ).toMatchObject({ - freshness: "stale", - reason: "ws_disconnected", - }); + expect(firstSocket.readyState).toBe(FakeWebSocket.CLOSED); const secondSocket = await waitForSocket( BINANCE_USDM_MARKET_WS_BASE_URL, diff --git a/tests/unit/binance-stream-protocol.test.ts b/tests/unit/binance-stream-protocol.test.ts index 21fdcac..3340a40 100644 --- a/tests/unit/binance-stream-protocol.test.ts +++ b/tests/unit/binance-stream-protocol.test.ts @@ -1,8 +1,29 @@ import { expect, test } from "bun:test"; +import type { BinanceMarketDefinition } from "../../src/adapters/binance/market-catalog.ts"; import { BinanceStreamProtocol } from "../../src/adapters/binance/stream-protocol.ts"; +const binanceUsdmMarket: BinanceMarketDefinition = { + venue: "binance", + family: "usdm", + symbol: "BTC/USDT:USDT", + id: "BTCUSDT", + type: "swap", + base: "BTC", + quote: "USDT", + settle: "USDT", + active: true, + contract: true, + linear: true, + contractSize: "1", + pricePrecision: 1, + amountPrecision: 3, + priceStep: "0.1", + amountStep: "0.001", + raw: {}, +}; + test("BinanceStreamProtocol routes funding only for markPriceUpdate events", () => { - const protocol = new BinanceStreamProtocol(); + const protocol = new BinanceStreamProtocol({ fundingRateStaleAfterMs: 100 }); expect(protocol.routeMessage({ s: "BTCUSDT", r: "0.00010000" })).toEqual({ kind: "ignore", @@ -22,3 +43,24 @@ test("BinanceStreamProtocol routes funding only for markPriceUpdate events", () }, }); }); + +test("BinanceStreamProtocol applies periodic reconnect liveness only to funding rate streams", () => { + const protocol = new BinanceStreamProtocol({ fundingRateStaleAfterMs: 100 }); + + expect( + protocol.livenessPolicy({ + channel: "fundingRate", + market: binanceUsdmMarket, + }), + ).toEqual({ + kind: "periodic", + staleAfterMs: 100, + onStale: "reconnect", + }); + expect( + protocol.livenessPolicy({ + channel: "l1book", + market: binanceUsdmMarket, + }), + ).toBeUndefined(); +}); diff --git a/tests/unit/subscription-multiplexer.test.ts b/tests/unit/subscription-multiplexer.test.ts index 7edf51b..6cfd3f0 100644 --- a/tests/unit/subscription-multiplexer.test.ts +++ b/tests/unit/subscription-multiplexer.test.ts @@ -2,6 +2,7 @@ import { beforeEach, expect, test } from "bun:test"; import { type MultiplexedStreamCallbacks, type MultiplexerSubscriptionHandle, + type StreamLivenessPolicy, SubscriptionMultiplexer, type VenueHeartbeat, type VenueStreamProtocol, @@ -208,6 +209,22 @@ const noAckProtocol: VenueStreamProtocol< }, }; +function createLivenessProtocol( + livenessPolicy: ( + descriptor: FakeDescriptor, + ) => StreamLivenessPolicy | undefined, + baseProtocol: VenueStreamProtocol< + FakeMessage, + FakeDescriptor, + FakePayload + > = protocol, +): VenueStreamProtocol { + return { + ...baseProtocol, + livenessPolicy, + }; +} + function createCallbacks(): { callbacks: MultiplexedStreamCallbacks; log: CallbackLog; @@ -1129,6 +1146,188 @@ test("quiet subscription stays fresh while the shared connection receives other ]); }); +test("periodic subscription stale restarts the socket and replays the subscription", async () => { + const clock = new FakeClock(); + const periodicProtocol = createLivenessProtocol((item) => + item.key === "a" + ? { kind: "periodic", staleAfterMs: 100, onStale: "reconnect" } + : undefined, + ); + const multiplexer = createMultiplexerWithProtocol(clock, periodicProtocol); + const callbacks = createCallbacks(); + const handle = multiplexer.subscribe(descriptor("a"), callbacks.callbacks); + + const socket = await openSocket("wss://fake.test/alpha"); + clock.advance(0); + ackFrame(socket, 0); + await handle.ready; + socket.emitJson({ key: "a", value: "A1" }); + + expect(callbacks.log.freshness).toEqual([{ freshness: "fresh" }]); + + clock.advance(99); + expect(socket.readyState).toBe(FakeWebSocket.OPEN); + + clock.advance(1); + expect(socket.readyState).toBe(FakeWebSocket.CLOSED); + expect(callbacks.log.freshness).toEqual([ + { freshness: "fresh" }, + { freshness: "stale", reason: "heartbeat_timeout" }, + ]); + expect(callbacks.log.disconnected).toBe(0); + + clock.advance(10); + const reconnectSocket = await waitForSocket("wss://fake.test/alpha", 1); + await Promise.resolve(); + clock.advance(0); + + expect(sentFrame(reconnectSocket, 0)).toEqual({ + op: "sub", + keys: ["a"], + }); + + ackFrame(reconnectSocket, 0); + reconnectSocket.emitJson({ key: "a", value: "A2" }); + + expect(callbacks.log.payloads.map((entry) => entry.payload.value)).toEqual([ + "A1", + "A2", + ]); + expect(callbacks.log.freshness).toEqual([ + { freshness: "fresh" }, + { freshness: "stale", reason: "heartbeat_timeout" }, + { freshness: "fresh" }, + ]); +}); + +test("periodic restart still disconnects unrelated subscriptions on the shared socket", async () => { + const clock = new FakeClock(); + const periodicProtocol = createLivenessProtocol((item) => + item.key === "a" + ? { kind: "periodic", staleAfterMs: 100, onStale: "reconnect" } + : undefined, + ); + const multiplexer = createMultiplexerWithProtocol(clock, periodicProtocol); + const periodic = createCallbacks(); + const unrelated = createCallbacks(); + + multiplexer.subscribe(descriptor("a"), periodic.callbacks); + multiplexer.subscribe(descriptor("b"), unrelated.callbacks); + + const socket = await openSocket("wss://fake.test/alpha"); + clock.advance(0); + ackFrame(socket, 0); + socket.emitJson({ key: "a", value: "A1" }); + socket.emitJson({ key: "b", value: "B1" }); + + clock.advance(100); + + expect(socket.readyState).toBe(FakeWebSocket.CLOSED); + expect(periodic.log.freshness).toEqual([ + { freshness: "fresh" }, + { freshness: "stale", reason: "heartbeat_timeout" }, + ]); + expect(periodic.log.disconnected).toBe(0); + expect(unrelated.log.freshness).toEqual([{ freshness: "fresh" }]); + expect(unrelated.log.disconnected).toBe(1); + + clock.advance(10); + const reconnectSocket = await waitForSocket("wss://fake.test/alpha", 1); + await Promise.resolve(); + clock.advance(0); + + expect(sentFrame(reconnectSocket, 0)).toEqual({ + op: "sub", + keys: ["a", "b"], + }); +}); + +test("periodic restart no-op does not leave disconnect suppression behind", async () => { + const clock = new FakeClock(); + const periodicProtocol = createLivenessProtocol(() => ({ + kind: "periodic", + staleAfterMs: 100, + onStale: "reconnect", + })); + const multiplexer = createMultiplexerWithProtocol(clock, periodicProtocol); + const callbacks = createCallbacks(); + + multiplexer.subscribe(descriptor("a"), callbacks.callbacks); + + const socket = await openSocket("wss://fake.test/alpha"); + clock.advance(0); + ackFrame(socket, 0); + socket.emitJson({ key: "a", value: "A1" }); + socket.readyState = FakeWebSocket.CLOSING; + + clock.advance(100); + + expect(callbacks.log.freshness).toEqual([ + { freshness: "fresh" }, + { freshness: "stale", reason: "heartbeat_timeout" }, + ]); + expect(callbacks.log.disconnected).toBe(0); + + socket.disconnect(); + + expect(callbacks.log.disconnected).toBe(1); +}); + +test("default subscription stale does not restart an open socket", async () => { + const clock = new FakeClock(); + const multiplexer = createMultiplexer(clock); + const callbacks = createCallbacks(); + const handle = multiplexer.subscribe(descriptor("a"), callbacks.callbacks); + + const socket = await openSocket("wss://fake.test/alpha"); + clock.advance(0); + ackFrame(socket, 0); + await handle.ready; + socket.emitJson({ key: "a", value: "A1" }); + + clock.advance(100); + + expect(socket.readyState).toBe(FakeWebSocket.OPEN); + expect(socketsForUrl("wss://fake.test/alpha")).toHaveLength(1); + expect(callbacks.log.freshness).toEqual([ + { freshness: "fresh" }, + { freshness: "stale", reason: "heartbeat_timeout" }, + ]); +}); + +test("periodic liveness starts after a no-ACK subscribe frame is sent", async () => { + const clock = new FakeClock(); + const periodicNoAckProtocol = createLivenessProtocol( + () => ({ kind: "periodic", staleAfterMs: 100, onStale: "reconnect" }), + noAckProtocol, + ); + const multiplexer = createMultiplexerWithProtocol( + clock, + periodicNoAckProtocol, + ); + const callbacks = createCallbacks(); + const handle = multiplexer.subscribe(descriptor("a"), callbacks.callbacks); + + const socket = await openSocket("wss://fake.test/alpha"); + clock.advance(0); + + expect(sentFrame(socket, 0)).toEqual({ op: "sub", keys: ["a"] }); + await handle.ready; + + clock.advance(100); + + expect(socket.readyState).toBe(FakeWebSocket.CLOSED); + expect(callbacks.log.freshness).toEqual([]); + expect(callbacks.log.disconnected).toBe(0); + + clock.advance(10); + const reconnectSocket = await waitForSocket("wss://fake.test/alpha", 1); + await Promise.resolve(); + clock.advance(0); + + expect(sentFrame(reconnectSocket, 0)).toEqual({ op: "sub", keys: ["a"] }); +}); + test("control frames are batched and throttled per connection", async () => { const clock = new FakeClock(); const multiplexer = createMultiplexer(clock, 1);