diff --git a/scripts/validation/opcua-lifecycle-memory/sdk1-comparator/BaselineMetrics.cs b/scripts/validation/opcua-lifecycle-memory/sdk1-comparator/BaselineMetrics.cs new file mode 100644 index 000000000..5937eebf1 --- /dev/null +++ b/scripts/validation/opcua-lifecycle-memory/sdk1-comparator/BaselineMetrics.cs @@ -0,0 +1,145 @@ +using System.Globalization; +using System.Collections; +using System.Diagnostics; +using System.Reflection; +using Opc.Ua; +using Opc.Ua.Server; +using Shouldly; +internal static class BaselineMetrics +{ + private static FieldInfo Field(object owner, string name) + { + for (var type = owner.GetType(); type is not null; type = type.BaseType) + if (type.GetField(name, BindingFlags.Instance | BindingFlags.NonPublic | BindingFlags.DeclaredOnly) is { } member) return member; + throw new InvalidOperationException("Missing exact-.182 diagnostic field: " + name); + } + private static T Required(object owner, string name) => Field(owner, name).GetValue(owner) is T value ? value + : throw new InvalidOperationException("Unexpected exact-.182 diagnostic field type: " + name); + internal static (int Browse, int History) Points(ISession session) + { + using (Required(session, "m_lock").EnterScope()) + return (Count(session, "m_browseContinuationPoints"), Count(session, "m_historyContinuationPoints")); + } + private static int Count(object owner, string name) => Field(owner, name).GetValue(owner) switch + { null => 0, ICollection values => values.Count, _ => throw new InvalidOperationException("Unexpected continuation collection type") }; + internal static int CachedStoreReferences(CertificateValidator validator) + { + var gate = Required(validator, "m_semaphore"); + if (!gate.Wait(TimeSpan.FromSeconds(5))) throw new TimeoutException("Original validator diagnostic lock did not settle"); + try + { + return new[] { "m_trustedCertificateStore", "m_issuerCertificateStore", "m_rejectedCertificateStore" } + .Select(name => Field(validator, name).GetValue(validator)) + .Count(value => value switch + { + null => false, + CertificateStoreIdentifier store => Field(store, "m_store").GetValue(store) switch + { null => false, ICertificateStore => true, _ => throw new InvalidOperationException("Original cached store type changed") }, + _ => throw new InvalidOperationException("Original validator store identifier type changed") + }); + } + finally { gate.Release(); } + } + internal static int Counter(object manager, string name) + { + var property = manager.GetType().GetProperty(name, BindingFlags.Instance | BindingFlags.Public | BindingFlags.NonPublic) + ?? throw new InvalidOperationException("Missing original node-manager diagnostic property: " + name); + return property.PropertyType == typeof(int) && property.GetValue(manager) is int value ? value + : throw new InvalidOperationException("Unexpected original node-manager diagnostic property type: " + name); + } + internal static async Task<(int Total, int Pki)> DescriptorsAsync(CancellationToken ct, params string[] roots) + { + ct.ThrowIfCancellationRequested(); + if (OperatingSystem.IsWindows()) return await WindowsDescriptors.CountAsync(ct, roots); + 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 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 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)); + } + + internal 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); + } + +} diff --git a/scripts/validation/opcua-lifecycle-memory/sdk1-comparator/BaselineRig.cs b/scripts/validation/opcua-lifecycle-memory/sdk1-comparator/BaselineRig.cs new file mode 100644 index 000000000..7508152f3 --- /dev/null +++ b/scripts/validation/opcua-lifecycle-memory/sdk1-comparator/BaselineRig.cs @@ -0,0 +1,107 @@ +using System.Net; +using System.Net.Sockets; +using System.Reflection; +using Microsoft.Extensions.Logging.Abstractions; +using Opc.Ua; +using ZB.MOM.WW.OtOpcUa.Commons.OpcUa; +using ZB.MOM.WW.OtOpcUa.Configuration.Entities; +using ZB.MOM.WW.OtOpcUa.Configuration.Enums; +using ZB.MOM.WW.OtOpcUa.Core.Authorization; +using ZB.MOM.WW.OtOpcUa.OpcUaServer; +using ZB.MOM.WW.OtOpcUa.OpcUaServer.Security; + +internal sealed class BaselineRig : IAsyncDisposable +{ + internal const string Cluster = "SITE-A", RawDeviceFolder = "Plant/Modbus/dev1", RawAlarmPath = "Plant/Modbus/dev1/temp_hi"; + internal const string EqA = "EqA", EqB = "EqB", KeyPrincipal = "event-probe", KeyId = "k-server-events"; + internal static string KeyPassword => Environment.GetEnvironmentVariable("OTOPCUA_BASELINE_CANONICAL_KEY") + ?? throw new InvalidOperationException("An ephemeral fixture credential is required"); + private readonly OpcUaApplicationHostOptions _options; + private OpcUaApplicationHost _host = null!; + private bool _disposed; + internal OtOpcUaSdkServer Server { get; private set; } = null!; + internal OtOpcUaNodeManager Nm => Server.NodeManager!; + internal SdkAddressSpaceSink Sink { get; private set; } = null!; + internal string EndpointUrl => $"opc.tcp://127.0.0.1:{_options.OpcUaPort}/OtOpcUa"; + internal string PkiRoot { get; } + internal CertificateValidator Validator + { + get + { + var applicationField = typeof(OpcUaApplicationHost).GetField("_application", BindingFlags.NonPublic | BindingFlags.Instance) + ?? throw new InvalidOperationException("Original host application field is missing"); + return applicationField.GetValue(_host) is Opc.Ua.Configuration.ApplicationInstance application + && application.ApplicationConfiguration.CertificateValidator is CertificateValidator validator ? validator + : throw new InvalidOperationException("Original host application field type changed"); + } + } + internal BaselineRig(string directory) + { + PkiRoot = Path.Combine(directory, "server-pki"); + using var listener = new TcpListener(IPAddress.Loopback, 0); + listener.Start(); + _options = new OpcUaApplicationHostOptions + { + ApplicationName = "OtOpcUa.SDK2LifecycleSoak", ApplicationUri = $"urn:OtOpcUa.SDK2LifecycleSoak.{Guid.NewGuid():N}", + OpcUaPort = ((IPEndPoint)listener.LocalEndpoint).Port, PublicHostname = "127.0.0.1", PkiStoreRoot = PkiRoot, + EnabledSecurityProfiles = [OpcUaSecurityProfile.None, OpcUaSecurityProfile.Basic256Sha256SignAndEncrypt], + AutoAcceptUntrustedClientCertificates = true, + }; + } + internal async Task BootAsync(CancellationToken ct) + { + Server = new OtOpcUaSdkServer(); + _host = new OpcUaApplicationHost(_options, NullLogger.Instance, new Auth()); + await _host.StartAsync(Server, ct); + Canonical.Ledger.Record("host-start"); + Sink = new SdkAddressSpaceSink(Nm); + } + internal async Task RestartAsync(CancellationToken ct) + { + await _host.DisposeAsync(); + Canonical.Ledger.Record("host-stop"); + await BootAsync(ct); + } + internal DataPlaneAuthorityState Enforce(bool enabled, ApiKey[] keys, params NodeAcl[] grants) + { + var state = new DataPlaneAuthorityState(); + state.Apply(Cluster, "rev-1", keys, grants, new Dictionary> + { + [new AclScopeRef(NodeAclScopeKind.Equipment, "eq-" + EqA)] = [new NodeAclPath([NodeAclRealms.Uns, EqA])], + [new AclScopeRef(NodeAclScopeKind.Equipment, "eq-" + EqB)] = [new NodeAclPath([NodeAclRealms.Uns, EqB])] + }); + Nm.DataPlaneAuthority = state; + Nm.Authorizer = new DataPlaneAuthorizer(state, new TriePermissionEvaluator(state.Tries), enabled, NullLogger.Instance, TimeProvider.System); + Canonical.Ledger.Record("authority-restore"); + return state; + } + internal static ApiKey Key() => new() { ApiKeyId = KeyId, Name = KeyId, SecretHash = "hash", Principal = KeyPrincipal, + Enabled = true, CreatedUtc = DateTime.UtcNow.AddDays(-1) }; + internal static NodeAcl ClusterGrant(string principal, string id, NodePermissions permissions) => new() + { NodeAclId = id, ClusterId = Cluster, Principal = principal, ScopeKind = NodeAclScopeKind.Cluster, PermissionFlags = permissions }; + internal void MaterialiseNative() + { + Sink.EnsureFolder(RawDeviceFolder, null, "dev1", AddressSpaceRealm.Raw); + Sink.MaterialiseAlarmCondition(RawAlarmPath, RawDeviceFolder, "temp_hi", "OffNormalAlarm", 700, AddressSpaceRealm.Raw, isNative: true); + Sink.EnsureFolder(EqA, null, "A", AddressSpaceRealm.Uns); + Sink.EnsureFolder(EqB, null, "B", AddressSpaceRealm.Uns); + Sink.WireAlarmNotifiers(RawAlarmPath, AddressSpaceRealm.Raw, [EqA, EqB], AddressSpaceRealm.Uns); + } + public async ValueTask DisposeAsync() + { + if (_disposed) return; + _disposed = true; + try + { + if (_host is not null) { await _host.DisposeAsync(); Canonical.Ledger.Record("host-stop"); } + } + finally { if (Directory.Exists(PkiRoot)) Directory.Delete(PkiRoot, true); } + } + private sealed class Auth : IOpcUaUserAuthenticator + { + public Task AuthenticateUserNameAsync(string username, string password, CancellationToken ct) => + Task.FromResult(username == "apikey" && password.StartsWith($"otua_{KeyId}_", StringComparison.Ordinal) + ? OpcUaUserAuthResult.AllowKey(KeyPrincipal, [OpcUaDataPlaneRoles.AlarmAck, OpcUaDataPlaneRoles.WriteOperate], KeyId) + : OpcUaUserAuthResult.Deny("Invalid fixture key")); + } +} diff --git a/scripts/validation/opcua-lifecycle-memory/sdk1-comparator/CanonicalOperations.cs b/scripts/validation/opcua-lifecycle-memory/sdk1-comparator/CanonicalOperations.cs new file mode 100644 index 000000000..ab34907ea --- /dev/null +++ b/scripts/validation/opcua-lifecycle-memory/sdk1-comparator/CanonicalOperations.cs @@ -0,0 +1,244 @@ +using System.Collections.Concurrent; +using System.Diagnostics; +using System.Text; +using Opc.Ua; +using Opc.Ua.Client; +using Opc.Ua.Configuration; +using Shouldly; +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; +internal static partial class Canonical +{ + internal static (int Browse, int History) ClosedListObservation; + private static void Configure(BaselineRig rig) + { + Ledger.Record("configure"); + rig.MaterialiseNative(); + rig.Nm.HistorianDataSource = new 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, [BaselineRig.Key()], Grant()); + } + + private static ZB.MOM.WW.OtOpcUa.Configuration.Entities.NodeAcl Grant() => BaselineRig.ClusterGrant( + BaselineRig.KeyPrincipal, "soak-grant", NodePermissions.Browse | NodePermissions.Read | NodePermissions.Subscribe + | NodePermissions.WriteOperate | NodePermissions.HistoryRead); + + private static void MaterialiseValues(BaselineRig rig) + { + rig.Sink.EnsureFolder(BaselineRig.RawDeviceFolder, null, "Device", AddressSpaceRealm.Raw); + rig.Sink.EnsureVariable(Scalar, BaselineRig.RawDeviceFolder, "Scalar", "Int32", true, AddressSpaceRealm.Raw); + rig.Sink.EnsureVariable(ArrayTag, BaselineRig.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($"{BaselineRig.RawDeviceFolder}/Browse{i}", + BaselineRig.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(BaselineRig 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 Ledger.Call("subscription-create", () => 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 Ledger.Call("subscription-apply", () => 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 Ledger.Call("write", () => owner.WriteAsync(null, new[] + { + new WriteValue { NodeId = scalar, AttributeId = Attributes.Value, Value = new DataValue(new Variant(cycle)) }, + new WriteValue { NodeId = array, AttributeId = Attributes.Value, Value = new DataValue(new Variant(values)) }, + new WriteValue { NodeId = uns, AttributeId = Attributes.Value, Value = new DataValue(new Variant(cycle)) }, + }, ct)); + writes.Results.ToList().ShouldAllBe(status => status == StatusCodes.Good); + (await Ledger.Call("read-value", () => owner.ReadValueAsync(scalar, ct))).Value.ShouldBeOfType().ShouldBe(cycle); + (await Ledger.Call("read-value", () => owner.ReadValueAsync(array, ct))).Value.ShouldBeOfType().ShouldBe(values); + (await Ledger.Call("read-value", () => owner.ReadValueAsync(uns, ct))).Value.ShouldBeOfType().ShouldBe(cycle); + await EventuallyAsync(() => seen.TryGetValue(scalar, out var value) && ScalarValue(value) == cycle + && seen.ContainsKey(array) && seen.TryGetValue(uns, out var mirror) && ScalarValue(mirror) == cycle, ct); + + // One explicitly released and one deliberately abandoned browse point. + var browse = await BrowseAsync(owner, ct); + RequirePoint(browse); + (await Ledger.Call("browse-next", () => owner.BrowseNextAsync(null, true, [browse], ct))).Results[0].StatusCode.ShouldBe(StatusCodes.Good); + RequirePoint(await BrowseAsync(owner, ct)); + // Production history points: cross-session denial, single-use resume, explicit release and close cleanup. + var first = (await PageAsync(owner, default, false, ct)).ContinuationPoint; + RequirePoint(first); + (await PageAsync(other, first, false, ct)).StatusCode.ShouldBe(StatusCodes.BadContinuationPointInvalid); + var next = (await PageAsync(owner, first, false, ct)).ContinuationPoint; + RequirePoint(next); + (await PageAsync(owner, next, true, ct)).StatusCode.ShouldBe(StatusCodes.Good); + for (var i = 0; i < 6; i++) RequirePoint((await PageAsync(owner, default, false, ct)).ContinuationPoint); + 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(); + Ledger.Record("writable-off"); + rig.Nm.UpdateTagAttributes(Scalar, true, null, "Int32", false, null, AddressSpaceRealm.Raw).ShouldBeTrue(); + Ledger.Record("writable-on"); + rig.Nm.RemoveVariableNode(ArrayTag, AddressSpaceRealm.Raw).ShouldBeTrue(); + Ledger.Record("variable-remove"); + rig.Sink.EnsureVariable(ArrayTag, BaselineRig.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(); + Ledger.Record("key-revoke"); + authority.Apply(BaselineRig.Cluster, $"revoked-{cycle}", [], [Grant()], + new Dictionary>()); + var denied = await Ledger.Call("read-service", () => owner.ReadAsync(null, 0, TimestampsToReturn.Neither, + [new ReadValueId { NodeId = scalar, AttributeId = Attributes.Value }], ct)); + denied.Results[0].StatusCode.ShouldBe(StatusCodes.BadUserAccessDenied); + rig.Enforce(true, [BaselineRig.Key()], Grant()); + + proxy.Pause(); + Ledger.Record("network-pause"); + var stalled = Stopwatch.StartNew(); + string stalledStatus; + try + { + using var timeout = CancellationTokenSource.CreateLinkedTokenSource(ct); + timeout.CancelAfter(TimeSpan.FromSeconds(1)); + await Ledger.Call("read-value", () => 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 Ledger.Call("read-value", () => owner.ReadValueAsync(scalar, ct))).StatusCode.ShouldBe(StatusCodes.Good); + await LogAsync(evidence, new { kind = "network-stall-recovered", cycle, status = stalledStatus, seconds = stalled.Elapsed.TotalSeconds }); + await Ledger.Call("subscription-delete", () => subscription.DeleteAsync(true, ct)); + await Ledger.Call("subscription-remove", () => owner.RemoveSubscriptionAsync(subscription, ct)); + rig.Sink.RebuildAddressSpace(); + Ledger.Record("address-space-rebuild"); + Configure(rig); + Ledger.Record("cycle-wire-body-completed"); + } + finally + { + proxy.Resume(); + await Ledger.Call("session-close", () => owner.CloseAsync(ct)).WaitAsync(TimeSpan.FromSeconds(10), ct); + await Ledger.Call("session-close", () => other.CloseAsync(ct)).WaitAsync(TimeSpan.FromSeconds(10), ct); + // Comparator (lmxopcua#799): the held-reference zero-closed-list comparison is an SDK2-only contract. SDK 1.5's + // Session.Close keeps its lists until Dispose (baseline-canonical/README.md, OWNERSHIP.md: closed sessions and + // points are collected once harness roots drop). Record the counts each cycle instead of asserting; the session + // must still leave the SessionManager, which the loop's quiescent zero-checks assert. + var retained = ownedSessions.Select(ContinuationCounts).ToArray(); + ClosedListObservation = (retained.Sum(x => x.Browse), retained.Sum(x => x.History)); + } + } + + private static async Task RestartWithActiveResourcesAsync(BaselineRig 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 Ledger.Call("subscription-create", () => subscription.CreateAsync(ct)); + subscription.AddItem(new MonitoredItem(subscription.DefaultItem) + { + StartNodeId = Raw(client, Scalar), AttributeId = Attributes.Value, SamplingInterval = 50, + }); + await Ledger.Call("subscription-apply", () => subscription.ApplyChangesAsync(ct)); + PublicCounts(rig).ShouldBe((1, 1, 1)); + RequirePoint(await BrowseAsync(client, ct)); + RequirePoint((await PageAsync(client, default, false, ct)).ContinuationPoint); + var oldInternal = rig.Server.CurrentInstance; + var oldSessionManager = oldInternal.SessionManager; + var oldSubscriptionManager = oldInternal.SubscriptionManager; + var oldSessions = oldSessionManager.GetSessions().ToArray(); + var oldSubscriptions = oldSubscriptionManager.GetSubscriptions().ToArray(); + var oldValidator = rig.Validator; + var oldStores = BaselineMetrics.CachedStoreReferences(oldValidator); + 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)); + await LogAsync(CurrentEvidence, new { kind = "original-validator-after-stop", cachedStoreReferencesBefore = oldStores, + cachedStoreReferencesAfter = BaselineMetrics.CachedStoreReferences(oldValidator), + disposalFlagObservable = false, reason = "SDK1.5 has no CertificateManager disposal API or disposed flag" }); + PublicCounts(rig).ShouldBe((0, 0, 0)); + } + + private static async Task ConnectAsync(BaselineRig 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 Ledger.Call("session-create", () => new DefaultSessionFactory(telemetry).CreateAsync(config, + new ConfiguredEndpoint(null, endpoint, EndpointConfiguration.Create(config)), false, false, + "SDK2LifecycleSoak", 15_000, new UserIdentity("apikey", Encoding.UTF8.GetBytes(BaselineRig.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 Ledger.Call("browse", () => session.BrowseAsync(null, null, 1, [new BrowseDescription + { + NodeId = Raw(session, BaselineRig.RawDeviceFolder), BrowseDirection = BrowseDirection.Forward, + ReferenceTypeId = ReferenceTypeIds.HierarchicalReferences, IncludeSubtypes = true, + ResultMask = (uint)BrowseResultMask.All, + }], ct))).Results[0].ContinuationPoint; + + private static async Task PageAsync(ClientSession session, byte[]? point, + bool release, CancellationToken ct) => (await Ledger.Call("history-read", () => 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, + new QualifiedName(BrowseNames.EventId))] }, + }), TimestampsToReturn.Both, release, + [new HistoryReadValueId { NodeId = Raw(session, BaselineRig.RawDeviceFolder), ContinuationPoint = point }], ct))).Results[0]; + +} diff --git a/scripts/validation/opcua-lifecycle-memory/sdk1-comparator/Program.cs b/scripts/validation/opcua-lifecycle-memory/sdk1-comparator/Program.cs new file mode 100644 index 000000000..ed7f54b11 --- /dev/null +++ b/scripts/validation/opcua-lifecycle-memory/sdk1-comparator/Program.cs @@ -0,0 +1,318 @@ +using System.Collections.Concurrent; +using System.Diagnostics; +using System.Net; +using System.Net.Sockets; +using System.Reflection; +using System.Runtime.InteropServices; +using System.Text.Json; +using Opc.Ua; +using Opc.Ua.Configuration; +using Opc.Ua.Server; +using Shouldly; +using ZB.MOM.WW.OtOpcUa.Commons.OpcUa; +using ZB.MOM.WW.OtOpcUa.Core.Abstractions; +using CoreHistoryReadResult = ZB.MOM.WW.OtOpcUa.Core.Abstractions.HistoryReadResult; +return await Canonical.ComparatorAsync(args); +internal static partial class Canonical +{ + private const string Scalar = BaselineRig.RawDeviceFolder + "/SoakScalar"; + private const string ArrayTag = BaselineRig.RawDeviceFolder + "/SoakArray"; + private const string UnsScalar = "SoakEquipment/Scalar"; + private static readonly DateTime Epoch = new(2026,10,1,0,0,0,DateTimeKind.Utc); + internal static readonly OperationLedger Ledger = new(); + private static StreamWriter CurrentEvidence = null!; + + /// + /// lmxopcua#799 SDK 1.5.378.182 comparator: the canonical late-slope loop (same cycle body, same every-fourth-cycle + /// active restart, same quiescent sample after each cycle, same cycle-3 cold baseline and caps with a latched first + /// rejection, same 30 s cycle / 25 s restart / 120 s startup bounds, same 7,200 s phase) driven against the immutable + /// efb9197a SDK1 publication. Comparator only: acceptance=false. Differences from the SDK2 test are listed in + /// README.md (standalone process instead of the VSTest testhost; no SDK2 certificate-manager disposal observables; + /// closed-session continuation lists recorded rather than asserted; managed-monotonic guard latched, not fatal). + /// + internal static async Task ComparatorAsync(string[] args) + { + if (args.Length != 3 || args[2] != "--comparator") throw new InvalidOperationException("Usage: 7200 --comparator"); + var root = Path.GetFullPath(args[0]); var seconds = int.Parse(args[1]); + // The measured phase matches the 7200 s late-slope run; a 60 s smoke exists only to qualify the tooling (never a result). + if (seconds != 7200 && !(seconds == 60 && Environment.GetEnvironmentVariable("OTOPCUA_COMPARATOR_SMOKE") == "1")) + throw new InvalidOperationException("The comparator phase matches the 7200 s late-slope run"); + if (Directory.Exists(root)) throw new InvalidOperationException("Use a fresh private artifact directory"); + Directory.CreateDirectory(root); + await using var output = new StreamWriter(new FileStream(Path.Combine(root, "resources.jsonl"), FileMode.CreateNew, FileAccess.Write, FileShare.Read)); + CurrentEvidence = output; + var sdk = typeof(NodeId).Assembly.GetCustomAttribute()?.InformationalVersion; + if (sdk is null || !sdk.StartsWith("1.5.378.182+", StringComparison.Ordinal)) throw new InvalidOperationException("The immutable .182 SDK is required"); + var clientRoot = Path.Combine(root, "private-fixture", "client-pki"); + var startup = Stopwatch.StartNew(); + using var startupDeadline = new CancellationTokenSource(TimeSpan.FromSeconds(120)); + var telemetry = DefaultTelemetry.Create(_ => { }); + var config = new ApplicationConfiguration(telemetry) + { + ApplicationName = "SDK2LifecycleSoak", ApplicationUri = $"urn:SDK2LifecycleSoak:{Guid.NewGuid():N}", ApplicationType = ApplicationType.Client, + SecurityConfiguration = new SecurityConfiguration + { + ApplicationCertificate = new CertificateIdentifier { StoreType = CertificateStoreType.Directory, StorePath = Path.Combine(clientRoot, "own"), SubjectName = "CN=SDK2LifecycleSoak" }, + TrustedIssuerCertificates = new CertificateTrustList { StoreType = CertificateStoreType.Directory, StorePath = Path.Combine(clientRoot, "issuer") }, + TrustedPeerCertificates = new CertificateTrustList { StoreType = CertificateStoreType.Directory, StorePath = Path.Combine(clientRoot, "trusted") }, + RejectedCertificateStore = new CertificateTrustList { StoreType = CertificateStoreType.Directory, StorePath = Path.Combine(clientRoot, "rejected") }, AutoAcceptUntrustedCertificates = true + }, + ClientConfiguration = new ClientConfiguration { DefaultSessionTimeout = 15000 }, TransportQuotas = new TransportQuotas { OperationTimeout = 5000 } + }; + var application = new ApplicationInstance(telemetry) { ApplicationConfiguration = config }; + await config.ValidateAsync(ApplicationType.Client, startupDeadline.Token); + await application.CheckApplicationInstanceCertificatesAsync(true, null, startupDeadline.Token); + var rig = new BaselineRig(Path.Combine(root, "private-fixture")); + await rig.BootAsync(startupDeadline.Token); Configure(rig); + var startupSeconds = startup.Elapsed.TotalSeconds; + var watch = Stopwatch.StartNew(); + await LogAsync(output, new + { + kind = "start", utc = DateTimeOffset.UtcNow, processId = Environment.ProcessId, mode = "sdk1-comparator", requestedSeconds = seconds, + build = Environment.GetEnvironmentVariable("OTOPCUA_COMPARATOR_SOURCE"), sourcePublication = "efb9197a", sdk, + runtime = Environment.Version.ToString(), architecture = RuntimeInformation.ProcessArchitecture.ToString(), startupSeconds, + limits = new { retainedManagedGrowthBytes = 64L << 20, rssGrowthBytes = 128L << 20, fdGrowth = 32 }, acceptance = false, + unsupportedObservables = new[] { "SDK2 CertificateManager disposed flag", "SDK2 manager-owned cache-clear proof", "ApplicationInstance.DisposeAsync" }, + }); + var cycles = 0; var restarts = 0; var lastCompleted = 0d; + Dictionary? baseline = null; object? firstRejection = null; object? monotonicRejection = null; + var retained = new Queue(); + var failures = new List(); + try + { + while (lastCompleted < seconds) + { + var cycle = Stopwatch.StartNew(); + using var cycleDeadline = new CancellationTokenSource(TimeSpan.FromSeconds(30)); + var ct = cycleDeadline.Token; + await ExerciseAsync(rig, config, ++cycles, output, ct); + await EventuallyAsync(() => PublicCounts(rig) == (0, 0, 0), ct); + var sample = await CaptureAsync(rig, config, cycles, watch.Elapsed.TotalSeconds, clientRoot, ct); + await LogAsync(output, new { kind = "resource-observation", resources = sample }); + foreach (var zero in new[] { "Sessions", "Subscriptions", "Items", "BrowsePoints", "HistoryPoints", "LiveDataItems", "LiveEventItems", "MonitoredNodes", "PkiDescriptors" }) + Convert.ToInt64(sample[zero]).ShouldBe(0, $"quiescent {zero} must be zero"); + if (cycles >= 3) + { + baseline ??= sample; + foreach (var steady in new[] { "Variables", "Folders", "Conditions", "ServerCachedStoreReferences", "ClientCachedStoreReferences" }) + sample[steady].ShouldBe(baseline[steady], $"quiescent {steady} must stay at its cycle-3 value"); + var exceeded = new List(); + if ((int)sample["Descriptors"] > (int)baseline["Descriptors"] + 32) exceeded.Add("Descriptors"); + if ((long)sample["ManagedBytes"] > (long)baseline["ManagedBytes"] + (64L << 20)) exceeded.Add("ManagedBytes"); + if ((long)sample["RssBytes"] > (long)baseline["RssBytes"] + (128L << 20)) exceeded.Add("RssBytes"); + if (exceeded.Count > 0 && firstRejection is null) + { + firstRejection = new { cycle = cycles, seconds = watch.Elapsed.TotalSeconds, exceededFields = exceeded, resources = sample, baseline }; + await LogAsync(output, new { kind = "budget-rejected", cycle = cycles, seconds = watch.Elapsed.TotalSeconds, exceededFields = exceeded, resources = sample, baseline }); + } + retained.Enqueue((long)sample["ManagedBytes"]); + if (retained.Count > 24) retained.Dequeue(); + var window = retained.ToArray(); + if (window.Length == 24 && monotonicRejection is null && window.Zip(window.Skip(1)).All(p => p.Second > p.First) && window[^1] - window[0] > (16L << 20)) + { + monotonicRejection = new { cycle = cycles, seconds = watch.Elapsed.TotalSeconds, first = window[0], last = window[^1] }; + await LogAsync(output, new { kind = "managed-monotonic-rejected", cycle = cycles, seconds = watch.Elapsed.TotalSeconds, first = window[0], last = window[^1] }); + } + } + await LogAsync(output, new { kind = firstRejection is null ? "quiescent" : "quiescent-budget-rejected", cycleSeconds = cycle.Elapsed.TotalSeconds, + closedSessionPointsWhileHeld = new { browse = ClosedListObservation.Browse, history = ClosedListObservation.History }, resources = sample }); + if (cycles % 4 == 0) + { + var restart = Stopwatch.StartNew(); + using var restartDeadline = CancellationTokenSource.CreateLinkedTokenSource(ct); + restartDeadline.CancelAfter(TimeSpan.FromSeconds(25)); + await RestartWithActiveResourcesAsync(rig, config, restartDeadline.Token); + restartDeadline.Token.ThrowIfCancellationRequested(); + var bounded = restart.Elapsed.TotalSeconds; + bounded.ShouldBeLessThanOrEqualTo(25); + Configure(rig); restarts++; + await LogAsync(output, new { kind = "restart", cycles, restarts, seconds = restart.Elapsed.TotalSeconds, boundedRestartSeconds = bounded }); + } + ct.ThrowIfCancellationRequested(); + cycle.Elapsed.TotalSeconds.ShouldBeLessThanOrEqualTo(30); + lastCompleted = watch.Elapsed.TotalSeconds; + await LogAsync(output, new { kind = "comparator-cycle", cycle = cycles, restarts, elapsedSeconds = lastCompleted, cycleSeconds = cycle.Elapsed.TotalSeconds }); + } + } + catch (Exception error) + { + failures.Add(error.GetType().FullName!); + await LogAsync(output, new { kind = "failed", cycles, restarts, seconds = watch.Elapsed.TotalSeconds, exceptionType = error.GetType().FullName, + message = error.Message.Length > 600 ? error.Message[..600] : error.Message, + uaStatus = error is ServiceResultException ua ? ua.StatusCode.ToString() : null, ledger = Ledger.Snapshot() }); + Console.Error.WriteLine(error); + } + var phaseSeconds = watch.Elapsed.TotalSeconds; + var shutdown = Stopwatch.StartNew(); + var shutdownVerified = false; + try + { + await rig.DisposeAsync().AsTask().WaitAsync(TimeSpan.FromSeconds(20)); + await Task.Run(() => { if (Directory.Exists(clientRoot)) Directory.Delete(clientRoot, true); }).WaitAsync(TimeSpan.FromSeconds(5)); + using var probe = new CancellationTokenSource(TimeSpan.FromSeconds(10)); + (await BaselineMetrics.DescriptorsAsync(probe.Token, clientRoot, rig.PkiRoot)).Pki.ShouldBe(0); + Directory.Exists(clientRoot).ShouldBeFalse(); Directory.Exists(rig.PkiRoot).ShouldBeFalse(); + shutdownVerified = shutdown.Elapsed.TotalSeconds <= 60; + } + catch (Exception error) { failures.Add("shutdown:" + error.GetType().FullName); Console.Error.WriteLine(error); } + await LogAsync(output, new + { + kind = "comparator-result", mode = "sdk1-comparator", cycles, restarts, cycleSeconds = phaseSeconds, shutdownSeconds = shutdown.Elapsed.TotalSeconds, + durationCompleted = lastCompleted >= seconds && failures.Count == 0, shutdownVerified, coldBudgetRejected = firstRejection is not null, + firstRejection, managedMonotonicRejected = monotonicRejection, fatalFailure = failures.Count != 0, failureTypes = failures, + applicationInstanceDisposal = "unsupported on SDK 1.5 (no DisposeAsync)", ledger = Ledger.Snapshot(), acceptance = false, + }); + return failures.Count != 0 ? 1 : firstRejection is not null || monotonicRejection is not null ? 3 : 0; + } + + private static async Task> CaptureAsync(BaselineRig rig, ApplicationConfiguration config, int cycle, double seconds, + string clientRoot, CancellationToken ct) + { + var counts = PublicCounts(rig); + var points = rig.Server.CurrentInstance.SessionManager.GetSessions().Select(ContinuationCounts).ToArray(); + var descriptors = await BaselineMetrics.DescriptorsAsync(ct, clientRoot, rig.PkiRoot); + var retainedBytes = GC.GetTotalMemory(forceFullCollection: true); + var memory = GC.GetGCMemoryInfo(); + using var process = Process.GetCurrentProcess(); + process.Refresh(); + return new Dictionary + { + ["Cycle"] = cycle, ["Seconds"] = seconds, ["Sessions"] = counts.Sessions, ["Subscriptions"] = counts.Subscriptions, ["Items"] = counts.Items, + ["BrowsePoints"] = points.Sum(x => x.Browse), ["HistoryPoints"] = points.Sum(x => x.History), + ["LiveDataItems"] = BaselineMetrics.Counter(rig.Nm, "LiveItemCount"), ["LiveEventItems"] = BaselineMetrics.Counter(rig.Nm, "LiveEventItemCount"), + ["MonitoredNodes"] = BaselineMetrics.Counter(rig.Nm, "MonitoredNodeCount"), ["Variables"] = OptionalCounter(rig.Nm, "VariableCount"), + ["Folders"] = OptionalCounter(rig.Nm, "FolderCount"), ["Conditions"] = OptionalCounter(rig.Nm, "AlarmConditionCount"), + ["ServerCachedStoreReferences"] = BaselineMetrics.CachedStoreReferences(rig.Validator), + ["ClientCachedStoreReferences"] = BaselineMetrics.CachedStoreReferences(config.CertificateValidator), + ["Descriptors"] = descriptors.Total, ["PkiDescriptors"] = descriptors.Pki, ["ManagedBytes"] = retainedBytes, ["RssBytes"] = process.WorkingSet64, + ["PrivateBytes"] = process.PrivateMemorySize64, ["AllocatedBytes"] = GC.GetTotalAllocatedBytes(), + ["Gen0"] = GC.CollectionCount(0), ["Gen1"] = GC.CollectionCount(1), ["Gen2"] = GC.CollectionCount(2), ["GcCommittedBytes"] = memory.TotalCommittedBytes, + ["GcHeapBytes"] = memory.HeapSizeBytes, ["GcFragmentedBytes"] = memory.FragmentedBytes, ["PinnedObjects"] = memory.PinnedObjectsCount, + ["PendingFinalizers"] = memory.FinalizationPendingCount, ["JitMethods"] = System.Runtime.JitInfo.GetCompiledMethodCount(), + ["JitIlBytes"] = System.Runtime.JitInfo.GetCompiledILBytes(), ["ActiveTimers"] = Timer.ActiveCount, ["Threads"] = process.Threads.Count, + }; + } + + private static int OptionalCounter(object manager, string name) => + manager.GetType().GetProperty(name, BindingFlags.Instance | BindingFlags.Public | BindingFlags.NonPublic)?.GetValue(manager) is int value ? value : -1; + + private static int ScalarValue(Variant value)=>value.Value is int result?result:throw new InvalidOperationException("Expected scalar Int32 CLR shape"); + private static void RequirePoint(byte[]? point)=>(point is {Length:>0}).ShouldBeTrue(); + private static (int Browse,int History) ContinuationCounts(ISession session)=>BaselineMetrics.Points(session); + private static (int Sessions,int Subscriptions,int Items) PublicCounts(BaselineRig rig) + {var subscriptions=rig.Server.CurrentInstance.SubscriptionManager.GetSubscriptions();return(rig.Server.CurrentInstance.SessionManager.GetSessions().Count,subscriptions.Count,subscriptions.Sum(s=>s.MonitoredItemCount));} + private static async Task EventuallyAsync(Func condition,CancellationToken ct) + {var elapsed=Stopwatch.StartNew();while(!condition()&&elapsed.Elapsed WriteAsync(string id,object? value,AddressSpaceRealm realm,CancellationToken ct)=>Task.FromResult(new NodeWriteOutcome(true,null));} + internal sealed class SqlLikeEventSource(List events) : IHistorianDataSource + { + private int _eventReads; + internal int EventReads => Volatile.Read(ref _eventReads); + private readonly List _all = + events.OrderBy(e => e.EventTimeUtc).ThenBy(e => e.EventId, StringComparer.Ordinal).ToList(); + + public async Task ReadEventsAsync( + string? sourceName, DateTime startUtc, DateTime endUtc, int maxEvents, CancellationToken ct) + { + Interlocked.Increment(ref _eventReads); + await Task.Delay(1, ct).ConfigureAwait(false); // complete on a pool thread, like a real backend + var window = _all.Where(e => e.EventTimeUtc >= startUtc && e.EventTimeUtc < endUtc).ToList(); + var cap = maxEvents <= 0 ? 1000 : maxEvents; + return new HistoricalEventsResult(window.Take(cap).ToList(), window.Count > cap ? new byte[] { 1 } : null); + } + + public bool SupportsReverseEvents => true; + + /// The #675 overload: reverse is (lo, hi] by (time, id) descending. + public async Task ReadEventsAsync( + string? sourceName, DateTime startUtc, DateTime endUtc, int maxEvents, bool reverse, CancellationToken ct) + { + if (!reverse) return await ReadEventsAsync(sourceName, startUtc, endUtc, maxEvents, ct).ConfigureAwait(false); + await Task.Delay(1, ct).ConfigureAwait(false); + var (lo, hi) = startUtc <= endUtc ? (startUtc, endUtc) : (endUtc, startUtc); + var window = _all.Where(e => e.EventTimeUtc > lo && e.EventTimeUtc <= hi).Reverse().ToList(); + var cap = maxEvents <= 0 ? 1000 : maxEvents; + return new HistoricalEventsResult(window.Take(cap).ToList(), window.Count > cap ? new byte[] { 1 } : null); + } + + public Task ReadRawAsync(string r, DateTime s, DateTime e, uint m, CancellationToken ct) => + NullHistorianDataSource.Instance.ReadRawAsync(r, s, e, m, ct); + public Task ReadProcessedAsync( + string r, DateTime s, DateTime e, TimeSpan i, HistoryAggregateType a, CancellationToken ct) => + NullHistorianDataSource.Instance.ReadProcessedAsync(r, s, e, i, a, ct); + public Task ReadAtTimeAsync(string r, IReadOnlyList t, CancellationToken ct) => + NullHistorianDataSource.Instance.ReadAtTimeAsync(r, t, ct); + public HistorianHealthSnapshot GetHealthSnapshot() => NullHistorianDataSource.Instance.GetHealthSnapshot(); + public void Dispose() { } + } + /// 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(); } + } + } +} + +internal sealed class OperationLedger +{ + private readonly ConcurrentDictionary _counts=new(); + internal void Record(string name)=>_counts.AddOrUpdate(name,1,(_,count)=>count+1); + internal async Task Call(string name,Func> operation) + {Record(name+".attempted");try{var value=await operation();Record(name+".returned");return value;}catch{Record(name+".faulted");throw;}} + internal async Task Call(string name,Func operation) + {Record(name+".attempted");try{await operation();Record(name+".returned");}catch{Record(name+".faulted");throw;}} + internal IReadOnlyDictionary Snapshot()=>new SortedDictionary(_counts); +} diff --git a/scripts/validation/opcua-lifecycle-memory/sdk1-comparator/README.md b/scripts/validation/opcua-lifecycle-memory/sdk1-comparator/README.md new file mode 100644 index 000000000..0860a4c98 --- /dev/null +++ b/scripts/validation/opcua-lifecycle-memory/sdk1-comparator/README.md @@ -0,0 +1,29 @@ +# SDK 1.5.378.182 comparator (lmxopcua#799) + +The owner's #799 scope asks for the matched SDK 1.5 comparator numbers next to the SDK2 acceptance runs. This +comparator drives the canonical workload against the **immutable repaired SDK1 baseline `efb9197a`** (includes +the #807 terminal-disposal fix), published RID-specific (`win-x64`, framework-dependent) from that commit and +referenced, never rebuilt. It is `acceptance=false` by construction; it is evidence for comparison only. + +Sources: `BaselineRig.cs`, `BaselineMetrics.cs`, `CanonicalOperations.cs`, `Program.cs` are copies of +`../baseline-canonical/` (the reviewed SDK1 port of the canonical operations, ported from the SDK2 soak at +`75057d8c`), with exactly these changes: + +| Change | Why | +|---|---| +| `Program.cs`: the 1..4-cycle / 120 s short-smoke entry is replaced by `ComparatorAsync`, the late-slope loop: same cycle body, quiescent sample after every cycle, active restart every fourth cycle (25 s bound), 30 s cycle bound, 120 s startup bound, 7,200 s phase, cycle-3 cold baseline with the unchanged +64 MiB managed / +128 MiB RSS / +32 descriptor caps and a latched first rejection, then bounded shutdown checks. | The canonical late-slope protocol applied to SDK1. | +| `CanonicalOperations.cs`: the held-reference zero closed-session continuation-list assertion is recorded per cycle (`closedSessionPointsWhileHeld`) instead of asserted; the 250 ms observation delay is removed. | SDK 1.5 `Session.Close` keeps the lists until `Dispose`; the baseline-canonical README/OWNERSHIP.md show this is not the SDK1 close contract and the objects are collected once harness roots drop. Asserting it would end the comparator at cycle 1. | +| The 24-sample managed-monotonic guard is latched (recorded) rather than fatal. | So the comparator yields numbers for the full phase even if the old SDK grows; it is not a pass/fail gate. | +| Steady inventories use SDK1 observables: `Server/ClientCachedStoreReferences` (validator cached store references) in place of SDK2 manager cached stores; no `ApplicationInstance.DisposeAsync` / manager-disposed checks. | Not present in SDK 1.5 (reported unsupported, never as zero). | +| `WindowsDescriptors.cs`: byte-for-byte the SDK2 test's Windows descriptor observer. | Same descriptor/PKI semantics as the SDK2 Windows run. | +| Samples also carry `PrivateBytes`. | Descriptive. | + +Known non-matched elements: the comparator is a standalone self-contained executable, while the SDK2 workload runs +inside the VSTest testhost (xUnit v3), so absolute RSS includes different host overhead; compare growth relative to +each run's own cycle-3 baseline. `OPCUA_CERTIFICATE_LEAK_TRACKING` has no SDK1 equivalent. + +```bash +python3 build.py --host "$sdk1HostWinX64" --client "$sdk1ClientWinX64" --assertions "$dirWithShouldly" --output "$build" +# on Windows: +powershell -NoProfile -ExecutionPolicy Bypass -File launch.ps1 -Bin "$build\publish" -Output "$run" -SourceRevision "$rev" +``` diff --git a/scripts/validation/opcua-lifecycle-memory/sdk1-comparator/WindowsDescriptors.cs b/scripts/validation/opcua-lifecycle-memory/sdk1-comparator/WindowsDescriptors.cs new file mode 100644 index 000000000..6515a668d --- /dev/null +++ b/scripts/validation/opcua-lifecycle-memory/sdk1-comparator/WindowsDescriptors.cs @@ -0,0 +1,132 @@ +using System.ComponentModel; +using System.Runtime.InteropServices; +using System.Runtime.Versioning; +using Microsoft.Win32.SafeHandles; + +/// Byte-for-byte the Windows descriptor observer of Sdk2LifecycleWindowsDescriptors.cs (lmxopcua#799), hosted for the SDK1 comparator. +internal static class WindowsDescriptors +{ + [SupportedOSPlatform("windows")] + internal static Task<(int Total, int Pki)> CountAsync(CancellationToken ct, IReadOnlyList roots) + { + var result = new TaskCompletionSource<(int Total, int Pki)>(TaskCreationOptions.RunContinuationsAsynchronously); + var thread = new Thread(() => + { + try { result.TrySetResult(WindowsHandleCounts(roots)); } + catch (Exception error) { result.TrySetException(error); } + }) { IsBackground = true, Name = "soak-descriptor-probe" }; + thread.Start(); + return BoundAsync(); + + async Task<(int Total, int Pki)> BoundAsync() + { + try { return await result.Task.WaitAsync(TimeSpan.FromSeconds(5), ct).ConfigureAwait(false); } + catch (TimeoutException) { throw new TimeoutException("Descriptor probe exceeded its handle-enumeration deadline."); } + } + } + + [SupportedOSPlatform("windows")] + private static (int Total, int Pki) WindowsHandleCounts(IReadOnlyList roots) + { + // Canonicalise the roots first and close those directory handles before the snapshot, so they are not counted. + var canonicalRoots = roots.Select(WindowsCanonicalRoot).Distinct(StringComparer.OrdinalIgnoreCase).ToArray(); + var self = (nuint)Environment.ProcessId; + var length = 1 << 20; + while (true) + { + var buffer = Marshal.AllocHGlobal(length); + try + { + var status = NtQuerySystemInformation(SystemExtendedHandleInformation, buffer, length, out var needed); + if (status == StatusInfoLengthMismatch) + { + length = Math.Max(length * 2, needed + (1 << 16)); + continue; + } + if (status != 0) throw new Win32Exception(RtlNtStatusToDosError(status), $"NtQuerySystemInformation failed (0x{status:X8})."); + var count = (long)Marshal.ReadIntPtr(buffer); + var entries = buffer + 2 * IntPtr.Size; + var entrySize = Marshal.SizeOf(); + int total = 0, pki = 0; + var path = new char[32768]; + for (long i = 0; i < count; i++) + { + var entry = Marshal.PtrToStructure(entries + (nint)(i * entrySize)); + if (entry.UniqueProcessId != self) continue; + total++; + var handle = (nint)entry.HandleValue; + // GetFileType is answered by the I/O manager without the file-object lock; only disk files and + // directories go on to the name query (pipes, sockets and consoles cannot hold PKI material). + if (GetFileType(handle) != FileTypeDisk) continue; + var name = FinalPath(handle, path); + if (name is not null && WithinWindowsRoots(name, canonicalRoots)) pki++; + } + return (total, pki); + } + finally { Marshal.FreeHGlobal(buffer); } + } + } + + [SupportedOSPlatform("windows")] + private static string WindowsCanonicalRoot(string root) + { + var full = Path.TrimEndingDirectorySeparator(Path.GetFullPath(root)); + if (!Directory.Exists(full)) return BaselineMetrics.CanonicalPath(full); + using var directory = CreateFileW(full, 0, FileShareAll, 0, OpenExisting, FileFlagBackupSemantics, 0); + if (directory.IsInvalid) throw new Win32Exception(Marshal.GetLastWin32Error(), "Cannot open a PKI root for canonicalisation."); + return FinalPath(directory.DangerousGetHandle(), new char[32768]) + ?? throw new IOException("A PKI root has no final path."); + } + + private static string? FinalPath(nint handle, char[] buffer) + { + var written = GetFinalPathNameByHandleW(handle, buffer, buffer.Length, 0); + if (written == 0 || written >= buffer.Length) return null; // Closed/reused between snapshot and query, or not a named file. + var name = new string(buffer, 0, (int)written); + if (name.StartsWith(@"\\?\UNC\", StringComparison.Ordinal)) return @"\\" + name[8..]; + return name.StartsWith(@"\\?\", StringComparison.Ordinal) ? name[4..] : name; + } + + private static bool WithinWindowsRoots(string path, IReadOnlyList roots) + { + path = Path.TrimEndingDirectorySeparator(path); + return roots.Any(root => path.Equals(root, StringComparison.OrdinalIgnoreCase) + || path.StartsWith(root + Path.DirectorySeparatorChar, StringComparison.OrdinalIgnoreCase)); + } + + private const int SystemExtendedHandleInformation = 64; + private const int StatusInfoLengthMismatch = unchecked((int)0xC0000004); + private const uint FileTypeDisk = 1; + private const uint FileShareAll = 7; + private const uint OpenExisting = 3; + private const uint FileFlagBackupSemantics = 0x02000000; + + [StructLayout(LayoutKind.Sequential)] + private struct HandleEntry + { + public nint Object; + public nuint UniqueProcessId; + public nuint HandleValue; + public uint GrantedAccess; + public ushort CreatorBackTraceIndex; + public ushort ObjectTypeIndex; + public uint HandleAttributes; + public uint Reserved; + } + + [DllImport("ntdll.dll")] + private static extern int NtQuerySystemInformation(int informationClass, nint information, int length, out int returnLength); + + [DllImport("ntdll.dll")] + private static extern int RtlNtStatusToDosError(int status); + + [DllImport("kernel32.dll", SetLastError = true)] + private static extern uint GetFileType(nint handle); + + [DllImport("kernel32.dll", SetLastError = true, CharSet = CharSet.Unicode)] + private static extern uint GetFinalPathNameByHandleW(nint handle, [Out] char[] path, int length, uint flags); + + [DllImport("kernel32.dll", SetLastError = true, CharSet = CharSet.Unicode)] + private static extern SafeFileHandle CreateFileW(string name, uint access, uint share, nint security, uint disposition, + uint flags, nint template); +} diff --git a/scripts/validation/opcua-lifecycle-memory/sdk1-comparator/build.py b/scripts/validation/opcua-lifecycle-memory/sdk1-comparator/build.py new file mode 100644 index 000000000..444f68d0c --- /dev/null +++ b/scripts/validation/opcua-lifecycle-memory/sdk1-comparator/build.py @@ -0,0 +1,80 @@ +#!/usr/bin/env python3 +"""Publish the SDK 1.5.378.182 comparator (lmxopcua#799) self-contained against an immutable SDK1 publication. + +Like ../baseline-canonical/build.py: never rebuilds product code. References come from the given RID-specific +framework-dependent publications of the SDK1 host and client (host wins duplicate names; native DLLs are skipped as +references and copied beside the output), plus Shouldly and its assertion dependencies. The comparator is published +self-contained for the requested RID on the same pinned runtime as the SDK2 run (10.0.14). +""" +import argparse +import hashlib +import json +from pathlib import Path +import shutil +import subprocess +import xml.etree.ElementTree as ET + +RUNTIME = "10.0.14" +SOURCES = ["Program.cs", "BaselineRig.cs", "BaselineMetrics.cs", "CanonicalOperations.cs", "WindowsDescriptors.cs"] + + +def managed(path): + data = path.read_bytes() + if data[:2] != b"MZ": + return False + pe = int.from_bytes(data[0x3C:0x40], "little") + magic = int.from_bytes(data[pe + 24:pe + 26], "little") + directories = pe + 24 + (112 if magic == 0x20B else 96) + clr = directories + 14 * 8 + return int.from_bytes(data[clr:clr + 4], "little") != 0 + + +parser = argparse.ArgumentParser(description=__doc__) +parser.add_argument("--host", type=Path, required=True) +parser.add_argument("--client", type=Path, required=True) +parser.add_argument("--assertions", type=Path, required=True) +parser.add_argument("--rid", default="win-x64") +parser.add_argument("--output", type=Path, required=True) +args = parser.parse_args() +root = args.output.resolve() +root.mkdir(parents=True, exist_ok=False) +source = Path(__file__).resolve().parent +references, natives = {}, {} +for directory in [args.host, args.client]: + for dll in sorted(directory.resolve(strict=True).glob("*.dll")): + (references if managed(dll) else natives).setdefault(dll.name, dll) +for name in ["Shouldly.dll", "DiffEngine.dll", "EmptyFiles.dll", "EmptyFiles.Xml.dll"]: + path = args.assertions.resolve(strict=True) / name + if path.exists(): + references.setdefault(name, path) +project = ET.Element("Project", Sdk="Microsoft.NET.Sdk") +properties = ET.SubElement(project, "PropertyGroup") +for key, value in {"OutputType": "Exe", "TargetFramework": "net10.0", "Nullable": "enable", "ImplicitUsings": "enable", + "TreatWarningsAsErrors": "true", "EnableDefaultCompileItems": "false", "AssemblyName": "sdk1-comparator", + "RuntimeIdentifier": args.rid, "SelfContained": "true", "RuntimeFrameworkVersion": RUNTIME, + "NoWarn": "CA1416"}.items(): + ET.SubElement(properties, key).text = value +items = ET.SubElement(project, "ItemGroup") +ET.SubElement(items, "FrameworkReference", Include="Microsoft.AspNetCore.App") +for name, path in references.items(): + reference = ET.SubElement(items, "Reference", Include=path.stem) + ET.SubElement(reference, "HintPath").text = str(path) +for name in SOURCES: + ET.SubElement(items, "Compile", Include=str(source / name)) +path = root / "comparator.csproj" +ET.indent(project) +ET.ElementTree(project).write(path, encoding="unicode") +publish = root / "publish" +with (root / "build.log").open("w") as log: + result = subprocess.run(["dotnet", "publish", str(path), "-c", "Release", "-o", str(publish), "-m:2", "--nologo"], + stdout=log, stderr=subprocess.STDOUT) +if result.returncode == 0: + for name, native in natives.items(): + if not (publish / name).exists(): + shutil.copy2(native, publish / name) +manifest = {"diagnosticOnly": True, "productRebuilt": False, "rid": args.rid, "runtime": RUNTIME, + "references": {n: hashlib.sha256(p.read_bytes()).hexdigest() for n, p in references.items()}, + "natives": {n: hashlib.sha256(p.read_bytes()).hexdigest() for n, p in natives.items()}, + "sources": {n: hashlib.sha256((source / n).read_bytes()).hexdigest() for n in SOURCES}} +(root / "provenance.json").write_text(json.dumps(manifest, indent=2) + "\n") +raise SystemExit(result.returncode) diff --git a/scripts/validation/opcua-lifecycle-memory/sdk1-comparator/launch.ps1 b/scripts/validation/opcua-lifecycle-memory/sdk1-comparator/launch.ps1 new file mode 100644 index 000000000..791644d95 --- /dev/null +++ b/scripts/validation/opcua-lifecycle-memory/sdk1-comparator/launch.ps1 @@ -0,0 +1,48 @@ +<# +.SYNOPSIS + Runs the SDK 1.5.378.182 comparator (lmxopcua#799) for 7,200 s on Windows with the same envelope as ../windows/launch.ps1: + private task directory for TEMP and fixture PKI, ephemeral fixture credential, external 30 s sampler, 7,440 s outer deadline. +#> +param( + [Parameter(Mandatory)] [string] $Bin, + [Parameter(Mandatory)] [string] $Output, + [Parameter(Mandatory)] [string] $SourceRevision +) +$ErrorActionPreference = 'Stop' +if (Test-Path $Output) { throw 'Use a fresh output directory' } +New-Item -ItemType Directory -Path $Output | Out-Null +$Output = (Resolve-Path $Output).Path +New-Item -ItemType Directory -Path (Join-Path $Output 'tmp') | Out-Null +& icacls $Output /inheritance:r /grant:r "$($env:USERNAME):(OI)(CI)F" '*S-1-5-18:(OI)(CI)F' | Out-Null +$exe = Join-Path (Resolve-Path $Bin).Path 'sdk1-comparator.exe' +$started = Get-Date +$bytes = New-Object byte[] 3; [System.Security.Cryptography.RandomNumberGenerator]::Create().GetBytes($bytes) +$env:OTOPCUA_BASELINE_CANONICAL_KEY = 'otua_k-server-events_' + (($bytes | ForEach-Object { $_.ToString('x2') }) -join '') +$env:OTOPCUA_COMPARATOR_SOURCE = $SourceRevision +$env:TEMP = Join-Path $Output 'tmp'; $env:TMP = $env:TEMP +@{ kind = 'comparator-launch'; sourceRevision = $SourceRevision; exe = $exe + exeSha256 = (Get-FileHash -Algorithm SHA256 $exe).Hash.ToLowerInvariant() + coreClrSha256 = (Get-FileHash -Algorithm SHA256 (Join-Path (Split-Path $exe) 'coreclr.dll')).Hash.ToLowerInvariant() + startedUtc = $started.ToUniversalTime().ToString('o'); credentialBytes = $env:OTOPCUA_BASELINE_CANONICAL_KEY.Length } | + ConvertTo-Json | Set-Content -Encoding UTF8 (Join-Path $Output 'launch.json') +$psi = New-Object System.Diagnostics.ProcessStartInfo $exe +$psi.Arguments = "`"$(Join-Path $Output 'run')`" 7200 --comparator" +$psi.UseShellExecute = $false; $psi.RedirectStandardOutput = $true; $psi.RedirectStandardError = $true +$proc = [System.Diagnostics.Process]::Start($psi) +$outTask = $proc.StandardOutput.ReadToEndAsync(); $errTask = $proc.StandardError.ReadToEndAsync() +$sampler = Join-Path $Output 'external-samples.jsonl' +while (-not $proc.WaitForExit(30000)) { + $elapsed = ((Get-Date) - $started).TotalSeconds + $row = [ordered]@{ utc = (Get-Date).ToUniversalTime().ToString('o'); launcherElapsedSeconds = [math]::Round($elapsed, 1) } + try { + $p = Get-Process -Id $proc.Id -ErrorAction Stop + $row.workingSet64 = $p.WorkingSet64; $row.privateMemorySize64 = $p.PrivateMemorySize64 + $row.peakWorkingSet64 = $p.PeakWorkingSet64; $row.handleCount = $p.HandleCount; $row.threads = $p.Threads.Count + } catch { $row.targetGone = $true } + ($row | ConvertTo-Json -Compress) | Add-Content -Encoding UTF8 $sampler + if ($elapsed -gt 7440) { & taskkill /PID $proc.Id /T /F | Out-Null; 'outer-deadline' | Set-Content (Join-Path $Output 'launcher-outcome.txt') } +} +Set-Content -Encoding UTF8 (Join-Path $Output 'stdout.log') $outTask.Result +Set-Content -Encoding UTF8 (Join-Path $Output 'stderr.log') $errTask.Result +"$($proc.ExitCode)" | Set-Content -Encoding ASCII (Join-Path $Output 'exit-status.txt') +'done' | Add-Content (Join-Path $Output 'launcher-outcome.txt')