update health check faster

This commit is contained in:
wphan
2023-11-29 16:12:08 -08:00
parent a40bf1dfe4
commit 196e43d1b6
2 changed files with 43 additions and 63 deletions

View File

@@ -12,10 +12,9 @@ import {
commitHash, commitHash,
driftEnv, driftEnv,
endpoint, endpoint,
getSlotHealthCheckInfo,
wsEndpoint, wsEndpoint,
} from '..'; } from '..';
import { E_ALREADY_LOCKED, tryAcquire } from 'async-mutex'; import { SlotSource } from '@drift-labs/sdk';
/** /**
* Creates {count} buckets of size {increment} starting from {start}. Each bucket stores the count of values within its "size". * Creates {count} buckets of size {increment} starting from {start}. Each bucket stores the count of values within its "size".
@@ -129,7 +128,8 @@ let lastHealthCheckPerformed = Date.now() - healthCheckInterval;
* We may be hit by multiple sources performing health checks on us, so this middleware will latch * We may be hit by multiple sources performing health checks on us, so this middleware will latch
* to its health state and only update every `healthCheckInterval`. * to its health state and only update every `healthCheckInterval`.
*/ */
const handleHealthCheck = async (req, res, next) => { const handleHealthCheck = (slotSource: SlotSource) => {
return async (_req, res, _next) => {
if (Date.now() < lastHealthCheckPerformed + healthCheckInterval) { if (Date.now() < lastHealthCheckPerformed + healthCheckInterval) {
if (lastHealthCheckState) { if (lastHealthCheckState) {
res.writeHead(200); res.writeHead(200);
@@ -140,11 +140,8 @@ const handleHealthCheck = async (req, res, next) => {
// always check if last check was unhealthy (give it another chance to recover) // always check if last check was unhealthy (give it another chance to recover)
} }
const { lastSlotReceived, lastSlotReceivedMutex } = getSlotHealthCheckInfo();
try {
await tryAcquire(lastSlotReceivedMutex).runExclusive(async () => {
// healthy if slot has advanced since the last check // healthy if slot has advanced since the last check
const lastSlotReceived = slotSource.getSlot();
lastHealthCheckState = lastSlotReceived > lastHealthCheckSlot; lastHealthCheckState = lastSlotReceived > lastHealthCheckSlot;
if (!lastHealthCheckState) { if (!lastHealthCheckState) {
logger.error( logger.error(
@@ -156,7 +153,6 @@ const handleHealthCheck = async (req, res, next) => {
lastHealthCheckSlot = lastSlotReceived; lastHealthCheckSlot = lastSlotReceived;
lastHealthCheckPerformed = Date.now(); lastHealthCheckPerformed = Date.now();
});
if (!lastHealthCheckState) { if (!lastHealthCheckState) {
healthStatus = HEALTH_STATUS.UnhealthySlotSubscriber; healthStatus = HEALTH_STATUS.UnhealthySlotSubscriber;
@@ -169,24 +165,7 @@ const handleHealthCheck = async (req, res, next) => {
healthStatus = HEALTH_STATUS.Ok; healthStatus = HEALTH_STATUS.Ok;
res.writeHead(200); res.writeHead(200);
res.end('OK'); res.end('OK');
} catch (e) { };
if (e === E_ALREADY_LOCKED) {
// someone else is updating the slot, just return the latched state
if (lastHealthCheckState) {
logger.error(`DlobServer is healthy (busy mutex)`);
res.writeHead(200);
res.end('OK');
} else {
logger.error(`DlobServer is not healthy (busy mutex)`);
res.writeHead(500);
res.end(`NOK`);
}
} else {
logger.error(`Error in health check: ${e.message}`);
console.error(e);
next(e);
}
}
}; };
export { export {

View File

@@ -32,7 +32,6 @@ import {
import { logger, setLogLevel } from './utils/logger'; import { logger, setLogLevel } from './utils/logger';
import { Mutex } from 'async-mutex';
import * as http from 'http'; import * as http from 'http';
import { handleHealthCheck } from './core/metrics'; import { handleHealthCheck } from './core/metrics';
import { handleResponseTime } from './core/middleware'; import { handleResponseTime } from './core/middleware';
@@ -136,13 +135,6 @@ logger.info(`Commit: ${commitHash}`);
let MARKET_SUBSCRIBERS: SubscriberLookup = {}; let MARKET_SUBSCRIBERS: SubscriberLookup = {};
const lastSlotReceivedMutex = new Mutex();
let lastSlotReceived: number;
export const getSlotHealthCheckInfo = () => {
return { lastSlotReceived, lastSlotReceivedMutex };
};
const initializeAllMarketSubscribers = async (driftClient: DriftClient) => { const initializeAllMarketSubscribers = async (driftClient: DriftClient) => {
const markets: SubscriberLookup = {}; const markets: SubscriberLookup = {};
@@ -272,10 +264,6 @@ const main = async () => {
logger.error(e); logger.error(e);
}); });
setInterval(async () => {
lastSlotReceived = slotSource.getSlot();
}, ORDERBOOK_UPDATE_INTERVAL);
logger.info(`Initializing DLOB Provider...`); logger.info(`Initializing DLOB Provider...`);
const initDLOBProviderStart = Date.now(); const initDLOBProviderStart = Date.now();
await dlobProvider.subscribe(); await dlobProvider.subscribe();
@@ -284,9 +272,15 @@ const main = async () => {
); );
logger.info(`dlob provider size ${dlobProvider.size()}`); logger.info(`dlob provider size ${dlobProvider.size()}`);
if (useWebsocket) { if (useWebsocket) {
setInterval(async () => { const recursiveFetch = (delay = WS_FALLBACK_FETCH_INTERVAL) => {
await dlobProvider.fetch(); setTimeout(() => {
}, WS_FALLBACK_FETCH_INTERVAL); dlobProvider.fetch().then(() => {
// eslint-disable-next-line @typescript-eslint/no-unused-vars
recursiveFetch();
});
}, delay);
};
recursiveFetch();
} }
logger.info(`Initializing DLOBSubscriber...`); logger.info(`Initializing DLOBSubscriber...`);
@@ -302,7 +296,14 @@ const main = async () => {
`DLOBSubscriber initialized in ${Date.now() - initDlobSubscriberStart} ms` `DLOBSubscriber initialized in ${Date.now() - initDlobSubscriberStart} ms`
); );
logger.info(`Initializing all market subscribers...`);
const initAllMarketSubscribersStart = Date.now();
MARKET_SUBSCRIBERS = await initializeAllMarketSubscribers(driftClient); MARKET_SUBSCRIBERS = await initializeAllMarketSubscribers(driftClient);
logger.info(
`All market subscribers initialized in ${
Date.now() - initAllMarketSubscribersStart
} ms`
);
const handleStartup = async (_req, res, _next) => { const handleStartup = async (_req, res, _next) => {
if (driftClient.isSubscribed && dlobProvider.size() > 0) { if (driftClient.isSubscribed && dlobProvider.size() > 0) {
@@ -314,9 +315,9 @@ const main = async () => {
} }
}; };
app.get('/health', handleHealthCheck); app.get('/health', handleHealthCheck(slotSource));
app.get('/startup', handleStartup); app.get('/startup', handleStartup);
app.get('/', handleHealthCheck); app.get('/', handleHealthCheck(slotSource));
app.get('/orders/json/raw', async (_req, res, next) => { app.get('/orders/json/raw', async (_req, res, next) => {
try { try {