lamassu/lamassu-server
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Merge pull request #727 from chaotixkilla/feat-btc-transaction-batching

Browse source 
BTC transaction batching
This commit is contained in:
Rafael Taranto 2021-11-24 22:56:09 +00:00 • committed by GitHub
commit cc8c48ff4c
19 changed files with 331 additions and 28 deletions
Download patch file Download diff file Expand all files Collapse all files

6
lib/cash-in/cash-in-low.js
View file

@ -102,7 +102,7 @@ function diff (oldTx, newTx) {
}
function ensureRatchet (oldField, newField, fieldKey) {
const monotonic = ['cryptoAtoms', 'fiat', 'cashInFeeCrypto', 'send', 'sendConfirmed', 'operatorCompleted', 'timedout', 'txVersion']
const monotonic = ['cryptoAtoms', 'fiat', 'cashInFeeCrypto', 'send', 'sendConfirmed', 'operatorCompleted', 'timedout', 'txVersion', 'batched']
const free = ['sendPending', 'error', 'errorCode', 'customerId']
if (_.isNil(oldField)) return true
@ -138,11 +138,11 @@ function nilEqual (a, b) {
function isClearToSend (oldTx, newTx) {
const now = Date.now()
return newTx.send &&
return (newTx.send || newTx.batched) &&
(!oldTx || (!oldTx.sendPending && !oldTx.sendConfirmed)) &&
(newTx.created > now - PENDING_INTERVAL_MS)
}
function isFinalTxStage (txChanges) {
return txChanges.send
return txChanges.send || txChanges.batched
}

16
lib/cash-in/cash-in-tx.js
View file

@ -120,7 +120,18 @@ function postProcess (r, pi, isBlacklisted, addressReuse, failedWalletScore) {
if (!cashInLow.isClearToSend(r.dbTx, r.tx)) return Promise.resolve({})
return pi.sendCoins(r.tx)
.then(txObj => ({
.then(txObj => {
if (txObj.batched) {
return {
batched: true,
batchTime: 'now()^',
sendPending: true,
error: null,
errorCode: null
}
}
return {
txHash: txObj.txid,
fee: txObj.fee,
sendConfirmed: true,
@ -128,7 +139,8 @@ function postProcess (r, pi, isBlacklisted, addressReuse, failedWalletScore) {
sendPending: false,
error: null,
errorCode: null
}))
}
})
.catch(err => {
// Important: We don't know what kind of error this is
// so not safe to assume that funds weren't sent.

2
lib/new-admin/graphql/types/transaction.type.js
View file

@ -45,6 +45,8 @@ const typeDef = gql`
discount: Int
txCustomerPhotoPath: String
txCustomerPhotoAt: Date
batched: Boolean
batchTime: Date
}
type Filter {

25
lib/plugins.js
View file

@ -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')
@ -237,6 +238,7 @@ function plugins (settings, deviceId) {
const pingPromise = recordPing(deviceTime, machineVersion, machineModel)
const currentConfigVersionPromise = fetchCurrentConfigVersion()
const currentAvailablePromoCodes = loyalty.getNumberOfAvailablePromoCodes()
const supportsBatchingPromise = cryptoCodes.map(c => wallet.supportsBatching(settings, c))
const timezoneObj = { utcOffset: timezone[0], dstOffset: timezone[1] }
const promises = [
@ -245,6 +247,7 @@ function plugins (settings, deviceId) {
currentConfigVersionPromise,
timezoneObj
].concat(
supportsBatchingPromise,
tickerPromises,
balancePromises,
testnetPromises,
@ -257,9 +260,11 @@ function plugins (settings, deviceId) {
const configVersion = arr[2]
const tz = arr[3]
const cryptoCodesCount = cryptoCodes.length
const tickers = arr.slice(4, cryptoCodesCount + 4)
const balances = arr.slice(cryptoCodesCount + 4, 2 * cryptoCodesCount + 4)
const testNets = arr.slice(2 * cryptoCodesCount + 4, arr.length - 2)
const batchableCoinsRes = arr.slice(4, cryptoCodesCount + 4)
const batchableCoins = batchableCoinsRes.map(it => ({ batchable: it }))
const tickers = arr.slice(cryptoCodesCount + 4, 2 * cryptoCodesCount + 4)
const balances = arr.slice(2 * cryptoCodesCount + 4, 3 * cryptoCodesCount + 4)
const testNets = arr.slice(3 * cryptoCodesCount + 4, arr.length - 1)
const coinParams = _.zip(cryptoCodes, testNets)
const coinsWithoutRate = _.map(mapCoinSettings, coinParams)
const areThereAvailablePromoCodes = arr[arr.length - 1] > 0
@ -268,7 +273,7 @@ function plugins (settings, deviceId) {
cassettes,
rates: buildRates(tickers),
balances: buildBalances(balances),
coins: _.zipWith(_.assign, coinsWithoutRate, tickers),
coins: _.zipWith(_.assign, _.zipWith(_.assign, coinsWithoutRate, tickers), batchableCoins),
configVersion,
areThereAvailablePromoCodes,
timezone: tz
@ -277,7 +282,19 @@ function plugins (settings, deviceId) {
}
function sendCoins (tx) {
return wallet.supportsBatching(settings, tx.cryptoCode)
.then(supportsBatching => {
if (supportsBatching) {
return transactionBatching.addTransactionToBatch(tx)
.then(() => ({
batched: true,
sendPending: false,
error: null,
errorCode: null
}))
}
return wallet.sendCoins(settings, tx)
})
}
function recordPing (deviceTime, version, model) {

10
lib/plugins/wallet/bitcoincashd/bitcoincashd.js
View file

@ -10,6 +10,8 @@ const cryptoRec = coinUtils.getCryptoCurrency('BCH')
const configPath = coinUtils.configPath(cryptoRec, options.blockchainDir)
const unitScale = cryptoRec.unitScale
const SUPPORTS_BATCHING = false
function rpcConfig () {
try {
const config = jsonRpc.parseConf(configPath)
@ -129,11 +131,17 @@ function cryptoNetwork (account, cryptoCode, settings, operatorId) {
.then(() => parseInt(rpcConfig.port, 10) === 18332 ? 'test' : 'main')
}
function supportsBatching (cryptoCode) {
return checkCryptoCode(cryptoCode)
.then(() => SUPPORTS_BATCHING)
}
module.exports = {
balance,
sendCoins,
newAddress,
getStatus,
newFunding,
cryptoNetwork
cryptoNetwork,
supportsBatching
}

32
lib/plugins/wallet/bitcoind/bitcoind.js
View file

@ -9,6 +9,8 @@ const { utils: coinUtils } = require('lamassu-coins')
const cryptoRec = coinUtils.getCryptoCurrency('BTC')
const unitScale = cryptoRec.unitScale
const SUPPORTS_BATCHING = true
const rpcConfig = jsonRpc.rpcConfig(cryptoRec)
function fetch (method, params) {
@ -79,6 +81,27 @@ 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((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()
throw err
})
}
function newAddress (account, info, tx, settings, operatorId) {
return checkCryptoCode(info.cryptoCode)
.then(() => fetch('getnewaddress'))
@ -149,6 +172,11 @@ function fetchRBF (txId) {
})
}
function supportsBatching (cryptoCode) {
return checkCryptoCode(cryptoCode)
.then(() => SUPPORTS_BATCHING)
}
module.exports = {
balance,
sendCoins,
@ -157,5 +185,7 @@ module.exports = {
newFunding,
cryptoNetwork,
fetchRBF,
estimateFee
estimateFee,
sendCoinsBatch,
supportsBatching
}

10
lib/plugins/wallet/bitgo/bitgo.js
View file

@ -14,6 +14,8 @@ const NAME = 'BitGo'
const SUPPORTED_COINS = ['BTC', 'ZEC', 'LTC', 'BCH', 'DASH']
const BCH_CODES = ['BCH', 'TBCH']
const SUPPORTS_BATCHING = false
function buildBitgo (account) {
const env = account.environment === 'test' ? 'test' : 'prod'
return new BitGo.BitGo({ accessToken: account.token.trim(), env, userAgent: userAgent })
@ -157,6 +159,11 @@ function cryptoNetwork (account, cryptoCode, settings, operatorId) {
.then(() => account.environment === 'test' ? 'test' : 'main')
}
function supportsBatching (cryptoCode) {
return checkCryptoCode(cryptoCode)
.then(() => SUPPORTS_BATCHING)
}
module.exports = {
NAME,
balance,
@ -164,5 +171,6 @@ module.exports = {
newAddress,
getStatus,
newFunding,
cryptoNetwork
cryptoNetwork,
supportsBatching
}

10
lib/plugins/wallet/dashd/dashd.js
View file

@ -9,6 +9,8 @@ const E = require('../../../error')
const cryptoRec = coinUtils.getCryptoCurrency('DASH')
const unitScale = cryptoRec.unitScale
const SUPPORTS_BATCHING = false
const rpcConfig = jsonRpc.rpcConfig(cryptoRec)
function fetch (method, params) {
@ -112,10 +114,16 @@ function newFunding (account, cryptoCode, settings, operatorId) {
}))
}
function supportsBatching (cryptoCode) {
return checkCryptoCode(cryptoCode)
.then(() => SUPPORTS_BATCHING)
}
module.exports = {
balance,
sendCoins,
newAddress,
getStatus,
newFunding
newFunding,
supportsBatching
}

10
lib/plugins/wallet/geth/base.js
View file

@ -17,6 +17,8 @@ const paymentPrefixPath = "m/44'/60'/0'/0'"
const defaultPrefixPath = "m/44'/60'/1'/0'"
let lastUsedNonces = {}
const SUPPORTS_BATCHING = false
module.exports = {
NAME,
balance,
@ -29,7 +31,8 @@ module.exports = {
newFunding,
privateKey,
isStrictAddress,
connect
connect,
supportsBatching
}
function connect (url) {
@ -222,3 +225,8 @@ function newFunding (account, cryptoCode, settings, operatorId) {
}))
})
}
function supportsBatching (cryptoCode) {
return checkCryptoCode(cryptoCode)
.then(() => SUPPORTS_BATCHING)
}

10
lib/plugins/wallet/litecoind/litecoind.js
View file

@ -9,6 +9,8 @@ const E = require('../../../error')
const cryptoRec = coinUtils.getCryptoCurrency('LTC')
const unitScale = cryptoRec.unitScale
const SUPPORTS_BATCHING = false
const rpcConfig = jsonRpc.rpcConfig(cryptoRec)
function fetch (method, params) {
@ -112,10 +114,16 @@ function newFunding (account, cryptoCode, settings, operatorId) {
}))
}
function supportsBatching (cryptoCode) {
return checkCryptoCode(cryptoCode)
.then(() => SUPPORTS_BATCHING)
}
module.exports = {
balance,
sendCoins,
newAddress,
getStatus,
newFunding
newFunding,
supportsBatching
}

32
lib/plugins/wallet/mock-wallet/mock-wallet.js
View file

@ -1,8 +1,12 @@
const _ = require('lodash/fp')
const BN = require('../../../bn')
const E = require('../../../error')
const { utils: coinUtils } = require('lamassu-coins')
const consoleLogLevel = require('console-log-level')
const NAME = 'FakeWallet'
const BATCHABLE_COINS = ['BTC']
const SECONDS = 1000
const PUBLISH_TIME = 3 * SECONDS
@ -57,7 +61,25 @@ function sendCoins (account, tx, settings, operatorId) {
})
}
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.plus(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: '<txHash>', fee: BN(0) })
}, 2000)
})
}
function newAddress () {
t0 = Date.now()
return Promise.resolve('<Fake address, don\'t send>')
}
@ -90,11 +112,17 @@ function getStatus (account, tx, requested, settings, operatorId) {
return Promise.resolve({status: 'confirmed'})
}
function supportsBatching (cryptoCode) {
return Promise.resolve(_.includes(cryptoCode, BATCHABLE_COINS))
}
module.exports = {
NAME,
balance,
sendCoinsBatch,
sendCoins,
newAddress,
getStatus,
newFunding
newFunding,
supportsBatching
}

10
lib/plugins/wallet/zcashd/zcashd.js
View file

@ -10,6 +10,8 @@ const E = require('../../../error')
const cryptoRec = coinUtils.getCryptoCurrency('ZEC')
const unitScale = cryptoRec.unitScale
const SUPPORTS_BATCHING = false
const rpcConfig = jsonRpc.rpcConfig(cryptoRec)
function fetch (method, params) {
@ -138,10 +140,16 @@ function newFunding (account, cryptoCode, settings, operatorId) {
}))
}
function supportsBatching (cryptoCode) {
return checkCryptoCode(cryptoCode)
.then(() => SUPPORTS_BATCHING)
}
module.exports = {
balance,
sendCoins,
newAddress,
getStatus,
newFunding
newFunding,
supportsBatching
}

3
lib/poller.js
View file

@ -17,6 +17,7 @@ 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
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(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)

29
lib/tx-batching-processing.js Normal file
View file

@ -0,0 +1,29 @@
const _ = require('lodash/fp')
const txBatching = require('./tx-batching')
const wallet = require('./wallet')
function submitBatch (settings, batch) {
txBatching.getBatchTransactions(batch)
.then(txs => {
wallet.sendCoinsBatch(settings, txs, batch.crypto_code)
.then(res => txBatching.confirmSentBatch(batch, res))
.catch(err => txBatching.setErroredBatch(batch, err.message))
})
}
function processBatches (settings, lifecycle) {
return txBatching.getBatchesByStatus(['open'])
.then(batches => {
_.each(batch => {
const elapsedMS = batch.time_elapsed * 1000
if (elapsedMS >= lifecycle) {
return txBatching.closeTransactionBatch(batch)
.then(() => submitBatch(settings, batch))
}
}, batches)
})
}
module.exports = processBatches

78
lib/tx-batching.js Normal file
View file

@ -0,0 +1,78 @@
const _ = require('lodash/fp')
const pgp = require('pg-promise')()
const uuid = require('uuid')
const BN = require('./bn')
const db = require('./db')
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 db.tx(t => {
const newBatchId = uuid.v4()
const q1 = t.none(`INSERT INTO transaction_batches (id, crypto_code) VALUES ($1, $2)`, [newBatchId, tx.cryptoCode])
const q2 = t.none(sql2, [newBatchId, tx.id])
return t.batch([q1, q2])
})
}
return db.none(sql2, [batch.id, tx.id])
})
}
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, tx) {
return db.tx(t => {
const q1 = t.none(`UPDATE transaction_batches SET status='sent', error_message=NULL WHERE id=$1`, [batch.id])
const q2 = t.none(`UPDATE cash_in_txs SET tx_hash=$1, fee=$2, send=true, send_confirmed=true, send_time=now(), send_pending=false, error=NULL, error_code=NULL WHERE batch_id=$3`, [tx.txid, tx.fee.toString(), batch.id])
return t.batch([q1, q2])
})
}
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 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 getOpenBatchCryptoValue (cryptoCode) {
const sql = `SELECT * FROM transaction_batches WHERE crypto_code=$1 AND status='open' ORDER BY created_at DESC LIMIT 1`
return db.oneOrNone(sql, [cryptoCode])
.then(batch => {
if (_.isNil(batch)) return Promise.resolve([])
return db.any(`SELECT * FROM cash_in_txs WHERE batch_id=$1`, [batch.id])
})
.then(txs => _.reduce((acc, tx) => acc.plus(tx.cash_in_fee_crypto).plus(tx.crypto_atoms), BN(0), txs))
}
module.exports = {
addTransactionToBatch,
closeTransactionBatch,
confirmSentBatch,
setErroredBatch,
getBatchTransactions,
getBatchesByStatus,
getOpenBatchCryptoValue
}

33
lib/wallet.js
View file

@ -13,6 +13,8 @@ const ph = require('./plugin-helper')
const layer2 = require('./layer2')
const httpError = require('./route-helpers').httpError
const logger = require('./logger')
const { getOpenBatchCryptoValue } = require('./tx-batching')
const BN = require('./bn')
const FETCH_INTERVAL = 5000
const INSUFFICIENT_FUNDS_CODE = 570
@ -46,7 +48,9 @@ const lastBalance = {}
function _balance (settings, cryptoCode) {
return fetchWallet(settings, cryptoCode)
.then(r => r.wallet.balance(r.account, cryptoCode, settings, r.operatorId))
.then(balance => ({ balance, timestamp: Date.now() }))
.then(balance => Promise.all([balance, supportsBatching(settings, cryptoCode)]))
.then(([balance, supportsBatching]) => Promise.all([balance, supportsBatching ? getOpenBatchCryptoValue(cryptoCode) : Promise.resolve(BN(0))]))
.then(([balance, reservedBalance]) => ({ balance: balance.minus(reservedBalance), reservedBalance, timestamp: Date.now() }))
.then(r => {
lastBalance[cryptoCode] = r
return r
@ -75,6 +79,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))
@ -210,6 +232,11 @@ function isStrictAddress (settings, cryptoCode, toAddress) {
})
}
function supportsBatching (settings, cryptoCode) {
return fetchWallet(settings, cryptoCode)
.then(r => r.wallet.supportsBatching(cryptoCode))
}
const coinFilter = ['ETH']
const balance = (settings, cryptoCode) => {
@ -230,11 +257,13 @@ const balanceFiltered = mem(_balance, {
module.exports = {
balance,
sendCoins,
sendCoinsBatch,
newAddress,
getStatus,
isStrictAddress,
sweep,
isHd,
newFunding,
cryptoNetwork
cryptoNetwork,
supportsBatching
}

24
migrations/1621556014244-add-btc-tx-batching.js Normal file
View file

@ -0,0 +1,24 @@
var db = require('./db')
exports.up = function (next) {
var sql = [
`CREATE TYPE transaction_batch_status AS ENUM('open', 'ready', 'failed', 'sent')`,
`CREATE TABLE transaction_batches (
id UUID PRIMARY KEY,
crypto_code TEXT NOT NULL,
status transaction_batch_status NOT NULL DEFAULT 'open',
created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
closed_at TIMESTAMPTZ,
error_message TEXT
)`,
`ALTER TABLE cash_in_txs ADD COLUMN batch_id UUID REFERENCES transaction_batches(id)`,
`ALTER TABLE cash_in_txs ADD COLUMN batched BOOLEAN NOT NULL DEFAULT false`,
`ALTER TABLE cash_in_txs ADD COLUMN batch_time TIMESTAMPTZ`
]
db.multi(sql, next)
}
exports.down = function (next) {
next()
}

2
new-lamassu-admin/src/pages/Transactions/Transactions.js
View file

@ -110,6 +110,8 @@ const GET_TRANSACTIONS = gql`
discount
customerId
isAnonymous
batched
batchTime
}
}
`

1
new-lamassu-admin/src/pages/Transactions/helper.js
View file

@ -11,6 +11,7 @@ const getCashInStatus = it => {
if (it.hasError) return 'Error'
if (it.sendConfirmed) return 'Sent'
if (it.expired) return 'Expired'
if (it.batched) return 'Batched'
return 'Pending'
}