Implementing a self-tuning limiter

First, build a controller that moves the concurrency ceiling itself. It does not send requests; it only manages the single number of how many calls are allowed in flight right now.

import asyncio
import time
 
class AdaptiveLimiter:
    def __init__(self, start=8, floor=2, ceil=96, cooldown=4.0):
        self.limit = float(start)   # current allowed concurrency
        self.floor = floor          # lower bound (avoid stalling)
        self.ceil = ceil            # upper bound (protect TPM)
        self.cooldown = cooldown    # seconds to suppress increase after a drop
        self._last_drop = 0.0
        self._sem = asyncio.Semaphore(start)
        self._lock = asyncio.Lock()
 
    def _resize(self, new_limit):
        new_limit = max(self.floor, min(self.ceil, new_limit))
        delta = int(new_limit) - int(self.limit)
        self.limit = new_limit
        # adjust the semaphore's permit count by the delta only
        if delta > 0:
            for _ in range(delta):
                self._sem.release()
        # shrinking is absorbed as in-flight permits return naturally
 
    async def on_success(self):
        async with self._lock:
            if time.monotonic() - self._last_drop < self.cooldown:
                return  # do not grow during cooldown
            self._resize(self.limit + 1)            # Additive Increase
 
    async def on_429(self):
        async with self._lock:
            self._last_drop = time.monotonic()
            self._resize(self.limit * 0.5)          # Multiplicative Decrease
 
    async def acquire(self):
        await self._sem.acquire()
 
    def release(self):
        self._sem.release()

The key is that on_429 records _last_drop and on_success skips the increase during cooldown. Without it, a success right after a halving flips you back into growth, the sawtooth oscillates at high frequency, and you keep stepping on 429s. This is exactly where I got burned in production: adding the cooldown changed the 429 rate by roughly threefold.

Draining the batch through a bounded worker pool

With the controller in hand, build a worker pool that actually moves the tasks. Do not hand the whole set to asyncio.gather at once: thirty thousand coroutines spawn simultaneously, pressure memory and the event loop, and leave concurrency entirely to the semaphore so behavior becomes hard to read. Instead, run a fixed number of workers that each pull one item at a time from a queue.

async def run_bulk(items, handler, limiter, workers=128):
    queue = asyncio.Queue()
    for it in items:
        queue.put_nowait(it)
    results, dead = [], []
 
    async def worker():
        while True:
            try:
                item = queue.get_nowait()
            except asyncio.QueueEmpty:
                return
            await limiter.acquire()
            try:
                out = await handler(item)      # one Gemini call
                await limiter.on_success()
                results.append((item, out))
            except RateLimited:
                await limiter.on_429()
                queue.put_nowait(item)         # 429: lower concurrency, requeue
            except PermanentError as e:
                dead.append((item, str(e)))    # unrecoverable: dead-letter
            finally:
                limiter.release()
                queue.task_done()
 
    await asyncio.gather(*[worker() for _ in range(workers)])
    return results, dead

The workers count is the logical upper bound; the limiter holds the effective concurrency. Even with 128 workers, if the semaphore only grants 12 permits, only 12 run at once. This two-layer design lets you provision plenty of workers for peak demand while the real flow rate is decided solely by what the API tells you.

Never throwing work away: dead-letters and resumable checkpoints

The most painful thing in bulk processing is getting to item 29,000, having the process die, and starting over. To prevent that, always keep a checkpoint that appends processed IDs as successes land. On re-run, load the processed set first and queue only the unprocessed items.

import json, pathlib
 
CKPT = pathlib.Path("checkpoint.jsonl")
 
def load_done():
    if not CKPT.exists():
        return set()
    return {json.loads(l)["id"] for l in CKPT.read_text().splitlines() if l}
 
def mark_done(item_id, payload):
    with CKPT.open("a") as f:
        f.write(json.dumps({"id": item_id, "out": payload}) + "\n")
 
# before launching run_bulk
done = load_done()
pending = [it for it in all_items if it["id"] not in done]

RateLimited lowers concurrency and requeues; 500/503/timeout retries in place with exponential backoff; and only failures that exceed the retry budget are quarantined into the dead-letter list (dead). When you eyeball the dead-letter the next morning, the cause is usually concentrated in one spot. In my case it was emoji-heavy reviews tripping the safety filter, and simply adding an input-sanitizing pre-step cut the dead-letter by about 40%. Keeping failures instead of discarding them stockpiles the seeds of exactly this kind of improvement.

Before / After: from fixed 64-way to adaptive control

On the real review-classification batch for the wallpaper apps (about 30,000 items, gemini-2.5-flash), the numbers moved clearly when I switched from fixed to adaptive concurrency.

At fixed 64-way, the 429 rate sat around 11%, dropped items (left unprocessed at the end) ran a steady 2,000 to 3,000, and a full run averaged 42 minutes. Because every 429 blindly retried the same item, wasted billed tokens piled up too.

After moving to adaptive control, concurrency hovered roughly between 18 and 40 depending on conditions, the 429 rate fell to 0.3%, dropped items went to zero thanks to checkpoints and the dead-letter, and a full run shortened to an average of 19 minutes. Even against the "safe" fixed 8-way (a 63-minute run), it is more than three times faster and does not break. Speed and safety were never a trade-off; the cause was pinning the flow rate to a fixed guess.

This carries straight over to any "drain it overnight" job, like monthly AdMob report aggregation or review monitoring. When a batch finishes in 19 minutes instead of 42, the whole morning sequence, down to the Slack notifications and roll-ups, gets a notch quieter.

A pre-production checklist

Finally, here is what I always confirm before putting this on a nightly batch. Keep the concurrency floor at 2 or more to avoid stalling. Derive the ceiling from TPM, so average tokens times max concurrency stays under your per-minute TPM budget. Never let non-429 errors trigger a decrease. Append to the checkpoint immediately on success, so buffering does not lose items. And run the practice of reviewing, not deleting, the dead-letter each morning.

The time I used to spend tuning a fixed value dropped to nearly zero with this design, because the line tracks the quota on its own even when requests get heavy. If you are also wrestling with a batch in the small hours, I hope it makes for a quieter morning.