feat(connections): detect server-initiated socket drops with a liveness watchdog

The Rust+ library raises no event when the server closes the socket, so a silent
drop previously went unnoticed until the next heartbeat (up to HeartbeatInterval,
default 60s) — a long window of a dead-but-"connected" socket.

Add IsConnected to IRustServerConnection (backed by RustPlusSocket.IsConnected)
and a WatchLivenessAsync watchdog that races the heartbeat loop and returns
Unreachable as soon as the socket is no longer open, bounded by the new
ConnectionOptions.LivenessPollInterval (default 5s).

Test: ServerClosesSocket_LivenessWatchdog_DrivesReconnect (heartbeat keeps
answering Ok, so only the watchdog can detect the drop).

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

Author: HandyS11

src/RustPlusBot.Features.Connections/ConnectionOptions.cs

@@ -44,4 +44,11 @@ public sealed class ConnectionOptions
 
     /// <summary>How often to poll managed devices for reachability changes while connected. Default 5m.</summary>
     public TimeSpan ReachabilityPollInterval { get; set; } = TimeSpan.FromMinutes(5);
+
+    /// <summary>
+    /// 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 <see cref="HeartbeatInterval"/>). Default 5s.
+    /// </summary>
+    public TimeSpan LivenessPollInterval { get; set; } = TimeSpan.FromSeconds(5);
 }

src/RustPlusBot.Features.Connections/Listening/IRustServerConnection.cs

@@ -5,6 +5,13 @@ namespace RustPlusBot.Features.Connections.Listening;
 /// <summary>One live Rust+ socket to a single server, driven by one player credential.</summary>
 internal interface IRustServerConnection : IAsyncDisposable
 {
+    /// <summary>
+    /// Whether the underlying socket is currently open. Cheap and synchronous: the Rust+ library raises no
+    /// event when the <em>server</em> 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.
+    /// </summary>
+    bool IsConnected { get; }
+
     /// <summary>Connects within <paramref name="timeout"/>.</summary>
     /// <param name="timeout">How long to wait for the connection.</param>
     /// <param name="cancellationToken">A cancellation token.</param>

src/RustPlusBot.Features.Connections/Listening/RustPlusSocketSource.cs

@@ -34,6 +34,8 @@ public IRustServerConnection Create(string ip, int port, ulong steamId, string p
     /// <summary>Returned when the player token is unusable; reports the credential as rejected and does nothing else.</summary>
     private sealed class RejectedConnection : IRustServerConnection
     {
+        public bool IsConnected => false;
+
         public Task<SocketConnectOutcome> ConnectAsync(TimeSpan timeout, CancellationToken cancellationToken) =>
             Task.FromResult(SocketConnectOutcome.AuthRejected);
 
@@ -152,6 +154,11 @@ private sealed partial class RustPlusServerConnection : IRustServerConnection
         private readonly ILogger _logger;
         private readonly RustPlus _rustPlus;
 
+        /// <inheritdoc />
+        // 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 RustPlusServerConnection(string ip,
             int port,
             ulong steamId,

src/RustPlusBot.Features.Connections/Supervisor/ConnectionSupervisor.cs

@@ -611,17 +611,25 @@ void OnClanChanged(object? sender, ClanProbeResult probe)
             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 +646,34 @@ void OnClanChanged(object? sender, ClanProbeResult probe)
         }
     }
 
+    /// <summary>
+    /// Polls <see cref="IRustServerConnection.IsConnected"/> while connected and returns
+    /// <see cref="ReconnectReason.Unreachable"/> 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.
+    /// </summary>
+    /// <param name="key">The (guild, server) routing key.</param>
+    /// <param name="connection">The live connection whose liveness is watched.</param>
+    /// <param name="ct">Cancels when the connected window ends.</param>
+    /// <returns><see cref="ReconnectReason.Unreachable"/> on a detected drop, else
+    /// <see cref="ReconnectReason.Stopped"/> when cancelled.</returns>
+    private async Task<ReconnectReason> WatchLivenessAsync(
+        (ulong Guild, Guid Server) key,
+        IRustServerConnection connection,
+        CancellationToken ct)
+    {
+        while (!ct.IsCancellationRequested)
+        {
+            await Task.Delay(_options.LivenessPollInterval, ct).ConfigureAwait(false);
+            if (!connection.IsConnected)
+            {
+                LogSocketDropped(logger, key.Server);
+                return ReconnectReason.Unreachable;
+            }
+        }
+
+        return ReconnectReason.Stopped;
+    }
+
     private async Task<ReconnectReason> RunHeartbeatLoopAsync(
         (ulong Guild, Guid Server) key,
         IRustServerConnection connection,
@@ -1442,6 +1478,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.")]

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<ConnectionSecurity>();
         services.AddSingleton<ConnectionSupervisor>();
@@ -258,6 +259,41 @@ public async Task Connect_MonumentsTimeout_DoesNotKillLoop_AndStillReconnects()
         Assert.True(source.CreateCount >= 2);
     }
 
+    /// <summary>
+    /// The Rust+ library raises no event when the SERVER closes the socket; only
+    /// <see cref="IRustServerConnection.IsConnected"/> 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.
+    /// </summary>
+    [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");
+    }
+
     /// <summary>
     /// 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

tests/RustPlusBot.Features.Connections.Tests/Fakes/FakeRustSocketSource.cs

@@ -195,6 +195,10 @@ internal sealed class FakeConnection(SocketConnectOutcome outcome, FakeRustSocke
         private IReadOnlyList<MapMarkerSnapshot> _lastMarkers = [];
         private bool _markerScriptStarted;
 
+        /// <summary>Liveness flag consulted by the supervisor's watchdog. Set false to simulate a
+        /// server-initiated socket close (which the real library signals via <c>IsConnected</c>, not an event).</summary>
+        public bool IsConnected { get; set; } = true;
+
         /// <summary>Gets the messages sent via <see cref="SendTeamMessageAsync"/>.</summary>
         public List<string> SentMessages { get; } = [];