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", 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/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/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) => { 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;