From 9541d276a0e69d87d62dd4c9b2a33ff161da3c71 Mon Sep 17 00:00:00 2001 From: "Jeremy D. Miller" Date: Tue, 4 Aug 2026 16:46:05 -0500 Subject: [PATCH 1/2] Back-pressure observability: a latched listener says so, and block errors reach real logging MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Three hardenings from chasing a permanently-latched ingest listener (JasperFx/CritterWatch#922 — one 'too busy' line, no resume ever, queue to 288k): - BackPressureAgent logs a periodic warning (about once a minute) while a listener stays latched, carrying the QueueCount and restart threshold the resume decision reads. It used to log exactly one line at latch time and nothing ever again. - The timer-driven back-pressure check is now exception-safe: a throw from StartAsync during an attempted resume was an unobserved ValueTask fault — the listener simply never resumed with nothing in the logs. Now logged and retried on the next interval. - BufferedReceiver/DurableReceiver wire the receiving block's OnError to ILogger, and ShardedExecutionBlock propagates OnError to its slot blocks. The JasperFx default sink is stderr; a terminally-faulted block (jasperfx#506) freezes QueueCount — which permanently latches a back-pressured listener — so it now logs at Critical. Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_016v2Aijyo8MX2AdPUZL5VtG --- .../Transports/BackPressureAgentTests.cs | 51 ++++++++++++++++++- .../Partitioning/ShardedExecutionBlock.cs | 16 ++++++ .../Runtime/WorkerQueues/BufferedReceiver.cs | 37 ++++++++++++-- .../Runtime/WorkerQueues/DurableReceiver.cs | 38 ++++++++++++-- src/Wolverine/Transports/BackPressureAgent.cs | 51 ++++++++++++++++--- src/Wolverine/Transports/ListeningAgent.cs | 2 +- 6 files changed, 180 insertions(+), 15 deletions(-) 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/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(); } } From fc74c86630be9d1f9fb0c0057092f1bc75139ca4 Mon Sep 17 00:00:00 2001 From: "Jeremy D. Miller" Date: Tue, 4 Aug 2026 19:27:21 -0500 Subject: [PATCH 2/2] Fix four defects in Bug_189 that made its failures undiagnosable (GH-3832) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit CIRabbitMQ failed on this PR with Bug_189...be_able_to_start_up_with_large_number_of_messages_waiting_on_you — System.TimeoutException : The operation has timed out. The test has four separate problems, and together they explain both the failure and why it has never been diagnosable. 1. The retry passed vacuously. Bug189Handler keeps _source, _count and _expected as statics on a static class, and WaitForCompletion never reset them. Those survive for the life of the worker process, and the supervisor retries a failed test inside that same process, so attempt 2 got an already-completed TaskCompletionSource back and returned without receiving a single message. That is how this test reports "passed on attempt 2" on green main runs (e.g. run 30950507401) while re-verifying nothing. WaitForCompletion now resets the count and installs a fresh source, so a retry is a real retry. 2. A failed receiver startup was invisible. receiverTask is deliberately fire-and-forget -- StartAsync can block draining a full queue under ProcessInline -- but it was never observed at all, so a host that failed to start produced no error: the waiter simply ran out its 120 seconds and reported a bare TimeoutException with the real cause discarded, which is exactly the message CI produced. The two are now raced so a startup fault is rethrown as itself, and the task's exception is always observed. 3. _count++ on a volatile int is a read-modify-write, and five inline listeners run the handler concurrently, so increments were being lost. Now Interlocked.Increment. 4. Two threads could both clear the threshold and call SetResult, the second throwing InvalidOperationException from inside a message handler. Now TrySetResult. The timeout message also carries how many messages were actually handled, so the next failure distinguishes "nothing was consumed" from "consumption was slow". Note what this does not claim: locally the test finishes in ~830ms against a 120s budget and is green 5 of 5, so the CI timing failure is not reproducible here and is not asserted to be fixed. What changes is that the next failure will say how far it got instead of nothing at all. Full local Wolverine.RabbitMQ.Tests: 487 passed, Bug_189 among them. Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_01WHAuhdWS3XeAk16swV9G8m --- ...e_are_many_messages_in_queue_on_startup.cs | 55 +++++++++++++++++-- 1 file changed, 49 insertions(+), 6 deletions(-) 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); } } }