diff --git a/.gitignore b/.gitignore index 3f820bc..acde1c6 100644 --- a/.gitignore +++ b/.gitignore @@ -142,4 +142,4 @@ src/**.js.map lib src/playground.ts -dump.rdb \ No newline at end of file +*dump.rdb \ No newline at end of file diff --git a/src/core/metrics.ts b/src/core/metrics.ts index 4eaabca..6032d9e 100644 --- a/src/core/metrics.ts +++ b/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; } diff --git a/src/index.ts b/src/index.ts index 891e3ac..e7d1a5a 100644 --- a/src/index.ts +++ b/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,