[CINFRA] Fix propagation of writeConcern to collections and groups in Replication2 (arangodb#20434)

Author: mchacki

arangod/Cluster/AgencyCache.cpp

@@ -799,7 +799,7 @@ AgencyCache::change_set_t AgencyCache::changedSince(
   if (get_rest) {  // All the rest, i.e. All keys excluding the usual suspects
     static std::vector<std::string> const exc{
         "Analyzers", "Collections",    "Databases",
-        "Views",     "ReplicatedLogs", "ReplicatedStates"};
+        "Views",     "ReplicatedLogs", "CollectionGroups"};
     auto keys = _readDB.nodePtr(AgencyCommHelper::path(what))->keys();
     keys.erase(std::remove_if(std::begin(keys), std::end(keys),
                               [&](auto const& x) {

arangod/Replication2/Supervision/CollectionGroupSupervision.cpp

@@ -933,13 +933,15 @@ struct TransactionBuilder {
   }
 
   void operator()(UpdateCollectionGroupInPlan const& action) {
-    auto write = env.write().emplace_object(
-        basics::StringUtils::concatT("/arango/Plan/CollectionGroups/", database,
-                                     "/", action.id.id(),
-                                     "/attributes/mutable"),
-        [&](VPackBuilder& builder) {
-          velocypack::serialize(builder, action.spec);
-        });
+    auto write =
+        env.write()
+            .emplace_object(basics::StringUtils::concatT(
+                                "/arango/Plan/CollectionGroups/", database, "/",
+                                action.id.id(), "/attributes/mutable"),
+                            [&](VPackBuilder& builder) {
+                              velocypack::serialize(builder, action.spec);
+                            })
+            .inc("/arango/Plan/Version");
     env = write.precs()
               .isNotEmpty(basics::StringUtils::concatT(
                   "/arango/Target/CollectionGroups/", database, "/", gid.id()))

arangod/Sharding/ShardingInfo.cpp

@@ -322,7 +322,9 @@ LogicalCollection* ShardingInfo::collection() const noexcept {
   return _collection;
 }
 
-void ShardingInfo::toVelocyPack(VPackBuilder& result, bool translateCids,
+void ShardingInfo::toVelocyPack(VPackBuilder& result,
+                                bool ignoreCollectionGroupAttributes,
+                                bool translateCids,
                                 bool includeShardsEntry) const {
   result.add(StaticStrings::NumberOfShards, VPackValue(_numberOfShards));
 
@@ -345,18 +347,21 @@ void ShardingInfo::toVelocyPack(VPackBuilder& result, bool translateCids,
     result.close();  // shards
   }
 
-  if (isSatellite()) {
-    result.add(StaticStrings::ReplicationFactor,
-               VPackValue(StaticStrings::Satellite));
-  } else {
-    result.add(StaticStrings::ReplicationFactor,
-               VPackValue(_replicationFactor));
+  if (!ignoreCollectionGroupAttributes) {
+    // For replication Two this class is not responsible for the following
+    // attributes.
+    if (isSatellite()) {
+      result.add(StaticStrings::ReplicationFactor,
+                 VPackValue(StaticStrings::Satellite));
+    } else {
+      result.add(StaticStrings::ReplicationFactor,
+                 VPackValue(_replicationFactor));
+    }
+    // minReplicationFactor deprecated in 3.6
+    result.add(StaticStrings::WriteConcern, VPackValue(_writeConcern));
+    result.add(StaticStrings::MinReplicationFactor, VPackValue(_writeConcern));
   }
 
-  // minReplicationFactor deprecated in 3.6
-  result.add(StaticStrings::WriteConcern, VPackValue(_writeConcern));
-  result.add(StaticStrings::MinReplicationFactor, VPackValue(_writeConcern));
-
   if (!_distributeShardsLike.empty()) {
     if (ServerState::instance()->isCoordinator()) {
       // We either want to expose _distributeShardsLike if we're either on a

arangod/Sharding/ShardingInfo.h

@@ -60,7 +60,9 @@ class ShardingInfo {
   std::string shardingStrategyName() const;
 
   LogicalCollection* collection() const noexcept;
-  void toVelocyPack(arangodb::velocypack::Builder& result, bool translateCids,
+
+  void toVelocyPack(arangodb::velocypack::Builder& result,
+                    bool ignoreCollectionGroupAttributes, bool translateCids,
                     bool includeShardsEntry = true) const;
 
   std::string const& distributeShardsLike() const noexcept;

arangod/VocBase/LogicalCollection.cpp

@@ -27,14 +27,13 @@
 #include "ApplicationFeatures/ApplicationServer.h"
 #include "Aql/QueryCache.h"
 #include "Basics/DownCast.h"
-#include "Basics/ReadLocker.h"
 #include "Basics/StaticStrings.h"
 #include "Basics/VelocyPackHelper.h"
 #include "Basics/WriteLocker.h"
+#include "Cluster/ClusterCollectionMethods.h"
 #include "Cluster/ClusterFeature.h"
 #include "Cluster/ClusterInfo.h"
 #include "Cluster/ClusterMethods.h"
-#include "Cluster/ClusterCollectionMethods.h"
 #include "Cluster/FollowerInfo.h"
 #include "Cluster/ServerState.h"
 #include "Logger/LogMacros.h"
@@ -57,7 +56,6 @@
 #include "VocBase/Properties/UserInputCollectionProperties.h"
 #include "VocBase/Validators.h"
 #include "velocypack/Builder.h"
-#include "VocBase/Properties/UserInputCollectionProperties.h"
 
 #ifdef USE_ENTERPRISE
 #include "Enterprise/Sharding/ShardingStrategyEE.h"
@@ -366,6 +364,20 @@ size_t LogicalCollection::replicationFactor() const noexcept {
 }
 
 size_t LogicalCollection::writeConcern() const noexcept {
+  if (_groupId.has_value() && ServerState::instance()->isDBServer()) {
+    TRI_ASSERT(replicationVersion() == replication::Version::TWO)
+        << "Set a groupId although we are not in Replication Two";
+    auto& ci = vocbase().server().getFeature<ClusterFeature>().clusterInfo();
+
+    auto const& group = ci.getCollectionGroupById(
+        replication2::agency::CollectionGroupId{_groupId.value()});
+    if (group) {
+      return group->attributes.mutableAttributes.writeConcern;
+    }
+    // If we cannot find a group this means the shard is dropped
+    // while we are still in this method. Let's just take the
+    // value at creation then.
+  }
   TRI_ASSERT(_sharding != nullptr);
   return _sharding->writeConcern();
 }
@@ -804,6 +816,16 @@ Result LogicalCollection::appendVPack(velocypack::Builder& build,
               VPackValue(std::to_string(planId().id())));
   }
 
+  std::shared_ptr<replication2::agency::CollectionGroupPlanSpecification const>
+      group;
+  if (_groupId.has_value()) {
+    TRI_ASSERT(replicationVersion() == replication::Version::TWO)
+        << "Set a groupId although we are not in Replication Two";
+    auto& ci = vocbase().server().getFeature<ClusterFeature>().clusterInfo();
+    group = ci.getCollectionGroupById(
+        replication2::agency::CollectionGroupId{_groupId.value()});
+  }
+  bool isReplicationTWO = group != nullptr;
 #ifdef USE_ENTERPRISE
   if (isSmart() && type() == TRI_COL_TYPE_EDGE &&
       ServerState::instance()->isRunningInCluster()) {
@@ -816,25 +838,54 @@ Result LogicalCollection::appendVPack(velocypack::Builder& build,
     }
     edgeCollection->shardMapToVelocyPack(build);
     bool includeShardsEntry = false;
-    _sharding->toVelocyPack(build, ctx != Serialization::List,
+    _sharding->toVelocyPack(build, isReplicationTWO, ctx != Serialization::List,
                             includeShardsEntry);
   } else {
-    _sharding->toVelocyPack(build, ctx != Serialization::List);
+    _sharding->toVelocyPack(build, isReplicationTWO,
+                            ctx != Serialization::List);
   }
 #else
-  _sharding->toVelocyPack(build, ctx != Serialization::List);
+  _sharding->toVelocyPack(build, isReplicationTWO, ctx != Serialization::List);
 #endif
-
-  includeVelocyPackEnterprise(build);
-  TRI_ASSERT(build.isOpenObject());
-
-  if (replicationVersion() == replication::Version::TWO &&
-      _groupId.has_value()) {
+  if (group) {
     build.add("groupId", VPackValue(_groupId.value()));
     if (_replicatedStateId) {
       build.add("replicatedStateId", VPackValue(*_replicatedStateId));
     }
+    // For replication1 the _sharding is responsible.
+    // For TWO the group contains those attributes
+
+    // Future TODO: It would be nice if we could use a "serializeInPlace"
+    // method Which does serialization but not the open/close object.
+    TRI_ASSERT(!build.hasKey(StaticStrings::ReplicationFactor))
+        << "replicationFactor already serialized from _sharding";
+    TRI_ASSERT(!build.hasKey(StaticStrings::WriteConcern))
+        << "writeConcern already serialized from _sharding";
+    TRI_ASSERT(!build.hasKey(StaticStrings::MinReplicationFactor))
+        << "minReplicationFactor already serialized from _sharding";
+    if (group->attributes.mutableAttributes.replicationFactor == 0) {
+      build.add(StaticStrings::ReplicationFactor,
+                VPackValue(StaticStrings::Satellite));
+      // For Backwards Compatibility, WriteConcern should return 0 here for
+      // satellites
+      build.add(StaticStrings::WriteConcern, VPackValue(0));
+      // minReplicationFactor deprecated in 3.6
+      build.add(StaticStrings::MinReplicationFactor, VPackValue(0));
+
+    } else {
+      build.add(
+          StaticStrings::ReplicationFactor,
+          VPackValue(group->attributes.mutableAttributes.replicationFactor));
+      build.add(StaticStrings::WriteConcern,
+                VPackValue(group->attributes.mutableAttributes.writeConcern));
+      // minReplicationFactor deprecated in 3.6
+      build.add(StaticStrings::MinReplicationFactor,
+                VPackValue(group->attributes.mutableAttributes.writeConcern));
+    }
   }
+
+  includeVelocyPackEnterprise(build);
+  TRI_ASSERT(build.isOpenObject());
   // We leave the object open
   return {};
 }
@@ -904,7 +955,7 @@ Result LogicalCollection::properties(velocypack::Slice slice) {
   }
 
   size_t replicationFactor = _sharding->replicationFactor();
-  size_t writeConcern = _sharding->writeConcern();
+  size_t writeConcern = this->writeConcern();
   VPackSlice replicationFactorSlice =
       slice.get(StaticStrings::ReplicationFactor);
 
@@ -1004,7 +1055,7 @@ Result LogicalCollection::properties(velocypack::Slice slice) {
       if ((ServerState::instance()->isCoordinator() ||
            (ServerState::instance()->isSingleServer() &&
             (isSatellite() || isSmart()))) &&
-          writeConcern != _sharding->writeConcern()) {  // check if changed
+          writeConcern != this->writeConcern()) {  // check if changed
         if (!_sharding->distributeShardsLike().empty()) {
           CollectionNameResolver resolver(vocbase());
           std::string name = resolver.getCollectionNameCluster(DataSourceId{

tests/js/client/shell/shell-collection-properties-propagation-cluster.js

@@ -156,6 +156,12 @@ function CollectionPropertiesPropagationSuite() {
         internal.sleep(1);
       }
       assertEqual(2, servers.length);
+
+      // For ReplicationVersion 2 shards and followerCollections
+      // all report the change in ReplicationFactor.
+      // For ReplicationVersion 1 only the Leading collection
+      // knows about the change.
+      const expectedRF = db._properties().replicationVersion === "2" ? 2 : 3;
 
       // shards do not see the replicationFactor update
       let keys;
@@ -168,7 +174,7 @@ function CollectionPropertiesPropagationSuite() {
         assertTrue(keys.length > 0);
         let found = 0;
         keys.forEach((s) => {
-          if (p[s].replicationFactor === 3) {
+          if (p[s].replicationFactor === expectedRF) {
             ++found;
           }
         });
@@ -179,28 +185,18 @@ function CollectionPropertiesPropagationSuite() {
       }
 
       keys.forEach((s) => {
-        assertEqual(3, p[s].replicationFactor, {s, p});
+        assertEqual(expectedRF, p[s].replicationFactor, {s, p});
       });
 
       p = c.properties();
       assertEqual(proto, p.distributeShardsLike);
-      let expectedRF;
-      if (db._properties().replicationVersion === "2") {
-        // with replication 2, the collection's replicationFactor is tied to the replicationFactor
-        // of the collection group, so it will report the correct value#
-        assertEqual(2, p.replicationFactor, p);
-      } else {
-        // with replication1, dependent collection won't see the replicationFactor update
-        assertEqual(3, p.replicationFactor, p);
-      }
+      assertEqual(expectedRF, p.replicationFactor, p);
 
-      // the individual shards currently do not report the updated replicationFactor,
-      // neither for replication 1 nor 2
       p = propertiesOnDBServers(c);
       keys = Object.keys(p);
       assertTrue(keys.length > 0);
       keys.forEach((s) => {
-        assertEqual(3, p[s].replicationFactor, {s, p});
+        assertEqual(expectedRF, p[s].replicationFactor, {s, p});
       });
     },
 
@@ -210,7 +206,7 @@ function CollectionPropertiesPropagationSuite() {
       db[proto].properties({ writeConcern: 1 });
 
       let p = db[proto].properties();
-      assertEqual(1, p.writeConcern);
+      assertEqual(1, p.writeConcern, p);
 
       // shards will see the writeConcern update
       let keys;
@@ -238,16 +234,24 @@ function CollectionPropertiesPropagationSuite() {
       });
 
       // dependent collection won't see the writeConcern update
+      const isReplicationTwo = db._properties().replicationVersion === "2";
+      // In Replication1 the followers won't see the writeConcern update
+      // as they are not changed
+      // In Replication2 the followers share the replicated log with the
+      // leader, and the log holds the writeConcern, hence all are updated.
+      // The above holds true for the collection as well as the corresponding
+      // shards.
+      const expectedFollowerWriteConcern = isReplicationTwo ? 1 : 2;
       p = c.properties();
       assertEqual(proto, p.distributeShardsLike);
-      assertEqual(2, p.writeConcern);
 
-      // nor do its shards
+      assertEqual(expectedFollowerWriteConcern, p.writeConcern);
+
       p = propertiesOnDBServers(c);
       keys = Object.keys(p);
       assertTrue(keys.length > 0);
       keys.forEach((s) => {
-        assertEqual(2, p[s].writeConcern, {s, p});
+        assertEqual(expectedFollowerWriteConcern, p[s].writeConcern, {s, p});
       });
     },