livekit
diff --git a/‎EventThreadingModel.md‎
Lines changed: 64 additions & 0 deletions b/‎EventThreadingModel.md‎
Lines changed: 64 additions & 0 deletions
diff --git a/‎EventThreadingModel.md.meta‎
Lines changed: 7 additions & 0 deletions b/‎EventThreadingModel.md.meta‎
Lines changed: 7 additions & 0 deletions
diff --git a/‎Runtime/Scripts/DataStream.cs‎
Lines changed: 39 additions & 17 deletions b/‎Runtime/Scripts/DataStream.cs‎
Lines changed: 39 additions & 17 deletions
diff --git a/‎Runtime/Scripts/Internal/FFIClient.cs‎
Lines changed: 87 additions & 16 deletions b/‎Runtime/Scripts/Internal/FFIClient.cs‎
Lines changed: 87 additions & 16 deletions
diff --git a/‎Runtime/Scripts/Internal/FfiInstruction.cs‎
Lines changed: 6 additions & 3 deletions b/‎Runtime/Scripts/Internal/FfiInstruction.cs‎
Lines changed: 6 additions & 3 deletions
@@ -0,0 +1,64 @@
+## Summary
+
+FFI events whose handlers don't touch Unity APIs run directly on the FFI callback thread instead of being marshaled to Unity's main thread via `SynchronizationContext.Post`. 
+
+`FFICallback` previously usually routes Rust events (except `AudioStreamEvent`) through `_context.Post` to Unity's main thread. That's the safe default for handlers that touch Unity APIs (`Texture2D`, `GameObject`, `Transform`, …) but it costs one frame of latency for handlers that don't. Four categories of events can skip that:
+
+1. **Audio stream events** that are written to the audio stream ring buffer and consumed on audio thread.
+2. **One-shot async completions** that only flip `IsDone` on a `YieldInstruction` — `SetMetadata`, `UnpublishTrack`, all stream `Read/Write/Close` ops.
+3. **Stream reader chunk events** that just append bytes/strings to an internal buffer.
+4. **Log batches** — `UnityEngine.Debug.unityLogger` is documented thread-safe; the post hop adds latency without benefit, especially during error storms or `LK_VERBOSE` noise.
+
+## Logic in code:
+
+```csharp
+internal static void RouteFfiEvent(FfiEvent response)
+{
+    if (_isDisposed) return;
+
+    // 1. Per-event-type fast paths — invoke handler directly on FFI thread.
+    if (response.MessageCase == FfiEvent.MessageOneofCase.AudioStreamEvent)      { ...; return; }
+    if (response.MessageCase == FfiEvent.MessageOneofCase.Logs)                  { ...; return; }
+    if (response.MessageCase == FfiEvent.MessageOneofCase.ByteStreamReaderEvent) { ...; return; }
+    if (response.MessageCase == FfiEvent.MessageOneofCase.TextStreamReaderEvent) { ...; return; }
+
+    // 2. One-shot completion fast path — opted-in pending callbacks complete inline.
+    var requestAsyncId = ExtractRequestAsyncId(response);
+    if (requestAsyncId.HasValue && Instance.TrySkipDispatch(requestAsyncId.Value, response))
+        return;
+
+    // 3. Fallback — post to Unity's main-thread sync context.
+    Instance._context?.Post(static (resp) =>
+    {
+        var r = resp as FfiEvent;
+        if (r == null) return;
+        DispatchEvent(r);
+    }, response);
+}
+```
+
+## Event Table
+
+| Event | Where it runs | Why |
+| --- | --- | --- |
+| `AudioStreamEvent` | **FFI thread** (unchanged) | Audio thread consumes the data; main-thread latency would hurt timing |
+| `Logs` | **FFI thread** (new) | `Debug.unityLogger` is thread-safe; logs reach console immediately during panics / errors |
+| `ByteStreamReaderEvent` | **FFI thread** (new) | Internal buffer is now lock-protected; chunks land without frame delay |
+| `TextStreamReaderEvent` | **FFI thread** (new) | Same lock as byte path (shared `ReadIncrementalInstructionBase`) |
+| One-shot completions via `FfiInstruction<T>` | **FFI thread** (new) | `SetLocalMetadata`, `SetLocalName`, `SetLocalAttributes`, `UnpublishTrack` — only flip `IsDone`/`IsError` |
+| One-shot completions via `FfiStreamInstruction<T>` | **FFI thread** (new) | `ByteStreamWriter.Write/Close`, `TextStreamWriter.Write/Close` |
+| One-shot completions via `FfiStreamResultInstruction<T,U>` | **FFI thread** (new) | `ByteStreamReader.ReadAll/WriteToFile`, `TextStreamReader.ReadAll` |
+| `RoomEvent` | Main thread | Fires user-facing `ParticipantConnected`, `TrackPublished`, etc. |
+| `TrackEvent` | Main thread | (No subscribers today; main-thread default for safety) |
+| `RpcMethodInvocation` | Main thread | User RPC handlers commonly touch game state |
+| `Disconnect` | Main thread | UI updates typical |
+| `VideoStreamEvent` | Main thread | Internal buffering is fast; user-facing raw delivery deferred (see follow-ups) |
+| `DataTrackStreamEvent` | Main thread | Deferred until a concrete consumer asks |
+| `Connect` (one-shot) | Main thread | Bespoke handler fires participant-connected events |
+| `PublishTrack` (one-shot) | Main thread | Bespoke handler |
+| `GetStats` (one-shot) | Main thread | Bespoke handler |
+| `CaptureAudioFrame` (one-shot) | Main thread | Bespoke handler |
+| `PerformRpc` (one-shot) | Main thread | Bespoke handler surfaces response |
+| `SendText` / `SendFile` (one-shot) | Main thread | Bespoke handlers |
+| `TextStreamOpen` / `ByteStreamOpen` (one-shot) | Main thread | Bespoke handlers return writer objects |
+| `PublishDataTrack` (one-shot) | Main thread | Bespoke handler |
@@ -80,6 +80,11 @@ public abstract class ReadIncrementalInstructionBase<TContent> : StreamYieldInst
         private readonly Queue<TContent> _pendingChunks = new();
         private TContent _latestChunk;
 
+        // Chunk events arrive on the FFI thread; Reset() and the LatestChunk getter
+        // run on the main-thread coroutine. _gate serializes mutations of the queue,
+        // _latestChunk, IsCurrentReadDone, IsEos, and Error across both sides.
+        private readonly object _gate = new();
+
         /// <summary>
         /// Error that occurred on the last read, if any.
         /// </summary>
@@ -94,8 +99,11 @@ protected TContent LatestChunk
         {
             get
             {
-                if (Error != null) throw Error;
-                return _latestChunk;
+                lock (_gate)
+                {
+                    if (Error != null) throw Error;
+                    return _latestChunk;
+                }
             }
         }
 
@@ -108,34 +116,48 @@ protected ReadIncrementalInstructionBase(FfiHandle readerHandle)
 
         protected void OnChunk(TContent content)
         {
-            if (IsCurrentReadDone)
-            {
-                // Consumer hasn't yielded since the last chunk; buffer until Reset().
-                _pendingChunks.Enqueue(content);
-            }
-            else
+            lock (_gate)
             {
-                _latestChunk = content;
-                IsCurrentReadDone = true;
+                if (IsCurrentReadDone)
+                {
+                    // Consumer hasn't yielded since the last chunk; buffer until Reset().
+                    _pendingChunks.Enqueue(content);
+                }
+                else
+                {
+                    _latestChunk = content;
+                    IsCurrentReadDone = true;
+                }
             }
         }
 
         public override void Reset()
         {
-            base.Reset();
-            if (_pendingChunks.Count > 0)
+            // base.Reset() must run under the same lock as OnChunk, otherwise the
+            // window between IsCurrentReadDone=false (from base) and the dequeue
+            // below lets a producer race in, write _latestChunk, and have its
+            // chunk immediately overwritten by the dequeue. That race lost ~4% of
+            // chunks under stress before this fix.
+            lock (_gate)
             {
-                _latestChunk = _pendingChunks.Dequeue();
-                IsCurrentReadDone = true;
+                base.Reset();
+                if (_pendingChunks.Count > 0)
+                {
+                    _latestChunk = _pendingChunks.Dequeue();
+                    IsCurrentReadDone = true;
+                }
             }
         }
 
         protected void OnEos(Proto.StreamError protoError)
         {
-            IsEos = true;
-            if (protoError != null)
+            lock (_gate)
             {
-                Error = new StreamError(protoError);
+                IsEos = true;
+                if (protoError != null)
+                {
+                    Error = new StreamError(protoError);
+                }
             }
         }
     }
 
@@ -186,7 +186,8 @@ internal void RegisterPendingCallback<TCallback>(
             ulong requestAsyncId,
             Func<FfiEvent, TCallback?> selector,
             Action<TCallback> onComplete,
-            Action? onCancel = null
+            Action? onCancel = null,
+            bool dispatchToMainThread = true
         ) where TCallback : class
         {
             // Request registration must happen before the request is sent. That ordering is what
@@ -198,7 +199,13 @@ internal void RegisterPendingCallback<TCallback>(
             //
             // Duplicate IDs are treated as a hard error because they would allow two unrelated
             // requests to compete for the same completion slot.
-            var pending = new PendingCallback<TCallback>(selector, onComplete, onCancel);
+            //
+            // dispatchToMainThread defaults to true: completion runs on Unity's main thread via
+            // SynchronizationContext.Post, which is safe for any onComplete that touches Unity
+            // APIs or fires user events. Pass false when the onComplete only mutates volatile
+            // YieldInstruction state — RouteFfiEvent will then run it inline on the FFI callback
+            // thread instead of paying a frame of latency for the post.
+            var pending = new PendingCallback<TCallback>(selector, onComplete, onCancel, dispatchToMainThread);
             if (!pendingCallbacks.TryAdd(requestAsyncId, pending))
             {
                 throw new InvalidOperationException($"Duplicate pending callback for request_async_id={requestAsyncId}");
@@ -278,6 +285,15 @@ static unsafe void FFICallback(UIntPtr data, UIntPtr size)
 
             var respData = new Span<byte>(data.ToPointer()!, (int)size.ToUInt64());
             var response = FfiEvent.Parser!.ParseFrom(respData);
+            RouteFfiEvent(response);
+        }
+
+        // Routing logic split out from FFICallback so tests can drive it from a
+        // chosen thread without going through the P/Invoke entry point. Running
+        // production traffic still always lands here via FFICallback above.
+        internal static void RouteFfiEvent(FfiEvent response)
+        {
+            if (_isDisposed) return;
 
             // Audio stream events are handled directly on the FFI callback thread
             // to bypass the main thread, since the audio thread consumes the data
@@ -287,6 +303,47 @@ static unsafe void FFICallback(UIntPtr data, UIntPtr size)
                 return;
             }
 
+            // Log batches are forwarded directly. UnityEngine.Debug.unityLogger is
+            // documented thread-safe; Unity's logger queues to its console drain
+            // internally. Skipping the main-thread post means logs reach the
+            // console without a one-frame delay — useful during error storms,
+            // panics, or LK_VERBOSE noise where the post queue could otherwise
+            // back up.
+            if (response.MessageCase == FfiEvent.MessageOneofCase.Logs)
+            {
+                Utils.HandleLogBatch(response.Logs);
+                return;
+            }
+
+            // Byte stream reader events feed an internal incremental-read buffer that
+            // already serializes mutations under its own lock. Skipping the main-thread
+            // post lets chunks land in the buffer immediately rather than waiting for
+            // the next frame drain.
+            if (response.MessageCase == FfiEvent.MessageOneofCase.ByteStreamReaderEvent)
+            {
+                Instance.ByteStreamReaderEventReceived?.Invoke(response.ByteStreamReaderEvent!);
+                return;
+            }
+
+            // Same treatment for text stream readers — they share
+            // ReadIncrementalInstructionBase<TContent> with the byte path, so the
+            // lock added there already protects all state mutations.
+            if (response.MessageCase == FfiEvent.MessageOneofCase.TextStreamReaderEvent)
+            {
+                Instance.TextStreamReaderEventReceived?.Invoke(response.TextStreamReaderEvent!);
+                return;
+            }
+
+            // One-shot completions registered with dispatchToMainThread:false also bypass the
+            // main thread. The pending callback's onComplete only mutates volatile
+            // YieldInstruction fields, so resolving it here saves up to one frame of latency
+            // on async ops like SetMetadata / UnpublishTrack / stream Read/Write/Close.
+            var requestAsyncId = ExtractRequestAsyncId(response);
+            if (requestAsyncId.HasValue && Instance.TrySkipDispatch(requestAsyncId.Value, response))
+            {
+                return;
+            }
+
             // Run on the main thread, the order of execution is guaranteed by Unity
             // It uses a Queue internally
             Instance._context?.Post(static (resp) =>
@@ -316,9 +373,6 @@ private static void DispatchEvent(FfiEvent ffiEvent)
 
             switch (ffiEvent.MessageCase)
             {
-                case FfiEvent.MessageOneofCase.Logs:
-                    Utils.HandleLogBatch(ffiEvent.Logs);
-                    break;
                 case FfiEvent.MessageOneofCase.PublishData:
                     break;
                 case FfiEvent.MessageOneofCase.RoomEvent:
@@ -338,15 +392,6 @@ private static void DispatchEvent(FfiEvent ffiEvent)
                 case FfiEvent.MessageOneofCase.VideoStreamEvent:
                     Instance.VideoStreamEventReceived?.Invoke(ffiEvent.VideoStreamEvent!);
                     break;
-                case FfiEvent.MessageOneofCase.AudioStreamEvent:
-                    Instance.AudioStreamEventReceived?.Invoke(ffiEvent.AudioStreamEvent!);
-                    break;
-                case FfiEvent.MessageOneofCase.ByteStreamReaderEvent:
-                    Instance.ByteStreamReaderEventReceived?.Invoke(ffiEvent.ByteStreamReaderEvent!);
-                    break;
-                case FfiEvent.MessageOneofCase.TextStreamReaderEvent:
-                    Instance.TextStreamReaderEventReceived?.Invoke(ffiEvent.TextStreamReaderEvent!);
-                    break;
                 case FfiEvent.MessageOneofCase.DataTrackStreamEvent:
                     Instance.DataTrackStreamEventReceived?.Invoke(ffiEvent.DataTrackStreamEvent!);
                     break;
@@ -357,7 +402,7 @@ private static void DispatchEvent(FfiEvent ffiEvent)
             }
         }
 
-        private bool TryDispatchPendingCallback(ulong requestAsyncId, FfiEvent ffiEvent)
+        internal bool TryDispatchPendingCallback(ulong requestAsyncId, FfiEvent ffiEvent)
         {
             // Remove-first dispatch is the key race-proofing step.
             //
@@ -385,6 +430,26 @@ private bool TryDispatchPendingCallback(ulong requestAsyncId, FfiEvent ffiEvent)
             return false;
         }
 
+        // Inline-completion fast path for one-shot callbacks whose onComplete only
+        // mutates volatile YieldInstruction state (no Unity APIs, no user-event
+        // invocations). Returning true means the caller has been fully handled here
+        // — no further dispatch needed. Returning false means the caller should fall
+        // through to its normal main-thread post path. Same race model as
+        // TryDispatchPendingCallback: the side that wins TryRemove is the only side
+        // that may invoke the completion.
+        internal bool TrySkipDispatch(ulong requestAsyncId, FfiEvent ffiEvent)
+        {
+            if (!pendingCallbacks.TryGetValue(requestAsyncId, out var pending))
+            {
+                return false;
+            }
+            if (pending.DispatchToMainThread)
+            {
+                return false;
+            }
+            return TryDispatchPendingCallback(requestAsyncId, ffiEvent);
+        }
+
         private static ulong? ExtractRequestAsyncId(FfiEvent ffiEvent)
         {
             // This switch is only concerned with one-shot async completion callbacks that echo
@@ -420,6 +485,7 @@ private bool TryDispatchPendingCallback(ulong requestAsyncId, FfiEvent ffiEvent)
 
         private abstract class PendingCallbackBase
         {
+            public abstract bool DispatchToMainThread { get; }
             public abstract bool TryComplete(FfiEvent ffiEvent);
             public abstract void Cancel();
         }
@@ -429,16 +495,21 @@ private sealed class PendingCallback<TCallback> : PendingCallbackBase where TCal
             private readonly Func<FfiEvent, TCallback?> selector;
             private readonly Action<TCallback> onComplete;
             private readonly Action? onCancel;
+            private readonly bool dispatchToMainThread;
+
+            public override bool DispatchToMainThread => dispatchToMainThread;
 
             public PendingCallback(
                 Func<FfiEvent, TCallback?> selector,
                 Action<TCallback> onComplete,
-                Action? onCancel
+                Action? onCancel,
+                bool dispatchToMainThread
             )
             {
                 this.selector = selector;
                 this.onComplete = onComplete;
                 this.onCancel = onCancel;
+                this.dispatchToMainThread = dispatchToMainThread;
             }
 
             public override bool TryComplete(FfiEvent ffiEvent)
 
@@ -28,7 +28,8 @@ internal FfiInstruction(
                 {
                     IsError = true;
                     IsDone = true;
-                });
+                },
+                dispatchToMainThread: false);
         }
     }
 
@@ -63,7 +64,8 @@ internal FfiStreamInstruction(
                     Error = new StreamError("Canceled");
                     IsError = true;
                     IsDone = true;
-                });
+                },
+                dispatchToMainThread: false);
         }
     }
 
@@ -116,7 +118,8 @@ internal FfiStreamResultInstruction(
                     Error = new StreamError("Canceled");
                     IsError = true;
                     IsDone = true;
-                });
+                },
+                dispatchToMainThread: false);
         }
     }
 }
Original file line number	Diff line number	Diff line change
`@@ -80,6 +80,11 @@ public abstract class ReadIncrementalInstructionBase<TContent> : StreamYieldInst`
`80`	`80`	`private readonly Queue<TContent> _pendingChunks = new();`
`81`	`81`	`private TContent _latestChunk;`
`82`	`82`
	`83`	`+ // Chunk events arrive on the FFI thread; Reset() and the LatestChunk getter`
	`84`	`+ // run on the main-thread coroutine. _gate serializes mutations of the queue,`
	`85`	`+ // _latestChunk, IsCurrentReadDone, IsEos, and Error across both sides.`
	`86`	`+ private readonly object _gate = new();`
	`87`	`+`
`83`	`88`	`/// <summary>`
`84`	`89`	`/// Error that occurred on the last read, if any.`
`85`	`90`	`/// </summary>`
`@@ -94,8 +99,11 @@ protected TContent LatestChunk`
`94`	`99`	`{`
`95`	`100`	`get`
`96`	`101`	`{`
`97`		`- if (Error != null) throw Error;`
`98`		`- return _latestChunk;`
	`102`	`+ lock (_gate)`
	`103`	`+ {`
	`104`	`+ if (Error != null) throw Error;`
	`105`	`+ return _latestChunk;`
	`106`	`+ }`
`99`	`107`	`}`
`100`	`108`	`}`
`101`	`109`
`@@ -108,34 +116,48 @@ protected ReadIncrementalInstructionBase(FfiHandle readerHandle)`
`108`	`116`
`109`	`117`	`protected void OnChunk(TContent content)`
`110`	`118`	`{`
`111`		`- if (IsCurrentReadDone)`
`112`		`- {`
`113`		`- // Consumer hasn't yielded since the last chunk; buffer until Reset().`
`114`		`- _pendingChunks.Enqueue(content);`
`115`		`- }`
`116`		`- else`
	`119`	`+ lock (_gate)`
`117`	`120`	`{`
`118`		`- _latestChunk = content;`
`119`		`- IsCurrentReadDone = true;`
	`121`	`+ if (IsCurrentReadDone)`
	`122`	`+ {`
	`123`	`+ // Consumer hasn't yielded since the last chunk; buffer until Reset().`
	`124`	`+ _pendingChunks.Enqueue(content);`
	`125`	`+ }`
	`126`	`+ else`
	`127`	`+ {`
	`128`	`+ _latestChunk = content;`
	`129`	`+ IsCurrentReadDone = true;`
	`130`	`+ }`
`120`	`131`	`}`
`121`	`132`	`}`
`122`	`133`
`123`	`134`	`public override void Reset()`
`124`	`135`	`{`
`125`		`- base.Reset();`
`126`		`- if (_pendingChunks.Count > 0)`
	`136`	`+ // base.Reset() must run under the same lock as OnChunk, otherwise the`
	`137`	`+ // window between IsCurrentReadDone=false (from base) and the dequeue`
	`138`	`+ // below lets a producer race in, write _latestChunk, and have its`
	`139`	`+ // chunk immediately overwritten by the dequeue. That race lost ~4% of`
	`140`	`+ // chunks under stress before this fix.`
	`141`	`+ lock (_gate)`
`127`	`142`	`{`
`128`		`- _latestChunk = _pendingChunks.Dequeue();`
`129`		`- IsCurrentReadDone = true;`
	`143`	`+ base.Reset();`
	`144`	`+ if (_pendingChunks.Count > 0)`
	`145`	`+ {`
	`146`	`+ _latestChunk = _pendingChunks.Dequeue();`
	`147`	`+ IsCurrentReadDone = true;`
	`148`	`+ }`
`130`	`149`	`}`
`131`	`150`	`}`
`132`	`151`
`133`	`152`	`protected void OnEos(Proto.StreamError protoError)`
`134`	`153`	`{`
`135`		`- IsEos = true;`
`136`		`- if (protoError != null)`
	`154`	`+ lock (_gate)`
`137`	`155`	`{`
`138`		`- Error = new StreamError(protoError);`
	`156`	`+ IsEos = true;`
	`157`	`+ if (protoError != null)`
	`158`	`+ {`
	`159`	`+ Error = new StreamError(protoError);`
	`160`	`+ }`
`139`	`161`	`}`
`140`	`162`	`}`
`141`	`163`	`}`
Original file line number	Diff line number	Diff line change
`@@ -28,7 +28,8 @@ internal FfiInstruction(`
`28`	`28`	`{`
`29`	`29`	`IsError = true;`
`30`	`30`	`IsDone = true;`
`31`		`- });`
	`31`	`+ },`
	`32`	`+ dispatchToMainThread: false);`
`32`	`33`	`}`
`33`	`34`	`}`
`34`	`35`
`@@ -63,7 +64,8 @@ internal FfiStreamInstruction(`
`63`	`64`	`Error = new StreamError("Canceled");`
`64`	`65`	`IsError = true;`
`65`	`66`	`IsDone = true;`
`66`		`- });`
	`67`	`+ },`
	`68`	`+ dispatchToMainThread: false);`
`67`	`69`	`}`
`68`	`70`	`}`
`69`	`71`
`@@ -116,7 +118,8 @@ internal FfiStreamResultInstruction(`
`116`	`118`	`Error = new StreamError("Canceled");`
`117`	`119`	`IsError = true;`
`118`	`120`	`IsDone = true;`
`119`		`- });`
	`121`	`+ },`
	`122`	`+ dispatchToMainThread: false);`
`120`	`123`	`}`
`121`	`124`	`}`
`122`	`125`	`}`