fixes for sync

arangod/Replication/InitialSyncer.cpp

@@ -262,6 +262,11 @@ int InitialSyncer::run (string& errorMsg,
 
   sendFinishBatch();
 
+  if (res != TRI_ERROR_NO_ERROR &&
+      errorMsg.empty()) {
+    errorMsg = TRI_errno_string(res);
+  }
+
   return res;
 }
 
@@ -1240,13 +1245,16 @@ int InitialSyncer::handleSyncKeys (std::string const& keysId,
           int res = TRI_ERROR_NO_ERROR;
           auto mptr = idx->lookupKey(documentKey.c_str());
 
-          if (mptr == nullptr) {
+          if (mptr == nullptr || isEdge) {
+            // in case of an edge collection we must always update
             TRI_document_edge_t* e = nullptr;
             TRI_document_edge_t edge;
+            std::string from;
+            std::string to;
 
             if (isEdge) {
-              std::string const from = JsonHelper::getStringValue(documentJson, TRI_VOC_ATTRIBUTE_FROM, "");
+              from = JsonHelper::getStringValue(documentJson, TRI_VOC_ATTRIBUTE_FROM, "");
-              std::string const to   = JsonHelper::getStringValue(documentJson, TRI_VOC_ATTRIBUTE_TO, "");
+              to   = JsonHelper::getStringValue(documentJson, TRI_VOC_ATTRIBUTE_TO, "");
 
               // parse _from
               if (! DocumentHelper::parseDocumentId(*trx.resolver(), from.c_str(), edge._fromCid, &edge._fromKey)) {
@@ -1263,6 +1271,11 @@ int InitialSyncer::handleSyncKeys (std::string const& keysId,
 
             // INSERT
             if (res == TRI_ERROR_NO_ERROR) {
+              if (mptr != nullptr && isEdge) {
+                // must remove existing edge first
+                TRI_RemoveShapedJsonDocumentCollection(trx.trxCollection(), (TRI_voc_key_t) documentKey.c_str(), 0, nullptr, &policy, false, false);
+              }
+
               res = TRI_InsertShapedJsonDocumentCollection(trx.trxCollection(), (TRI_voc_key_t) documentKey.c_str(), rid, nullptr, &result, shaped, e, false, false, true);
             }
           }
@@ -1285,6 +1298,40 @@ int InitialSyncer::handleSyncKeys (std::string const& keysId,
   return TRI_ERROR_NO_ERROR;
 }
 
+////////////////////////////////////////////////////////////////////////////////
+/// @brief changes the properties of a collection, based on the JSON provided
+////////////////////////////////////////////////////////////////////////////////
+
+int InitialSyncer::changeCollection (TRI_vocbase_col_t* col,
+                                     TRI_json_t const* json) {
+
+  bool waitForSync      = JsonHelper::getBooleanValue(json, "waitForSync", false);
+  bool doCompact        = JsonHelper::getBooleanValue(json, "doCompact", true);
+  int maximalSize       = JsonHelper::getNumericValue<int>(json, "maximalSize", TRI_JOURNAL_DEFAULT_MAXIMAL_SIZE);
+  uint32_t indexBuckets = JsonHelper::getNumericValue<uint32_t>(json, "indexBuckets", TRI_DEFAULT_INDEX_BUCKETS);
+
+  try {
+    triagens::arango::CollectionGuard guard(_vocbase, col->_cid);
+
+    TRI_col_info_t parameters;
+
+    // only need to set these three properties as the others cannot be updated on the fly
+    parameters._doCompact    = doCompact;
+    parameters._maximalSize  = maximalSize;
+    parameters._waitForSync  = waitForSync;
+    parameters._indexBuckets = indexBuckets;
+
+    bool doSync = _vocbase->_settings.forceSyncProperties;
+    return TRI_UpdateCollectionInfo(_vocbase, guard.collection()->_collection, &parameters, doSync);
+  }
+  catch (triagens::basics::Exception const& ex) {
+    return ex.code();
+  }
+  catch (...) {
+    return TRI_ERROR_INTERNAL;
+  }
+}
+
 ////////////////////////////////////////////////////////////////////////////////
 /// @brief handle the information about a collection
 ////////////////////////////////////////////////////////////////////////////////
@@ -1421,7 +1468,7 @@ int InitialSyncer::handleCollection (TRI_json_t const* parameters,
 
       if (col != nullptr) {
         // collection is already present
-        return TRI_ERROR_NO_ERROR;
+        return changeCollection(col, parameters);
       }
     }
 

arangod/Replication/InitialSyncer.h

@@ -231,6 +231,13 @@ namespace triagens {
                             TRI_voc_tick_t,
                             std::string&);
 
+////////////////////////////////////////////////////////////////////////////////
+/// @brief changes the properties of a collection, based on the JSON provided
+////////////////////////////////////////////////////////////////////////////////
+
+        int changeCollection (TRI_vocbase_col_t*,
+                              struct TRI_json_t const*);
+
 ////////////////////////////////////////////////////////////////////////////////
 /// @brief handle the information about a collection
 ////////////////////////////////////////////////////////////////////////////////

js/server/tests/replication-sync.js

@@ -117,6 +117,211 @@ function ReplicationSuite () {
       db._drop(cn);
     },
 
+////////////////////////////////////////////////////////////////////////////////
+/// @brief test collection properties
+////////////////////////////////////////////////////////////////////////////////
+
+    testProperties : function () {
+      connectToMaster();
+
+      compare(
+        function (state) {
+          var c = db._create(cn);
+
+          c.properties({ indexBuckets: 32, waitForSync: true, journalSize: 16 * 1024 * 1024 });
+        },
+        function (state) {
+          // don't create the collection on the slave
+        },
+        function (state) {
+          var c = db._collection(cn);
+          var p = c.properties();
+          assertEqual(32, p.indexBuckets);
+          assertTrue(p.waitForSync);
+          assertEqual(16 * 1024 * 1024, p.journalSize);
+        },
+        true
+      );
+    },
+
+////////////////////////////////////////////////////////////////////////////////
+/// @brief test collection properties
+////////////////////////////////////////////////////////////////////////////////
+
+    testPropertiesOther : function () {
+      connectToMaster();
+
+      compare(
+        function (state) {
+          var c = db._create(cn);
+
+          c.properties({ indexBuckets: 32, waitForSync: true, journalSize: 16 * 1024 * 1024 });
+        },
+        function (state) {
+          // create the collection on the slave, but with default properties
+          db._create(cn);
+        },
+        function (state) {
+          var c = db._collection(cn);
+          var p = c.properties();
+          assertEqual(32, p.indexBuckets);
+          assertTrue(p.waitForSync);
+          assertEqual(16 * 1024 * 1024, p.journalSize);
+        },
+        true
+      );
+    },
+
+////////////////////////////////////////////////////////////////////////////////
+/// @brief test with indexes
+////////////////////////////////////////////////////////////////////////////////
+
+    testCreateIndexes : function () {
+      connectToMaster();
+
+      compare(
+        function (state) {
+          var c = db._create(cn), i;
+
+          for (i = 0; i < 5000; ++i) {
+            c.save({ _key: "test" + i, "value1" : i, "value2": "test" + i });
+          }
+
+          c.ensureIndex({ type: "hash", fields: [ "value1", "value2" ] });
+          c.ensureIndex({ type: "skiplist", fields: [ "value1" ] });
+
+          state.checksum = collectionChecksum(cn);
+          state.count = collectionCount(cn);
+          assertEqual(5000, state.count);
+        },
+        function (state) {
+          // already create the collection on the slave
+          var c = db._create(cn);
+          
+          for (var i = 0; i < 5000; ++i) {
+            c.save({ _key: "test" + i, "value1" : "test" + i, "value2": i });
+          }
+        },
+        function (state) {
+          assertEqual(state.count, collectionCount(cn));
+          assertEqual(state.checksum, collectionChecksum(cn));
+
+          var idx = db._collection(cn).getIndexes();
+          assertEqual(3, idx.length); // primary + hash + skiplist
+          for (var i = 1; i < idx.length; ++i) {
+            assertFalse(idx[i].unique);
+            assertFalse(idx[i].sparse);
+
+            if (idx[i].type === 'hash') {
+              assertEqual("hash", idx[i].type);
+              assertEqual([ "value1", "value2" ], idx[i].fields);
+            }
+            else {
+              assertEqual("skiplist", idx[i].type);
+              assertEqual([ "value1" ], idx[i].fields);
+            }
+          }
+        },
+        true
+      );
+    },
+
+////////////////////////////////////////////////////////////////////////////////
+/// @brief test with edges
+////////////////////////////////////////////////////////////////////////////////
+
+    testEdges : function () {
+      connectToMaster();
+
+      compare(
+        function (state) {
+          var c = db._createEdgeCollection(cn), i;
+
+          for (i = 0; i < 100; ++i) {
+            c.save(cn + "/test" + i, cn + "/test" + (i % 10), { _key: "test" + i, "value1" : i, "value2": "test" + i });
+          }
+
+          state.checksum = collectionChecksum(cn);
+          state.count = collectionCount(cn);
+          assertEqual(100, state.count);
+        },
+        function (state) {
+          // already create the collection on the slave
+          var c = db._createEdgeCollection(cn);
+          
+          for (var i = 0; i < 100; ++i) {
+            c.save(cn + "/test" + i, cn + "/test" + (i % 10), { _key: "test" + i, "value1" : i, "value2": "test" + i });
+          }
+        },
+        function (state) {
+          assertEqual(state.count, collectionCount(cn));
+          assertEqual(state.checksum, collectionChecksum(cn));
+
+          var c = db._collection(cn);
+          assertEqual(3, c.type());
+
+          for (var i = 0; i < 100; ++i) {
+            var doc = c.document("test" + i);
+            assertEqual(cn + "/test" + i, doc._from);
+            assertEqual(cn + "/test" + (i % 10), doc._to);
+            assertEqual(i, doc.value1);
+            assertEqual("test" + i, doc.value2);
+          }
+        },
+        true
+      );
+    },
+
+////////////////////////////////////////////////////////////////////////////////
+/// @brief test with edges differences
+////////////////////////////////////////////////////////////////////////////////
+
+    testEdgesDifferences : function () {
+      connectToMaster();
+
+      compare(
+        function (state) {
+          var c = db._createEdgeCollection(cn), i;
+
+          for (i = 0; i < 100; ++i) {
+            c.save(cn + "/test" + i, cn + "/test" + (i % 10), { _key: "test" + i, "value1" : i, "value2": "test" + i });
+          }
+
+          state.checksum = collectionChecksum(cn);
+          state.count = collectionCount(cn);
+          assertEqual(100, state.count);
+        },
+        function (state) {
+          // already create the collection on the slave
+          var c = db._createEdgeCollection(cn);
+          
+          for (var i = 0; i < 200; ++i) {
+            c.save(
+              cn + "/test" + (i + 1), 
+              cn + "/test" + (i % 11), 
+              { _key: "test" + i, "value1" : i, "value2": "test" + i }
+            );
+          }
+        },
+        function (state) {
+          assertEqual(state.count, collectionCount(cn));
+          assertEqual(state.checksum, collectionChecksum(cn));
+
+          var c = db._collection(cn);
+          assertEqual(3, c.type());
+
+          for (var i = 0; i < 100; ++i) {
+            var doc = c.document("test" + i);
+            assertEqual(cn + "/test" + i, doc._from);
+            assertEqual(cn + "/test" + (i % 10), doc._to);
+            assertEqual(i, doc.value1);
+            assertEqual("test" + i, doc.value2);
+          }
+        },
+        true
+      );
+    },
+
 ////////////////////////////////////////////////////////////////////////////////
 /// @brief test non-present collection
 ////////////////////////////////////////////////////////////////////////////////