Repository: tower-rs/tower
Pull request #681 (2022-08-03 22:48:40)

retry: Change `Policy` to accept `&mut self`
This changes the `Policy` trait in the `retry` layer to accept `&mut
self` instead of `&self` and changes the output type of the returned
future to `()`. The motivation for this change is to simplify
the trait a bit. By the trait methods having mutable references it means
that for each retry request session you can mutate the local policy.
This is because the policy is cloned for each individual request that
arrives into the retry middleware. In addition, this allows the `Policy`
trait to be object safe.

cc @rcoh 

LucioFranco: cc @olix0r 
olix0r: I'm a little unclear on this:

> By the trait methods having mutable references it means
that for each retry request session you can mutate the local policy.
This is because the policy is cloned for each individual request that
arrives into the retry middleware

So, we clone the policy for each request AND it is mutable? How does that work? If it's cloned for each request, we'd need an `Arc<Mutex<..>>` for state to be held across all clones? Or is the mutability only intended to support mutating a single cloned replica?
LucioFranco: @olix0r By each request I mean request session, maybe that is a better way to word it. Aka we only clone the policy once iirc. So if you want data to be shared across all instances of the retry middleware then you'd need to `Arc<Mutex<..>`.
LucioFranco: > the one thing I might consider is making the returned type T=() I'm thinking about token bucket use cases where the policy might want to take and then return a token or something...but it's not immediately obvious to me how that would get wired

@rcoh I would imagine you can do anything with` &mut self`? and `Arc` if you need it shared.
LucioFranco: Ok this is ready for another round of reviews, CI is only failing due to a rustc beta/nightly bug.

Patch:
diff --git a/tower/src/retry/budget.rs b/tower/src/retry/budget.rs
index 468c4ca..35f7fa7 100644
--- a/tower/src/retry/budget.rs
+++ b/tower/src/retry/budget.rs
@@ -41,9 +41,9 @@
 //! }
 //!
 //! impl<E> Policy<Req, Res, E> for RetryPolicy {
-//!     type Future = future::Ready<Self>;
+//!     type Future = future::Ready<()>;
 //!
-//!     fn retry(&self, req: &mut Req, result: &mut Result<Res, E>) -> Option<Self::Future> {
+//!     fn retry(&mut self, req: &mut Req, result: &mut Result<Res, E>) -> Option<Self::Future> {
 //!         match result {
 //!             Ok(_) => {
 //!                 // Treat all `Response`s as success,
@@ -60,12 +60,12 @@
 //!                 }
 //!
 //!                 // Try again!
-//!                 Some(future::ready(self.clone()))
+//!                 Some(future::ready(()))
 //!             }
 //!         }
 //!     }
 //!
-//!     fn clone_request(&self, req: &Req) -> Option<Req> {
+//!     fn clone_request(&mut self, req: &Req) -> Option<Req> {
 //!         Some(req.clone())
 //!     }
 //! }
diff --git a/tower/src/retry/future.rs b/tower/src/retry/future.rs
index c23e6e3..7e22184 100644
--- a/tower/src/retry/future.rs
+++ b/tower/src/retry/future.rs
@@ -34,11 +34,11 @@ pin_project! {
             future: F
         },
         // Polling the future from [`Policy::retry`]
-        Checking {
+        Waiting {
             #[pin]
-            checking: P
+            waiting: P
         },
-        // Polling [`Service::poll_ready`] after [`Checking`] was OK.
+        // Polling [`Service::poll_ready`] after [`Waiting`] was OK.
         Retrying,
     }
 }
@@ -63,8 +63,8 @@ where
 
 impl<P, S, Request> Future for ResponseFuture<P, S, Request>
 where
-    P: Policy<Request, S::Response, S::Error> + Clone,
-    S: Service<Request> + Clone,
+    P: Policy<Request, S::Response, S::Error>,
+    S: Service<Request>,
 {
     type Output = Result<S::Response, S::Error>;
 
@@ -77,8 +77,8 @@ where
                     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 });
+                            Some(waiting) => {
+                                this.state.set(State::Waiting { waiting });
                             }
                             None => return Poll::Ready(result),
                         }
@@ -87,12 +87,9 @@ where
                         return Poll::Ready(result);
                     }
                 }
-                StateProj::Checking { checking } => {
-                    this.retry
-                        .as_mut()
-                        .project()
-                        .policy
-                        .set(ready!(checking.poll(cx)));
+                StateProj::Waiting { waiting } => {
+                    ready!(waiting.poll(cx));
+
                     this.state.set(State::Retrying);
                 }
                 StateProj::Retrying => {
diff --git a/tower/src/retry/mod.rs b/tower/src/retry/mod.rs
index a9e2738..597f3e4 100644
--- a/tower/src/retry/mod.rs
+++ b/tower/src/retry/mod.rs
@@ -17,9 +17,28 @@ pin_project! {
     /// Configure retrying requests of "failed" responses.
     ///
     /// A [`Policy`] classifies what is a "failed" response.
+    ///
+    /// # Clone
+    ///
+    /// This middleware requires that the inner `Service` implements [`Clone`],
+    /// because the `Service` must be stored in each [`ResponseFuture`] in
+    /// order to retry the request in the event of a failure. If the inner
+    /// `Service` type does not implement `Clone`, the [`Buffer`] middleware
+    /// can be added to make any `Service` cloneable.
+    ///
+    /// [`Buffer`]: crate::buffer::Buffer
+    ///
+    /// The `Policy` must also implement `Clone`. This middleware will
+    /// clone the policy for each _request session_. This means a new clone
+    /// of the policy will be created for each initial request and any subsequent
+    /// retries of that request. Therefore, any state stored in the `Policy` instance
+    /// is for that request session only. In order to share data across request
+    /// sessions, that shared state may be stored in an [`Arc`], so that all clones
+    /// of the `Policy` type reference the same instance of the shared state.
+    ///
+    /// [`Arc`]: std::sync::Arc
     #[derive(Clone, Debug)]
     pub struct Retry<P, S> {
-        #[pin]
         policy: P,
         service: S,
     }
diff --git a/tower/src/retry/policy.rs b/tower/src/retry/policy.rs
index 20adc3d..06710c7 100644
--- a/tower/src/retry/policy.rs
+++ b/tower/src/retry/policy.rs
@@ -14,9 +14,9 @@ use std::future::Future;
 /// struct Attempts(usize);
 ///
 /// impl<E> Policy<Req, Res, E> for Attempts {
-///     type Future = future::Ready<Self>;
+///     type Future = future::Ready<()>;
 ///
-///     fn retry(&self, req: &mut Req, result: &mut Result<Res, E>) -> Option<Self::Future> {
+///     fn retry(&mut self, req: &mut Req, result: &mut Result<Res, E>) -> Option<Self::Future> {
 ///         match result {
 ///             Ok(_) => {
 ///                 // Treat all `Response`s as success,
@@ -28,7 +28,8 @@ use std::future::Future;
 ///                 // But we limit the number of attempts...
 ///                 if self.0 > 0 {
 ///                     // Try again!
-///                     Some(future::ready(Attempts(self.0 - 1)))
+///                     self.0 -= 1;
+///                     Some(future::ready(()))
 ///                 } else {
 ///                     // Used all our attempts, no retry...
 ///                     None
@@ -37,14 +38,14 @@ use std::future::Future;
 ///         }
 ///     }
 ///
-///     fn clone_request(&self, req: &Req) -> Option<Req> {
+///     fn clone_request(&mut self, req: &Req) -> Option<Req> {
 ///         Some(req.clone())
 ///     }
 /// }
 /// ```
-pub trait Policy<Req, Res, E>: Sized {
+pub trait Policy<Req, Res, E> {
     /// The [`Future`] type returned by [`Policy::retry`].
-    type Future: Future<Output = Self>;
+    type Future: Future<Output = ()>;
 
     /// Check the policy if a certain request should be retried.
     ///
@@ -53,8 +54,10 @@ pub trait Policy<Req, Res, E>: Sized {
     ///
     /// If the request should **not** be retried, return `None`.
     ///
-    /// If the request *should* be retried, return `Some` future of a new
-    /// policy that would apply for the next request attempt.
+    /// If the request *should* be retried, return `Some` future that will delay
+    /// the next retry of the request. This can be used to sleep for a certain
+    /// duration, to wait for some external condition to be met before retrying,
+    /// or resolve right away, if the request should be retried immediately.
     ///
     /// ## Mutating Requests
     ///
@@ -77,10 +80,14 @@ pub trait Policy<Req, Res, E>: Sized {
     ///
     /// [`Service::Response`]: crate::Service::Response
     /// [`Service::Error`]: crate::Service::Error
-    fn retry(&self, req: &mut Req, result: &mut Result<Res, E>) -> Option<Self::Future>;
+    fn retry(&mut self, req: &mut Req, result: &mut Result<Res, E>) -> Option<Self::Future>;
 
     /// Tries to clone a request before being passed to the inner service.
     ///
     /// If the request cannot be cloned, return [`None`].
-    fn clone_request(&self, req: &Req) -> Option<Req>;
+    fn clone_request(&mut self, req: &Req) -> Option<Req>;
 }
+
+// Ensure `Policy` is object safe
+#[cfg(test)]
+fn _obj_safe(_: Box<dyn Policy<(), (), (), Future = futures::future::Ready<()>>>) {}


Test patch:
diff --git a/tower/tests/builder.rs b/tower/tests/builder.rs
index 90bb70a..574766b 100644
--- a/tower/tests/builder.rs
+++ b/tower/tests/builder.rs
@@ -43,13 +43,13 @@ where
     Req: Clone,
     E: Into<Box<dyn std::error::Error + Send + Sync + 'static>>,
 {
-    type Future = Ready<Self>;
+    type Future = Ready<()>;
 
-    fn retry(&self, _req: &mut Req, _result: &mut Result<Res, E>) -> Option<Self::Future> {
+    fn retry(&mut self, _req: &mut Req, _result: &mut Result<Res, E>) -> Option<Self::Future> {
         None
     }
 
-    fn clone_request(&self, req: &Req) -> Option<Req> {
+    fn clone_request(&mut self, req: &Req) -> Option<Req> {
         Some(req.clone())
     }
 }
diff --git a/tower/tests/retry/main.rs b/tower/tests/retry/main.rs
index d0d7a8a..7ce220b 100644
--- a/tower/tests/retry/main.rs
+++ b/tower/tests/retry/main.rs
@@ -122,16 +122,16 @@ type Handle = mock::Handle<Req, Res>;
 struct RetryErrors;
 
 impl Policy<Req, Res, Error> for RetryErrors {
-    type Future = future::Ready<Self>;
-    fn retry(&self, _: &mut Req, result: &mut Result<Res, Error>) -> Option<Self::Future> {
+    type Future = future::Ready<()>;
+    fn retry(&mut self, _: &mut Req, result: &mut Result<Res, Error>) -> Option<Self::Future> {
         if result.is_err() {
-            Some(future::ready(RetryErrors))
+            Some(future::ready(()))
         } else {
             None
         }
     }
 
-    fn clone_request(&self, req: &Req) -> Option<Req> {
+    fn clone_request(&mut self, req: &Req) -> Option<Req> {
         Some(*req)
     }
 }
@@ -140,16 +140,17 @@ impl Policy<Req, Res, Error> for RetryErrors {
 struct Limit(usize);
 
 impl Policy<Req, Res, Error> for Limit {
-    type Future = future::Ready<Self>;
-    fn retry(&self, _: &mut Req, result: &mut Result<Res, Error>) -> Option<Self::Future> {
+    type Future = future::Ready<()>;
+    fn retry(&mut 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)))
+            self.0 -= 1;
+            Some(future::ready(()))
         } else {
             None
         }
     }
 
-    fn clone_request(&self, req: &Req) -> Option<Req> {
+    fn clone_request(&mut self, req: &Req) -> Option<Req> {
         Some(*req)
     }
 }
@@ -158,18 +159,18 @@ impl Policy<Req, Res, Error> for Limit {
 struct UnlessErr(InnerError);
 
 impl Policy<Req, Res, Error> for UnlessErr {
-    type Future = future::Ready<Self>;
-    fn retry(&self, _: &mut Req, result: &mut Result<Res, Error>) -> Option<Self::Future> {
+    type Future = future::Ready<()>;
+    fn retry(&mut 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()))
+                Some(future::ready(()))
             } else {
                 None
             }
         })
     }
 
-    fn clone_request(&self, req: &Req) -> Option<Req> {
+    fn clone_request(&mut self, req: &Req) -> Option<Req> {
         Some(*req)
     }
 }
@@ -178,12 +179,12 @@ impl Policy<Req, Res, Error> for UnlessErr {
 struct CannotClone;
 
 impl Policy<Req, Res, Error> for CannotClone {
-    type Future = future::Ready<Self>;
-    fn retry(&self, _: &mut Req, _: &mut Result<Res, Error>) -> Option<Self::Future> {
+    type Future = future::Ready<()>;
+    fn retry(&mut self, _: &mut Req, _: &mut Result<Res, Error>) -> Option<Self::Future> {
         unreachable!("retry cannot be called since request isn't cloned");
     }
 
-    fn clone_request(&self, _req: &Req) -> Option<Req> {
+    fn clone_request(&mut self, _req: &Req) -> Option<Req> {
         None
     }
 }
@@ -199,21 +200,20 @@ impl Policy<Req, Res, Error> for MutatingPolicy
 where
     Error: From<&'static str>,
 {
-    type Future = future::Ready<Self>;
+    type Future = future::Ready<()>;
 
-    fn retry(&self, req: &mut Req, result: &mut Result<Res, Error>) -> Option<Self::Future> {
+    fn retry(&mut 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,
-            }))
+            self.remaining -= 1;
+            Some(future::ready(()))
         }
     }
 
-    fn clone_request(&self, req: &Req) -> Option<Req> {
+    fn clone_request(&mut self, req: &Req) -> Option<Req> {
         Some(*req)
     }
 }