From c3984d5d052525b36466fde1dcba80da658d3ca0 Mon Sep 17 00:00:00 2001 From: achamayou Date: Sun, 30 Aug 2026 19:47:29 +0100 Subject: [PATCH 1/4] Reject stale-view writes before local commit A transaction whose view changed while it was committing could apply its writes to the local store and only then be refused replication, leaving state that never reaches consensus - contradicting the documented contract that a failed transaction is rolled back. Validate the view the transaction captured atomically with the allocation of its version, under the same lock a rollback takes, so it is refused before any map is modified. If allocation wins the race instead, the rollback observes the new version and truncates the writes. The unused caller-supplied version resolver is removed: it had no callers and would have bypassed this check. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> Copilot-Session: 75d99c5d-6efa-4048-8032-8c78b97208d9 --- CHANGELOG.md | 12 ++++++++ python/pyproject.toml | 2 +- src/kv/apply_changes.h | 53 ++++++++++++++++++++------------- src/kv/committable_tx.h | 37 ++++++++++++++++------- src/kv/kv_types.h | 3 +- src/kv/store.h | 18 +++++++++-- src/kv/test/kv_test.cpp | 66 +++++++++++++++++++++++++++++++++++++++++ src/node/rpc/frontend.h | 3 +- src/node/snapshotter.h | 2 +- 9 files changed, 158 insertions(+), 38 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 126fdff584c8..a0a08dd2dcd5 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -5,6 +5,18 @@ All notable changes to this project will be documented in this file. The format is based on [Keep a Changelog](http://keepachangelog.com/en/1.0.0/) and this project adheres to [Semantic Versioning](http://semver.org/spec/v2.0.0.html). +## [7.0.14] + +[7.0.14]: https://github.com/microsoft/CCF/releases/tag/ccf-7.0.14 + +### Fixed + +- 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). + +### Changed + +- `ccf::kv::CommittableTx::commit()` no longer takes a caller-supplied version resolver. The parameter had no callers, and bypassed the view check above. Callers passing `nullptr` for it should remove the argument (#8242). + ## [7.0.13] [7.0.13]: https://github.com/microsoft/CCF/releases/tag/ccf-7.0.13 diff --git a/python/pyproject.toml b/python/pyproject.toml index 18462482f034..7529d0383b9b 100644 --- a/python/pyproject.toml +++ b/python/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta" [project] name = "ccf" -version = "7.0.13" +version = "7.0.14" authors = [ { name="CCF Team", email="CCF-Sec@microsoft.com" }, ] diff --git a/src/kv/apply_changes.h b/src/kv/apply_changes.h index dc64c5dd3dc1..a82a8d12091f 100644 --- a/src/kv/apply_changes.h +++ b/src/kv/apply_changes.h @@ -21,8 +21,9 @@ namespace ccf::kv // version which can have a conflict with the transaction. using VersionLastNewMap = Version; - using VersionResolver = std::function( - bool tx_contains_new_map)>; + using VersionResolution = std::tuple; + using VersionResolver = + std::function(bool tx_contains_new_map)>; static inline std::optional apply_changes( OrderedChanges& changes, @@ -118,32 +119,42 @@ namespace ccf::kv { // Get the version number to be used for this commit. ccf::kv::Version version_last_new_map = 0; - std::tie(version, version_last_new_map) = - version_resolver_fn(!new_maps.empty()); - - // Transfer ownership of these new maps to their target stores, iff we - // have writes to them - for (const auto& [map_name, map_ptr] : new_maps) + const auto version_resolution = version_resolver_fn(!new_maps.empty()); + if (version_resolution.has_value()) { - const auto it = views.find(map_name); - if (it != views.end() && it->second->has_writes()) - { - map_ptr->get_store()->add_dynamic_map(version, map_ptr); - } + std::tie(version, version_last_new_map) = version_resolution.value(); } - - for (auto& [view_name, view_ptr] : views) + else { - view_ptr->commit(version, track_deletes_on_missing_keys); + ok = false; } - // Collect ConsensusHooks - for (auto& [view_name, view_ptr] : views) + if (ok) { - auto hook_ptr = view_ptr->post_commit(); - if (hook_ptr != nullptr) + // Transfer ownership of these new maps to their target stores, iff we + // have writes to them + for (const auto& [map_name, map_ptr] : new_maps) + { + const auto it = views.find(map_name); + if (it != views.end() && it->second->has_writes()) + { + map_ptr->get_store()->add_dynamic_map(version, map_ptr); + } + } + + for (auto& [view_name, view_ptr] : views) + { + view_ptr->commit(version, track_deletes_on_missing_keys); + } + + // Collect ConsensusHooks + for (auto& [view_name, view_ptr] : views) { - hooks.push_back(std::move(hook_ptr)); + auto hook_ptr = view_ptr->post_commit(); + if (hook_ptr != nullptr) + { + hooks.push_back(std::move(hook_ptr)); + } } } } diff --git a/src/kv/committable_tx.h b/src/kv/committable_tx.h index e47877e983bd..f0e1cb866b29 100644 --- a/src/kv/committable_tx.h +++ b/src/kv/committable_tx.h @@ -160,9 +160,9 @@ namespace ccf::kv * * A transaction can either succeed and replicate * (`ccf::kv::CommitResult::SUCCESS`), fail because of a conflict with other - * transactions (`ccf::kv::CommitResult::FAIL_CONFLICT`), or succeed - * locally, but fail to replicate - * (`ccf::kv::CommitResult::FAIL_NO_REPLICATE`). + * transactions (`ccf::kv::CommitResult::FAIL_CONFLICT`), or fail to + * replicate (`ccf::kv::CommitResult::FAIL_NO_REPLICATE`). A transaction + * whose commit term is stale is rejected before its writes are applied. * * Transactions that fail are rolled back, no matter the reason. * @@ -170,8 +170,6 @@ namespace ccf::kv */ CommitResult commit( const ccf::ClaimsDigest& claims = ccf::empty_claims(), - std::function(bool has_new_map)> - version_resolver = nullptr, WriteSetObserver write_set_observer = nullptr) { if (committed) @@ -216,13 +214,25 @@ namespace ccf::kv std::optional new_maps_conflict_version = std::nullopt; bool track_deletes_on_missing_keys = false; + bool commit_term_changed = false; + std::optional expected_rollback_count; auto c = apply_changes( all_changes, - version_resolver == nullptr ? - [&](bool has_new_map) { - return pimpl->store->next_version(has_new_map); - } : - version_resolver, + [&](bool has_new_map) { + auto resolution = + pimpl->store->next_version(has_new_map, pimpl->commit_view); + commit_term_changed = !resolution.has_value(); + if (!resolution.has_value()) + { + return std::optional{}; + } + + const auto [resolved_version, previous_last_new_map, rollback_count] = + resolution.value(); + expected_rollback_count = rollback_count; + return std::optional( + std::in_place, resolved_version, previous_last_new_map); + }, hooks, pimpl->created_maps, new_maps_conflict_version, @@ -239,6 +249,13 @@ namespace ccf::kv { // This Tx is now in a dead state. Caller should create a new Tx and try // again. + if (commit_term_changed) + { + LOG_TRACE_FMT( + "Could not commit transaction because its commit term changed"); + return CommitResult::FAIL_NO_REPLICATE; + } + LOG_TRACE_FMT("Could not commit transaction due to conflict"); return CommitResult::FAIL_CONFLICT; } diff --git a/src/kv/kv_types.h b/src/kv/kv_types.h index 96de7bc58572..552c7a643d9f 100644 --- a/src/kv/kv_types.h +++ b/src/kv/kv_types.h @@ -709,7 +709,8 @@ namespace ccf::kv virtual void unlock_map_set() = 0; virtual Version next_version() = 0; - virtual std::tuple next_version(bool commit_new_map) = 0; + virtual std::optional> next_version( + bool commit_new_map, Term expected_commit_term) = 0; virtual ccf::TxID next_txid() = 0; virtual Version current_version() = 0; diff --git a/src/kv/store.h b/src/kv/store.h index 2a4175280836..577040344ca4 100644 --- a/src/kv/store.h +++ b/src/kv/store.h @@ -1176,9 +1176,23 @@ namespace ccf::kv return rollback_count == count; } - std::tuple next_version(bool commit_new_map) override + std::optional> next_version( + bool commit_new_map, Term expected_commit_term) override { std::lock_guard vguard(version_lock); + // If rollback updates the term before this lock is acquired, reject the + // transaction before map writes are applied. If version allocation wins + // the race, rollback observes the new version and truncates those writes. + if (term_of_next_version != expected_commit_term) + { + LOG_DEBUG_FMT( + "Refusing to assign a version to a transaction from term {} because " + "the current term is {}", + expected_commit_term, + term_of_next_version); + return std::nullopt; + } + Version v = next_version_unsafe(); auto previous_last_new_map = last_new_map; @@ -1187,7 +1201,7 @@ namespace ccf::kv last_new_map = v; } - return std::make_tuple(v, previous_last_new_map); + return std::make_tuple(v, previous_last_new_map, rollback_count); } Version next_version() override diff --git a/src/kv/test/kv_test.cpp b/src/kv/test/kv_test.cpp index a356b968e6d1..e9c047d97717 100644 --- a/src/kv/test/kv_test.cpp +++ b/src/kv/test/kv_test.cpp @@ -2980,6 +2980,72 @@ TEST_CASE("Store clear") } } +TEST_CASE("Stale-view writes are rejected before local application") +{ + ccf::kv::Store store; + store.set_encryptor(std::make_shared()); + auto consensus = std::make_shared(); + store.set_consensus(consensus); + + constexpr ccf::kv::Term initial_term = 2; + constexpr auto key = "key"; + MapTypes::StringString map("public:map"); + store.initialise_term(initial_term); + + { + auto tx = store.create_tx(); + tx.rw(map)->put(key, "initial"); + REQUIRE(tx.commit() == ccf::kv::CommitResult::SUCCESS); + } + + const auto baseline_txid = store.current_txid(); + const auto baseline_replica_count = consensus->replica.size(); + + auto stale_tx = store.create_tx(); + stale_tx.rw(map)->put(key, "stale"); + + const auto new_term = initial_term + 1; + store.rollback(baseline_txid, new_term); + + REQUIRE(stale_tx.commit() == ccf::kv::CommitResult::FAIL_NO_REPLICATE); + CHECK(store.current_txid() == baseline_txid); + CHECK(consensus->replica.size() == baseline_replica_count); + { + auto tx = store.create_read_only_tx(); + CHECK(tx.ro(map)->get(key) == "initial"); + } + + { + auto tx = store.create_tx(); + tx.rw(map)->put(key, "fresh"); + REQUIRE(tx.commit() == ccf::kv::CommitResult::SUCCESS); + } + + CHECK(store.current_txid() == ccf::TxID(new_term, baseline_txid.seqno + 1)); + CHECK(consensus->replica.size() == baseline_replica_count + 1); + { + auto tx = store.create_read_only_tx(); + CHECK(tx.ro(map)->get(key) == "fresh"); + } + + const auto before_dynamic_map_txid = store.current_txid(); + auto stale_dynamic_map_tx = store.create_tx(); + stale_dynamic_map_tx.rw("public:new_map") + ->put(key, "stale"); + + store.rollback(before_dynamic_map_txid, new_term + 1); + + REQUIRE( + stale_dynamic_map_tx.commit() == ccf::kv::CommitResult::FAIL_NO_REPLICATE); + CHECK(store.current_txid() == before_dynamic_map_txid); + CHECK(store.get_map(store.current_version(), "public:new_map") == nullptr); + + auto fresh_dynamic_map_tx = store.create_tx(); + fresh_dynamic_map_tx.rw("public:new_map") + ->put(key, "fresh"); + REQUIRE(fresh_dynamic_map_tx.commit() == ccf::kv::CommitResult::SUCCESS); +} + TEST_CASE("Reported TxID after commit") { ccf::kv::Store kv_store; diff --git a/src/node/rpc/frontend.h b/src/node/rpc/frontend.h index 01a7d28af1a3..9faccf096bd8 100644 --- a/src/node/rpc/frontend.h +++ b/src/node/rpc/frontend.h @@ -900,8 +900,7 @@ namespace ccf }; } - ccf::kv::CommitResult result = - tx.commit(ctx->claims, nullptr, ws_observer); + ccf::kv::CommitResult result = tx.commit(ctx->claims, ws_observer); switch (result) { diff --git a/src/node/snapshotter.h b/src/node/snapshotter.h index 5b437c2b1901..03a0d93703bc 100644 --- a/src/node/snapshotter.h +++ b/src/node/snapshotter.h @@ -277,7 +277,7 @@ namespace ccf commit_evidence = commit_evidence_; }; - auto rc = tx.commit(cd, nullptr, capture_ws_digest_and_commit_evidence); + auto rc = tx.commit(cd, capture_ws_digest_and_commit_evidence); if (rc != ccf::kv::CommitResult::SUCCESS) { LOG_FAIL_FMT( From 66282cfb4e2561a9d3b784e43de0af22c7267af8 Mon Sep 17 00:00:00 2001 From: achamayou Date: Sun, 30 Aug 2026 20:06:28 +0100 Subject: [PATCH 2/4] Order chunk metadata and snapshot scheduling with rollback Ledger chunk sizes were appended, and snapshot scheduling rolled back, outside the lock guarding the rollback epoch. A transaction could therefore restore chunk metadata for an entry a concurrent view change had already discarded, leaving the chunker permanently ahead of the store and skewing every later chunk boundary. Take the version lock for both the rollback and the append, and skip the append when the batch's rollback epoch or view no longer holds. A rollback can only discard a batch's writes by truncating, which moves the epoch on; a rollback that does not truncate may still move the view, which consensus rejects - so both are checked. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> Copilot-Session: 75d99c5d-6efa-4048-8032-8c78b97208d9 --- CHANGELOG.md | 1 + src/kv/store.h | 43 ++++++++++++----- src/kv/test/kv_test.cpp | 102 ++++++++++++++++++++++++++++++++++++++++ 3 files changed, 135 insertions(+), 11 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index a0a08dd2dcd5..986abd8955ea 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -11,6 +11,7 @@ and this project adheres to [Semantic Versioning](http://semver.org/spec/v2.0.0. ### Fixed +- 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). ### Changed diff --git a/src/kv/store.h b/src/kv/store.h index 577040344ca4..81f4a88242ef 100644 --- a/src/kv/store.h +++ b/src/kv/store.h @@ -659,16 +659,6 @@ namespace ccf::kv // at the specified version. // No transactions can be prepared or committed during rollback. - if (snapshotter) - { - snapshotter->rollback(tx_id.seqno); - } - - if (chunker) - { - chunker->rolled_back_to(tx_id.seqno); - } - std::lock_guard mguard(maps_lock); { @@ -696,6 +686,16 @@ namespace ccf::kv if (tx_id.seqno >= version) { + if (snapshotter) + { + snapshotter->rollback(tx_id.seqno); + } + if (chunker) + { + // Keep this ordered with append_entry_size() below, so a commit + // cannot restore chunk metadata after this rollback. + chunker->rolled_back_to(tx_id.seqno); + } return; } @@ -707,6 +707,16 @@ namespace ccf::kv unset_flag_unsafe(StoreFlag::SNAPSHOT_AT_NEXT_SIGNATURE); rollback_count++; pending_txs.clear(); + if (snapshotter) + { + snapshotter->rollback(tx_id.seqno); + } + if (chunker) + { + // Keep this ordered with append_entry_size() below, so a commit + // cannot restore chunk metadata after this rollback. + chunker->rolled_back_to(tx_id.seqno); + } auto e = get_encryptor(); if (e) { @@ -1083,7 +1093,18 @@ namespace ccf::kv if (chunker) { - chunker->append_entry_size(data_shared->size()); + std::lock_guard vguard(version_lock); + // A rollback can only discard this batch's writes by truncating, + // which requires its target to be below `version` and therefore + // increments rollback_count. A rollback that does not truncate + // leaves the writes intact, but may still move the term on, which + // consensus will reject - so both are checked here. + if ( + previous_rollback_count == rollback_count && + replication_view == term_of_next_version) + { + chunker->append_entry_size(data_shared->size()); + } } LOG_DEBUG_FMT( diff --git a/src/kv/test/kv_test.cpp b/src/kv/test/kv_test.cpp index e9c047d97717..7a5741a448f9 100644 --- a/src/kv/test/kv_test.cpp +++ b/src/kv/test/kv_test.cpp @@ -3424,6 +3424,108 @@ TEST_CASE("Range") } } +// Exposes the version the chunker has recorded entries up to, which is the +// state a rollback and a concurrent commit can disagree about. +class InspectableChunker : public ccf::kv::LedgerChunker +{ +public: + ccf::kv::Version current_version() + { + ccf::pal::MutexGuard guard(chunker_lock); + return current_tx_version; + } +}; + +// A PendingTx which rolls the store back while Store::commit() is midway +// through the batch it belongs to. Store::commit() calls this after releasing +// version_lock, so it reproduces a rollback landing between a batch being +// assembled and its chunk metadata being recorded, without needing threads. +class RollingBackPendingTx : public ccf::kv::PendingTx +{ + ccf::TxID txid; + ccf::kv::Store& store; + MapTypes::StringString& table; + ccf::TxID rollback_to; + ccf::kv::Term rollback_term; + +public: + RollingBackPendingTx( + ccf::TxID txid_, + ccf::kv::Store& store_, + MapTypes::StringString& table_, + ccf::TxID rollback_to_, + ccf::kv::Term rollback_term_) : + txid(txid_), + store(store_), + table(table_), + rollback_to(rollback_to_), + rollback_term(rollback_term_) + {} + + ccf::kv::PendingTxInfo call() override + { + auto tx = store.create_reserved_tx(txid); + tx.rw(table)->put("key", "value"); + auto info = tx.commit_reserved(); + store.rollback(rollback_to, rollback_term); + return info; + } +}; + +TEST_CASE("Chunk metadata is not restored by a batch a rollback discarded") +{ + 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"); + + INFO("Commit an ordinary transaction to establish a baseline"); + { + auto tx = store.create_tx(); + tx.rw(map)->put("key", "initial"); + REQUIRE(tx.commit() == ccf::kv::CommitResult::SUCCESS); + } + + const auto baseline_txid = store.current_txid(); + REQUIRE(chunker->current_version() == baseline_txid.seqno); + + INFO( + "A batch whose writes are discarded by a rollback must not leave chunk " + "metadata behind"); + { + const auto reserved = store.next_txid(); + REQUIRE(reserved.seqno == baseline_txid.seqno + 1); + + // The rollback target is below the reserved version, so it truncates and + // moves the rollback epoch on - exactly what a real election would do. + store.commit( + reserved, + std::make_unique( + reserved, store, map, baseline_txid, initial_term + 1), + false); + } + + CHECK(store.current_txid() == baseline_txid); + CHECK(chunker->current_version() == baseline_txid.seqno); + + INFO( + "The next transaction is chunked against its own version, with no " + "accumulated offset from the discarded batch"); + { + auto tx = store.create_tx(); + tx.rw(map)->put("key", "fresh"); + REQUIRE(tx.commit() == ccf::kv::CommitResult::SUCCESS); + } + + CHECK(chunker->current_version() == store.current_version()); +} + TEST_CASE("Ledger entry chunk request") { ccf::kv::Store store; From 4562e3c186cdcf0b4e9830102d8676dfffa9b08a Mon Sep 17 00:00:00 2001 From: achamayou Date: Sun, 30 Aug 2026 20:29:02 +0100 Subject: [PATCH 3/4] Stop a rollback moving chunk metadata forward A rollback whose target is at or beyond the store's own version discards nothing, but still reset the chunker to that target. The chunker can legitimately lag the store, because an entry is allocated a version well before its size is recorded, so this moved the chunker forward past the store and left a permanent offset that skewed every later chunk boundary. Clamp the target to the store's version, so a rollback can only ever move chunk metadata back. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> Copilot-Session: 75d99c5d-6efa-4048-8032-8c78b97208d9 --- CHANGELOG.md | 1 + src/kv/store.h | 8 +++++--- src/kv/test/kv_test.cpp | 45 +++++++++++++++++++++++++++++++++++++++++ 3 files changed, 51 insertions(+), 3 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 986abd8955ea..0095245ac6e1 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -11,6 +11,7 @@ and this project adheres to [Semantic Versioning](http://semver.org/spec/v2.0.0. ### Fixed +- 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/store.h b/src/kv/store.h index 81f4a88242ef..91f884f54547 100644 --- a/src/kv/store.h +++ b/src/kv/store.h @@ -692,9 +692,11 @@ namespace ccf::kv } if (chunker) { - // Keep this ordered with append_entry_size() below, so a commit - // cannot restore chunk metadata after this rollback. - chunker->rolled_back_to(tx_id.seqno); + // Nothing local is discarded here, but the chunker may still be + // behind `version` if an allocated entry has not reached + // append_entry_size() yet. Clamp so this can only ever move the + // chunker back, never forward past the Store. + chunker->rolled_back_to(std::min(tx_id.seqno, version)); } return; } diff --git a/src/kv/test/kv_test.cpp b/src/kv/test/kv_test.cpp index 7a5741a448f9..adc5479c3359 100644 --- a/src/kv/test/kv_test.cpp +++ b/src/kv/test/kv_test.cpp @@ -3526,6 +3526,51 @@ TEST_CASE("Chunk metadata is not restored by a batch a rollback discarded") CHECK(chunker->current_version() == store.current_version()); } +TEST_CASE("A rollback never moves chunk metadata past the store's version") +{ + 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"); + + { + auto tx = store.create_tx(); + tx.rw(map)->put("key", "initial"); + REQUIRE(tx.commit() == ccf::kv::CommitResult::SUCCESS); + } + + const auto version = store.current_version(); + REQUIRE(chunker->current_version() == version); + + SUBCASE("Rollback to the current version") + { + store.rollback(store.current_txid(), initial_term + 1); + } + + SUBCASE("Rollback beyond the current version") + { + store.rollback({initial_term, version + 3}, initial_term + 1); + } + + // Neither discards anything, so neither may move the chunker. + CHECK(store.current_version() == version); + CHECK(chunker->current_version() == version); + + INFO("Later entries are still recorded against their own version"); + { + auto tx = store.create_tx(); + tx.rw(map)->put("key", "fresh"); + REQUIRE(tx.commit() == ccf::kv::CommitResult::SUCCESS); + } + CHECK(chunker->current_version() == store.current_version()); +} + TEST_CASE("Ledger entry chunk request") { ccf::kv::Store store; From 03bb4a113bfe19a201561fc963344359af01125f Mon Sep 17 00:00:00 2001 From: achamayou Date: Sun, 30 Aug 2026 20:58:53 +0100 Subject: [PATCH 4/4] Guard rollback-sensitive transaction flags A transaction may request a forced ledger chunk, or arm a snapshot at the next signature. Both were applied after the transaction's writes, with no recheck that a concurrent view change had not already discarded them, so a transaction that never reached the ledger could still leave a chunk boundary behind, or arm a snapshot. Apply both under version_lock via Store::apply_tx_flags, refusing when the transaction's view or rollback epoch no longer holds, and attach the forced chunk to the transaction's own version rather than whichever version the store had since reached. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> Copilot-Session: 75d99c5d-6efa-4048-8032-8c78b97208d9 --- CHANGELOG.md | 1 + src/kv/committable_tx.h | 44 ++++++++++++++++-------- src/kv/kv_types.h | 6 ++++ src/kv/store.h | 28 ++++++++++++++++ src/kv/test/kv_test.cpp | 74 +++++++++++++++++++++++++++++++++++++++++ 5 files changed, 140 insertions(+), 13 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 0095245ac6e1..bf0dc9788281 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -11,6 +11,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 f0e1cb866b29..9fd93946b393 100644 --- a/src/kv/committable_tx.h +++ b/src/kv/committable_tx.h @@ -263,26 +263,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 552c7a643d9f..93e3e4702c39 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 91f884f54547..bea12a17e9dd 100644 --- a/src/kv/store.h +++ b/src/kv/store.h @@ -1418,6 +1418,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 adc5479c3359..b65547831992 100644 --- a/src/kv/test/kv_test.cpp +++ b/src/kv/test/kv_test.cpp @@ -3571,6 +3571,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;