Feat/back 1667

package.json

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

src/dex-helper/dummy-dex-helper.ts

@@ -299,6 +299,8 @@ export class DummyDexHelper implements IDexHelper {
   getLogger: LoggerConstructor;
   web3Provider: Web3;
   getTokenUSDPrice: (token: Token, amount: bigint) => Promise<number>;
+  dexStatePublisher: any; // FIXME
+  dexStateSubscriber: any; // FIXME
 
   constructor(network: number, rpcUrl?: string) {
     this.config = new ConfigHelper(false, generateConfig(network), 'is');

src/dex-helper/idex-helper.ts

@@ -24,4 +24,6 @@ export interface IDexHelper {
   blockManager: IBlockManager;
   getLogger: LoggerConstructor;
   getTokenUSDPrice: (token: Token, amount: bigint) => Promise<number>;
+  dexStateSubscriber: any; // temp ?
+  dexStatePublisher: any; // temp ?
 }

src/dex-utils.ts

@@ -0,0 +1,5 @@
+export class DexPoolNotFoundError extends Error {
+  constructor(dexKey: string, poolIdentifier: string) {
+    super(`Pool ${poolIdentifier} not found for DEX ${dexKey}`);
+  }
+}

src/dex/idex.ts

@@ -207,6 +207,14 @@ export interface IDexPricing<ExchangeData> {
   // useful for RFQ system
   isBlacklisted?(userAddress?: Address): AsyncOrSync<boolean>;
 
+  addPoolGenerateState?({
+    poolIdentifier,
+    blockNumber,
+  }: {
+    poolIdentifier: string;
+    blockNumber: number;
+  }): AsyncOrSync<null | unknown>;
+
   // blacklist a specific userAddress from exchange
   setBlacklist?(userAddress?: Address): AsyncOrSync<boolean>;
 }

src/dex/pancakeswap-v3/pancakeswap-v3-pool.ts

@@ -29,8 +29,9 @@ import {
   _reduceTickBitmap,
   _reduceTicks,
 } from '../uniswap-v3/contract-math/utils';
+import { StatefulDualSynchronizer } from '../../stateful-dual-synchronizer';
 
-export class PancakeSwapV3EventPool extends StatefulEventSubscriber<PoolState> {
+export class PancakeSwapV3EventPool extends StatefulDualSynchronizer<PoolState> {
   handlers: {
     [event: string]: (
       event: any,
@@ -61,7 +62,7 @@ export class PancakeSwapV3EventPool extends StatefulEventSubscriber<PoolState> {
 
   constructor(
     readonly dexHelper: IDexHelper,
-    parentName: string,
+    dexKey: string,
     readonly stateMultiContract: Contract,
     readonly erc20Interface: Interface,
     protected readonly factoryAddress: Address,
@@ -73,14 +74,8 @@ export class PancakeSwapV3EventPool extends StatefulEventSubscriber<PoolState> {
     readonly poolInitCodeHash: string,
     readonly poolDeployer?: string,
   ) {
-    super(
-      parentName,
-      `${token0}_${token1}_${feeCode}`,
-      dexHelper,
-      logger,
-      true,
-      mapKey,
-    );
+    const poolIdentifier = `${token0}_${token1}_${feeCode}`;
+    super(dexKey, poolIdentifier, dexHelper, logger);
     this.feeCodeAsString = feeCode.toString();
     this.token0 = token0.toLowerCase();
     this.token1 = token1.toLowerCase();

src/dex/pancakeswap-v3/pancakeswap-v3.ts

@@ -62,6 +62,7 @@ import {
   PancakeswapV3Factory,
 } from './pancakeswap-v3-factory';
 import { extractReturnAmountPosition } from '../../executor/utils';
+import { DexPoolNotFoundError } from '../../dex-utils';
 
 type PoolPairsInfo = {
   token0: Address;
@@ -293,7 +294,10 @@ export class PancakeswapV3
         },
       });
     } catch (e) {
-      if (e instanceof Error && e.message.endsWith('Pool does not exist')) {
+      if (
+        (e instanceof Error && e.message.endsWith('Pool does not exist')) ||
+        e instanceof DexPoolNotFoundError
+      ) {
         // no need to await we want the set to have the pool key but it's not blocking
         this.dexHelper.cache.zadd(
           this.notExistingPoolSetKey,
@@ -368,6 +372,22 @@ export class PancakeswapV3
     return true;
   }
 
+  async addPoolGenerateState({
+    poolIdentifier,
+    blockNumber,
+  }: {
+    poolIdentifier: string;
+    blockNumber: number;
+  }): Promise<PoolState | null> {
+    const [token0, token1, fee] = poolIdentifier.split('_');
+
+    const pool = await this.getPool(token0, token1, BigInt(fee), blockNumber);
+
+    if (!pool) return null;
+
+    return pool.getState(blockNumber);
+  }
+
   async getPoolIdentifiers(
     srcToken: Token,
     destToken: Token,
@@ -661,7 +681,7 @@ export class PancakeswapV3
           if (state.liquidity <= 0n) {
             if (state.liquidity < 0) {
               this.logger.error(
-                `${this.dexKey}-${this.network}: ${pool.poolAddress} pool has negative liquidity: ${state.liquidity}. Find with key: ${pool.mapKey}`,
+                `${this.dexKey}-${this.network}: ${pool.poolAddress} pool has negative liquidity: ${state.liquidity}. Find with key: ${pool.poolIdentifier}`,
               );
             }
             this.logger.trace(`pool have 0 liquidity`);

src/dex/uniswap-v2/uniswap-v2.ts

@@ -3,7 +3,6 @@ import { pack } from '@ethersproject/solidity';
 import _ from 'lodash';
 import { AsyncOrSync, DeepReadonly } from 'ts-essentials';
 import erc20ABI from '../../abi/erc20.json';
-import { StatefulEventSubscriber } from '../../stateful-event-subscriber';
 import {
   AdapterExchangeParam,
   Address,
@@ -69,6 +68,7 @@ import {
   hexConcat,
 } from 'ethers/lib/utils';
 import { BigNumber } from 'ethers';
+import { StatefulDualSynchronizer } from '../../stateful-dual-synchronizer';
 
 const rebaseTokens = _rebaseTokens as { chainId: number; address: string }[];
 
@@ -124,11 +124,12 @@ export interface UniswapV2Pair {
   pool?: UniswapV2EventPool;
 }
 
-export class UniswapV2EventPool extends StatefulEventSubscriber<UniswapV2PoolState> {
+export class UniswapV2EventPool extends StatefulDualSynchronizer<UniswapV2PoolState> {
   decoder = (log: Log) => this.iface.parseLog(log);
 
   constructor(
-    parentName: string,
+    public parentName: string,
+    public poolIdentifier: string,
     protected dexHelper: IDexHelper,
     private poolAddress: Address,
     private token0: Token,
@@ -142,15 +143,7 @@ export class UniswapV2EventPool extends StatefulEventSubscriber<UniswapV2PoolSta
     private feesMultiCallDecoder?: (values: any[]) => number,
     private iface: Interface = uniswapV2PoolIface,
   ) {
-    super(
-      parentName,
-      (token0.symbol || token0.address) +
-        '-' +
-        (token1.symbol || token1.address) +
-        ' pool',
-      dexHelper,
-      logger,
-    );
+    super(parentName, poolIdentifier, dexHelper, logger);
   }
 
   protected processLog(
@@ -301,10 +294,11 @@ export class UniswapV2
     reserves1: string,
     feeCode: number,
     blockNumber: number,
-  ) {
+  ): Promise<UniswapV2EventPool | null> {
     const { callEntry, callDecoder } = this.getFeesMultiCallData(pair) || {};
     pair.pool = new UniswapV2EventPool(
       this.dexKey,
+      this._getPoolIdentifier(pair.token0, pair.token1),
       this.dexHelper,
       pair.exchange!,
       pair.token0,
@@ -318,9 +312,50 @@ export class UniswapV2
     );
     pair.pool.addressesSubscribed.push(pair.exchange!);
 
-    await pair.pool.initialize(blockNumber, {
+    try {
+      await pair.pool.initialize(
+        blockNumber,
+        /*, {
       state: { reserves0, reserves1, feeCode },
-    });
+    }
+      */
+      );
+      return pair.pool;
+    } catch (e) {
+      this.logger.error(`Error_addPool could not add pool with error:`, e);
+      delete pair.pool;
+      return null;
+    }
+  }
+
+  async addPoolGenerateState({
+    poolIdentifier,
+    blockNumber,
+  }: {
+    poolIdentifier: string;
+    blockNumber: number;
+  }): Promise<UniswapV2PoolState | null> {
+    // poolIdentifier is in the format of 'uniswapv2_<token0>_<token1>'
+    const [, token0, token1] = poolIdentifier.split('_');
+    const pair = await this.findPair(
+      { address: token0, decimals: 0 },
+      { address: token1, decimals: 0 },
+    );
+    if (!(pair && pair.exchange)) return null;
+
+    const [pairState] = await this.getManyPoolReserves([pair], blockNumber);
+
+    const pool = await this.addPool(
+      pair,
+      pairState.reserves0,
+      pairState.reserves1,
+      pairState.feeCode,
+      blockNumber,
+    );
+
+    if (!pool) return null;
+
+    return pool.getState(blockNumber);
   }
 
   async getBuyPrice(
@@ -479,7 +514,9 @@ export class UniswapV2
           pairState.feeCode,
           blockNumber,
         );
-      } else pair.pool.setState(pairState, blockNumber);
+      } else {
+        // pair.pool.setState(pairState, blockNumber); // TEMP: temporarily disabled to prevent confusions with new system
+      }
     }
   }
 
@@ -525,6 +562,19 @@ export class UniswapV2
     };
   }
 
+  _getPoolIdentifier(_from: Token, _to: Token) {
+    const from = this.dexHelper.config.wrapETH(_from);
+    const to = this.dexHelper.config.wrapETH(_to);
+
+    const tokenAddress = [from.address.toLowerCase(), to.address.toLowerCase()]
+      .sort((a, b) => (a > b ? 1 : -1))
+      .join('_');
+
+    const poolIdentifier = `${this.dexKey}_${tokenAddress}`;
+
+    return poolIdentifier.toLowerCase();
+  }
+
   async getPoolIdentifiers(
     _from: Token,
     _to: Token,

src/dex/uniswap-v3/uniswap-v3-pool.ts

@@ -4,10 +4,7 @@ import { Interface } from '@ethersproject/abi';
 import { ethers } from 'ethers';
 import { assert, DeepReadonly } from 'ts-essentials';
 import { Log, Logger, BlockHeader, Address } from '../../types';
-import {
-  InitializeStateOptions,
-  StatefulEventSubscriber,
-} from '../../stateful-event-subscriber';
+import { InitializeStateOptions } from '../../stateful-event-subscriber';
 import { IDexHelper } from '../../dex-helper/idex-helper';
 import {
   PoolState,
@@ -27,8 +24,9 @@ import { TickBitMap } from './contract-math/TickBitMap';
 import { uint256ToBigInt } from '../../lib/decoders';
 import { decodeStateMultiCallResultWithRelativeBitmaps } from './utils';
 import { _reduceTickBitmap, _reduceTicks } from './contract-math/utils';
+import { StatefulDualSynchronizer } from '../../stateful-dual-synchronizer';
 
-export class UniswapV3EventPool extends StatefulEventSubscriber<PoolState> {
+export class UniswapV3EventPool extends StatefulDualSynchronizer<PoolState> {
   handlers: {
     [event: string]: (
       event: any,
@@ -79,7 +77,7 @@ export class UniswapV3EventPool extends StatefulEventSubscriber<PoolState> {
       poolKey = `${poolKey}_${tickSpacing}`;
     }
 
-    super(parentName, poolKey, dexHelper, logger, true, mapKey);
+    super(parentName, poolKey, dexHelper, logger);
     this.feeCodeAsString = feeCode.toString();
     this.token0 = token0.toLowerCase();
     this.token1 = token1.toLowerCase();

src/dex/uniswap-v3/uniswap-v3.ts

@@ -72,6 +72,7 @@ import { OptimalSwapExchange } from '@paraswap/core';
 import { OnPoolCreatedCallback, UniswapV3Factory } from './uniswap-v3-factory';
 import { hexConcat, hexlify, hexZeroPad, hexValue } from 'ethers/lib/utils';
 import { extractReturnAmountPosition } from '../../executor/utils';
+import { DexPoolNotFoundError } from '../../dex-utils';
 
 type PoolPairsInfo = {
   token0: Address;
@@ -321,7 +322,10 @@ export class UniswapV3
         },
       });
     } catch (e) {
-      if (e instanceof Error && e.message.endsWith('Pool does not exist')) {
+      if (
+        (e instanceof Error && e.message.endsWith('Pool does not exist')) ||
+        e instanceof DexPoolNotFoundError
+      ) {
         // no need to await we want the set to have the pool key but it's not blocking
         this.dexHelper.cache.zadd(
           this.notExistingPoolSetKey,
@@ -443,6 +447,28 @@ export class UniswapV3
     return true;
   }
 
+  async addPoolGenerateState({
+    poolIdentifier,
+    blockNumber,
+  }: {
+    poolIdentifier: string;
+    blockNumber: number;
+  }): Promise<PoolState | null> {
+    const [token0, token1, fee, tickSpacing] = poolIdentifier.split('_');
+
+    const pool = await this.getPool(
+      token0,
+      token1,
+      BigInt(fee),
+      blockNumber,
+      tickSpacing !== undefined ? BigInt(tickSpacing) : undefined,
+    );
+
+    if (!pool) return null;
+
+    return pool.getState(blockNumber);
+  }
+
   async getPoolIdentifiers(
     srcToken: Token,
     destToken: Token,
@@ -763,7 +789,7 @@ export class UniswapV3
           if (state.liquidity <= 0n) {
             if (state.liquidity < 0) {
               this.logger.error(
-                `${this.dexKey}-${this.network}: ${pool.poolAddress} pool has negative liquidity: ${state.liquidity}. Find with key: ${pool.mapKey}`,
+                `${this.dexKey}-${this.network}: ${pool.poolAddress} pool has negative liquidity: ${state.liquidity}. Find with key: ${pool.poolIdentifier}`,
               );
             }
             this.logger.trace(`pool have 0 liquidity`);