From 0c4d732ee1113d3104d7da78419437da732bb23a Mon Sep 17 00:00:00 2001 From: = Date: Fri, 24 Jul 2026 04:36:51 +0200 Subject: [PATCH 1/3] fix: keep event consumers alive across transient handler failures MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Three faults observed live tonight, all the same class: an exception escaping into a consumer's await-foreach ends that subscription for the rest of the process, so the feature it drives goes silently dead until the bot restarts. 1. "Map connection-status loop faulted. InvalidOperationException: GetMap returned no data." — ConnectionSupervisor.GetMonumentsAsync forwarded the connection-level call (documented "Throws on failure") straight through a query seam whose own contract promised "or an empty list". It was the sole query in RustPlusSocketSource without a degrade-to-null/empty catch. It now catches, logs and returns [], and the failure message names the Rust+ error code instead of a bare "returned no data". 2. "ConnectionStatusChanged consumer faulted. TimeoutException" from a Discord REST edit — the Workspace consumer died, leaving the info channels stale. 3. The same shape existed, unguarded, in 21 consumer loops across 11 services. Add EventBusConsumption.ConsumeAsync(handler, onHandlerFailure, ct): it subscribes, guards each handler call, reports failures and keeps consuming; only cancellation of the loop token ends the subscription. Abstractions stays dependency-free (the callback is Action, so each service keeps its own logger). Every vulnerable consumer now routes through it; each service's outer try/catch stays as a backstop. ClansHostedService and CommandsHostedService already isolate per event inline and are left alone — migrating them would lose the guild/server ids in their failure logs. Also fixes the unrelated log line that started this: Rust+ sends the token "apartmentcomplex", which no entry in MonumentTokenMap mapped, so apartment complexes rendered with no icon ("No monument icon for token ...; mapped type: null"). MonumentType.ApartmentsComplex and Apartments_Complex.svg were already in RustMapsApi.Assets — only the token table had not caught up. Tests: EventBusConsumptionTests (a failing handler costs its own event only; cancellation still ends the subscription), WorkspaceConsumerResilienceTests and MapHostedServiceTests (both written failing first, reproducing the live exceptions), plus a monuments-seam and a token-map case. Co-Authored-By: Claude Opus 4.8 (1M context) --- .../Connections/IRustServerQuery.cs | 5 +- .../Events/EventBusConsumption.cs | 55 ++++++++++ .../Hosting/AlarmsHostedService.cs | 51 ++++----- .../Hosting/ChatHostedService.cs | 72 ++++++------ .../Hosting/ConnectionHostedService.cs | 23 ++-- .../Listening/RustPlusSocketSource.cs | 6 +- .../Supervisor/ConnectionSupervisor.cs | 21 +++- .../Hosting/EventsHostedService.cs | 27 ++--- .../Assets/MonumentTokenMap.cs | 1 + .../Hosting/InfoMapHostedService.cs | 11 +- .../Hosting/MapHostedService.cs | 68 +++++++++--- .../Hosting/PlayersHostedService.cs | 11 +- .../Hosting/StorageMonitorsHostedService.cs | 43 +++----- .../Hosting/SwitchesHostedService.cs | 51 ++++----- .../Hosting/WipesHostedService.cs | 27 ++--- .../Hosting/WorkspaceHostedService.cs | 88 +++++++-------- .../Events/EventBusConsumptionTests.cs | 70 ++++++++++++ .../Fakes/FakeRustSocketSource.cs | 16 ++- .../ServerQueryTests.cs | 21 ++++ .../Hosting/MapHostedServiceTests.cs | 103 ++++++++++++++++++ .../MonumentTokenMapTests.cs | 1 + .../WorkspaceConsumerResilienceTests.cs | 67 ++++++++++++ 22 files changed, 594 insertions(+), 244 deletions(-) create mode 100644 src/RustPlusBot.Abstractions/Events/EventBusConsumption.cs create mode 100644 tests/RustPlusBot.Abstractions.Tests/Events/EventBusConsumptionTests.cs create mode 100644 tests/RustPlusBot.Features.Map.Tests/Hosting/MapHostedServiceTests.cs create mode 100644 tests/RustPlusBot.Features.Workspace.Tests/Hosting/WorkspaceConsumerResilienceTests.cs diff --git a/src/RustPlusBot.Abstractions/Connections/IRustServerQuery.cs b/src/RustPlusBot.Abstractions/Connections/IRustServerQuery.cs index 27fa751e..3c574816 100644 --- a/src/RustPlusBot.Abstractions/Connections/IRustServerQuery.cs +++ b/src/RustPlusBot.Abstractions/Connections/IRustServerQuery.cs @@ -53,11 +53,12 @@ public interface IRustServerQuery /// The world snapshot, or null. Task GetWorldAsync(ulong guildId, Guid serverId, CancellationToken cancellationToken); - /// Gets the server's monuments, or an empty list when there is no live socket. + /// 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. /// The owning guild snowflake. /// The target server id. /// A cancellation token. - /// The monuments, or an empty list when there is no live socket. + /// The monuments, or an empty list when there is no live socket or the fetch fails/times out. Task> GetMonumentsAsync( ulong guildId, Guid serverId, diff --git a/src/RustPlusBot.Abstractions/Events/EventBusConsumption.cs b/src/RustPlusBot.Abstractions/Events/EventBusConsumption.cs new file mode 100644 index 00000000..5be073d5 --- /dev/null +++ b/src/RustPlusBot.Abstractions/Events/EventBusConsumption.cs @@ -0,0 +1,55 @@ +namespace RustPlusBot.Abstractions.Events; + +/// +/// Consumption helpers for subscribers. +/// +/// +/// Every long-running consumer in this solution wraps its await foreach in a broad catch that logs +/// and then exits: 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 keeps the +/// subscription alive: a failed handler costs its own event and nothing more. +/// +public static class EventBusConsumption +{ + /// + /// Consumes every published , reporting and skipping the ones whose + /// handler throws. Only cancellation of ends the subscription. + /// + /// The event type to consume. + /// The bus to subscribe to. + /// Handles one event; its failures are reported, not propagated. + /// Reports a handler failure (typically a log call). + /// Ends the subscription when cancelled. + /// A task that completes when the subscription ends. + public static async Task ConsumeAsync( + this IEventBus eventBus, + Func handle, + Action onHandlerFailure, + CancellationToken cancellationToken) + where TEvent : notnull + { + ArgumentNullException.ThrowIfNull(eventBus); + ArgumentNullException.ThrowIfNull(handle); + ArgumentNullException.ThrowIfNull(onHandlerFailure); + + await foreach (var evt in eventBus.SubscribeAsync(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); + } + } + } +} diff --git a/src/RustPlusBot.Features.Alarms/Hosting/AlarmsHostedService.cs b/src/RustPlusBot.Features.Alarms/Hosting/AlarmsHostedService.cs index 15fc6629..54d822d1 100644 --- a/src/RustPlusBot.Features.Alarms/Hosting/AlarmsHostedService.cs +++ b/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(cancellationToken) - .ConfigureAwait(false)) - { - await coordinator.HandlePairedAsync(evt, cancellationToken).ConfigureAwait(false); - } + await eventBus.ConsumeAsync(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(cancellationToken) - .ConfigureAwait(false)) - { - await relay.HandleTriggeredAsync(evt, cancellationToken).ConfigureAwait(false); - } + await eventBus.ConsumeAsync(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(cancellationToken) - .ConfigureAwait(false)) - { - await relay.HandleConnectionStatusAsync(evt, cancellationToken).ConfigureAwait(false); - } + await eventBus.ConsumeAsync(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(cancellationToken) - .ConfigureAwait(false)) - { - await relay.HandleReachabilityChangedAsync(evt, cancellationToken).ConfigureAwait(false); - } + await eventBus.ConsumeAsync(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(cancellationToken) - .ConfigureAwait(false)) - { - await relay.HandleStateObservedAsync(evt, cancellationToken).ConfigureAwait(false); - } + await eventBus.ConsumeAsync(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(cancellationToken) - .ConfigureAwait(false)) - { - await purger.HandleServerWipedAsync(evt, cancellationToken).ConfigureAwait(false); - } + await eventBus.ConsumeAsync(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); diff --git a/src/RustPlusBot.Features.Chat/Hosting/ChatHostedService.cs b/src/RustPlusBot.Features.Chat/Hosting/ChatHostedService.cs index 04ad9969..b2c150f0 100644 --- a/src/RustPlusBot.Features.Chat/Hosting/ChatHostedService.cs +++ b/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(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( + (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(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(); - 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(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(); + 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); diff --git a/src/RustPlusBot.Features.Connections/Hosting/ConnectionHostedService.cs b/src/RustPlusBot.Features.Connections/Hosting/ConnectionHostedService.cs index e4f2c3b9..a80bc667 100644 --- a/src/RustPlusBot.Features.Connections/Hosting/ConnectionHostedService.cs +++ b/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(cancellationToken) - .ConfigureAwait(false)) - { - await supervisor.EnsureConnectionAsync(registered.GuildId, registered.ServerId, cancellationToken) - .ConfigureAwait(false); - } + await eventBus.ConsumeAsync( + (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(cancellationToken) - .ConfigureAwait(false)) - { - await supervisor.EnsureConnectionAsync(changed.GuildId, changed.ServerId, cancellationToken) - .ConfigureAwait(false); - } + await eventBus.ConsumeAsync( + (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); } diff --git a/src/RustPlusBot.Features.Connections/Listening/RustPlusSocketSource.cs b/src/RustPlusBot.Features.Connections/Listening/RustPlusSocketSource.cs index 0dc59c44..47f2b35b 100644 --- a/src/RustPlusBot.Features.Connections/Listening/RustPlusSocketSource.cs +++ b/src/RustPlusBot.Features.Connections/Listening/RustPlusSocketSource.cs @@ -714,7 +714,11 @@ public async Task> 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(); diff --git a/src/RustPlusBot.Features.Connections/Supervisor/ConnectionSupervisor.cs b/src/RustPlusBot.Features.Connections/Supervisor/ConnectionSupervisor.cs index 80f78185..ade23d55 100644 --- a/src/RustPlusBot.Features.Connections/Supervisor/ConnectionSupervisor.cs +++ b/src/RustPlusBot.Features.Connections/Supervisor/ConnectionSupervisor.cs @@ -311,8 +311,25 @@ public async Task> 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. + LogMonumentsFetchFailed(logger, ex, serverId); + return []; + } } /// diff --git a/src/RustPlusBot.Features.Events/Hosting/EventsHostedService.cs b/src/RustPlusBot.Features.Events/Hosting/EventsHostedService.cs index 06c919a8..22aa08b8 100644 --- a/src/RustPlusBot.Features.Events/Hosting/EventsHostedService.cs +++ b/src/RustPlusBot.Features.Events/Hosting/EventsHostedService.cs @@ -115,11 +115,9 @@ private async Task ConsumeMarkerEventsAsync(CancellationToken cancellationToken) { try { - await foreach (var evt in eventBus.SubscribeAsync(cancellationToken) - .ConfigureAwait(false)) - { - await relay.RelayAsync(evt, cancellationToken).ConfigureAwait(false); - } + await eventBus.ConsumeAsync(relay.RelayAsync, + ex => LogHandlerFailed(logger, ex, nameof(MapMarkersChangedEvent)), cancellationToken) + .ConfigureAwait(false); } catch (OperationCanceledException) { @@ -137,11 +135,9 @@ private async Task ConsumeRigEventsAsync(CancellationToken cancellationToken) { try { - await foreach (var evt in eventBus.SubscribeAsync(cancellationToken) - .ConfigureAwait(false)) - { - await relay.RelayRigAsync(evt, cancellationToken).ConfigureAwait(false); - } + await eventBus.ConsumeAsync(relay.RelayRigAsync, + ex => LogHandlerFailed(logger, ex, nameof(RigStateChangedEvent)), cancellationToken) + .ConfigureAwait(false); } catch (OperationCanceledException) { @@ -159,11 +155,9 @@ private async Task ConsumeConnectionStatusEventsAsync(CancellationToken cancella { try { - await foreach (var evt in eventBus.SubscribeAsync(cancellationToken) - .ConfigureAwait(false)) - { - await ClearIfDisconnectedAsync(evt, cancellationToken).ConfigureAwait(false); - } + await eventBus.ConsumeAsync(ClearIfDisconnectedAsync, + ex => LogHandlerFailed(logger, ex, nameof(ConnectionStatusChangedEvent)), cancellationToken) + .ConfigureAwait(false); } catch (OperationCanceledException) { @@ -193,6 +187,9 @@ private async Task ClearIfDisconnectedAsync(ConnectionStatusChangedEvent evt, Ca } } + [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 = "Event relay loop faulted.")] private static partial void LogRelayLoopFaulted(ILogger logger, Exception exception); diff --git a/src/RustPlusBot.Features.Map/Assets/MonumentTokenMap.cs b/src/RustPlusBot.Features.Map/Assets/MonumentTokenMap.cs index 3985e9d0..2116e652 100644 --- a/src/RustPlusBot.Features.Map/Assets/MonumentTokenMap.cs +++ b/src/RustPlusBot.Features.Map/Assets/MonumentTokenMap.cs @@ -19,6 +19,7 @@ public static class MonumentTokenMap { ["AbandonedMilitaryBase"] = MonumentType.MilitaryBaseA, ["airfield_display_name"] = MonumentType.Airfield, + ["apartmentcomplex"] = MonumentType.ApartmentsComplex, ["arctic_base_a"] = MonumentType.ArcticResearchBaseA, ["arctic_base_b"] = MonumentType.ArcticResearchBaseA, ["bandit_camp"] = MonumentType.BanditTown, diff --git a/src/RustPlusBot.Features.Map/Hosting/InfoMapHostedService.cs b/src/RustPlusBot.Features.Map/Hosting/InfoMapHostedService.cs index 51e5990d..ac3fb976 100644 --- a/src/RustPlusBot.Features.Map/Hosting/InfoMapHostedService.cs +++ b/src/RustPlusBot.Features.Map/Hosting/InfoMapHostedService.cs @@ -163,11 +163,9 @@ private async Task ConsumeConnectionStatusAsync(CancellationToken cancellationTo { try { - await foreach (var evt in eventBus.SubscribeAsync(cancellationToken) - .ConfigureAwait(false)) - { - await OnConnectionStatusAsync(evt, cancellationToken).ConfigureAwait(false); - } + await eventBus.ConsumeAsync(OnConnectionStatusAsync, + ex => LogHandlerFailed(logger, ex, nameof(ConnectionStatusChangedEvent)), cancellationToken) + .ConfigureAwait(false); } catch (OperationCanceledException) { @@ -209,6 +207,9 @@ private async Task OnConnectionStatusAsync(ConnectionStatusChangedEvent evt, Can coordinator.Register(new RustMapsMapKey((int)world.WorldSize, (int)world.Seed), evt.GuildId, evt.ServerId); } + [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 = "Info-map tick loop faulted.")] private static partial void LogTickFaulted(ILogger logger, Exception exception); diff --git a/src/RustPlusBot.Features.Map/Hosting/MapHostedService.cs b/src/RustPlusBot.Features.Map/Hosting/MapHostedService.cs index 949a0a01..9a113c71 100644 --- a/src/RustPlusBot.Features.Map/Hosting/MapHostedService.cs +++ b/src/RustPlusBot.Features.Map/Hosting/MapHostedService.cs @@ -98,11 +98,10 @@ private async Task ConsumeMarkerEventsAsync(CancellationToken cancellationToken) { try { - await foreach (var evt in eventBus.SubscribeAsync(cancellationToken) - .ConfigureAwait(false)) - { - await RefreshAsync(evt.GuildId, evt.ServerId, cancellationToken).ConfigureAwait(false); - } + await eventBus.ConsumeAsync( + (evt, ct) => RefreshAsync(evt.GuildId, evt.ServerId, ct), + ex => LogHandlerFailed(logger, ex, nameof(MapMarkersChangedEvent)), cancellationToken) + .ConfigureAwait(false); } catch (OperationCanceledException) { @@ -128,11 +127,10 @@ private async Task ConsumeSettingsEventsAsync(CancellationToken cancellationToke { try { - await foreach (var evt in eventBus.SubscribeAsync(cancellationToken) - .ConfigureAwait(false)) - { - await RefreshAsync(evt.GuildId, evt.ServerId, cancellationToken).ConfigureAwait(false); - } + await eventBus.ConsumeAsync( + (evt, ct) => RefreshAsync(evt.GuildId, evt.ServerId, ct), + ex => LogHandlerFailed(logger, ex, nameof(MapSettingsChangedEvent)), cancellationToken) + .ConfigureAwait(false); } catch (OperationCanceledException) { @@ -162,7 +160,8 @@ private async Task RunPeriodicRefreshAsync(CancellationToken cancellationToken) await Task.Delay(options.Value.MapRefreshInterval, cancellationToken).ConfigureAwait(false); foreach (var (guild, server) in _connected.Keys) { - await RefreshAsync(guild, server, cancellationToken).ConfigureAwait(false); + await RunGuardedAsync(() => RefreshAsync(guild, server, cancellationToken), + guild, server, cancellationToken).ConfigureAwait(false); } } } @@ -178,6 +177,38 @@ private async Task RunPeriodicRefreshAsync(CancellationToken cancellationToken) } } + /// + /// Runs one unit of loop work, absorbing any failure so a single bad refresh degrades that refresh + /// only. Without this, an exception escaping into a consumer's await foreach ends the + /// subscription for the rest of the process — the #map then stays frozen until the bot restarts. + /// + /// The refresh (or status handling) to run. + /// The owning guild snowflake, for the failure log. + /// The target server id, for the failure log. + /// A cancellation token; a real shutdown still propagates. + /// A task that completes when the work has run or failed. + private async Task RunGuardedAsync( + Func work, + ulong guildId, + Guid serverId, + CancellationToken cancellationToken) + { + try + { + await work().ConfigureAwait(false); + } + catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested) + { + throw; // Shutdown: let the loop's own handler end it. + } +#pragma warning disable CA1031 // Broad catch: any refresh failure must cost one repaint, never the loop. + catch (Exception ex) +#pragma warning restore CA1031 + { + LogRefreshFailed(logger, ex, guildId, serverId); + } + } + private async Task RefreshAsync(ulong guildId, Guid serverId, CancellationToken cancellationToken) { if (!_throttle.ShouldRefresh(guildId, serverId, options.Value.MapRefreshInterval)) @@ -204,11 +235,9 @@ private async Task ConsumeConnectionStatusEventsAsync(CancellationToken cancella { try { - await foreach (var evt in eventBus.SubscribeAsync(cancellationToken) - .ConfigureAwait(false)) - { - await OnConnectionStatusAsync(evt, cancellationToken).ConfigureAwait(false); - } + await eventBus.ConsumeAsync(OnConnectionStatusAsync, + ex => LogHandlerFailed(logger, ex, nameof(ConnectionStatusChangedEvent)), cancellationToken) + .ConfigureAwait(false); } catch (OperationCanceledException) { @@ -245,6 +274,13 @@ private async Task OnConnectionStatusAsync(ConnectionStatusChangedEvent evt, Can await RefreshAsync(evt.GuildId, evt.ServerId, cancellationToken).ConfigureAwait(false); } + [LoggerMessage(Level = LogLevel.Error, Message = "Handling {EventType} failed; skipping that repaint.")] + private static partial void LogHandlerFailed(ILogger logger, Exception exception, string eventType); + + [LoggerMessage(Level = LogLevel.Warning, + Message = "Refreshing the map for guild {GuildId}, server {ServerId} failed; skipping this repaint.")] + private static partial void LogRefreshFailed(ILogger logger, Exception exception, ulong guildId, Guid serverId); + [LoggerMessage(Level = LogLevel.Error, Message = "Map marker loop faulted.")] private static partial void LogMarkerLoopFaulted(ILogger logger, Exception exception); diff --git a/src/RustPlusBot.Features.Players/Hosting/PlayersHostedService.cs b/src/RustPlusBot.Features.Players/Hosting/PlayersHostedService.cs index 9de6ec17..d5054b63 100644 --- a/src/RustPlusBot.Features.Players/Hosting/PlayersHostedService.cs +++ b/src/RustPlusBot.Features.Players/Hosting/PlayersHostedService.cs @@ -50,11 +50,9 @@ private async Task ConsumeAsync(CancellationToken cancellationToken) { try { - await foreach (var evt in eventBus.SubscribeAsync(cancellationToken) - .ConfigureAwait(false)) - { - await relay.RelayAsync(evt, cancellationToken).ConfigureAwait(false); - } + await eventBus.ConsumeAsync(relay.RelayAsync, + ex => LogHandlerFailed(logger, ex, nameof(PlayerStateChangedEvent)), cancellationToken) + .ConfigureAwait(false); } catch (OperationCanceledException) { @@ -68,6 +66,9 @@ private async Task ConsumeAsync(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 = "Player relay loop faulted.")] private static partial void LogRelayLoopFaulted(ILogger logger, Exception exception); } diff --git a/src/RustPlusBot.Features.StorageMonitors/Hosting/StorageMonitorsHostedService.cs b/src/RustPlusBot.Features.StorageMonitors/Hosting/StorageMonitorsHostedService.cs index ea8d21ca..546349e8 100644 --- a/src/RustPlusBot.Features.StorageMonitors/Hosting/StorageMonitorsHostedService.cs +++ b/src/RustPlusBot.Features.StorageMonitors/Hosting/StorageMonitorsHostedService.cs @@ -66,11 +66,9 @@ private async Task ConsumePairedAsync(CancellationToken cancellationToken) { try { - await foreach (var evt in eventBus.SubscribeAsync(cancellationToken) - .ConfigureAwait(false)) - { - await coordinator.HandlePairedAsync(evt, cancellationToken).ConfigureAwait(false); - } + await eventBus.ConsumeAsync(coordinator.HandlePairedAsync, + ex => LogHandlerFailed(logger, ex, nameof(StorageMonitorPairedEvent)), cancellationToken) + .ConfigureAwait(false); } catch (OperationCanceledException) { @@ -88,11 +86,9 @@ private async Task ConsumeTriggeredAsync(CancellationToken cancellationToken) { try { - await foreach (var evt in eventBus.SubscribeAsync(cancellationToken) - .ConfigureAwait(false)) - { - await relay.HandleTriggeredAsync(evt, cancellationToken).ConfigureAwait(false); - } + await eventBus.ConsumeAsync(relay.HandleTriggeredAsync, + ex => LogHandlerFailed(logger, ex, nameof(StorageMonitorTriggeredEvent)), cancellationToken) + .ConfigureAwait(false); } catch (OperationCanceledException) { @@ -110,11 +106,9 @@ private async Task ConsumeStatusAsync(CancellationToken cancellationToken) { try { - await foreach (var evt in eventBus.SubscribeAsync(cancellationToken) - .ConfigureAwait(false)) - { - await relay.HandleConnectionStatusAsync(evt, cancellationToken).ConfigureAwait(false); - } + await eventBus.ConsumeAsync(relay.HandleConnectionStatusAsync, + ex => LogHandlerFailed(logger, ex, nameof(ConnectionStatusChangedEvent)), cancellationToken) + .ConfigureAwait(false); } catch (OperationCanceledException) { @@ -132,11 +126,9 @@ private async Task ConsumeReachabilityChangedAsync(CancellationToken cancellatio { try { - await foreach (var evt in eventBus.SubscribeAsync(cancellationToken) - .ConfigureAwait(false)) - { - await relay.HandleReachabilityChangedAsync(evt, cancellationToken).ConfigureAwait(false); - } + await eventBus.ConsumeAsync(relay.HandleReachabilityChangedAsync, + ex => LogHandlerFailed(logger, ex, nameof(DeviceReachabilityChangedEvent)), cancellationToken) + .ConfigureAwait(false); } catch (OperationCanceledException) { @@ -154,11 +146,9 @@ private async Task ConsumeWipedAsync(CancellationToken cancellationToken) { try { - await foreach (var evt in eventBus.SubscribeAsync(cancellationToken) - .ConfigureAwait(false)) - { - await purger.HandleServerWipedAsync(evt, cancellationToken).ConfigureAwait(false); - } + await eventBus.ConsumeAsync(purger.HandleServerWipedAsync, + ex => LogHandlerFailed(logger, ex, nameof(ServerWipedEvent)), cancellationToken) + .ConfigureAwait(false); } catch (OperationCanceledException) { @@ -172,6 +162,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 = "Storage monitor pairing loop faulted.")] private static partial void LogPairedLoopFaulted(ILogger logger, Exception exception); diff --git a/src/RustPlusBot.Features.Switches/Hosting/SwitchesHostedService.cs b/src/RustPlusBot.Features.Switches/Hosting/SwitchesHostedService.cs index 0ec822bb..fd0537c4 100644 --- a/src/RustPlusBot.Features.Switches/Hosting/SwitchesHostedService.cs +++ b/src/RustPlusBot.Features.Switches/Hosting/SwitchesHostedService.cs @@ -68,11 +68,9 @@ private async Task ConsumePairedAsync(CancellationToken cancellationToken) { try { - await foreach (var evt in eventBus.SubscribeAsync(cancellationToken) - .ConfigureAwait(false)) - { - await coordinator.HandlePairedAsync(evt, cancellationToken).ConfigureAwait(false); - } + await eventBus.ConsumeAsync(coordinator.HandlePairedAsync, + ex => LogHandlerFailed(logger, ex, nameof(SwitchPairedEvent)), cancellationToken) + .ConfigureAwait(false); } catch (OperationCanceledException) { @@ -90,11 +88,9 @@ private async Task ConsumeStateAsync(CancellationToken cancellationToken) { try { - await foreach (var evt in eventBus.SubscribeAsync(cancellationToken) - .ConfigureAwait(false)) - { - await relay.HandleStateChangedAsync(evt, cancellationToken).ConfigureAwait(false); - } + await eventBus.ConsumeAsync(relay.HandleStateChangedAsync, + ex => LogHandlerFailed(logger, ex, nameof(SwitchStateChangedEvent)), cancellationToken) + .ConfigureAwait(false); } catch (OperationCanceledException) { @@ -112,11 +108,9 @@ private async Task ConsumeStatusAsync(CancellationToken cancellationToken) { try { - await foreach (var evt in eventBus.SubscribeAsync(cancellationToken) - .ConfigureAwait(false)) - { - await relay.HandleConnectionStatusAsync(evt, cancellationToken).ConfigureAwait(false); - } + await eventBus.ConsumeAsync(relay.HandleConnectionStatusAsync, + ex => LogHandlerFailed(logger, ex, nameof(ConnectionStatusChangedEvent)), cancellationToken) + .ConfigureAwait(false); } catch (OperationCanceledException) { @@ -134,11 +128,9 @@ private async Task ConsumeDeviceTriggeredAsync(CancellationToken cancellationTok { try { - await foreach (var evt in eventBus.SubscribeAsync(cancellationToken) - .ConfigureAwait(false)) - { - await relay.HandleDeviceTriggeredAsync(evt, cancellationToken).ConfigureAwait(false); - } + await eventBus.ConsumeAsync(relay.HandleDeviceTriggeredAsync, + ex => LogHandlerFailed(logger, ex, nameof(SmartDeviceTriggeredEvent)), cancellationToken) + .ConfigureAwait(false); } catch (OperationCanceledException) { @@ -156,11 +148,9 @@ private async Task ConsumeReachabilityChangedAsync(CancellationToken cancellatio { try { - await foreach (var evt in eventBus.SubscribeAsync(cancellationToken) - .ConfigureAwait(false)) - { - await relay.HandleReachabilityChangedAsync(evt, cancellationToken).ConfigureAwait(false); - } + await eventBus.ConsumeAsync(relay.HandleReachabilityChangedAsync, + ex => LogHandlerFailed(logger, ex, nameof(DeviceReachabilityChangedEvent)), cancellationToken) + .ConfigureAwait(false); } catch (OperationCanceledException) { @@ -178,11 +168,9 @@ private async Task ConsumeWipedAsync(CancellationToken cancellationToken) { try { - await foreach (var evt in eventBus.SubscribeAsync(cancellationToken) - .ConfigureAwait(false)) - { - await purger.HandleServerWipedAsync(evt, cancellationToken).ConfigureAwait(false); - } + await eventBus.ConsumeAsync(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 = "Switch device-triggered relay loop faulted.")] private static partial void LogDeviceLoopFaulted(ILogger logger, Exception exception); diff --git a/src/RustPlusBot.Features.Wipes/Hosting/WipesHostedService.cs b/src/RustPlusBot.Features.Wipes/Hosting/WipesHostedService.cs index 04d27e8e..c5d217b9 100644 --- a/src/RustPlusBot.Features.Wipes/Hosting/WipesHostedService.cs +++ b/src/RustPlusBot.Features.Wipes/Hosting/WipesHostedService.cs @@ -58,16 +58,12 @@ private async Task ConsumeStatusAsync(CancellationToken cancellationToken) { try { - await foreach (var evt in eventBus.SubscribeAsync(cancellationToken) - .ConfigureAwait(false)) - { - if (!evt.IsConnected || evt.WasConnected) - { - continue; - } - - await detector.CheckAsync(evt.GuildId, evt.ServerId, cancellationToken).ConfigureAwait(false); - } + await eventBus.ConsumeAsync( + (evt, ct) => !evt.IsConnected || evt.WasConnected + ? Task.CompletedTask + : detector.CheckAsync(evt.GuildId, evt.ServerId, ct), + ex => LogHandlerFailed(logger, ex, nameof(ConnectionStatusChangedEvent)), cancellationToken) + .ConfigureAwait(false); } catch (OperationCanceledException) { @@ -85,11 +81,9 @@ private async Task ConsumeWipedAsync(CancellationToken cancellationToken) { try { - await foreach (var evt in eventBus.SubscribeAsync(cancellationToken) - .ConfigureAwait(false)) - { - await announcer.HandleServerWipedAsync(evt, cancellationToken).ConfigureAwait(false); - } + await eventBus.ConsumeAsync(announcer.HandleServerWipedAsync, + ex => LogHandlerFailed(logger, ex, nameof(ServerWipedEvent)), cancellationToken) + .ConfigureAwait(false); } catch (OperationCanceledException) { @@ -103,6 +97,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 = "Wipe connection-status loop faulted.")] private static partial void LogStatusLoopFaulted(ILogger logger, Exception exception); diff --git a/src/RustPlusBot.Features.Workspace/Hosting/WorkspaceHostedService.cs b/src/RustPlusBot.Features.Workspace/Hosting/WorkspaceHostedService.cs index e458b1ee..03ae039d 100644 --- a/src/RustPlusBot.Features.Workspace/Hosting/WorkspaceHostedService.cs +++ b/src/RustPlusBot.Features.Workspace/Hosting/WorkspaceHostedService.cs @@ -144,23 +144,35 @@ private async Task OnChannelDestroyedAsync(SocketChannel channel) } } + /// + /// Reconciles one server in its own scope. Every consumer below runs this through + /// , which absorbs its failures: the reconcile + /// talks to Discord over REST, where a timeout or a 5xx is routine, and one of those must never end + /// the subscription that drives the channels. + /// + /// The owning guild snowflake. + /// The server to reconcile. + /// A cancellation token. + /// A task that completes when the reconcile has run. + private async Task ReconcileServerAsync(ulong guildId, Guid serverId, CancellationToken cancellationToken) + { + var scope = scopeFactory.CreateAsyncScope(); + await using (scope.ConfigureAwait(false)) + { + var reconciler = scope.ServiceProvider.GetRequiredService(); + await reconciler.ReconcileServerAsync(guildId, serverId, cancellationToken).ConfigureAwait(false); + } + } + private async Task ConsumeConnectionStatusAsync(CancellationToken cancellationToken) { - // If this loop faults (broad catch), the consumer exits permanently and info channels stop - // updating until the host restarts. Acceptable: the reconciler is idempotent and a restart heals. try { - await foreach (var changed in eventBus.SubscribeAsync(cancellationToken) - .ConfigureAwait(false)) - { - var scope = scopeFactory.CreateAsyncScope(); - await using (scope.ConfigureAwait(false)) - { - var reconciler = scope.ServiceProvider.GetRequiredService(); - await reconciler.ReconcileServerAsync(changed.GuildId, changed.ServerId, cancellationToken) - .ConfigureAwait(false); - } - } + await eventBus.ConsumeAsync( + (evt, ct) => ReconcileServerAsync(evt.GuildId, evt.ServerId, ct), + ex => logger.LogError(ex, "Handling {EventType} failed; skipping that reconcile.", + nameof(ConnectionStatusChangedEvent)), + cancellationToken).ConfigureAwait(false); } catch (OperationCanceledException) { @@ -176,17 +188,11 @@ private async Task ConsumeServerCredentialsAsync(CancellationToken cancellationT { try { - await foreach (var changed in eventBus.SubscribeAsync(cancellationToken) - .ConfigureAwait(false)) - { - var scope = scopeFactory.CreateAsyncScope(); - await using (scope.ConfigureAwait(false)) - { - var reconciler = scope.ServiceProvider.GetRequiredService(); - await reconciler.ReconcileServerAsync(changed.GuildId, changed.ServerId, cancellationToken) - .ConfigureAwait(false); - } - } + await eventBus.ConsumeAsync( + (evt, ct) => ReconcileServerAsync(evt.GuildId, evt.ServerId, ct), + ex => logger.LogError(ex, "Handling {EventType} failed; skipping that reconcile.", + nameof(ServerCredentialsChangedEvent)), + cancellationToken).ConfigureAwait(false); } catch (OperationCanceledException) { @@ -200,21 +206,13 @@ await reconciler.ReconcileServerAsync(changed.GuildId, changed.ServerId, cancell private async Task ConsumeInfoMapReadyAsync(CancellationToken cancellationToken) { - // If this loop faults (broad catch), the consumer exits permanently and the #info map message stops - // updating until the host restarts. Acceptable: the reconciler is idempotent and a restart heals. try { - await foreach (var ready in eventBus.SubscribeAsync(cancellationToken) - .ConfigureAwait(false)) - { - var scope = scopeFactory.CreateAsyncScope(); - await using (scope.ConfigureAwait(false)) - { - var reconciler = scope.ServiceProvider.GetRequiredService(); - await reconciler.ReconcileServerAsync(ready.GuildId, ready.ServerId, cancellationToken) - .ConfigureAwait(false); - } - } + await eventBus.ConsumeAsync( + (evt, ct) => ReconcileServerAsync(evt.GuildId, evt.ServerId, ct), + ex => logger.LogError(ex, "Handling {EventType} failed; skipping that reconcile.", + nameof(InfoMapReadyEvent)), + cancellationToken).ConfigureAwait(false); } catch (OperationCanceledException) { @@ -234,17 +232,11 @@ private async Task ConsumeServerRegisteredAsync(CancellationToken cancellationTo // long after startup. try { - await foreach (var registered in eventBus.SubscribeAsync(cancellationToken) - .ConfigureAwait(false)) - { - var scope = scopeFactory.CreateAsyncScope(); - await using (scope.ConfigureAwait(false)) - { - var reconciler = scope.ServiceProvider.GetRequiredService(); - await reconciler.ReconcileServerAsync(registered.GuildId, registered.ServerId, cancellationToken) - .ConfigureAwait(false); - } - } + await eventBus.ConsumeAsync( + (evt, ct) => ReconcileServerAsync(evt.GuildId, evt.ServerId, ct), + ex => logger.LogError(ex, "Handling {EventType} failed; skipping that reconcile.", + nameof(ServerRegisteredEvent)), + cancellationToken).ConfigureAwait(false); } catch (OperationCanceledException) { diff --git a/tests/RustPlusBot.Abstractions.Tests/Events/EventBusConsumptionTests.cs b/tests/RustPlusBot.Abstractions.Tests/Events/EventBusConsumptionTests.cs new file mode 100644 index 00000000..05292261 --- /dev/null +++ b/tests/RustPlusBot.Abstractions.Tests/Events/EventBusConsumptionTests.cs @@ -0,0 +1,70 @@ +using RustPlusBot.Abstractions.Events; + +namespace RustPlusBot.Abstractions.Tests.Events; + +public sealed class EventBusConsumptionTests +{ + [Fact] + public async Task A_failing_handler_costs_its_own_event_only() + { + var handled = new List(); + var failures = new List(); + using var cts = new CancellationTokenSource(); + + var consume = new ScriptedBus(new Ping(1), new Ping(2), new Ping(3)).ConsumeAsync( + (evt, _) => + { + handled.Add(evt.Id); + return evt.Id == 1 ? throw new TimeoutException("The operation has timed out.") : Task.CompletedTask; + }, + failures.Add, + cts.Token); + + while (handled.Count < 3) + { + await Task.Delay(10); + } + + await cts.CancelAsync(); + await Assert.ThrowsAnyAsync(() => consume); + Assert.Equal([1, 2, 3], handled); + Assert.Single(failures); + Assert.IsType(failures[0]); + } + + [Fact] + public async Task Cancellation_ends_the_subscription() + { + using var cts = new CancellationTokenSource(); + await cts.CancelAsync(); + + var consume = new ScriptedBus(new Ping(1)).ConsumeAsync( + (_, _) => Task.CompletedTask, _ => Assert.Fail("no handler failure expected"), cts.Token); + + await Assert.ThrowsAnyAsync(() => consume); + } + + private sealed record Ping(int Id); + + /// A bus whose subscription replays a fixed sequence, then blocks until cancelled. + /// The events to yield, in order. + private sealed class ScriptedBus(params Ping[] events) : IEventBus + { + public ValueTask PublishAsync(TEvent @event, CancellationToken cancellationToken = default) + where TEvent : notnull => ValueTask.CompletedTask; + + public async IAsyncEnumerable SubscribeAsync( + [System.Runtime.CompilerServices.EnumeratorCancellation] + CancellationToken cancellationToken = default) + where TEvent : notnull + { + foreach (var evt in events.Cast()) + { + cancellationToken.ThrowIfCancellationRequested(); + yield return evt; + } + + await Task.Delay(Timeout.Infinite, cancellationToken).ConfigureAwait(false); + } + } +} diff --git a/tests/RustPlusBot.Features.Connections.Tests/Fakes/FakeRustSocketSource.cs b/tests/RustPlusBot.Features.Connections.Tests/Fakes/FakeRustSocketSource.cs index 7b2277bd..739ca1cf 100644 --- a/tests/RustPlusBot.Features.Connections.Tests/Fakes/FakeRustSocketSource.cs +++ b/tests/RustPlusBot.Features.Connections.Tests/Fakes/FakeRustSocketSource.cs @@ -262,6 +262,11 @@ internal sealed class FakeConnection(SocketConnectOutcome outcome, FakeRustSocke /// (a per-request timeout) instead of returning . public bool MonumentsTimeout { get; set; } + /// When set, throws this instead of returning + /// — models the real socket's "GetMap returned no data" throw when + /// the Rust+ endpoint answers with an error (rate limit, no map, …). + public Exception? MonumentsFault { get; set; } + /// The bytes returned by . Defaults to null. public byte[]? MapImageResult { get; set; } @@ -414,10 +419,13 @@ public Task> GetMapMarkersAsync(TimeSpan timeou Task.FromResult(World); public Task> GetMonumentsAsync(TimeSpan timeout, - CancellationToken cancellationToken = default) => - MonumentsTimeout - ? Task.FromException>(new OperationCanceledException()) - : Task.FromResult(MonumentsResult); + CancellationToken cancellationToken = default) + { + var fault = MonumentsFault ?? (MonumentsTimeout ? new OperationCanceledException() : null); + return fault is null + ? Task.FromResult(MonumentsResult) + : Task.FromException>(fault); + } public Task GetMapImageAsync(TimeSpan timeout, CancellationToken cancellationToken = default) => Task.FromResult(MapImageResult); diff --git a/tests/RustPlusBot.Features.Connections.Tests/ServerQueryTests.cs b/tests/RustPlusBot.Features.Connections.Tests/ServerQueryTests.cs index c8e03528..4a20e23a 100644 --- a/tests/RustPlusBot.Features.Connections.Tests/ServerQueryTests.cs +++ b/tests/RustPlusBot.Features.Connections.Tests/ServerQueryTests.cs @@ -258,6 +258,27 @@ public async Task GetMonumentsAsync_returns_empty_when_no_live_socket() Assert.Empty(result); } + [Fact] + public async Task GetMonumentsAsync_returns_empty_when_the_endpoint_fails() + { + // The query seam promises degradation ("or an empty list"), but the socket-level call throws on a + // failed GetMap. Left unguarded, that throw escapes into the map pipeline and kills its event loop. + var source = new FakeRustSocketSource(); + var (provider, supervisor) = CreateHarness(source); + await using var _ = provider; + var serverId = await SeedServerWithActiveAsync(provider, steamId: 555UL); + + using var cts = new CancellationTokenSource(TimeSpan.FromSeconds(30)); + await supervisor.EnsureConnectionAsync(10UL, serverId, cts.Token); + await WaitUntilAsync(() => supervisor.HasLiveSocket(10UL, serverId), cts.Token); + source.LastConnection!.MonumentsFault = new InvalidOperationException("GetMap returned no data."); + + var result = await supervisor.GetMonumentsAsync(10UL, serverId, cts.Token); + + Assert.Empty(result); + await supervisor.StopAllAsync(); + } + private static async Task WaitUntilAsync(Func condition, CancellationToken ct) { while (!condition()) diff --git a/tests/RustPlusBot.Features.Map.Tests/Hosting/MapHostedServiceTests.cs b/tests/RustPlusBot.Features.Map.Tests/Hosting/MapHostedServiceTests.cs new file mode 100644 index 00000000..bc591b2b --- /dev/null +++ b/tests/RustPlusBot.Features.Map.Tests/Hosting/MapHostedServiceTests.cs @@ -0,0 +1,103 @@ +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Logging.Abstractions; +using Microsoft.Extensions.Options; +using NSubstitute; +using RustMapsApi.V4.Assets; +using RustPlusBot.Abstractions.Connections; +using RustPlusBot.Abstractions.Events; +using RustPlusBot.Abstractions.Time; +using RustPlusBot.Domain.Connections; +using RustPlusBot.Features.Events.State; +using RustPlusBot.Features.Map.Assets; +using RustPlusBot.Features.Map.Composing; +using RustPlusBot.Features.Map.Hosting; +using RustPlusBot.Features.Map.Posting; +using RustPlusBot.Features.Map.Rendering; +using RustPlusBot.Features.Workspace.Locating; +using RustPlusBot.Persistence.Connections; +using RustPlusBot.Persistence.Map; + +namespace RustPlusBot.Features.Map.Tests.Hosting; + +public sealed class MapHostedServiceTests +{ + private const ulong Guild = 1UL; + private static readonly Guid Server = Guid.NewGuid(); + + /// A clock that jumps a minute per read so the refresh throttle never gates a test refresh. + private static IClock AdvancingClock() + { + var ticks = 0; + var clock = Substitute.For(); + clock.UtcNow.Returns(_ => DateTimeOffset.UnixEpoch + TimeSpan.FromMinutes(ticks++)); + return clock; + } + + private static IServiceScopeFactory ScopeFactory(IConnectionStore store) => + new ServiceCollection() + .AddScoped(_ => store) + .AddScoped(_ => Substitute.For()) + .BuildServiceProvider() + .GetRequiredService(); + + private static IConnectionStore ConnectedStore() + { + var store = Substitute.For(); + store.GetStateAsync(Guild, Server, Arg.Any()) + .Returns(new ConnectionState + { + GuildId = Guild, RustServerId = Server, Status = ConnectionStatus.Connected + }); + return store; + } + + private static MapComposer Composer(IServiceScopeFactory scopeFactory) => + new(new BaseMapCache([]), Substitute.For(), Substitute.For(), + Substitute.For(), + new MapRenderer(new MonumentIconSource(new MonumentAssetSource(), NullLogger.Instance)), + scopeFactory); + + [Fact] + public async Task Status_loop_survives_a_faulting_refresh_and_keeps_consuming_events() + { + // A transient failure anywhere in the refresh path (a Rust+ query, Discord, …) used to escape the + // await-foreach and end the subscription for the rest of the process: the #map then went silent + // until restart. One bad refresh must degrade that refresh only. + var locator = Substitute.For(); + var calls = 0; + locator.GetChannelIdAsync(Guild, Server, Arg.Any()) + .Returns(_ => ++calls == 1 + ? throw new InvalidOperationException("GetMap returned no data.") + : null); + + var scopeFactory = ScopeFactory(ConnectedStore()); + var pipeline = new MapPipeline(Composer(scopeFactory), new BaseMapCache([]), locator, + Substitute.For()); + var bus = new InMemoryEventBus(); + var options = Options.Create(new MapOptions + { + MapRefreshInterval = TimeSpan.FromMinutes(30) + }); + var service = new MapHostedService(bus, pipeline, AdvancingClock(), options, scopeFactory, + NullLogger.Instance); + + await service.StartAsync(CancellationToken.None); + try + { + // The bus drops events published before the subscriber is live, so publish until observed. + var deadline = DateTimeOffset.UtcNow.AddSeconds(5); + while (DateTimeOffset.UtcNow < deadline && Volatile.Read(ref calls) < 2) + { + await bus.PublishAsync(new ConnectionStatusChangedEvent(Guild, Server, true, false)); + await Task.Delay(20); + } + } + finally + { + await service.StopAsync(CancellationToken.None); + } + + Assert.True(Volatile.Read(ref calls) >= 2, + $"the status loop stopped consuming after the first refresh threw (calls: {calls})"); + } +} diff --git a/tests/RustPlusBot.Features.Map.Tests/MonumentTokenMapTests.cs b/tests/RustPlusBot.Features.Map.Tests/MonumentTokenMapTests.cs index 859fe27e..1d1be3d4 100644 --- a/tests/RustPlusBot.Features.Map.Tests/MonumentTokenMapTests.cs +++ b/tests/RustPlusBot.Features.Map.Tests/MonumentTokenMapTests.cs @@ -13,6 +13,7 @@ public sealed class MonumentTokenMapTests [InlineData("bandit_camp", MonumentType.BanditTown)] [InlineData("radtown", MonumentType.Radtown)] [InlineData("jungle_ziggurat", MonumentType.JungleZigguratA)] + [InlineData("apartmentcomplex", MonumentType.ApartmentsComplex)] [InlineData("train_tunnel_display_name", MonumentType.TunnelEntrance)] [InlineData("train_tunnel_link_display_name", MonumentType.TunnelEntranceTransition)] [InlineData("arctic_base_b", MonumentType.ArcticResearchBaseA)] diff --git a/tests/RustPlusBot.Features.Workspace.Tests/Hosting/WorkspaceConsumerResilienceTests.cs b/tests/RustPlusBot.Features.Workspace.Tests/Hosting/WorkspaceConsumerResilienceTests.cs new file mode 100644 index 00000000..6df6f335 --- /dev/null +++ b/tests/RustPlusBot.Features.Workspace.Tests/Hosting/WorkspaceConsumerResilienceTests.cs @@ -0,0 +1,67 @@ +using Discord.WebSocket; +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Logging.Abstractions; +using NSubstitute; +using RustPlusBot.Abstractions.Events; +using RustPlusBot.Features.Workspace.Hosting; +using RustPlusBot.Features.Workspace.Reconciler; +using RustPlusBot.Features.Workspace.Registry; + +namespace RustPlusBot.Features.Workspace.Tests.Hosting; + +public sealed class WorkspaceConsumerResilienceTests +{ + private static (WorkspaceHostedService Service, ServiceProvider Provider, InMemoryEventBus Bus) Build( + IWorkspaceReconciler reconciler) + { + var services = new ServiceCollection(); + services.AddScoped(_ => reconciler); + services.AddSingleton(Substitute.For()); // StartAsync resolves this up front. + var provider = services.BuildServiceProvider(); + + var bus = new InMemoryEventBus(); + var service = new WorkspaceHostedService(new DiscordSocketClient(), bus, + provider.GetRequiredService(), + NullLogger.Instance); + return (service, provider, bus); + } + + /// + /// A Discord REST timeout inside the reconcile is routine, but it used to escape the consumer's + /// await-foreach and end the subscription for the rest of the process — every later connect/disconnect + /// then left the info channels stale until the bot restarted. + /// + [Fact] + public async Task ConnectionStatus_consumer_survives_a_faulting_reconcile() + { + var calls = 0; + var reconciler = Substitute.For(); + reconciler.ReconcileServerAsync(Arg.Any(), Arg.Any(), Arg.Any()) + .Returns(_ => Interlocked.Increment(ref calls) == 1 + ? throw new TimeoutException("The operation has timed out.") + : Task.CompletedTask); + + var (service, provider, bus) = Build(reconciler); + await using var _ = provider; + + await service.StartAsync(CancellationToken.None); + try + { + // The in-process bus does not replay; re-publish until the consumer has handled two events. + var deadline = DateTimeOffset.UtcNow.AddSeconds(20); + while (DateTimeOffset.UtcNow < deadline && Volatile.Read(ref calls) < 2) + { + await bus.PublishAsync( + new ConnectionStatusChangedEvent(10UL, Guid.NewGuid(), IsConnected: true, WasConnected: false)); + await Task.Delay(20); + } + } + finally + { + await service.StopAsync(CancellationToken.None); + } + + Assert.True(Volatile.Read(ref calls) >= 2, + $"the consumer stopped after the first reconcile threw (calls: {calls})"); + } +} From 60454c94cac0dc7691a9b034c9aced0ea10ee732 Mon Sep 17 00:00:00 2001 From: = Date: Fri, 24 Jul 2026 11:28:08 +0200 Subject: [PATCH 2/3] fix(connections): use a query-specific log for the monuments seam MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Copilot review: GetMonumentsAsync (the query seam used by map rendering) reused LogMonumentsFetchFailed, whose message says "rig detection disabled for this connection" — misleading for a non-rig caller. Give the seam its own generic LogMonumentsQueryFailed and keep the rig-specific message on the rig path. Co-Authored-By: Claude Opus 4.8 (1M context) --- .../Supervisor/ConnectionSupervisor.cs | 6 +++++- 1 file changed, 5 insertions(+), 1 deletion(-) diff --git a/src/RustPlusBot.Features.Connections/Supervisor/ConnectionSupervisor.cs b/src/RustPlusBot.Features.Connections/Supervisor/ConnectionSupervisor.cs index ade23d55..aa2fb10a 100644 --- a/src/RustPlusBot.Features.Connections/Supervisor/ConnectionSupervisor.cs +++ b/src/RustPlusBot.Features.Connections/Supervisor/ConnectionSupervisor.cs @@ -327,7 +327,7 @@ public async Task> GetMonumentsAsync( // 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. - LogMonumentsFetchFailed(logger, ex, serverId); + LogMonumentsQueryFailed(logger, ex, serverId); return []; } } @@ -1515,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); From 4942b9ec4f84751e9d928ef1e7868e6f624f5dc1 Mon Sep 17 00:00:00 2001 From: = Date: Fri, 24 Jul 2026 11:36:51 +0200 Subject: [PATCH 3/3] test(map): stop the status-loop regression test flaking on slow CI AdvancingClock jumped 1 minute per read while MapRefreshInterval is 30 minutes, so MapRefreshThrottle gated every refresh after the first: the counter only re-incremented once ~30 clock reads accumulated. Fast enough locally, but on Windows CI fewer than 30 events were consumed inside the 5s deadline, so calls stuck at 1 and the assert failed. Advance an hour per read (past the interval) so the throttle passes every refresh and each event reaches the counter. Co-Authored-By: Claude Opus 4.8 (1M context) --- .../Hosting/MapHostedServiceTests.cs | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/tests/RustPlusBot.Features.Map.Tests/Hosting/MapHostedServiceTests.cs b/tests/RustPlusBot.Features.Map.Tests/Hosting/MapHostedServiceTests.cs index bc591b2b..574b9b5a 100644 --- a/tests/RustPlusBot.Features.Map.Tests/Hosting/MapHostedServiceTests.cs +++ b/tests/RustPlusBot.Features.Map.Tests/Hosting/MapHostedServiceTests.cs @@ -24,12 +24,13 @@ public sealed class MapHostedServiceTests private const ulong Guild = 1UL; private static readonly Guid Server = Guid.NewGuid(); - /// A clock that jumps a minute per read so the refresh throttle never gates a test refresh. + /// A clock that jumps an hour per read — past below, + /// so the throttle passes every refresh and each consumed event reaches the counter (never gates). private static IClock AdvancingClock() { var ticks = 0; var clock = Substitute.For(); - clock.UtcNow.Returns(_ => DateTimeOffset.UnixEpoch + TimeSpan.FromMinutes(ticks++)); + clock.UtcNow.Returns(_ => DateTimeOffset.UnixEpoch + TimeSpan.FromHours(ticks++)); return clock; }