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
    self._last_inflation_us: Optional[int] = None

async def pace(self, nbytes: int) -> None:
    """Sleep off what `nbytes` overdraws from the allowance as it stands.

    The static leg: no verdict is folded in, so on a purely adaptive
    pacer this holds the rate the verdicts last set. 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
    await self._bucket(nbytes)

async def pace_verdict(self, nbytes: int, congested: Optional[bool],
                       inflation_us: Optional[int] = None) -> None:
    """Adaptive leg: fold one `UplinkGauge` verdict in, then drain.

    ``congested=None`` (no fresh sample) holds the rate and still drains
    the bucket; True/False are one `_gauge_backoff` step, with the
    gauge's measured inflation telling a draining queue from a standing
    one. The cut is gentle — a delay verdict fires at a bounded queue
    where a loss-like signal would mean one already overflowed — and the
    growth step is proportional rather than a fixed 8 KiB: verdicts
    arrive at the gauge's ping cadence, a few per second, 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,
            inflation_us=inflation_us)
    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, inflation_us: Optional[int] = None) -> 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 and further cuts for a drain window: a
    sample inside it, clear or congested, still reflects the queue that
    cut is draining. Resuming on a clear would keep the bottleneck queue
    standing, and the cut never relieves the stream sharing the link;
    cutting again on each congested sample would take a fat buffer's
    worth of samples to the floor before it empties, and the recovery
    from there is additive. Past the window a queue still standing draws
    another full cut, but one the gauge measures as shrinking is already
    draining at the rate in force and gets a quarter of it: the full cut
    would only trade the drain's last seconds for a recovery that takes
    many times longer. The epoch closes only on a clear
    sample past that window: ending it on one inside 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
        now = time.monotonic()
        draining = (inflation_us is not None and self._last_inflation_us is not None
                    and inflation_us < self._last_inflation_us)
        self._last_inflation_us = inflation_us
        if not self._congested:
            self._congested = True
            self._probe_ceiling = max(self.rate_bps, 2 * self._RATE_FLOOR)
        elif now < self._hold_until:
            return
        elif draining:
            cut = 1 - (1 - cut) / 4
        self.rate_bps = max(self.rate_bps * cut, self._RATE_FLOOR)
        self._hold_until = now + 1.5
        return
    self._last_inflation_us = None
    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)