perf(metrics): pull-model worker queue gauge (fixes last-writer-wins), Interlocked command counters

docs/Metrics.md

@@ -72,7 +72,7 @@ Observable gauges are pull-based; the `Meter` invokes the supplied callback when
 |------------|--------------|-------------|
 | `mxgateway.sessions.open` | `_openSessions` | Currently open sessions tracked by `SessionManager`. |
 | `mxgateway.workers.running` | `_workersRunning` | Worker clients in a running state. |
-| `mxgateway.events.worker_queue.depth` | `_workerEventQueueDepth` | Undelivered worker events held by `WorkerClient` — staged *and* queued (GWC-24). Incremented when the read loop stages an event, decremented when the consumer reads it, so a backlog stuck in the staging channel is visible rather than invisible. |
+| `mxgateway.events.worker_queue.depth` | `_workerEventQueueDepthSources` (summed on demand) | Undelivered worker events held by `WorkerClient` — staged *and* queued (GWC-24) — summed across every live client at collection time (GWC-30). Each client owns an interlocked counter incremented when the read loop stages an event and decremented when the consumer reads it, and registers it as a gauge source for its lifetime, so a backlog stuck in a staging channel is visible and concurrent sessions add up instead of overwriting one another. |
 | `mxgateway.events.grpc_stream_queue.depth` | `_eventStreamBacklogSources` (summed on demand) | Live backlog buffered across every active `EventStreamService` subscriber, summed from the subscribers' channel `Count` at collection time. |
 
 ## Snapshot Shape
@@ -111,7 +111,7 @@ The scalar fields mirror the counters and gauges. The four dictionaries provide
 - `EventsBySession` keys by `sessionId`; entries are removed via `RemoveSessionEvents` when a session closes so the map does not grow without bound.
 - `RetryAttemptsByArea` keys by the resilience `area` tag, e.g. `worker_startup`.
 
-`EventsReceived` is read with `Interlocked.Read(ref _eventsReceived)` because `EventReceived` increments it via `Interlocked.Increment` outside the lock to keep the event-ingestion path non-blocking.
+`EventsReceived` is read with `Interlocked.Read(ref _eventsReceived)` because `EventReceived` increments it via `Interlocked.Increment` outside the lock to keep the event-ingestion path non-blocking. `CommandsStarted`, `CommandsSucceeded`, `CommandsFailed`, and `CommandFailuresByMethod` are read the same way: the command counters run two-to-three times per gRPC call, so they are recorded with `Interlocked` and a `ConcurrentDictionary` rather than under `_syncRoot` (GWC-30). The two queue depths are pulled from their registered sources before the lock is taken, since those delegates reach into subscriber channels and worker clients.
 
 ## Recording Sites
 
@@ -146,7 +146,7 @@ _metrics.RemoveSessionEvents(session.SessionId);
 - `RecordWorkerStoppedOnce` calls `WorkerStopped(reason)` exactly once per worker, guarding against double-counting on simultaneous fault and exit signals.
 - `WorkerKilled(reason)` when the client forcibly terminates the worker.
 - `HeartbeatFailed(SessionId)` per missed heartbeat.
-- `SetWorkerEventQueueDepth(queueDepth)` when the read loop stages an event and when the consumer reads one, so the gauge tracks staged + queued events.
+- `RegisterWorkerEventQueueDepthSource(...)` once at construction, disposed in `DisposeAsync`. The client's own `_eventQueueDepth` is incremented when the read loop stages an event and decremented when the consumer reads one, so the gauge tracks staged + queued events without either hot-path step calling into `GatewayMetrics`.
 - `EventReceived(SessionId, workerEvent.Event.Family.ToString())` for each worker event.
 - `QueueOverflow("worker-events")` when the timed write into the bounded consumer channel exceeds `EventChannelFullModeTimeout`, and `QueueOverflow("worker-event-staging")` when the staging channel is full at its `2 × EventChannelCapacity` bound. The two labels distinguish a stalled consumer from one that merely drains too slowly; both fault the session with `ProtocolViolation`.
 

src/ZB.MOM.WW.MxGateway.Server/Metrics/GatewayMetrics.cs

@@ -28,7 +28,9 @@ public sealed class GatewayMetrics : IDisposable
     private readonly Histogram<double> _workerStartupLatencyHistogram;
     private readonly Histogram<double> _commandLatencyHistogram;
     private readonly Histogram<double> _eventStreamSendLatencyHistogram;
-    private readonly Dictionary<string, long> _commandFailuresByMethod = new(StringComparer.OrdinalIgnoreCase);
+    // Concurrent (not Dictionary + _syncRoot) because CommandFailed runs on every failing gRPC call:
+    // the command counters are recorded outside the lock, so their breakdown map must be too.
+    private readonly ConcurrentDictionary<string, long> _commandFailuresByMethod = new(StringComparer.OrdinalIgnoreCase);
     private readonly ConcurrentDictionary<string, long> _eventsByFamily = new(StringComparer.OrdinalIgnoreCase);
     private readonly ConcurrentDictionary<string, long> _eventsBySession = new(StringComparer.Ordinal);
     private readonly Dictionary<string, long> _retryAttemptsByArea = new(StringComparer.OrdinalIgnoreCase);
@@ -41,9 +43,16 @@ public sealed class GatewayMetrics : IDisposable
     private readonly ConcurrentDictionary<long, Func<int>> _eventStreamBacklogSources = new();
     private long _nextEventStreamBacklogSourceId;
 
+    // GWC-30: the same pull model for the worker event queue depth. It replaces a pushed scalar that
+    // every WorkerClient wrote twice per event (staged, consumed) under _syncRoot — a process-wide
+    // lock on the hottest path, and last-writer-wins across sessions, so the gauge reported one
+    // arbitrary session's backlog instead of the gateway's. Each client registers a source returning
+    // its own undelivered depth; the gauge sums them at collection time only.
+    private readonly ConcurrentDictionary<long, Func<int>> _workerEventQueueDepthSources = new();
+    private long _nextWorkerEventQueueDepthSourceId;
+
     private int _openSessions;
     private int _workersRunning;
-    private int _workerEventQueueDepth;
     private int _alarmProviderMode;
     private long _sessionsOpened;
     private long _sessionsClosed;
@@ -201,10 +210,10 @@ public sealed class GatewayMetrics : IDisposable
     /// <param name="method">Name of the command method.</param>
     public void CommandStarted(string method)
     {
-        lock (_syncRoot)
+        // GWC-30: the three command counters run two-to-three times per gRPC call, so they use
-        {
+        // Interlocked rather than _syncRoot — the same idiom as EventReceived. Nothing here needs a
-            _commandsStarted++;
+        // consistent multi-field view; GetSnapshot reads each with Interlocked.Read.
-        }
+        Interlocked.Increment(ref _commandsStarted);
 
         _commandsStartedCounter.Add(1, new KeyValuePair<string, object?>("method", method));
     }
@@ -216,10 +225,7 @@ public sealed class GatewayMetrics : IDisposable
     /// <param name="duration">Elapsed time to complete the command.</param>
     public void CommandSucceeded(string method, TimeSpan duration)
     {
-        lock (_syncRoot)
+        Interlocked.Increment(ref _commandsSucceeded);
-        {
-            _commandsSucceeded++;
-        }
 
         KeyValuePair<string, object?> methodTag = new("method", method);
         _commandsSucceededCounter.Add(1, methodTag);
@@ -234,11 +240,8 @@ public sealed class GatewayMetrics : IDisposable
     /// <param name="duration">Elapsed time before command failed.</param>
     public void CommandFailed(string method, string category, TimeSpan duration)
     {
-        lock (_syncRoot)
+        Interlocked.Increment(ref _commandsFailed);
-        {
-            _commandsFailed++;
         Increment(_commandFailuresByMethod, method);
-        }
 
         KeyValuePair<string, object?> methodTag = new("method", method);
         KeyValuePair<string, object?> categoryTag = new("category", category);
@@ -275,29 +278,24 @@ public sealed class GatewayMetrics : IDisposable
     }
 
     /// <summary>
-    /// Sets the worker event queue depth; delegates to SetWorkerEventQueueDepth.
+    ///     Registers a live depth source for the worker event queue-depth gauge and returns a handle
+    ///     that removes it when disposed. Each <c>WorkerClient</c> registers once and reports its own
+    ///     undelivered (staged + queued) event count, so the gauge is the gateway-wide sum instead of
+    ///     the last value any one session happened to push (GWC-30).
     /// </summary>
-    /// <param name="depth">Queue depth value.</param>
+    /// <param name="depth">
-    public void SetEventQueueDepth(int depth)
+    ///     Returns this worker client's current undelivered event count. Invoked only at collection
+    ///     time; must be cheap and non-blocking (a <see cref="Volatile.Read(ref int)"/> of an
+    ///     interlocked counter). Negative readings — which a racing decrement can produce — are
+    ///     clamped to zero when summed.
+    /// </param>
+    /// <returns>A handle whose disposal unregisters the source. Safe to dispose more than once.</returns>
+    public IDisposable RegisterWorkerEventQueueDepthSource(Func<int> depth)
     {
-        SetWorkerEventQueueDepth(depth);
+        ArgumentNullException.ThrowIfNull(depth);
-    }
+        long id = Interlocked.Increment(ref _nextWorkerEventQueueDepthSourceId);
-
+        _workerEventQueueDepthSources[id] = depth;
-    /// <summary>
+        return new GaugeSourceRegistration(_workerEventQueueDepthSources, id);
-    /// Sets the worker event queue depth to the given value.
-    /// </summary>
-    /// <param name="depth">Queue depth value.</param>
-    public void SetWorkerEventQueueDepth(int depth)
-    {
-        if (depth < 0)
-        {
-            throw new ArgumentOutOfRangeException(nameof(depth), depth, "Queue depth cannot be negative.");
-        }
-
-        lock (_syncRoot)
-        {
-            _workerEventQueueDepth = depth;
-        }
     }
 
     /// <summary>
@@ -318,7 +316,7 @@ public sealed class GatewayMetrics : IDisposable
         ArgumentNullException.ThrowIfNull(backlog);
         long id = Interlocked.Increment(ref _nextEventStreamBacklogSourceId);
         _eventStreamBacklogSources[id] = backlog;
-        return new EventStreamBacklogRegistration(this, id);
+        return new GaugeSourceRegistration(_eventStreamBacklogSources, id);
     }
 
     /// <summary>
@@ -460,21 +458,23 @@ public sealed class GatewayMetrics : IDisposable
     /// <returns>The current metrics snapshot.</returns>
     public GatewayMetricsSnapshot GetSnapshot()
     {
-        // Compute the live gRPC stream backlog outside _syncRoot: the sources are the subscriber
+        // Compute the live queue depths outside _syncRoot: the sources are the subscriber channels'
-        // channels' Count (their own locks) and must not run under this lock. GWC-15.
+        // Count (their own locks) and the worker clients' interlocked counters, neither of which may
+        // run under this lock. GWC-15, GWC-30.
+        int workerEventQueueDepth = GetWorkerEventQueueDepth();
         int grpcEventStreamQueueDepth = GetGrpcEventStreamQueueDepth();
         lock (_syncRoot)
         {
             return new GatewayMetricsSnapshot(
                 OpenSessions: _openSessions,
                 WorkersRunning: _workersRunning,
-                WorkerEventQueueDepth: _workerEventQueueDepth,
+                WorkerEventQueueDepth: workerEventQueueDepth,
                 GrpcEventStreamQueueDepth: grpcEventStreamQueueDepth,
                 SessionsOpened: _sessionsOpened,
                 SessionsClosed: _sessionsClosed,
-                CommandsStarted: _commandsStarted,
+                CommandsStarted: Interlocked.Read(ref _commandsStarted),
-                CommandsSucceeded: _commandsSucceeded,
+                CommandsSucceeded: Interlocked.Read(ref _commandsSucceeded),
-                CommandsFailed: _commandsFailed,
+                CommandsFailed: Interlocked.Read(ref _commandsFailed),
                 EventsReceived: Interlocked.Read(ref _eventsReceived),
                 QueueOverflows: _queueOverflows,
                 Faults: _faults,
@@ -521,22 +521,19 @@ public sealed class GatewayMetrics : IDisposable
         }
     }
 
-    private int GetWorkerEventQueueDepth()
+    // Sums the undelivered event backlog across every live worker client (GWC-30).
-    {
+    private int GetWorkerEventQueueDepth() => SumSources(_workerEventQueueDepthSources);
-        lock (_syncRoot)
-        {
-            return _workerEventQueueDepth;
-        }
-    }
 
-    // Sums the live backlog across every registered event-stream subscriber. Runs at collection
+    // Sums the live backlog across every registered event-stream subscriber.
-    // time (ObservableGauge scrape) or when GetSnapshot projects the value — never on the
+    private int GetGrpcEventStreamQueueDepth() => SumSources(_eventStreamBacklogSources);
-    // per-event path. Enumerating ConcurrentDictionary.Values never throws on concurrent
+
+    // Runs at collection time (ObservableGauge scrape) or when GetSnapshot projects the value —
+    // never on a per-event path. Enumerating ConcurrentDictionary.Values never throws on concurrent
     // register/unregister; a source removed mid-enumeration simply drops from this sample.
-    private int GetGrpcEventStreamQueueDepth()
+    private static int SumSources(ConcurrentDictionary<long, Func<int>> sources)
     {
         int total = 0;
-        foreach (Func<int> source in _eventStreamBacklogSources.Values)
+        foreach (Func<int> source in sources.Values)
         {
             int value = source();
             if (value > 0)
@@ -548,11 +545,6 @@ public sealed class GatewayMetrics : IDisposable
         return total;
     }
 
-    private void UnregisterEventStreamBacklogSource(long id)
-    {
-        _eventStreamBacklogSources.TryRemove(id, out _);
-    }
-
     private int GetAlarmProviderMode()
     {
         lock (_syncRoot)
@@ -572,9 +564,10 @@ public sealed class GatewayMetrics : IDisposable
         values.AddOrUpdate(key, 1, static (_, currentValue) => currentValue + 1);
     }
 
-    // Handle returned by RegisterEventStreamBacklogSource. Disposal (once) removes the source
+    // Handle returned by the pull-model gauge registrations. Disposal (once) removes the source from
-    // from the gauge's live sum. Idempotent so a double dispose from a stream teardown is safe.
+    // that gauge's live sum. Idempotent so a double dispose from a stream or worker-client teardown
-    private sealed class EventStreamBacklogRegistration(GatewayMetrics metrics, long id) : IDisposable
+    // is safe, and shared by both gauges so the two registrations cannot drift apart.
+    private sealed class GaugeSourceRegistration(ConcurrentDictionary<long, Func<int>> sources, long id) : IDisposable
     {
         private int _disposed;
 
@@ -582,7 +575,7 @@ public sealed class GatewayMetrics : IDisposable
         {
             if (Interlocked.Exchange(ref _disposed, 1) == 0)
             {
-                metrics.UnregisterEventStreamBacklogSource(id);
+                sources.TryRemove(id, out _);
             }
         }
     }

src/ZB.MOM.WW.MxGateway.Server/Workers/WorkerClient.cs

@@ -40,6 +40,12 @@ public sealed class WorkerClient : IWorkerClient
     private readonly ConcurrentDictionary<string, PendingCommand> _pendingCommands = new(StringComparer.Ordinal);
     private readonly SemaphoreSlim _pendingCommandSlots;
     private readonly CancellationTokenSource _stopCts = new();
+
+    // GWC-30: this client's contribution to the gateway-wide worker queue-depth gauge. Registered
+    // once here and read only when the gauge is scraped, so staging and consuming an event cost an
+    // Interlocked on _eventQueueDepth and nothing else — previously each of those two hot-path steps
+    // called into GatewayMetrics and took its process-wide lock. Null when metrics are disabled.
+    private readonly IDisposable? _eventQueueDepthRegistration;
     // Touched only by WriteLoopAsync — the single consumer of _outboundEnvelopes — so it needs no
     // interlocking. See WriteLoopAsync for why the stamp happens there rather than at construction.
     private ulong _nextSequence;
@@ -111,6 +117,8 @@ public sealed class WorkerClient : IWorkerClient
                 FullMode = BoundedChannelFullMode.Wait,
                 AllowSynchronousContinuations = false,
             });
+        _eventQueueDepthRegistration = _metrics?.RegisterWorkerEventQueueDepthSource(
+            () => Volatile.Read(ref _eventQueueDepth));
         _lastHeartbeatAt = _timeProvider.GetUtcNow();
     }
 
@@ -299,8 +307,8 @@ public sealed class WorkerClient : IWorkerClient
     {
         await foreach (WorkerEvent workerEvent in _events.Reader.ReadAllAsync(cancellationToken).ConfigureAwait(false))
         {
-            int queueDepth = Math.Max(0, Interlocked.Decrement(ref _eventQueueDepth));
+            // No metrics call on the hot path: the gauge pulls _eventQueueDepth when scraped (GWC-30).
-            _metrics?.SetWorkerEventQueueDepth(queueDepth);
+            Interlocked.Decrement(ref _eventQueueDepth);
             yield return workerEvent;
         }
     }
@@ -371,6 +379,9 @@ public sealed class WorkerClient : IWorkerClient
         }
 
         _disposed = true;
+        // Drop out of the worker queue-depth gauge before teardown: whatever this client still holds
+        // is about to be discarded, and the sum must not keep counting a client that is going away.
+        _eventQueueDepthRegistration?.Dispose();
         KillOwnedProcess("Dispose");
         _stopCts.Cancel();
         _outboundEnvelopes.Writer.TryComplete();
@@ -598,8 +609,7 @@ public sealed class WorkerClient : IWorkerClient
         {
             // Counted here rather than at the _events write so the single gauge reports total
             // undelivered events (staged + queued). ReadEventsCoreAsync decrements on consumer read.
-            int queueDepth = Interlocked.Increment(ref _eventQueueDepth);
+            Interlocked.Increment(ref _eventQueueDepth);
-            _metrics?.SetWorkerEventQueueDepth(queueDepth);
             return;
         }
 

src/ZB.MOM.WW.MxGateway.Tests/Metrics/GatewayMetricsTests.cs

@@ -19,7 +19,10 @@ public sealed class GatewayMetricsTests
         metrics.CommandFailed("WriteSecured", "AuthorizationFailed", TimeSpan.FromMilliseconds(12));
         metrics.EventReceived("session-1", "OnDataChange");
         metrics.EventReceived("session-1", "OnDataChange");
-        metrics.SetWorkerEventQueueDepth(7);
+        // GWC-30: the worker queue-depth gauge sums one live source per worker client, so the two
+        // registrations below stand in for two concurrent sessions holding 3 and 4 events.
+        using IDisposable workerDepthSourceA = metrics.RegisterWorkerEventQueueDepthSource(static () => 3);
+        using IDisposable workerDepthSourceB = metrics.RegisterWorkerEventQueueDepthSource(static () => 4);
         // GWC-15: the gRPC stream queue-depth gauge sums live backlog sources at collection time
         // rather than tracking a pushed running total. Register a source reporting 3.
         using IDisposable backlogSource = metrics.RegisterEventStreamBacklogSource(static () => 3);
@@ -54,16 +57,111 @@ public sealed class GatewayMetricsTests
         Assert.Equal(2, snapshot.EventsBySession["session-1"]);
     }
 
-    /// <summary>Verifies that negative queue depth is rejected.</summary>
+    /// <summary>
+    ///     GWC-30: the worker queue-depth gauge sums every registered source rather than holding a
+    ///     single pushed scalar, so concurrent sessions add up instead of overwriting one another,
+    ///     and a disposed registration (a worker client going away) drops out of the sum. Disposal
+    ///     is idempotent because a client's dispose path can run twice.
+    /// </summary>
     [Fact]
-    public void SetEventQueueDepth_RejectsNegativeDepth()
+    public void WorkerEventQueueDepthSources_SumAcrossRegistrationsAndDropOnDispose()
     {
         using GatewayMetrics metrics = new();
 
-        ArgumentOutOfRangeException exception = Assert.Throws<ArgumentOutOfRangeException>(
+        IDisposable firstSource = metrics.RegisterWorkerEventQueueDepthSource(static () => 3);
-            () => metrics.SetWorkerEventQueueDepth(-1));
+        using IDisposable secondSource = metrics.RegisterWorkerEventQueueDepthSource(static () => 4);
 
-        Assert.Equal("depth", exception.ParamName);
+        Assert.Equal(7, metrics.GetSnapshot().WorkerEventQueueDepth);
+
+        firstSource.Dispose();
+        Assert.Equal(4, metrics.GetSnapshot().WorkerEventQueueDepth);
+
+        firstSource.Dispose();
+        Assert.Equal(4, metrics.GetSnapshot().WorkerEventQueueDepth);
+    }
+
+    /// <summary>
+    ///     A depth source reads a lock-free counter that a racing decrement can momentarily push
+    ///     below zero, so the sum clamps each reading instead of rejecting it — the pull model has
+    ///     no caller to throw back at.
+    /// </summary>
+    [Fact]
+    public void WorkerEventQueueDepthSources_ClampNegativeReadingsToZero()
+    {
+        using GatewayMetrics metrics = new();
+
+        using IDisposable negativeSource = metrics.RegisterWorkerEventQueueDepthSource(static () => -5);
+        using IDisposable positiveSource = metrics.RegisterWorkerEventQueueDepthSource(static () => 4);
+
+        Assert.Equal(4, metrics.GetSnapshot().WorkerEventQueueDepth);
+    }
+
+    /// <summary>
+    ///     Verifies the exported gauge keeps the name <c>mxgateway.events.worker_queue.depth</c> and
+    ///     reports the summed sources, so the pull-model rework is invisible to exporters.
+    /// </summary>
+    [Fact]
+    public void WorkerEventQueueDepthGauge_ReportsSummedSources()
+    {
+        using GatewayMetrics metrics = new();
+        using MeterListener listener = new();
+
+        int? capturedDepth = null;
+
+        listener.InstrumentPublished = (instrument, meterListener) =>
+        {
+            if (ReferenceEquals(instrument.Meter, metrics.Meter)
+                && instrument.Name == "mxgateway.events.worker_queue.depth")
+            {
+                meterListener.EnableMeasurementEvents(instrument);
+            }
+        };
+        listener.SetMeasurementEventCallback<int>(
+            (instrument, measurement, _, _) =>
+            {
+                if (ReferenceEquals(instrument.Meter, metrics.Meter)
+                    && instrument.Name == "mxgateway.events.worker_queue.depth")
+                {
+                    capturedDepth = measurement;
+                }
+            });
+        listener.Start();
+
+        using IDisposable firstSource = metrics.RegisterWorkerEventQueueDepthSource(static () => 5);
+        using IDisposable secondSource = metrics.RegisterWorkerEventQueueDepthSource(static () => 6);
+        listener.RecordObservableInstruments();
+
+        Assert.Equal(11, capturedDepth);
+    }
+
+    /// <summary>
+    ///     The command counters are incremented with <see cref="Interlocked"/> rather than under the
+    ///     process-wide metrics lock, so this asserts no increment is lost when every gRPC thread
+    ///     records at once.
+    /// </summary>
+    [Fact]
+    public void CommandCounters_CountEveryConcurrentInvocation()
+    {
+        const int workers = 8;
+        const int perWorker = 500;
+        using GatewayMetrics metrics = new();
+
+        Parallel.For(0, workers, _ =>
+        {
+            for (int index = 0; index < perWorker; index++)
+            {
+                metrics.CommandStarted("Register");
+                metrics.CommandSucceeded("Register", TimeSpan.FromMilliseconds(1));
+                metrics.CommandFailed("WriteSecured", "AuthorizationFailed", TimeSpan.FromMilliseconds(1));
+            }
+        });
+
+        GatewayMetricsSnapshot snapshot = metrics.GetSnapshot();
+
+        Assert.Equal(workers * perWorker, snapshot.CommandsStarted);
+        Assert.Equal(workers * perWorker, snapshot.CommandsSucceeded);
+        Assert.Equal(workers * perWorker, snapshot.CommandsFailed);
+        Assert.Equal(workers * perWorker, snapshot.CommandFailuresByMethod["WriteSecured"]);
     }
 
     /// <summary>