consensus data quarantining

crates/mysten-common/Cargo.toml

@@ -19,3 +19,4 @@ sui-types.workspace = true
 tracing.workspace = true
 prometheus.workspace = true
 anyhow.workspace = true
+itertools.workspace = true

crates/mysten-common/src/sync/notify_read.rs

@@ -116,19 +116,37 @@ impl<K: Eq + Hash + Clone, V: Clone> NotifyRead<K, V> {
     }
 }
 
-impl<K: Eq + Hash + Clone + Unpin, V: Clone + Unpin> NotifyRead<K, V> {
+impl<K: Eq + Hash + Clone + Unpin + std::fmt::Debug, V: Clone + Unpin> NotifyRead<K, V> {
     pub async fn read(&self, keys: &[K], fetch: impl FnOnce(&[K]) -> Vec<Option<V>>) -> Vec<V> {
         let registrations = self.register_all(keys);
 
         let results = fetch(keys);
 
-        let results = results
-            .into_iter()
-            .zip(registrations)
-            .map(|(a, r)| match a {
+        let results =
+            itertools::izip!(keys, results.into_iter(), registrations).map(|(k, a, r)| match a {
                 // Note that Some() clause also drops registration that is already fulfilled
                 Some(ready) => Either::Left(futures::future::ready(ready)),
-                None => Either::Right(r),
+                None => Either::Right({
+                    #[cfg(msim)]
+                    {
+                        let mut r = r;
+                        async move {
+                            loop {
+                                tokio::select! {
+                                    _ = tokio::time::sleep(tokio::time::Duration::from_millis(1000)) => {
+                                        tracing::debug!("Waiting for notify_read of {:?}", k);
+                                    }
+                                    r = &mut r => {
+                                        break r;
+                                    }
+                                }
+                            }
+                        }
+                    }
+
+                    #[cfg(not(msim))]
+                    r
+                })
             });
 
         join_all(results).await

crates/sui-benchmark/tests/simtest.rs

@@ -168,9 +168,13 @@ mod test {
                     test_cluster_clone.start_node(validator).await;
                 }
             }
+            info!("restarter_task finished");
         });
+        info!("xxx step");
         test_simulated_load(test_cluster.clone(), 330).await;
+        info!("xxx step");
         restarter_task.await.unwrap();
+        info!("xxx step");
         test_cluster.wait_for_epoch_all_nodes(1).await;
     }
 
@@ -1196,6 +1200,8 @@ mod test {
             assert!(!results.unique_move_functions_called.is_empty());
         });
 
+        info!("joining bench_task and surfer_task");
         let _ = futures::join!(bench_task, surfer_task);
+        info!("complete");
     }
 }

crates/sui-core/src/authority.rs

@@ -22,6 +22,7 @@ use move_binary_format::binary_config::BinaryConfig;
 use move_binary_format::CompiledModule;
 use move_core_types::annotated_value::MoveStructLayout;
 use move_core_types::language_storage::ModuleId;
+use mysten_common::fatal;
 use mysten_metrics::{TX_TYPE_SHARED_OBJ_TX, TX_TYPE_SINGLE_WRITER_TX};
 use parking_lot::Mutex;
 use prometheus::{
@@ -1561,6 +1562,8 @@ impl AuthorityState {
                 .force_reload_system_packages(&BuiltInFramework::all_package_ids());
         }
 
+        epoch_store.remove_shared_version_assignments(&tx_key);
+
         // commit_certificate finished, the tx is fully committed to the store.
         tx_guard.commit_tx();
 
@@ -3021,15 +3024,6 @@ impl AuthorityState {
             .enqueue_certificates(certs, epoch_store)
     }
 
-    pub(crate) fn enqueue_with_expected_effects_digest(
-        &self,
-        certs: Vec<(VerifiedExecutableTransaction, TransactionEffectsDigest)>,
-        epoch_store: &AuthorityPerEpochStore,
-    ) {
-        self.transaction_manager
-            .enqueue_with_expected_effects_digest(certs, epoch_store)
-    }
-
     fn create_owner_index_if_empty(
         &self,
         genesis_objects: &[Object],
@@ -3116,6 +3110,7 @@ impl AuthorityState {
         epoch_start_configuration: EpochStartConfiguration,
         accumulator: Arc<StateAccumulator>,
         expensive_safety_check_config: &ExpensiveSafetyCheckConfig,
+        epoch_last_checkpoint: CheckpointSequenceNumber,
     ) -> SuiResult<Arc<AuthorityPerEpochStore>> {
         Self::check_protocol_version(
             supported_protocol_versions,
@@ -3181,6 +3176,7 @@ impl AuthorityState {
                 new_committee,
                 epoch_start_configuration,
                 expensive_safety_check_config,
+                epoch_last_checkpoint,
             )
             .await?;
         assert_eq!(new_epoch_store.epoch(), new_epoch);
@@ -3211,6 +3207,11 @@ impl AuthorityState {
             self.get_backing_package_store().clone(),
             self.get_object_store().clone(),
             &self.config.expensive_safety_check_config,
+            self.checkpoint_store
+                .get_epoch_last_checkpoint(epoch_store.epoch())
+                .unwrap()
+                .map(|c| *c.sequence_number())
+                .unwrap_or_default(),
         );
         let new_epoch = new_epoch_store.epoch();
         self.transaction_manager.reconfigure(new_epoch);
@@ -5140,6 +5141,7 @@ impl AuthorityState {
         new_committee: Committee,
         epoch_start_configuration: EpochStartConfiguration,
         expensive_safety_check_config: &ExpensiveSafetyCheckConfig,
+        epoch_last_checkpoint: CheckpointSequenceNumber,
     ) -> SuiResult<Arc<AuthorityPerEpochStore>> {
         let new_epoch = new_committee.epoch;
         info!(new_epoch = ?new_epoch, "re-opening AuthorityEpochTables for new epoch");
@@ -5156,6 +5158,7 @@ impl AuthorityState {
             self.get_object_store().clone(),
             expensive_safety_check_config,
             cur_epoch_store.get_chain_identifier(),
+            epoch_last_checkpoint,
         );
         self.epoch_store.store(new_epoch_store.clone());
         Ok(new_epoch_store)
@@ -5255,6 +5258,14 @@ impl RandomnessRoundReceiver {
         let transaction = VerifiedExecutableTransaction::new_system(transaction, epoch);
         let digest = *transaction.digest();
 
+        // Randomness state updates contain the full bls signature for the random round,
+        // which cannot necessarily be reconstructed again later. Therefore we must immediately
+        // persist this transaction. If we crash before its outputs are committed, this
+        // ensures we will be able to re-execute it.
+        self.authority_state
+            .get_cache_commit()
+            .persist_transaction(&transaction);
+
         // Send transaction to TransactionManager for execution.
         self.authority_state
             .transaction_manager()
@@ -5294,9 +5305,9 @@ impl RandomnessRoundReceiver {
 
             let effects = effects.pop().expect("should return effects");
             if *effects.status() != ExecutionStatus::Success {
-                panic!("failed to execute randomness state update transaction at epoch {epoch}, round {round}: {effects:?}");
+                fatal!("failed to execute randomness state update transaction at epoch {epoch}, round {round}: {effects:?}");
             }
-            debug!("successfully executed randomness state update transaction at epoch {epoch}, round {round}");
+            debug!("successfully executed randomness state update transaction at epoch {epoch}, round {round}: {effects:?}");
         });
     }
 }