From 73c0d09198e6890ea81673039cc20db56214354e Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?S=C3=A9rgio=20Salgado?= Date: Mon, 24 May 2021 02:53:36 +0100 Subject: [PATCH] feat: add transaction batching module feat: plugin sendCoins batching support feat: batching processing on poller feat: mock-wallet batching fix: bitcoin tx batching fix: transaction batching db table --- lib/plugins.js | 10 +++ lib/plugins/wallet/bitcoind/bitcoind.js | 12 ++- lib/plugins/wallet/mock-wallet/mock-wallet.js | 25 ++++++ lib/poller.js | 3 + lib/tx-batching.js | 84 +++++++++++++++++++ lib/wallet.js | 22 ++++- .../1621556014244-add-btc-tx-batching.js | 4 +- 7 files changed, 154 insertions(+), 6 deletions(-) create mode 100644 lib/tx-batching.js diff --git a/lib/plugins.js b/lib/plugins.js index a0ae0b8a..7f0ed759 100644 --- a/lib/plugins.js +++ b/lib/plugins.js @@ -22,6 +22,7 @@ const machineLoader = require('./machine-loader') const customers = require('./customers') const commissionMath = require('./commission-math') const loyalty = require('./loyalty') +const transactionBatching = require('./tx-batching') const { cassetteMaxCapacity, CASH_OUT_DISPENSE_READY, CONFIRMATION_CODE } = require('./constants') @@ -277,6 +278,15 @@ function plugins (settings, deviceId) { } function sendCoins (tx) { + if (wallet.supportsBatching(settings, tx.cryptoCode)) { + return transactionBatching.addTransactionToBatch(tx) + .then(() => ({ + batched: true, + sendPending: false, + error: null, + errorCode: null + })) + } return wallet.sendCoins(settings, tx) } diff --git a/lib/plugins/wallet/bitcoind/bitcoind.js b/lib/plugins/wallet/bitcoind/bitcoind.js index d374b0fd..516e7f39 100644 --- a/lib/plugins/wallet/bitcoind/bitcoind.js +++ b/lib/plugins/wallet/bitcoind/bitcoind.js @@ -81,13 +81,18 @@ function sendCoins (account, tx, settings, operatorId, feeMultiplier) { function sendCoinsBatch (account, txs, cryptoCode) { return checkCryptoCode(cryptoCode) + .then(() => calculateFeeDiscount(feeMultiplier)) + .then(newFee => fetch('settxfee', [newFee])) .then(() => { const txAddressAmountPairs = _.map(tx => [tx.address, tx.cryptoAtoms.shift(-unitScale).toFixed(8)], txs) return Promise.all([JSON.stringify(_.fromPairs(txAddressAmountPairs))]) }) .then(([obj]) => fetch('sendmany', ['', obj])) - .then(res => ({ - txid: res.txid + .then((txId) => fetch('gettransaction', [txId])) + .then((res) => _.pick(['fee', 'txid'], res)) + .then((pickedObj) => ({ + fee: BN(pickedObj.fee).abs().shift(unitScale).round(), + txid: pickedObj.txid })) .catch(err => { if (err.code === -6) throw new E.InsufficientFundsError() @@ -173,5 +178,6 @@ module.exports = { newFunding, cryptoNetwork, fetchRBF, - estimateFee + estimateFee, + sendCoinsBatch } diff --git a/lib/plugins/wallet/mock-wallet/mock-wallet.js b/lib/plugins/wallet/mock-wallet/mock-wallet.js index 75c3c340..4a945869 100644 --- a/lib/plugins/wallet/mock-wallet/mock-wallet.js +++ b/lib/plugins/wallet/mock-wallet/mock-wallet.js @@ -1,3 +1,5 @@ +const _ = require('lodash/fp') + const BN = require('../../../bn') const E = require('../../../error') const { utils: coinUtils } = require('lamassu-coins') @@ -57,7 +59,29 @@ function sendCoins (account, tx, settings, operatorId) { }) } +<<<<<<< HEAD function newAddress (account, info, tx, settings, operatorId) { +======= +function sendCoinsBatch (account, txs, cryptoCode) { + sendCount = sendCount + txs.length + return new Promise((resolve, reject) => { + setTimeout(() => { + const cryptoSum = _.reduce((acc, value) => acc.add(value.crypto_atoms), BN(0), txs) + if (isInsufficient(cryptoSum, cryptoCode)) { + console.log('[%s] DEBUG: Mock wallet insufficient funds: %s', + cryptoCode, cryptoSum.toString()) + return reject(new E.InsufficientFundsError()) + } + + console.log('[%s] DEBUG: Mock wallet sending %s cryptoAtoms in a batch', + cryptoCode, cryptoSum.toString()) + return resolve({ txid: '', fee: BN(0) }) + }, 2000) + }) +} + +function newAddress () { +>>>>>>> feat: add transaction batching module t0 = Date.now() return Promise.resolve('') } @@ -93,6 +117,7 @@ function getStatus (account, tx, requested, settings, operatorId) { module.exports = { NAME, balance, + sendCoinsBatch, sendCoins, newAddress, getStatus, diff --git a/lib/poller.js b/lib/poller.js index e7c9948b..1c05bef1 100644 --- a/lib/poller.js +++ b/lib/poller.js @@ -17,6 +17,7 @@ const NodeCache = require('node-cache') const util = require('util') const db = require('./db') const state = require('./middlewares/state') +const batching = require('./tx-batching') const INCOMING_TX_INTERVAL = 30 * T.seconds const LIVE_INCOMING_TX_INTERVAL = 5 * T.seconds @@ -31,6 +32,7 @@ const SANCTIONS_INITIAL_DOWNLOAD_INTERVAL = 5 * T.minutes const SANCTIONS_UPDATE_INTERVAL = 1 * T.day const RADAR_UPDATE_INTERVAL = 5 * T.minutes const PRUNE_MACHINES_HEARTBEAT = 1 * T.day +const TRANSACTION_BATCH_LIFECYCLE = 20 * T.minutes const CHECK_NOTIFICATION_INTERVAL = 20 * T.seconds const PENDING_INTERVAL = 10 * T.seconds @@ -214,6 +216,7 @@ function doPolling (schema) { } addToQueue(cashOutTx.monitorUnnotified, UNNOTIFIED_INTERVAL, schema, QUEUE.FAST, settings) addToQueue(cashInTx.monitorPending, PENDING_INTERVAL, schema, QUEUE.FAST, settings) + addToQueue(batching.processBatches, UNNOTIFIED_INTERVAL, schema, QUEUE.FAST, settings, TRANSACTION_BATCH_LIFECYCLE) addToQueue(pi().sweepHd, SWEEP_HD_INTERVAL, schema, QUEUE.FAST, settings) addToQueue(pi().pong, PONG_INTERVAL, schema, QUEUE.FAST) addToQueue(pi().clearOldLogs, LOGS_CLEAR_INTERVAL, schema, QUEUE.SLOW) diff --git a/lib/tx-batching.js b/lib/tx-batching.js new file mode 100644 index 00000000..93cd5eb9 --- /dev/null +++ b/lib/tx-batching.js @@ -0,0 +1,84 @@ +const _ = require('lodash/fp') +const pgp = require('pg-promise')() +const uuid = require('uuid') + +const db = require('./db') +const wallet = require('./wallet') + +function createTransactionBatch (cryptoCode) { + const sql = `INSERT INTO transaction_batches (id, crypto_code) VALUES ($1, $2) RETURNING *` + + return db.one(sql, [uuid.v4(), cryptoCode]) +} + +function closeTransactionBatch (batch) { + const sql = `UPDATE transaction_batches SET status='ready', closed_at=now() WHERE id=$1` + + return db.none(sql, [batch.id]) +} + +function confirmSentBatch (batch) { + const sql = `UPDATE transaction_batches SET status='sent', error_message=NULL WHERE id=$1` + + return db.none(sql, [batch.id]) +} + +function setErroredBatch (batch, errorMsg) { + const sql = `UPDATE transaction_batches SET status='failed', error_message=$1 WHERE id=$2` + + return db.none(sql, [errorMsg, batch.id]) +} + +function addTransactionToBatch (tx) { + const sql = `SELECT * FROM transaction_batches WHERE crypto_code=$1 AND status='open' ORDER BY created_at DESC LIMIT 1` + const sql2 = `UPDATE cash_in_txs SET batch_id=$1 WHERE id=$2` + + return db.oneOrNone(sql, [tx.cryptoCode]) + .then(batch => { + if (_.isNil(batch)) + return createTransactionBatch(tx.cryptoCode) + return Promise.resolve(batch) + }) + .then(batch => db.none(sql2, [batch.id, tx.id])) +} + +function getBatchTransactions (batch) { + const sql = `SELECT * FROM cash_in_txs WHERE batch_id=$1` + return db.manyOrNone(sql, [batch.id]) +} + +function getBatchesByStatus (statuses) { + const sql = `SELECT *, EXTRACT(EPOCH FROM (now() - created_at)) as time_elapsed FROM transaction_batches WHERE status in ($1^)` + + return db.manyOrNone(sql, [_.map(pgp.as.text, statuses).join(',')]) +} + +function submitBatch (settings, batch) { + getBatchTransactions(batch) + .then(txs => { + wallet.sendCoinsBatch(settings, txs, batch.crypto_code) + .then(() => confirmSentBatch(batch)) + .catch(err => setErroredBatch(batch, err.message)) + }) +} + +function processBatches (settings, lifecycle) { + getBatchesByStatus(['open', 'failed']) + .then(batches => { + _.each(batch => { + const elapsedMS = batch.time_elapsed * 1000 + + if (elapsedMS >= lifecycle) { + return closeTransactionBatch(batch) + .then(() => submitBatch(settings, batch)) + } + }, batches) + }) +} + +module.exports = { + createTransactionBatch, + closeTransactionBatch, + addTransactionToBatch, + processBatches +} diff --git a/lib/wallet.js b/lib/wallet.js index 6ec32192..d1d3db82 100644 --- a/lib/wallet.js +++ b/lib/wallet.js @@ -75,6 +75,24 @@ function sendCoins (settings, tx) { }) } +function sendCoinsBatch (settings, txs, cryptoCode) { + return fetchWallet(settings, cryptoCode) + .then(r => { + return r.wallet.sendCoinsBatch(r.account, txs, cryptoCode) + .then(res => { + mem.clear(module.exports.balance) + return res + }) + }) + .catch(err => { + if (err.name === INSUFFICIENT_FUNDS_NAME) { + throw httpError(INSUFFICIENT_FUNDS_NAME, INSUFFICIENT_FUNDS_CODE) + } + + throw err + }) +} + function newAddress (settings, info, tx) { const walletAddressPromise = fetchWallet(settings, info.cryptoCode) .then(r => r.wallet.newAddress(r.account, info, tx, settings, r.operatorId)) @@ -235,11 +253,13 @@ const balanceFiltered = mem(_balance, { module.exports = { balance, sendCoins, + sendCoinsBatch, newAddress, getStatus, isStrictAddress, sweep, isHd, newFunding, - cryptoNetwork + cryptoNetwork, + supportsBatching } diff --git a/migrations/1621556014244-add-btc-tx-batching.js b/migrations/1621556014244-add-btc-tx-batching.js index 4750b0fb..69eafc24 100644 --- a/migrations/1621556014244-add-btc-tx-batching.js +++ b/migrations/1621556014244-add-btc-tx-batching.js @@ -2,7 +2,7 @@ var db = require('./db') exports.up = function (next) { var sql = [ - `CREATE TYPE transaction_batch_status AS ENUM('open', 'failed', 'sent')`, + `CREATE TYPE transaction_batch_status AS ENUM('open', 'ready', 'failed', 'sent')`, `CREATE TABLE transaction_batches ( id UUID PRIMARY KEY, crypto_code TEXT NOT NULL, @@ -11,7 +11,7 @@ exports.up = function (next) { closed_at TIMESTAMPTZ, error_message TEXT )`, - `ALTER TABLE cash_in_txs ADD COLUMN batch_id REFERENCES transaction_batches(id)` + `ALTER TABLE cash_in_txs ADD COLUMN batch_id UUID REFERENCES transaction_batches(id)` ] db.multi(sql, next)