near: add ability to backfill, refactor common code

watcher/scripts/backfillNear.ts

@@ -0,0 +1,83 @@
+import { sleep } from '@wormhole-foundation/wormhole-monitor-common';
+import { INITIAL_DEPLOYMENT_BLOCK_BY_CHAIN } from '@wormhole-foundation/wormhole-monitor-common/src/consts';
+import axios from 'axios';
+import { connect } from 'near-api-js';
+import { Provider } from 'near-api-js/lib/providers';
+import { BlockResult } from 'near-api-js/lib/providers/provider';
+import { NEAR_CONTRACT } from '../src/consts';
+import { storeVaasByBlock } from '../src/databases/utils';
+import {
+  NearExplorerTransactionRequestParams,
+  NearExplorerTransactionResponse,
+} from '../src/types/near';
+import { NearWatcher } from '../src/watchers/NearWatcher';
+
+// This script exists because NEAR RPC nodes do not support querying blocks older than 5 epochs
+// (~2.5 days): https://docs.near.org/api/rpc/setup#querying-historical-data. This script fetches
+// all transactions for the core bridge contract from the NEAR Explorer backend API and then uses
+// the archival RPC node to backfill messages in the given range.
+
+const BATCH_SIZE = 1000;
+const NEAR_ARCHIVE_RPC = 'https://archival-rpc.mainnet.near.org';
+const NEAR_EXPLORER_TRANSACTION_URL =
+  'https://backend-mainnet-1713.onrender.com/trpc/transaction.listByAccountId';
+
+const getArchivalRpcProvider = async (): Promise<Provider> => {
+  const connection = await connect({ nodeUrl: NEAR_ARCHIVE_RPC, networkId: 'mainnet' });
+  const provider = connection.connection.provider;
+
+  // sleep for 100ms between each request (do not parallelize calls with Promise.all)
+  for (const propName of Object.getOwnPropertyNames(Object.getPrototypeOf(provider))) {
+    if (typeof (provider as any)[propName] === 'function') {
+      (provider as any)[propName] = async (...args: any[]) => {
+        await sleep(100); // respect rate limits: 600req/min
+        return (provider as any)[propName](...args);
+      };
+    }
+  }
+
+  return provider;
+};
+
+const getExplorerTransactionsUrl = (timestamp: number, batchSize: number): string => {
+  const params: NearExplorerTransactionRequestParams = {
+    accountId: NEAR_CONTRACT,
+    limit: batchSize,
+    cursor: {
+      timestamp,
+      indexInChunk: 0,
+    },
+  };
+  return `${NEAR_EXPLORER_TRANSACTION_URL}?batch=1&input={"0":${JSON.stringify(params)}}`;
+};
+
+const backfillNear = async (fromBlock: number, toBlock: number): Promise<void> => {
+  const watcher = new NearWatcher();
+  fromBlock = Math.max(fromBlock, Number(INITIAL_DEPLOYMENT_BLOCK_BY_CHAIN[watcher.chain] ?? 0));
+  if (fromBlock > toBlock) return;
+
+  // use archival rpc to fetch specified transactions
+  const provider = await getArchivalRpcProvider();
+
+  // fetch all transactions for core bridge contract from explorer:
+  // https://github.com/near/near-explorer/blob/beead42ba2a91ad8d2ac3323c29b1148186eec98/backend/src/router/transaction/list.ts#L127
+  const timestamp = (await provider.block(toBlock)).header.timestamp;
+  const transactions = (
+    (await axios.get(getExplorerTransactionsUrl(timestamp, BATCH_SIZE)))
+      .data as NearExplorerTransactionResponse
+  )[0].result.data.items.filter((tx) => tx.status === 'success');
+
+  // filter out transactions that are not in the given block range
+  const blocks: BlockResult[] = [];
+  const blockHashes = [...new Set(transactions.map((tx) => tx.blockHash))];
+  blockHashes.forEach(async (hash) => {
+    const block = await provider.block(hash);
+    if (block.header.height >= fromBlock && block.header.height <= toBlock) {
+      blocks.push(block);
+    }
+  });
+
+  watcher.provider = provider;
+  const vaasByBlock = await watcher.getMessagesFromBlockResults(blocks);
+  await storeVaasByBlock(watcher.chain, vaasByBlock);
+};

watcher/src/consts.ts

@@ -49,6 +49,8 @@ export const ALGORAND_INFO = {
   token: '',
 };
 
+export const NEAR_CONTRACT = 'contract.wormhole_crypto.near';
+
 export const DB_SOURCE = process.env.DB_SOURCE || 'local';
 export const JSON_DB_FILE = process.env.JSON_DB_FILE || '../server/db.json';
 export const DB_LAST_BLOCK_FILE =

watcher/src/types/near.ts

@@ -0,0 +1,58 @@
+// https://nomicon.io/Standards/EventsFormat
+export type EventLog = {
+  event: string;
+  standard: string;
+  data?: unknown;
+  version?: string; // this is supposed to exist but is missing in WH logs
+};
+
+export type WormholePublishEventLog = {
+  standard: 'wormhole';
+  event: 'publish';
+  data: string;
+  nonce: number;
+  emitter: string;
+  seq: number;
+  block: number;
+};
+
+export const isWormholePublishEventLog = (log: EventLog): log is WormholePublishEventLog => {
+  return log.standard === 'wormhole' && log.event === 'publish';
+};
+
+export type NearExplorerTransactionResponse = {
+  id: string | null;
+  result: {
+    type: string;
+    data: {
+      items: NearExplorerTransaction[];
+    };
+  };
+}[];
+
+export type NearExplorerTransaction = {
+  hash: string;
+  signerId: string;
+  receiverId: string;
+  blockHash: string;
+  blockTimestamp: number;
+  actions: {
+    kind: string;
+    args: {
+      methodName: string;
+      args: string;
+      gas: number;
+      deposit: string;
+    };
+  }[];
+  status: string;
+};
+
+export type NearExplorerTransactionRequestParams = {
+  accountId: string;
+  limit: number;
+  cursor?: {
+    timestamp: number; // paginate with timestamp
+    indexInChunk: number;
+  };
+};

watcher/src/watchers/NearWatcher.ts

@@ -1,38 +1,57 @@
-// npx pretty-quick
-
 import { connect } from 'near-api-js';
-import { Provider } from 'near-api-js/lib/providers';
-import { ExecutionStatus } from 'near-api-js/lib/providers/provider';
-import { RPCS_BY_CHAIN } from '../consts';
+import { Provider, TypedError } from 'near-api-js/lib/providers';
+import { BlockResult, ExecutionStatus } from 'near-api-js/lib/providers/provider';
+import { NEAR_CONTRACT, RPCS_BY_CHAIN } from '../consts';
 import { VaasByBlock } from '../databases/types';
 import { makeBlockKey, makeVaaKey } from '../databases/utils';
+import { EventLog, isWormholePublishEventLog } from '../types/near';
 import { Watcher } from './Watcher';
 
-const NEAR_ARCHIVE_RPC = 'https://archival-rpc.mainnet.near.org';
-const NEAR_RPC = RPCS_BY_CHAIN.near!;
-const NEAR_CONTRACT = 'contract.wormhole_crypto.near';
-
 export class NearWatcher extends Watcher {
-  private provider: Provider | null = null;
+  provider: Provider | null = null;
 
   constructor() {
     super('near');
   }
 
-  public async getFinalizedBlockNumber(): Promise<number> {
+  async getFinalizedBlockNumber(): Promise<number> {
     this.logger.info(`fetching final block for ${this.chain}`);
     const provider = await this.getProvider();
     const block = await provider.block({ finality: 'final' });
     return block.header.height;
   }
 
-  public async getMessagesForBlocks(fromBlock: number, toBlock: number): Promise<VaasByBlock> {
+  async getMessagesForBlocks(fromBlock: number, toBlock: number): Promise<VaasByBlock> {
+    // assume toBlock was retrieved from getFinalizedBlockNumber and is finalized
     this.logger.info(`fetching info for blocks ${fromBlock} to ${toBlock}`);
     const provider = await this.getProvider();
-    const vaasByBlock: VaasByBlock = {};
+    const blocks: BlockResult[] = [];
+    let block = await provider.block(toBlock);
+    try {
+      blocks.push(block);
+      while (true) {
+        // traverse backwards via block hashes: https://github.com/wormhole-foundation/wormhole-monitor/issues/35
+        block = await provider.block(block.header.prev_hash);
+        if (block.header.height < fromBlock) break;
+        blocks.push(block);
+      }
+    } catch (e) {
+      if (e instanceof TypedError && e.type === 'HANDLER_ERROR') {
+        this.logger.error(
+          `block height for block ${block.header.prev_hash} is too old for rpc, use backfillNear to fetch blocks before height ${block.header.height}`
+        );
+      }
+
+      throw e;
+    }
 
-    for (let blockId = fromBlock; blockId <= toBlock; blockId++) {
-      const block = await provider.block({ blockId });
+    return this.getMessagesFromBlockResults(blocks);
+  }
+
+  async getMessagesFromBlockResults(blocks: BlockResult[]): Promise<VaasByBlock> {
+    const provider = await this.getProvider();
+    const vaasByBlock: VaasByBlock = {};
+    for (const block of blocks) {
       const chunks = await Promise.all(
         block.chunks.map(({ chunk_hash }) => provider.chunk(chunk_hash))
       );
@@ -64,34 +83,9 @@ export class NearWatcher extends Watcher {
 
   async getProvider(): Promise<Provider> {
     if (!this.provider) {
-      const connection = await connect({
-        nodeUrl: NEAR_ARCHIVE_RPC,
-        networkId: 'mainnet',
-      });
+      const connection = await connect({ nodeUrl: RPCS_BY_CHAIN.near!, networkId: 'mainnet' });
       this.provider = connection.connection.provider;
     }
     return this.provider;
   }
 }
-
-// https://nomicon.io/Standards/EventsFormat
-type EventLog = {
-  event: string;
-  standard: string;
-  data?: unknown;
-  version?: string; // this is supposed to exist but is missing in WH logs
-} & Partial<WormholePublishEventLog>;
-
-type WormholePublishEventLog = {
-  standard: 'wormhole';
-  event: 'publish';
-  data: string;
-  nonce: number;
-  emitter: string;
-  seq: number;
-  block: number;
-};
-
-const isWormholePublishEventLog = (log: EventLog): log is WormholePublishEventLog => {
-  return log.standard === 'wormhole' && log.event === 'publish';
-};