Merge pull request #1 from drift-labs/luke/dpe-1270-ui-network-bandwidth-improvement-1

Draft for changes to DlobSubscriber

.vscode/launch.json

@@ -0,0 +1,16 @@
+{
+    // Use IntelliSense to learn about possible attributes.
+    // Hover to view descriptions of existing attributes.
+    // For more information, visit: https://go.microsoft.com/fwlink/?linkid=830387
+    "version": "0.2.0",
+    "configurations": [
+        {
+            "type": "node",
+            "request": "attach",
+            "name": "Debug",
+            "port": 2230,
+            "timeout": 3000,
+            "restart": true,
+        },
+    ]
+}

package.json

@@ -1,52 +1,53 @@
 {
-	"name": "@drift-labs/dlob-server",
+  "name": "@drift-labs/dlob-server",
-	"version": "0.1.0",
+  "version": "0.1.0",
-	"author": "wphan",
+  "author": "wphan",
-	"main": "lib/index.js",
+  "main": "lib/index.js",
-	"license": "Apache-2.0",
+  "license": "Apache-2.0",
-	"dependencies": {
+  "dependencies": {
-		"@drift-labs/sdk": "2.30.0",
+    "@drift-labs/sdk": "2.30.0",
-		"@opentelemetry/api": "^1.1.0",
+    "@opentelemetry/api": "^1.1.0",
-		"@opentelemetry/auto-instrumentations-node": "^0.31.1",
+    "@opentelemetry/auto-instrumentations-node": "^0.31.1",
-		"@opentelemetry/exporter-prometheus": "^0.31.0",
+    "@opentelemetry/exporter-prometheus": "^0.31.0",
-		"@opentelemetry/sdk-node": "^0.31.0",
+    "@opentelemetry/sdk-node": "^0.31.0",
-		"@project-serum/anchor": "^0.19.1-beta.1",
+    "@project-serum/anchor": "^0.19.1-beta.1",
-		"@project-serum/serum": "^0.13.65",
+    "@project-serum/serum": "^0.13.65",
-		"@solana/web3.js": "^1.22.0",
+    "@solana/web3.js": "^1.22.0",
-		"async-mutex": "^0.4.0",
+    "async-mutex": "^0.4.0",
-		"commander": "^9.4.0",
+    "commander": "^9.4.0",
-		"compression": "^1.7.4",
+    "compression": "^1.7.4",
-		"cors": "^2.8.5",
+    "cors": "^2.8.5",
-		"dotenv": "^10.0.0",
+    "dotenv": "^10.0.0",
-		"express": "^4.18.2",
+    "express": "^4.18.2",
-		"express-rate-limit": "^6.7.0",
+    "express-rate-limit": "^6.7.0",
-		"morgan": "^1.10.0",
+    "morgan": "^1.10.0",
-		"response-time": "^2.3.2",
+    "response-time": "^2.3.2",
-		"typescript": "4.5.4",
+    "typescript": "4.5.4",
-		"winston": "^3.8.1"
+    "winston": "^3.8.1"
-	},
+  },
-	"devDependencies": {
+  "devDependencies": {
-		"@typescript-eslint/eslint-plugin": "^4.28.0",
+    "@typescript-eslint/eslint-plugin": "^4.28.0",
-		"@typescript-eslint/parser": "^4.28.0",
+    "@typescript-eslint/parser": "^4.28.0",
-		"eslint": "^7.29.0",
+    "eslint": "^7.29.0",
-		"eslint-config-prettier": "^8.3.0",
+    "eslint-config-prettier": "^8.3.0",
-		"eslint-plugin-prettier": "^3.4.0",
+    "eslint-plugin-prettier": "^3.4.0",
-		"husky": "^7.0.4",
+    "husky": "^7.0.4",
-		"prettier": "^2.4.1",
+    "prettier": "^2.4.1",
-		"ts-node": "^10.9.1"
+    "ts-node": "^10.9.1"
-	},
+  },
-	"scripts": {
+  "scripts": {
-		"prepare": "husky install",
+    "prepare": "husky install",
-		"build": "yarn clean && tsc",
+    "build": "yarn clean && tsc",
-		"clean": "rm -rf lib",
+    "clean": "rm -rf lib",
-		"start": "node lib/index.js",
+    "start": "node lib/index.js",
-		"dev": "ts-node src/index.ts",
+    "dev": "ts-node src/index.ts",
-		"dev:inspect": "yarn build && node --inspect ./lib/index.js",
+    "dev:inspect": "yarn build && node --inspect ./lib/index.js",
-		"example": "ts-node example/client.ts",
+    "dev:debug": "yarn build && node --inspect-brk --inspect=2230 ./lib/index.js",
-		"exampleWithSlot": "ts-node example/clientWithSlot.ts",
+    "example": "ts-node example/client.ts",
-		"prettify": "prettier --check './src/**/*.ts'",
+    "exampleWithSlot": "ts-node example/clientWithSlot.ts",
-		"prettify:fix": "prettier --write './src/**/*.ts'",
+    "prettify": "prettier --check './src/**/*.ts'",
-		"lint": "eslint . --ext ts --quiet",
+    "prettify:fix": "prettier --write './src/**/*.ts'",
-		"lint:fix": "eslint . --ext ts --fix"
+    "lint": "eslint . --ext ts --quiet",
-	}
+    "lint:fix": "eslint . --ext ts --fix"
+  }
 }

src/index.ts

@@ -24,7 +24,9 @@ import {
   PerpMarkets,
   DLOBSubscriber,
   MarketType,
-  isVariant,
+  SpotMarketConfig,
+  PhoenixSubscriber,
+  SerumSubscriber,
 } from "@drift-labs/sdk";
 
 import { Mutex } from "async-mutex";
@@ -216,6 +218,83 @@ const endpointResponseTimeHistogram = meter.createHistogram(
   }
 );
 
+const getPhoenixSubscriber = (
+  driftClient: DriftClient,
+  marketConfig: SpotMarketConfig,
+  accountLoader: BulkAccountLoader
+) => {
+  return new PhoenixSubscriber({
+    connection: driftClient.connection,
+    programId: new PublicKey(sdkConfig.PHOENIX),
+    marketAddress: marketConfig.phoenixMarket,
+    accountSubscription: {
+      type: "polling",
+      accountLoader,
+    },
+  });
+};
+
+const getSerumSubscriber = (
+  driftClient: DriftClient,
+  marketConfig: SpotMarketConfig,
+  accountLoader: BulkAccountLoader
+) => {
+  return new SerumSubscriber({
+    connection: driftClient.connection,
+    programId: new PublicKey(sdkConfig.SERUM_V3),
+    marketAddress: marketConfig.serumMarket,
+    accountSubscription: {
+      type: "polling",
+      accountLoader,
+    },
+  });
+};
+
+type SubscriberLookup = {
+  [marketIndex: number]: {
+    phoenix?: PhoenixSubscriber;
+    serum?: SerumSubscriber;
+  };
+};
+
+let MARKET_SUBSCRIBERS: SubscriberLookup = {};
+
+const initializeAllMarketSubscribers = async (
+  driftClient: DriftClient,
+  bulkAccountLoader: BulkAccountLoader
+) => {
+  const markets: SubscriberLookup = {};
+
+  for (const market of sdkConfig.SPOT_MARKETS) {
+    markets[market.marketIndex] = {
+      phoenix: undefined,
+      serum: undefined,
+    };
+
+    if (market.phoenixMarket) {
+      const phoenixSubscriber = getPhoenixSubscriber(
+        driftClient,
+        market,
+        bulkAccountLoader
+      );
+      await phoenixSubscriber.subscribe();
+      markets[market.marketIndex].phoenix = phoenixSubscriber;
+    }
+
+    if (market.serumMarket) {
+      const serumSubscriber = getSerumSubscriber(
+        driftClient,
+        market,
+        bulkAccountLoader
+      );
+      await serumSubscriber.subscribe();
+      markets[market.marketIndex].serum = serumSubscriber;
+    }
+  }
+
+  return markets;
+};
+
 const main = async () => {
   const wallet = getWallet();
   const clearingHousePublicKey = new PublicKey(sdkConfig.DRIFT_PROGRAM_ID);
@@ -313,6 +392,11 @@ const main = async () => {
     });
   });
 
+  MARKET_SUBSCRIBERS = await initializeAllMarketSubscribers(
+    driftClient,
+    bulkAccountLoader
+  );
+
   // start http server listening to /health endpoint using http package
   app.get("/health", handleResponseTime, async (req, res, next) => {
     try {
@@ -686,8 +770,15 @@ const main = async () => {
 
   app.get("/l2", handleResponseTime, async (req, res, next) => {
     try {
-      const { marketName, marketIndex, marketType, depth, includeVamm } =
+      const {
-        req.query;
+        marketName,
+        marketIndex,
+        marketType,
+        depth,
+        includeVamm,
+        includePhoenix,
+        includeSerum,
+      } = req.query;
 
       const { normedMarketType, normedMarketIndex, error } = validateDlobQuery(
         marketType as string,
@@ -699,13 +790,21 @@ const main = async () => {
         return;
       }
 
-      const l2 = dlobSubscriber.getL2({
+      const isSpot = getVariant(normedMarketType);
+
+      const l2 = await dlobSubscriber.getL2({
         marketIndex: normedMarketIndex,
         marketType: normedMarketType,
         depth: depth ? parseInt(depth as string) : 10,
-        includeVamm: includeVamm
+        includeVamm: `${includeVamm}`.toLowerCase() === "true",
-          ? (includeVamm as string).toLowerCase() === "true"
+        fallbackL2Generators: isSpot
-          : false,
+          ? [
+              `${includePhoenix}`.toLowerCase() === "true" &&
+                MARKET_SUBSCRIBERS[normedMarketIndex].phoenix,
+              `${includeSerum}`.toLowerCase() === "true" &&
+                MARKET_SUBSCRIBERS[normedMarketIndex].serum,
+            ].filter((a) => !!a)
+          : [],
       });
 
       for (const key of Object.keys(l2)) {