feat: allow horizontal scalability of lighthouse prover

packages/agents/lighthouse/package.json

@@ -9,7 +9,8 @@
     "src/**/*"
   ],
   "scripts": {
-    "start:prover": "LIGHTHOUSE_SERVICE=prover node --enable-source-maps dist/tasks/run.js",
+    "start:prover:pub": "LIGHTHOUSE_SERVICE=prover-pub node --enable-source-maps dist/tasks/run.js",
+    "start:prover:sub": "LIGHTHOUSE_SERVICE=prover-sub node --enable-source-maps dist/tasks/run.js",
     "start:process": "LIGHTHOUSE_SERVICE=process node --enable-source-maps dist/tasks/run.js",
     "start:propagate": "LIGHTHOUSE_SERVICE=propagate node --enable-source-maps dist/tasks/run.js",
     "start:sendoutboundroot": "LIGHTHOUSE_SERVICE=sendoutboundroot node --enable-source-maps dist/tasks/run.js",
@@ -34,6 +35,7 @@
     "@eth-optimism/sdk": "2.0.1",
     "@sinclair/typebox": "0.25.21",
     "@types/aws-lambda": "8.10.110",
+    "amqplib": "0.10.3",
     "datadog-lambda-js": "6.86.0",
     "dd-trace": "3.13.2",
     "dotenv": "16.0.3",

packages/agents/lighthouse/src/config.ts

@@ -12,7 +12,6 @@ const MIN_CARTOGRAPHER_POLL_INTERVAL = 30_000;
 const DEFAULT_CARTOGRAPHER_POLL_INTERVAL = 60_000;
 export const DEFAULT_PROVER_BATCH_SIZE = 1;
 export const DEFAULT_RELAYER_WAIT_TIME = 60_000 * 3600; // 1 hour
-export const DEFAULT_CONCURRENCY = 1;
 
 dotenvConfig();
 
@@ -35,6 +34,30 @@ export const TPollingConfig = Type.Object({
   cartographer: Type.Integer({ minimum: MIN_CARTOGRAPHER_POLL_INTERVAL }),
 });
 
+export const TMQConfig = Type.Object({
+  connection: Type.Object({
+    uri: Type.String(),
+  }),
+  exchange: Type.Object({
+    name: Type.String(),
+    type: Type.Union([Type.Literal("fanout"), Type.Literal("topic"), Type.Literal("direct")]),
+    publishTimeout: Type.Integer(),
+    persistent: Type.Boolean(),
+    durable: Type.Boolean(),
+  }),
+  subscriber: Type.Optional(Type.String()),
+  queueLimit: Type.Optional(Type.Number()),
+  prefetchSize: Type.Optional(Type.Number()),
+});
+
+export const TServerConfig = Type.Object({
+  prover: Type.Object({
+    port: Type.Integer({ minimum: 1, maximum: 65535 }),
+    host: Type.String({ format: "ipv4" }),
+  }),
+  adminToken: Type.String(),
+});
+
 export const NxtpLighthouseConfigSchema = Type.Object({
   hubDomain: Type.String(),
   chains: Type.Record(Type.String(), TChainConfig),
@@ -63,13 +86,15 @@ export const NxtpLighthouseConfigSchema = Type.Object({
   ),
   proverBatchSize: Type.Record(Type.String(), Type.Integer({ minimum: 1, maximum: 100 })),
   relayerWaitTime: Type.Integer({ minimum: 0 }),
-  concurrency: Type.Integer({ minimum: 0 }),
   service: Type.Union([
-    Type.Literal("prover"),
+    Type.Literal("prover-pub"),
+    Type.Literal("prover-sub"),
     Type.Literal("propagate"),
     Type.Literal("process"),
     Type.Literal("sendoutboundroot"),
   ]),
+  messageQueue: TMQConfig,
+  server: TServerConfig,
 });
 
 export type NxtpLighthouseConfig = Static<typeof NxtpLighthouseConfigSchema>;
@@ -162,11 +187,27 @@ export const getEnvConfig = (
       configJson.relayerWaitTime ||
       configFile.relayerWaitTime ||
       DEFAULT_RELAYER_WAIT_TIME,
-    concurrency: process.env.NXTP_PROVER_CONCURRENCY
-      ? +process.env.NXTP_PROVER_CONCURRENCY
-      : undefined || configJson.concurrency || configFile.concurrency || DEFAULT_CONCURRENCY,
+    messageQueue: process.env.MESSAGE_QUEUE
+      ? JSON.parse(process.env.MESSAGE_QUEUE)
+      : configJson.messageQueue ?? configFile.messageQueue,
+    server: {
+      prover: {
+        host:
+          process.env.PROVER_SUB_SERVER_HOST ||
+          configJson.server?.prover?.host ||
+          configFile.server?.prover?.host ||
+          "0.0.0.0",
+        port: process.env.PROVER_SUB_SERVER_PORT
+          ? +process.env.PROVER_SUB_SERVER_PORT
+          : undefined || configJson.server?.prover?.port || configFile.server?.prover?.port || 7072,
+      },
+      adminToken:
+        process.env.LH_SERVER_ADMIN_TOKEN ||
+        configJson.server?.adminToken ||
+        configFile.server?.adminToken ||
+        "blahblah",
+    },
   };
-
   nxtpConfig.cartographerUrl =
     nxtpConfig.cartographerUrl ??
     (nxtpConfig.environment === "production"

packages/agents/lighthouse/src/tasks/index.ts

@@ -3,20 +3,24 @@ import { contractDeployments } from "@connext/nxtp-txservice";
 
 import { getConfig } from "../config";
 
-import { makeProver } from "./prover";
+import { makeProverPublisher, makeProverSubscriber } from "./prover";
 import { makePropagate } from "./propagate";
 import { makeProcessFromRoot } from "./processFromRoot";
 import { makeSendOutboundRoot } from "./sendOutboundRoot";
 
-export const makeLighthouse = async () => {
+export const makeLighthouse = async (_service?: string) => {
   const chainData = await getChainData();
   if (!chainData) {
     throw new Error("Could not get chain data");
   }
   const config = await getConfig(chainData, contractDeployments);
-  switch (process.env.LIGHTHOUSE_SERVICE) {
-    case "prover":
-      await makeProver(config, chainData);
+  const service = _service ?? process.env.LIGHTHOUSE_SERVICE;
+  switch (service) {
+    case "prover-pub":
+      await makeProverPublisher(config, chainData);
+      break;
+    case "prover-sub":
+      await makeProverSubscriber(config, chainData);
       break;
     case "propagate":
       await makePropagate(config, chainData);

packages/agents/lighthouse/src/tasks/prover/bindings/index.ts

@@ -0,0 +1,34 @@
+import { AdminRequest } from "@connext/nxtp-utils";
+import fastify, { FastifyInstance, FastifyReply } from "fastify";
+
+import { getContext } from "../prover";
+
+export const bindHealthServer = async (): Promise<FastifyInstance> => {
+  const { config, logger } = getContext();
+
+  const server = fastify();
+
+  server.get("/ping", (_, res) => api.get.ping(res));
+
+  const address = await server.listen({ port: config.server.prover.port, host: config.server.prover.host });
+  logger.info(`Server listening at ${address}`);
+  return server;
+};
+
+export const api = {
+  auth: {
+    admin: (body: AdminRequest, res: FastifyReply, nested: (res: FastifyReply) => Promise<void>) => {
+      const { config } = getContext();
+      const { adminToken } = body;
+      if (adminToken !== config.server.adminToken) {
+        return res.status(401).send("Unauthorized to perform this operation");
+      }
+      return nested(res);
+    },
+  },
+  get: {
+    ping: async (res: FastifyReply) => {
+      return res.status(200).send("pong\n");
+    },
+  },
+};

packages/agents/lighthouse/src/tasks/prover/context.ts

@@ -2,6 +2,7 @@ import { ChainReader, ConnextContractInterfaces } from "@connext/nxtp-txservice"
 import { ChainData, Logger, RelayerType } from "@connext/nxtp-utils";
 import { Database } from "@connext/nxtp-adapters-database";
 import { Relayer } from "@connext/nxtp-adapters-relayer";
+import Broker from "amqplib";
 
 import { NxtpLighthouseConfig } from "../../config";
 
@@ -13,6 +14,7 @@ export type ProverContext = {
     contracts: ConnextContractInterfaces; // Used to read and write to smart contracts.
     relayers: { instance: Relayer; apiKey: string; type: RelayerType }[]; // Used to send txs to relayer.
     database: Database;
+    mqClient: Broker.Connection;
   };
   config: NxtpLighthouseConfig;
   chainData: Map<string, ChainData>;

packages/agents/lighthouse/src/tasks/prover/handler.ts

@@ -0,0 +1,72 @@
+import { createLoggingContext, getChainData } from "@connext/nxtp-utils";
+import { contractDeployments } from "@connext/nxtp-txservice";
+
+import { getConfig } from "../../config";
+
+import { getContext, makeProver } from "./prover";
+import { BrokerMessage } from "./operations/publisher";
+import { processMessages } from "./operations/consumer";
+
+export const handler = async (event: any): Promise<{ statusCode: number; body: string }> => {
+  const { requestContext, methodContext } = createLoggingContext("AmazonMQ.consumer");
+  try {
+    const chainData = await getChainData();
+    if (!chainData) {
+      return {
+        statusCode: 500,
+        body: "Failed to get chaind data",
+      };
+    }
+    const config = await getConfig(chainData, contractDeployments);
+    await makeProver(config, chainData);
+
+    const { logger } = getContext();
+
+    /*
+     * Example Rabbit MQ record events
+     * In the RabbitMQ example, pizzaQueue is the name of the RabbitMQ queue, and / is the name of the virtual host.
+     * When receiving messages, the event source lists messages under pizzaQueue::/.
+     *
+     * {
+     *   "eventSource": "aws:rmq",
+     *   "eventSourceArn": "arn:aws:mq:us-west-2:111122223333:broker:pizzaBroker:b-9bcfa592-423a-4942-879d-eb284b418fc8",
+     *   "rmqMessagesByQueue": {
+     *     "pizzaQueue::/": [
+     *       {
+     *         "basicProperties": {
+     *           ...
+     *         },
+     *         "redelivered": false,
+     *         "data": "eyJ0aW1lb3V0IjowLCJkYXRhIjoiQ1pybWYwR3c4T3Y0YnFMUXhENEUifQ=="
+     *       }
+     *     ]
+     *   }
+     * }
+     */
+    logger.info("Received an event from mq", requestContext, methodContext, { event });
+
+    const rmqMessagesByQueue = event.rmqMessagesByQueue as Record<string, any[]>;
+    const queues = Object.keys(rmqMessagesByQueue);
+    const brokerMessagesToProcess: BrokerMessage[] = [];
+    for (const queue of queues) {
+      const queueMessages = rmqMessagesByQueue[queue].map(
+        (msg: any) => JSON.parse(msg.data as string) as BrokerMessage,
+      );
+      brokerMessagesToProcess.push(...queueMessages);
+    }
+
+    for (const brokerMessage of brokerMessagesToProcess) {
+      await processMessages(brokerMessage, requestContext);
+    }
+    return {
+      statusCode: 200,
+      body: `Processed ${brokerMessagesToProcess.length} messages`,
+    };
+  } catch (err: unknown) {
+    console.error(`Message processing failed, error: ${err}`);
+    return {
+      statusCode: 500,
+      body: `Message processing failed, error: ${err}`,
+    };
+  }
+};

packages/agents/lighthouse/src/tasks/prover/index.ts

@@ -1 +1 @@
-export { makeProver } from "./prover";
+export { makeProverPublisher, makeProverSubscriber } from "./prover";