Improve logging and refactor

packages/agents/router/src/tasks/publisher/bindings/server/index.ts

@@ -1,4 +1,11 @@
-import { jsonifyError, NxtpError, ClearCacheRequestSchema, ClearCacheRequest, AdminRequest } from "@connext/nxtp-utils";
+import {
+  jsonifyError,
+  NxtpError,
+  ClearCacheRequestSchema,
+  ClearCacheRequest,
+  AdminRequest,
+  AdminSchema,
+} from "@connext/nxtp-utils";
 import fastify, { FastifyInstance, FastifyReply } from "fastify";
 import { register } from "prom-client";
 
@@ -11,6 +18,8 @@ export const bindServer = async (): Promise<FastifyInstance> => {
 
   server.get("/ping", (_, res) => api.get.ping(res));
 
+  server.get("/nonce", (_, res) => api.get.nonce(res));
+
   server.get("/metrics", (_, res) => api.get.metrics(res));
 
   server.post<{ Body: ClearCacheRequest }>(
@@ -19,20 +28,24 @@ export const bindServer = async (): Promise<FastifyInstance> => {
     async (req, res) => api.auth.admin(req.body, res, api.post.clearCache),
   );
 
+  server.post<{ Body: AdminRequest }>("/nonce", { schema: { body: AdminSchema } }, async (req, res) =>
+    api.auth.admin(req.body, res, api.post.nonce),
+  );
+
   const address = await server.listen({ port: config.server.pub.port, host: config.server.pub.host });
   logger.info(`Server listening at ${address}`);
   return server;
 };
 
 export const api = {
   auth: {
-    admin: (body: AdminRequest, res: FastifyReply, nested: (res: FastifyReply) => Promise<void>) => {
+    admin: (body: AdminRequest, res: FastifyReply, nested: (res: FastifyReply, req?: any) => 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);
+      return nested(res, body.additions);
     },
   },
   get: {
@@ -50,6 +63,24 @@ export const api = {
         return res.status(500).send(json);
       }
     },
+    nonce: async (res: FastifyReply) => {
+      const {
+        config: { chains },
+        logger,
+        adapters: { cache },
+      } = getContext();
+      try {
+        const result: any = {};
+        for (const domain of Object.keys(chains)) {
+          result[domain] = await cache.transfers.getLatestNonce(domain);
+        }
+        return res.status(200).send(JSON.stringify(result));
+      } catch (e: unknown) {
+        const json = jsonifyError(e as NxtpError);
+        logger.error("Failed to getLatestNonce", undefined, undefined, json);
+        return res.status(500).send(json);
+      }
+    },
   },
   post: {
     clearCache: async (res: FastifyReply) => {
@@ -59,5 +90,13 @@ export const api = {
       await cache.transfers.clear();
       return res.status(200).send();
     },
+    nonce: async (res: FastifyReply, req: any) => {
+      const {
+        adapters: { cache },
+      } = getContext();
+      const { domain, nonce } = req;
+      await cache.transfers.setLatestNonce(domain as string, nonce as number);
+      return res.status(200).send();
+    },
   },
 };

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

@@ -19,7 +19,7 @@ export const bindSubscriber = async (queueName: string, channel: Broker.Channel)
   logger.info("Binding subscriber for queue", requestContext, methodContext, { queue: queueName });
   interval(async () => {
     if (numberOfChild < maxChildCount) {
-      logger.debug("Trying to pull data from the queue", requestContext, methodContext);
+      logger.debug("Trying to pull data from the queue", requestContext, methodContext, { waitPeriod });
       try {
         const messages: Message[] = [];
         for (let i = 0; i < batchSize; i++) {

packages/agents/sequencer/src/config.ts

@@ -118,11 +118,13 @@ export const getEnvConfig = (
     executer: {
       batchSize: process.env.EXECUTER_BATCH_SIZE
         ? Number(process.env.EXECUTER_BATCH_SIZE)
-        : DEFAULT_EXECUTER_BATCH_SIZE,
+        : undefined || configFile?.executer?.batchSize || DEFAULT_EXECUTER_BATCH_SIZE,
       maxChildCount: process.env.MAX_CHILD_PROCESS
         ? Number(process.env.MAX_CHILD_PROCESS)
-        : DEFAULT_CHILD_PROCESS_COUNT,
-      waitPeriod: process.env.WAIT_PERIOD ? Number(process.env.WAIT_PERIOD) : DEFAULT_EXECUTER_WAIT_PERIOD,
+        : undefined || configFile?.executer?.maxChildCount || DEFAULT_CHILD_PROCESS_COUNT,
+      waitPeriod: process.env.WAIT_PERIOD
+        ? Number(process.env.WAIT_PERIOD)
+        : undefined || configFile?.executer?.waitPeriod || DEFAULT_EXECUTER_WAIT_PERIOD,
     },
   };
 

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

@@ -44,7 +44,7 @@ export const TMQExchangeConfig = Type.Object({
 export const TMQQueueConfig = Type.Object({
   name: Type.String(),
   limit: Type.Integer(),
-  queueLimit: Type.Integer(),
+  queueLimit: Type.Optional(Type.Integer()),
   deadLetter: Type.Optional(Type.String()),
   subscribe: Type.Boolean(),
 });

packages/agents/sequencer/test/mock.ts

@@ -98,6 +98,11 @@ export const mock = {
     database: {
       url: "http://example.com",
     },
+    executer: {
+      batchSize: 100,
+      maxChildCount: 5,
+      waitPeriod: 3000,
+    },
   }),
   adapters: {
     cache: (): SinonStubbedInstance<StoreManager> => {

packages/utils/src/types/api.ts

@@ -16,6 +16,7 @@ export enum ExecStatus {
 /// MARK - Shared API
 export const AdminSchema = Type.Object({
   adminToken: Type.String(),
+  additions: Type.Optional(Type.Any()),
 });
 export type AdminRequest = Static<typeof AdminSchema>;