Order chunk metadata and snapshot scheduling with rollback

CHANGELOG.md

@@ -15,7 +15,8 @@ and this project adheres to [Semantic Versioning](http://semver.org/spec/v2.0.0.
 
 ### 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).
+- 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
 

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<ccf::ds::Mutex> 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<ccf::ds::Mutex> 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(

src/kv/test/kv_test.cpp

@@ -3494,6 +3494,108 @@ TEST_CASE("Reserved transaction map creation is serialised with lookups")
       fmt::format("public:reserved_{}", reserved_txs - 1)) != nullptr);
 }
 
+// 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::ds::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<ccf::kv::NullTxEncryptor>());
+  auto consensus = std::make_shared<ccf::kv::test::PrimaryStubConsensus>();
+  store.set_consensus(consensus);
+  auto chunker = std::make_shared<InspectableChunker>();
+  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<RollingBackPendingTx>(
+        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;