fix(downloader): honor Retry-After with cancel-safe backoff, add fault-injection bench and tests

scripts/bench_downloader.py

@@ -43,6 +43,7 @@ from shutil import rmtree
 from typing import Any, Optional
 
 import click
+import fault_server  # local sibling (scripts/): fault-injection server + payload verifier
 from rich.console import Console
 from rich.table import Table
 
@@ -393,6 +394,174 @@ def print_report(rows: list[Row], runs: int, total_segments: int) -> None:
         )
 
 
+def fault_integrity(out_dir: Path, segments: int, seg_size: int) -> tuple[list[int], list[int]]:
+    """Verify every output file byte-for-byte against its seeded expected payload.
+
+    Returns (missing, corrupt). ``corrupt`` (present but not byte-identical) is the loud
+    CORRUPTION signal: a fault path that finalizes wrong bytes must fail the run, not pass.
+    """
+    missing: list[int] = []
+    corrupt: list[int] = []
+    for i in range(segments):
+        p = out_dir / f"seg_{i}.bin"
+        if not p.exists():
+            missing.append(i)
+            continue
+        if p.read_bytes() != fault_server.segment_payload(f"/seg/{i}", seg_size):
+            corrupt.append(i)
+    return missing, corrupt
+
+
+# per worker-count under a fault profile: workers, median_s, MB/s, missing, corrupt, raised,
+# requests seen, amplification (requests/segments), retries (requests-segments), status_counts,
+# then worst-across-runs sweep safety: strays, locked handles, live pool threads
+FaultRow = tuple[int, float, float, list[int], list[int], bool, int, float, int, dict[int, int], int, int, int]
+
+
+def print_fault_report(rows: list[FaultRow], runs: int, segments: int, profile: str) -> None:
+    plural = "s" if runs != 1 else ""
+    table = Table(title=f"fault profile '{profile}': {segments} segments (median of {runs} run{plural})")
+    table.add_column("workers", justify="right")
+    table.add_column("seconds", justify="right")
+    table.add_column("MB/s", justify="right")
+    table.add_column("status", justify="left")
+    table.add_column("requests", justify="right")
+    table.add_column("amp", justify="right")
+    table.add_column("sweep", justify="left")
+    for w, secs, mbps, missing, corrupt, raised, requests_seen, amp, _retries, _codes, strays, locked, threads in rows:
+        if corrupt:
+            status = f"[red]CORRUPTION {len(corrupt)} segs[/]"
+        elif missing:
+            status = f"[red]FAIL {len(missing)} incomplete[/]"
+        elif raised:
+            status = "[yellow]raised (all present)[/]"
+        else:
+            status = "[green]ok[/]"
+        if locked:
+            sweep_status = f"[red]{locked} LOCKED (WinError 32)[/]"
+        elif threads:
+            sweep_status = f"[yellow]{threads} pool threads live[/]"
+        elif strays:
+            sweep_status = f"{strays} strays"
+        else:
+            sweep_status = "[green]clean[/]"
+        table.add_row(str(w), f"{secs:.3f}", f"{mbps:.1f}", status, str(requests_seen), f"{amp:.2f}x", sweep_status)
+    console.print(table)
+    for w, _secs, _mbps, _missing, _corrupt, _raised, requests_seen, amp, retries, codes, *_rest in rows:
+        code_str = ", ".join(f"{c}:{n}" for c, n in codes.items()) or "none"
+        console.print(
+            f"amplification  workers={w:<3} {requests_seen} requests for {segments} segments "
+            f"({amp:.2f}x, {retries} excess requests (retries/hedges/probes/parts)), statuses [{code_str}]"
+        )
+
+
+def _procs_notice(procs: int, n_urls: int, session: Optional[Any]) -> None:
+    """Warn loudly when --procs cannot engage, so a run is never mistaken for a multiprocess test."""
+    if procs <= 1:
+        return
+    if n_urls < dl_mod.MP_MIN_SEGMENTS:
+        console.print(
+            f"[yellow]--procs {procs} will NOT engage: {n_urls} segments < MP_MIN_SEGMENTS "
+            f"({dl_mod.MP_MIN_SEGMENTS}); the single-process path runs instead[/]"
+        )
+    elif dl_mod._build_session_spec(session) is None:
+        console.print(
+            f"[yellow]--procs {procs} will NOT engage: session not rebuildable in child processes; "
+            f"the single-process path runs instead[/]"
+        )
+
+
+def run_fault_mode(
+    profile_name: str,
+    segments: int,
+    seg_size: int,
+    worker_counts: list[int],
+    runs: int,
+    session: Optional[Any],
+    adaptive: bool,
+    procs: int,
+    seed: int,
+) -> None:
+    """Start a FaultServer in-process, sweep worker counts, and report reliability + overhead.
+
+    Reuses run_batch/tail_droop/sweep_strays (via run_batch) and the report conventions, adding
+    request amplification and byte-exact integrity, the metrics that catch retry/hedge storms
+    and silent corruption against a misbehaving host.
+    """
+    profile = fault_server.resolve_profile(profile_name)
+    server = fault_server.FaultServer(profile, seg_size, segments, seed)
+    threading.Thread(target=server.serve_forever, daemon=True).start()
+    host, port = str(server.server_address[0]), int(server.server_address[1])
+    urls = [f"http://{host}:{port}/seg/{i}" for i in range(segments)]
+    console.print(
+        f"[cyan]Fault mode:[/] profile={profile_name} seed={seed} "
+        f"{segments} x {seg_size / MIB:.2f} MiB, impersonate={getattr(session, 'impersonate_name', None) or 'none'}"
+    )
+
+    rows: list[FaultRow] = []
+    try:
+        with tempfile.TemporaryDirectory(prefix="faultbench_") as root_str:
+            root = Path(root_str)
+            for w in worker_counts:
+                samples: list[tuple[float, float, list[int], list[int], bool, int, dict[int, int], int, int, int]] = []
+                for _ in range(runs):
+                    server.reset_counters()
+                    out_dir = Path(tempfile.mkdtemp(prefix="run_", dir=root))
+                    secs, _comps, err, strays, locked, pool_threads = run_batch(
+                        urls, out_dir, w, session, adaptive, procs
+                    )
+                    missing, corrupt = fault_integrity(out_dir, segments, seg_size)
+                    stats = server.stats()
+                    samples.append(
+                        (
+                            secs,
+                            downloaded_mib(out_dir),
+                            missing,
+                            corrupt,
+                            err is not None,
+                            stats["requests"],
+                            stats["status_counts"],
+                            strays,
+                            locked,
+                            pool_threads,
+                        )
+                    )
+                    rmtree(out_dir, ignore_errors=True)
+                median_s = statistics.median(s[0] for s in samples)
+                best = min(samples, key=lambda s: abs(s[0] - median_s))  # representative run near the median
+                mbps = best[1] / best[0] if best[0] else 0.0
+                requests_seen = best[5]
+                amp = requests_seen / segments if segments else 0.0
+                retries = max(0, requests_seen - segments)
+                rows.append(
+                    (
+                        w,
+                        median_s,
+                        mbps,
+                        best[2],
+                        best[3],
+                        best[4],
+                        requests_seen,
+                        amp,
+                        retries,
+                        best[6],
+                        max(s[7] for s in samples),
+                        max(s[8] for s in samples),
+                        max(s[9] for s in samples),
+                    )
+                )
+    finally:
+        server.stall_event.set()  # release any stalled handler threads before shutdown
+        server.shutdown()
+
+    print_fault_report(rows, runs, segments, profile_name)
+
+    broken = {row[0]: (row[3], row[4]) for row in rows if row[3] or row[4]}
+    if broken:
+        detail = "; ".join(f"workers={w}: missing {miss} corrupt {corr}" for w, (miss, corr) in broken.items())
+        raise click.ClickException(f"incomplete/corrupt downloads under fault profile: {detail}")
+
+
 @click.command(context_settings={"help_option_names": ["-h", "--help"]})
 @click.option("--segments", default=64, show_default=True, help="Synthetic segment count.")
 @click.option("--seg-size", default=2 * MIB, show_default=True, help="Bytes per synthetic segment.")
@@ -434,6 +603,12 @@ def print_report(rows: list[Row], runs: int, total_segments: int) -> None:
 @click.option(
     "--no-read1", is_flag=True, help="A/B: force racers back to blocking full-chunk reads (pre-fix behavior)."
 )
+@click.option(
+    "--fault-profile",
+    type=click.Choice(sorted(fault_server.PROFILES)),
+    help="Reliability mode: serve via the fault-injection server (fault_server.py) and verify integrity + amplification.",
+)
+@click.option("--seed", default=fault_server.DEFAULT_SEED, show_default=True, help="Fault RNG seed (--fault-profile).")
 def main(
     segments: int,
     seg_size: int,
@@ -452,6 +627,8 @@ def main(
     adaptive: bool,
     procs: int,
     no_read1: bool,
+    fault_profile: Optional[str],
+    seed: int,
 ) -> None:
     """Benchmark downloader throughput vs worker count, with optional fault injection."""
     worker_counts = [int(w) for w in workers.split(",") if w.strip()]
@@ -481,7 +658,7 @@ def main(
             saved["RACER_READ1"] = dl_mod.RACER_READ1
             dl_mod.RACER_READ1 = False
         else:
-            console.print("[yellow]--no-read1 ignored: this checkout has no RACER_READ1 (pre-fix behavior anyway)[/]")
+            console.print("[yellow]--no-read1 ignored: only applies to older checkouts that still have RACER_READ1[/]")
 
     if impersonate:
         from unshackle.core.session import session as make_session
@@ -489,6 +666,14 @@ def main(
         session = make_session(impersonate)
 
     try:
+        if fault_profile:
+            if urls_file:
+                console.print("[yellow]--urls-file ignored in --fault-profile mode[/]")
+            if any(faults.values()):
+                console.print("[yellow]--fault-* flags ignored in --fault-profile mode (use the profile knobs)[/]")
+            _procs_notice(procs, segments, session)
+            run_fault_mode(fault_profile, segments, seg_size, worker_counts, runs, session, adaptive, procs, seed)
+            return
         if urls_file:
             if any(faults.values()):
                 console.print("[yellow]fault flags ignored in --urls-file mode[/]")
@@ -496,6 +681,7 @@ def main(
             if not urls:
                 raise click.ClickException("--urls-file contained no URLs")
             total = len(urls)
+            _procs_notice(procs, len(urls), session)
             console.print(f"[cyan]URL-file mode:[/] {len(urls)} URLs, impersonate={impersonate or 'none'}")
         else:
             # deterministic shared buffer; generated once, served for every segment, and
@@ -507,6 +693,7 @@ def main(
             host, port = str(server.server_address[0]), int(server.server_address[1])
             urls = [f"http://{host}:{port}/seg/{i}" for i in range(segments)]
             active = {k: v for k, v in faults.items() if v}
+            _procs_notice(procs, segments, session)
             console.print(
                 f"[cyan]Synthetic mode:[/] {segments} x {seg_size / MIB:.2f} MiB "
                 f"(latency={latency_ms}ms, rate={rate_kib or 'unbounded'} KiB/s, "

scripts/bench_suite.sh

@@ -73,6 +73,16 @@ run scripts/bench_downloader.py --fault-503 2 --fast-timeouts --segments 8
 run scripts/bench_downloader.py --fault-reset 2 --fast-timeouts --segments 8
 run scripts/bench_downloader.py --fault-stall 1 --fast-timeouts --segments 8 || true
 
+# 2b. Fault-injection reliability + request amplification (fault_server.py; every run is
+# byte-integrity verified, so a CORRUPTION line here is a real finalize-wrong-bytes bug).
+# loopback is the zero-fault python-overhead ceiling; the rest exercise retry/hedge/tail paths.
+run scripts/bench_downloader.py --fault-profile loopback --segments 64 --workers 8,16 --runs 3
+run scripts/bench_downloader.py --fault-profile rate-limit --segments 32 --workers 8
+run scripts/bench_downloader.py --fault-profile flaky-first --segments 16 --workers 8 --fast-timeouts
+run scripts/bench_downloader.py --fault-profile reset --segments 16 --workers 8 --fast-timeouts
+run scripts/bench_downloader.py --fault-profile stall --segments 16 --workers 8 --fast-timeouts
+run scripts/bench_downloader.py --fault-profile tail-slow --segments 32 --workers 8 --adaptive --fast-timeouts
+
 # 3. Multiprocess path (>=24 segments engages spawned children; strided chunk split).
 # big batch: at small sizes the fixed spawn cost drowns the throughput signal
 run scripts/bench_downloader.py --segments 128 --workers 8 --procs 2
@@ -91,6 +101,18 @@ run scripts/bench_downloader.py --segments 32 --workers 8 --impersonate Chrome13
 echo; echo "==================== pytest tests/test_downloader_leaks.py ===================="
 uv run pytest tests/test_downloader_leaks.py -q || echo "(pytest failed or missing - run 'uv sync' to install dev deps)"
 
+# 6b. Low-core variant: pin to two cores to emulate a small box, so the python-overhead
+# ceiling and one fault profile can be read against the full-core numbers above. taskset is
+# util-linux (Linux only); skipped cleanly elsewhere.
+if command -v taskset >/dev/null 2>&1 && taskset -c 0-1 true 2>/dev/null; then
+  echo; echo "==================== low-core variant (taskset -c 0-1, 2 cores) ===================="
+  run_lowcore() { echo; echo "-------------------- taskset -c 0-1 $* --------------------"; taskset -c 0-1 uv run python "$@"; }
+  run_lowcore scripts/bench_downloader.py --fault-profile loopback --segments 64 --workers 8,16 --runs 3
+  run_lowcore scripts/bench_downloader.py --fault-profile rate-limit --segments 32 --workers 8
+else
+  echo; echo "==================== low-core variant skipped (taskset missing or CPUs 0-1 not schedulable) ===================="
+fi
+
 # 7. Live network tests (opt-in)
 if [[ "${1:-}" == "--live" ]]; then
   run scripts/live_manifest_test.py --workers 16 --track largest

scripts/fault_server.py

@@ -0,0 +1,379 @@
+#!/usr/bin/env python3
+"""
+Localhost fault-injection HTTP server for downloader reliability benchmarking.
+
+Serves deterministic pseudo-random segment payloads (seeded per URL path, so the
+bytes are reproducible and any output file can be hashed against them to prove
+integrity) and applies a configurable *fault profile* to those responses. Stall
+and reset faults are decided deterministically from a printed seed keyed on
+(path, attempt), so a retry can recover instead of failing forever; rate-limit
+windows are wall-clock relative to run start, so their timing is not seed-bound.
+
+Fault profiles:
+    stall        sleep N seconds mid-body on X% of (path, attempt)s
+    reset        RST the connection mid-body on X% of (path, attempt)s
+    rate-limit   reply 429 + Retry-After during recurring throttle windows
+                 (window_open seconds serving / window_closed seconds throttled)
+    tail-slow    heavily throttle the last K segments' bodies
+    flaky-first  fail attempt 1 per path (503), succeed on retry
+    loopback     no faults; payloads served from memory (overhead ceiling)
+
+Range requests are honored (206 + Content-Range) so byte-range / hedge / tail-boost
+download paths exercise correctly, and an unsatisfiable resume gets a 416.
+
+Standalone:
+    uv run python scripts/fault_server.py --profile rate-limit --segments 32
+    uv run python scripts/fault_server.py --profile stall --stall-secs 4 --port 8080
+
+Importable: build a ``FaultServer``, serve it on a thread, and hash output against
+``segment_payload(path, size)``.
+"""
+
+from __future__ import annotations
+
+import argparse
+import hashlib
+import json
+import random
+import socket
+import struct
+import threading
+import time
+from dataclasses import asdict, dataclass
+from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
+from typing import Any, Optional
+
+MIB = 1024 * 1024
+DEFAULT_SEED = 1729
+
+
+def segment_payload(path: str, size: int) -> bytes:
+    """Deterministic pseudo-random body for ``path``.
+
+    Pure and seed-stable: the bench regenerates this to byte-compare every output
+    file, so it must never depend on the fault seed, request order, or attempt.
+    """
+    seed = int.from_bytes(hashlib.sha256(path.encode("utf-8")).digest()[:8], "big")
+    return random.Random(seed).randbytes(size)
+
+
+@dataclass
+class FaultProfile:
+    """A named set of deterministic response faults. Presets live in ``PROFILES``."""
+
+    name: str
+    stall_pct: float = 0.0  # % of (path, attempt)s that stall mid-body
+    stall_secs: float = 3.0  # seconds to hold when a stall fires
+    reset_pct: float = 0.0  # % of (path, attempt)s that RST mid-body
+    rate_limit: bool = False  # 429 + Retry-After during closed windows
+    retry_after: int = 1  # Retry-After seconds advertised on a 429
+    window_open: float = 5.0  # seconds serving normally before a throttle window
+    window_closed: float = 5.0  # seconds replying 429 during a throttle window
+    body_rate_kib: int = 0  # throttle every segment body to this KiB/s (0 = unthrottled)
+    tail_slow: int = 0  # throttle the last K segments' bodies
+    tail_rate_kib: int = 256  # throttle rate for tail-slow segments (KiB/s)
+    flaky_first: bool = False  # fail attempt 1 per path, succeed after
+    flaky_status: int = 503  # status returned by a flaky-first failure
+
+
+PROFILES: dict[str, FaultProfile] = {
+    "loopback": FaultProfile("loopback"),
+    "stall": FaultProfile("stall", stall_pct=25.0, stall_secs=3.0),
+    "reset": FaultProfile("reset", reset_pct=25.0),
+    # short windows + a body throttle so a fast loopback batch actually spans several throttle
+    # windows and trips the 429 + Retry-After retry path (a real 900 MB/s batch would otherwise
+    # finish inside the first open window). Standalone use can widen the windows via overrides.
+    "rate-limit": FaultProfile(
+        "rate-limit", rate_limit=True, retry_after=1, window_open=1.0, window_closed=1.0, body_rate_kib=2048
+    ),
+    "tail-slow": FaultProfile("tail-slow", tail_slow=4, tail_rate_kib=256),
+    "flaky-first": FaultProfile("flaky-first", flaky_first=True, flaky_status=503),
+}
+
+
+class FaultServer(ThreadingHTTPServer):
+    """Threaded loopback server serving seeded payloads under a ``FaultProfile``.
+
+    Counters (total requests, per-status, per-path attempts) are the request-amplification
+    signal the bench reads: a retry/hedge storm against a throttled host shows up as
+    requests >> segment count. ``reset_counters`` re-baselines between benchmark runs.
+    """
+
+    allow_reuse_address = True
+    daemon_threads = True
+    request_queue_size = 128  # default 5 drops SYNs past 5 concurrent connects -> phantom 1s retransmit tail
+
+    def __init__(
+        self,
+        profile: FaultProfile,
+        seg_size: int,
+        total_segments: int,
+        seed: int = DEFAULT_SEED,
+        host: str = "127.0.0.1",
+        port: int = 0,
+    ) -> None:
+        self.profile = profile
+        self.seg_size = seg_size
+        self.total_segments = total_segments
+        self.seed = seed
+        self.stall_event = threading.Event()  # set at teardown to release stalled handlers
+        self._lock = threading.Lock()
+        self._request_count = 0
+        self._status_counts: dict[int, int] = {}
+        self._attempts: dict[str, int] = {}
+        self._window_origin = time.monotonic()
+        # precompute the known segment bodies so serving is a dict lookup (no per-request
+        # generation), making the loopback profile a true python-overhead ceiling
+        self._payloads: dict[str, bytes] = {
+            f"/seg/{i}": segment_payload(f"/seg/{i}", seg_size) for i in range(total_segments)
+        }
+        super().__init__((host, port), FaultHandler)
+
+    def handle_error(self, request: Any, client_address: Any) -> None:  # silence expected fault noise
+        pass
+
+    def payload(self, path: str) -> bytes:
+        cached = self._payloads.get(path)
+        if cached is not None:
+            return cached
+        with self._lock:  # unknown path (standalone use): generate + cache once
+            body = self._payloads.get(path)
+            if body is None:
+                body = segment_payload(path, self.seg_size)
+                self._payloads[path] = body
+            return body
+
+    def note_request(self) -> None:
+        with self._lock:
+            self._request_count += 1
+
+    def note_status(self, code: int) -> None:
+        with self._lock:
+            self._status_counts[code] = self._status_counts.get(code, 0) + 1
+
+    def note_attempt(self, path: str) -> int:
+        with self._lock:
+            n = self._attempts.get(path, 0) + 1
+            self._attempts[path] = n
+            return n
+
+    def in_throttle_window(self) -> bool:
+        p = self.profile
+        period = p.window_open + p.window_closed
+        if period <= 0:
+            return False
+        phase = (time.monotonic() - self._window_origin) % period
+        return phase >= p.window_open
+
+    def decides(self, kind: str, path: str, attempt: int, pct: float) -> bool:
+        """Deterministic per-(path, attempt) fault gate: fires for ``pct`` percent of them."""
+        if pct <= 0:
+            return False
+        digest = hashlib.sha256(f"{self.seed}:{kind}:{path}:{attempt}".encode("utf-8")).digest()
+        return (int.from_bytes(digest[:4], "big") / 0xFFFFFFFF) * 100.0 < pct
+
+    def reset_counters(self) -> None:
+        with self._lock:
+            self._request_count = 0
+            self._status_counts.clear()
+            self._attempts.clear()
+        self._window_origin = time.monotonic()  # realign throttle windows to this run's start
+
+    def stats(self) -> dict[str, Any]:
+        with self._lock:
+            max_attempts = max(self._attempts.values(), default=0)
+            return {
+                "requests": self._request_count,
+                "status_counts": dict(sorted(self._status_counts.items())),
+                "max_attempts": max_attempts,
+            }
+
+
+class FaultHandler(BaseHTTPRequestHandler):
+    protocol_version = "HTTP/1.1"
+
+    def log_message(self, *_: Any) -> None:  # silence per-request logging
+        pass
+
+    def _parse_range(self, total: int) -> Optional[tuple[int, int]]:
+        header = self.headers.get("Range")
+        if not header or not header.startswith("bytes="):
+            return None
+        start_s, _, end_s = header[len("bytes=") :].partition("-")
+        if not start_s:  # RFC 7233 suffix range: bytes=-N means the last N bytes
+            return max(0, total - int(end_s)), total - 1
+        start = int(start_s)
+        end = int(end_s) if end_s else total - 1
+        if start > end:  # inverted range: report unsatisfiable so _serve answers 416
+            return total, total - 1
+        return start, min(end, total - 1)
+
+    def do_GET(self) -> None:
+        server: FaultServer = self.server  # type: ignore[assignment]
+        server.note_request()
+        if not self.path.startswith("/seg/"):
+            self.send_error(404)
+            return
+        try:
+            idx = int(self.path.rsplit("/", 1)[-1])
+        except ValueError:
+            self.send_error(404)
+            return
+
+        attempt = server.note_attempt(self.path)
+        prof = server.profile
+
+        if prof.flaky_first and attempt == 1:  # first attempt per path always fails, retry recovers
+            self._send_status(prof.flaky_status)
+            return
+        if prof.rate_limit and server.in_throttle_window():  # recurring throttle window
+            self._send_429(prof.retry_after)
+            return
+
+        rate = prof.body_rate_kib
+        if prof.tail_slow and idx >= server.total_segments - prof.tail_slow:
+            rate = prof.tail_rate_kib
+        stall = server.decides("stall", self.path, attempt, prof.stall_pct)
+        reset = server.decides("reset", self.path, attempt, prof.reset_pct)
+        self._serve(server.payload(self.path), stall=stall, reset=reset, rate_kib=rate)
+
+    def _send_status(self, code: int) -> None:
+        server: FaultServer = self.server  # type: ignore[assignment]
+        server.note_status(code)
+        self.send_response(code)
+        self.send_header("Content-Length", "0")
+        self.end_headers()
+
+    def _send_429(self, retry_after: int) -> None:
+        server: FaultServer = self.server  # type: ignore[assignment]
+        server.note_status(429)
+        self.send_response(429)
+        self.send_header("Retry-After", str(retry_after))
+        self.send_header("Content-Length", "0")
+        self.end_headers()
+
+    def _serve(self, full: bytes, stall: bool = False, reset: bool = False, rate_kib: int = 0) -> None:
+        server: FaultServer = self.server  # type: ignore[assignment]
+        rng = self._parse_range(len(full))
+        if rng is not None and rng[0] >= len(full):  # unsatisfiable resume: a real CDN answers 416
+            server.note_status(416)
+            self.send_response(416)
+            self.send_header("Content-Range", f"bytes */{len(full)}")
+            self.send_header("Content-Length", "0")
+            self.end_headers()
+            return
+        if rng is not None:
+            start, end = rng
+            body = full[start : end + 1]
+            server.note_status(206)
+            self.send_response(206)
+            self.send_header("Content-Range", f"bytes {start}-{end}/{len(full)}")
+        else:
+            body = full
+            server.note_status(200)
+            self.send_response(200)
+        self.send_header("Content-Length", str(len(body)))
+        self.send_header("Accept-Ranges", "bytes")
+        self.send_header("Content-Type", "application/octet-stream")
+        self.end_headers()
+
+        try:
+            if reset:  # send half, then abort the TCP connection (RST via SO_LINGER=0)
+                half = max(1, len(body) // 2)
+                self.wfile.write(body[:half])
+                self.wfile.flush()
+                sock = self.connection
+                sock.setsockopt(socket.SOL_SOCKET, socket.SO_LINGER, struct.pack("ii", 1, 0))
+                sock.close()
+                self.close_connection = True
+            elif stall:  # send half, hold (a client READ_TIMEOUT shorter than stall_secs will retry), then finish
+                half = max(1, len(body) // 2)
+                self.wfile.write(body[:half])
+                self.wfile.flush()
+                server.stall_event.wait(server.profile.stall_secs)
+                self.wfile.write(body[half:])
+                self.wfile.flush()
+            else:
+                self._write_body(body, rate_kib)
+        except (BrokenPipeError, ConnectionError, OSError):
+            self.close_connection = True
+
+    def _write_body(self, body: bytes, rate_kib: int) -> None:
+        if rate_kib <= 0:
+            self.wfile.write(body)
+            return
+        chunk = 64 * 1024  # throttle: fixed-size chunks paced to rate_kib KiB/s
+        per_chunk = chunk / (rate_kib * 1024)
+        for off in range(0, len(body), chunk):
+            self.wfile.write(body[off : off + chunk])
+            self.wfile.flush()
+            time.sleep(per_chunk)
+
+
+def resolve_profile(name: str, overrides: Optional[dict[str, Any]] = None) -> FaultProfile:
+    """Return a copy of the named preset with any non-None ``overrides`` applied."""
+    if name not in PROFILES:
+        raise KeyError(f"unknown profile {name!r}; choices: {', '.join(sorted(PROFILES))}")
+    base = asdict(PROFILES[name])
+    for key, value in (overrides or {}).items():
+        if value is not None and key in base:
+            base[key] = value
+    return FaultProfile(**base)
+
+
+def _build_arg_parser() -> argparse.ArgumentParser:
+    p = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
+    p.add_argument("--port", type=int, default=0, help="Listen port (0 = auto).")
+    p.add_argument("--profile", choices=sorted(PROFILES), default="loopback", help="Fault profile to apply.")
+    p.add_argument("--segments", type=int, default=32, help="Number of /seg/i payloads to precompute.")
+    p.add_argument("--seg-size", type=int, default=2 * MIB, help="Bytes per segment payload.")
+    p.add_argument("--seed", type=int, default=DEFAULT_SEED, help="Fault RNG seed (printed for reproducibility).")
+    # profile knobs (override the chosen preset when given)
+    p.add_argument("--stall-pct", type=float, help="Override: %% of (path, attempt)s that stall.")
+    p.add_argument("--stall-secs", type=float, help="Override: seconds a stall holds.")
+    p.add_argument("--reset-pct", type=float, help="Override: %% of (path, attempt)s that RST.")
+    p.add_argument("--retry-after", type=int, help="Override: Retry-After seconds on a 429.")
+    p.add_argument("--window-open", type=float, help="Override: seconds serving before a throttle window.")
+    p.add_argument("--window-closed", type=float, help="Override: seconds replying 429 during a window.")
+    p.add_argument("--body-rate-kib", type=int, help="Override: throttle every body to this KiB/s (0 = off).")
+    p.add_argument("--tail-slow", type=int, help="Override: throttle the last K segments.")
+    p.add_argument("--tail-rate-kib", type=int, help="Override: tail-slow throttle rate (KiB/s).")
+    return p
+
+
+def main() -> None:
+    args = _build_arg_parser().parse_args()
+    overrides = {
+        "stall_pct": args.stall_pct,
+        "stall_secs": args.stall_secs,
+        "reset_pct": args.reset_pct,
+        "retry_after": args.retry_after,
+        "window_open": args.window_open,
+        "window_closed": args.window_closed,
+        "body_rate_kib": args.body_rate_kib,
+        "tail_slow": args.tail_slow,
+        "tail_rate_kib": args.tail_rate_kib,
+    }
+    profile = resolve_profile(args.profile, overrides)
+    server = FaultServer(profile, args.seg_size, args.segments, args.seed, port=args.port)
+    host, port = server.server_address[0], server.server_address[1]
+    config = {
+        "listen": f"http://{host}:{port}",
+        "seed": args.seed,
+        "segments": args.segments,
+        "seg_size": args.seg_size,
+        "base_url": f"http://{host}:{port}/seg/{{i}}",
+        "profile": asdict(profile),
+    }
+    print(json.dumps(config, indent=2), flush=True)
+    try:
+        server.serve_forever()
+    except KeyboardInterrupt:
+        pass
+    finally:
+        server.stall_event.set()
+        server.shutdown()
+
+
+if __name__ == "__main__":
+    main()