trade/trade-drift-dlob
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge pull request #41 from drift-labs/master

Browse Source 
redis fetching
This commit is contained in:
Nour Alharithi
2023-12-10 16:12:05 -08:00
committed by GitHub
parent a8f42ce14a 21c3e8ca8e
commit c79e092e24
6 changed files with 268 additions and 89 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
package.json
View File

@@ -6,7 +6,7 @@
"license": "Apache-2.0",
"dependencies": {
"@coral-xyz/anchor": "^0.29.0",
"@drift-labs/sdk": "2.49.0-beta.17",
"@drift-labs/sdk": "2.52.0-beta.0",
"@opentelemetry/api": "^1.1.0",
"@opentelemetry/auto-instrumentations-node": "^0.31.1",
"@opentelemetry/exporter-prometheus": "^0.31.0",

12
src/core/metrics.ts
View File

@@ -40,6 +40,8 @@ enum METRIC_TYPES {
gpa_fetch_duration = 'gpa_fetch_duration',
last_ws_message_received_ts = 'last_ws_message_received_ts',
account_updates_count = 'account_updates_count',
cache_hit_count = 'cache_hit_count',
cache_miss_count = 'cache_miss_count',
current_system_ts = 'current_system_ts',
health_status = 'health_status',
}
@@ -126,6 +128,14 @@ lastWsReceivedTsGauge.addCallback((obs: ObservableResult) => {
obs.observe(lastWsMsgReceivedTs, {});
});
const cacheHitCounter = meter.createCounter(METRIC_TYPES.cache_hit_count, {
description: 'Total redis cache hits',
});
const cacheMissCounter = meter.createCounter(METRIC_TYPES.cache_miss_count, {
description: 'Total redis cache misses',
});
const accountUpdatesCounter = meter.createCounter(
METRIC_TYPES.account_updates_count,
{
@@ -238,4 +248,6 @@ export {
handleHealthCheck,
setLastReceivedWsMsgTs,
accountUpdatesCounter,
cacheHitCounter,
cacheMissCounter,
};

30
src/dlob-subscriber/DLOBSubscriberIO.ts
View File

@@ -62,8 +62,8 @@ export class DLOBSubscriberIO extends DLOBSubscriber {
marketType: MarketType.PERP,
marketName: market.symbol,
depth: -1,
includeVamm: true,
numVammOrders: 100,
includeVamm: true,
updateOnChange: true,
fallbackL2Generators: [],
});
@@ -124,13 +124,39 @@ export class DLOBSubscriberIO extends DLOBSubscriber {
l2Args.marketType,
l2Args.marketIndex
);
const l2Formatted_depth100 = Object.assign({}, l2Formatted, {
bids: l2Formatted.bids.slice(0, 100),
asks: l2Formatted.asks.slice(0, 100),
});
const l2Formatted_depth20 = Object.assign({}, l2Formatted, {
bids: l2Formatted.bids.slice(0, 20),
asks: l2Formatted.asks.slice(0, 20),
});
const l2Formatted_depth5 = Object.assign({}, l2Formatted, {
bids: l2Formatted.bids.slice(0, 5),
asks: l2Formatted.asks.slice(0, 5),
});
this.redisClient.client.publish(
`orderbook_${marketType}_${l2Args.marketIndex}`,
JSON.stringify(l2Formatted)
);
this.redisClient.client.set(
`last_update_orderbook_${marketType}_${l2Args.marketIndex}`,
JSON.stringify(l2Formatted)
JSON.stringify(l2Formatted_depth100)
);
this.redisClient.client.set(
`last_update_orderbook_${marketType}_${l2Args.marketIndex}_depth_100`,
JSON.stringify(l2Formatted_depth100)
);
this.redisClient.client.set(
`last_update_orderbook_${marketType}_${l2Args.marketIndex}_depth_20`,
JSON.stringify(l2Formatted_depth20)
);
this.redisClient.client.set(
`last_update_orderbook_${marketType}_${l2Args.marketIndex}_depth_5`,
JSON.stringify(l2Formatted_depth5)
);
}
}

162
src/index.ts
View File

@@ -28,8 +28,6 @@ import {
initialize,
isVariant,
OrderSubscriber,
UserAccount,
Order,
} from '@drift-labs/sdk';
import { logger, setLogLevel } from './utils/logger';
@@ -39,7 +37,9 @@ import {
gpaFetchDurationHistogram,
handleHealthCheck,
accountUpdatesCounter,
cacheHitCounter,
setLastReceivedWsMsgTs,
cacheMissCounter,
} from './core/metrics';
import { handleResponseTime } from './core/middleware';
import {
@@ -59,8 +59,14 @@ import {
getDLOBProviderFromOrderSubscriber,
getDLOBProviderFromUserMap,
} from './dlobProvider';
import { RedisClient } from './utils/redisClient';
require('dotenv').config();
const REDIS_HOST = process.env.REDIS_HOST || 'localhost';
const REDIS_PORT = process.env.REDIS_PORT || '6379';
const REDIS_PASSWORD = process.env.REDIS_PASSWORD;
const driftEnv = (process.env.ENV || 'devnet') as DriftEnv;
const commitHash = process.env.COMMIT;
//@ts-ignore
@@ -77,6 +83,7 @@ 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 =
':remote-addr - :remote-user [:date[clf]] ":method :url HTTP/:http-version" :status :res[content-length] ":referrer" ":user-agent" :req[x-forwarded-for]';
@@ -264,6 +271,7 @@ const main = async () => {
'updateReceived',
(_pubkey: PublicKey, _slot: number, _dataType: 'raw' | 'decoded') => {
setLastReceivedWsMsgTs(Date.now());
// eslint-disable-next-line @typescript-eslint/no-unused-vars
updatesReceivedTotal++;
accountUpdatesCounter.add(1);
}
@@ -343,6 +351,13 @@ const main = async () => {
`DLOBSubscriber initialized in ${Date.now() - initDlobSubscriberStart} ms`
);
let redisClient: RedisClient;
if (useRedis) {
logger.info('Connecting to redis');
redisClient = new RedisClient(REDIS_HOST, REDIS_PORT, REDIS_PASSWORD);
await redisClient.connect();
}
logger.info(`Initializing all market subscribers...`);
const initAllMarketSubscribersStart = Date.now();
MARKET_SUBSCRIBERS = await initializeAllMarketSubscribers(driftClient);
@@ -742,6 +757,69 @@ const main = async () => {
adjustedDepth = '-1';
}
let l2Formatted: any;
if (useRedis) {
if (
!isSpot &&
`${includeVamm}`.toLowerCase() === 'true' &&
`${includeOracle}`.toLowerCase().toLowerCase() === 'true' &&
!grouping
) {
let redisL2: string;
if (parseInt(adjustedDepth as string) === 5) {
redisL2 = await redisClient.client.get(
`last_update_orderbook_perp_${normedMarketIndex}_depth_5`
);
} else if (parseInt(adjustedDepth as string) === 20) {
redisL2 = await redisClient.client.get(
`last_update_orderbook_perp_${normedMarketIndex}_depth_20`
);
} else if (parseInt(adjustedDepth as string) === 100) {
redisL2 = await redisClient.client.get(
`last_update_orderbook_perp_${normedMarketIndex}_depth_100`
);
}
if (
redisL2 &&
slotSource.getSlot() - parseInt(JSON.parse(redisL2).slot) < 10
)
l2Formatted = redisL2;
} else if (
isSpot &&
`${includeSerum}`.toLowerCase() === 'true' &&
`${includePhoenix}`.toLowerCase() === 'true' &&
`${includeOracle}`.toLowerCase() === 'true' &&
!grouping
) {
let redisL2: string;
if (parseInt(adjustedDepth as string) === 5) {
redisL2 = await redisClient.client.get(
`last_update_orderbook_spot_${normedMarketIndex}_depth_5`
);
} else if (parseInt(adjustedDepth as string) === 20) {
redisL2 = await redisClient.client.get(
`last_update_orderbook_spot_${normedMarketIndex}_depth_20`
);
} else if (parseInt(adjustedDepth as string) === 100) {
redisL2 = await redisClient.client.get(
`last_update_orderbook_spot_${normedMarketIndex}_depth_100`
);
}
if (
redisL2 &&
slotSource.getSlot() - parseInt(JSON.parse(redisL2).slot) < 10
)
l2Formatted = redisL2;
}
if (l2Formatted) {
cacheHitCounter.add(1);
res.writeHead(200);
res.end(l2Formatted);
return;
}
}
const l2 = dlobSubscriber.getL2({
marketIndex: normedMarketIndex,
marketType: normedMarketType,
@@ -783,7 +861,7 @@ const main = async () => {
res.end(JSON.stringify(l2Formatted));
} else {
// make the BNs into strings
const l2Formatted = l2WithBNToStrings(l2);
l2Formatted = l2WithBNToStrings(l2);
if (`${includeOracle}`.toLowerCase() === 'true') {
addOracletoResponse(
l2Formatted,
@@ -792,10 +870,10 @@ const main = async () => {
normedMarketIndex
);
}
}
cacheMissCounter.add(1);
res.writeHead(200);
res.end(JSON.stringify(l2Formatted));
}
} catch (err) {
next(err);
}
@@ -836,7 +914,8 @@ const main = async () => {
return;
}
const l2s = normedParams.map((normedParam) => {
const l2s = await Promise.all(
normedParams.map(async (normedParam) => {
const { normedMarketType, normedMarketIndex, error } =
validateDlobQuery(
driftClient,
@@ -859,6 +938,66 @@ const main = async () => {
adjustedDepth = '-1';
}
let l2Formatted: any;
if (useRedis) {
if (
marketType === 'perp' &&
normedParam['includeVamm'].toLowerCase() === 'true' &&
normedParam['includeOracle'].toLowerCase() === 'true' &&
!normedParam['grouping']
) {
let redisL2: string;
if (parseInt(adjustedDepth as string) === 5) {
redisL2 = await redisClient.client.get(
`last_update_orderbook_perp_${normedMarketIndex}_depth_5`
);
} else if (parseInt(adjustedDepth as string) === 20) {
redisL2 = await redisClient.client.get(
`last_update_orderbook_perp_${normedMarketIndex}_depth_20`
);
} else if (parseInt(adjustedDepth as string) === 100) {
redisL2 = await redisClient.client.get(
`last_update_orderbook_perp_${normedMarketIndex}_depth_100`
);
}
if (redisL2) {
const parsedRedisL2 = JSON.parse(redisL2);
if (slotSource.getSlot() - parseInt(parsedRedisL2.slot) < 10)
l2Formatted = parsedRedisL2;
}
} else if (
marketType === 'spot' &&
normedParam['includePhoenix'].toLowerCase() === 'true' &&
normedParam['includeSerum'].toLowerCase() === 'true' &&
!normedParam['grouping']
) {
let redisL2: string;
if (parseInt(adjustedDepth as string) === 5) {
redisL2 = await redisClient.client.get(
`last_update_orderbook_spot_${normedMarketIndex}_depth_5`
);
} else if (parseInt(adjustedDepth as string) === 20) {
redisL2 = await redisClient.client.get(
`last_update_orderbook_spot_${normedMarketIndex}_depth_20`
);
} else if (parseInt(adjustedDepth as string) === 100) {
redisL2 = await redisClient.client.get(
`last_update_orderbook_spot_${normedMarketIndex}_depth_100`
);
}
if (redisL2) {
const parsedRedisL2 = JSON.parse(redisL2);
if (slotSource.getSlot() - parseInt(parsedRedisL2.slot) < 10)
l2Formatted = parsedRedisL2;
}
}
if (l2Formatted) {
cacheHitCounter.add(1);
return l2Formatted;
}
}
const l2 = dlobSubscriber.getL2({
marketIndex: normedMarketIndex,
marketType: normedMarketType,
@@ -890,7 +1029,7 @@ const main = async () => {
parseInt(normedParam['grouping'] as string)
);
const l2Formatted = l2WithBNToStrings(
l2Formatted = l2WithBNToStrings(
groupL2(l2, groupingBN, finalDepth)
);
if (`${normedParam['includeOracle']}`.toLowerCase() === 'true') {
@@ -901,10 +1040,9 @@ const main = async () => {
normedMarketIndex
);
}
return l2Formatted;
} else {
// make the BNs into strings
const l2Formatted = l2WithBNToStrings(l2);
l2Formatted = l2WithBNToStrings(l2);
if (`${normedParam['includeOracle']}`.toLowerCase() === 'true') {
addOracletoResponse(
l2Formatted,
@@ -913,9 +1051,11 @@ const main = async () => {
normedMarketIndex
);
}
return l2Formatted;
}
});
cacheMissCounter.add(1);
return l2Formatted;
})
);
res.writeHead(200);
res.end(JSON.stringify({ l2s }));

5
src/publishers/dlobPublisher.ts
View File

@@ -36,8 +36,6 @@ import { GeyserOrderSubscriber } from '../grpc/OrderSubscriberGRPC';
require('dotenv').config();
const stateCommitment: Commitment = 'processed';
const ORDERBOOK_UPDATE_INTERVAL = 400;
const WS_FALLBACK_FETCH_INTERVAL = ORDERBOOK_UPDATE_INTERVAL * 10;
const driftEnv = (process.env.ENV || 'devnet') as DriftEnv;
const commitHash = process.env.COMMIT;
const REDIS_HOST = process.env.REDIS_HOST || 'localhost';
@@ -62,6 +60,9 @@ const useOrderSubscriber =
const useGrpc = process.env.USE_GRPC?.toLowerCase() === 'true';
const useWebsocket = process.env.USE_WEBSOCKET?.toLowerCase() === 'true';
const ORDERBOOK_UPDATE_INTERVAL = useGrpc ? 500 : 1000;
const WS_FALLBACK_FETCH_INTERVAL = 10_000;
logger.info(`RPC endpoint: ${endpoint}`);
logger.info(`WS endpoint: ${wsEndpoint}`);
logger.info(`DriftEnv: ${driftEnv}`);

8
yarn.lock
View File

@@ -115,10 +115,10 @@
enabled "2.0.x"
kuler "^2.0.0"
"@drift-labs/sdk@2.49.0-beta.17":
version "2.49.0-beta.17"
resolved "https://registry.yarnpkg.com/@drift-labs/sdk/-/sdk-2.49.0-beta.17.tgz#59d39712dcd2d0e0e6905b42c8efb743178277f3"
integrity sha512-0OLJFxDlKxIsQ32MuqVp/oHJFaeCy1IMigLmU92Zlapvjd8euR73VueiF4rGGROwrqO4IIamPgf68cD7D1V2HA==
"@drift-labs/sdk@2.52.0-beta.0":
version "2.52.0-beta.0"
resolved "https://registry.yarnpkg.com/@drift-labs/sdk/-/sdk-2.52.0-beta.0.tgz#03da2e192e3f2180238e5e464336e0fc1697c0e0"
integrity sha512-SnuXh06oJFzAvTQZ/dOjcWduhKC8rRMlyTE5Xqb3LfCcEkQ/B+XvxTItCSzAPh33qq0MMhZKjxGdqKezUxnV1w==
dependencies:
"@coral-xyz/anchor" "0.28.1-beta.2"
"@ellipsis-labs/phoenix-sdk" "^1.4.2"