Add a batch scheduling API: call() + add_many()/replace_many()

[bunchesofdonald]

Why

Docket's scheduling API was strictly one-execution-at-a-time: every add()/replace() costs ~3 Redis round-trips (per-key lock SET, _schedule EVALSHA, lock-release EVALSHA), so callers that fan out N schedules pay N×3 sequential round-trips on a single-threaded event loop. Motivating case: a caller replacing up to 200 keyed tasks per poll — up to ~600 sequential Redis ops.

Linear: CLOUD-4263

What

await docket.replace_many(
    docket.call(check_freshness, when=when, key=f"freshness-{d.id}")(d.id)
    for d in deployments
)

docket.call(...) mirrors the currying (and argument type-checking) of add/replace but returns a TaskCall spec; add_many()/replace_many() send the batch to Redis in pipelined chunks — O(N / chunk_size) round-trips instead of O(N).

Per-execution semantics are unchanged: each task is individually strike-checked, deduplicated by key, and scheduled atomically by the same _schedule Lua script, with the same metrics counters. Tracing emits one batch span (docket.add_many/docket.replace_many with docket.batch.count) instead of N spans. There is no atomicity across the batch — by design.

Notable details

• @redis_script now returns a RedisScript object instead of a wrapped function. Not out of any love for classes — the decorated script grew from one operation to three (__call__ for the immediate EVALSHA, enqueue() for the pipelined EVALSHA, enqueue_eval() for cluster's full-source EVAL) plus two pieces of data callers need (lua for SCRIPT LOAD, sha). The closure version was already leaking under one extra: wrapper.__lua__ = lua needed a # type: ignore, because pyright can't see attributes on a function type — so every _schedule.enqueue(...) call site would have needed a cast or a hand-maintained Protocol, and none of it would share the decorator's ParamSpec. As a generic RedisScript[P, R] (via Concatenate), all three entry points get the same fully-typed keyword contract as the declared function, checked by pyright with zero per-call-site annotations, and they share one _keys_and_argv so the KEYS/ARGV encoding (and the new positional-args guard) can't drift between the immediate and pipelined forms. It's a closure with names: no inheritance, no mutable state beyond what the old decorator closed over, and the hot call path is byte-for-byte the same strategy (pre-computed SHA, same encode loop, still bypassing Signature.bind).
• Redis Cluster uses full-source EVAL for the batch: redis-py blocks pipelined EVALSHA in cluster mode outright, and a node's script cache can't be relied on across failover/resharding anyway. Verified against a real local cluster; batch tests run un-skipped on the cluster CI leg.
• Per-execution error capture: pipelines run with raise_on_error=False; a Redis error for one task marks just that execution Disposition.FAILED with the exception on Execution.schedule_exception, so callers always know exactly which tasks landed.
• chunk_size (default 1,000; None to disable) bounds client-side buffering per round-trip, and composes with error capture: a failure in a later chunk leaves earlier chunks fully accounted for.
• The batch path skips the per-key {key}:lock the single path takes: the lock guards nothing beyond the single atomic script call (no other code acquires it) and pipelined invocations already serialize in Redis. Removing the vestigial lock from the single path is a possible follow-up, as is converting Agenda.scatter() to use add_many internally.
• Execution.schedule() was refactored into shared script-args/reply-application halves used by both paths, and _check_stricken/_record_schedule_metrics collapse the strike/metrics triplication in Docket.

Testing

• 26 new tests in tests/test_batch_scheduling.py (dispositions, ordering, dedup within and across batches, strikes, replace semantics, wire-level round-trip counting, chunking, chunk validation, per-execution and cross-chunk error capture), plus batch counter tests, batch span tests, and RedisScript enqueue/guard tests.
• Verified locally at 100% coverage with CI's exact flags on three legs: memory (855 passed), Redis 8.0, and Redis 8.6 cluster (860 passed).

🤖 Generated with Claude Code

[read-the-docs-community]

Documentation build overview

📚 docket | 🛠️ Build #33415650 | 📁 Comparing 1c30e06 against latest (67f9394)

  🔍 Preview build  

3 files changed

± api-reference/index.html
± getting-started/index.html
± task-patterns/index.html

[codecov-commenter]

❌ 1 Tests Failed:

Tests completed	Failed	Passed	Skipped
956	1	955	7

View the top 1 failed test(s) by shortest run time

tests/test_progress_pubsub.py::test_run_subscribe_both_state_and_progress

Stack Traces | 2.14s run time

async def collect_events():
>       async for event in execution.subscribe(ready=subscribed):  # pragma: no cover

tests/test_progress_pubsub.py:102: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
src/docket/execution.py:1301: in subscribe
    async for message in pubsub.listen():  # pragma: no cover
.venv/lib/python3.11.../redis/asyncio/client.py:1451: in listen
    response = await self.handle_message(await self.parse_response(block=True))
                                         ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
.venv/lib/python3.11.../redis/asyncio/client.py:1295: in parse_response
    response = await self._execute(
.venv/lib/python3.11.../redis/asyncio/client.py:1201: in _execute
    response = await conn.retry.call_with_retry(
.venv/lib/python3.11.../redis/asyncio/retry.py:69: in call_with_retry
    return await do()
           ^^^^^^^^^^
.venv/lib/python3.11.../redis/asyncio/connection.py:792: in read_response
    response = await self._parser.read_response(
.venv/lib/python3.11.../redis/_parsers/resp3.py:185: in read_response
    response = await self._read_response(
.venv/lib/python3.11.../redis/_parsers/resp3.py:197: in _read_response
    raw = await self._readline()
          ^^^^^^^^^^^^^^^^^^^^^^
.venv/lib/python3.11.../redis/_parsers/base.py:578: in _readline
    data = await self._stream.readline()
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
../............../_temp/uv-python-dir/cpython-3.11.15-linux-x86_64-gnu/lib/python3.11/asyncio/streams.py:566: in readline
    line = await self.readuntil(sep)
           ^^^^^^^^^^^^^^^^^^^^^^^^^
../............../_temp/uv-python-dir/cpython-3.11.15-linux-x86_64-gnu/lib/python3.11/asyncio/streams.py:658: in readuntil
    await self._wait_for_data('readuntil')
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <StreamReader eof transport=<_SelectorSocketTransport closed fd=25>>
func_name = 'readuntil'

    async def _wait_for_data(self, func_name):
        """Wait until feed_data() or feed_eof() is called.
    
        If stream was paused, automatically resume it.
        """
        # StreamReader uses a future to link the protocol feed_data() method
        # to a read coroutine. Running two read coroutines at the same time
        # would have an unexpected behaviour. It would not possible to know
        # which coroutine would get the next data.
        if self._waiter is not None:
            raise RuntimeError(
                f'{func_name}() called while another coroutine is '
                f'already waiting for incoming data')
    
        assert not self._eof, '_wait_for_data after EOF'
    
        # Waiting for data while paused will make deadlock, so prevent it.
        # This is essential for readexactly(n) for case when n > self._limit.
        if self._paused:
            self._paused = False
            self._transport.resume_reading()
    
        self._waiter = self._loop.create_future()
        try:
>           await self._waiter
E           asyncio.exceptions.CancelledError

../............../_temp/uv-python-dir/cpython-3.11.15-linux-x86_64-gnu/lib/python3.11/asyncio/streams.py:543: CancelledError

The above exception was the direct cause of the following exception:

execution = <docket.execution.Execution object at 0x7f6ec2aa3f50>

    async def test_run_subscribe_both_state_and_progress(execution: Execution):
        """Run.subscribe() should yield both state and progress events."""
        # Set up subscriber in background
        all_events: list[StateEvent | ProgressEvent] = []
        subscribed = asyncio.Event()
    
        async def collect_events():
            async for event in execution.subscribe(ready=subscribed):  # pragma: no cover
                all_events.append(event)
                # Stop after we get a running state and some progress
                if (
                    len(
                        [
                            e
                            for e in all_events
                            if e["type"] == "state" and e["state"] == ExecutionState.RUNNING
                        ]
                    )
                    > 0
                    and len([e for e in all_events if e["type"] == "progress"]) >= 3
                ):
                    break
    
        subscriber_task = asyncio.create_task(collect_events())
    
        # Wait for the SUBSCRIBE to be acknowledged.
        await subscribed.wait()
    
        # Publish mixed state and progress events
        await execution.claim("worker-1")
        await execution.progress.set_total(50)
        await execution.progress.increment(5)
    
        # Wait for subscriber to collect events
>       await asyncio.wait_for(subscriber_task, timeout=2.0)

tests/test_progress_pubsub.py:129: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

fut = <Task cancelled name='Task-3209' coro=<test_run_subscribe_both_state_and_progress.<locals>.collect_events() done, defined at .../docket/tests/test_progress_pubsub.py:101>>
timeout = 2.0

    async def wait_for(fut, timeout):
        """Wait for the single Future or coroutine to complete, with timeout.
    
        Coroutine will be wrapped in Task.
    
        Returns result of the Future or coroutine.  When a timeout occurs,
        it cancels the task and raises TimeoutError.  To avoid the task
        cancellation, wrap it in shield().
    
        If the wait is cancelled, the task is also cancelled.
    
        This function is a coroutine.
        """
        loop = events.get_running_loop()
    
        if timeout is None:
            return await fut
    
        if timeout <= 0:
            fut = ensure_future(fut, loop=loop)
    
            if fut.done():
                return fut.result()
    
            await _cancel_and_wait(fut, loop=loop)
            try:
                return fut.result()
            except exceptions.CancelledError as exc:
                raise exceptions.TimeoutError() from exc
    
        waiter = loop.create_future()
        timeout_handle = loop.call_later(timeout, _release_waiter, waiter)
        cb = functools.partial(_release_waiter, waiter)
    
        fut = ensure_future(fut, loop=loop)
        fut.add_done_callback(cb)
    
        try:
            # wait until the future completes or the timeout
            try:
                await waiter
            except exceptions.CancelledError:
                if fut.done():
                    return fut.result()
                else:
                    fut.remove_done_callback(cb)
                    # We must ensure that the task is not running
                    # after wait_for() returns.
                    # See https://bugs.python.org/issue32751
                    await _cancel_and_wait(fut, loop=loop)
                    raise
    
            if fut.done():
                return fut.result()
            else:
                fut.remove_done_callback(cb)
                # We must ensure that the task is not running
                # after wait_for() returns.
                # See https://bugs.python.org/issue32751
                await _cancel_and_wait(fut, loop=loop)
                # In case task cancellation failed with some
                # exception, we should re-raise it
                # See https://bugs.python.org/issue40607
                try:
                    return fut.result()
                except exceptions.CancelledError as exc:
>                   raise exceptions.TimeoutError() from exc
E                   TimeoutError

../............../_temp/uv-python-dir/cpython-3.11.15-linux-x86_64-gnu/lib/python3.11/asyncio/tasks.py:502: TimeoutError

To view more test analytics, go to the Test Analytics Dashboard
_{📋 Got 3 mins? Take this short survey to help us improve Test Analytics.}

[chrisguidry]

A thing of beauty