12. Batch/Coalesce API Calls

Problem

Fin's serving layer fields a steady stream of small inference requests. Rather than dispatching each one individually, it's more efficient to coalesce nearby requests into batches. Given a stream of incoming request timestamps (sorted ascending) and a max batch size plus a max wait window, group them into batches.

A batch closes when it reaches max_size OR the next request's timestamp would exceed max_wait from the batch's first request's timestamp, whichever comes first.

Source: src/12_coalesce_api_batches.py

def coalesce_batches(timestamps: list[float], max_size: int, max_wait: float) -> list[list[float]]: ...

>>> coalesce_batches([0, 1, 2, 3, 10, 11], max_size=3, max_wait=5)
[[0, 1, 2], [3], [10, 11]]
>>> coalesce_batches([0, 1, 2, 3], max_size=1, max_wait=100)
[[0], [1], [2], [3]]

Step-by-Step Approach

1. Because the input timestamps are already sorted ascending, you can process them with a single forward sweep — no sorting or lookahead structure needed.
2. Start a new batch at the current unbatched timestamp; call it first, the anchor for the wait-window check.
3. Greedily extend the batch with subsequent timestamps as long as two conditions both hold: the batch hasn't yet reached max_size, and the candidate timestamp is within max_wait of first (i.e. timestamps[j] - first <= max_wait).
4. As soon as either condition fails — batch is full, or the next timestamp would be too late relative to the batch's first element — close the batch and append it to the result.
5. Resume the outer scan from the first timestamp that didn't make it into the closed batch, and repeat until all timestamps are consumed.
6. Handle the empty-input case by returning an empty list of batches directly.

The key insight is that the wait window is always measured from a batch's first element, not the previous element — so it behaves like a fixed origin per batch rather than a true rolling gap, which is what lets you close a batch definitively and never have to revisit it (an O(n), single-pass greedy sweep, since every timestamp is examined exactly once as it's added to its batch).

Reference solution

def coalesce_batches(timestamps: list[float], max_size: int, max_wait: float) -> list[list[float]]:
    # Single linear sweep, greedily extending the current batch: O(n) time, O(1) extra space
    batches: list[list[float]] = []
    i = 0
    n = len(timestamps)
    while i < n:
        first = timestamps[i]
        batch = [first]
        j = i + 1
        # wait anchored to first (fixed origin), not previous elem; short-circuit avoids index error
        while j < n and len(batch) < max_size and timestamps[j] - first <= max_wait:
            batch.append(timestamps[j])
            j += 1
        batches.append(batch)
        # resume right where the closed batch left off, no re-scanning
        i = j
    return batches

Key Functions & Tricks

• Two-pointer sweep with i and j — i = j resumes exactly where the last batch closed, O(n) total.
• while j < n and len(batch) < max_size and timestamps[j] - first <= max_wait: — short-circuits left-to-right, so timestamps[j] is never evaluated out of range.
• Anchoring the wait check to first, not the previous element — fixed origin per batch, closes definitively in one pass.
• len(batch) — O(1) in CPython; a list's length is tracked, not counted each call.

How to Recognize This Pattern