Add writeSyncLog

Author: triracle97

orm.js

@@ -89,6 +89,10 @@ class Orm extends EventEmitter {
   getOptions(collectionName, dbName) {
   }
 
+  //mock
+  writeSyncLog(type, mess) {
+  }
+
   plugin(plugin, ...args) {
     const _this = new Proxy(Orm, proxyHandlerFactory(this));
     plugin(_this, ...args);

sync/sync-bulk-utils.js

@@ -1,5 +1,6 @@
 const _ = require('lodash')
 const jsonfn = require('json-fn')
+const { EVENT_CONSTANT } = require('./sync-log')
 
 module.exports = function (orm) {
   orm.doCreateBulk = doCreateBulk
@@ -163,6 +164,7 @@ module.exports = function (orm) {
     await orm.removeFakeOfCollection(col, {})
     while (true) {
       try {
+        orm.writeSyncLog(EVENT_CONSTANT.DO_BULK_QUERY, bulkOp.map(op => op._id))
         if (!bulkOp.length) return
         await orm(col).bulkWrite(bulkOp).direct()
         await orm('CommitData').updateOne({}, {  [`highestCommitIdOfCollection.${col}`]: _.last(bulkOp).id  })
@@ -173,9 +175,11 @@ module.exports = function (orm) {
           && e.errors[1].writeErrors.length && e.errors[1].writeErrors[0].err && e.errors[1].writeErrors[0].err.index !== undefined) {
           const index = e.errors[1].writeErrors[0].err.index
           await orm('CommitData').updateOne({}, { [`highestCommitIdOfCollection.${col}`]: bulkOp[index].id })
+          orm.writeSyncLog(EVENT_CONSTANT.BULK_ERROR, bulkOp[index]._id)
           orm.setHighestCommitIdOfCollection(col, bulkOp[index].id)
           bulkOp = bulkOp.slice(index + 1)
         } else {
+          orm.writeSyncLog(EVENT_CONSTANT.DO_BULK_QUERY, bulkOp.map(op => op._id))
           console.error('Wrong thing happened in bulk write', e.message)
           return
         }
@@ -187,6 +191,7 @@ module.exports = function (orm) {
     const bulkOp = {}
     if (!commits.length)
       return
+    orm.writeSyncLog(EVENT_CONSTANT.EXEC_BULK, commits.map(commit => commit._id))
     for (const commit of commits) {
       const { value: highestId } = await orm.emit('getHighestCommitId', commit.dbName)
       if (commit.id <= highestId) continue
@@ -197,6 +202,7 @@ module.exports = function (orm) {
           _query.chain[0].fn = convertedQuery[_query.chain[0].fn]
         const convertedChain = convertChain(_query.chain)
         convertedChain.id = commit.id
+        convertedChain._id = commit._id
         if (convertedChain) {
           const run = !(await orm.emit(`commit:handler:shouldNotExecCommand:${commit.collectionName}`, commit));
           if (!run) continue

sync/sync-flow.js

@@ -1,6 +1,7 @@
 const AwaitLock = require('await-lock').default;
 const _ = require('lodash');
 const jsonFn = require('json-fn')
+const { EVENT_CONSTANT } = require('./sync-log')
 
 module.exports = function (orm, role) {
   let masterDbMap = (orm.mode === 'multi' ? {} : false)
@@ -47,13 +48,15 @@ module.exports = function (orm, role) {
     }
     //todo: [process:commit] can return array
     let _commit = _.cloneDeep(commit)
+    orm.writeSyncLog(EVENT_CONSTANT.START_FLOW, commit)
     await orm.emit(`process:commit:${commit.collectionName}`, _commit, target)
     if (_commit.tags) {
       for (const tag of _commit.tags) {
         await orm.emit(`process:commit:${tag}`, _commit)
       }
     }
     commit = _commit;
+    orm.writeSyncLog(EVENT_CONSTANT.AFTER_FIRST_PROCESS, commit)
 
     let value
     if (!checkMaster(commit.dbName)) {
@@ -65,6 +68,7 @@ module.exports = function (orm, role) {
       commit.fromClient = (await orm.emit('getCommitDataId')).value
       commit.createdDate = new Date()
       commit._fakeId = fakeId
+      orm.writeSyncLog(EVENT_CONSTANT.SEND_TO_MASTER, commit._id)
       orm.emit('transport:toMaster', commit)
       fakeId += 1
       await orm('CommitData').updateOne({}, { fakeId })
@@ -73,6 +77,7 @@ module.exports = function (orm, role) {
       commit.fromMaster = true;
       const lock = new AwaitLock()
       await lock.acquireAsync()
+      orm.writeSyncLog(EVENT_CONSTANT.START_EXECUTE)
       orm.once(`commit:result:${commit._id.toString()}`, function (result) {
         value = result
         lock.release()
@@ -117,17 +122,20 @@ module.exports = function (orm, role) {
 
   orm.onQueue('update:Commit:c', 'fake-channel', async function (commit) {
     if (!commit.id) return
+    orm.writeSyncLog(EVENT_CONSTANT.FLOW_EXEC, commit._id)
     if (!checkMaster(commit.dbName)) {
       await orm.emit('commit:update-fake', commit);
     }
     const run = !(await orm.emit(`commit:handler:shouldNotExecCommand:${commit.collectionName}`, commit));
+    orm.writeSyncLog(EVENT_CONSTANT.FLOW_SHOULD_NOT_EXEC, commit._id + ' ' + run)
     let result
     if (run) {
       let query = orm.getQuery(commit)
       if (commit.dbName) query.name += `@${commit.dbName}`
       try {
         result = await orm.execChain(query)
       } catch (e) {
+        orm.writeSyncLog(EVENT_CONSTANT.FLOW_EXEC_ERROR, commit._id + ' ' + e.message)
         console.error('Error on query', jsonFn.stringify(query), 'is', e)
         await orm.emit('commit:report:errorExec', commit.id, e.message)
       }

sync/sync-log.js

@@ -0,0 +1,23 @@
+module.exports.EVENT_CONSTANT = {
+  START_FLOW: 0,
+  AFTER_FIRST_PROCESS: 1,
+  SEND_TO_MASTER: 2,
+  BUILD_FAKE: 3,
+  ADDED_TO_QUEUE: 4,
+  DO_SEND: 5,
+  FINISH_SEND: 6,
+  COMMIT_REQUEST: 7,
+
+  START_EXECUTE: 100,
+  CREATE_COMMIT: 101,
+  PERFECT_FORM: 102,
+
+  REQUIRE_SYNC: 199,
+  EXEC_BULK: 200,
+  DO_BULK_QUERY: 201,
+  BULK_ERROR: 202,
+  CREATE_COMMIT_CLIENT: 203,
+  FLOW_EXEC: 204,
+  FLOW_SHOULD_NOT_EXEC: 205,
+  FLOW_EXEC_ERROR: 206
+}

sync/sync-plugin-multi.js

@@ -10,6 +10,7 @@ const replaceMasterUtils = require('./sync-utils/replace-master')
 const storeOldCommits = require('./sync-utils/store-old-commits')
 const {handleExtraProps} = require('./sync-handle-extra-props');
 const AwaitLock = require('await-lock').default
+const { EVENT_CONSTANT } = require('./sync-log')
 
 const syncPlugin = function (orm) {
   const whitelist = []
@@ -420,6 +421,7 @@ const syncPlugin = function (orm) {
 
     orm.onQueue("commit:build-fake", 'fake-channel', async function (query, target, commit) {
       if (!commit.chain) return;
+      orm.writeSyncLog(EVENT_CONSTANT.BUILD_FAKE, commit._id)
       const isDeleteCmd = target.cmd.includes('delete') || target.cmd.includes('remove')
       const fakeCollectionName = 'Recovery' + commit.collectionName
       const _query = orm.getQuery(commit)
@@ -621,6 +623,8 @@ const syncPlugin = function (orm) {
 
   orm.onQueue('transport:requireSync:callback', async function (commits) {
     if (!commits || !commits.length) return
+    const _idsList = commits.map(commit => commit._id)
+    orm.writeSyncLog(EVENT_CONSTANT.REQUIRE_SYNC, _idsList)
     const archivedCommits = _.remove(commits, commit => commit.data && !!commit.data.__arc)
     if (commits.length > COMMIT_BULK_WRITE_THRESHOLD) {
       await validateCommits(commits)
@@ -719,6 +723,7 @@ const syncPlugin = function (orm) {
   orm.onQueue('createCommit', async function (commit) {
     if (lock.acquired)
       return
+    orm.writeSyncLog(orm.isMaster() ? EVENT_CONSTANT.CREATE_COMMIT : EVENT_CONSTANT.CREATE_COMMIT_CLIENT, commit._id)
     if (!commit.id) {
       let { value: highestId } = await orm.emit('getHighestCommitId', commit.dbName);
       if (highestIdInMemory)
@@ -738,6 +743,7 @@ const syncPlugin = function (orm) {
       if (orm.isMaster()) {
         commit.execDate = new Date()
         commit.isPending = true
+        orm.writeSyncLog(EVENT_CONSTANT.PERFECT_FORM, commit)
       }
       this.value = await orm(`Commit`, commit.dbName).create(commit);
     } catch (e) {
@@ -803,6 +809,7 @@ const syncPlugin = function (orm) {
 
   orm.onQueue('commitRequest', async function (commits) {
     try {
+      orm.writeSyncLog(EVENT_CONSTANT.COMMIT_REQUEST, commits.map(commit => commit._id))
       if (!commits || !commits.length)
         return
       for (let commit of commits) {

sync/sync-queue-commit.js

@@ -1,6 +1,7 @@
 const QUEUE_COMMIT_MODEL = 'QueueCommit'
 const dayjs = require('dayjs')
 const _ = require('lodash')
+const { EVENT_CONSTANT } = require('./sync-log')
 
 module.exports = function (orm) {
 	let disabled = false
@@ -43,6 +44,7 @@ module.exports = function (orm) {
 			commit,
 			dateAdded: new Date()
 		})
+		orm.writeSyncLog(EVENT_CONSTANT.ADDED_TO_QUEUE, commit._id)
 		if (orm.getQueue('queue:send').length <= 1)
 			orm.emit('queue:send')
 	})
@@ -51,6 +53,7 @@ module.exports = function (orm) {
 		const removedCommitUUID = _queueCommit.map((commit) => {
 			return commit._id
 		})
+		orm.writeSyncLog(EVENT_CONSTANT.FINISH_SEND, removedCommitUUID)
 		await orm(QUEUE_COMMIT_MODEL).deleteMany({ 'commit._id': { $in: removedCommitUUID } })
 	})
 
@@ -73,6 +76,11 @@ module.exports = function (orm) {
 		if (disabled) return
 		console.log('Try to send to master')
 		const data = await orm(QUEUE_COMMIT_MODEL).find()
+		try {
+			orm.writeSyncLog(EVENT_CONSTANT.DO_SEND, data.map(item => item.commit._id))
+		} catch (e) {
+			console.error(e)
+		}
 		await orm.emit('transport:send', data.map(item => item.commit))
 	}