watcher: refactor base watcher classes.

watcher/scripts/locateMessageGaps.ts

@@ -8,7 +8,7 @@ import {
 import { TIMEOUT } from '../src/consts';
 import { BigtableDatabase } from '../src/databases/BigtableDatabase';
 import { parseMessageId } from '../src/databases/utils';
-import { makeFinalizedWatcher } from '../src/watchers/utils';
+import { makeFinalizedVaaWatcher } from '../src/watchers/utils';
 import { Watcher } from '../src/watchers/Watcher';
 import { ChainId, Network, toChain, toChainId } from '@wormhole-foundation/sdk-base';
 
@@ -95,7 +95,7 @@ import { ChainId, Network, toChain, toChainId } from '@wormhole-foundation/sdk-b
       }
       let watcher: Watcher;
       try {
-        watcher = makeFinalizedWatcher(network, chainName);
+        watcher = makeFinalizedVaaWatcher(network, chainName);
       } catch (e) {
         console.error('skipping gap for unsupported chain', chainName);
         continue;

watcher/src/watchers/AlgorandWatcher.ts

@@ -18,7 +18,7 @@ export class AlgorandWatcher extends Watcher {
   indexerClient: algosdk.Indexer;
 
   constructor(network: Network) {
-    super(network, 'Algorand');
+    super(network, 'Algorand', 'vaa');
 
     if (!ALGORAND_INFO[this.network].algodServer) {
       throw new Error('ALGORAND_INFO.algodServer is not defined!');

watcher/src/watchers/AptosWatcher.ts

@@ -21,7 +21,7 @@ export class AptosWatcher extends Watcher {
   eventHandle: string;
 
   constructor(network: Network) {
-    super(network, 'Aptos');
+    super(network, 'Aptos', 'vaa');
     this.client = new AptosClient(RPCS_BY_CHAIN[this.network][this.chain]!);
     this.coreBridgeAddress = contracts.coreBridge(network, 'Aptos');
     this.eventHandle = `${this.coreBridgeAddress}::state::WormholeMessageHandle`;

watcher/src/watchers/ArbitrumWatcher.ts

@@ -12,9 +12,9 @@ export class ArbitrumWatcher extends EVMWatcher {
 
   constructor(network: Network) {
     if (network === 'Mainnet') {
-      super(network, 'Arbitrum');
+      super(network, 'Arbitrum', 'latest', 'vaa');
     } else {
-      super(network, 'ArbitrumSepolia');
+      super(network, 'ArbitrumSepolia', 'latest', 'vaa');
     }
 
     this.rpc = RPCS_BY_CHAIN[this.network][this.chain];
@@ -23,8 +23,8 @@ export class ArbitrumWatcher extends EVMWatcher {
     }
     this.evmWatcher =
       network === 'Mainnet'
-        ? new EVMWatcher(network, 'Ethereum', 'finalized')
-        : new EVMWatcher(network, 'Sepolia', 'finalized');
+        ? new EVMWatcher(network, 'Ethereum', 'finalized', 'vaa')
+        : new EVMWatcher(network, 'Sepolia', 'finalized', 'vaa');
     this.latestL2Finalized = 0;
     this.l1L2Map = new Map<number, number>();
     this.lastEthTime = 0;

watcher/src/watchers/BSCWatcher.ts

@@ -3,7 +3,7 @@ import { EVMWatcher } from './EVMWatcher';
 
 export class BSCWatcher extends EVMWatcher {
   constructor(network: Network) {
-    super(network, 'Bsc');
+    super(network, 'Bsc', 'latest', 'vaa');
   }
   async getFinalizedBlockNumber(): Promise<number> {
     const latestBlock = await super.getFinalizedBlockNumber();

watcher/src/watchers/CosmwasmWatcher.ts

@@ -16,7 +16,7 @@ export class CosmwasmWatcher extends Watcher {
   latestBlockHeight: number;
 
   constructor(network: Network, chain: PlatformToChains<'Cosmwasm'>) {
-    super(network, chain);
+    super(network, chain, 'vaa');
     if (chain === 'Injective') {
       throw new Error('Please use InjectiveExplorerWatcher for injective');
     }

watcher/src/watchers/EVMWatcher.ts

@@ -7,6 +7,7 @@ import { makeBlockKey, makeVaaKey } from '../databases/utils';
 import { Watcher } from './Watcher';
 import { Network, PlatformToChains, contracts } from '@wormhole-foundation/sdk-base';
 import { ethers_contracts } from '@wormhole-foundation/sdk-evm-core';
+import { Mode } from '@wormhole-foundation/wormhole-monitor-common';
 
 // This is the hash for topic[0] of the core contract event LogMessagePublished
 // https://github.com/wormhole-foundation/wormhole/blob/main/ethereum/contracts/Implementation.sol#L12
@@ -34,35 +35,13 @@ export class EVMWatcher extends Watcher {
   constructor(
     network: Network,
     chain: PlatformToChains<'Evm'>,
-    finalizedBlockTag: BlockTag = 'latest'
+    finalizedBlockTag: BlockTag,
+    mode: Mode
   ) {
-    super(network, chain);
+    super(network, chain, mode);
     this.lastTimestamp = 0;
     this.latestFinalizedBlockNumber = 0;
     this.finalizedBlockTag = finalizedBlockTag;
-    // Special cases for batch size
-    if (chain === 'Acala' || chain === 'Karura' || chain === 'Berachain') {
-      this.maximumBatchSize = 50;
-    } else if (
-      chain === 'Blast' ||
-      chain === 'Klaytn' ||
-      chain === 'Scroll' ||
-      chain === 'Snaxchain' ||
-      chain === 'Unichain' ||
-      chain === 'Worldchain' ||
-      chain === 'Monad' ||
-      chain === 'MonadDevnet' ||
-      chain === 'Ink' ||
-      chain === 'HyperEVM' ||
-      chain === 'Seievm' ||
-      chain === 'Xlayer'
-    ) {
-      this.maximumBatchSize = 10;
-    }
-    // Special cases for watch loop delay
-    if (chain === 'Berachain') {
-      this.watchLoopDelay = 1000;
-    }
   }
 
   async getBlock(blockNumberOrTag: number | BlockTag): Promise<Block> {

watcher/src/watchers/FTEVMWatcher.ts

@@ -1,21 +1,18 @@
 import knex, { Knex } from 'knex';
-import { Watcher } from './Watcher';
 import { Network } from '@wormhole-foundation/sdk-base';
 import { assertEnvironmentVariable } from '@wormhole-foundation/wormhole-monitor-common';
 import { FAST_TRANSFER_CONTRACTS, FTEVMChain } from '../fastTransfer/consts';
 import { ethers } from 'ethers';
-import { AXIOS_CONFIG_JSON, RPCS_BY_CHAIN } from '../consts';
+import { RPCS_BY_CHAIN } from '../consts';
 import { makeBlockKey } from '../databases/utils';
 import TokenRouterParser from '../fastTransfer/tokenRouter/parser';
 import SwapLayerParser from '../fastTransfer/swapLayer/parser';
-import { Block } from './EVMWatcher';
-import { BigNumber } from 'ethers';
-import axios from 'axios';
+import { EVMWatcher } from './EVMWatcher';
 import { sleep } from '@wormhole-foundation/wormhole-monitor-common';
 
 export type BlockTag = 'finalized' | 'safe' | 'latest';
 
-export class FTEVMWatcher extends Watcher {
+export class FTEVMWatcher extends EVMWatcher {
   finalizedBlockTag: BlockTag;
   lastTimestamp: number;
   latestFinalizedBlockNumber: number;
@@ -33,7 +30,7 @@ export class FTEVMWatcher extends Watcher {
     finalizedBlockTag: BlockTag = 'latest',
     isTest = false
   ) {
-    super(network, chain, 'ft');
+    super(network, chain, finalizedBlockTag, 'ft');
     this.lastTimestamp = 0;
     this.latestFinalizedBlockNumber = 0;
     this.finalizedBlockTag = finalizedBlockTag;
@@ -63,71 +60,6 @@ export class FTEVMWatcher extends Watcher {
     this.logger.debug('FTWatcher', network, chain, finalizedBlockTag);
   }
 
-  async getBlock(blockNumberOrTag: number | BlockTag): Promise<Block> {
-    const rpc = RPCS_BY_CHAIN[this.network][this.chain];
-    if (!rpc) {
-      throw new Error(`${this.chain} RPC is not defined!`);
-    }
-    let result = (
-      await axios.post(
-        rpc,
-        [
-          {
-            jsonrpc: '2.0',
-            id: 1,
-            method: 'eth_getBlockByNumber',
-            params: [
-              typeof blockNumberOrTag === 'number'
-                ? `0x${blockNumberOrTag.toString(16)}`
-                : blockNumberOrTag,
-              false,
-            ],
-          },
-        ],
-        AXIOS_CONFIG_JSON
-      )
-    )?.data?.[0];
-    if (result && result.result === null) {
-      // Found null block
-      if (
-        typeof blockNumberOrTag === 'number' &&
-        blockNumberOrTag < this.latestFinalizedBlockNumber - 1000
-      ) {
-        return {
-          hash: '',
-          number: BigNumber.from(blockNumberOrTag).toNumber(),
-          timestamp: BigNumber.from(this.lastTimestamp).toNumber(),
-        };
-      }
-    } else if (result && result.error && result.error.code === 6969) {
-      return {
-        hash: '',
-        number: BigNumber.from(blockNumberOrTag).toNumber(),
-        timestamp: BigNumber.from(this.lastTimestamp).toNumber(),
-      };
-    }
-    result = result?.result;
-    if (result && result.hash && result.number && result.timestamp) {
-      // Convert to Ethers compatible type
-      this.lastTimestamp = result.timestamp;
-      return {
-        hash: result.hash,
-        number: BigNumber.from(result.number).toNumber(),
-        timestamp: BigNumber.from(result.timestamp).toNumber(),
-      };
-    }
-    throw new Error(
-      `Unable to parse result of eth_getBlockByNumber for ${blockNumberOrTag} on ${rpc}`
-    );
-  }
-
-  async getFinalizedBlockNumber(): Promise<number> {
-    this.logger.info(`fetching block ${this.finalizedBlockTag}`);
-    const block: Block = await this.getBlock(this.finalizedBlockTag);
-    this.latestFinalizedBlockNumber = block.number;
-    return block.number;
-  }
-
   async getFtMessagesForBlocks(fromBlock: number, toBlock: number): Promise<string> {
     const tokenRouterPromise = this.tokenRouterParser.getFTResultsInRange(fromBlock, toBlock);
     const swapLayerPromise = this.swapLayerParser?.getFTSwapInRange(fromBlock, toBlock) || [];

watcher/src/watchers/InjectiveExplorerWatcher.ts

@@ -18,7 +18,7 @@ export class InjectiveExplorerWatcher extends Watcher {
   latestBlockHeight: number;
 
   constructor(network: Network) {
-    super(network, 'Injective');
+    super(network, 'Injective', 'vaa');
     this.rpc = RPCS_BY_CHAIN[this.network][this.chain];
     if (!this.rpc) {
       throw new Error(`${this.chain} RPC is not defined!`);

watcher/src/watchers/MoonbeamWatcher.ts

@@ -6,7 +6,7 @@ import { Network } from '@wormhole-foundation/sdk-base';
 
 export class MoonbeamWatcher extends EVMWatcher {
   constructor(network: Network) {
-    super(network, 'Moonbeam');
+    super(network, 'Moonbeam', 'latest', 'vaa');
   }
   async getFinalizedBlockNumber(): Promise<number> {
     const latestBlock = await super.getFinalizedBlockNumber();