From 117803871093e547d02add446d58685634aa19e4 Mon Sep 17 00:00:00 2001 From: Joe Uhren Date: Sun, 17 Jul 2022 16:49:02 -0600 Subject: [PATCH] Better handling of stop/kill sync processes -The SIGINT and SIGTERM signals are now being caught and handled in the sync.js file so that most options for syncing blocks, markets, peers, masternodes, etc. are now being gracefully shut down instead of killed in the middle of the process. This should help prevent data anomalies when you need to stop or kill a sync process -The update_tx_db function was moved from the database.js file into the sync.js file so that block syncs can now be gracefully stopped. The update_tx_db function was also copied to the benchmark.js -The save_tx function was moved into the module.exports for the database.js file so that it can now be called from outside the database.js file --- lib/database.js | 241 +++++++++++------------------------ scripts/benchmark.js | 93 +++++++++++++- scripts/sync.js | 296 +++++++++++++++++++++++++++++++++++-------- 3 files changed, 410 insertions(+), 220 deletions(-) diff --git a/lib/database.js b/lib/database.js index db4a168e..6fd14b12 100644 --- a/lib/database.js +++ b/lib/database.js @@ -13,8 +13,7 @@ var mongoose = require('mongoose'), settings = require('./settings'), locale = require('./locale'), fs = require('fs'), - coingecko = require('./apis/coingecko'), - async = require('async'); + coingecko = require('./apis/coingecko'); function find_address(hash, caseSensitive, cb) { if (caseSensitive) { @@ -114,82 +113,6 @@ function find_tx(txid, cb) { }); } -function save_tx(txid, blockheight, cb) { - lib.get_rawtransaction(txid, function(tx) { - if (tx && tx != 'There was an error. Check your console.') { - lib.prepare_vin(tx, function(vin, tx_type_vin) { - lib.prepare_vout(tx.vout, txid, vin, ((!settings.blockchain_specific.zksnarks.enabled || typeof tx.vjoinsplit === 'undefined' || tx.vjoinsplit == null) ? [] : tx.vjoinsplit), function(vout, nvin, tx_type_vout) { - lib.syncLoop(vin.length, function (loop) { - var i = loop.iteration(); - - // check if address is inside an array - if (Array.isArray(nvin[i].addresses)) { - // extract the address - nvin[i].addresses = nvin[i].addresses[0]; - } - - update_address(nvin[i].addresses, blockheight, txid, nvin[i].amount, 'vin', function() { - loop.next(); - }); - }, function() { - lib.syncLoop(vout.length, function (subloop) { - var t = subloop.iteration(); - - // check if address is inside an array - if (Array.isArray(vout[t].addresses)) { - // extract the address - vout[t].addresses = vout[t].addresses[0]; - } - - if (vout[t].addresses) { - update_address(vout[t].addresses, blockheight, txid, vout[t].amount, 'vout', function() { - subloop.next(); - }); - } else - subloop.next(); - }, function() { - lib.calculate_total(vout, function(total) { - var op_return = null; - // check if the op_return value should be decoded and saved - if (settings.transaction_page.show_op_return) { - // loop through vout to find the op_return value - tx.vout.forEach(function (vout_data) { - // check if the op_return value exists - if (vout_data.scriptPubKey != null && vout_data.scriptPubKey.asm != null && vout_data.scriptPubKey.asm.indexOf('OP_RETURN') > -1) { - // decode the op_return value - op_return = hex_to_ascii(vout_data.scriptPubKey.asm.replace('OP_RETURN', '').trim()); - } - }); - } - - var newTx = new Tx({ - txid: tx.txid, - vin: nvin, - vout: vout, - total: total.toFixed(8), - timestamp: tx.time, - blockhash: tx.blockhash, - blockindex: blockheight, - tx_type: (tx_type_vout == null ? tx_type_vin : tx_type_vout), - op_return: op_return - }); - - newTx.save(function(err) { - if (err) - return cb(err, false); - else - return cb(null, vout.length > 0); - }); - }); - }); - }); - }); - }); - } else - return cb('tx not found: ' + txid, false); - }); -} - function get_market_data(market, coin_symbol, pair_symbol, cb) { if (fs.existsSync('./lib/markets/' + market + '.js')) { exMarket = require('./markets/' + market); @@ -1156,92 +1079,6 @@ module.exports = { }); }, - // updates tx, address & richlist db's; called by sync.js - update_tx_db: function(coin, start, end, txes, timeout, check_only, cb) { - var complete = false; - var blocks_to_scan = []; - var task_limit_blocks = settings.sync.block_parallel_tasks; - var task_limit_txs = 1; - - // fix for invalid block height (skip genesis block as it should not have valid txs) - if (typeof start === 'undefined' || start < 1) - start = 1; - - if (task_limit_blocks < 1) - task_limit_blocks = 1; - - for (i = start; i < (end + 1); i++) - blocks_to_scan.push(i); - - async.eachLimit(blocks_to_scan, task_limit_blocks, function(block_height, next_block) { - if (!check_only && block_height % settings.sync.save_stats_after_sync_blocks === 0) { - Stats.updateOne({coin: coin}, { - last: block_height - 1, - txes: txes - }, function() {}); - } else if (check_only) { - console.log('Checking block ' + block_height + '...'); - } - - lib.get_blockhash(block_height, function(blockhash) { - if (blockhash) { - lib.get_block(blockhash, function(block) { - if (block) { - async.eachLimit(block.tx, task_limit_txs, function(txid, next_tx) { - Tx.findOne({txid: txid}, function(err, tx) { - if (tx) { - setTimeout( function() { - tx = null; - next_tx(); - }, timeout); - } else { - save_tx(txid, block_height, function(err, tx_has_vout) { - if (err) - console.log(err); - else - console.log('%s: %s', block_height, txid); - - if (tx_has_vout) - txes++; - - setTimeout( function() { - tx = null; - next_tx(); - }, timeout); - }); - } - }); - }, function() { - setTimeout( function() { - blockhash = null; - block = null; - next_block(); - }, timeout); - }); - } else { - console.log('Block not found: %s', blockhash); - - setTimeout( function() { - next_block(); - }, timeout); - } - }); - } else { - setTimeout( function() { - next_block(); - }, timeout); - } - }); - }, function() { - Stats.updateOne({coin: coin}, { - last: end, - txes: txes - }, function() { - return cb(); - }); - }); - }, - create_peer: function(params, cb) { var newPeer = new Peers(params); @@ -1667,5 +1504,81 @@ module.exports = { } }, + save_tx: function(txid, blockheight, cb) { + lib.get_rawtransaction(txid, function(tx) { + if (tx && tx != 'There was an error. Check your console.') { + lib.prepare_vin(tx, function(vin, tx_type_vin) { + lib.prepare_vout(tx.vout, txid, vin, ((!settings.blockchain_specific.zksnarks.enabled || typeof tx.vjoinsplit === 'undefined' || tx.vjoinsplit == null) ? [] : tx.vjoinsplit), function(vout, nvin, tx_type_vout) { + lib.syncLoop(vin.length, function (loop) { + var i = loop.iteration(); + + // check if address is inside an array + if (Array.isArray(nvin[i].addresses)) { + // extract the address + nvin[i].addresses = nvin[i].addresses[0]; + } + + update_address(nvin[i].addresses, blockheight, txid, nvin[i].amount, 'vin', function() { + loop.next(); + }); + }, function() { + lib.syncLoop(vout.length, function (subloop) { + var t = subloop.iteration(); + + // check if address is inside an array + if (Array.isArray(vout[t].addresses)) { + // extract the address + vout[t].addresses = vout[t].addresses[0]; + } + + if (vout[t].addresses) { + update_address(vout[t].addresses, blockheight, txid, vout[t].amount, 'vout', function() { + subloop.next(); + }); + } else + subloop.next(); + }, function() { + lib.calculate_total(vout, function(total) { + var op_return = null; + // check if the op_return value should be decoded and saved + if (settings.transaction_page.show_op_return) { + // loop through vout to find the op_return value + tx.vout.forEach(function (vout_data) { + // check if the op_return value exists + if (vout_data.scriptPubKey != null && vout_data.scriptPubKey.asm != null && vout_data.scriptPubKey.asm.indexOf('OP_RETURN') > -1) { + // decode the op_return value + op_return = hex_to_ascii(vout_data.scriptPubKey.asm.replace('OP_RETURN', '').trim()); + } + }); + } + + var newTx = new Tx({ + txid: tx.txid, + vin: nvin, + vout: vout, + total: total.toFixed(8), + timestamp: tx.time, + blockhash: tx.blockhash, + blockindex: blockheight, + tx_type: (tx_type_vout == null ? tx_type_vin : tx_type_vout), + op_return: op_return + }); + + newTx.save(function(err) { + if (err) + return cb(err, false); + else + return cb(null, vout.length > 0); + }); + }); + }); + }); + }); + }); + } else + return cb('tx not found: ' + txid, false); + }); + }, + fs: fs }; \ No newline at end of file diff --git a/scripts/benchmark.js b/scripts/benchmark.js index f191d5ea..35a60086 100644 --- a/scripts/benchmark.js +++ b/scripts/benchmark.js @@ -2,7 +2,10 @@ var mongoose = require('mongoose'), db = require('../lib/database'), Tx = require('../models/tx'), Address = require('../models/address'), - settings = require('../lib/settings'); + settings = require('../lib/settings'), + lib = require('../lib/explorer'), + Stats = require('../models/stats'), + async = require('async'); var COUNT = 5000; // number of blocks to index @@ -27,7 +30,93 @@ mongoose.connect(dbString, function(err) { Address.deleteMany({}, function(err2) { var s_timer = new Date().getTime(); - db.update_tx_db(settings.coin.name, 1, COUNT, 0, settings.sync.update_timeout, false, function() { + // updates tx, address & richlist db's + function update_tx_db(coin, start, end, txes, timeout, check_only, cb) { + var complete = false; + var blocks_to_scan = []; + var task_limit_blocks = settings.sync.block_parallel_tasks; + var task_limit_txs = 1; + + // fix for invalid block height (skip genesis block as it should not have valid txs) + if (typeof start === 'undefined' || start < 1) + start = 1; + + if (task_limit_blocks < 1) + task_limit_blocks = 1; + + for (i = start; i < (end + 1); i++) + blocks_to_scan.push(i); + + async.eachLimit(blocks_to_scan, task_limit_blocks, function(block_height, next_block) { + if (!check_only && block_height % settings.sync.save_stats_after_sync_blocks === 0) { + Stats.updateOne({coin: coin}, { + last: block_height - 1, + txes: txes + }, function() {}); + } else if (check_only) { + console.log('Checking block ' + block_height + '...'); + } + + lib.get_blockhash(block_height, function(blockhash) { + if (blockhash) { + lib.get_block(blockhash, function(block) { + if (block) { + async.eachLimit(block.tx, task_limit_txs, function(txid, next_tx) { + Tx.findOne({txid: txid}, function(err, tx) { + if (tx) { + setTimeout( function() { + tx = null; + next_tx(); + }, timeout); + } else { + db.save_tx(txid, block_height, function(err, tx_has_vout) { + if (err) + console.log(err); + else + console.log('%s: %s', block_height, txid); + + if (tx_has_vout) + txes++; + + setTimeout( function() { + tx = null; + next_tx(); + }, timeout); + }); + } + }); + }, function() { + setTimeout( function() { + blockhash = null; + block = null; + next_block(); + }, timeout); + }); + } else { + console.log('Block not found: %s', blockhash); + + setTimeout( function() { + next_block(); + }, timeout); + } + }); + } else { + setTimeout( function() { + next_block(); + }, timeout); + } + }); + }, function() { + Stats.updateOne({coin: coin}, { + last: end, + txes: txes + }, function() { + return cb(); + }); + }); + } + + update_tx_db(settings.coin.name, 1, COUNT, 0, settings.sync.update_timeout, false, function() { var e_timer = new Date().getTime(); Tx.countDocuments({}, function(txerr, txcount) { diff --git a/scripts/sync.js b/scripts/sync.js index aeedc4b6..05909a09 100644 --- a/scripts/sync.js +++ b/scripts/sync.js @@ -6,11 +6,25 @@ var mongoose = require('mongoose'), AddressTx = require('../models/addresstx'), Richlist = require('../models/richlist'), Stats = require('../models/stats'), - settings = require('../lib/settings'); + settings = require('../lib/settings'), + async = require('async'); var mode = 'update'; var database = 'index'; var block_start = 1; var lockCreated = false; +var stopSync = false; + +// prevent stopping of the sync script to be able to gracefully shut down +process.on('SIGINT', () => { + console.log('Stopping sync process.. Please wait..'); + stopSync = true; +}); + +// prevent killing of the sync script to be able to gracefully shut down +process.on('SIGTERM', () => { + console.log('Stopping sync process.. Please wait..'); + stopSync = true; +}); // displays usage and exits function usage() { @@ -52,6 +66,124 @@ function exit(exitCode) { } } +// updates tx, address & richlist db's +function update_tx_db(coin, start, end, txes, timeout, check_only, cb) { + var complete = false; + var blocks_to_scan = []; + var task_limit_blocks = settings.sync.block_parallel_tasks; + var task_limit_txs = 1; + + // fix for invalid block height (skip genesis block as it should not have valid txs) + if (typeof start === 'undefined' || start < 1) + start = 1; + + if (task_limit_blocks < 1) + task_limit_blocks = 1; + + for (i = start; i < (end + 1); i++) + blocks_to_scan.push(i); + + async.eachLimit(blocks_to_scan, task_limit_blocks, function(block_height, next_block) { + if (!check_only && block_height % settings.sync.save_stats_after_sync_blocks === 0) { + Stats.updateOne({coin: coin}, { + last: block_height - 1, + txes: txes + }, function() {}); + } else if (check_only) { + console.log('Checking block ' + block_height + '...'); + } + + lib.get_blockhash(block_height, function(blockhash) { + if (blockhash) { + lib.get_block(blockhash, function(block) { + if (block) { + async.eachLimit(block.tx, task_limit_txs, function(txid, next_tx) { + Tx.findOne({txid: txid}, function(err, tx) { + if (tx) { + setTimeout( function() { + tx = null; + + // check if the script is stopping + if (stopSync) { + // stop the loop + next_tx({}); + } else + next_tx(); + }, timeout); + } else { + db.save_tx(txid, block_height, function(err, tx_has_vout) { + if (err) + console.log(err); + else + console.log('%s: %s', block_height, txid); + + if (tx_has_vout) + txes++; + + setTimeout( function() { + tx = null; + + // check if the script is stopping + if (stopSync) { + // stop the loop + next_tx({}); + } else + next_tx(); + }, timeout); + }); + } + }); + }, function() { + setTimeout( function() { + blockhash = null; + block = null; + + // check if the script is stopping + if (stopSync) { + // stop the loop + next_block({}); + } else + next_block(); + }, timeout); + }); + } else { + console.log('Block not found: %s', blockhash); + + setTimeout( function() { + // check if the script is stopping + if (stopSync) { + // stop the loop + next_block({}); + } else + next_block(); + }, timeout); + } + }); + } else { + setTimeout( function() { + // check if the script is stopping + if (stopSync) { + // stop the loop + next_block({}); + } else + next_block(); + }, timeout); + } + }); + }, function() { + // check if the script stopped prematurely + if (!stopSync) { + Stats.updateOne({coin: coin}, { + last: end, + txes: txes + }, function() { + return cb(); + }); + } else + return cb(); + }); +} + function update_heavy(coin, height, count, heavycoin_enabled, cb) { if (heavycoin_enabled == true) { db.update_heavy(coin, height, count, function() { @@ -94,8 +226,14 @@ function get_last_usd_price() { if (err == null) { // update markets_last_updated value db.update_last_updated_stats(settings.coin.name, { markets_last_updated: Math.floor(new Date() / 1000) }, function(cb) { - console.log('Market sync complete'); - exit(0); + // check if the script stopped prematurely + if (stopSync) { + console.log('Market sync was stopped prematurely'); + exit(1); + } else { + console.log('Market sync complete'); + exit(0); + } }); } else { // display error msg @@ -270,30 +408,36 @@ if (lib.is_locked([database]) == false) { check_show_sync_message(stats.count); console.log('Starting resync of blockchain data.. Please wait..'); - db.update_tx_db(settings.coin.name, block_start, stats.count, stats.txes, settings.sync.update_timeout, false, function() { - // update blockchain_last_updated value - db.update_last_updated_stats(settings.coin.name, { blockchain_last_updated: Math.floor(new Date() / 1000) }, function(cb) { - db.update_richlist('received', function() { - db.update_richlist('balance', function() { - // update richlist_last_updated value - db.update_last_updated_stats(settings.coin.name, { richlist_last_updated: Math.floor(new Date() / 1000) }, function(cb) { - db.get_stats(settings.coin.name, function(nstats) { - // check for and update heavycoin data if applicable - update_heavy(settings.coin.name, stats.count, 20, settings.blockchain_specific.heavycoin.enabled, function(heavy) { - // check for and update network history data if applicable - update_network_history(nstats.last, settings.network_history.enabled, function(network_hist) { - // always check for and remove the sync msg if exists - db.remove_sync_message(); - - console.log('Reindex complete (block: %s)', nstats.last); - exit(0); + update_tx_db(settings.coin.name, block_start, stats.count, stats.txes, settings.sync.update_timeout, false, function() { + // check if the script stopped prematurely + if (stopSync) { + console.log('Reindex was stopped prematurely'); + exit(1); + } else { + // update blockchain_last_updated value + db.update_last_updated_stats(settings.coin.name, { blockchain_last_updated: Math.floor(new Date() / 1000) }, function(cb) { + db.update_richlist('received', function() { + db.update_richlist('balance', function() { + // update richlist_last_updated value + db.update_last_updated_stats(settings.coin.name, { richlist_last_updated: Math.floor(new Date() / 1000) }, function(cb) { + db.get_stats(settings.coin.name, function(nstats) { + // check for and update heavycoin data if applicable + update_heavy(settings.coin.name, stats.count, 20, settings.blockchain_specific.heavycoin.enabled, function(heavy) { + // check for and update network history data if applicable + update_network_history(nstats.last, settings.network_history.enabled, function(network_hist) { + // always check for and remove the sync msg if exists + db.remove_sync_message(); + + console.log('Reindex complete (block: %s)', nstats.last); + exit(0); + }); }); }); }); }); }); }); - }); + } }); }); }); @@ -303,11 +447,17 @@ if (lib.is_locked([database]) == false) { } else if (mode == 'check') { console.log('Checking blocks.. Please wait..'); - db.update_tx_db(settings.coin.name, block_start, stats.count, stats.txes, settings.sync.check_timeout, true, function() { - db.get_stats(settings.coin.name, function(nstats) { - console.log('Block check complete (block: %s)', nstats.last); - exit(0); - }); + update_tx_db(settings.coin.name, block_start, stats.count, stats.txes, settings.sync.check_timeout, true, function() { + // check if the script stopped prematurely + if (stopSync) { + console.log('Block check was stopped prematurely'); + exit(1); + } else { + db.get_stats(settings.coin.name, function(nstats) { + console.log('Block check complete (block: %s)', nstats.last); + exit(0); + }); + } }); } else if (mode == 'update') { // Get the last synced block index value @@ -317,30 +467,36 @@ if (lib.is_locked([database]) == false) { // Check if the sync msg should be shown check_show_sync_message(count - last); - db.update_tx_db(settings.coin.name, last, count, stats.txes, settings.sync.update_timeout, false, function() { - // update blockchain_last_updated value - db.update_last_updated_stats(settings.coin.name, { blockchain_last_updated: Math.floor(new Date() / 1000) }, function(cb) { - db.update_richlist('received', function() { - db.update_richlist('balance', function() { - // update richlist_last_updated value - db.update_last_updated_stats(settings.coin.name, { richlist_last_updated: Math.floor(new Date() / 1000) }, function(cb) { - db.get_stats(settings.coin.name, function(nstats) { - // check for and update heavycoin data if applicable - update_heavy(settings.coin.name, stats.count, 20, settings.blockchain_specific.heavycoin.enabled, function(heavy) { - // check for and update network history data if applicable - update_network_history(nstats.last, settings.network_history.enabled, function(network_hist) { - // always check for and remove the sync msg if exists - db.remove_sync_message(); - - console.log('Block update complete (block: %s)', nstats.last); - exit(0); + update_tx_db(settings.coin.name, last, count, stats.txes, settings.sync.update_timeout, false, function() { + // check if the script stopped prematurely + if (stopSync) { + console.log('Block sync was stopped prematurely'); + exit(1); + } else { + // update blockchain_last_updated value + db.update_last_updated_stats(settings.coin.name, { blockchain_last_updated: Math.floor(new Date() / 1000) }, function(cb) { + db.update_richlist('received', function() { + db.update_richlist('balance', function() { + // update richlist_last_updated value + db.update_last_updated_stats(settings.coin.name, { richlist_last_updated: Math.floor(new Date() / 1000) }, function(cb) { + db.get_stats(settings.coin.name, function(nstats) { + // check for and update heavycoin data if applicable + update_heavy(settings.coin.name, stats.count, 20, settings.blockchain_specific.heavycoin.enabled, function(heavy) { + // check for and update network history data if applicable + update_network_history(nstats.last, settings.network_history.enabled, function(network_hist) { + // always check for and remove the sync msg if exists + db.remove_sync_message(); + + console.log('Block sync complete (block: %s)', nstats.last); + exit(0); + }); }); }); }); }); }); }); - }); + } }); } else if (mode == 'reindex-rich') { console.log('Check richlist'); @@ -458,6 +614,13 @@ if (lib.is_locked([database]) == false) { country_code: peer.country_code }, function() { console.log('Updated peer %s:%s [%s/%s]', address, port.toString(), (i + 1).toString(), body.length.toString()); + + // check if the script is stopping + if (stopSync) { + // stop the loop + loop.break(true); + } + loop.next(); }); }); @@ -485,6 +648,13 @@ if (lib.is_locked([database]) == false) { country_code: geo.country_code }, function() { console.log('Added new peer %s:%s [%s/%s]', address, port.toString(), (i + 1).toString(), body.length.toString()); + + // check if the script is stopping + if (stopSync) { + // stop the loop + loop.break(true); + } + loop.next(); }); } @@ -495,8 +665,14 @@ if (lib.is_locked([database]) == false) { }, function() { // update network_last_updated value db.update_last_updated_stats(settings.coin.name, { network_last_updated: Math.floor(new Date() / 1000) }, function(cb) { - console.log('Peer sync complete'); - exit(0); + // check if the script stopped prematurely + if (stopSync) { + console.log('Peer sync was stopped prematurely'); + exit(1); + } else { + console.log('Peer sync complete'); + exit(0); + } }); }); } else { @@ -521,9 +697,15 @@ if (lib.is_locked([database]) == false) { var i = loop.iteration(); db.save_masternode((isObject ? body[objectKeys[i]] : body[i]), function(success) { - if (success) + if (success) { + // check if the script is stopping + if (stopSync) { + // stop the loop + loop.break(true); + } + loop.next(); - else { + } else { console.log('Error: Cannot save masternode %s.', (isObject ? (body[objectKeys[i]].payee ? body[objectKeys[i]].payee : 'UNKNOWN') : (body[i].addr ? body[i].addr : 'UNKNOWN'))); exit(1); } @@ -531,8 +713,14 @@ if (lib.is_locked([database]) == false) { }, function() { db.remove_old_masternodes(function(cb) { db.update_last_updated_stats(settings.coin.name, { masternodes_last_updated: Math.floor(new Date() / 1000) }, function(cb) { - console.log('Masternode sync complete'); - exit(0); + // check if the script stopped prematurely + if (stopSync) { + console.log('Masternode sync was stopped prematurely'); + exit(1); + } else { + console.log('Masternode sync complete'); + exit(0); + } }); }); }); @@ -595,13 +783,13 @@ if (lib.is_locked([database]) == false) { console.log('%s[%s]: Market data updated successfully.', key, pair_key); complete++; - if (complete == total_pairs) + if (complete == total_pairs || stopSync) get_last_usd_price(); } else { console.log('%s[%s] Error: %s', key, pair_key, err); complete++; - if (complete == total_pairs) + if (complete == total_pairs || stopSync) get_last_usd_price(); } }); @@ -609,7 +797,7 @@ if (lib.is_locked([database]) == false) { } else { console.log('Error: Entry for %s[%s] does not exist in markets database.', key, pair_key); complete++; - if (complete == total_pairs) + if (complete == total_pairs || stopSync) get_last_usd_price(); } }); @@ -620,7 +808,7 @@ if (lib.is_locked([database]) == false) { console.log('%s market not installed', key); complete++; - if (complete == total_pairs) + if (complete == total_pairs || stopSync) get_last_usd_price(); } }