Repository: tower-rs/tower
Pull request #584 (2021-04-30 20:35:12)

improve flexiblity of retry policy
`retry::Policy` is an effective way to expressing retries, however, there two use cases that
as it stands today, cannot be expressed:
- Altering the final response (eg. to record the fact that you ran out of retries)
- Altering the request (eg. to set a header to the server indicating that this request is a retry)

(Technically the second is possible with `clone_request`, but it's a little unclear _which_ request would actually get sent).

This change implements what I think is pretty close to the minimal update to make this possible, namely, `req`
and `Res` both become mutable references. This enables policies to mutate them during execution & enables both of the
use cases above without complicating the "simple path" callers who don't need this behavior.

**This is a breaking change.** However, the fixes are only a couple of `&mut` and potentially a call to `as_ref()`.

I've also considered (and would be totally happy) with an API that "mapped" over req & res, taking and returning ownership. This may actually even be slightly more flexible. However, I didn't want to leak this detail onto all users.

rcoh: The entire point is to change the variant (at least for me). Because there
are "success" cases the actually need to be retries and when I run out of
retries I need to change it to failure

On Sun, May 2, 2021, 8:13 AM David Pedersen ***@***.***>
wrote:

> ***@***.**** commented on this pull request.
>
> I have never used this middleware in practice but to me the change looks
> sensible. Though I'm wondering if someone else with more experience using
> Retry has run into other problems that could be solved, should we decide
> to make a breaking change. cc @hawkw <https://github.com/hawkw> @olix0r
> <https://github.com/olix0r> @LucioFranco <https://github.com/LucioFranco>
> @seanmonstar <https://github.com/seanmonstar>
>
> Had one small suggestion/question.
>
> I've also considered (and would be totally happy) with an API that
> "mapped" over req & res, taking and returning ownership. This may actually
> even be slightly more flexible. However, I didn't want to leak this detail
> onto all users.
>
> To return a new request/response the future returned by Policy::retry
> would have to change as well which might be too much of a change.
> ------------------------------
>
> In tower/src/retry/policy.rs
> <https://github.com/tower-rs/tower/pull/584#discussion_r624684352>:
>
> > @@ -58,7 +58,7 @@ pub trait Policy<Req, Res, E>: Sized {
>      ///
>      /// [`Service::Response`]: crate::Service::Response
>      /// [`Service::Error`]: crate::Service::Error
> -    fn retry(&self, req: &Req, result: Result<&Res, &E>) -> Option<Self::Future>;
> +    fn retry(&self, req: &mut Req, result: &mut Result<Res, E>) -> Option<Self::Future>;
>
> I guess getting a &mut Result<Res, E> allows users to change Err(_) into
> Ok(_) which might be kinda odd. Not sure. Feels like getting Result<&mut
> Res, &mut E> would fit the use-cases as well but not allow changing the
> Result variant.
>
> —
> You are receiving this because you authored the thread.
> Reply to this email directly, view it on GitHub
> <https://github.com/tower-rs/tower/pull/584#pullrequestreview-649909776>,
> or unsubscribe
> <https://github.com/notifications/unsubscribe-auth/AADYKZZKITAKKYCN6H47VJ3TLU6YBANCNFSM435NQWHQ>
> .
>

davidpdrsn: Alright cool. That does make sense. 

I kinda feel like if we are going to make a breaking change we should take care to design something that we are happy with. That might be this or it might be bigger changes. I haven't used Retry personally so I don't know which it is. There was some discussion about other potential changes to Retry [here](https://github.com/tower-rs/tower/pull/414).
LucioFranco: @rcoh maybe I am forgetting how this works but can't you return an error in the returned service future? 
hawkw: i actually _just_ ran into a use-case for a mutable request in `clone_request`. i was trying to implement a retry policy where an HTTP request body is buffered lazily as the first request's body is sent, rather than aggregating the whole body prior to sending the initial request.

the approach i was going to take was storing the buffered body on the initial request, and then sending the buffer on a `oneshot` channel when the first request is dropped so that the potential retry can take ownership of it. however, there isn't currently any way to get the receiver side of the oneshot _out_ of the initial request, without introducing a mutex or something for this purpose. making the request mutable in `clone_request` would very make this kind of thing much easier, so i think we should definitely make this change!
rcoh: @hawkw just to clarify, in this change the request becomes mutable in the `retry` method; I thought it was a little unclear from a user perspective _which_ request was which if you were to mutate it in `clone_request()` (basically, is the cloned request sent first or is the original sent first?)

Which is why I made it mutatable in `retry` because then it seemed more obvious that the request you just mutated would get sent next. I _think_ that should be equivalent with your try_clone use case, but we could also make it mutable in both cases?
hawkw: > @hawkw just to clarify, in this change the request becomes mutable in the `retry` method; I thought it was a little unclear from a user perspective _which_ request was which if you were to mutate it in `clone_request()` (basically, is the cloned request sent first or is the original sent first?)
> 
> Which is why I made it mutatable in `retry` because then it seemed more obvious that the request you just mutated would get sent next. I _think_ that should be equivalent with your try_clone use case, but we could also make it mutable in both cases?

oh, i misread the PR description, whoops! i'm going to just go ahead and try to implement a lazily buffered request body against this branch and report back? i think it may still be possible without _also_ touching `clone_request`, but we'll see...

another option would be to make the `Policy` itself mutable in both `clone_request()` and `retry()`, which would also give you an avenue for sharing state between the first and second request...
hawkw: It would be really nice to get this change into 0.5 (which we're currently working on). @rcoh, are you interested in updating this branch to build with the latest `master`, or should I?
rcoh: I can do the merge tomorrow assuming there's nothing too problematic
rcoh: merge is done—I assume you'll squash merge, but if not let me know and I can cleanup history
hawkw: > I assume you'll squash merge, but if not let me know and I can cleanup history

Yeah, that's right, we generally merge PRs by squashing, so you don't have to worry about it! :)

Patch:
diff --git a/tower/CHANGELOG.md b/tower/CHANGELOG.md
index 5bdc7e3..63263f8 100644
--- a/tower/CHANGELOG.md
+++ b/tower/CHANGELOG.md
@@ -11,9 +11,12 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 - **util**: Removed deprecated `ServiceExt::ready_and` method and `ReadyAnd`
   future ([#652])
+- **retry**: **Breaking Change** `retry::Policy::retry` now accepts `&mut Req` and `&mut Res` instead of the previous mutable versions. This
+  increases the flexibility of the retry policy. To update, update your method signature to include `mut` for both parameters. ([#584])
 
 
 [#652]: https://github.com/tower-rs/tower/pull/652
+[#584]: https://github.com/tower-rs/tower/pull/584
 
 # 0.4.12 (February 16, 2022)
 
diff --git a/tower/src/retry/future.rs b/tower/src/retry/future.rs
index d18a5ab..c23e6e3 100644
--- a/tower/src/retry/future.rs
+++ b/tower/src/retry/future.rs
@@ -74,9 +74,9 @@ where
         loop {
             match this.state.as_mut().project() {
                 StateProj::Called { future } => {
-                    let result = ready!(future.poll(cx));
-                    if let Some(ref req) = this.request {
-                        match this.retry.policy.retry(req, result.as_ref()) {
+                    let mut result = ready!(future.poll(cx));
+                    if let Some(req) = &mut this.request {
+                        match this.retry.policy.retry(req, &mut result) {
                             Some(checking) => {
                                 this.state.set(State::Checking { checking });
                             }
diff --git a/tower/src/retry/policy.rs b/tower/src/retry/policy.rs
index 6f8a5b7..20adc3d 100644
--- a/tower/src/retry/policy.rs
+++ b/tower/src/retry/policy.rs
@@ -16,7 +16,7 @@ use std::future::Future;
 /// impl<E> Policy<Req, Res, E> for Attempts {
 ///     type Future = future::Ready<Self>;
 ///
-///     fn retry(&self, req: &Req, result: Result<&Res, &E>) -> Option<Self::Future> {
+///     fn retry(&self, req: &mut Req, result: &mut Result<Res, E>) -> Option<Self::Future> {
 ///         match result {
 ///             Ok(_) => {
 ///                 // Treat all `Response`s as success,
@@ -56,9 +56,28 @@ pub trait Policy<Req, Res, E>: Sized {
     /// If the request *should* be retried, return `Some` future of a new
     /// policy that would apply for the next request attempt.
     ///
+    /// ## Mutating Requests
+    ///
+    /// The policy MAY chose to mutate the `req`: if the request is mutated, the
+    /// mutated request will be sent to the inner service in the next retry.
+    /// This can be helpful for use cases like tracking the retry count in a
+    /// header.
+    ///
+    /// ## Mutating Results
+    ///
+    /// The policy MAY chose to mutate the result. This enables the retry
+    /// policy to convert a failure into a success and vice versa. For example,
+    /// if the policy is used to poll while waiting for a state change, the
+    /// policy can switch the result to emit a specific error when retries are
+    /// exhausted.
+    ///
+    /// The policy can also record metadata on the request to include
+    /// information about the number of retries required or to record that a
+    /// failure failed after exhausting all retries.
+    ///
     /// [`Service::Response`]: crate::Service::Response
     /// [`Service::Error`]: crate::Service::Error
-    fn retry(&self, req: &Req, result: Result<&Res, &E>) -> Option<Self::Future>;
+    fn retry(&self, req: &mut Req, result: &mut Result<Res, E>) -> Option<Self::Future>;
 
     /// Tries to clone a request before being passed to the inner service.
     ///


Test patch:
diff --git a/tower/tests/builder.rs b/tower/tests/builder.rs
index 8bc8ce1..90bb70a 100644
--- a/tower/tests/builder.rs
+++ b/tower/tests/builder.rs
@@ -45,7 +45,7 @@ where
 {
     type Future = Ready<Self>;
 
-    fn retry(&self, _req: &Req, _result: Result<&Res, &E>) -> Option<Self::Future> {
+    fn retry(&self, _req: &mut Req, _result: &mut Result<Res, E>) -> Option<Self::Future> {
         None
     }
 
diff --git a/tower/tests/retry/main.rs b/tower/tests/retry/main.rs
index a07242b..d0d7a8a 100644
--- a/tower/tests/retry/main.rs
+++ b/tower/tests/retry/main.rs
@@ -90,6 +90,27 @@ async fn success_with_cannot_clone() {
     assert_ready_ok!(fut.poll(), "world");
 }
 
+#[tokio::test(flavor = "current_thread")]
+async fn retry_mutating_policy() {
+    let _t = support::trace_init();
+
+    let (mut service, mut handle) = new_service(MutatingPolicy { remaining: 2 });
+
+    assert_ready_ok!(service.poll_ready());
+    let mut fut = task::spawn(service.call("hello"));
+
+    assert_request_eq!(handle, "hello").send_response("world");
+    assert_pending!(fut.poll());
+    // the policy alters the request. in real life, this might be setting a header
+    assert_request_eq!(handle, "retrying").send_response("world");
+
+    assert_pending!(fut.poll());
+
+    assert_request_eq!(handle, "retrying").send_response("world");
+
+    assert_ready_err!(fut.poll(), "out of retries");
+}
+
 type Req = &'static str;
 type Res = &'static str;
 type InnerError = &'static str;
@@ -102,7 +123,7 @@ struct RetryErrors;
 
 impl Policy<Req, Res, Error> for RetryErrors {
     type Future = future::Ready<Self>;
-    fn retry(&self, _: &Req, result: Result<&Res, &Error>) -> Option<Self::Future> {
+    fn retry(&self, _: &mut Req, result: &mut Result<Res, Error>) -> Option<Self::Future> {
         if result.is_err() {
             Some(future::ready(RetryErrors))
         } else {
@@ -120,7 +141,7 @@ struct Limit(usize);
 
 impl Policy<Req, Res, Error> for Limit {
     type Future = future::Ready<Self>;
-    fn retry(&self, _: &Req, result: Result<&Res, &Error>) -> Option<Self::Future> {
+    fn retry(&self, _: &mut Req, result: &mut Result<Res, Error>) -> Option<Self::Future> {
         if result.is_err() && self.0 > 0 {
             Some(future::ready(Limit(self.0 - 1)))
         } else {
@@ -138,8 +159,8 @@ struct UnlessErr(InnerError);
 
 impl Policy<Req, Res, Error> for UnlessErr {
     type Future = future::Ready<Self>;
-    fn retry(&self, _: &Req, result: Result<&Res, &Error>) -> Option<Self::Future> {
-        result.err().and_then(|err| {
+    fn retry(&self, _: &mut Req, result: &mut Result<Res, Error>) -> Option<Self::Future> {
+        result.as_ref().err().and_then(|err| {
             if err.to_string() != self.0 {
                 Some(future::ready(self.clone()))
             } else {
@@ -158,7 +179,7 @@ struct CannotClone;
 
 impl Policy<Req, Res, Error> for CannotClone {
     type Future = future::Ready<Self>;
-    fn retry(&self, _: &Req, _: Result<&Res, &Error>) -> Option<Self::Future> {
+    fn retry(&self, _: &mut Req, _: &mut Result<Res, Error>) -> Option<Self::Future> {
         unreachable!("retry cannot be called since request isn't cloned");
     }
 
@@ -167,6 +188,36 @@ impl Policy<Req, Res, Error> for CannotClone {
     }
 }
 
+/// Test policy that changes the request to `retrying` during retries and the result to `"out of retries"`
+/// when retries are exhausted.
+#[derive(Clone)]
+struct MutatingPolicy {
+    remaining: usize,
+}
+
+impl Policy<Req, Res, Error> for MutatingPolicy
+where
+    Error: From<&'static str>,
+{
+    type Future = future::Ready<Self>;
+
+    fn retry(&self, req: &mut Req, result: &mut Result<Res, Error>) -> Option<Self::Future> {
+        if self.remaining == 0 {
+            *result = Err("out of retries".into());
+            None
+        } else {
+            *req = "retrying";
+            Some(future::ready(MutatingPolicy {
+                remaining: self.remaining - 1,
+            }))
+        }
+    }
+
+    fn clone_request(&self, req: &Req) -> Option<Req> {
+        Some(*req)
+    }
+}
+
 fn new_service<P: Policy<Req, Res, Error> + Clone>(
     policy: P,
 ) -> (mock::Spawn<tower::retry::Retry<P, Mock>>, Handle) {