Webrtc

VideoFrameProcessor

Base class for processing video frames through workflow.

Can be used independently for data-only processing (no video track output) or as a base for VideoTransformTrackWithLoop when video output is needed.

Source code in inference/core/interfaces/webrtc_worker/webrtc.py

class VideoFrameProcessor:
    """Base class for processing video frames through workflow.

    Can be used independently for data-only processing (no video track output)
    or as a base for VideoTransformTrackWithLoop when video output is needed.
    """

    def __init__(
        self,
        asyncio_loop: asyncio.AbstractEventLoop,
        workflow_configuration: WorkflowConfiguration,
        api_key: str,
        model_manager: Optional[ModelManager] = None,
        data_output: Optional[List[str]] = None,
        stream_output: Optional[str] = None,
        has_video_track: bool = True,
        declared_fps: float = 30,
        termination_date: Optional[datetime.datetime] = None,
        terminate_event: Optional[asyncio.Event] = None,
        heartbeat_callback: Optional[Callable[[], None]] = None,
        realtime_processing: bool = True,
        is_preview: bool = False,
    ):
        self._file_processing = False
        self._loop = asyncio_loop
        self._termination_date = termination_date
        self._terminate_event = terminate_event
        self.track: Optional[MediaStreamTrack] = None
        self._track_active: bool = False
        self._av_logging_set: bool = False
        self._received_frames = 0
        self._declared_fps = declared_fps
        self._stop_processing = False
        self._termination_reason: Optional[str] = None
        self._processing_complete_sent = False
        self.heartbeat_callback = heartbeat_callback

        self.has_video_track = has_video_track
        self.stream_output = stream_output
        self.data_channel: Optional[RTCDataChannel] = None

        # Video file upload support
        self.video_upload_handler: Optional[VideoFileUploadHandler] = None
        self._track_ready_event: asyncio.Event = asyncio.Event()
        self.realtime_processing = realtime_processing
        self._rotation_code: Optional[int] = None

        # Optional receiver-paced flow control (enabled only after first ACK is received)
        self._ack_last: int = 0
        # If ack=1 and window=4, server may produce/send up to frame 5.
        # Configurable via WEBRTC_DATACHANNEL_ACK_WINDOW env var.
        self._ack_window: int = WEBRTC_DATA_CHANNEL_ACK_WINDOW
        self._ack_event: asyncio.Event = asyncio.Event()

        if data_output is None:
            self.data_output = None
            self._data_mode = DataOutputMode.NONE
        elif isinstance(data_output, list):
            self.data_output = [f for f in data_output if f]
            if self.data_output == ["*"]:
                self._data_mode = DataOutputMode.ALL
            elif len(self.data_output) == 0:
                self._data_mode = DataOutputMode.NONE
            else:
                self._data_mode = DataOutputMode.SPECIFIC
        else:
            raise WebRTCConfigurationError(
                f"data_output must be list or None, got {type(data_output).__name__}"
            )

        self._validate_output_fields(workflow_configuration)

        self._inference_pipeline = InferencePipeline.init_with_workflow(
            video_reference=VideoFrameProducer,
            workflow_specification=workflow_configuration.workflow_specification,
            workspace_name=workflow_configuration.workspace_name,
            workflow_id=workflow_configuration.workflow_id,
            api_key=api_key,
            image_input_name=workflow_configuration.image_input_name,
            workflows_parameters=workflow_configuration.workflows_parameters,
            workflows_thread_pool_workers=workflow_configuration.workflows_thread_pool_workers,
            cancel_thread_pool_tasks_on_exit=workflow_configuration.cancel_thread_pool_tasks_on_exit,
            video_metadata_input_name=workflow_configuration.video_metadata_input_name,
            model_manager=model_manager,
            _is_preview=is_preview,
        )

    def set_track(self, track: MediaStreamTrack, rotation_code: Optional[int] = None):
        if not self.track:
            self.track = track
            self._rotation_code = rotation_code
            self._track_ready_event.set()

    async def close(self):
        self._track_active = False
        self._stop_processing = True
        # Clean up video upload handler if present
        if self.video_upload_handler is not None:
            await self.video_upload_handler.cleanup()

    def record_ack(self, ack: int) -> None:
        """Record cumulative ACK from the client.

        ACK semantics: client has fully handled all frames <= ack.
        Backwards compatible: pacing is disabled until we receive the first ACK.
        """
        try:
            ack_int = int(ack)
        except (TypeError, ValueError):
            logger.warning("Invalid ACK value: %s", ack)
            return
        if ack_int < 0:
            logger.warning("Invalid ACK value: %s", ack)
            return
        if ack_int > self._ack_last:
            if ack_int % 100 == 1:
                logger.info("ACK received: %s", ack_int)
            self._ack_last = ack_int
            self._ack_event.set()

    async def _wait_for_ack_window(self, next_frame_id: int) -> None:
        """Block frame production when too far ahead of client ACKs."""
        if self.realtime_processing or self._ack_last == 0:
            return

        wait_counter = 0
        while not self._stop_processing and next_frame_id > (
            self._ack_last + self._ack_window
        ):
            if self._check_termination():
                return
            if self.heartbeat_callback:
                self.heartbeat_callback()

            self._ack_event.clear()
            try:
                await asyncio.wait_for(self._ack_event.wait(), timeout=0.2)
            except asyncio.TimeoutError:
                wait_counter += 1
                if wait_counter % 5 == 1:
                    logger.info(
                        "Waiting for ACK window (next=%d, ack_last=%d, window=%d)",
                        next_frame_id,
                        self._ack_last,
                        self._ack_window,
                    )

    def _check_termination(self):
        """Check if we should terminate based on timeout.

        Does NOT set terminate_event — callers must call _signal_termination()
        after sending final data-channel messages to avoid a race with the
        cleanup task closing the peer connection.
        """
        if self._termination_date and self._termination_date < datetime.datetime.now():
            logger.info("Timeout reached, terminating inference pipeline")
            self._termination_reason = "timeout_reached"
            return True
        if self._terminate_event and self._terminate_event.is_set():
            logger.info("Terminate event set, terminating inference pipeline")
            return True
        return False

    def _signal_termination(self):
        if self._terminate_event:
            self._terminate_event.set()

    @staticmethod
    def serialize_outputs_sync(
        fields_to_send: List[str],
        workflow_output: Dict[str, Any],
        data_output_mode: DataOutputMode,
    ) -> Tuple[Dict[str, Any], List[str]]:
        """Serialize workflow outputs for WebRTC transmission."""
        serialized = {}
        serialization_errors = []

        for field_name in fields_to_send:
            if field_name not in workflow_output:
                serialization_errors.append(f"Output '{field_name}' not found")
                continue

            output_data = workflow_output[field_name]

            if data_output_mode == DataOutputMode.ALL and isinstance(
                output_data, WorkflowImageData
            ):
                continue

            try:
                serialized[field_name] = serialize_for_webrtc(output_data)
            except Exception as e:
                serialization_errors.append(f"{field_name}: {e}")
                serialized[field_name] = {"__serialization_error__": str(e)}
                logger.error("[SERIALIZE] Error: %s - %s", field_name, e)

        return serialized, serialization_errors

    async def _send_data_output(
        self,
        workflow_output: Dict[str, Any],
        frame_timestamp: datetime.datetime,
        frame: VideoFrame,
        errors: List[str],
    ):
        frame_id = self._received_frames

        if not self.data_channel or self.data_channel.readyState != "open":
            return

        video_metadata = WebRTCVideoMetadata(
            frame_id=frame_id,
            received_at=frame_timestamp.isoformat(),
            pts=frame.pts,
            time_base=frame.time_base,
            declared_fps=self._declared_fps,
            height=frame.height,
            width=frame.width,
        )

        webrtc_output = WebRTCOutput(
            serialized_output_data=None,
            video_metadata=video_metadata,
            errors=errors.copy(),
        )

        if self._data_mode == DataOutputMode.NONE:
            json_bytes = await asyncio.to_thread(
                lambda: json.dumps(webrtc_output.model_dump()).encode("utf-8")
            )
            await send_chunked_data(
                self.data_channel,
                frame_id,
                json_bytes,
                heartbeat_callback=self.heartbeat_callback,
            )
            return

        if self._data_mode == DataOutputMode.ALL:
            fields_to_send = list(workflow_output.keys())
        else:
            fields_to_send = self.data_output

        serialized_outputs, serialization_errors = await asyncio.to_thread(
            VideoFrameProcessor.serialize_outputs_sync,
            fields_to_send,
            workflow_output,
            self._data_mode,
        )

        webrtc_output.errors.extend(serialization_errors)
        if serialized_outputs:
            webrtc_output.serialized_output_data = serialized_outputs

        # TODO: use orjson
        json_bytes = await asyncio.to_thread(
            lambda: json.dumps(webrtc_output.model_dump(mode="json")).encode("utf-8")
        )

        if WEBRTC_GZIP_PREVIEW_FRAME_COMPRESSION:

            def compress_json():
                return gzip.compress(json_bytes, compresslevel=6)

            output_bytes = await asyncio.to_thread(compress_json)
        else:
            output_bytes = json_bytes

        success = await send_chunked_data(
            self.data_channel,
            frame_id,
            output_bytes,
            heartbeat_callback=self.heartbeat_callback,
        )
        if not success:
            logger.error("[SEND_OUTPUT] Frame %d failed", frame_id)

    async def _send_processing_complete(self):
        """Send final message indicating processing is complete.

        Also drains the data channel buffer to ensure delivery before the
        connection is closed.
        """
        if self._processing_complete_sent:
            return
        if not self.data_channel or self.data_channel.readyState != "open":
            return

        self._processing_complete_sent = True
        completion_output = WebRTCOutput(
            processing_complete=True,
            termination_reason=self._termination_reason,
            video_metadata=WebRTCVideoMetadata(
                frame_id=self._received_frames,
                received_at=datetime.datetime.now().isoformat(),
            ),
        )
        json_bytes = json.dumps(completion_output.model_dump()).encode("utf-8")
        await send_chunked_data(
            self.data_channel, self._received_frames + 1, json_bytes
        )
        if not await wait_for_buffer_drain(
            self.data_channel, timeout=2.0, low_threshold=0
        ):
            logger.warning(
                "Buffer drain timed out, processing_complete may not reach client"
            )

    async def process_frames_data_only(self):
        """Process frames for data extraction only, without video track output."""
        if not self._av_logging_set:
            av_logging.set_libav_level(av_logging.ERROR)
            self._av_logging_set = True

        try:
            while not self._stop_processing:
                await self._wait_for_ack_window(next_frame_id=self._received_frames + 1)
                if self._check_termination():
                    await self._send_processing_complete()
                    self._signal_termination()
                    break
                if self.heartbeat_callback:
                    self.heartbeat_callback()
                if not self.track or self.track.readyState == "ended":
                    break

                # Drain queue for realtime RTSP
                if (
                    isinstance(self.track, PlayerStreamTrack)
                    and self.realtime_processing
                ):
                    while self.track._queue.qsize() > 30:
                        self.track._queue.get_nowait()

                frame = await self.track.recv()
                self._received_frames += 1
                frame_timestamp = datetime.datetime.now()

                workflow_output, _, errors = await self._process_frame_async(
                    frame=frame,
                    frame_id=self._received_frames,
                    render_output=False,
                    include_errors_on_frame=False,
                )

                await self._send_data_output(
                    workflow_output, frame_timestamp, frame, errors
                )

        except asyncio.CancelledError as exc:
            # No one will catch this exception as it's executed in a create_task
            logger.info("[DATA_ONLY] Processing cancelled: %s", exc)
        except MediaStreamError as exc:
            logger.info("[DATA_ONLY] Media stream ended: %s", exc)
        except Exception as exc:
            logger.error(
                "[DATA_ONLY] Error at frame %d: %s", self._received_frames, exc
            )
        finally:
            await self._send_processing_complete()

    @staticmethod
    def _ensure_workflow_specification(
        workflow_configuration: WorkflowConfiguration, api_key: str
    ) -> None:
        has_specification = workflow_configuration.workflow_specification is not None
        has_workspace_and_workflow_id = (
            workflow_configuration.workspace_name is not None
            and workflow_configuration.workflow_id is not None
        )

        if not has_specification and not has_workspace_and_workflow_id:
            raise WebRTCConfigurationError(
                "Either 'workflow_specification' or both 'workspace_name' and 'workflow_id' must be provided"
            )

        if not has_specification and has_workspace_and_workflow_id:
            try:
                workflow_configuration.workflow_specification = (
                    get_workflow_specification(
                        api_key=api_key,
                        workspace_id=workflow_configuration.workspace_name,
                        workflow_id=workflow_configuration.workflow_id,
                    )
                )
                workflow_configuration.workspace_name = None
                workflow_configuration.workflow_id = None
            except Exception as e:
                raise WebRTCConfigurationError(
                    f"Failed to fetch workflow specification from API: {str(e)}"
                )

    def _validate_output_fields(
        self, workflow_configuration: WorkflowConfiguration
    ) -> None:
        if workflow_configuration.workflow_specification is None:
            return

        workflow_outputs = workflow_configuration.workflow_specification.get(
            "outputs", []
        )
        available_output_names = [o.get("name") for o in workflow_outputs]

        if self._data_mode == DataOutputMode.SPECIFIC:
            invalid_fields = [
                field
                for field in self.data_output
                if field not in available_output_names
            ]
            if invalid_fields:
                raise WebRTCConfigurationError(
                    f"Invalid data_output fields: {invalid_fields}. "
                    f"Available workflow outputs: {available_output_names}"
                )

        if self.stream_output and self.stream_output not in available_output_names:
            raise WebRTCConfigurationError(
                f"Invalid stream_output field: '{self.stream_output}'. "
                f"Available workflow outputs: {available_output_names}"
            )

    async def _process_frame_async(
        self,
        frame: VideoFrame,
        frame_id: int,
        stream_output: Optional[str] = None,
        render_output: bool = True,
        include_errors_on_frame: bool = True,
    ) -> Tuple[Dict[str, Any], Optional[VideoFrame], List[str]]:
        """Async wrapper for process_frame using executor."""

        if self._rotation_code is not None:
            frame = rotate_video_frame(frame, self._rotation_code)

        loop = asyncio.get_running_loop()
        return await loop.run_in_executor(
            None,
            process_frame,
            frame,
            frame_id,
            self._declared_fps,
            self._declared_fps,  # TODO: measure fps
            self._file_processing,
            self._inference_pipeline,
            stream_output,
            render_output,
            include_errors_on_frame,
        )

process_frames_data_only() async

Process frames for data extraction only, without video track output.

Source code in inference/core/interfaces/webrtc_worker/webrtc.py

async def process_frames_data_only(self):
    """Process frames for data extraction only, without video track output."""
    if not self._av_logging_set:
        av_logging.set_libav_level(av_logging.ERROR)
        self._av_logging_set = True

    try:
        while not self._stop_processing:
            await self._wait_for_ack_window(next_frame_id=self._received_frames + 1)
            if self._check_termination():
                await self._send_processing_complete()
                self._signal_termination()
                break
            if self.heartbeat_callback:
                self.heartbeat_callback()
            if not self.track or self.track.readyState == "ended":
                break

            # Drain queue for realtime RTSP
            if (
                isinstance(self.track, PlayerStreamTrack)
                and self.realtime_processing
            ):
                while self.track._queue.qsize() > 30:
                    self.track._queue.get_nowait()

            frame = await self.track.recv()
            self._received_frames += 1
            frame_timestamp = datetime.datetime.now()

            workflow_output, _, errors = await self._process_frame_async(
                frame=frame,
                frame_id=self._received_frames,
                render_output=False,
                include_errors_on_frame=False,
            )

            await self._send_data_output(
                workflow_output, frame_timestamp, frame, errors
            )

    except asyncio.CancelledError as exc:
        # No one will catch this exception as it's executed in a create_task
        logger.info("[DATA_ONLY] Processing cancelled: %s", exc)
    except MediaStreamError as exc:
        logger.info("[DATA_ONLY] Media stream ended: %s", exc)
    except Exception as exc:
        logger.error(
            "[DATA_ONLY] Error at frame %d: %s", self._received_frames, exc
        )
    finally:
        await self._send_processing_complete()

record_ack(ack)

Record cumulative ACK from the client.

ACK semantics: client has fully handled all frames <= ack. Backwards compatible: pacing is disabled until we receive the first ACK.

Source code in inference/core/interfaces/webrtc_worker/webrtc.py

def record_ack(self, ack: int) -> None:
    """Record cumulative ACK from the client.

    ACK semantics: client has fully handled all frames <= ack.
    Backwards compatible: pacing is disabled until we receive the first ACK.
    """
    try:
        ack_int = int(ack)
    except (TypeError, ValueError):
        logger.warning("Invalid ACK value: %s", ack)
        return
    if ack_int < 0:
        logger.warning("Invalid ACK value: %s", ack)
        return
    if ack_int > self._ack_last:
        if ack_int % 100 == 1:
            logger.info("ACK received: %s", ack_int)
        self._ack_last = ack_int
        self._ack_event.set()

serialize_outputs_sync(fields_to_send, workflow_output, data_output_mode) staticmethod

Serialize workflow outputs for WebRTC transmission.

Source code in inference/core/interfaces/webrtc_worker/webrtc.py

@staticmethod
def serialize_outputs_sync(
    fields_to_send: List[str],
    workflow_output: Dict[str, Any],
    data_output_mode: DataOutputMode,
) -> Tuple[Dict[str, Any], List[str]]:
    """Serialize workflow outputs for WebRTC transmission."""
    serialized = {}
    serialization_errors = []

    for field_name in fields_to_send:
        if field_name not in workflow_output:
            serialization_errors.append(f"Output '{field_name}' not found")
            continue

        output_data = workflow_output[field_name]

        if data_output_mode == DataOutputMode.ALL and isinstance(
            output_data, WorkflowImageData
        ):
            continue

        try:
            serialized[field_name] = serialize_for_webrtc(output_data)
        except Exception as e:
            serialization_errors.append(f"{field_name}: {e}")
            serialized[field_name] = {"__serialization_error__": str(e)}
            logger.error("[SERIALIZE] Error: %s - %s", field_name, e)

    return serialized, serialization_errors

VideoTransformTrackWithLoop

Bases: VideoStreamTrack, VideoFrameProcessor

Video track that processes frames through workflow and sends video back.

Inherits from both VideoStreamTrack (for WebRTC video track functionality) and VideoFrameProcessor (for workflow processing logic).

Source code in inference/core/interfaces/webrtc_worker/webrtc.py

class VideoTransformTrackWithLoop(VideoStreamTrack, VideoFrameProcessor):
    """Video track that processes frames through workflow and sends video back.

    Inherits from both VideoStreamTrack (for WebRTC video track functionality)
    and VideoFrameProcessor (for workflow processing logic).
    """

    def __init__(
        self,
        asyncio_loop: asyncio.AbstractEventLoop,
        workflow_configuration: WorkflowConfiguration,
        api_key: str,
        model_manager: Optional[ModelManager] = None,
        data_output: Optional[List[str]] = None,
        stream_output: Optional[str] = None,
        has_video_track: bool = True,
        declared_fps: float = 30,
        termination_date: Optional[datetime.datetime] = None,
        terminate_event: Optional[asyncio.Event] = None,
        heartbeat_callback: Optional[Callable[[], None]] = None,
        realtime_processing: bool = True,
        is_preview: bool = False,
        *args,
        **kwargs,
    ):
        VideoStreamTrack.__init__(self, *args, **kwargs)
        VideoFrameProcessor.__init__(
            self,
            asyncio_loop=asyncio_loop,
            workflow_configuration=workflow_configuration,
            api_key=api_key,
            data_output=data_output,
            stream_output=stream_output,
            has_video_track=has_video_track,
            declared_fps=declared_fps,
            termination_date=termination_date,
            terminate_event=terminate_event,
            model_manager=model_manager,
            heartbeat_callback=heartbeat_callback,
            realtime_processing=realtime_processing,
            is_preview=is_preview,
        )

    async def _auto_detect_stream_output(
        self, frame: VideoFrame, frame_id: int
    ) -> None:
        workflow_output_for_detect, _, _ = await self._process_frame_async(
            frame=frame,
            frame_id=frame_id,
            render_output=False,
            include_errors_on_frame=False,
        )
        detected_output = detect_image_output(workflow_output_for_detect)
        if detected_output:
            self.stream_output = detected_output
            logger.info(f"Auto-detected stream_output: {detected_output}")
        else:
            logger.warning("No image output detected, will use fallback")
            self.stream_output = ""

    async def recv(self):
        # Silencing swscaler warnings in multi-threading environment
        if not self._av_logging_set:
            av_logging.set_libav_level(av_logging.ERROR)
            self._av_logging_set = True

        if self.heartbeat_callback:
            self.heartbeat_callback()

        # Wait for track to be ready (video file upload case)
        if self.track is None:
            logger.info("[RECV] Track is None, waiting for track_ready_event...")
            await self._track_ready_event.wait()
            if self.track is None:
                logger.error("[RECV] Track still None after wait!")
                raise MediaStreamError("Track not available after wait")

        # Optional ACK pacing: block producing the next frame if we're too far ahead.
        await self._wait_for_ack_window(next_frame_id=self._received_frames + 1)

        if self._check_termination():
            logger.warning("[RECV] Termination triggered, closing gracefully")
            await self._send_processing_complete()
            self._signal_termination()
            reason = self._termination_reason or "terminate_event"
            raise MediaStreamError(f"Processing terminated: {reason}")

        # Drain queue if using PlayerStreamTrack (RTSP/video file)
        if isinstance(self.track, PlayerStreamTrack) and self.realtime_processing:
            queue_size = self.track._queue.qsize()
            if queue_size > 30:
                drained = 0
                while self.track._queue.qsize() > 30:
                    self.track._queue.get_nowait()
                    drained += 1
                logger.info(
                    "[RECV] Drained %d frames from queue (was %d)", drained, queue_size
                )

        try:
            frame: VideoFrame = await self.track.recv()
        except MediaStreamError:
            logger.info("[RECV] Track ended after %d frames", self._received_frames)
            await self._send_processing_complete()
            raise

        self._received_frames += 1
        frame_id = self._received_frames
        frame_timestamp = datetime.datetime.now()

        if self.stream_output is None and frame_id == 1:
            await self._auto_detect_stream_output(frame, frame_id)

        workflow_output, new_frame, errors = await self._process_frame_async(
            frame=frame,
            frame_id=frame_id,
            stream_output=self.stream_output,
            render_output=True,
            include_errors_on_frame=True,
        )

        new_frame.pts = frame.pts
        new_frame.time_base = frame.time_base

        await self._send_data_output(workflow_output, frame_timestamp, frame, errors)

        if errors:
            logger.warning("[RECV] Frame %d errors: %s", frame_id, errors)

        return new_frame

create_chunked_binary_message(frame_id, chunk_index, total_chunks, payload)

Create a binary message with standard 12-byte header.

Format: [frame_id: 4][chunk_index: 4][total_chunks: 4][payload: N] All integers are uint32 little-endian.

Source code in inference/core/interfaces/webrtc_worker/webrtc.py

def create_chunked_binary_message(
    frame_id: int, chunk_index: int, total_chunks: int, payload: bytes
) -> bytes:
    """Create a binary message with standard 12-byte header.

    Format: [frame_id: 4][chunk_index: 4][total_chunks: 4][payload: N]
    All integers are uint32 little-endian.
    """
    header = struct.pack("<III", frame_id, chunk_index, total_chunks)
    return header + payload

send_chunked_data(data_channel, frame_id, payload_bytes, chunk_size=CHUNK_SIZE, heartbeat_callback=None, buffer_timeout=120.0) async

Send payload via data channel with chunking and backpressure.

We chunk large payloads because WebRTC data channels have message size limits. We apply backpressure (wait for buffer to drain) to avoid overwhelming the network and causing ICE connection failures.

Heads up: buffer_timeout needs to be higher than WEBRTC_DATA_CHANNEL_BUFFER_DRAINING_DELAY! Otherwise we will timeout ourselves.

Source code in inference/core/interfaces/webrtc_worker/webrtc.py

async def send_chunked_data(
    data_channel: RTCDataChannel,
    frame_id: int,
    payload_bytes: bytes,
    chunk_size: int = CHUNK_SIZE,
    heartbeat_callback: Optional[Callable[[], None]] = None,
    buffer_timeout: float = 120.0,
) -> bool:
    """Send payload via data channel with chunking and backpressure.

    We chunk large payloads because WebRTC data channels have message size limits.
    We apply backpressure (wait for buffer to drain) to avoid overwhelming the
    network and causing ICE connection failures.

    Heads up: buffer_timeout needs to be higher than WEBRTC_DATA_CHANNEL_BUFFER_DRAINING_DELAY!
    Otherwise we will timeout ourselves.
    """

    if buffer_timeout <= WEBRTC_DATA_CHANNEL_BUFFER_DRAINING_DELAY:
        logger.warning(
            "[SEND_CHUNKED] buffer_timeout (%.2fs) <= WEBRTC_DATA_CHANNEL_BUFFER_DRAINING_DELAY (%.2fs), "
            "this will likely cause immediate timeouts during buffer drain",
            buffer_timeout,
            WEBRTC_DATA_CHANNEL_BUFFER_DRAINING_DELAY,
        )

    if data_channel.readyState != "open":
        return False

    payload_size = len(payload_bytes)
    total_chunks = (payload_size + chunk_size - 1) // chunk_size
    view = memoryview(payload_bytes)
    high_threshold = WEBRTC_DATA_CHANNEL_BUFFER_SIZE_LIMIT

    for chunk_index in range(total_chunks):
        if data_channel.readyState != "open":
            logger.error(
                "[SEND_CHUNKED] Channel closed at chunk %d/%d",
                chunk_index,
                total_chunks,
            )
            return False

        start = chunk_index * chunk_size
        end = min(start + chunk_size, payload_size)
        chunk_data = view[start:end]

        message = create_chunked_binary_message(
            frame_id, chunk_index, total_chunks, chunk_data
        )

        if data_channel.bufferedAmount > high_threshold:
            if not await wait_for_buffer_drain(
                data_channel, buffer_timeout, heartbeat_callback
            ):
                logger.error(
                    "[SEND_CHUNKED] Buffer drain failed at chunk %d/%d",
                    chunk_index,
                    total_chunks,
                )
                return False

        data_channel.send(message)

        if heartbeat_callback:
            heartbeat_callback()
        await asyncio.sleep(0.001)

    return True

wait_for_buffer_drain(data_channel, timeout=30.0, heartbeat_callback=None, low_threshold=None) async

Wait for data channel buffer to drain below threshold, with timeout.

We use a low threshold (1/4 of limit) instead of just below the limit to avoid hysteresis - constantly triggering this wait after sending just a few chunks.

And we wait WEBRTC_DATA_CHANNEL_BUFFER_DRAINING_DELAY to avoid starving the event loop.

Source code in inference/core/interfaces/webrtc_worker/webrtc.py

async def wait_for_buffer_drain(
    data_channel: RTCDataChannel,
    timeout: float = 30.0,
    heartbeat_callback: Optional[Callable[[], None]] = None,
    low_threshold: Optional[int] = None,
) -> bool:
    """Wait for data channel buffer to drain below threshold, with timeout.

    We use a low threshold (1/4 of limit) instead of just below the limit to avoid
    hysteresis - constantly triggering this wait after sending just a few chunks.

    And we wait WEBRTC_DATA_CHANNEL_BUFFER_DRAINING_DELAY to avoid starving the
    event loop.
    """
    if low_threshold is None:
        low_threshold = WEBRTC_DATA_CHANNEL_BUFFER_SIZE_LIMIT // 4

    start_time = asyncio.get_event_loop().time()

    while data_channel.bufferedAmount > low_threshold:
        elapsed = asyncio.get_event_loop().time() - start_time
        if elapsed > timeout:
            logger.error("[BUFFER_DRAIN] Timeout after %.1fs", timeout)
            return False
        if data_channel.readyState != "open":
            logger.error("[BUFFER_DRAIN] Channel closed: %s", data_channel.readyState)
            return False
        if heartbeat_callback:
            heartbeat_callback()
        await asyncio.sleep(WEBRTC_DATA_CHANNEL_BUFFER_DRAINING_DELAY)

    return True