docs: update README and tests for time index cleanup logic

- Expanded README to include details on interval tasks and the new `{prefix}:time_index` sorted set for tracking schedules.
- Updated cleanup logic in `ListRedisScheduleSource` to remove stale entries older than 5 minutes instead of 1 hour.
- Modified tests to reflect the new 5-minute threshold for cleanup, ensuring accurate verification of the time index behavior.

Author: tmkarthi

README.md

@@ -154,9 +154,26 @@ This is very ineficent and should not be used for high-volume schedules. Because
 This source holds values in lists.
 
 * For cron tasks it uses key `{prefix}:cron`.
+* For interval tasks it uses key `{prefix}:interval`.
 * For timed schedules it uses key `{prefix}:time:{time}` where `{time}` is actually time where schedules should run.
+* A sorted set at `{prefix}:time_index` tracks all time keys with their unix timestamps as scores, so that past time schedules can be discovered via `ZRANGEBYSCORE` instead of scanning all Redis keys. Stale entries (older than 5 minutes with empty time key lists) are cleaned up automatically.
 
-The main advantage of this approach is that we only fetch tasks we need to run at a given time and do not perform any excesive calls to redis.
+The main advantage of this approach is that we only fetch tasks we need to run at a given time and do not perform any excessive calls to redis.
+
+#### `populate_time_index`
+
+If you are upgrading from an older version that did not maintain the `{prefix}:time_index` sorted set, existing time keys will not be present in the index. Set `populate_time_index=True` once on startup to backfill the index via a one-time `SCAN`, then set it back to `False` for subsequent runs:
+
+```python
+# First run after upgrading — backfills the time index
+source = ListRedisScheduleSource(
+    "redis://localhost/1",
+    populate_time_index=True,
+)
+
+# All subsequent runs — no SCAN, uses the time index
+source = ListRedisScheduleSource("redis://localhost/1")
+```
 
 
 ### Migration from one source to another

taskiq_redis/list_schedule_source.py

@@ -154,20 +154,20 @@ async def _cleanup_time_index(self, redis: Redis) -> None:  # type: ignore[type-
         """
         Remove stale entries from the time index sorted set.
 
-        Only removes entries that are older than 1 hour AND whose
+        Only removes entries that are older than 5 minutes AND whose
         corresponding time key list is empty (or no longer exists).
         This avoids a race condition where an eager cleanup in
         delete_schedule could remove an index entry right as
         add_schedule is creating a new schedule at the same minute.
         """
-        one_hour_ago = (
+        five_minutes_ago = (
             datetime.datetime.now(datetime.timezone.utc)
-            - datetime.timedelta(hours=1)
+            - datetime.timedelta(minutes=5)
         ).timestamp()
         stale_keys: list[bytes] = await redis.zrangebyscore(
             self._get_time_index_key(),
             "-inf",
-            one_hour_ago,
+            five_minutes_ago,
         )
         for key in stale_keys:
             if await redis.llen(key) == 0:

tests/test_list_schedule_source.py

@@ -237,12 +237,13 @@ async def test_time_index_not_eagerly_cleaned_on_delete(redis_url: str) -> None:
 @pytest.mark.anyio
 async def test_cleanup_removes_old_empty_entries(redis_url: str) -> None:
     """Test that _cleanup_time_index removes index entries that are
-    older than 1 hour and whose time key lists are empty."""
+    older than 5 minutes and whose time key lists are empty."""
     prefix = uuid.uuid4().hex
-    with freeze_time("2025-01-01 00:00:00"):
+    with freeze_time("2025-01-01 00:10:00"):
         source = ListRedisScheduleSource(redis_url, prefix=prefix)
+        # 10 minutes before "now" — well past the 5-minute threshold.
         old_time = datetime.datetime(
-            2024, 12, 31, 22, 0, tzinfo=datetime.timezone.utc,
+            2025, 1, 1, 0, 0, tzinfo=datetime.timezone.utc,
         )
         schedule = ScheduledTask(
             task_name="test_task",
@@ -263,8 +264,8 @@ async def test_cleanup_removes_old_empty_entries(redis_url: str) -> None:
     async with Redis(connection_pool=source._connection_pool) as redis:
         assert await redis.zcard(source._get_time_index_key()) == 1
 
-    # Run cleanup directly — entry is > 1 hour old and empty.
-    with freeze_time("2025-01-01 00:00:00"):
+    # Run cleanup directly — entry is > 5 minutes old and empty.
+    with freeze_time("2025-01-01 00:10:00"):
         async with Redis(connection_pool=source._connection_pool) as redis:
             await source._cleanup_time_index(redis)
 
@@ -276,12 +277,12 @@ async def test_cleanup_removes_old_empty_entries(redis_url: str) -> None:
 @pytest.mark.anyio
 async def test_cleanup_keeps_non_empty_entries(redis_url: str) -> None:
     """Test that _cleanup_time_index does NOT remove index entries whose
-    time key lists still have schedules, even if older than 1 hour."""
+    time key lists still have schedules, even if older than 5 minutes."""
     prefix = uuid.uuid4().hex
-    with freeze_time("2025-01-01 00:00:00"):
+    with freeze_time("2025-01-01 00:10:00"):
         source = ListRedisScheduleSource(redis_url, prefix=prefix)
         old_time = datetime.datetime(
-            2024, 12, 31, 22, 0, tzinfo=datetime.timezone.utc,
+            2025, 1, 1, 0, 0, tzinfo=datetime.timezone.utc,
         )
         schedule = ScheduledTask(
             task_name="test_task",
@@ -292,8 +293,8 @@ async def test_cleanup_keeps_non_empty_entries(redis_url: str) -> None:
         )
         await source.add_schedule(schedule)
 
-    # Run cleanup — entry is > 1 hour old but list is NOT empty.
-    with freeze_time("2025-01-01 00:00:00"):
+    # Run cleanup — entry is > 5 minutes old but list is NOT empty.
+    with freeze_time("2025-01-01 00:10:00"):
         async with Redis(connection_pool=source._connection_pool) as redis:
             await source._cleanup_time_index(redis)
 
@@ -305,13 +306,13 @@ async def test_cleanup_keeps_non_empty_entries(redis_url: str) -> None:
 @pytest.mark.anyio
 async def test_cleanup_keeps_recent_empty_entries(redis_url: str) -> None:
     """Test that _cleanup_time_index does NOT remove index entries that
-    are less than 1 hour old, even if their time key lists are empty."""
+    are less than 5 minutes old, even if their time key lists are empty."""
     prefix = uuid.uuid4().hex
-    with freeze_time("2025-01-01 00:00:00"):
+    with freeze_time("2025-01-01 00:04:00"):
         source = ListRedisScheduleSource(redis_url, prefix=prefix)
-        # 30 minutes ago — within the 1-hour safety window.
+        # 2 minutes ago — within the 5-minute safety window.
         recent_time = datetime.datetime(
-            2024, 12, 31, 23, 30, tzinfo=datetime.timezone.utc,
+            2025, 1, 1, 0, 2, tzinfo=datetime.timezone.utc,
         )
         schedule = ScheduledTask(
             task_name="test_task",
@@ -323,8 +324,8 @@ async def test_cleanup_keeps_recent_empty_entries(redis_url: str) -> None:
         await source.add_schedule(schedule)
         await source.delete_schedule(schedule.schedule_id)
 
-    # Run cleanup — entry is empty but only 30 min old.
-    with freeze_time("2025-01-01 00:00:00"):
+    # Run cleanup — entry is empty but only 2 minutes old.
+    with freeze_time("2025-01-01 00:04:00"):
         async with Redis(connection_pool=source._connection_pool) as redis:
             await source._cleanup_time_index(redis)
 
@@ -403,18 +404,18 @@ async def test_populate_time_index_from_existing_keys(redis_url: str) -> None:
 async def test_post_send_triggers_cleanup(redis_url: str) -> None:
     """Test the full lifecycle: add schedule, get it, post_send it,
     then verify cleanup (triggered from delete_schedule) removes
-    the stale index entry when it's > 1 hour old."""
+    the stale index entry when it's > 5 minutes old."""
     prefix = uuid.uuid4().hex
 
-    with freeze_time("2025-01-01 02:00:00"):
+    with freeze_time("2025-01-01 00:10:00"):
         source = ListRedisScheduleSource(redis_url, prefix=prefix)
         schedule = ScheduledTask(
             task_name="test_task",
             labels={},
             args=[],
             kwargs={},
             time=datetime.datetime(
-                2025, 1, 1, 0, 30, tzinfo=datetime.timezone.utc,
+                2025, 1, 1, 0, 0, tzinfo=datetime.timezone.utc,
             ),
         )
         await source.add_schedule(schedule)
@@ -424,7 +425,7 @@ async def test_post_send_triggers_cleanup(redis_url: str) -> None:
         assert schedules == [schedule]
 
         # post_send -> delete_schedule -> _maybe_cleanup_time_index.
-        # The entry is > 1 hour old and the list becomes empty,
+        # The entry is > 5 minutes old and the list becomes empty,
         # so cleanup should remove it.
         for s in schedules:
             await source.post_send(s)
@@ -433,7 +434,7 @@ async def test_post_send_triggers_cleanup(redis_url: str) -> None:
             assert await redis.zcard(source._get_time_index_key()) == 0
 
     # Second run should return nothing.
-    with freeze_time("2025-01-01 02:01:00"):
+    with freeze_time("2025-01-01 00:11:00"):
         schedules = await source.get_schedules()
         assert schedules == []
 
@@ -443,10 +444,10 @@ async def test_cleanup_rate_limited(redis_url: str) -> None:
     """Test that _maybe_cleanup_time_index only runs once per minute."""
     prefix = uuid.uuid4().hex
 
-    with freeze_time("2025-01-01 02:00:00"):
+    with freeze_time("2025-01-01 00:10:00"):
         source = ListRedisScheduleSource(redis_url, prefix=prefix)
         old_time = datetime.datetime(
-            2025, 1, 1, 0, 30, tzinfo=datetime.timezone.utc,
+            2025, 1, 1, 0, 0, tzinfo=datetime.timezone.utc,
         )
         sched1 = ScheduledTask(
             task_name="task1",