TransferPacer

def __init__(self, static_bps: int = 0, adaptive: bool = False) -> None:
    self.static_bps = static_bps
    self.adaptive = adaptive
    self.rate_bps = static_bps or 256 * 1024
    self._tokens = self.rate_bps * 0.5
    self._ts = time.monotonic()
    self._congested = False
    self._probe_ceiling = None
    self._slow_start = True
    self._hold_until = 0.0

def connection_state(self, gauged: bool = True) -> Dict[str, Any]:
    """Per-transfer gauge state. The RTT floor is a property of one
    connection's path: sharing it would let a nearby client's short base
    RTT make a distant client's read as permanent congestion. Upload
    reads pass gauged=False: with no honest congestion signal they pace
    against the static cap alone and pass untouched in adaptive-only
    mode, like any other gaugeless connection."""
    return {"rtt_floor_us": None, "gauged": gauged}

async def pace(self, sock: Any, nbytes: int, conn: Dict[str, Any]) -> None:
    """Sample the gauge and sleep off what `nbytes` overdraws.

    After a long idle gap the remembered rate is stale, so the
    multiplicative ramp is re-entered (TCP's restart after idle): a link
    that got faster meanwhile is rediscovered in chunks, not minutes, and
    one that got slower is cut by the first gauge sample. The deficit is
    slept off here and paid back by the next call's elapsed-time refill;
    zeroing the balance after the sleep would credit the slept interval
    twice and double the delivered rate.
    """
    if not self.active:
        return
    if self.adaptive and conn["gauged"]:
        outq = _sock_unsent_bytes(sock) if sock is not None else None
        if outq is not None:
            self._gauge_backoff(
                congested=outq > 192 * 1024, clear=outq < 96 * 1024, cut=0.6)
        else:
            rtt = _sock_rtt_us(sock) if sock is not None else None
            if rtt:
                floor = conn["rtt_floor_us"] = (
                    rtt if conn["rtt_floor_us"] is None
                    else min(conn["rtt_floor_us"], rtt)
                )
                self._gauge_backoff(
                    congested=rtt > floor + 8000, clear=True, cut=0.5)
            else:
                conn["gauged"] = False
    if self.adaptive and not conn["gauged"] and not self.static_bps:
        return
    await self._bucket(nbytes)

async def pace_verdict(self, nbytes: int, congested: Optional[bool]) -> None:
    """Adaptive leg for a transfer gauged by its caller.

    Uploads use this: their congestion lives on the client's uplink,
    readable only over the session socket (`UplinkGauge`), so the caller
    supplies the verdict this side's probes cannot. ``congested=None``
    (no fresh sample) holds the rate and still drains the bucket;
    True/False are one `_gauge_backoff` step. The cut is gentler than the
    socket gauges' — a delay verdict fires at a bounded queue where a
    loss-like signal means one already overflowed — and the growth step
    is proportional rather than the fixed 8 KiB: verdicts arrive at the
    gauge's ping cadence, a few per second where the download gauges
    sample per chunk, and a fixed step at that cadence would take minutes
    to recover a fast link's post-cut rate. Together they hold the AIMD
    sawtooth's duty cycle near the line instead of near half of it.
    """
    if not self.active:
        return
    if self.adaptive and congested is not None:
        step = max(8 * 1024, int(self.rate_bps * 0.03))
        self._gauge_backoff(
            congested=congested, clear=not congested, cut=0.65, step=step)
    await self._bucket(nbytes)

async def _bucket(self, nbytes: int) -> None:
    """Drain `nbytes` from the token bucket, sleeping off any overdraw."""
    now = time.monotonic()
    if self.adaptive and now - self._ts > 10:
        self._slow_start = True
    limit = min(self.rate_bps, self._ceiling)
    self._tokens = min(limit * 0.5, self._tokens + (now - self._ts) * limit)
    self._ts = now
    self._tokens -= nbytes
    if self._tokens < 0:
        await asyncio.sleep(-self._tokens / limit)

def _gauge_backoff(self, congested: bool, clear: bool, cut: float,
                   step: int = 8 * 1024) -> None:
    """One congestion-control step on the shared allowance: a congested
    sample multiplies the rate down; a clear one probes upward —
    multiplicatively while no congestion has ever been seen (the initial
    ramp toward an unknown link rate), additively by ``step`` after
    (fine-grained probing near the working point, TCP's post-ssthresh
    split; the caller sizes the step to its sample cadence).

    The recovery ceiling arms ONCE per congestion epoch, from the rate at
    the epoch's first congested sample (ssthresh semantics): arming it per
    chunk lets a sustained spike ratchet the ceiling toward the floor, and
    computing it from the post-backoff rate pins recovery below the rate
    itself. Reaching the ceiling releases it so clear stretches keep
    probing past the last congested rate; that sawtooth is what keeps a
    link that gets faster later reachable.

    A cut also pauses growth for a drain window: resuming on the first
    clear sample keeps the bottleneck queue standing, and the cut never
    relieves the stream sharing the link. The epoch closes only on a
    clear sample past that window: a clear inside the hold still reflects
    the pre-cut queue draining, and ending the epoch there would let an
    oscillating gauge re-arm the ceiling from each freshly cut rate — the
    same ratchet, one flap at a time."""
    if congested:
        self._slow_start = False
        if not self._congested:
            self._congested = True
            self._probe_ceiling = max(self.rate_bps, 2 * self._RATE_FLOOR)
        self.rate_bps = max(self.rate_bps * cut, self._RATE_FLOOR)
        self._hold_until = time.monotonic() + 1.5
        return
    if not clear:
        return
    if time.monotonic() < self._hold_until:
        return
    self._congested = False
    ceiling = self._probe_ceiling
    if ceiling is not None and self.rate_bps >= ceiling:
        self._probe_ceiling = ceiling = None
    bound = min(
        ceiling if ceiling is not None else self.rate_bps * 4,
        self._ceiling,
    )
    if self._slow_start:
        self.rate_bps = min(self.rate_bps * 2, bound)
    else:
        self.rate_bps = min(self.rate_bps + step, bound)