Fix "Too many "Too Many Requests" log" (#761)

Fixed #747

 - Add dep compact_str v0.6.1 to binstalk-downloader
 - Impl new type `DelayRequest`
 - Handle 503/429 with wait duration > `MAX_RETRY_DURATION` by simply taking the min
 - Fix `Client::send_request_inner`: Ensure 503/429 get propagated to other requests
   
   even if the current requests reach its maximum retry and decides to
   simply return an error.

Signed-off-by: Jiahao XU <Jiahao_XU@outlook.com>

crates/binstalk-downloader/src/remote/delay_request.rs

@@ -0,0 +1,104 @@
+use std::{
+    collections::HashMap,
+    future::Future,
+    pin::Pin,
+    sync::Mutex,
+    task::{Context, Poll},
+};
+
+use compact_str::{CompactString, ToCompactString};
+use reqwest::{Request, Url};
+use tokio::{
+    sync::Mutex as AsyncMutex,
+    time::{sleep_until, Instant},
+};
+use tower::{Service, ServiceExt};
+
+#[derive(Debug)]
+pub(super) struct DelayRequest<S> {
+    inner: AsyncMutex<S>,
+    hosts_to_delay: Mutex<HashMap<CompactString, Instant>>,
+}
+
+impl<S> DelayRequest<S> {
+    pub(super) fn new(inner: S) -> Self {
+        Self {
+            inner: AsyncMutex::new(inner),
+            hosts_to_delay: Default::default(),
+        }
+    }
+
+    pub(super) fn add_urls_to_delay<'a, Urls>(&self, urls: Urls, deadline: Instant)
+    where
+        Urls: IntoIterator<Item = &'a Url>,
+    {
+        let mut hosts_to_delay = self.hosts_to_delay.lock().unwrap();
+
+        urls.into_iter().filter_map(Url::host_str).for_each(|host| {
+            hosts_to_delay
+                .entry(host.to_compact_string())
+                .and_modify(|old_dl| {
+                    *old_dl = deadline.max(*old_dl);
+                })
+                .or_insert(deadline);
+        });
+    }
+
+    fn wait_until_available(&self, url: &Url) -> impl Future<Output = ()> + Send + 'static {
+        let mut hosts_to_delay = self.hosts_to_delay.lock().unwrap();
+
+        let sleep = url
+            .host_str()
+            .and_then(|host| hosts_to_delay.get(host).map(|deadline| (*deadline, host)))
+            .and_then(|(deadline, host)| {
+                if deadline.elapsed().is_zero() {
+                    Some(sleep_until(deadline))
+                } else {
+                    // We have already gone past the deadline,
+                    // so we should remove it instead.
+                    hosts_to_delay.remove(host);
+                    None
+                }
+            });
+
+        async move {
+            if let Some(sleep) = sleep {
+                sleep.await;
+            }
+        }
+    }
+}
+
+impl<'this, S> Service<Request> for &'this DelayRequest<S>
+where
+    S: Service<Request> + Send,
+    S::Future: Send,
+{
+    type Response = S::Response;
+    type Error = S::Error;
+    // TODO: Replace this with `type_alias_impl_trait` once it stablises
+    // https://github.com/rust-lang/rust/issues/63063
+    type Future = Pin<Box<dyn Future<Output = Result<Self::Response, Self::Error>> + Send + 'this>>;
+
+    fn poll_ready(&mut self, _cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
+        Poll::Ready(Ok(()))
+    }
+
+    fn call(&mut self, req: Request) -> Self::Future {
+        let this = *self;
+
+        Box::pin(async move {
+            this.wait_until_available(req.url()).await;
+
+            // Reduce critical section:
+            //  - Construct the request before locking
+            //  - Once it is ready, call it and obtain
+            //    the future, then release the lock before
+            //    polling the future, which performs network I/O that could
+            //    take really long.
+            let future = this.inner.lock().await.ready().await?.call(req);
+
+            future.await
+        })
+    }
+}