Slotting Architecture · 11 min read

How to Configure Async Batch Recalculation for 1M+ SKUs

Recalculating velocity for a million-plus SKUs is not a scaling problem you solve by adding threads — it is a coordination problem. Load every SKU’s history into memory at once and the job dies with an OOM kill three hours in. Fire a million database round-trips with an unbounded asyncio.gather and you flood the connection pool, starve the event loop, and crater the same WMS the daytime pickers depend on. Skip checkpointing and a crash at 90% means starting over from zero, blowing the nightly maintenance window. This page builds a batch recalculation driver that chunks the SKU list, bounds concurrency with an asyncio.Semaphore, checkpoints completed chunks for resumability, and applies backpressure so the database is never overrun. It is the bulk-recompute counterpart within the Async Batch Processing for Velocity guide, part of the larger Velocity Data Ingestion & WMS Sync Pipelines architecture.

Chunked queue feeding a semaphore-bounded worker pool with a checkpoint store A left-to-right pipeline. A source of roughly one million SKU ids feeds a chunker that splits them into batches of 5000, producing about 200 chunks. The chunks pass through an asyncio.Semaphore gate configured with max concurrency 8, which admits at most eight chunks at a time into a worker pool of recalculation workers. Each worker recomputes velocity for its batch against the database, then writes the highest contiguously completed chunk index to a checkpoint store holding last_completed_chunk. A dashed feedback arc runs from the checkpoint store back to the chunker labelled resume from cursor, so a crashed run restarts at the next unprocessed chunk instead of rescoring the whole catalog. Chunk → bounded worker pool → checkpoint SKU id stream ~1,000,000 ids Chunker chunk_size = 5000 → 200 Semaphore max_concurrency = 8 at most 8 in flight worker pool recalc worker recalc worker recalc worker recalc worker checkpoint last chunk resume from cursor → next unprocessed chunk

Prerequisites

Confirm each before scheduling the job into your maintenance window:

  • Python 3.10+ — the driver uses asyncio, X | None unions, list[...] generics, and a dataclass config.
  • An async data access layer — an async database driver (asyncpg, aiomysql) or an async HTTP client. If the recalculation itself is CPU-bound, plan to offload it to a process pool (see the event-loop pitfall below).
  • A durable, writable checkpoint path — a local file on a persistent volume, or a small row in a control table. It must survive the process crashing, or resumability buys you nothing.
  • A known nightly budget — the wall-clock window (say four hours) and the concurrency the database tolerates without hurting daytime transactions. These two numbers set max_concurrency and chunk_size.
  • A source cadence you control — align the recalc window with when fresh data actually lands by tuning the WMS/ERP polling strategy so you are not rescoring against a half-synced feed.

Configuration Block

Three parameters govern throughput, safety, and resumability. chunk_size trades memory against per-chunk overhead; max_concurrency is the hard ceiling on simultaneous database load; checkpoint_path is where progress survives a crash.

# batch_recalc.yaml — one profile per facility recalculation job
batch_recalc:
  chunk_size: 5000                       # SKUs per batch; sets peak memory per in-flight chunk
  max_concurrency: 8                     # asyncio.Semaphore ceiling; the hard backpressure limit
  checkpoint_path: "velocity_checkpoint.json"  # durable cursor; last contiguously completed chunk
# Equivalent Python config dict consumed by the driver
from pathlib import Path

BATCH_RECALC = {
    "chunk_size": 5000,
    "max_concurrency": 8,
    "checkpoint_path": Path("velocity_checkpoint.json"),
}

Implementation

The driver never materializes the whole catalog: it iterates SKU ids in chunk_size slices, wraps each chunk’s work in a Semaphore-guarded coroutine, and advances the checkpoint only across contiguously completed chunks. That last detail is what makes resume correct — if chunk 7 finishes before chunk 5, the durable cursor stays at 4 until 5 lands, so a crash never skips an unfinished chunk.

from __future__ import annotations

import asyncio
import json
import logging
from dataclasses import dataclass
from pathlib import Path
from typing import Awaitable, Callable, Iterator

logger = logging.getLogger("velocity.batch")


@dataclass(frozen=True)
class BatchConfig:
    chunk_size: int = 5000
    max_concurrency: int = 8
    checkpoint_path: Path = Path("velocity_checkpoint.json")


def _chunks(sku_ids: list[str], size: int) -> Iterator[tuple[int, list[str]]]:
    for start in range(0, len(sku_ids), size):
        yield start // size, sku_ids[start:start + size]


def _load_cursor(path: Path) -> int:
    if path.exists():
        return int(json.loads(path.read_text())["last_completed_chunk"])
    return -1  # nothing done yet


async def recalc_all(
    sku_ids: list[str],
    recalc_chunk: Callable[[list[str]], Awaitable[None]],
    cfg: BatchConfig,
) -> int:
    """Recalculate velocity for every SKU with bounded concurrency and resumable checkpoints.

    SKUs are split into fixed-size chunks; an ``asyncio.Semaphore`` caps how many chunks run
    at once so the database and event loop are never flooded. The durable cursor advances only
    across contiguously completed chunks, so a crash resumes at the first unfinished chunk
    instead of rescoring the whole catalog. Returns the count of chunks completed this run.
    """
    sem = asyncio.Semaphore(cfg.max_concurrency)
    lock = asyncio.Lock()
    resume_from = _load_cursor(cfg.checkpoint_path)
    done = resume_from
    completed: set[int] = set()

    async def worker(index: int, batch: list[str]) -> None:
        nonlocal done
        async with sem:                       # backpressure: at most max_concurrency in flight
            await recalc_chunk(batch)
        async with lock:                      # single writer advances the contiguous cursor
            completed.add(index)
            while (done + 1) in completed:
                done += 1
            cfg.checkpoint_path.write_text(json.dumps({"last_completed_chunk": done}))
        logger.info("chunk %d done (%d skus), checkpoint=%d", index, len(batch), done)

    tasks = [
        asyncio.create_task(worker(i, batch))
        for i, batch in _chunks(sku_ids, cfg.chunk_size)
        if i > resume_from
    ]
    logger.info("scheduling %d chunks (resume_from=%d, conc=%d)",
                len(tasks), resume_from, cfg.max_concurrency)
    await asyncio.gather(*tasks)
    return len(tasks)

Step-by-Step Walkthrough

  1. Slice, never load. _chunks yields (index, batch) tuples by slicing the id list in chunk_size steps. Only the ids live in memory here; each worker fetches that chunk’s history on demand, so peak memory is bounded by chunk_size * max_concurrency, not the catalog size.
  2. Read the resume cursor first. _load_cursor reads last_completed_chunk from checkpoint_path (or -1 on a fresh run). The task list comprehension then skips every chunk with i <= resume_from, so a restarted job schedules only the unfinished tail.
  3. Bound concurrency with the Semaphore. Every worker enters async with sem before touching the database. With max_concurrency = 8, at most eight chunks are ever in flight regardless of how many tasks exist — this is the backpressure valve that protects the connection pool.
  4. Advance the cursor contiguously. Under a single asyncio.Lock, each finished chunk joins the completed set, then done walks forward only while the next index is present. Out-of-order completions are held back, guaranteeing the persisted cursor is a safe restart point.
  5. Gather and report. asyncio.gather awaits the scheduled tail. Because the Semaphore — not gather — governs concurrency, gathering hundreds of task handles is cheap; the tasks simply queue on the semaphore until a permit frees.

Verification

Drive the whole thing with a stub recalc_chunk that does no real work, over 25,000 SKUs at chunk_size = 5000. Expect five chunks completed and a checkpoint file pinned at index 4.

import asyncio
import json
import logging
from pathlib import Path

logging.basicConfig(level=logging.INFO)


async def _fake_recalc(batch: list[str]) -> None:
    await asyncio.sleep(0)   # stands in for an async DB round-trip

cp = Path("cp.json")
cp.unlink(missing_ok=True)
cfg = BatchConfig(chunk_size=5000, max_concurrency=4, checkpoint_path=cp)
skus = [f"SKU-{i}" for i in range(25_000)]

completed = asyncio.run(recalc_all(skus, _fake_recalc, cfg))
print(completed, json.loads(cp.read_text()))
assert completed == 5
assert json.loads(cp.read_text())["last_completed_chunk"] == 4

Sample expected output (chunk order may vary; the checkpoint always ends at 4):

INFO:velocity.batch:scheduling 5 chunks (resume_from=-1, conc=4)
INFO:velocity.batch:chunk 0 done (5000 skus), checkpoint=0
INFO:velocity.batch:chunk 1 done (5000 skus), checkpoint=1
INFO:velocity.batch:chunk 2 done (5000 skus), checkpoint=2
INFO:velocity.batch:chunk 3 done (5000 skus), checkpoint=3
INFO:velocity.batch:chunk 4 done (5000 skus), checkpoint=4
5 {'last_completed_chunk': 4}

Common Pitfalls

  • Memory blowup from eager loading. Fetching every SKU’s pick history up front — one giant query or a list comprehension over the whole catalog — is the classic OOM. Keep the driver iterating over ids only and let each worker pull its own chunk’s rows, so resident memory tracks chunk_size * max_concurrency rather than the million-row total.
  • Unbounded gather. await asyncio.gather(*[recalc(s) for s in all_skus]) with no semaphore launches every task’s I/O at once, exhausting the connection pool and triggering the WMS’s implicit throttle. The Semaphore is not optional decoration — it is the only thing between you and a self-inflicted outage. Bounding concurrency is why this pairs with an adaptive WMS/ERP polling strategy upstream.
  • No checkpoint, or one written before the chunk commits. Without a durable cursor a crash at 90% restarts from zero. Just as bad is persisting the cursor before the chunk’s writes commit — a crash then skips uncommitted work. Advance the checkpoint only after recalc_chunk returns, and only across contiguous indices.
  • Event-loop starvation from CPU-bound work. If recalc_chunk does heavy pandas/NumPy math directly in the coroutine it blocks the single event loop, and your eight “concurrent” chunks run serially. Offload the CPU portion with loop.run_in_executor(ProcessPoolExecutor(), ...) so await actually yields — otherwise the streaming, event-driven approach in Real-Time Streaming Velocity Updates may be the better tool for incremental recompute.

FAQ

How do I pick chunk_size and max_concurrency together?

Treat max_concurrency as the database’s limit and chunk_size as the memory limit, then let the wall-clock budget referee. Start from the connection pool: max_concurrency should sit at or below the pool size so no worker waits on a connection. Set chunk_size so that chunk_size * max_concurrency rows fit comfortably in memory with headroom. Measure one run, then scale chunk_size up to cut per-chunk overhead until you either hit the memory ceiling or stop gaining throughput.

Can this resume across separate nightly runs, not just crashes?

Yes — the checkpoint is just a durable cursor, so a run that exhausts its window can exit cleanly and the next night’s invocation reads last_completed_chunk and continues from the tail. For a true multi-night sweep, reset the cursor only when you start a fresh full recalculation; otherwise each run appends progress until the catalog is complete.