Implement a batcher that flushes when the batch is full or when a timeout expires. What breaks in the timer path?

TL;DR: Give each submission a future and detach the pending batch when it fills or its first-item timer expires. Run the handler outside the queue lock, settle every caller on success or failure, and keep a caller's cancellation from cancelling work shared with its peers.

How to approach it

Explain when a batch starts and who owns its timer. Trace both a full batch and a lone request, then show how the handler's results reach the waiting callers. Test cancellation and failure separately from the successful path, because unresolved futures can leave clients waiting indefinitely.

A strong answer

Each caller receives a future, an object it can wait on until the batch handler sets its result or error. A batching proxy can pass a high-traffic test while failing when requests arrive alone. Full batches use the size trigger; a lone request relies on the timer. In a design where the timer also awaits the handler, cancelling that timer during its own flush can interrupt the work before callers receive their results. The implementation below separates queue management from handler execution.

The implementation:

import asyncio, math

class Batcher:
    """Flush when the batch is full or when max_wait elapses, whichever comes first."""

    def __init__(self, handler, max_size=8, max_wait=0.05):
        if type(max_size) is not int or max_size < 1 or not math.isfinite(max_wait) or max_wait < 0:
            raise ValueError("positive integer size and finite nonnegative wait required")
        self.handler, self.max_size, self.max_wait = handler, max_size, max_wait
        self._queue = []                 # list of (item, future)
        self._flush_task = None          # the pending timer, if any
        self._lock = asyncio.Lock()
        self._running = set()            # in-flight handler tasks, kept from GC

    async def submit(self, item):
        fut = asyncio.get_running_loop().create_future()
        async with self._lock:
            self._queue.append((item, fut))
            if len(self._queue) >= self.max_size:
                self._dispatch(self._detach_locked())
            elif self._flush_task is None:
                self._flush_task = asyncio.create_task(self._timer())
        return await fut                 # the caller waits outside the lock

    async def _timer(self):
        await asyncio.sleep(self.max_wait)
        async with self._lock:
            self._dispatch(self._detach_locked())

    def _detach_locked(self):
        """Take the pending batch and cancel the timer. Cheap, synchronous, and
        the ONLY thing that happens under the lock."""
        if not self._queue:
            return None
        batch, self._queue = self._queue, []
        # take the reference and clear it BEFORE cancelling, and never cancel the task we
        # are currently running inside: that is the self-cancellation bug
        task, self._flush_task = self._flush_task, None
        if task is not None and task is not asyncio.current_task():
            task.cancel()
        return batch

    def _dispatch(self, batch):
        """Run the handler in its own task. The batch's fate must not depend on
        whichever caller happened to fill it: if that caller is cancelled while
        awaiting the handler, its peers would otherwise wait forever."""
        if not batch:
            return
        t = asyncio.create_task(self._run(batch))
        self._running.add(t)
        t.add_done_callback(self._running.discard)

    async def _run(self, batch):
        items = [i for i, _ in batch]
        try:
            results = list(await self.handler(items))
            if len(results) != len(batch):
                raise RuntimeError(
                    f"handler returned {len(results)} results for {len(batch)} items")
        except asyncio.CancelledError as exc:
            # NOT caught by `except Exception`: CancelledError derives from
            # BaseException. This is the path that used to strand a batch.
            self._settle(batch, exc=exc)
            raise
        except Exception as exc:
            self._settle(batch, exc=exc)   # every caller in the batch gets the failure
            return
        for (_, fut), r in zip(batch, results):
            if not fut.done():
                fut.set_result(r)

    @staticmethod
    def _settle(batch, exc):
        for _, fut in batch:
            if not fut.done():
                fut.set_exception(exc)

Running it:

10 submissions with max_size 4   -> [0, 2, 4, 6, 8, 10, 12, 14, 16, 18]
                                     the last two flushed on the timer
a single submission              -> flushed on the timer, returns
handler raising                  -> every caller in the batch receives RuntimeError
caller cancelled mid-flush       -> its PEERS still resolve
a third submit during a flush    -> enqueues immediately, timer starts
handler returns the wrong count  -> every caller gets a RuntimeError, none hang

Test each of these outcomes with a bounded deadline so a broken implementation cannot leave the test hanging.

The timer bug appears in a different design: the timer calls a flush method that cancels the timer and then awaits the handler. It has just cancelled itself. Cancellation is delivered at a subsequent suspension point, potentially before the handler settles the callers' futures. A size-triggered test can miss the problem because a separate submission task initiated that flush.

Here, detaching the batch is synchronous, and the handler runs in its own task. Clearing the timer reference before cancelling it prevents later code from mistaking the old timer for a pending one. The current-task check avoids self-cancellation. Keeping handler execution separate also means cancellation of the caller that filled a batch cannot strand its peers.

Batching Queues and Backpressure covers the pattern; Continuous Batching is the inference-specific version where the batch is re-formed every step rather than once.

The batching window delays dispatch, and event-loop scheduling can delay it further. It is therefore a target for when to start flushing, not a hard bound on response latency. Handler time and any downstream queueing come afterward.

Illustrative regular arrivals: max_size=8, max_wait=50 ms
200 requests/s: seven more arrivals take 7/200 = 35 ms
20 requests/s: seven more arrivals take 7/20 = 350 ms
At the lower rate, the 50 ms timer target arrives before the batch fills.

These are regular-arrival examples, not expected waits under arbitrary traffic. With Poisson arrivals, the uncapped expected time for seven more arrivals is also 7/r, but comparing that mean with the timeout does not tell us which trigger always wins. The first request and later requests also spend different times waiting. Measure the distribution of queue residence and completed response latency under the actual arrivals.

Agree on the handler's error contract. A raised exception can fail the whole batch, as it does here. A handler that returns per-item outcomes can instead fail one item while preserving the others. That is useful only when the backend can identify which operations succeeded.

Every unresolved future must receive an outcome on an error path. The code converts and validates the handler's result inside the try block, so None, a malformed iterable or the wrong result count cannot leave callers waiting. It handles CancelledError separately because that exception derives from BaseException. Already completed or cancelled futures are skipped.

A cancelled caller's item may still execute in the shared handler. The example protects its peers; it does not promise that cancellation reverses a side effect.

Batch when combining work improves measured throughput enough to justify the added wait. Shared weight reads can make inference batching useful, but compute, cache traffic and memory limits still change with batch size. A database handler may also benefit from a bulk operation. If the handler simply repeats independent work with no saving, reduce the window or skip batching.

This example leaves admission limits and graceful shutdown to the caller. Limiting only the pending queue is insufficient because dispatched batches can accumulate in _running; bound total admitted work or handler concurrency as well. A shutdown protocol must stop admission, drain or cancel pending and running work, and settle every remaining future.

What interviewers probe next

• "Why start the timer on the first item rather than on every item?" Because restarting it per item means a steady stream of arrivals never triggers the timer, and a batch could wait indefinitely.
• "What if the handler is slow?" Items arriving during a flush go into the new queue and start their own timer, so the batcher pipelines. Bounding the queue is a separate backpressure decision.
• "How would you add backpressure?" Bound total admitted work, including running batches, and define whether excess submissions wait or are rejected. A pending-queue limit alone does not bound handler tasks.
• "How do you test the timer path?" Drive the event loop with a controllable clock, or submit fewer items than max_size and assert the flush happens within a tolerance of max_wait. Use bounded waits so a failed test cannot hang indefinitely.

Common mistakes

• Cancelling the timer task from inside itself, which hangs every caller in that batch and only at low load.
• Restarting the timer on every submission, so a steady stream never flushes on time.
• Leaving a future unresolved on an error path, which hangs one caller silently.
• Choosing max_wait without reference to the arrival rate, so the batcher adds its full window and delivers batches of two.

Key takeaways

• Two triggers, one shared queue, one future per caller; the caller awaits outside the lock.
• The timer must not cancel itself: compare against the current task before cancelling, and clear the reference first.
• max_wait sets a dispatch target; event-loop delay, handler time and downstream queues add to response latency.
• Every exit path must resolve every future in the batch, and whether one failure fails all of them is a contract decision.