Merge pull request #125 from drift-labs/master

graceful dlob publisher exit

Dockerfile

@@ -1,4 +1,4 @@
-FROM public.ecr.aws/bitnami/node:16
+FROM public.ecr.aws/bitnami/node:18
 RUN apt-get install git
 ENV NODE_ENV=production
 RUN npm install -g yarn
@@ -10,5 +10,3 @@ RUN yarn
 RUN yarn build
 
 EXPOSE 9464
-
-CMD [ "yarn", "start:all" ]

package.json

@@ -6,7 +6,7 @@
   "license": "Apache-2.0",
   "dependencies": {
     "@coral-xyz/anchor": "^0.29.0",
-    "@drift-labs/sdk": "2.74.0-beta.12",
+    "@drift-labs/sdk": "2.76.0-beta.1",
     "@opentelemetry/api": "^1.1.0",
     "@opentelemetry/auto-instrumentations-node": "^0.31.1",
     "@opentelemetry/exporter-prometheus": "^0.31.0",
@@ -24,7 +24,6 @@
     "cors": "^2.8.5",
     "dotenv": "^10.0.0",
     "express": "^4.18.2",
-    "express-rate-limit": "^6.7.0",
     "ioredis": "^5.3.2",
     "morgan": "^1.10.0",
     "prom-client": "^15.0.0",

src/core/metrics.ts

@@ -45,6 +45,7 @@ export enum HEALTH_STATUS {
 	StaleBulkAccountLoader,
 	UnhealthySlotSubscriber,
 	LivenessTesting,
+	Restart,
 }
 
 const metricsPort =
@@ -169,6 +170,10 @@ let lastHealthCheckState = true; // true = healthy, false = unhealthy
 let lastHealthCheckPerformed = Date.now() - healthCheckInterval;
 let lastTimeHealthy = Date.now() - healthCheckInterval;
 
+export const setHealthStatus = (status: HEALTH_STATUS): void => {
+	healthStatus = status;
+};
+
 /**
  * Middleware that checks if we are in general healthy by checking that the bulk account loader slot
  * has changed recently.
@@ -185,6 +190,12 @@ const handleHealthCheck = (
 	slotSource: SlotSource
 ) => {
 	return async (_req, res, _next) => {
+		if (healthStatus === HEALTH_STATUS.Restart) {
+			res.writeHead(500);
+			res.end(`NOK`);
+			return;
+		}
+
 		if (Date.now() < lastHealthCheckPerformed + healthCheckInterval) {
 			if (lastHealthCheckState) {
 				res.writeHead(200);

src/dlob-subscriber/DLOBSubscriberIO.ts

@@ -16,6 +16,7 @@ import {
 	addOracletoResponse,
 	l2WithBNToStrings,
 } from '../utils/utils';
+import { setHealthStatus, HEALTH_STATUS } from '../core/metrics';
 
 type wsMarketArgs = {
 	marketIndex: number;
@@ -34,7 +35,7 @@ const SPOT_MAKRET_STALENESS_THRESHOLD = 20 * 60 * 1000;
 
 const PERP_MARKETS_TO_SKIP_SLOT_CHECK = {
 	'mainnet-beta': [17],
-	devnet: [17, 21],
+	devnet: [17, 21, 23],
 };
 
 const SPOT_MARKETS_TO_SKIP_SLOT_CHECK = {
@@ -62,7 +63,7 @@ export class DLOBSubscriberIO extends DLOBSubscriber {
 
 	constructor(
 		config: DLOBSubscriptionConfig & {
-			env: DriftEnv,
+			env: DriftEnv;
 			redisClient: RedisClient;
 			perpMarketInfos: wsMarketInfo[];
 			spotMarketInfos: wsMarketInfo[];
@@ -209,11 +210,11 @@ export class DLOBSubscriberIO extends DLOBSubscriber {
 				this.killSwitchSlotDiffThreshold &&
 			!skipSlotCheck
 		) {
-			console.log(`Killing process due to slot diffs for market ${marketName}: 
+			console.log(`Unhealthy process due to slot diffs for market ${marketName}: 
 				dlobProvider slot: ${slot}
 				oracle slot: ${l2Formatted['oracleData']['slot']}
 			`);
-			process.exit(1);
+			setHealthStatus(HEALTH_STATUS.Restart);
 		}
 
 		// Check if times and slots are too different for market
@@ -227,11 +228,11 @@ export class DLOBSubscriberIO extends DLOBSubscriber {
 			Date.now() - lastMarketSlotAndTime.ts > MAKRET_STALENESS_THRESHOLD &&
 			!skipSlotCheck
 		) {
-			console.log(`Killing process due to same slot for market ${marketName} after > ${MAKRET_STALENESS_THRESHOLD}ms: 
+			console.log(`Unhealthy process due to same slot for market ${marketName} after > ${MAKRET_STALENESS_THRESHOLD}ms: 
 				dlobProvider slot: ${slot}
 				market slot: ${l2Formatted['marketSlot']}
 			`);
-			process.exit(1);
+			setHealthStatus(HEALTH_STATUS.Restart);
 		} else if (
 			lastMarketSlotAndTime &&
 			l2Formatted['marketSlot'] !== lastMarketSlotAndTime.slot

src/index.ts

@@ -2,7 +2,6 @@ import { program } from 'commander';
 import compression from 'compression';
 import cors from 'cors';
 import express from 'express';
-import rateLimit from 'express-rate-limit';
 import morgan from 'morgan';
 
 import { Commitment, Connection, Keypair, PublicKey } from '@solana/web3.js';
@@ -99,10 +98,6 @@ const SLOT_STALENESS_TOLERANCE =
 	parseInt(process.env.SLOT_STALENESS_TOLERANCE) || 20;
 const ROTATION_COOLDOWN = parseInt(process.env.ROTATION_COOLDOWN) || 5000;
 const useWebsocket = process.env.USE_WEBSOCKET?.toLowerCase() === 'true';
-const rateLimitCallsPerSecond = process.env.RATE_LIMIT_CALLS_PER_SECOND
-	? parseInt(process.env.RATE_LIMIT_CALLS_PER_SECOND)
-	: 1;
-const loadTestAllowed = process.env.ALLOW_LOAD_TEST?.toLowerCase() === 'true';
 const useRedis = process.env.USE_REDIS?.toLowerCase() === 'true';
 
 const logFormat =
@@ -119,23 +114,6 @@ app.use(compression());
 app.set('trust proxy', 1);
 app.use(logHttp);
 app.use(handleResponseTime);
-if (!FEATURE_FLAGS.DISABLE_RATE_LIMIT) {
-	app.use(
-		rateLimit({
-			windowMs: 1000, // 1 second
-			max: rateLimitCallsPerSecond,
-			standardHeaders: false, // Return rate limit info in the `RateLimit-*` headers
-			legacyHeaders: false, // Disable the `X-RateLimit-*` headers
-			skip: (req, _res) => {
-				if (!loadTestAllowed) {
-					return false;
-				}
-
-				return req.headers['user-agent'].includes('k6');
-			},
-		})
-	);
-}
 
 // strip off /dlob, if the request comes from exchange history server LB
 app.use((req, _res, next) => {

src/publishers/dlobPublisher.ts

@@ -40,6 +40,8 @@ import {
 } from '../dlobProvider';
 import FEATURE_FLAGS from '../utils/featureFlags';
 import { GeyserOrderSubscriber } from '../grpc/OrderSubscriberGRPC';
+import express from 'express';
+import { handleHealthCheck } from '../core/metrics';
 
 require('dotenv').config();
 const stateCommitment: Commitment = 'confirmed';
@@ -49,6 +51,9 @@ const REDIS_HOST = process.env.REDIS_HOST || 'localhost';
 const REDIS_PORT = process.env.REDIS_PORT || '6379';
 const REDIS_PASSWORD = process.env.REDIS_PASSWORD;
 
+// Set up express for health checks
+const app = express();
+
 //@ts-ignore
 const sdkConfig = initialize({ env: process.env.ENV });
 let driftClient: DriftClient;
@@ -420,6 +425,31 @@ const main = async () => {
 		recursiveFetch();
 	}
 
+	const handleStartup = async (_req, res, _next) => {
+		if (driftClient.isSubscribed && dlobProvider.size() > 0) {
+			res.writeHead(200);
+			res.end('OK');
+		} else {
+			res.writeHead(500);
+			res.end('Not ready');
+		}
+	};
+
+	app.get(
+		'/health',
+		handleHealthCheck(2 * WS_FALLBACK_FETCH_INTERVAL, dlobProvider)
+	);
+	app.get('/startup', handleStartup);
+	app.get('/', handleHealthCheck(2 * WS_FALLBACK_FETCH_INTERVAL, dlobProvider));
+	const server = app.listen(8080);
+
+	// Default keepalive is 5s, since the AWS ALB timeout is 60 seconds, clients
+	// sometimes get 502s.
+	// https://shuheikagawa.com/blog/2019/04/25/keep-alive-timeout/
+	// https://stackoverflow.com/a/68922692
+	server.keepAliveTimeout = 61 * 1000;
+	server.headersTimeout = 65 * 1000;
+
 	console.log('DLOBSubscriber Publishing Messages');
 };
 

src/serverLite.ts

@@ -2,7 +2,6 @@ import { program } from 'commander';
 import compression from 'compression';
 import cors from 'cors';
 import express from 'express';
-import rateLimit from 'express-rate-limit';
 import morgan from 'morgan';
 
 import { Commitment, Connection } from '@solana/web3.js';
@@ -65,13 +64,9 @@ const sdkConfig = initialize({ env: process.env.ENV });
 
 const stateCommitment: Commitment = 'processed';
 const serverPort = process.env.PORT || 6969;
-const rateLimitCallsPerSecond = process.env.RATE_LIMIT_CALLS_PER_SECOND
-	? parseInt(process.env.RATE_LIMIT_CALLS_PER_SECOND)
-	: 1;
 
 const SLOT_STALENESS_TOLERANCE =
 	parseInt(process.env.SLOT_STALENESS_TOLERANCE) || 20;
-const loadTestAllowed = process.env.ALLOW_LOAD_TEST?.toLowerCase() === 'true';
 
 const logFormat =
 	':remote-addr - :remote-user [:date[clf]] ":method :url HTTP/:http-version" :status :res[content-length] ":referrer" ":user-agent" :req[x-forwarded-for]';
@@ -85,21 +80,6 @@ app.use(compression());
 app.set('trust proxy', 1);
 app.use(logHttp);
 app.use(handleResponseTime);
-app.use(
-	rateLimit({
-		windowMs: 1000, // 1 second
-		max: rateLimitCallsPerSecond,
-		standardHeaders: true, // Return rate limit info in the `RateLimit-*` headers
-		legacyHeaders: false, // Disable the `X-RateLimit-*` headers
-		skip: (req, _res) => {
-			if (!loadTestAllowed) {
-				return false;
-			}
-
-			return req.headers['user-agent'].includes('k6');
-		},
-	})
-);
 
 // Metrics defined here
 const bootTimeMs = Date.now();

yarn.lock

@@ -115,10 +115,10 @@
     enabled "2.0.x"
     kuler "^2.0.0"
 
-"@drift-labs/sdk@2.74.0-beta.12":
+"@drift-labs/sdk@2.76.0-beta.1":
-  version "2.74.0-beta.12"
+  version "2.76.0-beta.1"
-  resolved "https://registry.yarnpkg.com/@drift-labs/sdk/-/sdk-2.74.0-beta.12.tgz#86a4e0657c5b425e303c4d3684ad82fe1f7919ce"
+  resolved "https://registry.yarnpkg.com/@drift-labs/sdk/-/sdk-2.76.0-beta.1.tgz#d483e0ce5938a9fd81532a0ff23a49fe939710f2"
-  integrity sha512-YFr6jnQn1ANRrAiby5gdg93dHyKVG3wJ9bTe/U3p2T6PoxDzzhZF4/1oabewodGNia5PHvTRiA3rpbmTHdvWkQ==
+  integrity sha512-NAtfLcp6tbDaR5cXaPsYvqSqclfjFo3GrvupzMImbz2b0ilgeeToW+JwHRlLgcOZaj+9JG7ECi31kHaHmItf9w==
   dependencies:
     "@coral-xyz/anchor" "0.28.1-beta.2"
     "@ellipsis-labs/phoenix-sdk" "^1.4.2"
@@ -128,6 +128,7 @@
     "@solana/web3.js" "1.73.2"
     strict-event-emitter-types "^2.0.0"
     uuid "^8.3.2"
+    zstddec "^0.1.0"
 
 "@ellipsis-labs/phoenix-sdk@^1.4.2":
   version "1.4.2"
@@ -2520,11 +2521,6 @@ eventemitter3@^4.0.7:
   resolved "https://registry.yarnpkg.com/eventemitter3/-/eventemitter3-4.0.7.tgz#2de9b68f6528d5644ef5c59526a1b4a07306169f"
   integrity sha512-8guHBZCwKnFhYdHr2ysuRWErTwhoN2X8XELRlrRwpmfeY2jjuUN4taQMsULKUVo1K4DvZl+0pgfyoysHxvmvEw==
 
-express-rate-limit@^6.7.0:
-  version "6.7.0"
-  resolved "https://registry.yarnpkg.com/express-rate-limit/-/express-rate-limit-6.7.0.tgz#6aa8a1bd63dfe79702267b3af1161a93afc1d3c2"
-  integrity sha512-vhwIdRoqcYB/72TK3tRZI+0ttS8Ytrk24GfmsxDXK9o9IhHNO5bXRiXQSExPQ4GbaE5tvIS7j1SGrxsuWs+sGA==
-
 express@^4.18.2:
   version "4.18.2"
   resolved "https://registry.yarnpkg.com/express/-/express-4.18.2.tgz#3fabe08296e930c796c19e3c516979386ba9fd59"
@@ -4507,3 +4503,8 @@ yn@3.1.1:
   version "3.1.1"
   resolved "https://registry.yarnpkg.com/yn/-/yn-3.1.1.tgz#1e87401a09d767c1d5eab26a6e4c185182d2eb50"
   integrity sha512-Ux4ygGWsu2c7isFWe8Yu1YluJmqVhxqK2cLXNQA5AcC3QfbGNpM7fu0Y8b/z16pXLnFxZYvWhd3fhBY9DLmC6Q==
+
+zstddec@^0.1.0:
+  version "0.1.0"
+  resolved "https://registry.yarnpkg.com/zstddec/-/zstddec-0.1.0.tgz#7050f3f0e0c3978562d0c566b3e5a427d2bad7ec"
+  integrity sha512-w2NTI8+3l3eeltKAdK8QpiLo/flRAr2p8AGeakfMZOXBxOg9HIu4LVDxBi81sYgVhFhdJjv1OrB5ssI8uFPoLg==