feat(FluidDex): add liquidity proxy state refresh interval

Author: Ivan011001

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 = 30 * 1000;

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

@@ -15,8 +15,6 @@ import {
 import { Address } from '../../types';
 import { Contract } from 'ethers';
 
-const NON_POOL_THROTTLE_MS = 60 * 1000; // 1 minute
-
 export class FluidDexLiquidityProxy extends StatefulEventSubscriber<FluidDexLiquidityProxyState> {
   handlers: {
     [event: string]: (
@@ -36,9 +34,7 @@ export class FluidDexLiquidityProxy extends StatefulEventSubscriber<FluidDexLiqu
 
   resolverContract: Contract;
 
-  private poolAddresses: Set<string> = new Set();
-  private lastPoolEventBlockNumber: number = 0;
-  private lastNonPoolEventTimestamp: number = 0;
+  shouldUpdateState = false;
 
   constructor(
     readonly parentName: string,
@@ -63,35 +59,14 @@ export class FluidDexLiquidityProxy extends StatefulEventSubscriber<FluidDexLiqu
     this.handlers['LogOperate'] = this.handleOperate.bind(this);
   }
 
-  setPoolAddresses(addresses: string[]) {
-    this.poolAddresses = new Set(addresses.map(a => a.toLowerCase()));
-  }
-
   async handleOperate(
     event: any,
     state: DeepReadonly<FluidDexLiquidityProxyState>,
     log: Readonly<Log>,
   ): Promise<DeepReadonly<FluidDexLiquidityProxyState> | null> {
-    if (log.topics.length < 2) {
-      return null;
-    }
-
-    const userAddress = ('0x' + log.topics[1].slice(26)).toLowerCase();
+    this.shouldUpdateState = true;
 
-    if (this.poolAddresses.has(userAddress)) {
-      if (log.blockNumber === this.lastPoolEventBlockNumber) {
-        return null;
-      }
-      this.lastPoolEventBlockNumber = log.blockNumber;
-    } else {
-      const now = Date.now();
-      if (now - this.lastNonPoolEventTimestamp < NON_POOL_THROTTLE_MS) {
-        return null;
-      }
-      this.lastNonPoolEventTimestamp = now;
-    }
-
-    return this.generateState(log.blockNumber);
+    return null;
   }
 
   async processLog(
@@ -121,15 +96,6 @@ export class FluidDexLiquidityProxy extends StatefulEventSubscriber<FluidDexLiqu
     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>> {

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

@@ -32,7 +32,7 @@ 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;
@@ -53,6 +53,8 @@ export class FluidDex extends SimpleExchange implements IDex<FluidDexData> {
 
   readonly fluidDexPoolIface: Interface;
 
+  private reserveUpdateIntervalTask?: NodeJS.Timeout;
+
   constructor(
     readonly network: Network,
     readonly dexKey: string,
@@ -124,8 +126,28 @@ export class FluidDex extends SimpleExchange implements IDex<FluidDexData> {
       }),
     );
 
-    this.liquidityProxy.setPoolAddresses(this.pools.map(p => p.address));
     await this.liquidityProxy.initialize(blockNumber);
+
+    if (this.dexHelper.config.isSlave === true) {
+      this.reserveUpdateIntervalTask = setInterval(
+        this.updateReserves.bind(this),
+        RESERVE_REFRESH_INTERVAL_MS,
+      );
+    }
+  }
+
+  private async updateReserves(): Promise<void> {
+    if (!this.liquidityProxy.shouldUpdateState) return;
+    this.liquidityProxy.shouldUpdateState = false;
+
+    try {
+      const blockNumber = await this.dexHelper.provider.getBlockNumber();
+
+      const state = await this.liquidityProxy.generateState(blockNumber);
+      this.liquidityProxy.setState(state, blockNumber);
+    } catch (error) {
+      this.logger.error(`${this.dexKey}: Error updating reserves:`, error);
+    }
   }
 
   getAdapters(side: SwapSide) {
@@ -136,7 +158,6 @@ export class FluidDex extends SimpleExchange implements IDex<FluidDexData> {
     poolsFromFactory: readonly PoolWithDecimals[],
   ) {
     this.pools = this.generateFluidDexPoolsFromPoolsFactory(poolsFromFactory);
-    this.liquidityProxy.setPoolAddresses(this.pools.map(p => p.address));
     this.logger.info(`${this.dexKey}: pools list was updated ...`);
   }
 
@@ -1273,4 +1294,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;
+    }
+  }
 }