Skip to content

Commit 38b39ea

Browse files
dependabot[bot]HandyS11claude
authored
Bump jetbrains.resharper.globaltools from 2026.1.4 to 2026.2.0 (#67)
* Bump jetbrains.resharper.globaltools from 2026.1.4 to 2026.2.0 --- updated-dependencies: - dependency-name: jetbrains.resharper.globaltools dependency-version: 2026.2.0 dependency-type: direct:production update-type: version-update:semver-minor ... Signed-off-by: dependabot[bot] <support@github.com> * style: space after spread operator for ReSharper 2026.2 ReSharper 2026.2 normalises `..x` to `.. x` in collection expressions, so cleanupcode rewrote this line and the CI format gate failed. Apply the formatting the tool now produces; it already matches the spread style used elsewhere in the tests. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com> * fix(map): subscribe before StartAsync returns so connect events are not dropped InfoMapHostedService subscribed to ConnectionStatusChangedEvent from inside its background Task.Run, so the subscription only came into existence once the thread pool scheduled that task. Every event published in that window was silently dropped, and the server only got registered on the next generation poll tick. On a saturated host that window is seconds long -- precisely when connections are being established. Subscribe synchronously in StartAsync (InMemoryEventBus registers eagerly) and hand the stream to the loop task, via a new ConsumeAsync overload that drains an already-established subscription. This is what made InfoMapHostedServiceTests flaky on CI. The test papered over the drop by republishing in a 5s poll loop, but that loop needs the pool to take its next turn: under starvation a single `await Task.Delay(20)` swallowed the whole deadline, so exactly one event was published, dropped, and never retried. Reproduced by blocking every pool worker -- the test failed with `publishes=1 elapsed=10548ms`, matching the CI failure. Tests now publish once and wait on a signal instead of polling a wall clock. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com> --------- Signed-off-by: dependabot[bot] <support@github.com> Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com> Co-authored-by: = <valentin-clergue@orange.fr> Co-authored-by: Claude Opus 5 <noreply@anthropic.com>
1 parent d53f080 commit 38b39ea

6 files changed

Lines changed: 121 additions & 37 deletions

File tree

.config/dotnet-tools.json

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -10,7 +10,7 @@
1010
"rollForward": false
1111
},
1212
"jetbrains.resharper.globaltools": {
13-
"version": "2026.1.4",
13+
"version": "2026.2.0",
1414
"commands": [
1515
"jb"
1616
],

src/RustPlusBot.Abstractions/Events/EventBusConsumption.cs

Lines changed: 33 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -8,7 +8,7 @@ namespace RustPlusBot.Abstractions.Events;
88
/// and then <em>exits</em>: an exception escaping a handler ends the subscription for the rest of the
99
/// process, and the feature it drives goes silently dead until the host restarts. Handlers here talk to
1010
/// 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
11+
/// trade is never the one we want. Consuming through <c>ConsumeAsync</c> keeps the
1212
/// subscription alive: a failed handler costs its own event and nothing more.
1313
/// </remarks>
1414
public static class EventBusConsumption
@@ -23,18 +23,48 @@ public static class EventBusConsumption
2323
/// <param name="onHandlerFailure">Reports a handler failure (typically a log call).</param>
2424
/// <param name="cancellationToken">Ends the subscription when cancelled.</param>
2525
/// <returns>A task that completes when the subscription ends.</returns>
26-
public static async Task ConsumeAsync<TEvent>(
26+
public static Task ConsumeAsync<TEvent>(
2727
this IEventBus eventBus,
2828
Func<TEvent, CancellationToken, Task> handle,
2929
Action<Exception> onHandlerFailure,
3030
CancellationToken cancellationToken)
3131
where TEvent : notnull
3232
{
3333
ArgumentNullException.ThrowIfNull(eventBus);
34+
35+
return eventBus.SubscribeAsync<TEvent>(cancellationToken)
36+
.ConsumeAsync(handle, onHandlerFailure, cancellationToken);
37+
}
38+
39+
/// <summary>
40+
/// Consumes an already-established subscription, reporting and skipping the events whose handler throws.
41+
/// Only cancellation of <paramref name="cancellationToken"/> ends the subscription.
42+
/// </summary>
43+
/// <remarks>
44+
/// Take this overload when the subscription must exist before the caller returns — a hosted service that
45+
/// subscribes inside its background <c>Task.Run</c> is only subscribed once that task is scheduled, and
46+
/// every event published in the meantime is dropped. Call
47+
/// <see cref="IEventBus.SubscribeAsync{TEvent}"/> synchronously in <c>StartAsync</c> (it registers the
48+
/// subscription eagerly) and hand the stream here.
49+
/// </remarks>
50+
/// <typeparam name="TEvent">The event type being consumed.</typeparam>
51+
/// <param name="events">The subscription stream to drain.</param>
52+
/// <param name="handle">Handles one event; its failures are reported, not propagated.</param>
53+
/// <param name="onHandlerFailure">Reports a handler failure (typically a log call).</param>
54+
/// <param name="cancellationToken">Ends the subscription when cancelled.</param>
55+
/// <returns>A task that completes when the subscription ends.</returns>
56+
public static async Task ConsumeAsync<TEvent>(
57+
this IAsyncEnumerable<TEvent> events,
58+
Func<TEvent, CancellationToken, Task> handle,
59+
Action<Exception> onHandlerFailure,
60+
CancellationToken cancellationToken)
61+
where TEvent : notnull
62+
{
63+
ArgumentNullException.ThrowIfNull(events);
3464
ArgumentNullException.ThrowIfNull(handle);
3565
ArgumentNullException.ThrowIfNull(onHandlerFailure);
3666

37-
await foreach (var evt in eventBus.SubscribeAsync<TEvent>(cancellationToken).ConfigureAwait(false))
67+
await foreach (var evt in events.ConfigureAwait(false))
3868
{
3969
try
4070
{

src/RustPlusBot.Features.Map/Hosting/InfoMapHostedService.cs

Lines changed: 10 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -46,7 +46,12 @@ internal sealed partial class InfoMapHostedService(
4646
/// <inheritdoc />
4747
public Task StartAsync(CancellationToken cancellationToken)
4848
{
49-
_statusLoop = Task.Run(() => ConsumeConnectionStatusAsync(_cts.Token), CancellationToken.None);
49+
// Subscribe here rather than inside the loop task: Task.Run only queues the body, so a service that
50+
// subscribed in there would stay unsubscribed until the pool schedules it — and every
51+
// ConnectionStatusChangedEvent published in that window is dropped. On a saturated host that window
52+
// is seconds long, which is exactly when connections are being established.
53+
var statusEvents = eventBus.SubscribeAsync<ConnectionStatusChangedEvent>(_cts.Token);
54+
_statusLoop = Task.Run(() => ConsumeConnectionStatusAsync(statusEvents, _cts.Token), CancellationToken.None);
5055
_tickLoop = Task.Run(() => RunTickAsync(_cts.Token), CancellationToken.None);
5156
return Task.CompletedTask;
5257
}
@@ -159,11 +164,13 @@ await eventBus.PublishAsync(new InfoMapReadyEvent(guild, server), cancellationTo
159164
}
160165
}
161166

162-
private async Task ConsumeConnectionStatusAsync(CancellationToken cancellationToken)
167+
private async Task ConsumeConnectionStatusAsync(
168+
IAsyncEnumerable<ConnectionStatusChangedEvent> statusEvents,
169+
CancellationToken cancellationToken)
163170
{
164171
try
165172
{
166-
await eventBus.ConsumeAsync<ConnectionStatusChangedEvent>(OnConnectionStatusAsync,
173+
await statusEvents.ConsumeAsync(OnConnectionStatusAsync,
167174
ex => LogHandlerFailed(logger, ex, nameof(ConnectionStatusChangedEvent)), cancellationToken)
168175
.ConfigureAwait(false);
169176
}

src/RustPlusBot.Features.Workspace/Hosting/WorkspaceHostedService.cs

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -154,7 +154,7 @@ private async Task OnChannelDestroyedAsync(SocketChannel channel)
154154

155155
/// <summary>
156156
/// Reconciles one server in its own scope. Every consumer below runs this through
157-
/// <see cref="EventBusConsumption.ConsumeAsync{TEvent}"/>, which absorbs its failures: the reconcile
157+
/// <c>EventBusConsumption.ConsumeAsync</c>, which absorbs its failures: the reconcile
158158
/// talks to Discord over REST, where a timeout or a 5xx is routine, and one of those must never end
159159
/// the subscription that drives the channels.
160160
/// </summary>

tests/RustPlusBot.Features.Commands.Tests/Modules/CctvChoiceDriftTests.cs

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -14,7 +14,7 @@ private static IReadOnlyList<string> ChoiceValues()
1414
.GetParameters()[0];
1515
return
1616
[
17-
..parameter.GetCustomAttributes<ChoiceAttribute>()
17+
.. parameter.GetCustomAttributes<ChoiceAttribute>()
1818
.Select(c => (string)c.Value!)
1919
];
2020
}

tests/RustPlusBot.Features.Map.Tests/Hosting/InfoMapHostedServiceTests.cs

Lines changed: 75 additions & 28 deletions
Original file line numberDiff line numberDiff line change
@@ -38,15 +38,11 @@ private static IOptions<MapOptions> ShortPollOptions() => Options.Create(new Map
3838
});
3939

4040
/// <summary>
41-
/// A fake bus that records every published <see cref="InfoMapReadyEvent"/>. Deterministic, unlike
42-
/// subscribing to the real <see cref="InMemoryEventBus"/> from a background task: that bus drops
43-
/// events published before the subscriber is active, which races the service's first tick.
41+
/// A fake bus that records every published <see cref="InfoMapReadyEvent"/> and signals each arrival.
42+
/// Deterministic, unlike subscribing to the real <see cref="InMemoryEventBus"/> from a background task:
43+
/// that bus drops events published before the subscriber is active, which races the service's first tick.
4444
/// </summary>
45-
private static (IEventBus Bus, ConcurrentQueue<InfoMapReadyEvent> Received) CapturingBus()
46-
{
47-
var bus = new CapturingEventBus();
48-
return (bus, bus.Received);
49-
}
45+
private static CapturingEventBus CapturingBus() => new();
5046

5147
[Fact]
5248
public async Task Ready_key_publishes_InfoMapReadyEvent_once_per_requester()
@@ -70,7 +66,8 @@ public async Task Ready_key_publishes_InfoMapReadyEvent_once_per_requester()
7066
// Capture publishes off a substituted bus: the real InMemoryEventBus drops events published
7167
// before a subscriber is active, so collecting via a background subscription races the
7268
// service's first tick and flakes on slow CI runners.
73-
var (bus, received) = CapturingBus();
69+
using var bus = CapturingBus();
70+
var received = bus.Received;
7471

7572
var service = new InfoMapHostedService(
7673
bus, coordinator, driver, query, ShortPollOptions(),
@@ -79,11 +76,7 @@ public async Task Ready_key_publishes_InfoMapReadyEvent_once_per_requester()
7976
await service.StartAsync(CancellationToken.None);
8077
try
8178
{
82-
var deadline = DateTimeOffset.UtcNow.AddSeconds(5);
83-
while (DateTimeOffset.UtcNow < deadline && received.Count < 2)
84-
{
85-
await Task.Delay(20);
86-
}
79+
await bus.WaitForAsync(2);
8780
}
8881
finally
8982
{
@@ -125,7 +118,8 @@ public async Task Tick_registers_and_generates_connected_servers_without_any_con
125118
IReadOnlyList<(ulong GuildId, Guid ServerId)> connectable = [(GuildA, serverId)];
126119
store.ListConnectableServersAsync(Arg.Any<CancellationToken>()).Returns(connectable);
127120

128-
var (bus, received) = CapturingBus();
121+
using var bus = CapturingBus();
122+
var received = bus.Received;
129123

130124
var service = new InfoMapHostedService(
131125
bus, coordinator, driver, query, ShortPollOptions(),
@@ -134,11 +128,7 @@ public async Task Tick_registers_and_generates_connected_servers_without_any_con
134128
await service.StartAsync(CancellationToken.None);
135129
try
136130
{
137-
var deadline = DateTimeOffset.UtcNow.AddSeconds(5);
138-
while (DateTimeOffset.UtcNow < deadline && received.IsEmpty)
139-
{
140-
await Task.Delay(20);
141-
}
131+
await bus.WaitForAsync(1);
142132
}
143133
finally
144134
{
@@ -179,19 +169,19 @@ public async Task Connect_registers_the_servers_world_key_with_the_coordinator()
179169
GenerationPollInterval = TimeSpan.FromSeconds(30) // must not need to fire for this test to pass
180170
}
181171
});
172+
// Signals the registration instead of polling for it: a poll loop needs the thread pool to take its
173+
// next turn, so on a saturated CI runner the whole deadline can elapse inside one Task.Delay.
174+
var registered = new SignallingCoordinator(coordinator);
182175
var service = new InfoMapHostedService(
183-
bus, coordinator, driver, query, pollOptions,
176+
bus, registered, driver, query, pollOptions,
184177
ScopeFactory(connectionStore), NullLogger<InfoMapHostedService>.Instance);
185178

186179
await service.StartAsync(CancellationToken.None);
187180
try
188181
{
189-
var deadline = DateTimeOffset.UtcNow.AddSeconds(5);
190-
while (DateTimeOffset.UtcNow < deadline && coordinator.PendingKeys().Count == 0)
191-
{
192-
await bus.PublishAsync(new ConnectionStatusChangedEvent(GuildA, serverId, true, false));
193-
await Task.Delay(20);
194-
}
182+
// Published exactly once: StartAsync subscribes before it returns, so no event can be dropped.
183+
await bus.PublishAsync(new ConnectionStatusChangedEvent(GuildA, serverId, true, false));
184+
await registered.Registered.WaitAsync(TimeSpan.FromSeconds(30));
195185
}
196186
finally
197187
{
@@ -202,16 +192,51 @@ public async Task Connect_registers_the_servers_world_key_with_the_coordinator()
202192
Assert.Contains((GuildA, serverId), coordinator.Requesters(Key));
203193
}
204194

205-
private sealed class CapturingEventBus : IEventBus
195+
/// <summary>Forwards to a real coordinator and completes <see cref="Registered"/> on the first Register.</summary>
196+
/// <param name="inner">The coordinator that holds the real state.</param>
197+
private sealed class SignallingCoordinator(IRustMapsMapCoordinator inner) : IRustMapsMapCoordinator
198+
{
199+
private readonly TaskCompletionSource _registered =
200+
new(TaskCreationOptions.RunContinuationsAsynchronously);
201+
202+
public Task Registered => _registered.Task;
203+
204+
public void Register(RustMapsMapKey key, ulong guildId, Guid serverId)
205+
{
206+
inner.Register(key, guildId, serverId);
207+
_registered.TrySetResult();
208+
}
209+
210+
public RustMapsMapSnapshot Snapshot(RustMapsMapKey key) => inner.Snapshot(key);
211+
212+
public bool TrySetGenerating(RustMapsMapKey key, string? mapId) => inner.TrySetGenerating(key, mapId);
213+
214+
public void SetReady(RustMapsMapKey key, RustMapsReadyMap ready) => inner.SetReady(key, ready);
215+
216+
public void SetFailed(RustMapsMapKey key) => inner.SetFailed(key);
217+
218+
public void SetLimitReached(RustMapsMapKey key) => inner.SetLimitReached(key);
219+
220+
public IReadOnlyList<RustMapsMapKey> PendingKeys() => inner.PendingKeys();
221+
222+
public IReadOnlyList<(ulong Guild, Guid Server)> Requesters(RustMapsMapKey key) => inner.Requesters(key);
223+
}
224+
225+
private sealed class CapturingEventBus : IEventBus, IDisposable
206226
{
227+
private readonly SemaphoreSlim _published = new(0);
228+
207229
public ConcurrentQueue<InfoMapReadyEvent> Received { get; } = new();
208230

231+
public void Dispose() => _published.Dispose();
232+
209233
public ValueTask PublishAsync<TEvent>(TEvent @event, CancellationToken cancellationToken = default)
210234
where TEvent : notnull
211235
{
212236
if (@event is InfoMapReadyEvent ready)
213237
{
214238
Received.Enqueue(ready);
239+
_published.Release();
215240
}
216241

217242
return ValueTask.CompletedTask;
@@ -220,5 +245,27 @@ public ValueTask PublishAsync<TEvent>(TEvent @event, CancellationToken cancellat
220245
public IAsyncEnumerable<TEvent> SubscribeAsync<TEvent>(CancellationToken cancellationToken = default)
221246
where TEvent : notnull =>
222247
AsyncEnumerable.Empty<TEvent>(); // No live subscriptions needed: assertions read Received.
248+
249+
/// <summary>
250+
/// Waits for <paramref name="count"/> events. Signal-based rather than a poll loop: polling needs the
251+
/// thread pool to take its next turn, so on a saturated runner a whole wall-clock deadline can elapse
252+
/// inside a single Task.Delay. A timeout here is not asserted on — the caller's assertions report it.
253+
/// </summary>
254+
/// <param name="count">How many events to wait for.</param>
255+
public async Task WaitForAsync(int count)
256+
{
257+
using var timeout = new CancellationTokenSource(TimeSpan.FromSeconds(30));
258+
try
259+
{
260+
for (var i = 0; i < count; i++)
261+
{
262+
await _published.WaitAsync(timeout.Token);
263+
}
264+
}
265+
catch (OperationCanceledException)
266+
{
267+
// Timed out; the caller asserts on what actually arrived.
268+
}
269+
}
223270
}
224271
}

0 commit comments

Comments
 (0)