36 outgoing server delay

.eslintrc

@@ -31,6 +31,6 @@
     "prefer-template": 0,
     "no-restricted-syntax": 0,
     "strict": 0,
-    "no-underscore-dangle": ["error", {"allow": ["__set__"]}]
+    "no-underscore-dangle": ["warn", {"allow": ["__set__", "_*"]}]
   }
 }

src/app.js

@@ -578,6 +578,8 @@ const apiInit = () => {
     }))
   }
 
+  // @TODO check if server paused before returning as valid incoming server
+
   // -------------- CHECK IF SERVER IS PROCESSING ------------------------------------------------------------------------------------------
 
   app.get('/api/status', (req, res) => {

src/incoming.js

@@ -3,16 +3,16 @@
 const Client = require('bitcoin-core')
 const config = require('config')
 
-const EncryptionKeys = require('./lib/EncryptionKeys.js')
-const Logger = require('./lib/Logger.js')
-const PreFlight = require('./lib/PreFlight.js')
-const RefillOutgoing = require('./lib/RefillOutgoing.js')
-const SelectOutgoing = require('./lib/SelectOutgoing.js')
-const ReturnAllToSenders = require('./lib/ReturnAllToSenders.js')
-const PrepareIncoming = require('./lib/PrepareIncoming.js')
-const RetrieveSubchainAddresses = require('./lib/RetrieveSubchainAddresses.js')
-const ProcessIncoming = require('./lib/ProcessIncoming.js')
-const SpendToHolding = require('./lib/SpendToHolding.js')
+let EncryptionKeys = require('./lib/EncryptionKeys.js') //eslint-disable-line
+let Logger = require('./lib/Logger.js') //eslint-disable-line
+let PreFlight = require('./lib/PreFlight.js') //eslint-disable-line
+let RefillOutgoing = require('./lib/RefillOutgoing.js') //eslint-disable-line
+let SelectOutgoing = require('./lib/SelectOutgoing.js') //eslint-disable-line
+let ReturnAllToSenders = require('./lib/ReturnAllToSenders.js') //eslint-disable-line
+let PrepareIncoming = require('./lib/PrepareIncoming.js') //eslint-disable-line
+let RetrieveSubchainAddresses = require('./lib/RetrieveSubchainAddresses.js') //eslint-disable-line
+let ProcessIncoming = require('./lib/ProcessIncoming.js') //eslint-disable-line
+let SpendToHolding = require('./lib/SpendToHolding.js') //eslint-disable-line
 
 const settings = config.get('INCOMING')
 
@@ -41,10 +41,11 @@ IncomingServer.init = () => {
 
   Logger.writeLog('INC_000', 'server starting')
   EncryptionKeys.findKeysToRemove({ type: 'private' }, IncomingServer.startProcessing)
-  setInterval(() => {
+  IncomingServer.cron = setInterval(() => {
     if (IncomingServer.paused === false) {
       EncryptionKeys.findKeysToRemove({ type: 'private' }, IncomingServer.startProcessing)
     } else {
+      clearInterval(IncomingServer.cron)
       Logger.writeLog('INC_001', 'processing paused', { paused: IncomingServer.paused })
     }
   }, settings.scriptInterval)
@@ -80,6 +81,7 @@ IncomingServer.holdingProcessed = (success, data) => {
   if (!success) {
     Logger.writeLog('INC_004', 'failed to process the holding account', { success, data }, true)
     IncomingServer.processing = false
+    IncomingServer.paused = true
     return
   }
   SelectOutgoing.run({
@@ -96,6 +98,9 @@ IncomingServer.outgoingSelected = (success, data) => {
   }
 
   if (data.returnAllToSenders) {
+    if (data.pause) {
+      IncomingServer.paused = true
+    }
     ReturnAllToSenders.run({
       navClient: IncomingServer.navClient,
     }, IncomingServer.allPendingReturned)
@@ -116,10 +121,10 @@ IncomingServer.outgoingSelected = (success, data) => {
 }
 
 IncomingServer.allPendingReturned = (success, data) => {
-  console.log('STATUS: IncomingServer.allPendingReturned', success, data)
   if (!success) {
     Logger.writeLog('INC_006', 'failed to return all pending to sender', { success, data }, true)
     IncomingServer.processing = false
+    IncomingServer.paused = true
     return
   }
   Logger.writeLog('INC_007', 'returned all pending to sender', { success, data }, true)
@@ -128,17 +133,42 @@ IncomingServer.allPendingReturned = (success, data) => {
 }
 
 IncomingServer.currentBatchPrepared = (success, data) => {
-  if (!success || !data || !data.currentBatch || !data.currentFlattened || !data.numFlattened) {
+  if (!success || !data || !data.currentBatch || !data.currentFlattened || !data.numFlattened || !data.pendingToReturn) {
     IncomingServer.processing = false
     return
   }
   IncomingServer.runtime.currentBatch = data.currentBatch
   IncomingServer.runtime.currentFlattened = data.currentFlattened
   IncomingServer.runtime.numFlattened = data.numFlattened
+  IncomingServer.runtime.pendingToReturn = data.pendingToReturn
+
+  if (IncomingServer.runtime.pendingToReturn && IncomingServer.runtime.pendingToReturn.length > 0) {
+    Logger.writeLog('INC_011', 'failed to process some transactions', { success, data }, true)
+    ReturnAllToSenders.fromList({
+      navClient: IncomingServer.navClient,
+      transactionsToReturn: IncomingServer.runtime.pendingToReturn,
+    }, IncomingServer.pendingFailedReturned)
+    return
+  }
+  RetrieveSubchainAddresses.run({
+    subClient: IncomingServer.subClient,
+    chosenOutgoing: IncomingServer.runtime.chosenOutgoing,
+    numAddresses: IncomingServer.runtime.numFlattened,
+  }, IncomingServer.retrievedSubchainAddresses)
+}
+
+IncomingServer.pendingFailedReturned = (success, data) => {
+  if (!success) {
+    Logger.writeLog('INC_011A', 'failed to return failed pending to sender', { success, data }, true)
+    IncomingServer.paused = true
+    ReturnAllToSenders.run({
+      navClient: IncomingServer.navClient,
+    }, IncomingServer.allPendingReturned)
+  }
   RetrieveSubchainAddresses.run({
     subClient: IncomingServer.subClient,
     chosenOutgoing: IncomingServer.runtime.chosenOutgoing,
-    numAddresses: data.numFlattened,
+    numAddresses: IncomingServer.runtime.numFlattened,
   }, IncomingServer.retrievedSubchainAddresses)
 }
 
@@ -150,6 +180,7 @@ IncomingServer.retrievedSubchainAddresses = (success, data) => {
     }, IncomingServer.allPendingReturned)
     return
   }
+  // @TODO compile the correct transactions to return
   ProcessIncoming.run({
     currentBatch: IncomingServer.runtime.currentBatch,
     currentFlattened: IncomingServer.runtime.currentFlattened,
@@ -163,21 +194,38 @@ IncomingServer.retrievedSubchainAddresses = (success, data) => {
 
 IncomingServer.transactionsProcessed = (success, data) => {
   if (!success || !data) {
+    if (data && data.partialFailure) {
+      Logger.writeLog('INC_010A', 'failed part way through processing subchain transactions', { success, data }, true)
+      IncomingServer.paused = true
+      IncomingServer.processing = false
+      return
+    }
     Logger.writeLog('INC_010', 'failed to process transactions', { success, data }, true)
+    IncomingServer.paused = true
     ReturnAllToSenders.run({
       navClient: IncomingServer.navClient,
     }, IncomingServer.allPendingReturned)
     return
   }
 
-  IncomingServer.runtime.successfulSubTransactions = data.successfulSubTransactions
-  IncomingServer.runtime.transactionsToReturn = data.transactionsToReturn
+  IncomingServer.runtime.successfulTxGroups = data.successfulTxGroups
+  IncomingServer.runtime.txGroupsToReturn = data.txGroupsToReturn
+  IncomingServer.runtime.transactionsToReturn = []
 
-  if (IncomingServer.runtime.transactionsToReturn && IncomingServer.runtime.transactionsToReturn.length > 0) {
+  if (IncomingServer.runtime.txGroupsToReturn && IncomingServer.runtime.txGroupsToReturn.length > 0) {
     Logger.writeLog('INC_011', 'failed to process some transactions', { success, data }, true)
+
+    // extract the relevant transactions to return from the txGroupsToReturn
+    for (let i = 0; i < IncomingServer.runtime.txGroupsToReturn.length; i++) {
+      const txGroup = IncomingServer.runtime.txGroupsToReturn[i]
+      for (let j = 0; j < txGroup.transactions.length; j++) {
+        IncomingServer.runtime.transactionsToReturn.push(txGroup.transactions[j])
+      }
+    }
+
     ReturnAllToSenders.fromList({
       navClient: IncomingServer.navClient,
-      transactionsToReturn: data.transactionsToReturn,
+      transactionsToReturn: IncomingServer.runtime.transactionsToReturn,
     }, IncomingServer.failedTransactionsReturned)
     return
   }
@@ -186,8 +234,18 @@ IncomingServer.transactionsProcessed = (success, data) => {
 
 IncomingServer.failedTransactionsReturned = (success, data) => {
   if (!success) {
+    IncomingServer.paused = true
     Logger.writeLog('INC_012', 'failed to return failed transactions to sender', { success, data }, true)
   }
+  IncomingServer.runtime.successfulSubTransactions = []
+  // extract the relevant transactions to return from the txGroupsToReturn
+  for (let i = 0; i < IncomingServer.runtime.successfulTxGroups.length; i++) {
+    const txGroup = IncomingServer.runtime.successfulTxGroups[i]
+    for (let j = 0; j < txGroup.transactions.length; j++) {
+      IncomingServer.runtime.successfulSubTransactions.push(txGroup.transactions[j])
+    }
+  }
+
   SpendToHolding.run({
     successfulSubTransactions: IncomingServer.runtime.successfulSubTransactions,
     holdingEncrypted: IncomingServer.runtime.holdingEncrypted,

src/lib/GroupPartials.js

@@ -0,0 +1,168 @@
+const lodash = require('lodash')
+
+let Logger = require('./Logger.js') // eslint-disable-line
+let EncryptedData = require('./EncryptedData.js') // eslint-disable-line
+
+const GroupPartials = {}
+
+GroupPartials.run = (options, callback) => {
+  const required = ['currentPending', 'client']
+  if (lodash.intersection(Object.keys(options), required).length !== required.length) {
+    Logger.writeLog('GRP_001', 'invalid options', { options, required })
+    callback(false, { message: 'invalid options provided to GroupPartials.run' })
+    return
+  }
+
+  GroupPartials.runtime = {
+    client: options.client,
+    currentPending: options.currentPending,
+    remainingToDecrypt: options.currentPending,
+    transactionsToReturn: [],
+    readyToProcess: {},
+    partials: {},
+    callback,
+  }
+  GroupPartials.getDecryptedData()
+}
+
+GroupPartials.getDecryptedData = () => {
+  if (GroupPartials.runtime.remainingToDecrypt.length < 1) {
+    GroupPartials.checkPartials()
+    return
+  }
+
+  EncryptedData.getEncrypted({
+    transaction: GroupPartials.runtime.remainingToDecrypt[0],
+    client: GroupPartials.runtime.client,
+  }, GroupPartials.checkDecrypted)
+}
+
+GroupPartials.partialFailed = (transaction) => {
+  const returnIndex = lodash.findIndex(GroupPartials.runtime.transactionsToReturn, (tx) => tx.txid === transaction.txid)
+  if (returnIndex === -1) {
+    GroupPartials.runtime.transactionsToReturn.push(transaction)
+  }
+  GroupPartials.runtime.remainingToDecrypt.splice(0, 1)
+  GroupPartials.getDecryptedData()
+}
+
+GroupPartials.checkDecrypted = (success, data) => {
+  if (!success || !data || !data.decrypted || !data.transaction) {
+    Logger.writeLog('GRP_002', 'failed to decrypt transaction data', { success })
+    GroupPartials.partialFailed(GroupPartials.runtime.remainingToDecrypt[0])
+    return
+  }
+  if (!data.decrypted.n || !data.decrypted.t || !data.decrypted.p || !data.decrypted.o || !data.decrypted.u) {
+    Logger.writeLog('GRP_003', 'failed to receive correct encrypted params', {
+      n: data.decrypted.n, // @TODO remove logging n before deployment
+      t: data.decrypted.t,
+      p: data.decrypted.p,
+      o: data.decrypted.o,
+      u: data.decrypted.u,
+    })
+    GroupPartials.partialFailed(data.transaction)
+    return
+  }
+
+  GroupPartials.runtime.client.validateAddress(data.decrypted.n).then((addressInfo) => {
+    if (addressInfo.isvalid !== true) {
+      Logger.writeLog('GRP_003A', 'encrypted address invalid', { success, data })
+      GroupPartials.partialFailed(data.transaction)
+      return
+    }
+    GroupPartials.groupPartials(data.decrypted, data.transaction)
+  }).catch((err) => {
+    Logger.writeLog('GRP_003B', 'failed to decrypt transaction data', { success, error: err })
+    GroupPartials.partialFailed(data.transaction)
+    return
+  })
+}
+
+GroupPartials.groupPartials = (decrypted, transaction) => {
+  if (!GroupPartials.runtime.partials[decrypted.u]) {
+    GroupPartials.runtime.partials[decrypted.u] = {
+      destination: decrypted.n,
+      unique: decrypted.u,
+      timeDelay: parseInt(decrypted.t, 10),
+      parts: parseInt(decrypted.o, 10),
+      partsSum: 0,
+      amount: 0,
+      transactions: {},
+      readyToProcess: false,
+    }
+  }
+
+  if (GroupPartials.runtime.partials[decrypted.u].readyToProcess === true) {
+    Logger.writeLog('GRP_006', 'this partial group is already flagged as completed', {
+      partials: GroupPartials.runtime.partials[decrypted.u],
+      transaction,
+      n: decrypted.n, // @TODO remove logging n before deployment
+      t: decrypted.t,
+      p: decrypted.p,
+      o: decrypted.o,
+      u: decrypted.u,
+    })
+    GroupPartials.partialFailed(transaction)
+    return
+  }
+
+  if (GroupPartials.runtime.partials[decrypted.u].destination !== decrypted.n) {
+    Logger.writeLog('GRP_004', 'decrypted address different from other partials', {
+      partials: GroupPartials.runtime.partials[decrypted.u],
+      transaction,
+    })
+    GroupPartials.partialFailed(transaction)
+    return
+  }
+
+  if (GroupPartials.runtime.partials[decrypted.u].transactions[transaction.txid]) {
+    Logger.writeLog('GRP_005', 'txid already processed', {
+      partials: GroupPartials.runtime.partials[decrypted.u],
+      transaction,
+    })
+    GroupPartials.partialFailed(transaction)
+    return
+  }
+
+  const unsafeTotal = GroupPartials.runtime.partials[decrypted.u].amount += transaction.amount
+  const safeTotal = Math.round(unsafeTotal * 100000000) / 100000000
+
+  GroupPartials.runtime.partials[decrypted.u].amount = safeTotal
+  GroupPartials.runtime.partials[decrypted.u].partsSum += parseInt(decrypted.p, 10)
+  GroupPartials.runtime.partials[decrypted.u].transactions[transaction.txid] = {
+    txid: transaction.txid,
+    amount: transaction.amount,
+    part: decrypted.p,
+  }
+
+  const numParts = GroupPartials.runtime.partials[decrypted.u].parts
+  const numTransactions = lodash.size(GroupPartials.runtime.partials[decrypted.u].transactions)
+
+  if (numTransactions === numParts
+  && (numParts * (numParts + 1)) / 2 === GroupPartials.runtime.partials[decrypted.u].partsSum) {
+    GroupPartials.runtime.partials[decrypted.u].readyToProcess = true
+    GroupPartials.runtime.readyToProcess[decrypted.u] = GroupPartials.runtime.partials[decrypted.u]
+  }
+  GroupPartials.runtime.remainingToDecrypt.splice(0, 1)
+  GroupPartials.getDecryptedData()
+}
+
+GroupPartials.checkPartials = () => {
+  lodash.forEach(GroupPartials.runtime.partials, (partial) => {
+    if (!partial.readyToProcess) {
+      lodash.forEach(partial.transactions, (partialTx) => {
+        const returnIndex = lodash.findIndex(GroupPartials.runtime.transactionsToReturn, (tx) => tx.txid === partialTx.txid)
+        if (returnIndex === -1 && partialTx.confirmations > 120) { // if its not already flagged as returning & too old
+          GroupPartials.runtime.transactionsToReturn.push(partialTx)
+        }
+      })
+    }
+  })
+
+  GroupPartials.runtime.callback(true, {
+    readyToProcess: GroupPartials.runtime.readyToProcess,
+    transactionsToReturn: GroupPartials.runtime.transactionsToReturn,
+  })
+}
+
+module.exports = GroupPartials

src/lib/PayoutFee.js

@@ -42,6 +42,10 @@ PayoutFee.send = () => {
       address: PayoutFee.runtime.settings.anonTxFeeAddress,
       amount: txFeeAccrued,
     }, PayoutFee.sent)
+  }).catch((err) => {
+    Logger.writeLog('PAY_002A', 'error getting balance', { error: err })
+    PayoutFee.runtime.callback(false, { message: 'error getting balance' })
+    return
   })
 }