RanchiMall/flocore-node
3
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge pull request #19 from maraoz/feature/p2p-import

Browse Source 
p2p block and transaction importing
This commit is contained in:
Manuel Aráoz 2014-01-10 11:21:03 -08:00
commit 3c22a323c9
9 changed files with 317 additions and 393 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
.jshintrc
View File

@ -34,7 +34,7 @@
"inject", "inject",
"expect" "expect"
], ],
"indent": 2, // Specify indentation spacing "indent": false, // Specify indentation spacing
"devel": true, // Allow development statements e.g. `console.log();`. "devel": true, // Allow development statements e.g. `console.log();`.
"noempty": true // Prohibit use of empty blocks. "noempty": true // Prohibit use of empty blocks.
} }

14
app/models/Transaction.js
View File

@ -3,6 +3,7 @@
/** /**
* Module dependencies. * Module dependencies.
*/ */
var mongoose = require('mongoose'), var mongoose = require('mongoose'),
Schema = mongoose.Schema, Schema = mongoose.Schema,
async = require('async'), async = require('async'),
@ -59,31 +60,20 @@ TransactionSchema.statics.fromIdWithInfo = function(txid, cb) {
}; };
TransactionSchema.statics.createFromArray = function(txs, next) { TransactionSchema.statics.createFromArray = function(txs, next) {
var that = this; var that = this;
if (!txs) return next(); if (!txs) return next();
// console.log('exploding ', txs);
async.forEach( txs, async.forEach( txs,
function(tx, callback) { function(tx, callback) {
// console.log('procesing TX %s', tx);
that.create({ txid: tx }, function(err) { that.create({ txid: tx }, function(err) {
if (err && ! err.toString().match(/E11000/)) { if (err && ! err.toString().match(/E11000/)) {
return callback();
}
if (err) {
return callback(err); return callback(err);
} }
return callback(); return callback();
}); });
}, },
function(err) { function(err) {
if (err) return next(err); return next(err);
return next();
} }
); );
}; };

2
config/env/development.js vendored
View File

@ -10,7 +10,7 @@ module.exports = {
pass: 'real_mystery', pass: 'real_mystery',
protocol: 'http', protocol: 'http',
host: process.env.BITCOIND_HOST || '127.0.0.1', host: process.env.BITCOIND_HOST || '127.0.0.1',
port: process.env.BITCOIND_PORT || '8332', port: process.env.BITCOIND_PORT || '18332',
}, },
network: 'testnet', network: 'testnet',
} }

2
etc/bitcoind/bitcoin.conf
View File

@ -1,8 +1,6 @@
rpcuser=mystery rpcuser=mystery
rpcpassword=real_mystery rpcpassword=real_mystery
server=1 server=1
rpcport=8332
testnet=3
txindex=1 txindex=1
# Allow connections outsite localhost? # Allow connections outsite localhost?

356
lib/Sync.js
View File

@ -1,226 +1,212 @@
require('classtool'); 'use strict';
require('classtool');
/* We dont sync any contents from TXs, only their IDs are stored */ /* We dont sync any contents from TXs, only their IDs are stored */
var isSyncTxEnabled = 0; var isSyncTxEnabled = 0;
function spec(b) { function spec() {
var mongoose = require('mongoose'); var mongoose = require('mongoose');
var util = require('util'); var util = require('util');
var RpcClient = require('bitcore/RpcClient').class(); var RpcClient = require('bitcore/RpcClient').class();
var networks = require('bitcore/networks'); var networks = require('bitcore/networks');
var async = require('async'); var async = require('async');
var config = require('../config/config'); var config = require('../config/config');
var Block = require('../app/models/Block'); var Block = require('../app/models/Block');
var Transaction=require('../app/models/Transaction'); var Transaction = require('../app/models/Transaction');
function Sync(config) { function Sync(config) {
this.network = config.networkName == 'testnet' ? networks.testnet : networks.livenet; this.network = config.networkName === 'testnet' ? networks.testnet: networks.livenet;
} }
var progress_bar = function(string, current, total) { var progress_bar = function(string, current, total) {
console.log( util.format("\t%s %d/%d [%d%%]", console.log(util.format('\t%s %d/%d [%d%%]', string, current, total, parseInt(100 * current / total)));
string, current, total, parseInt(100 * current/total)) };
);
}
Sync.prototype.getNextBlock = function (blockHash,cb) { Sync.prototype.getNextBlock = function(blockHash, cb) {
var that = this; var that = this;
if (!blockHash) {
return cb();
}
this.rpc.getBlock(blockHash, function(err, blockInfo) {
if (err) return cb(err);
if (blockInfo.result.height % 1000 === 0) {
var h = blockInfo.result.height,
d = blockInfo.result.confirmations;
progress_bar('height', h, h + d);
}
if ( !blockHash ) { that.storeBlock(blockInfo.result, function(err) {
return cb(); if (!err) {
} var txs = blockInfo.result.tx;
that.storeTxs(txs, function(err) {
if (!err) {
return that.getNextBlock(blockInfo.result.nextblockhash, cb);
}
});
}
});
});
};
this.rpc.getBlock(blockHash, function(err, blockInfo) { Sync.prototype.storeBlock = function(block, cb) {
if (err) return cb(err); Block.create(block, function(err, inBlock) {
// E11000 => already exists
if (err && ! err.toString().match(/E11000/)) {
return cb(err);
}
cb();
});
};
if ( ! ( blockInfo.result.height % 1000) ) { Sync.prototype.storeTxs = function(txs, cb) {
var h = blockInfo.result.height, Transaction.createFromArray(txs, cb);
d = blockInfo.result.confirmations; };
progress_bar('height', h, h + d);
}
Block.create( blockInfo.result, function(err, inBlock) { Sync.prototype.syncBlocks = function(reindex, cb) {
var that = this;
var genesisHash = this.network.genesisBlock.hash.reverse().toString('hex');
// E11000 => already exists console.log('Syncing Blocks... ' + reindex);
if (err && ! err.toString().match(/E11000/)) { if (reindex) {
return cb(err); return this.getNextBlock(genesisHash, cb);
} }
if (inBlock) { Block.findOne({},
Transaction.createFromArray( blockInfo.result.tx,function (err) { {},
return that.getNextBlock(blockInfo.result.nextblockhash, cb); {
}); sort: {
} 'time': - 1
else }
return that.getNextBlock(blockInfo.result.nextblockhash, cb); },
}); function(err, block) {
}); if (err) return cb(err);
}
Sync.prototype.syncBlocks = function (reindex, cb) { var nextHash = block && block.hash ? block.hash: genesisHash;
var that = this; console.log('\tStarting at hash: ' + nextHash);
var genesisHash = this.network.genesisBlock.hash.reverse().toString('hex'); return that.getNextBlock(nextHash, cb);
});
};
console.log("Syncing Blocks..."); // This is not currently used. Transactions are represented by txid only
if (reindex) // in mongodb
return this.getNextBlock(genesisHash, cb); Sync.prototype.syncTXs = function(reindex, cb) {
var that = this;
Block.findOne({}, {}, { sort: { 'time' : -1 } }, function(err, block) { console.log('Syncing TXs...');
if (err) return cb(err); if (reindex) {
// TODO?
}
var nextHash = Transaction.find({
block && block.hash blockhash: null
? block.hash },
: genesisHash function(err, txs) {
; if (err) return cb(err);
var read = 0;
var pull = 0;
var write = 0;
var total = txs.length;
console.log('\tneed to pull %d txs', total);
console.log('\tStarting at hash: ' + nextHash); if (!total) return cb();
return that.getNextBlock(nextHash, cb);
});
}
async.each(txs, function(tx, next) {
if (!tx.txid) {
console.log('NO TXID skipping...', tx);
return next();
}
// This is not currently used. Transactions are represented by txid only if (read++ % 1000 === 0) progress_bar('read', read, total);
// in mongodb
Sync.prototype.syncTXs = function (reindex, cb) {
var that = this; that.rpc.getRawTransaction(tx.txid, 1, function(err, txInfo) {
console.log("Syncing TXs..."); if (pull++ % 1000 === 0) progress_bar('\tpull', pull, total);
if (reindex) {
// TODO?
}
if (!err && txInfo) {
Transaction.update({
txid: tx.txid
},
txInfo.result, function(err) {
if (err) return next(err);
Transaction.find({blockhash: null}, function(err, txs) { if (write++ % 1000 === 0) progress_bar('\t\twrite', write, total);
if (err) return cb(err);
var read = 0; return next();
var pull = 0; });
var write = 0; }
var total = txs.length; else return next();
console.log("\tneed to pull %d txs", total); });
},
function(err) {
if (err) return cb(err);
return cb(err);
});
});
};
if (!total) return cb(); Sync.prototype.init = function(opts) {
async.each(txs, mongoose.connect(config.db);
function(tx, next){ this.db = mongoose.connection;
if (! tx.txid) { this.rpc = new RpcClient(config.bitcoind);
console.log("NO TXID skipping...", tx);
return next();
}
if ( ! ( read++ % 1000) ) this.db.on('error', console.error.bind(console, 'connection error:'));
progress_bar('read', read, total);
};
that.rpc.getRawTransaction(tx.txid, 1, function(err, txInfo) { Sync.prototype.import_history = function(opts, next) {
if ( ! ( pull++ % 1000) ) var that = this;
progress_bar('\tpull', pull, total); this.db.once('open', function() {
async.series([
function(cb) {
if (opts.destroy) {
console.log('Deleting Blocks...');
that.db.collections.blocks.drop(cb);
} else {
cb();
}
},
function(cb) {
if (opts.destroy) {
console.log('Deleting TXs...');
that.db.collections.transactions.drop(cb);
} else {
cb();
}
},
function(cb) {
if (!opts.skip_blocks) {
that.syncBlocks(opts.reindex, cb);
} else {
cb();
}
},
function(cb) {
if (isSyncTxEnabled && ! opts.skip_txs) {
that.syncTXs(opts.reindex, cb);
}
else {
return cb();
}
}], function(err) {
return next(err);
});
});
};
if (!err && txInfo) { Sync.prototype.close = function() {
Transaction.update({txid: tx.txid}, txInfo.result, function(err) { console.log("closing connection");
if (err) return next(err); this.db.close();
};
if ( ! ( write++ % 1000) ) return Sync;
progress_bar('\t\twrite', write, total); }
return next();
});
}
else return next();
});
},
function(err){
if (err) return cb(err);
return cb(err);
}
);
});
}
Sync.prototype.start = function (opts, next) {
mongoose.connect(config.db);
var db = mongoose.connection;
this.rpc = new RpcClient(config.bitcoind);
var that = this;
db.on('error', console.error.bind(console, 'connection error:'));
db.once('open', function (){
async.series([
function(cb){
if (opts.destroy) {
console.log("Deleting Blocks...");
return db.collections['blocks'].drop(cb);
}
return cb();
},
function(cb){
if (opts.destroy) {
console.log("Deleting TXs...");
return db.collections['transactions'].drop(cb);
}
return cb();
},
function(cb) {
if (! opts.skip_blocks) {
that.syncBlocks(opts.reindex, function(err) {
if (err) {
return cb(err);
}
console.log("\tBlocks done.");
return cb();
});
}
else {
return cb();
}
},
function(cb) {
if ( isSyncTxEnabled && ! opts.skip_txs) {
that.syncTXs(opts.reindex, function(err) {
if (err) {
return cb(err);
}
return cb();
});
}
else {
return cb();
}
},
function(cb) {
db.close();
return cb();
},
],
function(err) {
if (err) {
db.close();
return next(err);
}
return next();
});
});
}
return Sync;
};
module.defineClass(spec); module.defineClass(spec);

274
p2p.js
View File

@ -1,5 +1,7 @@
'use strict'; 'use strict';
process.env.NODE_ENV = process.env.NODE_ENV || 'development';
var fs = require('fs'); var fs = require('fs');
var HeaderDB = require('./HeaderDB').class(); var HeaderDB = require('./HeaderDB').class();
var Block = require('bitcore/Block').class(); var Block = require('bitcore/Block').class();
@ -7,6 +9,8 @@ var CoinConst = require('bitcore/const');
var coinUtil = require('bitcore/util/util'); var coinUtil = require('bitcore/util/util');
var networks = require('bitcore/networks'); var networks = require('bitcore/networks');
var Parser = require('bitcore/util/BinaryParser').class(); var Parser = require('bitcore/util/BinaryParser').class();
var async = require('async');
var Sync = require('./lib/Sync').class();
var peerdb_fn = 'peerdb.json'; var peerdb_fn = 'peerdb.json';
@ -14,222 +18,154 @@ var peerdb = undefined;
var hdrdb = undefined; var hdrdb = undefined;
var network = networks.testnet; var network = networks.testnet;
var config = { var config = {
network : network.name network: network.name
}; };
var PeerManager = require('bitcore/PeerManager').createClass({ var PeerManager = require('bitcore/PeerManager').createClass({
config : config config: config
}); });
var Peer = require('bitcore/Peer').class(); var Peer = require('bitcore/Peer').class();
function peerdb_load() { function peerdb_load() {
try { try {
peerdb = JSON.parse(fs.readFileSync(peerdb_fn)); peerdb = JSON.parse(fs.readFileSync(peerdb_fn));
} catch (d) { } catch(d) {
console.warn('Unable to read peer db', peerdb_fn, 'creating new one.'); console.warn('Unable to read peer db', peerdb_fn, 'creating new one.');
peerdb = [ { peerdb = [{
ipv4 : '127.0.0.1', ipv4: '127.0.0.1',
port : 18333 port: 18333
}, ]; },
];
fs.writeFileSync(peerdb_fn, JSON.stringify(peerdb)); fs.writeFileSync(peerdb_fn, JSON.stringify(peerdb));
} }
} }
function hdrdb_load() function hdrdb_load() {
{ hdrdb = new HeaderDB({
hdrdb = new HeaderDB({network: network}); network: network
});
} }
function get_more_headers(info) { function get_more_headers(info) {
var conn = info.conn; var conn = info.conn;
var loc = hdrdb.locator(); var loc = hdrdb.locator();
conn.sendGetHeaders(loc, coinUtil.NULL_HASH); conn.sendGetHeaders(loc, coinUtil.NULL_HASH);
} }
function add_header(info, block) { function add_header(info, block) {
var hashStr = coinUtil.formatHashFull(block.calcHash()); var hashStr = coinUtil.formatHashFull(block.calcHash());
try { try {
hdrdb.add(block); hdrdb.add(block);
} catch (e) { } catch(e) {
return; return;
} }
} }
function handle_headers(info) { function handle_headers(info) {
console.log('handle headers'); console.log('handle headers');
var headers = info.message.headers; var headers = info.message.headers;
headers.forEach(function(hdr) { headers.forEach(function(hdr) {
add_header(info, hdr); add_header(info, hdr);
}); });
// We persist the header DB after each batch // We persist the header DB after each batch
//hdrdb.writeFile(hdrdb_fn); //hdrdb.writeFile(hdrdb_fn);
// Only one request per batch of headers we receive.
// Only one request per batch of headers we receive. get_more_headers(info);
get_more_headers(info);
} }
function handle_verack(info) { function handle_verack(info) {
var inv = { var inv = {
type : CoinConst.MSG.BLOCK, type: CoinConst.MSG.BLOCK,
hash : network.genesisBlock.hash, hash: network.genesisBlock.hash,
}; };
var invs = [ inv ]; var invs = [inv];
// Asks for the genesis block
// console.log('p2psync: Asking for the genesis block');
// info.conn.sendGetData(invs);
// Asks for the genesis block
// console.log('p2psync: Asking for the genesis block');
// info.conn.sendGetData(invs);
} }
function handle_inv(info) { function handle_inv(info) {
console.log('handle inv'); // TODO: should limit the invs to objects we haven't seen yet
// TODO: should limit the invs to objects we haven't seen yet var invs = info.message.invs;
var invs = info.message.invs; invs.forEach(function(inv) {
invs.forEach(function(inv) { console.log('Handle inv for a ' + CoinConst.MSG.to_str(inv.type));
console.log('Received inv for a '+CoinConst.MSG.to_str(inv.type)); });
} info.conn.sendGetData(invs);
);
console.log('requesting getData');
info.conn.sendGetData(invs);
} }
function handle_tx(info) { var sync = new Sync({
var tx = info.message.tx.getStandardizedObject(); networkName: networks.testnet
console.log('handle tx: '+JSON.stringify(tx)); });
sync.init();
function handle_tx(info) {
var tx = info.message.tx.getStandardizedObject();
console.log('Handle tx: ' + tx.hash);
sync.storeTxs([tx.hash], function(err) {
if (err) {
console.log('error in handle TX: ' + err);
}
});
} }
function handle_block(info) { function handle_block(info) {
console.log('handle block'); var block = info.message.block;
var block = info.message.block; var now = Math.round(new Date().getTime() / 1000);
add_header(info, block); var blockHash = coinUtil.formatHashFull(block.calcHash());
console.log('Handle block: ' + blockHash);
sync.storeBlock({
'hash': blockHash,
'time': now
},
function(err) {
if (err) {
console.log('error in handle Block: ' + err);
} else {
var hashes = block.txs.map(function(tx) {
return coinUtil.formatHashFull(tx.hash);
});
sync.storeTxs(hashes, function() {});
}
});
} }
function handle_connected(data) { function handle_connected(data) {
var peerman = data.pm; var peerman = data.pm;
var peers_n = peerman.peers.length; var peers_n = peerman.peers.length;
console.log('p2psync: Connected to ' + peers_n + ' peer' + (peers_n !== 1 ? 's' : '')); console.log('p2psync: Connected to ' + peers_n + ' peer' + (peers_n !== 1 ? 's': ''));
} }
function p2psync() { function p2psync() {
var peerman = new PeerManager(); var peerman = new PeerManager();
peerdb.forEach(function(datum) { peerdb.forEach(function(datum) {
var peer = new Peer(datum.ipv4, datum.port); var peer = new Peer(datum.ipv4, datum.port);
peerman.addPeer(peer); peerman.addPeer(peer);
}); });
peerman.on('connection', function(conn) { peerman.on('connection', function(conn) {
conn.on('verack', handle_verack); conn.on('verack', handle_verack);
conn.on('block', handle_block); conn.on('block', handle_block);
conn.on('headers', handle_headers); conn.on('headers', handle_headers);
conn.on('inv', handle_inv); conn.on('inv', handle_inv);
conn.on('tx', handle_tx); conn.on('tx', handle_tx);
}); });
peerman.on('connect', handle_connected); peerman.on('connect', handle_connected);
peerman.start(); peerman.start();
}
function filesync_block_buf(blkdir, fn, buf) {
var parser = new Parser(buf);
var block = new Block();
block.parse(parser, true);
var hashStr = coinUtil.formatHashFull(block.calcHash());
try {
hdrdb.add(block);
} catch (e) {
var height = hdrdb.size();
console.log('HeaderDB failed adding block #' + height + ', ' + hashStr);
console.log(' Reason: ' + e);
return;
}
var height = hdrdb.size() - 1;
if ((height % 1000) == 0)
console.log('HeaderDB added block #' + height + ', ' + hashStr);
}
function filesync_open_cb(err, fd, blkdir, fn) {
if (err)
throw err;
var hdrbuf = new Buffer(4 * 2);
while (1) {
// read 2x 32-bit header
var bread = fs.readSync(fd, hdrbuf, 0, 4 * 2, null);
if (bread < (4 * 2)) {
console.log('Short read/EOF, ending scan of ' + fn);
break;
}
// check magic matches
var magic = hdrbuf.slice(0, 4);
if (magic.toString() != network.magic.toString()) {
console.log('Invalid network magic, ending scan of ' + fn);
break;
}
// block size
var blkSize = hdrbuf.readUInt32LE(4);
if (blkSize > (1 * 1024 * 1024))
throw new Error('Invalid block size ' + blkSize);
// read raw block data
var blkBuf = new Buffer(blkSize);
bread = fs.readSync(fd, blkBuf, 0, blkSize, null);
if (bread != blkSize)
throw new Error('Failed to read block');
// process block
filesync_block_buf(blkdir, fn, blkBuf);
}
fs.closeSync(fd);
hdrdb.writeFile(hdrdb_fn);
console.log('Wrote header db');
}
function filesync_block_file(blkdir, fn) {
console.log('Scanning ' + fn + ' for block data.');
var pathname = blkdir + '/' + fn;
fs.open(pathname, 'r', function(err, fd) {
filesync_open_cb(err, fd, blkdir, fn);
});
}
function cmd_filesync_rd(err, files, blkdir) {
if (err)
throw err;
files = files.sort();
var scanned = 0;
files.forEach(function(fn) {
var re = /^blk\d+\.dat$/;
if (fn.match(re)) {
filesync_block_file(blkdir, fn);
scanned++;
}
});
console.log('Scanned ' + scanned + ' of ' + files.length + ' files in '
+ blkdir);
} }
function main() { function main() {
peerdb_load(); peerdb_load();
hdrdb_load(); hdrdb_load();
p2psync(); p2psync();
} }
main(); main();

5
package.json
View File

@ -11,7 +11,12 @@
{ {
"name": "Matias Alejo Garcia", "name": "Matias Alejo Garcia",
"email": "ematiu@gmail.com" "email": "ematiu@gmail.com"
},
{
"name": "Manuel Araoz",
"email": "manuelaraoz@gmail.com"
} }
], ],
"bugs": { "bugs": {
"url": "https://github.com/bitpay/mystery/issues" "url": "https://github.com/bitpay/mystery/issues"

1
test/model/block.js
View File

@ -36,7 +36,6 @@ describe('Block fromHashWithInfo', function(){
}); });
}); });
it('should poll block\'s info from bitcoind', function(done) { it('should poll block\'s info from bitcoind', function(done) {
var block2 = Block.fromHashWithInfo(TESTING_BLOCK, function(err, b2) { var block2 = Block.fromHashWithInfo(TESTING_BLOCK, function(err, b2) {
if (err) done(err); if (err) done(err);

52
util/sync.js
View File

@ -1,34 +1,44 @@
#!/usr/bin/env node # ! /usr/bin / env node
'use strict';
process.env.NODE_ENV = process.env.NODE_ENV || 'development'; process.env.NODE_ENV = process.env.NODE_ENV || 'development';
require('buffertools').extend(); require('buffertools').extend();
var SYNC_VERSION = '0.1'; var SYNC_VERSION = '0.1';
var program = require('commander'); var program = require('commander');
var Sync = require('../lib/Sync').class(); var Sync = require('../lib/Sync').class();
var async = require('async');
program program.version(SYNC_VERSION).option('-N --network [livenet]', 'Set bitcoin network [testnet]', 'testnet').option('-R --reindex', 'Force reindexing', '0').option('-D --destroy', 'Remove current DB', '0').option('--skip_blocks', 'Sync blocks').option('--skip_txs', 'Sync transactions').parse(process.argv);
.version(SYNC_VERSION)
.option('-N --network [livenet]', 'Set bitcoin network [testnet]', 'testnet')
.option('-R --reindex', 'Force reindexing', '0')
.option('-D --destroy', 'Remove current DB', '0')
.option('--skip_blocks', 'Sync blocks')
.option('--skip_txs', 'Sync transactions')
.parse(process.argv);
var sync = new Sync({ networkName: program.network }); var sync = new Sync({
networkName: program.network
});
if (program.remove) { if (program.remove) {
} }
sync.start( program, function(err){ async.series([
if (err) { function(cb) {
console.log("CRITICAL ERROR: ", err); sync.init(program);
} cb();
else { },
console.log('Done!'); function(cb) {
} sync.import_history(program, function(err) {
}); if (err) {
console.log('CRITICAL ERROR: ', err);
}
else {
console.log('Done!');
}
cb();
});
},
function(cb) {
sync.close();
cb();
}]);