Rolled-back follower writes can enter a later leader's replication batch

[achamayou]

Describe the bug

A transaction already rolled back locally can be queued by Store::commit() while the node is a follower, survive a subsequent election, and then be included in a fresh leader's replication batch.

This was reproduced against current CCF ee34e7b2e0d3ae7f7a43a913d8ecfe35f7d74a9e using the real Store + AFT + Merkle history fixture from #8238. After an ordinary fresh commit, the Store was at 4.2 while both history and AFT had advanced to seqno 3. The old write was absent from the local map. The stale transaction also incorrectly returned SUCCESS after its local write had been rolled back.

The relevant ordering is:

• Store::commit() lines 1001-1021 rejects a stale term only when is_primary() is true, then inserts the pending transaction.
• Lines 1033-1055 leave an entry queued behind a hole and return SUCCESS.
• Store::rollback() lines 687-710 returns without clearing pending_txs when the rollback target equals the current version.
• A subsequent fresh commit drains both its own entry and the stale pending entry into the new-view batch.

To Reproduce

Use #8238 at 344e89c59adac38087267bd9189b8e7b855bb328 as a harness overlay on the current revision above, rather than checking out the PR's older base. Add the following source to commit_concurrency_test alongside its existing threaded scenarios. This uses the PR's checkpoint to pause ordinary transactions after local application/serialization but before Store::commit().

#include "commit_concurrency/threaded/fixture.h"

#define DOCTEST_CONFIG_NO_SHORT_MACRO_NAMES
#include <doctest/doctest.h>
#include <thread>

using namespace ccf::kv::test;
using namespace std::chrono_literals;

class PausedWrite
{
  Checkpoint checkpoint;
  std::jthread worker;

public:
  ccf::kv::CommitResult result = ccf::kv::CommitResult::FAIL_CONFLICT;

  PausedWrite(CommitConcurrencyFixture& fixture, size_t key)
  {
    worker = std::jthread([this, &fixture, key]() {
      auto tx = fixture.store->create_tx();
      tx.rw(fixture.table)->put(key, key);
      result = tx.commit(
        ccf::empty_claims(), checkpoint_write_set_observer(checkpoint));
    });
    checkpoint.wait_until_paused();
  }

  ~PausedWrite()
  {
    resume();
  }

  void resume()
  {
    checkpoint.release();
    if (worker.joinable())
    {
      worker.join();
    }
  }
};

DOCTEST_TEST_CASE("Stale follower pending write survives election")
{
  CommitConcurrencyFixture fixture;
  const auto baseline = fixture.commit_signature();
  PausedWrite first(fixture, 10);
  PausedWrite second(fixture, 20);
  DOCTEST_REQUIRE(
    fixture.store->current_txid().seqno == baseline.seqno + 2);

  fixture.step_down();
  first.resume();
  DOCTEST_REQUIRE(
    first.result == ccf::kv::CommitResult::FAIL_NO_REPLICATE);
  DOCTEST_REQUIRE(fixture.store->current_txid() == baseline);

  second.resume();
  DOCTEST_CHECK(
    second.result == ccf::kv::CommitResult::FAIL_NO_REPLICATE);
  DOCTEST_REQUIRE(fixture.store->current_txid() == baseline);
  DOCTEST_REQUIRE_FALSE(read_value(*fixture.store, fixture.table, 20));

  fixture.raft->periodic(200ms);
  DOCTEST_REQUIRE(fixture.raft->is_primary());
  DOCTEST_REQUIRE(fixture.store->current_txid() == baseline);

  auto fresh = fixture.store->create_tx();
  fresh.rw(fixture.table)->put(30, 30);
  DOCTEST_REQUIRE(fresh.commit() == ccf::kv::CommitResult::SUCCESS);
  const auto local = fixture.store->current_txid();
  DOCTEST_CHECK(fixture.raft->get_last_idx() == local.seqno);
  DOCTEST_CHECK(fixture.history_txid().seqno == local.seqno);
  DOCTEST_CHECK_FALSE(read_value(*fixture.store, fixture.table, 20));
}

Pinned sequence:

1. Commit the baseline signature at 2.1.
2. Pause two ordinary writers after their writes receive seqnos 2 and 3.
3. Lose leadership.
4. Resume the first writer. AFT refuses replication as a follower and rolls the Store back to the baseline, discarding both local writes.
5. Resume the second writer while still a follower. Its old-view entry at seqno 3 is queued behind missing seqno 2, and the call returns SUCCESS.
6. Run the normal election path with periodic(200ms). The singleton fixture wins election; the rollback target equals the Store version, so the stale queue entry survives.
7. Commit one fresh write at 4.2. The pending batch also includes the discarded seqno 3 entry.

Observed output from the instrumented test:

baseline=2.1, stale result=SUCCESS, fresh=4.2, store=4.2,
history seqno=3, raft last_idx=3, rolled-back key present=false

replicated == local.seqno: 3 == 2 [fails]
history.seqno == local.seqno: 3 == 2 [fails]

Expected behavior

A transaction discarded by rollback must not be accepted into a later replication batch. Fresh writes after an election must leave the local Store, history, and AFT's replicated index consistent. The stale transaction should fail rather than return success for a discarded write.

Environment information

• CCF: ee34e7b2e0d3ae7f7a43a913d8ecfe35f7d74a9e (ccf-7.0.13-38-gee34e7b2e).
• Harness: [WIP] Systematic testing: Test the full KV + History + Consensus stack, with deterministic and randomly-explored interleavings #8238, 344e89c59adac38087267bd9189b8e7b855bb328.
• Ubuntu 26.04 under WSL; Clang 21.1.8; CMake 4.2.3; Debug build; USE_SNMALLOC=OFF.
• Real OS threads, real Store/AFT/history, fixture's null transaction encryptor and stub ledger/network boundaries.

Additional context

This is distinct from the stale-view-before-local-application case fixed by #8242: both transactions here receive their versions before leadership changes. The original #8238 threaded and scheduled suites passed at this revision; the new scenario keeps the node a follower long enough for an already-discarded second commit to arrive.

The delayed writer corresponds to preemption in an ordinary frontend commit. The fixture regains leadership via periodic(), not force_become_primary(). A full multi-node end-to-end reproduction has not been run. This issue concerns concurrency correctness; no production fix is included.

[achamayou]

Three-node reproduction, with real elections

The reproducer above loses leadership through the harness fixture's step_down(), which hands the node a hand-built RequestVote attributed to a peer that is in no configuration, and regains it through a single-node election. The fair objection is that neither view change is a real one, so I rebuilt the same sequence on a three-node service where nothing is synthesised.

Node A runs the same real ccf::kv::Store + ccf::MerkleTxHistory + aft::Aft stack. Nodes B and C are ordinary aft::Aft instances over aft::LoggingStubStore, as in the existing raft unit tests under src/consensus/aft/test. Every message is produced by one aft::Aft and delivered unmodified to another, with the same AppendEntries payload handling as src/consensus/aft/test/test_common.h.

The sequence:

1. A bootstraps the service, commits a transaction, signs, and both peers acknowledge, so the signature at 2.2 is globally committed.
2. Two ordinary transactions are in flight on A, paused between local apply and Store::commit(), holding seqnos 3 and 4.
3. B's election timer expires and it campaigns. A steps down on B's RequestVote, and B wins term 3 with real votes from A and C.
4. The first paused transaction reaches Store::commit(), AFT refuses to replicate, and the KV rolls back to 2.2, discarding both writes.
5. The second reaches Store::commit() after that rollback. The stale-view check at src/kv/store.h:1001 is gated on is_primary(), which is now false, so nothing rejects it. It is parked in pending_txs at seqno 4 and returns SUCCESS.
6. A's election timer expires and it wins term 4 with real votes from B and C. become_leader() rolls back to the last committable index, which A is already at, so store.h:687-710 returns early and pending_txs is not cleared.
7. One ordinary transaction in term 4 fills the gap, and the batch drains both it and the discarded write.

Worth noting that pre-vote is enabled throughout and does not prevent step 3. Aft::recv_request_vote_unsafe only applies the leader_id and voted_for checks when election_type == ElectionType::RegularVote, so a follower whose timer expires can always disrupt a healthy primary if its log is up to date.

Result

baseline=2.2, terms 2->3->4, stale result=SUCCESS, fresh=4.3, store=4.3,
history seqno=4, raft last_idx=4, discarded key present=false

raft_stale_pending.cpp:297: ERROR: CHECK( second.result == ccf::kv::CommitResult::FAIL_NO_REPLICATE ) is NOT correct!
raft_stale_pending.cpp:332: ERROR: CHECK( a.raft->get_last_idx() == local.seqno ) is NOT correct!  values: CHECK( 4 == 3 )
raft_stale_pending.cpp:333: ERROR: CHECK( a.history_txid().seqno == local.seqno ) is NOT correct!  values: CHECK( 4 == 3 )
raft_stale_pending.cpp:339: ERROR: CHECK( b.raft->ledger->ledger.size() == local.seqno ) is NOT correct!  values: CHECK( 4 == 3 )
raft_stale_pending.cpp:340: ERROR: CHECK( c.raft->ledger->ledger.size() == local.seqno ) is NOT correct!  values: CHECK( 4 == 3 )
raft_stale_pending.cpp:350: ERROR: CHECK( a.raft->ledger->ledger.size() > ledger_before ) is NOT correct!  values: CHECK( 4 > 4 )

Two consequences are only visible with peers present:

• The discarded write replicates to B and C like any other entry. Their ledgers hold 4 entries while A's own KV is at version 3, so a transaction A itself discarded is now held by the rest of the service, and can be committed by it.
• A's last_replicated has moved past its version. The next transaction A accepts returns SUCCESS, but Store::commit() finds no contiguous batch for it and never replicates it - the ledger does not grow. That transaction is silently lost.

Reproducer

Same setup as the original report (#8238 harness applied to ee34e7b2e, registered through the same register.cmake). Save as tests/concurrency_investigation/raft_stale_pending.cpp and add it to commit_concurrency_test.

raft_stale_pending.cpp

// Copyright (c) Microsoft Corporation. All rights reserved.
// Licensed under the Apache 2.0 License.

// A three-node raft scenario for https://github.com/microsoft/CCF/issues/8293.
//
// stale_pending.cpp shows the same defect on the single-node fixture, where
// leadership is lost through CommitConcurrencyFixture::step_down() (a
// hand-built RequestVote attributed to an unconfigured phantom peer) and
// regained through a single-node election. The obvious objection to that is
// that neither view change is a real one.
//
// Nothing is synthesised here. Node A loses leadership because a real peer
// times out and campaigns, and regains it by winning a real election with
// real votes. Every message is produced by an aft::Aft instance and delivered
// unmodified to another aft::Aft instance.
//
// Only node A runs the real ccf::kv::Store + ccf::MerkleTxHistory stack that
// is under test. Nodes B and C use aft::LoggingStubStore, as every other raft
// unit test under src/consensus/aft/test does - their job is to run the
// protocol (time out, campaign, vote, acknowledge), not to keep a KV.

#include "commit_concurrency/threaded/fixture.h"

#define DOCTEST_CONFIG_NO_SHORT_MACRO_NAMES
#include <doctest/doctest.h>
#include <map>
#include <thread>

namespace
{
  using namespace ccf::kv::test;
  using namespace std::chrono_literals;

  using Raft = CommitConcurrencyRaft;
  using Nodes = std::map<ccf::NodeId, Raft*>;

  const ccf::NodeId node_a("A");
  const ccf::NodeId node_b("B");
  const ccf::NodeId node_c("C");

  // Matches commit_concurrency_raft_settings()
  constexpr auto request_timeout = 10ms;
  constexpr auto election_timeout = 100ms;

  ccf::kv::Configuration::Nodes three_node_configuration()
  {
    ccf::kv::Configuration::Nodes configuration;
    configuration.try_emplace(node_a);
    configuration.try_emplace(node_b);
    configuration.try_emplace(node_c);
    return configuration;
  }

  aft::ChannelStubProxy& channels_of(Raft& raft)
  {
    return *static_cast<aft::ChannelStubProxy*>(raft.channels.get());
  }

  // Node A: the node under test, running the same real stack as
  // CommitConcurrencyFixture, but as one member of a three node service.
  struct RealStackNode
  {
    ccf::NodeId id;
    std::shared_ptr<ccf::crypto::ECKeyPair> node_kp =
      ccf::crypto::make_ec_key_pair();
    std::shared_ptr<ccf::crypto::ECKeyPair_OpenSSL> service_kp =
      std::dynamic_pointer_cast<ccf::crypto::ECKeyPair_OpenSSL>(
        ccf::crypto::make_ec_key_pair());
    std::shared_ptr<ccf::kv::Store> store = std::make_shared<ccf::kv::Store>();
    std::shared_ptr<ccf::MerkleTxHistory> history;
    std::shared_ptr<Raft> raft;
    CommitConcurrencyTable table{"public:table"};

    explicit RealStackNode(const ccf::NodeId& id_) : id(id_)
    {
      store->set_encryptor(std::make_shared<ccf::kv::NullTxEncryptor>());
      history = std::make_shared<ccf::MerkleTxHistory>(*store, id, *node_kp);

      constexpr size_t certificate_validity_period_days = 365;
      const auto valid_from = ccf::ds::to_x509_time_string(
        std::chrono::system_clock::now() - std::chrono::hours(24));
      const auto valid_to = ccf::crypto::compute_cert_valid_to_string(
        valid_from, certificate_validity_period_days);
      history->set_endorsed_certificate(
        node_kp->self_sign("CN=Node", valid_from, valid_to));
      history->set_service_signing_identity(
        service_kp, ccf::COSESignaturesConfig{});
      store->set_history(history);

      raft = std::make_shared<Raft>(
        commit_concurrency_raft_settings(),
        std::make_unique<aft::Adaptor<ccf::kv::Store>>(store),
        std::make_unique<aft::LedgerStubProxy>(id),
        std::make_shared<aft::ChannelStubProxy>(),
        std::make_shared<aft::State>(id),
        nullptr);
      store->set_consensus(raft);
      raft->add_configuration(0, three_node_configuration());
      raft->start_ticking();
    }

    ccf::TxID history_txid()
    {
      auto [txid, root, term_of_next_version] =
        history->get_replicated_state_txid_and_root();
      (void)root;
      (void)term_of_next_version;
      return txid;
    }
  };

  // Nodes B and C, which only need to run the protocol.
  struct PeerNode
  {
    ccf::NodeId id;
    std::shared_ptr<aft::LoggingStubStore> store;
    std::shared_ptr<Raft> raft;

    explicit PeerNode(const ccf::NodeId& id_) :
      id(id_),
      store(std::make_shared<aft::LoggingStubStore>(id_))
    {
      raft = std::make_shared<Raft>(
        commit_concurrency_raft_settings(),
        std::make_unique<aft::Adaptor<aft::LoggingStubStore>>(store),
        std::make_unique<aft::LedgerStubProxy>(id),
        std::make_shared<aft::ChannelStubProxy>(),
        std::make_shared<aft::State>(id),
        nullptr);
      raft->add_configuration(0, three_node_configuration());
      raft->start_ticking();
    }
  };

  // Delivers every message currently queued by `from` to its addressed
  // recipient, appending the ledger payload to AppendEntries just as
  // src/consensus/aft/test/test_common.h does for the other raft tests.
  size_t deliver_from(Nodes& nodes, const ccf::NodeId& from)
  {
    auto& sender = *nodes.at(from);
    auto& messages = channels_of(sender).messages;
    size_t delivered = 0;

    while (!messages.empty())
    {
      auto [target, contents] = messages.front();
      messages.pop_front();

      const uint8_t* header = contents.data();
      auto header_size = contents.size();
      if (
        serialized::peek<aft::RaftMsgType>(header, header_size) ==
        aft::raft_append_entries)
      {
        const auto append_entries =
          *reinterpret_cast<const aft::AppendEntries*>(contents.data());
        const auto payload =
          sender.ledger->get_append_entries_payload(append_entries);
        if (!payload.has_value())
        {
          // The sender has since truncated these entries, so this message is
          // simply dropped, as it would be on a real network.
          continue;
        }
        contents.insert(contents.end(), payload->begin(), payload->end());
      }

      nodes.at(target)->recv_message(from, contents.data(), contents.size());
      ++delivered;
    }

    return delivered;
  }

  // Runs message delivery to quiescence.
  size_t deliver_all(Nodes& nodes)
  {
    constexpr size_t max_rounds = 32;
    size_t total = 0;

    for (size_t round = 0; round < max_rounds; ++round)
    {
      size_t delivered = 0;
      for (const auto& [id, raft] : nodes)
      {
        (void)raft;
        delivered += deliver_from(nodes, id);
      }

      if (delivered == 0)
      {
        return total;
      }
      total += delivered;
    }

    throw std::logic_error("Message delivery did not reach quiescence");
  }

  // Lets the primary send whatever it has appended since the last flush, and
  // delivers the resulting exchange.
  void flush(Nodes& nodes, Raft& primary)
  {
    primary.periodic(request_timeout);
    deliver_all(nodes);
  }

  class PausedWrite
  {
    Checkpoint checkpoint;
    std::jthread worker;

  public:
    ccf::kv::CommitResult result = ccf::kv::CommitResult::FAIL_CONFLICT;

    PausedWrite(RealStackNode& node, size_t key)
    {
      worker = std::jthread([this, &node, key]() {
        auto tx = node.store->create_tx();
        tx.rw(node.table)->put(key, key);
        result = tx.commit(
          ccf::empty_claims(), checkpoint_write_set_observer(checkpoint));
      });
      checkpoint.wait_until_paused();
    }

    ~PausedWrite()
    {
      resume();
    }

    void resume()
    {
      checkpoint.release();
      if (worker.joinable())
      {
        worker.join();
      }
    }
  };
}

DOCTEST_TEST_CASE(
  "A real election sequence can promote a discarded write into a later term" *
  doctest::test_suite("stale_pending_investigation"))
{
  RealStackNode a(node_a);
  PeerNode b(node_b);
  PeerNode c(node_c);
  Nodes nodes{
    {node_a, a.raft.get()}, {node_b, b.raft.get()}, {node_c, c.raft.get()}};

  // A bootstraps the service, as the first node of a new CCF network does,
  // writes one transaction, and signs. Both peers then catch up and
  // acknowledge, so the signature becomes globally committed.
  a.raft->force_become_primary();
  const auto first_term = a.raft->get_view();
  {
    auto tx = a.store->create_tx();
    tx.rw(a.table)->put(0, 0);
    DOCTEST_REQUIRE(tx.commit() == ccf::kv::CommitResult::SUCCESS);
  }
  a.history->emit_signature();
  const auto baseline = a.store->current_txid();
  flush(nodes, *a.raft);
  DOCTEST_REQUIRE(a.raft->get_last_idx() == baseline.seqno);
  DOCTEST_REQUIRE(a.raft->get_committed_seqno() == baseline.seqno);

  // Two ordinary transactions are in flight on A. Each has taken a version
  // and applied to A's KV, but neither has reached Store::commit() yet.
  PausedWrite first(a, 10);
  PausedWrite second(a, 20);
  DOCTEST_REQUIRE(a.store->current_txid().seqno == baseline.seqno + 2);

  // B's election timer expires while those two are still in flight, and it
  // campaigns. This is an ordinary, unforced CCF view change.
  b.raft->periodic(election_timeout);
  deliver_all(nodes);
  DOCTEST_REQUIRE(b.raft->is_primary());
  DOCTEST_REQUIRE_FALSE(a.raft->is_primary());
  const auto second_term = a.raft->get_view();
  DOCTEST_REQUIRE(second_term > first_term);

  // The first in-flight transaction now reaches Store::commit(). A is no
  // longer primary, so raft refuses to replicate and rolls the KV back,
  // discarding both in-flight writes.
  first.resume();
  DOCTEST_REQUIRE(first.result == ccf::kv::CommitResult::FAIL_NO_REPLICATE);
  DOCTEST_REQUIRE(a.store->current_txid() == baseline);

  // The second one reaches Store::commit() after that rollback. Its write has
  // already been discarded, and it is for a term that is over, but the
  // stale-view check in Store::commit() is gated on is_primary(), which is now
  // false, so nothing rejects it. It is parked in pending_txs at a seqno that
  // the Store no longer has, and reports success.
  second.resume();
  DOCTEST_CHECK(second.result == ccf::kv::CommitResult::FAIL_NO_REPLICATE);
  DOCTEST_REQUIRE(a.store->current_txid() == baseline);
  DOCTEST_REQUIRE_FALSE(read_value(*a.store, a.table, 20));

  // A's own election timer expires and it wins the next election, with real
  // votes from B and C. become_leader() rolls back to the last committable
  // index, which A is already at, so the rollback returns without clearing
  // pending_txs, and the parked entry survives into the new term.
  a.raft->periodic(election_timeout);
  deliver_all(nodes);
  DOCTEST_REQUIRE(a.raft->is_primary());
  DOCTEST_REQUIRE(a.raft->get_view() > second_term);
  DOCTEST_REQUIRE(a.store->current_txid().seqno == baseline.seqno);

  // One ordinary transaction in the new term fills the gap in pending_txs, so
  // the batch drains both it and the discarded write.
  auto fresh = a.store->create_tx();
  fresh.rw(a.table)->put(30, 30);
  DOCTEST_REQUIRE(fresh.commit() == ccf::kv::CommitResult::SUCCESS);

  const auto local = a.store->current_txid();
  DOCTEST_MESSAGE(fmt::format(
    "baseline={}, terms {}->{}->{}, stale result={}, fresh={}, store={}, "
    "history seqno={}, raft last_idx={}, discarded key present={}",
    baseline.to_str(),
    first_term,
    second_term,
    a.raft->get_view(),
    static_cast<int>(second.result),
    fresh.get_txid()->to_str(),
    local.to_str(),
    a.history_txid().seqno,
    a.raft->get_last_idx(),
    read_value(*a.store, a.table, 20).has_value()));

  DOCTEST_CHECK(a.raft->get_last_idx() == local.seqno);
  DOCTEST_CHECK(a.history_txid().seqno == local.seqno);
  DOCTEST_CHECK_FALSE(read_value(*a.store, a.table, 20));

  // The consequences are not confined to A. The extra entry replicates to both
  // peers like any other, so their ledgers grow past A's own KV.
  flush(nodes, *a.raft);
  DOCTEST_CHECK(b.raft->ledger->ledger.size() == local.seqno);
  DOCTEST_CHECK(c.raft->ledger->ledger.size() == local.seqno);

  // And A's Store is now permanently ahead of itself: last_replicated has
  // moved past version, so the next transaction it accepts reports success
  // without ever being replicated.
  const auto ledger_before = a.raft->ledger->ledger.size();
  auto stranded = a.store->create_tx();
  stranded.rw(a.table)->put(40, 40);
  DOCTEST_CHECK(stranded.commit() == ccf::kv::CommitResult::SUCCESS);
  flush(nodes, *a.raft);
  DOCTEST_CHECK(a.raft->ledger->ledger.size() > ledger_before);
}

[achamayou]

Origin: not a recent regression

The && get_consensus()->is_primary() conjunct was added by 134b3c5dd ("AFT: fix race conditions", #2635, 7 June 2021) as a single incidental line in a 19-file commit that was otherwise about the BFT progress tracker:

-        if (txid.term != term)
+        if (txid.term != term && consensus->is_primary())

No accompanying KV test, no comment. Under BFT, backups also executed transactions locally, which is the most plausible reason for exempting non-primaries. BFT has since been removed - src/kv retains only two stale comments mentioning it - so the conjunct is vestigial.

The rollback early return that lets the parked entry survive is older still (aa85cd8e2, #1288, 2020). Everything since is refactoring (8ad0bd7e3 #2755, c85829821 #4998, c74548a34 #7477, 8e040f924 #8265).

Fix: four lines

Inside the existing version_lock guard in Store::commit, just after the stale-view check and before the pending_txs.insert:

if (txid.seqno > version)
{
  // A rollback has discarded this transaction's writes since it took its
  // version. Parking it would let a later batch replicate writes this store
  // no longer holds.
  return CommitResult::FAIL_NO_REPLICATE;
}

next_version_unsafe() (src/kv/store.h:168) increments version under version_lock and hands the result to the transaction, so every real transaction reaches Store::commit with txid.seqno <= version. Only rollback lowers version. So txid.seqno > version is an exact discriminator between the two cases the current code conflates:

• the view changed but this transaction's writes survive - proceed, let replicate() fail and let Aft::replicate roll back, preserving whatever AFT: fix race conditions #2635 was protecting;
• this transaction's writes were already discarded - reject, and never park.

All three consequences in the report above are addressed, because nothing stale ever enters pending_txs.

Measured

Suite	Before	After
kv_test including the #8295 repro	1 case failing, 5 assertions	all 75 passing
three-node aft::Aft scenario (comment above)	12 assertions failing	all passing
raft_test	passing	passing (1,001,609 assertions)
history_test	passing	passing
historical_queries_test	passing	passing

One assertion in #8238's own src/commit_concurrency/threaded/deterministic.cpp:307 starts failing. That harness never calls next_version(); it fabricates TxIDs and commits at seqnos above Store::version, which no real transaction does. Its "reserve a hole and park behind it" setup should allocate versions through the store. Worth raising on that PR independently of this fix.

Residual

If a rollback is followed by enough fresh version allocations that version climbs back above a still-in-flight stale seqno, the check does not fire. In practice a backup does not advance version from user writes, and a re-elected primary has a different term_of_next_version so the existing stale-view check fires. A complete fix would compare the transaction's rollback_count against the store's, which needs plumbing through AbstractStore::commit and looks disproportionate.