user and replica set entity manager

creator-node/compose/env/base.env

@@ -71,3 +71,5 @@ discoveryNodeUnhealthyBlockDiff=500
 maxBatchClockStatusBatchSize=5
 
 reconfigSPIdBlacklistString=
+
+entityManagerAddress=0x5b9b42d6e4B2e4Bf8d42Eba32D46918e10899B66

creator-node/src/config.js

@@ -386,6 +386,12 @@ const config = convict({
     env: 'dataRegistryAddress',
     default: null
   },
+  entityManagerAddress: {
+    doc: 'entity manager registry address',
+    format: String,
+    env: 'entityManagerAddress',
+    default: '0x2F99338637F027CFB7494E46B49987457beCC6E3'
+  },
   dataProviderUrl: {
     doc: 'data contracts web3 provider url',
     format: String,
@@ -464,6 +470,12 @@ const config = convict({
     env: 'considerNodeUnhealthy',
     default: false
   },
+  entityManagerReplicaSetEnabled: {
+    doc: 'whether or not to use entity manager to update the replica set',
+    format: Boolean,
+    env: 'entityManagerReplicaSetEnabled',
+    default: true
+  },
 
   /** sync / snapback configs */
 

creator-node/src/middlewares.js

@@ -752,7 +752,6 @@ async function getReplicaSetSpIDs({
     const RETRY_TIMEOUT_MS = 1000 // 1 seconds
 
     let errorMsg = null
-    let blockNumberIndexed = false
     for (let retry = 1; retry <= MAX_RETRIES; retry++) {
       logger.info(
         `${logPrefix} retry #${retry}/${MAX_RETRIES} || time from start: ${
@@ -761,15 +760,40 @@ async function getReplicaSetSpIDs({
       )
 
       try {
-        // will throw error if blocknumber not found
-        replicaSet =
-          await libs.contracts.UserReplicaSetManagerClient.getUserReplicaSetAtBlockNumber(
-            userId,
+        if (config.get('entityManagerReplicaSetEnabled')) {
+          // will throw error if blocknumber not found
+          const encodedUserId = libs.Utils.encodeHashId(userId)
+          const spResponse = await libs.discoveryProvider.getUserReplicaSet({
+            encodedUserId,
             blockNumber
-          )
-        errorMsg = null
-        blockNumberIndexed = true
-        break
+          })
+
+          if (spResponse) {
+            if (spResponse.primarySpID) {
+              replicaSet = {
+                primaryId: spResponse.primarySpID,
+                secondaryIds: [
+                  spResponse.secondary1SpID,
+                  spResponse.secondary2SpID
+                ]
+              }
+              errorMsg = null
+              break
+            } else {
+              // The blocknumber was indexed by discovery, but there's still no user replica set returned
+              errorMsg = `User replica not found in discovery`
+              break
+            }
+          }
+        } else {
+          replicaSet =
+            await libs.contracts.UserReplicaSetManagerClient.getUserReplicaSetAtBlockNumber(
+              userId,
+              blockNumber
+            )
+          errorMsg = null
+          break
+        }
       } catch (e) {
         errorMsg = e.message
       } // Ignore all errors until MAX_RETRIES exceeded
@@ -779,22 +803,14 @@ async function getReplicaSetSpIDs({
 
     // Error if indexed blockNumber but didn't find any replicaSet for user
     if (
-      blockNumberIndexed &&
-      (!replicaSet ||
-        !replicaSet.hasOwnProperty('primaryId') ||
-        !replicaSet.primaryId)
+      !replicaSet ||
+      !replicaSet.hasOwnProperty('primaryId') ||
+      !replicaSet.primaryId
     ) {
       throw new Error(
         `${logPrefix} ERROR || Failed to retrieve user from UserReplicaSetManager after ${MAX_RETRIES} retries. Aborting.`
       )
     }
-
-    // Error if failed to index target blockNumber
-    if (!blockNumberIndexed) {
-      throw new Error(
-        `${logPrefix} ERROR || Web3 provider failed to index target blockNumber ${blockNumber} after ${MAX_RETRIES} retries. Aborting. Error ${errorMsg}`
-      )
-    }
   } else if (ensurePrimary && selfSpID) {
     /**
      * If ensurePrimary required but no blockNumber provided, poll URSM until returned primary = selfSpID
@@ -814,12 +830,30 @@ async function getReplicaSetSpIDs({
       )
 
       try {
-        replicaSet =
-          await libs.contracts.UserReplicaSetManagerClient.getUserReplicaSet(
-            userId
-          )
-
-        errorMsg = null
+        if (config.get('entityManagerReplicaSetEnabled')) {
+          const encodedUserId = libs.Utils.encodeHashId(userId)
+          const spResponse = await libs.discoveryProvider.getUserReplicaSet({
+            encodedUserId
+          })
+
+          if (spResponse && spResponse.primarySpID) {
+            replicaSet = {
+              primaryId: spResponse.primarySpID,
+              secondaryIds: [
+                spResponse.secondary1SpID,
+                spResponse.secondary2SpID
+              ]
+            }
+            errorMsg = null
+          } else {
+            errorMsg = `User replica not found in discovery`
+          }
+        } else {
+          replicaSet =
+            await libs.contracts.UserReplicaSetManagerClient.getUserReplicaSet(
+              userId
+            )
+        }
 
         if (
           replicaSet &&
@@ -863,10 +897,24 @@ async function getReplicaSetSpIDs({
 
     let errorMsg = null
     try {
-      replicaSet =
-        await libs.contracts.UserReplicaSetManagerClient.getUserReplicaSet(
-          userId
-        )
+      if (config.get('entityManagerReplicaSetEnabled')) {
+        const encodedUserId = libs.Utils.encodeHashId(userId)
+        const spResponse = await libs.discoveryProvider.getUserReplicaSet({
+          encodedUserId
+        })
+
+        if (spResponse && spResponse.primarySpID) {
+          replicaSet = {
+            primaryId: spResponse.primarySpID,
+            secondaryIds: [spResponse.secondary1SpID, spResponse.secondary2SpID]
+          }
+        }
+      } else {
+        replicaSet =
+          await libs.contracts.UserReplicaSetManagerClient.getUserReplicaSet(
+            userId
+          )
+      }
     } catch (e) {
       errorMsg = e.message
     }

creator-node/src/routes/audiusUsers.js

@@ -1,6 +1,7 @@
 const express = require('express')
 const fs = require('fs-extra')
 
+const config = require('../config')
 const models = require('../models')
 const { saveFileFromBufferToDisk } = require('../fileManager')
 const {
@@ -113,16 +114,32 @@ router.post(
     // Verify that wallet of the user on the blockchain for the given ID matches the user attempting to update
     const serviceRegistry = req.app.get('serviceRegistry')
     const { libs } = serviceRegistry
-    const userResp = await libs.contracts.UserFactoryClient.getUser(
-      blockchainUserId
-    )
-    if (
-      !userResp?.wallet ||
-      userResp.wallet.toLowerCase() !== req.session.wallet.toLowerCase()
-    ) {
-      throw new Error(
-        `Owner wallet ${userResp.wallet} of blockchainUserId ${blockchainUserId} does not match the wallet of the user attempting to write this data: ${req.session.wallet}`
+    if (config.get('entityManagerReplicaSetEnabled')) {
+      const encodedUserId = libs.Utils.encodeHashId(blockchainUserId)
+      const spResponse = await libs.discoveryProvider.getUserReplicaSet({
+        encodedUserId,
+        blockNumber
+      })
+      if (
+        (spResponse?.wallet ?? '').toLowerCase() !==
+        req.session.wallet.toLowerCase()
+      ) {
+        throw new Error(
+          `Owner wallet ${spResponse?.wallet} of blockchainUserId ${blockchainUserId} does not match the wallet of the user attempting to write this data: ${req.session.wallet}`
+        )
+      }
+    } else {
+      const userResp = await libs.contracts.UserFactoryClient.getUser(
+        blockchainUserId
       )
+      if (
+        !userResp?.wallet ||
+        userResp.wallet.toLowerCase() !== req.session.wallet.toLowerCase()
+      ) {
+        throw new Error(
+          `Owner wallet ${userResp.wallet} of blockchainUserId ${blockchainUserId} does not match the wallet of the user attempting to write this data: ${req.session.wallet}`
+        )
+      }
     }
 
     const cnodeUserUUID = req.session.cnodeUserUUID

creator-node/src/services/initAudiusLibs.js

@@ -31,6 +31,7 @@ module.exports = async ({
   const ethRegistryAddress = config.get('ethRegistryAddress')
   const ethOwnerWallet = config.get('ethOwnerWallet')
   const dataRegistryAddress = config.get('dataRegistryAddress')
+  const entityManagerAddress = config.get('entityManagerAddress')
   const dataProviderUrl = config.get('dataProviderUrl')
   const delegatePrivateKey = config.get('delegatePrivateKey')
   const oldDelegatePrivateKey = config.get('oldDelegatePrivateKey')
@@ -73,7 +74,8 @@ module.exports = async ({
           // pass as array
           [dataProviderUrl],
           // TODO - formatting this private key here is not ideal
-          (oldDelegatePrivateKey || delegatePrivateKey).replace('0x', '')
+          (oldDelegatePrivateKey || delegatePrivateKey).replace('0x', ''),
+          entityManagerAddress
         )
       : null,
     discoveryProviderConfig: enableDiscovery

creator-node/src/services/stateMachineManager/stateReconciliation/updateReplicaSet.jobProcessor.ts

@@ -97,7 +97,7 @@ const updateReplicaSetJobProcessor = async function ({
       audiusLibs = await initAudiusLibs({
         enableEthContracts: true,
         enableContracts: true,
-        enableDiscovery: false,
+        enableDiscovery: true,
         enableIdentity: true,
         logger
       })
@@ -695,17 +695,53 @@ const _issueUpdateReplicaSetOp = async (
         return response
       }
 
-      await audiusLibs.contracts.UserReplicaSetManagerClient._updateReplicaSet(
-        userId,
-        newReplicaSetSPIds[0], // new primary
-        newReplicaSetSPIds.slice(1), // [new secondary1, new secondary2]
-        // This defaulting logic is for the edge case when an SP deregistered and can't be fetched from our mapping, so we use the SP ID from the user's old replica set queried from the chain
-        oldPrimarySpId || chainPrimarySpId,
-        [
-          oldSecondary1SpId || chainSecondarySpIds?.[0],
-          oldSecondary2SpId || chainSecondarySpIds?.[1]
-        ]
-      )
+      if (config.get('entityManagerReplicaSetEnabled')) {
+        logger.info(
+          `[_issueUpdateReplicaSetOp] updating replica set now ${
+            Date.now() - startTimeMs
+          }ms for userId=${userId} wallet=${wallet}`
+        )
+
+        const { blockNumber } =
+          await audiusLibs.User.updateEntityManagerReplicaSet({
+            userId,
+            primary: newReplicaSetSPIds[0], // new primary
+            secondaries: newReplicaSetSPIds.slice(1), // [new secondary1, new secondary2]
+            // This defaulting logic is for the edge case when an SP deregistered and can't be fetched from our mapping, so we use the SP ID from the user's old replica set queried from the chain
+            oldPrimary: oldPrimarySpId || chainPrimarySpId,
+            oldSecondaries: [
+              oldSecondary1SpId || chainSecondarySpIds?.[0],
+              oldSecondary2SpId || chainSecondarySpIds?.[1]
+            ]
+          })
+        logger.info(
+          `[_issueUpdateReplicaSetOp] did call audiusLibs.User.updateEntityManagerReplicaSet waiting for ${blockNumber}`
+        )
+        // Wait for blockhash/blockNumber to be indexed
+        try {
+          await audiusLibs.User.waitForReplicaSetDiscoveryIndexing(
+            userId,
+            newReplicaSetSPIds,
+            blockNumber
+          )
+        } catch (err) {
+          throw new Error(
+            `[_issueUpdateReplicaSetOp] waitForReplicaSetDiscovery Indexing Unable to confirm updated replica set for user ${userId}`
+          )
+        }
+      } else {
+        await audiusLibs.contracts.UserReplicaSetManagerClient._updateReplicaSet(
+          userId,
+          newReplicaSetSPIds[0], // new primary
+          newReplicaSetSPIds.slice(1), // [new secondary1, new secondary2]
+          // This defaulting logic is for the edge case when an SP deregistered and can't be fetched from our mapping, so we use the SP ID from the user's old replica set queried from the chain
+          oldPrimarySpId || chainPrimarySpId,
+          [
+            oldSecondary1SpId || chainSecondarySpIds?.[0],
+            oldSecondary2SpId || chainSecondarySpIds?.[1]
+          ]
+        )
+      }
 
       response.issuedReconfig = true
       logger.info(
@@ -807,8 +843,25 @@ const _canReconfig = async ({
 }: CanReconfigParams): Promise<CanReconfigReturnValue> => {
   let error
   try {
-    const { primaryId: chainPrimarySpId, secondaryIds: chainSecondarySpIds } =
-      await libs.contracts.UserReplicaSetManagerClient.getUserReplicaSet(userId)
+    let chainPrimarySpId, chainSecondarySpIds
+    if (config.get('entityManagerReplicaSetEnabled')) {
+      const encodedUserId = libs.Utils.encodeHashId(userId)
+      const spResponse = await libs.discoveryProvider.getUserReplicaSet({
+        encodedUserId
+      })
+      chainPrimarySpId = spResponse?.primarySpID
+      chainSecondarySpIds = [
+        spResponse?.secondary1SpID,
+        spResponse?.secondary2SpID
+      ]
+    } else {
+      const response =
+        await libs.contracts.UserReplicaSetManagerClient.getUserReplicaSet(
+          userId
+        )
+      chainPrimarySpId = response.primaryId
+      chainSecondarySpIds = response.secondaryIds
+    }
 
     if (
       !chainPrimarySpId ||

creator-node/test/lib/libsMock.js

@@ -1,4 +1,5 @@
 const sinon = require('sinon')
+const { encode, decode } = require('../../src/hashids')
 
 function getLibsMock() {
   const libsMock = {
@@ -67,11 +68,34 @@ function getLibsMock() {
     Playlist: {
       getPlaylists: sinon.mock().atLeast(1)
     },
+    Utils: {
+      encodeHashId: sinon.mock().atLeast(1)
+    },
     discoveryProvider: {
-      discoveryProviderEndpoint: 'http://docker.for.mac.localhost:5000'
+      discoveryProviderEndpoint: 'http://docker.for.mac.localhost:5000',
+      getUserReplicaSet: sinon.mock().atLeast(1)
     }
   }
 
+  libsMock.Utils.encodeHashId.callsFake((id) => {
+    return encode(id)
+  })
+
+  libsMock.discoveryProvider.getUserReplicaSet.callsFake(({ encodedUserId }) => {
+    const user_id = decode(encodedUserId)
+    return {
+      user_id,
+      "wallet": '0xadd36bad12002f1097cdb7ee24085c28e960fc32',
+      "primary": 'http://mock-cn1.audius.co',
+      "secondary1": 'http://mock-cn2.audius.co',
+      "secondary2": 'http://mock-cn3.audius.co',
+      "primarySpID": 1,
+      "secondary1SpID": 2,
+      "secondary2SpID": 3
+    }
+  })
+
+
   libsMock.contracts.UserReplicaSetManagerClient.getUserReplicaSet.returns({
     primaryId: 1,
     secondaryIds: [2, 3]