diff --git a/CHANGELOG.md b/CHANGELOG.md index dfd7ef5bcda..d4fbab30970 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -15,6 +15,7 @@ and this project adheres to [Semantic Versioning](http://semver.org/spec/v2.0.0. ### Fixed +- A transaction's `force_ledger_chunk` and `snapshot_at_next_signature` flags are no longer applied once a concurrent view change has discarded the transaction's writes, which previously left a chunk boundary, or an armed snapshot, for a transaction no longer present in the ledger. The forced chunk is also attached to the transaction's own version rather than whichever version the store had reached (#8245). - A rollback whose target is at or beyond the store's own version no longer moves ledger chunk metadata forward past it, which previously left a permanent offset skewing later chunk boundaries (#8244). - Ledger chunk metadata and snapshot scheduling are no longer restored by a transaction whose writes a concurrent view change has already discarded. Both are now updated under the same lock as the rollback, and skipped when the transaction's rollback epoch or view no longer holds (#8243). - A transaction whose view changed while it was committing could apply its writes to the local key-value store and then fail to replicate, leaving state that never reached consensus. The transaction's view is now validated atomically with the allocation of its version, so it is rejected before any map is modified, and `ccf::kv::CommitResult::FAIL_NO_REPLICATE` no longer implies a locally applied write (#8242). diff --git a/src/kv/committable_tx.h b/src/kv/committable_tx.h index 2381021d9c5..444fa796d8d 100644 --- a/src/kv/committable_tx.h +++ b/src/kv/committable_tx.h @@ -241,6 +241,7 @@ namespace ccf::kv bool track_deletes_on_missing_keys = false; bool commit_term_changed = false; std::optional c; + std::optional expected_rollback_count; { MapSetLockGuard map_set_guard(*pimpl->store, maps_created); c = apply_changes( @@ -254,9 +255,12 @@ namespace ccf::kv return std::optional{}; } - const auto& resolved = resolution.value(); + const auto + [resolved_version, previous_last_new_map, rollback_count] = + resolution.value(); + expected_rollback_count = rollback_count; return std::optional( - std::in_place, std::get<0>(resolved), std::get<1>(resolved)); + std::in_place, resolved_version, previous_last_new_map); }, hooks, pimpl->created_maps, @@ -284,26 +288,44 @@ namespace ccf::kv committed = true; version = c.value(); - if (tx_flag_enabled(TxFlag::LEDGER_CHUNK_AT_NEXT_SIGNATURE)) + const auto force_ledger_chunk = + tx_flag_enabled(TxFlag::LEDGER_CHUNK_AT_NEXT_SIGNATURE); + const auto snapshot_at_next_signature = + tx_flag_enabled(TxFlag::SNAPSHOT_AT_NEXT_SIGNATURE); + + if (version == NoVersion) { - auto chunker = pimpl->store->get_chunker(); - if (chunker) + // Read-only transaction. It has no version to attach a ledger chunk + // to, but a requested snapshot must still be armed, as it was before + // these flags became rollback-sensitive. + if (snapshot_at_next_signature) { - chunker->force_end_of_chunk(version); + pimpl->store->set_flag( + AbstractStore::StoreFlag::SNAPSHOT_AT_NEXT_SIGNATURE); + unset_tx_flag(TxFlag::SNAPSHOT_AT_NEXT_SIGNATURE); } + return CommitResult::SUCCESS; } - if (tx_flag_enabled(TxFlag::SNAPSHOT_AT_NEXT_SIGNATURE)) + // These side effects outlive this transaction, so they must not be + // applied if a concurrent rollback has already discarded its writes. + if (force_ledger_chunk || snapshot_at_next_signature) { - pimpl->store->set_flag( - AbstractStore::StoreFlag::SNAPSHOT_AT_NEXT_SIGNATURE); - unset_tx_flag(TxFlag::SNAPSHOT_AT_NEXT_SIGNATURE); - } + if (!expected_rollback_count.has_value()) + { + throw std::logic_error( + "Transaction was allocated a version without a rollback count"); + } - if (version == NoVersion) - { - // Read-only transaction - return CommitResult::SUCCESS; + if (!pimpl->store->apply_tx_flags( + version, + pimpl->commit_view, + expected_rollback_count.value(), + force_ledger_chunk, + snapshot_at_next_signature)) + { + return CommitResult::FAIL_NO_REPLICATE; + } } // From here, we have received a unique commit version and made diff --git a/src/kv/kv_types.h b/src/kv/kv_types.h index 552c7a643d9..93e3e4702c3 100644 --- a/src/kv/kv_types.h +++ b/src/kv/kv_types.h @@ -744,6 +744,12 @@ namespace ccf::kv std::unique_ptr pending_tx, bool globally_committable) = 0; virtual bool check_rollback_count(Version count) = 0; + virtual bool apply_tx_flags( + Version version, + Term expected_term, + Version expected_rollback_count, + bool force_ledger_chunk, + bool snapshot_at_next_signature) = 0; virtual std::unique_ptr snapshot_unsafe_maps( Version v) = 0; diff --git a/src/kv/store.h b/src/kv/store.h index 717c6dfaa94..8d43f21a056 100644 --- a/src/kv/store.h +++ b/src/kv/store.h @@ -1417,6 +1417,34 @@ namespace ccf::kv return {this, term_of_last_version, tx_id, rollback_count}; } + bool apply_tx_flags( + Version tx_version, + Term expected_term, + Version expected_rollback_count, + bool force_ledger_chunk, + bool snapshot_at_next_signature) override + { + std::lock_guard vguard(version_lock); + if ( + term_of_next_version != expected_term || + rollback_count != expected_rollback_count) + { + return false; + } + + if (force_ledger_chunk && chunker) + { + chunker->force_end_of_chunk(tx_version); + } + + if (snapshot_at_next_signature) + { + set_flag_unsafe(StoreFlag::SNAPSHOT_AT_NEXT_SIGNATURE); + } + + return true; + } + void set_flag(StoreFlag f) override { std::lock_guard vguard(version_lock); diff --git a/src/kv/test/kv_test.cpp b/src/kv/test/kv_test.cpp index e3c80499207..0c08b85cc41 100644 --- a/src/kv/test/kv_test.cpp +++ b/src/kv/test/kv_test.cpp @@ -3641,6 +3641,80 @@ TEST_CASE("A rollback never moves chunk metadata past the store's version") CHECK(chunker->current_version() == store.current_version()); } +TEST_CASE("Rollback-sensitive transaction flags are not restored") +{ + ccf::kv::Store store; + store.set_encryptor(std::make_shared()); + auto consensus = std::make_shared(); + store.set_consensus(consensus); + auto chunker = std::make_shared(); + store.set_chunker(chunker); + + constexpr ccf::kv::Term initial_term = 2; + store.initialise_term(initial_term); + MapTypes::StringString map("public:map"); + + for (const auto* value : {"first", "second"}) + { + auto tx = store.create_tx(); + tx.rw(map)->put("key", value); + REQUIRE(tx.commit() == ccf::kv::CommitResult::SUCCESS); + } + + const auto discarded = store.current_txid(); + REQUIRE(store.check_rollback_count(0)); + + INFO("Flags from a transaction a rollback discarded are dropped"); + { + // A view change truncates the transaction's write away, then the + // transaction reaches the point where it would apply its flags. + store.rollback({initial_term, discarded.seqno - 1}, initial_term + 1); + REQUIRE(store.check_rollback_count(1)); + + CHECK_FALSE(store.apply_tx_flags( + discarded.seqno, + discarded.view, + 0, + /* force_ledger_chunk */ true, + /* snapshot_at_next_signature */ true)); + + CHECK_FALSE(store.flag_enabled( + ccf::kv::AbstractStore::StoreFlag::SNAPSHOT_AT_NEXT_SIGNATURE)); + CHECK_FALSE(chunker->is_chunk_end_requested(discarded.seqno)); + } + + INFO("Flags from a transaction still in its own epoch are applied"); + { + auto tx = store.create_tx(); + tx.rw(map)->put("key", "replacement"); + REQUIRE(tx.commit() == ccf::kv::CommitResult::SUCCESS); + const auto replacement = store.current_txid(); + + // Commit a later transaction, so that the store's version has moved on by + // the time the earlier transaction applies its flags. + { + auto later = store.create_tx(); + later.rw(map)->put("other", "later"); + REQUIRE(later.commit() == ccf::kv::CommitResult::SUCCESS); + } + REQUIRE(store.current_txid().seqno == replacement.seqno + 1); + + CHECK(store.apply_tx_flags( + replacement.seqno, + replacement.view, + 1, + /* force_ledger_chunk */ true, + /* snapshot_at_next_signature */ true)); + + CHECK(store.flag_enabled( + ccf::kv::AbstractStore::StoreFlag::SNAPSHOT_AT_NEXT_SIGNATURE)); + + INFO("The chunk is requested at the transaction's own version"); + CHECK(chunker->is_chunk_end_requested(replacement.seqno)); + CHECK_FALSE(chunker->is_chunk_end_requested(replacement.seqno - 1)); + } +} + TEST_CASE("Ledger entry chunk request") { ccf::kv::Store store;