From d19e33e840edc309d57c903b5aa90489447851b0 Mon Sep 17 00:00:00 2001 From: wphan Date: Tue, 15 Jul 2025 10:19:55 -0700 Subject: [PATCH] refactor health checks and prettify/lint --- src/core/metrics.ts | 172 ++++++++++++++++++-------- src/index.ts | 9 +- src/publishers/dlobPublisher.ts | 7 +- src/serverLite.ts | 12 +- src/utils/constants.ts | 2 +- src/utils/offload.ts | 5 +- src/utils/tests/auctionParams.test.ts | 63 ++++++---- src/utils/types.ts | 6 +- src/utils/utils.ts | 6 +- 9 files changed, 171 insertions(+), 111 deletions(-) diff --git a/src/core/metrics.ts b/src/core/metrics.ts index d32294f..c68f809 100644 --- a/src/core/metrics.ts +++ b/src/core/metrics.ts @@ -39,6 +39,7 @@ enum METRIC_TYPES { current_system_ts = 'current_system_ts', health_status = 'health_status', incoming_requests_count = 'incoming_requests_count', + kinesis_records_sent = 'kinesis_records_sent', } export enum HEALTH_STATUS { @@ -165,72 +166,123 @@ const responseStatusCounter = meter.createCounter( } ); -const healthCheckInterval = 2000; -let lastHealthCheckSlot = -1; -let lastHealthCheckState = true; // true = healthy, false = unhealthy -let lastHealthCheckPerformed = Date.now() - healthCheckInterval; -let lastTimeHealthy = Date.now() - healthCheckInterval; +const kinesisRecordsSentGauge = meter.createObservableGauge( + METRIC_TYPES.kinesis_records_sent, + { + description: 'Number of records successfully sent to Kinesis stream', + } +); -export const setHealthStatus = (status: HEALTH_STATUS): void => { - healthStatus = status; +let kinesisRecordsSent: KinesisMetric = { + stream: '', + count: 0, +}; + +const setKinesisRecordsSent = (count: number, stream: string): void => { + kinesisRecordsSent = { + stream, + count, + }; +}; + +kinesisRecordsSentGauge.addCallback((obs: ObservableResult) => { + obs.observe(kinesisRecordsSent.count, { stream: kinesisRecordsSent.stream }); +}); + +/** + * Health check configuration + */ +const HEALTH_CHECK_CONFIG = { + CHECK_INTERVAL_MS: 2000, + // Maximum time allowed between slot updates + MAX_SLOT_STALENESS_MS: 5000, + // Minimum expected slot advancement rate (slots per second) + MIN_SLOT_RATE: 1, +} as const; + +/** + * Tracks the health state of the slot subscriber + */ +type HealthState = { + lastSlot: number; + lastSlotTimestamp: number; +}; + +const globalHealthState: HealthState = { + lastSlot: -1, + lastSlotTimestamp: Date.now(), }; /** - * Middleware that checks if we are in general healthy by checking that the bulk account loader slot - * has changed recently. - * - * 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`. - * - * A grace period is also used to report unhealthy only if we have been unhealthy beyond the grace period. + * Evaluates if the current state is healthy based on slot progression */ -const handleHealthCheck = ( - healthCheckGracePeriod: number, - slotSource: SlotSource -) => { +function evaluateHealth(currentSlot: number): { + isHealthy: boolean; + reason?: string; +} { + const now = Date.now(); + + // First health check + if (globalHealthState.lastSlot === -1) { + globalHealthState.lastSlot = currentSlot; + globalHealthState.lastSlotTimestamp = now; + return { isHealthy: true }; + } + + const timeDelta = now - globalHealthState.lastSlotTimestamp; + const slotDelta = currentSlot - globalHealthState.lastSlot; + + // Update state if slot has progressed + if (currentSlot > globalHealthState.lastSlot) { + globalHealthState.lastSlot = currentSlot; + globalHealthState.lastSlotTimestamp = now; + } + + // Check if slots are too stale + if (timeDelta > HEALTH_CHECK_CONFIG.MAX_SLOT_STALENESS_MS) { + return { + isHealthy: false, + reason: `No slot updates in ${timeDelta}ms (max ${HEALTH_CHECK_CONFIG.MAX_SLOT_STALENESS_MS}ms)`, + }; + } + + // Check if slot update rate is too low + const slotRate = (slotDelta / timeDelta) * 1000; // Convert to per second + if (slotRate < HEALTH_CHECK_CONFIG.MIN_SLOT_RATE) { + return { + isHealthy: false, + reason: `Slot update rate ${slotRate.toFixed( + 2 + )} slots/sec below minimum ${HEALTH_CHECK_CONFIG.MIN_SLOT_RATE}`, + }; + } + + return { isHealthy: true }; +} + +/** + * Middleware that checks the health of the slot subscriber. + * Health is determined by: + * 1. Regular slot updates + * 2. Minimum slot update rate + */ +const handleHealthCheck = (slotSource: SlotSource) => { return async (_req, res, _next) => { - if (healthStatus === HEALTH_STATUS.Restart) { - logger.error(`Health status: Restart`); - res.writeHead(500); - res.end(`NOK`); - return; - } + const currentSlot = slotSource.getSlot(); + const { isHealthy, reason } = evaluateHealth(currentSlot); - if (Date.now() < lastHealthCheckPerformed + healthCheckInterval) { - if (lastHealthCheckState) { - res.writeHead(200); - res.end('OK'); - lastHealthCheckPerformed = Date.now(); - return; - } - // always check if last check was unhealthy (give it another chance to recover) - } - - // healthy if slot has advanced since the last check - const lastSlotReceived = slotSource.getSlot(); - const inGracePeriod = - Date.now() - lastTimeHealthy <= healthCheckGracePeriod; - lastHealthCheckState = lastSlotReceived > lastHealthCheckSlot; - if (!lastHealthCheckState) { + if (!isHealthy) { logger.error( - `Unhealthy: lastSlot: ${lastSlotReceived}, lastHealthCheckSlot: ${lastHealthCheckSlot}, timeSinceLastCheck: ${ - Date.now() - lastHealthCheckPerformed - } ms, sinceLastTimeHealthy: ${ - Date.now() - lastTimeHealthy - } ms, inGracePeriod: ${inGracePeriod}` + `Unhealthy: ${reason}, ` + + `Details: currentSlot=${currentSlot}, ` + + `lastSlot=${globalHealthState.lastSlot}, ` + + `timeSinceLastSlot=${ + Date.now() - globalHealthState.lastSlotTimestamp + }ms` ); - } else { - lastTimeHealthy = Date.now(); - } - - lastHealthCheckSlot = lastSlotReceived; - lastHealthCheckPerformed = Date.now(); - - if (!lastHealthCheckState && !inGracePeriod) { setHealthStatus(HEALTH_STATUS.UnhealthySlotSubscriber); - res.writeHead(500); - res.end(`NOK`); + res.end('NOK'); return; } @@ -240,6 +292,15 @@ const handleHealthCheck = ( }; }; +export const setHealthStatus = (status: HEALTH_STATUS): void => { + healthStatus = status; +}; + +interface KinesisMetric { + stream: string; + count: number; +} + export { endpointResponseTimeHistogram, gpaFetchDurationHistogram, @@ -250,4 +311,5 @@ export { accountUpdatesCounter, cacheHitCounter, runtimeSpecsGauge, + setKinesisRecordsSent, }; diff --git a/src/index.ts b/src/index.ts index 0158648..4cde8f3 100644 --- a/src/index.ts +++ b/src/index.ts @@ -265,12 +265,9 @@ const main = async (): Promise => { } }; - app.get( - '/health', - handleHealthCheck(2 * WS_FALLBACK_FETCH_INTERVAL, dlobProvider) - ); + app.get('/health', handleHealthCheck(dlobProvider)); app.get('/startup', handleStartup); - app.get('/', handleHealthCheck(2 * WS_FALLBACK_FETCH_INTERVAL, dlobProvider)); + app.get('/', handleHealthCheck(dlobProvider)); app.get('/priorityFees', async (req, res, next) => { try { @@ -1065,4 +1062,4 @@ async function recursiveTryCatch(f: () => Promise) { recursiveTryCatch(() => main()); -export { commitHash, driftClient, driftEnv, endpoint, sdkConfig, wsEndpoint }; \ No newline at end of file +export { commitHash, driftClient, driftEnv, endpoint, sdkConfig, wsEndpoint }; diff --git a/src/publishers/dlobPublisher.ts b/src/publishers/dlobPublisher.ts index c4e4bc8..2de68cb 100644 --- a/src/publishers/dlobPublisher.ts +++ b/src/publishers/dlobPublisher.ts @@ -652,12 +652,9 @@ const main = async () => { res.end(await register.metrics()); }); - app.get( - '/health', - handleHealthCheck(WS_FALLBACK_FETCH_INTERVAL, dlobProvider) - ); + app.get('/health', handleHealthCheck(dlobProvider)); app.get('/startup', handleStartup); - app.get('/', handleHealthCheck(WS_FALLBACK_FETCH_INTERVAL, dlobProvider)); + app.get('/', handleHealthCheck(dlobProvider)); const server = app.listen(8080); // Default keepalive is 5s, since the AWS ALB timeout is 60 seconds, clients diff --git a/src/serverLite.ts b/src/serverLite.ts index 7fd921f..d96d6fd 100644 --- a/src/serverLite.ts +++ b/src/serverLite.ts @@ -56,8 +56,6 @@ const sdkConfig = initialize({ env: process.env.ENV }); const stateCommitment: Commitment = 'confirmed'; const serverPort = process.env.PORT || 6969; -export const ORDERBOOK_UPDATE_INTERVAL = 1000; -const WS_FALLBACK_FETCH_INTERVAL = ORDERBOOK_UPDATE_INTERVAL * 60; const SLOT_STALENESS_TOLERANCE = parseInt(process.env.SLOT_STALENESS_TOLERANCE) || 100000; @@ -155,15 +153,9 @@ const main = async (): Promise => { } }; - app.get( - '/health', - handleHealthCheck(2 * WS_FALLBACK_FETCH_INTERVAL, slotSubscriber) - ); + app.get('/health', handleHealthCheck(slotSubscriber)); app.get('/startup', handleStartup); - app.get( - '/', - handleHealthCheck(2 * WS_FALLBACK_FETCH_INTERVAL, slotSubscriber) - ); + app.get('/', handleHealthCheck(slotSubscriber)); app.get('/l3', async (req, res, next) => { try { diff --git a/src/utils/constants.ts b/src/utils/constants.ts index 67e3418..99e9d70 100644 --- a/src/utils/constants.ts +++ b/src/utils/constants.ts @@ -1,4 +1,4 @@ -import { AuctionParamArgs } from "./types"; +import { AuctionParamArgs } from './types'; export const MEASURED_ENDPOINTS = [ '/priorityFees', diff --git a/src/utils/offload.ts b/src/utils/offload.ts index 3277792..05a0482 100644 --- a/src/utils/offload.ts +++ b/src/utils/offload.ts @@ -5,6 +5,7 @@ import { } from '@aws-sdk/client-kinesis'; import { ConfiguredRetryStrategy } from '@aws-sdk/util-retry'; import { logger } from '../utils/logger'; +import { setKinesisRecordsSent } from '../core/metrics'; type EventType = 'DLOBSnapshot' | 'DLOBL3Snapshot'; @@ -83,9 +84,7 @@ export const OffloadQueue = () => { } const successCount = batch.length - (response.FailedRecordCount || 0); - logger.info( - `Successfully sent ${successCount} records to Kinesis stream: ${kinesisStream}` - ); + setKinesisRecordsSent(successCount, kinesisStream); } catch (error) { logger.error('Error processing queue:', error); } finally { diff --git a/src/utils/tests/auctionParams.test.ts b/src/utils/tests/auctionParams.test.ts index ab9ea1d..a34037d 100644 --- a/src/utils/tests/auctionParams.test.ts +++ b/src/utils/tests/auctionParams.test.ts @@ -1,11 +1,11 @@ import { describe, it, expect, jest, beforeEach } from '@jest/globals'; -import { +import { BN, - PositionDirection, - MarketType, - ZERO, - QUOTE_PRECISION, - PRICE_PRECISION + PositionDirection, + MarketType, + ZERO, + QUOTE_PRECISION, + PRICE_PRECISION, } from '@drift-labs/sdk'; import { createMarketBasedAuctionParams, @@ -16,7 +16,7 @@ import { describe('Auction Parameters Functions', () => { describe('createMarketBasedAuctionParams', () => { it('should apply market-based logic for major PERP markets (0, 1, 2)', () => { - [0, 1, 2].forEach(marketIndex => { + [0, 1, 2].forEach((marketIndex) => { const args = { marketIndex, marketType: 'perp' as any, @@ -35,7 +35,7 @@ describe('Auction Parameters Functions', () => { }); it('should apply market-based logic for minor PERP markets (>2)', () => { - [3, 4, 5, 10].forEach(marketIndex => { + [3, 4, 5, 10].forEach((marketIndex) => { const args = { marketIndex, marketType: 'perp' as any, @@ -54,7 +54,7 @@ describe('Auction Parameters Functions', () => { }); it('should apply market-based logic for SPOT markets when "marketBased" is passed', () => { - [0, 1, 2, 5, 10].forEach(marketIndex => { + [0, 1, 2, 5, 10].forEach((marketIndex) => { const args = { marketIndex, marketType: 'spot' as any, @@ -206,7 +206,7 @@ describe('Auction Parameters Functions', () => { it('should successfully calculate prices with mock L2 orderbook data', async () => { const solPrice = 160; // $160 SOL price - + mockDriftClient.getOracleDataForPerpMarket.mockReturnValue({ price: new BN(solPrice).mul(PRICE_PRECISION), }); @@ -294,7 +294,7 @@ describe('Auction Parameters Functions', () => { ); expect(result.success).toBe(true); - + // With $1,000 at $160 per SOL, we should get ~6.25 SOL const baseAmount = result.data?.marketOrderParams.baseAmount; expect(baseAmount).toBeDefined(); @@ -311,7 +311,7 @@ describe('Auction Parameters Functions', () => { it('should handle different market types correctly', async () => { const usdcPrice = 1; // $1 USDC price - + mockDriftClient.getOracleDataForSpotMarket.mockReturnValue({ price: new BN(usdcPrice).mul(PRICE_PRECISION), }); @@ -349,7 +349,7 @@ describe('Auction Parameters Functions', () => { it('should handle different directions correctly', async () => { const solPrice = 160; // $160 SOL price - + mockDriftClient.getOracleDataForPerpMarket.mockReturnValue({ price: new BN(solPrice).mul(PRICE_PRECISION), }); @@ -381,13 +381,15 @@ describe('Auction Parameters Functions', () => { ); expect(result.success).toBe(true); - expect(result.data?.marketOrderParams.direction).toBe(PositionDirection.SHORT); + expect(result.data?.marketOrderParams.direction).toBe( + PositionDirection.SHORT + ); expect(result.data?.estimatedPrices.entryPrice.gt(ZERO)).toBe(true); }); it('should handle various order sizes with L2 depth', async () => { const solPrice = 160; // $160 SOL price - + mockDriftClient.getOracleDataForPerpMarket.mockReturnValue({ price: new BN(solPrice).mul(PRICE_PRECISION), }); @@ -397,12 +399,12 @@ describe('Auction Parameters Functions', () => { if (key.includes('dlob_orderbook')) { return { bids: [ - { price: '159950000', size: '500000000' }, // $159.95, 0.5 SOL + { price: '159950000', size: '500000000' }, // $159.95, 0.5 SOL { price: '159900000', size: '1000000000' }, // $159.90, 1 SOL { price: '159850000', size: '2000000000' }, // $159.85, 2 SOL ], asks: [ - { price: '160050000', size: '500000000' }, // $160.05, 0.5 SOL + { price: '160050000', size: '500000000' }, // $160.05, 0.5 SOL { price: '160100000', size: '1000000000' }, // $160.10, 1 SOL { price: '160150000', size: '2000000000' }, // $160.15, 2 SOL ], @@ -442,9 +444,13 @@ describe('Auction Parameters Functions', () => { expect(largeResult.success).toBe(true); expect(largeResult.data?.estimatedPrices.entryPrice.gt(ZERO)).toBe(true); - + // Large orders should have higher price impact - expect(largeResult.data?.estimatedPrices.priceImpact.gte(smallResult.data?.estimatedPrices.priceImpact)).toBe(true); + expect( + largeResult.data?.estimatedPrices.priceImpact.gte( + smallResult.data?.estimatedPrices.priceImpact + ) + ).toBe(true); }); it('should handle zero oracle price scenario', async () => { @@ -600,12 +606,16 @@ describe('Auction Parameters Functions', () => { }, ]; - scenarios.forEach(scenario => { + scenarios.forEach((scenario) => { const result = createMarketBasedAuctionParams(scenario.input); - - expect(result.auctionStartPriceOffsetFrom).toBe(scenario.expectedOffset.from); - expect(result.auctionStartPriceOffset).toBe(scenario.expectedOffset.value); - + + expect(result.auctionStartPriceOffsetFrom).toBe( + scenario.expectedOffset.from + ); + expect(result.auctionStartPriceOffset).toBe( + scenario.expectedOffset.value + ); + // Verify all original parameters are preserved expect(result.marketIndex).toBe(scenario.input.marketIndex); expect(result.marketType).toBe(scenario.input.marketType); @@ -652,7 +662,7 @@ describe('Auction Parameters Functions', () => { }; const auctionParams = createMarketBasedAuctionParams(input); - + // Simulate the structure that would come from deriveMarketOrderParams const mockDerivedParams = { auctionStartPriceOffsetFrom: auctionParams.auctionStartPriceOffsetFrom, @@ -666,7 +676,8 @@ describe('Auction Parameters Functions', () => { isOracleOrder: true, }; - const formattedResponse = formatAuctionParamsForResponse(mockDerivedParams); + const formattedResponse = + formatAuctionParamsForResponse(mockDerivedParams); // Verify the complete flow expect(formattedResponse.auctionStartPriceOffsetFrom).toBe('mark'); // Market-based for major market diff --git a/src/utils/types.ts b/src/utils/types.ts index 1f514ce..c836a98 100644 --- a/src/utils/types.ts +++ b/src/utils/types.ts @@ -1,5 +1,5 @@ -import { AssetType, MarketTypeStr } from "@drift-labs/sdk"; -import { TradeOffsetPrice } from "@drift/common"; +import { AssetType, MarketTypeStr } from '@drift-labs/sdk'; +import { TradeOffsetPrice } from '@drift/common'; export type AuctionParamArgs = { // mandatory args @@ -21,4 +21,4 @@ export type AuctionParamArgs = { auctionEndPriceOffsetFrom?: TradeOffsetPrice; additionalEndPriceBuffer?: string; userOrderId?: number; -}; \ No newline at end of file +}; diff --git a/src/utils/utils.ts b/src/utils/utils.ts index 612f1a0..9f6e758 100644 --- a/src/utils/utils.ts +++ b/src/utils/utils.ts @@ -656,14 +656,16 @@ export function createMarketBasedAuctionParams( // Resolve "marketBased" values and undefined values (both should use market-based logic) const resolvedAuctionStartPriceOffsetFrom = - args.auctionStartPriceOffsetFrom === 'marketBased' || args.auctionStartPriceOffsetFrom === undefined + args.auctionStartPriceOffsetFrom === 'marketBased' || + args.auctionStartPriceOffsetFrom === undefined ? isMajorMarket ? 'mark' : 'bestOffer' : args.auctionStartPriceOffsetFrom; const resolvedAuctionStartPriceOffset = - args.auctionStartPriceOffset === 'marketBased' || args.auctionStartPriceOffset === undefined + args.auctionStartPriceOffset === 'marketBased' || + args.auctionStartPriceOffset === undefined ? isMajorMarket ? 0 : -0.1