Reject stale-view writes before local commit

CHANGELOG.md

@@ -9,10 +9,15 @@ and this project adheres to [Semantic Versioning](http://semver.org/spec/v2.0.0.
 
 [7.0.14]: https://github.com/microsoft/CCF/releases/tag/ccf-7.0.14
 
+### Fixed
+
+- If the view changed while a transaction was committing, the transaction 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 and C++ applications built against it now require C++23. The supported minimum Clang version remains 18.1.2. (#8234)
 - `sandbox.sh` now derives node configuration defaults and CLI descriptions from the `cchost` configuration schema, rather than using defaults selected by the end-to-end test infrastructure. This changes the sandbox defaults for signature delay (100 ms -> 1000 ms), election timeout (4000 ms -> 5000 ms), ledger chunk size (5000000 bytes -> `5MB`, or 5242880 bytes), initial node and service certificate validity (90 days -> 1 day), and tick interval (1 ms -> 10 ms). Environment variables used by the test infrastructure no longer override sandbox defaults; for example, use the existing `--election-timeout-ms` option instead of `ELECTION_TIMEOUT_MS` (#8176).
+- `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).
 
 ### Fixed
 

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<std::tuple<Version, VersionLastNewMap>(
-    bool tx_contains_new_map)>;
+  using VersionResolution = std::tuple<Version, VersionLastNewMap>;
+  using VersionResolver =
+    std::function<std::optional<VersionResolution>(bool tx_contains_new_map)>;
 
   static inline std::optional<Version> 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));
+          }
         }
       }
     }

src/kv/committable_tx.h

@@ -189,18 +189,16 @@ 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.
      *
      * @return transaction outcome
      */
     CommitResult commit(
       const ccf::ClaimsDigest& claims = ccf::empty_claims(),
-      std::function<std::tuple<Version, Version>(bool has_new_map)>
-        version_resolver = nullptr,
       WriteSetObserver write_set_observer = nullptr)
     {
       if (committed)
@@ -241,16 +239,25 @@ namespace ccf::kv
       std::optional<Version> new_maps_conflict_version = std::nullopt;
 
       bool track_deletes_on_missing_keys = false;
+      bool commit_term_changed = false;
       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,
+          [&](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<VersionResolution>{};
+            }
+
+            const auto& resolved = resolution.value();
+            return std::optional<VersionResolution>(
+              std::in_place, std::get<0>(resolved), std::get<1>(resolved));
+          },
           hooks,
           pimpl->created_maps,
           new_maps_conflict_version,
@@ -263,6 +270,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;
       }

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<Version, Version> next_version(bool commit_new_map) = 0;
+    virtual std::optional<std::tuple<Version, Version, Version>> next_version(
+      bool commit_new_map, Term expected_commit_term) = 0;
     virtual ccf::TxID next_txid() = 0;
 
     virtual Version current_version() = 0;

src/kv/store.h

@@ -1176,9 +1176,23 @@ namespace ccf::kv
       return rollback_count == count;
     }
 
-    std::tuple<Version, Version> next_version(bool commit_new_map) override
+    std::optional<std::tuple<Version, Version, Version>> next_version(
+      bool commit_new_map, Term expected_commit_term) override
     {
       std::lock_guard<ccf::pal::Mutex> 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

src/kv/test/kv_test.cpp

@@ -2982,6 +2982,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<ccf::kv::NullTxEncryptor>());
+  auto consensus = std::make_shared<ccf::kv::test::PrimaryStubConsensus>();
+  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<MapTypes::StringString>("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<MapTypes::StringString>("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;

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)
           {

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(