SnapbackSM - Phase 1 - Recurring Sync (#892)

Enable recurring sync operation separate from UserReplicaSetManager contract through SnapbackSM
AudiusProject · Oct 13, 2020 · 759b2c4 · 759b2c4
1 parent 8cb4ceb
commit 759b2c4
Show file tree

Hide file tree

Showing 14 changed files with 670 additions and 68 deletions.
diff --git a/creator-node/compose/docker-compose.yml b/creator-node/compose/docker-compose.yml
@@ -26,6 +26,7 @@ services:
       - delegateOwnerWallet=${delegateOwnerWallet}
       - delegatePrivateKey=${delegatePrivateKey}
       - creatorNodeEndpoint=${creatorNodeEndpoint}
+      - snapbackDevModeEnabled=${snapbackDevModeEnabled}
     env_file:
       - ./env/base.env
     ports:

diff --git a/creator-node/src/config.js b/creator-node/src/config.js
@@ -2,6 +2,18 @@ const axios = require('axios')
 const convict = require('convict')
 const fs = require('fs')
 
+// Custom boolean format used to ensure that empty string '' is evaluated as false
+// https://github.com/mozilla/node-convict/issues/380
+convict.addFormat({
+  name: 'BooleanCustom',
+  validate: function (val) {
+    return (typeof val === 'boolean') || (typeof val === 'string')
+  },
+  coerce: function (val) {
+    return val === true || val === 'true'
+  }
+})
+
 // Define a schema
 const config = convict({
   dbUrl: {
@@ -365,6 +377,12 @@ const config = convict({
     format: 'nat',
     env: 'rehydrateMaxConcurrency',
     default: 10
+  },
+  snapbackDevModeEnabled: {
+    doc: 'TEST ONLY. DO NOT CONFIGURE MANUALLY. Disables automatic secondary sync issuing in order to test SnapbackSM.',
+    format: 'BooleanCustom',
+    env: 'snapbackDevModeEnabled',
+    default: false
   }
 
   // unsupported options at the moment

diff --git a/creator-node/src/middlewares.js b/creator-node/src/middlewares.js
@@ -95,7 +95,7 @@ async function ensurePrimaryMiddleware (req, res, next) {
  * @dev - Is not a middleware so it can be run before responding to client.
  */
 async function triggerSecondarySyncs (req) {
-  if (config.get('isUserMetadataNode')) return
+  if (config.get('isUserMetadataNode') || config.get('snapbackDevModeEnabled')) return
   try {
     if (!req.session.nodeIsPrimary || !req.session.creatorNodeEndpoints || !Array.isArray(req.session.creatorNodeEndpoints)) return
     const [primary, ...secondaries] = req.session.creatorNodeEndpoints

diff --git a/creator-node/src/routes/nodeSync.js b/creator-node/src/routes/nodeSync.js
@@ -168,6 +168,9 @@ module.exports = function (app) {
     const immediate = (req.body.immediate === true || req.body.immediate === 'true')
     // option to sync just the db records as opposed to db records and files on disk, defaults to false
     const dbOnlySync = (req.body.db_only_sync === true || req.body.db_only_sync === 'true')
+    // Log if initiated from SnapbackSM
+    const stateMachineInitiatedSync = (req.body.state_machine === true || req.body.state_machine === 'true')
+    if (stateMachineInitiatedSync) req.logger.info(`SnapbackSM sync initiated for ${walletPublicKeys} from ${creatorNodeEndpoint}`)
 
     if (!immediate) {
       req.logger.info('debounce time', config.get('debounceTime'))
@@ -308,20 +311,22 @@ async function _nodesync (req, walletPublicKeys, creatorNodeEndpoint, dbOnlySync
           transaction
         })
         const fetchedLatestBlockNumber = fetchedCNodeUser.latestBlockNumber
+        const fetchedLatestClockVal = fetchedCNodeUser.clock
 
         // Delete any previously stored data for cnodeUser in reverse table dependency order (cannot be parallelized).
         if (cnodeUser) {
           // Ensure imported data has higher blocknumber than already stored.
-          // TODO - replace this check with a clock check (!!!)
           const latestBlockNumber = cnodeUser.latestBlockNumber
-
-          if (!dbOnlySync) {
-            if ((fetchedLatestBlockNumber === -1 && latestBlockNumber !== -1) ||
-              (fetchedLatestBlockNumber !== -1 && fetchedLatestBlockNumber <= latestBlockNumber)
-            ) {
-              throw new Error(`Imported data is outdated, will not sync. Imported latestBlockNumber \
-                ${fetchedLatestBlockNumber} Self latestBlockNumber ${latestBlockNumber}`)
-            }
+          const latestClockValue = cnodeUser.clock
+
+          if (latestClockValue > fetchedLatestClockVal) {
+            throw new Error(`Imported data is outdated, will not sync. Imported latestBlockNumber \
+              ${fetchedLatestBlockNumber} Self latestBlockNumber ${latestBlockNumber}. \
+              fetched latestClockVal: ${fetchedLatestClockVal}, self latestClockVal: ${latestClockValue}`)
+          } else if (latestClockValue === fetchedLatestClockVal) {
+            // Already to update, no sync necessary
+            req.logger.info(`User ${fetchedWalletPublicKey} already up to date! fetchedLatestClockVal=${fetchedLatestClockVal}, latestClockValue=${latestClockValue}`)
+            continue
           }
 
           const cnodeUserUUID = cnodeUser.cnodeUserUUID

diff --git a/creator-node/src/routes/users.js b/creator-node/src/routes/users.js
@@ -170,4 +170,23 @@ module.exports = function (app) {
 
     return successResponse({ clockValue })
   }))
+  /**
+   * Returns latest clock value stored in CNodeUsers entry given wallet, or -1 if no entry found
+   */
+  app.post('/users/batch_clock_status', handleResponse(async (req, res) => {
+    const { walletPublicKeys } = req.body
+    const cnodeUsers = await models.CNodeUser.findAll({
+      where: {
+        walletPublicKey: {
+          [models.Sequelize.Op.in]: walletPublicKeys
+        }
+      }
+    })
+    return successResponse({
+      users: cnodeUsers.map(cnodeUser => ({
+        walletPublicKey: cnodeUser.walletPublicKey,
+        clock: cnodeUser.clock
+      }))
+    })
+  }))
 }