fix revision conflict handling

arangod/MMFiles/MMFilesCollection.cpp

@@ -2628,8 +2628,10 @@ int MMFilesCollection::insert(transaction::Methods* trx,
   operation.setRevisions(DocumentDescriptor(),
                          DocumentDescriptor(revisionId, doc.begin()));
 
+  MMFilesDocumentPosition old;
+
   try {
-    insertRevision(revisionId, marker->vpack(), 0, true, true);
+    old = insertRevision(revisionId, marker->vpack(), 0, true, true);
     // and go on with the insertion...
   } catch (basics::Exception const& ex) {
     return ex.code();
@@ -2667,6 +2669,9 @@ int MMFilesCollection::insert(transaction::Methods* trx,
       // from the list of revisions
       try {
         removeRevision(revisionId, false);
+        if (old) {
+          insertRevision(old, true);
+        }
       } catch (...) {
       }
       throw;
@@ -2674,6 +2679,10 @@ int MMFilesCollection::insert(transaction::Methods* trx,
 
     if (res != TRI_ERROR_NO_ERROR) {
       operation.revert(trx);
+        
+      if (old) {
+        insertRevision(old, true);
+      }
     }
   }
 
@@ -2738,10 +2747,14 @@ uint8_t const* MMFilesCollection::lookupRevisionVPackConditional(TRI_voc_rid_t r
   return vpack; 
 }
 
-void MMFilesCollection::insertRevision(TRI_voc_rid_t revisionId, uint8_t const* dataptr, TRI_voc_fid_t fid, bool isInWal, bool shouldLock) {
+MMFilesDocumentPosition MMFilesCollection::insertRevision(TRI_voc_rid_t revisionId, uint8_t const* dataptr, TRI_voc_fid_t fid, bool isInWal, bool shouldLock) {
   TRI_ASSERT(revisionId != 0);
   TRI_ASSERT(dataptr != nullptr);
-  _revisionsCache.insert(revisionId, dataptr, fid, isInWal, shouldLock);
+  return _revisionsCache.insert(revisionId, dataptr, fid, isInWal, shouldLock);
+}
+
+void MMFilesCollection::insertRevision(MMFilesDocumentPosition const& position, bool shouldLock) {
+  return _revisionsCache.insert(position, shouldLock);
 }
 
 void MMFilesCollection::updateRevision(TRI_voc_rid_t revisionId, uint8_t const* dataptr, TRI_voc_fid_t fid, bool isInWal) {

arangod/MMFiles/MMFilesCollection.h

@@ -29,6 +29,7 @@
 #include "Indexes/IndexLookupContext.h"
 #include "MMFiles/MMFilesDatafileStatistics.h"
 #include "MMFiles/MMFilesDitch.h"
+#include "MMFiles/MMFilesDocumentPosition.h"
 #include "MMFiles/MMFilesRevisionsCache.h"
 #include "VocBase/KeyGenerator.h"
 #include "VocBase/ManagedDocumentResult.h"
@@ -352,9 +353,11 @@ class MMFilesCollection final : public PhysicalCollection {
                         TRI_voc_rid_t newRevisionId,
                         velocypack::Slice const& newDoc);
 
-  void insertRevision(TRI_voc_rid_t revisionId, uint8_t const* dataptr,
+  MMFilesDocumentPosition insertRevision(TRI_voc_rid_t revisionId, uint8_t const* dataptr,
                                          TRI_voc_fid_t fid, bool isInWal, bool shouldLock);
 
+  void insertRevision(MMFilesDocumentPosition const& position, bool shouldLock);
+
   void updateRevision(TRI_voc_rid_t revisionId, uint8_t const* dataptr,
                       TRI_voc_fid_t fid, bool isInWal);
 

arangod/MMFiles/MMFilesRevisionsCache.cpp

@@ -93,7 +93,7 @@ void MMFilesRevisionsCache::clear() {
   _positions.truncate([](MMFilesDocumentPosition&) { return true; });
 }
 
-void MMFilesRevisionsCache::insert(TRI_voc_rid_t revisionId, uint8_t const* dataptr, TRI_voc_fid_t fid, bool isInWal, bool shouldLock) {
+MMFilesDocumentPosition MMFilesRevisionsCache::insert(TRI_voc_rid_t revisionId, uint8_t const* dataptr, TRI_voc_fid_t fid, bool isInWal, bool shouldLock) {
   TRI_ASSERT(revisionId != 0);
   TRI_ASSERT(dataptr != nullptr);
 
@@ -101,9 +101,17 @@ void MMFilesRevisionsCache::insert(TRI_voc_rid_t revisionId, uint8_t const* data
   int res = _positions.insert(nullptr, MMFilesDocumentPosition(revisionId, dataptr, fid, isInWal));
 
   if (res != TRI_ERROR_NO_ERROR) {
-    _positions.removeByKey(nullptr, &revisionId);
+    MMFilesDocumentPosition old = _positions.removeByKey(nullptr, &revisionId);
     _positions.insert(nullptr, MMFilesDocumentPosition(revisionId, dataptr, fid, isInWal));
+    return old;
   }
+
+  return MMFilesDocumentPosition();
+}
+
+void MMFilesRevisionsCache::insert(MMFilesDocumentPosition const& position, bool shouldLock) {
+  CONDITIONAL_WRITE_LOCKER(locker, _lock, shouldLock);
+  _positions.insert(nullptr, position);
 }
 
 void MMFilesRevisionsCache::update(TRI_voc_rid_t revisionId, uint8_t const* dataptr, TRI_voc_fid_t fid, bool isInWal) {

arangod/MMFiles/MMFilesRevisionsCache.h

@@ -46,7 +46,8 @@ class MMFilesRevisionsCache {
   size_t memoryUsage();
   void clear();
   MMFilesDocumentPosition lookup(TRI_voc_rid_t revisionId) const;
-  void insert(TRI_voc_rid_t revisionId, uint8_t const* dataptr, TRI_voc_fid_t fid, bool isInWal, bool shouldLock);
+  MMFilesDocumentPosition insert(TRI_voc_rid_t revisionId, uint8_t const* dataptr, TRI_voc_fid_t fid, bool isInWal, bool shouldLock);
+  void insert(MMFilesDocumentPosition const& position, bool shouldLock);
   void update(TRI_voc_rid_t revisionId, uint8_t const* dataptr, TRI_voc_fid_t fid, bool isInWal);
   bool updateConditional(TRI_voc_rid_t revisionId, TRI_df_marker_t const* oldPosition, TRI_df_marker_t const* newPosition, TRI_voc_fid_t newFid, bool isInWal);
   void remove(TRI_voc_rid_t revisionId);