Merge pull request #479 from matiu/prod2-fix-dead-cache

only cache addresses with balanceSat=0

app/controllers/addresses.js

@@ -12,6 +12,8 @@ var async = require('async');
 var MAX_BATCH_SIZE = 100;
 var RPC_CONCURRENCY = 5;
 
+var SIZE_TO_ENABLE_DEAD_CACHE = 500;
+
 var tDb = require('../../lib/TransactionDb').default();
 
 var checkSync = function(req, res) {
@@ -47,8 +49,9 @@ var getAddrs = function(req, res, next) {
     var addrStrs = req.param('addrs');
     var s = addrStrs.split(',');
     if (s.length === 0) return as;
+    var enableDeadAddresses = s.length > SIZE_TO_ENABLE_DEAD_CACHE;
     for (var i = 0; i < s.length; i++) {
-      var a = new Address(s[i]);
+      var a = new Address(s[i], enableDeadAddresses);
       as.push(a);
     }
   } catch (e) {
@@ -120,11 +123,20 @@ exports.multiutxo = function(req, res, next) {
   }
 };
 
+var stime = 0;
+var logtime = function(str, reset) {
+  if (reset || !stime)
+    stime = Date.now();
+
+  console.log('TIME:', str, ": ", Date.now() - stime);
+};
+
+var cache = {};
 exports.multitxs = function(req, res, next) {
   if (!checkSync(req, res)) return;
+  //logtime('Start', 1);
 
   function processTxs(txs, from, to, cb) {
-    txs = _.uniq(_.flatten(txs), 'txid');
     var nbTxs = txs.length;
 
     if (_.isUndefined(from) && _.isUndefined(to)) {
@@ -141,14 +153,6 @@ exports.multitxs = function(req, res, next) {
     if (to < 0) to = 0;
     if (from > nbTxs) from = nbTxs;
     if (to > nbTxs) to = nbTxs;
-
-    txs.sort(function(a, b) {
-      var b = (b.firstSeenTs || b.ts)+ b.txid;
-      var a = (a.firstSeenTs || a.ts)+ a.txid;
-      if (a > b) return -1;
-      if (a < b) return 1;
-      return 0;
-    });
     txs = txs.slice(from, to);
 
     var txIndex = {};
@@ -178,6 +182,8 @@ exports.multitxs = function(req, res, next) {
         }
 
         callback();
+      }, {
+        noExtraInfo: true
       });
     }, function(err) {
       if (err) return cb(err);
@@ -186,6 +192,19 @@ exports.multitxs = function(req, res, next) {
       // no longer at bitcoind (for example a double spend)
 
       var transactions = _.compact(_.pluck(txs, 'info'));
+      //rm not used items
+      _.each(transactions, function(t) {
+        t.vin = _.map(t.vin, function(i) {
+          return _.pick(i, ['addr', 'valueSat']);
+        });
+        t.vout = _.map(t.vout, function(o) {
+          return _.pick(o, ['scriptPubKey', 'value']);
+        });
+        delete t.locktime;
+        delete t.version;
+      });
+
+
       transactions = {
         totalItems: nbTxs,
         from: +from,
@@ -198,8 +217,20 @@ exports.multitxs = function(req, res, next) {
 
   var from = req.param('from');
   var to = req.param('to');
+  var addrStrs = req.param('addrs');
+
+  if (cache[addrStrs] && from > 0) {
+    //logtime('Cache hit');
+    txs = cache[addrStrs];
+    return processTxs(txs, from, to, function(err, transactions) {
+      //logtime('After process Txs');
+      if (err) return common.handleErrors(err, res)
+      res.jsonp(transactions);
+    });
+  };
 
   var as = getAddrs(req, res, next);
+  //logtime('After getAddrs');
   if (as) {
     var txs = [];
     async.eachLimit(as, RPC_CONCURRENCY, function(a, callback) {
@@ -215,7 +246,27 @@ exports.multitxs = function(req, res, next) {
     }, function(err) { // finished callback
       if (err) return common.handleErrors(err, res);
 
+      var MAX = 9999999999;
+      txs = _.uniq(_.flatten(txs), 'txid');
+      txs.sort(function(a, b) {
+        var b = (b.ts || b.firstSeenTs || MAX) + b.txid;
+        var a = (a.ts || b.firstSeenTs || MAX) + a.txid;
+        if (a > b) return -1;
+        if (a < b) return 1;
+        return 0;
+      });
+
+      if (!cache[addrStrs] || from == 0) {
+        cache[addrStrs] = txs;
+        // 5 min. just to purge memory. Cache is overwritten in from=0 requests.
+        setTimeout(function() {
+          console.log('Deleting cache:', addrStrs.substr(0, 20));
+          delete cache[addrStrs];
+        }, 5 * 60 * 1000);
+      }
+
       processTxs(txs, from, to, function(err, transactions) {
+        //logtime('After process Txs');
         if (err) return common.handleErrors(err, res);
         res.jsonp(transactions);
       });

app/models/Address.js

@@ -13,8 +13,20 @@ var TransactionDb = imports.TransactionDb || require('../../lib/TransactionDb').
 var BlockDb = imports.BlockDb || require('../../lib/BlockDb').default();
 var config = require('../../config/config');
 var CONCURRENCY = 5;
+var DAYS_TO_DEAD = 40;
+var MAX_CACHE_KEYS = 50000;
+
+var deadCache = {};
+
+function Address(addrStr, deadCacheEnable) {
+
+  if (deadCacheEnable && deadCache[addrStr]) {
+    //    console.log('DEAD CACHE HIT:', addrStr, deadCache[addrStr].cached);
+    return deadCache[addrStr];
+  }
+
+  this.deadCacheEnable = deadCacheEnable;
 
-function Address(addrStr) {
   this.balanceSat = 0;
   this.totalReceivedSat = 0;
   this.totalSentSat = 0;
@@ -76,6 +88,39 @@ function Address(addrStr) {
 
 }
 
+
+Address.deleteDeadCache = function(addrStr) {
+  if (deadCache[addrStr]) {
+    console.log('Deleting Dead Address Cache', addrStr);
+    delete deadCache[addrStr];
+  }
+};
+
+
+Address.prototype.setCache = function() {
+
+  this.cached = true;
+  deadCache[this.addrStr] = this;
+
+  var size = _.keys(deadCache).length;
+
+  console.log('%%%%%%%% cache size:', size); //TODO
+  if (size > MAX_CACHE_KEYS) {
+    console.log('%%%%%%%% deleting ~ 20% of the entries...');
+
+    var skip = _.random(4);
+
+    for (var prop in deadCache)
+      if (!(skip++ % 5))
+        delete deadCache[prop];
+
+    size = _.keys(deadCache).length;
+    console.log('%%%%%%%% cache size after delete:', size); //TODO
+  }
+  // TODO expire it...
+};
+
+
 Address.prototype.getObj = function() {
   // Normalize json address
   return {
@@ -165,8 +210,22 @@ Address.prototype.update = function(next, opts) {
   if (!('ignoreCache' in opts))
     opts.ignoreCache = config.ignoreCache;
 
+  if (opts.onlyUnspent && opts.includeTxInfo)
+    return cb('Bad params');
+
+  if (!opts.ignoreCache && this.cached) {
+    if (opts.onlyUnspent && this.unspent) {
+      return next();
+    }
+
+    if (opts.includeTxInfo && this.transactions.length && this.balanceSat == 0) {
+      return next();
+    }
+  }
+
   // should collect txList from address?
   var txList = opts.txLimit === 0 ? null : [];
+  var lastUsage, now = Date.now() / 1000;
 
   var tDb = TransactionDb;
   var bDb = BlockDb;
@@ -180,13 +239,15 @@ Address.prototype.update = function(next, opts) {
         // console.log('[Address.js.161:txOut:]',txOut); //TODO
         if (err) return next(err);
         if (opts.onlyUnspent) {
-          txOut = txOut.filter(function(x) {
+
+          var unspent = _.filter(txOut, function(x) {
             return !x.spentTxId;
           });
-          tDb.fillScriptPubKey(txOut, function() {
+
+          tDb.fillScriptPubKey(unspent, function() {
             //_.filter will filterout unspend without scriptPubkey
             //(probably from double spends)
-            self.unspent = _.filter(txOut.map(function(x) {
+            self.unspent = _.filter(unspent.map(function(x) {
               return {
                 address: self.addrStr,
                 txid: x.txid,
@@ -197,15 +258,38 @@ Address.prototype.update = function(next, opts) {
                 confirmations: x.isConfirmedCached ? (config.safeConfirmations) : x.confirmations,
                 confirmationsFromCache: !!x.isConfirmedCached,
               };
-            }), 'scriptPubKey');;
+            }), 'scriptPubKey');
+
+            if (self.deadCacheEnable && txOut.length && !self.unspent.length) {
+              // console.log('$$$$$$$$$$$$$$$$$$$$$$$$$$$    ',self.addrStr); //TODO
+              // console.log('[Address.js.242] NO UNSPENT:', self.addrStr, txOut.length); //TODO
+              // Asumes that addresses are ordered by Ts;
+              lastUsage = txOut[txOut.length - 1].spentTs || now;
+
+              var daysOld = (now - lastUsage) / (3600 * 24);
+              // console.log('[Address.js.253:dayOlds:]',daysOld); //TODO
+              var isOldEnough = daysOld > DAYS_TO_DEAD;
+
+              // console.log('[Address.js.246:isOldEnough:]', isOldEnough, lastUsage, now); //TODO
+
+              if (isOldEnough) {
+                self.setCache();
+              }
+            }
             return next();
           });
         } else {
+
           txOut.forEach(function(txItem) {
             self._addTxItem(txItem, txList, opts.includeTxInfo);
           });
           if (txList)
             self.transactions = txList;
+
+          if (self.deadCacheEnable && self.cached && self.balanceSat == 0) {
+            self.setCache();
+          }
+
           return next();
         }
       });

lib/Sync.js

@@ -6,6 +6,8 @@ var config        = imports.config || require('../config/config');
 var bitcore       = require('bitcore');
 var networks      = bitcore.networks;
 var async         = require('async');
+var _ = require('lodash');
+var Address = require('../app/models/Address');
 
 var logger = require('./logger').logger;
 var d      = logger.log;
@@ -293,7 +295,14 @@ Sync.prototype.setBranchConnectedBackwards = function(fromHash, cb) {
 
 //Store unconfirmed TXs
 Sync.prototype.storeTx = function(tx, cb) {
-  this.txDb.add(tx, cb);
+  this.txDb.add(tx, function(err, related) {
+    if (related) {
+      _.each(related, function(v,k){
+        Address.deleteDeadCache(k);
+      });
+    }
+    return cb(err, related);
+  });
 };
 
 

lib/TransactionDb.js

@@ -1,6 +1,7 @@
 'use strict';
 
 var imports = require('soop').imports();
+var _ = require('lodash');
 
 
 
@@ -115,9 +116,9 @@ TransactionDb.prototype.fromTxId = function(txid, cb) {
 
   // outs.
   db.createReadStream({
-    start: k,
-    end: k + '~'
-  })
+      start: k,
+      end: k + '~'
+    })
     .on('data', function(data) {
       var k = data.key.split('-');
       var v = data.value.split(':');
@@ -135,9 +136,9 @@ TransactionDb.prototype.fromTxId = function(txid, cb) {
 
       var k = SPENT_PREFIX + txid + '-';
       db.createReadStream({
-        start: k,
-        end: k + '~'
-      })
+          start: k,
+          end: k + '~'
+        })
         .on('data', function(data) {
           var k = data.key.split('-');
           var j = idx[parseInt(k[2])];
@@ -164,9 +165,9 @@ TransactionDb.prototype._fillSpent = function(info, cb) {
 
   var k = SPENT_PREFIX + info.txid + '-';
   db.createReadStream({
-    start: k,
-    end: k + '~'
-  })
+      start: k,
+      end: k + '~'
+    })
     .on('data', function(data) {
       var k = data.key.split('-');
       self._addSpentInfo(info.vout[k[2]], k[3], k[4], data.value);
@@ -180,7 +181,7 @@ TransactionDb.prototype._fillSpent = function(info, cb) {
 };
 
 
-TransactionDb.prototype._fillOutpoints = function(txInfo, cb) {
+TransactionDb.prototype._fillOutpoints = function(txInfo, cb, opts) {
   var self = this;
 
   if (!txInfo || txInfo.isCoinBase) return cb();
@@ -224,7 +225,7 @@ TransactionDb.prototype._fillOutpoints = function(txInfo, cb) {
           i.doubleSpentTxID = null;
         }
         return c_in();
-      });
+      }, opts);
     },
     function() {
       if (!incompleteInputs) {
@@ -237,16 +238,23 @@ TransactionDb.prototype._fillOutpoints = function(txInfo, cb) {
     });
 };
 
-TransactionDb.prototype._getInfo = function(txid, next) {
+TransactionDb.prototype._getInfo = function(txid, next, opts) {
   var self = this;
+  opts = opts || {};
 
   Rpc.getTxInfo(txid, function(err, txInfo) {
     if (err) return next(err);
+
+
     self._fillOutpoints(txInfo, function() {
+
+      if (opts.noExtraInfo)
+        return next(null, txInfo);
+
       self._fillSpent(txInfo, function() {
         return next(null, txInfo);
       });
-    });
+    }, opts);
   });
 };
 
@@ -260,7 +268,7 @@ TransactionDb.prototype.fromIdInfoSimple = function(txid, cb) {
   });
 };
 
-TransactionDb.prototype.fromIdWithInfo = function(txid, cb) {
+TransactionDb.prototype.fromIdWithInfo = function(txid, cb, opts) {
   var self = this;
 
   self._getInfo(txid, function(err, info) {
@@ -270,11 +278,11 @@ TransactionDb.prototype.fromIdWithInfo = function(txid, cb) {
       txid: txid,
       info: info
     });
-  });
+  }, opts);
 };
 
 // Gets address info from an outpoint
-TransactionDb.prototype.fromTxIdN = function(txid, n, cb) {
+TransactionDb.prototype.fromTxIdN = function(txid, n, cb, opts) {
   var self = this;
   var k = OUTS_PREFIX + txid + '-' + n;
 
@@ -294,12 +302,15 @@ TransactionDb.prototype.fromTxIdN = function(txid, n, cb) {
       };
     }
 
+    if (opts.noExtraInfo)
+      return cb(null, ret);
+
     // spent?
     var k = SPENT_PREFIX + txid + '-' + n + '-';
     db.createReadStream({
-      start: k,
-      end: k + '~'
-    })
+        start: k,
+        end: k + '~'
+      })
       .on('data', function(data) {
         var k = data.key.split('-');
         self._addSpentInfo(ret, k[3], k[4], data.value);
@@ -318,9 +329,9 @@ TransactionDb.prototype.deleteCacheForAddress = function(addr, cb) {
   var k = ADDR_PREFIX + addr + '-';
   var dbScript = [];
   db.createReadStream({
-    start: k,
-    end: k + '~'
-  })
+      start: k,
+      end: k + '~'
+    })
     .on('data', function(data) {
       var v = data.value.split(':');
       dbScript.push({
@@ -384,22 +395,25 @@ TransactionDb.prototype.cacheScriptPubKey = function(txouts, cb) {
   //  console.log('[TransactionDb.js.381:cacheScriptPubKey:]'); //TODO
   var self = this;
   var dbScript = [];
-  for (var ii in txouts) {
-    var txout = txouts[ii];
+  _.each(txouts, function(txout) {
+
     //everything already cached?
-    if (txout.scriptPubKeyCached || txout.spentTxId) {
-      continue;
-    }
+    if (txout.scriptPubKeyCached || txout.spentTxId)
+      return;
+
+    // not hard confirmed ?
+    if (!txout.isConfirmedCached && !txout.confirmedWillBeCached)
+      return;
 
     if (txout.scriptPubKey) {
-      var infoToCache = [txout.value_sat, (txout.isConfirmedCached || txout.confirmedWillBeCached) ? 1 : 0, txout.scriptPubKey];
+      var infoToCache = [txout.value_sat, 1, txout.scriptPubKey];
       dbScript.push({
         type: 'put',
         key: txout.key,
         value: infoToCache.join(':'),
       });
     }
-  }
+  });
   db.batch(dbScript, cb);
 };
 
@@ -456,10 +470,10 @@ TransactionDb.prototype.fromAddr = function(addr, opts, cb) {
   var unique = {};
 
   db.createReadStream({
-    start: k,
-    end: k + '~',
-    limit: opts.txLimit > 0 ? opts.txLimit : -1, // -1 means not limit
-  })
+      start: k,
+      end: k + '~',
+      limit: opts.txLimit > 0 ? opts.txLimit : -1, // -1 means not limit
+    })
     .on('data', function(data) {
       var k = data.key.split('-');
       var index = k[3] + k[4];
@@ -485,9 +499,9 @@ TransactionDb.prototype.fromAddr = function(addr, opts, cb) {
         }), CONCURRENCY, function(o, e_c) {
           var k = SPENT_PREFIX + o.txid + '-' + o.index + '-';
           db.createReadStream({
-            start: k,
-            end: k + '~'
-          })
+              start: k,
+              end: k + '~'
+            })
             .on('data', function(data) {
               var k = data.key.split('-');
               self._addSpentInfo(o, k[3], k[4], data.value);
@@ -550,8 +564,12 @@ TransactionDb.prototype.getStandardizedTx = function(tx, time, isCoinBase) {
 
 TransactionDb.prototype.fillScriptPubKey = function(txouts, cb) {
   var self = this;
+  var allCached = true;
   // Complete utxo info
   async.eachLimit(txouts, CONCURRENCY, function(txout, a_c) {
+    if (txout.scriptPubKeyCached) return a_c();
+    allCached = false;
+
     self.fromIdInfoSimple(txout.txid, function(err, info) {
       if (!info || !info.vout) return a_c(err);
 
@@ -559,6 +577,8 @@ TransactionDb.prototype.fillScriptPubKey = function(txouts, cb) {
       return a_c();
     });
   }, function() {
+    if (allCached)
+      return cb();
     self.cacheScriptPubKey(txouts, cb);
   });
 };
@@ -578,14 +598,14 @@ TransactionDb.prototype.removeFromTxId = function(txid, cb) {
       },
       function(c) {
         db.createReadStream({
-          start: SPENT_PREFIX + txid + '-',
-          end: SPENT_PREFIX + txid + '~'
-        })
+            start: SPENT_PREFIX + txid + '-',
+            end: SPENT_PREFIX + txid + '~'
+          })
           .pipe(
             db.createWriteStream({
               type: 'del'
             })
-        ).on('close', c);
+          ).on('close', c);
       }
     ],
     function(err) {
@@ -707,10 +727,10 @@ TransactionDb.prototype.checkVersion02 = function(cb) {
   var k = 'txa-';
   var isV2 = 1;
   db.createReadStream({
-    start: k,
-    end: k + '~',
-    limit: 1,
-  })
+      start: k,
+      end: k + '~',
+      limit: 1,
+    })
     .on('data', function(data) {
       isV2 = 0;
     })
@@ -726,9 +746,9 @@ TransactionDb.prototype.migrateV02 = function(cb) {
   var c2 = 0;
   var N = 50000;
   db.createReadStream({
-    start: k,
-    end: k + '~'
-  })
+      start: k,
+      end: k + '~'
+    })
     .on('data', function(data) {
       var k = data.key.split('-');
       var v = data.value.split(':');