From 041fd206cb9adf78f4f3bb776329b2b8b51936a0 Mon Sep 17 00:00:00 2001 From: Hyoung-yoon Kim Date: Mon, 28 Sep 2026 11:20:09 -0400 Subject: [PATCH 1/3] chore(config): duplicate module name check --- .../data-proxy/src/config-parser.test.ts | 50 +++++++++++++++++++ .../data-proxy/src/config/config-parser.ts | 9 ++++ 2 files changed, 59 insertions(+) diff --git a/workspace/data-proxy/src/config-parser.test.ts b/workspace/data-proxy/src/config-parser.test.ts index bbc5023..1ad4980 100644 --- a/workspace/data-proxy/src/config-parser.test.ts +++ b/workspace/data-proxy/src/config-parser.test.ts @@ -879,6 +879,56 @@ describe("parseConfig", () => { } expect(module.wsUrl).toBe("wss://ws.okx.com:8443/ws/v5/public"); }); + + it("rejects duplicate module names", () => { + const [result] = Effect.runSync( + parseConfig({ + modules: [ + { name: "lighter_dup", type: "lighter", streamType: "ticker" }, + { + name: "lighter_dup", + type: "lighter", + streamType: "order_book", + }, + { name: "lighter_dup", type: "lighter", streamType: "trade" }, + ], + routes: [ + { + type: "lighter", + moduleName: "lighter_dup", + path: "/lighter/ticker/:markets", + fetchFromModule: "{:markets}", + }, + ], + }), + ); + + expect(result).toBeErrResult('Duplicate module name "lighter_dup"'); + }); + + it("allows several routes to share one module name", () => { + const [result] = Effect.runSync( + parseConfig({ + modules: [{ name: "lighter", type: "lighter" }], + routes: [ + { + type: "lighter", + moduleName: "lighter", + path: "/lighter/ticker/:markets", + fetchFromModule: "{:markets}", + }, + { + type: "lighter", + moduleName: "lighter", + path: "/lighter/trade/:markets", + fetchFromModule: "{:markets}", + }, + ], + }), + ); + + expect(result).toBeOkResult(); + }); }); describe("it should fail on unknown properties", () => { diff --git a/workspace/data-proxy/src/config/config-parser.ts b/workspace/data-proxy/src/config/config-parser.ts index 924babb..c6b06de 100644 --- a/workspace/data-proxy/src/config/config-parser.ts +++ b/workspace/data-proxy/src/config/config-parser.ts @@ -498,8 +498,17 @@ export const parseConfig = ( } const modules: Modules[] = []; + const seenModuleNames = new Set(); for (const module of config.modules) { + if (seenModuleNames.has(module.name)) { + return [ + Result.err(`Duplicate module name "${module.name}"`), + hasWarnings, + ]; + } + seenModuleNames.add(module.name); + const resolved = yield* Effect.either( Match.value(module).pipe( Match.when({ type: "pyth-lazer" }, (m) => { From 8c4a44721a07db27f0e0fdfa8c926c96d5cc25db Mon Sep 17 00:00:00 2001 From: Hyoung-yoon Kim Date: Mon, 28 Sep 2026 16:37:53 -0400 Subject: [PATCH 2/3] refactor(bybit): one symbol per (un)subscribe frame The ByBit module now sends one symbol per subscribe or unsubscribe frame since ByBit rejects all symbols in the request if any of the symbols is invalid. This change also allows the module's `maxSymbolsPerRequest` to go above 10, which was set based on ByBit's limit on the number of arguments for its spot endpoint. --- .../src/config/bybit-module-config.ts | 3 +- .../src/modules/ticker-module/README.md | 2 +- .../ticker-module/bybit-ws-client.test.ts | 58 ++++++++++--------- .../src/modules/ticker-module/bybit.test.ts | 22 +++---- .../src/modules/ticker-module/bybit.ts | 27 +++++---- 5 files changed, 59 insertions(+), 53 deletions(-) diff --git a/workspace/data-proxy/src/config/bybit-module-config.ts b/workspace/data-proxy/src/config/bybit-module-config.ts index 8c2628f..b2909a4 100644 --- a/workspace/data-proxy/src/config/bybit-module-config.ts +++ b/workspace/data-proxy/src/config/bybit-module-config.ts @@ -13,8 +13,7 @@ export const BybitModuleConfigSchema = v.strictObject({ maxMessages: 10, maxMessagesWindow: "1 second", }), - // Bybit allows up to 10 args for each subscription request. - maxSymbolsPerRequest: v.optional(v.number(), 10), + maxSymbolsPerRequest: v.optional(v.number(), 100), // Bybit drops the socket if no ping is sent for ~20 seconds. keepaliveInterval: keepaliveIntervalField("15 seconds"), }); diff --git a/workspace/data-proxy/src/modules/ticker-module/README.md b/workspace/data-proxy/src/modules/ticker-module/README.md index 1a732ed..b382bd1 100644 --- a/workspace/data-proxy/src/modules/ticker-module/README.md +++ b/workspace/data-proxy/src/modules/ticker-module/README.md @@ -47,7 +47,7 @@ Duration fields accept a number (ms) or a duration string (`"30 seconds"`). | --- | --- | --- | --- | | Binance | `wss://stream.binance.com:9443/stream` | `5` / `"1 second"` | `streamType` (default `"bookTicker"`): `bookTicker`, `aggTrade`, `trade`, `ticker`, `miniTicker`. No keepalive. | | OKX | `wss://ws.okx.com:8443/ws/v5/public` | `480` / `"1 hour"` | `keepaliveInterval` (default `"20 seconds"`). | -| Bybit | `wss://stream.bybit.com/v5/public/spot` | `5` / `"1 second"` | `keepaliveInterval` (default `"15 seconds"`). | +| Bybit | `wss://stream.bybit.com/v5/public/spot` | `5` / `"1 second"` | `keepaliveInterval` (default `"15 seconds"`). Subscribe and unsubscribe send one symbol per frame since Bybit rejects the whole args list when any symbol is invalid. | | Lighter | `wss://mainnet.zklighter.elliot.ai/stream?readonly=true` | `180` / `"1 minute"` | `keepaliveInterval` (default `"60 seconds"`). `?readonly=true` avoids geo-restriction. Keys are numeric market ids. | ### Route diff --git a/workspace/data-proxy/src/modules/ticker-module/bybit-ws-client.test.ts b/workspace/data-proxy/src/modules/ticker-module/bybit-ws-client.test.ts index 3e20a78..acdd95c 100644 --- a/workspace/data-proxy/src/modules/ticker-module/bybit-ws-client.test.ts +++ b/workspace/data-proxy/src/modules/ticker-module/bybit-ws-client.test.ts @@ -44,26 +44,31 @@ const tickerMessage = (frame: BybitPriceFrame) => data: frame, }); +const parseFrames = (frames: string[]) => + frames.map((frame) => JSON.parse(frame) as { op: string; args: string[] }); + describe("buildSubscribeFrame / buildUnsubscribeFrame", () => { - it("produces the documented subscribe control frame", () => { - expect(JSON.parse(buildSubscribeFrame(["BTCUSDT"]))).toEqual({ - op: "subscribe", - args: ["tickers.BTCUSDT"], - }); + it("produces the documented subscribe message", () => { + expect(parseFrames(buildSubscribeFrame(["BTCUSDT"]))).toEqual([ + { + op: "subscribe", + args: ["tickers.BTCUSDT"], + }, + ]); }); - it("batches multiple symbols into one subscribe frame", () => { - expect(JSON.parse(buildSubscribeFrame(["BTCUSDT", "ETHUSDT"]))).toEqual({ - op: "subscribe", - args: ["tickers.BTCUSDT", "tickers.ETHUSDT"], - }); + it("sends one subscribe frame per symbol", () => { + expect(parseFrames(buildSubscribeFrame(["BTCUSDT", "ETHUSDT"]))).toEqual([ + { op: "subscribe", args: ["tickers.BTCUSDT"] }, + { op: "subscribe", args: ["tickers.ETHUSDT"] }, + ]); }); - it("produces the documented unsubscribe control frame", () => { - expect(JSON.parse(buildUnsubscribeFrame(["BTCUSDT", "ETHUSDT"]))).toEqual({ - op: "unsubscribe", - args: ["tickers.BTCUSDT", "tickers.ETHUSDT"], - }); + it("sends one unsubscribe frame per symbol", () => { + expect(parseFrames(buildUnsubscribeFrame(["BTCUSDT", "ETHUSDT"]))).toEqual([ + { op: "unsubscribe", args: ["tickers.BTCUSDT"] }, + { op: "unsubscribe", args: ["tickers.ETHUSDT"] }, + ]); }); }); @@ -261,7 +266,7 @@ const startService = ( }).pipe(Logger.withMinimumLogLevel(LogLevel.None)); describe("createBybitWS", () => { - it("opens the WS at the configured url and batches the subscribe on open", async () => { + it("opens the WS at the configured url and subscribes each symbol on open", async () => { const { fiber, ws: service } = await Effect.runPromise( startService(baseConfig), ); @@ -274,10 +279,10 @@ describe("createBybitWS", () => { ws.triggerOpen(); await flush(); - expect(ws.sent.length).toBe(1); - const frame = parseControl(ws.sent[0]); - expect(frame.op).toBe("subscribe"); - expect(frame.args).toEqual(["tickers.BTCUSDT", "tickers.ETHUSDT"]); + expect(ws.sent.map(parseControl)).toEqual([ + { op: "subscribe", args: ["tickers.BTCUSDT"] }, + { op: "subscribe", args: ["tickers.ETHUSDT"] }, + ]); expect(await Effect.runPromise(service.hasError())).toBe(false); await Effect.runPromise(Fiber.interrupt(fiber)); @@ -341,7 +346,7 @@ describe("createBybitWS", () => { await Effect.runPromise(Fiber.interrupt(fiber)); }); - it("subscribe batches multiple new symbols into one frame", async () => { + it("subscribes each new symbol in its own frame", async () => { const { ws: service, fiber } = await Effect.runPromise( startService(baseConfig, []), ); @@ -356,13 +361,10 @@ describe("createBybitWS", () => { ); await flush(); - expect(ws.sent.length).toBe(1); - const frame = parseControl(ws.sent[0]); - expect(frame.op).toBe("subscribe"); - expect(frame.args).toEqual([ - "tickers.BTCUSDT", - "tickers.ETHUSDT", - "tickers.SOLUSDT", + expect(ws.sent.map(parseControl)).toEqual([ + { op: "subscribe", args: ["tickers.BTCUSDT"] }, + { op: "subscribe", args: ["tickers.ETHUSDT"] }, + { op: "subscribe", args: ["tickers.SOLUSDT"] }, ]); await Effect.runPromise(Fiber.interrupt(fiber)); diff --git a/workspace/data-proxy/src/modules/ticker-module/bybit.test.ts b/workspace/data-proxy/src/modules/ticker-module/bybit.test.ts index 940e0ff..dc91c28 100644 --- a/workspace/data-proxy/src/modules/ticker-module/bybit.test.ts +++ b/workspace/data-proxy/src/modules/ticker-module/bybit.test.ts @@ -130,11 +130,11 @@ describe("BybitModuleService.handleRequest", () => { ); const ws = FakeWebSocket.instances[0]; ws.triggerOpen(); - await waitFor(() => ws.sent.length >= 1, "subscribe frame"); - expect(parseControl(ws.sent[0]).args).toEqual([ - "tickers.ETHUSDT", - "tickers.BTCUSDT", - "tickers.DOGEUSDT", + await waitFor(() => subscribeFrames(ws).length >= 3, "subscribe frames"); + expect(subscribeFrames(ws).map(parseControl)).toEqual([ + { op: "subscribe", args: ["tickers.ETHUSDT"] }, + { op: "subscribe", args: ["tickers.BTCUSDT"] }, + { op: "subscribe", args: ["tickers.DOGEUSDT"] }, ]); ws.triggerMessage(tickerMessage(ethTicker)); @@ -305,13 +305,13 @@ describe("BybitModuleService lifecycle", () => { ); const ws = FakeWebSocket.instances[0]; ws.triggerOpen(); - await waitFor(() => ws.sent.length >= 1, "subscribe frame"); + await waitFor(() => subscribeFrames(ws).length >= 2, "subscribe frames"); - expect(parseControl(ws.sent[0]).op).toBe("subscribe"); - expect(parseControl(ws.sent[0]).args).toEqual([ - "tickers.BTCUSDT", - "tickers.ETHUSDT", - ]); + const subscribes = subscribeFrames(ws).map(parseControl); + expect(subscribes.every((frame) => frame.args.length === 1)).toBe(true); + expect( + [...new Set(subscribes.map((frame) => frame.args[0]))].sort(), + ).toEqual(["tickers.BTCUSDT", "tickers.ETHUSDT"]); }); it("unsubscribes an symbol once it has been idle past symbolsCleanupTtl", async () => { diff --git a/workspace/data-proxy/src/modules/ticker-module/bybit.ts b/workspace/data-proxy/src/modules/ticker-module/bybit.ts index 4f7d589..a6daab4 100644 --- a/workspace/data-proxy/src/modules/ticker-module/bybit.ts +++ b/workspace/data-proxy/src/modules/ticker-module/bybit.ts @@ -45,17 +45,22 @@ export interface BybitPriceFrame { [key: string]: unknown; } -export const buildSubscribeFrame = (symbols: string[]): string => - JSON.stringify({ - op: "subscribe", - args: symbols.map((symbol) => `${TICKERS_TOPIC_PREFIX}${symbol}`), - }); - -export const buildUnsubscribeFrame = (symbols: string[]): string => - JSON.stringify({ - op: "unsubscribe", - args: symbols.map((symbol) => `${TICKERS_TOPIC_PREFIX}${symbol}`), - }); +/** One message per symbol. Bybit rejects every symbol in the argument when one symbol is invalid. */ +export const buildSubscribeFrame = (symbols: string[]): string[] => + symbols.map((symbol) => + JSON.stringify({ + op: "subscribe", + args: [`${TICKERS_TOPIC_PREFIX}${symbol}`], + }), + ); + +export const buildUnsubscribeFrame = (symbols: string[]): string[] => + symbols.map((symbol) => + JSON.stringify({ + op: "unsubscribe", + args: [`${TICKERS_TOPIC_PREFIX}${symbol}`], + }), + ); const codeFromJson = (json: Record): string | null => { if (typeof json.ret_code === "string") return json.ret_code; From 5cd9995c908caa84f238f59f6bd393ed48ab9a00 Mon Sep 17 00:00:00 2001 From: Hyoung-yoon Kim Date: Tue, 29 Sep 2026 08:46:51 -0400 Subject: [PATCH 3/3] chore(data-proxy): bump version to v7.0.0 --- workspace/data-proxy/package.json | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/workspace/data-proxy/package.json b/workspace/data-proxy/package.json index 08b0101..0705403 100644 --- a/workspace/data-proxy/package.json +++ b/workspace/data-proxy/package.json @@ -1,7 +1,7 @@ { "name": "@seda-protocol/data-proxy", "main": "./src/index.ts", - "version": "6.1.0", + "version": "7.0.0", "devDependencies": { "@types/big.js": "^6.2.2", "@types/ws": "^8.18.1",