Dev

package.json

@@ -1,6 +1,6 @@
 {
 	"dependencies": {
-		"@d8x/perpetuals-sdk": "^1.3.1",
+		"@d8x/perpetuals-sdk": "^1.3.3",
 		"axios": "^1.6.0",
 		"body-parser": "1.20.1",
 		"cors": "^2.8.5",

packages/api/package.json

@@ -25,7 +25,7 @@
 		"url": "https://github.com/D8-X/d8x-trader-backend/issues"
 	},
 	"dependencies": {
-		"@d8x/perpetuals-sdk": "^1.3.1",
+		"@d8x/perpetuals-sdk": "^1.3.3",
 		"@ethersproject/providers": "^5.7.2",
 		"ethers": "^5.7.2",
 		"redis": "^4.6.9",

packages/api/src/D8XBrokerBackendApp.ts

@@ -3,7 +3,7 @@ import WebSocket, { WebSocketServer } from "ws";
 import { IncomingMessage } from "http";
 import dotenv from "dotenv";
 import SDKInterface from "./sdkInterface";
-import { extractErrorMsg, isValidAddress,isValidPerpSymbol } from "utils";
+import { extractErrorMsg, isValidAddress, isValidPerpSymbol } from "utils";
 import { Order, PerpetualState, NodeSDKConfig, MarginAccount } from "@d8x/perpetuals-sdk";
 import EventListener from "./eventListener";
 import BrokerIntegration from "./brokerIntegration";
@@ -209,13 +209,13 @@ export default class D8XBrokerBackendApp {
 		});
 
 		this.express.get("/exchange-info", async (req: Request, res: Response) => {
-			res.setHeader('Content-Type', 'application/json');
+			res.setHeader("Content-Type", "application/json");
 			try {
 				this.lastRequestTsMs = Date.now();
 				const rsp = await this.sdk.exchangeInfo();
 				res.send(D8XBrokerBackendApp.JSONResponse("exchange-info", "", rsp));
 			} catch (err: any) {
-				console.log("Error in /exchange-info")
+				console.log("Error in /exchange-info");
 				console.log(err);
 				res.send(
 					D8XBrokerBackendApp.JSONResponse("error", "exchange-info", {
@@ -228,15 +228,18 @@ export default class D8XBrokerBackendApp {
 		this.express.get("/open-orders", async (req: Request, res: Response) => {
 			// open-orders?traderAddr=0xCafee&symbol=BTC-USD-MATIC
 			let rsp;
-			res.setHeader('Content-Type', 'application/json');
+			res.setHeader("Content-Type", "application/json");
 			try {
 				this.lastRequestTsMs = Date.now();
 				let addr: string;
 				let symbol: string | undefined;
 				if (
-					typeof req.query.traderAddr != "string" || !isValidAddress(req.query.traderAddr) ||
-					(req.query.symbol != undefined && typeof req.query.symbol != "string") ||
-					(req.query.symbol!=undefined && !isValidPerpSymbol(req.query.symbol))
+					typeof req.query.traderAddr != "string" ||
+					!isValidAddress(req.query.traderAddr) ||
+					(req.query.symbol != undefined &&
+						typeof req.query.symbol != "string") ||
+					(req.query.symbol != undefined &&
+						!isValidPerpSymbol(req.query.symbol))
 				) {
 					throw new Error("wrong arguments. Requires traderAddr and symbol");
 				} else {
@@ -260,7 +263,7 @@ export default class D8XBrokerBackendApp {
 
 		this.express.get("/trading-fee", async (req: Request, res: Response) => {
 			let rsp;
-			res.setHeader('Content-Type', 'application/json');
+			res.setHeader("Content-Type", "application/json");
 			try {
 				this.lastRequestTsMs = Date.now();
 				let traderAddr: string;
@@ -295,13 +298,14 @@ export default class D8XBrokerBackendApp {
 			// http://localhost:3001/position-risk?traderAddr=0x9d5aaB428e98678d0E645ea4AeBd25f744341a05&symbol=BTC-USD-MATIC
 			// http://localhost:3001/position-risk?traderAddr=0x9d5aaB428e98678d0E645ea4AeBd25f744341a05&symbol=MATIC
 			let rsp;
-			res.setHeader('Content-Type', 'application/json');
+			res.setHeader("Content-Type", "application/json");
 			try {
 				this.lastRequestTsMs = Date.now();
 				let addr: string;
 				let symbol: string | undefined;
 				if (
-					typeof req.query.traderAddr != "string" || !isValidAddress(req.query.traderAddr) ||
+					typeof req.query.traderAddr != "string" ||
+					!isValidAddress(req.query.traderAddr) ||
 					(req.query.symbol != undefined && typeof req.query.symbol != "string")
 				) {
 					throw new Error("wrong arguments. Requires traderAddr");
@@ -313,7 +317,7 @@ export default class D8XBrokerBackendApp {
 				}
 			} catch (err: any) {
 				const usg = "position-risk?traderAddr=0xCafee&symbol=MATIC-USD-MATIC";
-				console.log("error for position-risk:", extractErrorMsg(err))
+				console.log("error for position-risk:", extractErrorMsg(err));
 				res.send(
 					D8XBrokerBackendApp.JSONResponse("error", "position-risk", {
 						error: "error for position risk",
@@ -327,13 +331,14 @@ export default class D8XBrokerBackendApp {
 			"/max-order-size-for-trader",
 			async (req: Request, res: Response) => {
 				let rsp: string;
-				res.setHeader('Content-Type', 'application/json');
+				res.setHeader("Content-Type", "application/json");
 				try {
 					this.lastRequestTsMs = Date.now();
 					let addr: string;
 					let symbol: string;
 					if (
-						typeof req.query.traderAddr != "string" || !isValidAddress(req.query.traderAddr) ||
+						typeof req.query.traderAddr != "string" ||
+						!isValidAddress(req.query.traderAddr) ||
 						typeof req.query.symbol != "string"
 					) {
 						throw new Error(
@@ -355,7 +360,10 @@ export default class D8XBrokerBackendApp {
 						);
 					}
 				} catch (err: any) {
-					console.log("error for max-order-size-for-trader:", extractErrorMsg(err));
+					console.log(
+						"error for max-order-size-for-trader:",
+						extractErrorMsg(err),
+					);
 					const usg =
 						"max-order-size-for-trader?traderAddr=0xCafee&symbol=MATIC-USD-MATIC";
 					res.send(
@@ -375,7 +383,7 @@ export default class D8XBrokerBackendApp {
 		this.express.get(
 			"/perpetual-static-info",
 			async (req: Request, res: Response) => {
-				res.setHeader('Content-Type', 'application/json');
+				res.setHeader("Content-Type", "application/json");
 				try {
 					this.lastRequestTsMs = Date.now();
 					if (typeof req.query.symbol != "string") {
@@ -391,7 +399,10 @@ export default class D8XBrokerBackendApp {
 					);
 				} catch (err: any) {
 					const usg = "perpetual-static-info?symbol=BTC-USD-MATIC";
-					console.log("error for max-order-size-for-trader:", extractErrorMsg(err));
+					console.log(
+						"error for max-order-size-for-trader:",
+						extractErrorMsg(err),
+					);
 					res.send(
 						D8XBrokerBackendApp.JSONResponse(
 							"error",
@@ -408,7 +419,7 @@ export default class D8XBrokerBackendApp {
 
 		// see test/post.test.ts for an example
 		this.express.post("/order-digest", async (req, res) => {
-			res.setHeader('Content-Type', 'application/json');
+			res.setHeader("Content-Type", "application/json");
 			try {
 				this.lastRequestTsMs = Date.now();
 				const orders: Order[] = <Order[]>req.body.orders;
@@ -428,7 +439,7 @@ export default class D8XBrokerBackendApp {
 		});
 
 		this.express.post("/position-risk-on-collateral-action", async (req, res) => {
-			res.setHeader('Content-Type', 'application/json');
+			res.setHeader("Content-Type", "application/json");
 			try {
 				this.lastRequestTsMs = Date.now();
 				const traderAddr: string = req.body.traderAddr;
@@ -451,8 +462,11 @@ export default class D8XBrokerBackendApp {
 			} catch (err: any) {
 				const usg =
 					"{traderAddr: string, amount: number, positionRisk: <MarginAccount struct>}";
-				console.log("error for position-risk-on-collateral-action:", extractErrorMsg(err));
-				res.setHeader('Content-Type', 'application/json');
+				console.log(
+					"error for position-risk-on-collateral-action:",
+					extractErrorMsg(err),
+				);
+				res.setHeader("Content-Type", "application/json");
 				res.send(
 					D8XBrokerBackendApp.JSONResponse(
 						"error",
@@ -467,7 +481,7 @@ export default class D8XBrokerBackendApp {
 		});
 
 		this.express.get("/add-collateral", async (req: Request, res: Response) => {
-			res.setHeader('Content-Type', 'application/json');
+			res.setHeader("Content-Type", "application/json");
 			try {
 				this.lastRequestTsMs = Date.now();
 				if (
@@ -484,7 +498,7 @@ export default class D8XBrokerBackendApp {
 			} catch (err: any) {
 				const usg = "add-collateral?symbol=MATIC-USDC-USDC&amount='110.4'";
 				console.log("error for add-collateral:", extractErrorMsg(err));
-				res.setHeader('Content-Type', 'application/json');
+				res.setHeader("Content-Type", "application/json");
 				res.send(
 					D8XBrokerBackendApp.JSONResponse("error", "add-collateral", {
 						error: "error for add-collateral",
@@ -495,16 +509,14 @@ export default class D8XBrokerBackendApp {
 		});
 
 		this.express.get("/order-book", async (req: Request, res: Response) => {
-			res.setHeader('Content-Type', 'application/json');
+			res.setHeader("Content-Type", "application/json");
 			try {
-				if (
-					typeof req.query.symbol != "string"
-				) {
-					throw new Error("wrong arguments. Requires a symbol of the form WOKB-USD-WOKB.");
+				if (typeof req.query.symbol != "string") {
+					throw new Error(
+						"wrong arguments. Requires a symbol of the form WOKB-USD-WOKB.",
+					);
 				}
-				const rsp = await this.sdk.queryOrderBooks(
-					req.query.symbol,
-				);
+				const rsp = await this.sdk.queryOrderBooks(req.query.symbol);
 				res.send(D8XBrokerBackendApp.JSONResponse("order-books", "", rsp));
 			} catch (err: any) {
 				const usg = "order-book?symbol=WOKB-USD-WOKB";
@@ -519,7 +531,7 @@ export default class D8XBrokerBackendApp {
 		});
 
 		this.express.get("/remove-collateral", async (req: Request, res: Response) => {
-			res.setHeader('Content-Type', 'application/json');
+			res.setHeader("Content-Type", "application/json");
 			try {
 				this.lastRequestTsMs = Date.now();
 				if (
@@ -546,7 +558,7 @@ export default class D8XBrokerBackendApp {
 		});
 
 		this.express.get("/available-margin", async (req: Request, res: Response) => {
-			res.setHeader('Content-Type', 'application/json');
+			res.setHeader("Content-Type", "application/json");
 
 			try {
 				this.lastRequestTsMs = Date.now();
@@ -576,7 +588,7 @@ export default class D8XBrokerBackendApp {
 		});
 
 		this.express.get("/cancel-order", async (req: Request, res: Response) => {
-			res.setHeader('Content-Type', 'application/json');
+			res.setHeader("Content-Type", "application/json");
 			try {
 				this.lastRequestTsMs = Date.now();
 				if (

packages/api/src/eventListener.ts

@@ -39,6 +39,7 @@ import {
 } from "utils/src/wsTypes";
 import SturdyWebSocket from "sturdy-websocket";
 import { Logger } from "winston";
+import { TrackedWebsocketsProvider } from "./providers";
 
 /**
  * Class that listens to blockchain events on
@@ -151,6 +152,12 @@ export default class EventListener extends IndexPriceInterface {
 		this.resetRPCWebsocket(this.wsRPC);
 		sdkInterface.registerObserver(this);
 		this.lastBlockChainEventTs = Date.now();
+
+		// run _update only for the first time to set fundingRates and
+		// openInterest
+		await this._update("Initial Update Call", true);
+
+		// Set to initialized only after all initializations are done
 		this.isInitialized = true;
 	}
 
@@ -172,14 +179,18 @@ export default class EventListener extends IndexPriceInterface {
 		}
 		this.rpcResetting = true;
 
-		this.stopListening();
 		this.wsRPC = newWsRPC;
 		this.logger.info("resetting WS RPC", { newWsRPC });
 
+		// Close the underlying websockets connection without issuing
+		// eth_unsubscribe calls. We do this because ethers implementation has a
+		// bug where it sends async eth_unsubscribe requests from within
+		// synchronous WebsocketProvider._stopEvent call. Therefore if we
+		// attempt to remove event listeners before closing the websocket
+		// connection, it will crash the service with unhandled promise error
+		// while attempting to send(eth_unsubscribe) on a closed connection.
 		if (this.currentWSRpcProvider !== undefined) {
-			// do not call this.currentWSRpcProvider.destroy(); since it messes
-			// up ws state and causes panic
-			this.currentWSRpcProvider.removeAllListeners();
+			this.currentWSRpcProvider.destroy();
 			this.logger.info("old rpc provider destroyed");
 		}
 
@@ -191,7 +202,7 @@ export default class EventListener extends IndexPriceInterface {
 
 		// Attempt to establish a ws connection to new RPC
 		this.logger.info("creating new websocket rpc provider");
-		this.currentWSRpcProvider = new providers.WebSocketProvider(this.wsConn!);
+		this.currentWSRpcProvider = new TrackedWebsocketsProvider(this.wsConn!);
 
 		// On provider error - retry after short cooldown
 		this.currentWSRpcProvider.on("error", (error: Error) => () => {
@@ -446,12 +457,17 @@ export default class EventListener extends IndexPriceInterface {
 
 	/**
 	 * Handles updates from sdk interface
-	 * @param msg from observable
+	 * @param msg
+	 * @param firstTimeUpdate - if true, allow to run the _update even if not
+	 * initialized
+	 * @returns
 	 */
-	protected async _update(msg: String) {
-		// we receive a message from the observable sdk
-		// on update exchange info; we update price info and inform subscribers
-		if (!this.isInitialized) {
+	protected async _update(msg: String, firstTimeUpdate: boolean = false) {
+		// we receive a message from the observable sdk on update exchange info;
+		// we update price info and inform subscribers. Whenever this is a first
+		// time update - allow it to pass through to initialize the funding rate
+		// and openInterest
+		if (!this.isInitialized && !firstTimeUpdate) {
 			return;
 		}
 		console.log("received update from sdkInterface", msg);
@@ -471,6 +487,12 @@ export default class EventListener extends IndexPriceInterface {
 					perp.markPrice,
 					perp.indexPrice,
 				);
+
+				console.log(`[_update] setting fundingRate and openInterest`, {
+					fundingRate: perp.currentFundingRateBps / 1e4,
+					openInterest: perp.openInterestBC,
+					perpetualId: perp.id,
+				});
 			}
 		}
 	}
@@ -611,7 +633,7 @@ export default class EventListener extends IndexPriceInterface {
 	 * @param symbol order book symbol
 	 */
 	private addOrderBookEventHandlers(symbol: string) {
-		const provider = new providers.WebSocketProvider(this.wsRPC);
+		const provider = this.currentWSRpcProvider!;
 		this.orderBookContracts[symbol] = new Contract(
 			this.traderInterface.getOrderBookAddress(symbol),
 			this.traderInterface.getABI("lob")!,
@@ -750,6 +772,11 @@ export default class EventListener extends IndexPriceInterface {
 		fMarkPricePremium: BigNumber,
 		fSpotIndexPrice: BigNumber,
 	): void {
+		if (!this.isInitialized) {
+			console.log("onUpdateMarkPrice: eventListener not initialized");
+			return;
+		}
+
 		this.lastBlockChainEventTs = Date.now();
 
 		const hash =
@@ -780,6 +807,7 @@ export default class EventListener extends IndexPriceInterface {
 
 		// update data in sdkInterface's exchangeInfo
 		const fundingRate = this.fundingRate.get(perpetualId) || 0;
+
 		const oi = this.openInterest.get(perpetualId) || 0;
 		const symbol = this.symbolFromPerpetualId(perpetualId);
 
@@ -818,7 +846,13 @@ export default class EventListener extends IndexPriceInterface {
 		newMarkPrice: number,
 		newIndexPrice: number,
 	) {
+		if (!this.isInitialized) {
+			console.log("updateMarkPrice: eventListener not initialized");
+			return;
+		}
+
 		const fundingRate = this.fundingRate.get(perpetualId) || 0;
+
 		const oi = this.openInterest.get(perpetualId) || 0;
 		const symbol = this.symbolFromPerpetualId(perpetualId);
 		const obj: PriceUpdate = {