Adaptive Batch Size Tuning Under Load

When to Use

Invoke this skill when persistently writing high-volume tick feeds or trading logs into downstream databases (TimescaleDB, ClickHouse, InfluxDB) or message queues (Kafka, Redis Streams). Hardcoding a static batch size is wrong across regimes — it causes high persistence latency when markets are quiet (waiting for fixed batch limits to fill) and DB I/O overload during flash crashes.

The skill produces an AdaptiveBatchTunerEngine whose job is to scale the batch size B_t and flush timeout T_flush in response to two signals:

1. Smoothed batch fill ratio at the flush boundary F_ewma = EWMA(depth_at_flush / B_t) — did the producer fill the batch before the timeout, or did the timeout cut a half-empty batch?
2. Smoothed downstream write latency (target: target_write_latency_ms), which acts as the throttle and outranks signal 1.

Why fullness, and not queue depth

The obvious signal — buffer occupancy against a nominal queue_capacity — does not work in this architecture, and getting this wrong inverts the controller. add_item hands the batch back the instant the buffer reaches B, so the buffer depth is bounded by B by construction: depth / queue_capacity can never exceed B / queue_capacity and therefore measures the tunable, not the load. Tuning on it is positive feedback on B itself, and under saturation it drives B down to B_min — the exact opposite of the intent. Batch fullness at the flush boundary is the signal that actually separates "the producer outran the batch size" (F = 1.0 ⇒ high load) from "the timeout expired half-empty" (F < 1.0 ⇒ low load).

When NOT to Use

• Single-shot or batch-mode loads — if the total record count is bounded and known upfront, fine-tune one batch size statically. The adaptive engine adds no value when input is finite.
• Synchronous request/response flows — this is fire-and-forget throughput tuning, not per-request optimisation. Use RPC-style tuning instead.
• Distributed stream processing stages (Flink, ksqlDB, Spark Structured Streaming) — they have their own internal flush predicates; this engine is for the client-side write path into the sink, not for the consumer side of an internal stage.
• Latency-critical tick-to-trade paths — those need synchronously bounded writes; the entire batch-and-flush taxonomy is wrong. See tick-to-trade-latency-measurement instead.
• When the sink is a queue with built-in batching (Kafka producer with linger.ms/batch.size) AND the queue is the only consumer — let Kafka's producer do the tuning.
• When strict cross-batch write ordering is required across multiple producer threads — the engine detaches each batch before returning it and runs on_flush outside its lock, so concurrent producers may write out of order. Use a single producer thread, or serialise inside your callback.

Prerequisites

• A downstream sink write function accepting batches of records (caller loop).
• Capacity constants declared up front:
  • B_min — minimum batch size (records; default 10)
  • B_max — maximum batch size (records; default 1000)
  • T_min / T_max — flush-timeout bounds (default 50ms / 1000ms; aligned with ClickHouse's adaptive-busy-timeout range)
  • L_target — downstream write-latency target (default 50ms). This is the binding constraint on expansion; set it to a latency your sink can actually sustain, not an aspiration.
• Queue bounds: max_queue_size is the hard cap that raises QueueFullError; queue_capacity is the denominator of the exported backpressure gauge and must be <= max_queue_size (validated).
• Sink latency instrumentation — caller must call record_write_latency(ms) on every write, successful or failed. Without it the throttle never fires and the batch expands until it hits B_max.
• A scheduler tick if the producer can go quiet — see Workflow step 5.

Workflow

1. Construct the engine with a TuningConfig:

   python

   copy

   from batch_tuner import AdaptiveBatchTunerEngine, TuningConfig
   
   tuner = AdaptiveBatchTunerEngine(TuningConfig(
       min_batch_size=10,
       max_batch_size=1000,
       initial_batch_size=100,
       target_write_latency_ms=50.0,
       queue_capacity=2000,
       max_queue_size=5000,
   ))

2. Produce-loop pattern (the contract):

   python

   copy

   try:
       batch = tuner.add_item(item)
   except QueueFullError as exc:
       handle_overload()  # backpressure / degrade / drop
       continue
   
   if batch is None:
       continue  # not yet a flush boundary
   
   t0 = time.monotonic()
   try:
       sink_write(batch)
   finally:
       tuner.record_write_latency((time.monotonic() - t0) * 1000.0)

   The finally matters: a failed write is still a latency observation, and skipping it on the error path is how the throttle goes blind exactly when the sink is sick.

3. Adapt batch size (under the hood, evaluated at each flush boundary):

   • High load (F_ewma > 0.70 and EWMA(L) ≤ L_target): B_{t+1} = min(B_max, ⌊B_t × 1.5⌋); reduce T_flush.
   • Low load (F_ewma < 0.10): B_{t+1} = max(B_min, ⌊B_t / 1.2⌋); extend T_flush.
   • Deadband (0.10 ≤ F_ewma ≤ 0.70): no tuning — EWMA smoothing plus the deadband prevents oscillation around the boundary.

4. Apply latency feedback — this is what closes the loop:

   • If EWMA(L) > L_target: B_{t+1} = max(B_min, ⌊B_t × 0.8⌋), and the high-load expansion branch is barred until latency returns under target. The bar is not decorative: expansion multiplies by 1.5 while the throttle multiplies by 0.8, and 1.5 × 0.8 = 1.2 > 1, so without it one throttle per flush can never undo one expansion and B ratchets to B_max with sink latency pinned above target.
   • The throttle fires even inside the deadband, which is its purpose.

5. Flush triggers:

   • Threshold: #items ≥ current_batch_size (evaluated inside add_item).

   • Timeout: elapsed ≥ current_flush_timeout_sec. The engine owns no timer thread, so this is only evaluated when you call in. If the producer can stall — thin instrument, feed outage, the lull after the close — buffered records would otherwise sit in memory indefinitely and die with the process. Drive tuner.flush_if_due() from a scheduler at an interval at or below min_flush_timeout_sec:

     python

     copy

     batch = tuner.flush_if_due()   # None if not yet due / nothing buffered
     if batch:
         sink_write_with_metric(batch)

   • Forced flush: tuner.flush_now() for checkpoint boundaries. Returns up to current_batch_size items and deliberately does not tune — a batch that is partial because you asked for it says nothing about producer speed.

6. Shutdown:

   python

   copy

   leftover = tuner.close()          # drains the WHOLE buffer, not one batch
   for chunk in chunked(leftover, sink_max_rows):
       sink_write(chunk)

   close() is not capped at current_batch_size, so the returned list may be larger than your sink accepts in one call — chunk it. After close(), add_item raises RuntimeError; call reset() to reuse the engine.

Full procedure with rationale: references/workflows.md. Concrete numeric thresholds: references/standards.md. Operational checklist: assets/checklist.md.

Common Pitfalls

• Tuning on queue depth instead of batch fullness → the controller inverts and drives the batch down under load. See "Why fullness, and not queue depth" above; this is the single most important design point in the skill.
• Assuming close() returns one batch → it drains the entire buffer, which may exceed what your sink accepts in a single call. Chunk the result. (The converse bug — capping the shutdown drain at current_batch_size — silently strands records, which for an order log is data loss.)
• Never calling flush_if_due() → the flush timeout is dead weight, and a stalled producer leaves records buffered until the process exits.
• Skipping record_write_latency() on the error path → the throttle goes blind exactly when the sink is failing.
• Unbounded max batch size → queued memory can exceed sink RAM. Always set max_queue_size; the engine raises QueueFullError at the cap.
• Feeding a non-finite latency → rejected with ValueError. A NaN would otherwise poison the EWMA permanently (NaN > target is always False, silently disabling the throttle for the life of the process) and serialise as invalid JSON in the metrics export.
• Rapid oscillations: if fullness flickers around one threshold and the engine thrashes, lower fill_ewma_alpha (lower alpha = more smoothing) — but never to 0.
• Latency oscillation under repeated record_write_latency() spikes: if smoothed latency sits at L_target - epsilon and individual writes push it slightly above, the engine repeatedly throttles and climbs. Set target_write_latency_ms from the typical sink latency, not the ideal.
• Blocking inside on_flush: the callback runs outside the engine lock, so it will not deadlock or block other producers — but it does run on the calling producer's thread, so a slow sink write there still stalls that producer. Prefer the returned-batch pattern for the actual write.
• ClickHouse async_insert wait_for_async_insert = 0 (fire-and-forget): record_write_latency() will see near-0 ms for "successful" flushes, masking real DB-side problems. Use wait_for_async_insert = 1.

Verification

Run the unit tests:

python -m unittest discover -s skills/adaptive-batch-size-tuning-under-load/scripts -v

43 tests. What they assert:

• Control-law direction (the regression that matters): saturating load expands the batch toward B_max and shortens T_flush; quiet, timeout-driven load shrinks it and lengthens T_flush.
• Closed loop: with a batch-size-dependent sink latency, the equilibrium batch size settles strictly inside (B_min, B_max) with smoothed latency at or under target.
• Deadband: batches cut ~50% full produce zero tuning transitions; exact boundary values (0.10, 0.70) are inside the deadband.
• Latency throttle: fires above target, not at exactly target, stops at B_min, and the EWMA is seeded with its first observation (hand-computed expected values, not a re-derivation of the implementation).
• Shutdown: close() drains records the batch size would have stranded, preserves order, is idempotent, and add_item after close() raises.
• Flush triggers: flush_if_due() releases an idle buffer; flush_now() returns a partial batch and does not tune.
• Bounded queue: add_item past max_queue_size raises QueueFullError and does not buffer the rejected item.
• Validation: bound ordering, alpha range, non-finite/negative latency, queue_capacity > max_queue_size, and mis-signed tuning multipliers.
• Concurrency: an on_flush callback may re-enter the engine without deadlocking; a raising callback does not lose the batch; 8 concurrent producers × 500 items lose and duplicate nothing.
• Status is strict-JSON serializable (allow_nan=False, Prometheus-ready).

Confirm with the operational checklist in assets/checklist.md before deploying.

Related Skills

• kafka-based-tick-distribution-at-scale — the parent batch architecture for Kafka paths; this skill is the client-side tuning companion.
• producer-consumer-tick-pipeline — the broader produce/consume pipeline; this skill is the leaf that decides when to flush to the sink.
• tick-buffering-burst-handling — what to do when the queue genuinely overflows; QueueFullError from this skill should be the trigger.
• backpressure-drop-degrade-policy — design the policy that decides what to do when QueueFullError fires.
• graceful-shutdown-draining-in-flight-ticks — the shutdown counterpart; close() is this engine's contribution to that drain.
• kill-switch-and-drawdown-circuit-breakers — strategy-level circuit breaker upstream of this engine; pair them so strategy stops suppress writes.
• latency-monitoring-percentile-based-slas — for monitoring the sink's P99/P999 against L_target.
• model-inference-latency-budget-for-live-trading — analogous pattern for model inference instead of DB writes.