Merge pull request #171 from Chia-Network/fix/mutex-on-await

feat: add mutex to wait for cadt sync to prevent stack buildup

.jshintrc

@@ -0,0 +1,3 @@
+{
+  "esversion": 11
+}

package.json

@@ -5,7 +5,7 @@
   "description": "",
   "main": "proxy.js",
   "scripts": {
-    "test": "jest --detectOpenHandles",
+    "test": "jest --forceExit",
     "test-ci": "jest --ci",
     "start": "node --no-warnings src/server.js",
     "prepare-binary": "rm -rf dist && mkdir dist",
@@ -31,6 +31,7 @@
   "dependencies": {
     "@chia-carbon/core-registry-config": "^1.0.2",
     "@chia-carbon/core-registry-logger": "^1.0.11",
+    "async-mutex": "^0.4.0",
     "body-parser": "^1.20.2",
     "chia-datalayer": "^2.0.0",
     "chia-root-resolver": "^1.0.0",

src/api/registry.js

@@ -4,6 +4,10 @@ const { CONFIG } = require("../config");
 const { logger } = require("../logger");
 const wallet = require("../chia/wallet");
 const utils = require("../utils");
+const constants = require("../constants.js");
+const { Mutex } = require("async-mutex");
+
+const mutex = new Mutex();
 
 const registryUri = utils.generateUriForHostAndPort(
   CONFIG().CADT.PROTOCOL,
@@ -520,74 +524,81 @@ const getOrgMetaData = async (orgUid) => {
 /**
  * Waits for the registry data to synchronize.
  *
+ * @param {object} [options] - Function options.
+ * @param {boolean} [options.throwOnEmptyRegistry=false] - Flag to throw error on empty registry.
  * @returns {Promise<void>}
  */
-const waitForRegistryDataSync = async () => {
-  if (process.env.NODE_ENV === "test") {
-    return;
-  }
-
-  await utils.waitFor(5000);
-  const dataLayerConfig = {};
-
-  if (CONFIG().CHIA.DATALAYER_HOST) {
-    dataLayerConfig.datalayer_host = CONFIG().CHIA.DATALAYER_HOST;
-  }
-
-  if (CONFIG().CHIA.WALLET_HOST) {
-    dataLayerConfig.wallet_host = CONFIG().CHIA.WALLET_HOST;
-  }
-
-  if (CONFIG().CHIA.CERTIFICATE_FOLDER_PATH) {
-    dataLayerConfig.certificate_folder_path =
-      CONFIG().CHIA.CERTIFICATE_FOLDER_PATH;
-  }
-
-  if (CONFIG().CHIA.ALLOW_SELF_SIGNED_CERTIFICATES) {
-    dataLayerConfig.allowUnverifiedCert =
-      CONFIG().CHIA.ALLOW_SELF_SIGNED_CERTIFICATES;
-  }
-
-  const datalayer = new Datalayer(dataLayerConfig);
-  const homeOrg = await getHomeOrg();
-
-  if (!homeOrg) {
-    logger.warn(
-      "Can not find the home org from the Registry. Please verify your Registry is running and you have created a Home Organization."
-    );
-    return waitForRegistryDataSync();
-  }
-
-  const onChainRegistryRoot = await datalayer.getRoot({
-    id: homeOrg.registryId,
-  });
-
-  if (!onChainRegistryRoot.confirmed) {
-    console.log("Waiting for Registry root to confirm");
-    return waitForRegistryDataSync();
-  }
-
-  if (onChainRegistryRoot.hash !== homeOrg.registryHash) {
-    console.log("Waiting for Registry to sync with latest regisry root.", {
-      onChainRoot: onChainRegistryRoot.hash,
-      homeOrgRegistryRoot: homeOrg.registryHash,
-    });
-    return waitForRegistryDataSync();
-  }
+const waitForRegistryDataSync = async (options = {}) => {
+  await mutex.waitForUnlock();
 
-  const onChainOrgRoot = await datalayer.getRoot({ id: homeOrg.orgUid });
+  if (!mutex.isLocked()) {
+    const releaseMutex = await mutex.acquire();
+    try {
+      const opts = { throwOnEmptyRegistry: false, ...options };
 
-  if (!onChainOrgRoot.confirmed) {
-    console.log("Waiting for Organization root to confirm");
-    return waitForRegistryDataSync();
-  }
+      if (process.env.NODE_ENV === "test") {
+        return;
+      }
 
-  if (onChainOrgRoot.hash !== homeOrg.orgHash) {
-    console.log("Waiting for Registry to sync with latest organization root.", {
-      onChainRoot: onChainOrgRoot.hash,
-      homeOrgRoot: homeOrg.orgHash,
-    });
-    return waitForRegistryDataSync();
+      while (true) {
+        await utils.waitFor(5000);
+
+        const config = CONFIG().CHIA;
+        const dataLayerConfig = {
+          datalayer_host: config.DATALAYER_HOST,
+          wallet_host: config.WALLET_HOST,
+          certificate_folder_path: config.CERTIFICATE_FOLDER_PATH,
+          allowUnverifiedCert: config.ALLOW_SELF_SIGNED_CERTIFICATES,
+        };
+
+        const datalayer = new Datalayer(dataLayerConfig);
+        const homeOrg = await getHomeOrg();
+
+        if (!homeOrg) {
+          logger.warn("Cannot find the home org from the Registry. Please verify your Registry is running and you have created a Home Organization.");
+          continue;
+        }
+
+        const onChainRegistryRoot = await datalayer.getRoot({ id: homeOrg.registryId });
+
+        if (!onChainRegistryRoot.confirmed) {
+          console.log("Waiting for Registry root to confirm");
+          continue;
+        }
+
+        if (onChainRegistryRoot.hash === constants.emptySingletonHash && opts.throwOnEmptyRegistry) {
+          throw new Error("Registry is empty. Please add some data to run auto retirement task.");
+        }
+
+        if (onChainRegistryRoot.hash !== homeOrg.registryHash) {
+          console.log("Waiting for Registry to sync with latest registry root.", {
+            onChainRoot: onChainRegistryRoot.hash,
+            homeOrgRegistryRoot: homeOrg.registryHash,
+          });
+          continue;
+        }
+
+        const onChainOrgRoot = await datalayer.getRoot({ id: homeOrg.orgUid });
+
+        if (!onChainOrgRoot.confirmed) {
+          console.log("Waiting for Organization root to confirm");
+          continue;
+        }
+
+        if (onChainOrgRoot.hash !== homeOrg.orgHash) {
+          console.log("Waiting for Registry to sync with latest organization root.", {
+            onChainRoot: onChainOrgRoot.hash,
+            homeOrgRoot: homeOrg.orgHash,
+          });
+          continue;
+        }
+
+        // Exit the loop if all conditions are met
+        break;
+      }
+    } finally {
+      releaseMutex();
+    }
   }
 };
 

src/constants.js

@@ -0,0 +1,4 @@
+module.exports = {
+  emptySingletonHash:
+    "0x0000000000000000000000000000000000000000000000000000000000000000",
+};

tests/create-token.test.js

@@ -49,9 +49,10 @@ describe("Create Token Process", () => {
       .resolves(true);
   });
 
-  afterEach(() => {
+  afterEach(async () => {
     sinon.restore();
     nock.cleanAll();
+    await new Promise(resolve => setTimeout(() => resolve(), 1000));
   });
 
   afterAll(async () => {

tests/tasks/sync-retirements.test.js

@@ -39,8 +39,9 @@ describe("Task: Sync Retirements", () => {
     sinon.stub(registry, "waitForRegistryDataSync").resolves();
   });
 
-  afterEach(() => {
+  afterEach(async () => {
     sinon.restore();
+    await new Promise(resolve => setTimeout(() => resolve(), 1000));
   });
 
   it("skips retirement task if no homeorg can be attained by the registry", async () => {