fix: concurrency issues caused by poller

lib/cash-out/cash-out-atomic.js

@@ -27,6 +27,12 @@ function atomic (tx, pi, fromClient) {
         const isStale = fromClient && oldTx && (oldTx.txVersion >= tx.txVersion)
         if (isStale) throw new E.StaleTxError({ txId: tx.id })
 
+        const isStaleFromServer = !fromClient && oldTx && (oldTx.txVersion > tx.txVersion)
+        if (isStaleFromServer) {
+          logger.info('trying to update older version of tx', JSON.stringify(tx))
+          return Promise.resolve()
+        }
+
         return preProcess(t, oldTx, tx, pi)
           .then(preProcessedTx => cashOutLow.upsert(t, oldTx, preProcessedTx))
       })

lib/cash-out/cash-out-low.js

@@ -59,6 +59,11 @@ function diff (oldTx, newTx) {
           return updatedTx['customerId'] = newTx['customerId']
         }
         return
+      case 'dispense':
+        if (!oldTx.dispense) {
+          return updatedTx[fieldKey] = newTx[fieldKey]
+        }
+        return
       default:
         return updatedTx[fieldKey] = newTx[fieldKey]
     }

lib/middlewares/populateSettings.js

@@ -75,7 +75,7 @@ const populateSettings = function (req, res, next) {
           .catch(next)
       }
 
-      logger.debug('Fetching and caching a specific config version')
+      logger.debug('Fetching a cached specific config version')
       req.settings = cachedVersionedSettings
       return next()
     }

lib/plugins.js

@@ -342,13 +342,6 @@ function plugins (settings, deviceId) {
     return wallet.newAddress(settings, info, tx)
   }
 
-  function dispenseAck (tx) {
-    const cashOutConfig = configManager.getCashOut(deviceId, settings.config)
-    const cassettes = [cashOutConfig.cassette1, cashOutConfig.cassette2, cashOutConfig.cassette3, cashOutConfig.cassette4]
-
-    return dbm.addDispense(deviceId, tx, cassettes)
-  }
-
   function fiatBalance (fiatCode, cryptoCode) {
     const commissions = configManager.getCommissions(cryptoCode, deviceId, settings.config)
     return Promise.all([
@@ -864,7 +857,6 @@ function plugins (settings, deviceId) {
     isHd,
     isZeroConf,
     getStatus,
-    dispenseAck,
     getPhoneCode,
     executeTrades,
     pong,

lib/poller.js

@@ -16,7 +16,6 @@ const settingsLoader = require('./new-settings-loader')
 const NodeCache = require('node-cache')
 const util = require('util')
 const db = require('./db')
-const state = require('./middlewares/state')
 const processBatches = require('./tx-batching-processing')
 
 const INCOMING_TX_INTERVAL = 30 * T.seconds
@@ -99,8 +98,8 @@ function reload (schema) {
   })
 }
 
-function pi () { return cachedVariables.get(asyncLocalStorage.getStore().get('schema')).pi }
+function pi () { return cachedVariables.get('public').pi }
-function settings () { return cachedVariables.get(asyncLocalStorage.getStore().get('schema')).settings }
+function settings () { return cachedVariables.get('public').settings }
 
 function initialSanctionsDownload () {
   const structs = sanctions.getStructs()
@@ -146,19 +145,39 @@ function initializeEachSchema (schemas = ['public']) {
   }, schemas)
 }
 
-function addToQueue (func, interval, schema, queue, ...vars) {
+function recursiveTimeout (func, timeout, ...vars) {
-  return schemaCallbacks.get(schema).push(setInterval(() => {
+  setTimeout(() => {
-    return queue.enqueue().then(() => {
+    let promise = null
-      // get plugins or settings from the cache every time func is run
+
     const loadVariables = vars.length > 0 && typeof vars[0] === 'function'
     if (loadVariables) {
       const funcVars = [...vars]
       funcVars[0] = vars[0]()
-        return func(...funcVars)
+      promise = func(...funcVars)
+    } else {
+      promise = func(...vars)
     }
-      return func(...vars)
+
-    }).catch(console.error)
+    promise.finally(() => {
-  }, interval))
+      recursiveTimeout(func, timeout, ...vars)
+    })
+  }, timeout)
+}
+
+function addToQueue (func, interval, schema, queue, ...vars) {
+  recursiveTimeout(func, interval, ...vars)
+  // return schemaCallbacks.get(schema).push(setInterval(() => {
+    // return queue.enqueue().then(() => {
+    //   // get plugins or settings from the cache every time func is run
+    //   const loadVariables = vars.length > 0 && typeof vars[0] === 'function'
+    //   if (loadVariables) {
+    //     const funcVars = [...vars]
+    //     funcVars[0] = vars[0]()
+    //     return func(...funcVars)
+    //   }
+    //   return func(...vars)
+    // }).catch(console.error)
+  // }, interval))
 }
 
 function doPolling (schema) {

package-lock.json

@@ -1,6 +1,6 @@
 {
   "name": "lamassu-server",
-  "version": "8.1.2",
+  "version": "8.1.3",
   "lockfileVersion": 1,
   "requires": true,
   "dependencies": {

package.json

@@ -2,7 +2,7 @@
   "name": "lamassu-server",
   "description": "bitcoin atm client server protocol module",
   "keywords": [],
-  "version": "8.1.2",
+  "version": "8.1.3",
   "license": "Unlicense",
   "author": "Lamassu (https://lamassu.is)",
   "dependencies": {