Skip to content

Commit a1e700c

Browse files
HandyS11claude
HandyS11
and
claude
authored
fix: stop one transient failure from permanently killing an event consumer (#64)
* fix: keep event consumers alive across transient handler failures 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<TEvent>(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<Exception>, 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) <noreply@anthropic.com> * fix(connections): use a query-specific log for the monuments seam 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) <noreply@anthropic.com> * 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) <noreply@anthropic.com> --------- Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
1 parent 2ffb18f commit a1e700c

22 files changed

Lines changed: 599 additions & 244 deletions

File tree

‎src/RustPlusBot.Abstractions/Connections/IRustServerQuery.cs‎

Lines changed: 3 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -53,11 +53,12 @@ public interface IRustServerQuery
5353
/// <returns>The world snapshot, or null.</returns>
5454
Task<WorldSnapshot?> GetWorldAsync(ulong guildId, Guid serverId, CancellationToken cancellationToken);
5555

56-
/// <summary>Gets the server's monuments, or an empty list when there is no live socket.</summary>
56+
/// <summary>Gets the server's monuments, or an empty list when there is no live socket or the fetch
57+
/// fails; never throws for a failed fetch.</summary>
5758
/// <param name="guildId">The owning guild snowflake.</param>
5859
/// <param name="serverId">The target server id.</param>
5960
/// <param name="cancellationToken">A cancellation token.</param>
60-
/// <returns>The monuments, or an empty list when there is no live socket.</returns>
61+
/// <returns>The monuments, or an empty list when there is no live socket or the fetch fails/times out.</returns>
6162
Task<IReadOnlyList<MonumentSnapshot>> GetMonumentsAsync(
6263
ulong guildId,
6364
Guid serverId,

‎src/RustPlusBot.Abstractions/Events/EventBusConsumption.cs‎

Lines changed: 55 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,55 @@
1+
namespace RustPlusBot.Abstractions.Events;
2+
3+
/// <summary>
4+
/// Consumption helpers for <see cref="IEventBus"/> subscribers.
5+
/// </summary>
6+
/// <remarks>
7+
/// Every long-running consumer in this solution wraps its <c>await foreach</c> in a broad catch that logs
8+
/// and then <em>exits</em>: an exception escaping a handler ends the subscription for the rest of the
9+
/// process, and the feature it drives goes silently dead until the host restarts. Handlers here talk to
10+
/// Discord and to the Rust+ server, where a timeout or a 5xx is routine rather than exceptional, so that
11+
/// trade is never the one we want. Consuming through <see cref="ConsumeAsync{TEvent}"/> keeps the
12+
/// subscription alive: a failed handler costs its own event and nothing more.
13+
/// </remarks>
14+
public static class EventBusConsumption
15+
{
16+
/// <summary>
17+
/// Consumes every published <typeparamref name="TEvent"/>, reporting and skipping the ones whose
18+
/// handler throws. Only cancellation of <paramref name="cancellationToken"/> ends the subscription.
19+
/// </summary>
20+
/// <typeparam name="TEvent">The event type to consume.</typeparam>
21+
/// <param name="eventBus">The bus to subscribe to.</param>
22+
/// <param name="handle">Handles one event; its failures are reported, not propagated.</param>
23+
/// <param name="onHandlerFailure">Reports a handler failure (typically a log call).</param>
24+
/// <param name="cancellationToken">Ends the subscription when cancelled.</param>
25+
/// <returns>A task that completes when the subscription ends.</returns>
26+
public static async Task ConsumeAsync<TEvent>(
27+
this IEventBus eventBus,
28+
Func<TEvent, CancellationToken, Task> handle,
29+
Action<Exception> onHandlerFailure,
30+
CancellationToken cancellationToken)
31+
where TEvent : notnull
32+
{
33+
ArgumentNullException.ThrowIfNull(eventBus);
34+
ArgumentNullException.ThrowIfNull(handle);
35+
ArgumentNullException.ThrowIfNull(onHandlerFailure);
36+
37+
await foreach (var evt in eventBus.SubscribeAsync<TEvent>(cancellationToken).ConfigureAwait(false))
38+
{
39+
try
40+
{
41+
await handle(evt, cancellationToken).ConfigureAwait(false);
42+
}
43+
catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested)
44+
{
45+
throw; // A real shutdown: let the caller's loop handler end it.
46+
}
47+
#pragma warning disable CA1031 // Broad catch: a handler failure must cost one event, never the subscription.
48+
catch (Exception ex)
49+
#pragma warning restore CA1031
50+
{
51+
onHandlerFailure(ex);
52+
}
53+
}
54+
}
55+
}

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

Lines changed: 21 additions & 30 deletions
Original file line numberDiff line numberDiff line change
@@ -68,11 +68,9 @@ private async Task ConsumePairedAsync(CancellationToken cancellationToken)
6868
{
6969
try
7070
{
71-
await foreach (var evt in eventBus.SubscribeAsync<AlarmPairedEvent>(cancellationToken)
72-
.ConfigureAwait(false))
73-
{
74-
await coordinator.HandlePairedAsync(evt, cancellationToken).ConfigureAwait(false);
75-
}
71+
await eventBus.ConsumeAsync<AlarmPairedEvent>(coordinator.HandlePairedAsync,
72+
ex => LogHandlerFailed(logger, ex, nameof(AlarmPairedEvent)), cancellationToken)
73+
.ConfigureAwait(false);
7674
}
7775
catch (OperationCanceledException)
7876
{
@@ -90,11 +88,9 @@ private async Task ConsumeTriggeredAsync(CancellationToken cancellationToken)
9088
{
9189
try
9290
{
93-
await foreach (var evt in eventBus.SubscribeAsync<SmartDeviceTriggeredEvent>(cancellationToken)
94-
.ConfigureAwait(false))
95-
{
96-
await relay.HandleTriggeredAsync(evt, cancellationToken).ConfigureAwait(false);
97-
}
91+
await eventBus.ConsumeAsync<SmartDeviceTriggeredEvent>(relay.HandleTriggeredAsync,
92+
ex => LogHandlerFailed(logger, ex, nameof(SmartDeviceTriggeredEvent)), cancellationToken)
93+
.ConfigureAwait(false);
9894
}
9995
catch (OperationCanceledException)
10096
{
@@ -112,11 +108,9 @@ private async Task ConsumeStatusAsync(CancellationToken cancellationToken)
112108
{
113109
try
114110
{
115-
await foreach (var evt in eventBus.SubscribeAsync<ConnectionStatusChangedEvent>(cancellationToken)
116-
.ConfigureAwait(false))
117-
{
118-
await relay.HandleConnectionStatusAsync(evt, cancellationToken).ConfigureAwait(false);
119-
}
111+
await eventBus.ConsumeAsync<ConnectionStatusChangedEvent>(relay.HandleConnectionStatusAsync,
112+
ex => LogHandlerFailed(logger, ex, nameof(ConnectionStatusChangedEvent)), cancellationToken)
113+
.ConfigureAwait(false);
120114
}
121115
catch (OperationCanceledException)
122116
{
@@ -134,11 +128,9 @@ private async Task ConsumeReachabilityChangedAsync(CancellationToken cancellatio
134128
{
135129
try
136130
{
137-
await foreach (var evt in eventBus.SubscribeAsync<DeviceReachabilityChangedEvent>(cancellationToken)
138-
.ConfigureAwait(false))
139-
{
140-
await relay.HandleReachabilityChangedAsync(evt, cancellationToken).ConfigureAwait(false);
141-
}
131+
await eventBus.ConsumeAsync<DeviceReachabilityChangedEvent>(relay.HandleReachabilityChangedAsync,
132+
ex => LogHandlerFailed(logger, ex, nameof(DeviceReachabilityChangedEvent)), cancellationToken)
133+
.ConfigureAwait(false);
142134
}
143135
catch (OperationCanceledException)
144136
{
@@ -156,11 +148,9 @@ private async Task ConsumeStateObservedAsync(CancellationToken cancellationToken
156148
{
157149
try
158150
{
159-
await foreach (var evt in eventBus.SubscribeAsync<SmartDeviceStateObservedEvent>(cancellationToken)
160-
.ConfigureAwait(false))
161-
{
162-
await relay.HandleStateObservedAsync(evt, cancellationToken).ConfigureAwait(false);
163-
}
151+
await eventBus.ConsumeAsync<SmartDeviceStateObservedEvent>(relay.HandleStateObservedAsync,
152+
ex => LogHandlerFailed(logger, ex, nameof(SmartDeviceStateObservedEvent)), cancellationToken)
153+
.ConfigureAwait(false);
164154
}
165155
catch (OperationCanceledException)
166156
{
@@ -178,11 +168,9 @@ private async Task ConsumeWipedAsync(CancellationToken cancellationToken)
178168
{
179169
try
180170
{
181-
await foreach (var evt in eventBus.SubscribeAsync<ServerWipedEvent>(cancellationToken)
182-
.ConfigureAwait(false))
183-
{
184-
await purger.HandleServerWipedAsync(evt, cancellationToken).ConfigureAwait(false);
185-
}
171+
await eventBus.ConsumeAsync<ServerWipedEvent>(purger.HandleServerWipedAsync,
172+
ex => LogHandlerFailed(logger, ex, nameof(ServerWipedEvent)), cancellationToken)
173+
.ConfigureAwait(false);
186174
}
187175
catch (OperationCanceledException)
188176
{
@@ -196,6 +184,9 @@ private async Task ConsumeWipedAsync(CancellationToken cancellationToken)
196184
}
197185
}
198186

187+
[LoggerMessage(Level = LogLevel.Error, Message = "Handling {EventType} failed; skipping that event.")]
188+
private static partial void LogHandlerFailed(ILogger logger, Exception exception, string eventType);
189+
199190
[LoggerMessage(Level = LogLevel.Error, Message = "Alarm pairing loop faulted.")]
200191
private static partial void LogPairedLoopFaulted(ILogger logger, Exception exception);
201192

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

Lines changed: 38 additions & 34 deletions
Original file line numberDiff line numberDiff line change
@@ -76,13 +76,12 @@ private async Task ConsumeTeamMessagesAsync(CancellationToken cancellationToken)
7676
{
7777
try
7878
{
79-
await foreach (var evt in eventBus.SubscribeAsync<TeamMessageReceivedEvent>(cancellationToken)
80-
.ConfigureAwait(false))
81-
{
82-
await relay.RelayAsync(
83-
new RelayedChatLine(ChatChannelKind.Team, evt.GuildId, evt.ServerId, evt.SenderName,
84-
evt.Message, evt.FromActivePlayer), cancellationToken).ConfigureAwait(false);
85-
}
79+
await eventBus.ConsumeAsync<TeamMessageReceivedEvent>(
80+
(evt, ct) => relay.RelayAsync(
81+
new RelayedChatLine(ChatChannelKind.Team, evt.GuildId, evt.ServerId, evt.SenderName,
82+
evt.Message, evt.FromActivePlayer), ct),
83+
ex => LogHandlerFailed(logger, ex, nameof(TeamMessageReceivedEvent)), cancellationToken)
84+
.ConfigureAwait(false);
8685
}
8786
catch (OperationCanceledException)
8887
{
@@ -100,33 +99,9 @@ private async Task ConsumeClanMessagesAsync(CancellationToken cancellationToken)
10099
{
101100
try
102101
{
103-
await foreach (var evt in eventBus.SubscribeAsync<ClanMessageReceivedEvent>(cancellationToken)
104-
.ConfigureAwait(false))
105-
{
106-
// Clan members arrive as Steam ids only; chat is where we learn their names. A failure here
107-
// is isolated so it cannot take down the relay loop below: a missed name is cosmetic, a
108-
// dead relay loop silences clan chat until the process restarts.
109-
try
110-
{
111-
var scope = scopeFactory.CreateAsyncScope();
112-
await using (scope.ConfigureAwait(false))
113-
{
114-
var store = scope.ServiceProvider.GetRequiredService<IClanStore>();
115-
await store.RecordNameAsync(evt.GuildId, evt.ServerId, evt.SenderSteamId, evt.SenderName,
116-
cancellationToken).ConfigureAwait(false);
117-
}
118-
}
119-
#pragma warning disable CA1031 // Broad catch: a name-recording failure must not kill the clan relay loop.
120-
catch (Exception ex)
121-
#pragma warning restore CA1031
122-
{
123-
LogClanNameRecordFailed(logger, ex);
124-
}
125-
126-
await relay.RelayAsync(
127-
new RelayedChatLine(ChatChannelKind.Clan, evt.GuildId, evt.ServerId, evt.SenderName,
128-
evt.Message, evt.FromActivePlayer), cancellationToken).ConfigureAwait(false);
129-
}
102+
await eventBus.ConsumeAsync<ClanMessageReceivedEvent>(HandleClanMessageAsync,
103+
ex => LogHandlerFailed(logger, ex, nameof(ClanMessageReceivedEvent)), cancellationToken)
104+
.ConfigureAwait(false);
130105
}
131106
catch (OperationCanceledException)
132107
{
@@ -140,6 +115,32 @@ await relay.RelayAsync(
140115
}
141116
}
142117

118+
private async Task HandleClanMessageAsync(ClanMessageReceivedEvent evt, CancellationToken cancellationToken)
119+
{
120+
// Clan members arrive as Steam ids only; chat is where we learn their names. A failure here is
121+
// isolated so it cannot cost the relay below: a missed name is cosmetic, a dropped clan line is not.
122+
try
123+
{
124+
var scope = scopeFactory.CreateAsyncScope();
125+
await using (scope.ConfigureAwait(false))
126+
{
127+
var store = scope.ServiceProvider.GetRequiredService<IClanStore>();
128+
await store.RecordNameAsync(evt.GuildId, evt.ServerId, evt.SenderSteamId, evt.SenderName,
129+
cancellationToken).ConfigureAwait(false);
130+
}
131+
}
132+
#pragma warning disable CA1031 // Broad catch: a name-recording failure must not cost the relay.
133+
catch (Exception ex)
134+
#pragma warning restore CA1031
135+
{
136+
LogClanNameRecordFailed(logger, ex);
137+
}
138+
139+
await relay.RelayAsync(
140+
new RelayedChatLine(ChatChannelKind.Clan, evt.GuildId, evt.ServerId, evt.SenderName,
141+
evt.Message, evt.FromActivePlayer), cancellationToken).ConfigureAwait(false);
142+
}
143+
143144
private async Task OnMessageReceivedAsync(SocketMessage message)
144145
{
145146
try
@@ -169,6 +170,9 @@ private async Task OnMessageReceivedAsync(SocketMessage message)
169170
}
170171
}
171172

173+
[LoggerMessage(Level = LogLevel.Error, Message = "Handling {EventType} failed; skipping that event.")]
174+
private static partial void LogHandlerFailed(ILogger logger, Exception exception, string eventType);
175+
172176
[LoggerMessage(Level = LogLevel.Error, Message = "Team message relay loop faulted.")]
173177
private static partial void LogTeamRelayLoopFaulted(ILogger logger, Exception exception);
174178

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

Lines changed: 11 additions & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -78,12 +78,10 @@ private async Task ConsumeRegisteredAsync(CancellationToken cancellationToken)
7878
{
7979
try
8080
{
81-
await foreach (var registered in eventBus.SubscribeAsync<ServerRegisteredEvent>(cancellationToken)
82-
.ConfigureAwait(false))
83-
{
84-
await supervisor.EnsureConnectionAsync(registered.GuildId, registered.ServerId, cancellationToken)
85-
.ConfigureAwait(false);
86-
}
81+
await eventBus.ConsumeAsync<ServerRegisteredEvent>(
82+
(registered, ct) => supervisor.EnsureConnectionAsync(registered.GuildId, registered.ServerId, ct),
83+
ex => LogHandlerFailed(logger, ex, nameof(ServerRegisteredEvent)), cancellationToken)
84+
.ConfigureAwait(false);
8785
}
8886
catch (OperationCanceledException)
8987
{
@@ -101,12 +99,10 @@ private async Task ConsumeCredentialsChangedAsync(CancellationToken cancellation
10199
{
102100
try
103101
{
104-
await foreach (var changed in eventBus.SubscribeAsync<ServerCredentialsChangedEvent>(cancellationToken)
105-
.ConfigureAwait(false))
106-
{
107-
await supervisor.EnsureConnectionAsync(changed.GuildId, changed.ServerId, cancellationToken)
108-
.ConfigureAwait(false);
109-
}
102+
await eventBus.ConsumeAsync<ServerCredentialsChangedEvent>(
103+
(changed, ct) => supervisor.EnsureConnectionAsync(changed.GuildId, changed.ServerId, ct),
104+
ex => LogHandlerFailed(logger, ex, nameof(ServerCredentialsChangedEvent)), cancellationToken)
105+
.ConfigureAwait(false);
110106
}
111107
catch (OperationCanceledException)
112108
{
@@ -120,6 +116,9 @@ await supervisor.EnsureConnectionAsync(changed.GuildId, changed.ServerId, cancel
120116
#pragma warning restore CA1031
121117
}
122118

119+
[LoggerMessage(Level = LogLevel.Error, Message = "Handling {EventType} failed; skipping that event.")]
120+
private static partial void LogHandlerFailed(ILogger logger, Exception exception, string eventType);
121+
123122
[LoggerMessage(Level = LogLevel.Error, Message = "Connection hosted-service loop faulted.")]
124123
private static partial void LogLoopFaulted(ILogger logger, Exception exception);
125124
}

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

Lines changed: 5 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -714,7 +714,11 @@ public async Task<IReadOnlyList<MonumentSnapshot>> GetMonumentsAsync(
714714
.ConfigureAwait(false);
715715
if (!response.IsSuccess || response.Data is null)
716716
{
717-
throw new InvalidOperationException("GetMap returned no data.");
717+
// Name the reason: the caller only ever sees this message, and "rate_limit" (three heavy
718+
// GetMap calls land back-to-back on connect) reads very differently from "no_map".
719+
throw new InvalidOperationException(
720+
"GetMap returned no data; error: " + (response.Error?.Code.ToString() ?? "none") +
721+
" (" + (response.Error?.Message ?? "no message") + ").");
718722
}
719723

720724
var monuments = new List<MonumentSnapshot>();

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

Lines changed: 23 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -311,8 +311,25 @@ public async Task<IReadOnlyList<MonumentSnapshot>> GetMonumentsAsync(
311311
return [];
312312
}
313313

314-
return await live.Connection.GetMonumentsAsync(_options.HeartbeatTimeout, cancellationToken)
315-
.ConfigureAwait(false);
314+
try
315+
{
316+
return await live.Connection.GetMonumentsAsync(_options.HeartbeatTimeout, cancellationToken)
317+
.ConfigureAwait(false);
318+
}
319+
catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested)
320+
{
321+
throw;
322+
}
323+
#pragma warning disable CA1031 // Broad catch: this seam promises degradation, so a failed fetch is "no monuments".
324+
catch (Exception ex)
325+
#pragma warning restore CA1031
326+
{
327+
// The connection-level call throws on a failed/timed-out GetMap (rate limit, no map, slow
328+
// endpoint). Callers here are render paths that must degrade to an icon-less map, never fault:
329+
// an escaping exception tears down the consuming loop for the rest of the process.
330+
LogMonumentsQueryFailed(logger, ex, serverId);
331+
return [];
332+
}
316333
}
317334

318335
/// <inheritdoc />
@@ -1498,6 +1515,10 @@ await eventBus.PublishAsync(
14981515
"Fetching monuments for oil-rig detection on server {ServerId} failed; rig detection disabled for this connection.")]
14991516
private static partial void LogMonumentsFetchFailed(ILogger logger, Exception exception, Guid serverId);
15001517

1518+
[LoggerMessage(Level = LogLevel.Warning,
1519+
Message = "Querying monuments for server {ServerId} failed; returning no monuments for this call.")]
1520+
private static partial void LogMonumentsQueryFailed(ILogger logger, Exception exception, Guid serverId);
1521+
15011522
[LoggerMessage(Level = LogLevel.Warning, Message = "Relaying a message to team chat for server {ServerId} failed.")]
15021523
private static partial void LogSendFailed(ILogger logger, Exception exception, Guid serverId);
15031524

0 commit comments

Comments
 (0)