[CON-225] Remove manual sync from snapbackSM

creator-node/src/middlewares.js

@@ -15,6 +15,9 @@ const { hasEnoughStorageSpace } = require('./fileManager')
 const { getMonitors, MONITORS } = require('./monitors/monitors')
 const { verifyRequesterIsValidSP } = require('./apiSigning')
 const BlacklistManager = require('./blacklistManager')
+const {
+  issueSyncRequestsUntilSynced
+} = require('./services/stateMachineManager/stateReconciliation/stateReconciliationUtils')
 
 /**
  * Ensure valid cnodeUser and session exist for provided session token
@@ -269,7 +272,7 @@ async function ensureStorageMiddleware(req, res, next) {
  */
 async function issueAndWaitForSecondarySyncRequests(req) {
   const serviceRegistry = req.app.get('serviceRegistry')
-  const { snapbackSM } = serviceRegistry
+  const { manualSyncQueue } = serviceRegistry
 
   // Parse request headers
   const pollingDurationMs =
@@ -332,11 +335,13 @@ async function issueAndWaitForSecondarySyncRequests(req) {
     const replicationStart = Date.now()
     try {
       const secondaryPromises = secondaries.map((secondary) => {
-        return snapbackSM.issueSyncRequestsUntilSynced(
+        return issueSyncRequestsUntilSynced(
+          primary,
           secondary,
           wallet,
           primaryClockVal,
-          pollingDurationMs
+          pollingDurationMs,
+          manualSyncQueue
         )
       })
 

creator-node/src/serviceRegistry.js

@@ -130,7 +130,13 @@ class ServiceRegistry {
     return this.blacklistManager
   }
 
-  _setupBullMonitoring(app) {
+  setupBullMonitoring(
+    app,
+    stateMonitoringQueue,
+    cNodeEndpointToSpIdMapQueue,
+    stateReconciliationQueue,
+    manualSyncQueue
+  ) {
     this.logInfo('Setting up Bull queue monitoring...')
 
     const { queue: syncProcessingQueue } = this.syncQueue
@@ -172,11 +178,9 @@ class ServiceRegistry {
       queues: [
         stateMonitoringAdapter,
         stateReconciliationAdapter,
-        new BullAdapter(this.cNodeEndpointToSpIdMapQueue, {
-          readOnlyMode: true
-        }),
+        new BullAdapter(manualSyncQueue, { readOnlyMode: true }),
+        new BullAdapter(cNodeEndpointToSpIdMapQueue, { readOnlyMode: true }),
         new BullAdapter(this.stateMachineQueue, { readOnlyMode: true }),
-        new BullAdapter(this.manualSyncQueue, { readOnlyMode: true }),
         new BullAdapter(this.recurringSyncQueue, { readOnlyMode: true }),
         new BullAdapter(syncProcessingQueue, { readOnlyMode: true }),
         new BullAdapter(asyncProcessingQueue, { readOnlyMode: true }),
@@ -289,7 +293,8 @@ class ServiceRegistry {
     const {
       stateMonitoringQueue,
       cNodeEndpointToSpIdMapQueue,
-      stateReconciliationQueue
+      stateReconciliationQueue,
+      manualSyncQueue
     } = await this.stateMachineManager.init(this.libs, this.prometheusRegistry)
     this.stateMonitoringQueue = stateMonitoringQueue
     this.cNodeEndpointToSpIdMapQueue = cNodeEndpointToSpIdMapQueue
@@ -298,6 +303,7 @@ class ServiceRegistry {
     // SyncQueue construction (requires L1 identity)
     // Note - passes in reference to instance of self (serviceRegistry), a very sub-optimal workaround
     this.syncQueue = new SyncQueue(config, this.redis, this)
+    this.manualSyncQueue = manualSyncQueue
 
     // L2URSMRegistration (requires L1 identity)
     // Retries indefinitely
@@ -313,7 +319,13 @@ class ServiceRegistry {
     await this.skippedCIDsRetryQueue.init()
 
     try {
-      this._setupBullMonitoring(app)
+      this.setupBullMonitoring(
+        app,
+        stateMonitoringQueue,
+        cNodeEndpointToSpIdMapQueue,
+        stateReconciliationQueue,
+        manualSyncQueue
+      )
     } catch (e) {
       this.logError(
         `Failed to initialize bull monitoring UI: ${e.message || e}. Skipping..`
@@ -441,10 +453,8 @@ class ServiceRegistry {
    */
   async _initSnapbackSM() {
     this.snapbackSM = new SnapbackSM(config, this.libs)
-    const { stateMachineQueue, manualSyncQueue, recurringSyncQueue } =
-      this.snapbackSM
+    const { stateMachineQueue, recurringSyncQueue } = this.snapbackSM
     this.stateMachineQueue = stateMachineQueue
-    this.manualSyncQueue = manualSyncQueue
     this.recurringSyncQueue = recurringSyncQueue
 
     let isInitialized = false

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

@@ -29,9 +29,8 @@ class StateMachineManager {
         audiusLibs.discoveryProvider.discoveryProviderEndpoint,
         prometheusRegistry
       )
-    const stateReconciliationQueue = await stateReconciliationManager.init(
-      prometheusRegistry
-    )
+    const { stateReconciliationQueue, manualSyncQueue } =
+      await stateReconciliationManager.init(prometheusRegistry)
 
     // Upon completion, make jobs record metrics and enqueue other jobs as necessary
     stateMonitoringQueue.on(
@@ -60,7 +59,8 @@ class StateMachineManager {
     return {
       stateMonitoringQueue,
       cNodeEndpointToSpIdMapQueue,
-      stateReconciliationQueue
+      stateReconciliationQueue,
+      manualSyncQueue
     }
   }
 

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

@@ -5,6 +5,9 @@ module.exports = {
   // Max number of completed/failed jobs to keep in redis for the state monitoring queue
   RECONCILIATION_QUEUE_HISTORY: 300,
 
+  // Max number of completed/failed jobs to keep in redis for the state monitoring queue
+  MANUAL_SYNC_QUEUE_HISTORY: 300,
+
   // Max millis to run a fetch cNodeEndpoint->spId mapping job for before marking it as stalled (1 minute)
   C_NODE_ENDPOINT_TO_SP_ID_MAP_QUEUE_MAX_JOB_RUNTIME_MS: 1000 * 60,
 
@@ -53,7 +56,9 @@ module.exports = {
     // Max number of completed/failed jobs to keep in redis for the cNodeEndpoint->spId map queue
     C_NODE_ENDPOINT_TO_SP_ID_MAP: 100,
     // Max number of completed/failed jobs to keep in redis for the state monitoring queue
-    STATE_RECONCILIATION: 300
+    STATE_RECONCILIATION: 300,
+    // Max number of completed/failed jobs to keep in redis for the manual sync queue
+    MANUAL_SYNC: 300
   }),
 
   QUEUE_NAMES: Object.freeze({
@@ -62,7 +67,9 @@ module.exports = {
     // Name of queue that only processes jobs to fetch the cNodeEndpoint->spId mapping,
     C_NODE_ENDPOINT_TO_SP_ID_MAP: 'c-node-to-endpoint-sp-id-map-queue',
     // Name of StateReconciliationQueue
-    STATE_RECONCILIATION: 'state-reconciliation-queue'
+    STATE_RECONCILIATION: 'state-reconciliation-queue',
+    // Name of ManualSyncQueue
+    MANUAL_SYNC: 'manual-sync-queue'
   }),
 
   JOB_NAMES: Object.freeze({