test(opcua): preserve original canonical adapter partial ownership proof
This commit is contained in:
@@ -0,0 +1,144 @@
|
||||
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();
|
||||
var canonicalRoots = roots.Select(root => CanonicalPath(root)).Distinct(StringComparer.Ordinal).ToArray();
|
||||
if (OperatingSystem.IsLinux())
|
||||
{
|
||||
var targets = Directory.EnumerateFileSystemEntries("/proc/self/fd")
|
||||
.Select(path => new FileInfo(path).LinkTarget).ToArray();
|
||||
return (targets.Length, targets.Count(path => IsWithinRoots(path, canonicalRoots)));
|
||||
}
|
||||
if (OperatingSystem.IsMacOS())
|
||||
{
|
||||
using var process = new Process { StartInfo = new ProcessStartInfo("/usr/sbin/lsof")
|
||||
{ RedirectStandardOutput = true, RedirectStandardError = true, UseShellExecute = false } };
|
||||
foreach (var arg in new[] { "-a", "-p", Environment.ProcessId.ToString(CultureInfo.InvariantCulture), "-Fnf" })
|
||||
process.StartInfo.ArgumentList.Add(arg);
|
||||
var text = await ReadDescriptorProbeAsync(process, TimeSpan.FromSeconds(5), ct);
|
||||
process.ExitCode.ShouldBe(0);
|
||||
var total = 0;
|
||||
var pki = 0;
|
||||
var descriptor = false;
|
||||
foreach (var line in text.Split('\n'))
|
||||
{
|
||||
if (line.StartsWith('f'))
|
||||
{
|
||||
descriptor = line.Length > 1 && char.IsDigit(line[1]);
|
||||
if (descriptor) total++;
|
||||
}
|
||||
else if (descriptor && line.StartsWith('n') && IsWithinRoots(line[1..], canonicalRoots)) pki++;
|
||||
}
|
||||
return (total, pki);
|
||||
}
|
||||
throw new PlatformNotSupportedException("This soak requires Linux procfs or macOS lsof for actual descriptor evidence.");
|
||||
}
|
||||
|
||||
private static async Task<string> ReadDescriptorProbeAsync(Process process, TimeSpan timeout, CancellationToken ct)
|
||||
{
|
||||
using var deadline = CancellationTokenSource.CreateLinkedTokenSource(ct);
|
||||
deadline.CancelAfter(timeout);
|
||||
process.Start();
|
||||
var stdout = process.StandardOutput.ReadToEndAsync(deadline.Token);
|
||||
var stderr = process.StandardError.ReadToEndAsync(deadline.Token);
|
||||
var exited = process.WaitForExitAsync(deadline.Token);
|
||||
try
|
||||
{
|
||||
// Waiting for exit alone does not bound reads: both redirected pipes must also finish.
|
||||
await Task.WhenAll(stdout, stderr, exited).WaitAsync(deadline.Token).ConfigureAwait(false);
|
||||
return await stdout.ConfigureAwait(false);
|
||||
}
|
||||
catch (OperationCanceledException)
|
||||
{
|
||||
try { if (!process.HasExited) process.Kill(entireProcessTree: true); }
|
||||
catch (InvalidOperationException) when (process.HasExited) { /* It exited between the check and kill. */ }
|
||||
await process.WaitForExitAsync(CancellationToken.None).WaitAsync(TimeSpan.FromSeconds(2)).ConfigureAwait(false);
|
||||
// Observe both cancelled drains; never leave a pipe read running after returning.
|
||||
try { await Task.WhenAll(stdout, stderr).WaitAsync(TimeSpan.FromSeconds(2)).ConfigureAwait(false); }
|
||||
catch (OperationCanceledException) { }
|
||||
if (ct.IsCancellationRequested) throw;
|
||||
throw new TimeoutException("Descriptor probe exceeded its process and redirected-stream deadline.");
|
||||
}
|
||||
}
|
||||
|
||||
private static bool IsWithinRoots(string? path, IReadOnlyList<string> roots)
|
||||
{
|
||||
if (path is null || !Path.IsPathFullyQualified(path)) return false;
|
||||
// procfs marks an unlinked but still open file; its containing physical directory still counts.
|
||||
const string deleted = " (deleted)";
|
||||
if (path.EndsWith(deleted, StringComparison.Ordinal)) path = path[..^deleted.Length];
|
||||
var canonical = CanonicalPath(path);
|
||||
return roots.Any(root => canonical.Equals(root, StringComparison.Ordinal)
|
||||
|| canonical.StartsWith(Path.EndsInDirectorySeparator(root) ? root : root + Path.DirectorySeparatorChar,
|
||||
StringComparison.Ordinal));
|
||||
}
|
||||
|
||||
private static string CanonicalPath(string path, int depth = 0)
|
||||
{
|
||||
if (depth > 32) throw new IOException("Descriptor path link resolution exceeded its bound.");
|
||||
var full = Path.GetFullPath(path);
|
||||
var current = Path.GetPathRoot(full)!;
|
||||
foreach (var component in full[current.Length..].Split(Path.DirectorySeparatorChar, StringSplitOptions.RemoveEmptyEntries))
|
||||
{
|
||||
current = Path.Combine(current, component);
|
||||
FileSystemInfo entry = Directory.Exists(current) ? new DirectoryInfo(current) : new FileInfo(current);
|
||||
if (entry.LinkTarget is not null)
|
||||
{
|
||||
var target = entry.ResolveLinkTarget(returnFinalTarget: true)
|
||||
?? throw new IOException("Descriptor path link resolution returned no target.");
|
||||
// Resolve parent links too: an alias target can itself begin with /var or /tmp.
|
||||
current = CanonicalPath(target.FullName, depth + 1);
|
||||
}
|
||||
}
|
||||
return Path.TrimEndingDirectorySeparator(current);
|
||||
}
|
||||
|
||||
}
|
||||
@@ -0,0 +1,102 @@
|
||||
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!;
|
||||
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()
|
||||
{
|
||||
await _host.DisposeAsync();
|
||||
Canonical.Ledger.Record("host-stop");
|
||||
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,252 @@
|
||||
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
|
||||
{
|
||||
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);
|
||||
foreach (var closed in ownedSessions)
|
||||
{
|
||||
var before = ContinuationCounts(closed);
|
||||
if (before != (0, 0))
|
||||
{
|
||||
await LogAsync(evidence, new { kind = "closed-session-observation", cycle,
|
||||
activeSessions = rig.Server.CurrentInstance.SessionManager.GetSessions().Count,
|
||||
browsePoints = before.Browse, historyPoints = before.History, diagnosticReferenceHeld = true });
|
||||
await Task.Delay(250, ct); // Read-only failure investigation; never turns the failed guard into a pass.
|
||||
var after = ContinuationCounts(closed);
|
||||
await LogAsync(evidence, new { kind = "closed-session-after-fixed-observation", cycle,
|
||||
observationMilliseconds = 250, browsePoints = after.Browse, historyPoints = after.History });
|
||||
}
|
||||
before.ShouldBe((0, 0));
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
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,19 @@
|
||||
using System.Diagnostics;
|
||||
using System.Reflection;
|
||||
using System.Text.Json;
|
||||
using Opc.Ua;
|
||||
using Opc.Ua.Server;
|
||||
using ZB.MOM.WW.OtOpcUa.OpcUaServer;
|
||||
var types = new[] { typeof(Session), typeof(CertificateValidator), typeof(CertificateStoreIdentifier),
|
||||
typeof(Opc.Ua.Configuration.ApplicationInstance), typeof(OtOpcUaSdkServer), typeof(OpcUaApplicationHost), typeof(OtOpcUaNodeManager) };
|
||||
Console.WriteLine(JsonSerializer.Serialize(new
|
||||
{
|
||||
sdk = typeof(NodeId).Assembly.GetCustomAttribute<AssemblyInformationalVersionAttribute>()?.InformationalVersion,
|
||||
sdkFile = FileVersionInfo.GetVersionInfo(typeof(NodeId).Assembly.Location).FileVersion,
|
||||
modules = types.Select(t => new { type = t.FullName, assembly = t.Assembly.GetName().Name, mvid = t.Module.ModuleVersionId,
|
||||
interfaces = t.GetInterfaces().Select(x => x.FullName),
|
||||
fields = t.GetFields(BindingFlags.NonPublic | BindingFlags.Public | BindingFlags.Instance | BindingFlags.DeclaredOnly)
|
||||
.Select(f => new { name = f.Name, type = f.FieldType.FullName }),
|
||||
lifecycle = t.GetMethods(BindingFlags.Public | BindingFlags.Instance).Where(m => m.Name.Contains("Dispose") || m.Name.Contains("Close") || m.Name.Contains("Stop"))
|
||||
.Select(m => m.ToString()) })
|
||||
}, new JsonSerializerOptions { WriteIndented = true }));
|
||||
@@ -0,0 +1,210 @@
|
||||
using System.Collections.Concurrent;
|
||||
using System.Diagnostics;
|
||||
using System.Net;
|
||||
using System.Net.Sockets;
|
||||
using System.Reflection;
|
||||
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.RunAsync(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!;
|
||||
internal static async Task<int> RunAsync(string[] args)
|
||||
{
|
||||
var root=Path.GetFullPath(args[0]); var cycles=int.Parse(args[1]);
|
||||
var isolatedRestart = args.Length == 3 && args[2] == "--active-restart-control";
|
||||
if (args.Length > 3 || (args.Length == 3 && !isolatedRestart)) throw new InvalidOperationException("Unknown short-control mode");
|
||||
if(cycles<1 || cycles>4) throw new InvalidOperationException("Only a 1..4-cycle short diagnostic smoke is authorized");
|
||||
if(Directory.Exists(root)) throw new InvalidOperationException("Use a fresh private artifact directory");
|
||||
Directory.CreateDirectory(root);
|
||||
if(!OperatingSystem.IsWindows()) File.SetUnixFileMode(root,UnixFileMode.UserRead|UnixFileMode.UserWrite|UnixFileMode.UserExecute);
|
||||
await using var output=new StreamWriter(new FileStream(Path.Combine(root,"resources.jsonl"),FileMode.CreateNew,FileAccess.Write,FileShare.Read));
|
||||
CurrentEvidence=output;
|
||||
using var deadline=new CancellationTokenSource(TimeSpan.FromSeconds(120)); var ct=deadline.Token;
|
||||
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 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,ct);await application.CheckApplicationInstanceCertificatesAsync(true,null,ct);
|
||||
await using var rig=new BaselineRig(Path.Combine(root,"private-fixture"));
|
||||
await rig.BootAsync(ct);Configure(rig);
|
||||
var watch=Stopwatch.StartNew();var completed=0;var restarts=0;
|
||||
await LogAsync(output,new {kind="start",sdk,productRebuilt=false,sourcePublication="83ed880e",requestedCycles=cycles,
|
||||
processId=Environment.ProcessId,acceptance=false,scope="canonical operations short smoke; ownership differences remain explicit",
|
||||
unsupportedObservables=new[]{"SDK2 CertificateManager disposed flag","SDK2 manager-owned cache-clear proof","ApplicationInstance.DisposeAsync"}});
|
||||
try
|
||||
{
|
||||
if (isolatedRestart)
|
||||
{
|
||||
if (cycles != 1) throw new InvalidOperationException("The isolated restart control performs exactly one restart");
|
||||
await RestartWithActiveResourcesAsync(rig,config,ct);Configure(rig);restarts++;
|
||||
await LogAsync(output,new {kind="isolated-active-restart-completed",normalCycles=0,restarts,
|
||||
acceptance=false,ownershipParity=false,ledger=Ledger.Snapshot()});
|
||||
return 2;
|
||||
}
|
||||
while(completed<cycles)
|
||||
{
|
||||
await ExerciseAsync(rig,config,completed+1,output,ct);completed++;
|
||||
await EventuallyAsync(()=>PublicCounts(rig)==(0,0,0),ct);
|
||||
var counts=PublicCounts(rig);var fd=await BaselineMetrics.DescriptorsAsync(ct,clientRoot,rig.PkiRoot);
|
||||
fd.Pki.ShouldBe(0);
|
||||
foreach(var name in new[]{"LiveItemCount","LiveEventItemCount","MonitoredNodeCount"}) BaselineMetrics.Counter(rig.Nm,name).ShouldBe(0);
|
||||
using var process=Process.GetCurrentProcess();process.Refresh();
|
||||
await LogAsync(output,new {kind="quiescent",cycle=completed,seconds=watch.Elapsed.TotalSeconds,
|
||||
sessions=counts.Sessions,subscriptions=counts.Subscriptions,items=counts.Items,
|
||||
pkiDescriptors=fd.Pki,descriptors=fd.Total,managedBytes=GC.GetTotalMemory(true),rssBytes=process.WorkingSet64,
|
||||
serverCachedStoreReferences=BaselineMetrics.CachedStoreReferences(rig.Validator),
|
||||
clientCachedStoreReferences=BaselineMetrics.CachedStoreReferences(config.CertificateValidator),jit=System.Runtime.JitInfo.GetCompiledMethodCount(),timers=Timer.ActiveCount});
|
||||
if(completed%4==0){await RestartWithActiveResourcesAsync(rig,config,ct);Configure(rig);restarts++;}
|
||||
}
|
||||
await LogAsync(output,new {kind="operations-completed",cycles=completed,restarts,seconds=watch.Elapsed.TotalSeconds,
|
||||
acceptance=false,ownershipParity=false,reason="Original SDK has no equivalent manager-disposal observable; review required",ledger=Ledger.Snapshot()});
|
||||
return 2;
|
||||
}
|
||||
catch(Exception error)
|
||||
{
|
||||
await LogAsync(output,new {kind="failed",cycles=completed,restarts,exceptionType=error.GetType().FullName,
|
||||
uaStatus=error is ServiceResultException ua ? ua.StatusCode.ToString() : null,ledger=Ledger.Snapshot(),acceptance=false});
|
||||
Console.Error.WriteLine(error);return 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 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)
|
||||
{
|
||||
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,124 @@
|
||||
# Original 1.5.378.182 canonical adapter — partial diagnostic proof
|
||||
|
||||
This folder compiles a standalone diagnostic against the **unchanged original
|
||||
83ed880e published host/client DLLs**. It does not rebuild or patch product code,
|
||||
restore a substitute SDK, edit a baseline oracle, install a replacement session
|
||||
factory, or approve a performance denominator. Only short 1–4-cycle smoke runs are
|
||||
implemented. No original warmed/full/two-hour run has been performed.
|
||||
|
||||
The immutable SDK reports `1.5.378.182+6cca32dd6d79c8b57d2e00e4d2071b991d2cf71b`.
|
||||
The source workload was ported from `Sdk2LifecycleSoakTests` at `75057d8c`.
|
||||
`evidence.json` records source/input hashes, exact SDK layout and short-run results.
|
||||
Raw logs and task-owned PKI stay outside the repository; credentials are not logged.
|
||||
|
||||
## Supported operation coverage
|
||||
|
||||
`CanonicalOperations.cs` preserves the candidate's operation ordering, status
|
||||
assertions, timeouts, model, grants, event data, page size and typed-array shape.
|
||||
The representation changes are explicit: SDK2 Variant/ArrayOf helpers become the
|
||||
original Variant constructor and CLR values; ByteString becomes byte[]; and
|
||||
QualifiedName.From becomes its original constructor. There are no collection
|
||||
shims or private-state mutations.
|
||||
|
||||
| Canonical operation per normal cycle | Original short control observation |
|
||||
|---|---|
|
||||
| Two encrypted opc.tcp / SignAndEncrypt sessions, Raw/UNS URI binding | Completed |
|
||||
| One subscription, three data items, notifications | Completed |
|
||||
| One Write service with Int32/Int16[]/UNS Int32 values, three readbacks | Completed with original Good/type assertions |
|
||||
| Two browse pages, one explicit release and one abandonment | Completed; active browse count 1 |
|
||||
| History first page, foreign-session denial, resume, explicit release, six abandoned points | All 10 HistoryRead services returned; active history count 6 |
|
||||
| Writable off/on, array removal/recreation/publication | Completed |
|
||||
| Key revocation, denied read, authority restoration | Completed |
|
||||
| Paused TCP read, one-second cancellation, resume and successful recovery read | Completed; original UA status 2156134400 on the canceled read |
|
||||
| Subscription deletion/removal and address-space rebuild/reconfiguration | Completed |
|
||||
| Both normal clients CloseAsync | Both returned; version-specific post-close list comparison differs |
|
||||
| Every-fourth-cycle restart with an active session/subscription/item, browse/history points | Separate isolated one-restart control completed |
|
||||
|
||||
The failed normal control attempted one whole wire body. Its ledger records
|
||||
2 connects, 1 subscription create/apply/delete/remove, 1 Write service (3 values),
|
||||
5 ReadValue attempts (4 returned, 1 expected canceled fault), 1 revoked Read
|
||||
service, 2 Browse calls, 1 BrowseNext release, 10 HistoryRead calls, and 2 CloseAsync
|
||||
returns. Surgery, revocation, stalled-network recovery and rebuild all occurred
|
||||
before the post-close comparison. No normal cycle is reported as ownership-complete.
|
||||
|
||||
The independent active-restart control does **not** stand in for a successful
|
||||
four-cycle sequence: it starts one client/subscription/item, abandons browse and
|
||||
history points, stops/reboots the same URI/port/PKI server, and checks the captured
|
||||
old managers/sessions/subscription items and new manager counts. Those checks
|
||||
completed. It returned **exit 2**, explicitly unsupported ownership parity,
|
||||
not a green acceptance result.
|
||||
|
||||
## Close and ownership differences — unresolved
|
||||
|
||||
The first normal smoke returned **exit 1** at the copied zero-closed-list
|
||||
comparison: a retained server Session still had **1 browse / 6 history** points.
|
||||
The server SessionManager already had **zero active sessions**. A separate fixed
|
||||
250 ms read-only observation saw the same lists; the original failed comparison
|
||||
was still enforced. The first failure artifacts were left untouched.
|
||||
|
||||
Exact immutable-binary inspection explains the internal difference:
|
||||
|
||||
- SessionManager.CloseSession removes the session and calls Session.Close.
|
||||
- Session.Close only deletes its diagnostics; it does not clear continuation lists.
|
||||
- Session.Dispose clears both lists and disposes their retained values.
|
||||
- SessionManager.Shutdown/Dispose disposes sessions still in its active dictionary.
|
||||
|
||||
These are **version-specific diagnostics**, not proof that the original violates
|
||||
the wire close contract, has an unbounded leak, or retains the same objects once
|
||||
all strong diagnostic references are dropped. The adapter deliberately holds the
|
||||
old Session references while checking these fields. The SDK2 zero-list invariant
|
||||
cannot silently become a shared cross-version acceptance contract.
|
||||
|
||||
The restart control follows the different active-session shutdown path: captured
|
||||
old session/point lists and subscription items became empty. The original
|
||||
validator's cached store **references** remained 1 before and after stop. A cached
|
||||
store reference is not a count of open handles or proof of a leak.
|
||||
|
||||
Original ApplicationInstance and CertificateValidator expose no Dispose API, and
|
||||
there is no SDK2 CertificateManager/m_disposed/core-store-clear equivalent. The
|
||||
adapter reports those observables as **unsupported**, never as zero. Actual PKI
|
||||
file descriptors can be measured with the same bounded/canonical-path observer
|
||||
as the candidate; those observations cannot invent the missing manager ownership
|
||||
proof. No manual Session.Dispose, store replacement or cache clearing was added
|
||||
to make an assertion pass.
|
||||
|
||||
Before any common lifecycle/acceptance protocol can be proposed, a separately
|
||||
reviewed short control must establish original SessionManager removal, rejection
|
||||
of the old authentication token, rejection of old CP tokens from a fresh session,
|
||||
then drop every strong session/point/test reference at an isolated helper boundary
|
||||
and perform bounded GC/weak-reference observation. Any retained IDisposable point
|
||||
resource needs direct source/ownership inspection. Certificate-store/native
|
||||
ownership gaps must remain explicit. This proposal has **not** been executed.
|
||||
|
||||
The standalone adapter also lacks the candidate's xUnit/VSTest host; equal process
|
||||
memory overhead is not proven. Earlier original-subset managed/timer growth is an
|
||||
observation, not an approved RSS denominator or a proven SDK1 defect. Resource and
|
||||
operation parity require Astra/root review before any long run; no cap, baseline
|
||||
or acceptance protocol changes are included here.
|
||||
|
||||
## Build and short controls
|
||||
|
||||
`build.py` selects original DLLs from the provided immutable directories (host
|
||||
wins duplicate names), plus Shouldly 4.3.0/assertion dependencies only. It emits
|
||||
SHA-256 provenance and a new external project. No production project/package file
|
||||
is touched. The inventory lists exact types, fields, lifecycle APIs and MVIDs
|
||||
without creating product instances.
|
||||
|
||||
```bash
|
||||
python3 scripts/validation/opcua-lifecycle-memory/baseline-canonical/build.py \
|
||||
--host "$originalHost" --client "$originalClient" --assertions "$assertionDlls" \
|
||||
--output /tmp/original-canonical-build
|
||||
python3 scripts/validation/opcua-lifecycle-memory/baseline-canonical/run_smoke.py \
|
||||
--adapter /tmp/original-canonical-build/bin/Release/net10.0/adapter.dll \
|
||||
--output /tmp/original-canonical-close --mode normal
|
||||
python3 scripts/validation/opcua-lifecycle-memory/baseline-canonical/run_smoke.py \
|
||||
--adapter /tmp/original-canonical-build/bin/Release/net10.0/adapter.dll \
|
||||
--output /tmp/original-canonical-active-restart --mode active-restart
|
||||
```
|
||||
|
||||
Add `--inventory` to build the read-only metadata inventory instead. The runner
|
||||
uses a fresh private directory and a temporary environment credential with the
|
||||
same prefix/byte length as the canonical fixture; it preserves the original child
|
||||
exit. Exit 1 is an observed control failure; exit 2 is operations completed with
|
||||
unsupported ownership parity. Neither is a passing baseline. The smoke driver
|
||||
has a hard 120-second bound and no long-run option.
|
||||
@@ -0,0 +1,43 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Compile diagnostic adapters against the immutable original publication; never build product code."""
|
||||
import argparse
|
||||
import hashlib
|
||||
import json
|
||||
from pathlib import Path
|
||||
import subprocess
|
||||
import xml.etree.ElementTree as ET
|
||||
|
||||
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, help='Directory containing Shouldly.dll; only assertion dependencies are selected')
|
||||
parser.add_argument('--output', type=Path, required=True)
|
||||
parser.add_argument('--inventory', action='store_true')
|
||||
args = parser.parse_args()
|
||||
root = args.output.resolve(); root.mkdir(parents=True, exist_ok=False)
|
||||
source = Path(__file__).resolve().parent
|
||||
references = {}
|
||||
for directory in [args.host, args.client]:
|
||||
for dll in sorted(directory.resolve(strict=True).glob('*.dll')):
|
||||
references.setdefault(dll.name, dll)
|
||||
for name in ['Shouldly.dll', 'DiffEngine.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'}.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)
|
||||
files = ['Inventory.cs'] if args.inventory else ['Program.cs','BaselineRig.cs','BaselineMetrics.cs','CanonicalOperations.cs']
|
||||
for name in files: ET.SubElement(items,'Compile',Include=str(source/name))
|
||||
path=root/'adapter.csproj'; ET.indent(project); ET.ElementTree(project).write(path,encoding='unicode')
|
||||
manifest={'diagnosticOnly':True,'productRebuilt':False,'references':{name:hashlib.sha256(p.read_bytes()).hexdigest() for name,p in references.items()},
|
||||
'sources':{name:hashlib.sha256((source/name).read_bytes()).hexdigest() for name in files}}
|
||||
(root/'provenance.json').write_text(json.dumps(manifest,indent=2)+'\n')
|
||||
with (root/'build.log').open('w') as log:
|
||||
result=subprocess.run(['dotnet','build',str(path),'-c','Release','-m:2'],stdout=log,stderr=subprocess.STDOUT)
|
||||
raise SystemExit(result.returncode)
|
||||
File diff suppressed because it is too large
Load Diff
@@ -0,0 +1,29 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Run one bounded original-only diagnostic control; preserve failures and unsupported exit 2."""
|
||||
import argparse
|
||||
import json
|
||||
import os
|
||||
from pathlib import Path
|
||||
import secrets
|
||||
import subprocess
|
||||
|
||||
parser=argparse.ArgumentParser(description=__doc__)
|
||||
parser.add_argument('--adapter',type=Path,required=True)
|
||||
parser.add_argument('--output',type=Path,required=True)
|
||||
parser.add_argument('--mode',choices=['normal','active-restart'],required=True)
|
||||
args=parser.parse_args()
|
||||
output=args.output.resolve()
|
||||
if output.exists():raise RuntimeError('Use a fresh private output directory')
|
||||
log=output.with_suffix('.log')
|
||||
environment=os.environ.copy()
|
||||
environment['OTOPCUA_BASELINE_CANONICAL_KEY']='otua_k-server-events_'+secrets.token_hex(3)
|
||||
command=['dotnet',str(args.adapter.resolve(strict=True)),str(output),'1']
|
||||
if args.mode=='active-restart':command+=['--active-restart-control']
|
||||
with os.fdopen(os.open(log,os.O_WRONLY|os.O_CREAT|os.O_EXCL,0o600),'w') as stream:
|
||||
try:result=subprocess.run(command,env=environment,stdout=stream,stderr=subprocess.STDOUT,timeout=120);exit_code=result.returncode
|
||||
except subprocess.TimeoutExpired:exit_code=124
|
||||
if output.exists():
|
||||
(output/'runner-receipt.json').write_text(json.dumps({'mode':args.mode,'exitCode':exit_code,'acceptance':False,
|
||||
'productRebuilt':False,'timeoutSeconds':120,'credentialBytes':len(environment['OTOPCUA_BASELINE_CANONICAL_KEY'].encode())},indent=2)+'\n')
|
||||
print(json.dumps({'mode':args.mode,'exitCode':exit_code,'acceptance':False}))
|
||||
raise SystemExit(exit_code)
|
||||
Reference in New Issue
Block a user