perf(dashboard): LRU cap on the shared live-read session's advised set

src/ZB.MOM.WW.MxGateway.Server/Dashboard/DashboardLiveDataService.cs

@@ -15,13 +15,26 @@ public sealed class DashboardLiveDataService : IDashboardLiveDataService, IAsync
 {
     private const string BackendName = "Galaxy";
     private const string ClientName = "mxgateway-dashboard";
+
+    // One browse page of tags plus headroom. Bounds the standing advise load the
+    // single dashboard worker carries — and the event churn that advise set feeds —
+    // however much of a galaxy an operator browses through in one sitting.
+    private const int MaxSubscribedTags = 256;
+
     private static readonly TimeSpan ReadTimeout = TimeSpan.FromSeconds(5);
 
     private readonly ISessionManager _sessionManager;
     private readonly IGatewayAlarmService _alarmService;
     private readonly ILogger<DashboardLiveDataService> _logger;
     private readonly SemaphoreSlim _gate = new(1, 1);
-    private readonly HashSet<string> _subscribed = new(StringComparer.OrdinalIgnoreCase);
+
+    // Least-recently-read-last advise set: the list holds every currently advised
+    // tag ordered most- to least-recently read, the dictionary indexes into it.
+    // Both are only ever touched under _gate, which already serialises all viewers.
+    private readonly Dictionary<string, LinkedListNode<SubscribedTag>> _subscribed =
+        new(StringComparer.OrdinalIgnoreCase);
+
+    private readonly LinkedList<SubscribedTag> _recency = new();
 
     private GatewaySession? _session;
     private int _serverHandle;
@@ -58,15 +71,15 @@ public sealed class DashboardLiveDataService : IDashboardLiveDataService, IAsync
             (GatewaySession session, int serverHandle) = await EnsureReadyAsync(cancellationToken)
                 .ConfigureAwait(false);
 
-            string[] toSubscribe = tagAddresses.Where(tag => !_subscribed.Contains(tag)).ToArray();
+            string[] toSubscribe = TouchAndCollectNewTags(tagAddresses, out int justReadCount);
             if (toSubscribe.Length > 0)
             {
-                await session.SubscribeBulkAsync(serverHandle, toSubscribe, cancellationToken)
+                await EvictForAsync(session, serverHandle, toSubscribe.Length, justReadCount, cancellationToken)
                     .ConfigureAwait(false);
-                foreach (string tag in toSubscribe)
-                {
-                    _subscribed.Add(tag);
-                }
+                IReadOnlyList<SubscribeResult> subscribeResults = await session
+                    .SubscribeBulkAsync(serverHandle, toSubscribe, cancellationToken)
+                    .ConfigureAwait(false);
+                TrackSubscribed(toSubscribe, subscribeResults);
             }
 
             IReadOnlyList<BulkReadResult> results = await session
@@ -107,6 +120,131 @@ public sealed class DashboardLiveDataService : IDashboardLiveDataService, IAsync
         return Task.FromResult(new DashboardAlarmQueryResult(alarms, error, _alarmService.WorkerProcessId));
     }
 
+    // Promotes every already-advised tag in this read to the front of the recency
+    // list and returns the tags that still need subscribing (distinct, in request
+    // order). `justReadCount` is how many distinct tags of this read were already
+    // advised — they now occupy the front of the list and must never be evicted to
+    // make room for the same read's new tags. Callers must hold _gate.
+    private string[] TouchAndCollectNewTags(IReadOnlyCollection<string> tagAddresses, out int justReadCount)
+    {
+        int touched = 0;
+        List<string> toSubscribe = [];
+        HashSet<string> seen = new(StringComparer.OrdinalIgnoreCase);
+
+        foreach (string tag in tagAddresses)
+        {
+            if (_subscribed.TryGetValue(tag, out LinkedListNode<SubscribedTag>? node))
+            {
+                if (!ReferenceEquals(node, _recency.First))
+                {
+                    _recency.Remove(node);
+                    _recency.AddFirst(node);
+                }
+
+                if (seen.Add(tag))
+                {
+                    touched++;
+                }
+            }
+            else if (seen.Add(tag))
+            {
+                toSubscribe.Add(tag);
+            }
+        }
+
+        justReadCount = touched;
+        return [.. toSubscribe];
+    }
+
+    // Drops least-recently-read tags off the back of the advise set until the
+    // incoming tags fit under MaxSubscribedTags, unadvising them on the worker in
+    // one batch. A failed unadvise must not fail the read: the tags are dropped
+    // from tracking regardless, and the session-invalidation path already handles
+    // gateway/worker drift. Callers must hold _gate.
+    private async Task EvictForAsync(
+        GatewaySession session,
+        int serverHandle,
+        int incomingCount,
+        int justReadCount,
+        CancellationToken cancellationToken)
+    {
+        int overflow = _subscribed.Count + incomingCount - MaxSubscribedTags;
+        int evictable = _subscribed.Count - justReadCount;
+        int evictCount = Math.Min(overflow, evictable);
+        if (evictCount <= 0)
+        {
+            return;
+        }
+
+        List<int> evictedHandles = new(evictCount);
+        for (int i = 0; i < evictCount && _recency.Last is { } oldest; i++)
+        {
+            _recency.RemoveLast();
+            _subscribed.Remove(oldest.Value.TagAddress);
+            if (oldest.Value.ItemHandle != 0)
+            {
+                evictedHandles.Add(oldest.Value.ItemHandle);
+            }
+        }
+
+        _logger.LogDebug(
+            "Dashboard advise set hit its cap of {Cap}; evicted {EvictedCount} least-recently-read tags.",
+            MaxSubscribedTags,
+            evictCount);
+
+        if (evictedHandles.Count == 0)
+        {
+            return;
+        }
+
+        try
+        {
+            await session.UnsubscribeBulkAsync(serverHandle, evictedHandles, cancellationToken)
+                .ConfigureAwait(false);
+        }
+        catch (Exception exception) when (exception is not OperationCanceledException)
+        {
+            _logger.LogDebug(
+                exception,
+                "Unadvising {EvictedCount} evicted dashboard tags failed; they stay dropped from tracking.",
+                evictedHandles.Count);
+        }
+    }
+
+    // Records the freshly advised tags as the most recently read, keeping each
+    // tag's item handle so eviction can unadvise it. Tags the worker failed to
+    // advise are still tracked (matching the pre-cap behaviour of not retrying
+    // them on every read) but carry no handle, so eviction just forgets them.
+    // Callers must hold _gate.
+    private void TrackSubscribed(IReadOnlyList<string> tagAddresses, IReadOnlyList<SubscribeResult> results)
+    {
+        Dictionary<string, int> handles = new(results.Count, StringComparer.OrdinalIgnoreCase);
+        foreach (SubscribeResult result in results)
+        {
+            if (result.WasSuccessful && !string.IsNullOrEmpty(result.TagAddress))
+            {
+                handles[result.TagAddress] = result.ItemHandle;
+            }
+        }
+
+        // Inserted back-to-front so the read's first tag ends up most recent.
+        for (int i = tagAddresses.Count - 1; i >= 0; i--)
+        {
+            string tag = tagAddresses[i];
+            handles.TryGetValue(tag, out int itemHandle);
+            _subscribed[tag] = _recency.AddFirst(new SubscribedTag(tag, itemHandle));
+        }
+    }
+
+    // Forgets the whole advise set without unadvising: every call site is one where
+    // the backing session (and with it every item handle) is already gone.
+    // Callers must hold _gate.
+    private void ClearSubscriptions()
+    {
+        _subscribed.Clear();
+        _recency.Clear();
+    }
+
     // Returns a Ready session + its Register server handle, opening a fresh
     // session when none exists or the current one is no longer usable. Callers
     // must hold _gate.
@@ -132,7 +270,7 @@ public sealed class DashboardLiveDataService : IDashboardLiveDataService, IAsync
             await CloseQuietlyAsync(existing.SessionId).ConfigureAwait(false);
         }
 
-        _subscribed.Clear();
+        ClearSubscriptions();
         _session = null;
 
         GatewaySession session = await _sessionManager.OpenSessionAsync(
@@ -178,7 +316,7 @@ public sealed class DashboardLiveDataService : IDashboardLiveDataService, IAsync
     {
         _session = null;
         _serverHandle = 0;
-        _subscribed.Clear();
+        ClearSubscriptions();
     }
 
     private async Task CloseQuietlyAsync(string sessionId)
@@ -212,4 +350,8 @@ public sealed class DashboardLiveDataService : IDashboardLiveDataService, IAsync
 
         _gate.Dispose();
     }
+
+    // One entry of the advise set. ItemHandle is the handle the worker bound for
+    // the tag, or 0 when the subscribe failed and there is nothing to unadvise.
+    private readonly record struct SubscribedTag(string TagAddress, int ItemHandle);
 }

src/ZB.MOM.WW.MxGateway.Tests/Gateway/Dashboard/DashboardLiveDataServiceTests.cs

@@ -0,0 +1,344 @@
+using System.Diagnostics.CodeAnalysis;
+using System.Runtime.CompilerServices;
+using Microsoft.Extensions.Logging.Abstractions;
+using ZB.MOM.WW.MxGateway.Contracts.Proto;
+using ZB.MOM.WW.MxGateway.Server.Dashboard;
+using ZB.MOM.WW.MxGateway.Server.Sessions;
+using ZB.MOM.WW.MxGateway.Server.Workers;
+using ZB.MOM.WW.MxGateway.Tests.TestSupport;
+
+namespace ZB.MOM.WW.MxGateway.Tests.Gateway.Dashboard;
+
+public sealed class DashboardLiveDataServiceTests
+{
+    // Mirrors DashboardLiveDataService.MaxSubscribedTags — the cap is private, so the
+    // tests drive it through the public read surface at exactly its documented size.
+    private const int MaxSubscribedTags = 256;
+
+    /// <summary>
+    /// Verifies a tag already in the advise set is not subscribed again on a later read.
+    /// </summary>
+    [Fact]
+    public async Task ReadAsync_WhenTagAlreadySubscribed_DoesNotResubscribe()
+    {
+        RecordingWorkerClient worker = new();
+        await using FakeSessionManager sessionManager = new(worker);
+        await using DashboardLiveDataService service = CreateService(sessionManager);
+
+        await service.ReadAsync(["Tank_001.PV", "Tank_002.PV"], CancellationToken.None);
+        DashboardLiveReadResult second = await service.ReadAsync(
+            ["Tank_001.PV", "Tank_002.PV"],
+            CancellationToken.None);
+
+        Assert.Null(second.Error);
+        Assert.Equal(["Tank_001.PV", "Tank_002.PV"], worker.SubscribedTags);
+        Assert.Empty(worker.UnsubscribedHandles);
+    }
+
+    /// <summary>
+    /// Verifies the advise set is capped: subscribing past the cap unadvises the
+    /// least-recently-read tag on the worker and leaves the re-read tag advised.
+    /// </summary>
+    [Fact]
+    public async Task ReadAsync_PastCap_EvictsLeastRecentlyReadTag()
+    {
+        RecordingWorkerClient worker = new();
+        await using FakeSessionManager sessionManager = new(worker);
+        await using DashboardLiveDataService service = CreateService(sessionManager);
+
+        string[] filler = CreateTagAddresses(MaxSubscribedTags);
+        await service.ReadAsync(filler, CancellationToken.None);
+
+        // Re-read the tag that is currently least recent, making it most recent: the
+        // tag read before it is now the eviction candidate.
+        await service.ReadAsync([filler[^1]], CancellationToken.None);
+        Assert.Equal(MaxSubscribedTags, worker.SubscribedTags.Count);
+
+        DashboardLiveReadResult overflow = await service.ReadAsync(
+            ["Overflow.PV"],
+            CancellationToken.None);
+
+        Assert.Null(overflow.Error);
+        Assert.Equal([worker.HandleFor(filler[^2])], worker.UnsubscribedHandles);
+        Assert.Equal("Overflow.PV", worker.SubscribedTags[^1]);
+        Assert.Equal(MaxSubscribedTags + 1, worker.SubscribedTags.Count);
+
+        // The evicted tag is no longer tracked and re-subscribes; the re-read one does not.
+        await service.ReadAsync([filler[^1], filler[^2]], CancellationToken.None);
+        Assert.Equal(filler[^2], worker.SubscribedTags[^1]);
+        Assert.Equal(MaxSubscribedTags + 2, worker.SubscribedTags.Count);
+    }
+
+    /// <summary>
+    /// Verifies a failed unadvise of an evicted tag does not fail the read, and the
+    /// evicted tag is dropped from tracking anyway.
+    /// </summary>
+    [Fact]
+    public async Task ReadAsync_WhenEvictionUnsubscribeFails_StillCompletesRead()
+    {
+        RecordingWorkerClient worker = new() { FailUnsubscribe = true };
+        await using FakeSessionManager sessionManager = new(worker);
+        await using DashboardLiveDataService service = CreateService(sessionManager);
+
+        string[] filler = CreateTagAddresses(MaxSubscribedTags);
+        await service.ReadAsync(filler, CancellationToken.None);
+
+        DashboardLiveReadResult overflow = await service.ReadAsync(
+            ["Overflow.PV"],
+            CancellationToken.None);
+
+        Assert.Null(overflow.Error);
+        Assert.Equal("Overflow.PV", Assert.Single(overflow.Values).TagAddress);
+        Assert.Equal(1, sessionManager.OpenCount);
+
+        // The evicted tag was dropped from tracking despite the failed unadvise.
+        await service.ReadAsync([filler[^1]], CancellationToken.None);
+        Assert.Equal(filler[^1], worker.SubscribedTags[^1]);
+    }
+
+    private static DashboardLiveDataService CreateService(ISessionManager sessionManager)
+    {
+        return new DashboardLiveDataService(
+            sessionManager,
+            new FakeGatewayAlarmService(),
+            NullLogger<DashboardLiveDataService>.Instance);
+    }
+
+    private static string[] CreateTagAddresses(int count)
+    {
+        string[] addresses = new string[count];
+        for (int i = 0; i < count; i++)
+        {
+            addresses[i] = $"Tank_{i:D4}.PV";
+        }
+
+        return addresses;
+    }
+
+    // Serves the dashboard service a single Ready session backed by the recording
+    // worker, so reads exercise the real GatewaySession bulk command path.
+    private sealed class FakeSessionManager(RecordingWorkerClient workerClient) : ISessionManager, IAsyncDisposable
+    {
+        private readonly List<GatewaySession> _sessions = [];
+
+        /// <summary>Gets the number of sessions the dashboard service opened.</summary>
+        public int OpenCount { get; private set; }
+
+        /// <inheritdoc />
+        public Task<GatewaySession> OpenSessionAsync(
+            SessionOpenRequest request,
+            string? clientIdentity,
+            string? ownerKeyId,
+            CancellationToken cancellationToken)
+        {
+            OpenCount++;
+            string sessionId = $"dashboard-session-{OpenCount}";
+            GatewaySession session = new(
+                sessionId,
+                "Galaxy",
+                $"mxgw-1-{sessionId}",
+                "nonce",
+                clientIdentity,
+                request.ClientSessionName,
+                request.ClientCorrelationId,
+                TimeSpan.FromSeconds(30),
+                TimeSpan.FromSeconds(5),
+                TimeSpan.FromSeconds(5),
+                DateTimeOffset.UnixEpoch);
+            session.AttachWorkerClient(workerClient);
+            session.MarkReady();
+            _sessions.Add(session);
+            return Task.FromResult(session);
+        }
+
+        /// <inheritdoc />
+        public bool TryGetSession(string sessionId, [MaybeNullWhen(false)] out GatewaySession session)
+        {
+            session = _sessions.Find(candidate => candidate.SessionId == sessionId);
+            return session is not null;
+        }
+
+        /// <inheritdoc />
+        public Task<SessionCloseResult> CloseSessionAsync(string sessionId, CancellationToken cancellationToken)
+        {
+            return Task.FromResult(new SessionCloseResult(sessionId, SessionState.Closed, AlreadyClosed: false));
+        }
+
+        /// <inheritdoc />
+        public Task<WorkerCommandReply> InvokeAsync(
+            string sessionId,
+            WorkerCommand command,
+            CancellationToken cancellationToken) =>
+            throw new NotSupportedException();
+
+        /// <inheritdoc />
+        public IAsyncEnumerable<WorkerEvent> ReadEventsAsync(string sessionId, CancellationToken cancellationToken) =>
+            throw new NotSupportedException();
+
+        /// <inheritdoc />
+        public Task<SessionCloseResult> KillWorkerAsync(
+            string sessionId,
+            string reason,
+            CancellationToken cancellationToken) =>
+            throw new NotSupportedException();
+
+        /// <inheritdoc />
+        public Task<int> CloseExpiredLeasesAsync(DateTimeOffset now, CancellationToken cancellationToken) =>
+            throw new NotSupportedException();
+
+        /// <inheritdoc />
+        public Task ShutdownAsync(CancellationToken cancellationToken) => throw new NotSupportedException();
+
+        /// <summary>Disposes every session handed to the dashboard service.</summary>
+        /// <returns>A task that represents the asynchronous operation.</returns>
+        public async ValueTask DisposeAsync()
+        {
+            foreach (GatewaySession session in _sessions)
+            {
+                await session.DisposeAsync().ConfigureAwait(false);
+            }
+        }
+    }
+
+    // Answers Register / SubscribeBulk / UnsubscribeBulk / ReadBulk with successful
+    // replies and records what the dashboard advised and unadvised.
+    private sealed class RecordingWorkerClient : IWorkerClient
+    {
+        private const int RegisteredServerHandle = 77;
+
+        private readonly Dictionary<string, int> _itemHandles = new(StringComparer.OrdinalIgnoreCase);
+        private int _nextItemHandle = 1000;
+
+        /// <inheritdoc />
+        public string SessionId => "dashboard-session-1";
+
+        /// <inheritdoc />
+        public int? ProcessId => 4242;
+
+        /// <inheritdoc />
+        public WorkerClientState State => WorkerClientState.Ready;
+
+        /// <inheritdoc />
+        public DateTimeOffset LastHeartbeatAt => DateTimeOffset.UnixEpoch;
+
+        /// <summary>Gets the tag addresses subscribed, in the order the dashboard asked for them.</summary>
+        public List<string> SubscribedTags { get; } = [];
+
+        /// <summary>Gets the item handles the dashboard unsubscribed, in order.</summary>
+        public List<int> UnsubscribedHandles { get; } = [];
+
+        /// <summary>Gets or sets a value indicating whether unsubscribe commands throw.</summary>
+        public bool FailUnsubscribe { get; set; }
+
+        /// <summary>Gets the item handle bound for a previously subscribed tag.</summary>
+        /// <param name="tagAddress">Tag address to look up.</param>
+        /// <returns>The bound item handle.</returns>
+        public int HandleFor(string tagAddress) => _itemHandles[tagAddress];
+
+        /// <inheritdoc />
+        public Task StartAsync(CancellationToken cancellationToken) => Task.CompletedTask;
+
+        /// <inheritdoc />
+        public Task<WorkerCommandReply> InvokeAsync(
+            WorkerCommand command,
+            TimeSpan timeout,
+            CancellationToken cancellationToken)
+        {
+            MxCommand mxCommand = command.Command
+                ?? throw new InvalidOperationException("The dashboard sent a command with no payload.");
+
+            MxCommandReply reply = new()
+            {
+                Kind = mxCommand.Kind,
+                ProtocolStatus = new ProtocolStatus { Code = ProtocolStatusCode.Ok },
+            };
+
+            switch (mxCommand.Kind)
+            {
+                case MxCommandKind.Register:
+                    reply.Register = new RegisterReply { ServerHandle = RegisteredServerHandle };
+                    break;
+                case MxCommandKind.SubscribeBulk:
+                    reply.SubscribeBulk = Subscribe(mxCommand.SubscribeBulk.TagAddresses);
+                    break;
+                case MxCommandKind.UnsubscribeBulk:
+                    if (FailUnsubscribe)
+                    {
+                        throw new InvalidOperationException("Simulated worker unsubscribe failure.");
+                    }
+
+                    UnsubscribedHandles.AddRange(mxCommand.UnsubscribeBulk.ItemHandles);
+                    reply.UnsubscribeBulk = new BulkSubscribeReply();
+                    break;
+                case MxCommandKind.ReadBulk:
+                    reply.ReadBulk = Read(mxCommand.ReadBulk.TagAddresses);
+                    break;
+                default:
+                    throw new NotSupportedException($"Unexpected dashboard command {mxCommand.Kind}.");
+            }
+
+            return Task.FromResult(new WorkerCommandReply { Reply = reply });
+        }
+
+        /// <inheritdoc />
+        public async IAsyncEnumerable<WorkerEvent> ReadEventsAsync(
+            [EnumeratorCancellation] CancellationToken cancellationToken)
+        {
+            await Task.CompletedTask.ConfigureAwait(false);
+            yield break;
+        }
+
+        /// <inheritdoc />
+        public Task ShutdownAsync(TimeSpan timeout, CancellationToken cancellationToken) => Task.CompletedTask;
+
+        /// <inheritdoc />
+        public void Kill(string reason)
+        {
+        }
+
+        /// <inheritdoc />
+        public ValueTask DisposeAsync() => ValueTask.CompletedTask;
+
+        private BulkSubscribeReply Subscribe(IEnumerable<string> tagAddresses)
+        {
+            BulkSubscribeReply subscribeReply = new();
+            foreach (string tagAddress in tagAddresses)
+            {
+                SubscribedTags.Add(tagAddress);
+                if (!_itemHandles.TryGetValue(tagAddress, out int itemHandle))
+                {
+                    itemHandle = _nextItemHandle++;
+                    _itemHandles[tagAddress] = itemHandle;
+                }
+
+                subscribeReply.Results.Add(new SubscribeResult
+                {
+                    ServerHandle = RegisteredServerHandle,
+                    TagAddress = tagAddress,
+                    ItemHandle = itemHandle,
+                    WasSuccessful = true,
+                });
+            }
+
+            return subscribeReply;
+        }
+
+        private BulkReadReply Read(IEnumerable<string> tagAddresses)
+        {
+            BulkReadReply readReply = new();
+            foreach (string tagAddress in tagAddresses)
+            {
+                readReply.Results.Add(new BulkReadResult
+                {
+                    ServerHandle = RegisteredServerHandle,
+                    TagAddress = tagAddress,
+                    ItemHandle = _itemHandles.TryGetValue(tagAddress, out int itemHandle) ? itemHandle : 0,
+                    WasSuccessful = true,
+                    Quality = 192,
+                });
+            }
+
+            return readReply;
+        }
+    }
+}