near: abide by rate limits, add logging

watcher/scripts/backfillNear.ts

@@ -2,6 +2,7 @@ import { ChainName, CONTRACTS } from '@certusone/wormhole-sdk/lib/cjs/utils/cons
 import { INITIAL_DEPLOYMENT_BLOCK_BY_CHAIN } from '@wormhole-foundation/wormhole-monitor-common';
 import { BlockResult } from 'near-api-js/lib/providers/provider';
 import { initDb } from '../src/databases/utils';
+import { logSameLine } from '../src/utils/log';
 import { getArchivalRpcProvider, getTransactionsByAccountId } from '../src/utils/near';
 import { getMessagesFromBlockResults } from '../src/watchers/NearWatcher';
 
@@ -24,23 +25,30 @@ const BATCH_SIZE = 1000;
   );
 
   // fetch all transactions for core bridge contract from explorer
+  console.log('Fetching transactions from NEAR Explorer...');
   const toBlock = await provider.block({ finality: 'final' });
   const transactions = await getTransactionsByAccountId(
     CONTRACTS.MAINNET.near.core,
     BATCH_SIZE,
-    toBlock.header.timestamp.toString()
+    toBlock.header.timestamp.toString().padEnd(19, '9') // pad to nanoseconds
   );
+  console.log(`Fetched ${transactions.length} transactions`);
 
   // filter out transactions that precede last seen block
   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.header.height) {
+  const blockHashes = [...new Set(transactions.map((tx) => tx.blockHash))]; // de-dup blocks
+  logSameLine('Fetching blocks...');
+  for (let i = 0; i < blockHashes.length; i++) {
+    logSameLine(`Fetching blocks... ${i + 1}/${blockHashes.length}`);
+    const block = await provider.block({ blockId: blockHashes[i] });
+    if (block.header.height > fromBlock && block.header.height <= toBlock.header.height) {
       blocks.push(block);
     }
-  });
+  }
 
+  logSameLine(`Fetched ${blocks.length} blocks\n`);
   const vaasByBlock = await getMessagesFromBlockResults(provider, blocks);
+  console.log(`Found ${Object.keys(vaasByBlock).length} messages`);
   await db.storeVaasByBlock(chain, vaasByBlock);
+  console.log('Uploaded messages to db successfully');
 })();

watcher/src/databases/JsonDatabase.ts

@@ -53,7 +53,7 @@ export class JsonDatabase extends Database {
     // this will always overwrite the "last" block, so take caution if manually backfilling gaps
     const blockInfos = Object.keys(vaasByBlock);
     if (blockInfos.length) {
-      this.lastBlockByChain[chainId] = blockInfos[blockInfos.length - 1];
+      this.lastBlockByChain[chainId] = blockInfos.sort()[blockInfos.length - 1];
       writeFileSync(this.dbLastBlockFile, JSON.stringify(this.lastBlockByChain), ENCODING);
     }
   }

watcher/src/utils/log.ts

@@ -0,0 +1,5 @@
+export const logSameLine = (message: string) => {
+  process.stdout.clearLine(0);
+  process.stdout.cursorTo(0);
+  process.stdout.write(message);
+};

watcher/src/utils/near.ts

@@ -50,7 +50,9 @@ export const getTransactionsByAccountId = async (
     ).data as GetTransactionsByAccountIdResponse
   )[0];
   if ('error' in res) throw new Error(res.error.message);
-  return res.result.data.items.filter(
-    (tx) => tx.status === 'success' && tx.actions.some((a) => a.kind === 'functionCall') // other actions don't generate logs
-  );
+  return res.result.data.items
+    .filter(
+      (tx) => tx.status === 'success' && tx.actions.some((a) => a.kind === 'functionCall') // other actions don't generate logs
+    )
+    .reverse(); // return chronological order
 };

watcher/src/watchers/NearWatcher.ts

@@ -6,7 +6,8 @@ import { RPCS_BY_CHAIN } from '../consts';
 import { VaasByBlock } from '../databases/types';
 import { makeBlockKey, makeVaaKey } from '../databases/utils';
 import { EventLog } from '../types/near';
-import { isWormholePublishEventLog } from '../utils/types';
+import { logSameLine } from '../utils/log';
+import { isWormholePublishEventLog } from '../utils/validation';
 import { Watcher } from './Watcher';
 
 export class NearWatcher extends Watcher {
@@ -66,10 +67,14 @@ export const getMessagesFromBlockResults = async (
   blocks: BlockResult[]
 ): Promise<VaasByBlock> => {
   const vaasByBlock: VaasByBlock = {};
-  for (const block of blocks) {
-    const chunks = await Promise.all(
-      block.chunks.map(({ chunk_hash }) => provider.chunk(chunk_hash))
-    );
+  logSameLine(`Fetching messages from ${blocks.length} blocks...`);
+  for (let i = 0; i < blocks.length; i++) {
+    logSameLine(`Fetching messages from block ${i + 1}/${blocks.length}...`);
+    const chunks = [];
+    for (const chunk of blocks[i].chunks) {
+      chunks.push(await provider.chunk(chunk.chunk_hash));
+    }
+
     const transactions = chunks.flatMap(({ transactions }) => transactions);
     for (const tx of transactions) {
       const outcome = await provider.txStatus(tx.hash, CONTRACTS.MAINNET.near.core);
@@ -84,7 +89,7 @@ export const getMessagesFromBlockResults = async (
           .map((log) => JSON.parse(log.slice(11)) as EventLog)
           .filter(isWormholePublishEventLog);
         for (const log of logs) {
-          const { height, timestamp } = block.header;
+          const { height, timestamp } = blocks[i].header;
           const blockKey = makeBlockKey(height.toString(), timestamp.toString());
           const vaaKey = makeVaaKey(tx.hash, 'near', log.emitter, log.seq.toString());
           vaasByBlock[blockKey] = [...(vaasByBlock[blockKey] || []), vaaKey];
@@ -93,5 +98,6 @@ export const getMessagesFromBlockResults = async (
     }
   }
 
+  logSameLine(`Fetched messages from ${blocks.length} blocks\n`);
   return vaasByBlock;
 };