Concurrent schedule deletion empties a Cluster source refresh #147

A normal concurrent delete_schedule() can make RedisClusterScheduleSource.get_schedules() raise TypeError, discarding unrelated valid schedules from that refresh. The source scans keys and then GETs them individually; a key removed after SCAN returns it yields None from GET, which is passed straight to the serializer. The standalone/Sentinel schedule sources already filter missing values.

The actual Taskiq scheduler helper (taskiq.cli.scheduler.run.get_schedules) catches this exception and returns an empty list. Thus the scheduler replaces this source's snapshot with no schedules until a later refresh. This does not crash the scheduler process, but can suppress valid due tasks during that interval.

Reproduction

Main 8f70bc85dd6a469cae7d3b467af7dea720c92434, Windows 11, Python 3.14.7, taskiq 0.13.0, redis-py 8.0.0 and an isolated real three-primary Redis 8.2.11 Cluster. TEST_REDIS_CLUSTER_URL points at a reachable startup node with all advertised addresses reachable.

This inserts an event barrier between a real SCAN result and its subsequent GET. It uses two real source/client instances and actual SCAN/DELETE/GET calls; no keys or Redis results are fabricated. The iterator wrapper only controls the timing and is restored afterward. It exercises the source directly and then the actual scheduler snapshot helper, not a full scheduler daemon.

import asyncio,os,uuid
from redis.asyncio import RedisCluster
from taskiq import ScheduledTask
from taskiq.cli.scheduler.run import get_schedules
from taskiq_redis import RedisClusterScheduleSource

async def scenario(via_scheduler):
    prefix=uuid.uuid4().hex
    reader=RedisClusterScheduleSource(os.environ['TEST_REDIS_CLUSTER_URL'],prefix=prefix)
    writer=RedisClusterScheduleSource(os.environ['TEST_REDIS_CLUSTER_URL'],prefix=prefix)
    gone=ScheduledTask(schedule_id='{0}:gone',task_name='gone',labels={},args=[],kwargs={},cron='* * * * *')
    live=ScheduledTask(schedule_id='{1}:live',task_name='live',labels={},args=[],kwargs={},cron='* * * * *')
    scanned,deleted=asyncio.Event(),asyncio.Event()
    original=RedisCluster.scan_iter
    key=f'{prefix}:{gone.schedule_id}'.encode()
    async def coordinated(client,*args,**kwargs):
        async for found in original(client,*args,**kwargs):
            if client is reader.redis and found==key:
                scanned.set()
                await deleted.wait()
            yield found
    pending=None
    try:
        await writer.add_schedule(gone)
        await writer.add_schedule(live)
        RedisCluster.scan_iter=coordinated
        pending=asyncio.create_task(get_schedules(reader) if via_scheduler else reader.get_schedules())
        await asyncio.wait_for(scanned.wait(),5)
        await writer.delete_schedule(gone.schedule_id)
        deleted.set()
        try:
            result=await asyncio.wait_for(pending,5)
            print('scheduler helper' if via_scheduler else 'source','returned:',[x.task_name for x in result],flush=True)
        except Exception as exc:print('source raised:',type(exc).__name__,str(exc),flush=True)
        print('unrelated live schedule still exists:',bool(await writer.redis.exists(f'{prefix}:{live.schedule_id}')),flush=True)
    finally:
        deleted.set()
        RedisCluster.scan_iter=original
        if pending is not None:
            pending.cancel()
            await asyncio.gather(pending,return_exceptions=True)
        for schedule in (gone,live):await writer.delete_schedule(schedule.schedule_id)
        await reader.shutdown()
        await writer.shutdown()

async def main():
    await scenario(False)
    await scenario(True)

asyncio.run(main())

Observed:

source raised: TypeError a bytes-like object is required, not 'NoneType'
unrelated live schedule still exists: True
scheduler helper returned: []
unrelated live schedule still exists: True

Expected: the deleted schedule is omitted; the live schedule remains in the returned snapshot. SCAN is not an atomic snapshot, so a missing GET is an ordinary concurrent-deletion outcome. Genuine malformed serialized records should still report an error.

I would like to guard only raw_schedule is None before decoding and add regressions for direct/scheduler-helper reads, surviving/all-removed records, later re-addition, and a corrupt-payload control. This changes no public API and does not redesign the scan/index approach discussed in #78/#121/#122. It is also independent of the shared result-client lifecycle in #145/#146.

I checked open/recent closed issues and PRs, including schedule/delete/None searches, and found no overlapping deletion-race fix. Could a maintainer assign this to cw-jlu? Investigation, reproduction and this report used OpenAI Codex assistance. I plan to submit a Draft PR pending independent and account-owner review.