MB-55199: Sort magma pendingRequests into key and ascending seqno order

Before handing over the batch of writes to magma, order by key and
seqno, so that when duplicates occur they are in seqno ascending order.

Change-Id: Ida6bdb0c6caa52a05d10d6167e27d2768c97222a
Reviewed-on: https://review.couchbase.org/c/kv_engine/+/185329
Reviewed-by: Paolo Cocchi <[email protected]>
Tested-by: Jim Walker <[email protected]>

Author: jimwwalker

engines/ep/src/kvstore/magma-kvstore/magma-kvstore.cc

@@ -293,22 +293,28 @@ bool MagmaKVStore::MagmaCompactionCB::canPurge(CollectionID collection) {
     return onlyThisCollection.value() == collection;
 }
 
-struct MagmaKVStoreTransactionContext : public TransactionContext {
-    MagmaKVStoreTransactionContext(KVStore& kvstore,
-                                   Vbid vbid,
-                                   std::unique_ptr<PersistenceCallback> cb)
-        : TransactionContext(kvstore, vbid, std::move(cb)) {
-    }
-    /**
-     * Container for pending Magma requests.
-     *
-     * Using deque as as the expansion behaviour is less aggressive compared to
-     * std::vector (MagmaRequest objects are ~176 bytes in size).
-     */
-    using PendingRequestQueue = std::deque<MagmaRequest>;
-
-    PendingRequestQueue pendingReqs;
-};
+MagmaKVStoreTransactionContext::MagmaKVStoreTransactionContext(
+        KVStore& kvstore, Vbid vbid, std::unique_ptr<PersistenceCallback> cb)
+    : TransactionContext(kvstore, vbid, std::move(cb)) {
+}
+
+void MagmaKVStoreTransactionContext::preparePendingRequests() {
+    // MB-55199: Magma requires key/seqno order, but ascending seqno.
+    // KV-engine flusher writes out the batch in key/seqno, but descending seqno
+    // order.
+    std::sort(pendingReqs.begin(),
+              pendingReqs.end(),
+              [](const auto& lhs, const auto& rhs) {
+                  const auto comp = lhs->getItem().getKey().compare(
+                          rhs->getItem().getKey());
+                  // When keys are equal, sort by seqno.
+                  if (comp == 0) {
+                      return lhs->getItem().getBySeqno() <
+                             rhs->getItem().getBySeqno();
+                  }
+                  return comp < 0;
+              });
+}
 
 std::pair<Status, bool> MagmaKVStore::compactionCallBack(
         MagmaKVStore::MagmaCompactionCB& cbCtx,
@@ -778,7 +784,8 @@ void MagmaKVStore::commitCallback(MagmaKVStoreTransactionContext& txnCtx,
                                   int errCode,
                                   kvstats_ctx&) {
     const auto flushSuccess = (errCode == Status::Code::Ok);
-    for (const auto& req : txnCtx.pendingReqs) {
+    for (const auto& reqPtr : txnCtx.pendingReqs) {
+        const auto& req = *reqPtr;
         size_t mutationSize = req.getRawKeyLen() + req.getBodySize() +
                               req.getDocMeta().size();
         st.io_num_write++;
@@ -900,7 +907,8 @@ void MagmaKVStore::set(TransactionContext& txnCtx, queued_item item) {
     }
 
     auto& ctx = static_cast<MagmaKVStoreTransactionContext&>(txnCtx);
-    ctx.pendingReqs.emplace_back(std::move(item));
+    ctx.pendingReqs.emplace_back(
+            std::make_unique<MagmaRequest>(std::move(item)));
 }
 
 GetValue MagmaKVStore::get(const DiskDocKey& key,
@@ -1111,7 +1119,8 @@ void MagmaKVStore::del(TransactionContext& txnCtx, queued_item item) {
     }
 
     auto& ctx = dynamic_cast<MagmaKVStoreTransactionContext&>(txnCtx);
-    ctx.pendingReqs.emplace_back(std::move(item));
+    ctx.pendingReqs.emplace_back(
+            std::make_unique<MagmaRequest>(std::move(item)));
 }
 
 void MagmaKVStore::delVBucket(Vbid vbid,
@@ -1487,6 +1496,7 @@ int MagmaKVStore::saveDocs(MagmaKVStoreTransactionContext& txnCtx,
     };
 
     auto& ctx = dynamic_cast<MagmaKVStoreTransactionContext&>(txnCtx);
+    ctx.preparePendingRequests();
 
     // Vector of updates to be written to the data store.
     WriteOps writeOps;
@@ -1495,7 +1505,8 @@ int MagmaKVStore::saveDocs(MagmaKVStoreTransactionContext& txnCtx,
     // TODO: Replace writeOps with Magma::WriteOperations when it
     // becomes available. This will allow us to pass pendingReqs
     // in and create the WriteOperation from the pendingReqs queue.
-    for (auto& req : ctx.pendingReqs) {
+    for (auto& reqPtr : ctx.pendingReqs) {
+        auto& req = *reqPtr;
         Slice valSlice{req.getBodyData(), req.getBodySize()};
 
         auto docMeta = req.getDocMeta();

engines/ep/src/kvstore/magma-kvstore/magma-kvstore.h

@@ -36,6 +36,7 @@ class Status;
 
 class MagmaKVStoreConfig;
 class MagmaMemoryTrackingProxy;
+class MagmaRequest;
 struct kvstats_ctx;
 struct MagmaKVStoreTransactionContext;
 struct MagmaScanResult;
@@ -829,3 +830,24 @@ class MagmaKVStore : public KVStore {
 private:
     EventuallyPersistentEngine* currEngine;
 };
+
+struct MagmaKVStoreTransactionContext : public TransactionContext {
+    MagmaKVStoreTransactionContext(KVStore& kvstore,
+                                   Vbid vbid,
+                                   std::unique_ptr<PersistenceCallback> cb);
+
+    /**
+     * Prepare the requests before handing them over to Magma. This currently
+     * orders by seqno (ascending) see MB-55199
+     */
+    void preparePendingRequests();
+
+    /**
+     * Container for pending Magma requests.
+     *
+     * Storing the pointer as preparePendingRequests will sort this container.
+     */
+    using PendingRequestQueue = std::deque<std::unique_ptr<MagmaRequest>>;
+
+    PendingRequestQueue pendingReqs;
+};

engines/ep/tests/module_tests/magma-kvstore_test.cc

@@ -13,6 +13,7 @@
 #include "configuration.h"
 #include "kvstore/magma-kvstore/kv_magma_common/magma-kvstore_metadata.h"
 #include "kvstore/magma-kvstore/magma-kvstore_config.h"
+#include "kvstore/magma-kvstore/magma-kvstore_iorequest.h"
 #include "kvstore/storage_common/storage_common/local_doc_constants.h"
 #include "kvstore_test.h"
 #include "programs/engine_testapp/mock_server.h"
@@ -742,4 +743,31 @@ TEST_F(MagmaKVStoreTest, scanAllVersions) {
     auto& cb =
             static_cast<CustomCallback<GetValue>&>(bySeq->getValueCallback());
     EXPECT_EQ(2, cb.getProcessedCount());
+}
+
+// Validate the ordering by seqno
+TEST_F(MagmaKVStoreTest, preparePendingRequests) {
+    MagmaKVStoreTransactionContext ctx(*kvstore, vbid, nullptr);
+
+    // seqnos 5, 4, 3, 2, 1
+    std::array<std::string, 5> keys = {{"bbb", "c", "bbb", "aaa", "bbb"}};
+    uint64_t seqno = 5;
+    for (const auto& k : keys) {
+        auto qi = makeCommittedItem(makeStoredDocKey(k), "value");
+        qi->setBySeqno(seqno--);
+        ctx.pendingReqs.push_back(
+                std::make_unique<MagmaRequest>(std::move(qi)));
+    }
+
+    ctx.preparePendingRequests();
+
+    std::array<std::pair<uint64_t, std::string>, 5> expected = {
+            {{2, "aaa"}, {1, "bbb"}, {3, "bbb"}, {5, "bbb"}, {4, "c"}}};
+    ASSERT_EQ(expected.size(), ctx.pendingReqs.size());
+    auto itr = expected.begin();
+    for (const auto& req : ctx.pendingReqs) {
+        EXPECT_EQ(itr->first, req->getItem().getBySeqno()) << req->getItem();
+        EXPECT_EQ(itr->second, req->getItem().getKey().c_str());
+        ++itr;
+    }
 }