Guard rollback-sensitive transaction flags

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

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<Version> c;
+      std::optional<Version> expected_rollback_count;
       {
         MapSetLockGuard map_set_guard(*pimpl->store, maps_created);
         c = apply_changes(
@@ -254,9 +255,12 @@ namespace ccf::kv
               return std::optional<VersionResolution>{};
             }
 
-            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<VersionResolution>(
-              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

src/kv/kv_types.h

@@ -744,6 +744,12 @@ namespace ccf::kv
       std::unique_ptr<PendingTx> 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<AbstractSnapshot> snapshot_unsafe_maps(
       Version v) = 0;

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<ccf::ds::Mutex> 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<ccf::ds::Mutex> vguard(version_lock);

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<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");
+
+  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;