test(opcua): add opt-in lifecycle and resource soak harness
This commit is contained in:
Joseph Doherty
+538
@@ -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;
|
||||
|
||||
/// <summary>
|
||||
/// 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.
|
||||
/// </summary>
|
||||
[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<Sample>();
|
||||
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<string>(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<AssemblyInformationalVersionAttribute>()?.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<NodeId, Variant>();
|
||||
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<short[]>().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<DataPlaneAuthorityState>();
|
||||
authority.Apply(ServerEventRig.Cluster, $"revoked-{cycle}", [], [Grant()],
|
||||
new Dictionary<AclScopeRef, IReadOnlyList<NodeAclPath>>());
|
||||
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<bool>(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<ClientSession> 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<ByteString> 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<Opc.Ua.HistoryReadResult> 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<System.Threading.Lock>(points, "m_lock").EnterScope())
|
||||
return (NullableCollectionCount(points, "m_browse"), NullableCollectionCount(points, "m_history"));
|
||||
}
|
||||
|
||||
private static object[] CertificateCores(object manager)
|
||||
{
|
||||
using (RequiredField<System.Threading.Lock>(manager, "m_certificatesLock").EnterScope())
|
||||
{
|
||||
var cores = new List<object>();
|
||||
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<IDictionary>(manager, "m_customCores").Values.Cast<object>());
|
||||
return cores.Distinct(System.Collections.Generic.ReferenceEqualityComparer.Instance).ToArray();
|
||||
}
|
||||
}
|
||||
|
||||
private static int CachedStores(object core) => RequiredField<IDictionary>(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<T>(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<bool> 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<NodeWriteOutcome> WriteAsync(string id, object? value, AddressSpaceRealm realm, CancellationToken ct) =>
|
||||
Task.FromResult(new NodeWriteOutcome(true, null));
|
||||
}
|
||||
|
||||
/// <summary>Task-owned TCP relay. Pause blocks forwarding in both directions without closing either socket.</summary>
|
||||
private sealed class PauseProxy : IAsyncDisposable
|
||||
{
|
||||
private readonly TcpListener _listener = new(IPAddress.Loopback, 0);
|
||||
private readonly CancellationTokenSource _stop = new();
|
||||
private readonly ConcurrentBag<TcpClient> _sockets = [];
|
||||
private readonly ConcurrentBag<Task> _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(); }
|
||||
}
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user