better health check

.gitignore

@@ -142,4 +142,4 @@ src/**.js.map
 lib
 src/playground.ts
 
-dump.rdb
+*dump.rdb

src/core/metrics.ts

@@ -7,7 +7,13 @@ import {
 	MeterProvider,
 	View,
 } from '@opentelemetry/sdk-metrics-base';
-import { commitHash, driftEnv, endpoint, wsEndpoint } from '..';
+import {
+	commitHash,
+	driftEnv,
+	endpoint,
+	getSlotHealthCheckInfo,
+	wsEndpoint,
+} from '..';
 
 /**
  * Creates {count} buckets of size {increment} starting from {start}. Each bucket stores the count of values within its "size".
@@ -110,15 +116,35 @@ const responseStatusCounter = meter.createCounter(
 	}
 );
 
-const startupTime = Date.now();
+const healthCheckInterval = 2000;
+let lastHealthCheckSlot = -1;
+let lastHealthCheckSlotUpdated = Date.now();
 const handleHealthCheck = async (req, res, next) => {
+	const { lastSlotReceived, lastSlotReceivedMutex } = getSlotHealthCheckInfo();
+
 	try {
 		if (req.url === '/health' || req.url === '/') {
-			if (Date.now() > startupTime + 60 * 1000) {
-				healthStatus = HEALTH_STATUS.LivenessTesting;
+			// check if a slot was received recently
+			let healthySlotSubscriber = false;
+			await lastSlotReceivedMutex.runExclusive(async () => {
+				const slotChanged = lastSlotReceived > lastHealthCheckSlot;
+				const slotChangedRecently =
+					Date.now() - lastHealthCheckSlotUpdated < healthCheckInterval;
+				healthySlotSubscriber = slotChanged || slotChangedRecently;
+				logger.debug(
+					`Slotsubscriber health check: lastSlotReceived: ${lastSlotReceived}, lastHealthCheckSlot: ${lastHealthCheckSlot}, slotChanged: ${slotChanged}, slotChangedRecently: ${slotChangedRecently}`
+				);
+				if (slotChanged) {
+					lastHealthCheckSlot = lastSlotReceived;
+					lastHealthCheckSlotUpdated = Date.now();
+				}
+			});
+			if (!healthySlotSubscriber) {
+				healthStatus = HEALTH_STATUS.UnhealthySlotSubscriber;
+				logger.error(`SlotSubscriber is not healthy`);
 
 				res.writeHead(500);
-				res.end('Testing liveness test fail');
+				res.end(`SlotSubscriber is not healthy`);
 				return;
 			}
 

src/index.ts

@@ -42,6 +42,7 @@ import {
 } from './utils/utils';
 import { handleResponseTime } from './core/middleware';
 import { handleHealthCheck } from './core/metrics';
+import { Mutex } from 'async-mutex';
 
 require('dotenv').config();
 const driftEnv = (process.env.ENV || 'devnet') as DriftEnv;
@@ -114,6 +115,13 @@ logger.info(`Commit:       ${commitHash}`);
 
 let MARKET_SUBSCRIBERS: SubscriberLookup = {};
 
+const lastSlotReceivedMutex = new Mutex();
+let lastSlotReceived: number;
+
+export const getSlotHealthCheckInfo = () => {
+	return { lastSlotReceived, lastSlotReceivedMutex };
+};
+
 const initializeAllMarketSubscribers = async (driftClient: DriftClient) => {
 	const markets: SubscriberLookup = {};
 
@@ -162,6 +170,7 @@ const main = async () => {
 		programID: clearingHousePublicKey,
 		accountSubscription: {
 			type: 'websocket',
+			resubTimeoutMs: 60000,
 		},
 		env: driftEnv,
 		userStats: true,
@@ -184,6 +193,11 @@ const main = async () => {
 	});
 
 	await slotSubscriber.subscribe();
+	slotSubscriber.eventEmitter.on('newSlot', async (slot: number) => {
+		await lastSlotReceivedMutex.runExclusive(async () => {
+			lastSlotReceived = slot;
+		});
+	});
 
 	const userMap = new UserMap(
 		driftClient,