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

fix(JSON RPC): Prevening processing of duplicate entries

package.json

@@ -28,7 +28,7 @@
     "bcryptjs": "^2.4.3",
     "glob": "^7.1.6",
     "@chris.troutner/ipfs": "2.0.2",
-    "ipfs-coord": "^6.6.3",
+    "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

@@ -38,6 +38,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

@@ -36,6 +36,11 @@ class JSONRPC {
     this.authController = new AuthController(localConfig)
     this.aboutController = new AboutController()
 
+    // Cache to store IDs of processed JSON RPC commands. Used to prevent
+    // duplicate processing.
+    this.msgCache = []
+    this.MSG_CACHE_SIZE = 30
+
     _this = this
   }
 
@@ -44,7 +49,7 @@ class JSONRPC {
   async router (str, from) {
     try {
       // console.log('router str: ', str)
-      // console.log('router from: ', from)
+      console.log('router from: ', from)
 
       // Exit quietly if 'from' is not specified.
       if (!from || typeof from !== 'string') {
@@ -56,6 +61,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.
@@ -63,6 +69,15 @@ class JSONRPC {
         return
       }
 
+      // Check for duplicate entries with same 'id' value.
+      const alreadyProcessed = _this._checkIfAlreadyProcessed(parsedData)
+      if (alreadyProcessed) {
+        return
+      } else {
+        console.log('\nProcessing instruction from: ', from)
+        console.log('parsedData: ', parsedData)
+      }
+
       // Added the property "from" to the parsedData object;
       // necessary for calculating rate limits (based on the IPFS ID).
       parsedData.from = from
@@ -119,6 +134,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 = {