[CON-201] Remove QueueInterfacer + increase RS update concurrency

creator-node/src/config.js

@@ -532,6 +532,12 @@ const config = convict({
     env: 'maxRecurringRequestSyncJobConcurrency',
     default: 5
   },
+  maxUpdateReplicaSetJobConcurrency: {
+    doc: 'Max bull queue concurrency for update replica set jobs',
+    format: 'nat',
+    env: 'maxUpdateReplicaSetJobConcurrency',
+    default: 15
+  },
   peerHealthCheckRequestTimeout: {
     doc: 'Timeout [ms] for checking health check route',
     format: 'nat',

creator-node/src/services/initAudiusLibs.js

@@ -8,7 +8,7 @@ const { logger: genericLogger } = require('../logging')
  *
  * Configures dataWeb3 to be internal to libs, logged in with delegatePrivateKey in order to write chain TX
  */
-module.exports = async (logger = genericLogger) => {
+module.exports = async (excludeDiscovery = false, logger = genericLogger) => {
   /**
    * Define all config variables
    */
@@ -64,10 +64,12 @@ module.exports = async (logger = genericLogger) => {
       // TODO - formatting this private key here is not ideal
       delegatePrivateKey.replace('0x', '')
     ),
-    discoveryProviderConfig: {
-      whitelist: discoveryProviderWhitelist,
-      unhealthyBlockDiff: discoveryNodeUnhealthyBlockDiff
-    },
+    discoveryProviderConfig: excludeDiscovery
+      ? null
+      : {
+          whitelist: discoveryProviderWhitelist,
+          unhealthyBlockDiff: discoveryNodeUnhealthyBlockDiff
+        },
     // If an identity service config is present, set up libs with the connection, otherwise do nothing
     identityServiceConfig: identityService
       ? AudiusLibs.configIdentityService(identityService)

creator-node/src/services/stateMachineManager/index.js

@@ -5,22 +5,17 @@ const { logger: baseLogger } = require('../../logging')
 const StateMonitoringManager = require('./stateMonitoring')
 const StateReconciliationManager = require('./stateReconciliation')
 const { RECONFIG_MODES, QUEUE_NAMES } = require('./stateMachineConstants')
-const QueueInterfacer = require('./QueueInterfacer')
 const makeOnCompleteCallback = require('./makeOnCompleteCallback')
 const CNodeToSpIdMapManager = require('./CNodeToSpIdMapManager')
 
 /**
  * Manages the queue for monitoring the state of Content Nodes and
  * the queue for reconciling anomalies in the state (syncs and replica set updates).
- * Use QueueInterfacer for interfacing with the queues.
  */
 class StateMachineManager {
   async init(audiusLibs, prometheusRegistry) {
     this.updateEnabledReconfigModesSet()
 
-    // TODO: Remove this and libs another way -- maybe init a new instance for each updateReplicaSet job
-    QueueInterfacer.init(audiusLibs)
-
     // Initialize class immediately since bull jobs are run on cadence even on deploy
     await CNodeToSpIdMapManager.updateCnodeEndpointToSpIdMap(
       audiusLibs.ethContracts

creator-node/src/services/stateMachineManager/stateMonitoring/fetchCNodeEndpointToSpIdMap.jobProcessor.js

@@ -1,4 +1,4 @@
-const QueueInterfacer = require('../QueueInterfacer')
+const initAudiusLibs = require('../../initAudiusLibs')
 const NodeToSpIdManager = require('../CNodeToSpIdMapManager')
 
 /**
@@ -11,8 +11,9 @@ const NodeToSpIdManager = require('../CNodeToSpIdMapManager')
 module.exports = async function ({ logger }) {
   let errorMsg = ''
   try {
+    const audiusLibs = await initAudiusLibs(true)
     await NodeToSpIdManager.updateCnodeEndpointToSpIdMap(
-      QueueInterfacer.getAudiusLibs().ethContracts
+      audiusLibs.ethContracts
     )
   } catch (e) {
     errorMsg = e.message || e.toString()

creator-node/src/services/stateMachineManager/stateReconciliation/index.js

@@ -115,17 +115,19 @@ class StateReconciliationManager {
       logger.error(`Job failed to complete. ID=${job?.id}. Error=${err}`)
     })
 
-    // Register the logic that gets executed to process each new job from the queue
+    // Register the logic that gets executed to process each new job from the queues
     manualSyncQueue.process(
       config.get('maxManualRequestSyncJobConcurrency'),
       processManualSync
     )
-
     recurringSyncQueue.process(
       config.get('maxRecurringRequestSyncJobConcurrency'),
       processRecurringSync
     )
-    updateReplicaSetQueue.process(1 /** concurrency */, processUpdateReplicaSet)
+    updateReplicaSetQueue.process(
+      config.get('maxUpdateReplicaSetJobConcurrency'),
+      processUpdateReplicaSet
+    )
   }
 
   /*

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

@@ -10,8 +10,8 @@ const {
 } = require('../stateMachineConstants')
 const { retrieveClockValueForUserFromReplica } = require('../stateMachineUtils')
 const CNodeToSpIdMapManager = require('../CNodeToSpIdMapManager')
-const QueueInterfacer = require('../QueueInterfacer')
 const { getNewOrExistingSyncReq } = require('./stateReconciliationUtils')
+const initAudiusLibs = require('../../initAudiusLibs')
 
 const reconfigNodeWhitelist = config.get('reconfigNodeWhitelist')
   ? new Set(config.get('reconfigNodeWhitelist').split(','))
@@ -48,14 +48,13 @@ module.exports = async function ({
    * on a new replica set. Also, the sync check logic is coupled with a user state on the userStateManager.
    * There will be an explicit clock value check on the newly selected replica set nodes instead.
    */
+  const audiusLibs = await initAudiusLibs(true)
   const { services: healthyServicesMap } =
-    await QueueInterfacer.getAudiusLibs().ServiceProvider.autoSelectCreatorNodes(
-      {
-        performSyncCheck: false,
-        whitelist: reconfigNodeWhitelist,
-        log: true
-      }
-    )
+    await audiusLibs.ServiceProvider.autoSelectCreatorNodes({
+      performSyncCheck: false,
+      whitelist: reconfigNodeWhitelist,
+      log: true
+    })
 
   const healthyNodes = Object.keys(healthyServicesMap || {})
   if (healthyNodes.length === 0)
@@ -99,6 +98,7 @@ module.exports = async function ({
         secondary1,
         secondary2,
         newReplicaSet,
+        audiusLibs,
         logger
       ))
   } catch (e) {
@@ -461,6 +461,7 @@ const _issueUpdateReplicaSetOp = async (
   secondary1,
   secondary2,
   newReplicaSet,
+  audiusLibs,
   logger
 ) => {
   const response = {
@@ -513,7 +514,7 @@ const _issueUpdateReplicaSetOp = async (
     // Submit chain tx to update replica set
     const startTimeMs = Date.now()
     try {
-      await QueueInterfacer.getAudiusLibs().contracts.UserReplicaSetManagerClient.updateReplicaSet(
+      await audiusLibs.contracts.UserReplicaSetManagerClient.updateReplicaSet(
         userId,
         newReplicaSetSPIds[0], // primary
         newReplicaSetSPIds.slice(1) // [secondary1, secondary2]

creator-node/test/updateReplicaSet.jobProcessor.test.js

@@ -93,9 +93,7 @@ describe('test updateReplicaSet job processor', function () {
       '../src/services/stateMachineManager/stateReconciliation/updateReplicaSet.jobProcessor.js',
       {
         '../../../config': config,
-        '../QueueInterfacer': {
-          getAudiusLibs: () => audiusLibsStub
-        },
+        '../../initAudiusLibs': sandbox.stub().resolves(audiusLibsStub),
         './stateReconciliationUtils': {
           getNewOrExistingSyncReq: getNewOrExistingSyncReqStub
         },