test(downloader): cover probe judging and the submission window

Assert that a probe is judged on the trailing tick, that a backoff is temporary and climbs back to the cap, and that a worker always finds its next segment staged. Documents adaptive-workers behaviour in the downloading guide.

docs/guide/downloading.md

@@ -734,8 +734,8 @@ Two related flags:
 | Flag | Purpose |
 | --- | --- |
 | `--workers N` | Threads used per track for segment downloads. Default depends on the downloader. |
-| `--adaptive-workers` | Opt-in: start with a moderate per-track worker count and ramp it up or back off based on measured CDN throughput and errors, capped at `--workers`. Off by default (fixed worker count). |
-| `--download-processes N` | Opt-in: split a large segment batch across `N` download processes, each with its own worker pool. A single process tops out around 1.3 Gb/s (Python interpreter limit); `2` reaches 2.5GbE line rate on fast CDNs. Only engages for batches of 24+ segments. Ignored while `--speed-limit` (or serve's `global_speed_limit`) is set: the cap is one shared budget, which extra processes cannot share, so the download stays in a single process. Default `1`. |
+| `--adaptive-workers` | Opt-in: start the per-track worker count at `--workers`, reduce it when the CDN returns errors, then raise it again while the extra workers add measured throughput. The count never goes above `--workers`. Off by default (fixed worker count). |
+| `--download-processes N` | Opt-in: split a large segment batch across `N` download processes, each with its own worker pool. A single process tops out around 1.3 Gb/s (Python interpreter limit); `2` reaches 2.5GbE line rate on fast CDNs. Only engages for batches of 24+ segments. Ignored while `--speed-limit` (or serve's `global_speed_limit`) is set: the cap is one shared budget, which extra processes cannot share, so the download stays in a single process. Also ignored when the service HTTP session carries state a child process cannot rebuild, such as a custom TLS adapter. The download stays in a single process instead of running with the wrong settings. Default `1`. |
 | `--continue-downloads` | Opt-in: keep completed segment files when a download fails so the next run resumes instead of restarting from zero. One-off enable of the [`continue_downloads`](../reference/configuration/download.md#continue_downloads) config option, which documents what can and cannot resume. |
 | `--downloads N` | Number of tracks downloaded concurrently. Default `1`. |
 | `--slow [MIN-MAX]` | Add a delay between titles to look more like a real device. `--slow` alone means 60-120s; `--slow 20-40` sets a custom range. Minimum 20s. |

tests/core/test_adaptive_workers.py

@@ -18,6 +18,7 @@ from unshackle.core.downloaders.requests import (
     plan_tail_parts,
     requests,
     split_ranges,
+    submission_window,
     tail_boost_engages,
 )
 from unshackle.core.tracks.track import DownloadContext
@@ -176,6 +177,75 @@ def test_tail_guard_still_halves_on_error_burst() -> None:
     assert c.update(now, 2) == 3  # guard active (2 < 6) but 3 errors -> 6 // 2
 
 
+PER_WORKER = 1_000_000.0  # bytes/sec each worker adds, so throughput scales with the target
+
+
+def run_scaling(c: AdaptiveWorkerController, now: float, seconds: float, work: int = 10_000) -> float:
+    """Advance ``seconds`` of a download whose throughput scales with the live target.
+
+    Records samples on the 0.5s grid at ``target * PER_WORKER`` bytes/sec and calls ``update``
+    on every step, so the speed window mixes old and new rates the way a real run does. ``work``
+    stands for the segments in flight or not yet started, kept high to hold the tail guard off.
+    """
+    end = now + seconds
+    while now < end - 1e-9:
+        now += DT
+        c.record_bytes(int(c.target * PER_WORKER * DT), now)
+        c.update(now, work)
+    return now
+
+
+def test_probe_kept_when_the_true_gain_clears_the_bar() -> None:
+    c = AdaptiveWorkerController(cap=16, start=10, 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) == 12  # first evaluation probes 10 -> 12
+    target, now = run_tick(c, rate * 1.2, now)
+    assert target == 14  # probe held, and the controller climbs again
+
+
+def test_recovers_to_the_cap_after_an_error_burst() -> None:
+    c = AdaptiveWorkerController(cap=16, start=16, tick=TICK, window=WINDOW)
+    now = 0.0
+    c.update(now)
+    now = run_scaling(c, now, WINDOW)
+    for _ in range(3):
+        c.record_error(now)
+    now = run_scaling(c, now, TICK)
+    assert c.target == 8, "error burst must halve the target 16 -> 8"
+
+    now = run_scaling(c, now, 60.0)
+    assert c.target == 16, "controller never returned to its cap after backing off"
+    now = run_scaling(c, now, 40.0)
+    assert c.target == 16, "controller left the cap on a link where every worker still pays"
+
+
+def test_reverts_when_the_probe_stops_paying_on_a_saturated_link() -> None:
+    c = AdaptiveWorkerController(cap=16, start=8, tick=TICK, window=WINDOW)
+    now = 0.0
+    c.update(now)
+    rate = 8 * PER_WORKER
+    now = fill_window(c, rate, now)
+    seen = [c.update(now, 10_000)]
+    for _ in range(8):
+        rate = min(c.target, 10) * PER_WORKER
+        end = now + TICK
+        while now < end - 1e-9:
+            now += DT
+            c.record_bytes(int(rate * DT), now)
+        seen.append(c.update(now, 10_000))
+    assert max(seen) <= 12  # one probe past the knee is the most it may hold
+    assert seen[-1] <= 12
+
+
+def test_submission_window_stages_a_spare_segment_for_every_worker() -> None:
+    assert submission_window(16) == 32
+    for target in (2, 4, 6, 16):
+        assert submission_window(target) == 2 * target
+
+
 def test_plan_tail_parts_full_coverage_no_gaps() -> None:
     size = 20 * 1024 * 1024
     parts = plan_tail_parts(size, spare=8)

tests/downloaders/test_submission_window.py

@@ -0,0 +1,105 @@
+"""Submission-window invariants for the segmented drain loop.
+
+Segments are not all submitted up front: that convoys on the pool's global lock and starves
+the event drain. The loop meters submission through a window it tops up. The window must hold
+two segments for every worker, so a worker that finishes one finds the next already staged
+instead of idling until the next drain cycle. Fixed and adaptive mode must agree on that
+ratio; adaptive sizes it from its live target instead of from ``max_workers``.
+
+The test stalls every response, so no future can complete and the queue settles at exactly
+the window. Counting ``pool.submit`` calls makes the queue depth observable: a blocked
+server only ever sees ``max_workers`` requests, whatever is staged behind them.
+"""
+
+import importlib
+import threading
+import time
+from concurrent.futures.thread import ThreadPoolExecutor
+from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
+
+import pytest
+
+dl = importlib.import_module("unshackle.core.downloaders.requests")
+
+BODY = b"segment-body" * 64
+MAX_WORKERS = 4
+SEG_COUNT = 20  # more than the window, so queue depth is the only bound
+
+pytestmark = pytest.mark.unit
+
+
+class _Handler(BaseHTTPRequestHandler):
+    def do_GET(self):
+        self.server.release.wait(timeout=10)
+        self.send_response(200)
+        self.send_header("Content-Length", str(len(BODY)))
+        self.end_headers()
+        self.wfile.write(BODY)
+
+    def log_message(self, *args):
+        pass
+
+
+@pytest.fixture
+def server():
+    srv = ThreadingHTTPServer(("127.0.0.1", 0), _Handler)
+    srv.release = threading.Event()
+    thread = threading.Thread(target=srv.serve_forever, daemon=True)
+    thread.start()
+    yield srv
+    srv.release.set()
+    srv.shutdown()
+    thread.join(timeout=5)
+
+
+@pytest.fixture(autouse=True)
+def clean_cancel():
+    dl.DOWNLOAD_CANCELLED.clear()
+    yield
+    dl.DOWNLOAD_CANCELLED.clear()
+
+
+@pytest.fixture
+def submissions(monkeypatch):
+    counter: list[int] = []
+
+    class _CountingPool(ThreadPoolExecutor):
+        def submit(self, fn, /, *args, **kwargs):
+            counter.append(1)  # list.append is atomic, so no lock is needed here
+            return super().submit(fn, *args, **kwargs)
+
+    monkeypatch.setattr(dl, "ThreadPoolExecutor", _CountingPool)
+    return counter
+
+
+@pytest.mark.parametrize("adaptive", [False, True])
+def test_queue_holds_two_segments_per_worker(server, submissions, tmp_path, adaptive):
+    urls = [{"url": f"http://127.0.0.1:{server.server_address[1]}/seg{i}.bin"} for i in range(SEG_COUNT)]
+    events: list[dict] = []
+
+    def run():
+        events.extend(
+            dl.requests(
+                urls,
+                output_dir=tmp_path,
+                filename="seg_{i:04}.bin",
+                max_workers=MAX_WORKERS,
+                adaptive=adaptive,
+            )
+        )
+
+    worker = threading.Thread(target=run, daemon=True)
+    worker.start()
+    try:
+        deadline = time.time() + 3.0
+        while time.time() < deadline and len(submissions) < 2 * MAX_WORKERS:
+            time.sleep(0.02)
+        time.sleep(0.2)  # settle, so an over-filled window is caught as well as an under-filled one
+        assert len(submissions) == 2 * MAX_WORKERS
+    finally:
+        server.release.set()
+        worker.join(timeout=30)
+
+    assert not worker.is_alive()
+    assert sum(event.get("advance", 0) for event in events) == SEG_COUNT
+    assert len(sorted(tmp_path.glob("seg_*.bin"))) == SEG_COUNT