MB-16181: Filters with deleted collections

Upstream integration revealed an issue with the VB::Filter.
If we construct a filter with a collection "X" and "X" is actually
marked as deleted, "X" should not be part of the filter.

We fix this by replacing VB::Manifest::doesCollectionExist with
VB::Manifest::isCollectionOpen which performs the correct checks.

Tests updated to cover this case.

Change-Id: I38d4ba025805da7653bf3a84053d4280f4f547d7
Reviewed-on: http://review.couchbase.org/78570
Reviewed-by: Dave Rigby <daver@couchbase.com>
Tested-by: Build Bot <build@couchbase.com>

Author: jimwwalker

src/collections/vbucket_filter.cc

@@ -51,19 +51,23 @@ Collections::VB::Filter::Filter(const Collections::Filter& filter,
     }
 
     for (const auto& c : filter.getFilter()) {
-        if (rh.doesCollectionExist({c.data(), c.size()})) {
+        if (rh.isCollectionOpen({c.data(), c.size()})) {
             auto m = std::make_unique<std::string>(c);
             cb::const_char_buffer b{m->data(), m->size()};
             this->filter.emplace(b, std::move(m));
         } else {
-            // The VB::Manifest no longer has the collection so we won't filter
-            // it
+            // The VB::Manifest doesn't gave the collection, or the collection
+            // is deleted
             LOG(EXTENSION_LOG_NOTICE,
-                "VB::Filter::Filter: dropping collection:%s as it's not in the "
-                "VB::Manifest",
+                "VB::Filter::Filter: dropping collection:%s as it's not open",
                 c.c_str());
         }
     }
+
+    // All collections dropped
+    if (this->filter.empty() && !defaultAllowed) {
+        throw std::invalid_argument("VB::Filter::Filter: nothing to filter");
+    }
 }
 
 bool Collections::VB::Filter::allow(::DocKey key) const {

src/collections/vbucket_manifest.h

@@ -105,10 +105,10 @@ class Manifest {
         }
 
         /**
-         * @returns true/false if the collection exists
+         * @returns true if collection is open, false if not or unknown
          */
-        bool doesCollectionExist(cb::const_char_buffer collection) const {
-            return manifest.doesCollectionExist(collection);
+        bool isCollectionOpen(cb::const_char_buffer collection) const {
+            return manifest.isCollectionOpen(collection);
         }
 
     private:
@@ -422,10 +422,14 @@ class Manifest {
     }
 
     /**
-     * @returns true/false if the collection exists
+     * @returns true is the collection isOpen - false if not (or doesn't exist)
      */
-    bool doesCollectionExist(cb::const_char_buffer collection) const {
-        return map.count(collection) != 0;
+    bool isCollectionOpen(cb::const_char_buffer collection) const {
+        auto itr = map.find(collection);
+        if (itr != map.end()) {
+            return itr->second->isOpen();
+        }
+        return false;
     }
 
 protected:

tests/module_tests/collections/filter_test.cc

@@ -197,6 +197,35 @@ TEST_F(CollectionsFilterTest, filter_basic2) {
 
 class CollectionsVBFilterTest : public CollectionsFilterTest {};
 
+/**
+ * Try and create filter for collections which exist, but have been deleted
+ * i.e. they aren't writable so should never feature in a new VB::Filter
+ */
+TEST_F(CollectionsVBFilterTest, deleted_collection) {
+    Collections::Manifest m1(
+            R"({"revision":0,"separator":"$",)"
+            R"("collections":["$default", "vegetable", "fruit", "meat", "dairy"]})");
+    Collections::Manifest m2(
+            R"({"revision":1,"separator":"$",)"
+            R"("collections":["$default", "meat", "dairy"]})");
+
+    // Create the "producer" level filter so that we in theory produce at least
+    // these collections
+    std::string jsonFilter = R"({"collections":["vegetable", "fruit"]})";
+    boost::optional<const std::string&> json(jsonFilter);
+    Collections::Filter f(json, m1);
+
+    Collections::VB::Manifest vbm({});
+    // push creates
+    vbm.wlock().update(vb, m1);
+    // push deletes, removing both filtered collections
+    vbm.wlock().update(vb, m2);
+
+    // Construction will fail as the filter would not match anything valid
+    EXPECT_THROW(std::make_unique<Collections::VB::Filter>(f, vbm),
+                 std::invalid_argument);
+}
+
 /**
  * Create a filter with collections and check we allow what should be allowed.
  */