[consensus] Revert consensus handler change where commits from the sa…

…me round are ignored (#18248)

When Narwhal is running it is still possible on restarts to resend a
commit from the same round. Mysticeti should not return more than one
commit per round unless something is broken so it is okay to keep this
check for now.

crates/sui-core/src/consensus_handler.rs

@@ -27,7 +27,7 @@ use sui_types::{
     sui_system_state::epoch_start_sui_system_state::EpochStartSystemStateTrait,
     transaction::{SenderSignedData, VerifiedTransaction},
 };
-use tracing::{debug, error, info, instrument, trace_span};
+use tracing::{debug, error, info, instrument, trace_span, warn};
 
 use crate::{
     authority::{
@@ -233,7 +233,19 @@ impl<C: CheckpointServiceNotify + Send + Sync> ConsensusHandler<C> {
 
         let round = consensus_output.leader_round();
 
-        assert!(round > last_committed_round);
+        // TODO: Remove this once narwhal is deprecated. For now mysticeti will not return
+        // more than one leader per round so we are not in danger of ignoring any commits.
+        assert!(round >= last_committed_round);
+        if last_committed_round == round {
+            // we can receive the same commit twice after restart
+            // It is critical that the writes done by this function are atomic - otherwise we can
+            // lose the later parts of a commit if we restart midway through processing it.
+            warn!(
+                "Ignoring consensus output for round {} as it is already committed. NOTE: This is only expected if Narwhal is running.",
+                round
+            );
+            return;
+        }
 
         /* (serialized, transaction, output_cert) */
         let mut transactions = vec![];
@@ -996,6 +1008,16 @@ mod tests {
             last_consensus_stats_1.stats.get_num_user_transactions(0),
             num_transactions as u64
         );
+
+        // WHEN processing the same output multiple times
+        // THEN the consensus stats do not update
+        for _ in 0..2 {
+            consensus_handler
+                .handle_consensus_output(consensus_output.clone())
+                .await;
+            let last_consensus_stats_2 = consensus_handler.last_consensus_stats.clone();
+            assert_eq!(last_consensus_stats_1, last_consensus_stats_2);
+        }
     }
 
     #[test]