Joseph Doherty
678 lines
41 KiB
C#
678 lines
41 KiB
C#
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);
|
|
// The required observation period excludes initial certificate creation and server startup.
|
|
watch.Restart();
|
|
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 = await CaptureAsync(rig, config.CertificateManager, watch.Elapsed, cycles, clientRoot, serverRoot, ct);
|
|
// Retain the failing observation too; only the later quiescent row proves every gate passed.
|
|
await LogAsync(evidence, new { kind = "resource-observation", resources = sample });
|
|
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);
|
|
(await DescriptorCountsAsync(ct, 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 oldSessionManager = oldInternal.SessionManager;
|
|
var oldSubscriptionManager = oldInternal.SubscriptionManager;
|
|
var oldSessions = oldSessionManager.GetSessions().ToArray();
|
|
var oldSubscriptions = oldSubscriptionManager.GetSubscriptions().ToArray();
|
|
var oldManager = rig.Server.CertificateManager!;
|
|
var oldCores = CertificateCores(oldManager);
|
|
await rig.RestartAsync(ct).WaitAsync(TimeSpan.FromSeconds(20), ct);
|
|
oldSessionManager.GetSessions().ShouldBeEmpty();
|
|
oldSubscriptionManager.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, long GcCommittedBytes, long GcHeapBytes, long GcFragmentedBytes,
|
|
long PinnedObjects, long PendingFinalizers, long JitMethods, long JitIlBytes, long ActiveTimers, int Threads);
|
|
|
|
private static async Task<Sample> CaptureAsync(ServerEventRig rig, object clientManager, TimeSpan elapsed, int cycle,
|
|
string clientRoot, string serverRoot, CancellationToken ct)
|
|
{
|
|
var counts = PublicCounts(rig);
|
|
var points = rig.Server.CurrentInstance.SessionManager.GetSessions().Select(ContinuationCounts).ToArray();
|
|
var descriptors = await DescriptorCountsAsync(ct, clientRoot, serverRoot);
|
|
var retained = GC.GetTotalMemory(forceFullCollection: true);
|
|
var memory = GC.GetGCMemoryInfo();
|
|
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), memory.TotalCommittedBytes,
|
|
memory.HeapSizeBytes, memory.FragmentedBytes, memory.PinnedObjectsCount, memory.FinalizationPendingCount,
|
|
System.Runtime.JitInfo.GetCompiledMethodCount(), System.Runtime.JitInfo.GetCompiledILBytes(), Timer.ActiveCount,
|
|
process.Threads.Count);
|
|
}
|
|
|
|
public static bool DescriptorProbeSupported => OperatingSystem.IsLinux() || OperatingSystem.IsMacOS();
|
|
|
|
[Fact(SkipUnless = nameof(DescriptorProbeSupported), Skip = "Descriptor probes require Linux procfs or macOS lsof.")]
|
|
public async Task Descriptor_probe_observes_an_open_PKI_file_through_a_symlink_alias_and_its_close()
|
|
{
|
|
var ct = TestContext.Current.CancellationToken;
|
|
var temporary = Path.Combine(OperatingSystem.IsMacOS() ? "/tmp" : Path.GetTempPath(), $"otua-pki-probe-{Guid.NewGuid():N}");
|
|
var pki = Path.Combine(temporary, "pki");
|
|
var sibling = Path.Combine(temporary, "pki-other");
|
|
var alias = Path.Combine(temporary, "pki-alias");
|
|
Directory.CreateDirectory(pki);
|
|
Directory.CreateDirectory(sibling);
|
|
try
|
|
{
|
|
// On macOS the temporary root itself can use /var -> /private/var. The additional
|
|
// task-owned alias also checks links whose target contains a linked parent component.
|
|
Directory.CreateSymbolicLink(alias, pki);
|
|
await using var outside = new FileStream(Path.Combine(sibling, "control"), FileMode.Create,
|
|
FileAccess.ReadWrite, FileShare.ReadWrite | FileShare.Delete);
|
|
CanonicalPath(alias).ShouldBe(CanonicalPath(pki));
|
|
var before = (await DescriptorCountsAsync(ct, pki, alias)).Pki;
|
|
before.ShouldBe(0, "a sibling sharing the PKI path prefix is outside the root");
|
|
int opened;
|
|
await using (var held = new FileStream(Path.Combine(alias, "control"), FileMode.Create,
|
|
FileAccess.ReadWrite, FileShare.ReadWrite | FileShare.Delete))
|
|
{
|
|
opened = (await DescriptorCountsAsync(ct, pki, alias)).Pki;
|
|
opened.ShouldBeGreaterThan(0, "the positive control must observe the task-owned file opened through the alias");
|
|
}
|
|
var closed = (await DescriptorCountsAsync(ct, pki, alias)).Pki;
|
|
closed.ShouldBe(0, "closing the positive control must release its descriptor");
|
|
TestContext.Current.TestOutputHelper!.WriteLine($"PKI descriptor control: before={before}, opened={opened}, closed={closed}; siblingExcluded=true; aliasResolved=true");
|
|
}
|
|
finally
|
|
{
|
|
if (Directory.Exists(alias)) Directory.Delete(alias);
|
|
if (Directory.Exists(temporary)) Directory.Delete(temporary, recursive: true);
|
|
}
|
|
}
|
|
|
|
[Fact(SkipUnless = nameof(DescriptorProbeSupported), Skip = "The bounded probe control uses a task-owned Unix process.")]
|
|
public async Task Descriptor_probe_times_out_and_reaps_a_process_that_keeps_both_streams_open()
|
|
{
|
|
var ct = TestContext.Current.CancellationToken;
|
|
using var process = new Process
|
|
{
|
|
StartInfo = new ProcessStartInfo("/bin/sh")
|
|
{
|
|
RedirectStandardOutput = true, RedirectStandardError = true, UseShellExecute = false,
|
|
ArgumentList = { "-c", "printf control; printf control >&2; exec /bin/sleep 30" },
|
|
},
|
|
};
|
|
var elapsed = Stopwatch.StartNew();
|
|
await Should.ThrowAsync<TimeoutException>(() => ReadDescriptorProbeAsync(process, TimeSpan.FromMilliseconds(250), ct));
|
|
elapsed.Elapsed.ShouldBeLessThan(TimeSpan.FromSeconds(4), "timeout includes stream drains and bounded process cleanup");
|
|
process.HasExited.ShouldBeTrue("the task-owned timed-out process must be killed and reaped before returning");
|
|
TestContext.Current.TestOutputHelper!.WriteLine($"Descriptor timeout control: deadlineMs=250, elapsedMs={elapsed.Elapsed.TotalMilliseconds:F0}, childExited={process.HasExited}");
|
|
}
|
|
|
|
private static async Task<(int Total, int Pki)> DescriptorCountsAsync(CancellationToken ct, params string[] roots)
|
|
{
|
|
ct.ThrowIfCancellationRequested();
|
|
var canonicalRoots = roots.Select(root => CanonicalPath(root)).Distinct(StringComparer.Ordinal).ToArray();
|
|
if (OperatingSystem.IsLinux())
|
|
{
|
|
var targets = Directory.EnumerateFileSystemEntries("/proc/self/fd")
|
|
.Select(path => new FileInfo(path).LinkTarget).ToArray();
|
|
return (targets.Length, targets.Count(path => IsWithinRoots(path, canonicalRoots)));
|
|
}
|
|
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);
|
|
var text = await ReadDescriptorProbeAsync(process, TimeSpan.FromSeconds(5), ct);
|
|
process.ExitCode.ShouldBe(0);
|
|
var total = 0;
|
|
var pki = 0;
|
|
var descriptor = false;
|
|
foreach (var line in text.Split('\n'))
|
|
{
|
|
if (line.StartsWith('f'))
|
|
{
|
|
descriptor = line.Length > 1 && char.IsDigit(line[1]);
|
|
if (descriptor) total++;
|
|
}
|
|
else if (descriptor && line.StartsWith('n') && IsWithinRoots(line[1..], canonicalRoots)) pki++;
|
|
}
|
|
return (total, pki);
|
|
}
|
|
throw new PlatformNotSupportedException("This soak requires Linux procfs or macOS lsof for actual descriptor evidence.");
|
|
}
|
|
|
|
private static async Task<string> ReadDescriptorProbeAsync(Process process, TimeSpan timeout, CancellationToken ct)
|
|
{
|
|
using var deadline = CancellationTokenSource.CreateLinkedTokenSource(ct);
|
|
deadline.CancelAfter(timeout);
|
|
process.Start();
|
|
var stdout = process.StandardOutput.ReadToEndAsync(deadline.Token);
|
|
var stderr = process.StandardError.ReadToEndAsync(deadline.Token);
|
|
var exited = process.WaitForExitAsync(deadline.Token);
|
|
try
|
|
{
|
|
// Waiting for exit alone does not bound reads: both redirected pipes must also finish.
|
|
await Task.WhenAll(stdout, stderr, exited).WaitAsync(deadline.Token).ConfigureAwait(false);
|
|
return await stdout.ConfigureAwait(false);
|
|
}
|
|
catch (OperationCanceledException)
|
|
{
|
|
try { if (!process.HasExited) process.Kill(entireProcessTree: true); }
|
|
catch (InvalidOperationException) when (process.HasExited) { /* It exited between the check and kill. */ }
|
|
await process.WaitForExitAsync(CancellationToken.None).WaitAsync(TimeSpan.FromSeconds(2)).ConfigureAwait(false);
|
|
// Observe both cancelled drains; never leave a pipe read running after returning.
|
|
try { await Task.WhenAll(stdout, stderr).WaitAsync(TimeSpan.FromSeconds(2)).ConfigureAwait(false); }
|
|
catch (OperationCanceledException) { }
|
|
if (ct.IsCancellationRequested) throw;
|
|
throw new TimeoutException("Descriptor probe exceeded its process and redirected-stream deadline.");
|
|
}
|
|
}
|
|
|
|
private static bool IsWithinRoots(string? path, IReadOnlyList<string> roots)
|
|
{
|
|
if (path is null || !Path.IsPathFullyQualified(path)) return false;
|
|
// procfs marks an unlinked but still open file; its containing physical directory still counts.
|
|
const string deleted = " (deleted)";
|
|
if (path.EndsWith(deleted, StringComparison.Ordinal)) path = path[..^deleted.Length];
|
|
var canonical = CanonicalPath(path);
|
|
return roots.Any(root => canonical.Equals(root, StringComparison.Ordinal)
|
|
|| canonical.StartsWith(Path.EndsInDirectorySeparator(root) ? root : root + Path.DirectorySeparatorChar,
|
|
StringComparison.Ordinal));
|
|
}
|
|
|
|
private static string CanonicalPath(string path, int depth = 0)
|
|
{
|
|
if (depth > 32) throw new IOException("Descriptor path link resolution exceeded its bound.");
|
|
var full = Path.GetFullPath(path);
|
|
var current = Path.GetPathRoot(full)!;
|
|
foreach (var component in full[current.Length..].Split(Path.DirectorySeparatorChar, StringSplitOptions.RemoveEmptyEntries))
|
|
{
|
|
current = Path.Combine(current, component);
|
|
FileSystemInfo entry = Directory.Exists(current) ? new DirectoryInfo(current) : new FileInfo(current);
|
|
if (entry.LinkTarget is not null)
|
|
{
|
|
var target = entry.ResolveLinkTarget(returnFinalTarget: true)
|
|
?? throw new IOException("Descriptor path link resolution returned no target.");
|
|
// Resolve parent links too: an alias target can itself begin with /var or /tmp.
|
|
current = CanonicalPath(target.FullName, depth + 1);
|
|
}
|
|
}
|
|
return Path.TrimEndingDirectorySeparator(current);
|
|
}
|
|
|
|
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(); }
|
|
}
|
|
}
|
|
}
|