Avoid time operations that can panic

We have reports of runtime panics (linkerd/linkerd2#7748) that sound a
lot like rust-lang/rust#86470. We don't have any evidence that these
panics originate in tower, but we have some potentialy flawed `Instant`
arithmetic that could panic in this way.

Even though this is almost definitely a bug in Rust, it seems most
prudent to actively avoid the uses of `Instant` that are prone to this
bug.

This change replaces uses of `Instant::elapsed` and `Instant::sub` with
calls to `Instant::saturating_duration_since` to prevent this class of
panic. These fixes should ultimately be made in the standard library,
but this change lets us avoid this problem while we wait for those
fixes.

See also hyperium/hyper#2746

Author: olix0r

tower/src/hedge/latency.rs

@@ -82,7 +82,7 @@ where
         let this = self.project();
 
         let rsp = ready!(this.inner.poll(cx)).map_err(Into::into)?;
-        let duration = Instant::now() - *this.start;
+        let duration = Instant::now().saturating_duration_since(*this.start);
         this.rec.record(duration);
         Poll::Ready(Ok(rsp))
     }

tower/src/hedge/rotating_histogram.rs

@@ -39,7 +39,7 @@ impl RotatingHistogram {
     }
 
     fn maybe_rotate(&mut self) {
-        let delta = Instant::now() - self.last_rotation;
+        let delta = Instant::now().saturating_duration_since(self.last_rotation);
         // TODO: replace with delta.duration_div when it becomes stable.
         let rotations = (nanos(delta) / nanos(self.period)) as u32;
         if rotations >= 2 {

tower/src/limit/rate/service.rs

@@ -73,7 +73,7 @@ where
         match self.state {
             State::Ready { .. } => return Poll::Ready(ready!(self.inner.poll_ready(cx))),
             State::Limited => {
-                if let Poll::Pending = Pin::new(&mut self.sleep).poll(cx) {
+                if Pin::new(&mut self.sleep).poll(cx).is_pending() {
                     tracing::trace!("rate limit exceeded; sleeping.");
                     return Poll::Pending;
                 }

tower/src/load/peak_ewma.rs

@@ -91,6 +91,7 @@ const NANOS_PER_MILLI: f64 = 1_000_000.0;
 impl<S, C> PeakEwma<S, C> {
     /// Wraps an `S`-typed service so that its load is tracked by the EWMA of its peak latency.
     pub fn new(service: S, default_rtt: Duration, decay_ns: f64, completion: C) -> Self {
+        debug_assert!(decay_ns > 0.0, "decay_ns must be positive");
         Self {
             service,
             decay_ns,
@@ -241,7 +242,7 @@ impl RttEstimate {
             recv_at,
             sent_at
         );
-        let rtt = nanos(recv_at - sent_at);
+        let rtt = nanos(recv_at.saturating_duration_since(sent_at));
 
         let now = Instant::now();
         debug_assert!(
@@ -264,7 +265,7 @@ impl RttEstimate {
             // prior estimate according to how much time has elapsed since the last
             // update. The inverse of the decay is used to scale the estimate towards the
             // observed RTT value.
-            let elapsed = nanos(now - self.update_at);
+            let elapsed = nanos(now.saturating_duration_since(self.update_at));
             let decay = (-elapsed / decay_ns).exp();
             let recency = 1.0 - decay;
             let next_estimate = (self.rtt_ns * decay) + (rtt * recency);

tower/src/retry/budget.rs

@@ -175,7 +175,7 @@ impl Bucket {
         let mut gen = self.generation.lock().expect("generation lock");
 
         let now = Instant::now();
-        let diff = now - gen.time;
+        let diff = now.saturating_duration_since(gen.time);
         if diff < self.window {
             // not expired yet
             return;