Testnet sync (#4338)

* feat: separate router and sequencer admin keys (#4316) * feat: add nonce to stableswap staging events * feat: add nonce to v0 stableswap events * fix: event id * feat: add nonce field to types * feat: add nonce to stableswap db migration * fix: database adapter for stableswap nonce * chore: add commitlint local setup * ci: update ci workflow for commitlint * feat: update subgraph & poller logic with nonce * feat: rebuild graph-client * fix: query with bigInt params * fix: carto poller unit tests * fix: remove unused checkpoint for asset updates * fix(test): unit tests for poller * chore: add code owners (#4318) * chore: add code owners * (chore) testing gpg sign * (chore) test signed commit * (chore) testing signed commit * ci: fix syntax * ci: try different install * ci: fix commitlint * ci: validate commit * feat: graph-client build * Don't commitlint testnet-prod and prod * Hot Fix: Update Linea goerli RPC * Fix: Disable commitlint in testnet-prod * feat: getSupportedAssets and tests added (#4324) * feat: getSupportedAssets and tests added * chore: use a mockable method * fix(lint): auto fix order problems * test(fix): expect * refactor: renaming --------- Co-authored-by: Prathmesh Khandelwal <prathmeshkhandelwal83@gmail.com> Co-authored-by: wanglonghong <wanglonghong4467@gmail.com> * Fix aggregate root no exist (#4335) * fix: use received aggregate root * fix: lint * feat: should ack and reject (#4336) * chore: change comparison (#4330) * Improve logging and refactor (#4337) * feat: expose nonce api and refactors * fix(utils): mock * test(fix): make optional --------- Co-authored-by: Liu <57480598+liu-zhipeng@users.noreply.github.com> Co-authored-by: Sanchay Mittal <sanchaymittal@gmail.com> Co-authored-by: preethamr <preethamr@users.noreply.github.com> Co-authored-by: Prathmesh Khandelwal <201952225@iiitvadodara.ac.in> Co-authored-by: Prathmesh Khandelwal <prathmeshkhandelwal83@gmail.com> Co-authored-by: wanglonghong <wanglonghong4467@gmail.com>
connext · May 25, 2023 · e9fc6b0 · e9fc6b0
1 parent b5ebaa4
commit e9fc6b0
Show file tree

Hide file tree

Showing 6 changed files with 55 additions and 8 deletions.
diff --git a/packages/agents/router/src/tasks/publisher/bindings/server/index.ts b/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();
+    },
   },
 };
diff --git a/packages/agents/sequencer/src/bindings/subscriber/auctions/index.ts b/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: Broker.GetMessage[] = [];
         for (let i = 0; i < batchSize; i++) {

diff --git a/packages/agents/sequencer/src/config.ts b/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,
     },
   };
 

diff --git a/packages/agents/sequencer/src/lib/entities/config.ts b/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(),
 });

diff --git a/packages/agents/sequencer/test/mock.ts b/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> => {

diff --git a/packages/utils/src/types/api.ts b/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>;