test(soak): #799 SDK 1.5.378.182 comparator for the late-slope workload
v2-ci / unit-tests (push) Has been cancelled

Copies the reviewed baseline-canonical SDK1 port and replaces its 1..4-cycle smoke entry with the late-slope
loop (same cycle body, every-fourth-cycle active restart, 30/25/120 s bounds, 7200 s phase, cycle-3 cold
baseline with unchanged +64 MiB managed / +128 MiB RSS / +32 descriptor caps, latched first rejection,
bounded shutdown checks). Run against the immutable efb9197a publication (#807 fix included), never rebuilt.
The held-reference closed-session list assertion is recorded instead (SDK1 Close keeps lists until Dispose,
per OWNERSHIP.md) and the managed-monotonic guard is latched, so the comparator yields numbers for the whole
phase. acceptance=false by construction. Windows descriptor observer shared byte-for-byte with the SDK2 test.

Claude-Session: https://claude.ai/code/session_01GT67ACmpMR7b9LGRPLijMi
This commit is contained in:
Joseph Doherty
2026-11-12 12:48:44 -05:00
parent aa60b3e34e
commit 2e9cccbdef
8 changed files with 1103 additions and 0 deletions
@@ -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<T>(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<System.Threading.Lock>(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<SemaphoreSlim>(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<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));
}
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);
}
}
@@ -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<OpcUaApplicationHost>.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<AclScopeRef, IReadOnlyList<NodeAclPath>>
{
[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<OpcUaUserAuthResult> 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"));
}
}
@@ -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<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 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<int>().ShouldBe(cycle);
(await Ledger.Call("read-value", () => owner.ReadValueAsync(array, ct))).Value.ShouldBeOfType<short[]>().ShouldBe(values);
(await Ledger.Call("read-value", () => owner.ReadValueAsync(uns, ct))).Value.ShouldBeOfType<int>().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<DataPlaneAuthorityState>();
Ledger.Record("key-revoke");
authority.Apply(BaselineRig.Cluster, $"revoked-{cycle}", [], [Grant()],
new Dictionary<AclScopeRef, IReadOnlyList<NodeAclPath>>());
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<ClientSession> 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<byte[]?> 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<Opc.Ua.HistoryReadResult> 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];
}
@@ -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!;
/// <summary>
/// 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).
/// </summary>
internal static async Task<int> ComparatorAsync(string[] args)
{
if (args.Length != 3 || args[2] != "--comparator") throw new InvalidOperationException("Usage: <fresh-output> 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<AssemblyInformationalVersionAttribute>()?.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<string, object>? baseline = null; object? firstRejection = null; object? monotonicRejection = null;
var retained = new Queue<long>();
var failures = new List<string>();
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<string>();
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<Dictionary<string, object>> 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<string, object>
{
["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<bool> condition,CancellationToken ct)
{var elapsed=Stopwatch.StartNew();while(!condition()&&elapsed.Elapsed<TimeSpan.FromSeconds(10))await Task.Delay(25,ct);condition().ShouldBeTrue();}
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));}
internal sealed class SqlLikeEventSource(List<HistoricalEvent> events) : IHistorianDataSource
{
private int _eventReads;
internal int EventReads => Volatile.Read(ref _eventReads);
private readonly List<HistoricalEvent> _all =
events.OrderBy(e => e.EventTimeUtc).ThenBy(e => e.EventId, StringComparer.Ordinal).ToList();
public async Task<HistoricalEventsResult> 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;
/// <summary>The #675 overload: reverse is <c>(lo, hi]</c> by (time, id) descending.</summary>
public async Task<HistoricalEventsResult> 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<CoreHistoryReadResult> ReadRawAsync(string r, DateTime s, DateTime e, uint m, CancellationToken ct) =>
NullHistorianDataSource.Instance.ReadRawAsync(r, s, e, m, ct);
public Task<CoreHistoryReadResult> ReadProcessedAsync(
string r, DateTime s, DateTime e, TimeSpan i, HistoryAggregateType a, CancellationToken ct) =>
NullHistorianDataSource.Instance.ReadProcessedAsync(r, s, e, i, a, ct);
public Task<CoreHistoryReadResult> ReadAtTimeAsync(string r, IReadOnlyList<DateTime> t, CancellationToken ct) =>
NullHistorianDataSource.Instance.ReadAtTimeAsync(r, t, ct);
public HistorianHealthSnapshot GetHealthSnapshot() => NullHistorianDataSource.Instance.GetHealthSnapshot();
public void Dispose() { }
}
/// <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(); }
}
}
}
internal sealed class OperationLedger
{
private readonly ConcurrentDictionary<string,long> _counts=new();
internal void Record(string name)=>_counts.AddOrUpdate(name,1,(_,count)=>count+1);
internal async Task<T> Call<T>(string name,Func<Task<T>> 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<Task> operation)
{Record(name+".attempted");try{await operation();Record(name+".returned");}catch{Record(name+".faulted");throw;}}
internal IReadOnlyDictionary<string,long> Snapshot()=>new SortedDictionary<string,long>(_counts);
}
@@ -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"
```
@@ -0,0 +1,132 @@
using System.ComponentModel;
using System.Runtime.InteropServices;
using System.Runtime.Versioning;
using Microsoft.Win32.SafeHandles;
/// <summary>Byte-for-byte the Windows descriptor observer of Sdk2LifecycleWindowsDescriptors.cs (lmxopcua#799), hosted for the SDK1 comparator.</summary>
internal static class WindowsDescriptors
{
[SupportedOSPlatform("windows")]
internal static Task<(int Total, int Pki)> CountAsync(CancellationToken ct, IReadOnlyList<string> 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<string> 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<HandleEntry>();
int total = 0, pki = 0;
var path = new char[32768];
for (long i = 0; i < count; i++)
{
var entry = Marshal.PtrToStructure<HandleEntry>(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<string> 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);
}
@@ -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)
@@ -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')