warrentyexpired/ModernUO
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

fix(crowdsec): stop deadlocking and racing shutdown on the flush path

Browse source 
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<ValueTask>) overload, so it bound to
   Task.Run<TResult>(Func<TResult>) and produced a Task<ValueTask> 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<Task>) and unwraps. A loop awaited
   once has nothing to gain from ValueTask.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
This commit is contained in:
Kamron Batman 2026-07-25 11:07:30 -07:00
parent 4d5ca1105d
commit 6f74532280
No known key found for this signature in database
GPG key ID: 7D81DF26D9A5D94A
2 changed files with 65 additions and 26 deletions
Download patch file Download diff file Expand all files Collapse all files

20
Projects/UOContent/Misc/CrowdSec/CrowdSecAlertClient.cs
View file

@ -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<CrowdSecLoginResponse>(_jsonOptions, token);
var login = await response.Content.ReadFromJsonAsync<CrowdSecLoginResponse>(_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);
}
/// <summary>
@ -113,10 +115,10 @@ public sealed class CrowdSecAlertClient : ICrowdSecAlertClient
private async ValueTask SendWithRetryAsync(Func<HttpRequestMessage> 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();
}

71
Projects/UOContent/Misc/CrowdSec/CrowdSecReporter.cs
View file

@ -130,14 +130,19 @@ public sealed class CrowdSecReporter : IBanReporter
}
/// <summary>
/// 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 <see cref="BuildAlerts"/> 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
/// <see cref="CrowdSecSettings.ManualBanDuration"/> 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.
/// </summary>
/// <remarks>
/// 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 <em>on the loop thread</em>.
/// <see cref="Stop"/> runs on the main thread, where <see cref="SynchronizationContext.Current"/> is
/// the <c>EventLoopContext</c>; 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
/// <see cref="Task.Run(Func{Task})"/> 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
/// <c>ConfigureAwait(false)</c>. The bounded <see cref="Task.Wait(TimeSpan)"/> is the backstop behind
/// the flush's own budget: a wedged send costs a few seconds of shutdown, never the process.
/// </remarks>
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");
}
}
/// <summary>
/// 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
/// <see cref="BuildAlerts"/> 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 <see cref="CrowdSecSettings.ManualBanDuration"/> elapses. One shared short budget
/// bounds the whole flush, and any leftover retracts still self-heal via that duration.
/// </summary>
private async Task FlushRemainingOnStopAsync(List<ReportItem> reports, List<ReportItem> 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<ValueTask>) overload, so a ValueTask-returning lambda binds to Task.Run<TResult> and
// yields a Task<ValueTask> 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<ReportItem> reports = [];
List<ReportItem> 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);
}
}
}