Merge pull request #30 from Permissionless-Software-Foundation/ct-unstable

fix(JSON RPC): Prevening processing of duplicate entries

package.json

@@ -1,6 +1,6 @@
 {
   "name": "ipfs-bch-wallet-service",
-  "version": "1.11.4",
+  "version": "1.11.12",
   "description": "IPFS and REST service providing blockchain access needed by a wallet. ",
   "main": "index.js",
   "scripts": {
@@ -28,7 +28,7 @@
     "bcryptjs": "^2.4.3",
     "glob": "^7.1.6",
     "@chris.troutner/ipfs": "2.0.2",
-    "ipfs-coord": "^6.6.0",
+    "ipfs-coord": "^6.6.4",
     "jsonrpc-lite": "^2.2.0",
     "jsonwebtoken": "^8.5.1",
     "jwt-bch-lib": "^1.3.0",

src/adapters/ipfs/index.js

@@ -46,6 +46,13 @@ class IPFS {
       return true
     } catch (err) {
       console.error('Error in adapters/ipfs/index.js/start()')
+
+      // If error is due to a lock file issue. Kill the process, so that
+      // Docker or pm2 has a chance to restart the service.
+      if (err.message.includes('Lock already being held')) {
+        process.exit(1)
+      }
+
       throw err
     }
   }

src/controllers/json-rpc/index.js

@@ -38,6 +38,11 @@ class JSONRPC {
     this.aboutController = new AboutController()
     this.bchController = new BCHController(localConfig)
 
+    // Cache to store IDs of processed JSON RPC commands. Used to prevent
+    // duplicate processing.
+    this.msgCache = []
+    this.MSG_CACHE_SIZE = 30
+
     _this = this
   }
 
@@ -45,8 +50,8 @@ class JSONRPC {
   // which controller to route the instruction to.
   async router (str, from) {
     try {
-      console.log('router str: ', str)
-      console.log('router from: ', from)
+      // console.log('router str: ', str)
+      // console.log('router from: ', from)
 
       // Exit quietly if 'from' is not specified.
       if (!from || typeof from !== 'string') {
@@ -58,6 +63,7 @@ class JSONRPC {
 
       // Attempt to parse the incoming data as a JSON RPC string.
       const parsedData = _this.jsonrpc.parse(str)
+      // console.log('\nrouter from: ', from)
       // console.log('parsedData: ', parsedData)
 
       // Exit quietly if the incoming string is an invalid JSON RPC string.
@@ -65,6 +71,12 @@ class JSONRPC {
         return
       }
 
+      // Check for duplicate entries with same 'id' value.
+      const alreadyProcessed = _this._checkIfAlreadyProcessed(parsedData)
+      if (alreadyProcessed) {
+        return
+      }
+
       // Added the property "from" to the parsedData object;
       // necessary for calculating rate limits (based on the IPFS ID).
       parsedData.from = from
@@ -124,6 +136,34 @@ class JSONRPC {
     }
   }
 
+  // Checks the ID of the JSON RPC call, to see if the message has already been
+  // processed. Returns true if the ID exists in the cache of processed messages.
+  // If the ID is new, the function adds it to the cache and return false.
+  _checkIfAlreadyProcessed (data) {
+    try {
+      const id = data.payload.id
+
+      // Check if the hash is in the array of already processed message.
+      const alreadyProcessed = this.msgCache.includes(id)
+
+      // Update the msgCache if this is a new message.
+      if (!alreadyProcessed) {
+        // Add the hash to the array.
+        this.msgCache.push(id)
+
+        // If the array is at its max size, then remove the oldest element.
+        if (this.msgCache.length > this.MSG_CACHE_SIZE) {
+          this.msgCache.shift()
+        }
+      }
+
+      return alreadyProcessed
+    } catch (err) {
+      console.error('Error in _checkIfAlreadyProcessed: ', err)
+      return false
+    }
+  }
+
   // The default JSON RPC response if the incoming command could not be routed.
   defaultResponse () {
     const errorObj = {