ByBit one symbol per frame

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",

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

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"),
 });

workspace/data-proxy/src/config/config-parser.ts

@@ -498,8 +498,17 @@ export const parseConfig = (
 		}
 
 		const modules: Modules[] = [];
+		const seenModuleNames = new Set<string>();
 
 		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) => {

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

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));

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 () => {

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, unknown>): string | null => {
 	if (typeof json.ret_code === "string") return json.ret_code;