More load options - assets and throttleMs (#351)

CHANGELOG.md

@@ -3,6 +3,10 @@
 All notable changes to this project will be documented in this file.
 Version changes are pinned to SDK releases.
 
+## [1.20.5]
+
+- More load options - assets and throttleMs. ([#351](https://github.com/zetamarkets/sdk/pull/351))
+
 ## [1.20.4]
 
 - Add PNL to TradeEventV3 and PositionSize to ApplyFundingEvent. ([#344](https://github.com/zetamarkets/sdk/pull/344))

examples/basic/basic-example.ts

@@ -51,14 +51,17 @@ async function main() {
   // Airdropping SOL.
   await connection.requestAirdrop(wallet.publicKey, 100_000_000);
 
-  const loadExchangeConfig = types.defaultLoadExchangeConfig(
+  let loadExchangeConfig = types.defaultLoadExchangeConfig(
     Network.DEVNET,
     connection,
     utils.defaultCommitment(),
     0,
     true
   );
 
+  // Optionally add a throttle for startup load to prevent rate-limiting on free-tier RPCs
+  // loadExchangeConfig.throttleMs = 1000;
+
   await fetch(`${SERVER_URL}/faucet/USDC`, {
     method: "post",
     body: JSON.stringify({

src/exchange.ts

@@ -371,7 +371,11 @@ export class Exchange {
     if (this.isSetup) {
       throw "Exchange already setup";
     }
-    this._assets = assets.allAssets(loadConfig.network);
+    if (loadConfig.loadAssets) {
+      this._assets = loadConfig.loadAssets;
+    } else {
+      this._assets = assets.allAssets();
+    }
     this._provider = new anchor.AnchorProvider(
       loadConfig.connection,
       wallet instanceof types.DummyWallet ? null : wallet,
@@ -637,10 +641,23 @@ export class Exchange {
       this.subscribeOracle(this.assets, callback),
     ]);
 
-    const accFetches = (await Promise.all(allPromises)).slice(
-      0,
-      this.assets.length + 1
-    );
+    // If throttleMs is passed, do each promise slowly with a delay, else load everything at once
+    let accFetches = [];
+    if (loadConfig.throttleMs > 0) {
+      console.log(
+        `Fetching ${allPromises.length} core accounts with ${loadConfig.throttleMs}ms sleep inbetween each fetch...`
+      );
+      for (var prom of allPromises) {
+        await utils.sleep(loadConfig.throttleMs);
+        accFetches.push(await Promise.resolve(prom));
+      }
+      accFetches = accFetches.slice(0, this.assets.length + 1);
+    } else {
+      accFetches = (await Promise.all(allPromises)).slice(
+        0,
+        this.assets.length + 1
+      );
+    }
 
     let perpSyncQueueAccs: PerpSyncQueue[] = [];
     for (let i = 0; i < accFetches.length - 1; i++) {
@@ -652,26 +669,51 @@ export class Exchange {
       );
     }
 
-    await Promise.all(
-      this.assets.map(async (asset, i) => {
-        return this.getSubExchange(asset).load(
+    // If throttleMs is passed, load sequentially with some delay, else load as fast as possible
+    if (loadConfig.throttleMs > 0) {
+      for (var asset of this.assets) {
+        await utils.sleep(loadConfig.throttleMs);
+        await this.getSubExchange(asset).load(
           asset,
           this.opts,
-          [perpSyncQueueAccs[i]],
+          [perpSyncQueueAccs[this.assets.indexOf(asset)]],
           loadConfig.loadFromStore,
-          loadConfig.throttleMs,
           callback
         );
-      })
-    );
+      }
+    } else {
+      await Promise.all(
+        this.assets.map(async (asset, i) => {
+          return this.getSubExchange(asset).load(
+            asset,
+            this.opts,
+            [perpSyncQueueAccs[i]],
+            loadConfig.loadFromStore,
+            callback
+          );
+        })
+      );
+    }
 
-    await Promise.all(
-      this._assets.map(async (a) => {
-        await this.getPerpMarket(a).serumMarket.updateDecoded(
+    if (loadConfig.throttleMs > 0) {
+      console.log(
+        `Updating ${this.assets.length} markets with ${loadConfig.throttleMs}ms sleep inbetween each update...`
+      );
+      for (var asset of this.assets) {
+        await utils.sleep(loadConfig.throttleMs);
+        await this.getPerpMarket(asset).serumMarket.updateDecoded(
           this.connection as unknown as ConnectionZstd
         );
-      })
-    );
+      }
+    } else {
+      await Promise.all(
+        this._assets.map(async (a) => {
+          await this.getPerpMarket(a).serumMarket.updateDecoded(
+            this.connection as unknown as ConnectionZstd
+          );
+        })
+      );
+    }
 
     for (var se of this.getAllSubExchanges()) {
       // Only subscribe to the orderbook for assets provided in the override
@@ -700,6 +742,11 @@ export class Exchange {
     await this.updateExchangeState();
 
     this._isInitialized = true;
+
+    // An extra log because the throttleMs path is slower and logs more along the way
+    if (loadConfig.throttleMs > 0) {
+      console.log("Exchange load complete");
+    }
   }
 
   private addSubExchange(asset: Asset, subExchange: SubExchange) {

src/market.ts

@@ -73,7 +73,6 @@ export class ZetaGroupMarkets {
   public static async load(
     asset: Asset,
     opts: ConfirmOptions,
-    throttleMs: number,
     loadFromStore: boolean
   ): Promise<ZetaGroupMarkets> {
     let instance = new ZetaGroupMarkets(asset);

src/subexchange.ts

@@ -126,14 +126,13 @@ export class SubExchange {
 
   /**
    * Loads a fresh instance of the subExchange object using on chain state.
-   * @param throttle    Whether to sleep on market loading for rate limit reasons.
+   * @param throttleMs    Whether to sleep on market loading for rate limit reasons.
    */
   public async load(
     asset: Asset,
     opts: ConfirmOptions,
     fetchedAccs: any[],
     loadFromStore: boolean,
-    throttleMs = 0,
     callback?: (asset: Asset, event: EventType, slot: number, data: any) => void
   ) {
     console.info(`Loading ${assetToName(asset)} subExchange.`);
@@ -144,12 +143,7 @@ export class SubExchange {
 
     this._perpSyncQueue = fetchedAccs[0] as PerpSyncQueue;
 
-    this._markets = await ZetaGroupMarkets.load(
-      asset,
-      opts,
-      throttleMs,
-      loadFromStore
-    );
+    this._markets = await ZetaGroupMarkets.load(asset, opts, loadFromStore);
 
     Exchange.riskCalculator.updateMarginRequirements(asset);
 

src/types.ts

@@ -496,6 +496,7 @@ export interface LoadExchangeConfig {
   throttleMs: number;
   loadFromStore: boolean;
   TIFBufferSeconds: number;
+  loadAssets?: Asset[];
 }
 
 export function defaultLoadExchangeConfig(
@@ -506,7 +507,8 @@ export function defaultLoadExchangeConfig(
   loadFromStore = false,
   orderbookConnection: Connection = undefined,
   orderbookAssetSubscriptionOverride: Asset[] = undefined,
-  TIFBufferSeconds: number = undefined
+  TIFBufferSeconds: number = undefined,
+  loadAssets: Asset[] = undefined
 ): LoadExchangeConfig {
   return {
     network,
@@ -517,6 +519,7 @@ export function defaultLoadExchangeConfig(
     throttleMs,
     loadFromStore,
     TIFBufferSeconds,
+    loadAssets,
   };
 }