[forge] Latency baseline with faulty validators at 4k TPS

Experiment branch for testing leader reputation improvements.

- Last 2 validators (by ordered index) skip proposals 10% of the time
  (both regular and opt proposals on round % 10 == 0)
- realistic_env_max_load at ConstTps 4k, no fullnodes, no epoch change
- 15 min duration override (bypasses CI default of 480s)
- Add duration_override field to ForgeConfig for test-controlled duration

Not for merge to main.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

Author: danielxiangzl

.github/workflows/docker-build-test.yaml

@@ -270,7 +270,7 @@ jobs:
       GIT_SHA: ${{ needs.determine-docker-build-metadata.outputs.gitSha }}
       FORGE_TEST_SUITE: realistic_env_max_load
       IMAGE_TAG: ${{ needs.determine-docker-build-metadata.outputs.gitSha }}
-      FORGE_RUNNER_DURATION_SECS: 480
+      FORGE_RUNNER_DURATION_SECS: 1800
       COMMENT_HEADER: forge-e2e
       # Use the cache ID as the Forge namespace so we can limit Forge test concurrency on k8s, since Forge
       # test lifecycle is separate from that of GHA. This protects us from the case where many Forge tests are triggered

consensus/src/round_manager.rs

@@ -482,8 +482,25 @@ impl RoundManager {
                 .expect("Sending to a self loopback unbounded channel cannot fail");
         }
 
+        // HACK: Simulate proposer failures. The last 2 validators (by ordered index)
+        // skip their proposal 10% of the time (round % 10 == 0).
+        let should_skip_proposal = {
+            let author = self.proposal_generator.author();
+            let ordered = self.epoch_state.verifier.get_ordered_account_addresses();
+            let n = ordered.len();
+            let idx = ordered.iter().position(|a| *a == author);
+            matches!(idx, Some(i) if i >= n - 2) && new_round_event.round % 10 == 0
+        };
+        if should_skip_proposal {
+            warn!(
+                self.new_log(LogEvent::NewRound),
+                "HACK: skipping proposal for round {}", new_round_event.round
+            );
+        }
+
         // If the current proposer is the leading, try to propose a regular block if not opt proposed already
         if is_current_proposer
+            && !should_skip_proposal
             && self
                 .proposal_generator
                 .can_propose_in_round(new_round_event.round)
@@ -1427,9 +1444,19 @@ impl RoundManager {
 
         let parent = parent_vote.vote_data().proposed().clone();
         let opt_proposal_round = parent.round() + 1;
-        if self
-            .proposer_election
-            .is_valid_proposer(self.proposal_generator.author(), opt_proposal_round)
+        // HACK: Skip opt proposal for faulty validators too.
+        let should_skip_opt = {
+            let author = self.proposal_generator.author();
+            let ordered = self.epoch_state.verifier.get_ordered_account_addresses();
+            let n = ordered.len();
+            let idx = ordered.iter().position(|a| *a == author);
+            matches!(idx, Some(i) if i >= n - 2) && opt_proposal_round % 10 == 0
+        };
+
+        if !should_skip_opt
+            && self
+                .proposer_election
+                .is_valid_proposer(self.proposal_generator.author(), opt_proposal_round)
         {
             let expected_grandparent_round = parent
                 .round()

testsuite/forge-cli/src/suites/land_blocking.rs

@@ -17,7 +17,8 @@ pub(crate) fn get_land_blocking_test(
 ) -> Option<ForgeConfig> {
     let test = match test_name {
         "land_blocking" | "realistic_env_max_load" => {
-            realistic_env_max_load_test(duration, test_cmd, 7, 0, 3)
+            realistic_env_max_load_test(duration, test_cmd, 7, 0, 0)
+                .with_duration_override(Duration::from_secs(900))
         },
         "compat" => compat(),
         "framework_upgrade" => framework_upgrade(),

testsuite/forge-cli/src/suites/realistic_environment.rs

@@ -432,30 +432,18 @@ pub(crate) fn realistic_env_max_load_test(
     }
 
     // Create the test
-    let mempool_backlog = if ha_proxy { 28000 } else { 38000 };
     ForgeConfig::default()
         .with_initial_validator_count(NonZeroUsize::new(num_validators).unwrap())
         .with_initial_fullnode_count(num_vfns)
         .add_network_test(wrap_with_realistic_env(num_validators, TwoTrafficsTest {
             inner_traffic: EmitJobRequest::default()
-                .mode(EmitJobMode::MaxLoad { mempool_backlog })
+                .mode(EmitJobMode::ConstTps { tps: 4000 })
                 .init_gas_price_multiplier(20),
-            inner_success_criteria: SuccessCriteria::new(
-                if ha_proxy {
-                    7000
-                } else if long_running {
-                    // This is for forge stable
-                    11000
-                } else {
-                    // During land time we want to be less strict, otherwise we flaky fail
-                    10000
-                },
-            ),
+            inner_success_criteria: SuccessCriteria::new(3500),
         }))
         .with_genesis_helm_config_fn(Arc::new(move |helm_values| {
-            // Have single epoch change in land blocking, and a few on long-running
-            helm_values["chain"]["epoch_duration_secs"] =
-                (if long_running { 600 } else { 300 }).into();
+            // No epoch change so measurements are stable.
+            helm_values["chain"]["epoch_duration_secs"] = (24 * 3600).into();
             helm_values["chain"]["on_chain_consensus_config"] =
                 serde_yaml::to_value(OnChainConsensusConfig::default_for_genesis())
                     .expect("must serialize");

testsuite/forge/src/config.rs

@@ -10,7 +10,7 @@ use aptos_config::config::{
     OverrideNodeConfig,
 };
 use aptos_framework::ReleaseBundle;
-use std::{num::NonZeroUsize, sync::Arc};
+use std::{num::NonZeroUsize, sync::Arc, time::Duration};
 
 /// A PFN deployment configuration. Each entry maps to one helm release via the PFN deployer.
 pub struct PfnDeployment {
@@ -72,6 +72,9 @@ pub struct ForgeConfig {
     /// Retain debug logs and above for all nodes instead of just the first 5 nodes
     pub retain_debug_logs: bool,
 
+    /// Override the CLI-specified test duration
+    pub duration_override: Option<Duration>,
+
     /// URL to download the trusted setup blob for chunky DKG into the validator init-container
     pub decryption_setup_blob_url: Option<String>,
 }
@@ -199,6 +202,11 @@ impl ForgeConfig {
         self
     }
 
+    pub fn with_duration_override(mut self, duration: Duration) -> Self {
+        self.duration_override = Some(duration);
+        self
+    }
+
     pub fn with_multi_region_config(mut self) -> Self {
         self.multi_region_config = true;
         self
@@ -435,6 +443,7 @@ impl Default for ForgeConfig {
             fullnode_resource_override: NodeResourceOverride::default(),
             retain_debug_logs: false,
             decryption_setup_blob_url: None,
+            duration_override: None,
         }
     }
 }

testsuite/forge/src/runner.rs

@@ -311,7 +311,7 @@ impl<'cfg, F: Factory> Forge<'cfg, F> {
                 &initial_version,
                 &genesis_version,
                 self.tests.genesis_config.as_ref(),
-                self.global_duration + Duration::from_secs(NAMESPACE_CLEANUP_DURATION_BUFFER_SECS),
+                self.tests.duration_override.unwrap_or(self.global_duration) + Duration::from_secs(NAMESPACE_CLEANUP_DURATION_BUFFER_SECS),
                 self.tests.genesis_helm_config_fn.clone(),
                 self.tests.build_node_helm_config_fn(retain_debug_logs),
                 self.tests.existing_db_tag.clone(),
@@ -343,12 +343,16 @@ impl<'cfg, F: Factory> Forge<'cfg, F> {
 
             let logs_location = swarm.logs_location();
             let swarm = Arc::new(tokio::sync::RwLock::new(swarm));
+            let effective_duration = self
+                .tests
+                .duration_override
+                .unwrap_or(self.global_duration);
             for test in self.filter_tests(&self.tests.network_tests) {
                 let network_ctx = NetworkContext::new(
                     CoreContext::from_rng(&mut rng),
                     swarm.clone(),
                     &mut report,
-                    self.global_duration,
+                    effective_duration,
                     self.tests.emit_job_request.clone(),
                     self.tests.success_criteria.clone(),
                 );