Fix garbage collector race condition in storage v2

Summary:
The garbage collector had a race condition when it would delete deltas that
were in the middle of an object's delta chain. In the process of deleting
(unlinking) the delta, the garbage collector previously wouldn't acquire any
locks. That operation was then racing with the standard MVCC
`CreateAndLinkDelta` function that adds a new delta into the chain.
Fortunately, `CreateAndLinkDelta` always does its modifications while holding a
lock to the owner of the chain (either a vertex or an edge) so this change just
adds the lock acquiring to the garbage collector.

Reviewers: teon.banek

Reviewed By: teon.banek

Subscribers: pullbot

Differential Revision: https://phabricator.memgraph.io/D2582

src/storage/v2/delta.hpp

@@ -90,6 +90,24 @@ class PreviousPtr {
   std::atomic<uintptr_t> storage_;
 };
 
+inline bool operator==(const PreviousPtr::Pointer &a,
+                       const PreviousPtr::Pointer &b) {
+  if (a.type != b.type) return false;
+  switch (a.type) {
+    case PreviousPtr::Type::VERTEX:
+      return a.vertex == b.vertex;
+    case PreviousPtr::Type::EDGE:
+      return a.edge == b.edge;
+    case PreviousPtr::Type::DELTA:
+      return a.delta == b.delta;
+  }
+}
+
+inline bool operator!=(const PreviousPtr::Pointer &a,
+                       const PreviousPtr::Pointer &b) {
+  return !(a == b);
+}
+
 struct Delta {
   enum class Action {
     // Used for both Vertex and Edge

src/storage/v2/mvcc.hpp

@@ -92,11 +92,22 @@ inline void CreateAndLinkDelta(Transaction *transaction, TObj *object,
   // chains are valid at all times. The chains must be valid at all times
   // because garbage collection (which traverses the chains) is done
   // concurrently (as well as other execution threads).
+
+  // 1. We need to set the next delta of the new delta to the existing delta.
+  delta->next.store(object->delta, std::memory_order_release);
+  // 2. We need to set the previous delta of the new delta to the object.
   delta->prev.Set(object);
+  // 3. We need to set the previous delta of the existing delta to the new
+  // delta. After this point the garbage collector will be able to see the new
+  // delta but won't modify it until we are done with all of our modifications.
   if (object->delta) {
     object->delta->prev.Set(delta);
   }
-  delta->next.store(object->delta, std::memory_order_release);
+  // 4. Finally, we need to set the object's delta to the new delta. The garbage
+  // collector and other transactions will acquire the object lock to read the
+  // delta from the object. Because the lock is held during the whole time this
+  // modification is being done, everybody else will wait until we are fully
+  // done with our modification before they read the object's delta value.
   object->delta = delta;
 }
 

src/storage/v2/storage.cpp

@@ -1086,8 +1086,9 @@ void Storage::CollectGarbage() {
       transaction = &committed_transactions_ptr->front();
     }
 
-    if (transaction->commit_timestamp->load(std::memory_order_acquire) >=
-        oldest_active_start_timestamp) {
+    auto commit_timestamp =
+        transaction->commit_timestamp->load(std::memory_order_acquire);
+    if (commit_timestamp >= oldest_active_start_timestamp) {
       break;
     }
 
@@ -1109,6 +1110,18 @@ void Storage::CollectGarbage() {
     // When processing a delta that is the first one in its chain, we
     // obtain the corresponding vertex or edge lock, and then verify that this
     // delta still is the first in its chain.
+    // When processing a delta that is in the middle of the chain we only
+    // process the final delta of the given transaction in that chain. We
+    // determine the owner of the chain (either a vertex or an edge), obtain the
+    // corresponding lock, and then verify that this delta is still in the same
+    // position as it was before taking the lock.
+    //
+    // Even though the delta chain is lock-free (both `next` and `prev`) the
+    // chain should not be modified without taking the lock from the object that
+    // owns the chain (either a vertex or an edge). Modifying the chain without
+    // taking the lock will cause subtle race conditions that will leave the
+    // chain in a broken state.
+    // The chain can be only read without taking any locks.
 
     for (Delta &delta : transaction->deltas) {
       while (true) {
@@ -1143,6 +1156,38 @@ void Storage::CollectGarbage() {
             break;
           }
           case PreviousPtr::Type::DELTA: {
+            if (prev.delta->timestamp->load(std::memory_order_release) ==
+                commit_timestamp) {
+              // The delta that is newer than this one is also a delta from this
+              // transaction. We skip the current delta and will remove it as a
+              // part of the suffix later.
+              break;
+            }
+            std::unique_lock<utils::SpinLock> guard;
+            {
+              // We need to find the parent object in order to be able to use
+              // its lock.
+              auto parent = prev;
+              while (parent.type == PreviousPtr::Type::DELTA) {
+                parent = parent.delta->prev.Get();
+              }
+              switch (parent.type) {
+                case PreviousPtr::Type::VERTEX:
+                  guard =
+                      std::unique_lock<utils::SpinLock>(parent.vertex->lock);
+                  break;
+                case PreviousPtr::Type::EDGE:
+                  guard = std::unique_lock<utils::SpinLock>(parent.edge->lock);
+                  break;
+                case PreviousPtr::Type::DELTA:
+                  LOG(FATAL) << "Invalid database state!";
+              }
+            }
+            if (delta.prev.Get() != prev) {
+              // Something changed, we could now be the first delta in the
+              // chain.
+              continue;
+            }
             Delta *prev_delta = prev.delta;
             prev_delta->next.store(nullptr, std::memory_order_release);
             break;