improvements and more tests for replication (#4624)

arangod/Replication/Syncer.cpp

@@ -660,6 +660,13 @@ Result Syncer::createIndex(VPackSlice const& slice) {
     return Result(TRI_ERROR_ARANGO_COLLECTION_NOT_FOUND,
                   "did not find collection for index");
   }
+  
+  VPackBuilder s;
+  s.openObject();
+  s.add("objectId", VPackSlice::nullSlice());
+  s.close();
+  VPackBuilder merged = VPackCollection::merge(indexSlice, s.slice(),
+                                               /*mergeValues*/true, /*nullMeansRemove*/true);
 
   try {
     SingleCollectionTransaction trx(transaction::StandaloneContext::Create(vocbase),
@@ -674,7 +681,7 @@ Result Syncer::createIndex(VPackSlice const& slice) {
     auto physical = trx.documentCollection()->getPhysical();
     TRI_ASSERT(physical != nullptr);
     std::shared_ptr<arangodb::Index> idx;
-    res = physical->restoreIndex(&trx, indexSlice, idx);
+    res = physical->restoreIndex(&trx, merged.slice(), idx);
     res = trx.finish(res);
 
     return res;

arangod/Replication/TailingSyncer.cpp

@@ -314,14 +314,18 @@ Result TailingSyncer::processDocument(TRI_replication_operation_e type,
   // extract "rev"
   VPackSlice const rev = doc.get(StaticStrings::RevString);
 
-  if (!rev.isString()) {
+  if (!rev.isNone() && !rev.isString()) {
+    // _rev is an optional attribute
     return Result(TRI_ERROR_REPLICATION_INVALID_RESPONSE, "invalid document revision format");
   }
 
   _documentBuilder.clear();
   _documentBuilder.openObject();
   _documentBuilder.add(StaticStrings::KeyString, key);
-  _documentBuilder.add(StaticStrings::RevString, rev);
+  if (rev.isString()) {
+    // _rev is an optional attribute
+    _documentBuilder.add(StaticStrings::RevString, rev);
+  }
   _documentBuilder.close();
 
   VPackSlice const old = _documentBuilder.slice();

js/server/tests/replication/replication-ongoing-32.js

@@ -356,6 +356,73 @@ function BaseTestConfig() {
         }
       );
     },
+    
+    ////////////////////////////////////////////////////////////////////////////////
+    /// @brief test insert
+    ////////////////////////////////////////////////////////////////////////////////
+
+    testInsert: function() {
+      connectToMaster();
+
+      compare(
+        function(state) {
+          db._create(cn);
+        },
+
+        function(state) {
+          for (let i = 0; i < 1000; ++i) {
+            db[cn].insert({ _key: "test" + i, value: i });
+          }
+
+          state.checksum = collectionChecksum(cn);
+          state.count = collectionCount(cn);
+          assertEqual(1000, state.count);
+        },
+
+        function(state) {
+        },
+
+        function(state) {
+          assertEqual(state.count, collectionCount(cn));
+          assertEqual(state.checksum, collectionChecksum(cn));
+        }
+      );
+    },
+    
+    ////////////////////////////////////////////////////////////////////////////////
+    /// @brief test remove
+    ////////////////////////////////////////////////////////////////////////////////
+
+    testRemove: function() {
+      connectToMaster();
+
+      compare(
+        function(state) {
+          db._create(cn);
+        },
+
+        function(state) {
+          for (let i = 0; i < 1000; ++i) {
+            db[cn].insert({ _key: "test" + i, value: i });
+          }
+          for (let i = 0; i < 100; ++i) {
+            db[cn].remove("test" + i);
+          }
+
+          state.checksum = collectionChecksum(cn);
+          state.count = collectionCount(cn);
+          assertEqual(900, state.count);
+        },
+
+        function(state) {
+        },
+
+        function(state) {
+          assertEqual(state.count, collectionCount(cn));
+          assertEqual(state.checksum, collectionChecksum(cn));
+        }
+      );
+    },
 
     ////////////////////////////////////////////////////////////////////////////////
     /// @brief test long transaction, blocking