[release/2.x] Cherry pick: Add mutex locks to sample indexing strategies (#4075)

.threading_canary

@@ -1 +1 @@
-This looks like a 'job' for Threading Canary!
+This looks like a "job" for Threading Canary!

include/ccf/indexing/strategies/seqnos_by_key_in_memory.h

@@ -2,6 +2,7 @@
 // Licensed under the Apache 2.0 License.
 #pragma once
 
+#include "ccf/ds/mutex.h"
 #include "ccf/indexing/strategies/visit_each_entry_in_map.h"
 #include "ccf/seq_no_collection.h"
 
@@ -15,6 +16,9 @@ namespace ccf::indexing::strategies
     // Value is every SeqNo which talks about that key.
     std::unordered_map<ccf::ByteVector, SeqNoCollection> seqnos_by_key;
 
+    // Mutex guarding access to seqnos_by_key
+    ccf::Mutex lock;
+
     void visit_entry(
       const ccf::TxID& tx_id,
       const ccf::ByteVector& k,

src/consensus/aft/test/logging_stub.h

@@ -17,6 +17,8 @@ namespace aft
   protected:
     ccf::NodeId _id;
 
+    std::mutex ledger_access;
+
   public:
     std::vector<std::vector<uint8_t>> ledger;
     uint64_t skip_count = 0;
@@ -31,6 +33,8 @@ namespace aft
       kv::Term term,
       kv::Version index)
     {
+      std::lock_guard<std::mutex> lock(ledger_access);
+
       // The payload that we eventually deserialise must include the
       // ledger entry as well as the View and Index that identify it. In
       // the real entries, they are nested in the payload and the IV. For
@@ -73,6 +77,8 @@ namespace aft
 
     std::optional<std::vector<uint8_t>> get_entry_by_idx(size_t idx)
     {
+      std::lock_guard<std::mutex> lock(ledger_access);
+
       // Ledger indices are 1-based, hence the -1
       if (idx > 0 && idx <= ledger.size())
       {

src/indexing/enclave_lfs_access.h

@@ -6,6 +6,7 @@
 #include "ccf/crypto/sha256.h"
 #include "ccf/crypto/symmetric_key.h"
 #include "ccf/ds/hex.h"
+#include "ccf/ds/mutex.h"
 #include "ds/messaging.h"
 #include "indexing/lfs_interface.h"
 #include "indexing/lfs_ringbuffer_types.h"
@@ -86,6 +87,7 @@ namespace ccf::indexing
     using PendingResult = std::weak_ptr<FetchResult>;
 
     std::unordered_map<LFSKey, PendingResult> pending;
+    ccf::Mutex pending_access;
 
     ringbuffer::WriterPtr to_host;
 
@@ -136,6 +138,7 @@ namespace ccf::indexing
         dispatcher, LFSMsg::response, [this](const uint8_t* data, size_t size) {
           auto [obfuscated, encrypted] =
             ringbuffer::read_message<LFSMsg::response>(data, size);
+          std::lock_guard<ccf::Mutex> guard(pending_access);
           auto it = pending.find(obfuscated);
           if (it != pending.end())
           {
@@ -193,6 +196,7 @@ namespace ccf::indexing
         [this](const uint8_t* data, size_t size) {
           auto [obfuscated] =
             ringbuffer::read_message<LFSMsg::not_found>(data, size);
+          std::lock_guard<ccf::Mutex> guard(pending_access);
           auto it = pending.find(obfuscated);
           if (it != pending.end())
           {
@@ -258,6 +262,7 @@ namespace ccf::indexing
     FetchResultPtr fetch(const LFSKey& key) override
     {
       const auto obfuscated = obfuscate_key(key);
+      std::lock_guard<ccf::Mutex> guard(pending_access);
       auto it = pending.find(obfuscated);
 
       FetchResultPtr result;

src/indexing/strategies/seqnos_by_key_bucketed.cpp

@@ -5,6 +5,7 @@
 
 #include "ccf/ds/hex.h"
 #include "ccf/ds/logger.h"
+#include "ccf/ds/mutex.h"
 #include "ds/lru.h"
 #include "ds/serialized.h"
 #include "indexing/lfs_interface.h"
@@ -32,6 +33,8 @@ namespace ccf::indexing::strategies
     using BucketValue = std::pair<FetchResultPtr, SeqNoCollection>;
     LRU<BucketKey, BucketValue> old_results;
 
+    ccf::Mutex results_access;
+
     std::string name;
 
     std::shared_ptr<AbstractLFSAccess> lfs_access;
@@ -206,6 +209,8 @@ namespace ccf::indexing::strategies
 
       while (true)
       {
+        std::lock_guard<ccf::Mutex> guard(results_access);
+
         const auto bucket_key = std::make_pair(serialised_key, from_range);
         const auto old_it = old_results.find(bucket_key);
         if (old_it != old_results.end())
@@ -350,6 +355,8 @@ namespace ccf::indexing::strategies
   {
     const auto range = impl->get_range_for(tx_id.seqno);
 
+    std::lock_guard<ccf::Mutex> guard(impl->results_access);
+
     auto it = impl->current_results.find(k);
     if (it != impl->current_results.end())
     {

src/indexing/strategies/seqnos_by_key_in_memory.cpp

@@ -8,6 +8,7 @@ namespace ccf::indexing::strategies
   void SeqnosByKey_InMemory_Untyped::visit_entry(
     const ccf::TxID& tx_id, const ccf::ByteVector& k, const ccf::ByteVector& v)
   {
+    std::lock_guard<ccf::Mutex> guard(lock);
     seqnos_by_key[k].insert(tx_id.seqno);
   }
 
@@ -18,6 +19,7 @@ namespace ccf::indexing::strategies
       ccf::SeqNo to,
       std::optional<size_t> max_seqnos)
   {
+    std::lock_guard<ccf::Mutex> guard(lock);
     const auto it = seqnos_by_key.find(serialised_key);
     if (it != seqnos_by_key.end())
     {