GH-4186: report a NativeAck listener's real queue depth and last receipt

src/Testing/CoreTests/Runtime/WorkerQueues/native_ack_queue_depth_4186.cs

@@ -0,0 +1,193 @@
+using System.Reflection;
+using JasperFx.Core;
+using Microsoft.Extensions.DependencyInjection;
+using Microsoft.Extensions.Hosting;
+using Shouldly;
+using Wolverine;
+using Wolverine.Configuration;
+using Wolverine.Runtime;
+using Wolverine.Runtime.WorkerQueues;
+using Wolverine.Transports;
+using Wolverine.Transports.Stub;
+using Wolverine.Util;
+using Xunit;
+
+namespace CoreTests.Runtime.WorkerQueues;
+
+/// <summary>
+/// GH-4186. ListeningAgent read the receiver's depth through ILocalQueue, and neither NativeAckReceiver nor
+/// InlineReceiver is one -- deliberately, because a delivery in those two modes settles against the listener that
+/// brought it rather than against a local queue. Both maintain a real depth over a real block, and both used to
+/// contribute a constant 0 to EndpointHealthSnapshot: a saturated NativeAck listener was indistinguishable from
+/// an idle one, and rendered downstream as a reassuring green zero rather than as "unknown".
+///
+/// The companion half is LastQueueActivityAt, whose change-detection heuristic has exactly one writer --
+/// BackPressureAgent -- which correctly does not run for either mode (see Endpoint.ShouldEnforceBackPressure),
+/// leaving the timestamp frozen at listener construction forever.
+/// </summary>
+public class native_ack_queue_depth_4186 : IAsyncLifetime
+{
+    private IHost _host = null!;
+    private WolverineRuntime theRuntime = null!;
+
+    public async ValueTask InitializeAsync()
+    {
+        _host = await Host.CreateDefaultBuilder()
+            .UseWolverine(opts => opts.Discovery.IncludeType<NativeAckPingHandler>())
+            .StartAsync(TestContext.Current.CancellationToken);
+
+        theRuntime = (WolverineRuntime)_host.Services.GetRequiredService<IWolverineRuntime>();
+        NativeAckPingHandler.Handled.Clear();
+        NativeAckPingHandler.Gate = null;
+        NativeAckPingHandler.Entered = null;
+    }
+
+    public async ValueTask DisposeAsync()
+    {
+        NativeAckPingHandler.Gate = null;
+        NativeAckPingHandler.Entered = null;
+        await _host.StopAsync();
+        _host.Dispose();
+    }
+
+    [Fact]
+    public async Task depth_and_receipt_activity_reach_the_endpoint_health_snapshot()
+    {
+        var endpoint = new NativeAckStubEndpoint("na-4186", new StubTransport()) { IsListener = true };
+        endpoint.Mode = EndpointMode.NativeAck;
+
+        await theRuntime.Endpoints.StartListenerAsync(endpoint, CancellationToken.None);
+        var agent = theRuntime.Endpoints.FindListeningAgent(endpoint.Uri).ShouldNotBeNull();
+
+        // Guard against a vacuous pass: a BufferedReceiver here already reported its depth before GH-4186,
+        // and the test would prove nothing.
+        receiverOf(agent).ShouldBeOfType<NativeAckReceiver>();
+
+        var idle = snapshotFor(endpoint.Uri);
+        idle.QueueCount.ShouldBe(0);
+
+        // Not a wait on anything -- pure clock separation, so that the "moved" assertion below cannot pass or
+        // fail on DateTimeOffset.UtcNow's granularity (~15ms on Windows) rather than on the fix.
+        await Task.Delay(50.Milliseconds(), TestContext.Current.CancellationToken);
+
+        var gate = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
+        NativeAckPingHandler.Gate = gate;
+
+        try
+        {
+            // Deliberately not awaited: with every handler parked on the gate, a large enough batch will fill
+            // the block's bounded channel and the post itself will block. Backgrounding it keeps this test
+            // agnostic about that capacity.
+            var flood = Task.Run(() => agent.EnqueueDirectlyAsync(
+                Enumerable.Range(0, 200).Select(i => pingEnvelope("flood-" + i)).ToArray()), TestContext.Current.CancellationToken);
+
+            await waitUntil(() => snapshotFor(endpoint.Uri).QueueCount > 0,
+                "the NativeAck listener never reported a non-zero QueueCount");
+
+            var saturated = snapshotFor(endpoint.Uri);
+
+            // The whole point of the issue: the depth is real, and it is now visible.
+            saturated.QueueCount.ShouldBeGreaterThan(0);
+
+            // ...and the listener no longer looks like it has been idle since boot.
+            saturated.LastQueueActivityAt.ShouldNotBeNull();
+            saturated.LastQueueActivityAt.Value.ShouldBeGreaterThan(idle.LastQueueActivityAt!.Value);
+
+            gate.SetResult();
+            await flood;
+        }
+        finally
+        {
+            gate.TrySetResult();
+        }
+
+        // And it is a live number rather than a one-time stamp -- it comes back down as the block drains.
+        await waitUntil(() => snapshotFor(endpoint.Uri).QueueCount == 0,
+            "the NativeAck listener's reported QueueCount never returned to zero");
+    }
+
+    [Fact]
+    public async Task inline_receiver_depth_reaches_the_snapshot_too()
+    {
+        var endpoint = new StubEndpoint("inline-4186", new StubTransport()) { IsListener = true };
+        endpoint.Mode = EndpointMode.Inline;
+
+        await theRuntime.Endpoints.StartListenerAsync(endpoint, CancellationToken.None);
+        var agent = theRuntime.Endpoints.FindListeningAgent(endpoint.Uri).ShouldNotBeNull();
+        receiverOf(agent).ShouldBeOfType<InlineReceiver>();
+
+        snapshotFor(endpoint.Uri).QueueCount.ShouldBe(0);
+
+        var gate = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
+        var entered = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
+        NativeAckPingHandler.Gate = gate;
+        NativeAckPingHandler.Entered = entered;
+
+        try
+        {
+            // Inline invokes the pipeline on the caller's stack, so this cannot be awaited before asserting.
+            var inFlight = Task.Run(() => agent.EnqueueDirectlyAsync([pingEnvelope("inline")]), TestContext.Current.CancellationToken);
+
+            await entered.Task.WaitAsync(5.Seconds(), TestContext.Current.CancellationToken);
+
+            // Inline has no queue, but it does have in-flight work, and that is the number an operator wants.
+            snapshotFor(endpoint.Uri).QueueCount.ShouldBe(1);
+
+            gate.SetResult();
+            await inFlight;
+        }
+        finally
+        {
+            gate.TrySetResult();
+        }
+    }
+
+    /// <summary>
+    /// Shape 1 in the issue -- making NativeAckReceiver implement ILocalQueue -- would have been the smaller
+    /// diff and the wrong one: ListeningAgent.EnqueueDirectlyAsync type-switches on ILocalQueue *before* it
+    /// reaches the NativeAck branch that GH-4011 added, so a NativeAck receiver claiming to be a local queue
+    /// would silently take the wrong path on every DLQ replay.
+    /// </summary>
+    [Fact]
+    public void a_native_ack_receiver_reports_a_depth_without_claiming_to_be_a_local_queue()
+    {
+        var endpoint = new NativeAckStubEndpoint("na-4186-shape", new StubTransport());
+        endpoint.Mode = EndpointMode.NativeAck;
+        endpoint.Compile(theRuntime);
+
+        var receiver = new NativeAckReceiver(endpoint, theRuntime,
+            new HandlerPipeline(theRuntime, theRuntime, endpoint));
+
+        receiver.ShouldBeAssignableTo<IHasQueueDepth>();
+        receiver.ShouldNotBeAssignableTo<ILocalQueue>();
+    }
+
+    private EndpointHealthSnapshot snapshotFor(Uri uri)
+    {
+        return theRuntime.Endpoints.CollectEndpointHealth()
+            .Single(x => x.Uri == uri && x.Direction == EndpointDirection.Listening);
+    }
+
+    private static Envelope pingEnvelope(string name)
+    {
+        return new Envelope(new NativeAckPing(name)) { MessageType = typeof(NativeAckPing).ToMessageTypeName() };
+    }
+
+    private static async Task waitUntil(Func<bool> condition, string failure)
+    {
+        var deadline = DateTimeOffset.UtcNow.AddSeconds(10);
+        while (DateTimeOffset.UtcNow < deadline)
+        {
+            if (condition()) return;
+            await Task.Delay(25);
+        }
+
+        throw new TimeoutException(failure);
+    }
+
+    private static IReceiver receiverOf(IListeningAgent agent)
+    {
+        var field = typeof(ListeningAgent).GetField("_receiver", BindingFlags.NonPublic | BindingFlags.Instance)!;
+        return (IReceiver)field.GetValue(agent)!;
+    }
+}

src/Wolverine/Runtime/Partitioning/GlobalPartitionedInterceptor.cs

@@ -11,7 +11,7 @@ namespace Wolverine.Runtime.Partitioning;
 /// that match global partitioning rules and re-route them through Wolverine's
 /// message routing for proper partition assignment.
 /// </summary>
-internal class GlobalPartitionedInterceptor : IReceiver
+internal class GlobalPartitionedInterceptor : IReceiver, IHasQueueDepth
 {
     private readonly IReceiver _inner;
     private readonly IWolverineRuntime _runtime;
@@ -28,6 +28,13 @@ public GlobalPartitionedInterceptor(IReceiver inner, IWolverineRuntime runtime)
 
     public IHandlerPipeline Pipeline => _inner.Pipeline;
 
+    // GH-4186. A pass-through wrapper has to pass the depth through too. Without this the wrapped receiver's
+    // depth stopped at the wrapper and every endpoint in a global-partitioned topology reported a constant 0 --
+    // which for a Buffered or Durable receiver also meant BackPressureAgent could never fire.
+    public int QueueCount => _inner is IHasQueueDepth q ? q.QueueCount : 0;
+
+    public DateTimeOffset? LastReceivedAt => (_inner as IHasQueueDepth)?.LastReceivedAt;
+
     public async ValueTask ReceivedAsync(IListener listener, Envelope[] messages)
     {
         var passThrough = new List<Envelope>();

src/Wolverine/Runtime/WorkerQueues/IHasQueueDepth.cs

@@ -0,0 +1,34 @@
+namespace Wolverine.Runtime.WorkerQueues;
+
+/// <summary>
+/// GH-4186. Optional capability for an <see cref="Wolverine.Transports.IReceiver"/> that holds deliveries in an
+/// in-memory structure and can say how deep it currently is. <see cref="ILocalQueue"/> extends this, so
+/// <c>BufferedReceiver</c> and <c>DurableReceiver</c> satisfy it for free; the receivers that are deliberately
+/// *not* local queues -- <c>InlineReceiver</c> and <c>NativeAckReceiver</c> -- implement it directly.
+///
+/// <para>
+/// This is a separate interface rather than a widening of <see cref="ILocalQueue"/> because
+/// <c>ListeningAgent.EnqueueDirectlyAsync</c> type-switches on <see cref="ILocalQueue"/> to decide how a replayed
+/// envelope re-enters a listener, and a native-ack receiver needs its own branch there (GH-4011) precisely
+/// because it must not be enqueued into like a local queue. Reporting a depth had to be separable from being one.
+/// </para>
+/// </summary>
+public interface IHasQueueDepth
+{
+    /// <summary>
+    /// How many messages this receiver is currently holding in memory. Read by <c>ListeningAgent.QueueCount</c>,
+    /// which is what reaches <see cref="Wolverine.Configuration.EndpointHealthSnapshot.QueueCount"/> and the
+    /// back pressure checks.
+    /// </summary>
+    int QueueCount { get; }
+
+    /// <summary>
+    /// GH-4186. Approximate timestamp of the last delivery this receiver accepted, stamped on receipt rather than
+    /// inferred from a depth change. Null for receivers that do not stamp it, in which case
+    /// <c>ListeningAgent.LastQueueActivityAt</c> falls back to the <c>BackPressureAgent</c>-driven change-detection
+    /// heuristic. The modes that have no <c>BackPressureAgent</c> at all -- <c>Inline</c> and <c>NativeAck</c>,
+    /// see <see cref="Wolverine.Configuration.Endpoint.ShouldEnforceBackPressure"/> -- stamp it, because for them
+    /// nothing else ever would.
+    /// </summary>
+    DateTimeOffset? LastReceivedAt => null;
+}

src/Wolverine/Runtime/WorkerQueues/ILocalQueue.cs

@@ -9,8 +9,9 @@ public interface ILocalReceiver
     ValueTask EnqueueAsync(Envelope envelope);
 }
 
-public interface ILocalQueue : IReceiver, ILocalReceiver
+// GH-4186: QueueCount now comes from IHasQueueDepth, which the receivers that are NOT local queues
+// (InlineReceiver, NativeAckReceiver) can implement without also claiming they can be enqueued into.
+public interface ILocalQueue : IReceiver, ILocalReceiver, IHasQueueDepth
 {
-    int QueueCount { get; }
     Uri Uri { get; }
 }

src/Wolverine/Runtime/WorkerQueues/InlineReceiver.cs

@@ -7,7 +7,7 @@
 
 namespace Wolverine.Runtime.WorkerQueues;
 
-internal class InlineReceiver : IReceiver, ILatchedReceiver
+internal class InlineReceiver : IReceiver, ILatchedReceiver, IHasQueueDepth
 {
     private readonly ILogger _logger;
     private readonly Endpoint _endpoint;
@@ -34,8 +34,34 @@ public InlineReceiver(Endpoint endpoint, IWolverineRuntime runtime, IHandlerPipe
 
     public IHandlerPipeline Pipeline => _pipeline;
 
+    // GH-4186. Reachable through IHasQueueDepth. There is no queue in this mode -- the count is what is
+    // currently executing -- but that is still the number an operator wants, and it used to contribute a
+    // constant 0 to EndpointHealthSnapshot because nothing read it.
     public int QueueCount => Volatile.Read(ref _inFlightCount);
 
+    /// <summary>
+    /// GH-4186. Stamped on receipt. Inline runs no BackPressureAgent (see
+    /// <see cref="Endpoint.ShouldEnforceBackPressure"/>), and that agent is the only thing that advances
+    /// ListeningAgent's change-detection heuristic, so without this LastQueueActivityAt never moved off the
+    /// listener's construction time.
+    /// </summary>
+    public DateTimeOffset? LastReceivedAt
+    {
+        get
+        {
+            var ticks = Interlocked.Read(ref _lastReceivedTicks);
+            return ticks == 0 ? null : new DateTimeOffset(ticks, TimeSpan.Zero);
+        }
+    }
+
+    // A DateTimeOffset is too wide to write atomically, so the stamp is kept as UTC ticks.
+    private long _lastReceivedTicks;
+
+    private void stampReceipt()
+    {
+        Interlocked.Exchange(ref _lastReceivedTicks, DateTimeOffset.UtcNow.UtcTicks);
+    }
+
     public void Dispose()
     {
         // Nothing
@@ -74,6 +100,7 @@ public async ValueTask ReceivedAsync(IListener listener, Envelope[] messages)
     {
         if (messages.Length == 0) return;
 
+        stampReceipt();
         Interlocked.Add(ref _inFlightCount, messages.Length);
 
         foreach (var envelope in messages)
@@ -91,6 +118,7 @@ public async ValueTask ReceivedAsync(IListener listener, Envelope[] messages)
 
     public async ValueTask ReceivedAsync(IListener listener, Envelope envelope)
     {
+        stampReceipt();
         Interlocked.Increment(ref _inFlightCount);
 
         try

src/Wolverine/Runtime/WorkerQueues/NativeAckReceiver.cs

@@ -36,7 +36,7 @@ namespace Wolverine.Runtime.WorkerQueues;
 /// this mode exists to provide.
 /// </para>
 /// </summary>
-internal class NativeAckReceiver : IReceiver, IFaultTrackingReceiver, ILatchedReceiver
+internal class NativeAckReceiver : IReceiver, IFaultTrackingReceiver, ILatchedReceiver, IHasQueueDepth
 {
     private readonly RetryBlock<Envelope> _completeBlock;
     private readonly RetryBlock<Envelope> _deferBlock;
@@ -113,8 +113,34 @@ public NativeAckReceiver(Endpoint endpoint, IWolverineRuntime runtime, IHandlerP
 
     public IHandlerPipeline Pipeline { get; }
 
+    // GH-4186. Reachable through IHasQueueDepth rather than ILocalQueue: this receiver is deliberately not a
+    // local queue -- it cannot be enqueued into, because every delivery settles against the listener that
+    // brought it -- but its lane depth is exactly the saturation signal an operator wants from this mode, and
+    // it used to contribute a constant 0 to EndpointHealthSnapshot.
     public int QueueCount => (int)_receivingBlock.Count;
 
+    /// <summary>
+    /// GH-4186. Stamped on receipt rather than derived from a depth change, because this mode runs no
+    /// BackPressureAgent and the change-detection heuristic that agent drives is the only other writer -- so
+    /// LastQueueActivityAt sat frozen at listener construction for the whole life of the process.
+    /// </summary>
+    public DateTimeOffset? LastReceivedAt
+    {
+        get
+        {
+            var ticks = Interlocked.Read(ref _lastReceivedTicks);
+            return ticks == 0 ? null : new DateTimeOffset(ticks, TimeSpan.Zero);
+        }
+    }
+
+    // A DateTimeOffset is too wide to write atomically, so the stamp is kept as UTC ticks.
+    private long _lastReceivedTicks;
+
+    private void stampReceipt(DateTimeOffset now)
+    {
+        Interlocked.Exchange(ref _lastReceivedTicks, now.UtcTicks);
+    }
+
     /// <summary>CritterWatch#942 -- set when the receiving block faults terminally (jasperfx#506).</summary>
     public bool HasFaulted { get; private set; }
 
@@ -149,6 +175,7 @@ public async ValueTask ReceivedAsync(IListener listener, Envelope[] messages)
         }
 
         var now = DateTimeOffset.Now;
+        stampReceipt(now);
 
         // GH-4091. TWO passes, deliberately. Posting blocks once the execution block is at capacity, so a
         // single admit-then-post loop leaves every envelope after the first blocked one untracked -- with its
@@ -172,7 +199,10 @@ public async ValueTask ReceivedAsync(IListener listener, Envelope[] messages)
 
     public async ValueTask ReceivedAsync(IListener listener, Envelope envelope)
     {
-        if (await admitAsync(listener, envelope, DateTimeOffset.Now).ConfigureAwait(false) is { } entry)
+        var now = DateTimeOffset.Now;
+        stampReceipt(now);
+
+        if (await admitAsync(listener, envelope, now).ConfigureAwait(false) is { } entry)
         {
             await postAllAsync([entry]).ConfigureAwait(false);
         }