[CON-225] Remove manual sync from snapbackSM

creator-node/src/middlewares.js

@@ -269,7 +269,7 @@ async function ensureStorageMiddleware(req, res, next) {
  */
 async function issueAndWaitForSecondarySyncRequests(req) {
   const serviceRegistry = req.app.get('serviceRegistry')
-  const { snapbackSM } = serviceRegistry
+  const { StateMachineManager, manualSyncQueue } = serviceRegistry
 
   // Parse request headers
   const pollingDurationMs =
@@ -332,11 +332,12 @@ async function issueAndWaitForSecondarySyncRequests(req) {
     const replicationStart = Date.now()
     try {
       const secondaryPromises = secondaries.map((secondary) => {
-        return snapbackSM.issueSyncRequestsUntilSynced(
+        return StateMachineManager.issueSyncRequestsUntilSynced(
           secondary,
           wallet,
           primaryClockVal,
-          pollingDurationMs
+          pollingDurationMs,
+          manualSyncQueue
         )
       })
 

creator-node/src/serviceRegistry.js

@@ -133,7 +133,8 @@ class ServiceRegistry {
     app,
     stateMonitoringQueue,
     cNodeEndpointToSpIdMapQueue,
-    stateReconciliationQueue
+    stateReconciliationQueue,
+    manualSyncQueue
   ) {
     this.logInfo('Setting up Bull queue monitoring...')
 
@@ -178,7 +179,7 @@ class ServiceRegistry {
         stateReconciliationAdapter,
         new BullAdapter(cNodeEndpointToSpIdMapQueue, { readOnlyMode: true }),
         new BullAdapter(this.stateMachineQueue, { readOnlyMode: true }),
-        new BullAdapter(this.manualSyncQueue, { readOnlyMode: true }),
+        new BullAdapter(manualSyncQueue, { readOnlyMode: true }),
         new BullAdapter(this.recurringSyncQueue, { readOnlyMode: true }),
         new BullAdapter(syncProcessingQueue, { readOnlyMode: true }),
         new BullAdapter(asyncProcessingQueue, { readOnlyMode: true }),
@@ -289,12 +290,14 @@ class ServiceRegistry {
     const {
       stateMonitoringQueue,
       cNodeEndpointToSpIdMapQueue,
-      stateReconciliationQueue
+      stateReconciliationQueue,
+      manualSyncQueue
     } = await this.stateMachineManager.init(this.libs, this.prometheusRegistry)
 
     // 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
@@ -314,7 +317,8 @@ class ServiceRegistry {
         app,
         stateMonitoringQueue,
         cNodeEndpointToSpIdMapQueue,
-        stateReconciliationQueue
+        stateReconciliationQueue,
+        manualSyncQueue
       )
     } catch (e) {
       this.logError(
@@ -443,10 +447,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

@@ -1,13 +1,18 @@
 const _ = require('lodash')
+const axios = require('axios')
 
+const Utils = require('../utils')
 const config = require('../../config')
 const { logger: baseLogger } = require('../../logging')
 const StateMonitoringManager = require('./stateMonitoring')
 const StateReconciliationManager = require('./stateReconciliation')
 const NodeToSpIdManager = require('./CNodeToSpIdMapManager')
-const { RECONFIG_MODES } = require('./stateMachineConstants')
+const { RECONFIG_MODES, SyncType } = require('./stateMachineConstants')
 const QueueInterfacer = require('./QueueInterfacer')
 const makeOnCompleteCallback = require('./makeOnCompleteCallback')
+const {
+  getNewOrExistingSyncReq
+} = require('./stateReconciliation/stateReconciliationUtils')
 
 /**
  * Manages the queue for monitoring the state of Content Nodes and
@@ -29,9 +34,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 +64,8 @@ class StateMachineManager {
     return {
       stateMonitoringQueue,
       cNodeEndpointToSpIdMapQueue,
-      stateReconciliationQueue
+      stateReconciliationQueue,
+      manualSyncQueue
     }
   }
 
@@ -133,6 +138,98 @@ class StateMachineManager {
     this.highestEnabledReconfigMode = highestEnabledReconfigMode
     this.enabledReconfigModesSet = enabledReconfigModesSet
   }
+
+  /**
+   * Issues syncRequest for user against secondary, and polls for replication up to primary
+   * If secondary fails to sync within specified timeoutMs, will error
+   */
+  async issueSyncRequestsUntilSynced(
+    secondaryUrl,
+    wallet,
+    primaryClockVal,
+    timeoutMs,
+    queue
+  ) {
+    // Issue syncRequest before polling secondary for replication
+    const { duplicateSyncReq, syncReqToEnqueue } = getNewOrExistingSyncReq({
+      userWallet: wallet,
+      secondaryEndpoint: secondaryUrl,
+      primaryEndpoint: this.endpoint,
+      syncType: SyncType.Manual,
+      immediate: true
+    })
+    if (!_.isEmpty(duplicateSyncReq)) {
+      // Log duplicate and return
+      baseLogger.warn(`Duplicate sync request: ${duplicateSyncReq}`)
+      return
+    } else if (!_.isEmpty(syncReqToEnqueue)) {
+      const { jobName, jobData } = syncReqToEnqueue
+      await queue.add(jobName, jobData)
+    } else {
+      // Log error that the sync request couldn't be created and return
+      baseLogger.error(
+        `Failed to create manual sync request: ${duplicateSyncReq}`
+      )
+      return
+    }
+
+    // Poll clock status and issue syncRequests until secondary is caught up or until timeoutMs
+    const start = Date.now()
+    while (Date.now() - start < timeoutMs) {
+      try {
+        // Retrieve secondary clock status for user
+        const secondaryClockStatusResp = await axios({
+          method: 'get',
+          baseURL: secondaryUrl,
+          url: `/users/clock_status/${wallet}`,
+          responseType: 'json',
+          timeout: 1000 // 1000ms = 1s
+        })
+        const { clockValue: secondaryClockVal, syncInProgress } =
+          secondaryClockStatusResp.data.data
+
+        // If secondary is synced, return successfully
+        if (secondaryClockVal >= primaryClockVal) {
+          return
+
+          // Else, if a sync is not already in progress on the secondary, issue a new SyncRequest
+        } else if (!syncInProgress) {
+          const { duplicateSyncReq, syncReqToEnqueue } =
+            getNewOrExistingSyncReq({
+              userWallet: wallet,
+              secondaryEndpoint: secondaryUrl,
+              primaryEndpoint: this.endpoint,
+              syncType: SyncType.Manual
+            })
+          if (!_.isEmpty(duplicateSyncReq)) {
+            // Log duplicate and return
+            baseLogger.warn(`Duplicate sync request: ${duplicateSyncReq}`)
+            return
+          } else if (!_.isEmpty(syncReqToEnqueue)) {
+            const { jobName, jobData } = syncReqToEnqueue
+            await queue.add(jobName, jobData)
+          } else {
+            // Log error that the sync request couldn't be created and return
+            baseLogger.error(
+              `Failed to create manual sync request: ${duplicateSyncReq}`
+            )
+            return
+          }
+        }
+
+        // Give secondary some time to process ongoing or newly enqueued sync
+        // NOTE - we might want to make this timeout longer
+        await Utils.timeout(500)
+      } catch (e) {
+        // do nothing and let while loop continue
+      }
+    }
+
+    // This condition will only be hit if the secondary has failed to sync within timeoutMs
+    throw new Error(
+      `Secondary ${secondaryUrl} did not sync up to primary for user ${wallet} within ${timeoutMs}ms`
+    )
+  }
 }
 
 module.exports = StateMachineManager