From 73ee3ef20dd4cea42c9863eb3863c9e593c42144 Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Tue, 10 Nov 2026 12:02:25 -0500 Subject: [PATCH] test(opcua): add opt-in lifecycle and resource soak harness --- .../Sdk2LifecycleSoakTests.cs | 538 ++++++++++++++++++ 1 file changed, 538 insertions(+) create mode 100644 tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/Sdk2LifecycleSoakTests.cs diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/Sdk2LifecycleSoakTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/Sdk2LifecycleSoakTests.cs new file mode 100644 index 00000000..0b6cbd26 --- /dev/null +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/Sdk2LifecycleSoakTests.cs @@ -0,0 +1,538 @@ +using System.Collections; +using System.Collections.Concurrent; +using System.Diagnostics; +using System.Globalization; +using System.Net; +using System.Net.Sockets; +using System.Reflection; +using System.Text; +using System.Text.Json; +using Opc.Ua; +using Opc.Ua.Client; +using Opc.Ua.Configuration; +using Opc.Ua.Server; +using Shouldly; +using Xunit; +using ZB.MOM.WW.OtOpcUa.Commons.OpcUa; +using ZB.MOM.WW.OtOpcUa.Configuration.Enums; +using ZB.MOM.WW.OtOpcUa.Core.Abstractions; +using ZB.MOM.WW.OtOpcUa.Core.Authorization; +using ClientSession = Opc.Ua.Client.ISession; +using Subscription = Opc.Ua.Client.Subscription; +using MonitoredItem = Opc.Ua.Client.MonitoredItem; +using ServerSession = Opc.Ua.Server.ISession; + +namespace ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests; + +/// +/// Opt-in Task 12 lifecycle evidence. Diagnostic mode is deliberately separate from the two-hour gate. +/// Run only this test in an otherwise idle test process; RSS/FD counts describe the whole process. +/// No key, certificate, store path, token, or OPC UA payload is written to the JSONL evidence. +/// +[Trait("Category", "E2E")] +public sealed class Sdk2LifecycleSoakTests +{ + private const string Scalar = ServerEventRig.RawDeviceFolder + "/SoakScalar"; + private const string ArrayTag = ServerEventRig.RawDeviceFolder + "/SoakArray"; + private const string UnsScalar = "SoakEquipment/Scalar"; + private static readonly DateTime Epoch = new(2026, 10, 1, 0, 0, 0, DateTimeKind.Utc); + public static bool Enabled => Environment.GetEnvironmentVariable("OTOPCUA_SDK2_SOAK") == "1"; + + [Fact(SkipUnless = nameof(Enabled), Skip = "Opt in with OTOPCUA_SDK2_SOAK=1 and an explicit diagnostic/full mode.")] + public async Task Repeated_lifecycles_return_resources_to_a_steady_baseline() + { + var mode = Environment.GetEnvironmentVariable("OTOPCUA_SDK2_SOAK_MODE"); + (mode is "diagnostic" or "full").ShouldBeTrue("explicit diagnostic or full mode is required"); + var seconds = int.Parse(Environment.GetEnvironmentVariable("OTOPCUA_SDK2_SOAK_SECONDS") + ?? (mode == "full" ? "7200" : "90"), CultureInfo.InvariantCulture); + if (mode == "full") seconds.ShouldBeGreaterThanOrEqualTo(7200); + else seconds.ShouldBeInRange(60, 120); + var output = Environment.GetEnvironmentVariable("OTOPCUA_SDK2_SOAK_OUTPUT") + ?? throw new InvalidOperationException("An absolute, fresh OTOPCUA_SDK2_SOAK_OUTPUT JSONL path is required."); + Path.IsPathFullyQualified(output).ShouldBeTrue(); + Directory.CreateDirectory(Path.GetDirectoryName(output)!); + await using var evidence = new StreamWriter(new FileStream(output, FileMode.CreateNew, FileAccess.Write, FileShare.Read)); + using var run = CancellationTokenSource.CreateLinkedTokenSource(TestContext.Current.CancellationToken); + run.CancelAfter(TimeSpan.FromSeconds(seconds + 120)); + var ct = run.Token; + var watch = Stopwatch.StartNew(); + var clientRoot = Path.Combine(Path.GetTempPath(), $"otua-soak-client-{Guid.NewGuid():N}"); + ApplicationInstance? application = null; + ServerEventRig? rig = null; + string? serverRoot = null; + var cycles = 0; + var restarts = 0; + Sample? baseline = null; + var retained = new Queue(); + try + { + var telemetry = DefaultTelemetry.Create(_ => { }); + var security = TestClientSecurity.Build(clientRoot); + security.ApplicationCertificate!.SubjectName = "CN=SDK2LifecycleSoak"; + var config = new ApplicationConfiguration(telemetry) + { + ApplicationName = "SDK2LifecycleSoak", ApplicationUri = $"urn:SDK2LifecycleSoak:{Guid.NewGuid():N}", + ApplicationType = ApplicationType.Client, SecurityConfiguration = security, + ClientConfiguration = new ClientConfiguration { DefaultSessionTimeout = 15_000 }, + TransportQuotas = new TransportQuotas { OperationTimeout = 5000 }, + }; + application = new ApplicationInstance(telemetry) { ApplicationConfiguration = config }; + await config.ValidateAsync(ApplicationType.Client, ct); + await application.CheckApplicationInstanceCertificatesAsync(true, null, ct); + rig = await ServerEventRig.StartAsync("SDK2LifecycleSoak", encrypted: true, ct); + serverRoot = RequiredField(rig, "_pkiRoot"); + Configure(rig); + await LogAsync(evidence, new + { + kind = "start", utc = DateTimeOffset.UtcNow, processId = Environment.ProcessId, + mode, requestedSeconds = seconds, minimumFullSeconds = 7200, + build = Environment.GetEnvironmentVariable("OTOPCUA_SOAK_BUILD_SHA"), + sdk = typeof(OperationContext).Assembly.GetCustomAttribute()?.InformationalVersion, + diagnosticInstrumentation = "SDK2.0.1 CP lists and certificate-validation cached stores; read-only", + limits = new { retainedManagedGrowthBytes = 64L << 20, rssGrowthBytes = 128L << 20, fdGrowth = 32 }, + coverage = "encrypted sessions; scalar/Int16[] Raw+UNS; subscription removal; browse abandonment; history release/cross-session/close; surgical changes/rebuild; revocation; paused TCP read/recovery; same-PKI restart", + }); + while (watch.Elapsed < TimeSpan.FromSeconds(seconds)) + { + var cycle = Stopwatch.StartNew(); + await ExerciseAsync(rig, config, ++cycles, evidence, ct); + await EventuallyAsync(() => PublicCounts(rig) == (0, 0, 0), ct); + var sample = Capture(rig, config.CertificateManager, watch.Elapsed, cycles, clientRoot, serverRoot); + sample.Sessions.ShouldBe(0); sample.Subscriptions.ShouldBe(0); sample.Items.ShouldBe(0); + sample.BrowsePoints.ShouldBe(0); sample.HistoryPoints.ShouldBe(0); + sample.LiveDataItems.ShouldBe(0); sample.LiveEventItems.ShouldBe(0); sample.MonitoredNodes.ShouldBe(0); + sample.PkiDescriptors.ShouldBe(0, "quiescent certificate file descriptors must be closed"); + if (cycles >= 3) + { + baseline ??= sample; + sample.CachedCertificateStores.ShouldBe(baseline.CachedCertificateStores); + sample.Variables.ShouldBe(baseline.Variables); sample.Folders.ShouldBe(baseline.Folders); + sample.Conditions.ShouldBe(baseline.Conditions); + sample.Descriptors.ShouldBeLessThanOrEqualTo(baseline.Descriptors + 32); + sample.ManagedBytes.ShouldBeLessThanOrEqualTo(baseline.ManagedBytes + (64L << 20)); + sample.RssBytes.ShouldBeLessThanOrEqualTo(baseline.RssBytes + (128L << 20)); + retained.Enqueue(sample); + if (retained.Count > 24) retained.Dequeue(); + if (retained.Count == 24) + { + var window = retained.ToArray(); + var strictlyIncreasing = window.Zip(window.Skip(1)).All(pair => pair.Second.ManagedBytes > pair.First.ManagedBytes); + (strictlyIncreasing && window[^1].ManagedBytes - window[0].ManagedBytes > (16L << 20)).ShouldBeFalse( + "retained managed memory must not grow monotonically across 24 quiescent cycles"); + } + } + await LogAsync(evidence, new { kind = "quiescent", cycleSeconds = cycle.Elapsed.TotalSeconds, resources = sample }); + if (cycles % 4 == 0) + { + var restart = Stopwatch.StartNew(); + await RestartWithActiveResourcesAsync(rig, config, ct).WaitAsync(TimeSpan.FromSeconds(25), ct); + Configure(rig); + restarts++; + await LogAsync(evidence, new { kind = "restart", cycles, restarts, seconds = restart.Elapsed.TotalSeconds }); + } + } + cycles.ShouldBeGreaterThanOrEqualTo(4, "diagnostic mode must exercise at least one actual restart"); + restarts.ShouldBeGreaterThan(0); + watch.Elapsed.TotalSeconds.ShouldBeGreaterThanOrEqualTo(seconds); + var finalClientCores = CertificateCores(config.CertificateManager); + var finalServerCores = CertificateCores(rig.Server.CertificateManager!); + await rig.DisposeAsync().AsTask().WaitAsync(TimeSpan.FromSeconds(20), ct); + rig = null; + await application.DisposeAsync().AsTask().WaitAsync(TimeSpan.FromSeconds(20), ct); + application = null; + finalClientCores.Sum(CachedStores).ShouldBe(0); + finalServerCores.Sum(CachedStores).ShouldBe(0); + DescriptorCounts(clientRoot, serverRoot).Pki.ShouldBe(0); + await LogAsync(evidence, new { kind = "complete", utc = DateTimeOffset.UtcNow, mode, cycles, restarts, seconds = watch.Elapsed.TotalSeconds, + satisfiesTwoHourDuration = mode == "full" && watch.Elapsed.TotalSeconds >= 7200 }); + } + catch (Exception error) + { + await LogAsync(evidence, new { kind = "failed", cycles, restarts, seconds = watch.Elapsed.TotalSeconds, + exceptionType = error.GetType().FullName }); + throw; + } + finally + { + if (rig is not null) await rig.DisposeAsync().AsTask().WaitAsync(TimeSpan.FromSeconds(20), TestContext.Current.CancellationToken); + if (application is not null) await application.DisposeAsync().AsTask().WaitAsync(TimeSpan.FromSeconds(20), TestContext.Current.CancellationToken); + if (Directory.Exists(clientRoot)) Directory.Delete(clientRoot, true); + if (serverRoot is not null) Directory.Exists(serverRoot).ShouldBeFalse("the task-owned server PKI must be released"); + } + } + + private static void Configure(ServerEventRig rig) + { + rig.MaterialiseNative(); + rig.Nm.HistorianDataSource = new EventHistoryPagingOverTheWireTests.SqlLikeEventSource(Enumerable.Range(0, 40) + .Select(i => new HistoricalEvent($"soak-{i}", "source", Epoch.AddSeconds(i), Epoch, "soak", 500)).ToList()); + rig.Nm.NodeWriteGateway = new AcceptWrites(); + MaterialiseValues(rig); + rig.Enforce(true, [ServerEventRig.Key()], Grant()); + } + + private static ZB.MOM.WW.OtOpcUa.Configuration.Entities.NodeAcl Grant() => ServerEventRig.ClusterGrant( + ServerEventRig.KeyPrincipal, "soak-grant", NodePermissions.Browse | NodePermissions.Read | NodePermissions.Subscribe + | NodePermissions.WriteOperate | NodePermissions.HistoryRead); + + private static void MaterialiseValues(ServerEventRig rig) + { + rig.Sink.EnsureFolder(ServerEventRig.RawDeviceFolder, null, "Device", AddressSpaceRealm.Raw); + rig.Sink.EnsureVariable(Scalar, ServerEventRig.RawDeviceFolder, "Scalar", "Int32", true, AddressSpaceRealm.Raw); + rig.Sink.EnsureVariable(ArrayTag, ServerEventRig.RawDeviceFolder, "Array", "Int16", true, AddressSpaceRealm.Raw, + isArray: true, arrayLength: 4); + rig.Sink.EnsureFolder("SoakEquipment", null, "Equipment", AddressSpaceRealm.Uns); + rig.Sink.EnsureVariable(UnsScalar, "SoakEquipment", "Scalar", "Int32", true, AddressSpaceRealm.Uns); + for (var i = 0; i < 16; i++) rig.Sink.EnsureVariable($"{ServerEventRig.RawDeviceFolder}/Browse{i}", + ServerEventRig.RawDeviceFolder, $"Browse{i}", "Int32", false, AddressSpaceRealm.Raw); + rig.Sink.WriteValue(Scalar, 0, OpcUaStatus.Good, DateTime.UtcNow, AddressSpaceRealm.Raw); + rig.Sink.WriteValue(ArrayTag, new short[] { 0, 1, 2, 3 }, OpcUaStatus.Good, DateTime.UtcNow, AddressSpaceRealm.Raw); + rig.Sink.WriteValue(UnsScalar, 0, OpcUaStatus.Good, DateTime.UtcNow, AddressSpaceRealm.Uns); + } + + private static async Task ExerciseAsync(ServerEventRig rig, ApplicationConfiguration config, int cycle, + StreamWriter evidence, CancellationToken ct) + { + await using var proxy = new PauseProxy(new Uri(rig.EndpointUrl).Port); + using var owner = await ConnectAsync(rig, config, proxy.Port, ct); + using var other = await ConnectAsync(rig, config, null, ct); + var ownedSessions = rig.Server.CurrentInstance.SessionManager.GetSessions().ToArray(); + ownedSessions.Length.ShouldBe(2); + var scalar = Raw(owner, Scalar); + var array = Raw(owner, ArrayTag); + var unsNs = owner.NamespaceUris.GetIndex(V3NodeIds.UnsNamespaceUri); + unsNs.ShouldBeGreaterThan(0); + var uns = new NodeId(UnsScalar, (ushort)unsNs); + owner.NamespaceUris.GetString(scalar.NamespaceIndex).ShouldBe(V3NodeIds.RawNamespaceUri); + owner.NamespaceUris.GetString(uns.NamespaceIndex).ShouldBe(V3NodeIds.UnsNamespaceUri); + using var subscription = new Subscription(owner.DefaultSubscription) { PublishingInterval = 100, KeepAliveCount = 10 }; + owner.AddSubscription(subscription); + try + { + await subscription.CreateAsync(ct); + var seen = new ConcurrentDictionary(); + foreach (var id in new[] { scalar, array, uns }) + { + var item = new MonitoredItem(subscription.DefaultItem) { StartNodeId = id, AttributeId = Attributes.Value, + SamplingInterval = 50, QueueSize = 10 }; + item.Notification += (sender, _) => + { + foreach (var value in sender.DequeueValues()) if (StatusCode.IsGood(value.StatusCode)) seen[id] = value.WrappedValue; + }; + subscription.AddItem(item); + } + await subscription.ApplyChangesAsync(ct); + subscription.MonitoredItems.ShouldAllBe(item => ServiceResult.IsGood(item.Status.Error)); + PublicCounts(rig).ShouldBe((2, 1, 3)); + var values = new short[] { (short)(cycle % 30000), 2, 3, 4 }; + var writes = await owner.WriteAsync(null, new[] + { + new WriteValue { NodeId = scalar, AttributeId = Attributes.Value, Value = new DataValue(Variant.From(cycle)) }, + new WriteValue { NodeId = array, AttributeId = Attributes.Value, Value = new DataValue(SdkTestValues.FromClr(values)) }, + new WriteValue { NodeId = uns, AttributeId = Attributes.Value, Value = new DataValue(Variant.From(cycle)) }, + }, ct); + writes.Results.ToList().ShouldAllBe(status => status == StatusCodes.Good); + (await owner.ReadValueAsync(scalar, ct)).WrappedValue.GetInt32().ShouldBe(cycle); + (await owner.ReadValueAsync(array, ct)).ClrValue().ShouldBeOfType().ShouldBe(values); + (await owner.ReadValueAsync(uns, ct)).WrappedValue.GetInt32().ShouldBe(cycle); + await EventuallyAsync(() => seen.TryGetValue(scalar, out var value) && value.GetInt32() == cycle + && seen.ContainsKey(array) && seen.TryGetValue(uns, out var mirror) && mirror.GetInt32() == cycle, ct); + + // One explicitly released and one deliberately abandoned browse point. + var browse = await BrowseAsync(owner, ct); + browse.IsEmpty.ShouldBeFalse(); + (await owner.BrowseNextAsync(null, true, [browse], ct)).Results[0].StatusCode.ShouldBe(StatusCodes.Good); + (await BrowseAsync(owner, ct)).IsEmpty.ShouldBeFalse(); + // Production history points: cross-session denial, single-use resume, explicit release and close cleanup. + var first = (await PageAsync(owner, default, false, ct)).ContinuationPoint; + first.IsEmpty.ShouldBeFalse(); + (await PageAsync(other, first, false, ct)).StatusCode.ShouldBe(StatusCodes.BadContinuationPointInvalid); + var next = (await PageAsync(owner, first, false, ct)).ContinuationPoint; + next.IsEmpty.ShouldBeFalse(); + (await PageAsync(owner, next, true, ct)).StatusCode.ShouldBe(StatusCodes.Good); + for (var i = 0; i < 6; i++) (await PageAsync(owner, default, false, ct)).ContinuationPoint.IsEmpty.ShouldBeFalse(); + var held = ownedSessions.Select(ContinuationCounts).ToArray(); + held.Sum(count => count.Browse).ShouldBeGreaterThan(0); + held.Sum(count => count.History).ShouldBeGreaterThan(0); + var active = PublicCounts(rig); + await LogAsync(evidence, new { kind = "active", cycle, sessions = active.Sessions, + subscriptions = active.Subscriptions, items = active.Items, browsePoints = held.Sum(x => x.Browse), historyPoints = held.Sum(x => x.History) }); + + rig.Nm.UpdateTagAttributes(Scalar, false, null, "Int32", false, null, AddressSpaceRealm.Raw).ShouldBeTrue(); + rig.Nm.UpdateTagAttributes(Scalar, true, null, "Int32", false, null, AddressSpaceRealm.Raw).ShouldBeTrue(); + rig.Nm.RemoveVariableNode(ArrayTag, AddressSpaceRealm.Raw).ShouldBeTrue(); + rig.Sink.EnsureVariable(ArrayTag, ServerEventRig.RawDeviceFolder, "Array", "Int16", true, AddressSpaceRealm.Raw, + isArray: true, arrayLength: 4); + rig.Sink.WriteValue(ArrayTag, values, OpcUaStatus.Good, DateTime.UtcNow, AddressSpaceRealm.Raw); + var authority = rig.Nm.DataPlaneAuthority.ShouldBeOfType(); + authority.Apply(ServerEventRig.Cluster, $"revoked-{cycle}", [], [Grant()], + new Dictionary>()); + var denied = await owner.ReadAsync(null, 0, TimestampsToReturn.Neither, + [new ReadValueId { NodeId = scalar, AttributeId = Attributes.Value }], ct); + denied.Results[0].StatusCode.ShouldBe(StatusCodes.BadUserAccessDenied); + rig.Enforce(true, [ServerEventRig.Key()], Grant()); + + proxy.Pause(); + var stalled = Stopwatch.StartNew(); + string stalledStatus; + try + { + using var timeout = CancellationTokenSource.CreateLinkedTokenSource(ct); + timeout.CancelAfter(TimeSpan.FromSeconds(1)); + await owner.ReadValueAsync(scalar, timeout.Token).WaitAsync(TimeSpan.FromSeconds(5), ct); + throw new InvalidOperationException("The deliberately paused network unexpectedly completed the read."); + } + catch (OperationCanceledException) when (!ct.IsCancellationRequested) { stalledStatus = "OperationCanceled"; } + // Exact-tag ChannelAsyncOperation maps a canceled completion waiter to BadRequestInterrupted. + catch (ServiceResultException error) when (error.StatusCode == StatusCodes.BadTimeout + || error.StatusCode == StatusCodes.BadRequestCancelledByClient || error.StatusCode == StatusCodes.BadRequestTimeout + || error.StatusCode == StatusCodes.BadRequestInterrupted) { stalledStatus = error.StatusCode.ToString(); } + finally { proxy.Resume(); } + (await owner.ReadValueAsync(scalar, ct)).StatusCode.ShouldBe(StatusCodes.Good); + await LogAsync(evidence, new { kind = "network-stall-recovered", cycle, status = stalledStatus, seconds = stalled.Elapsed.TotalSeconds }); + await subscription.DeleteAsync(true, ct); + await owner.RemoveSubscriptionAsync(subscription, ct); + rig.Sink.RebuildAddressSpace(); + Configure(rig); + } + finally + { + proxy.Resume(); + await owner.CloseAsync(ct).WaitAsync(TimeSpan.FromSeconds(10), ct); + await other.CloseAsync(ct).WaitAsync(TimeSpan.FromSeconds(10), ct); + foreach (var closed in ownedSessions) + { + ContinuationCounts(closed).ShouldBe((0, 0)); + } + } + } + + private static async Task RestartWithActiveResourcesAsync(ServerEventRig rig, ApplicationConfiguration config, + CancellationToken ct) + { + using var client = await ConnectAsync(rig, config, null, ct); + using var subscription = new Subscription(client.DefaultSubscription) { PublishingInterval = 100 }; + client.AddSubscription(subscription); + await subscription.CreateAsync(ct); + subscription.AddItem(new MonitoredItem(subscription.DefaultItem) + { + StartNodeId = Raw(client, Scalar), AttributeId = Attributes.Value, SamplingInterval = 50, + }); + await subscription.ApplyChangesAsync(ct); + PublicCounts(rig).ShouldBe((1, 1, 1)); + (await BrowseAsync(client, ct)).IsEmpty.ShouldBeFalse(); + (await PageAsync(client, default, false, ct)).ContinuationPoint.IsEmpty.ShouldBeFalse(); + var oldInternal = rig.Server.CurrentInstance; + var oldSessions = oldInternal.SessionManager.GetSessions().ToArray(); + var oldSubscriptions = oldInternal.SubscriptionManager.GetSubscriptions().ToArray(); + var oldManager = rig.Server.CertificateManager!; + var oldCores = CertificateCores(oldManager); + await rig.RestartAsync(ct).WaitAsync(TimeSpan.FromSeconds(20), ct); + oldInternal.SessionManager.GetSessions().ShouldBeEmpty(); + oldInternal.SubscriptionManager.GetSubscriptions().ShouldBeEmpty(); + oldSubscriptions.ShouldAllBe(item => item.MonitoredItemCount == 0); + foreach (var session in oldSessions) ContinuationCounts(session).ShouldBe((0, 0)); + RequiredField(oldManager, "m_disposed").ShouldBeTrue(); + oldCores.Sum(CachedStores).ShouldBe(0, "stopped server validators must release every cached store"); + PublicCounts(rig).ShouldBe((0, 0, 0)); + } + + private static async Task ConnectAsync(ServerEventRig rig, ApplicationConfiguration config, + int? proxyPort, CancellationToken ct) + { + var telemetry = DefaultTelemetry.Create(_ => { }); + var endpoint = await CoreClientUtils.SelectEndpointAsync(config, rig.EndpointUrl, true, telemetry, ct) + ?? throw new InvalidOperationException("No encrypted endpoint."); + endpoint.SecurityMode.ShouldBe(MessageSecurityMode.SignAndEncrypt); + if (proxyPort is { } port) endpoint.EndpointUrl = $"opc.tcp://127.0.0.1:{port}/OtOpcUa"; + return await new DefaultSessionFactory(telemetry).CreateAsync(config, + new ConfiguredEndpoint(null, endpoint, EndpointConfiguration.Create(config)), false, false, + "SDK2LifecycleSoak", 15_000, new UserIdentity("apikey", Encoding.UTF8.GetBytes(ServerEventRig.KeyPassword)), default, ct); + } + + private static NodeId Raw(ClientSession session, string id) + { + var ns = session.NamespaceUris.GetIndex(V3NodeIds.RawNamespaceUri); + ns.ShouldBeGreaterThan(0); + return new NodeId(id, (ushort)ns); + } + + private static async Task BrowseAsync(ClientSession session, CancellationToken ct) => + (await session.BrowseAsync(null, null, 1, [new BrowseDescription + { + NodeId = Raw(session, ServerEventRig.RawDeviceFolder), BrowseDirection = BrowseDirection.Forward, + ReferenceTypeId = ReferenceTypeIds.HierarchicalReferences, IncludeSubtypes = true, + ResultMask = (uint)BrowseResultMask.All, + }], ct)).Results[0].ContinuationPoint; + + private static async Task PageAsync(ClientSession session, ByteString point, + bool release, CancellationToken ct) => (await session.HistoryReadAsync(null, new ExtensionObject(new ReadEventDetails + { + StartTime = Epoch.AddDays(-1), EndTime = Epoch.AddDays(1), NumValuesPerNode = 4, + Filter = new EventFilter { SelectClauses = [new SimpleAttributeOperand(ObjectTypeIds.BaseEventType, + QualifiedName.From(BrowseNames.EventId))] }, + }), TimestampsToReturn.Both, release, + [new HistoryReadValueId { NodeId = Raw(session, ServerEventRig.RawDeviceFolder), ContinuationPoint = point }], ct)).Results[0]; + + private static (int Sessions, int Subscriptions, int Items) PublicCounts(ServerEventRig rig) + { + var subscriptions = rig.Server.CurrentInstance.SubscriptionManager.GetSubscriptions(); + return (rig.Server.CurrentInstance.SessionManager.GetSessions().Count, subscriptions.Count, + subscriptions.Sum(subscription => subscription.MonitoredItemCount)); + } + + // Exact-tag diagnostic instrumentation only. Fail if fields/types change; never replace or mutate a store. + private static (int Browse, int History) ContinuationCounts(ServerSession session) + { + var points = session.ContinuationPoints; + using (RequiredField(points, "m_lock").EnterScope()) + return (NullableCollectionCount(points, "m_browse"), NullableCollectionCount(points, "m_history")); + } + + private static object[] CertificateCores(object manager) + { + using (RequiredField(manager, "m_certificatesLock").EnterScope()) + { + var cores = new List(); + foreach (var field in new[] { "m_peerCore", "m_userCore", "m_httpsCore" }) + if (Field(manager, field).GetValue(manager) is { } core) cores.Add(core); + cores.AddRange(RequiredField(manager, "m_customCores").Values.Cast()); + return cores.Distinct(System.Collections.Generic.ReferenceEqualityComparer.Instance).ToArray(); + } + } + + private static int CachedStores(object core) => RequiredField(core, "m_stores").Count; + private static int NullableCollectionCount(object owner, string name) => Field(owner, name).GetValue(owner) switch + { + null => 0, + ICollection items => items.Count, + _ => throw new InvalidOperationException($"SDK diagnostic field {name} no longer has the expected collection type."), + }; + private static FieldInfo Field(object owner, string name) => owner.GetType().GetField(name, + BindingFlags.Instance | BindingFlags.NonPublic) ?? throw new InvalidOperationException($"Missing SDK diagnostic field {name}."); + private static T RequiredField(object owner, string name) => Field(owner, name).GetValue(owner) is T value ? value + : throw new InvalidOperationException($"SDK diagnostic field {name} no longer has the expected type {typeof(T).Name}."); + + private sealed record Sample(int Cycle, double Seconds, int Sessions, int Subscriptions, int Items, + int BrowsePoints, int HistoryPoints, int LiveDataItems, int LiveEventItems, int MonitoredNodes, + int Variables, int Folders, int Conditions, int CachedCertificateStores, + int Descriptors, int PkiDescriptors, long ManagedBytes, long RssBytes, long AllocatedBytes, + int Gen0, int Gen1, int Gen2); + + private static Sample Capture(ServerEventRig rig, object clientManager, TimeSpan elapsed, int cycle, + string clientRoot, string serverRoot) + { + var counts = PublicCounts(rig); + var points = rig.Server.CurrentInstance.SessionManager.GetSessions().Select(ContinuationCounts).ToArray(); + var descriptors = DescriptorCounts(clientRoot, serverRoot); + var retained = GC.GetTotalMemory(forceFullCollection: true); + using var process = Process.GetCurrentProcess(); + process.Refresh(); + return new Sample(cycle, elapsed.TotalSeconds, counts.Sessions, counts.Subscriptions, counts.Items, + points.Sum(x => x.Browse), points.Sum(x => x.History), rig.Nm.LiveItemCount, rig.Nm.LiveEventItemCount, + rig.Nm.MonitoredNodeCount, rig.Nm.VariableCount, rig.Nm.FolderCount, rig.Nm.AlarmConditionCount, + CertificateCores(rig.Server.CertificateManager!).Sum(CachedStores) + CertificateCores(clientManager).Sum(CachedStores), + descriptors.Total, descriptors.Pki, retained, process.WorkingSet64, GC.GetTotalAllocatedBytes(), + GC.CollectionCount(0), GC.CollectionCount(1), GC.CollectionCount(2)); + } + + private static (int Total, int Pki) DescriptorCounts(params string[] roots) + { + if (OperatingSystem.IsLinux()) + { + var targets = Directory.EnumerateFileSystemEntries("/proc/self/fd").Select(path => new FileInfo(path).LinkTarget).ToArray(); + return (targets.Length, targets.Count(path => path is not null && roots.Any(root => path.StartsWith(root, StringComparison.Ordinal)))); + } + if (OperatingSystem.IsMacOS()) + { + using var process = new Process { StartInfo = new ProcessStartInfo("/usr/sbin/lsof") + { RedirectStandardOutput = true, RedirectStandardError = true, UseShellExecute = false } }; + foreach (var arg in new[] { "-a", "-p", Environment.ProcessId.ToString(CultureInfo.InvariantCulture), "-Fnf" }) + process.StartInfo.ArgumentList.Add(arg); + process.Start(); + var text = process.StandardOutput.ReadToEnd(); + _ = process.StandardError.ReadToEnd(); + if (!process.WaitForExit(5000)) { process.Kill(); throw new TimeoutException("Descriptor probe stalled."); } + process.ExitCode.ShouldBe(0); + var lines = text.Split('\n'); + return (lines.Count(line => line.Length > 1 && line[0] == 'f' && char.IsDigit(line[1])), + lines.Count(line => line.StartsWith('n') && roots.Any(root => line.AsSpan(1).StartsWith(root, StringComparison.Ordinal)))); + } + throw new PlatformNotSupportedException("This soak requires Linux procfs or macOS lsof for actual descriptor evidence."); + } + + private static async Task EventuallyAsync(Func condition, CancellationToken ct) + { + var deadline = Stopwatch.StartNew(); + while (!condition() && deadline.Elapsed < TimeSpan.FromSeconds(10)) await Task.Delay(25, ct); + condition().ShouldBeTrue("resource or notification condition did not settle within ten seconds"); + } + private static async Task LogAsync(StreamWriter output, object value) + { + await output.WriteLineAsync(JsonSerializer.Serialize(value)); + await output.FlushAsync(); + } + private sealed class AcceptWrites : IOpcUaNodeWriteGateway + { + public Task WriteAsync(string id, object? value, AddressSpaceRealm realm, CancellationToken ct) => + Task.FromResult(new NodeWriteOutcome(true, null)); + } + + /// Task-owned TCP relay. Pause blocks forwarding in both directions without closing either socket. + private sealed class PauseProxy : IAsyncDisposable + { + private readonly TcpListener _listener = new(IPAddress.Loopback, 0); + private readonly CancellationTokenSource _stop = new(); + private readonly ConcurrentBag _sockets = []; + private readonly ConcurrentBag _pumps = []; + private readonly Task _accept; + private TaskCompletionSource? _pause; + internal int Port => ((IPEndPoint)_listener.LocalEndpoint).Port; + internal PauseProxy(int targetPort) + { + _listener.Start(); + _accept = AcceptAsync(targetPort); + } + internal void Pause() => Volatile.Write(ref _pause, new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously)); + internal void Resume() => Interlocked.Exchange(ref _pause, null)?.TrySetResult(); + private async Task AcceptAsync(int port) + { + try + { + while (!_stop.IsCancellationRequested) + { + var incoming = await _listener.AcceptTcpClientAsync(_stop.Token); + _sockets.Add(incoming); + var outgoing = new TcpClient(); _sockets.Add(outgoing); + await outgoing.ConnectAsync(IPAddress.Loopback, port, _stop.Token); + _pumps.Add(PumpAsync(incoming.GetStream(), outgoing.GetStream())); + _pumps.Add(PumpAsync(outgoing.GetStream(), incoming.GetStream())); + } + } + catch (OperationCanceledException) when (_stop.IsCancellationRequested) { } + } + private async Task PumpAsync(NetworkStream source, NetworkStream destination) + { + var buffer = new byte[16 * 1024]; + try + { + while (true) + { + var length = await source.ReadAsync(buffer, _stop.Token); + if (length == 0) return; + if (Volatile.Read(ref _pause) is { } paused) await paused.Task.WaitAsync(_stop.Token); + await destination.WriteAsync(buffer.AsMemory(0, length), _stop.Token); + } + } + catch (OperationCanceledException) when (_stop.IsCancellationRequested) { } + catch (IOException) when (_stop.IsCancellationRequested) { } + } + public async ValueTask DisposeAsync() + { + Resume(); _stop.Cancel(); _listener.Stop(); + foreach (var socket in _sockets) socket.Dispose(); + try { await _accept; await Task.WhenAll(_pumps); } + finally { _stop.Dispose(); } + } + } +}