feat(seatbelt): introduce hedging resilience middleware (#298)

## Hedging Middleware

Adds a **hedging** resilience middleware to `seatbelt` that reduces tail
latency by launching additional concurrent requests when the original is
slow. The first acceptable result wins; remaining in-flight requests are
  cancelled.

   ### How it works

   1. The original request is sent immediately.
2. After a configurable delay, additional hedging attempts launch
concurrently.
3. A user-provided recovery callback classifies each result —
non-recoverable results are returned immediately.

   ### Hedging delay

   | Method | Behavior |
   |--------|----------|
| **`hedging_delay(duration)`** | Waits a fixed duration before each
hedging attempt *(default: 500ms)* |
| **`hedging_delay(Duration::ZERO)`** | Launches all hedging attempts at
once |
| **`hedging_delay_with(fn)`** | Computes delay per attempt via callback
|

   ### Configuration

Uses a **type-state builder** pattern — `clone_input` and `recovery` are
enforced at compile time before the layer can be constructed.

   ```rust
   let stack = (
       Hedging::layer("my_hedging", &context)
           .clone_input()
           .recovery_with(|result, _| match result {
               Ok(_) => RecoveryInfo::never(),
               Err(_) => RecoveryInfo::retry(),
           })
           .hedging_delay(Duration::from_secs(1))
           .max_hedged_attempts(2),
       Execute::new(my_operation),
   );

Runtime configuration is also supported via HedgingConfig
(serde-compatible):

   .config(&HedgingConfig {
       enabled: true,
       hedging_delay: Duration::from_millis(500),
       max_hedged_attempts: 1,
   })
```

---------

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
Co-authored-by: Copilot <198982749+Copilot@users.noreply.github.com>

Author: 3 people

.github/workflows/main.yml

@@ -74,7 +74,7 @@ jobs:
         run: cargo sort --check --grouped --workspace
       - name: Readmes
         if: success() || failure()
-        run: cargo workspaces exec --ignore-private cargo doc2readme --expand-macros --all-features --check --lib --template ../README.j2
+        run: cargo workspaces exec --ignore-private cargo doc2readme --check --lib --template ../README.j2
         env:
           # Disable sccache: macro expansion (rustc -Zunpretty=expanded) does not
           # produce the artifact files sccache expects to cache, causing it to fail.

.spelling

@@ -1,4 +1,4 @@
-306
+307
 →
 0.X.Y
 100k
@@ -125,6 +125,7 @@ grey
 gRPC
 Hardcoded
 hashers
+hedges
 hotfix
 hotfixes
 How-tos

Cargo.lock

@@ -43,13 +43,15 @@ timeout = []
 retry = ["dep:fastrand"]
 breaker = ["dep:fastrand"]
 fallback = []
+hedging = ["dep:futures-util"]
 serde = ["dep:serde", "dep:jiff"]
 metrics = ["dep:opentelemetry", "opentelemetry/metrics"]
 logs = ["dep:tracing"]
 tower-service = ["dep:tower-service"]
 
 [dependencies]
 fastrand = { workspace = true, optional = true }
+futures-util = { workspace = true, optional = true, features = ["alloc"] }
 jiff = { workspace = true, features = ["std", "serde"], optional = true }
 layered = { workspace = true }
 opentelemetry = { workspace = true, optional = true }
@@ -65,10 +67,11 @@ alloc_tracker.workspace = true
 criterion.workspace = true
 fastrand.workspace = true
 futures = { workspace = true, features = ["executor"] }
+futures-util = { workspace = true, features = ["alloc"] }
 http.workspace = true
 insta = { workspace = true, features = ["json"] }
 jiff = { workspace = true, default-features = true, features = ["serde"] }
-layered = { workspace = true, features = ["tower-service"] }
+layered = { workspace = true, features = ["tower-service", "dynamic-service"] }
 mutants.workspace = true
 ohno = { workspace = true, features = ["app-err"] }
 opentelemetry = { workspace = true, default-features = false, features = ["metrics"] }
@@ -113,6 +116,10 @@ required-features = ["retry", "timeout", "metrics"]
 name = "tower"
 required-features = ["retry", "timeout", "tower-service"]
 
+[[example]]
+name = "hedging"
+required-features = ["hedging"]
+
 [[example]]
 name = "breaker"
 required-features = ["breaker", "metrics"]
@@ -145,5 +152,10 @@ name = "breaker"
 harness = false
 required-features = ["breaker"]
 
+[[bench]]
+name = "hedging"
+harness = false
+required-features = ["hedging"]
+
 [lints]
 workspace = true

crates/seatbelt/Cargo.toml

@@ -89,8 +89,9 @@ for each module for details on how to use them.
 
 * [`timeout`][__link8] - Middleware that cancels long-running operations.
 * [`retry`][__link9] - Middleware that automatically retries failed operations.
-* [`breaker`][__link10] - Middleware that prevents cascading failures.
-* [`fallback`][__link11] - Middleware that replaces invalid output with a user-defined alternative.
+* [`hedging`][__link10] - Middleware that reduces tail latency via additional concurrent execution.
+* [`breaker`][__link11] - Middleware that prevents cascading failures.
+* [`fallback`][__link12] - Middleware that replaces invalid output with a user-defined alternative.
 
 ## Middleware Ordering
 
@@ -142,34 +143,37 @@ let service = ServiceBuilder::new()
 
 ## Examples
 
-Runnable examples covering each middleware and common composition patterns:
+Examples covering each middleware and common composition patterns:
 
-* [`timeout`][__link12]: Basic timeout that cancels long-running operations.
-* [`timeout_advanced`][__link13]: Dynamic timeout durations and timeout callbacks.
-* [`retry`][__link14]: Automatic retry with input cloning and recovery classification.
-* [`retry_advanced`][__link15]: Custom input cloning with attempt metadata injection.
-* [`retry_outage`][__link16]: Input restoration from errors when cloning is not possible.
-* [`breaker`][__link17]: Circuit breaker that monitors failure rates.
-* [`fallback`][__link18]: Substitutes default values for invalid outputs.
-* [`resilience_pipeline`][__link19]: Composing retry and timeout with metrics.
-* [`tower`][__link20]: Tower `ServiceBuilder` integration.
-* [`config`][__link21]: Loading settings from a [JSON file][__link22].
+* [`timeout`][__link13]: Basic timeout that cancels long-running operations.
+* [`timeout_advanced`][__link14]: Dynamic timeout duration and timeout callbacks.
+* [`retry`][__link15]: Automatic retry with input cloning and recovery classification.
+* [`retry_advanced`][__link16]: Custom input cloning with attempt metadata injection.
+* [`retry_outage`][__link17]: Input restoration from errors when cloning is not possible.
+* [`breaker`][__link18]: Circuit breaker that monitors failure rates.
+* [`hedging`][__link19]: Hedging slow requests with parallel attempts to reduce tail latency.
+* [`fallback`][__link20]: Substitutes default values for invalid outputs.
+* [`resilience_pipeline`][__link21]: Composing retry and timeout with metrics.
+* [`tower`][__link22]: Tower `ServiceBuilder` integration.
+* [`config`][__link23]: Loading settings from a [JSON file][__link24].
 
 ## Features
 
 This crate provides several optional features that can be enabled in your `Cargo.toml`:
 
-* **`timeout`** - Enables the [`timeout`][__link23] middleware for canceling long-running operations.
-* **`retry`** - Enables the [`retry`][__link24] middleware for automatically retrying failed operations with
+* **`timeout`** - Enables the [`timeout`][__link25] middleware for canceling long-running operations.
+* **`retry`** - Enables the [`retry`][__link26] middleware for automatically retrying failed operations with
   configurable backoff strategies, jitter, and recovery classification.
-* **`breaker`** - Enables the [`breaker`][__link25] middleware for preventing cascading failures.
-* **`fallback`** - Enables the [`fallback`][__link26] middleware for replacing invalid output with a
+* **`hedging`** - Enables the [`hedging`][__link27] middleware for reducing tail latency via additional
+  concurrent requests with configurable delay modes.
+* **`breaker`** - Enables the [`breaker`][__link28] middleware for preventing cascading failures.
+* **`fallback`** - Enables the [`fallback`][__link29] middleware for replacing invalid output with a
   user-defined alternative.
 * **`metrics`** - Exposes the OpenTelemetry metrics API for collecting and reporting metrics.
 * **`logs`** - Enables structured logging for resilience middleware using the `tracing` crate.
 * **`serde`** - Enables `serde::Serialize` and `serde::Deserialize` implementations for
   configuration types.
-* **`tower-service`** - Enables [`tower_service::Service`][__link27] trait implementations for all
+* **`tower-service`** - Enables [`tower_service::Service`][__link30] trait implementations for all
   resilience middleware.
 
 
@@ -178,29 +182,32 @@ This crate provides several optional features that can be enabled in your `Cargo
 This crate was developed as part of <a href="../..">The Oxidizer Project</a>. Browse this crate's <a href="https://github.com/microsoft/oxidizer/tree/main/crates/seatbelt">source code</a>.
 </sub>
 
- [__cargo_doc2readme_dependencies_info]: ggGkYW0CYXSEGy4k8ldDFPOhG2VNeXtD5nnKG6EPY6OfW5wBG8g18NOFNdxpYXKEG-h5X_wYKkOdGzmzySp-X6wWGyOeAMrS0ICyG1NcDWN9F0QhYWSFgmdsYXllcmVkZTAuMy4wgmtyZWNvdmVyYWJsZWUwLjEuMYJoc2VhdGJlbHRlMC4zLjGCZHRpY2tlMC4yLjGCbXRvd2VyX3NlcnZpY2VlMC4zLjM
+ [__cargo_doc2readme_dependencies_info]: ggGkYW0CYXSEGy4k8ldDFPOhG2VNeXtD5nnKG6EPY6OfW5wBG8g18NOFNdxpYXKEG4STOoP2_1kjG_YFfNmBbFbQG7vz1CXQ1d10GxetSJWWkufEYWSFgmdsYXllcmVkZTAuMy4wgmtyZWNvdmVyYWJsZWUwLjEuMYJoc2VhdGJlbHRlMC4zLjGCZHRpY2tlMC4yLjGCbXRvd2VyX3NlcnZpY2VlMC4zLjM
  [__link0]: https://crates.io/crates/layered/0.3.0
  [__link1]: https://docs.rs/layered/0.3.0/layered/?search=Stack
- [__link10]: https://docs.rs/seatbelt/0.3.1/seatbelt/breaker/index.html
- [__link11]: https://docs.rs/seatbelt/0.3.1/seatbelt/fallback/index.html
- [__link12]: https://github.com/microsoft/oxidizer/blob/main/crates/seatbelt/examples/timeout.rs
- [__link13]: https://github.com/microsoft/oxidizer/blob/main/crates/seatbelt/examples/timeout_advanced.rs
- [__link14]: https://github.com/microsoft/oxidizer/blob/main/crates/seatbelt/examples/retry.rs
- [__link15]: https://github.com/microsoft/oxidizer/blob/main/crates/seatbelt/examples/retry_advanced.rs
- [__link16]: https://github.com/microsoft/oxidizer/blob/main/crates/seatbelt/examples/retry_outage.rs
- [__link17]: https://github.com/microsoft/oxidizer/blob/main/crates/seatbelt/examples/breaker.rs
- [__link18]: https://github.com/microsoft/oxidizer/blob/main/crates/seatbelt/examples/fallback.rs
- [__link19]: https://github.com/microsoft/oxidizer/blob/main/crates/seatbelt/examples/resilience_pipeline.rs
+ [__link10]: https://docs.rs/seatbelt/0.3.1/seatbelt/hedging/index.html
+ [__link11]: https://docs.rs/seatbelt/0.3.1/seatbelt/breaker/index.html
+ [__link12]: https://docs.rs/seatbelt/0.3.1/seatbelt/fallback/index.html
+ [__link13]: https://github.com/microsoft/oxidizer/blob/main/crates/seatbelt/examples/timeout.rs
+ [__link14]: https://github.com/microsoft/oxidizer/blob/main/crates/seatbelt/examples/timeout_advanced.rs
+ [__link15]: https://github.com/microsoft/oxidizer/blob/main/crates/seatbelt/examples/retry.rs
+ [__link16]: https://github.com/microsoft/oxidizer/blob/main/crates/seatbelt/examples/retry_advanced.rs
+ [__link17]: https://github.com/microsoft/oxidizer/blob/main/crates/seatbelt/examples/retry_outage.rs
+ [__link18]: https://github.com/microsoft/oxidizer/blob/main/crates/seatbelt/examples/breaker.rs
+ [__link19]: https://github.com/microsoft/oxidizer/blob/main/crates/seatbelt/examples/hedging.rs
  [__link2]: https://docs.rs/tick/0.2.1/tick/?search=Clock
- [__link20]: https://github.com/microsoft/oxidizer/blob/main/crates/seatbelt/examples/tower.rs
- [__link21]: https://github.com/microsoft/oxidizer/blob/main/crates/seatbelt/examples/config.rs
- [__link22]: https://github.com/microsoft/oxidizer/blob/main/crates/seatbelt/examples/config.json
- [__link23]: https://docs.rs/seatbelt/0.3.1/seatbelt/timeout/index.html
- [__link24]: https://docs.rs/seatbelt/0.3.1/seatbelt/retry/index.html
- [__link25]: https://docs.rs/seatbelt/0.3.1/seatbelt/breaker/index.html
- [__link26]: https://docs.rs/seatbelt/0.3.1/seatbelt/fallback/index.html
- [__link27]: https://docs.rs/tower_service/0.3.3/tower_service/?search=Service
+ [__link20]: https://github.com/microsoft/oxidizer/blob/main/crates/seatbelt/examples/fallback.rs
+ [__link21]: https://github.com/microsoft/oxidizer/blob/main/crates/seatbelt/examples/resilience_pipeline.rs
+ [__link22]: https://github.com/microsoft/oxidizer/blob/main/crates/seatbelt/examples/tower.rs
+ [__link23]: https://github.com/microsoft/oxidizer/blob/main/crates/seatbelt/examples/config.rs
+ [__link24]: https://github.com/microsoft/oxidizer/blob/main/crates/seatbelt/examples/config.json
+ [__link25]: https://docs.rs/seatbelt/0.3.1/seatbelt/timeout/index.html
+ [__link26]: https://docs.rs/seatbelt/0.3.1/seatbelt/retry/index.html
+ [__link27]: https://docs.rs/seatbelt/0.3.1/seatbelt/hedging/index.html
+ [__link28]: https://docs.rs/seatbelt/0.3.1/seatbelt/breaker/index.html
+ [__link29]: https://docs.rs/seatbelt/0.3.1/seatbelt/fallback/index.html
  [__link3]: https://crates.io/crates/tick/0.2.1
+ [__link30]: https://docs.rs/tower_service/0.3.3/tower_service/?search=Service
  [__link4]: https://docs.rs/seatbelt/0.3.1/seatbelt/?search=ResilienceContext
  [__link5]: https://docs.rs/seatbelt/0.3.1/seatbelt/?search=ResilienceContext
  [__link6]: https://docs.rs/recoverable/0.1.1/recoverable/?search=RecoveryInfo

crates/seatbelt/README.md

@@ -0,0 +1,86 @@
+// Copyright (c) Microsoft Corporation.
+// Licensed under the MIT License.
+#![expect(missing_docs, reason = "benchmark code")]
+
+use alloc_tracker::{Allocator, Session};
+use criterion::{Criterion, criterion_group, criterion_main};
+use futures::executor::block_on;
+use layered::{Execute, Service, Stack};
+use seatbelt::hedging::Hedging;
+use seatbelt::{RecoveryInfo, ResilienceContext};
+use tick::Clock;
+
+#[global_allocator]
+static ALLOCATOR: Allocator<std::alloc::System> = Allocator::system();
+
+fn entry(c: &mut Criterion) {
+    let mut group = c.benchmark_group("hedging");
+    let session = Session::new();
+
+    // No hedging (baseline)
+    let service = Execute::new(|v: Input| async move { Output::from(v) });
+    let operation = session.operation("no-hedging");
+    group.bench_function("no-hedging", |b| {
+        b.iter(|| {
+            let _span = operation.measure_thread();
+            _ = block_on(service.execute(Input));
+        });
+    });
+
+    // With hedging (delay, no recovery needed)
+    let context = ResilienceContext::new(Clock::new_frozen());
+
+    let service = (
+        Hedging::layer("bench", &context)
+            .clone_input()
+            .recovery_with(|_, _| RecoveryInfo::never()),
+        Execute::new(|v: Input| async move { Output::from(v) }),
+    )
+        .into_service();
+
+    let operation = session.operation("with-hedging-delay");
+    group.bench_function("with-hedging-delay", |b| {
+        b.iter(|| {
+            let _span = operation.measure_thread();
+            _ = block_on(service.execute(Input));
+        });
+    });
+
+    // With hedging disabled (max_hedged_attempts = 0)
+    let context = ResilienceContext::new(Clock::new_frozen());
+
+    let service = (
+        Hedging::layer("bench", &context)
+            .clone_input()
+            .recovery_with(|_, _| RecoveryInfo::never())
+            .max_hedged_attempts(0),
+        Execute::new(|v: Input| async move { Output::from(v) }),
+    )
+        .into_service();
+
+    let operation = session.operation("with-hedging-passthrough");
+    group.bench_function("with-hedging-passthrough", |b| {
+        b.iter(|| {
+            let _span = operation.measure_thread();
+            _ = block_on(service.execute(Input));
+        });
+    });
+
+    group.finish();
+    session.print_to_stdout();
+}
+
+criterion_group!(benches, entry);
+criterion_main!(benches);
+
+#[derive(Debug, Clone)]
+struct Input;
+
+#[derive(Debug, Clone)]
+struct Output;
+
+impl From<Input> for Output {
+    fn from(_input: Input) -> Self {
+        Self
+    }
+}

crates/seatbelt/benches/hedging.rs

@@ -2,4 +2,6 @@
 
 When adding a new example, update [`README.md`](README.md) with a link to the new file and a
 brief one-line description. Keep the list in the same order: timeout, retry, breaker, fallback,
-combined pipelines, then configuration.
+combined pipelines, then configuration. 
+
+Also synchronize the `Examples` section in the [lib.rs](../lib.rs) documentation.

crates/seatbelt/examples/AGENTS.md

@@ -1,13 +1,14 @@
 # Examples
 
-Runnable examples covering each middleware and common composition patterns:
+Examples covering each middleware and common composition patterns:
 
 - [`timeout`](https://github.com/microsoft/oxidizer/blob/main/crates/seatbelt/examples/timeout.rs): Basic timeout that cancels long-running operations.
-- [`timeout_advanced`](https://github.com/microsoft/oxidizer/blob/main/crates/seatbelt/examples/timeout_advanced.rs): Dynamic timeout durations and timeout callbacks.
+- [`timeout_advanced`](https://github.com/microsoft/oxidizer/blob/main/crates/seatbelt/examples/timeout_advanced.rs): Dynamic timeout duration and timeout callbacks.
 - [`retry`](https://github.com/microsoft/oxidizer/blob/main/crates/seatbelt/examples/retry.rs): Automatic retry with input cloning and recovery classification.
 - [`retry_advanced`](https://github.com/microsoft/oxidizer/blob/main/crates/seatbelt/examples/retry_advanced.rs): Custom input cloning with attempt metadata injection.
 - [`retry_outage`](https://github.com/microsoft/oxidizer/blob/main/crates/seatbelt/examples/retry_outage.rs): Input restoration from errors when cloning is not possible.
 - [`breaker`](https://github.com/microsoft/oxidizer/blob/main/crates/seatbelt/examples/breaker.rs): Circuit breaker that monitors failure rates.
+- [`hedging`](https://github.com/microsoft/oxidizer/blob/main/crates/seatbelt/examples/hedging.rs): Hedging slow requests with parallel attempts to reduce tail latency.
 - [`fallback`](https://github.com/microsoft/oxidizer/blob/main/crates/seatbelt/examples/fallback.rs): Substitutes default values for invalid outputs.
 - [`resilience_pipeline`](https://github.com/microsoft/oxidizer/blob/main/crates/seatbelt/examples/resilience_pipeline.rs): Composing retry and timeout with metrics.
 - [`tower`](https://github.com/microsoft/oxidizer/blob/main/crates/seatbelt/examples/tower.rs): Tower `ServiceBuilder` integration.

crates/seatbelt/examples/README.md

@@ -0,0 +1,76 @@
+// Copyright (c) Microsoft Corporation.
+// Licensed under the MIT License.
+
+//! Hedging middleware example demonstrating how a slow primary request is hedged
+//! with a faster secondary request that completes first.
+
+use std::sync::atomic::{AtomicU32, Ordering};
+use std::time::Duration;
+
+use layered::{Execute, Service, Stack};
+use seatbelt::hedging::Hedging;
+use seatbelt::{RecoveryInfo, ResilienceContext};
+use tick::Clock;
+use tracing_subscriber::layer::SubscriberExt;
+use tracing_subscriber::util::SubscriberInitExt;
+
+static CALL_COUNT: AtomicU32 = AtomicU32::new(0);
+
+#[tokio::main]
+async fn main() {
+    // Set up tracing subscriber for logs to console
+    tracing_subscriber::registry().with(tracing_subscriber::fmt::layer()).init();
+
+    let clock = Clock::new_tokio();
+    let context = ResilienceContext::new(&clock).use_logs();
+
+    let op_clock = clock.clone();
+
+    // Configure hedging: if the original request hasn't completed after 200ms,
+    // launch a hedging request. The first successful response wins.
+    let stack = (
+        Hedging::layer("my_hedging", &context)
+            .clone_input()
+            .max_hedged_attempts(4)
+            .recovery_with(|_output: &String, _args| RecoveryInfo::never())
+            .hedging_delay(Duration::from_millis(200))
+            .on_execute(|_input, args| {
+                println!(
+                    "[execute] launching attempt {} (last: {})",
+                    args.attempt().index(),
+                    args.attempt().is_last()
+                );
+            }),
+        Execute::new(move |input: String| {
+            let clock = op_clock.clone();
+            async move { slow_then_fast_operation(input, &clock).await }
+        }),
+    );
+
+    let service = stack.into_service();
+
+    println!("[main] sending request...");
+    let start = std::time::Instant::now();
+
+    let output = service.execute("hello".to_string()).await;
+    println!("[main] result: {output} (took {:?})", start.elapsed());
+}
+
+/// Simulates a service where the first two calls are slow (500ms) and subsequent
+/// calls (the second hedging attempt onwards) are fast (50ms). The fast hedging
+/// attempt completes before the slow ones, demonstrating how hedging reduces tail latency.
+async fn slow_then_fast_operation(input: String, clock: &Clock) -> String {
+    let call = CALL_COUNT.fetch_add(1, Ordering::Relaxed);
+
+    if call < 2 {
+        // Original request and first hedge: simulate a slow response
+        println!("[service] attempt {call}: slow path (500ms)");
+        clock.delay(Duration::from_millis(500)).await;
+        format!("{input} - slow response")
+    } else {
+        // Second hedging request: simulate a fast response
+        println!("[service] attempt {call}: fast path (50ms)");
+        clock.delay(Duration::from_millis(50)).await;
+        format!("{input} - fast response")
+    }
+}