(pouchdb#4372) Removing stemmed revisions

When a document has more revisions than the revlimit the revisions are
now not only stemm from the revision tree but also removed from the
database.

src/adapters/idb/bulkDocs.js

@@ -235,7 +235,10 @@ function idbBulkDocs(dbOpts, req, opts, api, idb, Changes, callback) {
     function afterPutDoc(e) {
       if (isUpdate && api.auto_compaction) {
         autoCompact(docInfo);
+      } else if (docInfo.stemmedRevs.length) {
+        compactRevs(docInfo.stemmedRevs, docInfo.metadata.id, txn);
       }
+
       metadata.seq = e.target.result;
       // Current _rev is calculated from _rev_tree on read
       delete metadata.rev;

src/adapters/leveldb/index.js

@@ -445,6 +445,8 @@ function LevelPouch(opts, callback) {
     var newEdits = opts.new_edits;
     var results = new Array(req.docs.length);
     var fetchedDocs = new collections.Map();
+    var stemmedRevs = new collections.Map();
+
     var txn = new LevelTransaction();
     var docCountDelta = 0;
     var newUpdateSeq = db._updateSeq;
@@ -545,16 +547,13 @@ function LevelPouch(opts, callback) {
       });
     }
 
-    function autoCompact(callback) {
-
+    function compact(revsMap, callback) {
       var promise = Promise.resolve();
-
-      fetchedDocs.forEach(function (metadata, docId) {
+      revsMap.forEach(function (revs, docId) {
         // TODO: parallelize, for now need to be sequential to
         // pass orphaned attachment tests
         promise = promise.then(function () {
           return new Promise(function (resolve, reject) {
-            var revs = compactTree(metadata);
             api._doCompactionNoLock(docId, revs, {ctx: txn}, function (err) {
               /* istanbul ignore if */
               if (err) {
@@ -571,11 +570,20 @@ function LevelPouch(opts, callback) {
       }, callback);
     }
 
+    function autoCompact(callback) {
+      var revsMap = new collections.Map();
+      fetchedDocs.forEach(function (metadata, docId) {
+        revsMap.set(docId, compactTree(metadata));
+      });
+      compact(revsMap, callback);
+    }
+
     function finish() {
       if (api.auto_compaction) {
         return autoCompact(complete);
+      } else {
+          compact(stemmedRevs, complete);
       }
-      return complete();
     }
 
     function writeDoc(docInfo, winningRev, winningRevIsDeleted, newRevIsDeleted,
@@ -595,6 +603,10 @@ function LevelPouch(opts, callback) {
         docInfo.data._deleted = true;
       }
 
+      if (docInfo.stemmedRevs.length) {
+          stemmedRevs.set(docInfo.metadata.id, docInfo.stemmedRevs);
+      }
+
       var attachments = docInfo.data._attachments ?
         Object.keys(docInfo.data._attachments) :
         [];
@@ -660,7 +672,7 @@ function LevelPouch(opts, callback) {
         if (seq) {
           // check that there aren't any existing revisions with the same
           // revision id, else we shouldn't do anything
-          return callback2();
+          return callback2(null, docInfo.revsStemmed);
         }
         seq = ++newUpdateSeq;
         docInfo.metadata.rev_map[docInfo.metadata.rev] =
@@ -684,7 +696,7 @@ function LevelPouch(opts, callback) {
           rev: winningRev
         };
         fetchedDocs.set(docInfo.metadata.id, docInfo.metadata);
-        callback2();
+        callback2(null, docInfo.revsStemmed);
       }
 
       if (!attachments.length) {

src/adapters/websql/bulkDocs.js

@@ -220,17 +220,14 @@ function websqlBulkDocs(dbOpts, req, opts, api, db, Changes, callback) {
       finish();
     }
 
-    function autoCompact() {
+    function dataWritten(tx, seq) {
+      var id = docInfo.metadata.id;
       if (!isUpdate || !api.auto_compaction) {
-        return; // nothing to do
+        compactRevs(compactTree(docInfo.metadata), id, tx);
+      } else if (docInfo.stemmedRevs.length) {
+        compactRevs(docInfo.stemmedRevs, id, tx);
       }
-      var id = docInfo.metadata.id;
-      var revsToDelete = compactTree(docInfo.metadata);
-      compactRevs(revsToDelete, id, tx);
-    }
 
-    function dataWritten(tx, seq) {
-      autoCompact();
       docInfo.metadata.seq = seq;
       delete docInfo.metadata.rev;
 
@@ -242,7 +239,6 @@ function websqlBulkDocs(dbOpts, req, opts, api, db, Changes, callback) {
         : 'INSERT INTO ' + DOC_STORE +
       ' (id, winningseq, max_seq, json) VALUES (?,?,?,?);';
       var metadataStr = safeJsonStringify(docInfo.metadata);
-      var id = docInfo.metadata.id;
       var params = isUpdate ?
         [metadataStr, seq, winningRev, id] :
         [id, seq, seq, metadataStr];

src/deps/docs/processDocs.js

@@ -96,6 +96,7 @@ function processDocs(revLimit, docInfos, api, fetchedDocs, tx, results,
         // Ensure stemming applies to new writes as well
         var merged = merge([], currentDoc.metadata.rev_tree[0], revLimit);
         currentDoc.metadata.rev_tree = merged.tree;
+        currentDoc.stemmedRevs = merged.stemmedRevs || [];
         insertDoc(currentDoc, resultsIdx, docWritten);
       }
     }

src/deps/docs/updateDoc.js

@@ -42,6 +42,7 @@ function updateDoc(revLimit, prev, docInfo, results,
 
   var newRev = docInfo.metadata.rev;
   docInfo.metadata.rev_tree = merged.tree;
+  docInfo.stemmedRevs = merged.stemmedRevs || [];
   /* istanbul ignore else */
   if (prev.rev_map) {
     docInfo.metadata.rev_map = prev.rev_map; // used only by leveldb

src/deps/merge/index.js

@@ -1,5 +1,5 @@
 // for a better overview of what this is doing, read:
-// https://github.com/apache/couchdb/blob/master/src/couchdb/couch_key_tree.erl
+// https://github.com/apache/couchdb-couch/blob/master/src/couch_key_tree.erl
 //
 // But for a quick intro, CouchDB uses a revision tree to store a documents
 // history, A -> B -> C, when a document has conflicts, that is a branch in the
@@ -10,6 +10,7 @@
 // Tree = [Key, Opts, [Tree, ...]], in particular single node: [Key, []]
 
 import rootToLeaf from './rootToLeaf';
+import traverseRevTree from './traverseRevTree';
 
 function sortByPos(a, b) {
   return a.pos - b.pos;
@@ -185,6 +186,8 @@ function doMerge(tree, path, dontExpand) {
 function stem(tree, depth) {
   // First we break out the tree into a complete list of root to leaf paths
   var paths = rootToLeaf(tree);
+  var maybeStem = {};
+
   var result;
   for (var i = 0, len = paths.length; i < len; i++) {
     // Then for each path, we cut off the start of the path based on the
@@ -196,6 +199,12 @@ function stem(tree, depth) {
       pos: path.pos + numStemmed,
       ids: pathToTree(stemmed, numStemmed)
     };
+
+    for (var s = 0; s < numStemmed; s++) {
+        var rev = (path.pos + s) + '-' + stemmed[s].id;
+        maybeStem[rev] = true;
+    }
+
     // Then we remerge all those flat trees together, ensuring that we dont
     // connect trees that would go beyond the depth limit
     if (result) {
@@ -204,15 +213,26 @@ function stem(tree, depth) {
       result = [stemmedNode];
     }
   }
-  return result;
+
+  traverseRevTree(result, function (isLeaf, pos, revHash) {
+      // some revisions may have been removed in a branch but not in another
+      delete maybeStem[pos + '-' + revHash];
+  });
+
+  return {
+      tree: result,
+      revs: Object.keys(maybeStem),
+  };
 }
 
 function merge(tree, path, depth) {
   var newTree = doMerge(tree, path);
+  var stemmed = stem(newTree.tree, depth);
   return {
-    tree: stem(newTree.tree, depth),
+    tree: stemmed.tree,
+    stemmedRevs: stemmed.revs,
     conflicts: newTree.conflicts
   };
 }
 
-export default merge;
+export default merge;