UplinkGauge

Congestion verdicts for one client upload, timed end to end over the uploader's own session websocket(s).

Nothing on this side can see the client's uplink queue — it stands in the client's kernel and first hop — but every session connection from that client (the data WebSocket, or the WebRTC signaling socket) crosses the same uplink bottleneck as the upload. So the gauge sends a protocol ping every PING_INTERVAL on each session socket and times the pong: the reply rides back through the standing queue, and its round trip inflating past a per-session floor by more than INFLATION_US is the verdict a delay-based scavenger backs off on (LEDBAT's discipline: take what the link has spare, yield the moment anything else waits). Browsers answer pings in the network process, so no page or client-code cooperation is needed. The rejected alternative — reading flat out to measure the client's rate, then settling on a fixed share of it — filled the very queue it existed to prevent, latched a one-shot estimate, and held idle-screen uploads to half the link. Kernel srtt (TCP_INFO) was rejected too: it times the first hop's ACK, so any relaying middlebox — a reverse proxy, a tunnel — blinds it, where the pong crosses end to end.

The end-to-end trip also spans both event loops, so a server too loaded to serve the stream reads as congestion and sheds the upload first — load protection for free. INFLATION_US sits above scheduler jitter and below what a session feels: the queue the sawtooth sustains stays imperceptible.

Ping payloads are per-session counters; the pong echoes its ping's payload (RFC 6455), so note_pong matches replies to send times through _UPLINK_SESSIONS and stray pongs — aiohttp's own heartbeat's among them — fall through unmatched. A ping failure drops that socket; a gauge with no sockets left reports not alive and the caller stops pacing — a vanished session leaves nothing to protect.

Attributes

Functions

def __init__(self, conns: List[List[Any]]) -> None:
    """``conns``: ``[websocket, session state, last-consumed seq]`` per
    session connection sharing the uploader's uplink."""
    self._conns = conns
    self._last_ping = 0.0

async def sample(self) -> Optional[bool]:
    """Ping on cadence and return one verdict across the gauged sockets.

    Returns:
        True when any session's round trip stands above its floor by more
        than `INFLATION_US`, False when at least one session answered
        fresh and none congested, None when no fresh pong arrived since
        the last call (the caller holds its rate).
    """
    now = time.monotonic()
    if now - self._last_ping >= self.PING_INTERVAL:
        self._last_ping = now
        for conn in list(self._conns):
            ws, state, _last = conn
            payload = state["next_ping"].to_bytes(8, "big")
            state["next_ping"] += 1
            pending = state["pending"]
            pending[payload] = now
            while len(pending) > self._PENDING_MAX:
                pending.pop(next(iter(pending)))
            try:
                await ws.ping(payload)
            except Exception:
                self._conns.remove(conn)
    verdict: Optional[bool] = None
    for conn in self._conns:
        _ws, state, last = conn
        if state["seq"] == last or state["rtt_us"] is None:
            continue
        conn[2] = state["seq"]
        rtt = state["rtt_us"]
        floor = _observe_rtt_floor(state, rtt, now)
        congested = rtt > floor + self.INFLATION_US
        verdict = congested if verdict is None else (verdict or congested)
    return verdict