diff --git a/CHANGELOG.md b/CHANGELOG.md index 254084b..2fa8895 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,55 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] +## [0.5.0] - 2026-08-16 + +Retry can finally stop an attempt that hangs. Breaking. + +### Added +- `Retry::attempt_timeout(duration, on_timeout)`. Bounds a single attempt, so an operation that + hangs is finally stopped. When it fires the in-flight future is dropped and the attempt is + treated as a failure, feeding the normal backoff and the normal `.when(..)` predicate. + `on_timeout` supplies the error to report, because the operation never returned one. + + The wait runs on the injected `Clock`, not on Tokio directly, so a timeout is testable on a mock + clock with no real time. `tokio::time::timeout` cannot be. + [ADR007](https://github.com/azeemshaik025/mettle/blob/main/docs/adr/ADR007.md) covers why the + error comes from a closure rather than making `RetryError::error()` an `Option`, and why there is + no blocking equivalent. + +### Fixed +- `max_elapsed` documented honestly. It is checked *between* attempts, so on its own it never + bounded an operation that hangs, while the README said "give up after ~30s total". A future that + is never ready gave the budget nothing to act on. Pairing it with `attempt_timeout` is what makes + the budget enforceable; the blocking twin says plainly that it has no equivalent and points at + the call's own timeout setting instead. + +### Changed +- **Breaking:** `Retry` and `RetryFuture` take one more type parameter, for the on-timeout handler. + Only affects code that names those types; `retry(..)` and every builder method are unchanged. +- The crate description and docs now say what mettle is — retry, answered end to end — rather than + "a resilience toolkit", and no longer claim that timeout and circuit breaking are planned. Timeout + shipped here; the circuit breaker was built and deliberately not shipped. Scope, including what + has been refused and why, is in + [docs/ROADMAP.md](https://github.com/azeemshaik025/mettle/blob/main/docs/ROADMAP.md). + +### Upgrading + +Nothing to do unless you *name* `Retry` or `RetryFuture`, which mostly means storing one in a +struct field or writing a function that returns one. `retry(..)` and every builder method are +unchanged, so the common inline use compiles as-is. + +```diff +-fn build() -> Retry bool> { ++fn build() -> Retry bool, fn() -> E> { +``` + +The new parameter is the on-timeout handler. When no timeout is configured it is the function +pointer `fn() -> E` that `retry(..)` seeds, and it is never called. A default (`Q = NoTimeout`) was +tried and does not work: `NoTimeout` cannot implement `Fn() -> E`, so the no-timeout case would +need a second `IntoFuture` impl and the two would be seen as potentially overlapping. +[ADR007](https://github.com/azeemshaik025/mettle/blob/main/docs/adr/ADR007.md) has the detail. + ## [0.4.0] - 2026-08-09 Retry now reports why it gave up, not just what failed last. Breaking. diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md index 4177edc..a7b642c 100644 --- a/CONTRIBUTING.md +++ b/CONTRIBUTING.md @@ -15,15 +15,37 @@ cargo fmt --all # format cargo clippy --all-targets --all-features -- -D warnings ``` -Keep the whole feature matrix green: +Keep the whole feature matrix green. CI runs `test`, `clippy`, `docs` and `msrv` against each of +these four configurations, so a change that only builds under `--all-features` still fails: ```sh +cargo test --all-features +cargo test # defaults cargo test --no-default-features --features async cargo test --no-default-features --features blocking ``` -MSRV is 1.85. CI runs all of the above plus a docs build (`-D warnings`) and a -semver-compatibility check, so run them locally before opening a PR. +MSRV is 1.85, and it applies to tests and examples too: + +```sh +cargo +1.85.0 check --all-features --all-targets +``` + +The two checks that only CI used to catch, and that are worth running before you open a PR: + +```sh +RUSTDOCFLAGS='-D warnings' cargo doc --all-features --no-deps # broken intra-doc links +cargo semver-checks # cargo install cargo-semver-checks +``` + +`cargo doc` is the only thing that catches a broken intra-doc link; tests, clippy and MSRV all pass +straight through them. + +**If your change is breaking, bump the version in the same branch.** `cargo semver-checks` runs on +the pull request and compares your manifest against the published crate, so a breaking change left +at the current version fails CI no matter how good it is. In 0.x that means a minor bump: 0.4.0 to +0.5.0. Finalize the `CHANGELOG.md` heading with a date in the same commit and leave an empty +`## [Unreleased]` above it. ## Licensing diff --git a/Cargo.lock b/Cargo.lock index e45640b..c239368 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -10,7 +10,7 @@ checksum = "da7c62ceae207dd37ea5b845da6a0696c799f85e97da1ab5b7910be3c1c80223" [[package]] name = "mettle" -version = "0.4.0" +version = "0.5.0" dependencies = [ "fastrand", "pin-project-lite", diff --git a/Cargo.toml b/Cargo.toml index d5577ca..a5b786d 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,9 +1,9 @@ [package] name = "mettle" -version = "0.4.0" +version = "0.5.0" edition = "2024" rust-version = "1.85" -description = "A resilience toolkit for Rust: async and blocking retries with configurable backoff and jitter." +description = "Retry for Rust, async and blocking: backoff, jitter, per-attempt timeouts, and a clock you can mock." readme = "README.md" license = "MIT OR Apache-2.0" keywords = ["retry", "backoff", "jitter", "resilience", "tokio"] diff --git a/README.md b/README.md index 64e994e..7e10c46 100644 --- a/README.md +++ b/README.md @@ -5,9 +5,11 @@ [![CI](https://github.com/azeemshaik025/mettle/actions/workflows/ci.yml/badge.svg)](https://github.com/azeemshaik025/mettle/actions/workflows/ci.yml) [![license](https://img.shields.io/crates/l/mettle.svg)](#license) -**A resilience toolkit for Rust.** +**Retry for Rust, answered end to end.** -Composable, testable primitives for handling failure. [Documentation](https://docs.rs/mettle). +How long to wait, when to stop waiting on one attempt, when to give up, and what to report when +it's over. Every policy decision is a pure function of an injected clock, so a 30-second budget is +testable in microseconds with no real time passing. [Documentation](https://docs.rs/mettle). ## Install @@ -25,16 +27,27 @@ cargo add mettle --no-default-features --features blocking ```rust use mettle::retry; -use std::time::Duration; -// Retry with sensible defaults (exponential backoff, up to 3 retries), -// then override only what you need. +// Sensible defaults: exponential backoff, up to 3 retries. +let body = retry(|| async { fetch(&url).await }).await?; +``` + +Then override only what you need: + +```rust let body = retry(|| async { fetch(&url).await }) - .when(|e: &FetchError| e.is_transient()) // skip permanent errors - .max_elapsed(Duration::from_secs(30)) // give up after ~30s total + .when(|e: &FetchError| e.is_transient()) // skip permanent errors + .attempt_timeout(Duration::from_secs(5), || FetchError::Timeout) + .max_elapsed(Duration::from_secs(30)) .await?; ``` +`attempt_timeout` bounds a single try, dropping the in-flight future and feeding the timeout into +the normal backoff. Reach for it whenever the call can hang: `max_elapsed` is only consulted +*between* attempts, so on its own it cannot stop a call that never returns. Its second argument is +the error to report, since a timed-out attempt never returned one of its own — for `io::Error` that +is `|| ErrorKind::TimedOut.into()`. + No async runtime? The blocking twin is identical but ends in `.call()` instead of `.await`. Retrying on a fixed schedule means every client that failed together retries together, so a service @@ -74,15 +87,24 @@ match retry(|| async { fetch(&url).await }).await { Only want the underlying error? `.map_err(RetryError::into_error)`. -## Tools - -Each tool comes with a runnable example. Start there: +## Examples -- **retry**: async [examples/retry.rs](https://github.com/azeemshaik025/mettle/blob/main/examples/retry.rs) · blocking [examples/blocking_retry.rs](https://github.com/azeemshaik025/mettle/blob/main/examples/blocking_retry.rs) +Runnable, and the fastest way in: +[examples/retry.rs](https://github.com/azeemshaik025/mettle/blob/main/examples/retry.rs) (async) · +[examples/blocking_retry.rs](https://github.com/azeemshaik025/mettle/blob/main/examples/blocking_retry.rs) +(blocking). Retries emit `tracing` events out of the box (target `mettle::retry`). Install any subscriber (e.g. `tracing_subscriber::fmt::init()`) to see them. +## Scope + +mettle does retry, and does it completely, rather than being a shallow toolkit of five tools. The +test for anything new is whether it's about a failed call; that's what keeps bulkheads, rate +limiting and caching out. What's planned and what's been refused, with the reasoning, is in +[docs/ROADMAP.md](https://github.com/azeemshaik025/mettle/blob/main/docs/ROADMAP.md); design +decisions are in [docs/adr](https://github.com/azeemshaik025/mettle/tree/main/docs/adr). + ## Status v0.x, with async (Tokio) and blocking APIs. Expect breaking changes before 1.0. diff --git a/docs/ROADMAP.md b/docs/ROADMAP.md new file mode 100644 index 0000000..132d9aa --- /dev/null +++ b/docs/ROADMAP.md @@ -0,0 +1,126 @@ +# Roadmap and scope + +What mettle intends to build, and what it refuses to. The refusals matter more than the plans: a +crate with an edge is worth more than a crate with everything, and every feature costs a feature +matrix, a blocking twin, and an ADR forever. + +This file is the record. If something isn't here, it hasn't been decided. + +## What mettle is + +**Retry, solved completely, and the only one you can test exactly without sleeping.** + +Not a toolkit of five shallow tools. One problem, answered end to end: how long to wait, when to +stop waiting on one try, when the fleet should stop retrying, when to try again before the first +finishes, and what to report when it's over. Every policy decision is a pure function of an injected +time source, so you can test your configuration at its boundaries with no real clock. + +That gives a test for every future request: **is this about a failed call?** If not, it's out. That +test is what refused most of the list below, and it's the reason this file exists. + +The second, smaller claim is shape. Operations are `FnMut() -> Fut` factories, so nothing needs +`Clone`, `Send`, or `'static`. That's why `tower::retry` doesn't work for tonic users whose +`http::Request` isn't `Clone`. + +## Shipped + +| | | +|------|-------| +| retry, async and blocking twins over one decision core | 0.1 | +| exponential backoff, validated config, saturating arithmetic | 0.1 | +| jitter (full) and `DecorrelatedBackoff` | 0.3.0 | +| `RetryError` with attempts, elapsed, and why it stopped | 0.4.0 | +| `attempt_timeout`, so a hung attempt is finally bounded | 0.5.0 | + +## Planned + +In order. Each is a prerequisite for the ones under it. + +**1. Retry budget.** +A shared, rate-limited allowance for retries, so a partial outage can't turn into a retry storm. +Universal outside Rust (gRPC `retryThrottling`, Envoy `budget_percent`, Linkerd `retryRatio`, +Finagle `RetryBudget`, AWS retry quotas) and effectively absent inside it: the only working +implementation is buried in `aws-smithy-runtime` behind about twenty crates. + +This is the one the competition structurally cannot copy. A budget needs a handle shared across +calls; `backon`, `tokio-retry`, `tryhard` and `again` all hand out per-call values. Verified against +`backon` (25M downloads/90d): no budget, no hedging, no per-attempt timeout. + +Two decisions belong in the ADR before any code. Use a rate-with-TTL budget rather than the timeless +token bucket, because "does the budget refill after sixty idle seconds" is then a hand-fed-`Instant` +test nobody else in Rust can write, and because a budget that only refills on traffic locks out a +low-QPS caller forever. And name the debit ordering: `when(err)` runs before the backoff is drawn, +so a naive implementation debits for a retry that may never happen. + +**2. `cargo-mutants` in CI.** +Not a feature. The worst bugs this crate has had (the cleared window, the generation token, the +half-open livelock, the mock clock advancing on creation) were all mutation-detectable and all found +by hand. Do this before hedging. + +**3. Hedging, one backup request.** +Send a backup when the first request passes a latency threshold, take whichever answers first. Not a +separate tool: it's speculative retry. Weak incumbents — `tower::hedge` has had no functional change +since 2022, and the standalone crates have four-figure lifetime downloads. + +Strictly after the budget. Hedging at a fixed delay doubles load on a degraded backend exactly when +it's degraded, which is why gRPC gates hedging on `retryThrottling` and Finagle builds its backup +requests on `RetryBudget`. + +Only ever one backup. `#![forbid(unsafe_code)]` means there's no safe projection from +`Pin<&mut [Option; N]>` to element *i*, and boxing would violate ADR001. + +## Refused + +Recorded so nobody relitigates them. Each can be revisited if a real user asks, but the default is +no. + +- **Circuit breaker.** Built, tested, and not shipped — see below. The one refusal we reached by + building the thing. +- **Bulkhead.** Reads no clock, so it exercises none of what this crate is for, and + `Semaphore::try_acquire` is five lines. A queueing bulkhead is worse: it needs async queueing + machinery mettle doesn't have, and the blocking twin is untestable because `Condvar::wait_timeout` + has no injection point. +- **General rate limiting.** `governor` (13.6M downloads/90d) and `ratelimit` (4M) own this, and + `ratelimit` already implements the injected-clock thesis. Being third here would turn our + differentiator into table stakes. +- **Adaptive concurrency limits (AIMD, Vegas, Gradient).** The value is the published control law, + not deterministic replay. Deterministic testing proves a limiter is repeatable, not well tuned. +- **Standalone timeout.** `tokio::time::timeout` exists, and a blocking version would force + `Send + 'static` onto the caller's operation. What was genuinely missing was the *per-attempt* + bound, which shipped as `attempt_timeout` (ADR007). +- **Fallback.** `Result::or_else` already is the combinator. +- **Cache.** Owned by `moka` and `foyer`, and there's no decision logic to make pure, so the + architecture buys nothing. +- **Health checks.** A fleet concern that belongs with discovery and load balancing. mettle has no + notion of a set of endpoints. +- **Ambient deadline propagation.** The hard part is a task-local convention that survives `spawn`, + which is a runtime and ecosystem problem. +- **Graceful degradation.** A posture, not a primitive. + +## The circuit breaker, in detail + +It exists, on the `circuit-breaker` branch: a pure state machine, a shared wrapper, RAII permits, an +injected `Now`, 100 tests, and a half-open livelock found and fixed. It is not shipped. + +The demand is roughly fiftyfold smaller than retry — about 1.2M downloads per 90 days across every +Rust breaker (`failsafe`, `recloser`, `circuit_breaker`) against about 60M for the retry crates. +Infrastructure has also absorbed much of the use case: Istio, Linkerd and Envoy break circuits at +the proxy, so anyone on a mesh already has one. + +And it doesn't fit the spine. Everything else here answers "a call failed, now what". A breaker +answers "should I call at all", which is why it was the only thing needing shared mutable state, a +new trait, and a concurrency story. Shipping it would double the public surface for a tool the +evidence says few would reach for. + +The work isn't wasted; it's how we found out. If a user asks, the branch is ready. + +## The ceiling + +One maintainer, with a commitment to deterministic tests and mutation-checked fixes. The binding +constraint isn't lines of code, it's feature matrix times twin count times ADR surface. Every +async-only feature is a documented exception to the twin rule: `attempt_timeout` is the first, and +hedging would be the second, which is as many as this crate should have. + +Budget for correction too. The circuit breaker needed a fix worth roughly a fifth of its size +*after* it was otherwise complete, and that was the half-open livelock, which no per-call assertion +could have caught. Assume the same on the retry budget. diff --git a/docs/adr/ADR007.md b/docs/adr/ADR007.md new file mode 100644 index 0000000..42c33e8 --- /dev/null +++ b/docs/adr/ADR007.md @@ -0,0 +1,86 @@ +# ADR007: Bounding one attempt, and where the error comes from + +**Status:** Accepted + +## Context + +`max_elapsed` lives inside `should_retry_after`, which is only reached when an attempt returns. An +operation that hangs never returns, so the budget never got a chance to apply. A dead TCP peer, a +lost response, a service that accepts your connection and then goes quiet: all of them produce a +future that is simply never ready, and the retry sat there forever. + +That was not a missing feature. `README.md` said "give up after ~30s total" and the crate did not +do it. Measured before the fix: a 500 ms budget still running at 3 seconds, stopped only by an +external `tokio::time::timeout`. + +## Decisions + +**1. `attempt_timeout` bounds the attempt; `max_elapsed` keeps bounding the whole.** +Two separate limits, because they answer different questions. `attempt_timeout` drops the in-flight +future, which in Rust is real cancellation, and turns the attempt into a failure that feeds the +normal backoff and the normal `.when(..)` predicate. + +The two are also not independent, which is the part worth noticing: a timed-out attempt returns +control to the loop, so `max_elapsed` is finally consulted. The end-to-end check reports +`reason = max_elapsed` where before it hung. The per-attempt bound is what makes the total budget +enforceable at all. + +**2. The caller supplies the error, through `on_timeout`.** +A timed-out attempt has no `E`, because the operation never returned one. Three ways out, and only +one of them is cheap for the people who never use this: + +- Make `RetryError::error()` return `Option<&E>`. Every caller then unwraps an `Option` in the + middle of matching their own error enum, forever, to serve a case most of them never hit. +- Require `E: From`, which puts a bound on the whole builder and rules out + `E = String` and `E = anyhow::Error`, exactly the types ADR006 went out of its way to keep. +- Take `on_timeout: impl Fn() -> E`. + +We took the third. It keeps one error type flowing through `.when(..)`, the tracing event and the +final `RetryError`, and a timed-out attempt travels the identical path to a returned one. It also +reads as documentation at the call site: `|| MyError::Timeout` says what a timeout *means* in this +caller's vocabulary, which is a thing the library could not have guessed. + +The cost is a second argument on the builder method, and that the handler is a `Fn` rather than a +`FnOnce`, since it may fire on every attempt. + +**3. The deadline runs on the injected `Clock`, not on Tokio.** +`Clock::sleep` already existed for backoff, so the timeout reuses it. That is the whole reason to +have this rather than telling people to wrap their operation in `tokio::time::timeout`: with the +injected clock, a five second timeout is testable in microseconds and the test asserts the exact +sleep sequence. `tokio::time::timeout` cannot be tested that way without real time or careful use +of `start_paused`, and it ties the operation to Tokio. + +**4. `Retry` and `RetryFuture` gain a type parameter, and that is breaking.** +The handler has to be stored, and `E` is not a parameter of `Retry` — it comes from the operation's +return type. So the handler needs its own parameter, and `retry()` seeds it with `fn() -> E`, a real +function-pointer type that satisfies `Fn() -> E` whether or not a timeout is configured. That keeps +one `IntoFuture` impl rather than two overlapping ones. + +We tried a defaulted `Q = NoTimeout` to avoid the break. It does not work: `NoTimeout` cannot +implement `Fn() -> E`, so the no-timeout case would need a second `IntoFuture` impl and the two +would be seen as potentially overlapping. + +The break is narrow. `retry(..)` and every builder method are unchanged; only code that *names* +`Retry` or `RetryFuture`, such as storing one in a struct field, has to add a parameter. + +**5. There is no blocking equivalent, and the docs say so.** +Interrupting a blocking closure means running it on another thread, which forces `Send + 'static` +onto the caller's operation. `op_may_be_non_send` and `op_may_borrow_non_static_data` exist to +forbid exactly that, and ADR001 chose a hand-written future partly to avoid those bounds. + +So the blocking `max_elapsed` doc states plainly that it is checked between attempts, that there is +no equivalent, and that the call itself is where to set a bound: `TcpStream::set_read_timeout`, or +whatever the client offers. An honest limitation documented beats a feature that quietly changes +what the caller's operation is allowed to capture. + +## Consequences + +The crate now delivers the time bound its README claimed. `max_elapsed` alone still cannot stop a +hang, and both its doc comments now say so rather than implying otherwise. + +One thing the work exposed, worth keeping in mind beyond this change: the test mock advanced its +virtual clock when a sleep was *created* rather than when it was *polled*. That was invisible while +retry only ever built a sleep it intended to await. Arming a deadline that a fast operation never +polls made it visible, and it had been quietly inflating nothing only by luck. Real timers cost +nothing if dropped unpolled, and the mock now matches. A mock that diverges from the thing it +stands in for is a bug waiting for a new usage pattern to find it. diff --git a/docs/adr/README.md b/docs/adr/README.md index f2acee0..e80f212 100644 --- a/docs/adr/README.md +++ b/docs/adr/README.md @@ -12,3 +12,6 @@ changes, edit its file and say what changed and why. | [004](ADR004.md) | Jitter, a fourth dependency, and `Clone` | Accepted | | [005](ADR005.md) | Making the tested path as usable as the production path | Accepted | | [006](ADR006.md) | `RetryError` and the `Error` bound | Accepted | +| [007](ADR007.md) | Bounding one attempt, and where the error comes from | Accepted | + +What's planned and what's deliberately refused lives in [../ROADMAP.md](../ROADMAP.md). diff --git a/examples/blocking_retry.rs b/examples/blocking_retry.rs index 3324139..f422cd9 100644 --- a/examples/blocking_retry.rs +++ b/examples/blocking_retry.rs @@ -4,6 +4,11 @@ //! plain `Result` (no `async`) and you finish with `.call()` instead of `.await`, so no runtime //! is needed. //! +//! One method is missing here on purpose: there is no `.attempt_timeout`. Interrupting a blocking +//! call means running it on another thread, which would force `Send + 'static` onto your +//! operation. Bound the call itself instead — `TcpStream::set_read_timeout`, or whatever your +//! client offers. +//! //! Run with: `cargo run --example blocking_retry` use std::sync::atomic::{AtomicUsize, Ordering::SeqCst}; diff --git a/examples/retry.rs b/examples/retry.rs index 21a6adc..755f7af 100644 --- a/examples/retry.rs +++ b/examples/retry.rs @@ -9,6 +9,7 @@ //! - `.when(pred)` retries only the errors `pred` accepts (default: every error) //! - `.backoff(cfg)` swaps or tunes the backoff strategy //! - `.max_elapsed(d)` gives up once the next wait would push total time past `d` +//! - `.attempt_timeout(d, on_timeout)` bounds a single attempt, so a hang can't stall the loop //! - `.clock(clock)` supplies the time source (default: Tokio) //! //! The default schedule is deterministic, which means a fleet of clients that failed together @@ -68,6 +69,21 @@ async fn main() -> Result<(), mettle::BackoffConfigError> { .await; println!("5. decorrel.: {result:?}"); // every delay at least 20ms + // 6) A call that hangs. Without `.attempt_timeout` this never returns: `.max_elapsed` is only + // checked between attempts, and an attempt that never finishes never gets there. Bounding + // the attempt turns the hang into an ordinary failure, which lets the budget apply. + let result = retry(hangs_forever) + .attempt_timeout(Duration::from_millis(100), || FetchError::Timeout) + .max_elapsed(Duration::from_millis(300)) + .await; + let err = result.unwrap_err(); + println!( + "6. hung call: gave up after {} attempts in {:?}, reason={}", + err.attempts(), + err.elapsed(), + err.stop_reason().as_str() + ); + Ok(()) } @@ -88,6 +104,12 @@ async fn always_404() -> Result<&'static str, FetchError> { Err(FetchError::NotFound) } +/// A call that never returns: a dead peer, a lost response, a service that accepted the connection +/// and then went quiet. +async fn hangs_forever() -> Result<&'static str, FetchError> { + std::future::pending().await +} + /// A snappier exponential backoff: 20 ms first delay, defaults for the rest. fn fast_backoff() -> ExponentialBackoff { ExponentialBackoff::new(ExponentialBackoffConfig { diff --git a/src/blocking/retry.rs b/src/blocking/retry.rs index aa69dd5..05b8925 100644 --- a/src/blocking/retry.rs +++ b/src/blocking/retry.rs @@ -72,6 +72,12 @@ impl Retry { } /// Give up once this much total time has elapsed (default: no limit). + /// + /// Checked *between* attempts, when one returns. It cannot interrupt an attempt that is still + /// running, and there is no blocking equivalent of the async `attempt_timeout`: interrupting a + /// blocking closure would mean running it on another thread, which would force + /// `Send + 'static` onto your operation, and two tests here exist to forbid exactly that. Bound the call itself instead, with whatever it offers, such as + /// `TcpStream::set_read_timeout` or your client's own timeout setting. pub fn max_elapsed(mut self, budget: Duration) -> Self { self.max_elapsed = Some(budget); self diff --git a/src/lib.rs b/src/lib.rs index 4b8c56a..eaf22bd 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -1,10 +1,13 @@ //! # mettle //! -//! A resilience toolkit for Rust: composable, testable primitives for handling failure, so you -//! don't hand-roll retry-and-backoff logic in every project. +//! Retry for Rust, answered end to end: how long to wait, when to stop waiting on one attempt, +//! when to give up, and what to report when it's over. Every policy decision is a pure function of +//! an injected clock, so a thirty second budget is testable in microseconds with no real time +//! passing. //! -//! Available now: `retry()` (async) and `blocking::retry()` (sync), both with configurable -//! backoff and optional jitter. Timeout and circuit breaking are planned. +//! `retry()` (async) and `blocking::retry()` (sync) share one decision core, so the two decide +//! alike. Operations are `FnMut() -> Fut` factories rather than values, so nothing you retry has to +//! be `Clone`, `Send`, or `'static`. //! //! # Quickstart //! @@ -23,9 +26,14 @@ //! # async fn fetch() -> Result { Ok(1) } //! ``` //! -//! Override the backoff, clock, retry predicate (`.when`), or time budget (`.max_elapsed`) with -//! the builder methods, then `.await`. No async runtime? The blocking twin is identical but ends -//! in `.call()` instead of `.await`. +//! Override the backoff, clock, retry predicate (`.when`), per-attempt bound +//! (`.attempt_timeout`), or total time budget (`.max_elapsed`) with the builder methods, then +//! `.await`. No async runtime? The blocking twin is identical but ends in `.call()` instead of +//! `.await`. +//! +//! Reach for `.attempt_timeout` whenever the operation can hang. `.max_elapsed` is only consulted +//! between attempts, so on its own it cannot stop a call that never returns; the two together are +//! what make a total budget enforceable. //! //! # Jitter //! diff --git a/src/retry.rs b/src/retry.rs index 775313b..81e7d57 100644 --- a/src/retry.rs +++ b/src/retry.rs @@ -31,12 +31,13 @@ use crate::shared::{Decision, give_up, should_retry_after, trace_retry}; /// ([`backoff`](Retry::backoff), [`clock`](Retry::clock), [`when`](Retry::when), /// [`max_elapsed`](Retry::max_elapsed)), then `.await` it (it implements [`IntoFuture`]). #[must_use = "a `Retry` does nothing until you `.await` it"] -pub struct Retry { +pub struct Retry { op: F, backoff: B, clock: C, when: P, max_elapsed: Option, + attempt_timeout: Option<(Duration, Q)>, } /// Start retrying `op`, with sensible defaults for everything else: exponential backoff, @@ -66,7 +67,12 @@ pub struct Retry { /// .map_err(RetryError::into_error) /// # } /// ``` -pub fn retry(op: F) -> Retry bool> +// Five parameters, and a caller never writes this type: `retry(op)` is always used inline or +// through `.await`. A public alias would be another name to learn for no gain. +#[allow(clippy::type_complexity)] +pub fn retry( + op: F, +) -> Retry bool, fn() -> E> where F: FnMut() -> Fut, Fut: Future>, @@ -77,48 +83,97 @@ where clock: TokioClock, when: (|_| true) as fn(&E) -> bool, max_elapsed: None, + attempt_timeout: None, } } -impl Retry { +impl Retry { /// Override the backoff strategy (any [`Backoff`]). - pub fn backoff(self, backoff: B2) -> Retry { + pub fn backoff(self, backoff: B2) -> Retry { Retry { backoff, op: self.op, when: self.when, clock: self.clock, max_elapsed: self.max_elapsed, + attempt_timeout: self.attempt_timeout, } } /// Override the clock (any [`Clock`]), e.g. a mock clock in tests. - pub fn clock(self, clock: C2) -> Retry { + pub fn clock(self, clock: C2) -> Retry { Retry { clock, op: self.op, backoff: self.backoff, when: self.when, max_elapsed: self.max_elapsed, + attempt_timeout: self.attempt_timeout, } } /// Give up once this much total time has elapsed (default: no limit). + /// + /// Checked *between* attempts, when one returns. It cannot interrupt an attempt that is still + /// running, so on its own it does not bound an operation that hangs. Pair it with + /// [`attempt_timeout`](Retry::attempt_timeout), which bounds each attempt and thereby gives + /// this budget something to act on. pub fn max_elapsed(mut self, budget: Duration) -> Self { self.max_elapsed = Some(budget); self } } -impl Retry +impl Retry where F: FnMut() -> Fut, Fut: Future>, { + /// Bound how long a single attempt may take. + /// + /// `max_elapsed` is only consulted *between* attempts, so on its own it cannot stop an + /// operation that hangs: a dead TCP peer produces a future that is simply never ready, and the + /// budget never gets a chance to apply. This does stop it. When `timeout` passes, the in-flight + /// future is dropped (real cancellation in Rust) and the attempt is treated as a failure, so it + /// feeds the normal backoff sequence and the normal `.when(..)` predicate. + /// + /// `on_timeout` supplies the error to report for a timed-out attempt, because the operation + /// never returned one. Mapping it into your own error type keeps everything downstream, from + /// `.when(..)` to the final [`RetryError`], working on one type. + /// + /// The wait runs on the injected [`Clock`], not on Tokio directly, so a mock clock makes this + /// testable without real time. + /// + /// ```no_run + /// # use mettle::retry; + /// # use std::time::Duration; + /// # #[derive(Debug)] enum MyError { Timeout, Other } + /// # async fn fetch() -> Result { Ok(1) } + /// # async fn demo() { + /// let out = retry(fetch) + /// .attempt_timeout(Duration::from_secs(5), || MyError::Timeout) + /// .await; + /// # let _ = out; + /// # } + /// ``` + pub fn attempt_timeout(self, timeout: Duration, on_timeout: Q2) -> Retry + where + Q2: Fn() -> E, + { + Retry { + attempt_timeout: Some((timeout, on_timeout)), + op: self.op, + backoff: self.backoff, + clock: self.clock, + when: self.when, + max_elapsed: self.max_elapsed, + } + } + /// Only retry errors for which `predicate` returns `true` (default: retry all). /// /// The predicate sees each `&E`, so `e`'s type is inferred; no annotation needed. - pub fn when(self, predicate: P2) -> Retry + pub fn when(self, predicate: P2) -> Retry where P2: Fn(&E) -> bool, { @@ -128,21 +183,23 @@ where backoff: self.backoff, clock: self.clock, max_elapsed: self.max_elapsed, + attempt_timeout: self.attempt_timeout, } } } -impl IntoFuture for Retry +impl IntoFuture for Retry where C: Clock, P: Fn(&E) -> bool, + Q: Fn() -> E, E: std::fmt::Debug, F: FnMut() -> Fut, B: Backoff, Fut: Future>, { type Output = Result>; - type IntoFuture = RetryFuture; + type IntoFuture = RetryFuture; fn into_future(self) -> Self::IntoFuture { RetryFuture { @@ -158,6 +215,7 @@ where clock: self.clock, backoff: self.backoff, max_elapsed: self.max_elapsed, + attempt_timeout: self.attempt_timeout, } } } @@ -167,7 +225,7 @@ pin_project! { /// /// It borrows only what the operation borrows, and is `Send` only when its parts are, so /// the operation may borrow local state and need not be `Send`. - pub struct RetryFuture { + pub struct RetryFuture { op: F, when: P, clock: C, @@ -179,6 +237,7 @@ pin_project! { state: RetryState, max_elapsed: Option, + attempt_timeout: Option<(Duration, Q)>, } } @@ -189,14 +248,17 @@ pin_project! { enum RetryState { Idle, Sleeping { #[pin] delay: S }, - Attempting { #[pin] fut: Fut }, + // The timeout future is `None` unless `attempt_timeout` was set, so the common path + // allocates and polls nothing extra. + Attempting { #[pin] fut: Fut, #[pin] deadline: Option }, } } -impl Future for RetryFuture +impl Future for RetryFuture where B: Backoff, P: Fn(&E) -> bool, + Q: Fn() -> E, E: std::fmt::Debug, F: FnMut() -> Fut, C: Clock, @@ -214,14 +276,39 @@ where // the clock starts. RetryStateProj::Idle => { *this.start = Some(this.clock.now()); - RetryState::Attempting { fut: (this.op)() } + let deadline = this + .attempt_timeout + .as_ref() + .map(|(d, _)| this.clock.sleep(*d)); + RetryState::Attempting { + fut: (this.op)(), + deadline, + } } // An attempt is in flight — drive it. - RetryStateProj::Attempting { fut } => match fut.poll(cx) { - Poll::Pending => return Poll::Pending, - Poll::Ready(Ok(value)) => return Poll::Ready(Ok(value)), - Poll::Ready(Err(err)) => { + RetryStateProj::Attempting { fut, deadline } => { + // The operation gets first look; a result that is already available wins even + // if the deadline is also up, so a race never discards a finished call. + let err = match fut.poll(cx) { + Poll::Ready(Ok(value)) => return Poll::Ready(Ok(value)), + Poll::Ready(Err(err)) => err, + // Pair the deadline with its handler, so the arm that needs both is the + // only one that can run. Nothing to unwrap, and no unreachable panic. + Poll::Pending => { + match (deadline.as_pin_mut(), this.attempt_timeout.as_ref()) { + (Some(d), Some((_, on_timeout))) => match d.poll(cx) { + Poll::Pending => return Poll::Pending, + // Out of time. Synthesising the error here is what lets a + // timed-out attempt travel the same path as a returned one. + Poll::Ready(()) => on_timeout(), + }, + // No timeout configured, so the attempt runs unbounded. + _ => return Poll::Pending, + } + } + }; + { let elapsed = || { this.start.map_or(Duration::ZERO, |s| { this.clock.now().saturating_duration_since(s) @@ -253,12 +340,21 @@ where } } } - }, + } // Backing off — wait, then start the next attempt. RetryStateProj::Sleeping { delay } => match delay.poll(cx) { Poll::Pending => return Poll::Pending, - Poll::Ready(()) => RetryState::Attempting { fut: (this.op)() }, + Poll::Ready(()) => { + let deadline = this + .attempt_timeout + .as_ref() + .map(|(d, _)| this.clock.sleep(*d)); + RetryState::Attempting { + fut: (this.op)(), + deadline, + } + } }, }; this.state.set(next); @@ -300,16 +396,43 @@ mod tests { } } impl Clock for MockClock { - type Sleep = std::future::Ready<()>; + type Sleep = MockSleep; fn now(&self) -> Instant { self.now_calls.fetch_add(1, SeqCst); self.start + *self.elapsed.lock().unwrap() } - fn sleep(&self, dur: Duration) -> std::future::Ready<()> { - self.log.lock().unwrap().push(dur); - *self.elapsed.lock().unwrap() += dur; - std::future::ready(()) + fn sleep(&self, dur: Duration) -> MockSleep { + MockSleep { + dur, + clock: self.clone(), + fired: false, + } + } + } + + /// A sleep that only advances the virtual clock when it is actually polled. + /// + /// Creating a `tokio::time::Sleep` and dropping it unpolled costs nothing, so the mock has to + /// behave the same way. It used to record on creation, which was fine while retry only ever + /// built a sleep it intended to await. `attempt_timeout` arms a deadline that a fast operation + /// never polls, and that made the difference visible. + struct MockSleep { + dur: Duration, + clock: MockClock, + fired: bool, + } + + impl Future for MockSleep { + type Output = (); + fn poll(mut self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<()> { + if !self.fired { + self.fired = true; + let dur = self.dur; + self.clock.log.lock().unwrap().push(dur); + *self.clock.elapsed.lock().unwrap() += dur; + } + Poll::Ready(()) } } @@ -826,6 +949,71 @@ mod tests { ); } + #[tokio::test] + async fn attempt_timeout_bounds_an_operation_that_hangs() { + // The bug this exists for: `max_elapsed` is only consulted between attempts, so on its own + // it never stops a future that is simply never ready. A dead TCP peer looks exactly like + // this. Runs on the mock clock, so it is instant and exact. + let clock = MockClock::new(); + let out: Result> = retry(std::future::pending::>) + .backoff(backoff(2)) + .attempt_timeout(secs(5), || "timed out") + .clock(clock.clone()) + .await; + + let err = out.unwrap_err(); + assert_eq!(*err.error(), "timed out"); + assert_eq!(err.attempts(), 3); // the initial attempt plus 2 retries, each timed out + assert_eq!(err.stop_reason(), StopReason::RetriesExhausted); + // 3 attempts x 5s of waiting, plus the 1s and 2s backoff sleeps between them. + assert_eq!( + clock.slept(), + vec![secs(5), secs(1), secs(5), secs(2), secs(5)] + ); + } + + #[tokio::test] + async fn a_fast_operation_never_sees_the_timeout() { + // The deadline must not fire for work that finishes, and must not cost a sleep. + let clock = MockClock::new(); + let out: Result> = retry(|| async { Ok(7) }) + .attempt_timeout(secs(5), || "timed out") + .clock(clock.clone()) + .await; + assert_eq!(out.unwrap(), 7); + assert!(clock.slept().is_empty()); + } + + #[tokio::test] + async fn a_returned_error_wins_over_an_expired_deadline() { + // If the operation is ready in the same poll the deadline is up, the real result wins. + // Otherwise a race would throw away a call that actually completed. + let clock = MockClock::new(); + let out: Result> = retry(|| async { Err("real error") }) + .backoff(backoff(1)) + .attempt_timeout(Duration::ZERO, || "timed out") + .clock(clock.clone()) + .await; + assert_eq!(*out.unwrap_err().error(), "real error"); + } + + #[tokio::test] + async fn attempt_timeout_feeds_the_when_predicate() { + // A timed-out attempt travels the same path as a returned error, so `.when` can refuse it. + let clock = MockClock::new(); + let out: Result> = retry(std::future::pending::>) + .backoff(backoff(5)) + .attempt_timeout(secs(5), || "timed out") + .when(|e: &&str| *e != "timed out") + .clock(clock.clone()) + .await; + + let err = out.unwrap_err(); + assert_eq!(err.stop_reason(), StopReason::NotRetryable); + assert_eq!(err.attempts(), 1); + assert_eq!(clock.slept(), vec![secs(5)]); // one timeout, then it gave up + } + #[tokio::test] async fn accepts_a_borrowed_or_shared_clock() { // `.clock(c)` takes the clock by value, so without the reference impls a test could hand