Reset `last_committable` on rollback (#3780)

.daily_canary

@@ -1 +1 @@
-Quack Quack
+Piou

src/kv/kv_types.h

@@ -759,7 +759,7 @@ namespace kv
     virtual bool must_force_ledger_chunk(Version version) = 0;
     virtual bool must_force_ledger_chunk_unsafe(Version version) = 0;
 
-    virtual size_t commit_gap() = 0;
+    virtual size_t committable_gap() = 0;
 
     enum class Flag : uint8_t
     {

src/kv/store.h

@@ -47,7 +47,10 @@ namespace kv
     Term term_of_last_version = 0;
 
     Version last_replicated = 0;
+    // Version of the latest committable entry committed in this term and by
+    // _this_ store. Always reset on rollback.
     Version last_committable = 0;
+
     Version rollback_count = 0;
 
     std::unordered_map<Version, std::tuple<std::unique_ptr<PendingTx>, bool>>
@@ -518,7 +521,6 @@ namespace kv
         std::lock_guard<std::mutex> vguard(version_lock);
         version = v;
         last_replicated = v;
-        last_committable = v;
       }
 
       if (h)
@@ -636,7 +638,7 @@ namespace kv
 
         version = tx_id.version;
         last_replicated = tx_id.version;
-        last_committable = tx_id.version;
+        last_committable = 0;
         unset_flag_unsafe(Flag::LEDGER_CHUNK_AT_NEXT_SIGNATURE);
         unset_flag_unsafe(Flag::SNAPSHOT_AT_NEXT_SIGNATURE);
         rollback_count++;
@@ -1103,7 +1105,7 @@ namespace kv
       return {term_of_next_version, version};
     }
 
-    size_t commit_gap() override
+    size_t committable_gap() override
     {
       std::lock_guard<std::mutex> vguard(version_lock);
       return version - last_committable;

src/node/history.h

@@ -540,7 +540,8 @@ namespace ccf
             auto consensus = self->store.get_consensus();
             if (
               (consensus != nullptr) && consensus->can_replicate() &&
-              self->store.commit_gap() > 0 && time > time_of_last_signature &&
+              self->store.committable_gap() > 0 &&
+              time > time_of_last_signature &&
               (time - time_of_last_signature) > sig_ms_interval)
             {
               should_emit_signature = true;
@@ -725,12 +726,12 @@ namespace ccf
     void try_emit_signature() override
     {
       std::unique_lock<std::mutex> mguard(signature_lock, std::defer_lock);
-      if (store.commit_gap() < sig_tx_interval || !mguard.try_lock())
+      if (store.committable_gap() < sig_tx_interval || !mguard.try_lock())
       {
         return;
       }
 
-      if (store.commit_gap() >= sig_tx_interval)
+      if (store.committable_gap() >= sig_tx_interval)
       {
         mguard.unlock();
         emit_signature();