Migrate Narwhal failpoints to simtest, add more (MystenLabs#10458)

## Description 

- Adds failpoints for all storage operations and inbound RPCs.
- Adds delays in Certifier.

## Test Plan 

Is tests.

crates/sui-benchmark/tests/simtest.rs

@@ -4,7 +4,7 @@
 #[cfg(msim)]
 mod test {
 
-    use rand::{thread_rng, Rng};
+    use rand::{distributions::uniform::SampleRange, thread_rng, Rng};
     use std::str::FromStr;
     use std::sync::{Arc, Mutex};
     use std::time::{Duration, Instant};
@@ -137,8 +137,26 @@ mod test {
         }
     }
 
+    async fn delay_failpoint<R>(range_ms: R, probability: f64)
+    where
+        R: SampleRange<u64>,
+    {
+        let duration = {
+            let mut rng = thread_rng();
+            if rng.gen_range(0.0..1.0) < probability {
+                info!("Matched probability threshold for delay failpoint. Delaying...");
+                Some(Duration::from_millis(rng.gen_range(range_ms)))
+            } else {
+                None
+            }
+        };
+        if let Some(duration) = duration {
+            tokio::time::sleep(duration).await;
+        }
+    }
+
     #[sim_test(config = "test_config()")]
-    async fn test_simulated_load_reconfig_crashes() {
+    async fn test_simulated_load_reconfig_with_crashes_and_delays() {
         sui_protocol_config::ProtocolConfig::poison_get_for_min_version();
         let test_cluster = build_test_cluster(4, 1000).await;
 
@@ -169,6 +187,21 @@ mod test {
                 handle_failpoint(dead_validator.clone(), client_node, 0.01);
             }
         });
+
+        // Narwhal fail points.
+        let dead_validator = dead_validator_orig.clone();
+        register_fail_points(
+            &[
+                "narwhal-rpc-response",
+                "narwhal-store-before-write",
+                "narwhal-store-after-write",
+            ],
+            move || {
+                handle_failpoint(dead_validator.clone(), client_node, 0.001);
+            },
+        );
+        register_fail_point_async("narwhal-delay", || delay_failpoint(10..20, 0.001));
+
         test_simulated_load(test_cluster, 120).await;
     }
 

crates/workspace-hack/Cargo.toml

@@ -204,8 +204,7 @@ ethnum = { version = "1", default-features = false }
 event-listener = { version = "2", default-features = false }
 expect-test = { version = "1", default-features = false }
 eyre = { version = "0.6" }
-fail-9fbad63c4bcf4a8f = { package = "fail", version = "0.4", default-features = false }
-fail-d8f496e17d97b5cb = { package = "fail", version = "0.5", default-features = false }
+fail = { version = "0.4", default-features = false }
 fallible-iterator = { version = "0.2" }
 fastcrypto = { git = "https://github.com/MystenLabs/fastcrypto", rev = "84fd7c7428c5f59185aecc56a2e0a006e8e07de1", features = ["copy_key"] }
 fastcrypto-zkp = { git = "https://github.com/MystenLabs/fastcrypto", rev = "84fd7c7428c5f59185aecc56a2e0a006e8e07de1", default-features = false }
@@ -896,8 +895,7 @@ ethnum = { version = "1", default-features = false }
 event-listener = { version = "2", default-features = false }
 expect-test = { version = "1", default-features = false }
 eyre = { version = "0.6" }
-fail-9fbad63c4bcf4a8f = { package = "fail", version = "0.4", default-features = false }
-fail-d8f496e17d97b5cb = { package = "fail", version = "0.5", default-features = false }
+fail = { version = "0.4", default-features = false }
 fallible-iterator = { version = "0.2" }
 fastcrypto = { git = "https://github.com/MystenLabs/fastcrypto", rev = "84fd7c7428c5f59185aecc56a2e0a006e8e07de1", features = ["copy_key"] }
 fastcrypto-derive = { git = "https://github.com/MystenLabs/fastcrypto", rev = "84fd7c7428c5f59185aecc56a2e0a006e8e07de1", default-features = false }

narwhal/benchmark/README.md

@@ -348,16 +348,3 @@ plot_params = {
 The first graph ('latency') plots the latency versus the throughput. It shows that the latency is low until a fairly neat threshold after which it drastically increases. Determining this threshold is crucial to understanding the limits of the system.
 
 Another challenge is comparing apples-to-apples between different deployments of the system. The challenge here is again that latency and throughput are interdependent, as a result a throughput/number of nodes chart could be tricky to produce fairly. The way to do it is to define a maximum latency and measure the throughput at this point instead of simply pushing every system to its peak throughput (where latency is meaningless). The second graph ('tps') plots the maximum achievable throughput under a maximum latency for different numbers of nodes.
-
-## Benchmarks with failpoints
-
-[Fail points](https://docs.rs/failpoints/latest/failpoints/) are code instrumentations that allow errors and other behavior to be injected dynamically at runtime, primarily for testing purposes. Fail points are flexible and can be configured to exhibit a variety of behavior, including panics, early returns, and sleeping. They can be controlled both programmatically and via the environment, and can be triggered conditionally and probabilistically.
-
-Search the code base for `fail_point!()` macro to see the list instrumented failpoints.
-
-Example usage
-
-```
-$ FAILPOINTS='rpc-response-delay=5%sleep(10000);request-batch=5%return;report-our-batch=5%return;request-vote=5%return;certificate-store-panic=.01%return;certificate-store=5%return' \
-  fab failpoints
-```

narwhal/network/Cargo.toml

@@ -16,6 +16,7 @@ parking_lot = "0.12.1"
 quinn-proto = "^0.9.2"
 prometheus = "0.13.3"
 rand = { version = "0.8.5", features = ["small_rng"] }
+sui-macros = { path = "../../crates/sui-macros" }
 tokio = { workspace = true, features = ["rt", "net", "sync", "macros", "time"] }
 tracing = "0.1.36"
 types = { path = "../types", package = "narwhal-types" }
@@ -31,7 +32,6 @@ anyhow = "1.0.65"
 axum.workspace = true
 axum-server = "0.4.2"
 tower = "0.4.13"
-fail = "0.5.1"
 
 [dev-dependencies]
 bincode = "1.3.3"

narwhal/network/src/failpoints.rs

@@ -1,7 +1,7 @@
 // Copyright (c) Mysten Labs, Inc.
 // SPDX-License-Identifier: Apache-2.0
 use anemo_tower::callback::{MakeCallbackHandler, ResponseHandler};
-use fail::fail_point;
+use sui_macros::fail_point;
 
 #[derive(Clone, Default)]
 pub struct FailpointsMakeCallbackHandler {}
@@ -24,11 +24,7 @@ pub struct FailpointsResponseHandler {}
 
 impl ResponseHandler for FailpointsResponseHandler {
     fn on_response(self, _response: &anemo::Response<bytes::Bytes>) {
-        // TODO: Use tokio::sleep() instead of built in sleep()/delay()
-        // Warning: if this failpoint is used with the default sleep()
-        // or delay() it could end up blocking the system and causing other
-        // unintended effects.
-        fail_point!("rpc-response-delay");
+        fail_point!("narwhal-rpc-response");
     }
 
     fn on_error<E>(self, _error: &E) {}

narwhal/network/src/p2p.rs

@@ -216,12 +216,6 @@ impl WorkerRpc for anemo::Network {
 
         let peer_id = PeerId(peer.0.to_bytes());
 
-        fail::fail_point!("request-batch", |_| {
-            Err(format_err!(
-                "Injected error in request batch from peer {peer_id}"
-            ))
-        });
-
         let peer = self
             .peer(peer_id)
             .ok_or_else(|| format_err!("Network has no connection with peer {peer_id}"))?;

narwhal/node/Cargo.toml

@@ -44,7 +44,6 @@ workspace-hack = { version = "0.1", path = "../../crates/workspace-hack" }
 
 anemo.workspace = true
 reqwest = { version = "0.11.13", default_features= false, features = ["json", "rustls-tls"] }
-fail = "0.5.1"
 
 [dev-dependencies]
 pretty_assertions = "1.3.0"

narwhal/node/src/main.rs

@@ -248,15 +248,6 @@ async fn run(
     worker_keypair: NetworkKeyPair,
     registry: Registry,
 ) -> Result<(), eyre::Report> {
-    // Only enabled if failpoints feature flag is set
-    let _failpoints_scenario: fail::FailScenario<'_>;
-    if fail::has_failpoints() {
-        warn!("Failpoints are enabled");
-        _failpoints_scenario = fail::FailScenario::setup();
-    } else {
-        info!("Failpoints are not enabled");
-    }
-
     let workers_file = matches.value_of("workers").unwrap();
     let parameters_file = matches.value_of("parameters");
     let store_path = matches.value_of("store").unwrap();

narwhal/primary/Cargo.toml

@@ -38,6 +38,7 @@ network = { path = "../network", package = "narwhal-network" }
 types = { path = "../types", package = "narwhal-types" }
 storage = { path = "../storage", package = "narwhal-storage" }
 store = { path = "../../crates/typed-store", package = "typed-store" }
+sui-macros = { path = "../../crates/sui-macros" }
 mysten-network.workspace = true
 workspace-hack = { version = "0.1", path = "../../crates/workspace-hack" }
 
@@ -46,7 +47,6 @@ mysten-metrics = { path = "../../crates/mysten-metrics" }
 
 anemo.workspace = true
 anemo-tower.workspace = true
-fail = "0.5.1"
 
 [dev-dependencies]
 dashmap = "5.4.0"

narwhal/primary/src/certifier.rs

@@ -13,6 +13,7 @@ use network::anemo_ext::NetworkExt;
 use std::sync::Arc;
 use std::time::Duration;
 use storage::{CertificateStore, HeaderStore};
+use sui_macros::fail_point_async;
 use tokio::{
     sync::oneshot,
     task::{JoinHandle, JoinSet},
@@ -129,12 +130,6 @@ impl Certifier {
         let peer_id = anemo::PeerId(target.0.to_bytes());
         let peer = network.waiting_peer(peer_id);
 
-        fail::fail_point!("request-vote", |_| {
-            Err(DagError::NetworkError(format!(
-                "Injected error in request vote for {header}"
-            )))
-        });
-
         let mut client = PrimaryToPrimaryClient::new(peer);
 
         let mut missing_parents: Vec<CertificateDigest> = Vec::new();
@@ -388,6 +383,7 @@ impl Certifier {
                     let signature_service = self.signature_service.clone();
                     let metrics = self.metrics.clone();
                     let network = self.network.clone();
+                    fail_point_async!("narwhal-delay");
                     self.propose_header_tasks.spawn(monitored_future!(Self::propose_header(
                         name,
                         committee,
@@ -407,6 +403,7 @@ impl Certifier {
                 Some(result) = self.propose_header_tasks.join_next() => {
                     match result {
                         Ok(Ok(certificate)) => {
+                            fail_point_async!("narwhal-delay");
                             self.synchronizer.accept_own_certificate(certificate).await
                         },
                         Ok(Err(e)) => Err(e),

narwhal/primary/src/primary.rs

@@ -1122,13 +1122,6 @@ impl WorkerToPrimary for WorkerReceiverHandler {
     ) -> Result<anemo::Response<()>, anemo::rpc::Status> {
         let message = request.into_body();
 
-        fail::fail_point!("report-our-batch", |_| {
-            Err(anemo::rpc::Status::internal(format!(
-                "Injected error in report our batch from worker_id {}",
-                message.worker_id
-            )))
-        });
-
         let (tx_ack, rx_ack) = oneshot::channel();
         let response = self
             .tx_our_digests

narwhal/storage/Cargo.toml

@@ -18,10 +18,10 @@ tracing = "0.1.36"
 crypto = { path = "../crypto", package = "narwhal-crypto" }
 types = { path = "../types", package = "narwhal-types" }
 store = { path = "../../crates/typed-store", package = "typed-store" }
+sui-macros = { path = "../../crates/sui-macros" }
 config = { path = "../config", package = "narwhal-config" }
 workspace-hack = { version = "0.1", path = "../../crates/workspace-hack" }
 prometheus = "0.13.3"
-fail = "0.5.1"
 lru = "0.10"
 parking_lot = "0.12.1"
 tap = "1.0.1"