Merge branch 'neo' into master

t3rm1n4l · t3rm1n4l · commit f8a33459f122 · 2022-03-22T14:37:18.000+05:30
Commits: dc531f2 MB-47139 magma-kvstore: Adapt to SetMaxOpenFiles API change ac0da7c MB-51513: Merge branch 'couchbase/cheshire-cat' into 'couchbase/neo' cc160cb MB-51513: Revert "MB-34280: Set max DCP name to 200 characters" 5cbcd28 MB-51513: Merge 'couchbase/cheshire-cat' into 'couchbase/neo' 09f70f6 MB-51414: Simplify handling of snapshot ranges in processItems() db53ff0 MB-50874: Merge branch 'mad-hatter' into cheshire-cat Change-Id: Ic6195a595038776f2c7014f9c1241b96f12bfa90
diff --git a/daemon/mcbp_validators.cc b/daemon/mcbp_validators.cc
@@ -576,14 +576,6 @@ static Status dcp_open_validator(Cookie& cookie) {
                                 "consumer_name must be a string");
                         return Status::Einval;
                     }
-                    auto nm = kv.value().get<std::string>();
-                    if (nm.size() > cb::limits::MaxDcpName) {
-                        cookie.setErrorContext(
-                                "consumer_name limit is " +
-                                std::to_string(cb::limits::MaxDcpName) +
-                                " characters");
-                        return Status::Einval;
-                    }
                 } else {
                     cookie.setErrorContext("Unsupported JSON property " +
                                            kv.key());
@@ -595,14 +587,6 @@ static Status dcp_open_validator(Cookie& cookie) {
             return Status::Einval;
         }
     }
-
-    if (size_t(cookie.getHeader().getKeylen()) > cb::limits::MaxDcpName) {
-        cookie.setErrorContext("Dcp name limit is " +
-                               std::to_string(cb::limits::MaxDcpName) +
-                               " characters");
-        return Status::Einval;
-    }
-
     return verify_common_dcp_restrictions(cookie);
 }
 
diff --git a/docs/dcp/documentation/commands/open-connection.md b/docs/dcp/documentation/commands/open-connection.md
@@ -43,12 +43,7 @@ Flags are specified as a bitmask in network byte order with the following bits d
 
 When setting the Producer or Consumer flag the sender is telling the server what type of connection will be created. For example, if the Producer type is set then the sender of the Open Connection message will be a Consumer.
 
-The connection name is specified using the key field. When selecting a name the
-only requirement is that the name take up no more space than 200 bytes. It is
-recommended that the name uses that ASCII character set and uses alpha-numeric
-characters. It is highly advantageous for improved supportability Couchbase
-Server that the connection names embed as much contextual information as
-possible from the client.
+The connection name is specified using the key field. When selecting a name the only requirement is that the name take up no more space than 256 bytes. It is recommended that the name uses that ASCII character set and uses alpha-numeric characters. It is highly advantageous for improved supportability Couchbase Server that the connection names embed as much contextual information as possible from the client.
 
 As of version 6.5, the _value_ can be used to specify additional information
 about the connection to be opened. If non-empty, the value must be a JSON
diff --git a/engines/ep/src/checkpoint_manager.cc b/engines/ep/src/checkpoint_manager.cc
@@ -971,13 +971,10 @@ CheckpointManager::ItemsForCursor CheckpointManager::getItemsForCursor(
         if (enteredNewCp) {
             const auto& checkpoint = **cursor.getCheckpoint();
             result.checkpointType = checkpoint.getCheckpointType();
-            result.ranges.push_back(
-                    {{checkpoint.getSnapshotStartSeqno(),
-                      checkpoint.getSnapshotEndSeqno()},
-                     checkpoint.getHighCompletedSeqno(),
-                     checkpoint.getHighPreparedSeqno(),
-                     checkpoint.getState() ==
-                             checkpoint_state::CHECKPOINT_CLOSED});
+            result.ranges.push_back({{checkpoint.getSnapshotStartSeqno(),
+                                      checkpoint.getSnapshotEndSeqno()},
+                                     checkpoint.getHighCompletedSeqno(),
+                                     checkpoint.getHighPreparedSeqno()});
             enteredNewCp = false;
 
             // As we cross into new checkpoints, update the maxDeletedRevSeqno
diff --git a/engines/ep/src/checkpoint_manager.h b/engines/ep/src/checkpoint_manager.h
@@ -60,8 +60,6 @@ struct CheckpointSnapshotRange {
     // HPS that should be flushed when the entire range has been persisted.
     // This is the seqno of the latest prepare in this checkpoint.
     std::optional<uint64_t> highPreparedSeqno = {};
-    // Is the checkpoint closed
-    bool isClosed = false;
 };
 
 /**
diff --git a/engines/ep/src/connhandler.h b/engines/ep/src/connhandler.h
@@ -81,9 +81,6 @@ struct ConnCounter {
 
 class ConnHandler : public DcpConnHandlerIface {
 public:
-    /// The maximum length of a DCP stat name
-    static constexpr size_t MaxDcpStatNameLength = 47;
-
     enum class PausedReason : uint8_t {
         BufferLogFull,
         Initializing,
diff --git a/engines/ep/src/dcp/active_stream.cc b/engines/ep/src/dcp/active_stream.cc
@@ -1180,27 +1180,12 @@ void ActiveStream::processItems(
             if (qi->getOperation() == queue_op::checkpoint_end) {
                 // At the end of each checkpoint remove its snapshot range, so
                 // we don't use it to set nextSnapStart for the next checkpoint.
-                // If statement ensures that we have indeed completed the
-                // snapshot range by checking that checkpoint_end seqno is
-                // greater than the end of the range. Also account for an edge
-                // case where we have closed checkpoint who's checkpoint_end is
-                // equal to snapshot end seqno. This can happen when snapshots
-                // partially received by a replica which is then promoted to
-                // active
+                // We can just erase the range at the head of ranges as every
+                // time as CheckpointManager::getItemsForCursor() will always
+                // ensure there is a snapshot range for if there is a
+                // queue_op::checkpoint_end in the items it returns.
                 auto rangeItr = outstandingItemsResult.ranges.begin();
-                auto snapEnd = static_cast<int64_t>(rangeItr->getEnd());
-                if (snapEnd < qi->getBySeqno() ||
-                    (rangeItr->isClosed && snapEnd == qi->getBySeqno())) {
-                    outstandingItemsResult.ranges.erase(rangeItr);
-                } else {
-                    log(spdlog::level::level_enum::err,
-                        "ActiveStream::processItems checkpoint_end:{} "
-                        "should not be in the current snapshot range "
-                        "s:{}->e:{}",
-                        qi->getBySeqno(),
-                        rangeItr->getStart(),
-                        rangeItr->getEnd());
-                }
+                outstandingItemsResult.ranges.erase(rangeItr);
             }
 
             if (qi->getOperation() == queue_op::checkpoint_start) {
diff --git a/engines/ep/src/kvstore/magma-kvstore/magma-kvstore.cc b/engines/ep/src/kvstore/magma-kvstore/magma-kvstore.cc
@@ -597,10 +597,6 @@ MagmaKVStore::MagmaKVStore(MagmaKVStoreConfig& configuration)
             configuration.getMagmaMemoryQuotaLowWaterMarkRatio();
 
     configuration.setStore(this);
-    {
-        cb::UseArenaMallocSecondaryDomain domainGuard;
-        magma::SetMaxOpenFiles(configuration.getMaxFileDescriptors());
-    }
 
     // To save memory only allocate counters for the number of vBuckets that
     // this shard will have to deal with
@@ -665,6 +661,7 @@ MagmaKVStore::MagmaKVStore(MagmaKVStoreConfig& configuration)
 
     magma = std::make_unique<MagmaMemoryTrackingProxy>(configuration.magmaCfg);
 
+    magma->SetMaxOpenFiles(configuration.getMaxFileDescriptors());
     setMaxDataSize(configuration.getBucketQuota());
     setMagmaFragmentationPercentage(
             configuration.getMagmaFragmentationPercentage());
diff --git a/engines/ep/src/kvstore/magma-kvstore/magma-memory-tracking-proxy.cc b/engines/ep/src/kvstore/magma-kvstore/magma-memory-tracking-proxy.cc
@@ -363,6 +363,11 @@ void MagmaMemoryTrackingProxy::EnableBlockCache(bool enable) {
     magma->EnableBlockCache(enable);
 }
 
+void MagmaMemoryTrackingProxy::SetMaxOpenFiles(size_t n, bool blocking) {
+    cb::UseArenaMallocSecondaryDomain domainGuard;
+    magma->SetMaxOpenFiles(n, blocking);
+}
+
 void MagmaMemoryTrackingProxy::SetMemoryQuota(const size_t quota) {
     cb::UseArenaMallocSecondaryDomain domainGuard;
     magma->SetMemoryQuota(quota);
diff --git a/engines/ep/src/kvstore/magma-kvstore/magma-memory-tracking-proxy.h b/engines/ep/src/kvstore/magma-kvstore/magma-memory-tracking-proxy.h
@@ -176,6 +176,7 @@ class MagmaMemoryTrackingProxy {
     magma::Status Rollback(const magma::Magma::KVStoreID kvID,
                            magma::Magma::SeqNo rollbackSeqno,
                            magma::Magma::RollbackCallback callback);
+    void SetMaxOpenFiles(size_t n, bool blocking = false);
     void SetFragmentationRatio(double fragRatio);
     void EnableBlockCache(bool enable);
     void SetMemoryQuota(const size_t quota);
diff --git a/engines/ep/tests/module_tests/collections/collections_seqno_advance.cc b/engines/ep/tests/module_tests/collections/collections_seqno_advance.cc
@@ -131,17 +131,6 @@ class CollectionsSeqnoAdvanced
     }
 
     void setupOneOperation(InputType type, ForStream fs) {
-        switch (fs) {
-        case ForStream::Yes: {
-            queueOperation(currentSeqno, type, myCollection);
-            break;
-        }
-        case ForStream::No: {
-            queueOperation(currentSeqno, type, CollectionEntry::vegetable);
-            break;
-        }
-        }
-
         switch (type) {
         case InputType::Mutation:
         case InputType::Prepare:
@@ -152,71 +141,89 @@ class CollectionsSeqnoAdvanced
         case InputType::CPEnd:
             break;
         }
+
+        switch (fs) {
+        case ForStream::Yes: {
+            queueOperation(type, myCollection);
+            break;
+        }
+        case ForStream::No: {
+            queueOperation(type, CollectionEntry::vegetable);
+            break;
+        }
+        }
     }
 
-    void queueOperation(uint64_t seqno, InputType type, CollectionID cid) {
+    void queueOperation(InputType type, CollectionID cid) {
         switch (type) {
         case InputType::Mutation: {
-            queueMutation(seqno, cid);
+            queueMutation(cid);
             break;
         }
         case InputType::Prepare: {
-            queuePrepare(seqno, cid);
+            queuePrepare(cid);
             break;
         }
         case InputType::CPEndStart: {
-            queueCPEnd(seqno);
-            queueCPStart(seqno);
+            queueCPEnd();
+            queueCPStart();
             break;
         }
         case InputType::CPStart: {
-            queueCPStart(seqno);
+            queueCPStart();
             break;
         }
         case InputType::CPEnd: {
-            queueCPEnd(seqno);
+            queueCPEnd();
             break;
         }
         }
     }
 
-    void queueMutation(uint64_t seqno, CollectionID cid) {
+    void queueMutation(CollectionID cid) {
         auto item = makeCommittedItem(
-                makeStoredDocKey(std::to_string(seqno), cid), "value");
-        item->setBySeqno(seqno);
+                makeStoredDocKey(std::to_string(currentSeqno), cid), "value");
+        item->setBySeqno(currentSeqno);
         input.items.emplace_back(item);
     }
 
-    void queuePrepare(uint64_t seqno, CollectionID cid) {
+    void queuePrepare(CollectionID cid) {
         auto item = makePendingItem(
-                makeStoredDocKey(std::to_string(seqno), cid), "value");
-        item->setBySeqno(seqno);
+                makeStoredDocKey(std::to_string(currentSeqno), cid), "value");
+        item->setBySeqno(currentSeqno);
         input.items.emplace_back(item);
     }
 
-    void queueCPStart(uint64_t seqno) {
+    void queueCPStart() {
         queue_op checkpoint_op = queue_op::checkpoint_start;
         StoredDocKey key(to_string(checkpoint_op), CollectionID::System);
-        queued_item qi(new Item(key, vbid, checkpoint_op, 1, seqno));
-
-        input.ranges.push_back(
-                {{static_cast<uint64_t>(input.items.back()->getBySeqno()),
-                  seqno},
-                 {},
-                 {}});
-
+        queued_item qi(new Item(key, vbid, checkpoint_op, 1, currentSeqno + 1));
         input.items.emplace_back(qi);
+        checkStart = currentSeqno + 1;
+        input.ranges.push_back({{currentSeqno + 1, currentSeqno + 1}, {}, {}});
     }
 
-    void queueCPEnd(uint64_t seqno) {
+    void queueCPEnd() {
+        auto endSeqno = currentSeqno;
+        if (!input.items.empty()) {
+            endSeqno = static_cast<uint64_t>(input.items.back()->getBySeqno());
+        }
+        if (input.ranges.empty()) {
+            input.ranges.push_back({{1, endSeqno}, {}, {}});
+        } else {
+            input.ranges.back().range.setEnd(endSeqno);
+        }
+
         queue_op checkpoint_op = queue_op::checkpoint_end;
         StoredDocKey key(to_string(checkpoint_op), CollectionID::System);
-        queued_item qi(new Item(key, vbid, checkpoint_op, 1, seqno));
+        queued_item qi(new Item(key, vbid, checkpoint_op, 1, currentSeqno + 1));
         input.items.emplace_back(qi);
     }
 
     // Starting seqno, each operation will increment this
     uint64_t currentSeqno{1};
+    uint64_t checkStart{1};
+    uint64_t checkEnd{1};
 
     std::shared_ptr<MockDcpProducer> producer;
     std::shared_ptr<MockActiveStream> stream;
diff --git a/engines/ep/tests/module_tests/dcp_stream_test.cc b/engines/ep/tests/module_tests/dcp_stream_test.cc
@@ -1146,7 +1146,7 @@ TEST_P(StreamTest, ProcessItemsCheckpointStartIsLastItem) {
     result.items.push_back(dummy);
     result.items.emplace_back(new Item(
             makeStoredDocKey("end"), vbid, queue_op::checkpoint_end, 1, 10));
-    result.ranges.push_back({{0, 9}, {}, {}, true});
+    result.ranges.push_back({{0, 9}, {}, {}});
 
     stream->public_processItems(result);
     result.items.clear();
@@ -1164,13 +1164,13 @@ TEST_P(StreamTest, ProcessItemsCheckpointStartIsLastItem) {
                                                 queue_op::checkpoint_end,
                                                 1,
                                                 /*seqno*/ 11)));
-    result.ranges.push_back({{10, 10}, {}, {}, true});
+    result.ranges.push_back({{10, 10}, {}, {}});
     result.items.push_back(queued_item(new Item(makeStoredDocKey("start"),
                                                 vbid,
                                                 queue_op::checkpoint_start,
                                                 2,
                                                 /*seqno*/ 11)));
-    result.ranges.push_back({{11, 11}, {}, {}, true});
+    result.ranges.push_back({{11, 11}, {}, {}});
 
     // Test - call processItems() twice: once with the items above, then with
     // a single mutation.
diff --git a/engines/ep/tests/module_tests/dcp_test.cc b/engines/ep/tests/module_tests/dcp_test.cc
diff --git a/include/memcached/limits.h b/include/memcached/limits.h
diff --git a/tests/mcbp/mcbp_dcp_test.cc b/tests/mcbp/mcbp_dcp_test.cc
