_VideoRelay

def __init__(self, server: "DataStreamingServer", display_id: str,
             ws: web.WebSocketResponse, budget: int) -> None:
    self.server = server
    self.display_id = display_id
    self.ws = ws
    self.budget = budget
    self.backlog: deque = deque()
    self.backlog_bytes = 0
    self.live_rows: set[int] = set()
    self.stopped = False
    self._wake = asyncio.Event()
    self._task: Optional[asyncio.Task] = None
    self._next_sync_req = 0.0
    self.verdict = ConnectionVerdict()
    self.sent: Dict[int, None] = {}
    self.lost_run = 0
    self.lost_since = 0.0
    self.lost_first: Optional[int] = None
    self.lost_told = False
    self.since_key = 0
    self.numbered = False
    self.written = 0
    self.marks: deque = deque(maxlen=LOST_FRAME_MEMORY)
    self.gauged = False
    self.pinged = 0.0
    self.pongs = 0
    self.rtt_floor: Optional[float] = None
    self.settled_floor: Optional[float] = None
    self.recent_rtts: Deque[Tuple[float, float]] = deque()
    self.anchored = False
    self.tracked = False
    self.gated = False
    self.lost_gated = False
    self.pad_bps = 0.0
    self.pad_until = 0.0
    self.pad_next = 0.0
    self.pad_pinged = 0.0

def start(self) -> None:
    self._task = asyncio.create_task(
        self._run(), name=f"VideoRelay:{self.display_id}")

def _judge(self, missed: int) -> None:
    """Count one chunk offered or held back and `missed` chunks that will not
    reach this client, telling its page when its connection verdict changes."""
    poor = self.verdict.note(1, missed, time.monotonic())
    if poor is not None:
        _spawn_background_task(_broadcast_to_clients(
            self.server.clients, f"CONNECTION {'poor' if poor else 'ok'}", watched=True, only=id(self.ws)))

def stop(self) -> None:
    """Graceful: an in-flight send completes — canceling mid-frame would
    tear the websocket framing on a socket that stays open for control."""
    self.stopped = True
    self.backlog.clear()
    self.backlog_bytes = 0
    self._leave_common()
    self._wake.set()

def flush_for_gate(self, counted: bool = True, item: Optional[dict] = None) -> None:
    """ACK backpressure engaged: drop the undrained backlog and `item`, the
    chunk held back. A delivering client of a stream naming each frame's
    reference (`tracked`) keeps its rows: the frames dropped open a run, and
    once the gate lifts it resumes on the frame predicting past it (`ungate`).
    An anchor predicting from a frame the client was sent still goes: the
    encoder, which marks it over the anchor before (`settle`), predicts past
    the run from it, where without it only a key frame would do, and the run
    restarts after it. Otherwise every row is gated, so the client resumes
    only at the IDR that _set_backpressure_enabled requests when the gate
    lifts. The chunk held back and the backlog count against the connection
    unless `counted` is False (a stalled client: a page that stopped acking,
    such as a hidden tab, is no judge of its link)."""
    items = [*self.backlog, *([item] if item is not None else ())]
    dropped = [i['data'] for i in items]
    dropped = [d for d in dropped if len(d) >= 12 and d[0] == 0x04]
    if (counted and self.tracked and (self.gated or self.live_rows)
            and not any((d[1] & 0x0F) == 0x01 for d in dropped)):
        self.gated = True
        kept = []
        for it in items:
            data = it['data']
            if len(data) < 12 or data[0] != 0x04:
                continue
            frame_id = (data[2] << 8) | data[3]
            reference = (data[10] << 8) | data[11]
            if data[1] & FRAME_ANCHOR and reference != frame_id and reference in self.sent:
                if it is item:
                    self.sent[frame_id] = None
                    if len(self.sent) > VIDEO_RELAY_SENT_MEMORY:
                        del self.sent[next(iter(self.sent))]
                    self._hold(data, held=False)
                # A run from before it is over; one after it (the anchor still queued from
                # an earlier flush) stands.
                if self.lost_first is not None and 0 < (frame_id - self.lost_first) % 65536 < 32768:
                    self._end_lost()
                kept.append(it)
                continue
            self.sent.pop(frame_id, None)
            self._lose(frame_id)
        self._judge(1 + len(self.backlog) - len(kept))
        self.backlog.clear()
        self.backlog.extend(kept)
        self.backlog_bytes = sum(len(it['data']) for it in kept)
        if kept:
            self._wake.set()
        return
    if counted:
        self._judge(1 + len(self.backlog))
    self.gated = False
    if self.backlog or self.live_rows:
        self.backlog.clear()
        self.backlog_bytes = 0
        self.live_rows.clear()
        self._leave_common()

def ungate(self, resync: bool) -> bool:
    """The gate lifted: whether this client, whose acks caught up (`resync`),
    resumes on its own, on the frame predicting past the run its gate opened
    (`flush_for_gate`), or on the next where none was dropped. Otherwise every
    row waits for the IDR the lift asks for."""
    gated, self.gated = self.gated, False
    if resync and self.tracked and self.live_rows:
        if self.lost_first is not None:
            # The run's wait starts at the lift, and is the unanchored one: the
            # client has room, so the frame predicting past it comes at once.
            self.lost_since = time.monotonic()
            self.lost_gated = True
        return True
    if gated:
        self.live_rows.clear()
        self._leave_common()
    return False

def hold_sync(self) -> None:
    """A key frame for every row is already asked for (the gate's lift): the
    deltas dropped until it arrives ask for no other within the sync floor."""
    self._next_sync_req = time.monotonic() + VIDEO_RELAY_SYNC_FLOOR_SECONDS

def _want_sync(self) -> bool:
    """Rate-limit this relay's keyframe (re)requests to the sync floor."""
    now = time.monotonic()
    if now >= self._next_sync_req:
        self._next_sync_req = now + VIDEO_RELAY_SYNC_FLOOR_SECONDS
        return True
    return False

def _leave_common(self) -> None:
    """This client holds no frame the others do until its next key frame."""
    for held in (False, True):
        common = self.server.common_frames.get((self.display_id, held))
        if common is not None:
            common.leave(self)

def _lose(self, frame_id: int) -> None:
    """Count a frame lost to this client, starting a run where none is open."""
    self.lost_run += 1
    if self.lost_first is None:
        self.lost_first = frame_id
        self.lost_since = time.monotonic()
        self.lost_told = False

def _end_lost(self) -> None:
    """The run is over: the encoder predicted past it, or a key frame came."""
    self.lost_run = 0
    self.lost_first = None
    self.lost_since = 0.0
    self.lost_told = False
    self.lost_gated = False

def _frames_ahead(self) -> int:
    """This client's frames still on their way out of this host: those queued
    here, and those handed to its socket whose last byte it has not sent
    (`_ws_write_backlog`; a frame with fewer bytes behind it than are
    unsent is still in there)."""
    marks = self.marks
    if marks:
        unsent = _ws_write_backlog(self.ws)
        if unsent is None:
            marks.clear()
        while marks and self.written - marks[0] >= unsent:
            marks.popleft()
    return len(self.backlog) + len(marks)

def _lag(self) -> float:
    """How long the oldest ping written behind this client's frames
    (`ping_behind`) has gone unanswered past the path's round trip: the
    queue ahead of it on the path. A ping older than the newest answered
    one was lost, since the socket keeps them in order. 0 where none is
    outstanding or no pong came yet."""
    state = _uplink_session_state(self.ws)
    now = time.monotonic()
    if state["seq"] != self.pongs and state["rtt_us"] is not None:
        self.pongs = state["seq"]
        rtt = state["rtt_us"] / 1e6
        self.rtt_floor = _observe_rtt_floor(state, state["rtt_us"], now) / 1e6
        recent = self.recent_rtts
        recent.append((now, rtt))
        while recent and now - recent[0][0] > VIDEO_RELAY_SETTLE_SECONDS:
            recent.popleft()
        if self.settled_floor is not None and rtt < self.settled_floor:
            self.settled_floor = max(rtt, self.rtt_floor)
        least = min(r for _, r in recent)
        floor = max(self.rtt_floor, self.settled_floor or 0.0)
        if (now - recent[0][0] >= VIDEO_RELAY_SETTLE_SECONDS * 0.9
                and least - floor >= VIDEO_RELAY_LAG_SECONDS):
            data_logger.info(
                f"Video relay[{self.display_id}] for {client_permissions.get(self.ws, {}).get('remote_address')}: "
                f"round trip floor {floor * 1000:.0f} -> {least * 1000:.0f} ms (a queue on its path not its own)")
            self.settled_floor = least
    if self.settled_floor is not None and self.rtt_floor is not None:
        self.rtt_floor = max(self.rtt_floor, self.settled_floor)
    if self.rtt_floor is None:
        return 0.0
    answered = state.get("answered", 0.0)
    oldest = next((sent for sent in state["pending"].values() if sent > answered), None)
    return 0.0 if oldest is None else max(0.0, now - oldest - self.rtt_floor)

def _room(self) -> bool:
    """Whether this client has room for a frame (VIDEO_RELAY_ROOM_FRAMES)."""
    return (self._frames_ahead() < VIDEO_RELAY_ROOM_FRAMES
            and self._lag() < VIDEO_RELAY_LAG_SECONDS)

def _anchor_room(self) -> bool:
    """Whether this client is sent an anchor without room: not far behind, and its
    queue under VIDEO_RELAY_ANCHOR_SECONDS past the round trip."""
    return not self._far() and self._lag() < VIDEO_RELAY_ANCHOR_SECONDS

def _far(self) -> bool:
    """Whether this client is too far behind to be answered without room
    (VIDEO_RELAY_FAR_FRAMES)."""
    return (self._frames_ahead() >= VIDEO_RELAY_FAR_FRAMES
            or self._lag() >= VIDEO_RELAY_FAR_SECONDS)

def _forward(self, frame_id: int, reference: int, anchor: bool = False) -> Optional[bool]:
    """Whether a delta frame naming the frame it predicts from goes to this
    client; None where only a key frame brings the client back. The first
    frame predicting past an open run, which decodes on the client, ends it,
    and an `anchor` goes without room unless the client is too far behind."""
    # In H.264 a frame where frame_num may wrap is never left out.
    kept = self.numbered and not self.since_key % FRAME_NUM_WRAP
    if reference not in self.sent:
        self._lose(frame_id)
        return self._repair()
    if self.lost_first is not None:
        if not anchor or kept or self._anchor_room():
            self._end_lost()
            return True
        # An anchor left out predicts past the run: the encoder is told of it.
        self.lost_first = frame_id
        self.lost_told = False
        self._lose(frame_id)
        return self._repair()
    if self.server._owns_display(self.ws, self.display_id):
        return True
    self.gauged = True
    if anchor and self._anchor_room():
        return True
    if not kept and not self._room():
        self._lose(frame_id)
        return self._repair()
    return True

def _repair(self) -> Optional[bool]:
    """A frame of the open run was held back: tell the encoder of the run's
    first frame once the client has room for the frame predicting past it,
    or sooner where waiting would cost a key frame: at
    VIDEO_RELAY_REACH_FRAMES, while the encoder still holds the frame to
    predict from, and in H.264 VIDEO_RELAY_WRAP_LEAD frames ahead of a
    frame_num wrap (FRAME_NUM_WRAP), which is never left out. False while
    the run waits; None once it has waited VIDEO_RELAY_LOST_RECOVERY_SECONDS,
    or VIDEO_RELAY_ANCHORED_RECOVERY_SECONDS where anchors run and the run
    is not one a gate opened (`ungate`)."""
    limit = (VIDEO_RELAY_ANCHORED_RECOVERY_SECONDS if self.anchored and not self.lost_gated
             else VIDEO_RELAY_LOST_RECOVERY_SECONDS)
    if time.monotonic() - self.lost_since > limit:
        return None
    if not self.lost_told:
        due = (self.lost_run >= VIDEO_RELAY_REACH_FRAMES
               or (self.numbered and -self.since_key % FRAME_NUM_WRAP <= VIDEO_RELAY_WRAP_LEAD))
        if self._room() or (due and not self._far()):
            self.lost_told = self.server._relay_lost(
                self.ws, self.display_id, self.lost_first,
                self.lost_run <= VIDEO_RELAY_REACH_FRAMES)
    return False

def offer(self, item: dict) -> bool:
    """Accept, drop, or gate one encoded chunk.

    Runs on the event loop and never awaits.

    Args:
        item: The fan-out item (`data` memoryview, `owner` frame,
            `frame_id`).

    Returns:
        True when the caller should request a keyframe (data was dropped
        that only a sync point recovers).
    """
    data = item['data']
    size = len(data)
    is_video = size >= 12 and data[0] == 0x04
    is_idr = is_video and (data[1] & 0x0F) == 0x01
    dropped = False
    flushed = 0
    if (not is_idr and self.backlog
            and self.backlog_bytes + size > self.budget):
        flushed = len(self.backlog)
        self.backlog.clear()
        self.backlog_bytes = 0
        self.live_rows.clear()
        self._end_lost()
        self._leave_common()
        dropped = True
    deliver = True
    if is_video:
        row = (data[4] << 8) | data[5]
        frame_id = (data[2] << 8) | data[3]
        reference = (data[10] << 8) | data[11]
        if not is_idr:
            self.tracked = reference != frame_id
        if is_idr:
            self.since_key = 0
            self.numbered = (data[1] >> 4) == WIRE_H264
            self.anchored = False
            self.live_rows.add(row)
            self._end_lost()
            self.sent.clear()
            for held in (False, True):
                self.server.common_frames_for(self.display_id, held).join(self)
        elif row not in self.live_rows:
            deliver = False
            dropped = True
        elif reference != frame_id:
            # A whole-frame session names the frame each delta predicts from;
            # a stripe, or a session that cannot say, names its own id.
            self.since_key += 1
            anchor = bool(data[1] & FRAME_ANCHOR)
            self.anchored = self.anchored or anchor
            forward = self._forward(frame_id, reference, anchor)
            deliver = bool(forward)
            if forward is None:
                self.live_rows.clear()
                self._end_lost()
                self._leave_common()
                dropped = True
    if deliver:
        if is_video:
            self.sent[frame_id] = None
            if len(self.sent) > VIDEO_RELAY_SENT_MEMORY:
                del self.sent[next(iter(self.sent))]
        self.backlog.append(item)
        self.backlog_bytes += size
        self._wake.set()
        if is_video:
            self._hold(data, held=False)
    self._judge(flushed + (not deliver))
    return dropped and self._want_sync()

def _hold(self, data: Any, held: bool = True) -> None:
    """This client holds a video frame its relay wrote, or where not `held` was
    handed one its relay queued: a key frame, or a delta frame naming the one
    it predicts from, which this relay sends only where the client holds that
    one (`CommonFrames`). A relay that drops its queue leaves both."""
    frame_id = (data[2] << 8) | data[3]
    key = (data[1] & 0x0F) == 0x01
    if (((data[4] << 8) | data[5]) in self.live_rows
            and (key or ((data[10] << 8) | data[11]) != frame_id)):
        self.server.common_frames_for(self.display_id, held).hold(self, frame_id, key)

def set_padding(self, bps: float, until: float) -> None:
    """Pad this client's stream at `bps` until `until` (`PathProbe`), in chunks
    written while no frame waits, or stop padding where `bps` is 0."""
    start = bps > 0 and self.pad_bps <= 0
    self.pad_bps, self.pad_until = max(0.0, bps), until
    if start:
        self.pad_next = time.monotonic()
        self._wake.set()

def _padding_due(self, now: float) -> int:
    """The bytes of the padding chunk due now, or 0."""
    if self.pad_bps <= 0 or now >= self.pad_until or now < self.pad_next:
        return 0
    self.pad_next = max(self.pad_next + PAD_STEP_S, now - PAD_STEP_S)
    return max(PAD_CHUNK_MIN, min(PAD_CHUNK_MAX, int(self.pad_bps * PAD_STEP_S / 8)))

async def _write_padding(self, size: int) -> None:
    """Write one padding chunk to the display's owner, counted in what the display
    sent (`sent_bytes`, and `pad_bytes` apart), with a ping behind it each
    PAD_PING_S whose answer acks the bytes sent through it for the link steer
    (`_note_round_trip`)."""
    ds = self.server.display_clients.get(self.display_id)
    if ds is None or ds.get('ws') is not self.ws or not ds.get('backpressure_enabled', True):
        return
    ds['sent_bytes'] = ds.get('sent_bytes', 0) + size
    ds['pad_bytes'] = ds.get('pad_bytes', 0) + size
    sent_bytes, ws = ds['sent_bytes'], self.ws

    def answered(rtt_s: float) -> None:
        if ds.get('ws') is ws and 0.0 <= rtt_s * 1000.0 <= RTT_SAMPLE_SANE_MAX_MS:
            _note_round_trip(ds, rtt_s * 1000.0, sent_bytes, time.monotonic())

    await _send_live(self.ws, memoryview(_PADDING)[:size], f"Padding for '{self.display_id}'")
    now = time.monotonic()
    # Pinged only with no frame awaiting its ack: the browser answers a ping at once and the
    # page acks a frame on its own tick, so a ping behind an unacked frame would ack bytes past
    # it first and the acked history would run out of order.
    if now - self.pad_pinged >= PAD_PING_S and ds.get('unacked_since') is None:
        self.pad_pinged = now
        await ping_behind(self.ws, answered)

async def _run(self) -> None:
    """Drain the backlog onto the socket until stopped or the socket dies,
    padding the stream (`set_padding`) while it holds no frame."""
    try:
        while True:
            if self.stopped:
                return
            if not self.backlog:
                now = time.monotonic()
                pad = self._padding_due(now)
                if pad:
                    try:
                        await self._write_padding(pad)
                    except (ConnectionResetError, OSError, RuntimeError):
                        self.server.clients.discard(self.ws)
                        return
                    continue
                self._wake.clear()
                if self.pad_bps > 0 and now < self.pad_until:
                    with contextlib.suppress(asyncio.TimeoutError):
                        await asyncio.wait_for(self._wake.wait(), max(0.0, self.pad_next - now))
                else:
                    await self._wake.wait()
                continue
            item = self.backlog.popleft()
            data = item['data']
            self.backlog_bytes -= len(data)
            self.written += len(data)
            self.marks.append(self.written)
            # Stamped before the await, and only for the display's
            # registered client: that is what the ACK RTT math measures, and
            # the hand to the socket is where its stats read a frame as sent.
            ds = self.server.display_clients.get(self.display_id)
            if ds is not None and ds.get('ws') is self.ws:
                _note_send(ds, item['frame_id'], len(data))
                watch = self.server._stream_watches.get(self.display_id)
                if watch is not None:
                    watch.note_send(getattr(item.get('owner'), 'capture_ns', 0), len(data))
            try:
                await _send_live(self.ws, data, f"Video relay for '{self.display_id}'")
                if len(data) >= 12 and data[0] == 0x04 and not self.stopped:
                    self._hold(data)
                if self.gauged and time.monotonic() - self.pinged >= VIDEO_RELAY_PING_SECONDS:
                    self.pinged = time.monotonic()
                    await ping_behind(self.ws)
            except (ConnectionResetError, OSError, RuntimeError):
                self.server.clients.discard(self.ws)
                return
    finally:
        group = self.server.video_relay_groups.get(self.display_id)
        if group is not None and group.get(self.ws) is self:
            del group[self.ws]