_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

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

def stop(self) -> None:
    """Graceful: an in-flight send completes — cancelling 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._wake.set()

def flush_for_gate(self) -> None:
    """ACK backpressure engaged: drop the undrained backlog and gate every
    row, so the client resumes only at the IDR that
    _set_backpressure_enabled requests when the gate lifts."""
    if self.backlog or self.live_rows:
        self.backlog.clear()
        self.backlog_bytes = 0
        self.live_rows.clear()

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 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_h264 = size >= 10 and data[0] == 0x04
    is_idr = is_h264 and data[1] == 0x01
    dropped = False
    if (not is_idr and self.backlog
            and self.backlog_bytes + size > self.budget):
        self.backlog.clear()
        self.backlog_bytes = 0
        self.live_rows.clear()
        dropped = True
    deliver = True
    if is_h264:
        row = (data[4] << 8) | data[5]
        if is_idr:
            self.live_rows.add(row)
        elif row not in self.live_rows:
            deliver = False
            dropped = True
    if deliver:
        self.backlog.append(item)
        self.backlog_bytes += size
        self._wake.set()
    return dropped and self._want_sync()

async def _run(self) -> None:
    """Drain the backlog onto the socket until stopped or the socket dies."""
    try:
        while True:
            if self.stopped:
                return
            if not self.backlog:
                self._wake.clear()
                await self._wake.wait()
                continue
            item = self.backlog.popleft()
            data = item['data']
            self.backlog_bytes -= len(data)
            # Stamped before the await, and only for the display's
            # registered client: that is what the ACK RTT math measures.
            ds = self.server.display_clients.get(self.display_id)
            if ds is not None and ds.get('ws') is self.ws:
                fid = item['frame_id']
                ds['sent_timestamps'][fid] = time.monotonic()
                ds['last_sent_frame_id'] = fid
                ds['has_sent_any_frame'] = True
                if len(ds['sent_timestamps']) > SENT_FRAME_TIMESTAMP_HISTORY_SIZE:
                    ds['sent_timestamps'].popitem(last=False)
            try:
                await asyncio.wait_for(
                    self.ws.send_bytes(data),
                    timeout=SHARED_STREAM_SEND_TIMEOUT_SECONDS,
                )
            except asyncio.TimeoutError:
                # Checked before OSError: on 3.11+ TimeoutError subclasses it.
                data_logger.warning(
                    f"Video relay for '{self.display_id}' send stalled past "
                    f"{SHARED_STREAM_SEND_TIMEOUT_SECONDS}s; dropping client.")
                self.server.clients.discard(self.ws)
                _close_abandoned_ws(self.ws)
                return
            except (ConnectionResetError, OSError, RuntimeError):
                self.server.clients.discard(self.ws)
                return
            self.server._bytes_sent_in_interval += len(data)
    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]