Fixed block sync where blocks come in all at once.
This commit is contained in:
Chris Kleeschulte
parent
7a8bae64a4
commit
4bd3a06edc
@ -121,6 +121,7 @@ AddressService.prototype.getAddressSummary = function(address, options, callback
|
||||
result.totalReceived = Unit.fromSatoshis(result.totalReceivedSat).toBTC();
|
||||
result.totalSent = Unit.fromSatoshis(result.totalSentSat).toBTC();
|
||||
result.unconfirmedBalance = result.unconfirmedBalanceSat;
|
||||
result.transactions = _.uniq(result.transactions);
|
||||
callback(null, result);
|
||||
});
|
||||
|
||||
@ -129,7 +130,12 @@ AddressService.prototype.getAddressSummary = function(address, options, callback
|
||||
|
||||
txStream._transform = function(chunk, enc, callback) {
|
||||
|
||||
// TODO: what if the address in tx more than once? such as an address sending from itself to itself
|
||||
// in the case where an address appears in both an input -and-
|
||||
// an output (sending money to one's self or using the sending
|
||||
// address as the change address (not recommended), we will get
|
||||
// duplicates. We don't want to look up the tx again.
|
||||
// Luckily, due to the way leveldb stores keys, we should get
|
||||
// txids out in lexigraphical order, so we can use an LRU here
|
||||
var key = self._encoding.decodeAddressIndexKey(chunk);
|
||||
|
||||
self._tx.getTransaction(key.txid, options, function(err, tx) {
|
||||
@ -390,77 +396,114 @@ AddressService.prototype._getAddressHistory = function(address, options, callbac
|
||||
|
||||
};
|
||||
|
||||
AddressService.prototype._removeBlock = function(block, callback) {
|
||||
var self = this;
|
||||
async.eachSeries(block.txs, function(tx, next) {
|
||||
self._removeTx(tx, block, next);
|
||||
}, callback);
|
||||
};
|
||||
|
||||
AddressService.prototype._removeTx = function(tx, block, callback) {
|
||||
var self = this;
|
||||
async.parallelLimit([
|
||||
function(next) {
|
||||
async.eachOfSeries(tx.inputs, function(input, indext, next) {
|
||||
self._removeInput(input, tx, block, index, next);
|
||||
}, next);
|
||||
},
|
||||
function(next) {
|
||||
async.eachOfSeries(tx.outputs, function(output, index, next) {
|
||||
self._removeOutput(output, tx, block, index, next);
|
||||
}, next);
|
||||
}
|
||||
], 4, callback);
|
||||
};
|
||||
|
||||
AddressService.prototype._removeInput = function(input, tx, block, index, callback) {
|
||||
|
||||
var self = this;
|
||||
var address = input.getAddress();
|
||||
var removalOps = [];
|
||||
|
||||
if (!address) {
|
||||
return callback();
|
||||
}
|
||||
|
||||
address.network = self._network;
|
||||
address = address.toString();
|
||||
|
||||
removalOps.push({
|
||||
type: 'del',
|
||||
key: self._encoding.encodeAddressIndexKey(address, block.height, tx.txid(), index, 1, block.ts)
|
||||
});
|
||||
|
||||
// look up prev output of this input and put it back in the set of utxos
|
||||
self._transaction.getTransaction(input.prevout.txid(), function(err, _tx) {
|
||||
|
||||
if (err) {
|
||||
return callback(err);
|
||||
}
|
||||
|
||||
assert(_tx, 'Missing prev tx to insert back into the utxo set when reorging address index.');
|
||||
|
||||
removalOps.push({
|
||||
type: 'put',
|
||||
key: self._encoding.encodeUtxoIndexKey(address, _tx.txid(), input.prevout.index),
|
||||
value: self._encoding.encodeUtxoIndexValue(
|
||||
_tx.height,
|
||||
_tx.__inputValues[input.prevout.index],
|
||||
_tx.timestamp, _tx.outputs[input.prevout.index].script.toRaw())
|
||||
});
|
||||
|
||||
callback(null, removalOps);
|
||||
|
||||
});
|
||||
};
|
||||
|
||||
AddressService.prototype._removeOutput = function(output, tx, block, index, callback) {
|
||||
|
||||
var self = this;
|
||||
var address = output.getAddress();
|
||||
var removalOps = [];
|
||||
|
||||
if (!address) {
|
||||
return callback();
|
||||
}
|
||||
|
||||
address.network = self._network;
|
||||
address = address.toString();
|
||||
|
||||
removalOps.push({
|
||||
type: 'del',
|
||||
key: self._encoding.encodeAddressIndexKey(address, block.height, tx.txid(), index, 0, block.ts)
|
||||
});
|
||||
|
||||
//remove the utxo for this output from the collection
|
||||
removalOps.push({
|
||||
type: 'del',
|
||||
key: self._encoding.encodeUtxoIndexKey(address, tx.txid(), index)
|
||||
});
|
||||
|
||||
callback(null, removalOps);
|
||||
};
|
||||
|
||||
AddressService.prototype.onReorg = function(args, callback) {
|
||||
|
||||
var self = this;
|
||||
|
||||
var oldBlockList = args[1];
|
||||
|
||||
var removalOps = [];
|
||||
// for every tx, remove the address index key for every input and output
|
||||
// for every output record we remove, we need to replace it with the output pointer located in the input that spent it
|
||||
// TODO: DRY self up!
|
||||
async.eachSeries(oldBlockList, function(block, next) {
|
||||
// for every input record, we need to find its previous output and put it back into the utxo collection
|
||||
async.eachSeries(oldBlockList, self._removeBlock.bind(self), function(err, ops) {
|
||||
|
||||
async.eachSeries(block.txs, function(tx, next) {
|
||||
if (err) {
|
||||
return callback(err);
|
||||
}
|
||||
|
||||
async.parallelLimit([
|
||||
|
||||
function(next) {
|
||||
async.eachOfSeries(tx.inputs, function(index, input, next) {
|
||||
|
||||
var address = input.getAddress();
|
||||
|
||||
if (!address) {
|
||||
return next();
|
||||
}
|
||||
|
||||
address.network = self._network;
|
||||
address = address.toString();
|
||||
|
||||
removalOps.push({
|
||||
type: 'del',
|
||||
key: self._encoding.encodeAddressIndexKey(address, block.height, tx.txid(), index, 1, block.ts)
|
||||
});
|
||||
// look up prev out here
|
||||
removalOps.push({
|
||||
type: 'put',
|
||||
key: self._encoding.encodeUtxoIndexKey(address, tx.txid(), input.prevout.index),
|
||||
value: self._encoding.encodeUtxoIndexValue(prevTx.height,
|
||||
|
||||
});
|
||||
},
|
||||
function(next) {
|
||||
|
||||
async.eachOfSeries(tx.outputs, function(index, output, next) {
|
||||
|
||||
var address = output.getAddress();
|
||||
|
||||
if (!address) {
|
||||
return next();
|
||||
}
|
||||
|
||||
address.network = self._network;
|
||||
address = address.toString();
|
||||
|
||||
removalOps.push({
|
||||
type: 'del',
|
||||
key: self._encoding.encodeAddressIndexKey(address, block.height, tx.txid(), index, 0, block.ts)
|
||||
});
|
||||
|
||||
});
|
||||
}
|
||||
], 4, function(err) {
|
||||
|
||||
if (err) {
|
||||
return callback(err);
|
||||
}
|
||||
|
||||
});
|
||||
});
|
||||
callback(null, _.compact(_.flatten(ops)));
|
||||
});
|
||||
|
||||
callback(null, removalOps);
|
||||
};
|
||||
|
||||
AddressService.prototype.onBlock = function(block, callback) {
|
||||
@ -574,7 +617,6 @@ AddressService.prototype._processTransaction = function(tx, opts) {
|
||||
outputOperations.length <= tx.outputs.length * 2,
|
||||
'Output operations count is not reflective of what should be possible.');
|
||||
|
||||
|
||||
var inputOperations = tx.inputs.map(function(input, index) {
|
||||
return self._processInput(tx, input, index, _opts);
|
||||
});
|
||||
|
||||
@ -551,6 +551,13 @@ BlockService.prototype._onBlock = function(block) {
|
||||
return;
|
||||
}
|
||||
|
||||
|
||||
// this will prevent the block service from hammering us with blocks
|
||||
// before we are ready
|
||||
// this will be turned to false once all the blocks have been processed
|
||||
// by the services
|
||||
this._header.blockServiceSyncing = true;
|
||||
|
||||
// this service must receive blocks in order
|
||||
var prevHash = bcoin.util.revHex(block.prevBlock);
|
||||
if (this._tip.hash !== prevHash) {
|
||||
|
||||
@ -28,7 +28,7 @@ var HeaderService = function(options) {
|
||||
this.GENESIS_HASH = constants.BITCOIN_GENESIS_HASH[this.node.network];
|
||||
this._lastHeader = null;
|
||||
this.blockServiceSyncing = true;
|
||||
|
||||
this._blockQueue = [];
|
||||
};
|
||||
|
||||
inherits(HeaderService, BaseService);
|
||||
@ -235,60 +235,145 @@ HeaderService.prototype.getPublishEvents = function() {
|
||||
|
||||
};
|
||||
|
||||
// TODO: if blocks come in too rapidly, there can be an erroreous reorg situation
|
||||
HeaderService.prototype._onBlock = function(block) {
|
||||
// If blocks arrive in rapid succession from the p2p network,
|
||||
// then this handler could not be finished saving the previous block info
|
||||
// this should be a rare case since blocks aren't usually mined milliseconds apart
|
||||
HeaderService.prototype._queueBlock = function(_block) {
|
||||
|
||||
var self = this;
|
||||
|
||||
var hash = block.rhash();
|
||||
// mode = syncing, this header already saved
|
||||
if (self.blockServiceSyncing) {
|
||||
|
||||
return self._broadcast(_block);
|
||||
|
||||
}
|
||||
|
||||
// mode = fully synced, live blocks arriving
|
||||
if (!self._blockProcessor) {
|
||||
self._blockProcessor = setInterval(self._processBlocks.bind(self), 1000);
|
||||
}
|
||||
|
||||
// we just need to queue the block, the interval processor will take it from there
|
||||
self._blockQueue.push(_block);
|
||||
|
||||
};
|
||||
|
||||
|
||||
// this gets fired on a timer to process new blocks from the queue
|
||||
// under normal circumstances, the queue will only have 1-2 items in it, probably less
|
||||
// we need this in case there is a deluge of blocks from a peer, asynchronously
|
||||
// we will also just process one block per call to keep things simple
|
||||
HeaderService.prototype._processBlocks = function() {
|
||||
|
||||
var self = this;
|
||||
|
||||
// if we have no blocks to process, exit
|
||||
var block = self._blockQueue.shift();
|
||||
|
||||
if (!block) {
|
||||
return;
|
||||
}
|
||||
|
||||
// if the block service is busy, then exit and wait for the next call
|
||||
if (self.blockServiceSyncing) {
|
||||
return;
|
||||
}
|
||||
|
||||
// clear the interval because this process might span intervals
|
||||
// we don't want to turn this back on until we know the block header has been persisted
|
||||
clearInterval(self._blockProcessor);
|
||||
|
||||
// let's see if this de-queued block is in order and not reorging the chain
|
||||
var prevHash = bcoin.util.revHex(block.prevBlock);
|
||||
var newBlock = prevHash === self._lastHeader.hash;
|
||||
|
||||
|
||||
// common case, blocks arriving in order and not reorging
|
||||
if (newBlock) {
|
||||
return self._onSyncBlock(block);
|
||||
}
|
||||
|
||||
// if we are here, we've got special things to do. This is almost always a reorg sitch
|
||||
self._detectAndHandleReorg(block);
|
||||
|
||||
};
|
||||
|
||||
HeaderService.prototype._onSyncBlock = function(block) {
|
||||
|
||||
var self = this;
|
||||
|
||||
|
||||
self._syncBlock(block, function(err) {
|
||||
|
||||
if (err) {
|
||||
log.error(err);
|
||||
self.node.stop();
|
||||
}
|
||||
|
||||
self._broadcast(block);
|
||||
|
||||
// at this point we know the block header has been persisted, but
|
||||
// not necessarily that the services are done syncing
|
||||
self._blockProcessor = setInterval(self._processBlocks.bind(self), 1000);
|
||||
|
||||
});
|
||||
};
|
||||
|
||||
HeaderService.prototype._detectAndHandleReorg = function(block) {
|
||||
|
||||
var self = this;
|
||||
|
||||
// this is the rare case that a block comes to us out of order or is a reorg'ed block
|
||||
self._detectReorg(block, function(err, reorg) {
|
||||
|
||||
if (err) {
|
||||
log.error(err);
|
||||
return self.node.stop();
|
||||
}
|
||||
|
||||
// if not reorg, then this block is out of order and we can't do anything with it
|
||||
// TODO: does this ever happen? Can this ever happen?
|
||||
if (!reorg) {
|
||||
return log.warn('Block: ' + block.rhash() + ' arrived out of order, skipping.');
|
||||
}
|
||||
|
||||
var header = block.toHeaders().toJSON();
|
||||
header.timestamp = header.ts;
|
||||
header.prevHash = header.prevBlock;
|
||||
|
||||
self._handleReorg(block, header, function(err) {
|
||||
|
||||
if (err) {
|
||||
log.error(err);
|
||||
return self.node.stop();
|
||||
}
|
||||
|
||||
self._onSyncBlock(block);
|
||||
|
||||
});
|
||||
|
||||
});
|
||||
};
|
||||
|
||||
HeaderService.prototype._syncBlock = function(block, callback) {
|
||||
|
||||
var self = this;
|
||||
|
||||
var header = block.toHeaders().toJSON();
|
||||
header.timestamp = header.ts;
|
||||
header.prevHash = header.prevBlock;
|
||||
|
||||
if (newBlock) {
|
||||
log.debug('Header Service: new block: ' + block.rhash());
|
||||
|
||||
log.debug('Header Service: new block: ' + hash);
|
||||
self._saveHeaders(self._onHeader(header));
|
||||
self._saveHeaders(self._onHeader(header), function(err) {
|
||||
|
||||
}
|
||||
if (err) {
|
||||
return callback(err);
|
||||
}
|
||||
|
||||
// this is the rare case that a block comes to us out of order or is a reorg'ed block
|
||||
// in almost all cases, this will be a reorg
|
||||
if (!newBlock && !self.blockServiceSyncing) {
|
||||
|
||||
return self._detectReorg(block, function(err, reorg) {
|
||||
|
||||
if (err) {
|
||||
log.error(err);
|
||||
self.node.stop();
|
||||
return;
|
||||
}
|
||||
|
||||
if (reorg) {
|
||||
return self._handleReorg(block, header, function(err) {
|
||||
|
||||
if (err) {
|
||||
log.error(err);
|
||||
self.node.stop();
|
||||
return;
|
||||
}
|
||||
|
||||
self._saveHeaders(self._onHeader(header));
|
||||
|
||||
}); // this sets the last header
|
||||
}
|
||||
|
||||
self._broadcast(block);
|
||||
|
||||
});
|
||||
|
||||
}
|
||||
|
||||
setImmediate(function() {
|
||||
self._broadcast(block);
|
||||
self._onHeadersSave();
|
||||
callback();
|
||||
});
|
||||
|
||||
};
|
||||
@ -329,6 +414,7 @@ HeaderService.prototype._onHeader = function(header) {
|
||||
|
||||
HeaderService.prototype._onHeaders = function(headers) {
|
||||
|
||||
var self = this;
|
||||
log.debug('Header Service: Received: ' + headers.length + ' header(s).');
|
||||
|
||||
var dbOps = [];
|
||||
@ -339,19 +425,25 @@ HeaderService.prototype._onHeaders = function(headers) {
|
||||
|
||||
header = header.toObject();
|
||||
|
||||
var ops = this._onHeader(header);
|
||||
var ops = self._onHeader(header);
|
||||
|
||||
dbOps = dbOps.concat(ops);
|
||||
|
||||
this._tip.height = header.height;
|
||||
this._tip.hash = header.hash;
|
||||
self._tip.height = header.height;
|
||||
self._tip.hash = header.hash;
|
||||
}
|
||||
|
||||
this._saveHeaders(dbOps);
|
||||
self._saveHeaders(dbOps, function(err) {
|
||||
if (err) {
|
||||
log.error(err);
|
||||
return self.node.stop();
|
||||
}
|
||||
self._onHeadersSave();
|
||||
});
|
||||
|
||||
};
|
||||
|
||||
HeaderService.prototype._saveHeaders = function(dbOps) {
|
||||
HeaderService.prototype._saveHeaders = function(dbOps, callback) {
|
||||
|
||||
var tipOps = utils.encodeTip(this._tip, this.name);
|
||||
|
||||
@ -361,7 +453,7 @@ HeaderService.prototype._saveHeaders = function(dbOps) {
|
||||
value: tipOps.value
|
||||
});
|
||||
|
||||
this._db.batch(dbOps, this._onHeadersSave.bind(this));
|
||||
this._db.batch(dbOps, callback);
|
||||
};
|
||||
|
||||
HeaderService.prototype._onHeadersSave = function(err) {
|
||||
@ -405,6 +497,7 @@ HeaderService.prototype._onHeadersSave = function(err) {
|
||||
|
||||
log.debug('Header Service: emitting headers to block service.');
|
||||
|
||||
|
||||
self.emit('headers');
|
||||
});
|
||||
|
||||
@ -418,7 +511,7 @@ HeaderService.prototype._startBlockSubscription = function() {
|
||||
|
||||
this._subscribedBlock = true;
|
||||
|
||||
this._bus.on('p2p/block', this._onBlock.bind(this));
|
||||
this._bus.on('p2p/block', this._queueBlock.bind(this));
|
||||
this._bus.subscribe('p2p/block');
|
||||
|
||||
};
|
||||
|
||||
2
test/data/real-data-blocks.json
Normal file
2
test/data/real-data-blocks.json
Normal file
File diff suppressed because one or more lines are too long
@ -12,6 +12,7 @@ var bcoin = require('bcoin');
|
||||
describe('Address Service', function() {
|
||||
|
||||
var tx = Tx.fromRaw( '0100000004de9b4bb17f627096a9ee0b4528e4eae17df5b5c69edc29704c2e84a7371db29f010000006b483045022100f5b1a0d33b7be291c3953c25f8ae39d98601aa7099a8674daf638a08b86c7173022006ce372da5ad088a1cc6e5c49c2760a1b6f085eb1b51b502211b6bc9508661f9012102ec5e3731e54475dd2902326f43602a03ae3d62753324139163f81f20e787514cffffffff7a1d4e5fc2b8177ec738cd723a16cf2bf493791e55573445fc0df630fe5e2d64010000006b483045022100cf97f6cb8f126703e9768545dfb20ffb10ba78ae3d101aa46775f5a239b075fc02203150c4a89a11eaf5e404f4f96b62efa4455e9525765a025525c7105a7e47b6db012102c01e11b1d331f999bbdb83e8831de503cd52a01e3834a95ccafd615c67703d77ffffffff9e52447116415ca0d0567418a1a4ef8f27be3ff5a96bf87c922f3723d7db5d7c000000006b483045022100f6c117e536701be41a6b0b544d7c3b1091301e4e64a6265b6eb167b15d16959d022076916de4b115e700964194ce36a24cb9105f86482f4abbc63110c3f537cd5770012102ddf84cc7bee2d6a82ac09628a8ad4a26cd449fc528b81e7e6cc615707b8169dfffffffff5815d9750eb3572e30d6fd9df7afb4dbd76e042f3aa4988ac763b3fdf8397f80010000006a473044022028f4402b736066d93d2a32b28ccd3b7a21d84bb58fcd07fe392a611db94cdec5022018902ee0bf2c3c840c1b81ead4e6c87c88c48b2005bf5eea796464e561a620a8012102b6cdd1a6cd129ef796faeedb0b840fcd0ca00c57e16e38e46ee7028d59812ae7ffffffff0220a10700000000001976a914c342bcd1a7784d9842f7386b8b3b8a3d4171a06e88ac59611100000000001976a91449f8c749a9960dc29b5cbe7d2397cea7d26611bb88ac00000000', 'hex');
|
||||
var realBlocks = require('../../data/real-data-blocks.json');
|
||||
var blocks = require('../../data/blocks.json');
|
||||
var addressService;
|
||||
var sandbox;
|
||||
@ -225,17 +226,33 @@ describe('Address Service', function() {
|
||||
|
||||
describe('#onReorg', function() {
|
||||
|
||||
it('should reorg', function(done ) {
|
||||
it('should reorg when there is nothing to reorg', function(done ) {
|
||||
|
||||
var commonAncestorHeader = bcoin.block.fromRaw(blocks[5], 'hex').toHeaders().toJSON();
|
||||
var oldBlocks = [bcoin.block.fromRaw(blocks[6], 'hex')];
|
||||
|
||||
addressService.onReorg([commonAncestorHeader, oldBlocks], function(err, ops) {
|
||||
|
||||
expect(ops.length).to.equal(1);
|
||||
expect(ops[0].type).to.equal('del');
|
||||
|
||||
expect(ops.length).to.equal(0);
|
||||
done();
|
||||
|
||||
});
|
||||
|
||||
});
|
||||
|
||||
it.skip('should reorg when there is something to reorg', function(done ) {
|
||||
|
||||
var commonAncestorHeader = bcoin.block.fromRaw(realBlocks.block0, 'hex').toHeaders().toJSON();
|
||||
var oldBlocks = [bcoin.block.fromRaw(realBlocks.block1, 'hex')];
|
||||
|
||||
|
||||
var getTransaction = sandbox.stub().callsArgWith(1, null, tx);
|
||||
addressService._transaction = { getTransaction: getTransaction };
|
||||
addressService.onReorg([commonAncestorHeader, oldBlocks], function(err, ops) {
|
||||
|
||||
expect(ops.length).to.equal(0);
|
||||
done();
|
||||
|
||||
});
|
||||
|
||||
});
|
||||
|
||||
@ -101,6 +101,7 @@ describe('Block Service', function() {
|
||||
it('should process blocks', function() {
|
||||
var processBlock = sandbox.stub(blockService, '_processBlock');
|
||||
blockService._tip = { hash: block1.rhash(), height: 1 };
|
||||
blockService._header = { blockServiceSyncing: false };
|
||||
blockService._onBlock(block2);
|
||||
expect(processBlock.calledOnce).to.be.true;
|
||||
});
|
||||
@ -108,6 +109,7 @@ describe('Block Service', function() {
|
||||
it('should not process blocks', function() {
|
||||
var processBlock = sandbox.stub(blockService, '_processBlock');
|
||||
blockService._tip = { hash: block2.rhash(), height: 1 };
|
||||
blockService._header = { blockServiceSyncing: false };
|
||||
blockService._onBlock(block1);
|
||||
expect(processBlock.calledOnce).to.be.false;
|
||||
});
|
||||
|
||||
Loading…
Reference in New Issue
Block a user