diff --git a/src/Testing/CoreTests/Transports/BackPressureAgentTests.cs b/src/Testing/CoreTests/Transports/BackPressureAgentTests.cs index 5445433f4..cbeb601d0 100644 --- a/src/Testing/CoreTests/Transports/BackPressureAgentTests.cs +++ b/src/Testing/CoreTests/Transports/BackPressureAgentTests.cs @@ -1,3 +1,4 @@ +using Microsoft.Extensions.Logging; using NSubstitute; using Wolverine.Configuration; using Wolverine.Runtime.Agents; @@ -13,11 +14,26 @@ public class BackPressureAgentTests private readonly Endpoint theEndpoint = new TcpEndpoint(5555); private readonly IListeningAgent theListeningAgent = Substitute.For(); private readonly IWolverineObserver theObserver; + private readonly RecordingLogger theLogger = new(); public BackPressureAgentTests() { theObserver = Substitute.For(); - theBackPressureAgent = new BackPressureAgent(theListeningAgent, theEndpoint, theObserver); + theBackPressureAgent = new BackPressureAgent(theListeningAgent, theEndpoint, theObserver, theLogger); + } + + private class RecordingLogger : ILogger + { + public readonly List<(LogLevel Level, string Message)> Entries = new(); + + public IDisposable? BeginScope(TState state) where TState : notnull => null; + public bool IsEnabled(LogLevel logLevel) => true; + + public void Log(LogLevel logLevel, EventId eventId, TState state, Exception? exception, + Func formatter) + { + Entries.Add((logLevel, formatter(state, exception))); + } } [Fact] @@ -102,4 +118,37 @@ public async Task restart_when_too_busy_but_below_the_restart_threshold() await theListeningAgent.DidNotReceive().MarkAsTooBusyAndStopReceivingAsync(); await theListeningAgent.Received().StartAsync(); } + + [Fact] + public async Task warns_periodically_while_latched_and_not_draining() + { + // GH CritterWatch#922 — a latched listener used to log exactly one line when it stopped and + // then nothing forever. Operators need a periodic sign of life carrying the numbers the + // resume decision is made from. + theListeningAgent.Status.Returns(ListeningStatus.TooBusy); + theListeningAgent.QueueCount.Returns(theEndpoint.BufferingLimits.Restart + 100); + + for (var i = 0; i < BackPressureAgent.LatchedChecksPerReminder; i++) + { + await theBackPressureAgent.CheckNowAsync(); + } + + var warning = theLogger.Entries.ShouldHaveSingleItem(); + warning.Level.ShouldBe(LogLevel.Warning); + warning.Message.ShouldContain("still latched by back pressure"); + + // and it repeats on the next full interval rather than spamming every check + for (var i = 0; i < BackPressureAgent.LatchedChecksPerReminder; i++) + { + await theBackPressureAgent.CheckNowAsync(); + } + + theLogger.Entries.Count.ShouldBe(2); + + // recovering resets the cadence + theListeningAgent.QueueCount.Returns(theEndpoint.BufferingLimits.Restart); + await theBackPressureAgent.CheckNowAsync(); + await theListeningAgent.Received().StartAsync(); + theLogger.Entries.Count.ShouldBe(2); + } } \ No newline at end of file diff --git a/src/Transports/RabbitMQ/Wolverine.RabbitMQ.Tests/Bugs/Bug_189_fails_if_there_are_many_messages_in_queue_on_startup.cs b/src/Transports/RabbitMQ/Wolverine.RabbitMQ.Tests/Bugs/Bug_189_fails_if_there_are_many_messages_in_queue_on_startup.cs index 1fe03baba..ee3c9872f 100644 --- a/src/Transports/RabbitMQ/Wolverine.RabbitMQ.Tests/Bugs/Bug_189_fails_if_there_are_many_messages_in_queue_on_startup.cs +++ b/src/Transports/RabbitMQ/Wolverine.RabbitMQ.Tests/Bugs/Bug_189_fails_if_there_are_many_messages_in_queue_on_startup.cs @@ -58,6 +58,18 @@ await Host.CreateDefaultBuilder() try { + // The receiver host is deliberately not awaited before the waiter -- StartAsync can block + // draining a full queue under ProcessInline. But leaving it entirely unobserved meant a + // host that failed to *start* produced no error at all: the waiter simply ran out its 120 + // seconds and the test reported a bare "System.TimeoutException : The operation has timed + // out." with the real cause discarded. Race the two so a startup failure is rethrown as + // itself. + var finished = await Task.WhenAny(waiter, receiverTask); + if (ReferenceEquals(finished, receiverTask) && receiverTask.IsFaulted) + { + await receiverTask; + } + await waiter; } finally @@ -67,6 +79,11 @@ await Host.CreateDefaultBuilder() var host = await receiverTask; await host.StopAsync(TestContext.Current.CancellationToken); } + else + { + // Never leave the startup task's exception unobserved, whichever way we exit + _ = receiverTask.ContinueWith(t => _ = t.Exception, TaskScheduler.Default); + } } } @@ -74,15 +91,37 @@ public record Bug189(Guid Id); public static class Bug189Handler { - private static TaskCompletionSource _source = new TaskCompletionSource(); - private static volatile int _count = 0; + private static TaskCompletionSource _source = + new(TaskCreationOptions.RunContinuationsAsynchronously); + + private static int _count; private static int _expected; public static Task WaitForCompletion(int count, int millisecondTimeout) { + // Reset per invocation. These are statics on a static class, so they survive for the life + // of the worker process -- and the supervisor retries a failed test inside that same + // process. A stale, already-completed _source made attempt 2 return a finished task and + // "pass" without receiving anything, which is how this test has been showing up as + // "passed on attempt 2" on green main runs while never actually re-verifying the fix. + Interlocked.Exchange(ref _count, 0); _expected = count; + _source = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + + var source = _source; + + return source.Task.TimeoutAfterAsync(millisecondTimeout).ContinueWith(t => + { + // A bare TimeoutException says nothing about how far the run got, which is the one + // number separating "nothing was ever consumed" from "consumption was merely slow". + if (t.IsFaulted && t.Exception!.InnerException is TimeoutException) + { + throw new TimeoutException( + $"Only {Volatile.Read(ref _count)} of the expected {count} messages were handled within {millisecondTimeout}ms"); + } - return _source.Task.TimeoutAfterAsync(millisecondTimeout); + return t.GetAwaiter().GetResult(); + }, TaskScheduler.Default); } public static void Handle(Bug189 bug, Envelope envelope) @@ -95,11 +134,15 @@ public static void Handle(Bug189 bug, Envelope envelope) } } - _count++; + // Five inline listeners run this concurrently. `_count++` on a volatile int is a + // read-modify-write, so increments were being lost outright, and two threads could both + // clear the threshold and call SetResult -- the second throwing InvalidOperationException + // from inside a message handler. + var count = Interlocked.Increment(ref _count); - if (_count >= _expected) + if (count >= _expected) { - _source.SetResult(_count); + _source.TrySetResult(count); } } } diff --git a/src/Wolverine/Runtime/Partitioning/ShardedExecutionBlock.cs b/src/Wolverine/Runtime/Partitioning/ShardedExecutionBlock.cs index 91859e6bd..6050bd725 100644 --- a/src/Wolverine/Runtime/Partitioning/ShardedExecutionBlock.cs +++ b/src/Wolverine/Runtime/Partitioning/ShardedExecutionBlock.cs @@ -78,6 +78,22 @@ public override void Complete() public override uint Count => (uint)_slots.Sum(x => x.Count); + /// + /// Propagates to every slot block. Without this, a slot's escaping exception falls to the + /// JasperFx Block default sink (stderr) — invisible to anyone reading structured logs. + /// + public override Action OnError + { + get => _slots[0].OnError; + set + { + foreach (var slot in _slots) + { + slot.OnError = value; + } + } + } + public override ValueTask PostAsync(Envelope item) { // This first uses new "message grouping rules" to determine a GroupId diff --git a/src/Wolverine/Runtime/WorkerQueues/BufferedReceiver.cs b/src/Wolverine/Runtime/WorkerQueues/BufferedReceiver.cs index 3ebb4d9f1..b33c42dc7 100644 --- a/src/Wolverine/Runtime/WorkerQueues/BufferedReceiver.cs +++ b/src/Wolverine/Runtime/WorkerQueues/BufferedReceiver.cs @@ -59,9 +59,24 @@ protected BufferedReceiver(Endpoint endpoint, IWolverineRuntime runtime, IHandle (env, _) => env.Listener is { } l ? l.CompleteAsync(env).AsTask() : Task.CompletedTask, runtime.Logger, runtime.Cancellation); - _receivingBlock = endpoint.GroupShardingSlotNumber == null - ? new Block(endpoint.MaxDegreeOfParallelism, boundedCapacity, executeAsync) - : new ShardedExecutionBlock((int)endpoint.GroupShardingSlotNumber, runtime.Options.MessagePartitioning, boundedCapacity, executeAsync).DeserializeFirst(pipeline, runtime, this); + if (endpoint.GroupShardingSlotNumber == null) + { + _receivingBlock = new Block(endpoint.MaxDegreeOfParallelism, boundedCapacity, executeAsync); + } + else + { + var sharded = new ShardedExecutionBlock((int)endpoint.GroupShardingSlotNumber, + runtime.Options.MessagePartitioning, boundedCapacity, executeAsync); + sharded.OnError = onBlockError; + _receivingBlock = sharded.DeserializeFirst(pipeline, runtime, this); + } + + // Route block-level failures (an exception escaping the execution machinery itself, or the + // block faulting terminally per jasperfx#506) through real logging. The JasperFx default sink + // is stderr, which reads as a silent stall in any structured-logging deployment — and a + // faulted block freezes QueueCount, which permanently latches a back-pressured listener + // (GH CritterWatch#922). + _receivingBlock.OnError = onBlockError; if (endpoint.TryBuildDeadLetterSender(runtime, out var dlq)) { @@ -73,6 +88,22 @@ protected BufferedReceiver(Endpoint endpoint, IWolverineRuntime runtime, IHandle } } + private void onBlockError(Envelope? envelope, Exception ex) + { + // A terminal block fault (jasperfx#506) reports with a null item + if (envelope == null) + { + _logger.LogCritical(ex, + "The local worker queue for {Uri} has faulted and stopped processing. Messages buffered locally will not be executed", + Uri); + } + else + { + _logger.LogError(ex, "Error processing envelope {EnvelopeId} ({MessageType}) in the local worker queue for {Uri}", + envelope.Id, envelope.MessageType, Uri); + } + } + internal async Task executeAsync(Envelope envelope, CancellationToken _) { if (_latched && envelope.Listener != null) diff --git a/src/Wolverine/Runtime/WorkerQueues/DurableReceiver.cs b/src/Wolverine/Runtime/WorkerQueues/DurableReceiver.cs index a3c8573eb..63b27d00b 100644 --- a/src/Wolverine/Runtime/WorkerQueues/DurableReceiver.cs +++ b/src/Wolverine/Runtime/WorkerQueues/DurableReceiver.cs @@ -53,6 +53,23 @@ public DurableReceiver(Endpoint endpoint, IWolverineRuntime runtime, IHandlerPip Pipeline = pipeline; + void onBlockError(Envelope? envelope, Exception ex) + { + // A terminal block fault (jasperfx#506) reports with a null item + if (envelope == null) + { + _logger.LogCritical(ex, + "The local worker queue for {Uri} has faulted and stopped processing. Messages buffered locally will not be executed", + Uri); + } + else + { + _logger.LogError(ex, + "Error processing envelope {EnvelopeId} ({MessageType}) in the local worker queue for {Uri}", + envelope.Id, envelope.MessageType, Uri); + } + } + Func execute = async (envelope, _) => { if (_latched) @@ -78,9 +95,24 @@ public DurableReceiver(Endpoint endpoint, IWolverineRuntime runtime, IHandlerPip } }; - _receiver = endpoint.GroupShardingSlotNumber == null - ? new Block(endpoint.MaxDegreeOfParallelism, execute) - : new ShardedExecutionBlock((int)endpoint.GroupShardingSlotNumber, runtime.Options.MessagePartitioning, execute).DeserializeFirst(pipeline, runtime, this); + if (endpoint.GroupShardingSlotNumber == null) + { + _receiver = new Block(endpoint.MaxDegreeOfParallelism, execute); + } + else + { + var sharded = new ShardedExecutionBlock((int)endpoint.GroupShardingSlotNumber, + runtime.Options.MessagePartitioning, execute); + sharded.OnError = onBlockError; + _receiver = sharded.DeserializeFirst(pipeline, runtime, this); + } + + // Route block-level failures (an exception escaping the execution machinery itself, or the + // block faulting terminally per jasperfx#506) through real logging. The JasperFx default sink + // is stderr, which reads as a silent stall in any structured-logging deployment — and a + // faulted block freezes QueueCount, which permanently latches a back-pressured listener + // (GH CritterWatch#922). + _receiver.OnError = onBlockError; _deferBlock = new RetryBlock((env, _) => env.Listener!.DeferAsync(env).AsTask(), runtime.Logger, runtime.Cancellation); diff --git a/src/Wolverine/Transports/BackPressureAgent.cs b/src/Wolverine/Transports/BackPressureAgent.cs index 0f3c9919b..57da54af8 100644 --- a/src/Wolverine/Transports/BackPressureAgent.cs +++ b/src/Wolverine/Transports/BackPressureAgent.cs @@ -1,4 +1,5 @@ using System.Timers; +using Microsoft.Extensions.Logging; using Wolverine.Configuration; using Wolverine.Runtime.Agents; using Timer = System.Timers.Timer; @@ -7,16 +8,22 @@ namespace Wolverine.Transports; internal class BackPressureAgent : IDisposable { + // At the 2 second polling interval, this logs roughly once a minute while latched + internal const int LatchedChecksPerReminder = 30; + private readonly IListeningAgent _agent; private readonly Endpoint _endpoint; private readonly IWolverineObserver _observer; + private readonly ILogger _logger; private Timer? _timer; + private int _latchedChecks; - public BackPressureAgent(IListeningAgent agent, Endpoint endpoint, IWolverineObserver observer) + public BackPressureAgent(IListeningAgent agent, Endpoint endpoint, IWolverineObserver observer, ILogger logger) { _agent = agent; _endpoint = endpoint; _observer = observer; + _logger = logger; } public void Dispose() @@ -36,11 +43,24 @@ public void Start() private void TimerOnElapsed(object? sender, ElapsedEventArgs e) { -#pragma warning disable CS4014 -#pragma warning disable VSTHRD110 - CheckNowAsync(); -#pragma warning restore VSTHRD110 -#pragma warning restore CS4014 + _ = checkSafelyAsync(); + } + + private async Task checkSafelyAsync() + { + // An exception escaping CheckNowAsync from the timer used to be an unobserved ValueTask + // fault — a listener whose restart kept throwing simply never resumed, with nothing in the + // logs (GH CritterWatch#922). The timer keeps firing, so log and let the next interval retry. + try + { + await CheckNowAsync(); + } + catch (Exception ex) + { + _logger.LogError(ex, + "Back pressure check failed for listener at {Uri} (status {Status}, local count {Count}). Will retry on the next interval", + _endpoint.Uri, _agent.Status, _agent.QueueCount); + } } public async ValueTask CheckNowAsync() @@ -53,6 +73,8 @@ public async ValueTask CheckNowAsync() if (_agent.Status is ListeningStatus.Accepting or ListeningStatus.Unknown) { + _latchedChecks = 0; + if (_agent.QueueCount > _endpoint.BufferingLimits.Maximum) { await _observer.BackPressureTriggered(_endpoint, _agent); @@ -63,9 +85,24 @@ public async ValueTask CheckNowAsync() { if (_agent.QueueCount <= _endpoint.BufferingLimits.Restart) { + _latchedChecks = 0; await _agent.StartAsync(); await _observer.BackPressureLifted(_endpoint); } + else if (++_latchedChecks % LatchedChecksPerReminder == 0) + { + // A latched listener used to log exactly one line at the moment it stopped and then + // nothing forever — operators watching a queue grow for 40 minutes had no way to tell + // "still draining" from "wedged" (GH CritterWatch#922). Say so, with the numbers the + // resume decision is actually made from. + _logger.LogWarning( + "Listener at {Uri} is still latched by back pressure after {Seconds:N0}s: local count {Count} has not dropped to the restart threshold {Restart}. If the count is not falling, the local queue is not draining", + _endpoint.Uri, _latchedChecks * 2, _agent.QueueCount, _endpoint.BufferingLimits.Restart); + } + } + else + { + _latchedChecks = 0; } } -} \ No newline at end of file +} diff --git a/src/Wolverine/Transports/ListeningAgent.cs b/src/Wolverine/Transports/ListeningAgent.cs index 4c85e66b2..664f0c144 100644 --- a/src/Wolverine/Transports/ListeningAgent.cs +++ b/src/Wolverine/Transports/ListeningAgent.cs @@ -98,7 +98,7 @@ public ListeningAgent(Endpoint endpoint, WolverineRuntime runtime) if (endpoint.ShouldEnforceBackPressure()) { - _backPressureAgent = new BackPressureAgent(this, endpoint, runtime.Observer); + _backPressureAgent = new BackPressureAgent(this, endpoint, runtime.Observer, _logger); _backPressureAgent.Start(); } }