Fix too many 429 response

Fixed #1229

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

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

@@ -78,28 +78,19 @@ impl<S> DelayRequest<S> {
             });
     }
 
-    fn wait_until_available(&self, url: &Url) -> impl Future<Output = ()> + Send + 'static {
+    fn get_delay_until(&self, host: &str) -> Option<Instant> {
         let mut hosts_to_delay = self.hosts_to_delay.lock().unwrap();
 
-        let deadline = 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(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(deadline) = deadline {
-                sleep_until(deadline).await;
+        hosts_to_delay.get(host).copied().and_then(|delay_until| {
+            if delay_until.elapsed().is_zero() {
+                Some(delay_until)
+            } else {
+                // We have already gone past the deadline,
+                // so we should remove it instead.
+                hosts_to_delay.remove(host);
+                None
             }
-        }
+        })
     }
 }
 
@@ -122,15 +113,26 @@ where
         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);
+            let host = req.url().host_str();
+
+            let future = loop {
+                // Wait until we are ready to send next requests
+                // (client-side rate-limit throttler).
+                let mut guard = this.inner.lock().await;
+                let ready = guard.ready().await?;
+
+                if let Some(delay_until) = host.and_then(|host| this.get_delay_until(host)) {
+                    // If the host rate-limit us, then wait until then
+                    // and try again (server-side rate-limit throttler).
+                    //
+                    // Drop guard to reduce critical section.
+                    drop(guard);
+                    sleep_until(delay_until).await;
+                } else {
+                    // Reduce critical section by breaking here
+                    break ready.call(req);
+                }
+            };
 
             future.await
         })