Back-pressure observability: a latched listener says so, and block errors reach real logging

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<IListeningAgent>();
     private readonly IWolverineObserver theObserver;
+    private readonly RecordingLogger theLogger = new();
 
     public BackPressureAgentTests()
     {
         theObserver = Substitute.For<IWolverineObserver>();
-        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>(TState state) where TState : notnull => null;
+        public bool IsEnabled(LogLevel logLevel) => true;
+
+        public void Log<TState>(LogLevel logLevel, EventId eventId, TState state, Exception? exception,
+            Func<TState, Exception?, string> 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);
+    }
 }

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,22 +79,49 @@ 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);
+            }
         }
     }
 
     public record Bug189(Guid Id);
 
     public static class Bug189Handler
     {
-        private static TaskCompletionSource<int> _source = new TaskCompletionSource<int>();
-        private static volatile int _count = 0;
+        private static TaskCompletionSource<int> _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<int>(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);
             }
         }
     }

src/Wolverine/Runtime/Partitioning/ShardedExecutionBlock.cs

@@ -78,6 +78,22 @@ public override void Complete()
 
     public override uint Count => (uint)_slots.Sum(x => x.Count);
 
+    /// <summary>
+    /// 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.
+    /// </summary>
+    public override Action<Envelope, Exception> 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

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<Envelope>(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<Envelope>(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)

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<Envelope, CancellationToken, Task> execute = async (envelope, _) =>
         {
             if (_latched)
@@ -78,9 +95,24 @@ public DurableReceiver(Endpoint endpoint, IWolverineRuntime runtime, IHandlerPip
             }
         };
 
-        _receiver = endpoint.GroupShardingSlotNumber == null 
-            ? new Block<Envelope>(endpoint.MaxDegreeOfParallelism, execute)
-            : new ShardedExecutionBlock((int)endpoint.GroupShardingSlotNumber, runtime.Options.MessagePartitioning, execute).DeserializeFirst(pipeline, runtime, this);
+        if (endpoint.GroupShardingSlotNumber == null)
+        {
+            _receiver = new Block<Envelope>(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<Envelope>((env, _) => env.Listener!.DeferAsync(env).AsTask(), runtime.Logger,
             runtime.Cancellation);

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;
         }
     }
-}
+}

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();
         }
     }