// Copyright (c) Microsoft Corporation.
// Licensed under the MIT License.

#![allow(dead_code, reason = "This is a test module")]
#![allow(missing_docs, reason = "This is a test module")]
#![cfg(all(feature = "hedging", feature = "tower-service"))]

//! Integration tests for hedging middleware using only public API.

use std::future::poll_fn;
use std::sync::Arc;
use std::sync::atomic::{AtomicBool, AtomicU32, Ordering};
use std::time::Duration;

use layered::{Execute, Service, Stack};
use rstest::rstest;
use seatbelt::hedging::{Hedging, OnExecuteArgs};
use seatbelt::{RecoveryInfo, ResilienceContext};
use tick::{Clock, ClockControl};
use tower_service::Service as TowerService;

/// Helper to execute a service either via `layered::Service` or `tower_service::Service`.
async fn execute_service<S, In, Out, Err>(service: &mut S, input: In, use_tower: bool) -> Result<Out, Err>
where
    S: Service<In, Out = Result<Out, Err>> + TowerService<In, Response = Out, Error = Err>,
    S::Future: Send,
    In: Send + 'static,
    Out: Send + 'static,
    Err: Send + 'static,
{
    if use_tower {
        poll_fn(|cx| service.poll_ready(cx)).await?;
        service.call(input).await
    } else {
        service.execute(input).await
    }
}

#[rstest]
#[case::layered(false)]
#[case::tower(true)]
#[tokio::test]
async fn hedging_disabled_passes_through(#[case] use_tower: bool) {
    let clock = Clock::new_frozen();
    let counter = Arc::new(AtomicU32::new(0));
    let counter_clone = Arc::clone(&counter);

    let context: ResilienceContext<String, Result<String, String>> = ResilienceContext::new(&clock).name("test");
    let stack = (
        Hedging::layer("test_hedging", &context)
            .clone_input()
            .recovery_with(|_: &Result<String, String>, _| RecoveryInfo::retry())
            .disable(),
        Execute::new(move |v: String| {
            counter_clone.fetch_add(1, Ordering::SeqCst);
            async move { Ok::<_, String>(v) }
        }),
    );

    let mut service = stack.into_service();
    let result = execute_service(&mut service, "test".to_string(), use_tower).await;

    assert_eq!(result, Ok("test".to_string()));
    assert_eq!(counter.load(Ordering::SeqCst), 1);
}

#[rstest]
#[case::layered(false)]
#[case::tower(true)]
#[tokio::test]
async fn immediate_mode_all_run_concurrently(#[case] use_tower: bool) {
    let clock = ClockControl::default().auto_advance_timers(true).to_clock();
    let counter = Arc::new(AtomicU32::new(0));
    let counter_clone = Arc::clone(&counter);

    let context: ResilienceContext<String, Result<String, String>> = ResilienceContext::new(&clock).name("test");
    let stack = (
        Hedging::layer("test_hedging", &context)
            .clone_input()
            .recovery_with(|result: &Result<String, String>, _| match result {
                Ok(_) => RecoveryInfo::never(),
                Err(_) => RecoveryInfo::retry(),
            })
            .hedging_delay(Duration::ZERO)
            .max_hedged_attempts(2),
        Execute::new(move |v: String| {
            let count = counter_clone.fetch_add(1, Ordering::SeqCst);
            async move {
                if count < 2 {
                    Err::<String, String>(format!("error_{count}"))
                } else {
                    Ok(format!("success:{v}"))
                }
            }
        }),
    );

    let mut service = stack.into_service();
    let result = execute_service(&mut service, "test".to_string(), use_tower).await;

    assert_eq!(result, Ok("success:test".to_string()));
    // All 3 attempts (1 original + 2 hedging attempts) should have been launched
    assert_eq!(counter.load(Ordering::SeqCst), 3);
}

#[rstest]
#[case::layered(false)]
#[case::tower(true)]
#[tokio::test]
async fn immediate_mode_returns_first_success(#[case] use_tower: bool) {
    let clock = ClockControl::default().auto_advance_timers(true).to_clock();
    let counter = Arc::new(AtomicU32::new(0));
    let counter_clone = Arc::clone(&counter);

    let context: ResilienceContext<String, Result<String, String>> = ResilienceContext::new(&clock).name("test");
    let stack = (
        Hedging::layer("test_hedging", &context)
            .clone_input()
            .recovery_with(|result: &Result<String, String>, _| match result {
                Ok(_) => RecoveryInfo::never(),
                Err(_) => RecoveryInfo::retry(),
            })
            .hedging_delay(Duration::ZERO)
            .max_hedged_attempts(2),
        Execute::new(move |v: String| {
            counter_clone.fetch_add(1, Ordering::SeqCst);
            async move { Ok::<_, String>(format!("ok:{v}")) }
        }),
    );

    let mut service = stack.into_service();
    let result = execute_service(&mut service, "test".to_string(), use_tower).await;

    // First success should be returned
    assert_eq!(result, Ok("ok:test".to_string()));
}

#[rstest]
#[case::layered(false)]
#[case::tower(true)]
#[tokio::test]
async fn delay_mode_launches_hedging_attempt_after_timeout(#[case] use_tower: bool) {
    let clock = ClockControl::default().auto_advance_timers(true).to_clock();
    let counter = Arc::new(AtomicU32::new(0));
    let counter_clone = Arc::clone(&counter);

    let context: ResilienceContext<String, Result<String, String>> = ResilienceContext::new(&clock).name("test");
    let stack = (
        Hedging::layer("test_hedging", &context)
            .clone_input()
            .recovery_with(|result: &Result<String, String>, _| match result {
                Ok(_) => RecoveryInfo::never(),
                Err(_) => RecoveryInfo::retry(),
            })
            .hedging_delay(Duration::from_secs(1))
            .max_hedged_attempts(1),
        Execute::new(move |v: String| {
            let count = counter_clone.fetch_add(1, Ordering::SeqCst);
            async move {
                if count == 0 {
                    // First attempt fails
                    Err::<String, String>("transient".to_string())
                } else {
                    Ok(format!("hedged:{v}"))
                }
            }
        }),
    );

    let mut service = stack.into_service();
    let result = execute_service(&mut service, "test".to_string(), use_tower).await;

    assert_eq!(result, Ok("hedged:test".to_string()));
    assert_eq!(counter.load(Ordering::SeqCst), 2);
}

#[rstest]
#[case::layered(false)]
#[case::tower(true)]
#[tokio::test]
async fn dynamic_mode_computes_delay(#[case] use_tower: bool) {
    let clock = ClockControl::default().auto_advance_timers(true).to_clock();
    let counter = Arc::new(AtomicU32::new(0));
    let counter_clone = Arc::clone(&counter);

    let context: ResilienceContext<String, Result<String, String>> = ResilienceContext::new(&clock).name("test");
    let stack = (
        Hedging::layer("test_hedging", &context)
            .clone_input()
            .recovery_with(|result: &Result<String, String>, _| match result {
                Ok(_) => RecoveryInfo::never(),
                Err(_) => RecoveryInfo::retry(),
            })
            .hedging_delay_with(|_input, args| Duration::from_millis(100 * u64::from(args.attempt().index())))
            .max_hedged_attempts(2),
        Execute::new(move |v: String| {
            let count = counter_clone.fetch_add(1, Ordering::SeqCst);
            async move {
                if count < 2 {
                    Err::<String, String>(format!("fail_{count}"))
                } else {
                    Ok(format!("ok:{v}"))
                }
            }
        }),
    );

    let mut service = stack.into_service();
    let result = execute_service(&mut service, "test".to_string(), use_tower).await;

    assert_eq!(result, Ok("ok:test".to_string()));
    assert_eq!(counter.load(Ordering::SeqCst), 3);
}

#[rstest]
#[case::layered(false)]
#[case::tower(true)]
#[tokio::test]
async fn on_execute_callback_invoked(#[case] use_tower: bool) {
    let clock = ClockControl::default().auto_advance_timers(true).to_clock();
    let execute_calls = Arc::new(AtomicU32::new(0));
    let execute_calls_clone = Arc::clone(&execute_calls);

    let context: ResilienceContext<String, Result<String, String>> = ResilienceContext::new(&clock).name("test");
    let stack = (
        Hedging::layer("test_hedging", &context)
            .clone_input()
            .recovery_with(|_: &Result<String, String>, _| RecoveryInfo::retry())
            .hedging_delay(Duration::ZERO)
            .max_hedged_attempts(2)
            .on_execute(move |_input, _: OnExecuteArgs| {
                execute_calls_clone.fetch_add(1, Ordering::SeqCst);
            }),
        Execute::new(|v: String| async move { Ok::<_, String>(v) }),
    );

    let mut service = stack.into_service();
    let _result = execute_service(&mut service, "test".to_string(), use_tower).await;

    // on_execute should be called for each attempt (1 original + 2 hedging attempts)
    assert_eq!(execute_calls.load(Ordering::SeqCst), 3);
}

#[rstest]
#[case::layered(false)]
#[case::tower(true)]
#[tokio::test]
async fn no_hedging_configured_passes_through(#[case] use_tower: bool) {
    let clock = Clock::new_frozen();
    let counter = Arc::new(AtomicU32::new(0));
    let counter_clone = Arc::clone(&counter);

    let context: ResilienceContext<String, Result<String, String>> = ResilienceContext::new(&clock).name("test");
    let stack = (
        Hedging::layer("test_hedging", &context)
            .clone_input()
            .recovery_with(|_: &Result<String, String>, _| RecoveryInfo::retry())
            .max_hedged_attempts(0),
        Execute::new(move |v: String| {
            counter_clone.fetch_add(1, Ordering::SeqCst);
            async move { Ok::<_, String>(v) }
        }),
    );

    let mut service = stack.into_service();
    let result = execute_service(&mut service, "test".to_string(), use_tower).await;

    assert_eq!(result, Ok("test".to_string()));
    assert_eq!(counter.load(Ordering::SeqCst), 1);
}

#[rstest]
#[case::layered(false)]
#[case::tower(true)]
#[tokio::test]
async fn all_fail_returns_last_result(#[case] use_tower: bool) {
    let clock = ClockControl::default().auto_advance_timers(true).to_clock();

    let context: ResilienceContext<String, Result<String, String>> = ResilienceContext::new(&clock).name("test");
    let stack = (
        Hedging::layer("test_hedging", &context)
            .clone_input()
            .recovery_with(|_: &Result<String, String>, _| RecoveryInfo::retry())
            .hedging_delay(Duration::ZERO)
            .max_hedged_attempts(2),
        Execute::new(|_v: String| async move { Err::<String, String>("always_fail".to_string()) }),
    );

    let mut service = stack.into_service();
    let result = execute_service(&mut service, "test".to_string(), use_tower).await;

    assert_eq!(result, Err("always_fail".to_string()));
}

#[rstest]
#[case::layered(false)]
#[case::tower(true)]
#[tokio::test]
async fn clone_service_works_independently(#[case] use_tower: bool) {
    let clock = ClockControl::default().auto_advance_timers(true).to_clock();
    let call_count = Arc::new(AtomicU32::new(0));
    let call_count_clone = Arc::clone(&call_count);

    let context: ResilienceContext<String, Result<String, String>> = ResilienceContext::new(&clock).name("test");
    let stack = (
        Hedging::layer("test_hedging", &context)
            .clone_input()
            .recovery_with(|result: &Result<String, String>, _| match result {
                Ok(_) => RecoveryInfo::never(),
                Err(_) => RecoveryInfo::retry(),
            })
            .hedging_delay(Duration::ZERO)
            .max_hedged_attempts(1),
        Execute::new(move |v: String| {
            call_count_clone.fetch_add(1, Ordering::SeqCst);
            async move { Ok::<_, String>(format!("ok:{v}")) }
        }),
    );

    let mut service = stack.into_service();
    let mut cloned_service = service.clone();

    let result1 = execute_service(&mut service, "original".to_string(), use_tower).await;
    let result2 = execute_service(&mut cloned_service, "cloned".to_string(), use_tower).await;

    assert_eq!(result1, Ok("ok:original".to_string()));
    assert_eq!(result2, Ok("ok:cloned".to_string()));
}

#[rstest]
#[case::layered(false)]
#[case::tower(true)]
#[tokio::test]
async fn enable_if_skips_hedging(#[case] use_tower: bool) {
    let clock = ClockControl::default().auto_advance_timers(true).to_clock();
    let counter = Arc::new(AtomicU32::new(0));
    let counter_clone = Arc::clone(&counter);

    let context: ResilienceContext<String, Result<String, String>> = ResilienceContext::new(&clock).name("test");
    let stack = (
        Hedging::layer("test_hedging", &context)
            .clone_input()
            .recovery_with(|_: &Result<String, String>, _| RecoveryInfo::retry())
            .hedging_delay(Duration::ZERO)
            .max_hedged_attempts(2)
            .enable_if(|input| input.contains("hedging")),
        Execute::new(move |v: String| {
            counter_clone.fetch_add(1, Ordering::SeqCst);
            async move { Ok::<_, String>(v) }
        }),
    );

    let mut service = stack.into_service();

    // This should NOT be hedged (no "hedging" in input)
    let result = execute_service(&mut service, "normal".to_string(), use_tower).await;
    assert_eq!(result, Ok("normal".to_string()));
    assert_eq!(counter.load(Ordering::SeqCst), 1); // Only 1 call, no hedging attempts
}

#[rstest]
#[case::layered(false)]
#[case::tower(true)]
#[tokio::test]
async fn clone_returning_none_skips_hedging(#[case] use_tower: bool) {
    let clock = ClockControl::default().auto_advance_timers(true).to_clock();
    let counter = Arc::new(AtomicU32::new(0));
    let counter_clone = Arc::clone(&counter);

    let context: ResilienceContext<String, Result<String, String>> = ResilienceContext::new(&clock).name("test");
    let stack = (
        Hedging::layer("test_hedging", &context)
            .clone_input_with(|input, args| {
                // Only clone for the original request (attempt 0), refuse all hedging attempts.
                (args.attempt().index() == 0).then(|| input.clone())
            })
            .recovery_with(|result: &Result<String, String>, _| match result {
                Ok(_) => RecoveryInfo::never(),
                Err(_) => RecoveryInfo::retry(),
            })
            .hedging_delay(Duration::from_millis(10))
            .max_hedged_attempts(2),
        Execute::new(move |v: String| {
            let count = counter_clone.fetch_add(1, Ordering::SeqCst);
            async move {
                if count == 0 {
                    Err::<String, String>("transient".to_string())
                } else {
                    Ok(format!("hedged:{v}"))
                }
            }
        }),
    );

    let mut service = stack.into_service();
    let result = execute_service(&mut service, "test".to_string(), use_tower).await;

    // Clone refused all hedging attempts, so only the original (recoverable) result returns.
    assert_eq!(result, Err("transient".to_string()));
    assert_eq!(counter.load(Ordering::SeqCst), 1);
}

#[rstest]
#[case::layered(false)]
#[case::tower(true)]
#[tokio::test]
async fn handle_unavailable_continues_hedging(#[case] use_tower: bool) {
    let clock = ClockControl::default().auto_advance_timers(true).to_clock();
    let counter = Arc::new(AtomicU32::new(0));
    let counter_clone = Arc::clone(&counter);

    let context: ResilienceContext<String, Result<String, String>> = ResilienceContext::new(&clock).name("test");
    let stack = (
        Hedging::layer("test_hedging", &context)
            .clone_input()
            .recovery_with(|result: &Result<String, String>, _| match result {
                Ok(_) => RecoveryInfo::never(),
                Err(e) if e == "unavailable" => RecoveryInfo::unavailable(),
                Err(_) => RecoveryInfo::retry(),
            })
            .hedging_delay(Duration::ZERO)
            .max_hedged_attempts(2)
            .handle_unavailable(true),
        Execute::new(move |v: String| {
            let count = counter_clone.fetch_add(1, Ordering::SeqCst);
            async move {
                if count == 0 {
                    Err::<String, String>("unavailable".to_string())
                } else {
                    Ok(format!("ok:{v}"))
                }
            }
        }),
    );

    let mut service = stack.into_service();
    let result = execute_service(&mut service, "test".to_string(), use_tower).await;

    // With handle_unavailable(true), the "unavailable" error is treated as recoverable,
    // so hedging continues and a successful result is returned.
    assert_eq!(result, Ok("ok:test".to_string()));
}

#[rstest]
#[case::layered(false)]
#[case::tower(true)]
#[tokio::test]
async fn slow_original_accepted_after_hedging_launched(#[case] use_tower: bool) {
    // Scenario: the original request is slow and doesn't finish within the hedging delay.
    // A hedging attempt is launched and both return successful (non-recoverable) results, but the
    // original finishes first and its result is accepted.
    use tokio::sync::Notify;

    let clock = ClockControl::default().auto_advance_timers(true).to_clock();
    let counter = Arc::new(AtomicU32::new(0));
    let counter_clone = Arc::clone(&counter);

    // The original waits for this signal before completing.
    let original_go = Arc::new(Notify::new());
    let original_go_clone = Arc::clone(&original_go);

    // The hedging attempt waits for this signal (never sent) to simulate being slower.
    let hedging_block = Arc::new(Notify::new());

    let context: ResilienceContext<String, Result<String, String>> = ResilienceContext::new(&clock).name("test");
    let stack = (
        Hedging::layer("test_hedging", &context)
            .clone_input()
            .recovery_with(|result: &Result<String, String>, _| match result {
                Ok(_) => RecoveryInfo::never(),
                Err(_) => RecoveryInfo::retry(),
            })
            // Hedging attempt fires after 100ms
            .hedging_delay(Duration::from_millis(100))
            .max_hedged_attempts(1)
            .on_execute(move |_input, args: OnExecuteArgs| {
                // When the hedging attempt is about to launch, let the original complete.
                if args.attempt().index() > 0 {
                    original_go_clone.notify_one();
                }
            }),
        Execute::new(move |v: String| {
            let count = counter_clone.fetch_add(1, Ordering::SeqCst);
            let go = Arc::clone(&original_go);
            let block = Arc::clone(&hedging_block);
            async move {
                if count == 0 {
                    // Original: wait until the hedging attempt is launched, then succeed.
                    go.notified().await;
                    Ok::<_, String>(format!("original:{v}"))
                } else {
                    // Hedging attempt: also succeeds, but is slower than the original.
                    block.notified().await;
                    Ok(format!("hedging:{v}"))
                }
            }
        }),
    );

    let mut service = stack.into_service();
    let result = execute_service(&mut service, "test".to_string(), use_tower).await;

    // The original request wins because it finishes before the hedging attempt.
    assert_eq!(result, Ok("original:test".to_string()));
    // Both attempts were launched (original + 1 hedging attempt).
    assert_eq!(counter.load(Ordering::SeqCst), 2);
}

#[rstest]
#[case::layered(false)]
#[case::tower(true)]
#[tokio::test]
async fn non_cloneable_input_skips_hedging_and_invokes_on_execute(#[case] use_tower: bool) {
    let clock = ClockControl::default().auto_advance_timers(true).to_clock();
    let counter = Arc::new(AtomicU32::new(0));
    let counter_clone = Arc::clone(&counter);
    let execute_calls = Arc::new(AtomicU32::new(0));
    let execute_calls_clone = Arc::clone(&execute_calls);

    let context: ResilienceContext<String, Result<String, String>> = ResilienceContext::new(&clock).name("test");
    let stack = (
        Hedging::layer("test_hedging", &context)
            .clone_input_with(|_input, _args| None::<String>)
            .recovery_with(|_: &Result<String, String>, _| RecoveryInfo::retry())
            .hedging_delay(Duration::ZERO)
            .max_hedged_attempts(2)
            .on_execute(move |_input, _: OnExecuteArgs| {
                execute_calls_clone.fetch_add(1, Ordering::SeqCst);
            }),
        Execute::new(move |v: String| {
            counter_clone.fetch_add(1, Ordering::SeqCst);
            async move { Ok::<_, String>(v) }
        }),
    );

    let mut service = stack.into_service();
    let result = execute_service(&mut service, "test".to_string(), use_tower).await;

    // The original request should succeed without hedging.
    assert_eq!(result, Ok("test".to_string()));
    // Only 1 execution: no hedging attempts because clone returned None.
    assert_eq!(counter.load(Ordering::SeqCst), 1);
    // on_execute must still be called exactly once for the original request.
    assert_eq!(execute_calls.load(Ordering::SeqCst), 1);
}

#[rstest]
#[case::layered(false)]
#[case::tower(true)]
#[tokio::test]
async fn initial_attempt_is_last_when_no_hedging_attempts(#[case] use_tower: bool) {
    let clock = ClockControl::default().auto_advance_timers(true).to_clock();
    let observed_is_last = Arc::new(AtomicBool::new(false));
    let observed_clone = Arc::clone(&observed_is_last);

    let context: ResilienceContext<String, Result<String, String>> = ResilienceContext::new(&clock).name("test");
    let stack = (
        Hedging::layer("test_hedging", &context)
            .clone_input()
            .recovery_with(|_: &Result<String, String>, _| RecoveryInfo::never())
            .max_hedged_attempts(0)
            .on_execute(move |_input, args: OnExecuteArgs| {
                observed_clone.store(args.attempt().is_last(), Ordering::SeqCst);
            }),
        Execute::new(|v: String| async move { Ok::<_, String>(v) }),
    );

    let mut service = stack.into_service();
    let _result = execute_service(&mut service, "test".to_string(), use_tower).await;

    assert!(
        observed_is_last.load(Ordering::SeqCst),
        "initial attempt must have is_last == true when max_hedged_attempts == 0"
    );
}

#[rstest]
#[case::layered(false)]
#[case::tower(true)]
#[tokio::test]
async fn unavailable_returned_immediately_when_handle_unavailable_is_false(#[case] use_tower: bool) {
    let clock = ClockControl::default().auto_advance_timers(true).to_clock();
    let counter = Arc::new(AtomicU32::new(0));
    let counter_clone = Arc::clone(&counter);

    let context: ResilienceContext<String, Result<String, String>> = ResilienceContext::new(&clock).name("test");
    let stack = (
        Hedging::layer("test_hedging", &context)
            .clone_input()
            .recovery_with(|result: &Result<String, String>, _| match result {
                Ok(_) => RecoveryInfo::never(),
                Err(_) => RecoveryInfo::unavailable(),
            })
            .hedging_delay(Duration::ZERO)
            .max_hedged_attempts(2)
            .handle_unavailable(false),
        Execute::new(move |v: String| {
            let count = counter_clone.fetch_add(1, Ordering::SeqCst);
            async move {
                match count {
                    // Attempt 0: returns an error classified as Unavailable.
                    // With handle_unavailable=false this must be accepted immediately.
                    0 => Err::<String, String>("unavailable_error".into()),
                    // Attempt 1+: would succeed if reached.
                    _ => Ok(format!("success:{v}")),
                }
            }
        }),
    );

    let mut service = stack.into_service();
    let result = execute_service(&mut service, "test".to_string(), use_tower).await;

    // The unavailable error must be returned immediately (not recovered).
    assert_eq!(result, Err("unavailable_error".to_string()));
}

#[rstest]
#[case::layered(false)]
#[case::tower(true)]
#[tokio::test]
async fn launch_hedging_attempt_actually_launches_attempts(#[case] use_tower: bool) {
    let clock = ClockControl::default().auto_advance_timers(true).to_clock();
    let counter = Arc::new(AtomicU32::new(0));
    let counter_clone = Arc::clone(&counter);

    let context: ResilienceContext<String, Result<String, String>> = ResilienceContext::new(&clock).name("test");
    let stack = (
        Hedging::layer("test_hedging", &context)
            .clone_input()
            .recovery_with(|result: &Result<String, String>, _| match result {
                Ok(_) => RecoveryInfo::never(),
                Err(_) => RecoveryInfo::retry(),
            })
            .hedging_delay(Duration::ZERO)
            .max_hedged_attempts(1),
        Execute::new(move |v: String| {
            let count = counter_clone.fetch_add(1, Ordering::SeqCst);
            async move {
                match count {
                    // Original attempt fails with a recoverable error.
                    0 => Err::<String, String>("transient".into()),
                    // Hedging attempt succeeds.
                    _ => Ok(format!("hedged:{v}")),
                }
            }
        }),
    );

    let mut service = stack.into_service();
    let result = execute_service(&mut service, "test".to_string(), use_tower).await;

    // If launch_hedging_attempt is a no-op, only the original attempt runs and we
    // get the transient error back. With a real launch, the hedging attempt succeeds.
    assert_eq!(result, Ok("hedged:test".to_string()));
    assert!(counter.load(Ordering::SeqCst) >= 2, "at least 2 attempts must have been launched");
}

#[rstest]
#[case::layered(false)]
#[case::tower(true)]
#[tokio::test]
async fn invoke_on_execute_actually_calls_callback(#[case] use_tower: bool) {
    let clock = ClockControl::default().auto_advance_timers(true).to_clock();
    let execute_calls = Arc::new(AtomicU32::new(0));
    let execute_calls_clone = Arc::clone(&execute_calls);

    let context: ResilienceContext<String, Result<String, String>> = ResilienceContext::new(&clock).name("test");
    let stack = (
        Hedging::layer("test_hedging", &context)
            .clone_input()
            .recovery_with(|_: &Result<String, String>, _| RecoveryInfo::never())
            .max_hedged_attempts(0)
            .on_execute(move |_input, _: OnExecuteArgs| {
                execute_calls_clone.fetch_add(1, Ordering::SeqCst);
            }),
        Execute::new(|v: String| async move { Ok::<_, String>(v) }),
    );

    let mut service = stack.into_service();
    let _result = execute_service(&mut service, "test".to_string(), use_tower).await;

    assert_eq!(
        execute_calls.load(Ordering::SeqCst),
        1,
        "on_execute callback must be invoked at least once"
    );
}
