From 8116d3f42f68d76e840f715e88c4e9dde290301d Mon Sep 17 00:00:00 2001
From: zek <3463291+zek@users.noreply.github.com>
Date: Wed, 28 Oct 2026 13:27:57 -0500
Subject: [PATCH] fix(rtc): send and receive data packets without a new ctypes
 array type

publish_data built a `ctypes.c_byte * data_len` array type for each new
payload length, and the data packet receive path did the same with
`ctypes.POINTER(ctypes.c_byte * data_len)`. Each ctypes array type is a
reference cycle, so every packet of a new length left garbage that only
the cyclic garbage collector can free.

Pass the address of the bytes payload directly on the send side, and read
the received buffer with ctypes.string_at, as FfiClient.request already does.
---
 livekit-rtc/livekit/rtc/participant.py |  6 ++--
 livekit-rtc/livekit/rtc/room.py        |  7 +---
 livekit-rtc/tests/test_dc.py           | 44 ++++++++++++++++++++++++++
 3 files changed, 47 insertions(+), 10 deletions(-)

diff --git a/livekit-rtc/livekit/rtc/participant.py b/livekit-rtc/livekit/rtc/participant.py
index 6c54d533..07b038b3 100644
--- a/livekit-rtc/livekit/rtc/participant.py
+++ b/livekit-rtc/livekit/rtc/participant.py
@@ -14,7 +14,6 @@
 
 from __future__ import annotations
 
-import ctypes
 import asyncio
 import functools
 import inspect
@@ -42,7 +41,7 @@
 from ._proto.track_pb2 import (
     ParticipantTrackPermission,
 )
-from ._utils import BroadcastQueue
+from ._utils import BroadcastQueue, get_address
 from .track import LocalTrack
 from .track_publication import (
     LocalTrackPublication,
@@ -336,11 +335,10 @@ async def publish_data(
             payload = payload.encode("utf-8")
 
         data_len = len(payload)
-        cdata = (ctypes.c_byte * data_len)(*payload)
 
         req = proto_ffi.FfiRequest()
         req.publish_data.local_participant_handle = self._ffi_handle.handle
-        req.publish_data.data_ptr = ctypes.addressof(cdata)
+        req.publish_data.data_ptr = get_address(payload)
         req.publish_data.data_len = data_len
         req.publish_data.reliable = reliable
         req.publish_data.topic = topic
diff --git a/livekit-rtc/livekit/rtc/room.py b/livekit-rtc/livekit/rtc/room.py
index e58321eb..e533d205 100644
--- a/livekit-rtc/livekit/rtc/room.py
+++ b/livekit-rtc/livekit/rtc/room.py
@@ -1115,12 +1115,7 @@ def _on_room_event(self, event: proto_room.RoomEvent) -> None:
             if which_val == "user":
                 owned_buffer_info = packet.user.data
                 buffer_info = owned_buffer_info.data
-                native_data = ctypes.cast(
-                    buffer_info.data_ptr,
-                    ctypes.POINTER(ctypes.c_byte * buffer_info.data_len),
-                ).contents
-
-                data = bytes(native_data)
+                data = ctypes.string_at(buffer_info.data_ptr, buffer_info.data_len)
                 FfiHandle(owned_buffer_info.handle.id).dispose()
                 rparticipant = cast(
                     RemoteParticipant,
diff --git a/livekit-rtc/tests/test_dc.py b/livekit-rtc/tests/test_dc.py
index aff64e3f..7c4ddb0e 100644
--- a/livekit-rtc/tests/test_dc.py
+++ b/livekit-rtc/tests/test_dc.py
@@ -12,6 +12,8 @@
 from __future__ import annotations
 
 import asyncio
+import ctypes
+import gc
 from typing import Optional
 
 import pytest
@@ -277,3 +279,45 @@ async def test_data_targeted_with_topic() -> None:
     await _assert_no_data(other, other_collector)
 
     await asyncio.gather(sender.disconnect(), target.disconnect(), other.disconnect())
+
+
+@skip_if_no_credentials()  # type: ignore[untyped-decorator]
+@pytest.mark.asyncio
+async def test_publish_data_makes_no_ctypes_array_type() -> None:
+    """Sending and receiving payloads of new lengths makes no ctypes array type."""
+    room_name = unique_room_name("py-dc-array-type")
+
+    sender = rtc.Room()
+    receiver = rtc.Room()
+
+    await connect_room("sender", room_name, room=sender)
+    await connect_room("receiver", room_name, room=receiver)
+    await ensure_rooms_all_connected([sender, receiver])
+    await ensure_participants_visible(sender, ["receiver"])
+
+    collector = _DataCollector(receiver, sender_identity="sender")
+    lengths = range(1001, 1011)
+    payloads = [bytes([i]) * n for i, n in enumerate(lengths)]
+
+    gc.collect()
+    gc.disable()
+    try:
+        for payload in payloads:
+            await sender.local_participant.publish_data(payload)
+        await wait_until(
+            lambda: len(collector.packets) >= len(payloads),
+            message="receiver did not get all payloads",
+        )
+        array_types = [
+            o
+            for o in gc.get_objects()
+            if isinstance(o, type)
+            and issubclass(o, ctypes.Array)
+            and getattr(o, "_length_", None) in lengths
+        ]
+    finally:
+        gc.enable()
+    assert array_types == []
+    assert collector.payloads() == payloads
+
+    await asyncio.gather(sender.disconnect(), receiver.disconnect())