perf: adaptive target/upload side (concurrency, backpressure, throttle response)

[allamiro]

Why the target side needs this too

#100 and #101 cover adapting to what the SOURCES are doing. The upload side is currently the opposite of adaptive: batching.max_concurrent_batches is a fixed number (default 8, added in #94), applied identically whether the object store is answering in 5ms or 5 seconds.

That is fine while the engine is source-starved — which it is today, at 87-92% of the read cycle spent waiting (#99). It stops being fine the moment the read path is fixed (#96 A1/A2), because then the object store becomes the next thing that can saturate, and a fixed concurrency is wrong in both directions:

• too low when the backend is fast and the engine is behind — artificial ceiling
• too high when the backend is degraded — the engine piles on concurrent uploads against a struggling store, turning slow into failing

What to adapt on

Measured per-stage means (from the engine's own metrics, ~432 batches):

object_upload    7.75 ms
manifest_upload  5.20 ms
verify           2.55 ms

Those are already exported as vtop_stage_duration_seconds{stage=...}, so the control signal exists without new instrumentation. Candidates:

1. Latency-driven concurrency. Raise in-flight uploads while p95 upload latency is flat; back off when it climbs. Classic additive-increase / multiplicative-decrease.
2. Error/throttle-driven backoff. S3 answers overload with 503 SlowDown and 429. Treating those as a signal to reduce concurrency is strictly better than retrying at the same rate. Needs checking whether the current backend surfaces them distinguishably from other errors.
3. Backpressure into the read path. If uploads cannot keep up, reading faster only grows inflight_batches and memory. The read and target sides should share one notion of "we are behind" rather than each optimising alone. Related: test: 1M records/sec for 5 minutes and 5M records / sec for 1 minute soak — sustained backpressure behaviour #98 hypothesis 1 (unbounded inflight growth under sustained load).
4. Adaptive batch sizing. Research: what it would actually take to reach 1M 5M and even 20 M records/sec #92 notes fixed per-batch cost amortises over larger batches; if upload latency is dominated by per-request overhead, larger objects beat more concurrency. This overlaps perf: batches seal on the 60s age timer rather than filling #89 (seal policy) and perf: establish a benchmark matrix (record size, batch size, format, compression) #90 (benchmark matrix).

Sequencing — deliberately after the read fix

Do not start this until #96 A1/A2 lands and the 1M/s benchmark is re-run. Right now the target side is idle (0.33-1.6% CPU) and any adaptive controller would be tuning against a signal that is pure noise, because nothing is loading it.

This is exactly the mistake already made three times in this repo (#94, #97, and the reverted adaptive-poll attempt in #99): optimising a stage before establishing it was the constraint. The read path is the known constraint today. Fix it, re-measure, and only then find out whether the object store is the next one.

Explicit warning from the reverted attempt

The adaptive poll-window heuristic reverted in #99 failed because the adaptation made the thing it was adapting less able to succeed, with no recovery path — window shrank to 5ms, fetch could not complete, source read empty, window shrank further. 48x regression.

Any adaptive controller here must have:

• a floor that still permits forward progress
• a forced probe that lets it recover from an over-correction
• a test that pins recovery, not just the backoff

Related

• perf: source reads are serial for ALL adapters (kafka, file, syslog) — shared read-path design #96 read-path fix (blocks this)
• test: 1M records/sec for 5 minutes and 5M records / sec for 1 minute soak — sustained backpressure behaviour #98 soak test (would reveal whether inflight grows unbounded)
• perf: batches seal on the 60s age timer rather than filling #89 seal policy, perf: establish a benchmark matrix (record size, batch size, format, compression) #90 benchmark matrix, Research: what it would actually take to reach 1M 5M and even 20 M records/sec #92 throughput analysis

[allamiro]

Status audit — 2026-07-25

PARTIAL PREREQUISITES ONLY — keep open.

Merged foundation:

• bounded upload/verify concurrency exists through batching.max_concurrent_batches (default 8);
• stage latency and in-flight metrics exist;
• the read-path prerequisite perf: source reads are serial for ALL adapters (kafka, file, syslog) — shared read-path design #96 is closed.

Not implemented:

• no latency-driven/AIMD concurrency controller;
• no 429/503 SlowDown classification feeding concurrency reduction;
• no shared target/read backpressure controller;
• no adaptive batch-size controller;
• no floor + forced-probe recovery test.

The current code still reads a fixed max_concurrent_batches and passes it to buffer_unordered.

Waiting on performance evidence first: rerun after #96 and demonstrate that the target is now the constraint. If it is, implement the controller and pin both backoff and recovery behavior. There is no adaptive-target performance result yet.