Function _bucket_stats

tools/loadtest.py:203–233

(
    results: list[Result], bucket_seconds: float, duration: float
)

def _bucket_stats(
    results: list[Result], bucket_seconds: float, duration: float
) -> list[dict]:
    n_buckets = max(1, int(duration // bucket_seconds) + 1)
    buckets: list[list[Result]] = [[] for _ in range(n_buckets)]
    for r in results:
        b = int(r.t_end_s // bucket_seconds)
        if 0 <= b < n_buckets:
            buckets[b].append(r)
    rows = []
    for i, rs in enumerate(buckets):
        if not rs:
            rows.append({"bucket_s": i * bucket_seconds, "n": 0})
            continue
        oks = [r for r in rs if r.ok]
        lats = [r.latency_ms for r in oks]
        st = pcts(lats) if lats else {}
        rows.append(
            {
                "bucket_s": i * bucket_seconds,
                "n": len(rs),
                "ok": len(oks),
                "fail": len(rs) - len(oks),
                "rps": len(rs) / bucket_seconds,
                "p50": st.get("p50"),
                "p95": st.get("p95"),
                "p99": st.get("p99"),
                "max": st.get("max"),
            }
        )
    return rows


# --- HTTP load generation ---------------------------------------------------