zcash-hackworks
/
bitcore-node-zcash
mirror of https://github.com/zcash-hackworks/bitcore-node-zcash.git
1
0
Fork 0
Code Issues Projects Releases Wiki Activity

refactor tdb and bdb

This commit is contained in:
Matias Alejo Garcia 2014-05-23 21:23:44 -03:00
parent 4aaece66a5
commit f0f936d2d8
12 changed files with 2231 additions and 843 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

8
app/controllers/blocks.js
View File

@ -5,9 +5,11 @@
*/
var common = require('./common'),
async = require('async'),
BlockDb = require('../../lib/BlockDb');
BlockDb = require('../../lib/BlockDb'),
TransactionDb = require('../../lib/TransactionDb');
var bdb = new BlockDb();
var tdb = new TransactionDb();
/**
* Find block by hash ...
@ -17,7 +19,7 @@ exports.block = function(req, res, next, hash) {
if (err || !block)
return common.handleErrors(err, res, next);
else {
bdb.getPoolInfo(block.info.tx[0], function(info) {
tdb.getPoolInfo(block.info.tx[0], function(info) {
block.info.poolInfo = info;
req.block = block.info;
return next();
@ -67,7 +69,7 @@ var getBlock = function(blockhash, cb) {
};
}
bdb.getPoolInfo(block.info.tx[0], function(info) {
tdb.getPoolInfo(block.info.tx[0], function(info) {
block.info.poolInfo = info;
return cb(err, block.info);
});

1328
dev-util/node-tick-report Normal file
View File

File diff suppressed because it is too large Load Diff

10
dev-util/stats
View File

@ -58,3 +58,13 @@ first 10%
Mon Mar 10 11:59:25 ART 2014
10% de blk 0 (testnet)
=> 37s
Thu May 22 13:42:50 ART 2014 (base58check + toString opts + custom getStandardizedObject)
10% testnet
=> 29s
100% testnet
=> 17m6s (user time)

155
lib/BlockDb.js
View File

@ -1,6 +1,5 @@
'use strict';
var imports = require('soop').imports();
var ThisParent = imports.parent || require('events').EventEmitter;
var TIMESTAMP_PREFIX = 'bts-'; // bts-<ts> => <hash>
var PREV_PREFIX = 'bpr-'; // bpr-<hash> => <prev_hash>
var NEXT_PREFIX = 'bne-'; // bne-<hash> => <next_hash>
@ -8,6 +7,10 @@ var MAIN_PREFIX = 'bma-'; // bma-<hash> => <height> (0 is unconnect
var TIP = 'bti-'; // bti = <hash>:<height> last block on the chain
var LAST_FILE_INDEX = 'file-'; // last processed file index
// txid - blockhash mapping (only for confirmed txs, ONLY FOR BEST BRANCH CHAIN)
var IN_BLK_PREFIX = 'btx-'; //btx-<txid> = <block>
var MAX_OPEN_FILES = 500;
/**
@ -18,10 +21,15 @@ var levelup = require('levelup'),
var db = imports.db || levelup(config.leveldb + '/blocks',{maxOpenFiles: MAX_OPEN_FILES} );
var Rpc = imports.rpc || require('./Rpc');
var logger = require('./logger').logger;
var d = logger.log;
var info = logger.info;
var BlockDb = function() {
this.txDb = require('./TransactionDb').default();
BlockDb.super(this, arguments);
};
BlockDb.parent = ThisParent;
BlockDb.prototype.close = function(cb) {
db.close(cb);
@ -37,42 +45,139 @@ BlockDb.prototype.drop = function(cb) {
});
};
// adds a block. Does not update Next pointer in
// the block prev to the new block, nor TIP pointer
//
BlockDb.prototype.add = function(b, height, cb) {
var self = this;
BlockDb.prototype._addBlockScript = function(b, height) {
var time_key = TIMESTAMP_PREFIX +
( b.time || Math.round(new Date().getTime() / 1000) );
return db.batch()
.put(time_key, b.hash)
.put(MAIN_PREFIX + b.hash, height)
.put(PREV_PREFIX + b.hash, b.previousblockhash)
.write(function(err){
if (!err) {
self.emit('new_block', {blockid: b.hash});
}
cb(err);
return [
{
type: 'put',
key: time_key,
value: b.hash,
},
{
type: 'put',
key: MAIN_PREFIX + b.hash,
value: height,
},
{
type: 'put',
key:PREV_PREFIX + b.hash,
value: b.previousblockhash,
},
];
};
BlockDb.prototype._delTxsScript = function(txs) {
var dbScript =[];
for(var ii in txs){
dbScript.push({
type: 'del',
key: IN_BLK_PREFIX + txs[ii],
});
}
return dbScript;
};
BlockDb.prototype._addTxsScript = function(txs, hash, height) {
var dbScript =[];
for(var ii in txs){
dbScript.push({
type: 'put',
key: IN_BLK_PREFIX + txs[ii],
value: hash+':'+height,
});
}
return dbScript;
};
// Returns blockHash and height for a given txId (If the tx is on the MAIN chain).
BlockDb.prototype.getBlockForTx = function(txId, cb) {
db.get(IN_BLK_PREFIX + txId,function (err, val) {
if (err && err.notFound) return cb();
if (err) return cb(err);
var v = val.split(':');
return cb(err,v[0],parseInt(v[1]));
});
};
BlockDb.prototype._changeBlockHeight = function(hash, height, cb) {
var self = this;
var dbScript1 = this._setHeightScript(hash,height);
d('Getting TXS FROM %s to set it Main', hash);
this.fromHashWithInfo(hash, function(err, bi) {
if (!bi || !bi.info || !bi.info.tx)
throw new Error('unable to get info for block:'+ hash);
var dbScript2;
if (height>=0) {
dbScript2 = self._addTxsScript(bi.info.tx, hash, height);
info('\t%s %d Txs', 'Confirming', bi.info.tx.length);
} else {
dbScript2 = self._delTxsScript(bi.info.tx);
info('\t%s %d Txs', 'Unconfirming', bi.info.tx.length);
}
db.batch(dbScript2.concat(dbScript1),cb);
});
};
BlockDb.prototype.setBlockMain = function(hash, height, cb) {
this._changeBlockHeight(hash,height,cb);
};
BlockDb.prototype.setBlockNotMain = function(hash, cb) {
this._changeBlockHeight(hash,-1,cb);
};
// adds a block (and its txs). Does not update Next pointer in
// the block prev to the new block, nor TIP pointer
//
BlockDb.prototype.add = function(b, height, cb) {
d('adding block %s #d', b,height);
var dbScript = this._addBlockScript(b,height);
dbScript = dbScript.concat(this._addTxsScript(b.tx,b.hash, height));
this.txDb.addMany(b.tx, function(err) {
if (err) return cb(err);
db.batch(dbScript,cb);
});
};
BlockDb.prototype.getTip = function(cb) {
if (this.cachedTip){
var v = this.cachedTip.split(':');
return cb(null,v[0], parseInt(v[1]));
}
var self = this;
db.get(TIP, function(err, val) {
if (!val) return cb();
self.cachedTip = val;
var v = val.split(':');
return cb(err,v[0], parseInt(v[1]));
});
};
BlockDb.prototype.setTip = function(hash, height, cb) {
console.log('[BlockDb.js.75] TIP', hash, height); //TODO
db.put(TIP, hash + ':' + height, function(err) {
this.cachedTip = hash + ':' + height;
db.put(TIP, this.cachedTip, function(err) {
return cb(err);
});
};
BlockDb.prototype.getDepth = function(hash, cb) {
var v = this.cachedTip.split(':');
if (!v) throw new Error('getDepth called with not cachedTip');
this.getHeight(hash, function(err,h){
return cb(err,parseInt(v[1]) - h);
});
};
//mainly for testing
BlockDb.prototype.setPrev = function(hash, prevHash, cb) {
db.put(PREV_PREFIX + hash, prevHash, function(err) {
@ -119,11 +224,13 @@ BlockDb.prototype.getHeight = function(hash, cb) {
});
};
BlockDb.prototype.setHeight = function(hash, height, cb) {
if (!height) console.log('\tNew orphan: %s',hash);
db.put(MAIN_PREFIX + hash, height, function(err) {
return cb(err);
});
BlockDb.prototype._setHeightScript = function(hash, height) {
d('setHeight: %s #%d', hash,height);
return ([{
type: 'put',
key: MAIN_PREFIX + hash,
value: height,
}]);
};
BlockDb.prototype.setNext = function(hash, nextHash, cb) {

161
lib/HistoricSync.js
View File

@ -14,12 +14,28 @@ var Sync = require('./Sync');
var sockets = require('../app/controllers/socket.js');
var BlockExtractor = require('./BlockExtractor.js');
var buffertools = require('buffertools');
var Address = bitcore.Address;
var bitcoreUtil = bitcore.util;
var Bignum = bitcore.Bignum;
var Script = bitcore.Script;
// This is 0.1.2 = > c++ version of base58-native
var base58 = require('base58-native').base58Check;
var encodedData = require('soop').load('bitcore/util/EncodedData',{
base58: base58
});
var versionedData= require('soop').load('bitcore/util/VersionedData',{
parent: encodedData
});
var Address = require('soop').load('bitcore/lib/Address',{
parent: versionedData
});
var logger = require('./logger').logger;
var d = logger.log;
var info = logger.info;
var error = logger.error;
// var bitcoreUtil = require('bitcore/util/util');
// var Deserialize = require('bitcore/Deserialize');
var BAD_GEN_ERROR = 'Bad genesis block. Network mismatch between Insight and bitcoind? Insight is configured for:';
var BAD_GEN_ERROR_DB = 'Bad genesis block. Network mismatch between Insight and levelDB? Insight is configured for:';
@ -41,35 +57,26 @@ function HistoricSync(opts) {
this.sync = new Sync(opts);
}
function p() {
var args = [];
Array.prototype.push.apply(args, arguments);
args.unshift('[historic_sync]');
/*jshint validthis:true */
console.log.apply(this, args);
}
HistoricSync.prototype.showProgress = function() {
var self = this;
if ( self.status ==='syncing' &&
( self.syncedBlocks ) % self.step !== 1) return;
if (self.error) {
p('ERROR: ' + self.error);
}
if (self.error)
error(self.error);
else {
self.updatePercentage();
p(util.format('status: [%d%%]', self.syncPercentage));
info(util.format('status: [%d%%]', self.syncPercentage));
}
if (self.shouldBroadcast) {
sockets.broadcastSyncInfo(self.info());
}
// if (self.syncPercentage > 10) {
// process.exit(-1);
// }
//
// if (self.syncPercentage > 10) {
// process.exit(-1);
// }
};
@ -133,6 +140,61 @@ HistoricSync.prototype.getBlockFromRPC = function(cb) {
});
};
HistoricSync.prototype._fromBuffer = function (buf) {
var buf2 = buffertools.reverse(buf);
return parseInt(buf2.toString('hex'), 16);
};
HistoricSync.prototype.getStandardizedTx = function (tx, time) {
var self = this;
tx.txid = bitcoreUtil.formatHashFull(tx.getHash());
var ti=0;
tx.vin = tx.ins.map(function(txin) {
var ret = {n: ti++};
if (txin.isCoinBase()) {
ret.isCoinBase = true;
} else {
ret.txid = buffertools.reverse(new Buffer(txin.getOutpointHash())).toString('hex');
ret.vout = txin.getOutpointIndex();
}
return ret;
});
var to = 0;
tx.vout = tx.outs.map(function(txout) {
var val;
if (txout.s) {
var s = new Script(txout.s);
var addrs = new Address.fromScriptPubKey(s, config.network);
// support only for p2pubkey p2pubkeyhash and p2sh
if (addrs && addrs.length === 1) {
val = {addresses: [addrs[0].toString() ] };
}
}
return {
valueSat: self._fromBuffer(txout.v),
scriptPubKey: val,
n: to++,
};
});
tx.time = time;
return tx;
};
HistoricSync.prototype.getStandardizedBlock = function(b) {
var self = this;
var block = {
hash: bitcoreUtil.formatHashFull(b.getHash()),
previousblockhash: bitcoreUtil.formatHashFull(b.prev_hash),
time: b.timestamp,
};
block.tx = b.txs.map(function(tx){
return self.getStandardizedTx(tx, b.timestamp);
});
return block;
};
HistoricSync.prototype.getBlockFromFile = function(cb) {
var self = this;
@ -141,36 +203,7 @@ HistoricSync.prototype.getBlockFromFile = function(cb) {
//get Info
self.blockExtractor.getNextBlock(function(err, b) {
if (err || ! b) return cb(err);
blockInfo = b.getStandardizedObject(b.txs, self.network);
blockInfo.previousblockhash = blockInfo.prev_block;
var ti=0;
// Get TX Address
b.txs.forEach(function(t) {
var objTx = blockInfo.tx[ti++];
//add time from block
objTx.time = blockInfo.time;
var to=0;
t.outs.forEach( function(o) {
try {
var s = new Script(o.s);
var addrs = new Address.fromScriptPubKey(s, config.network);
// support only for p2pubkey p2pubkeyhash and p2sh
if (addrs.length === 1) {
objTx.out[to].addrStr = addrs[0];
}
} catch (e) {
console.log('WARN Could not processs: ' + objTx.txid ,e);
}
to++;
});
});
blockInfo = self.getStandardizedBlock(b);
self.sync.bDb.setLastFileIndex(self.blockExtractor.currentFileIndex, function(err) {
return cb(err,blockInfo);
});
@ -235,7 +268,7 @@ HistoricSync.prototype.updateStartBlock = function(next) {
self.sync.bDb.fromHashWithInfo(tip, function(err, bi) {
blockInfo = bi ? bi.info : {};
if (oldtip)
self.sync.setBlockHeight(oldtip, false, cb);
self.sync.setBlockHeight(oldtip, -1, cb);
else
return cb();
});
@ -249,16 +282,18 @@ HistoricSync.prototype.updateStartBlock = function(next) {
}
else {
oldtip = tip;
if (!tip)
throw new Error('Previous blockchain tip was not found on bitcoind. Please reset Insight DB. Tip was:'+tip)
tip = blockInfo.previousblockhash;
assert(tip);
p('Previous TIP is now orphan. Back to:' + tip);
info('Previous TIP is now orphan. Back to:' + tip);
ret = true;
}
return ret;
},
function(err) {
self.startBlock = tip;
p('Resuming sync from block:'+tip);
info('Resuming sync from block:'+tip);
return next(err);
}
);
@ -275,7 +310,7 @@ HistoricSync.prototype.prepareFileSync = function(opts, next) {
try {
self.blockExtractor = new BlockExtractor(config.bitcoind.dataDir, config.network);
} catch (e) {
p(e.message + '. Disabling file sync.');
info(e.message + '. Disabling file sync.');
return next();
}
@ -288,7 +323,7 @@ HistoricSync.prototype.prepareFileSync = function(opts, next) {
var h = self.genesis;
p('Seeking file to:' + self.startBlock);
info('Seeking file to:' + self.startBlock);
//forward till startBlock
async.whilst(
function() {
@ -320,18 +355,18 @@ HistoricSync.prototype.prepareRpcSync = function(opts, next) {
HistoricSync.prototype.showSyncStartMessage = function() {
var self = this;
p('Got ' + self.connectedCountDB +
info('Got ' + self.connectedCountDB +
' blocks in current DB, out of ' + self.blockChainHeight + ' block at bitcoind');
if (self.blockExtractor) {
p('bitcoind dataDir configured...importing blocks from .dat files');
p('First file index: ' + self.blockExtractor.currentFileIndex);
info('bitcoind dataDir configured...importing blocks from .dat files');
info('First file index: ' + self.blockExtractor.currentFileIndex);
}
else {
p('syncing from RPC (slow)');
info('syncing from RPC (slow)');
}
p('Starting from: ', self.startBlock);
info('Starting from: ', self.startBlock);
self.showProgress();
};
@ -389,7 +424,7 @@ HistoricSync.prototype.start = function(opts, next) {
var self = this;
if (self.status==='starting' || self.status==='syncing') {
p('## Wont start to sync while status is %s', self.status);
error('## Wont start to sync while status is %s', self.status);
return next();
}

6
lib/PeerSync.js
View File

@ -63,9 +63,9 @@ PeerSync.prototype.handleTx = function(info) {
if (self.shouldBroadcast) {
sockets.broadcastTx(tx);
if (tx.addrsToEmit) {
tx.addrsToEmit.forEach(function(a) {
sockets.broadcastAddressTx(a, tx.txid);
});
for(var ii in tx.addrsToEmit){
sockets.broadcastAddressTx(tx.addrsToEmit[ii], tx.txid);
}
}
}
}

140
lib/Sync.js
View File

@ -7,6 +7,12 @@ var bitcore = require('bitcore');
var networks = bitcore.networks;
var async = require('async');
var logger = require('./logger').logger;
var d = logger.log;
var info = logger.info;
var syncId = 0;
function Sync(opts) {
@ -15,6 +21,7 @@ function Sync(opts) {
this.bDb = require('./BlockDb').default();
this.txDb = require('./TransactionDb').default();
this.network = config.network === 'testnet' ? networks.testnet : networks.livenet;
this.cachedLastHash = null;
}
Sync.prototype.close = function(cb) {
@ -56,7 +63,8 @@ Sync.prototype.destroy = function(next) {
* \
* F-G-NEW
* 1) Set F-G as connected (mark TXs as valid)
* 2) Declare D-E orphans (and possible invalidate TXs on them)
* 2) Set new heights in F-G-NEW
* 3) Declare D-E orphans (and possible invalidate TXs on them)
*
*
* Case 3)
@ -77,22 +85,22 @@ Sync.prototype.storeTipBlock = function(b, allowReorgs, cb) {
if (!b) return cb();
var self = this;
var oldTip, oldNext, height, needReorg = false;
var oldTip, oldNext, oldHeight, needReorg = false, height = -1;
var newPrev = b.previousblockhash;
async.series([
function(c) {
// TODO? remove this check?
self.bDb.has(b.hash, function(err, val) {
return c(err ||
(val ? new Error('WARN: Ignoring already existing block:' + b.hash) : null));
});
},
function(c) {
if (!allowReorgs) return c();
if (!allowReorgs || newPrev === self.cachedLastHash) return c();
self.bDb.has(newPrev, function(err, val) {
// Genesis? no problem
if (!val && newPrev.match(/^0+$/)) return c();
return c(err ||
(!val ? new Error('NEED_SYNC Ignoring block with non existing prev:' + b.hash) : null));
});
@ -101,8 +109,8 @@ Sync.prototype.storeTipBlock = function(b, allowReorgs, cb) {
if (!allowReorgs) return c();
self.bDb.getTip(function(err, hash, h) {
oldTip = hash;
if (!hash) height = -1
else height = h || 0;
oldHeight = hash ? (h || 0) : -1
if (oldTip && newPrev !== oldTip) {
needReorg = true;
console.log('## REORG Triggered, tip mismatch');
@ -110,11 +118,7 @@ Sync.prototype.storeTipBlock = function(b, allowReorgs, cb) {
return c();
});
},
function(c) {
self.txDb.createFromBlock(b, height, function(err) {
return c(err);
});
},
function(c) {
if (!needReorg) return c();
self.bDb.getNext(newPrev, function(err, val) {
@ -123,17 +127,29 @@ Sync.prototype.storeTipBlock = function(b, allowReorgs, cb) {
return c();
});
},
function(c) {
self.bDb.add(b, height + 1, c);
function(c) {
if (!allowReorgs) return c();
if (needReorg) {
info('NEW TIP: %s NEED REORG (old tip: %s #%d)', b.hash, oldTip, oldHeight);
self.processReorg(oldTip, oldNext, newPrev, oldHeight, function(err, h) {
if (err) throw err;
height = h;
return c();
});
}
else {
height = oldHeight + 1;
return c();
}
},
function(c) {
if (!needReorg) return c();
console.log('NEW TIP: %s #%d NEED REORG (old tip: %s)', b.hash, height + 1, oldTip);
self.processReorg(oldTip, oldNext, newPrev, height + 1, c);
self.cachedLastHash = b.hash; // just for speed up.
self.bDb.add(b, height, c);
},
function(c) {
if (!allowReorgs) return c();
self.bDb.setTip(b.hash, height + 1, function(err) {
self.bDb.setTip(b.hash, height, function(err) {
return c(err);
});
},
@ -152,21 +168,20 @@ Sync.prototype.storeTipBlock = function(b, allowReorgs, cb) {
});
};
Sync.prototype.processReorg = function(oldTip, oldNext, newPrev, newHeight, cb) {
Sync.prototype.processReorg = function(oldTip, oldNext, newPrev, oldHeight, cb) {
var self = this;
var orphanizeFrom;
var orphanizeFrom, newHeight;
async.series([
function(c) {
self.bDb.getHeight(newPrev, function(err, height) {
if (!height) return c();
if (!height) return c(new Error('Could not found block:' + newPrev));
if (height<0) return c();
console.log('# Reorg Case 1)');
// case 1
newHeight = height + 1;
info('# Reorg Case 1) OldNext: %s NewHeight: %d', oldNext, newHeight);
orphanizeFrom = oldNext;
return c(err);
});
@ -174,10 +189,12 @@ Sync.prototype.processReorg = function(oldTip, oldNext, newPrev, newHeight, cb)
function(c) {
if (orphanizeFrom) return c();
console.log('# Reorg Case 2)');
self.setBranchConnectedBackwards(newPrev, newHeight-1, function(err, yHash, newYHashNext) {
info('# Reorg Case 2)');
self.setBranchConnectedBackwards(newPrev, function(err, yHash, newYHashNext, height) {
if (err) return c(err);
newHeight = height;
self.bDb.getNext(yHash, function(err, yHashNext) {
// Connect the new branch, and orphanize the old one.
orphanizeFrom = yHashNext;
self.bDb.setNext(yHash, newYHashNext, function(err) {
return c(err);
@ -187,26 +204,17 @@ Sync.prototype.processReorg = function(oldTip, oldNext, newPrev, newHeight, cb)
},
function(c) {
if (!orphanizeFrom) return c();
self.setBranchOrphan(orphanizeFrom, function(err) {
self._setBranchOrphan(orphanizeFrom, function(err) {
return c(err);
});
},
],
function(err) {
return cb(err);
return cb(err, newHeight);
});
};
//height = false => unconnnected
Sync.prototype.setBlockHeight = function(hash, height, cb) {
var self = this;
self.bDb.setHeight(hash, height, function(err) {
if (err) return cb(err);
return self.txDb.handleBlockChange(hash, height>-1? true : false, cb);
});
};
Sync.prototype.setBranchOrphan = function(fromHash, cb) {
Sync.prototype._setBranchOrphan = function(fromHash, cb) {
var self = this,
hashInterator = fromHash;
@ -215,7 +223,7 @@ Sync.prototype.setBranchOrphan = function(fromHash, cb) {
return hashInterator;
},
function(c) {
self.setBlockHeight(hashInterator, false, function(err) {
self.bDb.setBlockNotMain(hashInterator, function(err) {
if (err) return cb(err);
self.bDb.getNext(hashInterator, function(err, val) {
hashInterator = val;
@ -225,45 +233,53 @@ Sync.prototype.setBranchOrphan = function(fromHash, cb) {
}, cb);
};
Sync.prototype.setBranchConnectedBackwards = function(fromHash, initialHeight, cb) {
Sync.prototype.setBranchConnectedBackwards = function(fromHash, cb) {
//console.log('[Sync.js.219:setBranchConnectedBackwards:]',fromHash); //TODO
var self = this,
hashInterator = fromHash,
lastHash = fromHash,
isMain;
yHeight,
branch = [];
async.doWhilst(
function(c) {
self.setBlockHeight(hashInterator, initialHeight--, function(err) {
branch.unshift(hashInterator);
self.bDb.getPrev(hashInterator, function(err, val) {
if (err) return c(err);
self.bDb.getPrev(hashInterator, function(err, val) {
if (err) return c(err);
lastHash = hashInterator;
hashInterator = val;
self.bDb.getHeight(hashInterator, function(err, height) {
isMain = height ? 1 : 0;
return c();
});
lastHash = hashInterator;
hashInterator = val;
self.bDb.getHeight(hashInterator, function(err, height) {
yHeight = height;
return c();
});
});
},
function() {
return hashInterator && !isMain;
return hashInterator && !yHeight;
},
function(err) {
console.log('\tFound yBlock:', hashInterator);
return cb(err, hashInterator, lastHash);
}
);
function() {
info('\tFound yBlock: %s #%d', hashInterator, yHeight);
var heightIter = yHeight + 1;
var hashIter;
async.whilst(
function() {
hashIter = branch.shift();
return hashIter;
},
function(c) {
self.setBlockMain(hashIter, heightIter++, c);
},
function(err) {
return cb(err, hashInterator, lastHash, heightIter);
});
});
};
//Store unconfirmed TXs
Sync.prototype.storeTxs = function(txs, cb) {
var self = this;
self.txDb.createFromArray(txs, null, null, function(err) {
if (err) return cb(err);
return cb(err);
});
this.txDb.addMany(txs, cb);
};

299
lib/TransactionDb.js
View File

@ -1,15 +1,7 @@
'use strict';
var imports = require('soop').imports();
var PoolMatch = imports.poolMatch || require('soop').load('./PoolMatch',config);
// txid - blockhash mapping (only for confirmed txs)
var IN_BLK_PREFIX = 'txb-'; //txb-<txid> = <block>
var RECENT_ = 'tx-'; //tx-<block> = <txid>
pre-<height>- =txid
// to show tx outs
var OUTS_PREFIX = 'txo-'; //txo-<txid>-<n> => [addr, btc_sat]
@ -37,19 +29,13 @@ var bitcore = require('bitcore'),
config = require('../config/config'),
assert = require('assert');
var logger = require('./logger').logger;
var d = logger.log;
var info = logger.info;
var warn = logger.warn;
var db = imports.db || levelup(config.leveldb + '/txs',{maxOpenFiles: MAX_OPEN_FILES} );
var Script = bitcore.Script;
// This is 0.1.2 = > c++ version of base57-native
var base58 = require('base58-native').base58Check;
var encodedData = require('soop').load('bitcore/util/EncodedData',{
base58: base58
});
var versionedData= require('soop').load('bitcore/util/VersionedData',{
parent: encodedData
});
var Address = require('soop').load('bitcore/lib/Address',{
parent: versionedData
});
var PoolMatch = imports.poolMatch || require('soop').load('./PoolMatch',config);
var TransactionDb = function() {
TransactionDb.super(this, arguments);
@ -195,9 +181,8 @@ TransactionDb.prototype._fillOutpoints = function(info, cb) {
async.eachLimit(info.vin, CONCURRENCY, function(i, c_in) {
self.fromTxIdN(i.txid, i.vout, info.confirmations, function(err, ret) {
//console.log('[TransactionDb.js.154:ret:]',ret); //TODO
if (!ret || !ret.addr || !ret.valueSat) {
console.log('Could not get TXouts in %s,%d from %s ', i.txid, i.vout, info.txid);
info('Could not get TXouts in %s,%d from %s ', i.txid, i.vout, info.txid);
if (ret) i.unconfirmedInput = ret.unconfirmedInput;
incompleteInputs = 1;
return c_in(); // error not scalated
@ -258,7 +243,6 @@ TransactionDb.prototype._getInfo = function(txid, next) {
Rpc.getTxInfo(txid, function(err, info) {
if (err) return next(err);
self._fillOutpoints(info, function() {
self._fillSpent(info, function() {
return next(null, info);
@ -341,10 +325,13 @@ TransactionDb.prototype.fromTxIdN = function(txid, n, confirmations, cb) {
TransactionDb.prototype.fillConfirmations = function(o, cb) {
var self = this;
self.getHeight(o.txid, function(err, height) {
console.log('[TransactionDb.js.339]'); //TODO
self.getBlock(o.txid, function(err, hash) {
console.log('[TransactionDb.js.342]'); //TODO
if (err) return cb(err);
o.height = height;
o.isConfirmed = hash?1:0;
if (!o.spentTxId) return cb();
if (o.multipleSpentAttempts) {
@ -352,20 +339,20 @@ TransactionDb.prototype.fillConfirmations = function(o, cb) {
//TODO save it for later is height > 6
async.eachLimit(o.multipleSpentAttempts, CONCURRENCY,
function(oi, e_c) {
self.getHeight(oi.spentTxId, function(err, height) {
self.getBlock(oi.spentTxId, function(err, hash) {
if (err) return;
if (height) {
if (hash) {
o.spentTxId = oi.spentTxId;
o.index = oi.index;
o.spentHeight = height;
o.spentIsConfirmed = 1;
}
return e_c();
});
}, cb);
} else {
self.getHeight(o.spentTxId, function(err, height) {
self.getBlock(o.spentTxId, function(err, hash) {
if (err) return cb(err);
o.spentHeight = height;
o.spentIsConfirmed = hash?1:0;
return cb();
});
}
@ -459,215 +446,95 @@ TransactionDb.prototype.removeFromTxId = function(txid, cb) {
};
TransactionDb.prototype.adaptTxObject = function(txInfo) {
var self = this;
// adapt bitcore TX object to bitcoind JSON response
txInfo.txid = txInfo.hash;
TransactionDb.prototype._addScript = function(tx) {
var relatedAddrs = [];
var dbScript = [];
var ts = tx.time;
var txid = tx.txid;
var to = 0;
var tx = txInfo;
if (tx.outs) {
tx.outs.forEach(function(o) {
var s = new Script(o.s);
var addrs = new Address.fromScriptPubKey(s, config.network);
// support only for p2pubkey p2pubkeyhash and p2sh
if (addrs.length === 1) {
tx.out[to].addrStr = addrs[0].toString();
tx.out[to].n = to;
}
to++;
});
// Input Outpoints (mark them as spent)
if (!tx.isCoinBase){
for(var ii in tx.vin) {
var i = tx.vin[ii];
dbScript.push({
type: 'put',
key: SPENT_PREFIX + i.txid + '-' + i.vout + '-' + txid + '-' + i.n,
value: ts || 0,
});
}
}
var count = 0;
txInfo.vin = txInfo. in .map(function(txin) {
var i = {};
for(var ii in tx.vout) {
var o = tx.vout[ii];
if ((o.value||o.valueSat) &&
o.scriptPubKey &&
o.scriptPubKey.addresses &&
o.scriptPubKey.addresses[0] && !o.scriptPubKey.addresses[1] // TODO : not supported=> standard multisig
) {
var addr = o.scriptPubKey.addresses[0];
var sat = o.valueSat || (o.value * util.COIN).toFixed(0);
if (txin.coinbase) {
txInfo.isCoinBase = true;
} else {
i.txid = txin.prev_out.hash;
i.vout = txin.prev_out.n;
}
i.n = count++;
return i;
});
count = 0;
txInfo.vout = txInfo.out.map(function(txout) {
var o = {};
o.value = txout.value;
o.n = count++;
if (txout.addrStr) {
o.scriptPubKey = {};
o.scriptPubKey.addresses = [txout.addrStr];
}
return o;
});
relatedAddrs[addr]=1;
var k = OUTS_PREFIX + txid + '-' + o.n;
dbScript.push({
type: 'put',
key: k,
value: addr + ':' + sat,
},{
type: 'put',
key: ADDR_PREFIX + addr + '-' + txid + '-' + o.n,
value: sat + ':' + ts,
});
}
}
tx.relatedAddrs=relatedAddrs;
return dbScript;
};
TransactionDb.prototype.add = function(tx, blockhash, height, cb) {
var self = this;
var addrs = [];
if (typeof height === 'undefined')
throw new Error('add should received height');
if (tx.hash) self.adaptTxObject(tx);
var ts = tx.time;
async.series([
// Input Outpoints (mark them as spent)
function(p_c) {
if (tx.isCoinBase) return p_c();
async.forEachLimit(tx.vin, CONCURRENCY,
function(i, next_out) {
db.batch()
.put(SPENT_PREFIX + i.txid + '-' + i.vout + '-' + tx.txid + '-' + i.n,
ts || 0)
.write(next_out);
},
function(err) {
return p_c(err);
});
},
// Parse Outputs
function(p_c) {
async.forEachLimit(tx.vout, CONCURRENCY,
function(o, next_out) {
if (o.value && o.scriptPubKey &&
o.scriptPubKey.addresses &&
o.scriptPubKey.addresses[0] && !o.scriptPubKey.addresses[1] // TODO : not supported
) {
var addr = o.scriptPubKey.addresses[0];
var sat = Math.round(o.value * util.COIN);
if (addrs.indexOf(addr) === -1) {
addrs.push(addr);
}
// existed?
var k = OUTS_PREFIX + tx.txid + '-' + o.n;
db.get(k, function(err, val) {
if (!val || (err && err.notFound)) {
db.batch()
.put(k, addr + ':' + sat)
.put(ADDR_PREFIX + addr + '-' + tx.txid + '-' + o.n, sat + ':' + ts)
.write(next_out);
} else {
return next_out();
}
});
} else {
return next_out();
}
},
function(err) {
if (err) {
console.log('ERR at TX %s: %s', tx.txid, err);
return cb(err);
}
return p_c();
});
},
function(p_c) {
if (!blockhash) {
return p_c();
}
return self.setBlock(tx.txid, blockhash, p_c);
},
], function(err) {
if (addrs.length > 0 && !blockhash) {
// only emit if we are processing a single tx (not from a block)
tx.addrsToEmit=addrs;
}
return cb(err);
});
TransactionDb.prototype.add = function(tx, blockhash, cb) {
var dbScript = this._addScript(tx, blockhash);
db.batch(dbScript, cb);
};
TransactionDb.prototype._addManyFromObjs = function(txs, next) {
var dbScript = [];
TransactionDb.prototype.setBlock = function(txId, blockHash, c) {
if (!blockHash) return c();
db.batch()
.put(IN_BLK_PREFIX + txId, blockHash)
.write(c);
for(var ii in txs){
var s = this._addScript(txs[ii]);
dbScript = dbScript.concat(s);
}
db.batch(dbScript, next);
};
TransactionDb.prototype.getBlock = function(txId, c) {
db.get(IN_BLK_PREFIX + txId,c);
};
TransactionDb.prototype._addManyFromHashes = function(txs, next) {
var self=this;
var dbScript = [];
async.eachLimit(txs, CONCURRENCY, function(tx, each_cb) {
if (tx === genesisTXID)
return each_cb();
TransactionDb.prototype.handleBlockChange = function(hash, isConnected, cb) {
var toChange = [];
console.log('\tSearching Txs from block:' + hash);
var k = FROM_BLK_PREFIX + hash;
var k2 = IN_BLK_PREFIX;
db.createReadStream({
start: k,
end: k + '~'
})
.on('data', function(data) {
var ks = data.key.split('-');
toChange.push({
key: k2 + ks[2] + '-' + ks[1],
type: 'put',
value: height,
});
})
.on('error', function(err) {
return cb(err);
})
.on('end', function(err) {
if (err) return cb(err);
console.log('\t%s %d Txs', height ? 'Confirming' : 'Unconfirming', toChange.length);
db.batch(toChange, cb);
Rpc.getTxInfo(tx, function(err, inInfo) {
if (!inInfo) return each_cb(err);
dbScript = dbScript.concat(self._addScript(inInfo));
return each_cb();
});
};
// txs can be a [hashes] or [txObjects]
TransactionDb.prototype.createFromArray = function(txs, blockHash, height, next) {
var self = this;
if (!txs) return next();
async.forEachLimit(txs, CONCURRENCY, function(t, each_cb) {
if (typeof t === 'string') {
// TODO: parse it from networks.genesisTX?
if (t === genesisTXID) return each_cb();
Rpc.getTxInfo(t, function(err, inInfo) {
if (!inInfo) return each_cb(err);
return self.add(inInfo, blockHash, height, each_cb);
});
} else {
return self.add(t, blockHash, height, each_cb);
}
},
function(err) {
return next(err);
if (err) return next(err);
db.batch(dbScript,next);
});
};
TransactionDb.prototype.createFromBlock = function(b, height, next) {
var self = this;
if (!b || !b.tx) return next();
TransactionDb.prototype.addMany = function(txs, next) {
if (!txs) return next();
if (typeof height === 'undefined')
throw new Error('createFromBlock should received height');
var fn = (typeof txs[0] ==='string') ?
this._addManyFromHashes : this._addManyFromObjs;
return self.createFromArray(b.tx, b.hash, height, next);
return fn.apply(this,[txs, next]);
};

3
package.json
View File

@ -60,13 +60,13 @@
"soop": "=0.1.5",
"commander": "*",
"bignum": "*",
"winston": "*",
"express": "~3.4.7",
"buffertools": "*",
"should": "~2.1.1",
"socket.io": "~0.9.16",
"moment": "~2.5.0",
"sinon": "~1.7.3",
"chai": "~1.8.1",
"xmlhttprequest": "~1.6.0",
"bufferput": "git://github.com/bitpay/node-bufferput.git"
},
@ -80,6 +80,7 @@
"grunt-nodemon": "~0.2.0",
"grunt-mocha-test": "~0.8.1",
"should": "2.1.1",
"chai": "=1.9.1",
"grunt-markdown": "~0.5.0"
}
}

8
test/integration/01-transactionouts.js
View File

@ -29,8 +29,12 @@ describe('TransactionDb fromIdWithInfo', function(){
assert.equal(tx.txid, txid);
assert(!tx.info.isCoinBase);
for(var i=0; i<20; i++)
for(var i=0; i<20; i++) {
assert(parseFloat(tx.info.vin[i].value) === parseFloat(50), 'input '+i);
}
console.log('[01-transactionouts.js.34:tx:]',tx.info.vin[0]); //TODO
assert(tx.info.vin[0].addr === 'msGKGCy2i8wbKS5Fo1LbWUTJnf1GoFFG59', 'addr 0');
assert(tx.info.vin[1].addr === 'mfye7oHsdrHbydtj4coPXCasKad2eYSv5P', 'addr 1');
done();
@ -134,7 +138,7 @@ describe('TransactionDb Outs', function(){
assert.equal(readItems.length,0);
var unmatch=[];
txDb.createFromArray([v.txid], null, function(err) {
txDb.addMany([v.txid], function(err) {
if (err) return done(err);
txDb.fromTxId( v.txid, function(err, readItems) {

2
test/integration/02-transactionouts.js
View File

@ -27,7 +27,7 @@ describe('TransactionDb Expenses', function(){
function(txid,c_out) {
async.each(spentValid[txid],
function(i,c_in) {
txDb.createFromArray([i.txid], null, function(err) {
txDb.addMany([i.txid], function(err) {
return c_in();
});
},

954
test/integration/99-sync.js.descructive-test
View File

File diff suppressed because it is too large Load Diff