fix: expose participantIdentity on RoomEvent.DataReceived

[mariusgassen]

• handleDataPacket had packet.participantIdentity but dropped it before passing to handleUserPacket
• added as required param internally, optional trailing arg on RoomEventCallbacks.dataReceived
• allows callers to buffer and replay messages when the sender hasn't joined yet (data channel / participant join event race)

[changeset-bot]

⚠️ No Changeset found

Latest commit: 4cb1d92

Merging this PR will not cause a version bump for any packages. If these changes should not result in a new version, you're good to go. If these changes should result in a version bump, you need to add a changeset.

This PR includes no changesets

When changesets are added to this PR, you'll see the packages that this PR includes changesets for and the associated semver types

Click here to learn what changesets are, and how to add one.

Click here if you're a maintainer who wants to add a changeset to this PR

[devin-ai-integration]

✅ Devin Review: No Issues Found

Devin Review analyzed this PR and found no bugs or issues to report.

[lukasIO]

thanks for the PR. This race shouldn't occur anymore in practice as the server is buffering messages for some time until a participant becomes active.
In what scenarios do you see this happening?
Is there a way to consistently reproduce it?

[mariusgassen]

In what scenarios do you see this happening? Is there a way to consistently reproduce it?

Are you refering to this commit?
livekit/livekit@6a64df2

I was running livekit-server 1.13.2 when implementing a client side buffering of messages with unknown remote participant senders.
I will update to the latest server version and verify this makes our downstream implementation obsolete.

The original scenario was caused by our application's internal protocol. Participants would only accept messages from certain other participants. Without being able to check the sender (message sent directly after join, received before the sender was known as a remote participant), the messages were just discarded.

Edit:
I don't think the above mentioned commit solves our issue.
We use the Go server SDK to publish a data packet on the connected local participant LocalParticipant.PublishDataPacket, not on the RoomService.SendData API.
The message is delivered fine.

The race we hit is purely on the client side: the data channel message (WebRTC) arrives before the participant update (join) signal (WebSocket) and the RemoteParticipant is created. The RTCEngine event loop processes them in either order. We observe both events more or less consistently within a ~30ms window: data received (and dropped) - then participant join.

We can reproduce it by having a participant send a data message immediately on connected, before the receiving participant has had a chance to process the join signal.

Our solution is a client side buffering of messages with no known remote participant - requiring this PRs change.

We already observed and implemented something similar on the server component side:
The Go server SDK provides both - the Sender (*RemoteParticipant) and SenderIdentity (string) in the DataReceiveParams for processing incoming messages.
This way we can buffer messages within a timed window, in which the server-side participant awaits the ParticipantConnected event for the sender before discarding them completly.

[mariusgassen]

I quickly vibe-coded two clients connecting to a room where one immediately sends a message.

Running against a local LiveKit server 1.13.6 it's reliably reproducible - the data channel message arrives before the participant connected event - up to 2500ms late.

async function makeToken(apiKey: string, apiSecret: string, roomName: string, identity: string) {
  const t = new AccessToken(apiKey, apiSecret, { identity });
  t.addGrant({ room: roomName, roomJoin: true, canPublishData: true });
  return t.toJwt();
}

function setupReceiver(room: Room) {
  let dataReceivedAt: number | undefined;
  let raceDetected = false;

  room.on(RoomEvent.DataReceived, (_payload, participant: RemoteParticipant | undefined) => {
    dataReceivedAt = performance.now();
    if (!participant) {
      raceDetected = true;
      log('[receiver] DataReceived — participant: UNDEFINED (race hit!) — sender identity unknown to SDK', 'race');
    } else {
      log(`[receiver] DataReceived — participant: "${participant.identity}" (no race this time)`, 'ok');
    }
  });

  room.on(RoomEvent.ParticipantConnected, (p) => {
    const delta = dataReceivedAt !== undefined
      ? `${(performance.now() - dataReceivedAt).toFixed(1)}ms after DataReceived`
      : 'before DataReceived';
    log(`[receiver] ParticipantConnected — "${p.identity}" arrived ${delta}`);
  });

  room.on(RoomEvent.Connected, () => {
    log('[receiver] Connected event received');
  });

  return { raceDetected: () => raceDetected };
}

function setupSender(room: Room, sendDelayMs: number) {
  room.on(RoomEvent.Connected, () => {
    log(`[sender]   Connected event received — publishing data in ${sendDelayMs}ms`);
    setTimeout(
      () => room.localParticipant.publishData(new TextEncoder().encode('hi'), { reliable: true }),
      sendDelayMs,
    );
  });
}

async function runTest(serverUrl: string, apiKey: string, apiSecret: string, sendDelayMs: number) {
  const roomName = `event-race-${Date.now()}`;
  const receiver = new Room();
  const sender   = new Room();

  const { raceDetected } = setupReceiver(receiver);
  setupSender(sender, sendDelayMs);

  log('[receiver] connecting…');
  await receiver.connect(serverUrl, await makeToken(apiKey, apiSecret, roomName, 'receiver'));
  log('[receiver] connected');

  log('[sender] connecting…');
  await sender.connect(serverUrl, await makeToken(apiKey, apiSecret, roomName, 'sender'));

  await new Promise((r) => setTimeout(r, 5000));

  log('');
  log(
    raceDetected()
      ? '✗ Race detected: RemoteParticipant was undefined at DataReceived time.'
      : '✓ No race this run (timing-dependent — try again).',
    raceDetected() ? 'race' : 'ok',
  );

  await Promise.all([receiver.disconnect(), sender.disconnect()]);
}