fix: stop one transient failure from permanently killing an event consumer

src/RustPlusBot.Abstractions/Connections/IRustServerQuery.cs

@@ -53,11 +53,12 @@ public interface IRustServerQuery
     /// <returns>The world snapshot, or null.</returns>
     Task<WorldSnapshot?> GetWorldAsync(ulong guildId, Guid serverId, CancellationToken cancellationToken);
 
-    /// <summary>Gets the server's monuments, or an empty list when there is no live socket.</summary>
+    /// <summary>Gets the server's monuments, or an empty list when there is no live socket or the fetch
+    /// fails; never throws for a failed fetch.</summary>
     /// <param name="guildId">The owning guild snowflake.</param>
     /// <param name="serverId">The target server id.</param>
     /// <param name="cancellationToken">A cancellation token.</param>
-    /// <returns>The monuments, or an empty list when there is no live socket.</returns>
+    /// <returns>The monuments, or an empty list when there is no live socket or the fetch fails/times out.</returns>
     Task<IReadOnlyList<MonumentSnapshot>> GetMonumentsAsync(
         ulong guildId,
         Guid serverId,

src/RustPlusBot.Abstractions/Events/EventBusConsumption.cs

@@ -0,0 +1,55 @@
+namespace RustPlusBot.Abstractions.Events;
+
+/// <summary>
+/// Consumption helpers for <see cref="IEventBus"/> subscribers.
+/// </summary>
+/// <remarks>
+/// Every long-running consumer in this solution wraps its <c>await foreach</c> in a broad catch that logs
+/// and then <em>exits</em>: an exception escaping a handler ends the subscription for the rest of the
+/// process, and the feature it drives goes silently dead until the host restarts. Handlers here talk to
+/// Discord and to the Rust+ server, where a timeout or a 5xx is routine rather than exceptional, so that
+/// trade is never the one we want. Consuming through <see cref="ConsumeAsync{TEvent}"/> keeps the
+/// subscription alive: a failed handler costs its own event and nothing more.
+/// </remarks>
+public static class EventBusConsumption
+{
+    /// <summary>
+    /// Consumes every published <typeparamref name="TEvent"/>, reporting and skipping the ones whose
+    /// handler throws. Only cancellation of <paramref name="cancellationToken"/> ends the subscription.
+    /// </summary>
+    /// <typeparam name="TEvent">The event type to consume.</typeparam>
+    /// <param name="eventBus">The bus to subscribe to.</param>
+    /// <param name="handle">Handles one event; its failures are reported, not propagated.</param>
+    /// <param name="onHandlerFailure">Reports a handler failure (typically a log call).</param>
+    /// <param name="cancellationToken">Ends the subscription when cancelled.</param>
+    /// <returns>A task that completes when the subscription ends.</returns>
+    public static async Task ConsumeAsync<TEvent>(
+        this IEventBus eventBus,
+        Func<TEvent, CancellationToken, Task> handle,
+        Action<Exception> onHandlerFailure,
+        CancellationToken cancellationToken)
+        where TEvent : notnull
+    {
+        ArgumentNullException.ThrowIfNull(eventBus);
+        ArgumentNullException.ThrowIfNull(handle);
+        ArgumentNullException.ThrowIfNull(onHandlerFailure);
+
+        await foreach (var evt in eventBus.SubscribeAsync<TEvent>(cancellationToken).ConfigureAwait(false))
+        {
+            try
+            {
+                await handle(evt, cancellationToken).ConfigureAwait(false);
+            }
+            catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested)
+            {
+                throw; // A real shutdown: let the caller's loop handler end it.
+            }
+#pragma warning disable CA1031 // Broad catch: a handler failure must cost one event, never the subscription.
+            catch (Exception ex)
+#pragma warning restore CA1031
+            {
+                onHandlerFailure(ex);
+            }
+        }
+    }
+}

src/RustPlusBot.Features.Alarms/Hosting/AlarmsHostedService.cs

@@ -68,11 +68,9 @@ private async Task ConsumePairedAsync(CancellationToken cancellationToken)
     {
         try
         {
-            await foreach (var evt in eventBus.SubscribeAsync<AlarmPairedEvent>(cancellationToken)
-                               .ConfigureAwait(false))
-            {
-                await coordinator.HandlePairedAsync(evt, cancellationToken).ConfigureAwait(false);
-            }
+            await eventBus.ConsumeAsync<AlarmPairedEvent>(coordinator.HandlePairedAsync,
+                    ex => LogHandlerFailed(logger, ex, nameof(AlarmPairedEvent)), cancellationToken)
+                .ConfigureAwait(false);
         }
         catch (OperationCanceledException)
         {
@@ -90,11 +88,9 @@ private async Task ConsumeTriggeredAsync(CancellationToken cancellationToken)
     {
         try
         {
-            await foreach (var evt in eventBus.SubscribeAsync<SmartDeviceTriggeredEvent>(cancellationToken)
-                               .ConfigureAwait(false))
-            {
-                await relay.HandleTriggeredAsync(evt, cancellationToken).ConfigureAwait(false);
-            }
+            await eventBus.ConsumeAsync<SmartDeviceTriggeredEvent>(relay.HandleTriggeredAsync,
+                    ex => LogHandlerFailed(logger, ex, nameof(SmartDeviceTriggeredEvent)), cancellationToken)
+                .ConfigureAwait(false);
         }
         catch (OperationCanceledException)
         {
@@ -112,11 +108,9 @@ private async Task ConsumeStatusAsync(CancellationToken cancellationToken)
     {
         try
         {
-            await foreach (var evt in eventBus.SubscribeAsync<ConnectionStatusChangedEvent>(cancellationToken)
-                               .ConfigureAwait(false))
-            {
-                await relay.HandleConnectionStatusAsync(evt, cancellationToken).ConfigureAwait(false);
-            }
+            await eventBus.ConsumeAsync<ConnectionStatusChangedEvent>(relay.HandleConnectionStatusAsync,
+                    ex => LogHandlerFailed(logger, ex, nameof(ConnectionStatusChangedEvent)), cancellationToken)
+                .ConfigureAwait(false);
         }
         catch (OperationCanceledException)
         {
@@ -134,11 +128,9 @@ private async Task ConsumeReachabilityChangedAsync(CancellationToken cancellatio
     {
         try
         {
-            await foreach (var evt in eventBus.SubscribeAsync<DeviceReachabilityChangedEvent>(cancellationToken)
-                               .ConfigureAwait(false))
-            {
-                await relay.HandleReachabilityChangedAsync(evt, cancellationToken).ConfigureAwait(false);
-            }
+            await eventBus.ConsumeAsync<DeviceReachabilityChangedEvent>(relay.HandleReachabilityChangedAsync,
+                    ex => LogHandlerFailed(logger, ex, nameof(DeviceReachabilityChangedEvent)), cancellationToken)
+                .ConfigureAwait(false);
         }
         catch (OperationCanceledException)
         {
@@ -156,11 +148,9 @@ private async Task ConsumeStateObservedAsync(CancellationToken cancellationToken
     {
         try
         {
-            await foreach (var evt in eventBus.SubscribeAsync<SmartDeviceStateObservedEvent>(cancellationToken)
-                               .ConfigureAwait(false))
-            {
-                await relay.HandleStateObservedAsync(evt, cancellationToken).ConfigureAwait(false);
-            }
+            await eventBus.ConsumeAsync<SmartDeviceStateObservedEvent>(relay.HandleStateObservedAsync,
+                    ex => LogHandlerFailed(logger, ex, nameof(SmartDeviceStateObservedEvent)), cancellationToken)
+                .ConfigureAwait(false);
         }
         catch (OperationCanceledException)
         {
@@ -178,11 +168,9 @@ private async Task ConsumeWipedAsync(CancellationToken cancellationToken)
     {
         try
         {
-            await foreach (var evt in eventBus.SubscribeAsync<ServerWipedEvent>(cancellationToken)
-                               .ConfigureAwait(false))
-            {
-                await purger.HandleServerWipedAsync(evt, cancellationToken).ConfigureAwait(false);
-            }
+            await eventBus.ConsumeAsync<ServerWipedEvent>(purger.HandleServerWipedAsync,
+                    ex => LogHandlerFailed(logger, ex, nameof(ServerWipedEvent)), cancellationToken)
+                .ConfigureAwait(false);
         }
         catch (OperationCanceledException)
         {
@@ -196,6 +184,9 @@ private async Task ConsumeWipedAsync(CancellationToken cancellationToken)
         }
     }
 
+    [LoggerMessage(Level = LogLevel.Error, Message = "Handling {EventType} failed; skipping that event.")]
+    private static partial void LogHandlerFailed(ILogger logger, Exception exception, string eventType);
+
     [LoggerMessage(Level = LogLevel.Error, Message = "Alarm pairing loop faulted.")]
     private static partial void LogPairedLoopFaulted(ILogger logger, Exception exception);
 

src/RustPlusBot.Features.Chat/Hosting/ChatHostedService.cs

@@ -76,13 +76,12 @@ private async Task ConsumeTeamMessagesAsync(CancellationToken cancellationToken)
     {
         try
         {
-            await foreach (var evt in eventBus.SubscribeAsync<TeamMessageReceivedEvent>(cancellationToken)
-                               .ConfigureAwait(false))
-            {
-                await relay.RelayAsync(
-                    new RelayedChatLine(ChatChannelKind.Team, evt.GuildId, evt.ServerId, evt.SenderName,
-                        evt.Message, evt.FromActivePlayer), cancellationToken).ConfigureAwait(false);
-            }
+            await eventBus.ConsumeAsync<TeamMessageReceivedEvent>(
+                    (evt, ct) => relay.RelayAsync(
+                        new RelayedChatLine(ChatChannelKind.Team, evt.GuildId, evt.ServerId, evt.SenderName,
+                            evt.Message, evt.FromActivePlayer), ct),
+                    ex => LogHandlerFailed(logger, ex, nameof(TeamMessageReceivedEvent)), cancellationToken)
+                .ConfigureAwait(false);
         }
         catch (OperationCanceledException)
         {
@@ -100,33 +99,9 @@ private async Task ConsumeClanMessagesAsync(CancellationToken cancellationToken)
     {
         try
         {
-            await foreach (var evt in eventBus.SubscribeAsync<ClanMessageReceivedEvent>(cancellationToken)
-                               .ConfigureAwait(false))
-            {
-                // Clan members arrive as Steam ids only; chat is where we learn their names. A failure here
-                // is isolated so it cannot take down the relay loop below: a missed name is cosmetic, a
-                // dead relay loop silences clan chat until the process restarts.
-                try
-                {
-                    var scope = scopeFactory.CreateAsyncScope();
-                    await using (scope.ConfigureAwait(false))
-                    {
-                        var store = scope.ServiceProvider.GetRequiredService<IClanStore>();
-                        await store.RecordNameAsync(evt.GuildId, evt.ServerId, evt.SenderSteamId, evt.SenderName,
-                            cancellationToken).ConfigureAwait(false);
-                    }
-                }
-#pragma warning disable CA1031 // Broad catch: a name-recording failure must not kill the clan relay loop.
-                catch (Exception ex)
-#pragma warning restore CA1031
-                {
-                    LogClanNameRecordFailed(logger, ex);
-                }
-
-                await relay.RelayAsync(
-                    new RelayedChatLine(ChatChannelKind.Clan, evt.GuildId, evt.ServerId, evt.SenderName,
-                        evt.Message, evt.FromActivePlayer), cancellationToken).ConfigureAwait(false);
-            }
+            await eventBus.ConsumeAsync<ClanMessageReceivedEvent>(HandleClanMessageAsync,
+                    ex => LogHandlerFailed(logger, ex, nameof(ClanMessageReceivedEvent)), cancellationToken)
+                .ConfigureAwait(false);
         }
         catch (OperationCanceledException)
         {
@@ -140,6 +115,32 @@ await relay.RelayAsync(
         }
     }
 
+    private async Task HandleClanMessageAsync(ClanMessageReceivedEvent evt, CancellationToken cancellationToken)
+    {
+        // Clan members arrive as Steam ids only; chat is where we learn their names. A failure here is
+        // isolated so it cannot cost the relay below: a missed name is cosmetic, a dropped clan line is not.
+        try
+        {
+            var scope = scopeFactory.CreateAsyncScope();
+            await using (scope.ConfigureAwait(false))
+            {
+                var store = scope.ServiceProvider.GetRequiredService<IClanStore>();
+                await store.RecordNameAsync(evt.GuildId, evt.ServerId, evt.SenderSteamId, evt.SenderName,
+                    cancellationToken).ConfigureAwait(false);
+            }
+        }
+#pragma warning disable CA1031 // Broad catch: a name-recording failure must not cost the relay.
+        catch (Exception ex)
+#pragma warning restore CA1031
+        {
+            LogClanNameRecordFailed(logger, ex);
+        }
+
+        await relay.RelayAsync(
+            new RelayedChatLine(ChatChannelKind.Clan, evt.GuildId, evt.ServerId, evt.SenderName,
+                evt.Message, evt.FromActivePlayer), cancellationToken).ConfigureAwait(false);
+    }
+
     private async Task OnMessageReceivedAsync(SocketMessage message)
     {
         try
@@ -169,6 +170,9 @@ private async Task OnMessageReceivedAsync(SocketMessage message)
         }
     }
 
+    [LoggerMessage(Level = LogLevel.Error, Message = "Handling {EventType} failed; skipping that event.")]
+    private static partial void LogHandlerFailed(ILogger logger, Exception exception, string eventType);
+
     [LoggerMessage(Level = LogLevel.Error, Message = "Team message relay loop faulted.")]
     private static partial void LogTeamRelayLoopFaulted(ILogger logger, Exception exception);
 

src/RustPlusBot.Features.Connections/Hosting/ConnectionHostedService.cs

@@ -78,12 +78,10 @@ private async Task ConsumeRegisteredAsync(CancellationToken cancellationToken)
     {
         try
         {
-            await foreach (var registered in eventBus.SubscribeAsync<ServerRegisteredEvent>(cancellationToken)
-                               .ConfigureAwait(false))
-            {
-                await supervisor.EnsureConnectionAsync(registered.GuildId, registered.ServerId, cancellationToken)
-                    .ConfigureAwait(false);
-            }
+            await eventBus.ConsumeAsync<ServerRegisteredEvent>(
+                    (registered, ct) => supervisor.EnsureConnectionAsync(registered.GuildId, registered.ServerId, ct),
+                    ex => LogHandlerFailed(logger, ex, nameof(ServerRegisteredEvent)), cancellationToken)
+                .ConfigureAwait(false);
         }
         catch (OperationCanceledException)
         {
@@ -101,12 +99,10 @@ private async Task ConsumeCredentialsChangedAsync(CancellationToken cancellation
     {
         try
         {
-            await foreach (var changed in eventBus.SubscribeAsync<ServerCredentialsChangedEvent>(cancellationToken)
-                               .ConfigureAwait(false))
-            {
-                await supervisor.EnsureConnectionAsync(changed.GuildId, changed.ServerId, cancellationToken)
-                    .ConfigureAwait(false);
-            }
+            await eventBus.ConsumeAsync<ServerCredentialsChangedEvent>(
+                    (changed, ct) => supervisor.EnsureConnectionAsync(changed.GuildId, changed.ServerId, ct),
+                    ex => LogHandlerFailed(logger, ex, nameof(ServerCredentialsChangedEvent)), cancellationToken)
+                .ConfigureAwait(false);
         }
         catch (OperationCanceledException)
         {
@@ -120,6 +116,9 @@ await supervisor.EnsureConnectionAsync(changed.GuildId, changed.ServerId, cancel
 #pragma warning restore CA1031
     }
 
+    [LoggerMessage(Level = LogLevel.Error, Message = "Handling {EventType} failed; skipping that event.")]
+    private static partial void LogHandlerFailed(ILogger logger, Exception exception, string eventType);
+
     [LoggerMessage(Level = LogLevel.Error, Message = "Connection hosted-service loop faulted.")]
     private static partial void LogLoopFaulted(ILogger logger, Exception exception);
 }

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

@@ -714,7 +714,11 @@ public async Task<IReadOnlyList<MonumentSnapshot>> GetMonumentsAsync(
                 .ConfigureAwait(false);
             if (!response.IsSuccess || response.Data is null)
             {
-                throw new InvalidOperationException("GetMap returned no data.");
+                // Name the reason: the caller only ever sees this message, and "rate_limit" (three heavy
+                // GetMap calls land back-to-back on connect) reads very differently from "no_map".
+                throw new InvalidOperationException(
+                    "GetMap returned no data; error: " + (response.Error?.Code.ToString() ?? "none") +
+                    " (" + (response.Error?.Message ?? "no message") + ").");
             }
 
             var monuments = new List<MonumentSnapshot>();

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

@@ -311,8 +311,25 @@ public async Task<IReadOnlyList<MonumentSnapshot>> GetMonumentsAsync(
             return [];
         }
 
-        return await live.Connection.GetMonumentsAsync(_options.HeartbeatTimeout, cancellationToken)
-            .ConfigureAwait(false);
+        try
+        {
+            return await live.Connection.GetMonumentsAsync(_options.HeartbeatTimeout, cancellationToken)
+                .ConfigureAwait(false);
+        }
+        catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested)
+        {
+            throw;
+        }
+#pragma warning disable CA1031 // Broad catch: this seam promises degradation, so a failed fetch is "no monuments".
+        catch (Exception ex)
+#pragma warning restore CA1031
+        {
+            // The connection-level call throws on a failed/timed-out GetMap (rate limit, no map, slow
+            // endpoint). Callers here are render paths that must degrade to an icon-less map, never fault:
+            // an escaping exception tears down the consuming loop for the rest of the process.
+            LogMonumentsQueryFailed(logger, ex, serverId);
+            return [];
+        }
     }
 
     /// <inheritdoc />
@@ -1498,6 +1515,10 @@ await eventBus.PublishAsync(
             "Fetching monuments for oil-rig detection on server {ServerId} failed; rig detection disabled for this connection.")]
     private static partial void LogMonumentsFetchFailed(ILogger logger, Exception exception, Guid serverId);
 
+    [LoggerMessage(Level = LogLevel.Warning,
+        Message = "Querying monuments for server {ServerId} failed; returning no monuments for this call.")]
+    private static partial void LogMonumentsQueryFailed(ILogger logger, Exception exception, Guid serverId);
+
     [LoggerMessage(Level = LogLevel.Warning, Message = "Relaying a message to team chat for server {ServerId} failed.")]
     private static partial void LogSendFailed(ILogger logger, Exception exception, Guid serverId);