fix(opcua): preserve repeated alarm methods and owned write reverts

This commit is contained in:
Joseph Doherty
2026-11-10 11:00:31 -05:00
parent 8c422b9ccc
commit 09ac0e58cb
6 changed files with 414 additions and 28 deletions
@@ -0,0 +1,63 @@
using Opc.Ua;
namespace ZB.MOM.WW.OtOpcUa.OpcUaServer;
/// <summary>Preserves the application's repeat-method contract at the SDK's supported validation seam.</summary>
internal static class ApplicationAlarmConditions
{
internal static AlarmConditionState Create(string type, NodeState parent) => type switch
{
"OffNormalAlarm" => new OffNormal(parent),
"DiscreteAlarm" => new Discrete(parent),
_ => new Alarm(parent),
};
private static ServiceResult CurrentTrunkRepeat(ServiceResult result, AcknowledgeableConditionState condition,
ISystemContext context, ByteString eventId, LocalizedText comment, bool acknowledge)
{
var repeated = acknowledge ? StatusCodes.BadConditionBranchAlreadyAcked : StatusCodes.BadConditionBranchAlreadyConfirmed;
var callback = acknowledge ? condition.OnAcknowledge : condition.OnConfirm;
if (result.StatusCode != repeated || callback is null || eventId.IsEmpty
|| condition.BranchId is not { Value.IsNull: true }
|| condition.EventId is not { } current || current.Value != eventId
|| !ReferenceEquals(condition.GetEventByEventId(eventId), condition)
|| (acknowledge ? condition.AckedState?.Id?.Value : condition.ConfirmedState?.Id?.Value) != true)
return result;
// In exact SDK 2.0.1, the Already* guards precede the application callback. Invoke it once
// here; every other SDK validation result, superseded EventId and branch retains the base path.
try { return callback(context, condition, eventId, comment); }
catch (Exception error)
{
return ServiceResult.Create(error, StatusCodes.BadUnexpectedError,
acknowledge ? "Unexpected error acknowledging a Condition." : "Unexpected error confirming a Condition.");
}
}
private sealed class OffNormal(NodeState parent) : OffNormalAlarmState(parent)
{
protected override ServiceResult ProcessBeforeAcknowledge(ISystemContext context, ByteString id, LocalizedText comment) =>
CurrentTrunkRepeat(base.ProcessBeforeAcknowledge(context, id, comment), this, context, id, comment, true);
protected override ServiceResult ProcessBeforeConfirm(ISystemContext context, ByteString id, LocalizedText comment) =>
CurrentTrunkRepeat(base.ProcessBeforeConfirm(context, id, comment), this, context, id, comment, false);
protected override ConditionState CreateBranchInstance() => new OffNormal(this);
}
private sealed class Discrete(NodeState parent) : DiscreteAlarmState(parent)
{
protected override ServiceResult ProcessBeforeAcknowledge(ISystemContext context, ByteString id, LocalizedText comment) =>
CurrentTrunkRepeat(base.ProcessBeforeAcknowledge(context, id, comment), this, context, id, comment, true);
protected override ServiceResult ProcessBeforeConfirm(ISystemContext context, ByteString id, LocalizedText comment) =>
CurrentTrunkRepeat(base.ProcessBeforeConfirm(context, id, comment), this, context, id, comment, false);
protected override ConditionState CreateBranchInstance() => new Discrete(this);
}
private sealed class Alarm(NodeState parent) : AlarmConditionState(parent)
{
protected override ServiceResult ProcessBeforeAcknowledge(ISystemContext context, ByteString id, LocalizedText comment) =>
CurrentTrunkRepeat(base.ProcessBeforeAcknowledge(context, id, comment), this, context, id, comment, true);
protected override ServiceResult ProcessBeforeConfirm(ISystemContext context, ByteString id, LocalizedText comment) =>
CurrentTrunkRepeat(base.ProcessBeforeConfirm(context, id, comment), this, context, id, comment, false);
protected override ConditionState CreateBranchInstance() => new Alarm(this);
}
}
@@ -8,6 +8,32 @@ namespace ZB.MOM.WW.OtOpcUa.OpcUaServer;
/// </summary>
internal sealed class EquipmentVariableState(NodeState? parent, object ownerLock) : BaseDataVariableState(parent)
{
// Every field below is accessed under ownerLock. Tokens identify client attempts; the mutation
// revision detects a driver publication that re-enters the callback before its optimistic commit.
private long _nextWriteToken;
private long _mutationRevision;
private long? _activeWriteToken;
private long? _committedWriteToken;
internal readonly record struct WriteStamp(EquipmentVariableState Owner, long Token);
internal WriteStamp? ActiveWriteStamp => _activeWriteToken is { } token ? new(this, token) : null;
internal bool OwnsWrite(WriteStamp stamp) => ReferenceEquals(this, stamp.Owner) && _committedWriteToken == stamp.Token;
internal void InvalidateWriteOwnership()
{
_mutationRevision++;
_committedWriteToken = null;
}
internal void PublishValue(Variant value, StatusCode status, DateTimeUtc timestamp)
{
InvalidateWriteOwnership();
Value = value;
StatusCode = status;
Timestamp = timestamp;
ChangeMasks |= NodeStateChangeMasks.Value;
}
/// <inheritdoc />
protected override ServiceResult WriteValueAttribute(ISystemContext context, NumericRange indexRange,
Variant value, StatusCode statusCode, DateTimeUtc sourceTimestamp)
@@ -33,14 +59,20 @@ internal sealed class EquipmentVariableState(NodeState? parent, object ownerLock
if ((userAccess & AccessLevels.TimestampWrite) == 0) return StatusCodes.BadUserAccessDenied;
}
var result = write(context, this, indexRange, default, ref value, ref statusCode, ref sourceTimestamp);
var mutationBefore = _mutationRevision;
var token = ++_nextWriteToken;
var previousAttempt = _activeWriteToken;
_activeWriteToken = token;
ServiceResult result;
try { result = write(context, this, indexRange, default, ref value, ref statusCode, ref sourceTimestamp); }
finally { _activeWriteToken = previousAttempt; }
if (ServiceResult.IsBad(result)) return result;
// A reentrant driver update is newer than this attempt. Do not overwrite its tuple.
if (_mutationRevision != mutationBefore) return result;
// Value's first-touch setter can reset status. Commit in this order, under the shared lock.
Value = value;
StatusCode = statusCode;
Timestamp = sourceTimestamp == DateTimeUtc.MinValue ? DateTimeUtc.Now : sourceTimestamp;
ChangeMasks |= NodeStateChangeMasks.Value;
PublishValue(value, statusCode, sourceTimestamp == DateTimeUtc.MinValue ? DateTimeUtc.Now : sourceTimestamp);
_committedWriteToken = token;
return result;
}
}
@@ -1046,6 +1046,7 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2
// Value deliberately untouched — that IS the feature: the node keeps its last known value under a
// Bad status (OPC UA Part 8's last-known-value convention).
if (variable is EquipmentVariableState equipment) equipment.InvalidateWriteOwnership();
variable.StatusCode = new StatusCode(statusCode);
variable.Timestamp = sourceTimestampUtc;
variable.ClearChangeMasks(SystemContext, includeChildren: false);
@@ -1076,7 +1077,9 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2
{
if (!_variables.TryGetValue(key, out var variable)) return false;
variable.Value = SdkValueBoundary.FromClr(value, variable.DataType == DataTypeIds.Byte && variable.ValueRank == ValueRanks.OneDimension);
var payload = SdkValueBoundary.FromClr(value, variable.DataType == DataTypeIds.Byte && variable.ValueRank == ValueRanks.OneDimension);
if (variable is EquipmentVariableState equipment) equipment.InvalidateWriteOwnership();
variable.Value = payload;
// The caller's status code reaches the SDK verbatim, sub-code included. This used to
// re-expand a 3-state enum, so every Bad became a bare StatusCodes.Bad (0x80000000) and a
// driver's BadNoCommunication was indistinguishable from BadTypeMismatch at the client.
@@ -2347,7 +2350,8 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2
// presents ("apikey"), so an audit trail built on it cannot tell two consuming systems apart.
var clientUserId = identity is { Principal.Length: > 0 } ? identity.Principal : identity?.DisplayName;
return DispatchDeviceWrite(gateway, nodeKey, nodeBareId, nodeRealm, optimisticValue, priorValue, priorStatus, clientUserId);
return DispatchDeviceWrite(gateway, nodeKey, nodeBareId, nodeRealm, optimisticValue, priorValue, priorStatus, clientUserId,
(node as EquipmentVariableState)?.ActiveWriteStamp);
}
/// <summary>
@@ -2358,16 +2362,17 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2
/// here (review 01 L3 — it used to surface only as an <c>UnobservedTaskException</c> at finalization, with
/// the constant reason "write dispatch faulted" in the audit event). A gateway that throws SYNCHRONOUSLY
/// (review 01 H5) is refused inline with <c>BadDeviceFailure</c> before the SDK applies anything, rather
/// than escaping the handler under <c>Lock</c>. <c>RunContinuationsAsynchronously</c> guarantees the revert
/// never runs inline on the SDK write thread (the gateway can return a synchronously-completed task — e.g.
/// its boot-window "no DriverHostActor yet" branch), so <see cref="RevertOptimisticWriteIfNeeded"/> never
/// re-enters <c>Lock</c> while <c>CustomNodeManager2.Write</c> still holds it.
/// than escaping the handler under <c>Lock</c>. The default-scheduler continuation is queued without
/// <c>ExecuteSynchronously</c>; its manager-lock acquisition follows the optimistic commit. Production
/// corrections additionally require the exact node instance and committed write token, so a newer
/// equal-valued write or publication cannot be mistaken for this attempt.
/// </summary>
/// <returns><see cref="ServiceResult.Good"/> once the write is in flight (the SDK then applies the value
/// optimistically); a Bad result when the dispatch itself threw.</returns>
internal ServiceResult DispatchDeviceWrite(
IOpcUaNodeWriteGateway gateway, string nodeKey, string nodeBareId, AddressSpaceRealm nodeRealm,
object? optimisticValue, object? priorValue, StatusCode priorStatus, string? clientUserId)
object? optimisticValue, object? priorValue, StatusCode priorStatus, string? clientUserId,
EquipmentVariableState.WriteStamp? writeStamp = null)
{
Task<NodeWriteOutcome> dispatch;
try
@@ -2398,7 +2403,7 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2
outcome = new NodeWriteOutcome(false, reason);
}
RevertOptimisticWriteIfNeeded(nodeBareId, outcome, optimisticValue, priorValue, priorStatus, clientUserId, nodeRealm);
RevertOptimisticWriteIfNeeded(nodeBareId, outcome, optimisticValue, priorValue, priorStatus, clientUserId, nodeRealm, writeStamp);
},
CancellationToken.None, TaskContinuationOptions.RunContinuationsAsynchronously, TaskScheduler.Default);
@@ -2774,8 +2779,8 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2
/// items, and any failure is swallowed (logged to the SDK trace) so the revert is never broken by it.
/// </item>
/// </list>
/// Silent value-wise — this node manager carries no logger; the gateway logs the underlying write failure
/// and the SDK trace captures any audit-report failure.
/// A superseded production outcome is logged at Debug; the gateway records the underlying device failure
/// and audit construction/reporting failures are logged here.
/// <para><c>internal</c> (not private) so the behavioural test drives the blip-then-settle continuation
/// against a booted node manager — see <c>NodeManagerWriteRevertTests</c>.</para>
/// </summary>
@@ -2787,9 +2792,10 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2
/// <param name="clientUserId">The writing principal's user-id string (the identity's DisplayName), threaded
/// from <see cref="OnEquipmentTagWrite"/> to populate the audit event's ClientUserId; null when unknown.</param>
/// <param name="realm">The address-space realm (Raw or Uns) the written node lives in.</param>
/// <param name="writeStamp">The actual production node and accepted write attempt; absent only for legacy direct tests.</param>
internal void RevertOptimisticWriteIfNeeded(
string nodeId, NodeWriteOutcome outcome, object? optimisticValue, object? priorValue, StatusCode priorStatus,
string? clientUserId, AddressSpaceRealm realm)
string? clientUserId, AddressSpaceRealm realm, EquipmentVariableState.WriteStamp? writeStamp = null)
{
// Built under Lock if (and only if) a revert is performed, then reported AFTER Lock is released.
AuditWriteUpdateEventState? auditEvent = null;
@@ -2798,7 +2804,18 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2
lock (Lock)
{
if (!_variables.TryGetValue(key, out var variable)) return; // rebuilt/removed ⇒ no-op
if (!ShouldRevert(outcome, SdkValueBoundary.ToClr(variable.Value), optimisticValue)) return; // success, or poll moved it on
if (outcome.Success) return;
if (writeStamp is { } stamp)
{
if (variable is not EquipmentVariableState equipment || !equipment.OwnsWrite(stamp))
{
_logger.LogDebug("OtOpcUaNodeManager: ignored superseded write outcome for {NodeId}", nodeId);
return;
}
}
else if (!ShouldRevert(outcome, SdkValueBoundary.ToClr(variable.Value), optimisticValue)) return;
// The value guard above is retained only for legacy direct unit calls. Production writes
// carry exact node-instance/token ownership, so copied arrays and equal newer values are safe.
// Item B: surface a transient Bad-quality blip on the still-applied optimistic value, then restore
// the prior value/status. Both publishes are under this single Lock hold (mirrors WriteValue's
@@ -2808,7 +2825,9 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2
variable.Timestamp = now;
variable.ClearChangeMasks(SystemContext, includeChildren: false); // notify — the Bad blip
variable.Value = SdkValueBoundary.FromClr(priorValue, variable.DataType == DataTypeIds.Byte && variable.ValueRank == ValueRanks.OneDimension);
var prior = SdkValueBoundary.FromClr(priorValue, variable.DataType == DataTypeIds.Byte && variable.ValueRank == ValueRanks.OneDimension);
if (variable is EquipmentVariableState restored) restored.InvalidateWriteOwnership();
variable.Value = prior;
variable.StatusCode = priorStatus;
variable.Timestamp = now;
variable.ClearChangeMasks(SystemContext, includeChildren: false); // notify — restore prior
@@ -2913,14 +2932,8 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2
/// <summary>Map our domain <c>AlarmType</c> string to the matching SDK condition subtype. Script
/// alarms have no OPC limit/setpoint values, so limit-style types fall back to the base
/// <see cref="AlarmConditionState"/> (see <see cref="MaterialiseAlarmCondition"/> remarks).</summary>
private static AlarmConditionState CreateAlarmConditionOfType(string alarmType, NodeState parent) => alarmType switch
{
"OffNormalAlarm" => new OffNormalAlarmState(parent),
"DiscreteAlarm" => new DiscreteAlarmState(parent),
// "LimitAlarm" / "AlarmCondition" / unknown ⇒ base: a script-driven alarm has no OPC limits
// to populate, so the limit subtypes would carry unset High/Low children.
_ => new AlarmConditionState(parent),
};
private static AlarmConditionState CreateAlarmConditionOfType(string alarmType, NodeState parent) =>
ApplicationAlarmConditions.Create(alarmType, parent);
/// <summary>
/// The <c>AlarmConditionType_ShelvingState_*</c> method instance declarations a client calls
@@ -3323,6 +3336,7 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2
: default;
// Drop the stale (old-typed) value — mirrors EnsureVariable's fresh-node state so a client
// never reads a value whose runtime type contradicts the just-changed DataType.
if (v is EquipmentVariableState equipment) equipment.InvalidateWriteOwnership();
v.Value = Variant.Null;
v.StatusCode = StatusCodes.BadWaitingForInitialData;
v.Timestamp = DateTime.MinValue;
@@ -0,0 +1,158 @@
using Microsoft.Extensions.Logging.Abstractions;
using Opc.Ua;
using Opc.Ua.Server;
using Shouldly;
using Xunit;
using ZB.MOM.WW.OtOpcUa.Commons.OpcUa;
using ZB.MOM.WW.OtOpcUa.OpcUaServer.Security;
namespace ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests;
public sealed class Sdk2ConditionMethodCompatibilityTests : IAsyncLifetime
{
private readonly string _pki = Path.Combine(Path.GetTempPath(), $"otua-sdk2-methods-{Guid.NewGuid():N}");
private OpcUaApplicationHost _host = null!;
private OtOpcUaSdkServer _server = null!;
private OtOpcUaNodeManager _manager = null!;
private readonly List<AlarmCommand> _scripted = [];
private readonly List<NativeAlarmAck> _native = [];
public async ValueTask InitializeAsync()
{
var started = await OpcUaHostStart.StartAsync(port => new OpcUaApplicationHost(new OpcUaApplicationHostOptions
{
ApplicationName = "SDK2.MethodContracts", ApplicationUri = "urn:SDK2.MethodContracts",
OpcUaPort = port, PublicHostname = "localhost", PkiStoreRoot = _pki,
}, NullLogger<OpcUaApplicationHost>.Instance), () => new OtOpcUaSdkServer(), TestContext.Current.CancellationToken);
_host = started.Host;
_server = started.Server;
_manager = _server.NodeManager!;
_manager.AlarmCommandRouter = _scripted.Add;
_manager.NativeAlarmAckRouter = _native.Add;
}
public async ValueTask DisposeAsync()
{
await _host.DisposeAsync();
Directory.Delete(_pki, true);
}
[Theory]
[InlineData(false, null)]
[InlineData(false, "again")]
[InlineData(true, null)]
[InlineData(true, "again")]
public void Actual_SDK_method_preserves_current_trunk_reack_and_comment(bool native, string? comment)
{
var condition = Condition(native);
Call(condition, condition.Acknowledge!, condition.EventId!.Value, "first").ShouldBe(ServiceResult.Good);
if (!native) Call(condition, condition.Confirm!, condition.EventId.Value, "confirmed").ShouldBe(ServiceResult.Good);
var before = condition.EventId.Value;
Call(condition, condition.Acknowledge!, before, comment).ShouldBe(ServiceResult.Good);
condition.AckedState!.Id!.Value.ShouldBeTrue();
condition.ConfirmedState!.Id!.Value.ShouldBeTrue();
if (comment is null) condition.EventId.Value.ShouldBe(before);
else
{
condition.EventId.Value.ShouldNotBe(before);
condition.Comment!.Value.Text.ShouldBe(comment);
}
if (native) _native.Count.ShouldBe(2, "a native re-ack is an upstream retry");
else _scripted.Count(x => x.Operation == "Acknowledge").ShouldBe(2);
}
[Fact]
public void Actual_SDK_native_confirm_keeps_the_application_unsupported_status()
{
var condition = Condition(native: true);
var before = condition.EventId!.Value;
Call(condition, condition.Confirm!, before, null).StatusCode.ShouldBe(StatusCodes.BadNotSupported);
condition.ConfirmedState!.Id!.Value.ShouldBeTrue();
condition.EventId.Value.ShouldBe(before);
_native.ShouldBeEmpty();
_scripted.ShouldBeEmpty();
}
[Fact]
public void Current_trunk_reack_still_requires_app_authority_primary_and_native_preflight()
{
var condition = Condition(native: true);
Call(condition, condition.Acknowledge!, condition.EventId!.Value, null).ShouldBe(ServiceResult.Good);
var before = condition.EventId.Value;
Call(condition, condition.Acknowledge!, before, null, authorized: false).StatusCode.ShouldBe(StatusCodes.BadUserAccessDenied);
_manager.NativeAlarmAckPrimaryGate = () => false;
Call(condition, condition.Acknowledge!, before, null).StatusCode.ShouldBe(StatusCodes.BadInvalidState);
_manager.NativeAlarmAckPrimaryGate = () => true;
_manager.NativeAlarmAckPreflight = _ => "driver unavailable";
Call(condition, condition.Acknowledge!, before, null).StatusCode.ShouldBe(StatusCodes.BadUserAccessDenied);
condition.EventId.Value.ShouldBe(before);
_native.Count.ShouldBe(1);
}
[Theory]
[InlineData("empty")]
[InlineData("foreign")]
[InlineData("superseded")]
[InlineData("disabled")]
public void Actual_SDK_method_keeps_event_and_enabled_guards(string invalid)
{
var condition = Condition(native: false);
var original = condition.EventId!.Value;
Call(condition, condition.Acknowledge!, original, null).ShouldBe(ServiceResult.Good);
var current = condition.EventId.Value;
var token = invalid switch
{
"empty" => default,
"foreign" => ByteString.From(Guid.NewGuid().ToByteArray()),
"superseded" => original,
_ => current,
};
if (invalid == "disabled") condition.SetEnableState(Context(), false);
var routes = _scripted.Count;
ServiceResult.IsBad(Call(condition, condition.Acknowledge!, token, null)).ShouldBeTrue();
_scripted.Count.ShouldBe(routes);
condition.EventId.Value.ShouldBe(current);
}
[Theory]
[InlineData("OffNormalAlarm")]
[InlineData("DiscreteAlarm")]
[InlineData("AlarmCondition")]
public void An_acknowledged_branch_is_constructible_and_does_not_get_the_trunk_repeat_waiver(string type)
{
var condition = Condition(native: false, type);
var branch = condition.CreateBranch(Context(), new NodeId(Guid.NewGuid())).ShouldBeAssignableTo<AlarmConditionState>()!;
branch.EventId!.Value = ByteString.From(Guid.NewGuid().ToByteArray());
branch.SetAcknowledgedState(Context(), true);
var routes = _scripted.Count;
Call(condition, condition.Acknowledge!, branch.EventId!.Value, null).StatusCode.ShouldBe(StatusCodes.BadConditionBranchAlreadyAcked);
_scripted.Count.ShouldBe(routes);
}
private AlarmConditionState Condition(bool native, string type = "OffNormalAlarm")
{
var realm = native ? AddressSpaceRealm.Raw : AddressSpaceRealm.Uns;
_manager.EnsureFolder("eq", null, "Equipment", realm);
_manager.MaterialiseAlarmCondition("alarm", "eq", "Alarm", type, 700, realm, isNative: native);
_manager.WriteAlarmCondition("alarm", new AlarmConditionSnapshot(true, false, true, true,
AlarmShelvingKind.Unshelved, 700, "active"), DateTime.UtcNow, realm);
return _manager.TryGetAlarmCondition("alarm", realm)!;
}
private ServerSystemContext Context(bool authorized = true) => new(_server.CurrentInstance)
{
UserIdentity = new RoleCarryingUserIdentity(new AnonymousIdentityToken(),
authorized ? [OpcUaDataPlaneRoles.AlarmAck] : [], "operator", null),
};
private ServiceResult Call(AlarmConditionState condition, AddCommentMethodState method,
ByteString eventId, string? comment, bool authorized = true)
{
var argumentErrors = new List<ServiceResult>();
var result = method.Call(Context(authorized), condition.NodeId,
[Variant.From(eventId), Variant.From(comment is null ? default(LocalizedText) : new LocalizedText(comment))],
argumentErrors, new List<Variant>());
argumentErrors.All(ServiceResult.IsGood).ShouldBeTrue();
return result;
}
}
@@ -1,3 +1,4 @@
using Microsoft.Extensions.Logging;
using Microsoft.Extensions.Logging.Abstractions;
using Opc.Ua;
using Opc.Ua.Server;
@@ -15,6 +16,7 @@ public sealed class Sdk2EquipmentVariableTests : IAsyncLifetime
private OtOpcUaSdkServer _server = null!;
private OtOpcUaNodeManager _manager = null!;
private readonly RecordingGateway _gateway = new();
private readonly OutcomeLogger _outcomeLogger = new();
private static CancellationToken Ct => TestContext.Current.CancellationToken;
public async ValueTask InitializeAsync()
@@ -24,7 +26,7 @@ public sealed class Sdk2EquipmentVariableTests : IAsyncLifetime
{
ApplicationName = "SDK2.ValueContracts", ApplicationUri = "urn:SDK2.ValueContracts",
PublicHostname = "localhost", OpcUaPort = port, PkiStoreRoot = _pki,
}, NullLogger<OpcUaApplicationHost>.Instance), () => new OtOpcUaSdkServer(), Ct);
}, NullLogger<OpcUaApplicationHost>.Instance), () => new OtOpcUaSdkServer(_outcomeLogger), Ct);
_host = started.Host;
_server = started.Server;
_manager = _server.NodeManager!;
@@ -214,6 +216,123 @@ public sealed class Sdk2EquipmentVariableTests : IAsyncLifetime
scalarBytes.TypeInfo.ValueRank.ShouldBe(ValueRanks.Scalar);
}
[Theory]
[InlineData(AddressSpaceRealm.Raw)]
[InlineData(AddressSpaceRealm.Uns)]
public async Task An_immediate_failed_array_outcome_reverts_an_actual_node_write(AddressSpaceRealm realm)
{
var node = Node("Int16", array: true, realm: realm);
_manager.WriteValue("v", new short[] { 7, 8 }, OpcUaStatus.Good, DateTime.UtcNow, realm);
_manager.NodeWriteGateway = new OutcomeGateway(_ => Task.FromResult(new NodeWriteOutcome(false, "refused")));
Write(node, new short[] { 1, 2 }).ShouldBe(ServiceResult.Good);
var deadline = DateTime.UtcNow.AddSeconds(2);
while (DateTime.UtcNow < deadline && !SdkValueBoundary.ToClr(node.Value).ShouldBeOfType<short[]>().SequenceEqual(new short[] { 7, 8 }))
await Task.Delay(10, Ct);
SdkValueBoundary.ToClr(node.Value).ShouldBeOfType<short[]>().ShouldBe([7, 8]);
node.StatusCode.ShouldBe(StatusCodes.Good);
}
[Theory]
[InlineData("equal-client-write")]
[InlineData("equal-driver-poll")]
[InlineData("status-only")]
[InlineData("same-id-recreated")]
[InlineData("shape-update")]
public async Task An_old_failed_outcome_cannot_replace_a_newer_actual_node_tuple(string change)
{
var node = Node("Int16", array: true);
_manager.WriteValue("v", new short[] { 7, 8 }, OpcUaStatus.Good, DateTime.UtcNow, AddressSpaceRealm.Raw);
var first = new TaskCompletionSource<NodeWriteOutcome>(TaskCreationOptions.RunContinuationsAsynchronously);
var calls = 0;
_manager.NodeWriteGateway = new OutcomeGateway(_ => ++calls == 1
? first.Task : Task.FromResult(new NodeWriteOutcome(true, null)));
Write(node, new short[] { 1, 2 }).ShouldBe(ServiceResult.Good);
var stamp = new DateTime(2026, 11, 10, 12, 0, 0, DateTimeKind.Utc);
switch (change)
{
case "equal-client-write": Write(node, new short[] { 1, 2 }).ShouldBe(ServiceResult.Good); break;
case "equal-driver-poll":
_manager.WriteValue("v", new short[] { 1, 2 }, StatusCodes.UncertainLastUsableValue.Code, stamp, AddressSpaceRealm.Raw); break;
case "status-only":
_manager.WriteStatuses([new StatusWrite("v", StatusCodes.BadNoCommunication.Code, stamp, AddressSpaceRealm.Raw)]); break;
case "same-id-recreated":
_manager.RemoveVariableNode("v", AddressSpaceRealm.Raw).ShouldBeTrue();
node = Node("Int16", array: true);
Write(node, new short[] { 1, 2 }).ShouldBe(ServiceResult.Good); break;
case "shape-update":
_manager.UpdateTagAttributes("v", true, null, "Int32", true, 2, AddressSpaceRealm.Raw).ShouldBeTrue(); break;
}
var expected = node.Value;
var status = node.StatusCode;
var time = node.Timestamp;
first.SetResult(new NodeWriteOutcome(false, "older write failed"));
await _outcomeLogger.Ignored.Task.WaitAsync(TimeSpan.FromSeconds(5), Ct);
node.Value.ShouldBe(expected);
node.StatusCode.ShouldBe(status);
node.Timestamp.ShouldBe(time);
}
[Fact]
public async Task A_reentrant_driver_publication_wins_over_the_callback_optimistic_commit()
{
var node = Node("Int16", array: true);
_manager.WriteValue("v", new short[] { 7, 8 }, OpcUaStatus.Good, DateTime.UtcNow, AddressSpaceRealm.Raw);
var first = new TaskCompletionSource<NodeWriteOutcome>(TaskCreationOptions.RunContinuationsAsynchronously);
var stamp = new DateTime(2026, 11, 10, 12, 0, 0, DateTimeKind.Utc);
_manager.NodeWriteGateway = new OutcomeGateway(_ =>
{
_manager.WriteValue("v", new short[] { 1, 2 }, StatusCodes.UncertainLastUsableValue.Code, stamp, AddressSpaceRealm.Raw);
return first.Task;
});
Write(node, new short[] { 1, 2 }).ShouldBe(ServiceResult.Good);
node.StatusCode.ShouldBe(StatusCodes.UncertainLastUsableValue);
node.Timestamp.ToDateTime().ShouldBe(stamp);
first.SetResult(new NodeWriteOutcome(false, "old write failed"));
await _outcomeLogger.Ignored.Task.WaitAsync(TimeSpan.FromSeconds(5), Ct);
SdkValueBoundary.ToClr(node.Value).ShouldBeOfType<short[]>().ShouldBe([1, 2]);
node.StatusCode.ShouldBe(StatusCodes.UncertainLastUsableValue);
node.Timestamp.ToDateTime().ShouldBe(stamp);
}
[Theory]
[InlineData(false)]
[InlineData(true)]
public async Task A_refused_attempt_does_not_disown_an_older_committed_write(bool synchronousThrow)
{
var node = Node("Int16", array: true);
_manager.WriteValue("v", new short[] { 7, 8 }, OpcUaStatus.Good, DateTime.UtcNow, AddressSpaceRealm.Raw);
var first = new TaskCompletionSource<NodeWriteOutcome>(TaskCreationOptions.RunContinuationsAsynchronously);
_manager.NodeWriteGateway = new OutcomeGateway(_ => first.Task);
Write(node, new short[] { 1, 2 }).ShouldBe(ServiceResult.Good);
if (synchronousThrow)
{
_manager.NodeWriteGateway = new OutcomeGateway(_ => throw new InvalidOperationException("gateway threw"));
Write(node, new short[] { 3, 4 }).StatusCode.ShouldBe(StatusCodes.BadDeviceFailure);
}
else Write(node, new short[] { 3, 4 }, authorized: false).StatusCode.ShouldBe(StatusCodes.BadUserAccessDenied);
first.SetResult(new NodeWriteOutcome(false, "first write failed"));
var deadline = DateTime.UtcNow.AddSeconds(2);
while (DateTime.UtcNow < deadline && !SdkValueBoundary.ToClr(node.Value).ShouldBeOfType<short[]>().SequenceEqual(new short[] { 7, 8 }))
await Task.Delay(10, Ct);
SdkValueBoundary.ToClr(node.Value).ShouldBeOfType<short[]>().ShouldBe([7, 8]);
}
private sealed class OutcomeLogger : ILogger<OtOpcUaNodeManager>
{
internal TaskCompletionSource Ignored { get; } = new(TaskCreationOptions.RunContinuationsAsynchronously);
public IDisposable? BeginScope<TState>(TState state) where TState : notnull => null;
public bool IsEnabled(LogLevel level) => true;
public void Log<TState>(LogLevel level, EventId id, TState state, Exception? exception, Func<TState, Exception?, string> formatter)
{
if (formatter(state, exception).Contains("ignored superseded write outcome", StringComparison.Ordinal)) Ignored.TrySetResult();
}
}
private sealed class OutcomeGateway(Func<object?, Task<NodeWriteOutcome>> outcome) : IOpcUaNodeWriteGateway
{
public Task<NodeWriteOutcome> WriteAsync(string nodeId, object? value, AddressSpaceRealm realm, CancellationToken ct) => outcome(value);
}
private BaseDataVariableState Node(string type, bool array = false, bool refuse = false,
bool writable = true, AddressSpaceRealm realm = AddressSpaceRealm.Raw)
{
@@ -312,7 +312,7 @@ public sealed class SdkAddressSpaceSinkTests : IDisposable
var condition = nm.TryGetAlarmCondition("alm-1");
condition.ShouldNotBeNull();
// It is a REAL Part 9 alarm condition (subtype mapped from "OffNormalAlarm").
condition.ShouldBeOfType<OffNormalAlarmState>();
condition.ShouldBeAssignableTo<OffNormalAlarmState>();
// v3 dual-namespace: the sink's transitional default realm is UNS, so default-realm nodes live in the
// UNS namespace (index 3), not the Raw namespace nm.NamespaceIndex (index 2) reports.
var unsNs = nm.NamespaceIndexForRealm(AddressSpaceRealm.Uns);