feat: persistent ts relayer (#4831)

### Description

- Adjust long running TS relayer with retry queue instead of crashing.
- Adds a "whitelist" for relaying to specific message
senders/recipients.
- Adds a "symbol" flag borrowed from warp commands for filtering on a
specific warp route.

### Drive-by changes

None

### Related issues

- Enables warp route deployer to run the CLI relayer in the background
and test/share the warp UI.

### Backward compatibility

Yes

### Testing

CLI e2e tests

.changeset/real-phones-bake.md

@@ -0,0 +1,7 @@
+---
+'@hyperlane-xyz/infra': minor
+'@hyperlane-xyz/cli': minor
+'@hyperlane-xyz/sdk': minor
+---
+
+Implements persistent relayer for use in CLI

typescript/cli/scripts/run-e2e-test.sh

@@ -3,8 +3,7 @@
 function cleanup() {
   set +e
   pkill -f anvil
-  rm -rf /tmp/anvil2
-  rm -rf /tmp/anvil3
+  rm -rf ./tmp
   rm -f ./test-configs/anvil/chains/anvil2/addresses.yaml
   rm -f ./test-configs/anvil/chains/anvil3/addresses.yaml
   set -e

typescript/cli/src/commands/relayer.ts

@@ -1,35 +1,94 @@
-import { HyperlaneCore, HyperlaneRelayer } from '@hyperlane-xyz/sdk';
+import {
+  ChainMap,
+  HyperlaneCore,
+  HyperlaneRelayer,
+  RelayerCacheSchema,
+} from '@hyperlane-xyz/sdk';
+import { Address } from '@hyperlane-xyz/utils';
 
 import { CommandModuleWithContext } from '../context/types.js';
 import { log } from '../logger.js';
+import { tryReadJson, writeJson } from '../utils/files.js';
+import { getWarpCoreConfigOrExit } from '../utils/input.js';
 
-import { agentTargetsCommandOption } from './options.js';
+import {
+  agentTargetsCommandOption,
+  overrideRegistryUriCommandOption,
+  symbolCommandOption,
+  warpCoreConfigCommandOption,
+} from './options.js';
 import { MessageOptionsArgTypes } from './send.js';
 
+const DEFAULT_RELAYER_CACHE = `${overrideRegistryUriCommandOption.default}/relayer-cache.json`;
+
 export const relayerCommand: CommandModuleWithContext<
-  MessageOptionsArgTypes & { chains?: string }
+  MessageOptionsArgTypes & {
+    chains?: string;
+    cache: string;
+    symbol?: string;
+    warp?: string;
+  }
 > = {
   command: 'relayer',
-  describe: 'Run a Hyperlane message self-relayer',
+  describe: 'Run a Hyperlane message relayer',
   builder: {
     chains: agentTargetsCommandOption,
+    cache: {
+      describe: 'Path to relayer cache file',
+      type: 'string',
+      default: DEFAULT_RELAYER_CACHE,
+    },
+    symbol: symbolCommandOption,
+    warp: warpCoreConfigCommandOption,
   },
-  handler: async ({ context, chains }) => {
-    const chainsArray = chains
-      ? chains.split(',').map((_) => _.trim())
-      : undefined;
+  handler: async ({ context, cache, chains, symbol, warp }) => {
     const chainAddresses = await context.registry.getAddresses();
     const core = HyperlaneCore.fromAddressesMap(
       chainAddresses,
       context.multiProvider,
     );
 
-    const relayer = new HyperlaneRelayer({ core });
+    const chainsArray =
+      chains?.split(',').map((_) => _.trim()) ?? Object.keys(chainAddresses);
+
+    const whitelist: ChainMap<Address[]> = Object.fromEntries(
+      chainsArray.map((chain) => [chain, []]),
+    );
+
+    // add warp route addresses to whitelist
+    if (symbol || warp) {
+      const warpRoute = await getWarpCoreConfigOrExit({
+        context,
+        symbol,
+        warp,
+      });
+      warpRoute.tokens.forEach(
+        ({ chainName, addressOrDenom }) =>
+          (whitelist[chainName] = [addressOrDenom!]),
+      );
+    }
+
+    const relayer = new HyperlaneRelayer({ core, whitelist });
+    // TODO: fix merkle hook stubbing
+
+    const jsonCache = tryReadJson(cache);
+    if (jsonCache) {
+      try {
+        const parsedCache = RelayerCacheSchema.parse(jsonCache);
+        relayer.hydrate(parsedCache);
+      } catch (error) {
+        log(`Error hydrating cache: ${error}`);
+      }
+    }
+
     log('Starting relayer ...');
-    relayer.start(chainsArray);
+    relayer.start();
+
     process.once('SIGINT', () => {
-      relayer.stop(chainsArray);
       log('Stopping relayer ...');
+      relayer.stop();
+
+      writeJson(cache, relayer.cache);
       process.exit(0);
     });
   },

typescript/cli/src/commands/signCommands.ts

@@ -1,7 +1,7 @@
 // Commands that send tx and require a key to sign.
 // It's useful to have this listed here so the context
 // middleware can request keys up front when required.
-export const SIGN_COMMANDS = ['deploy', 'send', 'status', 'submit'];
+export const SIGN_COMMANDS = ['deploy', 'send', 'status', 'submit', 'relayer'];
 
 export function isSignCommand(argv: any): boolean {
   return (

typescript/cli/src/tests/commands/helpers.ts

@@ -1,3 +1,5 @@
+import { $ } from 'zx';
+
 import { ERC20Test__factory, ERC4626Test__factory } from '@hyperlane-xyz/core';
 import { ChainAddresses } from '@hyperlane-xyz/registry';
 import {
@@ -183,3 +185,26 @@ export async function sendWarpRouteMessageRoundTrip(
   await hyperlaneWarpSendRelay(chain1, chain2, warpCoreConfigPath);
   return hyperlaneWarpSendRelay(chain2, chain1, warpCoreConfigPath);
 }
+
+export async function hyperlaneSendMessage(
+  origin: string,
+  destination: string,
+) {
+  return $`yarn workspace @hyperlane-xyz/cli run hyperlane send message \
+        --registry ${REGISTRY_PATH} \
+        --origin ${origin} \
+        --destination ${destination} \
+        --key ${ANVIL_KEY} \
+        --verbosity debug \
+        --yes`;
+}
+
+export function hyperlaneRelayer(chains: string[], warp?: string) {
+  return $`yarn workspace @hyperlane-xyz/cli run hyperlane relayer \
+        --registry ${REGISTRY_PATH} \
+        --chains ${chains.join(',')} \
+        --warp ${warp ?? ''} \
+        --key ${ANVIL_KEY} \
+        --verbosity debug \
+        --yes`;
+}

typescript/cli/src/tests/commands/warp.ts

@@ -60,9 +60,10 @@ export async function hyperlaneWarpSendRelay(
   origin: string,
   destination: string,
   warpCorePath: string,
+  relay = true,
 ) {
   return $`yarn workspace @hyperlane-xyz/cli run hyperlane warp send \
-        --relay \
+        ${relay ? '--relay' : ''} \
         --registry ${REGISTRY_PATH} \
         --overrides " " \
         --origin ${origin} \

typescript/cli/src/tests/relay.e2e-test.ts

@@ -0,0 +1,82 @@
+import { TokenType } from '@hyperlane-xyz/sdk';
+
+import { writeYamlOrJson } from '../utils/files.js';
+
+import { hyperlaneCoreDeploy } from './commands/core.js';
+import {
+  REGISTRY_PATH,
+  hyperlaneRelayer,
+  hyperlaneSendMessage,
+} from './commands/helpers.js';
+import {
+  hyperlaneWarpDeploy,
+  hyperlaneWarpSendRelay,
+} from './commands/warp.js';
+
+const CHAIN_NAME_1 = 'anvil2';
+const CHAIN_NAME_2 = 'anvil3';
+
+const SYMBOL = 'ETH';
+
+const WARP_DEPLOY_OUTPUT = `${REGISTRY_PATH}/deployments/warp_routes/${SYMBOL}/${CHAIN_NAME_1}-${CHAIN_NAME_2}-config.yaml`;
+
+const EXAMPLES_PATH = './examples';
+const CORE_CONFIG_PATH = `${EXAMPLES_PATH}/core-config.yaml`;
+
+const TEST_TIMEOUT = 100_000; // Long timeout since these tests can take a while
+describe('hyperlane relayer e2e tests', async function () {
+  this.timeout(TEST_TIMEOUT);
+
+  before(async () => {
+    await hyperlaneCoreDeploy(CHAIN_NAME_1, CORE_CONFIG_PATH);
+    await hyperlaneCoreDeploy(CHAIN_NAME_2, CORE_CONFIG_PATH);
+
+    const warpConfig = {
+      anvil2: {
+        type: TokenType.native,
+        symbol: SYMBOL,
+      },
+      anvil3: {
+        type: TokenType.synthetic,
+        symbol: SYMBOL,
+      },
+    };
+
+    const warpConfigPath = './tmp/warp-route-config.yaml';
+    writeYamlOrJson(warpConfigPath, warpConfig);
+    await hyperlaneWarpDeploy(warpConfigPath);
+  });
+
+  describe('relayer', () => {
+    it('should relay core messages', async () => {
+      const process = hyperlaneRelayer([CHAIN_NAME_1, CHAIN_NAME_2]);
+
+      await hyperlaneSendMessage(CHAIN_NAME_1, CHAIN_NAME_2);
+      await hyperlaneSendMessage(CHAIN_NAME_2, CHAIN_NAME_1);
+
+      await process.kill('SIGINT');
+    });
+
+    it('should relay warp messages', async () => {
+      const process = hyperlaneRelayer(
+        [CHAIN_NAME_1, CHAIN_NAME_2],
+        WARP_DEPLOY_OUTPUT,
+      );
+
+      await hyperlaneWarpSendRelay(
+        CHAIN_NAME_1,
+        CHAIN_NAME_2,
+        WARP_DEPLOY_OUTPUT,
+        false,
+      );
+      await hyperlaneWarpSendRelay(
+        CHAIN_NAME_2,
+        CHAIN_NAME_1,
+        WARP_DEPLOY_OUTPUT,
+        false,
+      );
+
+      await process.kill('SIGINT');
+    });
+  });
+});

typescript/cli/src/utils/relay.ts

@@ -19,5 +19,6 @@ export function stubMerkleTreeConfig(
       },
     },
     ism: {},
+    backlog: [],
   });
 }

typescript/infra/scripts/relay.ts

@@ -10,11 +10,10 @@ const CACHE_PATH = process.env.RELAYER_CACHE ?? './relayer-cache.json';
 async function main() {
   const { environment } = await getArgs().argv;
   const { core } = await getHyperlaneCore(environment);
-  const relayer = new HyperlaneRelayer({ core });
 
-  // target subset of chains
-  // const chains = ['ethereum', 'polygon', 'bsc']
-  const chains = undefined;
+  // target subset of chains and senders/recipients
+  const whitelist = undefined;
+  const relayer = new HyperlaneRelayer({ core, whitelist });
 
   try {
     const contents = await readFile(CACHE_PATH, 'utf-8');
@@ -26,10 +25,10 @@ async function main() {
     console.error(`Failed to load cache from ${CACHE_PATH}`);
   }
 
-  relayer.start(chains);
+  relayer.start();
 
   process.once('SIGINT', async () => {
-    relayer.stop(chains);
+    relayer.stop();
 
     const cache = JSON.stringify(relayer.cache);
     await writeFile(CACHE_PATH, cache, 'utf-8');

typescript/sdk/src/core/HyperlaneCore.ts

@@ -204,9 +204,16 @@ export class HyperlaneCore extends HyperlaneApp<CoreFactories> {
       mailbox.on<DispatchEvent>(
         mailbox.filters.Dispatch(),
         (_sender, _destination, _recipient, message, event) => {
-          const parsed = HyperlaneCore.parseDispatchedMessage(message);
-          this.logger.info(`Observed message ${parsed.id} on ${originChain}`);
-          return handler(parsed, event);
+          const dispatched = HyperlaneCore.parseDispatchedMessage(message);
+
+          // add human readable chain names
+          dispatched.parsed.originChain = this.getOrigin(dispatched);
+          dispatched.parsed.destinationChain = this.getDestination(dispatched);
+
+          this.logger.info(
+            `Observed message ${dispatched.id} on ${originChain} to ${dispatched.parsed.destinationChain}`,
+          );
+          return handler(dispatched, event);
         },
       );
     });