Added regtest reorg tests,

lib/scaffold/start.js

@@ -102,6 +102,7 @@ function lookInBuiltInPath(req, service) {
     var serviceFile = path.resolve(__dirname, '../services/' + service.name);
     return req(serviceFile);
   } catch(e) {
+console.log(e);
     if(e.code !== 'MODULE_NOT_FOUND') {
       log.error(e);
     }

lib/services/address/index.js

@@ -11,6 +11,7 @@ var _ = bitcore.deps._;
 var Encoding = require('./encoding');
 var utils = require('../../utils');
 var Transform = require('stream').Transform;
+var assert = require('assert');
 
 var AddressService = function(options) {
   BaseService.call(this, options);
@@ -270,7 +271,6 @@ AddressService.prototype.start = function(callback) {
       return callback(err);
     }
     self._encoding = new Encoding(prefix);
-    self._startSubscriptions();
     callback();
   });
 };
@@ -371,80 +371,67 @@ AddressService.prototype._getAddressHistory = function(address, options, callbac
 
 };
 
-AddressService.prototype._startSubscriptions = function() {
+AddressService.prototype.onReorg = function(args, callback) {
 
-  if (this._subscribed) {
+  var self = this;
-    return;
-  }
 
-  this._subscribed = true;
+  var oldBlockList = args[1];
-  if (!this._bus) {
-    this._bus = this.node.openBus({remoteAddress: 'localhost'});
-  }
-
-  this._bus.on('block/reorg', this._onReorg.bind(this));
-  this._bus.subscribe('block/reorg');
-};
-
-AddressService.prototype._onReorg = function(commonAncestorHeader, oldBlockList) {
-
-  // if the common ancestor block height is greater than our own, then nothing to do for the reorg
-  if (this._tip.height <= commonAncestorHeader.height) {
-    return;
-  }
-
-  // set the tip to the common ancestor in case something goes wrong with the reorg
-  var tipOps = utils.encodeTip({ hash: commonAncestorHeader.hash, height: commonAncestorHeader.height }, this.name);
-
-  var removalOps = [{
-    type: 'put',
-    key: tipOps.key,
-    value: tipOps.value
-  }];
 
+  var removalOps = [];
   // for every tx, remove the address index key for every input and output
+  // TODO: DRY self up!
   for(var i = 0; i < oldBlockList.length; i++) {
+
     var block = oldBlockList[i];
     //txs
-    for(var j = 0; j < block.transactions.length; j++) {
+    for(var j = 0; j < block.txs.length; j++) {
-      var tx = block.transactions[j];
+
+      var tx = block.txs[j];
 
       //inputs
       var address;
+
       for(var k = 0; k < tx.inputs.length; k++) {
+
         var input = tx.inputs[k];
-        address = input.address;
+        address = input.getAddress();
 
         if (!address) {
           continue;
         }
 
+        address.network = self._network;
+        address = address.toString();
+
         removalOps.push({
           type: 'del',
-          key: this.encoding.encodeTransactionKey(address, block.height, tx.id, k, 1)
+          key: self._encoding.encodeAddressIndexKey(address, block.height, tx.txid(), k, 1, block.ts)
         });
       }
 
       //outputs
       for(k = 0; k < tx.outputs.length; k++) {
+
         var output = tx.outputs[k];
-        address = output.address;
+        address = output.getAddress();
 
         if (!address) {
           continue;
         }
 
+        address.network = self._network;
+        address = address.toString();
+
         removalOps.push({
           type: 'del',
-          key: this.encoding.encodeTransactionKey(address, block.height, tx.id, k, 0)
+          key: self._encoding.encodeAddressIndexKey(address, block.height, tx.txid(), k, 0, block.ts)
         });
 
       }
     }
   }
 
-  this._db.batch(removalOps);
+  callback(null, removalOps);
-
 };
 
 AddressService.prototype.onBlock = function(block, callback) {
@@ -477,8 +464,11 @@ AddressService.prototype._processInput = function(tx, input, opts) {
   address.network = this._network;
   address = address.toString();
   var txid = tx.txid();
+  var timestamp = this._timestamp.getTimestampSync(opts.block.rhash());
+  assert(timestamp, 'Must have a timestamp in order to process input.');
+
   // address index
-  var addressKey = this._encoding.encodeAddressIndexKey(address, opts.block.height, txid);
+  var addressKey = this._encoding.encodeAddressIndexKey(address, opts.block.height, txid, opts.inputIndex, 1, timestamp);
 
   var operations = [{
     type: 'put',
@@ -511,14 +501,19 @@ AddressService.prototype._processOutput = function(tx, output, index, opts) {
   address.network = this._network;
   address = address.toString();
   var txid = tx.txid();
-  var addressKey = this._encoding.encodeAddressIndexKey(address, opts.block.height, txid);
+  var timestamp = this._timestamp.getTimestampSync(opts.block.rhash());
+  assert(timestamp, 'Must have a timestamp in order to process output.');
+
+  var addressKey = this._encoding.encodeAddressIndexKey(address, opts.block.height, txid, opts.outputIndex, 0, timestamp);
+
   var utxoKey = this._encoding.encodeUtxoIndexKey(address, txid, index);
   var utxoValue = this._encoding.encodeUtxoIndexValue(
     opts.block.height,
     Unit.fromBTC(output.value).toSatoshis(),
-    this._timestamp.getTimestampSync(opts.block.rhash()),
+    timestamp,
     output.script.toRaw()
   );
+
   var operations = [{
     type: 'put',
     key: addressKey
@@ -546,7 +541,8 @@ AddressService.prototype._processTransaction = function(tx, opts) {
 
   outputOperations = _.flatten(_.compact(outputOperations));
 
-  var inputOperations = tx.inputs.map(function(input) {
+  var inputOperations = tx.inputs.map(function(input, index) {
+    _opts.inputIndex = index;
     return self._processInput(tx, input, _opts);
   });
 

lib/services/block/index.js

@@ -16,9 +16,10 @@ var BlockService = function(options) {
   BaseService.call(this, options);
 
   this._tip = null;
-  this._p2p = this.node.services.p2p;
   this._db = this.node.services.db;
+  this._p2p = this.node.services.p2p;
   this._header = this.node.services.header;
+  this._timestamp = this.node.services.timestamp;
 
   this._subscriptions = {};
   this._subscriptions.block = [];
@@ -31,7 +32,7 @@ var BlockService = function(options) {
 
 inherits(BlockService, BaseService);
 
-BlockService.dependencies = [ 'p2p', 'db', 'header' ];
+BlockService.dependencies = [ 'timestamp', 'p2p', 'db', 'header' ];
 
 // --- public prototype functions
 BlockService.prototype.getAPIMethods = function() {
@@ -112,12 +113,6 @@ BlockService.prototype.getPublishEvents = function() {
       scope: this,
       subscribe: this.subscribe.bind(this, 'block'),
       unsubscribe: this.unsubscribe.bind(this, 'block')
-    },
-    {
-      name: 'block/reorg',
-      scope: this,
-      subscribe: this.subscribe.bind(this, 'reorg'),
-      unsubscribe: this.unsubscribe.bind(this, 'reorg')
     }
   ];
 
@@ -228,7 +223,7 @@ BlockService.prototype._findCommonAncestor = function(hash, allHeaders, callback
     // test case
     function() {
 
-      return _oldTip !== _newTip || ++count <= allHeaders.size;
+      return _oldTip !== _newTip && ++count < allHeaders.size;
 
     },
     // get block
@@ -243,25 +238,41 @@ BlockService.prototype._findCommonAncestor = function(hash, allHeaders, callback
 
         // once we've found the old tip, we will find its prev and check to see if matches new tip's prev
         var block = self._encoding.decodeBlockValue(data);
-
+        // apply the block's height
-        // we will squirrel away the block because our services will need to remove it after we've found the common ancestor
+        var blockHdr = allHeaders.get(block.rhash());
-        oldBlocks.push(block);
+        if (!blockHdr) {
-
+          return next(new Error('Could not find block in list of headers: ' + block.rhash()));
-        // this is our current tip's prev hash
-        _oldTip = bcoin.util.revHex(block.prevBlock);
-
-        // our current headers have the correct state of the chain, so consult that for its prev hash
-        var header = allHeaders.get(_newTip);
-
-        if (!header) {
-          return next(new Error('Header missing from list of headers'));
         }
+        block.height = blockHdr.height;
+        assert(block.height >= 0, 'We mamaged to save a header with an incorrect height.');
 
-        // set new tip to the prev hash
+        // apply the block's timestamp
-        _newTip = header.prevHash;
+        self._timestamp.getTimestamp(block.rhash(), function(err, timestamp) {
 
-        next();
+          if (err || !timestamp) {
+            return next(err || new Error('missing timestamp'));
+          }
 
+          block.ts = timestamp;
+          // we will squirrel away the block because our services will need to remove it after we've found the common ancestor
+          oldBlocks.push(block);
+
+          // this is our current tip's prev hash
+          _oldTip = bcoin.util.revHex(block.prevBlock);
+
+          // our current headers have the correct state of the chain, so consult that for its prev aash
+          var header = allHeaders.get(_newTip);
+
+          if (!header) {
+            return next(new Error('Header missing from list of headers'));
+          }
+
+          // set new tip to the prev hash
+          _newTip = header.prevHash;
+
+          next();
+
+        });
       });
 
     }, function(err) {
@@ -319,7 +330,7 @@ BlockService.prototype._getHash = function(blockArg, callback) {
 
 };
 
-BlockService.prototype._handleReorg = function(hash, allHeaders) {
+BlockService.prototype._handleReorg = function(hash, allHeaders, block) {
 
   // hash is the hash of the new block that we are reorging to.
   assert(hash, 'We were asked to reorg to a non-existent hash.');
@@ -339,66 +350,27 @@ BlockService.prototype._handleReorg = function(hash, allHeaders) {
         ' (the forked block) could not be found. Bitcore-node must exit.');
 
       self.node.stop();
-
       return;
     }
 
     var commonAncestorHeader = allHeaders.get(commonAncestorHash);
-    log.warn('Block Service: A common ancestor block was found to at hash: ' + commonAncestorHeader);
 
-    self._broadcast(self.subscriptions.reorg, 'block/reorg', [commonAncestorHeader, oldBlocks]);
+    log.info('Block Service: A common ancestor block was found to at hash: ' + commonAncestorHeader.hash);
 
-    self._onReorg(commonAncestorHeader, oldBlocks);
+    self._processReorg(commonAncestorHeader, oldBlocks, block);
-
-    self._reorging = false;
-    self._sync();
 
   });
 
-};
-
-// get the blocks from our current tip to the given hash, non-inclusive
-BlockService.prototype._getOldBlocks = function(hash, callback) {
-
-  var blocks = [];
-
-  var _tip = this._tip.hash;
-
-  async.whilst(
-    function() {
-      return _tip !== hash;
-    },
-    function(next) {
-
-      this._get(this._encoding.encodeBlockKey(_tip), function(err, block) {
-
-        if (err) {
-          return callback(err);
-        }
-
-        if (!block) {
-          next(new Error('expected to find a block in database, but found none.'));
-          return;
-        }
-        blocks.push(block);
-        _tip = bcoin.util.revHex(block.prevBlock);
-      });
-    },
-    function(err) {
-      if (err) {
-        return callback(err);
-      }
-      callback(null, blocks);
-    });
 
 };
 
 // this JUST rewinds the chain back to the common ancestor block, nothing more
-BlockService.prototype._onReorg = function(commonAncestorHeader, oldBlockList) {
+BlockService.prototype._onReorg = function(commonAncestorHeader, oldBlockList, newBlock) {
 
   // set the tip to the common ancestor in case something goes wrong with the reorg
-  this._setTip({ hash: commonAncestorHeader.hash, height: commonAncestorHeader.height });
+  var self = this;
-  var tipOps = utils.encodeTip(this._tip, this.name);
+  self._setTip({ hash: commonAncestorHeader.hash, height: commonAncestorHeader.height });
+  var tipOps = utils.encodeTip(self._tip, self.name);
 
   var removalOps = [{
     type: 'put',
@@ -410,11 +382,18 @@ BlockService.prototype._onReorg = function(commonAncestorHeader, oldBlockList) {
   oldBlockList.forEach(function(block) {
     removalOps.push({
       type: 'del',
-      key: this.encoding.encodeBlockKey(block.rhash()),
+      key: self._encoding.encodeBlockKey(block.rhash()),
     });
   });
 
-  this._db.batch(removalOps);
+  self._db.batch(removalOps, function() {
+
+        self._reorging = false;
+
+        if (newBlock) {
+          self._onBlock(newBlock);
+        }
+  });
 
 };
 
@@ -423,6 +402,55 @@ BlockService.prototype._onAllHeaders = function() {
 };
 
 
+BlockService.prototype._processReorg = function(commonAncestorHeader, oldBlocks, newBlock) {
+
+  var self = this;
+  var operations = [];
+  var services = self.node.services;
+
+  async.eachSeries(
+    services,
+    function(mod, next) {
+      if(mod.onReorg) {
+        mod.onReorg.call(mod, [commonAncestorHeader, oldBlocks], function(err, ops) {
+          if (err) {
+            return next(err);
+          }
+          if (ops) {
+            operations = operations.concat(ops);
+          }
+          next();
+        });
+      } else {
+        setImmediate(next);
+      }
+    },
+
+    function(err) {
+
+      if (err) {
+        if (!self.node.stopping) {
+          log.error('Block Service: Error: ' + err);
+          self.node.stop();
+        }
+        return;
+      }
+
+      self._db.batch(operations, function(err) {
+
+        if (err && !self.node.stopping) {
+            log.error('Block Service: Error: ' + err);
+            self.node.stop();
+        }
+
+        self._onReorg(commonAncestorHeader, oldBlocks, newBlock);
+
+
+      });
+    }
+  );
+};
+
 BlockService.prototype._processBlock = function(block) {
 
   var self = this;
@@ -511,7 +539,6 @@ BlockService.prototype._onBlock = function(block) {
   if (this._tip.hash !== prevHash) {
     return;
   }
-
   log.debug('Block Service: new block: ' + block.rhash());
   block.height = this._tip.height + 1;
   this._processBlock(block);
@@ -522,10 +549,10 @@ BlockService.prototype._setListeners = function() {
 
   var self = this;
   self._header.once('headers', self._onAllHeaders.bind(self));
-  self._header.on('reorg', function(hash, headers) {
+  self._header.on('reorg', function(hash, headers, block) {
     if (!self._reorging && !this._initialSync) {
       log.debug('Block Service: detected a reorg from the header service.');
-      self._handleReorg(hash, headers);
+      self._handleReorg(hash, headers, block);
     }
   });
 

lib/services/header/encoding.js

@@ -35,7 +35,7 @@ Encoding.prototype.encodeHeaderValue = function(header) {
   var prevHash = new Buffer(header.prevHash, 'hex');
   var merkleRoot = new Buffer(header.merkleRoot, 'hex');
   var tsBuf = new Buffer(4);
-  tsBuf.writeUInt32BE(header.timestamp);
+  tsBuf.writeUInt32BE(header.timestamp || header.time);
   var bitsBuf = new Buffer(4);
   bitsBuf.writeUInt32BE(header.bits);
   var nonceBuf = new Buffer(4);

lib/services/header/index.js

@@ -212,7 +212,7 @@ HeaderService.prototype.start = function(callback) {
 };
 
 HeaderService.prototype.stop = function(callback) {
-  setImmediate(callback);
+  callback();
 };
 
 HeaderService.prototype._startHeaderSubscription = function() {
@@ -243,15 +243,13 @@ HeaderService.prototype._onBlock = function(block) {
   var prevHash = bcoin.util.revHex(block.prevBlock);
   var newBlock = prevHash === self._lastHeader.hash;
 
+  var header = block.toHeaders().toJSON();
+  header.timestamp = header.ts;
+  header.prevHash = header.prevBlock;
+
   if (newBlock) {
 
     log.debug('Header Service: new block: ' + hash);
-
-    var header = block.toHeaders().toJSON();
-
-    header.timestamp = header.ts;
-    header.prevHash = header.prevBlock;
-
     self._saveHeaders(self._onHeader(header));
 
   }
@@ -269,7 +267,8 @@ HeaderService.prototype._onBlock = function(block) {
       }
 
       if (reorg) {
-        self._handleReorg(block);
+        self._handleReorg(block, header); // this sets the last header
+        self._saveHeaders(self._onHeader(header));
         return;
       }
 
@@ -299,7 +298,9 @@ HeaderService.prototype._onHeader = function(header) {
 
   header.height = this._lastHeader.height + 1;
   header.chainwork = this._getChainwork(header, this._lastHeader).toString(16, 64);
-
+  if (!header.timestamp) {
+    header.timestamp = header.time;
+  }
   this._lastHeader = header;
 
   return [
@@ -461,7 +462,6 @@ HeaderService.prototype._detectReorg = function(block, callback) {
 
   assert(block, 'Block is needed to detect reorg.');
 
-console.log(bcoin.util.revHex(block.prevBlock));
   var key = this._encoding.encodeHeaderHashKey(bcoin.util.revHex(block.prevBlock));
 
   this._db.get(key, function(err, val) {
@@ -505,7 +505,7 @@ HeaderService.prototype._detectStartupReorg = function(callback) {
 
 };
 
-HeaderService.prototype._handleReorg = function(block) {
+HeaderService.prototype._handleReorg = function(block, header) {
 
   var self = this;
   self.getAllHeaders(function(err, headers) {
@@ -518,8 +518,12 @@ HeaderService.prototype._handleReorg = function(block) {
 
     var hash = headers.getIndex(self._originalTip.height).hash;
 
-    if (block) {
+    if (block && header) {
       hash = block.rhash();
+      self._lastHeader = headers.get(header.prevHash);
+      assert(self._lastHeader, 'Expected our reorg block to have a header entry, but it did not.');
+      headers.set(hash, header); // appends to the end
+      self.emit('reorg', hash, headers, block);
     }
 
     assert(hash, 'To reorg, we need a hash to reorg to.');

lib/services/insight-api

@@ -1 +1 @@
-/home/k/source/insight-api
+/Users/chrisk/source/insight-api

lib/services/mempool/index.js

@@ -69,28 +69,32 @@ MempoolService.prototype.start = function(callback) {
   });
 };
 
-MempoolService.prototype._onReorg = function(commonAncestorHeader, oldBlockList) {
+MempoolService.prototype.onReorg = function(args, callback) {
 
-  // set the tip to the common ancestor in case something goes wrong with the reorg
+  var oldBlockList = args[1];
-  this._setTip({ hash: commonAncestorHeader.hash, height: commonAncestorHeader.height });
-  var tipOps = utils.encodeTip(this._tip, this.name);
 
-  var removalOps = [{
+  var removalOps = [];
-    type: 'put',
-    key: tipOps.key,
-    value: tipOps.value
-  }];
 
-  // remove all the old blocks that we reorg from
+  for(var i = 0; i < oldBlockList.length; i++) {
-  oldBlockList.forEach(function(block) {
-    removalOps.push({
-      type: 'del',
-      key: this.encoding.encodeBlockKey(block.rhash()),
-    });
-  });
 
-  this._db.batch(removalOps);
+    var block = oldBlockList[i];
 
+    for(var j = 0; j < block.txs.length; j++) {
+
+      var tx = block.txs[j];
+      var key = this._encoding.encodeMempoolTransactionKey(tx.txid());
+      var value = this._encoding.encodeMempoolTransactionValue(tx);
+
+      removalOps.push({
+        type: 'put',
+        key: key,
+        value: value
+      });
+
+    }
+  }
+
+  callback(null, removalOps);
 };
 
 MempoolService.prototype._startSubscriptions = function() {
@@ -104,9 +108,6 @@ MempoolService.prototype._startSubscriptions = function() {
     this._bus = this.node.openBus({remoteAddress: 'localhost-mempool'});
   }
 
-  this._bus.on('block/reorg', this._onReorg.bind(this));
-  this._bus.subscribe('block/reorg');
-
   this._bus.on('p2p/transaction', this._onTransaction.bind(this));
   this._bus.subscribe('p2p/transaction');
 };

lib/services/p2p/index.js

@@ -213,6 +213,10 @@ P2P.prototype._initP2P = function() {
   this._maxPeers = this._options.maxPeers || 60;
   this._minPeers = this._options.minPeers || 0;
   this._configPeers = this._options.peers;
+
+  if (this.node.network === 'regtest') {
+    Networks.enableRegtest();
+  }
   this.messages = new p2p.Messages({ network: Networks.get(this.node.network) });
   this._peerHeights = [];
   this._peers = [];
@@ -243,6 +247,12 @@ P2P.prototype._onPeerBlock = function(peer, message) {
 };
 
 P2P.prototype._onPeerDisconnect = function(peer, addr) {
+
+  if (!this.node.stopping) {
+    this._connect();
+    return;
+  }
+
   this._removePeer(peer);
   log.info('Disconnected from peer: ' + addr.ip.v4);
 };

lib/services/timestamp/index.js

@@ -4,24 +4,20 @@ var BaseService = require('../../service');
 var Encoding = require('./encoding');
 var assert = require('assert');
 var _ = require('lodash');
-var index = require('../../index');
-var log = index.log;
 var LRU = require('lru-cache');
 
 var inherits = require('util').inherits;
-var utils = require('../../../lib/utils');
 
 function TimestampService(options) {
   BaseService.call(this, options);
   this._db = this.node.services.db;
-  this._tip = null;
   this._lastBlockTimestamp = 0;
   this._cache = new LRU(10);
 }
 
 inherits(TimestampService, BaseService);
 
-TimestampService.dependencies = [ 'db', 'block' ];
+TimestampService.dependencies = [ 'db' ];
 
 TimestampService.prototype.getAPIMethods = function() {
   return [
@@ -75,7 +71,6 @@ TimestampService.prototype.getBlockHashesByTimestamp = function(high, low, callb
 
 TimestampService.prototype.start = function(callback) {
   var self = this;
-  self._setListeners();
 
   self._db.getPrefix(self.name, function(err, prefix) {
 
@@ -86,67 +81,12 @@ TimestampService.prototype.start = function(callback) {
     self._prefix = prefix;
     self._encoding = new Encoding(self._prefix);
 
-    self._db.getServiceTip(self.name, function(err, tip) {
+    callback();
 
-      if (err) {
-        return callback(err);
-      }
-
-      self._tip = tip;
-      self._startSubscriptions();
-      callback();
-
-    });
   });
 
 };
 
-TimestampService.prototype._startSubscriptions = function() {
-
-  if (this._subscribed) {
-    return;
-  }
-
-  this._subscribed = true;
-  if (!this._bus) {
-    this._bus = this.node.openBus({remoteAddress: 'localhost'});
-  }
-
-  this._bus.on('block/reorg', this._onReorg.bind(this));
-  this._bus.subscribe('block/reorg');
-};
-
-TimestampService.prototype._sync = function() {
-
-  if (--this._p2pBlockCallsNeeded > 0) {
-
-    log.info('Blocks download progress: ' + this._numCompleted + '/' +
-      this._numNeeded + '  (' + (this._numCompleted/this._numNeeded*100).toFixed(2) + '%)');
-    this._p2p.getBlocks({ startHash: this._latestBlockHash });
-    return;
-
-  }
-
-};
-
-TimestampService.prototype._setListeners = function() {
-
-  var self = this;
-  self.on('reorg', self._onReorg.bind(self));
-
-};
-
-TimestampService.prototype._setTip = function(tip) {
-  log.debug('Timestamp Service: Setting tip to height: ' + tip.height);
-  log.debug('Timestamp Service: Setting tip to hash: ' + tip.hash);
-  this._tip = tip;
-  this._db.setServiceTip('block', this._tip);
-};
-
-TimestampService.prototype.stop = function(callback) {
-  setImmediate(callback);
-};
-
 TimestampService.prototype.onBlock = function(block, callback) {
 
   var operations = [];
@@ -160,12 +100,8 @@ TimestampService.prototype.onBlock = function(block, callback) {
 
   this._lastBlockTimestamp = ts;
 
-  this._tip.hash = hash;
-  this._tip.height++;
   this._cache.set(hash, ts);
 
-  var tipInfo = utils.encodeTip(this._tip, this.name);
-
   operations = operations.concat([
     {
       type: 'put',
@@ -176,58 +112,43 @@ TimestampService.prototype.onBlock = function(block, callback) {
       type: 'put',
       key: this._encoding.encodeBlockTimestampKey(hash),
       value: this._encoding.encodeBlockTimestampValue(ts)
-    },
-    {
-      type: 'put',
-      key: tipInfo.key,
-      value: tipInfo.value
     }
-
   ]);
 
-  setImmediate(function() {
+  callback(null, operations);
-    callback(null, operations);
-  });
 };
 
-TimestampService.prototype._onReorg = function(oldBlockList, newBlockList, commonAncestor) {
+TimestampService.prototype.onReorg = function(args, callback) {
 
-  // if the common ancestor block height is greater than our own, then nothing to do for the reorg
+  var self = this;
-  if (this._tip.height <= commonAncestor.header.height) {
+  var commonAncestorHeader = args[0];
-    return;
+  var oldBlockList = args[1];
-  }
-
-  // set the tip to the common ancestor in case something goes wrong with the reorg
-  var tipOps = utils.encodeTip({ hash: commonAncestor.hash, height: commonAncestor.header.height }, this.name);
-
-  var removalOps = [{
-    type: 'put',
-    key: tipOps.key,
-    value: tipOps.value
-  }];
 
+  var removalOps = [];
 
   // remove all the old blocks that we reorg from
   oldBlockList.forEach(function(block) {
     removalOps.concat([
       {
         type: 'del',
-        key: this.encoding.encodeTimestampBlockKey(block.ts),
+        key: self._encoding.encodeTimestampBlockKey(block.ts),
       },
       {
         type: 'del',
-        key: this.encoding.encodeBlockTimestampKey(block.rhash()),
+        key: self._encoding.encodeBlockTimestampKey(block.rhash()),
       }
     ]);
   });
 
-  this._db.batch(removalOps);
+  // look up the adjusted timestamp from our own database and set the lastTimestamp to it
+  self.getTimestamp(commonAncestorHeader.hash, function(err, timestamp) {
 
-  // set the last time stamp to the common ancestor
+    if (err) {
-  this._lastBlockTimestamp = commonAncestor.ts;
+      return callback(err);
-
+    }
-  //call onBlock for each of the new blocks
+    self._lastBlockTimestamp = timestamp;
-  newBlockList.forEach(this._onBlock.bind(this));
+    callback(null, removalOps);
+  });
 
 };
 
@@ -237,12 +158,23 @@ TimestampService.prototype.getTimestampSync = function(hash) {
 };
 
 TimestampService.prototype.getTimestamp = function(hash, callback) {
-  this._db.get(this._encoding.encodeBlockTimestampKey(hash), callback);
+  var self = this;
+  self._db.get(self._encoding.encodeBlockTimestampKey(hash), function(err, data) {
+    if (err) {
+      return callback(err);
+    }
+    callback(null, self._encoding.decodeBlockTimestampValue(data));
+  });
 };
 
 TimestampService.prototype.getHash = function(timestamp, callback) {
-  this._db.get(this._encoding.encodeTimestampBlockKey(timestamp), callback);
+  var self = this;
+  self._db.get(self._encoding.encodeTimestampBlockKey(timestamp), function(err, data) {
+    if (err) {
+      return callback(err);
+    }
+    callback(null, self._encoding.decodeTimestampBlockValue(data));
+  });
 };
 
-
 module.exports = TimestampService;

lib/services/transaction/index.js

@@ -234,19 +234,10 @@ TransactionService.prototype.start = function(callback) {
       return callback(err);
     }
 
-    self._db.getServiceTip(self.name, function(err, tip) {
+    self.prefix = prefix;
+    self._encoding = new Encoding(self.prefix);
+    callback();
 
-      if (err) {
-        return callback(err);
-      }
-
-      self._tip = tip;
-      self.prefix = prefix;
-      self._encoding = new Encoding(self.prefix);
-      self._startSubscriptions();
-      callback();
-
-    });
   });
 };
 
@@ -275,41 +266,31 @@ TransactionService.prototype.onBlock = function(block, callback) {
 
 };
 
-TransactionService.prototype._onReorg = function(commonAncestorHeader, oldBlockList) {
+TransactionService.prototype.onReorg = function(args, callback) {
 
   var self = this;
-  // if the common ancestor block height is greater than our own, then nothing to do for the reorg
-  if (self._tip.height <= commonAncestorHeader.height) {
-    return;
-  }
 
-  // set the tip to the common ancestor in case something goes wrong with the reorg
+  var oldBlockList = args[1];
-  var tipOps = utils.encodeTip({ hash: commonAncestorHeader.hash, height: commonAncestorHeader.height }, self.name);
 
-  var removalOps = [{
+  var removalOps = [];
-    type: 'put',
-    key: tipOps.key,
-    value: tipOps.value
-  }];
 
   for(var i = 0; i < oldBlockList.length; i++) {
+
     var block = oldBlockList[i];
-    for(var j = 0; j < block.transactions.length; j++) {
+
-      var tx = block.transactions[j];
+    for(var j = 0; j < block.txs.length; j++) {
+
+      var tx = block.txs[j];
+
       removalOps.push({
         type: 'del',
-        key: self._encoding.encodeTransactionKey(tx.id)
+        key: self._encoding.encodeTransactionKey(tx.txid())
       });
+
     }
   }
 
-  self._db.batch(removalOps, function(err) {
+  callback(null, removalOps);
-    if (err) {
-      log.error('Transaction Service: error removing operations during reorg.' + err);
-      self.node.stop();
-      return;
-    }
-  });
 
 };
 
@@ -337,19 +318,4 @@ TransactionService.prototype._processTransaction = function(tx, opts) {
 
 };
 
-TransactionService.prototype._startSubscriptions = function() {
-
-  if (this._subscribed) {
-    return;
-  }
-
-  this._subscribed = true;
-  if (!this._bus) {
-    this._bus = this.node.openBus({remoteAddress: 'localhost'});
-  }
-
-  this._bus.on('block/reorg', this._onReorg.bind(this));
-  this._bus.subscribe('block/reorg');
-};
-
 module.exports = TransactionService;

routes.txt

@@ -0,0 +1,34 @@
+  get('/addrs/:addrs/utxo', this.cacheShort(), addresses.checkAddrs.bind(addresses), addresses.multiutxo.bind(addresses));
+  get('/addr/:addr/utxo', this.cacheShort(), addresses.checkAddr.bind(addresses), addresses.utxo.bind(addresses));
+  post('/addrs/utxo', this.cacheShort(), addresses.checkAddrs.bind(addresses), addresses.multiutxo.bind(addresses));
+  get('/addrs/:addrs/txs', this.cacheShort(), addresses.checkAddrs.bind(addresses), addresses.multitxs.bind(addresses));
+  get('/addr/:addr', this.cacheShort(), addresses.checkAddr.bind(addresses), addresses.show.bind(addresses));
+  post('/addrs/txs', this.cacheShort(), addresses.checkAddrs.bind(addresses), addresses.multitxs.bind(addresses));
+
+  get('/addr/:addr/totalReceived', this.cacheShort(), addresses.checkAddr.bind(addresses), addresses.totalReceived.bind(addresses));
+  get('/addr/:addr/totalSent', this.cacheShort(), addresses.checkAddr.bind(addresses), addresses.totalSent.bind(addresses));
+  get('/addr/:addr/unconfirmedBalance', this.cacheShort(), addresses.checkAddr.bind(addresses), addresses.unconfirmedBalance.bind(addresses));
+
+  get('/tx/:txid', this.cacheShort(), transactions.show.bind(transactions));
+  param('txid', transactions.transaction.bind(transactions));
+  get('/txs', this.cacheShort(), transactions.list.bind(transactions));
+  post('/tx/send', transactions.send.bind(transactions));
+  get('/rawtx/:txid', this.cacheLong(), transactions.showRaw.bind(transactions));
+  param('txid', transactions.rawTransaction.bind(transactions));
+
+  get('/blocks', this.cacheShort(), blocks.list.bind(blocks));
+  get('/block/:blockHash', this.cacheShort(), blocks.checkBlockHash.bind(blocks), blocks.show.bind(blocks));
+  param('blockHash', blocks.block.bind(blocks));
+  param('blockHash', blocks.rawBlock.bind(blocks));
+  get('/block-index/:height', this.cacheShort(), blocks.blockIndex.bind(blocks));
+  param('height', blocks.blockIndex.bind(blocks));
+  get('/rawblock/:blockHash', this.cacheLong(), blocks.checkBlockHash.bind(blocks), blocks.showRaw.bind(blocks));
+
+  get('/status', this.cacheShort(), status.show.bind(status));
+  get('/sync', this.cacheShort(), status.sync.bind(status));
+  get('/peer', this.cacheShort(), status.peer.bind(status));
+  get('/version', this.cacheShort(), status.version.bind(status));
+  get('/messages/verify', messages.verify.bind(messages));
+  post('/messages/verify', messages.verify.bind(messages));
+  get('/utils/estimatefee', utils.estimateFee.bind(utils));
+  get('/currency', currency.index.bind(currency));

test/regtest/data/blocks.json

@@ -0,0 +1,10 @@
+[
+  "0000002006226e46111a0b59caaf126043eb5bbf28c34f3a5e332a1fc7b2b73cf188910f0520bf1be2e869376908f7c637214f9f0f9f4fd86c819a8c98a3c5f8d70a65f805ba9559ffff7f20000000000102000000010000000000000000000000000000000000000000000000000000000000000000ffffffff03510101ffffffff0200f2052a01000000232103e91a98edf4e457b5d5aa1e50e35ce6afb67bf7a9ff98b5c36dabbe994b080205ac0000000000000000266a24aa21a9ede2f61c3f71d1defd3fa999dfa36953755c690689799962b48bebd836974e8cf900000000",
+  "00000020bd5725c4bc8fb66032523ca18eb2c809ca1935ab35baf5a1bbaf60ef9a616b2b373d5a12492652b1d22c28ce5c6eb6b22f03c69db1eae9a368e1b544147583eb06ba9559ffff7f20030000000102000000010000000000000000000000000000000000000000000000000000000000000000ffffffff03520101ffffffff0200f2052a01000000232103e91a98edf4e457b5d5aa1e50e35ce6afb67bf7a9ff98b5c36dabbe994b080205ac0000000000000000266a24aa21a9ede2f61c3f71d1defd3fa999dfa36953755c690689799962b48bebd836974e8cf900000000",
+   "00000020af97cfd581a5f0da7925c346c371f6c58131a6b00a95060562f90607aa111e5f97c7b866d2503ee4982569a486f6e25023e72bef588bdef60a06fc740e50001906ba9559ffff7f20020000000102000000010000000000000000000000000000000000000000000000000000000000000000ffffffff03530101ffffffff0200f2052a01000000232103e91a98edf4e457b5d5aa1e50e35ce6afb67bf7a9ff98b5c36dabbe994b080205ac0000000000000000266a24aa21a9ede2f61c3f71d1defd3fa999dfa36953755c690689799962b48bebd836974e8cf900000000",
+   "000000201c9d9eb8064b9508b3cd10de5927cc47347a6899b66946ecc22d52b04c2f3c5f99435c3a200a2419695565432e3115a5ff0d57ed1975b9efe0c7ad1a709a6a4e07ba9559ffff7f20000000000102000000010000000000000000000000000000000000000000000000000000000000000000ffffffff03540101ffffffff0200f2052a01000000232103e91a98edf4e457b5d5aa1e50e35ce6afb67bf7a9ff98b5c36dabbe994b080205ac0000000000000000266a24aa21a9ede2f61c3f71d1defd3fa999dfa36953755c690689799962b48bebd836974e8cf900000000",
+   "00000020de37a9c9fbb1dc3ea5ca9e148e5310d639649814b45a424afe1d58a18db5d8327b529f7d58238330d55b82228ea14abcab965319de7e59c6534758c91c137fcb07ba9559ffff7f20020000000102000000010000000000000000000000000000000000000000000000000000000000000000ffffffff03550101ffffffff0200f2052a01000000232103e91a98edf4e457b5d5aa1e50e35ce6afb67bf7a9ff98b5c36dabbe994b080205ac0000000000000000266a24aa21a9ede2f61c3f71d1defd3fa999dfa36953755c690689799962b48bebd836974e8cf900000000",
+   "00000020e3aa93afbba267bec962a7a38bd064478e5c082e9a73e434b20544950ad8395b3b4a71816a4ce7973d0bcc6a583441fca260f71a175bd2887a9e3e40e4badcd355bb9559ffff7f20000000000102000000010000000000000000000000000000000000000000000000000000000000000000ffffffff03560101ffffffff0200f2052a010000002321023886024ea5984e57b35c3b339f5aee097819ac55235e4fd5822a6ad0a4de1b55ac0000000000000000266a24aa21a9ede2f61c3f71d1defd3fa999dfa36953755c690689799962b48bebd836974e8cf900000000",
+   "000000201a3c951a20b5d603144ce060c86e95fed1869524e66acfc46bdf08d96f6642094916aa5b965b0016fb8ba8b58e99f6b3edbe1a844aa7948adaccf7f28f08f914b9cb9559ffff7f20030000000102000000010000000000000000000000000000000000000000000000000000000000000000ffffffff03570101ffffffff0200f2052a01000000232102a8ef631320be3e6203329acba88fe0a663c19d59fee8240592dc2a32553a4159ac0000000000000000266a24aa21a9ede2f61c3f71d1defd3fa999dfa36953755c690689799962b48bebd836974e8cf900000000"
+
+]

test/regtest/data/blocks_reorg.json

@@ -0,0 +1,4 @@
+[
+   "000000201a3c951a20b5d603144ce060c86e95fed1869524e66acfc46bdf08d96f664209b4b1c32ec485f4ad27c5402a1b16a0b1135364b7c9b0dcf4276f9fa3fd215d1b08cc9559ffff7f20000000000102000000010000000000000000000000000000000000000000000000000000000000000000ffffffff03570101ffffffff0200f2052a01000000232102a5566542d1f0f202541d98755628a41dcd4416b50db820e2b04d5ecb0bd02b73ac0000000000000000266a24aa21a9ede2f61c3f71d1defd3fa999dfa36953755c690689799962b48bebd836974e8cf900000000"
+]
+

test/regtest/reorg.js

@@ -4,89 +4,273 @@ var expect = require('chai').expect;
 var sinon = require('sinon');
 var net = require('net');
 var spawn = require('child_process').spawn;
+var path = require('path');
+var rimraf = require('rimraf');
+var mkdirp = require('mkdirp');
+var fs = require('fs');
+var p2p = require('bitcore-p2p');
+var bitcore = require('bitcore-lib');
+var Networks = bitcore.Networks;
+var Header = bitcore.BlockHeader;
+var Block = bitcore.Block;
+var BcoinBlock = require('bcoin').block;
+var http = require('http');
 
+Networks.enableRegtest();
+var messages = new p2p.Messages({ network: Networks.get('regtest'), Block: BcoinBlock });
 var server;
-var headers = [];
+var rawBlocks = require('./data/blocks.json');
-var blocks = [];
+var rawReorgBlocks = require('./data/blocks_reorg.json')[0];
-var magic = new Buffer('00', 'hex'); // TODO find out what this is
-var messages = {
-  verack: new Buffer('76657273696f6e0000000000', 'hex'),
 
+var reorgBlock = BcoinBlock.fromRaw(rawReorgBlocks, 'hex');
+
+var blocks = rawBlocks.map(function(rawBlock) {
+  return new Block(new Buffer(rawBlock, 'hex'));
+});
+
+var headers = blocks.map(function(block) {
+  return block.header;
+});
+
+console.log(headers);
+
+var magic = new Buffer('fabfb5da', 'hex');
+
+var debug = true;
+var bitcoreDataDir = '/tmp/bitcore';
+
+var bitcore = {
+  configFile: {
+    file: bitcoreDataDir + '/bitcore-node.json',
+    conf: {
+      network: 'regtest',
+      port: 53001,
+      datadir: bitcoreDataDir,
+      services: [
+        'p2p',
+        'db',
+        'header',
+        'block',
+        'address',
+        'transaction',
+        'mempool',
+        'web',
+        'insight-api',
+        'fee',
+        'timestamp'
+      ],
+      servicesConfig: {
+        'p2p': {
+          'peers': [
+            { 'ip': { 'v4': '127.0.0.1' }, port: 18444 }
+          ]
+        },
+        'insight-api': {
+          'routePrefix': 'api'
+        }
+      }
+    }
+  },
+  httpOpts: {
+    protocol: 'http:',
+    hostname: 'localhost',
+    port: 53001,
+  },
+  opts: { cwd: bitcoreDataDir },
+  datadir: bitcoreDataDir,
+  exec: path.resolve(__dirname, '../../bin/bitcore-node'),
+  args: ['start'],
+  process: null
 };
 
 
-/*
+var blockIndex = 0;
-
+var tcpSocket;
-
-   comms path:
-
-      client = bitcore-node
-      server = my fake server
-
-      client -> version
-
-      server -> version
-
-      client -> verack
-
-      server -> verack
-
-      client -> getHeaders
-
-      server -> headers
-
-      client -> ?
-
-      server -> ?
-
-
-
-
-
-
-*/
-
-var startFakeNode = function(callback) {
 
+var startFakeNode = function() {
   server = net.createServer(function(socket) {
-    socket.write('hi\r\n');
+
+    tcpSocket = socket;
+    socket.on('end', function() {
+      console.log('bitcore-node has ended the connection');
+    });
+
+    socket.on('data', function(data) {
+
+      var command = data.slice(4, 16).toString('hex');
+      var message;
+
+      if (command === '76657273696f6e0000000000') { //version
+        message = messages.Version();
+      }
+
+      if (command === '76657261636b000000000000') { //verack
+        message = messages.VerAck();
+      }
+
+      if (command === '676574686561646572730000') { //getheaders
+        message = messages.Headers(headers, { BlockHeader: Header });
+      }
+
+      if (command === '676574626c6f636b73000000') { //getblocks
+        var block = blocks[blockIndex];
+        if (!block) {
+          return;
+        }
+        var blockHash = block.hash;
+        var inv = p2p.Inventory.forBlock(blockHash);
+        message = messages.Inventory([inv]);
+      }
+
+      if (command === '676574646174610000000000') { //getdata
+        var raw = rawBlocks[blockIndex++];
+        var blk = BcoinBlock.fromRaw(raw, 'hex');
+        message = messages.Block(blk, { Block: BcoinBlock });
+      }
+
+      if (message) {
+        socket.write(message.toBuffer());
+      }
+
+    });
+
     socket.pipe(socket);
   });
 
-  server.listen(1337, '127.0.0.1');
+  server.listen(18444, '127.0.0.1');
+};
+
+
+var shutdownFakeNode = function() {
+  server.close();
+};
+
+var shutdownBitcore = function(callback) {
+  if (bitcore.process) {
+    bitcore.process.kill();
+  }
   callback();
 };
 
+var startBitcore = function(callback) {
+
+  rimraf(bitcoreDataDir, function(err) {
+
+    if(err) {
+      return callback(err);
+    }
+
+    mkdirp(bitcoreDataDir, function(err) {
+
+      if(err) {
+        return callback(err);
+      }
+
+      fs.writeFileSync(bitcore.configFile.file, JSON.stringify(bitcore.configFile.conf));
+
+      var args = bitcore.args;
+      bitcore.process = spawn(bitcore.exec, args, bitcore.opts);
+
+      bitcore.process.stdout.on('data', function(data) {
+
+        if (debug) {
+          process.stdout.write(data.toString());
+        }
+
+      });
+      bitcore.process.stderr.on('data', function(data) {
+
+        if (debug) {
+          process.stderr.write(data.toString());
+        }
+
+      });
+
+      callback();
+    });
+
+  });
+
 
-var shutdownFakeNode = function(done) {
-  server.close();
-  done();
 };
 
-
-
 describe('Reorg', function() {
   // 1. spin up bitcore-node and have it connect to our custom tcp socket
   // 2. feed it a few headers
   // 3. feed it a few blocks
   // 4. feed it a block that reorgs
 
+  this.timeout(60000);
+
   before(function(done) {
-    startFakeNode(done);
+    startFakeNode();
+    startBitcore(done);
   });
 
   after(function(done) {
-    shutdownFakeNode(done);
+    shutdownFakeNode();
+    shutdownBitcore(done);
   });
 
   it('should reorg correctly', function(done) {
-    var client = new net.Socket();
+
-    client.connect(1337, '127.0.0.1');
+    // at this point we have a fully synced chain at height 7....
-    client.on('data', function(data) {
+    // we now want to send a new block number 7 whose prev hash is block 6 (it should be block 7)
-      console.log(data.toString());
+    // we then should reorg back to block 6 then back up to the new block 7
-      client.destroy();
+
-    });
+    setTimeout(function() {
-    done();
+
+        console.log('From Test: reorging to block: ' + reorgBlock.rhash());
+        rawBlocks.push(rawReorgBlocks);
+        var blockHash = reorgBlock.rhash();
+        var inv = p2p.Inventory.forBlock(blockHash);
+
+        var msg = messages.Inventory([inv]);
+        tcpSocket.write(msg.toBuffer());
+
+        setTimeout(function() {
+          var error;
+          var request = http.request('http://localhost:53001/api/block/' + reorgBlock.rhash(), function(res) {
+
+            if (res.statusCode !== 200 && res.statusCode !== 201) {
+              if (error) {
+                return;
+              }
+              return done('Error from bitcore-node webserver: ' + res.statusCode);
+            }
+
+            var resError;
+            var resData = '';
+
+            res.on('error', function(e) {
+              resError = e;
+            });
+
+            res.on('data', function(data) {
+              resData += data;
+            });
+
+            res.on('end', function() {
+              if (error) {
+                return;
+              }
+              var data = JSON.parse(resData);
+              done(resError, resData);
+            });
+
+          });
+
+          request.on('error', function(e) {
+            error = e;
+            done(error);
+          });
+
+          request.write('');
+          request.end();
+        }, 2000);
+    }, 2000);
+
+
   });
 });