diff --git a/Directory.Packages.props b/Directory.Packages.props index 47d6e06e..def38fa7 100644 --- a/Directory.Packages.props +++ b/Directory.Packages.props @@ -15,8 +15,8 @@ - - + + diff --git a/docs/superpowers/plans/2026-07-24-team-changed-push.md b/docs/superpowers/plans/2026-07-24-team-changed-push.md new file mode 100644 index 00000000..66f17b32 --- /dev/null +++ b/docs/superpowers/plans/2026-07-24-team-changed-push.md @@ -0,0 +1,862 @@ +# Push-driven team state via `team_changed` — Implementation Plan + +> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking. + +**Goal:** Move team-state detection (connect/disconnect/death/respawn/AFK) from the 2–5 s poll inside `PollMarkersAsync` onto RustPlusApi beta.6's `OnTeamChanged` push event, keeping a low-frequency team poll as an AFK safety tick. Marker/rig polling is unchanged. + +**Architecture:** RustPlusApi beta.6 now dispatches the `team_changed` broadcast as an `OnTeamChanged` event whose `TeamInfo` payload is the same model the polled `GetTeamInfoAsync` already maps. `RustPlusSocketSource` surfaces it as a new `IRustServerConnection.TeamChanged` event. `ConnectionSupervisor` runs `TeamStateTracker.Diff` on each pushed snapshot (instant transitions) and also on a slow `PollTeamAsync` loop (guarantees `Diff` runs so "BecameAfk" can't stall during broadcast silence). The team block is removed from the fast marker poll. + +**Tech Stack:** C# / .NET 10, xUnit, NSubstitute, RustPlusApi 2.0.0-beta.6. Build/test via `dtk` (DotnetTokenKiller). + +## Global Constraints + +- RustPlusApi and RustPlusApi.Fcm are pinned to `2.0.0-beta.6` in `Directory.Packages.props` (central package management — versions live there, `PackageReference` entries carry no `Version`). +- `TeamStateTracker` logic MUST NOT change — it is already lock-guarded and its AFK model is correct; this work only changes *how often* and *from where* `Diff` is called. +- Fire-and-forget publish handlers use `_shutdown.Token`, guard on `_disposed`, and swallow all exceptions into a `LoggerMessage` (mirror `PublishClanStateAsync`, `ConnectionSupervisor.cs:1189`). +- Broad `catch (Exception)` blocks in background loops keep the existing `#pragma warning disable CA1031` + justifying comment convention. +- Build: `dtk build`. Test one project with a filter: `dtk test tests/RustPlusBot.Features.Connections.Tests/RustPlusBot.Features.Connections.Tests.csproj --filter ""`. + +--- + +### Task 1: Upgrade RustPlusApi to beta.6 + +**Files:** +- Modify: `Directory.Packages.props:18-19` + +**Interfaces:** +- Produces: the `RustPlus.OnTeamChanged` event (`EventHandler`) and the `RustPlusApi.Data.Events.TeamChangedEventArg { ulong PlayerId; RustPlusApi.Data.TeamInfo TeamInfo }` record, available to later tasks. + +- [ ] **Step 1: Bump both package versions** + +In `Directory.Packages.props`, change the two lines: + +```xml + + +``` + +- [ ] **Step 2: Restore + build the whole solution to confirm beta.6 is source-compatible** + +Run: `dtk build` +Expected: build succeeds. beta.6 keeps the same public surface for the methods the bot already uses (`GetMapMarkersAsync`, `GetTeamInfoAsync`, `ProcessRequestAsync`, `ParseNotification`); it only *adds* `OnTeamChanged`. If any pre-existing call site fails to compile, fix it minimally in this task before proceeding. + +- [ ] **Step 3: Run the full connections test project to confirm no runtime regression from the bump** + +Run: `dtk test tests/RustPlusBot.Features.Connections.Tests/RustPlusBot.Features.Connections.Tests.csproj` +Expected: PASS (same green set as before the bump). + +- [ ] **Step 4: Commit** + +```bash +git add Directory.Packages.props +git commit -m "build(deps): bump RustPlusApi to 2.0.0-beta.6 (team_changed dispatch)" +``` + +--- + +### Task 2: Extract `TeamInfoMapping.ToSnapshot` and unit-test it + +Extract the inline `TeamInfo → TeamInfoSnapshot` mapping (currently in `RustPlusServerConnection.GetTeamInfoAsync`, `RustPlusSocketSource.cs:339-353`) into a reusable, directly-testable helper so both the poll and the new push handler share one mapping. + +**Files:** +- Create: `src/RustPlusBot.Features.Connections/Listening/TeamInfoMapping.cs` +- Modify: `src/RustPlusBot.Features.Connections/Listening/RustPlusSocketSource.cs:339-353` +- Modify: `tests/RustPlusBot.Features.Connections.Tests/RustPlusBot.Features.Connections.Tests.csproj` +- Create: `tests/RustPlusBot.Features.Connections.Tests/TeamInfoMappingTests.cs` + +**Interfaces:** +- Produces: `internal static class TeamInfoMapping` with `public static TeamInfoSnapshot ToSnapshot(RustPlusApi.Data.TeamInfo teamInfo)`. +- Consumes: `RustPlusApi.Data.TeamInfo { ulong LeaderSteamId; IEnumerable? Members; RustPlusApi.Data.Notes.DeathNote? DeathNote }`; `RustPlusApi.Data.MemberInfo { ulong SteamId; string? Name; float X; float Y; bool IsOnline; bool IsAlive; DateTime LastSpawnTime; DateTime LastDeathTime }`; `RustPlusApi.Data.Notes.DeathNote { float X; float Y }`. + +- [ ] **Step 1: Give the test project access to `RustPlusApi.Data` types** + +In `tests/RustPlusBot.Features.Connections.Tests/RustPlusBot.Features.Connections.Tests.csproj`, add inside the first `` (the `PackageReference` group): + +```xml + +``` + +- [ ] **Step 2: Write the failing test** + +Create `tests/RustPlusBot.Features.Connections.Tests/TeamInfoMappingTests.cs`: + +```csharp +using RustPlusBot.Features.Connections.Listening; + +namespace RustPlusBot.Features.Connections.Tests; + +public sealed class TeamInfoMappingTests +{ + [Fact] + public void ToSnapshot_MapsMembersLeaderAndDeathNote() + { + var teamInfo = new RustPlusApi.Data.TeamInfo + { + LeaderSteamId = 100UL, + DeathNote = new RustPlusApi.Data.Notes.DeathNote { X = 12f, Y = 34f }, + Members = + [ + new RustPlusApi.Data.MemberInfo + { + SteamId = 100UL, + Name = "Alice", + X = 1f, + Y = 2f, + IsOnline = true, + IsAlive = false, + LastSpawnTime = new DateTime(2026, 1, 1, 0, 0, 0, DateTimeKind.Unspecified), + LastDeathTime = new DateTime(2026, 1, 2, 0, 0, 0, DateTimeKind.Unspecified), + }, + ], + }; + + var snapshot = TeamInfoMapping.ToSnapshot(teamInfo); + + Assert.Equal(100UL, snapshot.LeaderSteamId); + Assert.Equal((12f, 34f), snapshot.DeathNote); + var m = Assert.Single(snapshot.Members); + Assert.Equal(100UL, m.SteamId); + Assert.Equal("Alice", m.Name); + Assert.True(m.IsOnline); + Assert.False(m.IsAlive); + // Unspecified game timestamps are read as UTC. + Assert.Equal(DateTimeKind.Utc, m.LastDeathTimeUtc.UtcDateTime.Kind); + Assert.Equal(new DateTimeOffset(2026, 1, 2, 0, 0, 0, TimeSpan.Zero), m.LastDeathTimeUtc); + } + + [Fact] + public void ToSnapshot_NullMembersAndNoDeathNote_YieldEmptyMembersAndNullNote() + { + var teamInfo = new RustPlusApi.Data.TeamInfo { LeaderSteamId = 7UL, Members = null, DeathNote = null }; + + var snapshot = TeamInfoMapping.ToSnapshot(teamInfo); + + Assert.Equal(7UL, snapshot.LeaderSteamId); + Assert.Empty(snapshot.Members); + Assert.Null(snapshot.DeathNote); + } +} +``` + +- [ ] **Step 3: Run the test to verify it fails** + +Run: `dtk test tests/RustPlusBot.Features.Connections.Tests/RustPlusBot.Features.Connections.Tests.csproj --filter "FullyQualifiedName~TeamInfoMappingTests"` +Expected: FAIL to compile — `TeamInfoMapping` does not exist. + +- [ ] **Step 4: Create the mapping helper** + +Create `src/RustPlusBot.Features.Connections/Listening/TeamInfoMapping.cs`: + +```csharp +using RustPlusBot.Abstractions.Connections; + +namespace RustPlusBot.Features.Connections.Listening; + +/// Maps a RustPlusApi to the bot's decoupled +/// . Shared by the polled read and the pushed team_changed event so +/// the two paths can never drift. +internal static class TeamInfoMapping +{ + /// Projects a RustPlusApi team-info model onto a . + /// The RustPlusApi team info (from a poll response or a team_changed broadcast). + /// The decoupled snapshot the tracker and events consume. + public static TeamInfoSnapshot ToSnapshot(RustPlusApi.Data.TeamInfo teamInfo) + { + // CONFIRMED (2.0.0-beta.6): MemberInfo exposes SteamId/Name?/X/Y/IsOnline/IsAlive plus + // LastSpawnTime/LastDeathTime as Unspecified-kind DateTimes that are actually UTC. + var members = (teamInfo.Members ?? []) + .Select(m => new TeamMemberSnapshot( + m.SteamId, + m.Name ?? string.Empty, + m.X, + m.Y, + m.IsOnline, + m.IsAlive, + new DateTimeOffset(DateTime.SpecifyKind(m.LastSpawnTime, DateTimeKind.Utc)), + new DateTimeOffset(DateTime.SpecifyKind(m.LastDeathTime, DateTimeKind.Utc)))) + .ToList(); + var deathNote = teamInfo.DeathNote is { } dn + ? ((float X, float Y)?)(dn.X, dn.Y) + : null; + return new TeamInfoSnapshot(teamInfo.LeaderSteamId, members, deathNote); + } +} +``` + +- [ ] **Step 5: Point `GetTeamInfoAsync` at the shared helper** + +In `src/RustPlusBot.Features.Connections/Listening/RustPlusSocketSource.cs`, replace the mapping block (lines 339-353) — everything from `var members = (response.Data.Members ?? [])` through `return new TeamInfoSnapshot(...)` — with: + +```csharp + return TeamInfoMapping.ToSnapshot(response.Data); +``` + +Leave the `if (!response.IsSuccess || response.Data is null) { return null; }` guard above it untouched. + +- [ ] **Step 6: Run the tests to verify they pass** + +Run: `dtk test tests/RustPlusBot.Features.Connections.Tests/RustPlusBot.Features.Connections.Tests.csproj --filter "FullyQualifiedName~TeamInfoMappingTests"` +Expected: PASS (both tests). + +- [ ] **Step 7: Run the full connections project to confirm the refactor broke nothing** + +Run: `dtk test tests/RustPlusBot.Features.Connections.Tests/RustPlusBot.Features.Connections.Tests.csproj` +Expected: PASS. + +- [ ] **Step 8: Commit** + +```bash +git add src/RustPlusBot.Features.Connections/Listening/TeamInfoMapping.cs \ + src/RustPlusBot.Features.Connections/Listening/RustPlusSocketSource.cs \ + tests/RustPlusBot.Features.Connections.Tests/TeamInfoMappingTests.cs \ + tests/RustPlusBot.Features.Connections.Tests/RustPlusBot.Features.Connections.Tests.csproj +git commit -m "refactor(connections): extract shared TeamInfoMapping.ToSnapshot" +``` + +--- + +### Task 3: Add the `TeamChanged` event to the connection abstraction and wire it up + +**Files:** +- Modify: `src/RustPlusBot.Features.Connections/Listening/IRustServerConnection.cs:172` (end of interface) +- Modify: `src/RustPlusBot.Features.Connections/Listening/RustPlusSocketSource.cs` (ctor ~172, `DisposeAsync` ~768, event declaration, new handler) +- Modify: `tests/RustPlusBot.Features.Connections.Tests/Fakes/FakeRustSocketSource.cs` + +**Interfaces:** +- Produces: `event EventHandler? IRustServerConnection.TeamChanged`; on the fake, `FakeConnection.RaiseTeamChanged(TeamInfoSnapshot snapshot)` and `int FakeConnection.TeamInfoCallCount`. +- Consumes: `TeamInfoMapping.ToSnapshot` (Task 2); `RustPlusApi.Data.Events.TeamChangedEventArg` (Task 1). + +- [ ] **Step 1: Declare the event on the interface** + +In `IRustServerConnection.cs`, add after the `StorageMonitorTriggered` event (line 171), before the closing brace: + +```csharp + + /// + /// Raised when the server pushes a team_changed broadcast (member join/leave, online/offline, + /// death/respawn, movement, leader change). Carries the full team snapshot — the same shape a + /// poll returns. + /// + event EventHandler? TeamChanged; +``` + +- [ ] **Step 2: Declare + raise the event in the real connection** + +In `RustPlusSocketSource.cs`, inside the `RustPlusServerConnection` class, add the event declaration next to the other public events (near the `SmartDeviceTriggered`/`StorageMonitorTriggered` declarations — search for `public event EventHandler? StorageMonitorTriggered;` and add below it): + +```csharp + public event EventHandler? TeamChanged; +``` + +Add the private handler next to `OnClanChanged` (near line 834): + +```csharp + private void OnTeamChanged(object? sender, RustPlusApi.Data.Events.TeamChangedEventArg e) => + TeamChanged?.Invoke(this, TeamInfoMapping.ToSnapshot(e.TeamInfo)); +``` + +Subscribe in the constructor event block (after `_rustPlus.OnClanChanged += OnClanChanged;`, line 172): + +```csharp + _rustPlus.OnTeamChanged += OnTeamChanged; +``` + +Unsubscribe in `DisposeAsync` (after `_rustPlus.OnClanChanged -= OnClanChanged;`, line 768): + +```csharp + _rustPlus.OnTeamChanged -= OnTeamChanged; +``` + +- [ ] **Step 3: Write the failing fake-connection test** + +Add to `tests/RustPlusBot.Features.Connections.Tests/RustPlusSocketSourceTests.cs` (new `[Fact]` inside the existing class): + +```csharp + [Fact] + public void FakeConnection_RaiseTeamChanged_InvokesSubscribers() + { + var source = new Fakes.FakeRustSocketSource(); + source.EnqueueConnect(SocketConnectOutcome.Connected); + var connection = (Fakes.FakeRustSocketSource.FakeConnection)source.Create("127.0.0.1", 28015, 1UL, "1"); + + TeamInfoSnapshot? received = null; + connection.TeamChanged += (_, s) => received = s; + var snapshot = new TeamInfoSnapshot(5UL, []); + connection.RaiseTeamChanged(snapshot); + + Assert.Same(snapshot, received); + } +``` + +Add `using RustPlusBot.Abstractions.Connections;` to the file's usings if not present. + +- [ ] **Step 4: Run it to verify it fails** + +Run: `dtk test tests/RustPlusBot.Features.Connections.Tests/RustPlusBot.Features.Connections.Tests.csproj --filter "FullyQualifiedName~FakeConnection_RaiseTeamChanged"` +Expected: FAIL to compile — `FakeConnection` has no `TeamChanged` / `RaiseTeamChanged`. + +- [ ] **Step 5: Extend the fake connection** + +In `tests/.../Fakes/FakeRustSocketSource.cs`, inside `FakeConnection`: + +Add the event next to the other event declarations (after `StorageMonitorTriggered`, line 296): + +```csharp + + /// Raised by to simulate a pushed team_changed broadcast. + public event EventHandler? TeamChanged; +``` + +Add a call counter property next to `TeamResult` (line 208): + +```csharp + + /// Number of times has been called (to assert team info is + /// no longer polled on the fast marker cadence). + public int TeamInfoCallCount { get; private set; } +``` + +Change `GetTeamInfoAsync` (line 310-311) to count calls: + +```csharp + public Task GetTeamInfoAsync(TimeSpan timeout, CancellationToken cancellationToken) + { + TeamInfoCallCount++; + return Task.FromResult(TeamResult); + } +``` + +Add the raise helper next to `RaiseClanChanged` (line 454): + +```csharp + + /// Raises to simulate a pushed team_changed broadcast. + /// The team snapshot to deliver. + public void RaiseTeamChanged(TeamInfoSnapshot snapshot) => TeamChanged?.Invoke(this, snapshot); +``` + +Finally, add a pre-`Create` setup hook on the OUTER `FakeRustSocketSource` (not the connection) so a test can configure a connection's `TeamResult` before the supervisor's poll loop starts — this eliminates the race between the test assigning `TeamResult` and the immediate priming poll. Add the property next to `LastConnection` (line 44): + +```csharp + /// Optional hook invoked on each new at creation time, before it is + /// returned and the poll loops start. Lets a test stage (or null) + /// without racing the immediate priming team poll. + internal Action? LastConnectionSetup { get; set; } +``` + +and invoke it inside `Create`, immediately before `LastConnection = connection;` (line 101): + +```csharp + LastConnectionSetup?.Invoke(connection); +``` + +- [ ] **Step 6: Run it to verify it passes** + +Run: `dtk test tests/RustPlusBot.Features.Connections.Tests/RustPlusBot.Features.Connections.Tests.csproj --filter "FullyQualifiedName~FakeConnection_RaiseTeamChanged"` +Expected: PASS. + +- [ ] **Step 7: Build the solution (the new interface member must be satisfied everywhere)** + +Run: `dtk build` +Expected: build succeeds — `RejectedConnection` and any other `IRustServerConnection` implementor compiles. If `RejectedConnection` (`RustPlusSocketSource.cs:35`) needs the member, add `public event EventHandler? TeamChanged;` to it (auto-implemented, never raised). + +- [ ] **Step 8: Commit** + +```bash +git add src/RustPlusBot.Features.Connections/Listening/IRustServerConnection.cs \ + src/RustPlusBot.Features.Connections/Listening/RustPlusSocketSource.cs \ + tests/RustPlusBot.Features.Connections.Tests/Fakes/FakeRustSocketSource.cs \ + tests/RustPlusBot.Features.Connections.Tests/RustPlusSocketSourceTests.cs +git commit -m "feat(connections): surface team_changed as IRustServerConnection.TeamChanged" +``` + +--- + +### Task 4: Add the `TeamPollInterval` option + +**Files:** +- Modify: `src/RustPlusBot.Features.Connections/ConnectionOptions.cs:43` (after `AfkEpsilon`) +- Modify: `tests/RustPlusBot.Features.Connections.Tests/ConnectionOptionsTests.cs` + +**Interfaces:** +- Produces: `ConnectionOptions.TeamPollInterval` (`TimeSpan`, default 30 s). + +- [ ] **Step 1: Write the failing default-value test** + +In `ConnectionOptionsTests.cs`, add an assertion (place it beside the existing marker-interval assertions around line 12-13): + +```csharp + Assert.Equal(TimeSpan.FromSeconds(30), o.TeamPollInterval); +``` + +- [ ] **Step 2: Run it to verify it fails** + +Run: `dtk test tests/RustPlusBot.Features.Connections.Tests/RustPlusBot.Features.Connections.Tests.csproj --filter "FullyQualifiedName~ConnectionOptionsTests"` +Expected: FAIL to compile — `TeamPollInterval` does not exist. + +- [ ] **Step 3: Add the option** + +In `ConnectionOptions.cs`, add after the `AfkEpsilon` property (line 43): + +```csharp + + /// + /// How often to poll team info as an AFK safety tick and self-heal, now that live team changes arrive + /// via the pushed team_changed event. Presence/death events are instant; this only bounds how long + /// a still player in a broadcast-silent team can take to be flagged AFK (worst case AfkThreshold + this). + /// Default 30s. + /// + public TimeSpan TeamPollInterval { get; set; } = TimeSpan.FromSeconds(30); +``` + +- [ ] **Step 4: Run it to verify it passes** + +Run: `dtk test tests/RustPlusBot.Features.Connections.Tests/RustPlusBot.Features.Connections.Tests.csproj --filter "FullyQualifiedName~ConnectionOptionsTests"` +Expected: PASS. + +- [ ] **Step 5: Commit** + +```bash +git add src/RustPlusBot.Features.Connections/ConnectionOptions.cs \ + tests/RustPlusBot.Features.Connections.Tests/ConnectionOptionsTests.cs +git commit -m "feat(connections): add TeamPollInterval (AFK safety-tick cadence)" +``` + +--- + +### Task 5: Drive team state from the push event + slow tick; remove it from the marker poll + +This is the core wiring. `ConnectionSupervisor.cs` gets: a `DimensionsHolder`, a shared `PublishTeamStateAsync` helper, an `OnTeamChanged` subscription, a new `PollTeamAsync` loop, and the removal of the team block from `PollMarkersAsync`. + +**Files:** +- Modify: `src/RustPlusBot.Features.Connections/Supervisor/ConnectionSupervisor.cs` (RunAsync ~608-660; `PollMarkersAsync` 725-786; add helpers + logger messages) +- Modify: `tests/RustPlusBot.Features.Connections.Tests/ConnectionSupervisorTests.cs` (harness option + new tests + a `WaitUntilAsync` helper) + +**Interfaces:** +- Consumes: `IRustServerConnection.TeamChanged` (Task 3); `ConnectionOptions.TeamPollInterval` (Task 4); `TeamStateTracker.Diff` (unchanged); `FakeConnection.RaiseTeamChanged` / `TeamInfoCallCount` / `TeamResult` (Task 3); the harness `CreateHarness`, `SeedAsync`, `WaitForStateAsync` (`ConnectionSupervisorTests.cs`). +- Produces: no new public API; behaviour change only. + +- [ ] **Step 1: Add an optional `teamPollInterval` to the test harness** + +In `ConnectionSupervisorTests.cs`, change the harness signature (line 24) and its options block. Replace: + +```csharp + private static Harness CreateHarness(FakeRustSocketSource source) + { +``` + +with: + +```csharp + private static Harness CreateHarness(FakeRustSocketSource source, TimeSpan? teamPollInterval = null) + { +``` + +and in the `Options.Create(new ConnectionOptions { ... })` block (lines 61-71) add one line after `LivenessPollInterval = TimeSpan.FromMilliseconds(20),`: + +```csharp + TeamPollInterval = teamPollInterval ?? TimeSpan.FromMilliseconds(20), +``` + +- [ ] **Step 2: Write the failing push-event test** + +Add to `ConnectionSupervisorTests.cs`. It connects, raises a `TeamChanged` prime snapshot, then a second snapshot where a member has gone offline, and asserts a `Disconnect` transition is published. (First `Diff` primes silently; the second yields the transition.) + +```csharp + [Fact] + public async Task TeamChanged_push_publishes_player_state_transition() + { + var source = new FakeRustSocketSource(); + source.EnqueueConnect(SocketConnectOutcome.Connected); + source.EnqueueHeartbeat(HeartbeatResult.Ok(1)); + // Make the slow poll inert: TeamResult = null means Diff(null) is a no-op, so the tracker's baseline + // is driven ONLY by the pushed snapshots below — no race between the priming poll and the pushes. + source.LastConnectionSetup = c => c.TeamResult = null; + // Large team-poll interval too, belt-and-suspenders. + await using var h = CreateHarness(source, teamPollInterval: TimeSpan.FromSeconds(30)); + var (serverId, _, _) = await SeedAsync(h.Provider); + + using var cts = new CancellationTokenSource(TimeSpan.FromSeconds(30)); + var captured = new System.Collections.Concurrent.ConcurrentQueue(); + var stream = h.Bus.SubscribeAsync(cts.Token); + var subTask = Task.Run(async () => + { + await foreach (var e in stream) + { + captured.Enqueue(e); + } + }, CancellationToken.None); + + await h.Supervisor.EnsureConnectionAsync(10UL, serverId, cts.Token); + var state = await WaitForStateAsync(h.Provider, serverId, s => s.Status == ConnectionStatus.Connected); + Assert.NotNull(state); + var conn = source.LastConnection!; + + var online = new TeamMemberSnapshot( + 100UL, "Alice", 1f, 1f, IsOnline: true, IsAlive: true, DateTimeOffset.UnixEpoch, DateTimeOffset.UnixEpoch); + var offline = online with { IsOnline = false }; + + conn.RaiseTeamChanged(new TeamInfoSnapshot(100UL, [online])); // prime (silent) + conn.RaiseTeamChanged(new TeamInfoSnapshot(100UL, [offline])); // -> Disconnect + + await WaitUntilAsync(() => !captured.IsEmpty, cts.Token); + + Assert.True(captured.TryDequeue(out var evt)); + var transition = Assert.Single(evt!.Transitions); + Assert.Equal(PlayerTransitionKind.Disconnect, transition.Kind); + Assert.Equal(100UL, transition.SteamId); + + await h.Supervisor.StopAllAsync(); + await cts.CancelAsync(); + try { await subTask; } catch (OperationCanceledException) { /* expected */ } + } +``` + +If the file has no `WaitUntilAsync`, add this private static helper to the class: + +```csharp + private static async Task WaitUntilAsync(Func predicate, CancellationToken ct) + { + while (!predicate()) + { + ct.ThrowIfCancellationRequested(); + await Task.Delay(15, ct); + } + } +``` + +Ensure these usings are present at the top of the file: `using RustPlusBot.Abstractions.Events;` (for `PlayerStateChangedEvent`, `PlayerTransitionKind`). + +- [ ] **Step 3: Run it to verify it fails** + +Run: `dtk test tests/RustPlusBot.Features.Connections.Tests/RustPlusBot.Features.Connections.Tests.csproj --filter "FullyQualifiedName~TeamChanged_push_publishes"` +Expected: FAIL — no `TeamChanged` subscription exists yet, so nothing is published and `WaitUntilAsync` runs to the 30 s timeout (`OperationCanceledException`). + +- [ ] **Step 4: Add the `DimensionsHolder` type** + +In `ConnectionSupervisor.cs`, add near the `LiveSocket` record (line 1579): + +```csharp + /// Mutable, thread-visible holder for the per-connected-window map dimensions. The marker poll + /// resolves these once off the critical path; the team push handler and team poll read them (possibly + /// null before resolution — PlayerStateChangedEvent tolerates a null and renders without a grid ref). + private sealed class DimensionsHolder + { + private volatile MapDimensions? _value; + + public MapDimensions? Value + { + get => _value; + set => _value = value; + } + } +``` + +- [ ] **Step 5: Add the shared publish helper** + +In `ConnectionSupervisor.cs`, add next to `PublishClanStateAsync` (after line ~1210). This is the single Diff-and-publish path used by both the push handler and the slow poll: + +```csharp + private async Task PublishTeamStateAsync( + (ulong Guild, Guid Server) key, + TeamStateTracker tracker, + DimensionsHolder dims, + TeamInfoSnapshot? snapshot) + { + if (_disposed) + { + return; + } + + try + { + var transitions = tracker.Diff(snapshot, clock.UtcNow, _options.AfkThreshold, _options.AfkEpsilon); + if (transitions.Count == 0) + { + return; + } + + var evt = new PlayerStateChangedEvent(key.Guild, key.Server, dims.Value, transitions); + // Supervisor-wide shutdown token, not a per-connection ct: a pushed team change should publish + // regardless of one connection's reconnect cycle (mirrors the chat/clan handlers). + await eventBus.PublishAsync(evt, _shutdown.Token).ConfigureAwait(false); + } + catch (OperationCanceledException) + { + // Shutting down. + } +#pragma warning disable CA1031 // Broad catch: a publish failure must not crash the socket callback or poll. + catch (Exception ex) +#pragma warning restore CA1031 + { + LogPublishTeamStateFailed(logger, ex, key.Server); + } + } +``` + +- [ ] **Step 6: Add the two `LoggerMessage` declarations** + +In `ConnectionSupervisor.cs`, next to the existing `LogMarkerPollFailed` declaration (search for `LogMarkerPollFailed`), add: + +```csharp + [LoggerMessage(Level = LogLevel.Warning, Message = "Team poll for server {ServerId} failed.")] + private static partial void LogTeamPollFailed(ILogger logger, Exception ex, Guid serverId); + + [LoggerMessage(Level = LogLevel.Warning, Message = "Publishing team state for server {ServerId} failed.")] + private static partial void LogPublishTeamStateFailed(ILogger logger, Exception ex, Guid serverId); +``` + +- [ ] **Step 7: Add the `PollTeamAsync` loop** + +In `ConnectionSupervisor.cs`, add after `PollMarkersAsync` (after line 786): + +```csharp + /// + /// Low-frequency team poll that runs the same as the pushed + /// team_changed handler. Live changes arrive via the push event; this loop exists solely to (a) prime the + /// baseline on connect and (b) guarantee Diff runs periodically so a still player in a + /// broadcast-silent team is still flagged AFK. Its first iteration runs immediately (prime), then it waits + /// between iterations. Degrades safely: a failed poll is + /// logged and skipped, never ending the loop. + /// + private async Task PollTeamAsync( + (ulong Guild, Guid Server) key, + IRustServerConnection connection, + TeamStateTracker tracker, + DimensionsHolder dims, + CancellationToken ct) + { + while (!ct.IsCancellationRequested) + { + try + { + var team = await connection.GetTeamInfoAsync(_options.HeartbeatTimeout, ct).ConfigureAwait(false); + await PublishTeamStateAsync(key, tracker, dims, team).ConfigureAwait(false); + } + catch (OperationCanceledException) when (ct.IsCancellationRequested) + { + return; // stopping + } +#pragma warning disable CA1031 // Broad catch: a failed team poll is logged and skipped; the loop survives. + catch (Exception ex) +#pragma warning restore CA1031 + { + LogTeamPollFailed(logger, ex, key.Server); + } + + await Task.Delay(_options.TeamPollInterval, ct).ConfigureAwait(false); + } + } +``` + +- [ ] **Step 8: Remove the team block from `PollMarkersAsync` and thread the dims holder** + +In `ConnectionSupervisor.cs`, change the `PollMarkersAsync` signature (line 725-729). Replace the `TeamStateTracker tracker` parameter with `DimensionsHolder dims`: + +```csharp + private async Task PollMarkersAsync( + (ulong Guild, Guid Server) key, + IRustServerConnection connection, + DimensionsHolder dims, + CancellationToken ct) + { +``` + +Immediately after `var dims = await connection.GetMapDimensionsAsync(...)` becomes a name clash — the parameter is now named `dims`. Rename the local dimension variable. Replace line 736: + +```csharp + var localDims = await connection.GetMapDimensionsAsync(_options.HeartbeatTimeout, ct).ConfigureAwait(false); + dims.Value = localDims; +``` + +Then update the two uses of the old `dims` local inside `PollMarkersAsync` — in the `PublishMarkerDeltaAsync(key, dims, ...)` call (line 755) and the `DetectRigActivationsAsync(key, current, rigs, dims, ...)` call (line 759) — to pass `localDims`: + +```csharp + await PublishMarkerDeltaAsync(key, localDims, previous, current, ct).ConfigureAwait(false); +``` + +```csharp + await DetectRigActivationsAsync(key, current, rigs, localDims, rigsInRadius, ct).ConfigureAwait(false); +``` + +Delete the team block (lines 761-768) entirely — the `var team = ...` through the `PlayerStateChangedEvent` publish: + +```csharp + var team = await connection.GetTeamInfoAsync(_options.HeartbeatTimeout, ct).ConfigureAwait(false); + var transitions = tracker.Diff(team, clock.UtcNow, _options.AfkThreshold, _options.AfkEpsilon); + if (transitions.Count > 0) + { + await eventBus.PublishAsync( + new PlayerStateChangedEvent(key.Guild, key.Server, dims, transitions), ct) + .ConfigureAwait(false); + } +``` + +(The `catch (OperationCanceledException) when (ct.IsCancellationRequested) { return; }` and the marker `catch` below it stay.) + +- [ ] **Step 9: Wire the holder, the push handler, and the team poll into `RunAsync`** + +In `ConnectionSupervisor.cs` `RunAsync`, after `var tracker = new TeamStateTracker();` (line 608), add: + +```csharp + var dims = new DimensionsHolder(); + + void OnTeamChanged(object? sender, TeamInfoSnapshot snapshot) + { + // Fire-and-forget: PublishTeamStateAsync catches everything internally. + _ = PublishTeamStateAsync(key, tracker, dims, snapshot); + } +``` + +Subscribe alongside the other handlers (after `connection.ClanChanged += OnClanChanged;`, line 613): + +```csharp + connection.TeamChanged += OnTeamChanged; +``` + +Change the marker-poll launch (line 624) to pass `dims` instead of `tracker`: + +```csharp + var markerPoll = Task.Run(() => PollMarkersAsync(key, connection, dims, pollCts.Token), + CancellationToken.None); +``` + +Add the team-poll launch right after the reachability-poll launch (after line 627): + +```csharp + var teamPoll = Task.Run(() => PollTeamAsync(key, connection, tracker, dims, pollCts.Token), + CancellationToken.None); +``` + +Add `teamPoll` to the `Task.WhenAll` join (line 646): + +```csharp + await Task.WhenAll(markerPoll, reachabilityPoll, teamPoll, heartbeat, liveness).ConfigureAwait(false); +``` + +Unsubscribe in the `finally` alongside the others (after `connection.ClanChanged -= OnClanChanged;`, line 659): + +```csharp + connection.TeamChanged -= OnTeamChanged; +``` + +- [ ] **Step 10: Run the push test — it should now pass** + +Run: `dtk test tests/RustPlusBot.Features.Connections.Tests/RustPlusBot.Features.Connections.Tests.csproj --filter "FullyQualifiedName~TeamChanged_push_publishes"` +Expected: PASS. + +- [ ] **Step 11: Add + run the slow-poll-tick test** + +Add to `ConnectionSupervisorTests.cs`. No `TeamChanged` event is raised; the slow poll must drive the transition purely from `TeamResult` changing between polls. + +```csharp + [Fact] + public async Task TeamPoll_tick_publishes_transition_without_any_push() + { + var source = new FakeRustSocketSource(); + source.EnqueueConnect(SocketConnectOutcome.Connected); + source.EnqueueHeartbeat(HeartbeatResult.Ok(1)); + // Fast team poll so the tick drives the transition quickly. + await using var h = CreateHarness(source, teamPollInterval: TimeSpan.FromMilliseconds(20)); + var (serverId, _, _) = await SeedAsync(h.Provider); + + var online = new TeamMemberSnapshot( + 100UL, "Alice", 1f, 1f, IsOnline: true, IsAlive: true, DateTimeOffset.UnixEpoch, DateTimeOffset.UnixEpoch); + source.LastConnectionSetup = c => c.TeamResult = new TeamInfoSnapshot(100UL, [online]); + + using var cts = new CancellationTokenSource(TimeSpan.FromSeconds(30)); + var captured = new System.Collections.Concurrent.ConcurrentQueue(); + var stream = h.Bus.SubscribeAsync(cts.Token); + var subTask = Task.Run(async () => + { + await foreach (var e in stream) + { + captured.Enqueue(e); + } + }, CancellationToken.None); + + await h.Supervisor.EnsureConnectionAsync(10UL, serverId, cts.Token); + var state = await WaitForStateAsync(h.Provider, serverId, s => s.Status == ConnectionStatus.Connected); + Assert.NotNull(state); + var conn = source.LastConnection!; + + // Wait for at least one poll to prime the baseline with the online snapshot, then flip to offline. + await WaitUntilAsync(() => conn.TeamInfoCallCount >= 1, cts.Token); + conn.TeamResult = new TeamInfoSnapshot(100UL, [online with { IsOnline = false }]); + + await WaitUntilAsync(() => !captured.IsEmpty, cts.Token); + Assert.True(captured.TryDequeue(out var evt)); + Assert.Contains(evt!.Transitions, t => t.Kind == PlayerTransitionKind.Disconnect && t.SteamId == 100UL); + + await h.Supervisor.StopAllAsync(); + await cts.CancelAsync(); + try { await subTask; } catch (OperationCanceledException) { /* expected */ } + } +``` + +This test uses the `LastConnectionSetup` hook (added to `FakeRustSocketSource` in Task 3, Step 5) so the initial online `TeamResult` is staged before the poll starts, then flips it to offline after the baseline is primed. + +Run: `dtk test tests/RustPlusBot.Features.Connections.Tests/RustPlusBot.Features.Connections.Tests.csproj --filter "FullyQualifiedName~TeamPoll_tick_publishes"` +Expected: PASS. + +- [ ] **Step 12: Add + run the efficiency test (team info is not on the marker cadence)** + +```csharp + [Fact] + public async Task TeamInfo_is_not_polled_on_the_fast_marker_cadence() + { + var source = new FakeRustSocketSource(); + source.EnqueueConnect(SocketConnectOutcome.Connected); + source.EnqueueHeartbeat(HeartbeatResult.Ok(1)); + // Marker cadence stays fast (20ms from the harness); team poll is slow. + await using var h = CreateHarness(source, teamPollInterval: TimeSpan.FromSeconds(10)); + var (serverId, _, _) = await SeedAsync(h.Provider); + + using var cts = new CancellationTokenSource(TimeSpan.FromSeconds(30)); + await h.Supervisor.EnsureConnectionAsync(10UL, serverId, cts.Token); + var state = await WaitForStateAsync(h.Provider, serverId, s => s.Status == ConnectionStatus.Connected); + Assert.NotNull(state); + var conn = source.LastConnection!; + + // Wait for the team poll's immediate first (priming) call, then let ~10 marker intervals elapse. + await WaitUntilAsync(() => conn.TeamInfoCallCount >= 1, cts.Token); + await Task.Delay(200, cts.Token); + + // Only the single priming poll ran; the 10s team interval hasn't elapsed, so the fast marker + // cadence did NOT trigger any further team-info reads. + Assert.Equal(1, conn.TeamInfoCallCount); + + await h.Supervisor.StopAllAsync(); + } +``` + +Run: `dtk test tests/RustPlusBot.Features.Connections.Tests/RustPlusBot.Features.Connections.Tests.csproj --filter "FullyQualifiedName~TeamInfo_is_not_polled"` +Expected: PASS. + +- [ ] **Step 13: Run the whole connections project (nothing else regressed)** + +Run: `dtk test tests/RustPlusBot.Features.Connections.Tests/RustPlusBot.Features.Connections.Tests.csproj` +Expected: PASS (all existing marker/rig/AFK/status-loop tests plus the four new ones). + +- [ ] **Step 14: Run the full solution test suite** + +Run: `dtk test` +Expected: PASS across all projects (the downstream Players tests that consume `PlayerStateChangedEvent` are unaffected — the event contract is unchanged). + +- [ ] **Step 15: Commit** + +```bash +git add src/RustPlusBot.Features.Connections/Supervisor/ConnectionSupervisor.cs \ + tests/RustPlusBot.Features.Connections.Tests/ConnectionSupervisorTests.cs \ + tests/RustPlusBot.Features.Connections.Tests/Fakes/FakeRustSocketSource.cs +git commit -m "feat(connections): drive team state from team_changed push + slow AFK tick" +``` + +--- + +## Notes for the implementer + +- **Concurrency:** `PublishTeamStateAsync` is called from the RustPlusApi dispatch thread (push) and the `PollTeamAsync` thread. `TeamStateTracker.Diff` is lock-guarded and swaps its baseline atomically, so the two never corrupt shared state. Ordering is NOT guaranteed, though: `PollTeamAsync` captures its snapshot before an `await` spanning a network round-trip, so a concurrent push can commit a newer baseline first and the poll's stale snapshot may then emit a rare, self-healing spurious presence transition (corrected on the next tick). This is the accepted cost of the slow-poll (Approach B) design; AFK timing is unaffected. +- **Do not** reintroduce a `GetTeamInfoAsync` call inside `PollMarkersAsync`; team state is now exclusively push + `PollTeamAsync`. +- **Line numbers** in this plan are from the pre-change tree; after early tasks they will drift. Anchor edits on the quoted code, not the numbers. +- **Out of scope:** the `GetMapMarkers returned no data` diagnostic-message gap noted in the design's §7 is a separate change — do not bundle it here. diff --git a/docs/superpowers/specs/2026-07-24-team-changed-push-design.md b/docs/superpowers/specs/2026-07-24-team-changed-push-design.md new file mode 100644 index 00000000..1db8f28b --- /dev/null +++ b/docs/superpowers/specs/2026-07-24-team-changed-push-design.md @@ -0,0 +1,140 @@ +# Push-driven team state via `team_changed` (RustPlusApi beta.6) + +**Date:** 2026-07-24 +**Status:** Approved design +**Branch:** `fix/consumer-loop-resilience` (or a dedicated follow-up branch) + +## 1. Problem & motivation + +Team-state detection (connect / disconnect / death / respawn / AFK) is currently +derived by **polling** `GetTeamInfoAsync` inside `PollMarkersAsync` +(`ConnectionSupervisor.cs:761`), on the marker cadence of every 2–5 s +(`MarkerPollInterval` = 5 s, `MarkerPollFastInterval` = 2 s). + +Two problems motivate a change: + +1. **Efficiency / latency.** A `getTeamInfo` request every 2–5 s per connected + server is wasteful, and presence/death events lag by up to one poll interval. +2. **Log noise.** RustPlusApi `beta.5` does not dispatch the `team_changed` + broadcast — its `ParseNotification` falls through to + `Logger.LogUnknownBroadcast`, producing recurring + `WRN Unknown broadcast received: RustPlusContracts.AppBroadcast`. `team_changed` + is the *only* unhandled broadcast field in `beta.5`, so every one of these + warnings is a dropped team update. + +RustPlusApi **beta.6** now parses `team_changed` and raises a new +`OnTeamChanged` event carrying `TeamChangedEventArg { PlayerId, TeamInfo }`, +where `TeamInfo` is produced by the **same `ToTeamInfo()` mapping** the polled +`GetTeamInfoAsync` already uses. Full field parity (members with +`X/Y/IsOnline/IsAlive/LastSpawnTime/LastDeathTime`, `LeaderSteamId`, `DeathNote`). + +## 2. Scope + +**In scope:** upgrade `beta.5 → beta.6`; move team-state detection from the fast +poll to the `OnTeamChanged` push event; retain a low-frequency team poll purely as +an AFK safety tick. + +**Explicitly out of scope:** marker/rig polling. The Rust+ protocol has **no +broadcast** for cargo-ship / patrol-heli / chinook / travelling-vendor markers; +the server only answers `getMapMarkers` on request. `PollMarkersAsync`'s marker +and rig-activation work is unchanged. This is therefore a *partial* migration — +"move team state to push", not "replace the polling system" wholesale. + +## 3. Architecture + +### 3.1 Connection layer — `IRustServerConnection` / `RustPlusSocketSource` + +- Add to the interface: + `event EventHandler? TeamChanged;` +- `RustPlusSocketSource`: + - Subscribe `_rustPlus.OnTeamChanged += OnTeamChanged` in the ctor event block + (near `RustPlusSocketSource.cs:172`); unsubscribe in `DisposeAsync` + (near line 768), symmetrically with the existing handlers. + - Extract the inline `TeamInfo → TeamInfoSnapshot` mapping currently in + `GetTeamInfoAsync` (lines 339–353) into a private static + `ToSnapshot(RustPlusApi.Data.TeamInfo)` helper. Both `GetTeamInfoAsync` and + the new handler reuse it: + `private void OnTeamChanged(object? s, TeamChangedEventArg e) => TeamChanged?.Invoke(this, ToSnapshot(e.TeamInfo));` + - Because the event and the poll share one mapping, there is zero behavioural + drift between pushed and polled team snapshots. +- The test fake `IRustServerConnection` gains the `TeamChanged` event plus a test + hook to raise it. + +### 3.2 Supervisor layer — `ConnectionSupervisor` + +- **Push handler.** Wire an `OnTeamChanged` handler in `RunAsync` alongside the + existing `OnTeamMessage` / `OnClanChanged` wiring (~line 609). On each event it + calls `tracker.Diff(snapshot, clock.UtcNow, _options.AfkThreshold, + _options.AfkEpsilon)` and, when transitions are non-empty, fire-and-forget + publishes `PlayerStateChangedEvent` using the same `_ = PublishAsync(...)` + error-isolation pattern as the other handlers. Subscribe on setup, unsubscribe + in the `finally` (lines 655–659 block). +- **Slow team poll (AFK tick + self-heal).** Split the team block (lines 761–768) + out of `PollMarkersAsync` into a new `PollTeamAsync` loop that runs the *same* + `Diff` + publish at a new `TeamPollInterval` (default 30 s). Its first iteration + runs immediately (to prime the baseline promptly), then it waits + `TeamPollInterval` between iterations. Launched as a fourth `Task.Run` next to + `markerPoll` / `reachabilityPoll` (line 624) and joined in the same + `Task.WhenAll` on exit (line 646). +- **`PollMarkersAsync`** loses its `tracker` parameter and the team block; markers + and rig-activation detection are otherwise untouched. +- **Shared `dims`.** `PlayerStateChangedEvent` needs `MapDimensions?`. `dims` stays + resolved lazily *off the critical connect path* (preserving the non-blocking + behaviour the comment at lines 731–735 protects), stored in a per-connected-window + holder that both the push handler and `PollTeamAsync` read. Events arriving before + `dims` resolves publish `dims: null`, which `PlayerStateChangedEvent` already + tolerates (nullable → renders without a grid reference). + +### 3.3 Options — `ConnectionOptions` + +- Add `public TimeSpan TeamPollInterval { get; set; } = TimeSpan.FromSeconds(30);` + +## 4. Concurrency & correctness + +- `TeamStateTracker.Diff` is now invoked from two threads: the RustPlusApi dispatch + thread (push) and the `PollTeamAsync` thread (tick). `TeamStateTracker` is already + guarded by `_gate` and swaps `_baseline` atomically, so concurrent calls serialize + and never corrupt shared state. Ordering is NOT guaranteed, though: + `PollTeamAsync` captures its snapshot before an `await` spanning a network + round-trip, so a concurrent push can commit a newer baseline first and the poll's + stale snapshot may then emit a rare, self-healing spurious presence transition + (corrected on the next tick). This is the accepted cost of the slow-poll + (Approach B) design; AFK timing is unaffected. +- **No change to the AFK model.** `UpdateAfk` still flags `BecameAfk` when + `Clock - stillSince >= Threshold`, evaluated whenever `Diff` runs. The slow poll + exists solely to guarantee `Diff` runs periodically, so "became AFK" cannot stall + during broadcast silence (a stationary player in a quiet/solo team emits no + `team_changed`). Worst-case detection latency for BecameAfk is + `AfkThreshold + TeamPollInterval`; presence/death/respawn/returned-from-AFK remain + instant via push. +- **Baseline priming.** Whichever of {first broadcast, first `PollTeamAsync` + iteration} runs first primes the tracker silently (`Diff` primes on first non-null + call); the other diffs against it. No spurious connect-event flood on connect. + +## 5. Testing + +- `RustPlusSocketSource`: raising `OnTeamChanged` surfaces a correctly-mapped + `TeamInfoSnapshot` — member fields, leader death note, offline and dead members. +- `ConnectionSupervisor`: a `TeamChanged` event carrying a moved / disconnected / + died member publishes the expected `PlayerStateChangedEvent`; and the fast marker + cadence no longer issues any `getTeamInfo` request. +- AFK-during-silence: with no broadcasts delivered, a stationary online player is + still flagged `BecameAfk` from the slow poll within + `AfkThreshold + TeamPollInterval`. +- Regression: existing marker/rig and status-loop tests remain green (the marker + loop is untouched apart from removing the team block). + +## 6. Rollout + +- Bump `RustPlusApi` and `RustPlusApi.Fcm` to `2.0.0-beta.6` in + `Directory.Packages.props`. +- The `Unknown broadcast received` warning disappears once beta.6 dispatches + `team_changed`; no code change is required to silence it. + +## 7. Out-of-scope follow-up (noted, not included here) + +The related investigation found that `RustPlusSocketSource.GetMapMarkersAsync` +throws a bare `"GetMapMarkers returned no data."` that discards the server's +`response.Error` code/message, making the `Marker poll ... failed` warning +non-diagnostic. That is a separate, independent fix and is **not** part of this +design. 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/ConnectionOptions.cs b/src/RustPlusBot.Features.Connections/ConnectionOptions.cs index bd817610..544cc4f3 100644 --- a/src/RustPlusBot.Features.Connections/ConnectionOptions.cs +++ b/src/RustPlusBot.Features.Connections/ConnectionOptions.cs @@ -42,6 +42,14 @@ public sealed class ConnectionOptions /// Movement tolerance (world units) below which a member is considered still. Default 1. public float AfkEpsilon { get; set; } = 1f; + /// + /// How often to poll team info as an AFK safety tick and self-heal, now that live team changes arrive + /// via the pushed team_changed event. Presence/death events are instant; this only bounds how long + /// a still player in a broadcast-silent team can take to be flagged AFK (worst case AfkThreshold + this). + /// Default 30s. + /// + public TimeSpan TeamPollInterval { get; set; } = TimeSpan.FromSeconds(30); + /// How often to poll managed devices for reachability changes while connected. Default 5m. public TimeSpan ReachabilityPollInterval { get; set; } = TimeSpan.FromMinutes(5); 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/IRustServerConnection.cs b/src/RustPlusBot.Features.Connections/Listening/IRustServerConnection.cs index 7ba22920..6f889d86 100644 --- a/src/RustPlusBot.Features.Connections/Listening/IRustServerConnection.cs +++ b/src/RustPlusBot.Features.Connections/Listening/IRustServerConnection.cs @@ -169,4 +169,11 @@ Task> GetMonumentsAsync(TimeSpan timeout, /// Raised when a managed storage monitor's contents change in-game; carries the entity id and the new contents. event EventHandler? StorageMonitorTriggered; + + /// + /// Raised when the server pushes a team_changed broadcast (member join/leave, online/offline, + /// death/respawn, movement, leader change). Carries the full team snapshot — the same shape a + /// poll returns. + /// + event EventHandler? TeamChanged; } diff --git a/src/RustPlusBot.Features.Connections/Listening/RustPlusSocketSource.cs b/src/RustPlusBot.Features.Connections/Listening/RustPlusSocketSource.cs index 0dc59c44..c8bdd9f8 100644 --- a/src/RustPlusBot.Features.Connections/Listening/RustPlusSocketSource.cs +++ b/src/RustPlusBot.Features.Connections/Listening/RustPlusSocketSource.cs @@ -138,6 +138,12 @@ public event EventHandler? StorageMonitorTriggered remove { _ = value; } } + public event EventHandler? TeamChanged + { + add { _ = value; } + remove { _ = value; } + } + public ValueTask DisposeAsync() => ValueTask.CompletedTask; } @@ -170,6 +176,7 @@ public RustPlusServerConnection(string ip, _rustPlus.OnStorageMonitorTriggered += OnStorageMonitorTriggered; _rustPlus.OnClanChatReceived += OnClanChatReceived; _rustPlus.OnClanChanged += OnClanChanged; + _rustPlus.OnTeamChanged += OnTeamChanged; } /// @@ -336,21 +343,7 @@ public async Task GetInfoAsync(TimeSpan timeout, CancellationTo return null; } - var members = (response.Data.Members ?? []) - .Select(m => new TeamMemberSnapshot( - m.SteamId, - m.Name ?? string.Empty, - m.X, - m.Y, - m.IsOnline, - m.IsAlive, - new DateTimeOffset(DateTime.SpecifyKind(m.LastSpawnTime, DateTimeKind.Utc)), - new DateTimeOffset(DateTime.SpecifyKind(m.LastDeathTime, DateTimeKind.Utc)))) - .ToList(); - var deathNote = response.Data.DeathNote is { } dn - ? ((float X, float Y)?)(dn.X, dn.Y) - : null; - return new TeamInfoSnapshot(response.Data.LeaderSteamId, members, deathNote); + return TeamInfoMapping.ToSnapshot(response.Data); } catch (OperationCanceledException) when (!cancellationToken.IsCancellationRequested) { @@ -466,6 +459,8 @@ public async Task PromoteToLeaderAsync(ulong steamId, public event EventHandler? StorageMonitorTriggered; + public event EventHandler? TeamChanged; + public async Task GetSmartDeviceInfoAsync(ulong entityId, SmartDeviceKind kind, TimeSpan timeout, @@ -714,7 +709,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(); @@ -762,6 +761,7 @@ public async ValueTask DisposeAsync() _rustPlus.OnStorageMonitorTriggered -= OnStorageMonitorTriggered; _rustPlus.OnClanChatReceived -= OnClanChatReceived; _rustPlus.OnClanChanged -= OnClanChanged; + _rustPlus.OnTeamChanged -= OnTeamChanged; try { // CONFIRMED: RustPlusSocket implements IAsyncDisposable in 2.0.0-beta.1. @@ -831,6 +831,14 @@ private void OnClanChanged(object? sender, ClanChangedEventArg e) => ClanChanged?.Invoke(this, e.ClanInfo is { } info ? ClanProbeResult.From(ClanMapping.ToSnapshot(info)) : ClanProbeResult.NoClan); + private void OnTeamChanged(object? sender, RustPlusApi.Data.Events.TeamChangedEventArg e) + { + if (e.TeamInfo is { } teamInfo) + { + TeamChanged?.Invoke(this, TeamInfoMapping.ToSnapshot(teamInfo)); + } + } + [LoggerMessage(Level = LogLevel.Warning, Message = "Rust+ socket connect failed.")] private static partial void LogConnectFailed(ILogger logger, Exception ex); diff --git a/src/RustPlusBot.Features.Connections/Listening/TeamInfoMapping.cs b/src/RustPlusBot.Features.Connections/Listening/TeamInfoMapping.cs new file mode 100644 index 00000000..2a3b8f52 --- /dev/null +++ b/src/RustPlusBot.Features.Connections/Listening/TeamInfoMapping.cs @@ -0,0 +1,33 @@ +using RustPlusBot.Abstractions.Connections; + +namespace RustPlusBot.Features.Connections.Listening; + +/// Maps a RustPlusApi to the bot's decoupled +/// . Shared by the polled read and the pushed team_changed event so +/// the two paths can never drift. +internal static class TeamInfoMapping +{ + /// Projects a RustPlusApi team-info model onto a . + /// The RustPlusApi team info (from a poll response or a team_changed broadcast). + /// The decoupled snapshot the tracker and events consume. + public static TeamInfoSnapshot ToSnapshot(RustPlusApi.Data.TeamInfo teamInfo) + { + // CONFIRMED (2.0.0-beta.6): MemberInfo exposes SteamId/Name?/X/Y/IsOnline/IsAlive plus + // LastSpawnTime/LastDeathTime as Unspecified-kind DateTimes that are actually UTC. + var members = (teamInfo.Members ?? []) + .Select(m => new TeamMemberSnapshot( + m.SteamId, + m.Name ?? string.Empty, + m.X, + m.Y, + m.IsOnline, + m.IsAlive, + new DateTimeOffset(DateTime.SpecifyKind(m.LastSpawnTime, DateTimeKind.Utc)), + new DateTimeOffset(DateTime.SpecifyKind(m.LastDeathTime, DateTimeKind.Utc)))) + .ToList(); + var deathNote = teamInfo.DeathNote is { } dn + ? ((float X, float Y)?)(dn.X, dn.Y) + : null; + return new TeamInfoSnapshot(teamInfo.LeaderSteamId, members, deathNote); + } +} diff --git a/src/RustPlusBot.Features.Connections/Supervisor/ConnectionSupervisor.cs b/src/RustPlusBot.Features.Connections/Supervisor/ConnectionSupervisor.cs index 80f78185..451c2311 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. + LogMonumentsQueryFailed(logger, ex, serverId); + return []; + } } /// @@ -589,11 +606,22 @@ void OnClanChanged(object? sender, ClanProbeResult probe) #pragma warning restore RCS1163 var tracker = new TeamStateTracker(); + var dims = new DimensionsHolder(); + +#pragma warning disable RCS1163 // Unused 'sender': required by the EventHandler delegate shape. + void OnTeamChanged(object? sender, TeamInfoSnapshot snapshot) + { + // Fire-and-forget: PublishTeamStateAsync catches everything internally. + _ = PublishTeamStateAsync(key, tracker, dims, snapshot); + } +#pragma warning restore RCS1163 + connection.TeamMessageReceived += OnTeamMessage; connection.SmartDeviceTriggered += OnSmartDevice; connection.StorageMonitorTriggered += OnStorage; connection.ClanMessageReceived += OnClanMessage; connection.ClanChanged += OnClanChanged; + connection.TeamChanged += OnTeamChanged; _liveSockets[key] = new LiveSocket(connection, activeSteamId, tracker); await PrimeDevicesAsync(key, connection, ct).ConfigureAwait(false); @@ -604,10 +632,12 @@ void OnClanChanged(object? sender, ClanProbeResult probe) await PublishClanStateAsync(key, clanProbe).ConfigureAwait(false); using var pollCts = CancellationTokenSource.CreateLinkedTokenSource(ct); - var markerPoll = Task.Run(() => PollMarkersAsync(key, connection, tracker, pollCts.Token), + var markerPoll = Task.Run(() => PollMarkersAsync(key, connection, dims, pollCts.Token), CancellationToken.None); var reachabilityPoll = Task.Run(() => PollReachabilityAsync(key, connection, pollCts.Token), CancellationToken.None); + var teamPoll = Task.Run(() => PollTeamAsync(key, connection, tracker, dims, pollCts.Token), + CancellationToken.None); // Race the heartbeat against a liveness watchdog: the Rust+ library raises no event when the SERVER // closes the socket, so without the watchdog a silent drop goes unnoticed until the next heartbeat // (up to a minute). Whichever signals a reason first wins; the finally cancels and joins the rest. @@ -625,8 +655,8 @@ void OnClanChanged(object? sender, ClanProbeResult probe) await pollCts.CancelAsync().ConfigureAwait(false); try { -#pragma warning disable VSTHRD003 // Suppress: all four tasks are owned by this connected window and explicitly joined on exit. - await Task.WhenAll(markerPoll, reachabilityPoll, heartbeat, liveness).ConfigureAwait(false); +#pragma warning disable VSTHRD003 // Suppress: all five tasks are owned by this connected window and explicitly joined on exit. + await Task.WhenAll(markerPoll, reachabilityPoll, teamPoll, heartbeat, liveness).ConfigureAwait(false); #pragma warning restore VSTHRD003 } catch (OperationCanceledException) @@ -640,6 +670,7 @@ void OnClanChanged(object? sender, ClanProbeResult probe) connection.StorageMonitorTriggered -= OnStorage; connection.ClanMessageReceived -= OnClanMessage; connection.ClanChanged -= OnClanChanged; + connection.TeamChanged -= OnTeamChanged; } } @@ -708,7 +739,7 @@ await PublishStatusAsync(key, ConnectionStatus.Connected, beat.PlayerCount, cred private async Task PollMarkersAsync( (ulong Guild, Guid Server) key, IRustServerConnection connection, - TeamStateTracker tracker, + DimensionsHolder dims, CancellationToken ct) { // Fetch map dimensions and oil-rig positions here, off the critical connect path: these are the two @@ -716,7 +747,8 @@ private async Task PollMarkersAsync( // this background poll means a slow map no longer delays the connection going live (heartbeat + chat // relay start immediately); marker/rig detection simply activates once these resolve. Both degrade // safely on timeout (dims -> null, rigs -> empty) without ending the poll. - var dims = await connection.GetMapDimensionsAsync(_options.HeartbeatTimeout, ct).ConfigureAwait(false); + var localDims = await connection.GetMapDimensionsAsync(_options.HeartbeatTimeout, ct).ConfigureAwait(false); + dims.Value = localDims; var rigs = await GetRigPositionsAsync(key.Server, connection, ct).ConfigureAwait(false); IReadOnlyList? previous = null; @@ -735,20 +767,11 @@ private async Task PollMarkersAsync( } else { - await PublishMarkerDeltaAsync(key, dims, previous, current, ct).ConfigureAwait(false); + await PublishMarkerDeltaAsync(key, localDims, previous, current, ct).ConfigureAwait(false); previous = current; } - await DetectRigActivationsAsync(key, current, rigs, dims, rigsInRadius, ct).ConfigureAwait(false); - - var team = await connection.GetTeamInfoAsync(_options.HeartbeatTimeout, ct).ConfigureAwait(false); - var transitions = tracker.Diff(team, clock.UtcNow, _options.AfkThreshold, _options.AfkEpsilon); - if (transitions.Count > 0) - { - await eventBus.PublishAsync( - new PlayerStateChangedEvent(key.Guild, key.Server, dims, transitions), ct) - .ConfigureAwait(false); - } + await DetectRigActivationsAsync(key, current, rigs, localDims, rigsInRadius, ct).ConfigureAwait(false); } catch (OperationCanceledException) when (ct.IsCancellationRequested) { @@ -768,6 +791,48 @@ await eventBus.PublishAsync( } } + /// + /// Low-frequency team poll that runs the same as the pushed + /// team_changed handler. Live changes arrive via the push event; this loop exists solely to (a) prime the + /// baseline on connect and (b) guarantee Diff runs periodically so a still player in a + /// broadcast-silent team is still flagged AFK. Its first iteration runs immediately (prime), then it waits + /// between iterations. Degrades safely: a failed poll is + /// logged and skipped, never ending the loop. + /// + /// The (guild, server) routing key. + /// The live connection to poll. + /// The shared AFK/online tracker whose baseline this poll also primes/diffs. + /// The connected window's dimensions holder, read for the published event. + /// Cancels when the connected window ends. + private async Task PollTeamAsync( + (ulong Guild, Guid Server) key, + IRustServerConnection connection, + TeamStateTracker tracker, + DimensionsHolder dims, + CancellationToken ct) + { + while (!ct.IsCancellationRequested) + { + try + { + var team = await connection.GetTeamInfoAsync(_options.HeartbeatTimeout, ct).ConfigureAwait(false); + await PublishTeamStateAsync(key, tracker, dims, team).ConfigureAwait(false); + } + catch (OperationCanceledException) when (ct.IsCancellationRequested) + { + return; // stopping + } +#pragma warning disable CA1031 // Broad catch: a failed team poll is logged and skipped; the loop survives. + catch (Exception ex) +#pragma warning restore CA1031 + { + LogTeamPollFailed(logger, ex, key.Server); + } + + await Task.Delay(_options.TeamPollInterval, ct).ConfigureAwait(false); + } + } + private async Task PublishMarkerDeltaAsync( (ulong Guild, Guid Server) key, MapDimensions? dims, @@ -1193,6 +1258,42 @@ private async Task PublishClanStateAsync((ulong Guild, Guid Server) key, ClanPro } } + private async Task PublishTeamStateAsync( + (ulong Guild, Guid Server) key, + TeamStateTracker tracker, + DimensionsHolder dims, + TeamInfoSnapshot? snapshot) + { + if (_disposed) + { + return; + } + + try + { + var transitions = tracker.Diff(snapshot, clock.UtcNow, _options.AfkThreshold, _options.AfkEpsilon); + if (transitions.Count == 0) + { + return; + } + + var evt = new PlayerStateChangedEvent(key.Guild, key.Server, dims.Value, transitions); + // Supervisor-wide shutdown token, not a per-connection ct: a pushed team change should publish + // regardless of one connection's reconnect cycle (mirrors the chat/clan handlers). + await eventBus.PublishAsync(evt, _shutdown.Token).ConfigureAwait(false); + } + catch (OperationCanceledException) + { + // Shutting down. + } +#pragma warning disable CA1031 // Broad catch: a publish failure must not crash the socket callback or poll. + catch (Exception ex) +#pragma warning restore CA1031 + { + LogPublishTeamStateFailed(logger, ex, key.Server); + } + } + private async Task PrimeDevicesAsync( (ulong Guild, Guid Server) key, IRustServerConnection connection, @@ -1486,6 +1587,12 @@ await eventBus.PublishAsync( [LoggerMessage(Level = LogLevel.Warning, Message = "Marker poll for server {ServerId} failed.")] private static partial void LogMarkerPollFailed(ILogger logger, Exception exception, Guid serverId); + [LoggerMessage(Level = LogLevel.Warning, Message = "Team poll for server {ServerId} failed.")] + private static partial void LogTeamPollFailed(ILogger logger, Exception exception, Guid serverId); + + [LoggerMessage(Level = LogLevel.Warning, Message = "Publishing team state for server {ServerId} failed.")] + private static partial void LogPublishTeamStateFailed(ILogger logger, Exception exception, Guid serverId); + [LoggerMessage(Level = LogLevel.Warning, Message = "Reachability poll for server {ServerId} failed.")] private static partial void LogReachabilityPollFailed(ILogger logger, Exception exception, Guid serverId); @@ -1498,6 +1605,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); @@ -1557,6 +1668,20 @@ private readonly record struct Prepared( private sealed record LiveSocket(IRustServerConnection Connection, ulong ActiveSteamId, TeamStateTracker Tracker); + /// Mutable, thread-visible holder for the per-connected-window map dimensions. The marker poll + /// resolves these once off the critical path; the team push handler and team poll read them (possibly + /// null before resolution — PlayerStateChangedEvent tolerates a null and renders without a grid ref). + private sealed class DimensionsHolder + { + private volatile MapDimensions? _value; + + public MapDimensions? Value + { + get => _value; + set => _value = value; + } + } + private sealed class Handle(CancellationTokenSource cts, Task runTask) : IAsyncDisposable { public ValueTask DisposeAsync() 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/ConnectionOptionsTests.cs b/tests/RustPlusBot.Features.Connections.Tests/ConnectionOptionsTests.cs index f0ab114b..b0c6589c 100644 --- a/tests/RustPlusBot.Features.Connections.Tests/ConnectionOptionsTests.cs +++ b/tests/RustPlusBot.Features.Connections.Tests/ConnectionOptionsTests.cs @@ -15,6 +15,7 @@ public void Defaults_match_the_2a_ii_design() Assert.Equal(TimeSpan.FromMinutes(15), o.RigActiveWindow); Assert.Equal(TimeSpan.FromMinutes(15), o.RigOfflineWindow); Assert.Equal(TimeSpan.FromSeconds(30), o.RigTickInterval); + Assert.Equal(TimeSpan.FromSeconds(30), o.TeamPollInterval); } /// Verifies that the reachability poll interval defaults to 5 minutes. diff --git a/tests/RustPlusBot.Features.Connections.Tests/ConnectionSupervisorTests.cs b/tests/RustPlusBot.Features.Connections.Tests/ConnectionSupervisorTests.cs index e5e04820..859eb728 100644 --- a/tests/RustPlusBot.Features.Connections.Tests/ConnectionSupervisorTests.cs +++ b/tests/RustPlusBot.Features.Connections.Tests/ConnectionSupervisorTests.cs @@ -21,7 +21,7 @@ namespace RustPlusBot.Features.Connections.Tests; public sealed class ConnectionSupervisorTests { - private static Harness CreateHarness(FakeRustSocketSource source) + private static Harness CreateHarness(FakeRustSocketSource source, TimeSpan? teamPollInterval = null) { var protector = Substitute.For(); protector.Unprotect(Arg.Any()).Returns(c => c.Arg()); @@ -68,6 +68,7 @@ private static Harness CreateHarness(FakeRustSocketSource source) MarkerPollInterval = TimeSpan.FromMilliseconds(20), MarkerPollFastInterval = TimeSpan.FromMilliseconds(20), LivenessPollInterval = TimeSpan.FromMilliseconds(20), + TeamPollInterval = teamPollInterval ?? TimeSpan.FromMilliseconds(20), })); services.AddSingleton(); services.AddSingleton(); @@ -741,6 +742,193 @@ await WaitUntilAsync( } } + [Fact] + public async Task TeamChanged_push_publishes_player_state_transition() + { + var source = new FakeRustSocketSource(); + source.EnqueueConnect(SocketConnectOutcome.Connected); + source.EnqueueHeartbeat(HeartbeatResult.Ok(1)); + // Make the slow poll inert: TeamResult = null means Diff(null) is a no-op, so the tracker's baseline + // is driven ONLY by the pushed snapshots below — no race between the priming poll and the pushes. + source.LastConnectionSetup = c => c.TeamResult = null; + // Large team-poll interval too, belt-and-suspenders. + await using var h = CreateHarness(source, teamPollInterval: TimeSpan.FromSeconds(30)); + var (serverId, _, _) = await SeedAsync(h.Provider); + + using var cts = new CancellationTokenSource(TimeSpan.FromSeconds(30)); + var captured = new System.Collections.Concurrent.ConcurrentQueue(); + var stream = h.Bus.SubscribeAsync(cts.Token); + var subTask = Task.Run(async () => + { + await foreach (var e in stream) + { + captured.Enqueue(e); + } + }, CancellationToken.None); + + await h.Supervisor.EnsureConnectionAsync(10UL, serverId, cts.Token); + var state = await WaitForStateAsync(h.Provider, serverId, s => s.Status == ConnectionStatus.Connected); + Assert.NotNull(state); + var conn = source.LastConnection!; + + var online = new TeamMemberSnapshot( + 100UL, "Alice", 1f, 1f, IsOnline: true, IsAlive: true, DateTimeOffset.UnixEpoch, DateTimeOffset.UnixEpoch); + var offline = online with + { + IsOnline = false + }; + + conn.RaiseTeamChanged(new TeamInfoSnapshot(100UL, [online])); // prime (silent) + conn.RaiseTeamChanged(new TeamInfoSnapshot(100UL, [offline])); // -> Disconnect + + await WaitUntilAsync(() => !captured.IsEmpty, cts.Token); + + Assert.True(captured.TryDequeue(out var evt)); + var transition = Assert.Single(evt!.Transitions); + Assert.Equal(PlayerTransitionKind.Disconnect, transition.Kind); + Assert.Equal(100UL, transition.SteamId); + + await h.Supervisor.StopAllAsync(); + await cts.CancelAsync(); + try { await subTask; } + catch (OperationCanceledException) + { + /* expected */ + } + } + + [Fact] + public async Task TeamPoll_tick_publishes_transition_without_any_push() + { + var source = new FakeRustSocketSource(); + source.EnqueueConnect(SocketConnectOutcome.Connected); + source.EnqueueHeartbeat(HeartbeatResult.Ok(1)); + // Fast team poll so the tick drives the transition quickly. + await using var h = CreateHarness(source, teamPollInterval: TimeSpan.FromMilliseconds(20)); + var (serverId, _, _) = await SeedAsync(h.Provider); + + var online = new TeamMemberSnapshot( + 100UL, "Alice", 1f, 1f, IsOnline: true, IsAlive: true, DateTimeOffset.UnixEpoch, DateTimeOffset.UnixEpoch); + source.LastConnectionSetup = c => c.TeamResult = new TeamInfoSnapshot(100UL, [online]); + + using var cts = new CancellationTokenSource(TimeSpan.FromSeconds(30)); + var captured = new System.Collections.Concurrent.ConcurrentQueue(); + var stream = h.Bus.SubscribeAsync(cts.Token); + var subTask = Task.Run(async () => + { + await foreach (var e in stream) + { + captured.Enqueue(e); + } + }, CancellationToken.None); + + await h.Supervisor.EnsureConnectionAsync(10UL, serverId, cts.Token); + var state = await WaitForStateAsync(h.Provider, serverId, s => s.Status == ConnectionStatus.Connected); + Assert.NotNull(state); + var conn = source.LastConnection!; + + // Wait for at least one poll to prime the baseline with the online snapshot, then flip to offline. + await WaitUntilAsync(() => conn.TeamInfoCallCount >= 1, cts.Token); + conn.TeamResult = new TeamInfoSnapshot(100UL, [ + online with + { + IsOnline = false + } + ]); + + await WaitUntilAsync(() => !captured.IsEmpty, cts.Token); + Assert.True(captured.TryDequeue(out var evt)); + Assert.Contains(evt!.Transitions, t => t.Kind == PlayerTransitionKind.Disconnect && t.SteamId == 100UL); + + await h.Supervisor.StopAllAsync(); + await cts.CancelAsync(); + try { await subTask; } + catch (OperationCanceledException) + { + /* expected */ + } + } + + [Fact] + public async Task TeamInfo_is_not_polled_on_the_fast_marker_cadence() + { + var source = new FakeRustSocketSource(); + source.EnqueueConnect(SocketConnectOutcome.Connected); + source.EnqueueHeartbeat(HeartbeatResult.Ok(1)); + // Marker cadence stays fast (20ms from the harness); team poll is slow. + await using var h = CreateHarness(source, teamPollInterval: TimeSpan.FromSeconds(10)); + var (serverId, _, _) = await SeedAsync(h.Provider); + + using var cts = new CancellationTokenSource(TimeSpan.FromSeconds(30)); + await h.Supervisor.EnsureConnectionAsync(10UL, serverId, cts.Token); + var state = await WaitForStateAsync(h.Provider, serverId, s => s.Status == ConnectionStatus.Connected); + Assert.NotNull(state); + var conn = source.LastConnection!; + + // Wait for the team poll's immediate first (priming) call, then let ~10 marker intervals elapse. + await WaitUntilAsync(() => conn.TeamInfoCallCount >= 1, cts.Token); + await Task.Delay(200, cts.Token); + + // Only the single priming poll ran; the 10s team interval hasn't elapsed, so the fast marker + // cadence did NOT trigger any further team-info reads. + Assert.Equal(1, conn.TeamInfoCallCount); + + await h.Supervisor.StopAllAsync(); + } + + [Fact] + public async Task TeamPoll_flags_afk_during_broadcast_silence() + { + var source = new FakeRustSocketSource(); + source.EnqueueConnect(SocketConnectOutcome.Connected); + source.EnqueueHeartbeat(HeartbeatResult.Ok(1)); + // A stationary, online, alive member — never moves, so it must become AFK once enough time passes. + var online = new TeamMemberSnapshot( + 100UL, "Alice", 1f, 1f, IsOnline: true, IsAlive: true, DateTimeOffset.UnixEpoch, DateTimeOffset.UnixEpoch); + source.LastConnectionSetup = c => c.TeamResult = new TeamInfoSnapshot(100UL, [online]); + // Fast tick so AFK is re-evaluated promptly once the clock advances. No team_changed push is ever raised. + await using var h = CreateHarness(source, teamPollInterval: TimeSpan.FromMilliseconds(20)); + var (serverId, _, _) = await SeedAsync(h.Provider); + + using var cts = new CancellationTokenSource(TimeSpan.FromSeconds(30)); + var captured = new System.Collections.Concurrent.ConcurrentQueue(); + var stream = h.Bus.SubscribeAsync(cts.Token); + var subTask = Task.Run(async () => + { + await foreach (var e in stream) + { + captured.Enqueue(e); + } + }, CancellationToken.None); + + await h.Supervisor.EnsureConnectionAsync(10UL, serverId, cts.Token); + var state = await WaitForStateAsync(h.Provider, serverId, s => s.Status == ConnectionStatus.Connected); + Assert.NotNull(state); + var conn = source.LastConnection!; + + // Let the poll prime the baseline at t=epoch (member still, not yet AFK). + await WaitUntilAsync(() => conn.TeamInfoCallCount >= 1, cts.Token); + + // Advance the fake clock past AfkThreshold (default 5m) WITHOUT any team_changed push. + var clock = h.Provider.GetRequiredService(); + clock.UtcNow.Returns(DateTimeOffset.UnixEpoch + TimeSpan.FromMinutes(6)); + + await WaitUntilAsync( + () => captured.SelectMany(e => e.Transitions).Any(t => t.Kind == PlayerTransitionKind.BecameAfk), + cts.Token); + Assert.Contains( + captured.SelectMany(e => e.Transitions), + t => t.Kind == PlayerTransitionKind.BecameAfk && t.SteamId == 100UL); + + await h.Supervisor.StopAllAsync(); + await cts.CancelAsync(); + try { await subTask; } + catch (OperationCanceledException) + { + /* expected */ + } + } + private static async Task WaitUntilAsync(Func condition, CancellationToken ct) { while (!condition()) diff --git a/tests/RustPlusBot.Features.Connections.Tests/Fakes/FakeRustSocketSource.cs b/tests/RustPlusBot.Features.Connections.Tests/Fakes/FakeRustSocketSource.cs index 7b2277bd..c556e6b8 100644 --- a/tests/RustPlusBot.Features.Connections.Tests/Fakes/FakeRustSocketSource.cs +++ b/tests/RustPlusBot.Features.Connections.Tests/Fakes/FakeRustSocketSource.cs @@ -43,6 +43,11 @@ internal sealed class FakeRustSocketSource : IRustSocketSource /// The connection produced by the most recent call (for driving inbound/inspecting sends). internal FakeConnection? LastConnection { get; private set; } + /// Optional hook invoked on each new at creation time, before it is + /// returned and the poll loops start. Lets a test stage (or null) + /// without racing the immediate priming team poll. + internal Action? LastConnectionSetup { get; set; } + public IRustServerConnection Create(string ip, int port, ulong steamId, string playerToken) { Interlocked.Increment(ref _createCount); @@ -98,6 +103,8 @@ public IRustServerConnection Create(string ip, int port, ulong steamId, string p _pendingClanProbe = null; } + LastConnectionSetup?.Invoke(connection); + LastConnection = connection; return connection; } @@ -207,6 +214,10 @@ internal sealed class FakeConnection(SocketConnectOutcome outcome, FakeRustSocke /// The snapshot returned by . Defaults to a non-null empty snapshot. public TeamInfoSnapshot? TeamResult { get; set; } = new(0UL, []); + /// Number of times has been called (to assert team info is + /// no longer polled on the fast marker cadence). + public int TeamInfoCallCount { get; private set; } + /// The result returned by . Defaults to true. public bool PromoteResult { get; set; } = true; @@ -262,6 +273,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; } @@ -290,6 +306,9 @@ internal sealed class FakeConnection(SocketConnectOutcome outcome, FakeRustSocke /// Raised by . public event EventHandler? StorageMonitorTriggered; + /// Raised by to simulate a pushed team_changed broadcast. + public event EventHandler? TeamChanged; + public Task ConnectAsync(TimeSpan timeout, CancellationToken cancellationToken) => Task.FromResult(outcome); @@ -302,8 +321,11 @@ public Task GetInfoAsync(TimeSpan timeout, CancellationToken ca public Task GetTimeAsync(TimeSpan timeout, CancellationToken cancellationToken) => Task.FromResult(TimeResult); - public Task GetTeamInfoAsync(TimeSpan timeout, CancellationToken cancellationToken) => - Task.FromResult(TeamResult); + public Task GetTeamInfoAsync(TimeSpan timeout, CancellationToken cancellationToken) + { + TeamInfoCallCount++; + return Task.FromResult(TeamResult); + } public Task SendTeamMessageAsync(string message, CancellationToken cancellationToken) { @@ -414,10 +436,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); @@ -445,6 +470,10 @@ public Task> GetMonumentsAsync(TimeSpan timeout, /// The probe result to raise. public void RaiseClanChanged(ClanProbeResult result) => ClanChanged?.Invoke(this, result); + /// Raises to simulate a pushed team_changed broadcast. + /// The team snapshot to deliver. + public void RaiseTeamChanged(TeamInfoSnapshot snapshot) => TeamChanged?.Invoke(this, snapshot); + /// Simulates an in-game smart-device state change. /// The smart-device entity id to raise the event for. /// The current active state carried on the trigger arg. diff --git a/tests/RustPlusBot.Features.Connections.Tests/RustPlusBot.Features.Connections.Tests.csproj b/tests/RustPlusBot.Features.Connections.Tests/RustPlusBot.Features.Connections.Tests.csproj index 5586b02b..f3eb42a5 100644 --- a/tests/RustPlusBot.Features.Connections.Tests/RustPlusBot.Features.Connections.Tests.csproj +++ b/tests/RustPlusBot.Features.Connections.Tests/RustPlusBot.Features.Connections.Tests.csproj @@ -6,6 +6,7 @@ + diff --git a/tests/RustPlusBot.Features.Connections.Tests/RustPlusSocketSourceTests.cs b/tests/RustPlusBot.Features.Connections.Tests/RustPlusSocketSourceTests.cs index 4c67b28e..67a0be08 100644 --- a/tests/RustPlusBot.Features.Connections.Tests/RustPlusSocketSourceTests.cs +++ b/tests/RustPlusBot.Features.Connections.Tests/RustPlusSocketSourceTests.cs @@ -1,4 +1,5 @@ using Microsoft.Extensions.Logging.Abstractions; +using RustPlusBot.Abstractions.Connections; using RustPlusBot.Features.Connections.Listening; namespace RustPlusBot.Features.Connections.Tests; @@ -26,4 +27,19 @@ public async Task Create_NonNumericToken_YieldsAuthRejectedConnection() Assert.Equal(SocketConnectOutcome.AuthRejected, await connection.ConnectAsync(TimeSpan.Zero, default)); } + + [Fact] + public void FakeConnection_RaiseTeamChanged_InvokesSubscribers() + { + var source = new Fakes.FakeRustSocketSource(); + source.EnqueueConnect(SocketConnectOutcome.Connected); + var connection = (Fakes.FakeRustSocketSource.FakeConnection)source.Create("127.0.0.1", 28015, 1UL, "1"); + + TeamInfoSnapshot? received = null; + connection.TeamChanged += (_, s) => received = s; + var snapshot = new TeamInfoSnapshot(5UL, []); + connection.RaiseTeamChanged(snapshot); + + Assert.Same(snapshot, received); + } } 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.Connections.Tests/TeamInfoMappingTests.cs b/tests/RustPlusBot.Features.Connections.Tests/TeamInfoMappingTests.cs new file mode 100644 index 00000000..d10094b4 --- /dev/null +++ b/tests/RustPlusBot.Features.Connections.Tests/TeamInfoMappingTests.cs @@ -0,0 +1,61 @@ +using RustPlusBot.Features.Connections.Listening; + +namespace RustPlusBot.Features.Connections.Tests; + +public sealed class TeamInfoMappingTests +{ + [Fact] + public void ToSnapshot_MapsMembersLeaderAndDeathNote() + { + var teamInfo = new RustPlusApi.Data.TeamInfo + { + LeaderSteamId = 100UL, + DeathNote = new RustPlusApi.Data.Notes.DeathNote + { + X = 12f, Y = 34f + }, + Members = + [ + new RustPlusApi.Data.MemberInfo + { + SteamId = 100UL, + Name = "Alice", + X = 1f, + Y = 2f, + IsOnline = true, + IsAlive = false, + LastSpawnTime = new DateTime(2026, 1, 1, 0, 0, 0, DateTimeKind.Unspecified), + LastDeathTime = new DateTime(2026, 1, 2, 0, 0, 0, DateTimeKind.Unspecified), + }, + ], + }; + + var snapshot = TeamInfoMapping.ToSnapshot(teamInfo); + + Assert.Equal(100UL, snapshot.LeaderSteamId); + Assert.Equal((12f, 34f), snapshot.DeathNote); + var m = Assert.Single(snapshot.Members); + Assert.Equal(100UL, m.SteamId); + Assert.Equal("Alice", m.Name); + Assert.True(m.IsOnline); + Assert.False(m.IsAlive); + // Unspecified game timestamps are read as UTC. + Assert.Equal(DateTimeKind.Utc, m.LastDeathTimeUtc.UtcDateTime.Kind); + Assert.Equal(new DateTimeOffset(2026, 1, 2, 0, 0, 0, TimeSpan.Zero), m.LastDeathTimeUtc); + } + + [Fact] + public void ToSnapshot_NullMembersAndNoDeathNote_YieldEmptyMembersAndNullNote() + { + var teamInfo = new RustPlusApi.Data.TeamInfo + { + LeaderSteamId = 7UL, Members = null, DeathNote = null + }; + + var snapshot = TeamInfoMapping.ToSnapshot(teamInfo); + + Assert.Equal(7UL, snapshot.LeaderSteamId); + Assert.Empty(snapshot.Members); + Assert.Null(snapshot.DeathNote); + } +} 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..574b9b5a --- /dev/null +++ b/tests/RustPlusBot.Features.Map.Tests/Hosting/MapHostedServiceTests.cs @@ -0,0 +1,104 @@ +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 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.FromHours(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})"); + } +}