Merge pull request #179 from GetStream/feature/uni-133-fix-leaving-a-call-can-take-long-time

sierpinskid · web-flow · commit 6a51a408ad73 · 2025-11-23T17:14:17.000+01:00
Feature/uni 133 fix leaving a call can take long time
diff --git a/Packages/StreamVideo/Runtime/Core/ConnectionState.cs b/Packages/StreamVideo/Runtime/Core/ConnectionState.cs
@@ -13,6 +13,11 @@ public enum ConnectionState
         /// If was Connected before and allowed by <see cref="IStreamVideoLowLevelClient.ReconnectStrategy"/> it will switch to WaitToReconnect and attempt to connect again after a timeout
         /// </summary>
         Disconnected,
+        
+        /// <summary>
+        /// Disconnecting current user in progress. Client can still reconnect after this.
+        /// </summary>
+        Disconnecting,
 
         /// <summary>
         /// Currently connecting to server, waiting for the connection handshake to complete
@@ -30,7 +35,7 @@ public enum ConnectionState
         Connected,
 
         /// <summary>
-        /// Connection is permanently closing. There will be no reconnects made. StreamChatClient is probably being disposed
+        /// Connection is permanently closing. There will be no reconnects made. StreamVideoClient is probably being disposed
         /// </summary>
         Closing,
     }
@@ -47,6 +52,7 @@ public static bool IsValidToConnect(this ConnectionState state)
                 case ConnectionState.Connecting:
                 case ConnectionState.Connected:
                 case ConnectionState.Closing:
+                case ConnectionState.Disconnecting:
                     return false;
                 case ConnectionState.Disconnected:
                 case ConnectionState.WaitToReconnect:
diff --git a/Packages/StreamVideo/Runtime/Core/LowLevelClient/CallingState.cs b/Packages/StreamVideo/Runtime/Core/LowLevelClient/CallingState.cs
@@ -51,6 +51,11 @@ public enum CallingState
         /// <summary>
         /// The call is in offline mode.
         /// </summary>
-        Offline
+        Offline,
+        
+        /// <summary>
+        /// Leaving operation in progress.
+        /// </summary>
+        Leaving,
     }
 }
diff --git a/Packages/StreamVideo/Runtime/Core/LowLevelClient/ReconnectScheduler.cs b/Packages/StreamVideo/Runtime/Core/LowLevelClient/ReconnectScheduler.cs
@@ -173,6 +173,8 @@ private void OnConnectionStateChanged(ConnectionState previous, ConnectionState
                     break;
                 case ConnectionState.Closing:
                     break;
+                case ConnectionState.Disconnecting:
+                    break;
                 default:
                     throw new ArgumentOutOfRangeException(nameof(current), current, null);
             }
diff --git a/Packages/StreamVideo/Runtime/Core/LowLevelClient/RtcSession.cs b/Packages/StreamVideo/Runtime/Core/LowLevelClient/RtcSession.cs
@@ -369,7 +369,10 @@ public async Task StartAsync(StreamCall call)
 
                 //StreamTodo: validate when this state should set
                 CallState = CallingState.Joined;
+                
+#if STREAM_DEBUG_ENABLED
                 _videoAudioSyncBenchmark?.Init(call);
+#endif
             }
             catch
             {
@@ -380,6 +383,14 @@ public async Task StartAsync(StreamCall call)
 
         public async Task StopAsync(string reason = "")
         {
+            if (CallState == CallingState.Leaving || CallState == CallingState.Offline)
+            {
+                return;
+            }
+
+            CallState = CallingState.Leaving;
+
+
             if (UseNativeAudioBindings)
             {
 #if STREAM_NATIVE_AUDIO
@@ -393,26 +404,20 @@ public async Task StopAsync(string reason = "")
                 {
                     // Trace leave call before leaving the call. Otherwise, stats are not send because SFU WS disconnects
                     _sfuTracer?.Trace(PeerConnectionTraceKey.LeaveCall, new { SessionId = SessionId, Reason = reason });
+
                     if (_statsSender != null) // This was null in tests
                     {
-                        await _statsSender.SendFinalStatsAsync();
+                        using (new TimeLogScope("Sending final stats on leave", _logs.Info))
+                        {
+                            await _statsSender.SendFinalStatsAsync();
+                        }
                     }
                 }
                 catch (Exception e)
                 {
                     _logs.Error($"Failed to send final stats on leave: {e.Message}");
                 }
 
-                _sfuWebSocket.SendLeaveCallRequest(reason);
-
-                for (int i = 0; i < 60; i++)
-                {
-                    if (_sfuWebSocket.SendQueueCount > 0)
-                    {
-                        await Task.Delay(5);
-                    }
-                }
-
 #if STREAM_DEBUG_ENABLED
                 if (_sfuWebSocket.SendQueueCount > 0)
                 {
@@ -424,9 +429,17 @@ public async Task StopAsync(string reason = "")
 
             ClearSession();
             //StreamTodo: check with js definition of "offline" 
+
+            using (new TimeLogScope("Sending leave call request & disconnect", _logs.Info))
+            {
+                await _sfuWebSocket.DisconnectAsync(WebSocketCloseStatus.NormalClosure, reason);
+            }
+            
             CallState = CallingState.Offline;
-            await _sfuWebSocket.DisconnectAsync(WebSocketCloseStatus.NormalClosure, "Video session stopped");
+
+#if STREAM_DEBUG_ENABLED
             _videoAudioSyncBenchmark?.Finish();
+#endif
         }
 
         //StreamTodo: call by call.reconnectOrSwitchSfu()
@@ -666,7 +679,8 @@ private IEnumerable<TrackSubscriptionDetails> GetDesiredTracksDetails()
                     var userId = GetUserId(participant);
                     if (string.IsNullOrEmpty(userId))
                     {
-                        _logs.Error($"Cannot subscribe to {trackType} - participant UserId is null or empty. SessionID: {participant.SessionId}");
+                        _logs.Error(
+                            $"Cannot subscribe to {trackType} - participant UserId is null or empty. SessionID: {participant.SessionId}");
                         continue;
                     }
 
diff --git a/Packages/StreamVideo/Runtime/Core/LowLevelClient/WebSockets/BasePersistentWebSocket.cs b/Packages/StreamVideo/Runtime/Core/LowLevelClient/WebSockets/BasePersistentWebSocket.cs
@@ -73,10 +73,10 @@ private void TryToReconnect()
             {
                 return;
             }
-            
-            #if STREAM_DEBUG_ENABLED
+
+#if STREAM_DEBUG_ENABLED
             Logs.Info($"{GetType()} TryToReconnect");
-            #endif
+#endif
 
             ConnectAsync().LogIfFailed();
         }
@@ -96,17 +96,24 @@ public Task ConnectAsync(CancellationToken cancellationToken = default)
             }
         }
 
-        public Task DisconnectAsync(WebSocketCloseStatus closeStatus, string closeMessage)
+        public async Task DisconnectAsync(WebSocketCloseStatus closeStatus, string closeMessage)
         {
-            //StreamTodo: ignore if already disconnected or disconnecting
-            OnDisconnecting();
+            if (ConnectionState == ConnectionState.Disconnected || ConnectionState == ConnectionState.Closing ||
+                ConnectionState == ConnectionState.Disconnecting)
+            {
+                return;
+            }
+
+            ConnectionState = ConnectionState.Disconnecting;
+            
+            await OnDisconnectingAsync(closeMessage);
 
             if (WebsocketClient == null)
             {
-                return Task.CompletedTask;
+                return;
             }
 
-            return WebsocketClient.DisconnectAsync(closeStatus, closeMessage);
+            await WebsocketClient.DisconnectAsync(closeStatus, closeMessage);
         }
 
         //StreamTodo: either move to coordinator or make generic and pass TMessageType and abstract deserializer
@@ -179,7 +186,7 @@ public void Dispose()
         protected IRequestUriFactory UriFactory { get; }
         protected IAuthProvider AuthProvider { get; }
         protected IReadOnlyDictionary<string, Action<string>> EventHandlers => _eventKeyToHandler;
-        
+
         protected abstract int HealthCheckMaxWaitingTime { get; }
         protected abstract int HealthCheckSendInterval { get; }
 
@@ -198,7 +205,7 @@ protected BasePersistentWebSocket(IWebsocketClient websocketClient, IReconnectSc
 
             WebsocketClient.ConnectionFailed += OnConnectionFailed;
             WebsocketClient.Disconnected += OnDisconnected;
-            
+
             _reconnectScheduler.ReconnectionScheduled += OnReconnectionScheduled;
         }
 
@@ -208,16 +215,16 @@ protected BasePersistentWebSocket(IWebsocketClient websocketClient, IReconnectSc
 
         protected void OnHealthCheckReceived()
         {
-            
 #if STREAM_DEBUG_ENABLED
             var timeSinceLast = Mathf.Round(TimeService.Time - _lastHealthCheckReceivedTime);
             //Logs.Info($"{LogsPrefix} Health check RECEIVED. Time since last: {timeSinceLast} seconds");
 #endif
             _lastHealthCheckReceivedTime = TimeService.Time;
         }
 
-        protected virtual void OnDisconnecting()
+        protected virtual Task OnDisconnectingAsync(string closeMessage)
         {
+            return Task.CompletedTask;
         }
 
         protected virtual void OnDisposing()
@@ -226,8 +233,10 @@ protected virtual void OnDisposing()
 
         private readonly object _websocketConnectionFailedFlagLock = new object();
         private readonly IReconnectScheduler _reconnectScheduler;
+
         private readonly Dictionary<string, Action<string>> _eventKeyToHandler =
             new Dictionary<string, Action<string>>();
+
         private readonly StringBuilder _logSb = new StringBuilder();
 
         private ConnectionState _connectionState;
@@ -275,7 +284,8 @@ private void MonitorHealthCheck()
             var timeSinceLastHealthCheck = TimeService.Time - _lastHealthCheckReceivedTime;
             if (timeSinceLastHealthCheck > HealthCheckMaxWaitingTime)
             {
-                Logs.Warning($"{LogsPrefix} Health check was not received since: {timeSinceLastHealthCheck}, reset connection");
+                Logs.Warning(
+                    $"{LogsPrefix} Health check was not received since: {timeSinceLastHealthCheck}, reset connection");
                 WebsocketClient
                     .DisconnectAsync(WebSocketCloseStatus.InternalServerError,
                         $"{LogsPrefix} Health check was not received since: {timeSinceLastHealthCheck}")
diff --git a/Packages/StreamVideo/Runtime/Core/LowLevelClient/WebSockets/CoordinatorWebSocket.cs b/Packages/StreamVideo/Runtime/Core/LowLevelClient/WebSockets/CoordinatorWebSocket.cs
@@ -103,11 +103,11 @@ protected override async Task ExecuteConnectAsync(CancellationToken cancellation
             await _connectUserTaskSource.Task;
         }
 
-        protected override void OnDisconnecting()
+        protected override async Task OnDisconnectingAsync(string closeMessage)
         {
             _connectUserTaskSource?.TrySetCanceled();
             
-            base.OnDisconnecting();
+            await base.OnDisconnectingAsync(closeMessage);
         }
 
         protected override void OnDisposing()
diff --git a/Packages/StreamVideo/Runtime/Core/LowLevelClient/WebSockets/SfuWebSocket.cs b/Packages/StreamVideo/Runtime/Core/LowLevelClient/WebSockets/SfuWebSocket.cs
@@ -11,6 +11,7 @@
 using StreamVideo.Libs.Logs;
 using StreamVideo.Libs.Serialization;
 using StreamVideo.Libs.Time;
+using StreamVideo.Libs.Utils;
 using StreamVideo.Libs.Websockets;
 using Error = StreamVideo.v1.Sfu.Events.Error;
 using ICETrickle = StreamVideo.v1.Sfu.Models.ICETrickle;
@@ -71,7 +72,7 @@ public void SendLeaveCallRequest(string reason = "")
         {
             if (string.IsNullOrEmpty(_sessionId))
             {
-                throw new ArgumentException($"{nameof(_sessionId)} is null or empty.");
+                return;
             }
 
             if (reason == null)
@@ -103,6 +104,11 @@ public void SendLeaveCallRequest(string reason = "")
 
         protected override void SendHealthCheck()
         {
+            if (ConnectionState != ConnectionState.Connected)
+            {
+                return;
+            }
+            
             var sfuRequest = new SfuRequest
             {
                 HealthCheckRequest = new HealthCheckRequest(),
@@ -114,6 +120,11 @@ protected override void SendHealthCheck()
 
         protected override async Task ExecuteConnectAsync(CancellationToken cancellationToken = default)
         {
+            if (ConnectionState == ConnectionState.Disconnecting || ConnectionState == ConnectionState.Closing)
+            {
+                Logs.Error($"Tried to connect to the {nameof(SfuWebSocket)} while disconnecting or closing. Aborting.");
+                return;
+            }
             //StreamTodo: validate session data
 
             if (string.IsNullOrEmpty(_sfuToken))
@@ -279,11 +290,26 @@ protected override void ProcessMessages()
             }
         }
 
-        protected override void OnDisconnecting()
+        protected override async Task OnDisconnectingAsync(string closeMessage)
         {
             _connectUserTaskSource?.TrySetCanceled();
+            
+            WebsocketClient.ClearSendQueue();
+            
+            using (new TimeLogScope("Sending leave call request", Logs.Info))
+            {
+                SendLeaveCallRequest(closeMessage);
+
+                for (int i = 0; i < 60; i++)
+                {
+                    if (SendQueueCount > 0)
+                    {
+                        await Task.Delay(5);
+                    }
+                }
+            }
 
-            base.OnDisconnecting();
+            await base.OnDisconnectingAsync(closeMessage);
         }
 
         protected override void OnDisposing()
diff --git a/Packages/StreamVideo/Runtime/Core/StatefulModels/IStreamCall.cs b/Packages/StreamVideo/Runtime/Core/StatefulModels/IStreamCall.cs
@@ -126,6 +126,9 @@ public interface IStreamCall : IStreamStatefulModel, IHasCustomData
         /// </summary>
         StreamCallType Type { get; }
 
+        /// <summary>
+        /// Does the user of this client own the call
+        /// </summary>
         bool IsLocalUserOwner { get; }
         
         /// <summary>
diff --git a/Packages/StreamVideo/Runtime/Core/StreamVideoClient.cs b/Packages/StreamVideo/Runtime/Core/StreamVideoClient.cs
@@ -451,6 +451,12 @@ private async Task LeaveCallAsync(IStreamCall call)
         {
             try
             {
+                var callState = InternalLowLevelClient.RtcSession.CallState;
+                if (callState == CallingState.Leaving || callState == CallingState.Offline)
+                {
+                    _logs.Warning($"{nameof(LeaveCallAsync)}: Call is already leaving or offline, skipping leave operation.");
+                    return;
+                }
                 await InternalLowLevelClient.RtcSession.StopAsync("User is leaving the call");
             }
             catch (Exception e)
diff --git a/Packages/StreamVideo/Runtime/Libs/Utils/TimeLogScope.cs b/Packages/StreamVideo/Runtime/Libs/Utils/TimeLogScope.cs
@@ -29,6 +29,7 @@ public void Dispose()
             _sb.Append(_name);
             _sb.Append(" - Executed in: ");
             _sb.Append(elapsed);
+            _sb.Append(" seconds");
 
             _logger(_sb.ToString());
         }
diff --git a/Packages/StreamVideo/Runtime/Libs/Websockets/IWebsocketClient.cs b/Packages/StreamVideo/Runtime/Libs/Websockets/IWebsocketClient.cs
@@ -27,5 +27,7 @@ public interface IWebsocketClient : IDisposable
         void Send(byte[] message);
         
         Task DisconnectAsync(WebSocketCloseStatus closeStatus, string closeMessage);
+
+        void ClearSendQueue();
     }
 }
diff --git a/Packages/StreamVideo/Runtime/Libs/Websockets/NativeWebSocketWrapper.cs b/Packages/StreamVideo/Runtime/Libs/Websockets/NativeWebSocketWrapper.cs
@@ -70,6 +70,11 @@ public async Task ConnectAsync(Uri serverUri)
 
         public Task DisconnectAsync(WebSocketCloseStatus closeStatus, string closeMessage) => DisconnectAsync();
 
+        public void ClearSendQueue()
+        {
+            throw new NotImplementedException();
+        }
+
         public async Task DisconnectAsync()
         {
             if (_webSocket == null)
diff --git a/Packages/StreamVideo/Runtime/Libs/Websockets/WebsocketClient.cs b/Packages/StreamVideo/Runtime/Libs/Websockets/WebsocketClient.cs
@@ -131,7 +131,7 @@ public async Task DisconnectAsync(WebSocketCloseStatus closeStatus, string close
         {
             LogInfoIfDebugMode("Disconnect");
             await TryDisposeResourcesAsync(closeStatus, closeMessage);
-            
+
             _receiveQueue.Clear();
             while (_sendQueue.TryTake(out _))
             {
@@ -140,6 +140,13 @@ public async Task DisconnectAsync(WebSocketCloseStatus closeStatus, string close
             Disconnected?.Invoke();
         }
 
+        public void ClearSendQueue()
+        {
+            while (_sendQueue.TryTake(out _))
+            {
+            }
+        }
+
         public void Dispose()
         {
             LogInfoIfDebugMode("Dispose");

Original file line number	Diff line number	Diff line change
`@@ -51,6 +51,11 @@ public enum CallingState`
`51`	`51`	`/// <summary>`
`52`	`52`	`/// The call is in offline mode.`
`53`	`53`	`/// </summary>`
`54`		`- Offline`
	`54`	`+ Offline,`
	`55`	`+`
	`56`	`+ /// <summary>`
	`57`	`+ /// Leaving operation in progress.`
	`58`	`+ /// </summary>`
	`59`	`+ Leaving,`
`55`	`60`	`}`
`56`	`61`	`}`
Original file line number	Diff line number	Diff line change
`@@ -173,6 +173,8 @@ private void OnConnectionStateChanged(ConnectionState previous, ConnectionState`
`173`	`173`	`break;`
`174`	`174`	`case ConnectionState.Closing:`
`175`	`175`	`break;`
	`176`	`+ case ConnectionState.Disconnecting:`
	`177`	`+ break;`
`176`	`178`	`default:`
`177`	`179`	`throw new ArgumentOutOfRangeException(nameof(current), current, null);`
`178`	`180`	`}`
Original file line number	Diff line number	Diff line change
`@@ -103,11 +103,11 @@ protected override async Task ExecuteConnectAsync(CancellationToken cancellation`
`103`	`103`	`await _connectUserTaskSource.Task;`
`104`	`104`	`}`
`105`	`105`
`106`		`- protected override void OnDisconnecting()`
	`106`	`+ protected override async Task OnDisconnectingAsync(string closeMessage)`
`107`	`107`	`{`
`108`	`108`	`_connectUserTaskSource?.TrySetCanceled();`
`109`	`109`
`110`		`- base.OnDisconnecting();`
	`110`	`+ await base.OnDisconnectingAsync(closeMessage);`
`111`	`111`	`}`
`112`	`112`
`113`	`113`	`protected override void OnDisposing()`
Original file line number	Diff line number	Diff line change
`@@ -451,6 +451,12 @@ private async Task LeaveCallAsync(IStreamCall call)`
`451`	`451`	`{`
`452`	`452`	`try`
`453`	`453`	`{`
	`454`	`+ var callState = InternalLowLevelClient.RtcSession.CallState;`
	`455`	`+ if (callState == CallingState.Leaving \|\| callState == CallingState.Offline)`
	`456`	`+ {`
	`457`	`+ _logs.Warning($"{nameof(LeaveCallAsync)}: Call is already leaving or offline, skipping leave operation.");`
	`458`	`+ return;`
	`459`	`+ }`
`454`	`460`	`await InternalLowLevelClient.RtcSession.StopAsync("User is leaving the call");`
`455`	`461`	`}`
`456`	`462`	`catch (Exception e)`
Original file line number	Diff line number	Diff line change
`@@ -29,6 +29,7 @@ public void Dispose()`
`29`	`29`	`_sb.Append(_name);`
`30`	`30`	`_sb.Append(" - Executed in: ");`
`31`	`31`	`_sb.Append(elapsed);`
	`32`	`+ _sb.Append(" seconds");`
`32`	`33`
`33`	`34`	`_logger(_sb.ToString());`
`34`	`35`	`}`