code review

Author: Richard Kuo (Onyx)

backend/onyx/connectors/slack/onyx_retry_handler.py

@@ -17,13 +17,9 @@ class OnyxRedisSlackRetryHandler(RetryHandler):
     """
     This class uses Redis to share a rate limit among multiple threads.
 
-    Threads that encounter a rate limit will observe the shared delay, increment the
-    shared delay with the retry value, and use the new shared value as a wait interval.
-
-    This has the effect of serializing calls when a rate limit is hit, which is what
-    needs to happens if the server punishes us with additional limiting when we make
-    a call too early. We believe this is what Slack is doing based on empirical
-    observation, meaning we see indefinite hangs if we're too aggressive.
+    This just sets the desired retry delay with TTL in redis. In conjunction with
+    a custom subclass of the client, the value is read and obeyed prior to an API call
+    and also serialized.
 
     Another way to do this is just to do exponential backoff. Might be easier?
 

backend/onyx/connectors/slack/onyx_slack_web_client.py

@@ -30,7 +30,6 @@ class OnyxSlackWebClient(WebClient):
     def __init__(
         self, delay_lock: str, delay_key: str, r: Redis, *args: Any, **kwargs: Any
     ) -> None:
-        """"""
         super().__init__(*args, **kwargs)
         self._delay_key = delay_key
         self._delay_lock = delay_lock
@@ -56,7 +55,6 @@ def _perform_urllib_http_request(
         start = time.monotonic()
         while True:
             acquired = lock.acquire(blocking_timeout=ONYX_SLACK_LOCK_BLOCKING_TIMEOUT)
-
             if acquired:
                 break
 
@@ -93,8 +91,10 @@ def _perform_urllib_http_request_internal(
         url: str,
         req: Request,
     ) -> Dict[str, Any]:
+        """Overrides the internal method which is mostly the direct call to
+        urllib/urlopen ... so this is a good place to perform our delay."""
 
-        # if we can get the lock, then read and extend the ttl
+        # read and execute the delay
         ttl_ms = cast(int, self._redis.pttl(self._delay_key))
         if ttl_ms < 0:  # negative values are error status codes ... see docs
             ttl_ms = 0

backend/tests/integration/connector_job_tests/slack/slack_api_utils.py

@@ -216,8 +216,13 @@ def get_and_provision_available_slack_channels(
 
     @staticmethod
     def build_slack_user_email_id_map(slack_client: WebClient) -> dict[str, str]:
-        users_results = slack_client.users_list()
-        users: list[dict[str, Any]] = users_results.get("members", [])
+        users: list[dict[str, Any]] = []
+
+        for users_results in make_paginated_slack_api_call(
+            slack_client.users_list,
+        ):
+            users.extend(users_results.get("members", []))
+
         user_email_id_map = {}
         for user in users:
             if not (email := user.get("profile", {}).get("email")):