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>

Author: HandyS11

src/RustPlusBot.Abstractions/Events/EventBusConsumption.cs

@@ -8,7 +8,7 @@ namespace RustPlusBot.Abstractions.Events;
 /// and then <em>exits</em>: an exception escaping a handler ends the subscription for the rest of the
 /// process, and the feature it drives goes silently dead until the host restarts. Handlers here talk to
 /// Discord and to the Rust+ server, where a timeout or a 5xx is routine rather than exceptional, so that
-/// trade is never the one we want. Consuming through <see cref="ConsumeAsync{TEvent}"/> keeps the
+/// trade is never the one we want. Consuming through <c>ConsumeAsync</c> keeps the
 /// subscription alive: a failed handler costs its own event and nothing more.
 /// </remarks>
 public static class EventBusConsumption
@@ -23,18 +23,48 @@ public static class EventBusConsumption
     /// <param name="onHandlerFailure">Reports a handler failure (typically a log call).</param>
     /// <param name="cancellationToken">Ends the subscription when cancelled.</param>
     /// <returns>A task that completes when the subscription ends.</returns>
-    public static async Task ConsumeAsync<TEvent>(
+    public static Task ConsumeAsync<TEvent>(
         this IEventBus eventBus,
         Func<TEvent, CancellationToken, Task> handle,
         Action<Exception> onHandlerFailure,
         CancellationToken cancellationToken)
         where TEvent : notnull
     {
         ArgumentNullException.ThrowIfNull(eventBus);
+
+        return eventBus.SubscribeAsync<TEvent>(cancellationToken)
+            .ConsumeAsync(handle, onHandlerFailure, cancellationToken);
+    }
+
+    /// <summary>
+    /// Consumes an already-established subscription, reporting and skipping the events whose handler throws.
+    /// Only cancellation of <paramref name="cancellationToken"/> ends the subscription.
+    /// </summary>
+    /// <remarks>
+    /// Take this overload when the subscription must exist before the caller returns — a hosted service that
+    /// subscribes inside its background <c>Task.Run</c> is only subscribed once that task is scheduled, and
+    /// every event published in the meantime is dropped. Call
+    /// <see cref="IEventBus.SubscribeAsync{TEvent}"/> synchronously in <c>StartAsync</c> (it registers the
+    /// subscription eagerly) and hand the stream here.
+    /// </remarks>
+    /// <typeparam name="TEvent">The event type being consumed.</typeparam>
+    /// <param name="events">The subscription stream to drain.</param>
+    /// <param name="handle">Handles one event; its failures are reported, not propagated.</param>
+    /// <param name="onHandlerFailure">Reports a handler failure (typically a log call).</param>
+    /// <param name="cancellationToken">Ends the subscription when cancelled.</param>
+    /// <returns>A task that completes when the subscription ends.</returns>
+    public static async Task ConsumeAsync<TEvent>(
+        this IAsyncEnumerable<TEvent> events,
+        Func<TEvent, CancellationToken, Task> handle,
+        Action<Exception> onHandlerFailure,
+        CancellationToken cancellationToken)
+        where TEvent : notnull
+    {
+        ArgumentNullException.ThrowIfNull(events);
         ArgumentNullException.ThrowIfNull(handle);
         ArgumentNullException.ThrowIfNull(onHandlerFailure);
 
-        await foreach (var evt in eventBus.SubscribeAsync<TEvent>(cancellationToken).ConfigureAwait(false))
+        await foreach (var evt in events.ConfigureAwait(false))
         {
             try
             {

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

@@ -46,7 +46,12 @@ internal sealed partial class InfoMapHostedService(
     /// <inheritdoc />
     public Task StartAsync(CancellationToken cancellationToken)
     {
-        _statusLoop = Task.Run(() => ConsumeConnectionStatusAsync(_cts.Token), CancellationToken.None);
+        // Subscribe here rather than inside the loop task: Task.Run only queues the body, so a service that
+        // subscribed in there would stay unsubscribed until the pool schedules it — and every
+        // ConnectionStatusChangedEvent published in that window is dropped. On a saturated host that window
+        // is seconds long, which is exactly when connections are being established.
+        var statusEvents = eventBus.SubscribeAsync<ConnectionStatusChangedEvent>(_cts.Token);
+        _statusLoop = Task.Run(() => ConsumeConnectionStatusAsync(statusEvents, _cts.Token), CancellationToken.None);
         _tickLoop = Task.Run(() => RunTickAsync(_cts.Token), CancellationToken.None);
         return Task.CompletedTask;
     }
@@ -159,11 +164,13 @@ await eventBus.PublishAsync(new InfoMapReadyEvent(guild, server), cancellationTo
         }
     }
 
-    private async Task ConsumeConnectionStatusAsync(CancellationToken cancellationToken)
+    private async Task ConsumeConnectionStatusAsync(
+        IAsyncEnumerable<ConnectionStatusChangedEvent> statusEvents,
+        CancellationToken cancellationToken)
     {
         try
         {
-            await eventBus.ConsumeAsync<ConnectionStatusChangedEvent>(OnConnectionStatusAsync,
+            await statusEvents.ConsumeAsync(OnConnectionStatusAsync,
                     ex => LogHandlerFailed(logger, ex, nameof(ConnectionStatusChangedEvent)), cancellationToken)
                 .ConfigureAwait(false);
         }

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

@@ -146,7 +146,7 @@ private async Task OnChannelDestroyedAsync(SocketChannel channel)
 
     /// <summary>
     /// Reconciles one server in its own scope. Every consumer below runs this through
-    /// <see cref="EventBusConsumption.ConsumeAsync{TEvent}"/>, which absorbs its failures: the reconcile
+    /// <c>EventBusConsumption.ConsumeAsync</c>, 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.
     /// </summary>

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

@@ -38,15 +38,11 @@ private static IOptions<MapOptions> ShortPollOptions() => Options.Create(new Map
     });
 
     /// <summary>
-    /// A fake bus that records every published <see cref="InfoMapReadyEvent"/>. Deterministic, unlike
-    /// subscribing to the real <see cref="InMemoryEventBus"/> from a background task: that bus drops
-    /// events published before the subscriber is active, which races the service's first tick.
+    /// A fake bus that records every published <see cref="InfoMapReadyEvent"/> and signals each arrival.
+    /// Deterministic, unlike subscribing to the real <see cref="InMemoryEventBus"/> from a background task:
+    /// that bus drops events published before the subscriber is active, which races the service's first tick.
     /// </summary>
-    private static (IEventBus Bus, ConcurrentQueue<InfoMapReadyEvent> Received) CapturingBus()
-    {
-        var bus = new CapturingEventBus();
-        return (bus, bus.Received);
-    }
+    private static CapturingEventBus CapturingBus() => new();
 
     [Fact]
     public async Task Ready_key_publishes_InfoMapReadyEvent_once_per_requester()
@@ -70,7 +66,8 @@ public async Task Ready_key_publishes_InfoMapReadyEvent_once_per_requester()
         // Capture publishes off a substituted bus: the real InMemoryEventBus drops events published
         // before a subscriber is active, so collecting via a background subscription races the
         // service's first tick and flakes on slow CI runners.
-        var (bus, received) = CapturingBus();
+        using var bus = CapturingBus();
+        var received = bus.Received;
 
         var service = new InfoMapHostedService(
             bus, coordinator, driver, query, ShortPollOptions(),
@@ -79,11 +76,7 @@ public async Task Ready_key_publishes_InfoMapReadyEvent_once_per_requester()
         await service.StartAsync(CancellationToken.None);
         try
         {
-            var deadline = DateTimeOffset.UtcNow.AddSeconds(5);
-            while (DateTimeOffset.UtcNow < deadline && received.Count < 2)
-            {
-                await Task.Delay(20);
-            }
+            await bus.WaitForAsync(2);
         }
         finally
         {
@@ -125,7 +118,8 @@ public async Task Tick_registers_and_generates_connected_servers_without_any_con
         IReadOnlyList<(ulong GuildId, Guid ServerId)> connectable = [(GuildA, serverId)];
         store.ListConnectableServersAsync(Arg.Any<CancellationToken>()).Returns(connectable);
 
-        var (bus, received) = CapturingBus();
+        using var bus = CapturingBus();
+        var received = bus.Received;
 
         var service = new InfoMapHostedService(
             bus, coordinator, driver, query, ShortPollOptions(),
@@ -134,11 +128,7 @@ public async Task Tick_registers_and_generates_connected_servers_without_any_con
         await service.StartAsync(CancellationToken.None);
         try
         {
-            var deadline = DateTimeOffset.UtcNow.AddSeconds(5);
-            while (DateTimeOffset.UtcNow < deadline && received.IsEmpty)
-            {
-                await Task.Delay(20);
-            }
+            await bus.WaitForAsync(1);
         }
         finally
         {
@@ -179,19 +169,19 @@ public async Task Connect_registers_the_servers_world_key_with_the_coordinator()
                 GenerationPollInterval = TimeSpan.FromSeconds(30) // must not need to fire for this test to pass
             }
         });
+        // Signals the registration instead of polling for it: a poll loop needs the thread pool to take its
+        // next turn, so on a saturated CI runner the whole deadline can elapse inside one Task.Delay.
+        var registered = new SignallingCoordinator(coordinator);
         var service = new InfoMapHostedService(
-            bus, coordinator, driver, query, pollOptions,
+            bus, registered, driver, query, pollOptions,
             ScopeFactory(connectionStore), NullLogger<InfoMapHostedService>.Instance);
 
         await service.StartAsync(CancellationToken.None);
         try
         {
-            var deadline = DateTimeOffset.UtcNow.AddSeconds(5);
-            while (DateTimeOffset.UtcNow < deadline && coordinator.PendingKeys().Count == 0)
-            {
-                await bus.PublishAsync(new ConnectionStatusChangedEvent(GuildA, serverId, true, false));
-                await Task.Delay(20);
-            }
+            // Published exactly once: StartAsync subscribes before it returns, so no event can be dropped.
+            await bus.PublishAsync(new ConnectionStatusChangedEvent(GuildA, serverId, true, false));
+            await registered.Registered.WaitAsync(TimeSpan.FromSeconds(30));
         }
         finally
         {
@@ -202,16 +192,51 @@ public async Task Connect_registers_the_servers_world_key_with_the_coordinator()
         Assert.Contains((GuildA, serverId), coordinator.Requesters(Key));
     }
 
-    private sealed class CapturingEventBus : IEventBus
+    /// <summary>Forwards to a real coordinator and completes <see cref="Registered"/> on the first Register.</summary>
+    /// <param name="inner">The coordinator that holds the real state.</param>
+    private sealed class SignallingCoordinator(IRustMapsMapCoordinator inner) : IRustMapsMapCoordinator
+    {
+        private readonly TaskCompletionSource _registered =
+            new(TaskCreationOptions.RunContinuationsAsynchronously);
+
+        public Task Registered => _registered.Task;
+
+        public void Register(RustMapsMapKey key, ulong guildId, Guid serverId)
+        {
+            inner.Register(key, guildId, serverId);
+            _registered.TrySetResult();
+        }
+
+        public RustMapsMapSnapshot Snapshot(RustMapsMapKey key) => inner.Snapshot(key);
+
+        public bool TrySetGenerating(RustMapsMapKey key, string? mapId) => inner.TrySetGenerating(key, mapId);
+
+        public void SetReady(RustMapsMapKey key, RustMapsReadyMap ready) => inner.SetReady(key, ready);
+
+        public void SetFailed(RustMapsMapKey key) => inner.SetFailed(key);
+
+        public void SetLimitReached(RustMapsMapKey key) => inner.SetLimitReached(key);
+
+        public IReadOnlyList<RustMapsMapKey> PendingKeys() => inner.PendingKeys();
+
+        public IReadOnlyList<(ulong Guild, Guid Server)> Requesters(RustMapsMapKey key) => inner.Requesters(key);
+    }
+
+    private sealed class CapturingEventBus : IEventBus, IDisposable
     {
+        private readonly SemaphoreSlim _published = new(0);
+
         public ConcurrentQueue<InfoMapReadyEvent> Received { get; } = new();
 
+        public void Dispose() => _published.Dispose();
+
         public ValueTask PublishAsync<TEvent>(TEvent @event, CancellationToken cancellationToken = default)
             where TEvent : notnull
         {
             if (@event is InfoMapReadyEvent ready)
             {
                 Received.Enqueue(ready);
+                _published.Release();
             }
 
             return ValueTask.CompletedTask;
@@ -220,5 +245,27 @@ public ValueTask PublishAsync<TEvent>(TEvent @event, CancellationToken cancellat
         public IAsyncEnumerable<TEvent> SubscribeAsync<TEvent>(CancellationToken cancellationToken = default)
             where TEvent : notnull =>
             AsyncEnumerable.Empty<TEvent>(); // No live subscriptions needed: assertions read Received.
+
+        /// <summary>
+        /// Waits for <paramref name="count"/> events. Signal-based rather than a poll loop: polling needs the
+        /// thread pool to take its next turn, so on a saturated runner a whole wall-clock deadline can elapse
+        /// inside a single Task.Delay. A timeout here is not asserted on — the caller's assertions report it.
+        /// </summary>
+        /// <param name="count">How many events to wait for.</param>
+        public async Task WaitForAsync(int count)
+        {
+            using var timeout = new CancellationTokenSource(TimeSpan.FromSeconds(30));
+            try
+            {
+                for (var i = 0; i < count; i++)
+                {
+                    await _published.WaitAsync(timeout.Token);
+                }
+            }
+            catch (OperationCanceledException)
+            {
+                // Timed out; the caller asserts on what actually arrived.
+            }
+        }
     }
 }