(#2818) - compaction removes orphaned attachments

Migrated idb/websql to have a new many-to-many table
of seqs to digests, fixed compaction of attachments
in all three adapters, added many new tests.

Author: nolanlawson

lib/adapters/idb.js

@@ -76,6 +76,7 @@ function LevelPouch(opts, callback) {
         running : false,
         docCount : -1
       };
+      db._compactionQueue = [];
       if (opts.db || opts.noMigrate) {
         afterDBCreated();
       } else {
@@ -577,13 +578,15 @@ function LevelPouch(opts, callback) {
 
       for (var i = 0; i < attachments.length; i++) {
         var key = attachments[i];
+        var att = doc.data._attachments[key];
 
-        if (doc.data._attachments[key].stub) {
-          recv++;
-          collectResults();
+        if (att.stub) {
+          // still need to update the refs mapping
+          var id = doc.data._id;
+          var rev = doc.data._rev;
+          saveAttachmentRefs(id, rev, att.digest, attachmentSaved);
           continue;
         }
-        var att = doc.data._attachments[key];
         var data;
         if (typeof att.data === 'string') {
           try {
@@ -650,18 +653,19 @@ function LevelPouch(opts, callback) {
         finish();
       }
     }
-
-    function saveAttachment(docInfo, digest, key, data, callback) {
-      var att = docInfo.data._attachments[key];
-      delete att.data;
-      att.digest = digest;
-      att.length = data.length;
+    
+    function saveAttachmentRefs(id, rev, digest, callback) {
       stores.attachmentStore.get(digest, function (err, oldAtt) {
-        if (err && err.name !== 'NotFoundError') {
-          return callback(err);
+        var newAttachment = false;
+        if (err) {
+          if (err.name !== 'NotFoundError') {
+            return callback(err);
+          } else {
+            newAttachment = true;
+          }
         }
 
-        var ref = [docInfo.metadata.id, docInfo.metadata.rev].join('@');
+        var ref = [id, rev].join('@');
         var newAtt = {};
 
         if (oldAtt) {
@@ -678,14 +682,37 @@ function LevelPouch(opts, callback) {
         }
 
         stores.attachmentStore.put(digest, newAtt, function (err) {
-          // do not try to store empty attachments
-          if (data.length === 0) {
+          if (err) {
             return callback(err);
           }
-          // doing this in batch causes a test to fail, wtf?
-          stores.binaryStore.put(digest, data, function (err) {
-            callback(err);
-          });
+          callback(null, newAttachment);
+        });
+      });
+    }
+
+    function saveAttachment(docInfo, digest, key, data, callback) {
+      var att = docInfo.data._attachments[key];
+      delete att.data;
+      att.digest = digest;
+      att.length = data.length;
+      var id = docInfo.metadata.id;
+      var rev = docInfo.metadata.rev;
+
+      saveAttachmentRefs(id, rev, digest, function (err, newAttachment) {
+        if (err) {
+          return callback(err);
+        }
+        // do not try to store empty attachments
+        if (data.length === 0) {
+          return callback(err);
+        }
+        if (!newAttachment) {
+          // small optimization - don't bother writing it again
+          return callback(err);
+        }
+        // doing this in batch causes a test to fail, wtf?
+        stores.binaryStore.put(digest, data, function (err) {
+          callback(err);
         });
       });
     }
@@ -984,15 +1011,34 @@ function LevelPouch(opts, callback) {
   };
 
   api._doCompaction = function (docId, rev_tree, revs, callback) {
+    if (!revs.length) {
+      return callback();
+    }
+    db._compactionQueue.push([docId, rev_tree, revs, callback]);
+    if (db._compactionQueue.length === 1) {
+      api._doCompactionSequentially(docId, rev_tree, revs, callback);
+    }
+  };
+
+  // execute compactions synchronously so we
+  // can correctly dedup attachments
+  api._doCompactionSequentially = function (docId, rev_tree, revs, callback) {
+    var originalCallback = callback;
+    callback = function (err) {
+      originalCallback(err);
+      process.nextTick(function () {
+        db._compactionQueue.shift();
+        if (db._compactionQueue.length) {
+          api._doCompactionSequentially.apply(api, db._compactionQueue[0]);
+        }
+      });
+    };
     stores.docStore.get(docId, function (err, metadata) {
       if (err) {
         return callback(err);
       }
       var seqs = metadata.rev_map; // map from rev to seq
       metadata.rev_tree = rev_tree;
-      if (!revs.length) {
-        return callback();
-      }
       var batch = [];
       batch.push({
         key: metadata.id,
@@ -1001,6 +1047,88 @@ function LevelPouch(opts, callback) {
         valueEncoding: vuvuEncoding,
         prefix: stores.docStore
       });
+
+      var digestMap = {};
+      var numDone = 0;
+      var overallErr;
+      function checkDone(err) {
+        if (err) {
+          overallErr = err;
+        }
+        if (++numDone === revs.length) { // done
+          if (overallErr) {
+            return callback(err);
+          }
+          deleteOrphanedAttachments();
+        }
+      }
+
+      function finish(err) {
+        if (err) {
+          return callback(err);
+        }
+        db.batch(batch, callback);
+      }
+
+      function deleteOrphanedAttachments() {
+        var possiblyOrphanedAttachments = Object.keys(digestMap);
+        if (!possiblyOrphanedAttachments.length) {
+          return finish();
+        }
+        var numDone = 0;
+        var overallErr;
+        function checkDone(err) {
+          if (err) {
+            overallErr = err;
+          }
+          if (++numDone === possiblyOrphanedAttachments.length) {
+            finish(overallErr);
+          }
+        }
+        var refsToDelete = new utils.Map();
+        revs.forEach(function (rev) {
+          refsToDelete.set(docId + '@' + rev, true);
+        });
+        possiblyOrphanedAttachments.forEach(function (digest) {
+          stores.attachmentStore.get(digest, function (err, attData) {
+            if (err) {
+              if (err.name === 'NotFoundError') {
+                return checkDone();
+              } else {
+                return checkDone(err);
+              }
+            }
+            var refs = Object.keys(attData.refs || {}).filter(function (ref) {
+              return !refsToDelete.has(ref);
+            });
+            var newRefs = {};
+            refs.forEach(function (ref) {
+              newRefs[ref] = true;
+            });
+            if (refs.length) { // not orphaned
+              batch.push({
+                key: digest,
+                type: 'put',
+                valueEncoding: 'json',
+                value: {refs: newRefs},
+                prefix: stores.attachmentStore
+              });
+            } else { // orphaned, can safely delete
+              batch = batch.concat([{
+                key: digest,
+                type: 'del',
+                prefix: stores.attachmentStore
+              }, {
+                key: digest,
+                type: 'del',
+                prefix: stores.binaryStore
+              }]);
+            }
+            checkDone();
+          });
+        });
+      }
+
       revs.forEach(function (rev) {
         var seq = seqs[rev];
         if (!seq) {
@@ -1011,8 +1139,22 @@ function LevelPouch(opts, callback) {
           type: 'del',
           prefix: stores.bySeqStore
         });
+        stores.bySeqStore.get(formatSeq(seq), function (err, doc) {
+          if (err) {
+            if (err.name === 'NotFoundError') {
+              return checkDone();
+            } else {
+              return checkDone(err);
+            }
+          }
+          var atts = Object.keys(doc._attachments || {});
+          atts.forEach(function (attName) {
+            var digest = doc._attachments[attName].digest;
+            digestMap[digest] = true;
+          });
+          checkDone();
+        });
       });
-      db.batch(batch, callback);
     });
   };