feat: enhance _get_previous_time_schedules to accept current_time parameter

- Updated the _get_previous_time_schedules method to take current_time as an argument, allowing for more precise cutoff calculations.
- Adjusted the logic to use the provided current_time for determining previous schedules, ensuring no overlap with the current window.
- Modified the call to _get_previous_time_schedules in the first run logic to pass the current_time parameter.

Author: tmkarthi

taskiq_redis/list_schedule_source.py

@@ -174,7 +174,10 @@ async def _cleanup_time_index(self, redis: Redis) -> None:  # type: ignore[type-
             if await redis.llen(key) == 0:
                 await redis.zrem(self._get_time_index_key(), key)
 
-    async def _get_previous_time_schedules(self) -> list[bytes]:
+    async def _get_previous_time_schedules(
+        self,
+        current_time: datetime.datetime,
+    ) -> list[bytes]:
         """
         Function that gets all timed schedules that are in the past.
 
@@ -187,11 +190,15 @@ async def _get_previous_time_schedules(self) -> list[bytes]:
 
         This function is called only during the first run to minimize
         the number of requests to the Redis server.
+
+        :param current_time: The reference time captured by the caller,
+            used to derive the cutoff so that the "previous" and "current"
+            windows never overlap.
         """
         logger.info("Getting previous time schedules")
-        minute_before = datetime.datetime.now(
-            datetime.timezone.utc,
-        ).replace(second=0, microsecond=0) - datetime.timedelta(
+        minute_before = current_time.replace(
+            second=0, microsecond=0,
+        ) - datetime.timedelta(
             minutes=1,
         )
         schedules = []
@@ -284,7 +291,7 @@ async def get_schedules(self) -> list["ScheduledTask"]:
         timed: list[bytes] = []
         # Only during first run, we need to get previous time schedules
         if not self._skip_past_schedules and self._is_first_run:
-            timed = await self._get_previous_time_schedules()
+            timed = await self._get_previous_time_schedules(current_time)
             self._is_first_run = False
         async with Redis(connection_pool=self._connection_pool) as redis:
             buffer = []