ci: testnet prod sync

README.md

@@ -61,7 +61,7 @@
 
 ## Connext Architecture
 
-The Connext architecture can be seen as a layered system, as follows:
+The Connext architecture can be seen as a layered system, as follows.
 
 | Layer                            | Protocol/Stakeholders                  |
 | -------------------------------- | -------------------------------------- |

packages/agents/sequencer/src/bindings/subscriber/auctions/index.ts

@@ -22,7 +22,7 @@ export const bindSubscriber = async (queueName: string, channel: Broker.Channel)
       if (consumeMessage) {
         const { transferId, type: messageType } = JSON.parse(consumeMessage.content.toString()) as Message;
         const { requestContext, methodContext } = createLoggingContext("Subscriber.consume", undefined, transferId);
-        logger.info("Executing the transfer", requestContext, methodContext, { message: consumeMessage });
+        logger.info("Executing the transfer", requestContext, methodContext, { transferId, messageType });
 
         if (messageType === MessageType.ExecuteFast) {
           await cache.auctions.setExecStatus(transferId, ExecStatus.Dequeued);
@@ -80,7 +80,7 @@ export const bindSubscriber = async (queueName: string, channel: Broker.Channel)
             }
           }
 
-          channel.reject(consumeMessage);
+          channel.reject(consumeMessage, false);
         }
       }
     },

packages/agents/sequencer/src/config.ts

@@ -75,11 +75,9 @@ export const getEnvConfig = (
       configJson.auctionWaitTime ||
       configFile.auctionWaitTime ||
       DEFAULT_AUCTION_WAIT_TIME,
-    executionWaitTime:
-      process.env.SEQ_EXECUTION_WAIT_TIME ||
-      configJson.executionWaitTime ||
-      configFile.executionWaitTime ||
-      DEFAULT_EXECUTION_WAIT_TIME,
+    executionWaitTime: process.env.SEQ_EXECUTION_WAIT_TIME
+      ? +process.env.SEQ_EXECUTION_WAIT_TIME
+      : undefined || configJson.executionWaitTime || configFile.executionWaitTime || DEFAULT_EXECUTION_WAIT_TIME,
     mode: {
       cleanup: process.env.SEQ_CLEANUP_MODE || configJson.mode?.cleanup || configFile.mode?.cleanup || false,
     },

packages/agents/sequencer/src/lib/entities/config.ts

@@ -61,6 +61,7 @@ export const TMessageQueueConfig = Type.Object({
   queues: Type.Array(TMQQueueConfig),
   bindings: Type.Array(TMQBindingConfig),
   executerTimeout: Type.Integer(),
+  prefetch: Type.Optional(Type.Integer()),
   publisher: Type.Optional(Type.String()),
   subscriber: Type.Optional(Type.String()),
 });

packages/agents/sequencer/src/lib/errors/executor.ts

@@ -2,7 +2,7 @@ import { ExecStatus, NxtpError } from "@connext/nxtp-utils";
 
 export class ExecutorDataExpired extends NxtpError {
   constructor(status: ExecStatus, context: any = {}) {
-    super("This lighthouse data has already expired.", { status, ...context }, ExecutorDataExpired.name);
+    super("This executor data has already expired.", { status, ...context }, ExecutorDataExpired.name);
   }
 }
 
@@ -26,7 +26,7 @@ export class MissingTransfer extends NxtpError {
 
 export class MissingExecutorData extends NxtpError {
   constructor(context: any = {}) {
-    super("No lighthouse data found in the cache", context, MissingExecutorData.name);
+    super("No executor data found in the cache", context, MissingExecutorData.name);
   }
 }
 

packages/agents/sequencer/src/lib/operations/execute/slowpath.ts

@@ -138,7 +138,7 @@ export const executeSlowPathData = async (
 
   // Ensure that the executor data for this transfer hasn't expired.
   const status = await cache.executors.getExecStatus(transferId);
-  if (status !== ExecStatus.Enqueued) {
+  if (status !== ExecStatus.Dequeued) {
     throw new ExecutorDataExpired(status, {
       transferId,
       executorData,

packages/agents/sequencer/src/sequencer.ts

@@ -31,7 +31,7 @@ const context: AppContext = {} as any;
 export const getContext = () => context;
 export const msgContentType = "application/json";
 export const SlippageErrorPatterns = ["dy < minDy", "Reverted 0x6479203c206d696e4479", "more than pool balance"]; // 0x6479203c206d696e4479 -- encoded hex string of "dy < minDy"
-
+export const DEFAULT_PREFETCH_SIZE = 100;
 /// MARK - Make Agents
 /**
  * Sets up and runs the sequencer publisher unit. Receives bids from router network and assigns to transfer by ID,
@@ -86,6 +86,9 @@ export const makeSubscriber = async () => {
   try {
     const mqClient = context.adapters.mqClient;
     const channel = await mqClient.createChannel();
+    const prefetchSize = context.config.messageQueue.prefetch ?? DEFAULT_PREFETCH_SIZE;
+    channel.prefetch(prefetchSize);
+
     await channel.assertExchange(
       context.config.messageQueue.exchanges[0].name,
       context.config.messageQueue.exchanges[0].type,

packages/agents/sequencer/test/lib/operations/execute/slowpath.spec.ts

@@ -181,7 +181,7 @@ describe("Operations:Execute:SlowPath", () => {
 
       getTransferStub.resolves(mockTransfer);
       getExecutorDataStub.resolves(mockExecutorData);
-      getExecStatusStub.resolves(ExecStatus.Enqueued);
+      getExecStatusStub.resolves(ExecStatus.Dequeued);
       getBackupDataStub.resolves([mockExecutorBackupData1, mockExecutorBackupData2]);
       sendExecuteSlowToRelayerStub.onCall(0).throws("Failed to send to the gelato");
       sendExecuteSlowToRelayerStub.onCall(1).resolves({ taskId: undefined, relayer: undefined });
@@ -208,7 +208,7 @@ describe("Operations:Execute:SlowPath", () => {
 
       getTransferStub.resolves(mockTransfer);
       getExecutorDataStub.resolves(mockExecutorData);
-      getExecStatusStub.resolves(ExecStatus.Enqueued);
+      getExecStatusStub.resolves(ExecStatus.Dequeued);
       getBackupDataStub.resolves([mockExecutorBackupData1, mockExecutorBackupData2]);
       sendExecuteSlowToRelayerStub.onCall(0).throws("Failed to send to the gelato");
       sendExecuteSlowToRelayerStub.onCall(1).resolves({ taskId: undefined, relayer: undefined });