From 89ef28f0b7ac140b0af525816be07cb08f0adb0d Mon Sep 17 00:00:00 2001 From: Braydon Fuller Date: Fri, 30 Oct 2015 16:57:01 -0400 Subject: [PATCH] Optimize mempool address index memory footprint - Adds default to store a large portion of the mempool index in leveldb - Includes an option to use memdown to have the mempool index in-memory --- integration/regtest-node.js | 13 +- lib/services/address/index.js | 470 +++++++++++++++++++--------- test/services/address/index.unit.js | 409 +++++++++++++++++++----- 3 files changed, 667 insertions(+), 225 deletions(-) diff --git a/integration/regtest-node.js b/integration/regtest-node.js index 4e39bb978..8b55db7f6 100644 --- a/integration/regtest-node.js +++ b/integration/regtest-node.js @@ -726,12 +726,15 @@ describe('Node Functionality', function() { node.services.bitcoind.sendTransaction(tx.serialize()); setImmediate(function() { - var length = node.services.address.mempoolOutputIndex[address].length; - length.should.equal(1); - should.exist(node.services.address.mempoolOutputIndex[address]); - done(); + var hashBuffer = bitcore.Address(address).hashBuffer; + node.services.address._getOutputsMempool(address, hashBuffer, function(err, outs) { + if (err) { + throw err; + } + outs.length.should.equal(1); + done(); + }); }); - }); }); diff --git a/lib/services/address/index.js b/lib/services/address/index.js index 3cc625102..fa3152174 100644 --- a/lib/services/address/index.js +++ b/lib/services/address/index.js @@ -1,13 +1,18 @@ 'use strict'; +var fs = require('fs'); var BaseService = require('../../service'); var inherits = require('util').inherits; var async = require('async'); +var mkdirp = require('mkdirp'); var index = require('../../'); var log = index.log; var errors = index.errors; var bitcore = require('bitcore-lib'); +var Networks = bitcore.Networks; var levelup = require('levelup'); +var leveldown = require('leveldown'); +var memdown = require('memdown'); var $ = bitcore.util.preconditions; var _ = bitcore.deps._; var Hash = bitcore.crypto.Hash; @@ -36,10 +41,14 @@ var AddressService = function(options) { this.node.services.bitcoind.on('tx', this.transactionHandler.bind(this)); this.node.services.bitcoind.on('txleave', this.transactionLeaveHandler.bind(this)); - this.mempoolOutputIndex = {}; - this.mempoolInputIndex = {}; - this.mempoolSpentIndex = {}; - + this._setMempoolIndexPath(); + if (options.mempoolMemoryIndex) { + this.levelupStore = memdown; + } else { + this.levelupStore = leveldown; + } + this.mempoolIndex = null; // Used for larger mempool indexes + this.mempoolSpentIndex = {}; // Used for small quick synchronous lookups }; inherits(AddressService, BaseService); @@ -55,9 +64,73 @@ AddressService.PREFIXES = { SPENTSMAP: new Buffer('05', 'hex') // Get the input that spends an output }; +AddressService.MEMPREFIXES = { + OUTPUTS: new Buffer('01', 'hex'), // Query mempool outputs by address + SPENTS: new Buffer('02', 'hex'), // Query mempool inputs by address + SPENTSMAP: new Buffer('03', 'hex') // Query mempool for the input that spends an output +}; + AddressService.SPACER_MIN = new Buffer('00', 'hex'); AddressService.SPACER_MAX = new Buffer('ff', 'hex'); +AddressService.prototype.start = function(callback) { + var self = this; + + async.series([ + function(next) { + // Flush any existing mempool index + if (fs.existsSync(self.mempoolIndexPath)) { + leveldown.destroy(self.mempoolIndexPath, next); + } else { + setImmediate(next); + } + }, + function(next) { + if (!fs.existsSync(self.mempoolIndexPath)) { + mkdirp(self.mempoolIndexPath, next); + } else { + setImmediate(next); + } + }, + function(next) { + self.mempoolIndex = levelup( + self.mempoolIndexPath, + { + db: self.levelupStore, + keyEncoding: 'binary', + valueEncoding: 'binary', + fillCache: false + }, + next + ); + } + ], callback); + +}; + +AddressService.prototype.stop = function(callback) { + // TODO Keep track of ongoing db requests before shutting down + this.mempoolIndex.close(callback); +}; + +/** + * This function will set `this.dataPath` based on `this.node.network`. + * @private + */ +AddressService.prototype._setMempoolIndexPath = function() { + $.checkState(this.node.datadir, 'Node is expected to have a "datadir" property'); + var regtest = Networks.get('regtest'); + if (this.node.network === Networks.livenet) { + this.mempoolIndexPath = this.node.datadir + '/bitcore-addressmempool.db'; + } else if (this.node.network === Networks.testnet) { + this.mempoolIndexPath = this.node.datadir + '/testnet3/bitcore-addressmempool.db'; + } else if (this.node.network === regtest) { + this.mempoolIndexPath = this.node.datadir + '/regtest/bitcore-addressmempool.db'; + } else { + throw new Error('Unknown network: ' + this.network); + } +}; + /** * Called by the Node to get the available API methods for this service, * that can be exposed over the JSON-RPC interface. @@ -139,7 +212,7 @@ AddressService.prototype.transactionOutputHandler = function(messages, tx, outpu */ AddressService.prototype.transactionLeaveHandler = function(txInfo) { var tx = bitcore.Transaction().fromBuffer(txInfo.buffer); - this.removeMempoolIndex(tx); + this.updateMempoolIndex(tx, false); }; /** @@ -166,7 +239,7 @@ AddressService.prototype.transactionHandler = function(txInfo) { // Update mempool index if (txInfo.mempool) { - this.updateMempoolIndex(tx); + this.updateMempoolIndex(tx, true); } for (var key in messages) { @@ -175,94 +248,21 @@ AddressService.prototype.transactionHandler = function(txInfo) { } }; -AddressService.prototype.removeMempoolIndex = function(tx) { - - var txid = tx.hash; - - var outputLength = tx.outputs.length; - for (var outputIndex = 0; outputIndex < outputLength; outputIndex++) { - var output = tx.outputs[outputIndex]; - if (!output.script) { - continue; - } - var addressInfo = this._extractAddressInfoFromScript(output.script); - if (!addressInfo) { - continue; - } - - var addressStr = bitcore.Address({ - hashBuffer: addressInfo.hashBuffer, - type: addressInfo.addressType, - network: this.node.network - }).toString(); - - // Remove from the mempool output index - if (this.mempoolOutputIndex[addressStr]) { - var txs = this.mempoolOutputIndex[addressStr]; - for (var t = 0; t < txs.length; t++) { - if (txs[t].txid === txid && txs[t].outputIndex === outputIndex) { - txs.splice(t, 1); - } - } - if (txs.length === 0) { - delete this.mempoolOutputIndex[addressStr]; - } - } - } - var inputLength = tx.inputs.length; - for (var inputIndex = 0; inputIndex < inputLength; inputIndex++) { - - var input = tx.inputs[inputIndex]; - - // Remove from the mempool spent index - var spentIndexKey = [input.prevTxId.toString('hex'), input.outputIndex].join('-'); - if (this.mempoolSpentIndex[spentIndexKey]) { - delete this.mempoolSpentIndex[spentIndexKey]; - } - - var address = input.script.toAddress(this.node.network); - if (!address) { - continue; - } - var inputAddressStr = address.toString(); - - // Remove from the mempool input index - if (this.mempoolInputIndex[inputAddressStr]) { - var inputTxs = this.mempoolInputIndex[inputAddressStr]; - for (var x = 0; x < inputTxs.length; x++) { - if (inputTxs[x].txid === txid && inputTxs[x].inputIndex === inputIndex) { - inputTxs.splice(x, 1); - } - } - if (inputTxs.length === 0) { - delete this.mempoolInputIndex[inputAddressStr]; - } - } - } -}; - /** * This function will update the mempool address index with the necessary - * information for further lookups. There are three indexes: - * - * mempoolOutputIndex, an object keyed by base58check encoded addresses with values: - * txid - A hex string of the transaction hash - * outputIndex - A number of the corresponding output - * satoshis - Total number of satoshis - * script - The script as a hex string - * - * mempoolInputIndex, an object keyed by base58check encoded addresses with values: - * txid - A hex string of the transaction hash - * inputIndex - A number of the corresponding input - * - * mempoolSpentIndex, an object keyed by - with (buffer) values: - * inputTxId - A 32 byte buffer of the input txid - * inputIndex - 4 bytes stored as UInt32BE - * + * information for further lookups. * @param {Transaction} - An instance of a Bitcore Transaction + * @param {Boolean} - Add/remove from the index */ -AddressService.prototype.updateMempoolIndex = function(tx) { - /* jshint maxstatements: 30 */ +AddressService.prototype.updateMempoolIndex = function(tx, add) { + /* jshint maxstatements: 100 */ + + var operations = []; + + var action = 'put'; + if (!add) { + action = 'del'; + } var txid = tx.hash; var txidBuffer = new Buffer(txid, 'hex'); @@ -278,21 +278,23 @@ AddressService.prototype.updateMempoolIndex = function(tx) { continue; } - var addressStr = bitcore.Address({ - hashBuffer: addressInfo.hashBuffer, - type: addressInfo.addressType, - network: this.node.network - }).toString(); + // Update output index + var outputIndexBuffer = new Buffer(4); + outputIndexBuffer.writeUInt32BE(outputIndex); - if (!this.mempoolOutputIndex[addressStr]) { - this.mempoolOutputIndex[addressStr] = []; - } + var outKey = Buffer.concat([ + AddressService.MEMPREFIXES.OUTPUTS, + addressInfo.hashBuffer, + txidBuffer, + outputIndexBuffer + ]); - this.mempoolOutputIndex[addressStr].push({ - txid: txid, - outputIndex: outputIndex, - satoshis: output.satoshis, - script: output._scriptBuffer.toString('hex') //TODO use a buffer + var outValue = this._encodeOutputValue(output.satoshis, output._scriptBuffer); + + operations.push({ + type: action, + key: outKey, + value: outValue }); } @@ -301,31 +303,71 @@ AddressService.prototype.updateMempoolIndex = function(tx) { var input = tx.inputs[inputIndex]; - // Update spent index - var spentIndexKey = [input.prevTxId.toString('hex'), input.outputIndex].join('-'); + var inputOutputIndexBuffer = new Buffer(4); + inputOutputIndexBuffer.writeUInt32BE(input.outputIndex); + + // Add an additional small spent index for fast synchronous lookups + var spentIndexSyncKey = this._encodeSpentIndexSyncKey( + input.prevTxId, + input.outputIndex + ); + if (add) { + this.mempoolSpentIndex[spentIndexSyncKey] = true; + } else { + delete this.mempoolSpentIndex[spentIndexSyncKey]; + } + // Add a more detailed spent index with values + var spentIndexKey = Buffer.concat([ + AddressService.MEMPREFIXES.SPENTSMAP, + input.prevTxId, + inputOutputIndexBuffer + ]); var inputIndexBuffer = new Buffer(4); inputIndexBuffer.writeUInt32BE(inputIndex); var inputIndexValue = Buffer.concat([ txidBuffer, inputIndexBuffer ]); - this.mempoolSpentIndex[spentIndexKey] = inputIndexValue; + operations.push({ + type: action, + key: spentIndexKey, + value: inputIndexValue + }); - var address = input.script.toAddress(this.node.network); - if (!address) { + // Update input index + var inputHashBuffer; + if (input.script.isPublicKeyHashIn()) { + inputHashBuffer = Hash.sha256ripemd160(input.script.chunks[1].buf); + } else if (input.script.isScriptHashIn()) { + inputHashBuffer = Hash.sha256ripemd160(input.script.chunks[input.script.chunks.length - 1].buf); + } else { continue; } - var inputAddressStr = address.toString(); - if (!this.mempoolInputIndex[inputAddressStr]) { - this.mempoolInputIndex[inputAddressStr] = []; - } - this.mempoolInputIndex[inputAddressStr].push({ - txid: tx.hash, // TODO use buffer - inputIndex: inputIndex + var inputKey = Buffer.concat([ + AddressService.MEMPREFIXES.SPENTS, + inputHashBuffer, + input.prevTxId, + inputOutputIndexBuffer + ]); + var inputValue = Buffer.concat([ + txidBuffer, + inputIndexBuffer + ]); + operations.push({ + type: action, + key: inputKey, + value: inputValue }); + } + this.mempoolIndex.batch(operations, function(err) { + if (err) { + return log.error(err); + } + }); + }; /** @@ -489,6 +531,16 @@ AddressService.prototype.blockHandler = function(block, addOutput, callback) { }); }; +AddressService.prototype._encodeSpentIndexSyncKey = function(txidBuffer, outputIndex) { + var outputIndexBuffer = new Buffer(4); + outputIndexBuffer.writeUInt32BE(outputIndex); + var key = Buffer.concat([ + txidBuffer, + outputIndexBuffer + ]); + return key.toString('binary'); +}; + AddressService.prototype._encodeOutputKey = function(hashBuffer, height, txidBuffer, outputIndex) { var heightBuffer = new Buffer(4); heightBuffer.writeUInt32BE(height); @@ -801,15 +853,9 @@ AddressService.prototype.getInputForOutput = function(txid, outputIndex, options txidBuffer = new Buffer(txid, 'hex'); } if (options.queryMempool) { - var spentIndexKey = [txid.toString('hex'), outputIndex].join('-'); - if (this.mempoolSpentIndex[spentIndexKey]) { - var mempoolValue = this.mempoolSpentIndex[spentIndexKey]; - var inputTxId = mempoolValue.slice(0, 32); - var inputIndex = mempoolValue.readUInt32BE(32); - return callback(null, { - inputTxId: inputTxId.toString('hex'), - inputIndex: inputIndex - }); + var spentIndexSyncKey = this._encodeSpentIndexSyncKey(txidBuffer, outputIndex); + if (this.mempoolSpentIndex[spentIndexSyncKey]) { + return this._getSpentMempool(txidBuffer, outputIndex, callback); } } var key = this._encodeInputKeyMap(txidBuffer, outputIndex); @@ -920,26 +966,97 @@ AddressService.prototype.getInputs = function(addressStr, options, callback) { } if(options.queryMempool) { - var mempoolInputs = self.mempoolInputIndex[addressStr]; - if (mempoolInputs) { - for(var i = 0; i < mempoolInputs.length; i++) { - var newInput = _.clone(mempoolInputs[i]); - newInput.address = addressStr; - newInput.height = -1; - newInput.confirmations = 0; - inputs.push(newInput); + self._getInputsMempool(addressStr, hashBuffer, function(err, mempoolInputs) { + if (err) { + return callback(err); } - } + inputs = inputs.concat(mempoolInputs); + callback(null, inputs); + }); + } else { + callback(null, inputs); } - callback(null, inputs); - }); return stream; }; +AddressService.prototype._getInputsMempool = function(addressStr, hashBuffer, callback) { + var self = this; + var mempoolInputs = []; + + var stream = self.mempoolIndex.createReadStream({ + gte: Buffer.concat([ + AddressService.MEMPREFIXES.SPENTS, + hashBuffer, + AddressService.SPACER_MIN + ]), + lte: Buffer.concat([ + AddressService.MEMPREFIXES.SPENTS, + hashBuffer, + AddressService.SPACER_MAX + ]), + valueEncoding: 'binary', + keyEncoding: 'binary' + }); + + stream.on('data', function(data) { + var txid = data.value.slice(0, 32); + var inputIndex = data.value.readUInt32BE(32); + var output = { + address: addressStr, + txid: txid.toString('hex'), //TODO use a buffer + inputIndex: inputIndex, + height: -1, + confirmations: 0 + }; + mempoolInputs.push(output); + }); + + var error; + + stream.on('error', function(streamError) { + if (streamError) { + error = streamError; + } + }); + + stream.on('close', function() { + if (error) { + return callback(error); + } + callback(null, mempoolInputs); + }); + +}; + +AddressService.prototype._getSpentMempool = function(txidBuffer, outputIndex, callback) { + var outputIndexBuffer = new Buffer(4); + outputIndexBuffer.writeUInt32BE(outputIndex); + var spentIndexKey = Buffer.concat([ + AddressService.MEMPREFIXES.SPENTSMAP, + txidBuffer, + outputIndexBuffer + ]); + + this.mempoolIndex.get( + spentIndexKey, + function(err, mempoolValue) { + if (err) { + return callback(err); + } + var inputTxId = mempoolValue.slice(0, 32); + var inputIndex = mempoolValue.readUInt32BE(32); + callback(null, { + inputTxId: inputTxId.toString('hex'), + inputIndex: inputIndex + }); + } + ); +}; + /** * Will give outputs for an address as an object with: * address - The base58check encoded address @@ -1033,24 +1150,75 @@ AddressService.prototype.getOutputs = function(addressStr, options, callback) { } if(options.queryMempool) { - var mempoolOutputs = self.mempoolOutputIndex[addressStr]; - if (mempoolOutputs) { - for(var i = 0; i < mempoolOutputs.length; i++) { - var newOutput = _.clone(mempoolOutputs[i]); - newOutput.address = addressStr; - newOutput.height = -1; - newOutput.confirmations = 0; - outputs.push(newOutput); + self._getOutputsMempool(addressStr, hashBuffer, function(err, mempoolOutputs) { + if (err) { + return callback(err); } - } + outputs = outputs.concat(mempoolOutputs); + callback(null, outputs); + }); + } else { + callback(null, outputs); } - callback(null, outputs); }); return stream; }; +AddressService.prototype._getOutputsMempool = function(addressStr, hashBuffer, callback) { + var self = this; + var mempoolOutputs = []; + + var stream = self.mempoolIndex.createReadStream({ + gte: Buffer.concat([ + AddressService.MEMPREFIXES.OUTPUTS, + hashBuffer, + AddressService.SPACER_MIN + ]), + lte: Buffer.concat([ + AddressService.MEMPREFIXES.OUTPUTS, + hashBuffer, + AddressService.SPACER_MAX + ]), + valueEncoding: 'binary', + keyEncoding: 'binary' + }); + + stream.on('data', function(data) { + // Format of data: prefix: 1, hashBuffer: 20, txid: 32, outputIndex: 4 + var txid = data.key.slice(21, 53); + var outputIndex = data.key.readUInt32BE(53); + var value = self._decodeOutputValue(data.value); + var output = { + address: addressStr, + txid: txid.toString('hex'), //TODO use a buffer + outputIndex: outputIndex, + height: -1, + satoshis: value.satoshis, + script: value.scriptBuffer.toString('hex'), //TODO use a buffer + confirmations: 0 + }; + mempoolOutputs.push(output); + }); + + var error; + + stream.on('error', function(streamError) { + if (streamError) { + error = streamError; + } + }); + + stream.on('close', function() { + if (error) { + return callback(error); + } + callback(null, mempoolOutputs); + }); + +}; + /** * Will give unspent outputs for an address or an array of addresses. * @param {Array|String} addresses - An array of addresses @@ -1141,8 +1309,8 @@ AddressService.prototype.isSpent = function(output, options, callback) { var txid = output.prevTxId ? output.prevTxId.toString('hex') : output.txid; var spent = self.node.services.bitcoind.isSpent(txid, output.outputIndex); if (!spent && queryMempool) { - var spentIndexKey = [txid, output.outputIndex].join('-'); - spent = self.mempoolSpentIndex[spentIndexKey] ? true : false; + var spentIndexSyncKey = this._encodeSpentIndexSyncKey(output.prevTxId, output.outputIndex); + spent = self.mempoolSpentIndex[spentIndexSyncKey] ? true : false; } setImmediate(function() { // TODO error should be the first argument? @@ -1150,6 +1318,7 @@ AddressService.prototype.isSpent = function(output, options, callback) { }); }; + /** * This will give the history for many addresses limited by a range of block heights (to limit * the database lookup times) and/or paginated to limit the results length. @@ -1248,8 +1417,11 @@ AddressService.prototype.getAddressSummary = function(address, options, callback for(var i = 0; i < outputs.length; i++) { // Bitcoind's isSpent only works for confirmed transactions var spentDB = self.node.services.bitcoind.isSpent(outputs[i].txid, outputs[i].outputIndex); - var spentIndexKey = [outputs[i].txid, outputs[i].outputIndex].join('-'); - var spentMempool = self.mempoolSpentIndex[spentIndexKey]; + var spentIndexSyncKey = self._encodeSpentIndexSyncKey( + new Buffer(outputs[i].txid, 'hex'), // TODO: get buffer directly + outputs[i].outputIndex + ); + var spentMempool = self.mempoolSpentIndex[spentIndexSyncKey]; txids.push(outputs[i]); diff --git a/test/services/address/index.unit.js b/test/services/address/index.unit.js index a63f66471..89fd0245d 100644 --- a/test/services/address/index.unit.js +++ b/test/services/address/index.unit.js @@ -19,6 +19,8 @@ var mockdb = { }; var mocknode = { + network: Networks.testnet, + datadir: 'testdir', db: mockdb, services: { bitcoind: { @@ -31,7 +33,10 @@ describe('Address Service', function() { var txBuf = new Buffer(txData[0], 'hex'); describe('#getAPIMethods', function() { it('should return the correct methods', function() { - var am = new AddressService({node: mocknode}); + var am = new AddressService({ + mempoolMemoryIndex: true, + node: mocknode + }); var methods = am.getAPIMethods(); methods.length.should.equal(7); }); @@ -39,7 +44,10 @@ describe('Address Service', function() { describe('#getPublishEvents', function() { it('will return an array of publish event objects', function() { - var am = new AddressService({node: mocknode}); + var am = new AddressService({ + mempoolMemoryIndex: true, + node: mocknode + }); am.subscribe = sinon.spy(); am.unsubscribe = sinon.spy(); var events = am.getPublishEvents(); @@ -75,7 +83,10 @@ describe('Address Service', function() { it('create a message for an address', function() { var txBuf = new Buffer('01000000010000000000000000000000000000000000000000000000000000000000000000ffffffff0704ffff001d0104ffffffff0100f2052a0100000043410496b538e853519c726a2c91e61ec11600ae1390813a627c66fb8be7947be63c52da7589379515d4e0a604f8141781e62294721166bf621e73a82cbf2342c858eeac00000000', 'hex'); var tx = bitcore.Transaction().fromBuffer(txBuf); - var am = new AddressService({node: mocknode}); + var am = new AddressService({ + mempoolMemoryIndex: true, + node: mocknode + }); am.node.network = Networks.livenet; var address = '12c6DSiU4Rq3P4ZxziKxzrL5LmMBrzjrJX'; var hashHex = bitcore.Address(address).hashBuffer.toString('hex'); @@ -94,7 +105,10 @@ describe('Address Service', function() { describe('#transactionHandler', function() { it('will pass outputs to transactionOutputHandler and call transactionEventHandler and balanceEventHandler', function() { var txBuf = new Buffer('01000000010000000000000000000000000000000000000000000000000000000000000000ffffffff0704ffff001d0104ffffffff0100f2052a0100000043410496b538e853519c726a2c91e61ec11600ae1390813a627c66fb8be7947be63c52da7589379515d4e0a604f8141781e62294721166bf621e73a82cbf2342c858eeac00000000', 'hex'); - var am = new AddressService({node: mocknode}); + var am = new AddressService({ + mempoolMemoryIndex: true, + node: mocknode + }); var address = '12c6DSiU4Rq3P4ZxziKxzrL5LmMBrzjrJX'; var message = {}; am.transactionOutputHandler = function(messages) { @@ -113,7 +127,10 @@ describe('Address Service', function() { describe('#_extractAddressInfoFromScript', function() { var am; before(function() { - am = new AddressService({node: mocknode}); + am = new AddressService({ + mempoolMemoryIndex: true, + node: mocknode + }); am.node.network = Networks.livenet; }); it('pay-to-publickey', function() { @@ -149,7 +166,10 @@ describe('Address Service', function() { var testBlock = bitcore.Block.fromString(blockData); before(function() { - am = new AddressService({node: mocknode}); + am = new AddressService({ + mempoolMemoryIndex: true, + node: mocknode + }); am.node.network = Networks.livenet; }); @@ -204,7 +224,10 @@ describe('Address Service', function() { }); }); it('should continue if output script is null', function(done) { - var am = new AddressService({node: mocknode, network: 'livenet'}); + var am = new AddressService({ + mempoolMemoryIndex: true, + node: mocknode, + }); var block = { __height: 345003, @@ -236,6 +259,7 @@ describe('Address Service', function() { var testBlock = bitcore.Block.fromString(blockData); var db = {}; var testnode = { + datadir: 'testdir', db: db, services: { bitcoind: { @@ -243,7 +267,10 @@ describe('Address Service', function() { } } }; - var am = new AddressService({node: testnode, network: 'livenet'}); + var am = new AddressService({ + mempoolMemoryIndex: true, + node: testnode + }); am.transactionEventHandler = sinon.spy(); am.balanceEventHandler = sinon.spy(); @@ -269,15 +296,46 @@ describe('Address Service', function() { }); }); + describe('#_encodeSpentIndexSyncKey', function() { + it('will encode to 36 bytes (string)', function() { + var am = new AddressService({ + mempoolMemoryIndex: true, + node: mocknode + }); + var txidBuffer = new Buffer('3b6bc2939d1a70ce04bc4f619ee32608fbff5e565c1f9b02e4eaa97959c59ae7', 'hex'); + var key = am._encodeSpentIndexSyncKey(txidBuffer, 12); + key.length.should.equal(36); + }); + it('will be able to decode encoded value', function() { + var am = new AddressService({ + mempoolMemoryIndex: true, + node: mocknode + }); + var txid = '3b6bc2939d1a70ce04bc4f619ee32608fbff5e565c1f9b02e4eaa97959c59ae7'; + var txidBuffer = new Buffer(txid, 'hex'); + var key = am._encodeSpentIndexSyncKey(txidBuffer, 12); + var keyBuffer = new Buffer(key, 'binary'); + keyBuffer.slice(0, 32).toString('hex').should.equal(txid); + var outputIndex = keyBuffer.readUInt32BE(32); + outputIndex.should.equal(12); + }); + }); + describe('#_encodeInputKeyMap/#_decodeInputKeyMap roundtrip', function() { var encoded; var outputTxIdBuffer = new Buffer('3b6bc2939d1a70ce04bc4f619ee32608fbff5e565c1f9b02e4eaa97959c59ae7', 'hex'); it('encode key', function() { - var am = new AddressService({node: mocknode}); + var am = new AddressService({ + mempoolMemoryIndex: true, + node: mocknode + }); encoded = am._encodeInputKeyMap(outputTxIdBuffer, 13); }); it('decode key', function() { - var am = new AddressService({node: mocknode}); + var am = new AddressService({ + mempoolMemoryIndex: true, + node: mocknode + }); var key = am._decodeInputKeyMap(encoded); key.outputTxId.toString('hex').should.equal(outputTxIdBuffer.toString('hex')); key.outputIndex.should.equal(13); @@ -288,11 +346,17 @@ describe('Address Service', function() { var encoded; var inputTxIdBuffer = new Buffer('3b6bc2939d1a70ce04bc4f619ee32608fbff5e565c1f9b02e4eaa97959c59ae7', 'hex'); it('encode key', function() { - var am = new AddressService({node: mocknode}); + var am = new AddressService({ + mempoolMemoryIndex: true, + node: mocknode + }); encoded = am._encodeInputValueMap(inputTxIdBuffer, 7); }); it('decode key', function() { - var am = new AddressService({node: mocknode}); + var am = new AddressService({ + mempoolMemoryIndex: true, + node: mocknode + }); var key = am._decodeInputValueMap(encoded); key.inputTxId.toString('hex').should.equal(inputTxIdBuffer.toString('hex')); key.inputIndex.should.equal(7); @@ -301,7 +365,10 @@ describe('Address Service', function() { describe('#transactionEventHandler', function() { it('will emit a transaction if there is a subscriber', function(done) { - var am = new AddressService({node: mocknode}); + var am = new AddressService({ + mempoolMemoryIndex: true, + node: mocknode + }); var emitter = new EventEmitter(); var address = bitcore.Address('1DzjESe6SLmAKVPLFMj6Sx1sWki3qt5i8N'); am.subscriptions['address/transaction'] = {}; @@ -335,7 +402,10 @@ describe('Address Service', function() { describe('#balanceEventHandler', function() { it('will emit a balance if there is a subscriber', function(done) { - var am = new AddressService({node: mocknode}); + var am = new AddressService({ + mempoolMemoryIndex: true, + node: mocknode + }); var emitter = new EventEmitter(); var address = bitcore.Address('1DzjESe6SLmAKVPLFMj6Sx1sWki3qt5i8N'); am.subscriptions['address/balance'][address.hashBuffer.toString('hex')] = [emitter]; @@ -358,7 +428,10 @@ describe('Address Service', function() { describe('#subscribe', function() { it('will add emitters to the subscribers array (transaction)', function() { - var am = new AddressService({node: mocknode}); + var am = new AddressService({ + mempoolMemoryIndex: true, + node: mocknode + }); var emitter = new EventEmitter(); var address = bitcore.Address('1DzjESe6SLmAKVPLFMj6Sx1sWki3qt5i8N'); @@ -378,7 +451,10 @@ describe('Address Service', function() { .should.deep.equal([emitter, emitter2]); }); it('will add an emitter to the subscribers array (balance)', function() { - var am = new AddressService({node: mocknode}); + var am = new AddressService({ + mempoolMemoryIndex: true, + node: mocknode + }); var emitter = new EventEmitter(); var name = 'address/balance'; var address = bitcore.Address('1DzjESe6SLmAKVPLFMj6Sx1sWki3qt5i8N'); @@ -400,7 +476,10 @@ describe('Address Service', function() { describe('#unsubscribe', function() { it('will remove emitter from subscribers array (transaction)', function() { - var am = new AddressService({node: mocknode}); + var am = new AddressService({ + mempoolMemoryIndex: true, + node: mocknode + }); var emitter = new EventEmitter(); var emitter2 = new EventEmitter(); var address = bitcore.Address('1DzjESe6SLmAKVPLFMj6Sx1sWki3qt5i8N'); @@ -411,7 +490,10 @@ describe('Address Service', function() { .should.deep.equal([emitter2]); }); it('will remove emitter from subscribers array (balance)', function() { - var am = new AddressService({node: mocknode}); + var am = new AddressService({ + mempoolMemoryIndex: true, + node: mocknode + }); var emitter = new EventEmitter(); var emitter2 = new EventEmitter(); var address = bitcore.Address('1DzjESe6SLmAKVPLFMj6Sx1sWki3qt5i8N'); @@ -422,7 +504,10 @@ describe('Address Service', function() { .should.deep.equal([emitter2]); }); it('should unsubscribe from all addresses if no addresses are specified', function() { - var am = new AddressService({node: mocknode}); + var am = new AddressService({ + mempoolMemoryIndex: true, + node: mocknode + }); var emitter = new EventEmitter(); var emitter2 = new EventEmitter(); var address1 = bitcore.Address('1KiW1A4dx1oRgLHtDtBjcunUGkYtFgZ1W'); @@ -439,7 +524,10 @@ describe('Address Service', function() { describe('#getBalance', function() { it('should sum up the unspent outputs', function(done) { - var am = new AddressService({node: mocknode}); + var am = new AddressService({ + mempoolMemoryIndex: true, + node: mocknode + }); var outputs = [ {satoshis: 1000}, {satoshis: 2000}, {satoshis: 3000} ]; @@ -452,7 +540,10 @@ describe('Address Service', function() { }); it('will handle error from unspent outputs', function(done) { - var am = new AddressService({node: mocknode}); + var am = new AddressService({ + mempoolMemoryIndex: true, + node: mocknode + }); am.getUnspentOutputs = sinon.stub().callsArgWith(2, new Error('error')); am.getBalance('someaddress', false, function(err) { should.exist(err); @@ -473,6 +564,8 @@ describe('Address Service', function() { } }; var testnode = { + network: Networks.testnet, + datadir: 'testdir', services: { db: db, bitcoind: { @@ -481,7 +574,10 @@ describe('Address Service', function() { } }; before(function() { - am = new AddressService({node: testnode}); + am = new AddressService({ + mempoolMemoryIndex: true, + node: testnode + }); }); it('will add mempool inputs on close', function(done) { @@ -492,6 +588,8 @@ describe('Address Service', function() { } }; var testnode = { + network: Networks.testnet, + datadir: 'testdir', services: { db: db, bitcoind: { @@ -499,19 +597,20 @@ describe('Address Service', function() { } } }; - var am = new AddressService({node: testnode}); + var am = new AddressService({ + mempoolMemoryIndex: true, + node: testnode + }); var args = { start: 15, end: 12, queryMempool: true }; - am.mempoolInputIndex[address] = [ - { - address: address, - height: -1, - confirmations: 0 - } - ]; + am._getInputsMempool = sinon.stub().callsArgWith(2, null, { + address: address, + height: -1, + confirmations: 0 + }); am.getInputs(address, args, function(err, inputs) { should.not.exist(err); inputs.length.should.equal(1); @@ -542,6 +641,7 @@ describe('Address Service', function() { am.node.services.bitcoind = { getMempoolInputs: sinon.stub().returns([]) }; + am._getInputsMempool = sinon.stub().callsArgWith(2, null, []); am.getInputs(address, args, function(err, inputs) { should.not.exist(err); inputs.length.should.equal(1); @@ -615,6 +715,82 @@ describe('Address Service', function() { }); + describe('#_getInputsMempool', function() { + var am; + var address = '1KiW1A4dx1oRgLHtDtBjcunUGkYtFgZ1W'; + var hashBuffer = bitcore.Address(address).hashBuffer; + var db = { + tip: { + __height: 1 + } + }; + var testnode = { + network: Networks.testnet, + datadir: 'testdir', + services: { + db: db, + bitcoind: { + on: sinon.stub() + } + } + }; + before(function() { + am = new AddressService({ + mempoolMemoryIndex: true, + node: testnode + }); + }); + it('it will handle error', function(done) { + var testStream = new EventEmitter(); + am.mempoolIndex = {}; + am.mempoolIndex.createReadStream = sinon.stub().returns(testStream); + + am._getInputsMempool(address, hashBuffer, function(err, outputs) { + should.exist(err); + err.message.should.equal('readstreamerror'); + done(); + }); + + testStream.emit('error', new Error('readstreamerror')); + setImmediate(function() { + testStream.emit('close'); + }); + }); + it('it will parse data', function(done) { + var testStream = new EventEmitter(); + am.mempoolIndex = {}; + am.mempoolIndex.createReadStream = sinon.stub().returns(testStream); + + am._getInputsMempool(address, hashBuffer, function(err, outputs) { + should.not.exist(err); + outputs.length.should.equal(1); + outputs[0].address.should.equal(address); + outputs[0].txid.should.equal(txid); + outputs[0].inputIndex.should.equal(5); + outputs[0].height.should.equal(-1); + outputs[0].confirmations.should.equal(0); + done(); + }); + + var txid = '5d32f0fff6871c377e00c16f48ebb5e89c723d0b9dd25f68fdda70c3392bee61'; + var inputIndex = 5; + var inputIndexBuffer = new Buffer(4); + inputIndexBuffer.writeUInt32BE(inputIndex); + var valueData = Buffer.concat([ + new Buffer(txid, 'hex'), + inputIndexBuffer + ]); + + // Note: key is not used currently + testStream.emit('data', { + value: valueData + }); + setImmediate(function() { + testStream.emit('close'); + }); + }); + }); + describe('#getOutputs', function() { var am; var address = '1KiW1A4dx1oRgLHtDtBjcunUGkYtFgZ1W'; @@ -625,6 +801,8 @@ describe('Address Service', function() { } }; var testnode = { + network: Networks.testnet, + datadir: 'testdir', services: { db: db, bitcoind: { @@ -637,7 +815,10 @@ describe('Address Service', function() { }; before(function() { - am = new AddressService({node: testnode}); + am = new AddressService({ + mempoolMemoryIndex: true, + node: testnode + }); }); it('will get outputs for an address and timestamp', function(done) { @@ -658,6 +839,7 @@ describe('Address Service', function() { return testStream; } }; + am._getOutputsMempool = sinon.stub().callsArgWith(2, null, []); am.getOutputs(address, args, function(err, outputs) { should.not.exist(err); outputs.length.should.equal(1); @@ -684,16 +866,17 @@ describe('Address Service', function() { createReadStream: sinon.stub().returns(readStream1) }; - am.mempoolOutputIndex = {}; - - am.mempoolOutputIndex[address] = [ + am._getOutputsMempool = sinon.stub().callsArgWith(2, null, [ { + address: address, + height: -1, + confirmations: 0, txid: 'aa2db23f670596e96ed94c405fd11848c8f236d266ee96da37ecd919e53b4371', satoshis: 307627737, script: '76a914f6db95c81dea3d10f0ff8d890927751bf7b203c188ac', outputIndex: 0 } - ]; + ]); am.getOutputs(address, options, function(err, outputs) { should.not.exist(err); @@ -762,6 +945,8 @@ describe('Address Service', function() { var db = {}; var testnode = { + network: Networks.testnet, + datadir: 'testdir', services: { db: db, bitcoind: { @@ -769,7 +954,10 @@ describe('Address Service', function() { } } }; - var am = new AddressService({node: testnode}); + var am = new AddressService({ + mempoolMemoryIndex: true, + node: testnode + }); am.getUnspentOutputsForAddress = function(address, queryMempool, callback) { var result = addresses[address]; if(result instanceof Error) { @@ -794,6 +982,8 @@ describe('Address Service', function() { var db = {}; var testnode = { + network: Networks.testnet, + datadir: 'testdir', db: db, services: { bitcoind: { @@ -801,7 +991,10 @@ describe('Address Service', function() { } } }; - var am = new AddressService({node: testnode}); + var am = new AddressService({ + mempoolMemoryIndex: true, + node: testnode + }); am.getUnspentOutputsForAddress = function(address, queryMempool, callback) { var result = addresses[address]; if(result instanceof Error) { @@ -827,6 +1020,8 @@ describe('Address Service', function() { var db = {}; var testnode = { + network: Networks.testnet, + datadir: 'testdir', db: db, services: { bitcoind: { @@ -834,7 +1029,10 @@ describe('Address Service', function() { } } }; - var am = new AddressService({node: testnode}); + var am = new AddressService({ + mempoolMemoryIndex: true, + node: testnode + }); am.getUnspentOutputsForAddress = function(address, queryMempool, callback) { var result = addresses[address]; if(result instanceof Error) { @@ -870,7 +1068,10 @@ describe('Address Service', function() { ]; var i = 0; - var am = new AddressService({node: mocknode}); + var am = new AddressService({ + mempoolMemoryIndex: true, + node: mocknode + }); am.getOutputs = sinon.stub().callsArgWith(2, null, outputs); am.isUnspent = function(output, options, callback) { callback(!outputs[i].spent); @@ -886,7 +1087,10 @@ describe('Address Service', function() { }); }); it('should handle an error from getOutputs', function(done) { - var am = new AddressService({node: mocknode}); + var am = new AddressService({ + mempoolMemoryIndex: true, + node: mocknode + }); am.getOutputs = sinon.stub().callsArgWith(2, new Error('error')); am.getUnspentOutputsForAddress('1KiW1A4dx1oRgLHtDtBjcunUGkYtFgZ1W', false, function(err, outputs) { should.exist(err); @@ -895,7 +1099,10 @@ describe('Address Service', function() { }); }); it('should handle when there are no outputs', function(done) { - var am = new AddressService({node: mocknode}); + var am = new AddressService({ + mempoolMemoryIndex: true, + node: mocknode + }); am.getOutputs = sinon.stub().callsArgWith(2, null, []); am.getUnspentOutputsForAddress('1KiW1A4dx1oRgLHtDtBjcunUGkYtFgZ1W', false, function(err, outputs) { should.exist(err); @@ -910,7 +1117,10 @@ describe('Address Service', function() { var am; before(function() { - am = new AddressService({node: mocknode}); + am = new AddressService({ + mempoolMemoryIndex: true, + node: mocknode + }); }); it('should give true when isSpent() gives false', function(done) { @@ -941,6 +1151,8 @@ describe('Address Service', function() { describe('#isSpent', function() { var db = {}; var testnode = { + network: Networks.testnet, + datadir: 'testdir', db: db, services: { bitcoind: { @@ -949,7 +1161,10 @@ describe('Address Service', function() { } }; it('should give true if bitcoind.isSpent gives true', function(done) { - var am = new AddressService({node: testnode}); + var am = new AddressService({ + mempoolMemoryIndex: true, + node: testnode + }); am.node.services.bitcoind = { isSpent: sinon.stub().returns(true), on: sinon.stub() @@ -960,7 +1175,10 @@ describe('Address Service', function() { }); }); it('should give true if bitcoind.isSpent is false and mempoolSpentIndex is true', function(done) { - var am = new AddressService({node: testnode}); + var am = new AddressService({ + mempoolMemoryIndex: true, + node: testnode + }); am.node.services.bitcoind = { isSpent: sinon.stub().returns(false), on: sinon.stub() @@ -971,15 +1189,23 @@ describe('Address Service', function() { prevTxId: new Buffer(txid, 'hex'), outputIndex: outputIndex }; - var spentKey = [txid, outputIndex].join('-'); - am.mempoolSpentIndex[spentKey] = new Buffer(5); + var outputIndexBuffer = new Buffer(4); + outputIndexBuffer.writeUInt32BE(outputIndex); + var spentKey = Buffer.concat([ + new Buffer(txid, 'hex'), + outputIndexBuffer + ]).toString('binary'); + am.mempoolSpentIndex[spentKey] = true; am.isSpent(output, {queryMempool: true}, function(spent) { spent.should.equal(true); done(); }); }); it('should give false if spent in mempool with queryMempool set to false', function(done) { - var am = new AddressService({node: testnode}); + var am = new AddressService({ + mempoolMemoryIndex: true, + node: testnode + }); am.node.services.bitcoind = { isSpent: sinon.stub().returns(false), on: sinon.stub() @@ -998,19 +1224,27 @@ describe('Address Service', function() { }); }); it('default to querying the mempool', function(done) { - var am = new AddressService({node: testnode}); + var am = new AddressService({ + mempoolMemoryIndex: true, + node: testnode + }); am.node.services.bitcoind = { isSpent: sinon.stub().returns(false), on: sinon.stub() }; - var txid = '3b6bc2939d1a70ce04bc4f619ee32608fbff5e565c1f9b02e4eaa97959c59ae7'; + var txidBuffer = new Buffer('3b6bc2939d1a70ce04bc4f619ee32608fbff5e565c1f9b02e4eaa97959c59ae7', 'hex'); var outputIndex = 0; var output = { - prevTxId: new Buffer(txid, 'hex'), + prevTxId: txidBuffer, outputIndex: outputIndex }; - var spentKey = [txid, outputIndex].join('-'); - am.mempoolSpentIndex[spentKey] = new Buffer(5); + var outputIndexBuffer = new Buffer(4); + outputIndexBuffer.writeUInt32BE(outputIndex); + var spentKey = Buffer.concat([ + txidBuffer, + outputIndexBuffer + ]).toString('binary'); + am.mempoolSpentIndex[spentKey] = true; am.isSpent(output, {}, function(spent) { spent.should.equal(true); done(); @@ -1029,7 +1263,10 @@ describe('Address Service', function() { var TestAddressService = proxyquire('../../../lib/services/address', { './history': TestAddressHistory }); - var am = new TestAddressService({node: mocknode}); + var am = new TestAddressService({ + mempoolMemoryIndex: true, + node: mocknode + }); am.getAddressHistory([], {}, function(err, history) { TestAddressHistory.prototype.get.callCount.should.equal(1); done(); @@ -1041,30 +1278,55 @@ describe('Address Service', function() { var tx = Transaction().fromBuffer(txBuf); before(function() { - am = new AddressService({node: mocknode}); + am = new AddressService({ + mempoolMemoryIndex: true, + node: mocknode + }); }); it('will update the input and output indexes', function() { - am.updateMempoolIndex(tx); - am.mempoolInputIndex['18Z29uNgWyUDtNyTKE1PaurbSR131EfANc'][0].txid.should.equal('45202ffdeb8344af4dec07cddf0478485dc65cc7d08303e45959630c89b51ea2'); - am.mempoolOutputIndex['12w93weN8oti3P1e5VYEuygqyujhADF7J5'][0].txid.should.equal('45202ffdeb8344af4dec07cddf0478485dc65cc7d08303e45959630c89b51ea2'); - Object.keys(am.mempoolSpentIndex).length.should.equal(14); - am.mempoolInputIndex['1JT7KDYwT9JY9o2vyqcKNSJgTWeKfV3ui8'].length.should.equal(12); - am.mempoolOutputIndex['12w93weN8oti3P1e5VYEuygqyujhADF7J5'].length.should.equal(1); + am.mempoolIndex = {}; + am.mempoolIndex.batch = function(operations, callback) { + callback.should.be.a('function'); + Object.keys(am.mempoolSpentIndex).length.should.equal(14); + for (var i = 0; i < operations.length; i++) { + operations[i].type.should.equal('put'); + } + var expectedValue = '45202ffdeb8344af4dec07cddf0478485dc65cc7d08303e45959630c89b51ea200000002'; + operations[7].value.toString('hex').should.equal(expectedValue); + var matches = 0; + for (var j = 0; j < operations.length; j++) { + var match = Buffer.concat([ + AddressService.MEMPREFIXES.SPENTS, + bitcore.Address('1JT7KDYwT9JY9o2vyqcKNSJgTWeKfV3ui8').hashBuffer + ]).toString('hex'); + + if (operations[j].key.slice(0, 21).toString('hex') === match) { + matches++; + } + } + matches.should.equal(12); + }; + am.updateMempoolIndex(tx, true); }); it('will remove the input and output indexes', function() { - am.removeMempoolIndex(tx); - should.not.exist(am.mempoolInputIndex['18Z29uNgWyUDtNyTKE1PaurbSR131EfANc']); - should.not.exist(am.mempoolOutputIndex['12w93weN8oti3P1e5VYEuygqyujhADF7J5']); - Object.keys(am.mempoolSpentIndex).length.should.equal(0); - should.not.exist(am.mempoolInputIndex['1JT7KDYwT9JY9o2vyqcKNSJgTWeKfV3ui8']); - should.not.exist(am.mempoolOutputIndex['12w93weN8oti3P1e5VYEuygqyujhADF7J5']); + am.mempoolIndex = {}; + am.mempoolIndex.batch = function(operations, callback) { + callback.should.be.a('function'); + Object.keys(am.mempoolSpentIndex).length.should.equal(0); + for (var i = 0; i < operations.length; i++) { + operations[i].type.should.equal('del'); + } + }; + am.updateMempoolIndex(tx, false); }); }); describe('#getAddressSummary', function() { var node = { + datadir: 'testdir', + network: Networks.testnet, services: { bitcoind: { isSpent: sinon.stub().returns(false), @@ -1101,13 +1363,18 @@ describe('Address Service', function() { } ]; - var as = new AddressService({node: node}); + var as = new AddressService({ + mempoolMemoryIndex: true, + node: node + }); as.getInputs = sinon.stub().callsArgWith(2, null, inputs); as.getOutputs = sinon.stub().callsArgWith(2, null, outputs); - as.mempoolSpentIndex = { - '689e9f543fa4aa5b2daa3b5bb65f9a00ad5aa1a2e9e1fc4e11061d85f2aa9bc5-0': true - }; - + var key = Buffer.concat([ + new Buffer('689e9f543fa4aa5b2daa3b5bb65f9a00ad5aa1a2e9e1fc4e11061d85f2aa9bc5', 'hex'), + new Buffer(Array(4)) + ]).toString('binary'); + as.mempoolSpentIndex = {}; + as.mempoolSpentIndex[key] = true; it('should handle unconfirmed and confirmed outputs and inputs', function(done) { as.getAddressSummary('mpkDdnLq26djg17s6cYknjnysAm3QwRzu2', {}, function(err, summary) { should.not.exist(err);