From eadef45ad78d1beda65830dbce8cb073623979e8 Mon Sep 17 00:00:00 2001 From: Harlan Crystal Date: Sat, 25 Jul 2026 21:22:20 -0700 Subject: [PATCH 1/4] Add IWebsocketClient.QueuedMessageCount (receive-queue depth) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Exposes how many received messages are waiting to be dequeued. Receiving runs on a background timer thread while consumers dequeue from Unity's main loop, so the depth is the only measure of how far behind the consumer is, and until now nothing could observe it. Used by the stalled-pump warning added later in this series. Deliberately read-only: there is no companion "discard the backlog" call, because a dropped protocol event cannot be recovered. The only catch-up mechanism is /sync, which the server rejects past roughly 1000 missed events, so discarding would leave local channel state permanently stale. Purely additive to SDK behavior, though note it is a new member on the public IWebsocketClient interface, so an integrator supplying their own implementation will need to add it. Implemented on both bundled ones: * WebsocketClient — ConcurrentQueue.Count (all platforms but WebGL) * NativeWebSocketWrapper — Queue.Count (UNITY_WEBGL only, so it is not covered by a non-WebGL compile) --- .../Libs/NativeWebSocket/NativeWebSocketWrapper.cs | 3 +++ .../StreamChat/Libs/Websockets/IWebsocketClient.cs | 9 +++++++++ .../StreamChat/Libs/Websockets/WebsocketClient.cs | 3 +++ 3 files changed, 15 insertions(+) diff --git a/Assets/Plugins/StreamChat/Libs/NativeWebSocket/NativeWebSocketWrapper.cs b/Assets/Plugins/StreamChat/Libs/NativeWebSocket/NativeWebSocketWrapper.cs index f25a6d86..29457eb0 100644 --- a/Assets/Plugins/StreamChat/Libs/NativeWebSocket/NativeWebSocketWrapper.cs +++ b/Assets/Plugins/StreamChat/Libs/NativeWebSocket/NativeWebSocketWrapper.cs @@ -35,6 +35,9 @@ public bool TryDequeueMessage(out string message) return message != null; } + /// + public int QueuedMessageCount => _messages.Count; + public async Task ConnectAsync(Uri serverUri, int timeout = 5) { if (_webSocket != null) diff --git a/Assets/Plugins/StreamChat/Libs/Websockets/IWebsocketClient.cs b/Assets/Plugins/StreamChat/Libs/Websockets/IWebsocketClient.cs index 8ee3130c..14f7742f 100644 --- a/Assets/Plugins/StreamChat/Libs/Websockets/IWebsocketClient.cs +++ b/Assets/Plugins/StreamChat/Libs/Websockets/IWebsocketClient.cs @@ -15,6 +15,15 @@ public interface IWebsocketClient : IDisposable bool TryDequeueMessage(out string message); + /// + /// Number of received messages waiting to be dequeued. Receiving runs on a + /// background timer while consumers dequeue from Unity's main loop, so this reports how + /// far behind the consumer is. Diagnostic only: the transport never drops messages, and + /// consumers must not either — a discarded protocol event cannot be recovered, since the + /// only catch-up mechanism (/sync) is itself limited to roughly 1000 missed events. + /// + int QueuedMessageCount { get; } + Task ConnectAsync(Uri serverUri, int timeout = 3); void Update(); diff --git a/Assets/Plugins/StreamChat/Libs/Websockets/WebsocketClient.cs b/Assets/Plugins/StreamChat/Libs/Websockets/WebsocketClient.cs index 0b30ab8d..e02bcba1 100644 --- a/Assets/Plugins/StreamChat/Libs/Websockets/WebsocketClient.cs +++ b/Assets/Plugins/StreamChat/Libs/Websockets/WebsocketClient.cs @@ -38,6 +38,9 @@ public WebsocketClient(ILogs logs, Encoding encoding = default, bool isDebugMode public bool TryDequeueMessage(out string message) => _receiveQueue.TryDequeue(out message); + /// + public int QueuedMessageCount => _receiveQueue.Count; + public async Task ConnectAsync(Uri serverUri, int timeout = 3) { if (IsConnected || IsConnecting) From d22c9a7c22697595628046827be49ad11dff5da8 Mon Sep 17 00:00:00 2001 From: Harlan Crystal Date: Sat, 25 Jul 2026 21:23:33 -0700 Subject: [PATCH 2/4] Cap websocket messages handled per Update MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit StreamChatLowLevelClient.Update drained the receive queue to empty every frame: while (_websocketClient.TryDequeueMessage(out var msg)) HandleNewWebsocketMessage(msg); Messages arrive on a background timer thread but are only handled here, on Unity's main loop, which the OS stops while the app is backgrounded. So the queue fills with nothing draining it, and the entire backlog is then handled in the FIRST frame after resume. Per-message cost in a real consumer is not small — in our app each message drives a persistence write, profile resolution, and a feed reload — so that frame locks the app up for seconds. This was a reported bug: iOS players backgrounding during chat traffic and returning to a frozen app. Now at most MaxMessagesHandledPerUpdate (20) are handled per Update and the rest stay queued for following frames. Nothing is dropped or reordered; one catastrophic frame becomes ~50 briefly degraded ones. The cap does not bind in steady state. WebsocketClient's reader enqueues at most one message per 50ms tick (UpdatesPerSecond = 20), while the drain runs at 20/frame = 600/s at 30fps — 30x headroom, so the loop still exits on an empty queue exactly as before. It only meters catch-up after a stall. WebGL's NativeWebSocketWrapper is browser-paced with no such ceiling, so there the cap does spread a burst over frames, which is the intent anyway. Happy to make the constant configurable via IStreamClientConfig if you would prefer that over a hardcoded value. --- .../LowLevelClient/StreamChatLowLevelClient.cs | 17 ++++++++++++++++- 1 file changed, 16 insertions(+), 1 deletion(-) diff --git a/Assets/Plugins/StreamChat/Core/LowLevelClient/StreamChatLowLevelClient.cs b/Assets/Plugins/StreamChat/Core/LowLevelClient/StreamChatLowLevelClient.cs index b45185e0..3b217d3c 100644 --- a/Assets/Plugins/StreamChat/Core/LowLevelClient/StreamChatLowLevelClient.cs +++ b/Assets/Plugins/StreamChat/Core/LowLevelClient/StreamChatLowLevelClient.cs @@ -415,12 +415,19 @@ public void Update(float deltaTime) _websocketClient.Update(); - while (_websocketClient.TryDequeueMessage(out var msg)) + // Bound the per-frame drain. The websocket receives on a background + // timer thread (see WebsocketClient) while this pump only runs on Unity's main + // loop, which stops while the app is backgrounded. Draining the whole backlog in + // one frame after a long background stalls that frame; spreading it over frames + // keeps the catch-up interactive. + int drained = 0; + while (drained < MaxMessagesHandledPerUpdate && _websocketClient.TryDequeueMessage(out string msg)) { #if STREAM_DEBUG_ENABLED _logs.Info(_authCredentials.UserId + " WS message: " + msg); #endif HandleNewWebsocketMessage(msg); + drained++; } } @@ -557,6 +564,14 @@ internal async Task ConnectUserAsync(string apiKey, string u private const string DefaultStreamAuthType = "jwt"; private const int HealthCheckMaxWaitingTime = 30; + // Max websocket messages handled per Update (i.e. per frame). Under + // WebsocketClient (every platform but WebGL) the reader enqueues at most one message per + // UpdatesPerSecond tick — 20/second — so at 30-60fps this drains 30-60x faster than + // messages can arrive: the cap never binds in steady state and only meters catch-up after + // the main loop was stalled. WebGL's NativeWebSocketWrapper is browser-paced with no such + // ceiling, so there the cap does spread a burst over frames, which is the intent anyway. + private const int MaxMessagesHandledPerUpdate = 20; + // For WebGL there is a slight delay when sending therefore we send HC event a bit sooner just in case private const int HealthCheckSendInterval = HealthCheckMaxWaitingTime - 1; From f4183b84d8a2033d6d878a4162e58c06968b3808 Mon Sep 17 00:00:00 2001 From: Harlan Crystal Date: Sat, 25 Jul 2026 21:27:01 -0700 Subject: [PATCH 3/4] Pace the reconnect missed-event replay through the per-frame budget MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit On every reconnect, StreamChatClient.OnConnected calls RestoreStateLostDuringDisconnect -> FetchAndProcessEventsSinceLastReceivedEvent, which fetches the events missed while disconnected and handled ALL of them in a single foreach, inline: foreach (var e in response.Events) { ...; HandleNewWebsocketMessage(serializedMsg); } That bypassed the per-frame drain cap from the previous commit entirely, so the cap only covered live socket messages and left the bigger offender untouched. This replay is at its largest precisely when an app returns from the background — the socket died while suspended, so the resume reconnect asks for everything missed, up to the ~1000-event /sync limit — and every event costs a deserialize plus full downstream handling. On iOS, where the process is suspended and no live backlog can accumulate, this replay IS the resume freeze. Now the events are queued into _pendingReplayEvents and drained by Update under the same MaxMessagesHandledPerUpdate budget as socket messages. Replay drains before live messages, so ordering is preserved within a frame (replayed events are older). Nothing is dropped. The queue is a plain Queue, matching the main-thread assumption the direct HandleNewWebsocketMessage call already made here. The queue is also cleared when the connection state goes to Disconnected. Those events are stale, and keeping them is self-compounding: _lastEventReceivedAt only advances as events are HANDLED, so the sync point captured on disconnect excludes everything still queued — the next /sync re-fetches exactly those events and enqueues them BEHIND the undrained remainder. Each reconnect flap would add another duplicate copy to an unbounded queue, and FIFO order would replay older events after newer ones had already been handled, dragging _lastEventReceivedAt backwards mid-drain. Clearing is safe precisely because the sync point does not count them: whatever is discarded is re-fetched on the next catch-up. --- .../StreamChatLowLevelClient.cs | 33 +++++++++++++++++-- 1 file changed, 31 insertions(+), 2 deletions(-) diff --git a/Assets/Plugins/StreamChat/Core/LowLevelClient/StreamChatLowLevelClient.cs b/Assets/Plugins/StreamChat/Core/LowLevelClient/StreamChatLowLevelClient.cs index 3b217d3c..abaf35b2 100644 --- a/Assets/Plugins/StreamChat/Core/LowLevelClient/StreamChatLowLevelClient.cs +++ b/Assets/Plugins/StreamChat/Core/LowLevelClient/StreamChatLowLevelClient.cs @@ -217,6 +217,17 @@ private set if (value == ConnectionState.Disconnected) { _disconnectionLastEventReceivedAt = _lastEventReceivedAt; + // Drop any replay events this connection never got through. They are + // stale and, worse, self-compounding if kept: _lastEventReceivedAt only advances + // as events are HANDLED, so the sync point captured on the line above excludes + // everything still queued here — the next /sync re-fetches exactly those events + // and enqueues them BEHIND the undrained remainder. Each reconnect flap would + // then add another duplicate copy to an unbounded queue, and the FIFO order would + // replay older events after newer ones had already been handled, dragging + // _lastEventReceivedAt backwards mid-drain so a drop in that window regressed the + // sync point again. Clearing is safe precisely because the sync point does not + // count them: whatever is discarded here is re-fetched on the next catch-up. + _pendingReplayEvents.Clear(); Disconnected?.Invoke(); } } @@ -421,6 +432,15 @@ public void Update(float deltaTime) // one frame after a long background stalls that frame; spreading it over frames // keeps the catch-up interactive. int drained = 0; + + // Reconnect-replay events are older than anything still on the socket, so they go + // first to keep events in order. They share the per-frame budget below. + while (drained < MaxMessagesHandledPerUpdate && _pendingReplayEvents.Count > 0) + { + HandleNewWebsocketMessage(_pendingReplayEvents.Dequeue()); + drained++; + } + while (drained < MaxMessagesHandledPerUpdate && _websocketClient.TryDequeueMessage(out string msg)) { #if STREAM_DEBUG_ENABLED @@ -480,8 +500,13 @@ public async Task FetchAndProcessEventsSinceLastReceivedEvent(IEnumerable ConnectUserAsync(string apiKey, string u private readonly Dictionary> _eventKeyToHandler = new Dictionary>(); + // Missed events fetched on reconnect, awaiting the per-frame handling budget + // in Update. Main-thread only, like the direct handling it replaced. + private readonly Queue _pendingReplayEvents = new Queue(); + private readonly object _websocketConnectionFailedFlagLock = new object(); private TaskCompletionSource _connectUserTaskSource; From 819a7d69db09816eb93c1339fd787c4fc01bbfa0 Mon Sep 17 00:00:00 2001 From: Harlan Crystal Date: Sat, 25 Jul 2026 21:30:15 -0700 Subject: [PATCH 4/4] Warn once when the receive backlog shows a stalled pump MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Logs once, when more than ReceiveBacklogWarningThreshold (500) received messages are waiting to be handled — i.e. the main loop stopped pumping long enough to fall far behind. Re-arms only after the backlog clears, so it cannot spam per frame. Counts the reconnect replay queue as well as the socket's. Both are drained by the same per-frame budget, so both contribute to how far behind we are, and the replay queue is the one that actually gets large: a /sync catch-up can hand back hundreds of events at once, whereas the socket queue is bounded by its ~20/s arrival rate against a 20/frame drain. Diagnostic ONLY. It deliberately does not discard the backlog, and neither should a consumer: reconnecting does not re-hydrate channel state (the sole catch-up is /sync, itself capped at roughly 1000 missed events), so discarding would drop those events permanently and leave local channel state silently stale — strictly worse than a slow drain. The backlog is bounded in practice anyway: health-check pings only go out from this Update, so a stalled client stops pinging and the server closes the socket within ~30s. Happy to make the threshold configurable via IStreamClientConfig if you would prefer that over a hardcoded value. --- Assets/Plugins/StreamChat/Changelog.txt | 2 + .../StreamChatLowLevelClient.cs | 39 +++++++++++++++++++ 2 files changed, 41 insertions(+) diff --git a/Assets/Plugins/StreamChat/Changelog.txt b/Assets/Plugins/StreamChat/Changelog.txt index 335bd26b..d16a3159 100644 --- a/Assets/Plugins/StreamChat/Changelog.txt +++ b/Assets/Plugins/StreamChat/Changelog.txt @@ -14,11 +14,13 @@ Fixes: Unreleased: Features: +* Add IWebsocketClient.QueuedMessageCount, the number of received messages waiting to be dequeued. Receiving runs on a background timer thread while the SDK dequeues from Unity's main loop, so this reports how far behind the main loop is. Diagnostic only - the transport never drops messages. Note: if you supply your own IWebsocketClient implementation you will need to add this member. * Add a public StreamApiException constructor (statusCode, code, errorMessage, moreInfo, duration, exceptionFields). StreamApiException is a public, catch-and-branch type (via the StreamApiExceptionExtensions.Is* helpers), but until now it could only be constructed inside the SDK from the internal APIErrorInternalDTO, so integrators could not build one to unit-test their own error handling (e.g. simulating a 403 / code 70 "no access to channels" response). The new constructor maps directly to the type's public properties and keeps APIErrorInternalDTO internal. * Add IStreamClientConfig.OptimisticMessageInsert (default true). When true (the existing behavior), a message you send is inserted into the local channel state and raised via IStreamChannel.MessageReceived immediately, before the server's message.new echo arrives. Set it to false to skip the optimistic local insert and wait for the server echo instead, so every participant - including the sender - observes messages in the same server-defined order. Useful when consistent cross-client ordering matters more than instant local feedback (e.g. a shared, broadcast-ordered feed). Fixes: +* Fix a multi-second frame stall when an app resumes from the background during chat traffic. Received websocket messages are handled only from Unity's main loop, which the OS stops while the app is backgrounded, so the whole accumulated backlog was handled in the first frame after resume - as was the entire /sync missed-event replay the reconnect fetches, which was handled inline in one loop. Both are now drained at up to 20 messages per Update, sharing one per-frame budget, with replayed (older) events going first so ordering is preserved. Nothing is dropped or reordered, and the cap does not bind in steady state. A one-off warning is logged if the combined backlog exceeds 500, which indicates the main loop stopped pumping. * TaskUtils.LogIfFailed now logs connectivity/transport failures as warnings instead of errors/exceptions. The SDK fire-and-forgets its connect, reconnect, and state-restore operations through LogIfFailed; when the device is offline these fail with HttpRequestException / WebException / SocketException / IOException / TimeoutException, which the reconnect flow recovers from - so surfacing them at error severity flooded crash/error reporting (Sentry, Bugsnag, etc.) with handled, non-actionable noise. Genuine (non-connectivity) failures still log as exceptions. Complements the connection-attempt-timeout fix from PR #213. v5.5.0: diff --git a/Assets/Plugins/StreamChat/Core/LowLevelClient/StreamChatLowLevelClient.cs b/Assets/Plugins/StreamChat/Core/LowLevelClient/StreamChatLowLevelClient.cs index abaf35b2..51472a69 100644 --- a/Assets/Plugins/StreamChat/Core/LowLevelClient/StreamChatLowLevelClient.cs +++ b/Assets/Plugins/StreamChat/Core/LowLevelClient/StreamChatLowLevelClient.cs @@ -426,6 +426,36 @@ public void Update(float deltaTime) _websocketClient.Update(); + // Report-only. We deliberately do NOT discard a large backlog: there is no + // recovery path that would make that safe. Reconnecting does not re-hydrate channel + // state — the only catch-up is /sync (see FetchAndProcessEventsSinceLastReceivedEvent), + // which the server rejects past roughly 1000 missed events, so discarding would drop + // the events permanently and leave local state silently stale. Draining a large + // backlog under the per-frame budget below is slow but correct. The backlog is bounded + // in practice anyway: health-check pings only go out from this Update, so a stalled + // client stops pinging and the server closes the socket within ~30s. + // Counts the reconnect replay queue as well as the socket's. Both are drained by the same + // per-frame budget below, so both contribute to how far behind we are — and the replay + // queue is the one that actually gets large, since a /sync catch-up can hand back + // hundreds of events at once whereas the socket queue is bounded by its ~20/s arrival + // rate against a 20/frame drain. Measuring only the socket meant the backlog this warning + // exists to surface was the one it could not see. + int backlog = _websocketClient.QueuedMessageCount + _pendingReplayEvents.Count; + if (backlog > ReceiveBacklogWarningThreshold) + { + if (!_receiveBacklogWarned) + { + _receiveBacklogWarned = true; + _logs.Warning( + $"Websocket receive backlog of {backlog} exceeds {ReceiveBacklogWarningThreshold}; " + + "draining it over multiple frames. Expect degraded frame time until it clears."); + } + } + else + { + _receiveBacklogWarned = false; + } + // Bound the per-frame drain. The websocket receives on a background // timer thread (see WebsocketClient) while this pump only runs on Unity's main // loop, which stops while the app is backgrounded. Draining the whole backlog in @@ -597,6 +627,12 @@ internal async Task ConnectUserAsync(string apiKey, string u // ceiling, so there the cap does spread a burst over frames, which is the intent anyway. private const int MaxMessagesHandledPerUpdate = 20; + // Backlog depth that gets a one-off warning so a stalled pump is visible in the + // field. Diagnostic only — nothing is discarded. Sized above any ordinary hitch: at the + // socket's 20 messages/second ceiling this is ~25 seconds of saturated traffic with the + // main loop stopped. + private const int ReceiveBacklogWarningThreshold = 500; + // For WebGL there is a slight delay when sending therefore we send HC event a bit sooner just in case private const int HealthCheckSendInterval = HealthCheckMaxWaitingTime - 1; @@ -619,6 +655,9 @@ internal async Task ConnectUserAsync(string apiKey, string u // in Update. Main-thread only, like the direct handling it replaced. private readonly Queue _pendingReplayEvents = new Queue(); + // Latches the backlog warning so it logs once per episode, not every frame. + private bool _receiveBacklogWarned; + private readonly object _websocketConnectionFailedFlagLock = new object(); private TaskCompletionSource _connectUserTaskSource;