From ae6fc8c5a057d6c4d652ad9b33ccbaf6e5e8e10e Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Daniel=20Sierpi=C5=84ski?= <33436839+sierpinskid@users.noreply.github.com> Date: Tue, 25 Aug 2026 23:19:33 +0200 Subject: [PATCH 1/3] Add time budget for draining WS events. If needed, the load will be split across multiple frames --- .../LowLevelClient/EventDrainStopwatch.cs | 24 ++ .../EventDrainStopwatch.cs.meta | 3 + .../IStreamChatLowLevelClient.cs | 5 + .../StreamChatLowLevelClient.cs | 208 +++++++++++- ...StreamChatLowLevelClientEventDrainTests.cs | 306 ++++++++++++++++++ ...mChatLowLevelClientEventDrainTests.cs.meta | 3 + .../Tests/StateSync/StateSyncCatchUpTests.cs | 21 +- 7 files changed, 551 insertions(+), 19 deletions(-) create mode 100644 Assets/Plugins/StreamChat/Core/LowLevelClient/EventDrainStopwatch.cs create mode 100644 Assets/Plugins/StreamChat/Core/LowLevelClient/EventDrainStopwatch.cs.meta create mode 100644 Assets/Plugins/StreamChat/Tests/LowLevelClient/StreamChatLowLevelClientEventDrainTests.cs create mode 100644 Assets/Plugins/StreamChat/Tests/LowLevelClient/StreamChatLowLevelClientEventDrainTests.cs.meta diff --git a/Assets/Plugins/StreamChat/Core/LowLevelClient/EventDrainStopwatch.cs b/Assets/Plugins/StreamChat/Core/LowLevelClient/EventDrainStopwatch.cs new file mode 100644 index 00000000..062cf971 --- /dev/null +++ b/Assets/Plugins/StreamChat/Core/LowLevelClient/EventDrainStopwatch.cs @@ -0,0 +1,24 @@ +using System.Diagnostics; + +namespace StreamChat.Core.LowLevelClient +{ + /// + /// Elapsed-time source for the per-frame event drain budget. Production uses + /// ; tests inject a fake so pacing is deterministic. + /// + internal interface IElapsedStopwatch + { + void Restart(); + + double ElapsedMilliseconds { get; } + } + + internal sealed class DiagnosticsElapsedStopwatch : IElapsedStopwatch + { + public void Restart() => _stopwatch.Restart(); + + public double ElapsedMilliseconds => _stopwatch.Elapsed.TotalMilliseconds; + + private readonly Stopwatch _stopwatch = new Stopwatch(); + } +} diff --git a/Assets/Plugins/StreamChat/Core/LowLevelClient/EventDrainStopwatch.cs.meta b/Assets/Plugins/StreamChat/Core/LowLevelClient/EventDrainStopwatch.cs.meta new file mode 100644 index 00000000..8b7e1da9 --- /dev/null +++ b/Assets/Plugins/StreamChat/Core/LowLevelClient/EventDrainStopwatch.cs.meta @@ -0,0 +1,3 @@ +fileFormatVersion: 2 +guid: e4d8a91c7b2f4e6a9d1c3b5f708294a6 +timeCreated: 1756152000 diff --git a/Assets/Plugins/StreamChat/Core/LowLevelClient/IStreamChatLowLevelClient.cs b/Assets/Plugins/StreamChat/Core/LowLevelClient/IStreamChatLowLevelClient.cs index a9709bd4..9d1c035e 100644 --- a/Assets/Plugins/StreamChat/Core/LowLevelClient/IStreamChatLowLevelClient.cs +++ b/Assets/Plugins/StreamChat/Core/LowLevelClient/IStreamChatLowLevelClient.cs @@ -103,6 +103,11 @@ void SetReconnectStrategySettings(ReconnectStrategy reconnectStrategy, float? ex [Obsolete("Use DisconnectAsync(DisconnectCause). true maps to UserLogout, false to ConnectionReleased.")] Task DisconnectAsync(bool permanent); + /// + /// Fetch missed events via /sync and apply them. The returned task completes when + /// those events have been processed, which may span several calls + /// after a large catch-up. Keep calling while awaiting. + /// Task FetchAndProcessEventsSinceLastReceivedEvent(IEnumerable channelCids); /// diff --git a/Assets/Plugins/StreamChat/Core/LowLevelClient/StreamChatLowLevelClient.cs b/Assets/Plugins/StreamChat/Core/LowLevelClient/StreamChatLowLevelClient.cs index 8d12e583..db00148e 100644 --- a/Assets/Plugins/StreamChat/Core/LowLevelClient/StreamChatLowLevelClient.cs +++ b/Assets/Plugins/StreamChat/Core/LowLevelClient/StreamChatLowLevelClient.cs @@ -218,6 +218,8 @@ private set if (value == ConnectionState.Disconnected) { _disconnectionLastEventReceivedAt = _lastEventReceivedAt; + ClearPendingHistoryEvents(); + _heldLiveMessage = null; RaiseDisconnected(); } } @@ -430,13 +432,7 @@ public void Update(float deltaTime) _websocketClient.Update(); - while (_websocketClient.TryDequeueMessage(out var msg)) - { -#if STREAM_DEBUG_ENABLED - _logs.Info(_authCredentials.UserId + " WS message: " + msg); -#endif - HandleNewWebsocketMessage(msg); - } + DrainPendingEvents(); } public bool IsLocalUser(User user) => user.Id == _authCredentials.UserId; @@ -469,8 +465,9 @@ public async Task FetchAndProcessEventsSinceLastReceivedEvent(IEnumerable 1000 events + //StreamTodo: according to Android SDK there's an error if there are > 1000 events + var generation = _historyDrainGeneration; var response = await ChannelApi.SyncAsync(new SyncRequest { ChannelCids = channelCids.ToList(), @@ -478,24 +475,24 @@ public async Task FetchAndProcessEventsSinceLastReceivedEvent(IEnumerable + /// Per-frame time budget for applying live WS + /sync history events. Tests may lower this. + /// + internal int EventDrainTimeBudgetMs { get; set; } = 3; + + /// + /// Secondary bound so a batch of cheap events cannot still dump a huge list in one frame. + /// + internal int EventDrainCountCap { get; set; } = 64; + + internal IElapsedStopwatch EventDrainStopwatch { get; set; } = new DiagnosticsElapsedStopwatch(); + internal async Task ConnectUserAsync(string apiKey, string userId, ITokenProvider tokenProvider, CancellationToken cancellationToken = default) { @@ -574,6 +583,7 @@ internal async Task ConnectUserAsync(string apiKey, string u private const string DefaultStreamAuthType = "jwt"; private const int HealthCheckMaxWaitingTime = 30; + private const int EventDrainHealthGrace = 8; // 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; @@ -595,6 +605,8 @@ internal async Task ConnectUserAsync(string apiKey, string u private readonly object _websocketConnectionFailedFlagLock = new object(); private readonly object _websocketDisconnectedFlagLock = new object(); + private readonly object _pendingHistoryEventsLock = new object(); + private readonly Queue _pendingHistoryEvents = new Queue(); /// /// Every write must happen on this thread @@ -628,6 +640,10 @@ internal async Task ConnectUserAsync(string apiKey, string u /// private DateTimeOffset? _disconnectionLastEventReceivedAt; + private string _heldLiveMessage; + private TaskCompletionSource _historyDrainTcs; + private int _historyDrainGeneration; + private async Task RefreshAuthTokenFromProvider() { #if STREAM_DEBUG_ENABLED @@ -1038,6 +1054,168 @@ private TEvent DeserializeEvent(string content, out TDto dto) return response; } + private void DrainPendingEvents() + { + EventDrainStopwatch.Restart(); + var processed = 0; + var countCap = EventDrainCountCap; + var timeBudgetMs = EventDrainTimeBudgetMs; + + bool BudgetExhausted() + => processed > 0 && (processed >= countCap || + EventDrainStopwatch.ElapsedMilliseconds >= timeBudgetMs); + + while (true) + { + if (HasPendingHistoryEvents()) + { + if (BudgetExhausted()) + { + break; + } + + if (!TryDequeueHistoryEvent(out var historyEvent)) + { + continue; + } + + ProcessHistoryEvent(historyEvent); + processed++; + TryCompleteHistoryDrainIfIdle(); + continue; + } + + if (!TryTakeLiveMessage(out var liveMsg)) + { + break; + } + +#if STREAM_DEBUG_ENABLED + _logs.Info(_authCredentials.UserId + " WS message: " + liveMsg); +#endif + + var isHealth = IsHealthCheckMessage(liveMsg); + var hardStop = processed >= countCap + EventDrainHealthGrace; + if (hardStop || (BudgetExhausted() && !isHealth)) + { + _heldLiveMessage = liveMsg; + break; + } + + HandleNewWebsocketMessage(liveMsg); + processed++; + } + + TryCompleteHistoryDrainIfIdle(); + } + + private bool TryTakeLiveMessage(out string msg) + { + if (_heldLiveMessage != null) + { + msg = _heldLiveMessage; + _heldLiveMessage = null; + return true; + } + + return _websocketClient.TryDequeueMessage(out msg); + } + + private bool IsHealthCheckMessage(string msg) + { + if (string.IsNullOrEmpty(msg)) + { + return false; + } + + try + { + return _serializer.TryPeekValue(msg, "type", out var type) + && type == WSEventType.HealthCheck; + } + catch (Exception) + { + return false; + } + } + + private Task EnqueueHistoryEvents(IEnumerable events) + { + lock (_pendingHistoryEventsLock) + { + foreach (var e in events) + { + _pendingHistoryEvents.Enqueue(e); + } + + if (_historyDrainTcs == null || _historyDrainTcs.Task.IsCompleted) + { + _historyDrainTcs = new TaskCompletionSource( + TaskCreationOptions.RunContinuationsAsynchronously); + } + + return _historyDrainTcs.Task; + } + } + + private bool HasPendingHistoryEvents() + { + lock (_pendingHistoryEventsLock) + { + return _pendingHistoryEvents.Count > 0; + } + } + + private bool TryDequeueHistoryEvent(out object historyEvent) + { + lock (_pendingHistoryEventsLock) + { + if (_pendingHistoryEvents.Count == 0) + { + historyEvent = null; + return false; + } + + historyEvent = _pendingHistoryEvents.Dequeue(); + return true; + } + } + + private void ProcessHistoryEvent(object historyEvent) + { + if (historyEvent == null) + { + return; + } + + var serializedMsg = historyEvent as string ?? _serializer.Serialize(historyEvent); + HandleNewWebsocketMessage(serializedMsg); + } + + private void TryCompleteHistoryDrainIfIdle() + { + lock (_pendingHistoryEventsLock) + { + if (_pendingHistoryEvents.Count > 0 || _historyDrainTcs == null) + { + return; + } + + _historyDrainTcs.TrySetResult(true); + } + } + + private void ClearPendingHistoryEvents() + { + lock (_pendingHistoryEventsLock) + { + _historyDrainGeneration++; + _pendingHistoryEvents.Clear(); + _historyDrainTcs?.TrySetResult(true); + _historyDrainTcs = null; + } + } + private void HandleNewWebsocketMessage(string msg) { const string ErrorKey = "error"; diff --git a/Assets/Plugins/StreamChat/Tests/LowLevelClient/StreamChatLowLevelClientEventDrainTests.cs b/Assets/Plugins/StreamChat/Tests/LowLevelClient/StreamChatLowLevelClientEventDrainTests.cs new file mode 100644 index 00000000..f148dc70 --- /dev/null +++ b/Assets/Plugins/StreamChat/Tests/LowLevelClient/StreamChatLowLevelClientEventDrainTests.cs @@ -0,0 +1,306 @@ +#if STREAM_TESTS_ENABLED +using System; +using System.Collections.Generic; +using System.Net.WebSockets; +using System.Threading.Tasks; +using NSubstitute; +using NUnit.Framework; +using StreamChat.Core; +using StreamChat.Core.Configs; +using StreamChat.Core.LowLevelClient; +using StreamChat.Core.LowLevelClient.Events; +using StreamChat.Libs.AppInfo; +using StreamChat.Libs.Auth; +using StreamChat.Libs.Http; +using StreamChat.Libs.Logs; +using StreamChat.Libs.NetworkMonitors; +using StreamChat.Libs.Serialization; +using StreamChat.Libs.Time; +using StreamChat.Libs.Websockets; + +namespace StreamChat.Tests.LowLevelClient +{ + internal class StreamChatLowLevelClientEventDrainTests + { + [SetUp] + public void Up() + { + _authCredentials = new AuthCredentials("api123", "token123", "user123"); + _mockWebsocketClient = Substitute.For(); + _mockHttpClient = Substitute.For(); + _mockTimeService = Substitute.For(); + _mockNetworkMonitor = Substitute.For(); + _mockApplicationInfo = Substitute.For(); + _mockLogs = new UnityLogs(); + _mockStreamClientConfig = Substitute.For(); + + _mockWebsocketClient.ConnectAsync(Arg.Any()).Returns(Task.CompletedTask); + _mockHttpClient + .SendHttpRequestAsync(Arg.Is(HttpMethodType.Post), Arg.Any(), Arg.Any()) + .Returns(new HttpResponse(true, 200, "{\"events\":[]}", null, null)); + } + + [TearDown] + public void TearDown() + { + for (int i = _resourcesToDispose.Count - 1; i >= 0; i--) + { + _resourcesToDispose[i].Dispose(); + } + + _resourcesToDispose.Clear(); + } + + [Test] + public void when_live_backlog_exceeds_count_cap_expect_one_update_handles_only_the_cap() + { + var client = CreateConnectedClient(); + client.EventDrainCountCap = 2; + client.EventDrainTimeBudgetMs = 10_000; + + var received = new List(); + client.MessageReceived += e => received.Add(e.Message.Id); + + EnqueueLiveMessages(Message("m1"), Message("m2"), Message("m3"), Message("m4"), Message("m5")); + client.Update(0.1f); + + Assert.AreEqual(new[] { "m1", "m2" }, received.ToArray()); + + client.Update(0.1f); + Assert.AreEqual(new[] { "m1", "m2", "m3", "m4" }, received.ToArray()); + + client.Update(0.1f); + Assert.AreEqual(new[] { "m1", "m2", "m3", "m4", "m5" }, received.ToArray()); + } + + [Test] + public void when_one_live_message_per_update_expect_it_drains_fully() + { + var client = CreateConnectedClient(); + client.EventDrainCountCap = 2; + + var received = 0; + client.MessageReceived += _ => received++; + + EnqueueLiveMessages(Message("m1")); + client.Update(0.1f); + + Assert.AreEqual(1, received); + + client.Update(0.1f); + Assert.AreEqual(1, received); + } + + [Test] + public void when_budget_spent_expect_live_health_check_still_handled() + { + var client = CreateConnectedClient(); + client.EventDrainCountCap = 1; + client.EventDrainTimeBudgetMs = 10_000; + + var messages = new List(); + var healthChecks = 0; + client.MessageReceived += e => messages.Add(e.Message.Id); + client.EventReceived += payload => + { + if (payload.Contains("health.check")) + { + healthChecks++; + } + }; + + EnqueueLiveMessages(Message("m1"), HealthCheck(), Message("m2")); + client.Update(0.1f); + + Assert.AreEqual(new[] { "m1" }, messages.ToArray()); + Assert.GreaterOrEqual(healthChecks, 1); + + client.Update(0.1f); + Assert.AreEqual(new[] { "m1", "m2" }, messages.ToArray()); + } + + [Test] + public void when_time_budget_exhausted_expect_remaining_live_events_deferred() + { + var client = CreateConnectedClient(); + var clock = new ManualElapsedStopwatch(); + client.EventDrainStopwatch = clock; + client.EventDrainTimeBudgetMs = 3; + client.EventDrainCountCap = 1000; + + var received = new List(); + client.MessageReceived += e => + { + received.Add(e.Message.Id); + clock.ElapsedMilliseconds = 10; + }; + + EnqueueLiveMessages(Message("m1"), Message("m2"), Message("m3")); + client.Update(0.1f); + + Assert.AreEqual(new[] { "m1" }, received.ToArray()); + + clock.ElapsedMilliseconds = 0; + client.Update(0.1f); + Assert.AreEqual(new[] { "m1", "m2" }, received.ToArray()); + } + + [Test] + public void when_history_and_live_queued_expect_history_processed_first() + { + var client = CreateConnectedClient(); + client.EventDrainCountCap = 10; + client.EventDrainTimeBudgetMs = 10_000; + + var received = new List(); + client.MessageReceived += e => received.Add(e.Message.Id); + + StubSyncEvents(MessageEventJson("h1"), MessageEventJson("h2")); + SetDisconnectionWatermark(client, new DateTimeOffset(2026, 8, 10, 11, 0, 0, TimeSpan.Zero)); + _mockTimeService.Now.Returns(new DateTimeOffset(2026, 8, 10, 12, 0, 0, TimeSpan.Zero)); + + EnqueueLiveMessages(Message("l1"), Message("l2")); + + var fetch = client.FetchAndProcessEventsSinceLastReceivedEvent(new[] { "messaging:test" }); + client.Update(0.1f); + + Assert.AreEqual(new[] { "h1", "h2", "l1", "l2" }, received.ToArray()); + Assert.IsTrue(fetch.IsCompleted); + fetch.GetAwaiter().GetResult(); + } + + [Test] + public void when_sync_returns_many_events_expect_they_span_updates_and_task_waits() + { + var client = CreateConnectedClient(); + client.EventDrainCountCap = 2; + client.EventDrainTimeBudgetMs = 10_000; + + var received = new List(); + client.MessageReceived += e => received.Add(e.Message.Id); + + StubSyncEvents(MessageEventJson("h1"), MessageEventJson("h2"), MessageEventJson("h3"), + MessageEventJson("h4"), MessageEventJson("h5")); + SetDisconnectionWatermark(client, new DateTimeOffset(2026, 8, 10, 11, 0, 0, TimeSpan.Zero)); + _mockTimeService.Now.Returns(new DateTimeOffset(2026, 8, 10, 12, 0, 0, TimeSpan.Zero)); + + var fetch = client.FetchAndProcessEventsSinceLastReceivedEvent(new[] { "messaging:test" }); + Assert.IsFalse(fetch.IsCompleted); + + client.Update(0.1f); + Assert.AreEqual(new[] { "h1", "h2" }, received.ToArray()); + Assert.IsFalse(fetch.IsCompleted); + + client.Update(0.1f); + Assert.AreEqual(new[] { "h1", "h2", "h3", "h4" }, received.ToArray()); + + client.Update(0.1f); + Assert.AreEqual(new[] { "h1", "h2", "h3", "h4", "h5" }, received.ToArray()); + Assert.IsTrue(fetch.IsCompleted); + fetch.GetAwaiter().GetResult(); + } + + [Test] + public void when_disconnect_mid_history_drain_expect_pending_cleared_and_no_double_apply() + { + var client = CreateConnectedClient(); + client.EventDrainCountCap = 1; + client.EventDrainTimeBudgetMs = 10_000; + SetupDisconnectRaisesDisconnected(); + + var received = new List(); + client.MessageReceived += e => received.Add(e.Message.Id); + + StubSyncEvents(MessageEventJson("h1"), MessageEventJson("h2"), MessageEventJson("h3")); + SetDisconnectionWatermark(client, new DateTimeOffset(2026, 8, 10, 11, 0, 0, TimeSpan.Zero)); + _mockTimeService.Now.Returns(new DateTimeOffset(2026, 8, 10, 12, 0, 0, TimeSpan.Zero)); + + var fetch = client.FetchAndProcessEventsSinceLastReceivedEvent(new[] { "messaging:test" }); + client.Update(0.1f); + Assert.AreEqual(new[] { "h1" }, received.ToArray()); + + client.DisconnectAsync(DisconnectCause.ConnectionReleased).GetAwaiter().GetResult(); + Assert.IsTrue(fetch.IsCompleted); + + client.Update(0.1f); + Assert.AreEqual(new[] { "h1" }, received.ToArray()); + } + + private readonly List _resourcesToDispose = new List(); + + private AuthCredentials _authCredentials; + private IWebsocketClient _mockWebsocketClient; + private IApplicationInfo _mockApplicationInfo; + private ILogs _mockLogs; + private ITimeService _mockTimeService; + private INetworkMonitor _mockNetworkMonitor; + private IHttpClient _mockHttpClient; + private IStreamClientConfig _mockStreamClientConfig; + + private StreamChatLowLevelClient CreateConnectedClient() + { + var client = new StreamChatLowLevelClient(_authCredentials, _mockWebsocketClient, _mockHttpClient, + new NewtonsoftJsonSerializer(), _mockTimeService, _mockNetworkMonitor, _mockApplicationInfo, + _mockLogs, _mockStreamClientConfig); + _resourcesToDispose.Add(client); + + EnqueueLiveMessages(HealthCheck()); + client.Connect(); + client.Update(0.2f); + Assert.AreEqual(ConnectionState.Connected, client.ConnectionState); + return client; + } + + private void EnqueueLiveMessages(params string[] messages) + { + var queue = new Queue(messages); + _mockWebsocketClient.TryDequeueMessage(out Arg.Any()).Returns(arg => + { + if (queue.Count == 0) + { + return false; + } + + arg[0] = queue.Dequeue(); + return true; + }); + } + + private void StubSyncEvents(params string[] eventJsonObjects) + { + var body = "{\"events\":[" + string.Join(",", eventJsonObjects) + "]}"; + _mockHttpClient + .SendHttpRequestAsync(Arg.Is(HttpMethodType.Post), Arg.Any(), Arg.Any()) + .Returns(new HttpResponse(true, 200, body, null, null)); + } + + private void SetupDisconnectRaisesDisconnected() + { + _mockWebsocketClient.When(_ => _.DisconnectAsync(Arg.Any(), Arg.Any())) + .Do(_ => { _mockWebsocketClient.Disconnected += Raise.Event(); }); + } + + private static void SetDisconnectionWatermark(StreamChatLowLevelClient client, DateTimeOffset value) + { + var field = typeof(StreamChatLowLevelClient).GetField("_disconnectionLastEventReceivedAt", + System.Reflection.BindingFlags.Instance | System.Reflection.BindingFlags.NonPublic); + Assert.IsNotNull(field); + field.SetValue(client, (DateTimeOffset?)value); + } + + private static string Message(string id) + => $"{{\"type\":\"message.new\",\"cid\":\"messaging:test\",\"message\":{{\"id\":\"{id}\",\"text\":\"{id}\",\"type\":\"regular\"}}}}"; + + private static string MessageEventJson(string id) => Message(id); + + private static string HealthCheck() => "{\"connection_id\":\"fakeId\",\"type\":\"health.check\"}"; + + private sealed class ManualElapsedStopwatch : IElapsedStopwatch + { + public double ElapsedMilliseconds { get; set; } + + public void Restart() => ElapsedMilliseconds = 0; + } + } +} +#endif diff --git a/Assets/Plugins/StreamChat/Tests/LowLevelClient/StreamChatLowLevelClientEventDrainTests.cs.meta b/Assets/Plugins/StreamChat/Tests/LowLevelClient/StreamChatLowLevelClientEventDrainTests.cs.meta new file mode 100644 index 00000000..9871d70e --- /dev/null +++ b/Assets/Plugins/StreamChat/Tests/LowLevelClient/StreamChatLowLevelClientEventDrainTests.cs.meta @@ -0,0 +1,3 @@ +fileFormatVersion: 2 +guid: b7c2e19f4a8d4c3e9f1b6a5d2c8e0f34 +timeCreated: 1756152100 diff --git a/Assets/Plugins/StreamChat/Tests/StateSync/StateSyncCatchUpTests.cs b/Assets/Plugins/StreamChat/Tests/StateSync/StateSyncCatchUpTests.cs index 0642ccfb..a39e41ea 100644 --- a/Assets/Plugins/StreamChat/Tests/StateSync/StateSyncCatchUpTests.cs +++ b/Assets/Plugins/StreamChat/Tests/StateSync/StateSyncCatchUpTests.cs @@ -1,6 +1,7 @@ #if STREAM_TESTS_ENABLED using System; using System.Reflection; +using System.Threading.Tasks; using NSubstitute; using NUnit.Framework; using StreamChat.Core.Configs; @@ -115,8 +116,7 @@ public void when_sync_replays_events_older_than_watermark_expect_watermark_not_r SetDisconnectionLastEventReceivedAt(now.AddHours(-1)); StubSyncResponseWithMessageEvent(now.AddHours(-1)); - _lowLevelClient.FetchAndProcessEventsSinceLastReceivedEvent(new[] { TestChannelCid }).GetAwaiter() - .GetResult(); + FetchAndPumpUntilDone(); Assert.AreEqual(now, GetLastEventReceivedAt()); } @@ -132,8 +132,7 @@ public void when_sync_replays_events_newer_than_watermark_expect_watermark_advan SetDisconnectionLastEventReceivedAt(now.AddHours(-2)); StubSyncResponseWithMessageEvent(replayedEventCreatedAt); - _lowLevelClient.FetchAndProcessEventsSinceLastReceivedEvent(new[] { TestChannelCid }).GetAwaiter() - .GetResult(); + FetchAndPumpUntilDone(); Assert.AreEqual(replayedEventCreatedAt, GetLastEventReceivedAt()); } @@ -177,6 +176,20 @@ private void SetLastEventReceivedAt(DateTimeOffset value) private DateTimeOffset? GetLastEventReceivedAt() => (DateTimeOffset?)GetPrivateField("_lastEventReceivedAt").GetValue(_lowLevelClient); + private void FetchAndPumpUntilDone() + { + var task = _lowLevelClient.FetchAndProcessEventsSinceLastReceivedEvent(new[] { TestChannelCid }); + var frames = 0; + while (!task.IsCompleted && frames < 1000) + { + _lowLevelClient.Update(0.1f); + frames++; + } + + Assert.IsTrue(task.IsCompleted, "FetchAndProcessEventsSinceLastReceivedEvent did not finish after Update pumping."); + task.GetAwaiter().GetResult(); + } + private static FieldInfo GetPrivateField(string name) { var field = typeof(StreamChatLowLevelClient).GetField(name, From 9f4220bc9f56d30ee79019b54d2b51302f0ded54 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Daniel=20Sierpi=C5=84ski?= <33436839+sierpinskid@users.noreply.github.com> Date: Thu, 27 Aug 2026 23:28:18 +0200 Subject: [PATCH 2/3] Complete CTS on the same frame --- .../LowLevelClient/StreamChatLowLevelClient.cs | 15 +++++++++++---- 1 file changed, 11 insertions(+), 4 deletions(-) diff --git a/Assets/Plugins/StreamChat/Core/LowLevelClient/StreamChatLowLevelClient.cs b/Assets/Plugins/StreamChat/Core/LowLevelClient/StreamChatLowLevelClient.cs index c9d442ea..b2e98ac0 100644 --- a/Assets/Plugins/StreamChat/Core/LowLevelClient/StreamChatLowLevelClient.cs +++ b/Assets/Plugins/StreamChat/Core/LowLevelClient/StreamChatLowLevelClient.cs @@ -1290,8 +1290,7 @@ internal Task EnqueueHistoryEvents(IEnumerable events) if (_historyDrainTcs == null || _historyDrainTcs.Task.IsCompleted) { - _historyDrainTcs = new TaskCompletionSource( - TaskCreationOptions.RunContinuationsAsynchronously); + _historyDrainTcs = new TaskCompletionSource(); } return _historyDrainTcs.Task; @@ -1333,6 +1332,7 @@ private void ProcessHistoryEvent(object historyEvent) private void TryCompleteHistoryDrainIfIdle() { + TaskCompletionSource tcs; lock (_pendingHistoryEventsLock) { if (_pendingHistoryEvents.Count > 0 || _historyDrainTcs == null) @@ -1340,19 +1340,26 @@ private void TryCompleteHistoryDrainIfIdle() return; } - _historyDrainTcs.TrySetResult(true); + tcs = _historyDrainTcs; + _historyDrainTcs = null; } + + // Outside the lock so FetchAndProcess resumes on this Update, not a later scheduler tick. + tcs.TrySetResult(true); } private void ClearPendingHistoryEvents() { + TaskCompletionSource tcs; lock (_pendingHistoryEventsLock) { _historyDrainGeneration++; _pendingHistoryEvents.Clear(); - _historyDrainTcs?.TrySetResult(true); + tcs = _historyDrainTcs; _historyDrainTcs = null; } + + tcs?.TrySetResult(true); } private TDto DeserializePayload(object payload) From 115ae2d816fa66bbe2739f958b0dd0b326ab4cd3 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Daniel=20Sierpi=C5=84ski?= <33436839+sierpinskid@users.noreply.github.com> Date: Fri, 28 Aug 2026 12:34:09 +0200 Subject: [PATCH 3/3] The stateful reconnect /sync path now paces history events across frames (both ReplayEvents and BatchStateUpdate), which was the actual iOS resume freeze. While that drain runs, the health-check timeout is held off for up to 120s so catch-up cannot starve health.check into a reconnect loop. --- .../StreamChatLowLevelClient.cs | 187 ++++++++++++++++-- .../StreamChat/Core/StreamChatClient.cs | 8 +- ...StreamChatLowLevelClientEventDrainTests.cs | 136 ++++++++++++- .../StateSync/StateRecoveryClientTests.cs | 109 ++++++++++ 4 files changed, 422 insertions(+), 18 deletions(-) diff --git a/Assets/Plugins/StreamChat/Core/LowLevelClient/StreamChatLowLevelClient.cs b/Assets/Plugins/StreamChat/Core/LowLevelClient/StreamChatLowLevelClient.cs index b2e98ac0..d03883fd 100644 --- a/Assets/Plugins/StreamChat/Core/LowLevelClient/StreamChatLowLevelClient.cs +++ b/Assets/Plugins/StreamChat/Core/LowLevelClient/StreamChatLowLevelClient.cs @@ -218,7 +218,10 @@ private set { _disconnectionLastEventReceivedAt = _lastEventReceivedAt; ClearPendingHistoryEvents(); - _heldLiveMessage = null; + + // _heldLiveMessage is deliberately NOT cleared. IWebsocketClient's receive queue + // survives the disconnect, so dropping just the one message we happened to hold + // would punch a hole in an otherwise intact sequence. RaiseDisconnected(); } } @@ -461,7 +464,7 @@ public async Task FetchAndProcessEventsSinceLastReceivedEvent(IEnumerable @@ -549,6 +552,8 @@ internal HistorySyncApplyResult ApplyHistoryEvents(IEnumerable events) return result; } + // Main-thread only. Must not interleave with the paced silent-batch drain: + // both share _isApplyingHistoryEvents and _historyMaxAppliedCreatedAt. _isApplyingHistoryEvents = true; _historyMaxAppliedCreatedAt = null; @@ -705,6 +710,14 @@ internal async Task ConnectUserAsync(string apiKey, string u private const string DefaultStreamAuthType = "jwt"; private const int HealthCheckMaxWaitingTime = 30; private const int EventDrainHealthGrace = 8; + private const int ReceiveBacklogWarningThreshold = 500; + + /// + /// How long a paced /sync drain may block live health.check before the receive-side + /// timeout is allowed to fire. Shorter reintroduces a reconnect loop; longer delays + /// detecting a genuinely dead socket. + /// + internal const float EventDrainMaxStallSeconds = 120f; // 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; @@ -727,7 +740,7 @@ internal async Task ConnectUserAsync(string apiKey, string u private readonly object _websocketConnectionFailedFlagLock = new object(); private readonly object _websocketDisconnectedFlagLock = new object(); private readonly object _pendingHistoryEventsLock = new object(); - private readonly Queue _pendingHistoryEvents = new Queue(); + private readonly Queue _pendingHistoryEvents = new Queue(); /// /// Every write must happen on this thread @@ -766,6 +779,8 @@ internal async Task ConnectUserAsync(string apiKey, string u private int _historyDrainGeneration; private bool _isApplyingHistoryEvents; private DateTimeOffset? _historyMaxAppliedCreatedAt; + private float? _historyDrainStartedAt; + private bool _receiveBacklogWarningArmed = true; private async Task RefreshAuthTokenFromProvider() { @@ -1140,8 +1155,7 @@ private void RegisterEventType(string key, try { #if STREAM_DEBUG_ENABLED - var ignoreKeys = new[] { WSEventType.HealthCheck }; - if (!ignoreKeys.Contains(key)) + if (key != WSEventType.HealthCheck) { _logs.Warning("WS event received KEY: " + key + " CONTENT: " + payload); } @@ -1229,9 +1243,11 @@ bool BudgetExhausted() _logs.Info(_authCredentials.UserId + " WS message: " + liveMsg); #endif - var isHealth = IsHealthCheckMessage(liveMsg); var hardStop = processed >= countCap + EventDrainHealthGrace; - if (hardStop || (BudgetExhausted() && !isHealth)) + + // IsHealthCheckMessage does a full JObject.Parse, so only pay for it when the + // budget is actually spent and we need to decide whether to make an exception. + if (hardStop || (BudgetExhausted() && !IsHealthCheckMessage(liveMsg))) { _heldLiveMessage = liveMsg; break; @@ -1242,6 +1258,7 @@ bool BudgetExhausted() } TryCompleteHistoryDrainIfIdle(); + WarnOnStalledPump(); } private bool TryTakeLiveMessage(out string msg) @@ -1275,6 +1292,35 @@ private bool IsHealthCheckMessage(string msg) } internal Task EnqueueHistoryEvents(IEnumerable events) + => EnqueueHistoryEventsForReplay(events); + + /// + /// Queue /sync history events to be replayed as normal events across several Update calls. + /// The returned task completes when the last one has been handled. + /// + internal Task EnqueueHistoryEventsForReplay(IEnumerable events) + => EnqueueHistoryEvents(events, silentBatch: false); + + /// + /// Queue /sync history events to be applied to local state without per-event public callbacks, + /// across several Update calls. The /sync watermark advances once, when the whole batch drains. + /// + internal Task EnqueueHistoryEventsForBatchApply(IEnumerable events) + => EnqueueHistoryEvents(events, silentBatch: true); + + private readonly struct PendingHistoryEvent + { + public PendingHistoryEvent(object payload, bool silentBatch) + { + Payload = payload; + SilentBatch = silentBatch; + } + + public readonly object Payload; + public readonly bool SilentBatch; + } + + private Task EnqueueHistoryEvents(IEnumerable events, bool silentBatch) { if (events == null) { @@ -1285,11 +1331,21 @@ internal Task EnqueueHistoryEvents(IEnumerable events) { foreach (var e in events) { - _pendingHistoryEvents.Enqueue(e); + _pendingHistoryEvents.Enqueue(new PendingHistoryEvent(e, silentBatch)); + } + + if (_pendingHistoryEvents.Count == 0) + { + return Task.CompletedTask; } if (_historyDrainTcs == null || _historyDrainTcs.Task.IsCompleted) { + // TrySetResult is called outside this lock. Do not add + // RunContinuationsAsynchronously: RestoreStateLostDuringDisconnect awaits this + // task and must continue on the same Update so the re-query starts only after + // history has been applied. An inline continuation cannot deadlock because we + // already released the lock. _historyDrainTcs = new TaskCompletionSource(); } @@ -1305,13 +1361,13 @@ private bool HasPendingHistoryEvents() } } - private bool TryDequeueHistoryEvent(out object historyEvent) + private bool TryDequeueHistoryEvent(out PendingHistoryEvent historyEvent) { lock (_pendingHistoryEventsLock) { if (_pendingHistoryEvents.Count == 0) { - historyEvent = null; + historyEvent = default; return false; } @@ -1320,14 +1376,43 @@ private bool TryDequeueHistoryEvent(out object historyEvent) } } - private void ProcessHistoryEvent(object historyEvent) + private void ProcessHistoryEvent(PendingHistoryEvent pending) { - if (historyEvent == null) + if (pending.Payload == null) + { + return; + } + + if (!pending.SilentBatch) { + try + { + HandleNewWebsocketMessage(pending.Payload); + } + catch (Exception e) + { + // A malformed /sync event must not throw out of Update. + _logs.Exception(e); + } + return; } - HandleNewWebsocketMessage(historyEvent); + _isApplyingHistoryEvents = true; + try + { + HandleNewWebsocketMessage(pending.Payload); + } + catch (Exception e) + { + // Do not abort the batch. The /sync last_sync_at cursor only moves to events that + // applied, so a failed event is retried on the next reconnect. + _logs.Exception(e); + } + finally + { + _isApplyingHistoryEvents = false; + } } private void TryCompleteHistoryDrainIfIdle() @@ -1344,7 +1429,17 @@ private void TryCompleteHistoryDrainIfIdle() _historyDrainTcs = null; } - // Outside the lock so FetchAndProcess resumes on this Update, not a later scheduler tick. + // End of a silent batch: advance the /sync cursor exactly once, to the newest event + // that actually applied. Mirrors the finally block in ApplyHistoryEvents. + if (_historyMaxAppliedCreatedAt.HasValue) + { + TryAdvanceLastEventReceivedAt(_historyMaxAppliedCreatedAt.Value, HistorySyncWatermarkSource); + } + + _historyMaxAppliedCreatedAt = null; + + // Outside the lock so FetchAndProcess / recovery resumes on this Update, not a later + // scheduler tick. tcs.TrySetResult(true); } @@ -1359,9 +1454,40 @@ private void ClearPendingHistoryEvents() _historyDrainTcs = null; } + // A partially applied, abandoned batch must not later advance the cursor past events + // that were dropped. + _historyMaxAppliedCreatedAt = null; + _historyDrainStartedAt = null; + tcs?.TrySetResult(true); } + private void WarnOnStalledPump() + { + int pendingHistory; + lock (_pendingHistoryEventsLock) + { + pendingHistory = _pendingHistoryEvents.Count; + } + + var backlog = pendingHistory + (_heldLiveMessage != null ? 1 : 0); + if (backlog > ReceiveBacklogWarningThreshold) + { + if (_receiveBacklogWarningArmed) + { + _receiveBacklogWarningArmed = false; + _logs.Warning( + $"{pendingHistory} chat events are still waiting to be applied. They are being paced across " + + $"frames ({EventDrainTimeBudgetMs}ms per Update). Nothing is dropped, but your event handlers " + + "may be too expensive, or Update is not being called often enough."); + } + + return; + } + + _receiveBacklogWarningArmed = true; + } + private TDto DeserializePayload(object payload) { if (payload is string content) @@ -1587,6 +1713,39 @@ private void UpdateHealthCheck() PingHealthCheck(); } + // A paced history drain blocks live messages, including health.check, so the receive + // timestamp cannot advance while it runs. Timing out here would clear the queue and + // make the next /sync re-fetch the same events - a loop that never catches up. Pings + // still go out above, so the server keeps the socket open. Bounded by + // EventDrainMaxStallSeconds so a genuinely dead socket is still detected. + if (HasPendingHistoryEvents()) + { + if (!_historyDrainStartedAt.HasValue) + { + _historyDrainStartedAt = _timeService.Time; + } + + if (_timeService.Time - _historyDrainStartedAt.Value < EventDrainMaxStallSeconds) + { + _lastHealthCheckReceivedTime = _timeService.Time; + return; + } + + _logs.Warning( + $"The /sync catch-up has been draining for more than {EventDrainMaxStallSeconds}s and is " + + "still behind. Letting the health-check timeout apply. Your event handlers are likely too " + + "expensive for the per-frame budget."); + + // We have been stamping _lastHealthCheckReceivedTime every frame to hold the + // timeout off. Expire it so the check below fires this frame instead of waiting + // another HealthCheckMaxWaitingTime after we stop stamping. + _lastHealthCheckReceivedTime = _timeService.Time - HealthCheckMaxWaitingTime - 1f; + } + else + { + _historyDrainStartedAt = null; + } + var timeSinceLastHealthCheck = _timeService.Time - _lastHealthCheckReceivedTime; if (timeSinceLastHealthCheck > HealthCheckMaxWaitingTime) { diff --git a/Assets/Plugins/StreamChat/Core/StreamChatClient.cs b/Assets/Plugins/StreamChat/Core/StreamChatClient.cs index 90eee900..769dab48 100644 --- a/Assets/Plugins/StreamChat/Core/StreamChatClient.cs +++ b/Assets/Plugins/StreamChat/Core/StreamChatClient.cs @@ -1237,7 +1237,9 @@ private void OnConnected(HealthCheckEventInternalDTO dto) /// in local state but stop receiving events. /// /// 1. /sync catch-up (best effort). Must run first: a replayed channel.truncated - /// would wipe messages fetched in step 2. + /// would wipe messages fetched in step 2. A large catch-up is paced across Update + /// calls; this method awaits that drain, so step 2 does not start until it finishes. + /// may therefore fire seconds after . /// 2. Re-query and re-watch, even if step 1 failed or was skipped. /// 3. Raise once. /// @@ -1384,11 +1386,11 @@ private async Task TryCatchUpWithHistoryAsync(IReadOnlyList recoveryChan if (InternalLowLevelClient.Config.StateRecoveryStrategy == StateRecoveryStrategy.BatchStateUpdate) { - InternalLowLevelClient.ApplyHistoryEvents(response.Events); + await InternalLowLevelClient.EnqueueHistoryEventsForBatchApply(response.Events); } else { - InternalLowLevelClient.ReplayHistoryEvents(response.Events); + await InternalLowLevelClient.EnqueueHistoryEventsForReplay(response.Events); } } catch (StreamApiException ex) when (ex.IsInputError()) diff --git a/Assets/Plugins/StreamChat/Tests/LowLevelClient/StreamChatLowLevelClientEventDrainTests.cs b/Assets/Plugins/StreamChat/Tests/LowLevelClient/StreamChatLowLevelClientEventDrainTests.cs index f148dc70..b99b3778 100644 --- a/Assets/Plugins/StreamChat/Tests/LowLevelClient/StreamChatLowLevelClientEventDrainTests.cs +++ b/Assets/Plugins/StreamChat/Tests/LowLevelClient/StreamChatLowLevelClientEventDrainTests.cs @@ -2,6 +2,7 @@ using System; using System.Collections.Generic; using System.Net.WebSockets; +using System.Reflection; using System.Threading.Tasks; using NSubstitute; using NUnit.Framework; @@ -226,6 +227,122 @@ public void when_disconnect_mid_history_drain_expect_pending_cleared_and_no_doub Assert.AreEqual(new[] { "h1" }, received.ToArray()); } + [Test] + public void when_health_timeout_elapses_during_history_drain_expect_connection_stays() + { + var client = CreateConnectedClient(); + client.EventDrainCountCap = 1; + client.EventDrainTimeBudgetMs = 10_000; + SetupDisconnectRaisesDisconnected(); + + EnqueueHistory(client, Message("h1"), Message("h2"), Message("h3"), Message("h4"), Message("h5")); + client.Update(0.1f); + Assert.AreEqual(ConnectionState.Connected, client.ConnectionState); + + _mockTimeService.Time.Returns(31f); + client.Update(0.1f); + + Assert.AreEqual(ConnectionState.Connected, client.ConnectionState); + } + + [Test] + public void when_history_drain_exceeds_max_stall_expect_health_timeout_fires() + { + var client = CreateConnectedClient(); + client.EventDrainCountCap = 1; + client.EventDrainTimeBudgetMs = 10_000; + SetupDisconnectRaisesDisconnected(); + + EnqueueHistory(client, Message("h1"), Message("h2"), Message("h3"), Message("h4"), Message("h5")); + client.Update(0.1f); + + _mockTimeService.Time.Returns(StreamChatLowLevelClient.EventDrainMaxStallSeconds + 1f); + client.Update(0.1f); + + Assert.AreNotEqual(ConnectionState.Connected, client.ConnectionState); + } + + [Test] + public void when_disconnect_with_held_live_message_expect_it_is_still_delivered() + { + var client = CreateConnectedClient(); + client.EventDrainCountCap = 1; + client.EventDrainTimeBudgetMs = 10_000; + SetupDisconnectRaisesDisconnected(); + + var received = new List(); + client.MessageReceived += e => received.Add(e.Message.Id); + + EnqueueLiveMessages(Message("m1"), Message("m2")); + client.Update(0.1f); + Assert.AreEqual(new[] { "m1" }, received.ToArray()); + + client.DisconnectAsync(DisconnectCause.ConnectionReleased).GetAwaiter().GetResult(); + client.Update(0.1f); + + Assert.AreEqual(new[] { "m1", "m2" }, received.ToArray()); + } + + [Test] + public void when_silent_batch_spans_updates_expect_watermark_advances_once_at_end() + { + var client = CreateConnectedClient(); + client.EventDrainCountCap = 1; + client.EventDrainTimeBudgetMs = 10_000; + + var t0 = new DateTimeOffset(2026, 8, 10, 12, 0, 0, TimeSpan.Zero); + var t1 = t0.AddMinutes(1); + var t2 = t0.AddMinutes(2); + + var received = 0; + client.MessageReceived += _ => received++; + + var drain = client.EnqueueHistoryEventsForBatchApply(new object[] + { + Message("m1", t0), + Message("m2", t2), + Message("m3", t1), + }); + + client.Update(0.1f); + Assert.AreEqual(0, received); + Assert.IsNull(GetLastEventReceivedAt(client)); + Assert.IsFalse(drain.IsCompleted); + + client.Update(0.1f); + Assert.AreEqual(0, received); + Assert.IsNull(GetLastEventReceivedAt(client)); + Assert.IsFalse(drain.IsCompleted); + + client.Update(0.1f); + Assert.AreEqual(0, received); + Assert.AreEqual(t2, GetLastEventReceivedAt(client)); + Assert.IsTrue(drain.IsCompleted); + } + + [Test] + public void when_held_live_then_history_enqueued_expect_history_before_held() + { + var client = CreateConnectedClient(); + client.EventDrainCountCap = 1; + client.EventDrainTimeBudgetMs = 10_000; + + var received = new List(); + client.MessageReceived += e => received.Add(e.Message.Id); + + EnqueueLiveMessages(Message("l1"), Message("l2")); + client.Update(0.1f); + Assert.AreEqual(new[] { "l1" }, received.ToArray()); + + EnqueueHistory(client, Message("h1")); + client.Update(0.1f); + + Assert.AreEqual(new[] { "l1", "h1" }, received.ToArray()); + + client.Update(0.1f); + Assert.AreEqual(new[] { "l1", "h1", "l2" }, received.ToArray()); + } + private readonly List _resourcesToDispose = new List(); private AuthCredentials _authCredentials; @@ -251,6 +368,11 @@ private StreamChatLowLevelClient CreateConnectedClient() return client; } + private void EnqueueHistory(StreamChatLowLevelClient client, params object[] events) + { + client.EnqueueHistoryEventsForReplay(events); + } + private void EnqueueLiveMessages(params string[] messages) { var queue = new Queue(messages); @@ -283,14 +405,26 @@ private void SetupDisconnectRaisesDisconnected() private static void SetDisconnectionWatermark(StreamChatLowLevelClient client, DateTimeOffset value) { var field = typeof(StreamChatLowLevelClient).GetField("_disconnectionLastEventReceivedAt", - System.Reflection.BindingFlags.Instance | System.Reflection.BindingFlags.NonPublic); + BindingFlags.Instance | BindingFlags.NonPublic); Assert.IsNotNull(field); field.SetValue(client, (DateTimeOffset?)value); } + private static DateTimeOffset? GetLastEventReceivedAt(StreamChatLowLevelClient client) + { + var field = typeof(StreamChatLowLevelClient).GetField("_lastEventReceivedAt", + BindingFlags.Instance | BindingFlags.NonPublic); + Assert.IsNotNull(field); + return (DateTimeOffset?)field.GetValue(client); + } + private static string Message(string id) => $"{{\"type\":\"message.new\",\"cid\":\"messaging:test\",\"message\":{{\"id\":\"{id}\",\"text\":\"{id}\",\"type\":\"regular\"}}}}"; + private static string Message(string id, DateTimeOffset createdAt) + => $"{{\"type\":\"message.new\",\"cid\":\"messaging:test\",\"created_at\":\"{createdAt:O}\"," + + $"\"message\":{{\"id\":\"{id}\",\"text\":\"{id}\",\"type\":\"regular\",\"created_at\":\"{createdAt:O}\"}}}}"; + private static string MessageEventJson(string id) => Message(id); private static string HealthCheck() => "{\"connection_id\":\"fakeId\",\"type\":\"health.check\"}"; diff --git a/Assets/Plugins/StreamChat/Tests/StateSync/StateRecoveryClientTests.cs b/Assets/Plugins/StreamChat/Tests/StateSync/StateRecoveryClientTests.cs index 6f786a59..c84d38c3 100644 --- a/Assets/Plugins/StreamChat/Tests/StateSync/StateRecoveryClientTests.cs +++ b/Assets/Plugins/StreamChat/Tests/StateSync/StateRecoveryClientTests.cs @@ -549,6 +549,95 @@ public void when_replay_recovery_syncs_message_new_expect_channel_message_receiv Assert.AreEqual(1, _recoveredEvents.Count); } + [Test] + public void when_replay_recovery_syncs_many_events_expect_they_span_updates() + { + _client.InternalLowLevelClient.EventDrainCountCap = 2; + _client.InternalLowLevelClient.EventDrainTimeBudgetMs = 10_000; + + Connect(); + var channel = WatchChannel("messaging:a"); + RespondWith(QueryChannelsEndpoint, QueryChannelsJson("messaging:a", "a")); + + var received = 0; + channel.MessageReceived += (_, __) => received++; + + ReconnectWithSync(ManySyncMessages("messaging:a", count: 5)); + + Assert.AreEqual(1, received, "Health check consumes one slot of the cap; only one history event fits."); + AssertQueryChannelsCallCount(0); + Assert.AreEqual(0, _recoveredEvents.Count); + + Update(); + Assert.AreEqual(3, received); + AssertQueryChannelsCallCount(0); + Assert.AreEqual(0, _recoveredEvents.Count); + + Update(); + Assert.AreEqual(5, received); + AssertQueryChannelsCallCount(1); + Assert.AreEqual(1, _recoveredEvents.Count); + } + + [Test] + public void when_silent_recovery_syncs_many_events_expect_they_span_updates_without_callbacks() + { + _config.StateRecoveryStrategy = StateRecoveryStrategy.BatchStateUpdate; + _client.InternalLowLevelClient.EventDrainCountCap = 2; + _client.InternalLowLevelClient.EventDrainTimeBudgetMs = 10_000; + + Connect(); + var channel = WatchChannel("messaging:a"); + RespondWith(QueryChannelsEndpoint, QueryChannelsJson("messaging:a", "a")); + + var received = 0; + channel.MessageReceived += (_, __) => received++; + + ReconnectWithSync(ManySyncMessages("messaging:a", count: 5)); + + Assert.AreEqual(0, received); + Assert.AreEqual(1, channel.Messages.Count); + AssertQueryChannelsCallCount(0); + Assert.AreEqual(0, _recoveredEvents.Count); + Assert.IsNull(GetLastEventReceivedAt(_client.InternalLowLevelClient)); + + Update(); + Assert.AreEqual(0, received); + Assert.AreEqual(3, channel.Messages.Count); + AssertQueryChannelsCallCount(0); + Assert.IsNull(GetLastEventReceivedAt(_client.InternalLowLevelClient)); + + Update(); + Assert.AreEqual(0, received); + Assert.AreEqual(5, channel.Messages.Count); + AssertQueryChannelsCallCount(1); + Assert.AreEqual(1, _recoveredEvents.Count); + Assert.AreEqual(new DateTimeOffset(2026, 8, 24, 12, 0, 4, TimeSpan.Zero), + GetLastEventReceivedAt(_client.InternalLowLevelClient)); + } + + [Test] + public void when_disconnect_mid_paced_recovery_expect_no_state_recovered() + { + _client.InternalLowLevelClient.EventDrainCountCap = 1; + _client.InternalLowLevelClient.EventDrainTimeBudgetMs = 10_000; + + Connect(); + WatchChannel("messaging:a"); + RespondWith(QueryChannelsEndpoint, QueryChannelsJson("messaging:a", "a")); + + ReconnectWithSync(ManySyncMessages("messaging:a", count: 5)); + Assert.AreEqual(0, _recoveredEvents.Count); + AssertQueryChannelsCallCount(0); + + DropConnection(); + Update(); + Update(); + + Assert.AreEqual(0, _recoveredEvents.Count); + AssertQueryChannelsCallCount(0); + } + [Test] public void when_silent_recovery_syncs_custom_event_expect_channel_custom_event_received() { @@ -624,6 +713,26 @@ private static string CustomEventJson(string cid, string type) private static string SyncEventsJson(params string[] events) => "{\"events\":[" + string.Join(",", events) + "]}"; + private static string ManySyncMessages(string cid, int count) + { + var start = new DateTimeOffset(2026, 8, 24, 12, 0, 0, TimeSpan.Zero); + var events = new string[count]; + for (var i = 0; i < count; i++) + { + events[i] = MessageNewJson(cid, $"msg-{i}", start.AddSeconds(i)); + } + + return SyncEventsJson(events); + } + + private static DateTimeOffset? GetLastEventReceivedAt(StreamChatLowLevelClient client) + { + var field = typeof(StreamChatLowLevelClient).GetField("_lastEventReceivedAt", + BindingFlags.Instance | BindingFlags.NonPublic); + Assert.IsNotNull(field, "Expected _lastEventReceivedAt to exist."); + return (DateTimeOffset?)field.GetValue(client); + } + private void ReconnectWithSync(string syncJson) { var now = new DateTimeOffset(2026, 8, 24, 12, 0, 0, TimeSpan.Zero);