Hold the map set while committing a reserved transaction

src/kv/committable_tx.h

@@ -28,6 +28,35 @@ namespace ccf::kv
     };
 
   protected:
+    class MapSetLockGuard
+    {
+    private:
+      AbstractStore& store;
+      const bool locked;
+
+    public:
+      MapSetLockGuard(AbstractStore& store_, bool should_lock) :
+        store(store_),
+        locked(should_lock)
+      {
+        if (locked)
+        {
+          store.lock_map_set();
+        }
+      }
+
+      ~MapSetLockGuard()
+      {
+        if (locked)
+        {
+          store.unlock_map_set();
+        }
+      }
+
+      MapSetLockGuard(const MapSetLockGuard&) = delete;
+      MapSetLockGuard& operator=(const MapSetLockGuard&) = delete;
+    };
+
     bool committed = false;
     bool success = false;
 
@@ -206,31 +235,26 @@ namespace ccf::kv
       // If this transaction creates any maps, ensure that commit gets a
       // consistent snapshot of the existing map set
       const bool maps_created = !pimpl->created_maps.empty();
-      if (maps_created)
-      {
-        this->pimpl->store->lock_map_set();
-      }
 
       ccf::kv::ConsensusHookPtrs hooks;
 
       std::optional<Version> new_maps_conflict_version = std::nullopt;
 
       bool track_deletes_on_missing_keys = false;
-      auto c = apply_changes(
-        all_changes,
-        version_resolver == nullptr ?
-          [&](bool has_new_map) {
-            return pimpl->store->next_version(has_new_map);
-          } :
-          version_resolver,
-        hooks,
-        pimpl->created_maps,
-        new_maps_conflict_version,
-        track_deletes_on_missing_keys);
-
-      if (maps_created)
-      {
-        this->pimpl->store->unlock_map_set();
+      std::optional<Version> c;
+      {
+        MapSetLockGuard map_set_guard(*pimpl->store, maps_created);
+        c = apply_changes(
+          all_changes,
+          version_resolver == nullptr ?
+            [&](bool has_new_map) {
+              return pimpl->store->next_version(has_new_map);
+            } :
+            version_resolver,
+          hooks,
+          pimpl->created_maps,
+          new_maps_conflict_version,
+          track_deletes_on_missing_keys);
       }
 
       success = c.has_value();
@@ -458,14 +482,26 @@ namespace ccf::kv
 
       std::vector<ConsensusHookPtr> hooks;
       bool track_deletes_on_missing_keys = false;
-      auto c = apply_changes(
-        all_changes,
-        [this](bool) { return std::make_tuple(version, version - 1); },
-        hooks,
-        pimpl->created_maps,
-        version,
-        track_deletes_on_missing_keys,
-        rollback_count);
+
+      // A reserved transaction can create maps too - the first signature
+      // creates the signature tables. As in commit(), hold the map set while
+      // applying, so that add_dynamic_map() does not mutate it underneath a
+      // concurrent reader.
+      const bool maps_created = !pimpl->created_maps.empty();
+
+      std::optional<Version> c;
+      {
+        MapSetLockGuard map_set_guard(*pimpl->store, maps_created);
+        c = apply_changes(
+          all_changes,
+          [this](bool) { return std::make_tuple(version, version - 1); },
+          hooks,
+          pimpl->created_maps,
+          version,
+          track_deletes_on_missing_keys,
+          rollback_count);
+      }
+
       success = c.has_value();
 
       if (!success)

src/kv/test/kv_test.cpp

@@ -20,7 +20,9 @@
 #undef FAIL
 #include <random>
 #include <set>
+#include <stop_token>
 #include <string>
+#include <thread>
 #include <vector>
 
 struct MapTypes
@@ -3358,6 +3360,74 @@ TEST_CASE("Range")
   }
 }
 
+// Reproduces the race between a reserved transaction creating a map (which
+// writes to the Store's map set) and a concurrent reader looking one up. The
+// write happens in Store::add_dynamic_map via commit_reserved; the read holds
+// maps_lock via Store::get_map. Only ThreadSanitizer can observe the failure,
+// so this asserts nothing about interleaving - it exists to give TSAN both
+// accesses concurrently.
+TEST_CASE("Reserved transaction map creation is serialised with lookups")
+{
+  ccf::kv::Store store;
+  store.set_encryptor(std::make_shared<ccf::kv::NullTxEncryptor>());
+
+  {
+    auto tx = store.create_tx();
+    tx.rw<MapTypes::StringString>("public:existing")->put("k", "v");
+    REQUIRE(tx.commit() == ccf::kv::CommitResult::SUCCESS);
+  }
+
+  // Ensure every exit path, including a failed REQUIRE, stops and joins
+  // readers.
+  std::vector<std::jthread> readers;
+  readers.reserve(4);
+  for (size_t r = 0; r < 4; ++r)
+  {
+    readers.emplace_back([&store](std::stop_token stop_token) {
+      size_t n = 0;
+      while (!stop_token.stop_requested())
+      {
+        // Takes maps_lock and searches the map set. Looks up names in the
+        // range being inserted, so the search path traverses the nodes
+        // add_dynamic_map is writing. Deliberately avoids current_version(),
+        // so this contends only on maps_lock.
+        (void)store.get_map(1, fmt::format("public:reserved_{}", n % 2000));
+        n++;
+      }
+    });
+  }
+
+  constexpr size_t reserved_txs = 2000;
+  for (size_t i = 0; i < reserved_txs; ++i)
+  {
+    // Each reserved transaction writes to a map that does not exist yet, so
+    // committing it adds to the Store's map set. Nothing else here may take
+    // maps_lock, or it would order the write against the readers and hide the
+    // race being reproduced.
+    auto tx = store.create_reserved_tx(store.next_txid());
+    tx.rw<MapTypes::StringString>(fmt::format("public:reserved_{}", i))
+      ->put("k", "v");
+    const auto [result, data, claims, commit_evidence, hooks] =
+      tx.commit_reserved();
+    REQUIRE(result == ccf::kv::CommitResult::SUCCESS);
+  }
+
+  for (auto& reader : readers)
+  {
+    reader.request_stop();
+  }
+  readers.clear();
+
+  // Confirm the writes really did extend the map set, so this exercises
+  // Store::add_dynamic_map rather than silently doing nothing.
+  REQUIRE(
+    store.get_map(store.current_version(), "public:reserved_0") != nullptr);
+  REQUIRE(
+    store.get_map(
+      store.current_version(),
+      fmt::format("public:reserved_{}", reserved_txs - 1)) != nullptr);
+}
+
 TEST_CASE("Ledger entry chunk request")
 {
   ccf::kv::Store store;