Joseph Doherty
253 lines
16 KiB
C#
253 lines
16 KiB
C#
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];
|
|
|
|
}
|