Fix miscellaneous state machine bugs

creator-node/src/serviceRegistry.js

@@ -110,11 +110,36 @@ class ServiceRegistry {
  const { queue: sessionExpirationQueue } = this.sessionExpirationQueue
  const { queue: skippedCidsRetryQueue } = this.skippedCIDsRetryQueue
 
+ // Make state machine queues truncate long data (they have jobs with large inputs and outputs)
+ const stateMonitoringAdapter = new BullAdapter(stateMonitoringQueue, {
+ readOnlyMode: true
+ })
+ const stateReconciliationAdapter = new BullAdapter(
+ stateReconciliationQueue,
+ { readOnlyMode: true }
+ )
+
+ // These queues have very large inputs and outputs, so we truncate job
+ // data and results that are nested >=5 levels or contain strings >=10,000 characters
+ stateMonitoringAdapter.setFormatter('data', this.truncateBull.bind(this))
+ stateMonitoringAdapter.setFormatter(
+ 'returnValue',
+ this.truncateBull.bind(this)
+ )
+ stateReconciliationAdapter.setFormatter(
+ 'data',
+ this.truncateBull.bind(this)
+ )
+ stateReconciliationAdapter.setFormatter(
+ 'returnValue',
+ this.truncateBull.bind(this)
+ )
+
  // Dashboard to view queues at /health/bull endpoint. See https://github.com/felixmosh/bull-board#hello-world
  createBullBoard({
  queues: [
-new BullAdapter(stateMonitoringQueue, { readOnlyMode: true }),
-new BullAdapter(stateReconciliationQueue, { readOnlyMode: true }),
+stateMonitoringAdapter,
+stateReconciliationAdapter,
  new BullAdapter(stateMachineQueue, { readOnlyMode: true }),
  new BullAdapter(manualSyncQueue, { readOnlyMode: true }),
  new BullAdapter(recurringSyncQueue, { readOnlyMode: true }),
@@ -126,13 +151,53 @@ class ServiceRegistry {
  new BullAdapter(sessionExpirationQueue, { readOnlyMode: true }),
  new BullAdapter(skippedCidsRetryQueue, { readOnlyMode: true })
  ],
- serverAdapter: serverAdapter
+ serverAdapter
  })
 
  serverAdapter.setBasePath('/health/bull')
  app.use('/health/bull', serverAdapter.getRouter())
  }
 
+ // Truncates large JSON data in Bull Board after 5 levels of nesting or 10,0000 characters.
+ // Replaces truncated data with [Array-Truncated], [Object-Truncated], or [String-Truncated].
+ // Taken from https://github.com/felixmosh/bull-board/pull/414#issuecomment-1134874761
+ truncateBull(dataToTruncate, curDepth = 0) {
+ if (
+ typeof dataToTruncate === 'object' &&
+ !Array.isArray(dataToTruncate) &&
+ dataToTruncate !== null &&
+ dataToTruncate !== undefined
+ ) {
+ if (curDepth < 5) {
+ const newDepth = curDepth + 1
+ const json = Object.assign({}, dataToTruncate)
+ Object.entries(dataToTruncate).forEach(([key, value]) => {
+ switch (typeof value) {
+ case 'string':
+ json[key] = value.length > 10000 ? '[String-Truncated]' : value
+ break
+ case 'object':
+ if (Array.isArray(value)) {
+ json[key] =
+ JSON.stringify(value).length > 10000
+ ? '[Array-Truncated]'
+ : value
+ } else {
+ json[key] = this.truncateBull(value, newDepth)
+ }
+ break
+ default:
+ json[key] = value
+ break
+ }
+ })
+ return json
+ }
+ return '[Object-Truncated]'
+ }
+ return dataToTruncate
+ }
+
  /**
  * Some services require the node server to be running in order to initialize. Run those here.
  * Specifically:

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

@@ -87,16 +87,22 @@ module.exports = async function ({
 
  // Combine each promise's updateReplicaSetOps into a job
  for (const updateReplicaSetOp of updateReplicaSetOps) {
+ const { wallet } = updateReplicaSetOp
+
  updateReplicaSetJobs.push({
  jobName: JOB_NAMES.UPDATE_REPLICA_SET,
  jobData: {
- wallet: updateReplicaSetOp.wallet,
+ wallet,
  userId: updateReplicaSetOp.user_id,
  primary: updateReplicaSetOp.primary,
  secondary1: updateReplicaSetOp.secondary1,
  secondary2: updateReplicaSetOp.secondary2,
  unhealthyReplicas: Array.from(updateReplicaSetOp.unhealthyReplicas),
- replicaSetNodesToUserClockStatusesMap
+ replicaSetNodesToUserClockStatusesMap:
+ _transformAndFilterNodeToClockValuesMapping(
+ replicaSetNodesToUserClockStatusesMap,
+ wallet
+ )
  }
  })
  }
@@ -295,3 +301,22 @@ const _findReplicaSetUpdatesForUser = async (
 
  return requiredUpdateReplicaSetOps
 }
+
+/**
+ * Transforms data type from ((K1,V1) => (K2,V2)) to (K1,V1[K2]), where K1 is the node endpoint,
+ * V1 is the mapping, and K2 is the wallet to filter by. Default to clock value of -1 whenever the given
+ * wallet isn't present in a node's mapping.
+ * @param {Object} replicaSetNodesToUserClockStatusesMap map of secondary endpoint strings to (map of user wallet strings to clock value of secondary for user)
+ * @param {string} wallet the wallet to filter clock values for (other wallets will be exlucded from the output)
+ * @returns mapping of node endpoint (string) to clock value (number) on that node for the given wallet
+ */
+const _transformAndFilterNodeToClockValuesMapping = (
+ replicaSetNodesToUserClockStatusesMap,
+ wallet
+) => {
+ return Object.fromEntries(
+ Object.entries(replicaSetNodesToUserClockStatusesMap).map(
+ ([node, clockValueMapping]) => [node, clockValueMapping[wallet] || -1]
+ )
+ )
+}

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

@@ -1,3 +1,5 @@
+const _ = require('lodash')
+
 const config = require('../../../config')
 const CNodeToSpIdMapManager = require('../CNodeToSpIdMapManager')
 const { SyncType, QUEUE_NAMES } = require('../stateMachineConstants')
@@ -199,8 +201,11 @@ const _findSyncsForUser = (
  primaryEndpoint: thisContentNodeEndpoint,
  syncType: SyncType.Recurring
  })
- if (syncReqToEnqueue) syncReqsToEnqueue.push(syncReqToEnqueue)
- else if (duplicateSyncReq) duplicateSyncReqs.push(duplicateSyncReq)
+ 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}`

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

@@ -41,18 +41,6 @@ module.exports = async function (
  replicaSetNodesToUserClockStatusesMap,
  enabledReconfigModes
 ) {
- _validateJobData(
- logger,
- primary,
- secondary1,
- secondary2,
- wallet,
- unhealthyReplicas,
- healthyNodes,
- replicaSetNodesToUserClockStatusesMap,
- enabledReconfigModes
- )
-
  /**
  * Fetch all the healthy nodes while disabling sync checks to select nodes for new replica set
  * Note: sync checks are disabled because there should not be any syncs occurring for a particular user
@@ -74,6 +62,18 @@ module.exports = async function (
  'Auto-selecting Content Nodes returned an empty list of healthy nodes.'
  )
 
+ _validateJobData(
+ logger,
+ primary,
+ secondary1,
+ secondary2,
+ wallet,
+ unhealthyReplicas,
+ healthyNodes,
+ replicaSetNodesToUserClockStatusesMap,
+ enabledReconfigModes
+ )
+
  let errorMsg = ''
  let issuedReconfig = false
  let syncJobsToEnqueue = []
@@ -144,7 +144,7 @@ module.exports = async function (
  * @param {string} param.wallet current user's wallet address
  * @param {Set<string>} param.unhealthyReplicasSet a set of endpoints of unhealthy replica set nodes
  * @param {string[]} param.healthyNodes array of healthy Content Node endpoints used for selecting new replica set
- * @param {Object} param.replicaSetNodesToUserClockStatusesMap map of secondary endpoint strings to (map of user wallet strings to clock value of secondary for user)
+ * @param {Object} param.replicaSetNodesToUserClockStatusesMap map of secondary endpoint strings to clock value of secondary for user whose replica set should be updated
  * @param {string[]} param.enabledReconfigModes array of which reconfig modes are enabled
  * @returns {Object}
  * {
@@ -179,7 +179,6 @@ const determineNewReplicaSet = async ({
 
  if (unhealthyReplicasSet.size === 1) {
  return determineNewReplicaSetWhenOneNodeIsUnhealthy(
- wallet,
  primary,
  secondary1,
  secondary2,
@@ -272,17 +271,15 @@ const _validateJobData = (
 
 /**
  * Determines new replica set when one node in the current replica set is unhealthy.
- * @param {*} wallet wallet address of user whose replica set contains 1 unhealthy node to be replaced
  * @param {*} primary user's current primary endpoint
  * @param {*} secondary1 user's current first secondary endpoint
  * @param {*} secondary2 user's current second secondary endpoint
  * @param {*} unhealthyReplicasSet a set of endpoints of unhealthy replica set nodes
- * @param {*} replicaSetNodesToUserClockStatusesMap map of secondary endpoint strings to (map of user wallet strings to clock value of secondary for user)
+ * @param {*} replicaSetNodesToUserClockStatusesMap map of secondary endpoint strings to clock value of secondary for user whose replica set should be updated
  * @param {*} newReplicaNode endpoint of node that will replace the unhealthy node
  * @returns reconfig info to update the user's replica set to replace the 1 unhealthy node
  */
 const determineNewReplicaSetWhenOneNodeIsUnhealthy = (
- wallet,
  primary,
  secondary1,
  secondary2,
@@ -295,8 +292,8 @@ const determineNewReplicaSetWhenOneNodeIsUnhealthy = (
  // value of the two secondaries as the new primary, leave the other as the first secondary, and select a new second secondary
  if (unhealthyReplicasSet.has(primary)) {
  const [newPrimary, currentHealthySecondary] =
- replicaSetNodesToUserClockStatusesMap[secondary1][wallet] >=
- replicaSetNodesToUserClockStatusesMap[secondary2][wallet]
+ replicaSetNodesToUserClockStatusesMap[secondary1] >=
+ replicaSetNodesToUserClockStatusesMap[secondary2]
  ? [secondary1, secondary2]
  : [secondary2, secondary1]
  return {

creator-node/test/findReplicaSetUpdates.jobProcessor.test.js

@@ -74,26 +74,22 @@ describe('test findReplicaSetUpdates job processor', function () {
 
  const DEFAULT_CLOCK_STATUSES_MAP = {
  [primary]: {
- [wallet]: 10
+ [wallet]: 10,
+ randomWallet: 10
  },
  [secondary1]: {
- [wallet]: 10
+ [wallet]: 10,
+ anotherWallet: 100
  },
  [secondary2]: {
  [wallet]: 10
  }
  }
 
- const CLOCK_STATUSES_MAP_WHERE_SECONDARY1_IS_BEHIND = {
- [primary]: {
- [wallet]: 10
- },
- [secondary1]: {
- [wallet]: 9
- },
- [secondary2]: {
- [wallet]: 10
- }
+ const CLOCK_STATUSES_MAP_FILTERED_TO_WALLET = {
+ [primary]: 10,
+ [secondary1]: 10,
+ [secondary2]: 10
  }
 
  function getJobProcessorStub(
@@ -118,7 +114,6 @@ describe('test findReplicaSetUpdates job processor', function () {
  cNodeEndpointToSpIdMap = DEFAULT_CNODE_ENDOINT_TO_SP_ID_MAP,
  userSecondarySyncMetricsMap = {},
  unhealthyPeers = [],
- replicaSetNodesToUserClockStatusesMap = DEFAULT_CLOCK_STATUSES_MAP,
  isPrimaryHealthyInExtraHealthCheck = true
  }) {
  const getCNodeEndpointToSpIdMapStub = sandbox
@@ -138,7 +133,7 @@ describe('test findReplicaSetUpdates job processor', function () {
  logger,
  users,
  unhealthyPeers,
- replicaSetNodesToUserClockStatusesMap,
+ replicaSetNodesToUserClockStatusesMap: DEFAULT_CLOCK_STATUSES_MAP,
  userSecondarySyncMetricsMap
  })
  }
@@ -163,8 +158,7 @@ describe('test findReplicaSetUpdates job processor', function () {
  secondary2,
  unhealthyReplicas: expectedUnhealthyReplicas,
  replicaSetNodesToUserClockStatusesMap:
- jobProcessorArgs?.replicaSetNodesToUserClockStatusesMap ||
- DEFAULT_CLOCK_STATUSES_MAP
+ CLOCK_STATUSES_MAP_FILTERED_TO_WALLET
  }
  }
  ]