feat: standardize balance memoizing speed
feat: add infura-only override for getStatus() fix: remove coin filtering on poller feat: add getStatus() response caching on infura
This commit is contained in:
Sérgio Salgado
parent
850f488963
commit
6f8561850a
5 changed files with 100 additions and 44 deletions
|
|
@ -25,7 +25,6 @@ module.exports = {
|
|||
|
||||
const STALE_INCOMING_TX_AGE = T.day
|
||||
const STALE_LIVE_INCOMING_TX_AGE = 10 * T.minutes
|
||||
const STALE_LIVE_INCOMING_TX_AGE_FILTER = 5 * T.minutes
|
||||
const MAX_NOTIFY_AGE = T.day
|
||||
const MIN_NOTIFY_AGE = 5 * T.minutes
|
||||
const INSUFFICIENT_FUNDS_CODE = 570
|
||||
|
|
@ -91,21 +90,17 @@ function postProcess (txVector, justAuthorized, pi) {
|
|||
return Promise.resolve({})
|
||||
}
|
||||
|
||||
function fetchOpenTxs (statuses, fromAge, toAge, applyFilter, coinFilter) {
|
||||
const notClause = applyFilter ? '' : 'not'
|
||||
function fetchOpenTxs (statuses, fromAge, toAge) {
|
||||
const sql = `select *
|
||||
from cash_out_txs
|
||||
where ((extract(epoch from (now() - created))) * 1000)>$1
|
||||
and ((extract(epoch from (now() - created))) * 1000)<$2
|
||||
${_.isEmpty(coinFilter)
|
||||
? ``
|
||||
: `and crypto_code ${notClause} in ($3^)`}
|
||||
and status in ($4^)`
|
||||
and status in ($3^)
|
||||
and error is distinct from 'Operator cancel'`
|
||||
|
||||
const coinClause = _.map(pgp.as.text, coinFilter).join(',')
|
||||
const statusClause = _.map(pgp.as.text, statuses).join(',')
|
||||
|
||||
return db.any(sql, [fromAge, toAge, coinClause, statusClause])
|
||||
return db.any(sql, [fromAge, toAge, statusClause])
|
||||
.then(rows => rows.map(toObj))
|
||||
}
|
||||
|
||||
|
|
@ -164,22 +159,18 @@ function getWalletScore (tx, pi) {
|
|||
return tx
|
||||
}
|
||||
|
||||
function monitorLiveIncoming (settings, applyFilter, coinFilter) {
|
||||
function monitorLiveIncoming (settings) {
|
||||
const statuses = ['notSeen', 'published', 'insufficientFunds']
|
||||
const toAge = applyFilter ? STALE_LIVE_INCOMING_TX_AGE_FILTER : STALE_LIVE_INCOMING_TX_AGE
|
||||
|
||||
return monitorIncoming(settings, statuses, 0, toAge, applyFilter, coinFilter)
|
||||
return monitorIncoming(settings, statuses, 0, STALE_LIVE_INCOMING_TX_AGE)
|
||||
}
|
||||
|
||||
function monitorStaleIncoming (settings, applyFilter, coinFilter) {
|
||||
function monitorStaleIncoming (settings) {
|
||||
const statuses = ['notSeen', 'published', 'authorized', 'instant', 'rejected', 'insufficientFunds']
|
||||
const fromAge = applyFilter ? STALE_LIVE_INCOMING_TX_AGE_FILTER : STALE_LIVE_INCOMING_TX_AGE
|
||||
|
||||
return monitorIncoming(settings, statuses, fromAge, STALE_INCOMING_TX_AGE, applyFilter, coinFilter)
|
||||
return monitorIncoming(settings, statuses, STALE_LIVE_INCOMING_TX_AGE, STALE_INCOMING_TX_AGE)
|
||||
}
|
||||
|
||||
function monitorIncoming (settings, statuses, fromAge, toAge, applyFilter, coinFilter) {
|
||||
return fetchOpenTxs(statuses, fromAge, toAge, applyFilter, coinFilter)
|
||||
function monitorIncoming (settings, statuses, fromAge, toAge) {
|
||||
return fetchOpenTxs(statuses, fromAge, toAge)
|
||||
.then(txs => pEachSeries(txs, tx => processTxStatus(tx, settings)))
|
||||
.catch(err => {
|
||||
if (err.code === dbErrorCodes.SERIALIZATION_FAILURE) {
|
||||
|
|
|
|||
|
|
@ -24,6 +24,11 @@ const RECEIPT = 'sms_receipt'
|
|||
|
||||
const WALLET_SCORE_THRESHOLD = 9
|
||||
|
||||
const BALANCE_FETCH_SPEED_MULTIPLIER = {
|
||||
NORMAL: 1,
|
||||
SLOW: 3
|
||||
}
|
||||
|
||||
module.exports = {
|
||||
anonymousCustomer,
|
||||
CASSETTE_MAX_CAPACITY,
|
||||
|
|
@ -39,5 +44,7 @@ module.exports = {
|
|||
CASH_OUT_MINIMUM_AMOUNT_OF_CASSETTES,
|
||||
CASH_OUT_MAXIMUM_AMOUNT_OF_CASSETTES,
|
||||
WALLET_SCORE_THRESHOLD,
|
||||
RECEIPT
|
||||
RECEIPT,
|
||||
PSQL_URL,
|
||||
BALANCE_FETCH_SPEED_MULTIPLIER
|
||||
}
|
||||
|
|
|
|||
|
|
@ -1,5 +1,9 @@
|
|||
const _ = require('lodash/fp')
|
||||
const base = require('../geth/base')
|
||||
const T = require('../../../time')
|
||||
const { BALANCE_FETCH_SPEED_MULTIPLIER } = require('../../../constants')
|
||||
|
||||
const REGULAR_TX_POLLING = 5 * T.seconds
|
||||
|
||||
const NAME = 'infura'
|
||||
|
||||
|
|
@ -12,4 +16,60 @@ function run (account) {
|
|||
base.connect(endpoint)
|
||||
}
|
||||
|
||||
module.exports = _.merge(base, { NAME, run })
|
||||
const txsCache = {}
|
||||
setInterval(() => {
|
||||
_.forEach(it => {
|
||||
// A transaction not being updated for over an hour means that it stopped being fetched on poll,
|
||||
// which means that it probably was cancelled by the operator or something else caused the
|
||||
// transaction to stop being listened to. So, trim the cache to save memory
|
||||
const lastReq = _.isNil(txsCache[it].lastReqTime) ? Date.now() : new Date(txsCache[it].lastReqTime)
|
||||
const timePassedSinceReq = Date.now() - lastReq
|
||||
if (timePassedSinceReq > T.hour) delete txsCache[it]
|
||||
}, _.keys(txsCache))
|
||||
}, T.minute)
|
||||
|
||||
function shouldGetStatus (tx) {
|
||||
const timePassedSinceTx = Date.now() - new Date(tx.created)
|
||||
const timePassedSinceReq = Date.now() - new Date(txsCache[tx.id].lastReqTime)
|
||||
|
||||
// Allow for infura to gradually lower the amount of requests based on the time passed since the transaction
|
||||
// Until first 5 minutes - 1/2 regular polling speed
|
||||
// Until first 10 minutes - 1/4 regular polling speed
|
||||
// Until first hour - 1/8 polling speed
|
||||
// Until first two hours - 1/12 polling speed
|
||||
// Until first four hours - 1/16 polling speed
|
||||
// Until first day - 1/24 polling speed
|
||||
// After first day - 1/32 polling speed
|
||||
if (timePassedSinceTx < 5 * T.minutes) return _.isNil(txsCache[tx.id].res) || timePassedSinceReq > 2 * REGULAR_TX_POLLING
|
||||
if (timePassedSinceTx < 10 * T.minutes) return _.isNil(txsCache[tx.id].res) || timePassedSinceReq > 4 * REGULAR_TX_POLLING
|
||||
if (timePassedSinceTx < 1 * T.hour) return _.isNil(txsCache[tx.id].res) || timePassedSinceReq > 8 * REGULAR_TX_POLLING
|
||||
if (timePassedSinceTx < 2 * T.hours) return _.isNil(txsCache[tx.id].res) || timePassedSinceReq > 12 * REGULAR_TX_POLLING
|
||||
if (timePassedSinceTx < 4 * T.hours) return _.isNil(txsCache[tx.id].res) || timePassedSinceReq > 16 * REGULAR_TX_POLLING
|
||||
if (timePassedSinceTx < 1 * T.day) return _.isNil(txsCache[tx.id].res) || timePassedSinceReq > 24 * REGULAR_TX_POLLING
|
||||
return _.isNil(txsCache[tx.id].res) || timePassedSinceReq > 32 * REGULAR_TX_POLLING
|
||||
}
|
||||
|
||||
// Override geth's getStatus function to allow for different polling timing
|
||||
function getStatus (account, tx, requested, settings, operatorId) {
|
||||
if (_.isNil(txsCache[tx.id])) {
|
||||
txsCache[tx.id] = { lastReqTime: Date.now() }
|
||||
}
|
||||
|
||||
// return last available response
|
||||
if (!shouldGetStatus(tx)) {
|
||||
return Promise.resolve(txsCache[tx.id].res)
|
||||
}
|
||||
|
||||
return base.getStatus(account, tx, requested, settings, operatorId)
|
||||
.then(res => {
|
||||
if (res.status === 'confirmed') {
|
||||
delete txsCache[tx.id] // Transaction reached final status, can trim it from the caching obj
|
||||
} else {
|
||||
txsCache[tx.id].lastReqTime = Date.now()
|
||||
txsCache[tx.id].res = res
|
||||
}
|
||||
return res
|
||||
})
|
||||
}
|
||||
|
||||
module.exports = _.merge(base, { NAME, run, getStatus, fetchSpeed: BALANCE_FETCH_SPEED_MULTIPLIER.SLOW })
|
||||
|
|
|
|||
|
|
@ -21,8 +21,6 @@ const processBatches = require('./tx-batching-processing')
|
|||
|
||||
const INCOMING_TX_INTERVAL = 30 * T.seconds
|
||||
const LIVE_INCOMING_TX_INTERVAL = 5 * T.seconds
|
||||
const INCOMING_TX_INTERVAL_FILTER = 1 * T.minute
|
||||
const LIVE_INCOMING_TX_INTERVAL_FILTER = 10 * T.seconds
|
||||
const UNNOTIFIED_INTERVAL = 10 * T.seconds
|
||||
const SWEEP_HD_INTERVAL = T.minute
|
||||
const TRADE_INTERVAL = 60 * T.seconds
|
||||
|
|
@ -60,7 +58,6 @@ const QUEUE = {
|
|||
SLOW: SLOW_QUEUE
|
||||
}
|
||||
|
||||
const coinFilter = ['ETH']
|
||||
const schemaCallbacks = new Map()
|
||||
|
||||
const cachedVariables = new NodeCache({
|
||||
|
|
@ -167,24 +164,16 @@ function doPolling (schema) {
|
|||
pi().executeTrades()
|
||||
pi().pong()
|
||||
pi().clearOldLogs()
|
||||
cashOutTx.monitorLiveIncoming(settings(), false, coinFilter)
|
||||
cashOutTx.monitorStaleIncoming(settings(), false, coinFilter)
|
||||
if (!_.isEmpty(coinFilter)) {
|
||||
cashOutTx.monitorLiveIncoming(settings(), true, coinFilter)
|
||||
cashOutTx.monitorStaleIncoming(settings(), true, coinFilter)
|
||||
}
|
||||
cashOutTx.monitorLiveIncoming(settings())
|
||||
cashOutTx.monitorStaleIncoming(settings())
|
||||
cashOutTx.monitorUnnotified(settings())
|
||||
pi().sweepHd()
|
||||
notifier.checkNotification(pi())
|
||||
updateCoinAtmRadar()
|
||||
|
||||
addToQueue(pi().executeTrades, TRADE_INTERVAL, schema, QUEUE.FAST)
|
||||
addToQueue(cashOutTx.monitorLiveIncoming, LIVE_INCOMING_TX_INTERVAL, schema, QUEUE.FAST, settings, false, coinFilter)
|
||||
addToQueue(cashOutTx.monitorStaleIncoming, INCOMING_TX_INTERVAL, schema, QUEUE.FAST, settings, false, coinFilter)
|
||||
if (!_.isEmpty(coinFilter)) {
|
||||
addToQueue(cashOutTx.monitorLiveIncoming, LIVE_INCOMING_TX_INTERVAL_FILTER, schema, QUEUE.FAST, settings, true, coinFilter)
|
||||
addToQueue(cashOutTx.monitorStaleIncoming, INCOMING_TX_INTERVAL_FILTER, schema, QUEUE.FAST, settings, true, coinFilter)
|
||||
}
|
||||
addToQueue(cashOutTx.monitorLiveIncoming, LIVE_INCOMING_TX_INTERVAL, schema, QUEUE.FAST, settings)
|
||||
addToQueue(cashOutTx.monitorStaleIncoming, INCOMING_TX_INTERVAL, schema, QUEUE.FAST, settings)
|
||||
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)
|
||||
|
|
|
|||
|
|
@ -14,6 +14,7 @@ const httpError = require('./route-helpers').httpError
|
|||
const logger = require('./logger')
|
||||
const { getOpenBatchCryptoValue } = require('./tx-batching')
|
||||
const BN = require('./bn')
|
||||
const { BALANCE_FETCH_SPEED_MULTIPLIER } = require('./constants')
|
||||
|
||||
const FETCH_INTERVAL = 5000
|
||||
const INSUFFICIENT_FUNDS_CODE = 570
|
||||
|
|
@ -254,20 +255,28 @@ function checkBlockchainStatus (settings, cryptoCode) {
|
|||
.then(({ checkBlockchainStatus }) => checkBlockchainStatus(cryptoCode))
|
||||
}
|
||||
|
||||
const coinFilter = ['ETH']
|
||||
|
||||
const balance = (settings, cryptoCode) => {
|
||||
if (_.includes(coinFilter, cryptoCode)) return balanceFiltered(settings, cryptoCode)
|
||||
return balanceUnfiltered(settings, cryptoCode)
|
||||
return fetchWallet(settings, cryptoCode)
|
||||
.then(r => r.wallet.fetchSpeed ?? BALANCE_FETCH_SPEED_MULTIPLIER.NORMAL)
|
||||
.then(multiplier => {
|
||||
switch (multiplier) {
|
||||
case BALANCE_FETCH_SPEED_MULTIPLIER.NORMAL:
|
||||
return balanceNormal(settings, cryptoCode)
|
||||
case BALANCE_FETCH_SPEED_MULTIPLIER.SLOW:
|
||||
return balanceSlow(settings, cryptoCode)
|
||||
default:
|
||||
throw new Error()
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
const balanceUnfiltered = mem(_balance, {
|
||||
maxAge: FETCH_INTERVAL,
|
||||
const balanceNormal = mem(_balance, {
|
||||
maxAge: BALANCE_FETCH_SPEED_MULTIPLIER.NORMAL * FETCH_INTERVAL,
|
||||
cacheKey: (settings, cryptoCode) => cryptoCode
|
||||
})
|
||||
|
||||
const balanceFiltered = mem(_balance, {
|
||||
maxAge: 3 * FETCH_INTERVAL,
|
||||
const balanceSlow = mem(_balance, {
|
||||
maxAge: BALANCE_FETCH_SPEED_MULTIPLIER.SLOW * FETCH_INTERVAL,
|
||||
cacheKey: (settings, cryptoCode) => cryptoCode
|
||||
})
|
||||
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue