From 6f7453228028e8f509ab70e20c5fededc089ce2c Mon Sep 17 00:00:00 2001 From: Kamron Batman <3953314+kamronbatman@users.noreply.github.com> Date: Sat, 25 Jul 2026 11:07:30 -0700 Subject: [PATCH] fix(crowdsec): stop deadlocking and racing shutdown on the flush path Two defects in CrowdSecReporter shutdown, both reachable only on a shard that actually has CrowdSec configured with items still queued. 1. FlushRemainingOnStop blocked with GetAwaiter().GetResult() on a ValueTask. Beyond being unsupported in general -- an IValueTaskSource-backed ValueTask does not wait there, it throws -- it hung. Stop() runs on the main thread, where Main installs EventLoopContext as SynchronizationContext.Current, and nothing in CrowdSecAlertClient used ConfigureAwait(false), so the send's continuation was posted to a queue only LoopContext.ExecuteTasks() drains -- and by HandleClosed() the loop has stopped ticking. The thread waited on a continuation only that thread could run. The 3s budget did not help: it completes the HTTP call, not the resumption. Run the flush through Task.Run so the whole chain lives on the pool with no context to capture, block once on a real Task, and bound it with Wait so a wedged send costs a few seconds of shutdown rather than the process. Add ConfigureAwait(false) across the client and drain loop per rule #10 so the pool hop is defense in depth rather than the only thing holding it up. 2. Start() did Task.Run(() => DrainLoop(...)) where DrainLoop returned ValueTask. There is no Task.Run(Func) overload, so it bound to Task.Run(Func) and produced a Task that completes at the first suspending await, not when the loop exits -- silently upcast by the Task _drainTask field. Stop()'s drain-exited handshake was therefore a no-op (Wait returned true in ~0ms), letting the flush read a SingleReader channel concurrently with a live drain loop, which is precisely the hazard that handshake documents itself as preventing. It also made the loop fire-and-forget and swallowed any fault escaping it. Return Task so it binds to Task.Run(Func) and unwraps. A loop awaited once has nothing to gain from ValueTask. Co-Authored-By: Claude Opus 5 (1M context) --- .../Misc/CrowdSec/CrowdSecAlertClient.cs | 20 +++--- .../Misc/CrowdSec/CrowdSecReporter.cs | 71 ++++++++++++++----- 2 files changed, 65 insertions(+), 26 deletions(-) diff --git a/Projects/UOContent/Misc/CrowdSec/CrowdSecAlertClient.cs b/Projects/UOContent/Misc/CrowdSec/CrowdSecAlertClient.cs index 38af6b319..af853244a 100644 --- a/Projects/UOContent/Misc/CrowdSec/CrowdSecAlertClient.cs +++ b/Projects/UOContent/Misc/CrowdSec/CrowdSecAlertClient.cs @@ -1,4 +1,4 @@ -/************************************************************************* +/************************************************************************* * ModernUO * * Copyright 2019-2026 - ModernUO Development Team * * Email: hi@modernuo.com * @@ -64,10 +64,12 @@ public sealed class CrowdSecAlertClient : ICrowdSecAlertClient } var request = new CrowdSecLoginRequest { MachineId = _machineId, Password = _password }; - using var response = await _http.PostAsJsonAsync("/v1/watchers/login", request, _jsonOptions, token); + using var response = await _http.PostAsJsonAsync("/v1/watchers/login", request, _jsonOptions, token) + .ConfigureAwait(false); response.EnsureSuccessStatusCode(); - var login = await response.Content.ReadFromJsonAsync(_jsonOptions, token); + var login = await response.Content.ReadFromJsonAsync(_jsonOptions, token) + .ConfigureAwait(false); _token = login?.Token ?? throw new InvalidOperationException("CrowdSec login returned no token."); _tokenExpiresUtc = DateTime.TryParse(login.Expire, out var exp) ? exp.ToUniversalTime() : DateTime.UtcNow.AddHours(1); } @@ -90,7 +92,7 @@ public sealed class CrowdSecAlertClient : ICrowdSecAlertClient }; Authorize(message); return message; - }, token); + }, token).ConfigureAwait(false); } public async ValueTask DeleteDecisionsAsync(string origin, IPAddress ip, CancellationToken token) @@ -101,7 +103,7 @@ public sealed class CrowdSecAlertClient : ICrowdSecAlertClient var message = new HttpRequestMessage(HttpMethod.Delete, query); Authorize(message); return message; - }, token); + }, token).ConfigureAwait(false); } /// @@ -113,10 +115,10 @@ public sealed class CrowdSecAlertClient : ICrowdSecAlertClient private async ValueTask SendWithRetryAsync(Func build, CancellationToken token) { - await EnsureAuthAsync(token); + await EnsureAuthAsync(token).ConfigureAwait(false); using var first = build(); - using var response = await _http.SendAsync(first, token); + using var response = await _http.SendAsync(first, token).ConfigureAwait(false); if (response.StatusCode != HttpStatusCode.Unauthorized) { response.EnsureSuccessStatusCode(); @@ -125,9 +127,9 @@ public sealed class CrowdSecAlertClient : ICrowdSecAlertClient // Token rejected mid-flight: force a re-login and retry once. _token = null; - await EnsureAuthAsync(token); + await EnsureAuthAsync(token).ConfigureAwait(false); using var retry = build(); - using var retryResponse = await _http.SendAsync(retry, token); + using var retryResponse = await _http.SendAsync(retry, token).ConfigureAwait(false); retryResponse.EnsureSuccessStatusCode(); } diff --git a/Projects/UOContent/Misc/CrowdSec/CrowdSecReporter.cs b/Projects/UOContent/Misc/CrowdSec/CrowdSecReporter.cs index 7b632b384..84cfbdf31 100644 --- a/Projects/UOContent/Misc/CrowdSec/CrowdSecReporter.cs +++ b/Projects/UOContent/Misc/CrowdSec/CrowdSecReporter.cs @@ -130,14 +130,19 @@ public sealed class CrowdSecReporter : IBanReporter } /// - /// Best-effort bounded flush of whatever contribution items are still queued at shutdown. Runs on a - /// fresh, short-lived token — NOT the drain loop's (already-cancelled) token — since a cancelled token - /// would make the send fail immediately. Pending reports are deduped via and - /// sent as a single batch; pending retracts are issued as individual (deduped) DELETEs so an admin's - /// explicit unban still propagates on a clean shutdown instead of lingering until - /// elapses. One shared short budget bounds the whole - /// flush: a shutdown must never hang on this, and any leftover retracts still self-heal via that duration. + /// Best-effort bounded flush of whatever contribution items are still queued at shutdown. /// + /// + /// Shutdown is the one place where blocking is the right answer — the loop has stopped ticking, so + /// there is no later tick to resume on — but it must not block on the loop thread. + /// runs on the main thread, where is + /// the EventLoopContext; an await that captured it would post its continuation to a queue + /// nothing pumps any more, and we would wait here forever. Running the flush through + /// puts the whole chain on the pool, where there is no context to + /// capture, so correctness does not depend on every await in the client remembering + /// ConfigureAwait(false). The bounded is the backstop behind + /// the flush's own budget: a wedged send costs a few seconds of shutdown, never the process. + /// private void FlushRemainingOnStop() { if (_queue == null || _client == null) @@ -159,7 +164,32 @@ public sealed class CrowdSecReporter : IBanReporter return; } - // One shared, short budget bounds the entire flush so shutdown can never hang on it. + try + { + if (!Task.Run(() => FlushRemainingOnStopAsync(reports, retracts)).Wait(TimeSpan.FromSeconds(4))) + { + logger.Warning( + "CrowdSec flush-on-stop timed out; {Count} item(s) not contributed", + reports.Count + retracts.Count + ); + } + } + catch (Exception e) + { + logger.Warning(e, "CrowdSec flush-on-stop failed"); + } + } + + /// + /// Runs on a fresh, short-lived token — NOT the drain loop's (already-cancelled) token — since a + /// cancelled token would make the send fail immediately. Pending reports are deduped via + /// and sent as a single batch; pending retracts are issued as individual + /// (deduped) DELETEs so an admin's explicit unban still propagates on a clean shutdown instead of + /// lingering until elapses. One shared short budget + /// bounds the whole flush, and any leftover retracts still self-heal via that duration. + /// + private async Task FlushRemainingOnStopAsync(List reports, List retracts) + { using var flushCts = new CancellationTokenSource(TimeSpan.FromSeconds(3)); if (reports.Count > 0) @@ -167,7 +197,7 @@ public sealed class CrowdSecReporter : IBanReporter var alerts = BuildAlerts(reports, _settings, DateTime.UtcNow); try { - _client.PostAlertsAsync(alerts, flushCts.Token).GetAwaiter().GetResult(); + await _client.PostAlertsAsync(alerts, flushCts.Token).ConfigureAwait(false); } catch (Exception e) { @@ -191,7 +221,7 @@ public sealed class CrowdSecReporter : IBanReporter try { - _client.DeleteDecisionsAsync(_settings.Origin, retract.Ip, flushCts.Token).GetAwaiter().GetResult(); + await _client.DeleteDecisionsAsync(_settings.Origin, retract.Ip, flushCts.Token).ConfigureAwait(false); } catch (Exception e) { @@ -226,7 +256,12 @@ public sealed class CrowdSecReporter : IBanReporter SingleReader = true }); - private async ValueTask DrainLoop(CancellationToken token) + // Returns Task, not ValueTask, precisely because Start() hands it to Task.Run: there is no + // Task.Run(Func) overload, so a ValueTask-returning lambda binds to Task.Run and + // yields a Task that completes at the FIRST await instead of when the loop exits. That + // would make Stop()'s drain-exited handshake a no-op and let the flush race this loop on a + // SingleReader channel. A loop awaited once has nothing to gain from ValueTask anyway. + private async Task DrainLoop(CancellationToken token) { var reader = _queue.Reader; @@ -234,13 +269,13 @@ public sealed class CrowdSecReporter : IBanReporter { try { - if (!await reader.WaitToReadAsync(token)) + if (!await reader.WaitToReadAsync(token).ConfigureAwait(false)) { return; } // Coalesce a burst before flushing. - await Task.Delay(_settings.FlushInterval, token); + await Task.Delay(_settings.FlushInterval, token).ConfigureAwait(false); List reports = []; List retracts = []; @@ -252,7 +287,8 @@ public sealed class CrowdSecReporter : IBanReporter if (reports.Count > 0) { var alerts = BuildAlerts(reports, _settings, DateTime.UtcNow); - if (!await SendWithBoundedRetryAsync(() => _client.PostAlertsAsync(alerts, token), token)) + if (!await SendWithBoundedRetryAsync(() => _client.PostAlertsAsync(alerts, token), token) + .ConfigureAwait(false)) { RecordSendFailure(alerts.Count); } @@ -261,7 +297,8 @@ public sealed class CrowdSecReporter : IBanReporter foreach (var retract in retracts) { var ip = retract.Ip; - if (!await SendWithBoundedRetryAsync(() => _client.DeleteDecisionsAsync(_settings.Origin, ip, token), token)) + if (!await SendWithBoundedRetryAsync(() => _client.DeleteDecisionsAsync(_settings.Origin, ip, token), token) + .ConfigureAwait(false)) { RecordSendFailure(1); } @@ -293,7 +330,7 @@ public sealed class CrowdSecReporter : IBanReporter { try { - await send(); + await send().ConfigureAwait(false); return true; } catch (OperationCanceledException) @@ -310,7 +347,7 @@ public sealed class CrowdSecReporter : IBanReporter var delay = _retryDelays[attempt]; logger.Warning(e, "CrowdSec send failed (attempt {Attempt}); retrying in {Delay}", attempt + 1, delay); - await Task.Delay(delay, token); + await Task.Delay(delay, token).ConfigureAwait(false); } } }