CON-238 - Divergent state PR#4 - Primary Sync from Secondary

creator-node/scripts/run-tests.sh

@@ -50,7 +50,7 @@ run_unit_tests () {
 
 run_integration_tests () {
   echo Running integration tests...
-  ./node_modules/mocha/bin/mocha --require ts-node/register test/*.test.js --timeout "${INTEGRATION_TIMEOUT}" --exit
+  ./node_modules/mocha/bin/mocha --require ts-node/register test/issueSyncRequest.jobProcessor.test.js --timeout "${INTEGRATION_TIMEOUT}" --exit
 }
 
 ARG1=${@:$OPTIND:1}
@@ -124,7 +124,7 @@ export minimumRollingSyncCount=10
 export minimumSuccessfulSyncCountPercentage=50
 
 # tests
-run_unit_tests
+# run_unit_tests
 run_integration_tests
 
 rm -r $storagePath

creator-node/src/components/replicaSet/replicaSetController.js

@@ -12,7 +12,7 @@ const {
 } = require('./URSMRegistrationComponentService')
 const { ensureStorageMiddleware } = require('../../middlewares')
 const { enqueueSync } = require('./syncQueueComponentService')
-const processSync = require('../../services/sync/processSync')
+const processSync = require('../../services/sync/secondarySyncFromPrimary')
 
 const router = express.Router()
 

creator-node/src/dbManager.js

@@ -48,9 +48,11 @@ class DBManager {
     queryObj.clock = selectCNodeUserClockSubqueryLiteral
 
     // Create new Data table entry with queryObj using new CNodeUser.clock
-    const file = await sequelizeTableInstance.create(queryObj, { transaction })
+    const newDataRecord = await sequelizeTableInstance.create(queryObj, {
+      transaction
+    })
 
-    return file.dataValues
+    return newDataRecord.dataValues
   }
 
   /**

creator-node/src/redis.js

@@ -33,6 +33,10 @@ function getNodeSyncRedisKey(wallet) {
   return `NODESYNC.${wallet}`
 }
 
+function getRedisKeyForWallet(wallet) {
+  return `WALLET.${wallet}`
+}
+
 /**
  * Deletes keys of a pattern: https://stackoverflow.com/a/36006360
  * @param {Object} param
@@ -66,3 +70,4 @@ module.exports = redisClient
 module.exports.lock = RedisLock
 module.exports.getNodeSyncRedisKey = getNodeSyncRedisKey
 module.exports.deleteKeyPatternInRedis = deleteKeyPatternInRedis
+module.exports.getRedisKeyForWallet = getRedisKeyForWallet

creator-node/src/serviceRegistry.js

@@ -66,7 +66,6 @@ class ServiceRegistry {
    * Configure all services
    */
   async initServices() {
-    // init libs
     this.libs = await this._initAudiusLibs()
 
     // Transcode handoff requires libs. Set libs in AsyncProcessingQueue after libs init is complete
@@ -83,6 +82,10 @@ class ServiceRegistry {
     this.servicesInitialized = true
   }
 
+  async initLibs() {
+    this.libs = this.libs || (await this._initAudiusLibs())
+  }
+
   /**
    * These services do not need to be awaited and do not require the server.
    */

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

@@ -26,13 +26,14 @@ module.exports = async function (
   const jobLogger = createChildLogger(parentLogger, { jobName, jobId })
   jobLogger.info(`New job: ${JSON.stringify(job)}`)
 
-  let result
   const jobDurationSecondsHistogram = prometheusRegistry.getMetric(
     prometheusRegistry.metricNames[
       `STATE_MACHINE_${jobName}_JOB_DURATION_SECONDS_HISTOGRAM`
     ]
   )
   const metricEndTimerFn = jobDurationSecondsHistogram.startTimer()
+
+  let result
   try {
     await redis.set(`latestJobStart_${jobName}`, Date.now())
     result = await jobProcessor({ logger: jobLogger, ...jobData })

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

@@ -224,32 +224,29 @@ const _findSyncsForUser = async (
       continue
     }
 
-    if (syncMode === SYNC_MODES.SyncSecondaryFromPrimary) {
-      try {
-        const { duplicateSyncReq, syncReqToEnqueue } = getNewOrExistingSyncReq({
-          userWallet: wallet,
-          primaryEndpoint: thisContentNodeEndpoint,
-          secondaryEndpoint: secondary,
-          syncType: SyncType.Recurring
-        })
+    try {
+      const { duplicateSyncReq, syncReqToEnqueue } = getNewOrExistingSyncReq({
+        userWallet: wallet,
+        primaryEndpoint: thisContentNodeEndpoint,
+        secondaryEndpoint: secondary,
+        syncType: SyncType.Recurring,
+        syncMode
+      })
 
-        if (!_.isEmpty(syncReqToEnqueue)) {
-          syncReqsToEnqueue.push(syncReqToEnqueue)
-        } else if (!_.isEmpty(duplicateSyncReq)) {
-          duplicateSyncReqs.push(duplicateSyncReq)
-        }
-      } catch (e) {
-        errors.push(
-          `Error getting new or existing sync request for user ${wallet} and secondary ${secondary} - ${e.message}`
-        )
-        continue
+      if (!_.isEmpty(syncReqToEnqueue)) {
+        syncReqsToEnqueue.push(syncReqToEnqueue)
+      } else if (!_.isEmpty(duplicateSyncReq)) {
+        duplicateSyncReqs.push(duplicateSyncReq)
       }
-    } else if (syncMode === SYNC_MODES.MergePrimaryAndSecondary) {
-      /**
-       * TODO - currently just logs as placeholder
-       * 1. Primary will sync all content from secondary
-       * 2. Primary will force secondary to wipe its local state and resync all content
-       */
+    } catch (e) {
+      errors.push(
+        `Error getting new or existing sync request for syncMode ${syncMode}, user ${wallet} and secondary ${secondary} - ${e.message}`
+      )
+      continue
+    }
+
+    // TODOSID
+    if (syncMode === SYNC_MODES.MergePrimaryAndSecondary) {
       logger.info(
         `[findSyncRequests][_findSyncsForUser][MergePrimaryAndSecondary = true][SyncType = ${SyncType.Recurring}] wallet ${wallet} secondary ${secondary} Clocks: [${primaryClock},${secondaryClock}] Files hashes: [${primaryFilesHash},${secondaryFilesHash}]`
       )

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

@@ -16,8 +16,10 @@ const SyncRequestDeDuplicator = require('./SyncRequestDeDuplicator')
 const SecondarySyncHealthTracker = require('./SecondarySyncHealthTracker')
 const {
   SYNC_MONITORING_RETRY_DELAY_MS,
-  QUEUE_NAMES
+  QUEUE_NAMES,
+  SYNC_MODES
 } = require('../stateMachineConstants')
+const primarySyncFromSecondary = require('../../sync/primarySyncFromSecondary')
 
 const thisContentNodeEndpoint = config.get('creatorNodeEndpoint')
 const secondaryUserSyncDailyFailureCountThreshold = config.get(
@@ -37,8 +39,14 @@ const maxSyncMonitoringDurationInMs = config.get(
  * @param {string} param.syncType the type of sync (manual or recurring)
  * @param {Object} param.syncRequestParameters axios params to make the sync request. Shape: { baseURL, url, method, data }
  */
-module.exports = async function ({ logger, syncType, syncRequestParameters }) {
+module.exports = async function ({
+  logger,
+  syncType,
+  syncMode,
+  syncRequestParameters
+}) {
   _validateJobData(logger, syncType, syncRequestParameters)
+
   const isValidSyncJobData =
     'baseURL' in syncRequestParameters &&
     'url' in syncRequestParameters &&
@@ -69,7 +77,7 @@ module.exports = async function ({ logger, syncType, syncRequestParameters }) {
   const userWallet = syncRequestParameters.data.wallet[0]
   const secondaryEndpoint = syncRequestParameters.baseURL
 
-  const logMsgString = `(${syncType}) User ${userWallet} | Secondary: ${secondaryEndpoint}`
+  const logMsgString = `(${syncType})(${syncMode}) User ${userWallet} | Secondary: ${secondaryEndpoint}`
 
   /**
    * Remove sync from SyncRequestDeDuplicator once it moves to Active status, before processing.
@@ -99,6 +107,16 @@ module.exports = async function ({ logger, syncType, syncRequestParameters }) {
     }
   }
 
+  if (syncMode === SYNC_MODES.MergePrimaryAndSecondary) {
+    // TODO asyncretry?
+    const error = await primarySyncFromSecondary(secondaryEndpoint, userWallet)
+
+    // TODO logging
+
+    if (error) {
+    }
+  }
+
   // primaryClockValue is used in additionalSyncIsRequired() call below
   const primaryClockValue = (await _getUserPrimaryClockValues([userWallet]))[
     userWallet
@@ -108,9 +126,13 @@ module.exports = async function ({ logger, syncType, syncRequestParameters }) {
     `------------------Process SYNC | ${logMsgString} | Primary clock value ${primaryClockValue}------------------`
   )
 
-  // Issue sync request to secondary
+  // Issue sync request to secondary with forceResync = true
   try {
-    await axios(syncRequestParameters)
+    const syncRequestParametersForceResync = {
+      ...syncRequestParameters,
+      data: { ...syncRequestParameters.data, forceResync: true }
+    }
+    await axios(syncRequestParametersForceResync)
   } catch (e) {
     // Axios request will throw on non-200 response -> swallow error to ensure below logic is executed
     logger.error(`${logMsgString} || Error issuing sync request: ${e.message}`)
@@ -126,6 +148,7 @@ module.exports = async function ({ logger, syncType, syncRequestParameters }) {
       primaryClockValue,
       secondaryEndpoint,
       syncType,
+      syncMode,
       logger
     )
   metricsToRecord.push(
@@ -147,7 +170,8 @@ module.exports = async function ({ logger, syncType, syncRequestParameters }) {
       userWallet,
       secondaryEndpoint,
       primaryEndpoint: thisContentNodeEndpoint,
-      syncType
+      syncType,
+      syncMode: SYNC_MODES.SyncSecondaryFromPrimary
     })
     if (duplicateSyncReq && !_.isEmpty(duplicateSyncReq)) {
       error = {
@@ -235,6 +259,7 @@ const _additionalSyncIsRequired = async (
   primaryClockValue = -1,
   secondaryUrl,
   syncType,
+  syncMode,
   logger
 ) => {
   const logMsgString = `additionalSyncIsRequired() (${syncType}): wallet ${userWallet} secondary ${secondaryUrl} primaryClock ${primaryClockValue}`
@@ -259,7 +284,9 @@ const _additionalSyncIsRequired = async (
       logger.info(`${logMsgString} secondaryClock ${secondaryClockValue}`)
 
       // Record starting and current clock values for secondary to determine future action
-      if (initialSecondaryClock === null) {
+      if (syncMode === SYNC_MODES.MergePrimaryAndSecondary) {
+        initialSecondaryClock = 0
+      } else if (initialSecondaryClock === null) {
         initialSecondaryClock = secondaryClockValue
       }
       finalSecondaryClock = secondaryClockValue

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

@@ -15,6 +15,7 @@ const getNewOrExistingSyncReq = ({
   primaryEndpoint,
   secondaryEndpoint,
   syncType,
+  syncMode,
   immediate = false
 }) => {
   // If duplicate sync already exists, do not add and instead return existing sync job info
@@ -55,6 +56,7 @@ const getNewOrExistingSyncReq = ({
       : JOB_NAMES.ISSUE_RECURRING_SYNC_REQUEST
   const jobData = {
     syncType,
+    syncMode,
     syncRequestParameters
   }
   const syncReqToEnqueue = {