lmxopcua/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests/NodeManagerHistoryReadConcurrencyTests.cs

using System.Diagnostics;
using Microsoft.Extensions.Logging.Abstractions;
using Opc.Ua;
using Opc.Ua.Server;
using Shouldly;
using Xunit;
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
using HistorianRead = ZB.MOM.WW.OtOpcUa.Core.Abstractions.HistoryReadResult;
using SdkHistoryReadResult = Opc.Ua.HistoryReadResult;

namespace ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests;

/// <summary>
///     Arch-review 03/S3 — the node-manager's HistoryRead arms must serve a multi-node batch with BOUNDED
///     per-node parallelism, a per-request deadline, and a process-wide concurrent-batch limiter, rather
///     than block-bridging the gateway per node sequentially on an SDK request thread. Boots the same real
///     <see cref="OtOpcUaSdkServer"/> harness as <see cref="NodeManagerHistoryReadTests"/> and drives the
///     public <c>HistoryRead</c> with a delaying fake <see cref="IHistorianDataSource"/> that records the
///     maximum observed concurrent reads (deterministic — no raw-timing dependence for the primary
///     assertions).
/// </summary>
public sealed class NodeManagerHistoryReadConcurrencyTests : IDisposable
{
    private static CancellationToken Ct => TestContext.Current.CancellationToken;

    private readonly string _pkiRoot = Path.Combine(
        Path.GetTempPath(),
        $"otopcua-historyread-conc-{Guid.NewGuid():N}");

    /// <summary>
    ///     RED repro (03/S3): a single Raw HistoryRead naming 8 historized nodes against a 150 ms-per-read
    ///     backend must serve at least two nodes concurrently. On the current sequential <c>foreach</c> the
    ///     max observed concurrency is pinned at 1, so this fails.
    /// </summary>
    [Fact]
    public async Task Raw_batch_is_served_with_bounded_parallelism()
    {
        var (host, server) = await BootAsync();
        var nm = server.NodeManager!;
        var fake = new ConcurrencyTrackingHistorianDataSource(TimeSpan.FromMilliseconds(150));
        nm.HistorianDataSource = fake;

        var nodeIds = MaterializeHistorizedNodes(nm, count: 8);

        var details = new ReadRawModifiedDetails
        {
            StartTime = DateTime.UtcNow.AddHours(-1),
            EndTime = DateTime.UtcNow,
            NumValuesPerNode = 10,
            IsReadModified = false,
        };

        var sw = Stopwatch.StartNew();
        await Task.Run(() => InvokeHistoryRead(server, nm, details, nodeIds), Ct)
            .WaitAsync(TimeSpan.FromSeconds(15), Ct);
        sw.Stop();

        // Primary (deterministic): the batch was served with >1 node in flight at once.
        fake.MaxObservedConcurrency.ShouldBeGreaterThanOrEqualTo(2);

        await host.DisposeAsync();
    }

    /// <summary>Materialises <paramref name="count"/> historized Float variables and returns their NodeIds.</summary>
    private static NodeId[] MaterializeHistorizedNodes(OtOpcUaNodeManager nm, int count)
    {
        var ids = new NodeId[count];
        for (var i = 0; i < count; i++)
        {
            var key = $"eq/tag{i}";
            nm.EnsureVariable(key, parentFolderNodeId: null, displayName: $"Tag{i}", dataType: "Float",
                writable: false, historianTagname: $"WW.Tag{i}");
            ids[i] = nm.TryGetVariable(key)!.NodeId;
        }

        return ids;
    }

    /// <summary>Issues one HistoryRead batch for all supplied node ids in order.</summary>
    private static (IList<SdkHistoryReadResult> Results, IList<ServiceResult> Errors) InvokeHistoryRead(
        OtOpcUaSdkServer server, OtOpcUaNodeManager nm, HistoryReadDetails details, params NodeId[] nodeIds)
    {
        var context = new OperationContext(
            new RequestHeader(), secureChannelContext: null, RequestType.HistoryRead, identity: null);

        var nodesToRead = nodeIds.Select(id => new HistoryReadValueId { NodeId = id }).ToList();
        var results = Enumerable.Repeat<SdkHistoryReadResult>(null!, nodeIds.Length).ToList();
        var errors = Enumerable.Repeat<ServiceResult>(null!, nodeIds.Length).ToList();

        nm.HistoryRead(
            context,
            details,
            TimestampsToReturn.Both,
            releaseContinuationPoints: false,
            nodesToRead,
            results,
            errors);

        return (results, errors);
    }

    /// <summary>
    ///     A delaying fake historian source that tracks the maximum number of reads in flight at once
    ///     (via interlocked live-counter) so a test can prove per-node parallelism deterministically.
    /// </summary>
    private sealed class ConcurrencyTrackingHistorianDataSource(TimeSpan delay) : IHistorianDataSource
    {
        private int _current;
        private int _max;

        /// <summary>The highest number of concurrently in-flight reads observed.</summary>
        public int MaxObservedConcurrency => Volatile.Read(ref _max);

        private async Task<HistorianRead> TrackAsync(CancellationToken ct)
        {
            var now = Interlocked.Increment(ref _current);
            UpdateMax(now);
            try
            {
                await Task.Delay(delay, ct).ConfigureAwait(false);
            }
            finally
            {
                Interlocked.Decrement(ref _current);
            }

            return new HistorianRead(
                new[] { new DataValueSnapshot(1.0f, StatusCodes.Good, DateTime.UtcNow, DateTime.UtcNow) }, null);
        }

        private void UpdateMax(int observed)
        {
            int cur;
            while ((cur = Volatile.Read(ref _max)) < observed &&
                   Interlocked.CompareExchange(ref _max, observed, cur) != cur)
            {
            }
        }

        public Task<HistorianRead> ReadRawAsync(
            string fullReference, DateTime startUtc, DateTime endUtc, uint maxValuesPerNode,
            CancellationToken cancellationToken) => TrackAsync(cancellationToken);

        public Task<HistorianRead> ReadProcessedAsync(
            string fullReference, DateTime startUtc, DateTime endUtc, TimeSpan interval,
            HistoryAggregateType aggregate, CancellationToken cancellationToken) => TrackAsync(cancellationToken);

        public Task<HistorianRead> ReadAtTimeAsync(
            string fullReference, IReadOnlyList<DateTime> timestampsUtc, CancellationToken cancellationToken) =>
            TrackAsync(cancellationToken);

        public Task<HistoricalEventsResult> ReadEventsAsync(
            string? sourceName, DateTime startUtc, DateTime endUtc, int maxEvents,
            CancellationToken cancellationToken) =>
            Task.FromResult(new HistoricalEventsResult(Array.Empty<HistoricalEvent>(), null));

        public HistorianHealthSnapshot GetHealthSnapshot() => NullHistorianDataSource.Instance.GetHealthSnapshot();

        public void Dispose()
        {
        }
    }

    private async Task<(OpcUaApplicationHost Host, OtOpcUaSdkServer Server)> BootAsync()
    {
        var host = new OpcUaApplicationHost(
            new OpcUaApplicationHostOptions
            {
                ApplicationName = "OtOpcUa.HistoryReadConcTest",
                ApplicationUri = $"urn:OtOpcUa.HistoryReadConcTest:{Guid.NewGuid():N}",
                OpcUaPort = AllocateFreePort(),
                PublicHostname = "localhost",
                PkiStoreRoot = _pkiRoot,
            },
            NullLogger<OpcUaApplicationHost>.Instance);

        var server = new OtOpcUaSdkServer();
        await host.StartAsync(server, Ct);
        return (host, server);
    }

    private static int AllocateFreePort()
    {
        using var listener = new System.Net.Sockets.TcpListener(System.Net.IPAddress.Loopback, 0);
        listener.Start();
        var port = ((System.Net.IPEndPoint)listener.LocalEndpoint).Port;
        listener.Stop();
        return port;
    }

    /// <summary>Cleans up the PKI root directory.</summary>
    public void Dispose()
    {
        if (Directory.Exists(_pkiRoot))
        {
            try { Directory.Delete(_pkiRoot, recursive: true); }
            catch { /* best-effort cleanup */ }
        }
    }
}