Merge pull request #1154 from VeloraDEX/fix/fluid-dex-state-generation

fix(BACK-2167): reduce number of state generations

Author: Ivan011001

package.json

@@ -1,6 +1,6 @@
 {
   "name": "@paraswap/dex-lib",
-  "version": "4.8.20",
+  "version": "4.8.21",
   "main": "build/index.js",
   "types": "build/index.d.ts",
   "repository": "https://github.com/paraswap/paraswap-dex-lib",

src/dex/balancer-v3/balancer-v3.ts

@@ -32,8 +32,8 @@ import { removeCircularStepPairs } from './utils';
 
 const MAX_UINT256 =
   '115792089237316195423570985008687907853269984665640564039457584007913129639935';
-const POOL_UPDATE_TTL = 1 * 60; // 1min
-const RATE_UPDATE_TTL = 1 * 60; // 1min
+const POOL_UPDATE_TTL = 2 * 60; // 2min
+const RATE_UPDATE_TTL = 2 * 60; // 2min
 const HOOK_UPDATE_TTL = 5 * 60; // 5min
 
 type DeepMutable<T> = {

src/dex/fluid-dex/constants.ts

@@ -1,2 +1,4 @@
 // on-chain we use 1e4 but use extra buffer to avoid reverts
 export const MIN_SWAP_LIQUIDITY = 8500n;
+
+export const RESERVE_REFRESH_INTERVAL_MS = 10 * 1000;

src/dex/fluid-dex/fluid-dex-events.test.ts

@@ -40,7 +40,7 @@ describe('FluidDex EventPool Mainnet', function () {
     const eventsToTest: Record<Address, EventMappings> = {
       '0x52aa899454998be5b000ad077a46bbe360f4e497': {
         LogOperate: [
-          21190399, 21190405, 21190420, 21190452, 21190454, 21190465, 21190506,
+          24619596, 24619595, 24619593, 24619592, 24619591, 24619590, 24619589,
         ],
       },
     };
@@ -86,7 +86,7 @@ describe('FluidDex EventPool Mainnet', function () {
 
     const eventsToTest: Record<Address, EventMappings> = {
       '0x91716C4EDA1Fb55e84Bf8b4c7085f84285c19085': {
-        LogDexDeployed: [21199929],
+        LogDexDeployed: [24611892],
       },
     };
 

src/dex/fluid-dex/fluid-dex-liquidity-proxy.ts

@@ -1,130 +1,66 @@
-import { Interface } from '@ethersproject/abi';
-import { DeepReadonly } from 'ts-essentials';
-import { Log, Logger } from '../../types';
-import { bigIntify, catchParseLogError } from '../../utils';
-import { StatefulEventSubscriber } from '../../stateful-event-subscriber';
+import { Logger } from '../../types';
+import { bigIntify, Utils } from '../../utils';
 import { IDexHelper } from '../../dex-helper/idex-helper';
-import ResolverABI from '../../abi/fluid-dex/resolver.abi.json';
-import LiquidityABI from '../../abi/fluid-dex/liquidityUserModule.abi.json';
 import {
   CommonAddresses,
   FluidDexLiquidityProxyState,
   PoolReserve,
   PoolReserveResponse,
 } from './types';
-import { Address } from '../../types';
 import { Contract } from 'ethers';
+import ResolverABI from '../../abi/fluid-dex/resolver.abi.json';
 
-export class FluidDexLiquidityProxy extends StatefulEventSubscriber<FluidDexLiquidityProxyState> {
-  handlers: {
-    [event: string]: (
-      event: any,
-      state: DeepReadonly<FluidDexLiquidityProxyState>,
-      log: Readonly<Log>,
-    ) => Promise<DeepReadonly<FluidDexLiquidityProxyState> | null>;
-  } = {};
-
-  logDecoder: (log: Log) => any;
-
-  addressesSubscribed: Address[];
-
-  readonly liquidityIface = new Interface(LiquidityABI);
-
-  readonly resolverIface = new Interface(ResolverABI);
+const STATE_CACHE_KEY = 'liquidity_proxy_state';
+const STATE_TTL_SECONDS = 600; // 10 minutes
+const LOCAL_CACHE_TTL_SECONDS = 5; // 5 seconds
 
-  resolverContract: Contract;
+export class FluidDexLiquidityProxy {
+  readonly resolverContract: Contract;
 
   constructor(
-    readonly parentName: string,
+    readonly dexKey: string,
     readonly commonAddresses: CommonAddresses,
     protected network: number,
     readonly dexHelper: IDexHelper,
-    logger: Logger,
+    readonly logger: Logger,
   ) {
-    super(parentName, 'liquidity proxy', dexHelper, logger);
-
-    this.logDecoder = (log: Log) => this.liquidityIface.parseLog(log);
-
     this.resolverContract = new Contract(
       this.commonAddresses.resolver,
       ResolverABI,
       this.dexHelper.provider,
     );
-
-    this.addressesSubscribed = [commonAddresses.liquidityProxy];
-
-    // Add handlers
-    this.handlers['LogOperate'] = this.handleOperate.bind(this);
-  }
-
-  /**
-   * Handle a trade rate change on the pool.
-   */
-  async handleOperate(
-    event: any,
-    state: DeepReadonly<FluidDexLiquidityProxyState>,
-    log: Readonly<Log>,
-  ): Promise<DeepReadonly<FluidDexLiquidityProxyState> | null> {
-    return this.generateState(log.blockNumber);
-  }
-
-  /**
-   * The function is called every time any of the subscribed
-   * addresses release log. The function accepts the current
-   * state, updates the state according to the log, and returns
-   * the updated state.
-   * @param state - Current state of event subscriber
-   * @param log - Log released by one of the subscribed addresses
-   * @returns Updates state of the event subscriber after the log
-   */
-  async processLog(
-    state: DeepReadonly<FluidDexLiquidityProxyState>,
-    log: Readonly<Log>,
-  ): Promise<DeepReadonly<FluidDexLiquidityProxyState> | null> {
-    try {
-      const event = this.logDecoder(log);
-      if (event.name in this.handlers) {
-        return await this.handlers[event.name](event, state, log);
-      }
-    } catch (e) {
-      catchParseLogError(e, this.logger);
-    }
-
-    return null;
   }
 
-  async getStateOrGenerate(
-    blockNumber: number,
-  ): Promise<FluidDexLiquidityProxyState> {
-    let state = this.getState(blockNumber);
-    if (!state) {
-      state = await this.generateState(blockNumber);
-      this.setState(state, blockNumber);
-    }
-    return state;
-  }
-
-  /**
-   * The function generates state using on-chain calls. This
-   * function is called to regenerate state if the event based
-   * system fails to fetch events and the local state is no
-   * more correct.
-   * @param blockNumber - Blocknumber for which the state should
-   * should be generated
-   * @returns state of the event subscriber at blocknumber
-   */
-  async generateState(
-    blockNumber: number,
-  ): Promise<DeepReadonly<FluidDexLiquidityProxyState>> {
+  async fetchAndSetState(blockNumber: number): Promise<void> {
     const rawResult =
       await this.resolverContract.callStatic.getAllPoolsReservesAdjusted({
         blockTag: blockNumber,
       });
 
-    const convertedResult = this.convertToFluidDexPoolState(rawResult);
-    this.logger.info(`${this.parentName}: ${this.name}: generating state...`);
+    const state = this.convertToFluidDexPoolState(rawResult);
+
+    await this.dexHelper.cache.setex(
+      this.dexKey,
+      this.network,
+      STATE_CACHE_KEY,
+      STATE_TTL_SECONDS,
+      Utils.Serialize(state),
+    );
+  }
+
+  async getState(): Promise<FluidDexLiquidityProxyState | null> {
+    const cached = await this.dexHelper.cache.getAndCacheLocally(
+      this.dexKey,
+      this.network,
+      STATE_CACHE_KEY,
+      LOCAL_CACHE_TTL_SECONDS,
+    );
 
-    return convertedResult;
+    if (cached) {
+      return Utils.Parse(cached) as FluidDexLiquidityProxyState;
+    }
+
+    return null;
   }
 
   private convertToFluidDexPoolState(

src/dex/fluid-dex/fluid-dex.ts

@@ -32,12 +32,13 @@ import { BigNumber } from 'ethers';
 import { sqrt } from './utils';
 import { FluidDexLiquidityProxy } from './fluid-dex-liquidity-proxy';
 import { FluidDexEventPool } from './fluid-dex-pool';
-import { MIN_SWAP_LIQUIDITY } from './constants';
+import { MIN_SWAP_LIQUIDITY, RESERVE_REFRESH_INTERVAL_MS } from './constants';
 
 export class FluidDex extends SimpleExchange implements IDex<FluidDexData> {
   readonly hasConstantPriceLargeAmounts = false;
   readonly needWrapNative = false;
   readonly isFeeOnTransferSupported = false;
+  readonly isStatePollingDex = true;
   public static dexKeysWithNetwork: { key: string; networks: Network[] }[] =
     getDexKeysWithNetwork(FluidDexConfig);
 
@@ -53,6 +54,8 @@ export class FluidDex extends SimpleExchange implements IDex<FluidDexData> {
 
   readonly fluidDexPoolIface: Interface;
 
+  private reserveUpdateIntervalTask?: NodeJS.Timeout;
+
   constructor(
     readonly network: Network,
     readonly dexKey: string,
@@ -124,7 +127,25 @@ export class FluidDex extends SimpleExchange implements IDex<FluidDexData> {
       }),
     );
 
-    await this.liquidityProxy.initialize(blockNumber);
+    if (!this.dexHelper.config.isSlave) {
+      void this.updateReserves();
+
+      if (!this.reserveUpdateIntervalTask) {
+        this.reserveUpdateIntervalTask = setInterval(
+          this.updateReserves.bind(this),
+          RESERVE_REFRESH_INTERVAL_MS,
+        );
+      }
+    }
+  }
+
+  private async updateReserves(): Promise<void> {
+    try {
+      const blockNumber = await this.dexHelper.provider.getBlockNumber();
+      await this.liquidityProxy.fetchAndSetState(blockNumber);
+    } catch (error) {
+      this.logger.error(`${this.dexKey}: Error updating reserves:`, error);
+    }
   }
 
   getAdapters(side: SwapSide) {
@@ -191,9 +212,14 @@ export class FluidDex extends SimpleExchange implements IDex<FluidDexData> {
 
       if (!pools.length) return null;
 
-      const liquidityProxyState = await this.liquidityProxy.getStateOrGenerate(
-        blockNumber,
-      );
+      const liquidityProxyState = await this.liquidityProxy.getState();
+
+      if (!liquidityProxyState) {
+        this.logger.warn(
+          `${this.dexKey}-${this.network}: Liquidity proxy state is not available`,
+        );
+        return null;
+      }
 
       const poolsPrices = await Promise.all(
         pools.map(async pool => {
@@ -324,8 +350,6 @@ export class FluidDex extends SimpleExchange implements IDex<FluidDexData> {
         return eventPool;
       }),
     );
-
-    await this.liquidityProxy.updatePoolState(blockNumber);
   }
 
   // Returns list of top pools based on liquidity. Max
@@ -347,7 +371,7 @@ export class FluidDex extends SimpleExchange implements IDex<FluidDexData> {
         return [];
       }
 
-      const liquidityProxyState = this.liquidityProxy.getStaleState();
+      const liquidityProxyState = await this.liquidityProxy.getState();
 
       if (!liquidityProxyState) {
         return [];
@@ -1271,4 +1295,11 @@ export class FluidDex extends SimpleExchange implements IDex<FluidDexData> {
   private abs(value: bigint): bigint {
     return value < 0 ? -value : value;
   }
+
+  releaseResources(): void {
+    if (this.reserveUpdateIntervalTask) {
+      clearInterval(this.reserveUpdateIntervalTask);
+      this.reserveUpdateIntervalTask = undefined;
+    }
+  }
 }