diff --git a/src/RustPlusBot.Features.Connections/ConnectionOptions.cs b/src/RustPlusBot.Features.Connections/ConnectionOptions.cs index 0c523ca3..bd817610 100644 --- a/src/RustPlusBot.Features.Connections/ConnectionOptions.cs +++ b/src/RustPlusBot.Features.Connections/ConnectionOptions.cs @@ -44,4 +44,11 @@ public sealed class ConnectionOptions /// How often to poll managed devices for reachability changes while connected. Default 5m. public TimeSpan ReachabilityPollInterval { get; set; } = TimeSpan.FromMinutes(5); + + /// + /// How often the liveness watchdog checks whether the socket is still open while connected. The Rust+ + /// library raises no event when the server closes the socket, so this bounds how long a silent drop goes + /// unnoticed (independently of ). Default 5s. + /// + public TimeSpan LivenessPollInterval { get; set; } = TimeSpan.FromSeconds(5); } diff --git a/src/RustPlusBot.Features.Connections/Listening/IRustServerConnection.cs b/src/RustPlusBot.Features.Connections/Listening/IRustServerConnection.cs index ad5e8454..7ba22920 100644 --- a/src/RustPlusBot.Features.Connections/Listening/IRustServerConnection.cs +++ b/src/RustPlusBot.Features.Connections/Listening/IRustServerConnection.cs @@ -5,6 +5,13 @@ namespace RustPlusBot.Features.Connections.Listening; /// One live Rust+ socket to a single server, driven by one player credential. internal interface IRustServerConnection : IAsyncDisposable { + /// + /// Whether the underlying socket is currently open. Cheap and synchronous: the Rust+ library raises no + /// event when the server closes the socket, so a liveness watchdog polls this to detect a drop + /// promptly instead of waiting for the next (up to a minute apart) heartbeat. + /// + bool IsConnected { get; } + /// Connects within . /// How long to wait for the connection. /// A cancellation token. diff --git a/src/RustPlusBot.Features.Connections/Listening/RustPlusSocketSource.cs b/src/RustPlusBot.Features.Connections/Listening/RustPlusSocketSource.cs index c43c6a42..0dc59c44 100644 --- a/src/RustPlusBot.Features.Connections/Listening/RustPlusSocketSource.cs +++ b/src/RustPlusBot.Features.Connections/Listening/RustPlusSocketSource.cs @@ -34,6 +34,8 @@ public IRustServerConnection Create(string ip, int port, ulong steamId, string p /// Returned when the player token is unusable; reports the credential as rejected and does nothing else. private sealed class RejectedConnection : IRustServerConnection { + public bool IsConnected => false; + public Task ConnectAsync(TimeSpan timeout, CancellationToken cancellationToken) => Task.FromResult(SocketConnectOutcome.AuthRejected); @@ -170,6 +172,11 @@ public RustPlusServerConnection(string ip, _rustPlus.OnClanChanged += OnClanChanged; } + /// + // CONFIRMED (2.0.0-beta.5): RustPlusSocket.IsConnected == (_webSocket?.State == WebSocketState.Open), + // so it flips to false as soon as the server closes the socket (the library raises no event for that). + public bool IsConnected => _rustPlus.IsConnected; + public async Task ConnectAsync(TimeSpan timeout, CancellationToken cancellationToken) { using var timeoutCts = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken); @@ -380,7 +387,10 @@ public async Task GetClanInfoAsync(TimeSpan timeout, Cancellati { // CONFIRMED (2.0.0-beta.4): GetClanInfoAsync returns Task>, exposing // IsSuccess, Data, and Error?.Code as the response accessors. - var response = await _rustPlus.GetClanInfoAsync(timeoutCts.Token).ConfigureAwait(false); + // .WaitAsync guards the timeout even if the beta call doesn't internally honor the token + // (matching the other probes): on a dead socket this must never block the connect path. + var response = await _rustPlus.GetClanInfoAsync(timeoutCts.Token) + .WaitAsync(timeoutCts.Token).ConfigureAwait(false); return ClanMapping.FromResponse(response.IsSuccess, response.Error?.Code, response.Data); } catch (OperationCanceledException) when (!cancellationToken.IsCancellationRequested) diff --git a/src/RustPlusBot.Features.Connections/Supervisor/ConnectionSupervisor.cs b/src/RustPlusBot.Features.Connections/Supervisor/ConnectionSupervisor.cs index e340d79d..80f78185 100644 --- a/src/RustPlusBot.Features.Connections/Supervisor/ConnectionSupervisor.cs +++ b/src/RustPlusBot.Features.Connections/Supervisor/ConnectionSupervisor.cs @@ -558,9 +558,6 @@ void OnTeamMessage(object? sender, TeamChatLine line) } #pragma warning restore RCS1163 - var dims = await connection.GetMapDimensionsAsync(_options.HeartbeatTimeout, ct).ConfigureAwait(false); - var rigs = await GetRigPositionsAsync(key.Server, connection, ct).ConfigureAwait(false); - #pragma warning disable RCS1163 // Unused 'sender': required by the EventHandler delegate shape. void OnSmartDevice(object? sender, SmartDeviceTrigger trigger) { @@ -607,21 +604,29 @@ void OnClanChanged(object? sender, ClanProbeResult probe) await PublishClanStateAsync(key, clanProbe).ConfigureAwait(false); using var pollCts = CancellationTokenSource.CreateLinkedTokenSource(ct); - var markerPoll = Task.Run(() => PollMarkersAsync(key, connection, dims, rigs, tracker, pollCts.Token), + var markerPoll = Task.Run(() => PollMarkersAsync(key, connection, tracker, pollCts.Token), CancellationToken.None); var reachabilityPoll = Task.Run(() => PollReachabilityAsync(key, connection, pollCts.Token), CancellationToken.None); + // Race the heartbeat against a liveness watchdog: the Rust+ library raises no event when the SERVER + // closes the socket, so without the watchdog a silent drop goes unnoticed until the next heartbeat + // (up to a minute). Whichever signals a reason first wins; the finally cancels and joins the rest. + var heartbeat = RunHeartbeatLoopAsync(key, connection, credentialId, pollCts.Token); + var liveness = WatchLivenessAsync(key, connection, pollCts.Token); try { - return await RunHeartbeatLoopAsync(key, connection, credentialId, ct).ConfigureAwait(false); +#pragma warning disable VSTHRD003 // Suppress: heartbeat and liveness are owned by this connected window and joined below. + var winner = await Task.WhenAny(heartbeat, liveness).ConfigureAwait(false); + return await winner.ConfigureAwait(false); +#pragma warning restore VSTHRD003 } finally { await pollCts.CancelAsync().ConfigureAwait(false); try { -#pragma warning disable VSTHRD003 // Suppress: markerPoll and reachabilityPoll are owned by this connected window and explicitly joined on exit. - await Task.WhenAll(markerPoll, reachabilityPoll).ConfigureAwait(false); +#pragma warning disable VSTHRD003 // Suppress: all four tasks are owned by this connected window and explicitly joined on exit. + await Task.WhenAll(markerPoll, reachabilityPoll, heartbeat, liveness).ConfigureAwait(false); #pragma warning restore VSTHRD003 } catch (OperationCanceledException) @@ -638,6 +643,42 @@ void OnClanChanged(object? sender, ClanProbeResult probe) } } + /// + /// Polls while connected and returns + /// as soon as the socket is no longer open. This is the only + /// prompt signal for a server-initiated close, which the Rust+ library does not surface as an event. + /// + /// The (guild, server) routing key. + /// The live connection whose liveness is watched. + /// Cancels when the connected window ends. + /// on a detected drop, else + /// when cancelled. + private async Task WatchLivenessAsync( + (ulong Guild, Guid Server) key, + IRustServerConnection connection, + CancellationToken ct) + { + try + { + while (!ct.IsCancellationRequested) + { + await Task.Delay(_options.LivenessPollInterval, ct).ConfigureAwait(false); + if (!connection.IsConnected) + { + LogSocketDropped(logger, key.Server); + return ReconnectReason.Unreachable; + } + } + } + catch (OperationCanceledException) + { + // The connected window is ending (stop/reconnect): Task.Delay throws on cancellation. + // Return Stopped so the WhenAny winner is a clean reason rather than a faulted task. + } + + return ReconnectReason.Stopped; + } + private async Task RunHeartbeatLoopAsync( (ulong Guild, Guid Server) key, IRustServerConnection connection, @@ -667,11 +708,17 @@ await PublishStatusAsync(key, ConnectionStatus.Connected, beat.PlayerCount, cred private async Task PollMarkersAsync( (ulong Guild, Guid Server) key, IRustServerConnection connection, - MapDimensions? dims, - IReadOnlyList rigs, TeamStateTracker tracker, CancellationToken ct) { + // Fetch map dimensions and oil-rig positions here, off the critical connect path: these are the two + // heavy full-map downloads, and on a degraded map endpoint they can stall for seconds. Doing them in + // this background poll means a slow map no longer delays the connection going live (heartbeat + chat + // relay start immediately); marker/rig detection simply activates once these resolve. Both degrade + // safely on timeout (dims -> null, rigs -> empty) without ending the poll. + var dims = await connection.GetMapDimensionsAsync(_options.HeartbeatTimeout, ct).ConfigureAwait(false); + var rigs = await GetRigPositionsAsync(key.Server, connection, ct).ConfigureAwait(false); + IReadOnlyList? previous = null; var rigsInRadius = new HashSet(); while (!ct.IsCancellationRequested) @@ -703,14 +750,16 @@ await eventBus.PublishAsync( .ConfigureAwait(false); } } - catch (OperationCanceledException) + catch (OperationCanceledException) when (ct.IsCancellationRequested) { return; // stopping } -#pragma warning disable CA1031 // Broad catch: a failed poll is logged and skipped; the previous snapshot is retained. +#pragma warning disable CA1031 // Broad catch: a failed poll (incl. a per-request timeout) is logged and skipped; the previous snapshot is retained. catch (Exception ex) #pragma warning restore CA1031 { + // A per-request timeout (OperationCanceledException with ct NOT cancelled) must NOT end the + // poll loop — that would silently stop marker/rig/AFK detection for the rest of the connection. LogMarkerPollFailed(logger, ex, key.Server); } @@ -767,14 +816,16 @@ private async Task PollReachabilityAsync( { current = await ReadAllReachabilityAsync(key, connection, ct).ConfigureAwait(false); } - catch (OperationCanceledException) + catch (OperationCanceledException) when (ct.IsCancellationRequested) { throw; } -#pragma warning disable CA1031 // Broad catch: a failed reachability sweep is logged and retried next cycle. +#pragma warning disable CA1031 // Broad catch: a failed reachability sweep (incl. a per-request timeout) is logged and retried next cycle. catch (Exception ex) #pragma warning restore CA1031 { + // A per-request timeout (OperationCanceledException with ct NOT cancelled) must be retried next + // cycle, not rethrown — rethrowing would tear the sweep down for the rest of the connection. LogReachabilityPollFailed(logger, ex, key.Server); continue; } @@ -942,14 +993,17 @@ private async Task> GetRigPositionsAsync( return rigs; } - catch (OperationCanceledException) + catch (OperationCanceledException) when (ct.IsCancellationRequested) { + // Real shutdown/reconnect of THIS connection: propagate so the loop tears down. throw; } -#pragma warning disable CA1031 // Broad catch: a monuments-fetch failure just disables rig detection this window. +#pragma warning disable CA1031 // Broad catch: a monuments-fetch failure (incl. a per-request timeout) just disables rig detection this window. catch (Exception ex) #pragma warning restore CA1031 { + // A per-request timeout surfaces as OperationCanceledException with ct NOT cancelled; it must + // degrade rig detection for this window, never terminate the connection loop. LogMonumentsFetchFailed(logger, ex, serverId); // rig detection degrades gracefully for this window return []; } @@ -1435,6 +1489,10 @@ await eventBus.PublishAsync( [LoggerMessage(Level = LogLevel.Warning, Message = "Reachability poll for server {ServerId} failed.")] private static partial void LogReachabilityPollFailed(ILogger logger, Exception exception, Guid serverId); + [LoggerMessage(Level = LogLevel.Information, + Message = "Socket for server {ServerId} closed by the server; reconnecting.")] + private static partial void LogSocketDropped(ILogger logger, Guid serverId); + [LoggerMessage(Level = LogLevel.Warning, Message = "Fetching monuments for oil-rig detection on server {ServerId} failed; rig detection disabled for this connection.")] diff --git a/tests/RustPlusBot.Features.Connections.Tests/ConnectionSupervisorTests.cs b/tests/RustPlusBot.Features.Connections.Tests/ConnectionSupervisorTests.cs index 0d43d133..e5e04820 100644 --- a/tests/RustPlusBot.Features.Connections.Tests/ConnectionSupervisorTests.cs +++ b/tests/RustPlusBot.Features.Connections.Tests/ConnectionSupervisorTests.cs @@ -67,6 +67,7 @@ private static Harness CreateHarness(FakeRustSocketSource source) HeartbeatTimeout = TimeSpan.FromMilliseconds(200), MarkerPollInterval = TimeSpan.FromMilliseconds(20), MarkerPollFastInterval = TimeSpan.FromMilliseconds(20), + LivenessPollInterval = TimeSpan.FromMilliseconds(20), })); services.AddSingleton(); services.AddSingleton(); @@ -231,6 +232,68 @@ public async Task Heartbeat_Unreachable_ReconnectsAndRecovers() Assert.True(source.CreateCount >= 2); } + /// + /// A per-request timeout in a connected window (here the marker poll's oil-rig monuments fetch) surfaces + /// as an even though the connection token is not cancelled. It + /// must degrade rig detection for this window, NOT terminate the whole connection loop — otherwise a + /// transient slow map endpoint permanently kills the connection with no reconnect (the real-world bug). + /// + [Fact] + public async Task Connect_MonumentsTimeout_DoesNotKillLoop_AndStillReconnects() + { + var source = new FakeRustSocketSource(); + source.EnqueueConnect(SocketConnectOutcome.Connected); + source.TimeoutOnMonumentsOnce(); // the marker poll's rig fetch times out on the FIRST connection only + source.EnqueueHeartbeat(HeartbeatResult.Ok(2)); // first heartbeat -> Connected + source.EnqueueHeartbeat(HeartbeatResult.Unreachable); // next heartbeat -> drop + source.EnqueueConnect(SocketConnectOutcome.Connected); // reconnect + source.EnqueueHeartbeat(HeartbeatResult.Ok(4)); + await using var h = CreateHarness(source); + var (serverId, _, _) = await SeedAsync(h.Provider); + + await h.Supervisor.EnsureConnectionAsync(10UL, serverId); + + var recovered = await WaitForStateAsync( + h.Provider, serverId, s => s.Status == ConnectionStatus.Connected && s.PlayerCount == 4); + Assert.NotNull(recovered); + Assert.True(source.CreateCount >= 2); + } + + /// + /// The Rust+ library raises no event when the SERVER closes the socket; only + /// flips. The liveness watchdog must notice that and + /// drive a reconnect promptly, without waiting for the (up to a minute apart) heartbeat — here the fake's + /// heartbeat keeps reporting Ok, so only the watchdog can detect the drop. + /// + [Fact] + public async Task ServerClosesSocket_LivenessWatchdog_DrivesReconnect() + { + var source = new FakeRustSocketSource(); + source.EnqueueConnect(SocketConnectOutcome.Connected); + source.EnqueueHeartbeat(HeartbeatResult.Ok(2)); // first heartbeat -> Connected; held thereafter + source.EnqueueConnect(SocketConnectOutcome.Connected); // reconnect after the watchdog fires + await using var h = CreateHarness(source); + var (serverId, _, _) = await SeedAsync(h.Provider); + + await h.Supervisor.EnsureConnectionAsync(10UL, serverId); + var connected = await WaitForStateAsync( + h.Provider, serverId, s => s.Status == ConnectionStatus.Connected && s.PlayerCount == 2); + Assert.NotNull(connected); + var live = source.LastConnection; + Assert.NotNull(live); + + // Simulate a server-initiated close: the socket is no longer open, but the heartbeat still answers Ok. + live.IsConnected = false; + + var deadline = DateTimeOffset.UtcNow.AddSeconds(30); + while (source.CreateCount < 2 && DateTimeOffset.UtcNow < deadline) + { + await Task.Delay(15); + } + + Assert.True(source.CreateCount >= 2, "the liveness watchdog should have driven a reconnect"); + } + /// /// WasConnected must be computed from in-process state (the DB status survives restarts and /// would claim Connected at boot): statuses before the first Connected carry false; the drop diff --git a/tests/RustPlusBot.Features.Connections.Tests/Fakes/FakeRustSocketSource.cs b/tests/RustPlusBot.Features.Connections.Tests/Fakes/FakeRustSocketSource.cs index a84c9677..7b2277bd 100644 --- a/tests/RustPlusBot.Features.Connections.Tests/Fakes/FakeRustSocketSource.cs +++ b/tests/RustPlusBot.Features.Connections.Tests/Fakes/FakeRustSocketSource.cs @@ -29,6 +29,7 @@ internal sealed class FakeRustSocketSource : IRustSocketSource private HeartbeatResult _lastHeartbeat = HeartbeatResult.Ok(0); private ClanProbeResult? _pendingClanProbe; private IReadOnlyList _pendingMonuments = []; + private bool _pendingMonumentsTimeout; /// Number of times has been called. Safe to read from any thread. public int CreateCount => Volatile.Read(ref _createCount); @@ -60,6 +61,11 @@ public IRustServerConnection Create(string ip, int port, ulong steamId, string p connection.MonumentsResult = _pendingMonuments; _pendingMonuments = []; + // Transfer any pre-staged monuments timeout so the marker poll's rig fetch throws for the NEXT + // connection only (mimicking a real per-request timeout, which surfaces as OperationCanceledException). + connection.MonumentsTimeout = _pendingMonumentsTimeout; + _pendingMonumentsTimeout = false; + // Transfer any pre-staged storage contents so they are in place before the prime loop starts. foreach (var (entityId, contents) in _pendingStorageContents) { @@ -119,6 +125,14 @@ public void EnqueueMarkers(IReadOnlyList markers) => /// The monument list to return from . public void SetMonuments(IReadOnlyList monuments) => _pendingMonuments = monuments; + /// + /// Makes the NEXT connection's throw + /// , simulating a per-request timeout (the outer connection + /// token is NOT cancelled). The supervisor issues this fetch from its background marker poll, not the + /// connect path. Applies to the next connection only. Call before . + /// + public void TimeoutOnMonumentsOnce() => _pendingMonumentsTimeout = true; + /// /// Pre-stages the probe result returned by for the NEXT /// connection created by . Transferred to the new connection at creation time, before @@ -244,6 +258,10 @@ internal sealed class FakeConnection(SocketConnectOutcome outcome, FakeRustSocke /// The monuments returned by . Defaults to empty. public IReadOnlyList MonumentsResult { get; set; } = []; + /// When true, throws + /// (a per-request timeout) instead of returning . + public bool MonumentsTimeout { get; set; } + /// The bytes returned by . Defaults to null. public byte[]? MapImageResult { get; set; } @@ -253,6 +271,10 @@ internal sealed class FakeConnection(SocketConnectOutcome outcome, FakeRustSocke /// Messages sent to in-game clan chat through this fake. public List SentClanMessages { get; } = []; + /// Liveness flag consulted by the supervisor's watchdog. Set false to simulate a + /// server-initiated socket close (which the real library signals via IsConnected, not an event). + public bool IsConnected { get; set; } = true; + /// Raised when a team chat message arrives on this connection. public event EventHandler? TeamMessageReceived; @@ -393,7 +415,9 @@ public Task> GetMapMarkersAsync(TimeSpan timeou public Task> GetMonumentsAsync(TimeSpan timeout, CancellationToken cancellationToken = default) => - Task.FromResult(MonumentsResult); + MonumentsTimeout + ? Task.FromException>(new OperationCanceledException()) + : Task.FromResult(MonumentsResult); public Task GetMapImageAsync(TimeSpan timeout, CancellationToken cancellationToken = default) => Task.FromResult(MapImageResult);