Merge upstream changes and fix fork sync

.github/workflows/fork-sync.yaml

@@ -8,13 +8,17 @@ on:
 jobs:
   sync:
     runs-on: ubuntu-latest
+    permissions:
+      contents: write
+      pull-requests: write
     timeout-minutes: 30 # `yarn test` takes longer time
     steps:
       - uses: tgymnich/[email protected]
         with:
-          owner: across-protocol
-          repo: relayer
+          owner: LiskHQ
+          repo: across-relayer
           head: master
           base: master
-          pr_title: Sync fork repo
+          auto_merge: false
+          pr_title: Sync with upstream repo
           pr_message: Merge latest changes from upstream repo

package.json

@@ -12,7 +12,7 @@
   "dependencies": {
     "@across-protocol/constants": "^3.1.14",
     "@across-protocol/contracts": "^3.0.10",
-    "@across-protocol/sdk": "^3.1.27",
+    "@across-protocol/sdk": "^3.1.31",
     "@arbitrum/sdk": "^3.1.3",
     "@aws-sdk/client-kms": "^3.592.0",
     "@aws-sdk/client-s3": "^3.592.0",
@@ -30,7 +30,6 @@
     "@types/express": "^4.17.21",
     "@uma/common": "2.33.0",
     "@uma/logger": "^1.3.0",
-    "async": "^3.2.4",
     "axios": "^1.6.1",
     "dotenv": "^16.3.1",
     "ethers": "^5.7.2",
@@ -77,7 +76,6 @@
     "@nomiclabs/hardhat-waffle": "^2.0.5",
     "@typechain/ethers-v5": "^10.2.1",
     "@typechain/hardhat": "^6.1.6",
-    "@types/async-retry": "^1.4.5",
     "@types/bluebird": "^3.5.38",
     "@types/chai": "^4.3.5",
     "@types/minimist": "^1.2.2",

src/libexec/RelayerSpokePoolIndexer.ts

@@ -1,6 +1,5 @@
 import assert from "assert";
 import minimist from "minimist";
-import { setTimeout } from "node:timers/promises";
 import { Contract, Event, EventFilter, providers as ethersProviders, utils as ethersUtils } from "ethers";
 import { utils as sdkUtils } from "@across-protocol/sdk";
 import * as utils from "../../scripts/utils";
@@ -36,7 +35,9 @@ type ScraperOpts = {
 
 const { NODE_SUCCESS, NODE_APP_ERR } = utils;
 
-const INDEXER_POLLING_PERIOD = 2000; // ms; time to sleep between checking for exit request via SIGHUP.
+const INDEXER_POLLING_PERIOD = 2_000; // ms; time to sleep between checking for exit request via SIGHUP.
+const WS_PING_INTERVAL = 20_000; // ms
+const WS_PONG_TIMEOUT = WS_PING_INTERVAL / 2;
 
 let logger: winston.Logger;
 let chain: string;
@@ -199,7 +200,7 @@ async function listen(
   });
 
   do {
-    await setTimeout(INDEXER_POLLING_PERIOD);
+    await sdkUtils.delay(INDEXER_POLLING_PERIOD);
   } while (!stop);
 }
 
@@ -291,31 +292,82 @@ async function run(argv: string[]): Promise<void> {
   const providers = getWSProviders(chainId, quorum);
   let nProviders = providers.length;
   assert(providers.length > 0, `Insufficient providers for ${chain} (required ${quorum} by quorum)`);
+
   providers.forEach((provider) => {
-    provider._websocket.on("error", (err) => {
-      const _provider = getOriginFromURL(provider.connection.url);
-      const at = "RelayerSpokePoolIndexer::run";
-      let message = `Caught ${chain} provider error.`;
-      let log = logger.debug;
-      if (--nProviders < quorum) {
+    const { _websocket: ws } = provider;
+    const _provider = getOriginFromURL(provider.connection.url);
+    let interval: NodeJS.Timer | undefined;
+    let timeout: NodeJS.Timeout | undefined;
+
+    const closeProvider = () => {
+      if (interval) {
+        clearInterval(interval);
+        interval = undefined;
+      }
+
+      if (timeout) {
+        clearTimeout(timeout);
+        timeout = undefined;
+      }
+
+      if (!stop && --nProviders < quorum) {
         stop = true;
-        log = logger.warn;
-        message += " Insufficient providers to continue.";
+        logger.warn({
+          at: "RelayerSpokePoolIndexer::run",
+          message: `Insufficient ${chain} providers to continue.`,
+          quorum,
+          nProviders,
+        });
       }
-      log({ at, message, provider: _provider, quorum, nProviders, err });
+    };
+
+    // On connection, start an interval timer to periodically ping the remote end.
+    ws.on("open", () => {
+      interval = setInterval(() => {
+        ws.ping();
+        timeout = setTimeout(() => {
+          logger.warn({
+            at: "RelayerSpokePoolIndexer::run",
+            message: `Timed out on ${chain} provider.`,
+            provider: _provider,
+          });
+          ws.terminate();
+        }, WS_PONG_TIMEOUT);
+      }, WS_PING_INTERVAL);
     });
 
-    provider._websocket.on("close", () => {
+    // Pong received; cancel the timeout.
+    ws.on("pong", () => {
+      if (timeout) {
+        clearTimeout(timeout);
+        timeout = undefined;
+      }
+    });
+
+    // Oops, something went wrong.
+    ws.on("error", (err) => {
+      const at = "RelayerSpokePoolIndexer::run";
+      const message = `Caught ${chain} provider error.`;
+      logger.debug({ at, message, provider: _provider, quorum, nProviders, err });
+      closeProvider();
+    });
+
+    // Websocket is gone.
+    ws.on("close", () => {
       logger.debug({
         at: "RelayerSpokePoolIndexer::run",
         message: `${chain} provider connection closed.`,
-        provider: getOriginFromURL(provider.connection.url),
+        provider: _provider,
       });
+      closeProvider();
     });
   });
 
   logger.debug({ at: "RelayerSpokePoolIndexer::run", message: `Starting ${chain} listener.`, events, opts });
   await listen(eventMgr, spokePool, events, providers, opts);
+
+  // Cleanup where possible.
+  providers.forEach((provider) => provider._websocket.terminate());
 }
 
 if (require.main === module) {

src/relayer/index.ts

@@ -29,15 +29,14 @@ export async function runRelayer(_logger: winston.Logger, baseSigner: Signer): P
   // Explicitly don't log ignoredAddresses because it can be huge and can overwhelm log transports.
   const { ignoredAddresses: _ignoredConfig, ...loggedConfig } = config;
   logger.debug({ at: "Relayer#run", message: "Relayer started 🏃‍♂️", loggedConfig, relayerRun });
-
-  logger.info({ at: "Relayer#run", message: "Starting relayer API server." });
-  await runAPIServer(logger);
-
   const relayerClients = await constructRelayerClients(logger, config, baseSigner);
   const relayer = new Relayer(await baseSigner.getAddress(), logger, relayerClients, config);
   const simulate = !config.sendingRelaysEnabled;
   const enableSlowFills = config.sendingSlowRelaysEnabled;
 
+  logger.info({ at: "Relayer#run", message: "Starting relayer API server." });
+  await runAPIServer(logger);
+
   let run = 1;
   let txnReceipts: { [chainId: number]: Promise<string[]> };
   try {