Fix signature rollback race (#7746)

Co-authored-by: Amaury Chamayou <amchamay@microsoft.com>

Author: cjen1-msft

src/kv/committable_tx.h

@@ -414,7 +414,14 @@ namespace ccf::kv
 
       if (!success)
       {
-        throw std::logic_error("Failed to commit reserved transaction");
+        if (pimpl->store->check_rollback_count(rollback_count))
+        {
+          throw std::logic_error("Failed to commit reserved transaction");
+        }
+
+        committed = true;
+        return {
+          CommitResult::FAIL_NO_REPLICATE, {}, ccf::empty_claims(), {}, {}};
       }
 
       ccf::crypto::Sha256Hash commit_evidence_digest;

src/kv/store.h

@@ -642,6 +642,9 @@ namespace ccf::kv
 
         version = tx_id.seqno;
         last_replicated = tx_id.seqno;
+        // In practice rollback is only called at signature seqnos, so
+        // clamping here restores the latest committable entry
+        last_committable = std::min(last_committable, tx_id.seqno);
         unset_flag_unsafe(StoreFlag::SNAPSHOT_AT_NEXT_SIGNATURE);
         rollback_count++;
         pending_txs.clear();
@@ -992,14 +995,23 @@ namespace ccf::kv
         auto hooks_shared =
           std::make_shared<ccf::kv::ConsensusHookPtrs>(std::move(hooks_));
 
-        // NB: this cannot happen currently. Regular Tx only make it here if
-        // they did succeed, and signatures cannot conflict because they
-        // execute in order with a read_version that's version - 1, so even
-        // two contiguous signatures are fine
-        if (success_ != CommitResult::SUCCESS)
+        // A pending tx may fail here if rollback invalidated a reserved
+        // signature tx after it was dequeued from pending_txs.
+        if (success_ == CommitResult::FAIL_NO_REPLICATE)
         {
           LOG_DEBUG_FMT(
             "Failed Tx commit {}", previous_last_replicated + offset);
+          return success_;
+        }
+        // We should never fail from here, as normal txs have already succeeded
+        // and reserved txs only fail with FAIL_NO_REPLICATE
+        if (success_ != CommitResult::SUCCESS)
+        {
+          LOG_FAIL_FMT(
+            "Unexpected failure reason {} during commit of {}.{}",
+            static_cast<int>(success_),
+            txid.view,
+            txid.seqno);
         }
 
         if (h)

src/kv/test/stub_consensus.h

@@ -5,6 +5,7 @@
 #include "ccf/crypto/symmetric_key.h"
 #include "consensus/aft/impl/state.h"
 #include "kv/kv_types.h"
+#include "kv/store.h"
 
 #include <algorithm>
 #include <iostream>

src/node/test/history.cpp

@@ -18,6 +18,11 @@
 #include <doctest/doctest.h>
 #undef FAIL
 
+#include <condition_variable>
+#include <exception>
+#include <mutex>
+#include <thread>
+
 using MapT = ccf::kv::Map<size_t, size_t>;
 
 constexpr size_t certificate_validity_period_days = 365;
@@ -310,6 +315,55 @@ class TestPendingTx : public ccf::kv::PendingTx
   }
 };
 
+struct PausedSignatureCommit
+{
+  std::mutex lock;
+  std::condition_variable reserved_tx_created_cv;
+  std::condition_variable resume_cv;
+  bool reserved_tx_created = false;
+  bool resume = false;
+};
+
+class PausedReservedSignatureTx : public ccf::kv::PendingTx
+{
+  ccf::TxID txid;
+  ccf::kv::Store& store;
+  ccf::Signatures signatures;
+  ccf::SerialisedMerkleTree serialised_tree;
+  PausedSignatureCommit& paused;
+
+public:
+  PausedReservedSignatureTx(
+    ccf::TxID txid_, ccf::kv::Store& store_, PausedSignatureCommit& paused_) :
+    txid(txid_),
+    store(store_),
+    signatures(ccf::Tables::SIGNATURES),
+    serialised_tree(ccf::Tables::SERIALISED_MERKLE_TREE),
+    paused(paused_)
+  {}
+
+  ccf::kv::PendingTxInfo call() override
+  {
+    auto tx = store.create_reserved_tx(txid);
+    auto sig = tx.rw(signatures);
+    auto tree = tx.rw(serialised_tree);
+
+    sig->put(ccf::PrimarySignature(ccf::kv::test::PrimaryNodeId, txid.seqno));
+    tree->put({});
+
+    {
+      std::lock_guard<std::mutex> guard(paused.lock);
+      paused.reserved_tx_created = true;
+    }
+    paused.reserved_tx_created_cv.notify_one();
+
+    std::unique_lock<std::mutex> guard(paused.lock);
+    paused.resume_cv.wait(guard, [this]() { return paused.resume; });
+
+    return tx.commit_reserved();
+  }
+};
+
 TEST_CASE(
   "Batches containing but not ending on a committable transaction should not "
   "halt replication")
@@ -457,6 +511,87 @@ TEST_CASE(
   }
 }
 
+TEST_CASE(
+  "Reserved signature tx returns no-replicate if rolled back before "
+  "commit_reserved")
+{
+  ccf::kv::Store store;
+  auto encryptor = std::make_shared<ccf::kv::NullTxEncryptor>();
+  store.set_encryptor(encryptor);
+  auto consensus = std::make_shared<ccf::kv::test::PrimaryStubConsensus>();
+  store.set_consensus(consensus);
+  constexpr auto store_term = 2;
+  store.initialise_term(store_term);
+
+  MapT table("public:table");
+
+  INFO("Commit two normal transactions before emitting a signature");
+  {
+    auto tx = store.create_tx();
+    auto* txv = tx.rw(table);
+    txv->put(0, 1);
+    REQUIRE(tx.commit() == ccf::kv::CommitResult::SUCCESS);
+  }
+
+  {
+    auto tx = store.create_tx();
+    auto* txv = tx.rw(table);
+    txv->put(0, 2);
+    REQUIRE(tx.commit() == ccf::kv::CommitResult::SUCCESS);
+  }
+
+  REQUIRE(store.current_version() == 2);
+
+  auto txid = store.next_txid();
+  REQUIRE(txid == ccf::TxID(store_term, 3));
+
+  PausedSignatureCommit paused;
+  std::optional<ccf::kv::CommitResult> worker_result;
+
+  std::thread worker([&]() {
+    worker_result = store.commit(
+      txid,
+      std::make_unique<PausedReservedSignatureTx>(txid, store, paused),
+      true);
+  });
+
+  {
+    std::unique_lock<std::mutex> guard(paused.lock);
+    paused.reserved_tx_created_cv.wait(
+      guard, [&paused]() { return paused.reserved_tx_created; });
+  }
+
+  const auto new_term = store_term + 1;
+
+  INFO(
+    "Rollback after create_reserved_tx but before commit_reserved in a new "
+    "term");
+  store.rollback({store_term, 1}, new_term);
+  REQUIRE(store.commit_view() == new_term);
+  REQUIRE(store.current_txid() == ccf::TxID(store_term, 1));
+
+  {
+    std::lock_guard<std::mutex> guard(paused.lock);
+    paused.resume = true;
+  }
+  paused.resume_cv.notify_one();
+  worker.join();
+
+  REQUIRE(worker_result.has_value());
+  REQUIRE(worker_result.value() == ccf::kv::CommitResult::FAIL_NO_REPLICATE);
+  REQUIRE(store.current_txid() == ccf::TxID(store_term, 1));
+
+  INFO("A normal transaction can still commit after the failed signature path");
+  {
+    auto tx = store.create_tx();
+    auto* txv = tx.rw(table);
+    txv->put(0, 3);
+    REQUIRE(tx.commit() == ccf::kv::CommitResult::SUCCESS);
+  }
+
+  REQUIRE(store.current_txid() == ccf::TxID(new_term, 2));
+}
+
 TEST_CASE(
   "Check that rollback during replicate does not cause replication halts")
 {