Run each simulation in its own process; make simulator test state process-global (#27435)

## Description

The deterministic simulator is gaining a blocking-task pool, so a
simulation will run code across multiple threads. Per-simulation test
state kept in `thread_local!`s (fail-point and debug-fatal registries,
framework / advance-epoch / protocol-config injection overrides, the JWK
injector, checkpoint-timeout override, fork-stake and
synthetic-execution-time accounting, node-leak counter, sim IP/port
allocator, RNG seed) is invisible across those threads — a value a test
sets on the main thread is not seen on a pool thread. This moves all of
it to process globals, which requires at most one simulation per
process.

That single-sim-per-process model retires the in-process determinism
machinery: `MSIM_TEST_NUM` (many seeds in one process),
`MSIM_TEST_CHECK_DETERMINISM` / `#[sim_test(check_determinism)]` (run
twice in one process, compare RNG logs), and `init_static_initializers`
(which existed only to make that in-process double-run deterministic,
and whose own comment notes it has no effect on process-level
determinism). Determinism is now checked externally, by running a test
twice in separate processes and diffing the output
(`scripts/simtest/check-determinism.sh`, run in the `simtest` CI job);
multi-seed searches use `seed-search.py` / a seed loop.

Independent of, and landing ahead of, the blocking pool itself: process
globals behave identically single-threaded, so this is a no-op for
current simulator behavior.

## Test plan

- `scripts/simtest/check-determinism.sh` runs `test_net_determinism`
twice in two processes with the same seed and diffs the normalized logs;
verified locally it passes (byte-identical sim output). It runs in the
`simtest` job on every PR.
- Existing simulator tests continue to exercise the converted state; the
`simulator_tests` determinism tests still run (now once each, as
ordinary tests).

## Release notes

Check each box that your change affects. If none of the boxes relate to
your changes, release notes are not required.

- [ ] Protocol:
- [ ] Nodes (Validators and Full nodes):
- [ ] gRPC:
- [ ] JSON-RPC:
- [ ] GraphQL:
- [ ] CLI:
- [ ] Rust SDK:

---------

Co-authored-by: Mark Logan <mark@marklgn.com>

Author: mystenmark

.github/workflows/rust.yml

@@ -435,9 +435,9 @@ jobs:
       - name: cargo simtest
         run: |
           MSIM_TEST_SEED="$(printf "%lu\n" 0x$(git rev-parse HEAD | cut -c1-16))" scripts/simtest/cargo-simtest simtest --profile ci --cargo-quiet --no-fail-fast
-      - name: check new tests for flakiness
+      - name: check determinism
         run: |
-          scripts/simtest/stress-new-tests.sh
+          MSIM_TEST_SEED="$(printf "%lu\n" 0x$(git rev-parse HEAD | cut -c1-16))" scripts/simtest/check-determinism.sh
 
   simtest-mainnet:
     permissions:
@@ -473,9 +473,6 @@ jobs:
       - name: cargo simtest
         run: |
           MSIM_TEST_SEED="$(printf "%lu\n" 0x$(git rev-parse HEAD | cut -c1-16))" scripts/simtest/cargo-simtest simtest --profile ci --cargo-quiet --no-fail-fast
-      - name: check new tests for flakiness
-        run: |
-          scripts/simtest/stress-new-tests.sh
 
   # This job ensures that Move unit tests are run if there are changes
   # to Move code but not Rust code (If there are Rust changes, they

.github/workflows/simulator-nightly.yml

@@ -18,7 +18,7 @@ on:
         required: true
         default: main
       test_num:
-        description: "MSIM_TEST_NUM (test iterations)"
+        description: "Number of seeds to run (seed-search iterations)"
         type: string
         required: false
         default: "30"

crates/mysten-common/src/logging.rs

@@ -42,21 +42,17 @@ pub mod intercept_debug_fatal {
         pub callback: Arc<dyn Fn() + Send + Sync>,
     }
 
-    thread_local! {
-        static INTERCEPT_DEBUG_FATAL: Mutex<Option<DebugFatalCallback>> = Mutex::new(None);
-    }
+    static INTERCEPT_DEBUG_FATAL: Mutex<Option<DebugFatalCallback>> = Mutex::new(None);
 
     pub fn register_callback(message: &str, f: impl Fn() + Send + Sync + 'static) {
-        INTERCEPT_DEBUG_FATAL.with(|m| {
-            *m.lock().unwrap() = Some(DebugFatalCallback {
-                pattern: message.to_string(),
-                callback: Arc::new(f),
-            });
+        *INTERCEPT_DEBUG_FATAL.lock().unwrap() = Some(DebugFatalCallback {
+            pattern: message.to_string(),
+            callback: Arc::new(f),
         });
     }
 
     pub fn get_callback() -> Option<DebugFatalCallback> {
-        INTERCEPT_DEBUG_FATAL.with(|m| m.lock().unwrap().clone())
+        INTERCEPT_DEBUG_FATAL.lock().unwrap().clone()
     }
 }
 

crates/sui-config/src/local_ip_utils.rs

@@ -40,11 +40,12 @@ impl SimAddressManager {
 
 #[cfg(msim)]
 fn get_sim_address_manager() -> Arc<SimAddressManager> {
-    thread_local! {
-        // Uses Arc so that we could return a clone of the thread local singleton.
-        static SIM_ADDRESS_MANAGER: Arc<SimAddressManager> = Arc::new(SimAddressManager::new());
-    }
-    SIM_ADDRESS_MANAGER.with(|s| s.clone())
+    // Uses Arc so that we could return a clone of the process-global singleton.
+    static SIM_ADDRESS_MANAGER: std::sync::OnceLock<Arc<SimAddressManager>> =
+        std::sync::OnceLock::new();
+    SIM_ADDRESS_MANAGER
+        .get_or_init(|| Arc::new(SimAddressManager::new()))
+        .clone()
 }
 
 /// In simtest, we generate a new unique IP each time this function is called.

crates/sui-core/src/authority.rs

@@ -2873,15 +2873,14 @@ impl AuthorityState {
         >,
         fork_probability: f32,
     ) {
-        use std::cell::RefCell;
-        thread_local! {
-            static TOTAL_FAILING_STAKE: RefCell<u64> = RefCell::new(0);
-        }
+        static TOTAL_FAILING_STAKE: std::sync::Mutex<u64> = std::sync::Mutex::new(0);
         if !certificate.data().intent_message().value.is_system_tx() {
             let committee = epoch_store.committee();
             let cur_stake = (**committee).weight(&self.name);
             if cur_stake > 0 {
-                TOTAL_FAILING_STAKE.with_borrow_mut(|total_stake| {
+                {
+                    let mut total_stake = TOTAL_FAILING_STAKE.lock().unwrap();
+                    let total_stake = &mut *total_stake;
                     let already_forked = forked_validators
                         .lock()
                         .ok()
@@ -2937,7 +2936,7 @@ impl AuthorityState {
                             }
                         }
                     }
-                });
+                }
             }
         }
     }
@@ -6554,17 +6553,15 @@ impl TransactionKeyValueStoreTrait for AuthorityState {
 pub mod framework_injection {
     use move_binary_format::CompiledModule;
     use std::collections::BTreeMap;
-    use std::{cell::RefCell, collections::BTreeSet};
+    use std::collections::BTreeSet;
+    use std::sync::Mutex;
     use sui_framework::{BuiltInFramework, SystemPackage};
     use sui_types::base_types::{AuthorityName, ObjectID};
     use sui_types::is_system_package;
 
     type FrameworkOverrideConfig = BTreeMap<ObjectID, PackageOverrideConfig>;
 
-    // Thread local cache because all simtests run in a single unique thread.
-    thread_local! {
-        static OVERRIDE: RefCell<FrameworkOverrideConfig> = RefCell::new(FrameworkOverrideConfig::default());
-    }
+    static OVERRIDE: Mutex<FrameworkOverrideConfig> = Mutex::new(BTreeMap::new());
 
     type Framework = Vec<CompiledModule>;
 
@@ -6588,60 +6585,61 @@ pub mod framework_injection {
     }
 
     pub fn set_override(package_id: ObjectID, modules: Vec<CompiledModule>) {
-        OVERRIDE.with(|bs| {
-            bs.borrow_mut()
-                .insert(package_id, PackageOverrideConfig::Global(modules))
-        });
+        OVERRIDE
+            .lock()
+            .unwrap()
+            .insert(package_id, PackageOverrideConfig::Global(modules));
     }
 
     pub fn set_override_cb(package_id: ObjectID, func: PackageUpgradeCallback) {
-        OVERRIDE.with(|bs| {
-            bs.borrow_mut()
-                .insert(package_id, PackageOverrideConfig::PerValidator(func))
-        });
+        OVERRIDE
+            .lock()
+            .unwrap()
+            .insert(package_id, PackageOverrideConfig::PerValidator(func));
     }
 
     pub fn set_system_packages(packages: Vec<SystemPackage>) {
-        OVERRIDE.with(|bs| {
-            let mut new_packages_not_to_include: BTreeSet<_> =
-                BuiltInFramework::all_package_ids().into_iter().collect();
-            for pkg in &packages {
-                new_packages_not_to_include.remove(&pkg.id);
-            }
-            for pkg in packages {
-                bs.borrow_mut()
-                    .insert(pkg.id, PackageOverrideConfig::Global(pkg.modules()));
-            }
-            for empty_pkg in new_packages_not_to_include {
-                bs.borrow_mut()
-                    .insert(empty_pkg, PackageOverrideConfig::Global(vec![]));
-            }
-        });
+        let mut cfg = OVERRIDE.lock().unwrap();
+        let mut new_packages_not_to_include: BTreeSet<_> =
+            BuiltInFramework::all_package_ids().into_iter().collect();
+        for pkg in &packages {
+            new_packages_not_to_include.remove(&pkg.id);
+        }
+        for pkg in packages {
+            cfg.insert(pkg.id, PackageOverrideConfig::Global(pkg.modules()));
+        }
+        for empty_pkg in new_packages_not_to_include {
+            cfg.insert(empty_pkg, PackageOverrideConfig::Global(vec![]));
+        }
     }
 
     pub fn get_override_bytes(package_id: &ObjectID, name: AuthorityName) -> Option<Vec<Vec<u8>>> {
-        OVERRIDE.with(|cfg| {
-            cfg.borrow().get(package_id).and_then(|entry| match entry {
+        OVERRIDE
+            .lock()
+            .unwrap()
+            .get(package_id)
+            .and_then(|entry| match entry {
                 PackageOverrideConfig::Global(framework) => {
                     Some(compiled_modules_to_bytes(framework))
                 }
                 PackageOverrideConfig::PerValidator(func) => {
                     func(name).map(|fw| compiled_modules_to_bytes(&fw))
                 }
             })
-        })
     }
 
     pub fn get_override_modules(
         package_id: &ObjectID,
         name: AuthorityName,
     ) -> Option<Vec<CompiledModule>> {
-        OVERRIDE.with(|cfg| {
-            cfg.borrow().get(package_id).and_then(|entry| match entry {
+        OVERRIDE
+            .lock()
+            .unwrap()
+            .get(package_id)
+            .and_then(|entry| match entry {
                 PackageOverrideConfig::Global(framework) => Some(framework.clone()),
                 PackageOverrideConfig::PerValidator(func) => func(name),
             })
-        })
     }
 
     pub fn get_override_system_package(
@@ -6666,12 +6664,12 @@ pub mod framework_injection {
 
     pub fn get_extra_packages(name: AuthorityName) -> Vec<SystemPackage> {
         let built_in = BTreeSet::from_iter(BuiltInFramework::all_package_ids().into_iter());
-        let extra: Vec<ObjectID> = OVERRIDE.with(|cfg| {
-            cfg.borrow()
-                .keys()
-                .filter_map(|package| (!built_in.contains(package)).then_some(*package))
-                .collect()
-        });
+        let extra: Vec<ObjectID> = OVERRIDE
+            .lock()
+            .unwrap()
+            .keys()
+            .filter_map(|package| (!built_in.contains(package)).then_some(*package))
+            .collect();
 
         extra
             .into_iter()

crates/sui-core/src/authority/execution_time_estimator.rs

@@ -545,9 +545,7 @@ impl ExecutionTimeObserver {
             panic!("get_test_duration called in non-test configuration");
         }
 
-        thread_local! {
-            static PER_TEST_SEED: u64 = random::<u64>();
-        }
+        static PER_TEST_SEED: std::sync::OnceLock<u64> = std::sync::OnceLock::new();
 
         let mut hasher = std::collections::hash_map::DefaultHasher::new();
 
@@ -564,7 +562,7 @@ impl ExecutionTimeObserver {
             .is_some();
 
         if !checkpoint_digest_used {
-            PER_TEST_SEED.with(|seed| seed.hash(&mut hasher));
+            PER_TEST_SEED.get_or_init(random::<u64>).hash(&mut hasher);
         }
 
         key.hash(&mut hasher);

crates/sui-core/src/checkpoints/checkpoint_executor/utils.rs

@@ -107,19 +107,17 @@ pub struct CheckpointTimeoutConfig {
     pub warning_timeout: Duration,
 }
 
-// We use a thread local so that the config can be overridden on a per-test basis. This means
-// that get_scheduling_timeout() can be called multiple times in a multithreaded context, but
-// the function is still very cheap to call so this is okay.
-thread_local! {
-    static SCHEDULING_TIMEOUT: once_cell::sync::OnceCell<CheckpointTimeoutConfig> =
-        const { once_cell::sync::OnceCell::new() };
-}
+// The config can be overridden on a per-test basis. get_scheduling_timeout() can be called
+// multiple times in a multithreaded context, but the function is still very cheap to call so
+// this is okay.
+static SCHEDULING_TIMEOUT: std::sync::OnceLock<CheckpointTimeoutConfig> =
+    std::sync::OnceLock::new();
 
 #[cfg(msim)]
 pub fn init_checkpoint_timeout_config(config: CheckpointTimeoutConfig) {
-    SCHEDULING_TIMEOUT.with(|s| {
-        s.set(config).expect("SchedulingTimeoutConfig already set");
-    });
+    SCHEDULING_TIMEOUT
+        .set(config)
+        .expect("SchedulingTimeoutConfig already set");
 }
 
 fn get_scheduling_timeout() -> CheckpointTimeoutConfig {
@@ -145,7 +143,7 @@ fn get_scheduling_timeout() -> CheckpointTimeoutConfig {
         }
     }
 
-    SCHEDULING_TIMEOUT.with(|s| *s.get_or_init(inner))
+    *SCHEDULING_TIMEOUT.get_or_init(inner)
 }
 
 pub(super) fn assert_not_forked(

crates/sui-e2e-tests/tests/full_node_tests.rs

@@ -528,11 +528,6 @@ async fn test_full_node_sync_flood() {
     do_test_full_node_sync_flood().await
 }
 
-#[sim_test(check_determinism)]
-async fn test_full_node_sync_flood_determinism() {
-    do_test_full_node_sync_flood().await
-}
-
 async fn do_test_full_node_sync_flood() {
     let mut test_cluster = TestClusterBuilder::new()
         .disable_fullnode_pruning()

crates/sui-e2e-tests/tests/reconfiguration_tests.rs

@@ -99,7 +99,7 @@ async fn test_passive_reconfig_testnet_smoke_test() {
     do_test_passive_reconfig(Some(Chain::Testnet)).await;
 }
 
-#[sim_test(check_determinism)]
+#[sim_test]
 async fn test_passive_reconfig_determinism() {
     do_test_passive_reconfig(None).await;
 }

crates/sui-e2e-tests/tests/simulator_tests.rs

@@ -33,7 +33,7 @@ async fn make_fut(i: usize) -> usize {
     i
 }
 
-#[sim_test(check_determinism)]
+#[sim_test]
 async fn test_futures_ordered() {
     telemetry_subscribers::init_for_testing();
 
@@ -46,7 +46,7 @@ async fn test_futures_ordered() {
     debug!("final rng state: {}", OsRng.r#gen::<u32>());
 }
 
-#[sim_test(check_determinism)]
+#[sim_test]
 async fn test_futures_unordered() {
     telemetry_subscribers::init_for_testing();
 
@@ -61,7 +61,7 @@ async fn test_futures_unordered() {
     debug!("final rng state: {}", OsRng.r#gen::<u32>());
 }
 
-#[sim_test(check_determinism)]
+#[sim_test]
 async fn test_select_unbiased() {
     let mut f1 = FuturesUnordered::from_iter((0..200).map(make_fut));
     let mut f2 = FuturesUnordered::from_iter((0..200).map(make_fut));
@@ -92,7 +92,7 @@ async fn test_select_unbiased() {
     debug!("final rng state: {}", OsRng.r#gen::<u32>());
 }
 
-#[sim_test(check_determinism)]
+#[sim_test]
 async fn test_hash_collections() {
     telemetry_subscribers::init_for_testing();
 
@@ -123,7 +123,7 @@ async fn test_hash_collections() {
 
 // Test that starting up a network + fullnode, and sending one transaction through that network is
 // repeatable and deterministic.
-#[sim_test(check_determinism)]
+#[sim_test]
 async fn test_net_determinism() {
     let mut test_cluster = TestClusterBuilder::new().build().await;