machankura/mempool
3
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Slow down retries on Electrs errors in the main mempool loop.

Browse Source
This commit is contained in:
softsimon 2020-10-18 21:47:47 +07:00
parent 2ba7cd9ebd
commit 49f70ca28a
No known key found for this signature in database
GPG Key ID: 488D7DCFB5A430D7
4 changed files with 161 additions and 165 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

16
backend/src/api/bitcoin/electrs-api.ts
View File

@ -11,7 +11,7 @@ class ElectrsApi {
return new Promise((resolve, reject) => {
request(config.ELECTRS_API_URL + '/mempool', { json: true, timeout: 10000 }, (err, res, response) => {
if (err) {
reject(err);
reject('getMempoolInfo error: ' + err.message || err);
} else if (res.statusCode !== 200) {
reject(response);
} else {
@ -32,7 +32,7 @@ class ElectrsApi {
return new Promise((resolve, reject) => {
request(config.ELECTRS_API_URL + '/mempool/txids', { json: true, timeout: 10000, forever: true }, (err, res, response) => {
if (err) {
reject(err);
reject('getRawMempool error: ' + err.message || err);
} else if (res.statusCode !== 200) {
reject(response);
} else {
@ -50,7 +50,7 @@ class ElectrsApi {
return new Promise((resolve, reject) => {
request(config.ELECTRS_API_URL + '/tx/' + txId, { json: true, timeout: 10000, forever: true }, (err, res, response) => {
if (err) {
reject(err);
reject('getRawTransaction error: ' + err.message || err);
} else if (res.statusCode !== 200) {
reject(response);
} else {
@ -68,7 +68,7 @@ class ElectrsApi {
return new Promise((resolve, reject) => {
request(config.ELECTRS_API_URL + '/blocks/tip/height', { json: true, timeout: 10000 }, (err, res, response) => {
if (err) {
reject(err);
reject('getBlockHeightTip error: ' + err.message || err);
} else if (res.statusCode !== 200) {
reject(response);
} else {
@ -82,7 +82,7 @@ class ElectrsApi {
return new Promise((resolve, reject) => {
request(config.ELECTRS_API_URL + '/block/' + hash + '/txids', { json: true, timeout: 10000 }, (err, res, response) => {
if (err) {
reject(err);
reject('getTxIdsForBlock error: ' + err.message || err);
} else if (res.statusCode !== 200) {
reject(response);
} else {
@ -100,7 +100,7 @@ class ElectrsApi {
return new Promise((resolve, reject) => {
request(config.ELECTRS_API_URL + '/block-height/' + height, { json: true, timeout: 10000 }, (err, res, response) => {
if (err) {
reject(err);
reject('getBlockHash error: ' + err.message || err);
} else if (res.statusCode !== 200) {
reject(response);
} else {
@ -114,7 +114,7 @@ class ElectrsApi {
return new Promise((resolve, reject) => {
request(config.ELECTRS_API_URL + '/blocks/' + height, { json: true, timeout: 10000 }, (err, res, response) => {
if (err) {
reject(err);
reject('getBlocksFromHeight error: ' + err.message || err);
} else if (res.statusCode !== 200) {
reject(response);
} else {
@ -128,7 +128,7 @@ class ElectrsApi {
return new Promise((resolve, reject) => {
request(config.ELECTRS_API_URL + '/block/' + hash, { json: true, timeout: 10000 }, (err, res, response) => {
if (err) {
reject(err);
reject('getBlock error: ' + err.message || err);
} else if (res.statusCode !== 200) {
reject(response);
} else {

129
backend/src/api/blocks.ts
View File

@ -25,84 +25,79 @@ class Blocks {
this.newBlockCallbacks.push(fn);
}
public async updateBlocks() {
try {
const blockHeightTip = await bitcoinApi.getBlockHeightTip();
public async $updateBlocks() {
const blockHeightTip = await bitcoinApi.getBlockHeightTip();
if (this.blocks.length === 0) {
this.currentBlockHeight = blockHeightTip - config.INITIAL_BLOCK_AMOUNT;
if (this.blocks.length === 0) {
this.currentBlockHeight = blockHeightTip - config.INITIAL_BLOCK_AMOUNT;
} else {
this.currentBlockHeight = this.blocks[this.blocks.length - 1].height;
}
if (blockHeightTip - this.currentBlockHeight > config.INITIAL_BLOCK_AMOUNT * 2) {
logger.info(`${blockHeightTip - this.currentBlockHeight} blocks since tip. Fast forwarding to the ${config.INITIAL_BLOCK_AMOUNT} recent blocks`);
this.currentBlockHeight = blockHeightTip - config.INITIAL_BLOCK_AMOUNT;
}
if (!this.lastDifficultyAdjustmentTime) {
const heightDiff = blockHeightTip % 2016;
const blockHash = await bitcoinApi.getBlockHash(blockHeightTip - heightDiff);
const block = await bitcoinApi.getBlock(blockHash);
this.lastDifficultyAdjustmentTime = block.timestamp;
}
while (this.currentBlockHeight < blockHeightTip) {
if (this.currentBlockHeight === 0) {
this.currentBlockHeight = blockHeightTip;
} else {
this.currentBlockHeight = this.blocks[this.blocks.length - 1].height;
this.currentBlockHeight++;
logger.debug(`New block found (#${this.currentBlockHeight})!`);
}
if (blockHeightTip - this.currentBlockHeight > config.INITIAL_BLOCK_AMOUNT * 2) {
logger.info(`${blockHeightTip - this.currentBlockHeight} blocks since tip. Fast forwarding to the ${config.INITIAL_BLOCK_AMOUNT} recent blocks`);
this.currentBlockHeight = blockHeightTip - config.INITIAL_BLOCK_AMOUNT;
const blockHash = await bitcoinApi.getBlockHash(this.currentBlockHeight);
const block = await bitcoinApi.getBlock(blockHash);
const txIds = await bitcoinApi.getTxIdsForBlock(blockHash);
const mempool = memPool.getMempool();
let found = 0;
let notFound = 0;
const transactions: TransactionExtended[] = [];
for (let i = 0; i < txIds.length; i++) {
if (mempool[txIds[i]]) {
transactions.push(mempool[txIds[i]]);
found++;
} else {
logger.debug(`Fetching block tx ${i} of ${txIds.length}`);
const tx = await memPool.getTransactionExtended(txIds[i]);
if (tx) {
transactions.push(tx);
}
notFound++;
}
}
if (!this.lastDifficultyAdjustmentTime) {
const heightDiff = blockHeightTip % 2016;
const blockHash = await bitcoinApi.getBlockHash(blockHeightTip - heightDiff);
const block = await bitcoinApi.getBlock(blockHash);
logger.debug(`${found} of ${txIds.length} found in mempool. ${notFound} not found.`);
block.reward = transactions[0].vout.reduce((acc, curr) => acc + curr.value, 0);
block.coinbaseTx = this.stripCoinbaseTransaction(transactions[0]);
transactions.sort((a, b) => b.feePerVsize - a.feePerVsize);
block.medianFee = transactions.length > 1 ? Common.median(transactions.map((tx) => tx.feePerVsize)) : 0;
block.feeRange = transactions.length > 1 ? Common.getFeesInRange(transactions.slice(0, transactions.length - 1), 8) : [0, 0];
if (block.height % 2016 === 0) {
this.lastDifficultyAdjustmentTime = block.timestamp;
}
while (this.currentBlockHeight < blockHeightTip) {
if (this.currentBlockHeight === 0) {
this.currentBlockHeight = blockHeightTip;
} else {
this.currentBlockHeight++;
logger.debug(`New block found (#${this.currentBlockHeight})!`);
}
const blockHash = await bitcoinApi.getBlockHash(this.currentBlockHeight);
const block = await bitcoinApi.getBlock(blockHash);
const txIds = await bitcoinApi.getTxIdsForBlock(blockHash);
const mempool = memPool.getMempool();
let found = 0;
let notFound = 0;
const transactions: TransactionExtended[] = [];
for (let i = 0; i < txIds.length; i++) {
if (mempool[txIds[i]]) {
transactions.push(mempool[txIds[i]]);
found++;
} else {
logger.debug(`Fetching block tx ${i} of ${txIds.length}`);
const tx = await memPool.getTransactionExtended(txIds[i]);
if (tx) {
transactions.push(tx);
}
notFound++;
}
}
logger.debug(`${found} of ${txIds.length} found in mempool. ${notFound} not found.`);
block.reward = transactions[0].vout.reduce((acc, curr) => acc + curr.value, 0);
block.coinbaseTx = this.stripCoinbaseTransaction(transactions[0]);
transactions.sort((a, b) => b.feePerVsize - a.feePerVsize);
block.medianFee = transactions.length > 1 ? Common.median(transactions.map((tx) => tx.feePerVsize)) : 0;
block.feeRange = transactions.length > 1 ? Common.getFeesInRange(transactions.slice(0, transactions.length - 1), 8) : [0, 0];
if (block.height % 2016 === 0) {
this.lastDifficultyAdjustmentTime = block.timestamp;
}
this.blocks.push(block);
if (this.blocks.length > config.KEEP_BLOCK_AMOUNT) {
this.blocks.shift();
}
if (this.newBlockCallbacks.length) {
this.newBlockCallbacks.forEach((cb) => cb(block, txIds, transactions));
}
this.blocks.push(block);
if (this.blocks.length > config.KEEP_BLOCK_AMOUNT) {
this.blocks.shift();
}
} catch (err) {
logger.err('updateBlocks error' + err);
if (this.newBlockCallbacks.length) {
this.newBlockCallbacks.forEach((cb) => cb(block, txIds, transactions));
}
}
}

164
backend/src/api/mempool.ts
View File

@ -47,12 +47,8 @@ class Mempool {
}
}
public async updateMemPoolInfo() {
try {
this.mempoolInfo = await bitcoinApi.getMempoolInfo();
} catch (e) {
logger.err('Error getMempoolInfo ' + e.message || e);
}
public async $updateMemPoolInfo() {
this.mempoolInfo = await bitcoinApi.getMempoolInfo();
}
public getMempoolInfo(): MempoolInfo | undefined {
@ -93,100 +89,96 @@ class Mempool {
}
}
public async updateMempool() {
public async $updateMempool() {
logger.debug('Updating mempool');
const start = new Date().getTime();
let hasChange: boolean = false;
const currentMempoolSize = Object.keys(this.mempoolCache).length;
let txCount = 0;
try {
const transactions = await bitcoinApi.getRawMempool();
const diff = transactions.length - currentMempoolSize;
const newTransactions: TransactionExtended[] = [];
const transactions = await bitcoinApi.getRawMempool();
const diff = transactions.length - currentMempoolSize;
const newTransactions: TransactionExtended[] = [];
for (const txid of transactions) {
if (!this.mempoolCache[txid]) {
const transaction = await this.getTransactionExtended(txid);
if (transaction) {
this.mempoolCache[txid] = transaction;
txCount++;
if (this.inSync) {
this.txPerSecondArray.push(new Date().getTime());
this.vBytesPerSecondArray.push({
unixTime: new Date().getTime(),
vSize: transaction.vsize,
});
}
hasChange = true;
if (diff > 0) {
logger.debug('Fetched transaction ' + txCount + ' / ' + diff);
} else {
logger.debug('Fetched transaction ' + txCount);
}
newTransactions.push(transaction);
} else {
logger.debug('Error finding transaction in mempool.');
for (const txid of transactions) {
if (!this.mempoolCache[txid]) {
const transaction = await this.getTransactionExtended(txid);
if (transaction) {
this.mempoolCache[txid] = transaction;
txCount++;
if (this.inSync) {
this.txPerSecondArray.push(new Date().getTime());
this.vBytesPerSecondArray.push({
unixTime: new Date().getTime(),
vSize: transaction.vsize,
});
}
}
if ((new Date().getTime()) - start > config.MEMPOOL_REFRESH_RATE_MS * 10) {
break;
}
}
// Prevent mempool from clear on bitcoind restart by delaying the deletion
if ((config.NETWORK === 'mainnet' || !config.NETWORK)
&& this.mempoolProtection === 0 && transactions.length / currentMempoolSize <= 0.80) {
this.mempoolProtection = 1;
this.inSync = false;
logger.warn(`Mempool clear protection triggered because transactions.length: ${transactions.length} and currentMempoolSize: ${currentMempoolSize}.`);
setTimeout(() => {
this.mempoolProtection = 2;
logger.warn('Mempool clear protection resumed.');
}, 1000 * 60 * 2);
}
let newMempool = {};
const deletedTransactions: TransactionExtended[] = [];
if (this.mempoolProtection !== 1) {
this.mempoolProtection = 0;
// Index object for faster search
const transactionsObject = {};
transactions.forEach((txId) => transactionsObject[txId] = true);
// Replace mempool to separate deleted transactions
for (const tx in this.mempoolCache) {
if (transactionsObject[tx]) {
newMempool[tx] = this.mempoolCache[tx];
hasChange = true;
if (diff > 0) {
logger.debug('Fetched transaction ' + txCount + ' / ' + diff);
} else {
deletedTransactions.push(this.mempoolCache[tx]);
logger.debug('Fetched transaction ' + txCount);
}
newTransactions.push(transaction);
} else {
logger.debug('Error finding transaction in mempool.');
}
} else {
newMempool = this.mempoolCache;
}
const newTransactionsStripped = newTransactions.map((tx) => Common.stripTransaction(tx));
this.latestTransactions = newTransactionsStripped.concat(this.latestTransactions).slice(0, 6);
if (!this.inSync && transactions.length === Object.keys(newMempool).length) {
this.inSync = true;
logger.info('The mempool is now in sync!');
if ((new Date().getTime()) - start > config.MEMPOOL_REFRESH_RATE_MS * 10) {
break;
}
if (this.mempoolChangedCallback && (hasChange || deletedTransactions.length)) {
this.mempoolCache = newMempool;
this.mempoolChangedCallback(this.mempoolCache, newTransactions, deletedTransactions);
}
const end = new Date().getTime();
const time = end - start;
logger.debug(`New mempool size: ${Object.keys(newMempool).length} Change: ${diff}`);
logger.debug('Mempool updated in ' + time / 1000 + ' seconds');
} catch (err) {
logger.err('getRawMempool error. ' + err.message || err);
}
// Prevent mempool from clear on bitcoind restart by delaying the deletion
if ((config.NETWORK === 'mainnet' || !config.NETWORK)
&& this.mempoolProtection === 0 && transactions.length / currentMempoolSize <= 0.80) {
this.mempoolProtection = 1;
this.inSync = false;
logger.warn(`Mempool clear protection triggered because transactions.length: ${transactions.length} and currentMempoolSize: ${currentMempoolSize}.`);
setTimeout(() => {
this.mempoolProtection = 2;
logger.warn('Mempool clear protection resumed.');
}, 1000 * 60 * 2);
}
let newMempool = {};
const deletedTransactions: TransactionExtended[] = [];
if (this.mempoolProtection !== 1) {
this.mempoolProtection = 0;
// Index object for faster search
const transactionsObject = {};
transactions.forEach((txId) => transactionsObject[txId] = true);
// Replace mempool to separate deleted transactions
for (const tx in this.mempoolCache) {
if (transactionsObject[tx]) {
newMempool[tx] = this.mempoolCache[tx];
} else {
deletedTransactions.push(this.mempoolCache[tx]);
}
}
} else {
newMempool = this.mempoolCache;
}
const newTransactionsStripped = newTransactions.map((tx) => Common.stripTransaction(tx));
this.latestTransactions = newTransactionsStripped.concat(this.latestTransactions).slice(0, 6);
if (!this.inSync && transactions.length === Object.keys(newMempool).length) {
this.inSync = true;
logger.info('The mempool is now in sync!');
}
if (this.mempoolChangedCallback && (hasChange || deletedTransactions.length)) {
this.mempoolCache = newMempool;
this.mempoolChangedCallback(this.mempoolCache, newTransactions, deletedTransactions);
}
const end = new Date().getTime();
const time = end - start;
logger.debug(`New mempool size: ${Object.keys(newMempool).length} Change: ${diff}`);
logger.debug('Mempool updated in ' + time / 1000 + ' seconds');
}
private updateTxPerSecond() {

17
backend/src/index.ts
View File

@ -27,6 +27,7 @@ class Server {
private wss: WebSocket.Server | undefined;
private server: https.Server | http.Server | undefined;
private app: Express;
private retryOnElectrsErrorAfterSeconds = 5;
constructor() {
this.app = express();
@ -114,10 +115,18 @@ class Server {
}
async runMempoolIntervalFunctions() {
await memPool.updateMemPoolInfo();
await blocks.updateBlocks();
await memPool.updateMempool();
setTimeout(this.runMempoolIntervalFunctions.bind(this), config.ELECTRS_POLL_RATE_MS);
try {
await memPool.$updateMemPoolInfo();
await blocks.$updateBlocks();
await memPool.$updateMempool();
setTimeout(this.runMempoolIntervalFunctions.bind(this), config.ELECTRS_POLL_RATE_MS);
this.retryOnElectrsErrorAfterSeconds = 5;
} catch (e) {
this.retryOnElectrsErrorAfterSeconds *= 2;
this.retryOnElectrsErrorAfterSeconds = Math.min(this.retryOnElectrsErrorAfterSeconds, 3600);
logger.warn(`runMempoolIntervalFunctions error: ${(e.message || e)}. Retrying in ${this.retryOnElectrsErrorAfterSeconds} sec.`);
setTimeout(this.runMempoolIntervalFunctions.bind(this), 1000 * this.retryOnElectrsErrorAfterSeconds);
}
}
setUpWebsocketHandling() {