test(continuous): per-instance trigger-miss hook

crates/op-rbuilder-test-macros/src/lib.rs

@@ -3,16 +3,18 @@ use quote::{ToTokens, quote};
 use syn::{Expr, ItemFn, Meta, Token, parse_macro_input, punctuated::Punctuated};
 
 struct TestConfig {
-    args: Option<Expr>,   // Expression to pass to LocalInstance::new()
-    config: Option<Expr>, // NodeConfig<OpChainSpec> for new_with_config
-    multi_threaded: bool, // Whether to use multi_thread flavor
+    args: Option<Expr>,       // Expression to pass to LocalInstance::new()
+    config: Option<Expr>,     // NodeConfig<OpChainSpec> for new_with_config
+    test_hooks: Option<Expr>, // ContinuousTestHooks injected into the builder config
+    multi_threaded: bool,     // Whether to use multi_thread flavor
 }
 
 impl syn::parse::Parse for TestConfig {
     fn parse(input: syn::parse::ParseStream) -> syn::Result<Self> {
         let mut config = TestConfig {
             args: None,
             config: None,
+            test_hooks: None,
             multi_threaded: false,
         };
 
@@ -33,7 +35,7 @@ impl syn::parse::Parse for TestConfig {
                             return Err(syn::Error::new_spanned(
                                 path,
                                 format!(
-                                    "Unknown attribute '{}'. Use 'multi_threaded', 'args', or 'config'",
+                                    "Unknown attribute '{}'. Use 'multi_threaded', 'args', 'config', or 'test_hooks'",
                                     name
                                 ),
                             ));
@@ -47,11 +49,13 @@ impl syn::parse::Parse for TestConfig {
                             config.args = Some(nv.value);
                         } else if name == "config" {
                             config.config = Some(nv.value);
+                        } else if name == "test_hooks" {
+                            config.test_hooks = Some(nv.value);
                         } else {
                             return Err(syn::Error::new_spanned(
                                 nv.path,
                                 format!(
-                                    "Unknown attribute '{}'. Use 'multi_threaded', 'args', or 'config'",
+                                    "Unknown attribute '{}'. Use 'multi_threaded', 'args', 'config', or 'test_hooks'",
                                     name
                                 ),
                             ));
@@ -61,7 +65,7 @@ impl syn::parse::Parse for TestConfig {
                 _ => {
                     return Err(syn::Error::new_spanned(
                         arg,
-                        "Invalid attribute format. Use 'multi_threaded', 'args', or 'config'",
+                        "Invalid attribute format. Use 'multi_threaded', 'args', 'config', or 'test_hooks'",
                     ));
                 }
             }
@@ -71,44 +75,49 @@ impl syn::parse::Parse for TestConfig {
     }
 }
 
-fn generate_instance_init(args: &Option<Expr>, config: &Option<Expr>) -> proc_macro2::TokenStream {
-    let default_args = quote! {
-        {
-            let mut args = crate::args::OpRbuilderArgs::default();
-            args.flashblocks.flashblocks_port = crate::tests::get_available_port();
-            args.flashblocks.flashblocks_end_buffer_ms = 75;
-            args
-        }
-    };
-
-    let modify_args = |args_expr: &proc_macro2::TokenStream| {
-        quote! {
+fn generate_instance_init(
+    args: &Option<Expr>,
+    config: &Option<Expr>,
+    test_hooks: &Option<Expr>,
+) -> proc_macro2::TokenStream {
+    let args_init = match args {
+        None => quote! {
+            {
+                let mut args = crate::args::OpRbuilderArgs::default();
+                args.flashblocks.flashblocks_port = crate::tests::get_available_port();
+                args.flashblocks.flashblocks_end_buffer_ms = 75;
+                args
+            }
+        },
+        Some(args_expr) => quote! {
             {
                 let mut args = #args_expr;
                 args.flashblocks.flashblocks_port = crate::tests::get_available_port();
                 args.flashblocks.flashblocks_end_buffer_ms = 75;
                 args
             }
-        }
+        },
     };
 
-    match (args, config) {
+    match (config, test_hooks) {
         (None, None) => {
-            quote! { crate::tests::LocalInstance::new(#default_args).await? }
+            quote! { crate::tests::LocalInstance::new(#args_init).await? }
         }
-        (Some(args_expr), None) => {
-            let modified_args = modify_args(&quote! { #args_expr });
-            quote! { crate::tests::LocalInstance::new(#modified_args).await? }
+        (Some(config_expr), None) => {
+            quote! {
+                crate::tests::LocalInstance::new_with_config(#args_init, #config_expr).await?
+            }
         }
-        (None, Some(config_expr)) => {
+        (None, Some(hooks_expr)) => {
             quote! {
-                crate::tests::LocalInstance::new_with_config(#default_args, #config_expr).await?
+                crate::tests::LocalInstance::new_with_test_hooks(#args_init, #hooks_expr).await?
             }
         }
-        (Some(args_expr), Some(config_expr)) => {
-            let modified_args = modify_args(&quote! { #args_expr });
+        (Some(config_expr), Some(hooks_expr)) => {
             quote! {
-                crate::tests::LocalInstance::new_with_config(#modified_args, #config_expr).await?
+                crate::tests::LocalInstance::new_with_config_and_hooks(
+                    #args_init, #config_expr, #hooks_expr
+                ).await?
             }
         }
     }
@@ -122,7 +131,7 @@ pub fn rb_test(args: TokenStream, input: TokenStream) -> TokenStream {
     validate_signature(&input_fn);
 
     let test_name = &input_fn.sig.ident;
-    let instance_init = generate_instance_init(&config.args, &config.config);
+    let instance_init = generate_instance_init(&config.args, &config.config, &config.test_hooks);
 
     let test_attribute = if config.multi_threaded {
         quote! { #[tokio::test(flavor = "multi_thread")] }

crates/op-rbuilder/src/builder/continuous/interval.rs

@@ -13,6 +13,8 @@ use crate::{
 use alloy_primitives::B256;
 use reth_node_api::PayloadBuilderError;
 use reth_revm::{State, database::StateProviderDatabase};
+#[cfg(test)]
+use std::sync::Arc;
 use std::{ops::ControlFlow, time::Instant};
 use tokio::sync::{mpsc, oneshot};
 use tracing::{debug, field, metadata::Level, span};
@@ -123,7 +125,10 @@ where
             )
         };
 
-        let candidate_slot = SharedBest::new();
+        let candidate_slot = SharedBest::new(
+            #[cfg(test)]
+            Arc::clone(&self.force_take_miss_counter),
+        );
         let base_ctx = base_state.ctx.clone();
         let base_fb_state = base_state.fb_state.clone();
         let base_info = base_state.info.clone();

crates/op-rbuilder/src/builder/continuous/mod.rs

@@ -14,7 +14,16 @@ mod shared_best;
 mod transition;
 mod types;
 
+/// Test-only knobs for continuous build mode.
+///
+/// Threaded through [`BuilderConfig`](crate::builder::BuilderConfig) because the
+/// builder is constructed inside the node and is otherwise unreachable from a
+/// test. Always default (empty) outside `#[cfg(test)]`.
 #[cfg(test)]
-pub(crate) mod test_hooks {
-    pub(crate) use super::shared_best::test_hooks::force_next_take_misses;
+#[derive(Debug, Clone, Default)]
+pub struct ContinuousTestHooks {
+    /// Number of `SharedBest::take()` calls to force to miss, consumed once
+    /// across this builder's `SharedBest` instances. Deterministically
+    /// exercises the trigger-miss fallback path.
+    pub force_take_miss_count: u64,
 }

crates/op-rbuilder/src/builder/continuous/shared_best.rs

@@ -1,4 +1,6 @@
 use super::types::BestCandidate;
+#[cfg(test)]
+use std::sync::atomic::{AtomicU64, Ordering};
 use std::sync::{Arc, Mutex, MutexGuard};
 use tracing::warn;
 
@@ -59,60 +61,46 @@ impl<T: CandidateCounters> CandidateSlot<T> {
 /// build task. The build task writes on each improvement; the main loop takes
 /// on trigger to publish without awaiting task completion.
 #[derive(Clone)]
-pub(super) struct SharedBest(CandidateSlot<BestCandidate>);
+pub(super) struct SharedBest {
+    slot: CandidateSlot<BestCandidate>,
+    #[cfg(test)]
+    force_take_miss: Arc<AtomicU64>,
+}
 
 impl SharedBest {
-    pub(super) fn new() -> Self {
-        Self(CandidateSlot::new())
+    pub(super) fn new(#[cfg(test)] force_take_miss: Arc<AtomicU64>) -> Self {
+        Self {
+            slot: CandidateSlot::new(),
+            #[cfg(test)]
+            force_take_miss,
+        }
     }
 
     /// Take the current candidate (if any).
     pub(super) fn take(&self) -> Option<BestCandidate> {
         #[cfg(test)]
-        if test_hooks::should_force_take_miss() {
+        if self
+            .force_take_miss
+            .fetch_update(Ordering::AcqRel, Ordering::Acquire, |count| {
+                count.checked_sub(1)
+            })
+            .is_ok()
+        {
             return None;
         }
-        self.0.take()
+        self.slot.take()
     }
 
     pub(super) fn store(&self, candidate: BestCandidate) {
-        self.0.store(candidate);
+        self.slot.store(candidate);
     }
 
     pub(super) fn refresh_metrics(&self, candidates_evaluated: u64, candidates_improved: u64) {
-        self.0
+        self.slot
             .refresh_metrics(candidates_evaluated, candidates_improved);
     }
 }
 
-#[cfg(test)]
-pub(crate) mod test_hooks {
-    use std::sync::atomic::{AtomicU64, Ordering};
-
-    static FORCE_TAKE_MISS_COUNT: AtomicU64 = AtomicU64::new(0);
-
-    pub(crate) struct ForceTakeMissGuard;
-
-    impl Drop for ForceTakeMissGuard {
-        fn drop(&mut self) {
-            FORCE_TAKE_MISS_COUNT.store(0, Ordering::Release);
-        }
-    }
-
-    pub(crate) fn force_next_take_misses(count: u64) -> ForceTakeMissGuard {
-        FORCE_TAKE_MISS_COUNT.store(count, Ordering::Release);
-        ForceTakeMissGuard
-    }
-
-    pub(super) fn should_force_take_miss() -> bool {
-        FORCE_TAKE_MISS_COUNT
-            .fetch_update(Ordering::AcqRel, Ordering::Acquire, |count| {
-                count.checked_sub(1)
-            })
-            .is_ok()
-    }
-}
-
 #[cfg(test)]
 mod tests {
     use super::*;

crates/op-rbuilder/src/builder/mod.rs

@@ -37,7 +37,7 @@ pub use service::FlashblocksServiceBuilder;
 pub use state_root::StateRootCalculator;
 
 #[cfg(test)]
-pub(crate) use continuous::test_hooks as continuous_test_hooks;
+pub use continuous::ContinuousTestHooks;
 
 /// Configuration values that are applicable to any type of block builder.
 #[derive(Debug, Clone)]
@@ -109,6 +109,10 @@ pub struct BuilderConfig {
 
     /// Enable transaction tracking logs
     pub enable_tx_tracking_debug_logs: bool,
+
+    /// Test-only continuous-build hooks. Always default (empty) outside tests.
+    #[cfg(test)]
+    pub continuous_test_hooks: ContinuousTestHooks,
 }
 
 impl Default for BuilderConfig {
@@ -130,6 +134,8 @@ impl Default for BuilderConfig {
             flashblocks_config: FlashblocksConfig::default(),
             exclude_reverts_between_flashblocks: false,
             enable_tx_tracking_debug_logs: false,
+            #[cfg(test)]
+            continuous_test_hooks: ContinuousTestHooks::default(),
         }
     }
 }
@@ -157,6 +163,8 @@ impl TryFrom<OpRbuilderArgs> for BuilderConfig {
             flashblocks_config,
             exclude_reverts_between_flashblocks: args.exclude_reverts_between_flashblocks,
             enable_tx_tracking_debug_logs: false,
+            #[cfg(test)]
+            continuous_test_hooks: ContinuousTestHooks::default(),
         })
     }
 }

crates/op-rbuilder/src/builder/payload.rs

@@ -368,6 +368,9 @@ pub(crate) struct OpPayloadBuilderInner<Pool, Client, BuilderTx> {
     pool_change_epoch: Arc<AtomicU64>,
     /// Task executor used to offload blocking work.
     executor: Runtime,
+    /// Per-builder counter of remaining forced `SharedBest::take()` misses.
+    #[cfg(test)]
+    pub(crate) force_take_miss_counter: Arc<AtomicU64>,
 }
 
 impl<Pool, Client, BuilderTx> OpPayloadBuilderInner<Pool, Client, BuilderTx> {
@@ -458,6 +461,11 @@ where
             disable_state_root: config.flashblocks_config.disable_state_root,
             enable_incremental_state_root: config.flashblocks_config.enable_incremental_state_root,
         });
+        #[cfg(test)]
+        let force_take_miss_counter = Arc::new(AtomicU64::new(
+            config.continuous_test_hooks.force_take_miss_count,
+        ));
+
         Self {
             inner: Arc::new(OpPayloadBuilderInner {
                 builder_ctx,
@@ -471,6 +479,8 @@ where
                 task_metrics,
                 pool_change_epoch,
                 executor,
+                #[cfg(test)]
+                force_take_miss_counter,
             }),
         }
     }

crates/op-rbuilder/src/tests/flashblocks.rs

@@ -9,6 +9,7 @@ use tokio::time::timeout;
 
 use crate::{
     args::{FlashblocksArgs, OpRbuilderArgs},
+    builder::ContinuousTestHooks,
     tests::{
         BlockTransactionsExt, BundleOpts, ChainDriver, FLASHBLOCKS_NUMBER_ADDRESS,
         FlashblocksListener, TransactionBuilderExt, flashblocks_number_contract::FlashblocksNumber,
@@ -375,22 +376,27 @@ async fn smoke_continuous_resolve_after_publish_preserves_published_state(
     flashblocks_listener.stop().await
 }
 
-#[rb_test(args = OpRbuilderArgs {
-    chain_block_time: 1000,
-    enable_revert_protection: true,
-    flashblocks: FlashblocksArgs {
-        flashblocks_port: 1243,
-        flashblocks_addr: "127.0.0.1".into(),
-        flashblocks_block_time: 200,
-        flashblocks_continuous_build: true,
+// `test_hooks` forces the first SharedBest::take() on this builder instance to miss,
+// exercising the trigger-miss fallback path. It is injected into the builder config
+// rather than configured through production args.
+#[rb_test(
+    args = OpRbuilderArgs {
+        chain_block_time: 1000,
+        enable_revert_protection: true,
+        flashblocks: FlashblocksArgs {
+            flashblocks_port: 1243,
+            flashblocks_addr: "127.0.0.1".into(),
+            flashblocks_block_time: 200,
+            flashblocks_continuous_build: true,
+            ..Default::default()
+        },
         ..Default::default()
     },
-    ..Default::default()
-})]
+    test_hooks = ContinuousTestHooks { force_take_miss_count: 1 },
+)]
 async fn smoke_continuous_trigger_miss_fallback_publishes_candidate(
     rbuilder: LocalInstance,
 ) -> eyre::Result<()> {
-    let _force_miss = crate::builder::continuous_test_hooks::force_next_take_misses(1);
     let driver = rbuilder.driver().await?;
     let flashblocks_listener = rbuilder.spawn_flashblocks_listener();