diff --git a/src/kv/committable_tx.h b/src/kv/committable_tx.h index e47877e983b..e60666916ab 100644 --- a/src/kv/committable_tx.h +++ b/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 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 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 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 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) diff --git a/src/kv/test/kv_test.cpp b/src/kv/test/kv_test.cpp index 0bec4c7f466..609318d8cbb 100644 --- a/src/kv/test/kv_test.cpp +++ b/src/kv/test/kv_test.cpp @@ -20,7 +20,9 @@ #undef FAIL #include #include +#include #include +#include #include 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()); + + { + auto tx = store.create_tx(); + tx.rw("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 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(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;