updating the code with exponential backoff for when it tries to retrieve a user

Author: modernNeo

wall_e/extensions/leveling.py

@@ -7,19 +7,15 @@
 import discord
 import pytz
 from discord import NotFound, app_commands, Guild
-from discord.errors import DiscordServerError, HTTPException
+from discord.errors import DiscordException
 from discord.ext import commands, tasks
 
-from utilities.global_vars import bot, wall_e_config
-
-from wall_e_models.models import Level, UserPoint, UpdatedUser, ProfileBucketInProgress
-
 from utilities.embed import embed, COLOUR_MAPPING, WallEColour
 from utilities.file_uploading import start_file_uploading
+from utilities.global_vars import bot, wall_e_config
 from utilities.paginate import paginate_embed
 from utilities.setup_logger import Loggers
-
-BUGGY_USER = [234501345749630976]
+from wall_e_models.models import Level, UserPoint, UpdatedUser, ProfileBucketInProgress
 
 
 class Leveling(commands.Cog):
@@ -53,9 +49,10 @@ def __init__(self):
         self.council_channel = None
         self.levelling_website_avatar_channel = None
         self.bucket_update_in_progress = False
+        self.NUMBER_OF_RETRIEVAL_ATTEMPTS_PER_USER = 15
         self.ensure_xp_roles_exist_and_have_right_users.start()
         self.process_leveling_profile_data_for_lurkers.start()
-        self.process_outdated_profile_pics.start()
+        # self.process_outdated_profile_pics.start()
         self.process_leveling_profile_data_for_active_users.start()
 
     @commands.Cog.listener(name="on_ready")
@@ -67,7 +64,8 @@ async def upload_debug_logs(self):
         while self.guild is None:
             await asyncio.sleep(2)
         await start_file_uploading(
-            self.logger, self.guild, bot, wall_e_config, self.debug_log_file_absolute_path, "leveling_debug"
+            self.logger, self.guild, bot, wall_e_config, self.debug_log_file_absolute_path,
+            "leveling_debug"
         )
 
     @commands.Cog.listener(name="on_ready")
@@ -84,7 +82,8 @@ async def upload_error_logs(self):
         while self.guild is None:
             await asyncio.sleep(2)
         await start_file_uploading(
-            self.logger, self.guild, bot, wall_e_config, self.error_log_file_absolute_path, "leveling_error"
+            self.logger, self.guild, bot, wall_e_config, self.error_log_file_absolute_path,
+            "leveling_error"
         )
 
     @commands.Cog.listener(name="on_ready")
@@ -558,7 +557,7 @@ async def process_leveling_profile_data_for_lurkers(self):
             f"[Leveling process_leveling_profile_data_for_lurkers()] {user_ids_to_update} "
             f"potential updates retrieved for bucket {entry.bucket_number_completed}"
         )
-        await self._update_users(self.process_lurkers_logger, user_ids_to_update)
+        await self._update_users_with_given_ids(self.process_lurkers_logger, user_ids_to_update)
         await ProfileBucketInProgress.async_save(entry)
 
     async def _set_bucket_numbers(self, logger):
@@ -671,37 +670,40 @@ async def _get_current_bucket_number(self) -> ProfileBucketInProgress:
                 entry.bucket_number_completed = 1
         return entry
 
-    async def _update_users(self, logger, updated_user_ids):
+    async def _update_users_with_given_ids(self, logger, updated_user_ids):
         """
         iterates through the given list of user_ids and updates them
         :param updated_user_ids:
         :return:
         """
         total_number_of_updates_needed = len(updated_user_ids)
         for index, user_id in enumerate(updated_user_ids):
+            successful = False
             logger.debug(
-                f"[Leveling _update_users()] attempting to get updated user_point profile data for member {user_id} "
-                f"{index + 1}/{total_number_of_updates_needed} "
+                f"[Leveling _update_users_with_given_ids()] attempt "
+                f"{self.user_points[user_id].leveling_update_attempt} to get updated user_point profile data for "
+                f"member {user_id} {index + 1}/{total_number_of_updates_needed} "
             )
-            member = None
-            try:
-                member = await self.guild.fetch_member(user_id)
-            except (NotFound, DiscordServerError) as e:
-                try:
-                    logger.info(
-                        f"[Leveling _update_users()] got following error when fetching guild member {user_id}\n{e}"
-                    )
-                    member = await bot.fetch_user(user_id)
-                except DiscordServerError as e:
-                    logger.error(
-                        f"[Leveling _update_users()] got the following error when fetching member {user_id}\n{e}."
-                    )
-            except HTTPException as e:
-                logger.warn(
-                    f"[Leveling _update_users()] got the following error when fetching member {user_id}\n{e}."
+            while (
+                    not successful and
+                    self.user_points[user_id].leveling_update_attempt < self.NUMBER_OF_RETRIEVAL_ATTEMPTS_PER_USER
+            ):
+                self.user_points[user_id].leveling_update_attempt += 1
+                random_number_milliseconds = random.randint(0, 1000) / 1000
+                sleep_seconds = math.pow(
+                    2, self.user_points[user_id].leveling_update_attempt + random_number_milliseconds
+                )
+                logger.debug(
+                    f"[Leveling _update_users_with_given_ids()] Sleeping for {sleep_seconds} before attempt "
+                    f"{self.user_points[user_id].leveling_update_attempt} when trying to update user_point profile "
+                    f"data for member {user_id}."
                 )
-            if member:
-                await self._update_member_profile_data(logger, member, user_id, index, total_number_of_updates_needed)
+                await asyncio.sleep(sleep_seconds)
+                member = await self.get_user(logger, user_id)
+                if member:
+                    successful = await self._update_member_profile_data(
+                        logger, member, user_id, index, total_number_of_updates_needed
+                    )
 
     @tasks.loop(seconds=2)
     async def process_leveling_profile_data_for_active_users(self):
@@ -717,19 +719,59 @@ async def process_leveling_profile_data_for_active_users(self):
         for index, update_user in enumerate(updated_user_logs):
             updated_user_log_id = update_user[0]  # noqa: F841
             updated_user_id = update_user[1]
+            successful = False
             self.logger.debug(
-                f"[Leveling process_leveling_profile_data_for_active_users()] attempting to get updated "
-                f"user_point profile data for member {updated_user_id} "
-                f"{index + 1}/{total_number_of_updates_needed} "
+                f"[Leveling process_leveling_profile_data_for_active_users()] attempt "
+                f"{self.user_points[updated_user_id].leveling_update_attempt} to get data for user in UpdatedUser"
+                f" with id {updated_user_id} {index + 1}/{total_number_of_updates_needed} "
             )
+            while (
+                    not successful and
+                    self.user_points[updated_user_id].leveling_update_attempt <
+                    self.NUMBER_OF_RETRIEVAL_ATTEMPTS_PER_USER
+            ):
+                self.user_points[updated_user_id].leveling_update_attempt += 1
+                random_number_milliseconds = random.randint(0, 1000) / 1000
+                sleep_seconds = math.pow(
+                    2, self.user_points[updated_user_id].leveling_update_attempt + random_number_milliseconds
+                )
+                self.logger.debug(
+                    f"[Leveling process_leveling_profile_data_for_active_users()] Sleeping for {sleep_seconds} before"
+                    f" attempt {self.user_points[updated_user_id].leveling_update_attempt} when trying to get data"
+                    f" for user in UpdatedUser with id {updated_user_id}."
+                )
+                await asyncio.sleep(sleep_seconds)
+                member = await self.get_user(self.logger, updated_user_id)
+                if member:
+                    # cannot call _update_users_with_given_ids like process_outdated_profile_pics and
+                    # process_leveling_profile_data_for_lurkers because this task needs to be able to specify a
+                    # updated_user_log_id that will be deleted update_leveling_profile_info when the user is processed
+                    successful = await self._update_member_profile_data(
+                        self.logger, member, updated_user_id, index, total_number_of_updates_needed,
+                        updated_user_log_id=updated_user_log_id
+                    )
+
+    async def get_user(self, logger, user_id):
+        user = None
+        try:
+            user = await self.guild.fetch_member(user_id)
+        except NotFound as e:
             try:
-                member = await self.guild.fetch_member(updated_user_id)
-            except NotFound:
-                member = await bot.fetch_user(updated_user_id)
-            await self._update_member_profile_data(
-                self.logger, member, updated_user_id, index, total_number_of_updates_needed,
-                updated_user_log_id=updated_user_log_id
+                logger.info(
+                    f"[Leveling get_user()] unable to find guild member {user_id}\n{e}"
+                )
+                user = await bot.fetch_user(user_id)
+            except DiscordException as e:
+                logger.error(
+                    f"[Leveling get_user()] got the following error when fetching discord user "
+                    f"{user_id}\n{e}."
+                )
+        except DiscordException as e:
+            logger.error(
+                f"[Leveling get_user()] got the following error when fetching guild member "
+                f"{user_id}\n{e}."
             )
+        return user
 
     @tasks.loop(seconds=5)
     async def process_outdated_profile_pics(self):
@@ -747,7 +789,7 @@ async def process_outdated_profile_pics(self):
             f"[Leveling process_outdated_profile_pics()] {number_of_users_to_update} users with outdated CDN links"
             f" to update"
         )
-        await self._update_users(self.update_outdated_profile_pics_logger, user_ids_to_update)
+        await self._update_users_with_given_ids(self.update_outdated_profile_pics_logger, user_ids_to_update)
         self.process_outdated_profile_pics_in_progress = False
 
     async def _update_member_profile_data(self, logger, member, updated_user_id, index,
@@ -763,11 +805,10 @@ async def _update_member_profile_data(self, logger, member, updated_user_id, ind
          current loop
         :param updated_user_log_id: the ID of the UpdatedUser record to delete if this function was called as a
          result of member_update_listener's recording any detected changed in UpdatedUser
-        :return:
+        :return: True if member was successfully processed
         """
+        user_updated = False
         if member:
-            if member.id in BUGGY_USER:
-                return
             try:
                 if self.user_points[member.id].leveling_update_attempt >= 5:
                     log_level = logger.error if updated_user_log_id is None else logger.debug
@@ -812,13 +853,15 @@ async def _update_member_profile_data(self, logger, member, updated_user_id, ind
                     f" data in the database for member {member} with id [{updated_user_id}]"
                     f" {index + 1}/{total_number_of_updates_needed} due to error:\n{e}"
                 )
+            return user_updated
         else:
             logger.warn(
                 f"[Leveling _update_member_profile_data()] attempt "
                 f"{self.user_points[member.id].leveling_update_attempt}: unable to update the member profile data"
                 f" in the database for member {updated_user_id} {index + 1}/{total_number_of_updates_needed}"
             )
             await self.user_points[member.id].async_save()
+            return user_updated
 
     @app_commands.command(name="reset_attempts")
     @app_commands.checks.has_any_role("Bot_manager")