Merge pull request #515 from isocolsky/ref/locked-utxos

Discard recently spent inputs from utxo selection

lib/server.js

@@ -902,50 +902,80 @@ WalletService.prototype._getUtxosForCurrentWallet = function(addresses, cb) {
     return utxo.txid + '|' + utxo.vout
   };
 
-  async.waterfall([
+  var allAddresses, allUtxos, utxoIndex;
+
+  async.series([
 
     function(next) {
       if (_.isArray(addresses)) {
-        if (!_.isEmpty(addresses)) {
-          next(null, addresses);
-        } else {
-          next(null, []);
-        }
-      } else {
-        self.storage.fetchAddresses(self.walletId, next);
+        allAddresses = addresses;
+        return next();
       }
+      self.storage.fetchAddresses(self.walletId, function(err, addresses) {
+        allAddresses = addresses;
+        return next();
+      });
     },
-    function(addresses, next) {
-      if (addresses.length == 0) return next(null, []);
+    function(next) {
+      if (allAddresses.length == 0) return cb(null, []);
 
-      var addressStrs = _.pluck(addresses, 'address');
+      var addressStrs = _.pluck(allAddresses, 'address');
       self._getUtxos(addressStrs, function(err, utxos) {
         if (err) return next(err);
         if (utxos.length == 0) return next(null, []);
+        allUtxos = utxos;
+        utxoIndex = _.indexBy(allUtxos, utxoKey);
+        return next();
+      });
+    },
+    function(next) {
+      self.getPendingTxs({}, function(err, txps) {
+        if (err) return next(err);
 
-        self.getPendingTxs({}, function(err, txps) {
-          if (err) return next(err);
-
-          var lockedInputs = _.map(_.flatten(_.pluck(txps, 'inputs')), utxoKey);
-          var utxoIndex = _.indexBy(utxos, utxoKey);
-          _.each(lockedInputs, function(input) {
-            if (utxoIndex[input]) {
-              utxoIndex[input].locked = true;
-            }
-          });
-
-          // Needed for the clients to sign UTXOs
-          var addressToPath = _.indexBy(addresses, 'address');
-          _.each(utxos, function(utxo) {
-            utxo.path = addressToPath[utxo.address].path;
-            utxo.publicKeys = addressToPath[utxo.address].publicKeys;
-          });
-
-          return next(null, utxos);
+        var lockedInputs = _.map(_.flatten(_.pluck(txps, 'inputs')), utxoKey);
+        _.each(lockedInputs, function(input) {
+          if (utxoIndex[input]) {
+            utxoIndex[input].locked = true;
+          }
         });
+        return next();
       });
     },
-  ], cb);
+    function(next) {
+      var now = Math.floor(Date.now() / 1000);
+      // Fetch latest broadcasted txs and remove any spent inputs from the
+      // list of UTXOs returned by the block explorer. This counteracts any out-of-sync
+      // effects between broadcasting a tx and getting the list of UTXOs.
+      // This is especially true in the case of having multiple instances of the block explorer.
+      self.storage.fetchBroadcastedTxs(self.walletId, {
+        minTs: now - 24 * 3600,
+        limit: 100
+      }, function(err, txs) {
+        if (err) return next(err);
+        var spentInputs = _.map(_.flatten(_.pluck(txs, 'inputs')), utxoKey);
+        _.each(spentInputs, function(input) {
+          if (utxoIndex[input]) {
+            utxoIndex[input].spent = true;
+          }
+        });
+        allUtxos = _.reject(allUtxos, {
+          spent: true
+        });
+        return next();
+      });
+    },
+    function(next) {
+      // Needed for the clients to sign UTXOs
+      var addressToPath = _.indexBy(allAddresses, 'address');
+      _.each(allUtxos, function(utxo) {
+        utxo.path = addressToPath[utxo.address].path;
+        utxo.publicKeys = addressToPath[utxo.address].publicKeys;
+      });
+      return next();
+    },
+  ], function(err) {
+    return cb(err, allUtxos);
+  });
 };
 
 /**
@@ -1883,7 +1913,6 @@ WalletService.prototype.createTx = function(opts, cb) {
   var self = this;
 
   self._runLocked(cb, function(cb) {
-
     var wallet, txp, changeAddress;
     async.series([
 
@@ -2442,14 +2471,14 @@ WalletService.prototype.getPendingTxs = function(opts, cb) {
       txp.deleteLockTime = self.getRemainingDeleteLockTime(txp);
     });
 
-    async.each(txps, function(txp, a_cb) {
-      if (txp.status != 'accepted') return a_cb();
+    async.each(txps, function(txp, next) {
+      if (txp.status != 'accepted') return next();
 
       self._checkTxInBlockchain(txp, function(err, isInBlockchain) {
-        if (err || !isInBlockchain) return a_cb(err);
+        if (err || !isInBlockchain) return next(err);
         self._processBroadcast(txp, {
           byThirdParty: true
-        }, a_cb);
+        }, next);
       });
     }, function(err) {
       return cb(err, _.reject(txps, function(txp) {

lib/storage.js

@@ -296,6 +296,43 @@ Storage.prototype.fetchTxs = function(walletId, opts, cb) {
   });
 };
 
+/**
+ * fetchBroadcastedTxs. Times are in UNIX EPOCH (seconds)
+ *
+ * @param walletId
+ * @param opts.minTs
+ * @param opts.maxTs
+ * @param opts.limit
+ */
+Storage.prototype.fetchBroadcastedTxs = function(walletId, opts, cb) {
+  var self = this;
+
+  opts = opts || {};
+
+  var tsFilter = {};
+  if (_.isNumber(opts.minTs)) tsFilter.$gte = opts.minTs;
+  if (_.isNumber(opts.maxTs)) tsFilter.$lte = opts.maxTs;
+
+  var filter = {
+    walletId: walletId,
+    status: 'broadcasted',
+  };
+  if (!_.isEmpty(tsFilter)) filter.broadcastedOn = tsFilter;
+
+  var mods = {};
+  if (_.isNumber(opts.limit)) mods.limit = opts.limit;
+
+  this.db.collection(collections.TXS).find(filter, mods).sort({
+    createdOn: -1
+  }).toArray(function(err, result) {
+    if (err) return cb(err);
+    if (!result) return cb();
+    var txs = _.map(result, function(tx) {
+      return Model.TxProposal.fromObj(tx);
+    });
+    return self._completeTxData(walletId, txs, cb);
+  });
+};
 
 /**
  * Retrieves notifications after a specific id or from a given ts (whichever is more recent).

test/integration/server.js

@@ -3469,7 +3469,7 @@ describe('Wallet service', function() {
       var server, wallet;
       beforeEach(function(done) {
         // log.level = 'debug';
-        helpers.createAndJoinWallet(2, 3, function(s, w) {
+        helpers.createAndJoinWallet(1, 2, function(s, w) {
           server = s;
           wallet = w;
           done();
@@ -3644,7 +3644,7 @@ describe('Wallet service', function() {
         var _old1 = Defaults.UTXO_SELECTION_MIN_TX_AMOUNT_VS_UTXO_FACTOR;
         var _old2 = Defaults.MAX_TX_SIZE_IN_KB;
         Defaults.UTXO_SELECTION_MIN_TX_AMOUNT_VS_UTXO_FACTOR = 0.0001;
-        Defaults.MAX_TX_SIZE_IN_KB = 3;
+        Defaults.MAX_TX_SIZE_IN_KB = 2;
 
         helpers.stubUtxos(server, wallet, [100].concat(_.range(1, 20, 0)), function() {
           var txOpts = {
@@ -3826,7 +3826,7 @@ describe('Wallet service', function() {
               toAddress: '18PzpUFkFZE8zKWUPvfykkTxmB9oMR8qP7',
               amount: 200e2,
             }],
-            feePerKb: 50e2,
+            feePerKb: 80e2,
           };
           server.createTx(txOpts, function(err, txp) {
             should.exist(err);
@@ -3851,6 +3851,43 @@ describe('Wallet service', function() {
           });
         });
       });
+      it('should not use UTXOs of recently broadcasted txs', function(done) {
+        helpers.stubUtxos(server, wallet, [1, 1], function() {
+          var txOpts = {
+            outputs: [{
+              toAddress: '18PzpUFkFZE8zKWUPvfykkTxmB9oMR8qP7',
+              amount: 1.5e8,
+            }],
+            feePerKb: 100e2,
+          };
+          helpers.createAndPublishTx(server, txOpts, TestData.copayers[0].privKey_1H_0, function(txp) {
+            should.exist(txp);
+            var signatures = helpers.clientSign(txp, TestData.copayers[0].xPrivKey_44H_0H_0H);
+            server.signTx({
+              txProposalId: txp.id,
+              signatures: signatures,
+            }, function(err, txp) {
+              should.not.exist(err);
+              should.exist(txp);
+
+              helpers.stubBroadcast();
+              server.broadcastTx({
+                txProposalId: txp.id
+              }, function(err, txp) {
+                should.not.exist(err);
+                should.exist(txp.txid);
+                txp.status.should.equal('broadcasted');
+                server.createTx(txOpts, function(err, txp) {
+                  should.exist(err);
+                  err.code.should.equal('INSUFFICIENT_FUNDS');
+                  should.not.exist(txp);
+                  done();
+                });
+              });
+            });
+          });
+        });
+      });
     });
 
   });