perf(downloader): slow-start ramp for the adaptive worker controller

Warm up after half a tick and double the worker target while probes pay off, dropping to AIMD +2 on first plateau/error burst. 33% faster in shaped benchmarks; backoff semantics unchanged.

tests/core/test_adaptive_workers.py

@@ -1,6 +1,7 @@
-"""Unit tests for the opt-in adaptive worker controller (AIMD hill-climb over segment
-concurrency) and the plumbing that carries the flag into the downloader. Pure controller
-tests only. No threads, sockets, or real downloads; the clock is supplied explicitly."""
+"""Unit tests for the opt-in adaptive worker controller (slow-start then AIMD hill-climb
+over segment concurrency) and the plumbing that carries the flag into the downloader. Pure
+controller tests only. No threads, sockets, or real downloads; the clock is supplied
+explicitly."""
 
 from __future__ import annotations
 
@@ -54,15 +55,15 @@ def test_starts_at_min_six_cap() -> None:
     assert AdaptiveWorkerController(cap=1).target == 2  # clamped up to ADAPTIVE_MIN
 
 
-def test_no_op_before_full_window() -> None:
+def test_no_op_before_warmup() -> None:
     c = AdaptiveWorkerController(cap=16, tick=TICK, window=WINDOW)
     assert c.update(0.0) == 6  # arms last_tick before any samples are recorded
-    c.record_bytes(500_000, 0.0)
-    c.record_bytes(500_000, 5.0)
-    # a full tick has elapsed, but only 5s of window exists -> hold
-    assert c.update(5.0) == 6
-    c.record_bytes(500_000, 8.0)
-    assert c.update(9.0) == 6
+    c.record_bytes(500_000, 3.0)
+    # a full tick has elapsed since arming, but under half a tick of samples exists -> hold
+    assert c.update(4.5) == 6
+    c.record_bytes(500_000, 6.5)
+    # warmed up (first sample 3.0, now 8.5) -> slow-start probe fires
+    assert c.update(8.5) == 12
 
 
 def test_ramps_up_while_speed_improves() -> None:
@@ -71,7 +72,7 @@ def test_ramps_up_while_speed_improves() -> None:
     c.update(now)  # arm
     rate = 1_000_000.0
     now = _fill_window(c, rate, now)
-    targets = [c.update(now)]  # first evaluation: 6 -> 8
+    targets = [c.update(now)]  # first evaluation: slow-start doubles 6 -> 12
     for _ in range(6):
         rate *= 2  # sustained >10% windowed gain each tick
         target, now = _run_tick(c, rate, now)
@@ -81,16 +82,19 @@ def test_ramps_up_while_speed_improves() -> None:
     assert max(targets) <= 16
 
 
-def test_reverts_on_plateau() -> None:
+def test_reverts_on_plateau_and_ends_slow_start() -> None:
     c = AdaptiveWorkerController(cap=16, tick=TICK, window=WINDOW)
     now = 0.0
     c.update(now)
     rate = 1_000_000.0
     now = _fill_window(c, rate, now)
-    assert c.update(now) == 8  # first increase 6 -> 8
-    # hold the rate flat: <10% gain -> the increase is reverted and held
+    assert c.update(now) == 12  # first increase slow-starts 6 -> 12
+    # hold the rate flat: <10% gain -> the tick reverts the increase to its restore point
     target, now = _run_tick(c, rate, now)
     assert target == 6
+    # slow-start is over: the next successful probe is additive (+2), not a double
+    target, now = _run_tick(c, rate * 4, now)
+    assert target == 8
 
 
 def test_halves_and_cooldown_on_error_burst() -> None:
@@ -152,7 +156,7 @@ def test_tail_guard_none_path_unchanged() -> None:
     now = 0.0
     c.update(now)
     now = _fill_window(c, 1_000_000.0, now)
-    assert c.update(now) == 8  # unchanged: first evaluation 6 -> 8
+    assert c.update(now) == 12  # first evaluation slow-starts 6 -> 12
 
 
 def test_tail_guard_still_halves_on_error_burst() -> None:

unshackle/core/downloaders/requests.py

@@ -54,10 +54,10 @@ def _adaptive_chunk_size(content_length: int) -> int:
     return min(MAX_CHUNK, max(MIN_CHUNK, content_length // 4))
 
 
-# Adaptive worker controller (opt-in): AIMD hill-climb over segment concurrency
+# Adaptive worker controller (opt-in): slow-start then AIMD hill-climb over segment concurrency
 ADAPTIVE_TICK = 4.0  # seconds between target re-evaluations
 ADAPTIVE_START = 6  # initial worker target, clamped to cap
-ADAPTIVE_STEP = 2  # additive increase per tick
+ADAPTIVE_STEP = 2  # additive increase per tick once slow-start ends
 ADAPTIVE_MIN = 2  # never drop below this many workers
 ADAPTIVE_ERROR_BURST = 3  # errors within a tick that trigger multiplicative decrease
 ADAPTIVE_PLATEAU_GAIN = 1.10  # min speed ratio to justify keeping an increase
@@ -66,12 +66,12 @@ ADAPTIVE_PLATEAU_GAIN = 1.10  # min speed ratio to justify keeping an increase
 class AdaptiveWorkerController:
     """CDN-aware segment concurrency governor.
 
-    Pure and synchronous (no threads or sockets), so the AIMD policy is unit-testable.
-    ``update`` is evaluated once per ``tick`` and only after a full ``window`` of
-    throughput samples exists: it ramps the worker target up while the CDN keeps
-    rewarding extra concurrency, reverts the last increase on a plateau, and halves
-    (with a one-tick cooldown) on an error burst. Target is always clamped to
-    ``[ADAPTIVE_MIN, cap]``.
+    Pure and synchronous (no threads or sockets), so the policy is unit-testable.
+    ``update`` is evaluated once per ``tick``, starting once half a tick of throughput
+    samples exists. It slow-starts (doubles the target) while every probe keeps paying
+    off, then drops to AIMD (+``ADAPTIVE_STEP``) after the first plateau or error burst:
+    a plateau reverts the last increase, an error burst halves the target with a
+    one-tick cooldown. Target is always clamped to ``[ADAPTIVE_MIN, cap]``.
     """
 
     def __init__(
@@ -88,7 +88,9 @@ class AdaptiveWorkerController:
         self._last_tick: Optional[float] = None
         self._last_action: Optional[str] = None  # "increase" once we have probed upward
         self._speed_before_increase = 0.0
+        self._target_before_increase = self.target  # restore point for a plateau revert
         self._cooldown = False
+        self._ramping = True  # slow-start: double per probe until the first plateau or error burst
 
     def _prune(self, now: float) -> None:
         cutoff = now - self.window
@@ -114,11 +116,11 @@ class AdaptiveWorkerController:
             return 0.0
         return sum(n for _, n in self._samples) / span
 
-    def _full_window(self, now: float) -> bool:
-        # true once we have been observing for at least a full window; measured from the
-        # first-ever sample so continuous pruning (which keeps span just under window) does
-        # not perpetually starve the controller of a "full" window in real timing
-        return self._first_sample is not None and (now - self._first_sample) >= self.window
+    def _warmed_up(self, now: float) -> bool:
+        # half a tick of samples gives enough of a baseline to probe against; waiting for
+        # a full speed window would idle the ramp at the start of every download.
+        # measured from the first-ever sample so pruning cannot starve the check
+        return self._first_sample is not None and (now - self._first_sample) >= self.tick / 2
 
     def update(self, now: float, inflight_plus_remaining: Optional[int] = None) -> int:
         """Evaluate the policy at most once per tick; return the current target.
@@ -136,8 +138,8 @@ class AdaptiveWorkerController:
             return self.target
         self._last_tick = now
 
-        # hold until a full window of throughput history has accumulated
-        if not self._full_window(now):
+        # hold until half a tick of throughput history has accumulated
+        if not self._warmed_up(now):
             self._errors = 0
             return self.target
 
@@ -157,19 +159,24 @@ class AdaptiveWorkerController:
             self.target = max(ADAPTIVE_MIN, self.target // 2)
             self._cooldown = True
             self._last_action = None
+            self._ramping = False
             reason = "error_burst"
         elif inflight_plus_remaining is not None and inflight_plus_remaining < self.target:
             # tail guard: too little work to saturate the target, so a low measured speed
             # here reflects starvation rather than a plateau, so hold and skip the probe/revert this tick
             pass
         elif self._last_action == "increase" and speed_now < ADAPTIVE_PLATEAU_GAIN * self._speed_before_increase:
-            # last probe upward did not pay off, so revert it and hold
-            self.target = max(ADAPTIVE_MIN, self.target - ADAPTIVE_STEP)
+            # last probe upward did not pay off: revert it and hold. step size varies
+            # during slow-start, hence the recorded restore point rather than -STEP
+            self.target = max(ADAPTIVE_MIN, self._target_before_increase)
             self._last_action = None
+            self._ramping = False
             reason = "plateau"
         elif self.target < self.cap:
             self._speed_before_increase = speed_now
-            self.target = min(self.cap, self.target + ADAPTIVE_STEP)
+            self._target_before_increase = self.target
+            step = self.target if self._ramping else ADAPTIVE_STEP
+            self.target = min(self.cap, self.target + step)
             self._last_action = "increase"
             reason = "increase"
         else: