Use exponential backoff when missing Retry-After header.

Author: mephenor

src/ghga_service_commons/transports/factory.py

@@ -27,10 +27,15 @@ class CompositeTransportFactory:
     """TODO"""
 
     @classmethod
-    def _create_common_transport_layers(cls, config: CompositeConfig, transport):
+    def _create_common_transport_layers(
+        cls, config: CompositeConfig, limits: Limits | None = None
+    ):
         """TODO"""
+        base_transport = (
+            AsyncHTTPTransport(limits=limits) if limits else AsyncHTTPTransport()
+        )
         ratelimiting_transport = AsyncRatelimitingTransport(
-            config=config, transport=transport
+            config=config, transport=base_transport
         )
         retry_transport = AsyncRetryTransport(
             config=config, transport=ratelimiting_transport
@@ -42,19 +47,13 @@ def create_ratelimiting_retry_transport(
         cls, config: CompositeConfig, limits: Limits | None = None
     ) -> AsyncRetryTransport:
         """TODO"""
-        base_transport = (
-            AsyncHTTPTransport(limits=limits) if limits else AsyncHTTPTransport()
-        )
-        return cls._create_common_transport_layers(config, base_transport)
+        return cls._create_common_transport_layers(config)
 
     @classmethod
     def create_ratelimiting_retry_transport_with_cache(
         cls, config: CompositeCacheConfig, limits: Limits | None = None
-    ) -> AsyncRetryTransport:
+    ) -> AsyncCacheTransport:
         """TODO"""
-        base_transport = (
-            AsyncHTTPTransport(limits=limits) if limits else AsyncHTTPTransport()
-        )
         storage = AsyncInMemoryStorage(ttl=config.cache_ttl)
-        cache_tranport = AsyncCacheTransport(transport=base_transport, storage=storage)
-        return cls._create_common_transport_layers(config, cache_tranport)
+        retry_transport = cls._create_common_transport_layers(config)
+        return AsyncCacheTransport(transport=retry_transport, storage=storage)

src/ghga_service_commons/transports/ratelimiting.py

@@ -53,20 +53,21 @@ async def handle_async_request(self, request: httpx.Request) -> httpx.Response:
         # Caculate seconds since the last request has been fired and corresponding wait time
         time_elapsed = time.monotonic() - self._last_request_time
         remaining_wait = max(0, self._wait_time - time_elapsed)
-        log.info("Configured base wait time: %.3f s", self._wait_time)
-        log.info(
-            "Time elapsed since last request:%.3f.\nWaiting for at least %.3f s",
+        log.debug(
+            "Time elapsed since last request: %.3f s.\nRemaining wait time: %.3f s.",
             time_elapsed,
             remaining_wait,
         )
 
         # Add jitter to both cases and sleep
         if remaining_wait < self._jitter:
-            time.sleep(random.uniform(remaining_wait, self._jitter))  # noqa: S311
+            sleep_for = random.uniform(remaining_wait, self._jitter)  # noqa: S311
+            log.debug("Sleeping for %.3f s.")
+            time.sleep(sleep_for)
         else:
-            time.sleep(
-                random.uniform(remaining_wait, remaining_wait + self._jitter)  # noqa: S311
-            )
+            sleep_for = random.uniform(remaining_wait, remaining_wait + self._jitter)  # noqa: S311
+            log.debug("Sleeping for %.3f s.")
+            time.sleep(sleep_for)
 
         # Delegate call and update timestamp
         response = await self._transport.handle_async_request(request=request)
@@ -79,12 +80,13 @@ async def handle_async_request(self, request: httpx.Request) -> httpx.Response:
             retry_after = response.headers.get("Retry-After")
             if retry_after:
                 self._wait_time = float(retry_after)
-                log.info("Received retry after response: %.3f s", self._wait_time)
+                log.info("Received retry after response: %.3f s.", self._wait_time)
             else:
                 log.warning(
-                    "Retry-After header not present in 429 response, using fallback instead."
+                    "Retry-After header not present in 429 response.\nDelegating to underlying wait strategy."
                 )
-                self._wait_time = self._jitter
+                # Modify response headers to communicate intent to retry layer
+                response.headers["Should-Wait"] = "true"
             self._num_requests = 0
         elif self._reset_after and self._reset_after <= self._num_requests:
             self._wait_time = 0

src/ghga_service_commons/transports/retry.py

@@ -63,7 +63,6 @@ def _log_retry_stats(retry_state: RetryCallState):
     if time_passed := retry_state.seconds_since_start:
         retry_stats["seconds_elapsed"] = round(time_passed, 3)
 
-    log.warning(retry_state.fn)
     log.info(
         "Retry attempt number %i for function %s.",
         attempt_number,
@@ -75,11 +74,13 @@ def _log_retry_stats(retry_state: RetryCallState):
 class wait_exponential_ignore_429(wait_exponential):  # noqa: N801
     """Custom exponential backof strategy not waiting on 429 responses"""
 
-    def __call__(self, retry_state: RetryCallState) -> float:  # noqa: D102
+    def __call__(self, retry_state: RetryCallState) -> float:
+        """Copied from base class and adjusted"""
         if (
             retry_state.outcome
             and (result := retry_state.outcome.result())
             and isinstance(result, httpx.Response)
+            and not result.headers.get("Should-Wait")
         ):
             return 0
         try: