init order subscriber flag

This commit is contained in:
Chris Heaney
2023-11-28 19:53:36 -05:00
parent 96dc728330
commit 9baff82262
2 changed files with 120 additions and 30 deletions

65
src/dlobProvider.ts Normal file
View File

@@ -0,0 +1,65 @@
import {DLOB, OrderSubscriber, UserAccount, UserMap} from "@drift-labs/sdk";
import {PublicKey} from "@solana/web3.js";
export type DLOBProvider = {
subscribe() : Promise<void>;
getDLOB(slot: number): Promise<DLOB>;
getUniqueAuthorities(): PublicKey[];
getUserAccounts(): Generator<{userAccount: UserAccount, publicKey: PublicKey}>;
getUserAccount(publicKey: PublicKey): UserAccount | undefined;
size(): number;
}
export function getDLOBProviderFromUserMap(userMap:UserMap) : DLOBProvider {
return {
subscribe: async () => {
await userMap.subscribe();
},
getDLOB: async (slot: number) => {
return await userMap.getDLOB(slot);
},
getUniqueAuthorities: () => {
return userMap.getUniqueAuthorities();
},
getUserAccounts: function*() {
for (const user of userMap.values()) {
yield {userAccount: user.getUserAccount(), publicKey: user.getUserAccountPublicKey()};
}
},
getUserAccount: (publicKey) => {
return userMap.get(publicKey.toString())?.getUserAccount();
},
size: () => {
return userMap.size();
}
};
}
export function getDLOBProviderFromOrderSubscriber(orderSubscriber: OrderSubscriber) : DLOBProvider {
return {
subscribe: async () => {
await orderSubscriber.subscribe();
},
getDLOB: async (slot: number) => {
return await orderSubscriber.getDLOB(slot);
},
getUniqueAuthorities: () => {
const authorities = new Set<PublicKey>();
for (const { userAccount } of orderSubscriber.usersAccounts.values()) {
authorities.add(userAccount.authority);
}
return Array.from(authorities.values());
},
getUserAccounts: function*() {
for (const [key, {userAccount}] of orderSubscriber.usersAccounts.entries()) {
yield {userAccount: userAccount, publicKey: new PublicKey(key)};
}
},
getUserAccount: (publicKey) => {
return orderSubscriber.usersAccounts.get(publicKey.toString())?.userAccount;
},
size(): number {
return orderSubscriber.usersAccounts.size;
}
};
}

View File

@@ -27,6 +27,7 @@ import {
groupL2, groupL2,
initialize, initialize,
isVariant, isVariant,
OrderSubscriber,
} from '@drift-labs/sdk'; } from '@drift-labs/sdk';
import { logger, setLogLevel } from './utils/logger'; import { logger, setLogLevel } from './utils/logger';
@@ -46,6 +47,7 @@ import {
sleep, sleep,
validateDlobQuery, validateDlobQuery,
} from './utils/utils'; } from './utils/utils';
import {DLOBProvider, getDLOBProviderFromOrderSubscriber, getDLOBProviderFromUserMap} from "./dlobProvider";
require('dotenv').config(); require('dotenv').config();
const driftEnv = (process.env.ENV || 'devnet') as DriftEnv; const driftEnv = (process.env.ENV || 'devnet') as DriftEnv;
@@ -57,6 +59,7 @@ const stateCommitment: Commitment = 'processed';
const serverPort = process.env.PORT || 6969; const serverPort = process.env.PORT || 6969;
export const ORDERBOOK_UPDATE_INTERVAL = 1000; export const ORDERBOOK_UPDATE_INTERVAL = 1000;
const useWebsocket = process.env.USE_WEBSOCKET?.toLowerCase() === 'true'; const useWebsocket = process.env.USE_WEBSOCKET?.toLowerCase() === 'true';
const useOrderSubscriber = process.env.USE_ORDER_SUBSCRIBER?.toLowerCase() === 'true';
const rateLimitCallsPerSecond = process.env.RATE_LIMIT_CALLS_PER_SECOND const rateLimitCallsPerSecond = process.env.RATE_LIMIT_CALLS_PER_SECOND
? parseInt(process.env.RATE_LIMIT_CALLS_PER_SECOND) ? parseInt(process.env.RATE_LIMIT_CALLS_PER_SECOND)
@@ -196,6 +199,7 @@ const main = async () => {
type: 'polling', type: 'polling',
accountLoader: bulkAccountLoader, accountLoader: bulkAccountLoader,
}; };
slotSource = { slotSource = {
getSlot: () => bulkAccountLoader!.getSlot(), getSlot: () => bulkAccountLoader!.getSlot(),
}; };
@@ -206,6 +210,7 @@ const main = async () => {
}; };
slotSubscriber = new SlotSubscriber(connection); slotSubscriber = new SlotSubscriber(connection);
await slotSubscriber.subscribe(); await slotSubscriber.subscribe();
slotSource = { slotSource = {
getSlot: () => slotSubscriber!.getSlot(), getSlot: () => slotSubscriber!.getSlot(),
}; };
@@ -219,6 +224,40 @@ const main = async () => {
env: driftEnv, env: driftEnv,
}); });
let dlobProvider : DLOBProvider;
if (useOrderSubscriber) {
let subscriptionConfig;
if (useWebsocket) {
subscriptionConfig = {
type: "websocket",
};
} else {
subscriptionConfig = {
type: "polling",
frequency: ORDERBOOK_UPDATE_INTERVAL
};
}
const orderSubscriber = new OrderSubscriber({
driftClient,
subscriptionConfig,
});
dlobProvider = getDLOBProviderFromOrderSubscriber(orderSubscriber);
slotSource = {
getSlot: () => orderSubscriber.getSlot(),
};
} else {
const userMap = new UserMap(
driftClient,
driftClient.userAccountSubscriptionConfig,
false
);
dlobProvider = getDLOBProviderFromUserMap(userMap);
}
const dlobCoder = DLOBOrdersCoder.create(); const dlobCoder = DLOBOrdersCoder.create();
await driftClient.subscribe(); await driftClient.subscribe();
@@ -233,18 +272,13 @@ const main = async () => {
const userStatsMap = new UserStatsMap(driftClient); const userStatsMap = new UserStatsMap(driftClient);
logger.info(`Initializing userMap...`); logger.info(`Initializing DLOB Provider...`);
const initUserMapStart = Date.now(); const initUserMapStart = Date.now();
const userMap = new UserMap( await dlobProvider.subscribe();
driftClient, logger.info(`dlob provider initialized in ${Date.now() - initUserMapStart} ms`);
driftClient.userAccountSubscriptionConfig,
false
);
await userMap.subscribe();
logger.info(`userMap initialized in ${Date.now() - initUserMapStart} ms`);
const initUserStatsMapStarts = Date.now(); const initUserStatsMapStarts = Date.now();
await userStatsMap.sync(userMap.getUniqueAuthorities()); await userStatsMap.sync(dlobProvider.getUniqueAuthorities());
logger.info( logger.info(
`userStatsMap initialized in ${Date.now() - initUserStatsMapStarts} ms` `userStatsMap initialized in ${Date.now() - initUserStatsMapStarts} ms`
); );
@@ -253,7 +287,7 @@ const main = async () => {
const initDlobSubscriberStart = Date.now(); const initDlobSubscriberStart = Date.now();
const dlobSubscriber = new DLOBSubscriber({ const dlobSubscriber = new DLOBSubscriber({
driftClient, driftClient,
dlobSource: userMap, dlobSource: dlobProvider,
slotSource, slotSource,
updateFrequency: ORDERBOOK_UPDATE_INTERVAL, updateFrequency: ORDERBOOK_UPDATE_INTERVAL,
}); });
@@ -267,7 +301,7 @@ const main = async () => {
const handleStartup = async (_req, res, _next) => { const handleStartup = async (_req, res, _next) => {
if ( if (
driftClient.isSubscribed && driftClient.isSubscribed &&
userMap.size() > 0 && dlobProvider.size() > 0 &&
userStatsMap.size() > 0 userStatsMap.size() > 0
) { ) {
res.writeHead(200); res.writeHead(200);
@@ -299,16 +333,14 @@ const main = async () => {
}); });
} }
for (const user of userMap.values()) { for (const {userAccount, publicKey} of dlobProvider.getUserAccounts()) {
const userAccount = user.getUserAccount();
for (const order of userAccount.orders) { for (const order of userAccount.orders) {
if (isVariant(order.status, 'init')) { if (isVariant(order.status, 'init')) {
continue; continue;
} }
orders.push({ orders.push({
user: user.getUserAccountPublicKey().toBase58(), user: publicKey.toBase58(),
order: order, order: order,
}); });
} }
@@ -354,9 +386,7 @@ const main = async () => {
} }
oracles.push(oracleHuman); oracles.push(oracleHuman);
} }
for (const user of userMap.values()) { for (const {userAccount, publicKey} of dlobProvider.getUserAccounts()) {
const userAccount = user.getUserAccount();
for (const order of userAccount.orders) { for (const order of userAccount.orders) {
if (isVariant(order.status, 'init')) { if (isVariant(order.status, 'init')) {
continue; continue;
@@ -394,7 +424,7 @@ const main = async () => {
} }
orders.push({ orders.push({
user: user.getUserAccountPublicKey().toBase58(), user: publicKey.toBase58(),
order: orderHuman, order: orderHuman,
}); });
} }
@@ -418,16 +448,14 @@ const main = async () => {
try { try {
const dlobOrders: DLOBOrders = []; const dlobOrders: DLOBOrders = [];
for (const user of userMap.values()) { for (const {userAccount, publicKey} of dlobProvider.getUserAccounts()) {
const userAccount = user.getUserAccount();
for (const order of userAccount.orders) { for (const order of userAccount.orders) {
if (isVariant(order.status, 'init')) { if (isVariant(order.status, 'init')) {
continue; continue;
} }
dlobOrders.push({ dlobOrders.push({
user: user.getUserAccountPublicKey(), user: publicKey,
order, order,
} as DLOBOrder); } as DLOBOrder);
} }
@@ -468,9 +496,7 @@ const main = async () => {
const dlobOrders: DLOBOrders = []; const dlobOrders: DLOBOrders = [];
for (const user of userMap.values()) { for (const {userAccount, publicKey} of dlobProvider.getUserAccounts()) {
const userAccount = user.getUserAccount();
for (const order of userAccount.orders) { for (const order of userAccount.orders) {
if (isVariant(order.status, 'init')) { if (isVariant(order.status, 'init')) {
continue; continue;
@@ -486,7 +512,7 @@ const main = async () => {
} }
dlobOrders.push({ dlobOrders.push({
user: user.getUserAccountPublicKey(), user: publicKey,
order, order,
} as DLOBOrder); } as DLOBOrder);
} }
@@ -557,10 +583,9 @@ const main = async () => {
continue; continue;
} else { } else {
if (`${includeUserStats}`.toLowerCase() === 'true') { if (`${includeUserStats}`.toLowerCase() === 'true') {
const userAccount = side.userAccount.toBase58(); const userAccount = dlobProvider.getUserAccount(side.userAccount);
await userMap.mustGet(userAccount);
const userStats = await userStatsMap.mustGet( const userStats = await userStatsMap.mustGet(
userMap.getUserAuthority(userAccount)!.toBase58() userAccount.authority.toBase58()
); );
topMakers.add([ topMakers.add([
userAccount, userAccount,