srijs/rust-tokio-retry

Extensible, asynchronous retry behaviours for futures/tokio

tokio-retry

Extensible, asynchronous retry behaviours for the ecosystem of tokio libraries.

Documentation

Installation

Add this to your Cargo.toml:

[dependencies]
tokio-retry = "0.3"

Examples

use tokio_retry::Retry;
use tokio_retry::strategy::{ExponentialBackoff, jitter};

async fn action() -> Result<u64, ()> {
    // do some real-world stuff here...
    Err(())
}

#[tokio::main]
async fn main() -> Result<(), ()> {
    let retry_strategy = ExponentialBackoff::from_millis(10)
        .map(jitter) // add jitter to delays
        .take(3);    // limit to 3 retries

    let result = Retry::spawn(retry_strategy, action).await?;

    Ok(())
}

rust-tokio-retry's Issues

Is it possible to have an FnOnce action closure

In the code there is a trait bound that suggests that the action closure needs to be na FnMut. Is there a reason an FnOnce could not be used instead?

Doc at https://docs.rs/tokio-retry are out of date

The example at https://docs.rs/tokio-retry/0.2.0/tokio_retry/index.html show a code example which does not compile.
The "then" function does not exist on type retry.

let future = Retry::spawn(retry_strategy, action).then(|result| {
    println!("result {:?}", result);
    Ok(())
});

tokio::time::Interval based delay?

Tokio has the following structure: https://docs.rs/tokio/1.5.0/tokio/time/struct.Interval.html

It allows starting a timer, then doing some other work, and then sleeping the rest of the duration until that timer expires.

Having an option to use this instead of the existing sleep-based implementation would allow for the retry logic to have the semantics of "This future will perform its action (in the best case) every duration" instead of "this future will perform its action every duration + the time it takes to execute the action".

I would like it if this library supported this by adding IntervalRetry and IntervalRetryIf Future wrappers.

Ability to conditionally retry for certain error conditions

For example, if an HTTP API returns 503, you might retry. If it returns 401, you might not.

use log::{debug, warn};
use reqwest::{
    header::{HeaderMap, HeaderName, HeaderValue},
    Method,
};
use std::str::FromStr;

lazy_static::lazy_static! {
  static ref HTTP_CLIENT: reqwest::Client = reqwest::Client::new();
}

pub async fn http_request<T: for<'de> serde::Deserialize<'de>>(method_str: &str, url: &str, headers: &Vec<(String, String)>, payload: &Option<String>) -> Result<T, String> {
    debug!("http_get: url = {}", url);
    let mut headers_map = HeaderMap::new();
    for (key, value) in headers {
        headers_map.insert(HeaderName::from_str(key).unwrap(), HeaderValue::from_str(value).unwrap());
    }
    let method = Method::from_bytes(method_str.as_bytes()).unwrap();
    let request = if payload.is_some() {
        let payload = payload.as_ref().unwrap();
        HTTP_CLIENT.request(method, url).headers(headers_map).body(payload.clone())
    } else {
        HTTP_CLIENT.request(method, url).headers(headers_map)
    };
    let response = request.send().await;
    if response.is_err() {
        return Err(format!("{}", response.err().unwrap()));
    }
    let response = response.unwrap();
    let response_status = response.status().as_u16();
    let is_2xx = response_status >= 200 && response_status <= 299;
    if is_2xx == false {
        // TODO: how to not retry 4xx vs 5xx errors?
        return Err(format!("invalid response status: {}", response.status().as_u16()));
    }
    let stringified_response_body = response.text().await.unwrap();
    debug!("stringified_response_body = {}", stringified_response_body);
    let response_body: T = if stringified_response_body.len() > 0 {
        serde_json::from_str(&stringified_response_body).unwrap()
    } else {
        serde_json::from_str("null").unwrap()
    };
    return Ok(response_body);
}

pub async fn http_request_with_timeout<T: for<'de> serde::Deserialize<'de>>(method_str: &str, url: &str, headers: &Vec<(String, String)>, payload: &Option<String>) -> Result<T, String> {
    let timeout_seconds = 10;
    let request_future = http_request::<T>(method_str, url, headers, payload);
    let timeout_future = tokio::time::timeout(tokio::time::Duration::from_secs(timeout_seconds), request_future).await;
    if timeout_future.is_err() {
        return Err(String::from("timed out"));
    }
    let response = timeout_future.unwrap();
    return response;
}

pub async fn http_request_with_timeout_and_retries<T: for<'de> serde::Deserialize<'de>>(method_str: &str, url: &str, headers: &Vec<(String, String)>, payload: &Option<String>) -> T {
    let known_errors = vec![String::from("timed out"), String::from("invalid response status: 503")];
    let retry_interval_ms = 1000;
    let num_retries = 10;
    for attempt in 0..num_retries {
        let result = http_request_with_timeout::<T>(method_str, url, &headers, &payload).await;
        let is_success = result.is_ok();
        if is_success {
            return result.unwrap();
        }
        let error_message = result.err().unwrap();
        let is_error_known = known_errors.iter().cloned().position(|known_error| {
            return known_error == error_message;
        });
        if is_error_known.is_none() {
            panic!("unknown error: {}", error_message);
        }
        warn!("retry # {} / {}: {}", attempt, num_retries, error_message);
        tokio::time::sleep(tokio::time::Duration::from_millis(retry_interval_ms)).await;
    }
    panic!("request failed after {} retries", retry_interval_ms);
}

I currently have that but I'm trying to think of how to add logic for 5xx retries but not 4xx...

Support for the new tokio crate

At the moment the tokio-retry is based on top of tokio-core. But it has a modern replacement tokio

What do you think about porting crate to tokio?

Reset strategy when the action successfully poll?

From the code, it seems that this crate would push forward the strategy wait time regardless of whether the service has restored. If a service just intermittently fail, this doesn't feel like the right thing to do.

Maybe we can make strategy also a function, and reset it whenever the action part gets a successful poll?

Support for FnOnce closures

Examples breaks if action closure capture some data from outside. Please fix.

Panic in exponential backoff

When trying to use the tokio-retry with an exponential backoff, I got this panic:

thread 'main' panicked at 'attempt to multiply with overflow', /home/vorner/.cargo/registry/src/github.com-1ecc6299db9ec823/tokio-retry-0.0.5/src/strategies/exponential_backoff.rs:27
stack backtrace:
   0:     0x55adfc435fb3 - std::sys::imp::backtrace::tracing::imp::unwind_backtrace::h0c49f46a3545f908
                               at /checkout/src/libstd/sys/unix/backtrace/tracing/gcc_s.rs:49
   1:     0x55adfc4321a4 - std::sys_common::backtrace::_print::hcef39a9816714c4c
                               at /checkout/src/libstd/sys_common/backtrace.rs:71
   2:     0x55adfc439687 - std::panicking::default_hook::{{closure}}::h7c3c94835e02f846
                               at /checkout/src/libstd/sys_common/backtrace.rs:60
                               at /checkout/src/libstd/panicking.rs:355
   3:     0x55adfc43920b - std::panicking::default_hook::h0bf7bc3112fb107d
                               at /checkout/src/libstd/panicking.rs:371
   4:     0x55adfc439b5b - std::panicking::rust_panic_with_hook::ha27630c950090fec
                               at /checkout/src/libstd/panicking.rs:549
   5:     0x55adfc439a34 - std::panicking::begin_panic::heb97fa3158b71158
                               at /checkout/src/libstd/panicking.rs:511
   6:     0x55adfc439969 - std::panicking::begin_panic_fmt::h8144403278d84748
                               at /checkout/src/libstd/panicking.rs:495
   7:     0x55adfc4398f7 - rust_begin_unwind
                               at /checkout/src/libstd/panicking.rs:471
   8:     0x55adfc46444d - core::panicking::panic_fmt::h3b0cca53e68f9654
                               at /checkout/src/libcore/panicking.rs:69
   9:     0x55adfc464384 - core::panicking::panic::h4b991f5abe7d76d5
                               at /checkout/src/libcore/panicking.rs:49
  10:     0x55adfc07a795 - <tokio_retry::strategies::exponential_backoff::ExponentialBackoff as tokio_retry::strategy::RetryStrategy>::delay::ha8fcba7781f4e038
                               at /home/vorner/.cargo/registry/src/github.com-1ecc6299db9ec823/tokio-retry-0.0.5/src/strategies/exponential_backoff.rs:27
  11:     0x55adfc062a3b - <tokio_retry::strategy::limited_delay::LimitedDelay<S> as tokio_retry::strategy::RetryStrategy>::delay::h85a101e68b8e7153
                               at /home/vorner/.cargo/registry/src/github.com-1ecc6299db9ec823/tokio-retry-0.0.5/src/strategy/limited_delay.rs:19
  12:     0x55adfc061feb - <tokio_retry::strategy::jittered::Jittered<S> as tokio_retry::strategy::RetryStrategy>::delay::h541a52b8cb0cb7d3
                               at /home/vorner/.cargo/registry/src/github.com-1ecc6299db9ec823/tokio-retry-0.0.5/src/strategy/jittered.rs:18
  13:     0x55adfc0611b2 - <tokio_retry::future::RetryFuture<S, R, A, F>>::retry::h7f6c44d00636ed8f
                               at /home/vorner/.cargo/registry/src/github.com-1ecc6299db9ec823/tokio-retry-0.0.5/src/future.rs:114
  14:     0x55adfc05eb84 - <tokio_retry::future::RetryFuture<S, R, A, F> as futures::future::Future>::poll::h85617c13acb1b65f
                               at /home/vorner/.cargo/registry/src/github.com-1ecc6299db9ec823/tokio-retry-0.0.5/src/future.rs:140
  15:     0x55adfc0619c5 - <tokio_retry::future::RetryFuture<S, R, A, F>>::attempt::h624df1b195bdaf8e
                               at /home/vorner/.cargo/registry/src/github.com-1ecc6299db9ec823/tokio-retry-0.0.5/src/future.rs:110
  16:     0x55adfc05ecdb - <tokio_retry::future::RetryFuture<S, R, A, F> as futures::future::Future>::poll::h85617c13acb1b65f
                               at /home/vorner/.cargo/registry/src/github.com-1ecc6299db9ec823/tokio-retry-0.0.5/src/future.rs:144
  17:     0x55adfc01d5f4 - <futures::future::chain::Chain<A, B, C>>::poll::h8454fca0630cf7d1
                               at /home/vorner/.cargo/registry/src/github.com-1ecc6299db9ec823/futures-0.1.11/src/future/chain.rs:26
  18:     0x55adfc01880b - <futures::future::and_then::AndThen<A, B, F> as futures::future::Future>::poll::h47c3e4cf7ab9eb7f
                               at /home/vorner/.cargo/registry/src/github.com-1ecc6299db9ec823/futures-0.1.11/src/future/and_then.rs:32
  19:     0x55adfc022837 - <futures::future::map_err::MapErr<A, F> as futures::future::Future>::poll::hadffce42469c94a8
                               at /home/vorner/.cargo/registry/src/github.com-1ecc6299db9ec823/futures-0.1.11/src/future/map_err.rs:30
  20:     0x55adfc063c50 - <alloc::boxed::Box<F> as futures::future::Future>::poll::hcb56c6b2b4bf35dc
                               at /home/vorner/.cargo/registry/src/github.com-1ecc6299db9ec823/futures-0.1.11/src/future/mod.rs:106
  21:     0x55adfc01e954 - <futures::future::chain::Chain<A, B, C>>::poll::hdf328769d85559d3
                               at /home/vorner/.cargo/registry/src/github.com-1ecc6299db9ec823/futures-0.1.11/src/future/chain.rs:26
  22:     0x55adfc02633b - <futures::future::then::Then<A, B, F> as futures::future::Future>::poll::hc78fb7925a0ee080
                               at /home/vorner/.cargo/registry/src/github.com-1ecc6299db9ec823/futures-0.1.11/src/future/then.rs:32
  23:     0x55adfc021953 - <futures::future::loop_fn::LoopFn<A, F> as futures::future::Future>::poll::h19adbba2059d6edc
                               at /home/vorner/.cargo/registry/src/github.com-1ecc6299db9ec823/futures-0.1.11/src/future/loop_fn.rs:93
  24:     0x55adfc063c50 - <alloc::boxed::Box<F> as futures::future::Future>::poll::hcb56c6b2b4bf35dc
                               at /home/vorner/.cargo/registry/src/github.com-1ecc6299db9ec823/futures-0.1.11/src/future/mod.rs:106
  25:     0x55adfc068b6b - <futures::task_impl::Spawn<F>>::poll_future::{{closure}}::hec8f06d0ce4e9c0b
                               at /home/vorner/.cargo/registry/src/github.com-1ecc6299db9ec823/futures-0.1.11/src/task_impl/mod.rs:337
  26:     0x55adfc068bbd - <futures::task_impl::Spawn<T>>::enter::{{closure}}::h0149d9c8897c3a25
                               at /home/vorner/.cargo/registry/src/github.com-1ecc6299db9ec823/futures-0.1.11/src/task_impl/mod.rs:484
  27:     0x55adfc069602 - futures::task_impl::set::{{closure}}::ha611c92aba82b0d8
                               at /home/vorner/.cargo/registry/src/github.com-1ecc6299db9ec823/futures-0.1.11/src/task_impl/mod.rs:61
  28:     0x55adfc068f83 - <std::thread::local::LocalKey<T>>::with::h16340418cfe8a2a9
                               at /checkout/src/libstd/thread/local.rs:253
  29:     0x55adfc065c1e - futures::task_impl::set::h2e629adaef00391b
                               at /home/vorner/.cargo/registry/src/github.com-1ecc6299db9ec823/futures-0.1.11/src/task_impl/mod.rs:54
  30:     0x55adfc065463 - <futures::task_impl::Spawn<T>>::enter::h2e3d49fb3e877e29
                               at /home/vorner/.cargo/registry/src/github.com-1ecc6299db9ec823/futures-0.1.11/src/task_impl/mod.rs:484
  31:     0x55adfc065316 - <futures::task_impl::Spawn<F>>::poll_future::h52e6238f54e295d3
                               at /home/vorner/.cargo/registry/src/github.com-1ecc6299db9ec823/futures-0.1.11/src/task_impl/mod.rs:337
  32:     0x55adfc065e06 - tokio_core::reactor::Core::run::{{closure}}::h4841042cd993ea91
                               at /home/vorner/.cargo/registry/src/github.com-1ecc6299db9ec823/tokio-core-0.1.6/src/reactor/mod.rs:243
  33:     0x55adfc065f90 - <scoped_tls::ScopedKey<T>>::set::h982bddc3d783092f
                               at /home/vorner/.cargo/registry/src/github.com-1ecc6299db9ec823/scoped-tls-0.1.0/src/lib.rs:135
  34:     0x55adfc05dbcb - tokio_core::reactor::Core::run::h14de121c50c09099
                               at /home/vorner/.cargo/registry/src/github.com-1ecc6299db9ec823/tokio-core-0.1.6/src/reactor/mod.rs:242
  35:     0x55adfbff0f52 - aggregator::reactor::run::h003ce2f24890f7da
                               at /home/vorner/work/pakon/aggregator/src/reactor.rs:112
  36:     0x55adfbfe76f2 - pakon_aggregator::main::hb79cf260b59893f8
                               at /home/vorner/work/pakon/aggregator/src/main.rs:37
  37:     0x55adfc439855 - std::panicking::try::do_call::h689a21caeeef92aa
                               at /checkout/src/libcore/ops.rs:2606
                               at /checkout/src/libstd/panicking.rs:454
  38:     0x55adfc440a1a - __rust_maybe_catch_panic
                               at /checkout/src/libpanic_unwind/lib.rs:98
  39:     0x55adfc43a2fa - std::rt::lang_start::hf63d494cb7dd034c
                               at /checkout/src/libstd/panicking.rs:433
                               at /checkout/src/libstd/panic.rs:361
                               at /checkout/src/libstd/rt.rs:57
  40:     0x55adfbfe7b62 - main
  41:     0x7f09505391ff - __libc_start_main
  42:     0x55adfbf7dd29 - _start
  43:                0x0 - <unknown>

The retry strategy is created like this and let run for a while. Currently the action always fails:

    let retry = ExponentialBackoff::from_millis(10)
        .limit_delay(Duration::from_millis(5000))
        .jitter();

Please include an example of retrying something other than a function taking no parameters

I have an async operation that I'd like to retry that requires input parameters. Async closures are unstable, and an async block returns a dyn Future, not an FnMut(). Implementing Action directly is difficult because I have to name the type implementing Future returned by my method.
Is it possible to use this crate on stable if my operation requires parameters? If so, how?

Any plans for tokio 0.3 support?

Add support for tokio::timer::timeout::Timeout?

It looks like tokio_timer is deprecated. Is it possible to add support for tokio::timer::timout::Timeout instead?

TimerError when using tokio-retry 0.2 with old tokio-core crate

Hi,

When running the second example (for the old tokio_core crate) in the documentation (https://docs.rs/tokio-retry/0.2.0/tokio_retry/#using-the-tokio_core-crate) I get a TimerError and no retries:

result Err(TimerError(Error(Shutdown)))

I don't really understand enough about tokio-timer to understand the error. Is there a step missing from the example?

exponential backoff is too exponential

I found this counterintuitive:

#[test]
fn backoff() {
    let x: Vec<std::time::Duration> = tokio_retry::strategy::ExponentialBackoff::from_millis(50).take(5).collect();
    dbg!(x);
}

x = [
    50ms,
    2.5s,
    125s,
    6250s,
    312500s,
]

Looking at the source code I would have expected the calculation to be base * factor^retry but it seems that it's base^retry * factor. So to use this like I want it seems that I need to supply from_millis(2).factor(25).

Just swapping this would probably be a confusing breakage, perhaps from_millis could be replaced with from_base? Then it would probably have made me realize that I was doing something wrong.

limit_retries with ExponentialBackoff

Is it still possible to use limit_retries with ExponentialBackoff?

I would like an exponential back off but only for a couple of retries.

Immediate retry?

Hello,
Just checked sources - looks like Delay is the only option used for retry.

Are there any way to produce "Immediate" retry?

Thank you,