MB-33804 [SR]: Track pending Prepares in PassiveDurabilityMonitor

At DCP_PREPARE received, the PDM must track the new pending Prepare.
That is necessary for:
- implementing the in-order ACK logic
- takeover at Replica->Active vbstate transition

Change-Id: I99f892b1328ef2fb6c5ad0d4f1f296c2a6045bf0
Reviewed-on: http://review.couchbase.org/107900
Tested-by: Build Bot <[email protected]>
Reviewed-by: Dave Rigby <[email protected]>

Author: paolococchi

engines/ep/src/durability/active_durability_monitor.cc

@@ -113,7 +113,7 @@ void ActiveDurabilityMonitor::addSyncWrite(const void* cookie,
                 "ActiveDurabilityMonitor::addSyncWrite: Impossible");
     }
 
-    state.wlock()->addSyncWrite(cookie, item);
+    state.wlock()->addSyncWrite(cookie, std::move(item));
 
     // @todo: Missing step - check for satisfied SyncWrite, we may need to
     //     commit immediately in the no-replica scenario. Consider to do that in
@@ -486,9 +486,10 @@ void ActiveDurabilityMonitor::State::setReplicationTopology(
 }
 
 void ActiveDurabilityMonitor::State::addSyncWrite(const void* cookie,
-                                                  const queued_item& item) {
-    trackedWrites.emplace_back(cookie, item, *firstChain);
-    lastTrackedSeqno = item->getBySeqno();
+                                                  queued_item item) {
+    const auto seqno = item->getBySeqno();
+    trackedWrites.emplace_back(cookie, std::move(item), firstChain.get());
+    lastTrackedSeqno = seqno;
 }
 
 void ActiveDurabilityMonitor::State::removeExpired(

engines/ep/src/durability/active_durability_monitor.h

@@ -191,7 +191,7 @@ class ActiveDurabilityMonitor : public DurabilityMonitor {
     struct State {
         void setReplicationTopology(const nlohmann::json& topology);
 
-        void addSyncWrite(const void* cookie, const queued_item& item);
+        void addSyncWrite(const void* cookie, queued_item item);
 
         /**
          * Returns the next position for a node iterator.

engines/ep/src/durability/durability_monitor_impl.cc

@@ -20,23 +20,27 @@
 
 DurabilityMonitor::SyncWrite::SyncWrite(const void* cookie,
                                         queued_item item,
-                                        const ReplicationChain& chain)
+                                        const ReplicationChain* chain)
     : cookie(cookie),
       item(item),
-      majority(chain.majority),
       expiryTime(
               item->getDurabilityReqs().getTimeout()
                       ? std::chrono::steady_clock::now() +
                                 std::chrono::milliseconds(
                                         item->getDurabilityReqs().getTimeout())
                       : boost::optional<
-                                std::chrono::steady_clock::time_point>{}),
-      active(chain.active) {
-    // We are making a SyncWrite for tracking, we must have already ensured
-    // that the Durability Requirements can be met at this point.
-    Expects(chain.size() >= majority);
-    for (const auto& entry : chain.positions) {
-        acks[entry.first] = false;
+                                std::chrono::steady_clock::time_point>{}) {
+    if (chain) {
+        majority = chain->majority;
+        active = chain->active;
+
+        // We are making a SyncWrite for tracking, we must have already ensured
+        // that the Durability Requirements can be met at this point.
+        Expects(chain->size() >= majority);
+
+        for (const auto& entry : chain->positions) {
+            acks[entry.first] = false;
+        }
     }
 }
 

engines/ep/src/durability/durability_monitor_impl.h

@@ -34,13 +34,21 @@
 static const std::string UndefinedNode{};
 
 /**
- * Represents a tracked SyncWrite.
+ * Represents a tracked SyncWrite. It is mainly a wrapper around a pending
+ * Prepare item.
  */
 class DurabilityMonitor::SyncWrite {
 public:
+    /**
+     * @param (optional) cookie The cookie representing the client connection.
+     *     Necessary at Active for notifying the client at SyncWrite completion.
+     * @param item The pending Prepare being wrapped
+     * @param (optional) chain The repl-chain that the write is tracked against.
+     *     Necessary at Active for verifying the SW Durability Requirements.
+     */
     SyncWrite(const void* cookie,
               queued_item item,
-              const ReplicationChain& chain);
+              const ReplicationChain* chain);
 
     const StoredDocKey& getKey() const;
 
@@ -89,19 +97,19 @@ class DurabilityMonitor::SyncWrite {
 
     // This optimization eliminates the need of scanning the ACK map for
     // verifying Durability Requirements
-    Monotonic<uint8_t> ackCount;
+    Monotonic<uint8_t> ackCount{0};
 
     // Majority in the arithmetic definition: num-nodes / 2 + 1
-    const uint8_t majority;
+    uint8_t majority{0};
+
+    // Name of the active node in replication-chain. Used at Durability
+    // Requirements verification.
+    std::string active{UndefinedNode};
 
     // Used for enforcing the Durability Requirements Timeout. It is set
     // when this SyncWrite is added for tracking into the DurabilityMonitor.
     const boost::optional<std::chrono::steady_clock::time_point> expiryTime;
 
-    // Name of the active node in replication-chain. Used at Durability
-    // Requirements verification.
-    const std::string active;
-
     friend std::ostream& operator<<(std::ostream&, const SyncWrite&);
 };
 

engines/ep/src/durability/passive_durability_monitor.cc

@@ -16,6 +16,7 @@
  */
 
 #include "passive_durability_monitor.h"
+#include "durability_monitor_impl.h"
 
 #include "bucket_logger.h"
 #include "item.h"
@@ -56,9 +57,25 @@ int64_t PassiveDurabilityMonitor::getHighPreparedSeqno() const {
     return 0;
 }
 
+void PassiveDurabilityMonitor::addSyncWrite(queued_item item) {
+    auto durReq = item->getDurabilityReqs();
+
+    if (durReq.getLevel() == cb::durability::Level::None) {
+        throw std::invalid_argument(
+                "PassiveDurabilityMonitor::addSyncWrite: Level::None");
+    }
+
+    state.wlock()->trackedWrites.emplace_back(
+            nullptr /*cookie*/, std::move(item), nullptr /*firstChain*/);
+
+    /*
+     * @todo: Missing step - we may be able to increase the high_prepared_seqno
+     * and ack back back to the Active.
+     */
+}
+
 size_t PassiveDurabilityMonitor::getNumTracked() const {
-    // @todo-durability: return a correct value for this.
-    return 0;
+    return state.rlock()->trackedWrites.size();
 }
 
 void PassiveDurabilityMonitor::toOStream(std::ostream& os) const {

engines/ep/src/durability/passive_durability_monitor.h

@@ -17,6 +17,9 @@
 #pragma once
 
 #include "durability_monitor.h"
+#include "item.h"
+
+#include <folly/Synchronized.h>
 
 class VBucket;
 
@@ -38,11 +41,34 @@ class PassiveDurabilityMonitor : public DurabilityMonitor {
 
     int64_t getHighPreparedSeqno() const override;
 
-protected:
+    /**
+     * Add a pending Prepare for tracking into the PDM.
+     *
+     * @param item the queued_item
+     */
+    void addSyncWrite(queued_item item);
+
     size_t getNumTracked() const override;
 
+protected:
     void toOStream(std::ostream& os) const override;
 
+    /*
+     * This class embeds the state of a PDM. It has been designed for being
+     * wrapped by a folly::Synchronized<T>, which manages the read/write
+     * concurrent access to the T instance.
+     * Note: all members are public as accessed directly only by PDM, this is
+     * a protected struct. Avoiding direct access by PDM would require
+     * re-implementing most of the PDM functions into PDM::State and exposing
+     * them on the PDM::State public interface.
+     */
+    struct State {
+        /// The container of pending Prepares.
+        Container trackedWrites;
+    };
+
     // The VBucket owning this DurabilityMonitor instance
     VBucket& vb;
+
+    folly::Synchronized<State> state;
 };

engines/ep/src/kv_bucket.cc

@@ -897,10 +897,9 @@ ENGINE_ERROR_CODE KVBucket::setVBucketState_UNLOCKED(
         //     For now necessary at least for tests.
         // Durability: Re-set vb-state for applying the ReplicationChain
         //     encoded in 'meta'. This is for supporting the case where
-        //     ns_server issues a single set-vb-state call for creating an
-        //     Active VBucket.
+        //     ns_server issues a single set-vb-state call for creating a VB.
         // Note: Must be done /after/ the new VBucket has been added to vbMap.
-        if (to == vbucket_state_active) {
+        if (to == vbucket_state_active || to == vbucket_state_replica) {
             vbMap.setState(newvb, to, meta, vbStateLock);
         }
 

engines/ep/src/vbucket.h

@@ -2185,10 +2185,12 @@ class VBucket : public std::enable_shared_from_this<VBucket> {
 
     static double mutationMemThreshold;
 
-    friend class VBucketTestBase;
-    friend class VBucketDurabilityTest;
+    friend class ActiveDurabilityMonitorTest;
     friend class DurabilityMonitorTest;
+    friend class PassiveDurabilityMonitorTest;
     friend class SingleThreadedActiveStreamTest;
+    friend class VBucketTestBase;
+    friend class VBucketDurabilityTest;
 
     DISALLOW_COPY_AND_ASSIGN(VBucket);
 };