CON-328 Add manual mergePrimaryAndSecondary route for testing

creator-node/.gitignore

@@ -15,3 +15,4 @@ coverage/
 .nyc_output/
 compose/env/tmp*/
 file_storage/
+build/

creator-node/package.json

@@ -13,7 +13,8 @@
  "test:coverage": "nyc --reporter=lcov --reporter=text npm run test",
  "test:coverage:ci": "nyc --reporter=lcov --reporter=text npm run test:ci && nyc report --reporter=text-lcov | coveralls",
  "lint:fix": "eslint --fix --ext=js,ts src",
- "lint": "eslint --ext=js,ts src"
+ "lint": "eslint --ext=js,ts src",
+ "build": "./node_modules/.bin/tsc --project ./"
  },
  "keywords": [],
  "author": "",

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

@@ -12,6 +12,10 @@ const {
  respondToURSMRequestForSignature
 } = require('./URSMRegistrationComponentService')
 const { ensureStorageMiddleware } = require('../../middlewares')
+const {
+ SyncType,
+ SYNC_MODES
+} = require('../../services/stateMachineManager/stateMachineConstants')
 const {
  enqueueSync,
  processManualImmediateSync
@@ -159,6 +163,50 @@ const syncRouteController = async (req, res) => {
  return successResponse()
 }
 
+/**
+ * Adds a job to manualSyncQueue to issue a sync to secondary with syncMode MergePrimaryAndSecondary
+ * @notice This will only work if called on a primary for a user
+ */
+const mergePrimaryAndSecondaryController = async (req, res) => {
+ const serviceRegistry = req.app.get('serviceRegistry')
+ const manualSyncQueue = serviceRegistry.manualSyncQueue
+ const config = serviceRegistry.nodeConfig
+
+ const selfEndpoint = config.get('creatorNodeEndpoint')
+
+ const wallet = req.query.wallet
+ const endpoint = req.query.endpoint
+
+ if (!wallet || !endpoint) {
+ return errorResponseBadRequest(`Must provide wallet and endpoint params`)
+ }
+
+ const syncType = SyncType.Manual
+ const syncMode = SYNC_MODES.MergePrimaryAndSecondary
+ const immediate = true
+
+ const syncRequestParameters = {
+ baseURL: endpoint,
+ url: '/sync',
+ method: 'post',
+ data: {
+ wallet: [wallet],
+ creator_node_endpoint: selfEndpoint,
+ sync_type: syncType,
+ immediate,
+ from_manual_route: true
+ }
+ }
+
+ await manualSyncQueue.add({
+ syncType,
+ syncMode,
+ syncRequestParameters
+ })
+
+ return successResponse()
+}
+
 // Routes
 
 router.get(
@@ -170,5 +218,9 @@ router.post(
  ensureStorageMiddleware,
  handleResponse(syncRouteController)
 )
+router.post(
+ '/merge_primary_and_secondary',
+ handleResponse(mergePrimaryAndSecondaryController)
+)
 
 module.exports = router

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

@@ -208,8 +208,10 @@ async function _handleIssueSyncRequest({
  * Eventually should make this more robust, but proceeding with caution
  */
  if (syncMode === SYNC_MODES.MergePrimaryAndSecondary) {
- // Short-circuit if this syncMode is disabled
- if (!mergePrimaryAndSecondaryEnabled) {
+ const fromManualRoute = syncRequestParameters.data.from_manual_route
+
+ // Short-circuit if this syncMode is disabled or if manual route override not provided
+ if (!mergePrimaryAndSecondaryEnabled && !fromManualRoute) {
  return { result: 'success_mode_disabled' }
  }
 

creator-node/src/services/stateMachineManager/stateReconciliation/types.ts

@@ -6,6 +6,7 @@ export type SyncRequestAxiosData = {
  forceResync?: boolean
  timestamp?: string
  signature?: string
+ from_manual_route?: boolean
 }
 
 export type SyncRequestAxiosParams = {

creator-node/src/services/sync/primarySyncFromSecondary.js

@@ -30,7 +30,7 @@ module.exports = async function primarySyncFromSecondary({
 }) {
  const logPrefix = `[primarySyncFromSecondary][Wallet: ${wallet}][Secondary: ${secondary}]`
  const logger = genericLogger.child(logContext)
- logger.info(`[primarySyncFromSecondary] [Wallet: ${wallet}] Beginning...`)
+ logger.info(`${logPrefix} Beginning...`)
  const start = Date.now()
 
  // This is used only for logging record endpoint of requesting node
@@ -53,13 +53,21 @@ module.exports = async function primarySyncFromSecondary({
  )
 
  // TODO should be able to pass this through from StateMachine / caller
-const userReplicaSet = await getUserReplicaSet({
+let userReplicaSet = await getUserReplicaSet({
  wallet,
  selfEndpoint,
  logger,
  libs
  })
 
+ // Error if this node is not primary for user
+ if (userReplicaSet[0] !== selfEndpoint) {
+ throw new Error(`Failure - this node is not primary for user`)
+ }
+
+ // filter out current node from user's replica set
+ userReplicaSet = userReplicaSet.filter((url) => url !== selfEndpoint)
+
  // Keep importing data from secondary until full clock range has been retrieved
  let completed = false
  let exportClockRangeMin = 0
@@ -448,7 +456,7 @@ async function filterOutAlreadyPresentDBEntries({
  return filteredEntries
 }
 
-async function getUserReplicaSet({ wallet, selfEndpoint, libs, logger }) {
+async function getUserReplicaSet({ wallet, libs, logger }) {
  try {
  let userReplicaSet = await getCreatorNodeEndpoints({
  libs,
@@ -459,9 +467,6 @@ async function getUserReplicaSet({ wallet, selfEndpoint, libs, logger }) {
  myCnodeEndpoint: null
  })
 
- // filter out current node from user's replica set
- userReplicaSet = userReplicaSet.filter((url) => url !== selfEndpoint)
-
  // Spread + set uniq's the array
  userReplicaSet = [...new Set(userReplicaSet)]