fix(opcua): write-path robustness, injected clock, configurable quotas (B4)

Review 01 L3/H5/H4/P6/P4.

- L3/H5 OnEquipmentTagWrite's dispatch is factored into DispatchDeviceWrite:
  a gateway that throws synchronously is refused inline with
  BadDeviceFailure (it used to escape the handler under the SDK Lock), and
  a faulted/cancelled dispatch task has its exception observed and logged
  and carries the real reason into the revert/audit event instead of
  surfacing as UnobservedTaskException at finalization.
- H4 every timestamp the manager stamps (condition quality, timed-shelve
  expiry, method-transition events, carry-over, revert, revalidation)
  reads the injected Clock; RemainingShelveMs takes the clock reading.
- P6 OpcUa:Quotas:* (OpcUaQuotaOptions on OpcUaApplicationHostOptions):
  Min/MaxRequestThreadCount, MaxQueuedRequestCount with the former
  hard-coded 5/100/200 as defaults, and the TransportQuotas members as
  nullable overrides over the SDK defaults.
- P4 the SDK line exposes IHistoryReadAsyncNodeManager but CustomNodeManager2
  offers no async virtual to override (adopting it means re-implementing
  the base dispatcher), so the lever is the admission cap:
  MaxConcurrentHistoryReadBatches defaults to clamp(2*ProcessorCount, 4, 16).
  The Host still overrides it from ServerHistorianOptions (default 16) —
  follow-up F3.

Tests: NodeManagerWriteDispatchTests (sync throw refused; faulted task
reverts; success leaves the value; clock-stamped revert),
OpcUaApplicationHostQuotaTests.

Refs #732

Claude-Session: https://claude.ai/code/session_01Ub9hGmYLQvn9XETzw7CMsT
This commit is contained in:
Joseph Doherty
2026-10-22 10:22:29 -04:00
parent dca822dc59
commit 1578efd4a9
4 changed files with 402 additions and 31 deletions
@@ -105,6 +105,55 @@ public sealed class OpcUaApplicationHostOptions
/// is always element 0.
/// </summary>
public IList<string> PeerApplicationUris { get; set; } = new List<string>();
/// <summary>
/// SDK request-processing and transport quotas (<c>OpcUa:Quotas:*</c>, review 01 P6). The defaults are
/// exactly the values that used to be hard-coded (5 / 100 / 200 request threads and queue) and the SDK's
/// own transport defaults, so an absent section changes nothing.
/// </summary>
public OpcUaQuotaOptions Quotas { get; set; } = new();
}
/// <summary>
/// The <c>OpcUa:Quotas</c> section: the SDK's request-thread / queue ceilings and its transport quotas.
/// Request quotas carry the former hard-coded defaults; a transport quota left <c>null</c> keeps the SDK
/// default (<see cref="TransportQuotas"/>'s constructor values).
/// </summary>
public sealed class OpcUaQuotaOptions
{
/// <summary>Minimum SDK request-processing threads (default 5).</summary>
public int MinRequestThreadCount { get; set; } = 5;
/// <summary>Maximum SDK request-processing threads (default 100).</summary>
public int MaxRequestThreadCount { get; set; } = 100;
/// <summary>Requests the SDK queues before refusing with <c>BadTooManyOperations</c> (default 200) — the
/// burst ceiling review 01 P6 flagged.</summary>
public int MaxQueuedRequestCount { get; set; } = 200;
/// <summary>Per-operation timeout in milliseconds; <c>null</c> keeps the SDK default.</summary>
public int? OperationTimeoutMs { get; set; }
/// <summary>Largest string the transport decodes; <c>null</c> keeps the SDK default.</summary>
public int? MaxStringLength { get; set; }
/// <summary>Largest byte string the transport decodes; <c>null</c> keeps the SDK default.</summary>
public int? MaxByteStringLength { get; set; }
/// <summary>Largest array the transport decodes; <c>null</c> keeps the SDK default.</summary>
public int? MaxArrayLength { get; set; }
/// <summary>Largest message the transport accepts, in bytes; <c>null</c> keeps the SDK default.</summary>
public int? MaxMessageSize { get; set; }
/// <summary>Transport buffer size in bytes; <c>null</c> keeps the SDK default.</summary>
public int? MaxBufferSize { get; set; }
/// <summary>Secure-channel lifetime in milliseconds; <c>null</c> keeps the SDK default.</summary>
public int? ChannelLifetimeMs { get; set; }
/// <summary>Security-token lifetime in milliseconds; <c>null</c> keeps the SDK default.</summary>
public int? SecurityTokenLifetimeMs { get; set; }
}
/// <summary>
@@ -435,10 +484,8 @@ public sealed class OpcUaApplicationHost : IAsyncDisposable
var serverConfig = new ServerConfiguration
{
BaseAddresses = { $"opc.tcp://{_options.PublicHostname}:{_options.OpcUaPort}/OtOpcUa" },
MinRequestThreadCount = 5,
MaxRequestThreadCount = 100,
MaxQueuedRequestCount = 200,
};
ApplyRequestQuotas(serverConfig, _options.Quotas);
foreach (var policy in BuildSecurityPolicies(_options.EnabledSecurityProfiles))
{
@@ -469,7 +516,7 @@ public sealed class OpcUaApplicationHost : IAsyncDisposable
RejectedCertificateStore = new CertificateTrustList { StoreType = "Directory", StorePath = Path.Combine(_options.PkiStoreRoot, "rejected") },
AutoAcceptUntrustedCertificates = _options.AutoAcceptUntrustedClientCertificates,
},
TransportQuotas = new TransportQuotas(),
TransportQuotas = BuildTransportQuotas(_options.Quotas),
ClientConfiguration = new ClientConfiguration(),
TraceConfiguration = new TraceConfiguration(),
};
@@ -479,6 +526,40 @@ public sealed class OpcUaApplicationHost : IAsyncDisposable
return config;
}
/// <summary>
/// Applies the request-processing quotas (<c>OpcUa:Quotas:*</c>, review 01 P6 — they were hard-coded at
/// 5 / 100 / 200) to the SDK server configuration.
/// </summary>
/// <param name="serverConfig">The configuration to set.</param>
/// <param name="quotas">The configured quotas.</param>
internal static void ApplyRequestQuotas(ServerConfiguration serverConfig, OpcUaQuotaOptions quotas)
{
serverConfig.MinRequestThreadCount = quotas.MinRequestThreadCount;
serverConfig.MaxRequestThreadCount = quotas.MaxRequestThreadCount;
serverConfig.MaxQueuedRequestCount = quotas.MaxQueuedRequestCount;
}
/// <summary>
/// Builds the SDK transport quotas from <c>OpcUa:Quotas:*</c>: every member starts at the SDK default and
/// only a configured (non-null) value replaces it, so an absent section is byte-identical to the former
/// <c>new TransportQuotas()</c>.
/// </summary>
/// <param name="quotas">The configured quotas.</param>
/// <returns>The transport quotas to install.</returns>
internal static TransportQuotas BuildTransportQuotas(OpcUaQuotaOptions quotas)
{
var t = new TransportQuotas();
if (quotas.OperationTimeoutMs is { } operationTimeout) t.OperationTimeout = operationTimeout;
if (quotas.MaxStringLength is { } maxString) t.MaxStringLength = maxString;
if (quotas.MaxByteStringLength is { } maxByteString) t.MaxByteStringLength = maxByteString;
if (quotas.MaxArrayLength is { } maxArray) t.MaxArrayLength = maxArray;
if (quotas.MaxMessageSize is { } maxMessage) t.MaxMessageSize = maxMessage;
if (quotas.MaxBufferSize is { } maxBuffer) t.MaxBufferSize = maxBuffer;
if (quotas.ChannelLifetimeMs is { } channelLifetime) t.ChannelLifetime = channelLifetime;
if (quotas.SecurityTokenLifetimeMs is { } tokenLifetime) t.SecurityTokenLifetime = tokenLifetime;
return t;
}
/// <summary>
/// Maps each configured <see cref="OpcUaSecurityProfile"/> to a SDK
/// <see cref="ServerSecurityPolicy"/>. Duplicate profiles are silently de-duped because
@@ -581,9 +581,23 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2
/// Process-wide cap on concurrently-served HistoryRead batches (arch-review 03/S3). A batch that
/// cannot acquire a slot within its (bounded) wait budget fast-fails every handle with
/// <c>BadTooManyOperations</c>. Mirrors <c>ServerHistorianOptions.MaxConcurrentHistoryReadBatches</c>;
/// the Host sets it at <c>StartAsync</c>. The default (16) survives when the section is absent.
/// the Host sets it at <c>StartAsync</c>. The default (<see cref="DefaultMaxConcurrentHistoryReadBatches"/>)
/// survives when the section is absent.
/// <para>
/// Review 01 P4: every HistoryRead / HistoryUpdate arm is sync-over-async on an SDK request thread —
/// the SDK line pins this manager to the synchronous <c>CustomNodeManager2.HistoryRead</c> dispatcher
/// (its <c>IHistoryReadAsyncNodeManager</c> is a separate interface, adapted for sync managers, that
/// offers no async virtual to override; adopting it means re-implementing the base's handle validation
/// and continuation-point plumbing). So the admission cap is the lever: with a slow historian a
/// batch holds its request thread for up to <see cref="HistoryReadDeadline"/>, and 16 of them on a
/// small node is a third of <c>MaxRequestThreadCount</c>. The default therefore scales with the
/// machine — <c>clamp(2 × ProcessorCount, 4, 16)</c> — instead of a flat 16.
/// </para>
/// </summary>
public int MaxConcurrentHistoryReadBatches { get; set; } = 16;
public int MaxConcurrentHistoryReadBatches { get; set; } = DefaultMaxConcurrentHistoryReadBatches;
/// <summary>The CPU-scaled default of <see cref="MaxConcurrentHistoryReadBatches"/>: <c>clamp(2 × ProcessorCount, 4, 16)</c>.</summary>
public static int DefaultMaxConcurrentHistoryReadBatches { get; } = Math.Clamp(Environment.ProcessorCount * 2, 4, 16);
/// <summary>
/// Per-request wall-clock deadline for a HistoryRead batch (arch-review 03/S3). A node still in
@@ -1058,7 +1072,7 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2
SystemContext,
shelved: state.Shelving != AlarmShelvingKind.Unshelved,
oneShot: state.Shelving == AlarmShelvingKind.OneShot,
shelvingTime: state.Shelving == AlarmShelvingKind.Timed ? RemainingShelveMs(state.UnshelveAtUtc) : 0);
shelvingTime: state.Shelving == AlarmShelvingKind.Timed ? RemainingShelveMs(state.UnshelveAtUtc, UtcNow) : 0);
}
condition.SetSeverity(SystemContext, ToPart9Severity(state.Severity));
condition.Message.Value = new LocalizedText(state.Message);
@@ -1105,11 +1119,12 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2
/// at most <c>int.MaxValue</c> (the SDK timer's range).
/// </summary>
/// <param name="unshelveAtUtc">The engine's expiry instant, or null.</param>
/// <param name="nowUtc">The manager's clock reading (<see cref="Clock"/>), so a test clock drives it.</param>
/// <returns>The shelving time in milliseconds.</returns>
internal static double RemainingShelveMs(DateTime? unshelveAtUtc)
internal static double RemainingShelveMs(DateTime? unshelveAtUtc, DateTime nowUtc)
{
if (unshelveAtUtc is not { } at) return 0;
var remaining = (at - DateTime.UtcNow).TotalMilliseconds;
var remaining = (at - nowUtc).TotalMilliseconds;
return Math.Clamp(Math.Ceiling(remaining), 1, int.MaxValue);
}
@@ -1476,7 +1491,7 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2
if (alarm.Quality is not null)
{
alarm.Quality.Value = isNative ? StatusCodes.BadWaitingForInitialData : StatusCodes.Good;
if (alarm.Quality.SourceTimestamp is not null) alarm.Quality.SourceTimestamp.Value = DateTime.UtcNow;
if (alarm.Quality.SourceTimestamp is not null) alarm.Quality.SourceTimestamp.Value = UtcNow;
}
// Initial state via the SDK setters (T14: basic state only, NO event firing).
@@ -1538,7 +1553,7 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2
var (operation, unshelveAt) =
!shelving ? ("Unshelve", (DateTime?)null)
: oneShot ? ("OneShotShelve", null)
: ("TimedShelve", DateTime.UtcNow + TimeSpan.FromMilliseconds(shelvingTime));
: ("TimedShelve", UtcNow + TimeSpan.FromMilliseconds(shelvingTime));
return HandleAlarmCommand(context, condition, operation, comment: null, unshelveAt);
};
// The SDK's auto-unshelve timer (armed by SetShelvingState for the real remaining time since the
@@ -1635,9 +1650,9 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2
/// </remarks>
/// <param name="condition">The live native condition, about to be discarded.</param>
/// <returns>Its carried state.</returns>
private static NativeConditionCarryOver CaptureCarryOver(AlarmConditionState condition) => new(
private NativeConditionCarryOver CaptureCarryOver(AlarmConditionState condition) => new(
ReadConditionDelta(condition),
condition.Time?.Value ?? DateTime.UtcNow,
condition.Time?.Value ?? UtcNow,
condition.ActiveState?.TransitionTime?.Value,
condition.Quality?.SourceTimestamp?.Value,
condition.Comment?.Value,
@@ -1953,7 +1968,7 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2
}
if (alwaysReport || ReadConditionDelta(alarm) != before)
ReportConditionEvent(alarm, DateTime.UtcNow);
ReportConditionEvent(alarm, UtcNow);
return result;
}
@@ -2223,20 +2238,60 @@ 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;
// Fire-and-forget — MUST NOT block under Lock. On a FAILED outcome, compare-and-revert (off-Lock
// continuation). A faulted/cancelled WriteAsync is treated as a failure so the optimistic value never
// sticks when the route never resolved a real outcome. RunContinuationsAsynchronously 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 RevertOptimisticWriteIfNeeded never
// re-enters lock (Lock) while CustomNodeManager2.Write still holds it.
_ = gateway.WriteAsync(nodeKey, optimisticValue, nodeRealm, CancellationToken.None)
.ContinueWith(
t =>
return DispatchDeviceWrite(gateway, nodeKey, nodeBareId, nodeRealm, optimisticValue, priorValue, priorStatus, clientUserId);
}
/// <summary>
/// The dispatch half of <see cref="OnEquipmentTagWrite"/>: start the device write and arm the off-Lock
/// revert. Fire-and-forget — MUST NOT block under <c>Lock</c>. On a FAILED outcome, compare-and-revert
/// (off-Lock continuation). A faulted/cancelled <c>WriteAsync</c> is treated as a failure so the optimistic
/// value never sticks when the route never resolved a real outcome; its exception is OBSERVED and logged
/// 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.
/// </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)
{
Task<NodeWriteOutcome> dispatch;
try
{
dispatch = gateway.WriteAsync(nodeKey, optimisticValue, nodeRealm, CancellationToken.None);
}
catch (Exception ex)
{
_logger.LogWarning(ex, "OtOpcUaNodeManager: write dispatch for {NodeId} threw synchronously; the write is refused", nodeKey);
return new ServiceResult(StatusCodes.BadDeviceFailure, $"write dispatch threw: {ex.Message}");
}
_ = dispatch.ContinueWith(
t =>
{
NodeWriteOutcome outcome;
if (t.IsCompletedSuccessfully)
{
var outcome = t.IsCompletedSuccessfully ? t.Result : new NodeWriteOutcome(false, "write dispatch faulted");
RevertOptimisticWriteIfNeeded(nodeBareId, outcome, optimisticValue, priorValue, priorStatus, clientUserId, nodeRealm);
},
CancellationToken.None, TaskContinuationOptions.RunContinuationsAsynchronously, TaskScheduler.Default);
outcome = t.Result;
}
else
{
// Reading t.Exception observes it, so the failure never reaches UnobservedTaskException.
var fault = t.Exception?.GetBaseException();
var reason = fault is not null ? $"write dispatch faulted: {fault.Message}"
: t.IsCanceled ? "write dispatch cancelled" : "write dispatch faulted";
_logger.LogWarning(fault, "OtOpcUaNodeManager: write dispatch for {NodeId} did not complete — {Reason}", nodeKey, reason);
outcome = new NodeWriteOutcome(false, reason);
}
RevertOptimisticWriteIfNeeded(nodeBareId, outcome, optimisticValue, priorValue, priorStatus, clientUserId, nodeRealm);
},
CancellationToken.None, TaskContinuationOptions.RunContinuationsAsynchronously, TaskScheduler.Default);
return ServiceResult.Good;
}
@@ -2647,13 +2702,14 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2
// 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
// node-update shape). See the method remarks for the queue-coalescing caveat.
var now = UtcNow;
variable.StatusCode = StatusCodes.BadDeviceFailure;
variable.Timestamp = DateTime.UtcNow;
variable.Timestamp = now;
variable.ClearChangeMasks(SystemContext, includeChildren: false); // notify — the Bad blip
variable.Value = priorValue;
variable.StatusCode = priorStatus;
variable.Timestamp = DateTime.UtcNow;
variable.Timestamp = now;
variable.ClearChangeMasks(SystemContext, includeChildren: false); // notify — restore prior
// Item C: build the audit event-state while we hold the node reference + Lock, but DON'T report yet.
@@ -2710,7 +2766,7 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2
severity: EventSeverity.Medium,
message: new LocalizedText($"Inbound write rejected by device: {reason}"),
status: false,
actionTimestamp: DateTime.UtcNow);
actionTimestamp: UtcNow);
// Write-specific fields (AuditWriteUpdateEventType). AttributeId = Value; IndexRange empty (full write);
// OldValue = the real pre-write value; NewValue = the value the client attempted. The AuditWriteUpdate
@@ -5079,7 +5135,7 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2
{
if (!AclEnforced || (_liveItems.IsEmpty && _eventItems.IsEmpty)) return;
var now = DateTime.UtcNow;
var now = UtcNow;
var stoodDown = 0;
var restored = 0;
@@ -0,0 +1,160 @@
using Microsoft.Extensions.Logging.Abstractions;
using Opc.Ua;
using Shouldly;
using Xunit;
using ZB.MOM.WW.OtOpcUa.Commons.OpcUa;
namespace ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests;
/// <summary>
/// Review 01 L3 / H5 / H4 — the device-write dispatch behind <see cref="OtOpcUaNodeManager.OnEquipmentTagWrite"/>
/// and the manager's clock: a gateway that throws synchronously is refused inline (the handler runs under
/// the SDK <c>Lock</c>, and the throw used to escape it), a faulted dispatch task is observed and still
/// reverts the optimistic value, and the timestamps the manager stamps come from the injected
/// <see cref="OtOpcUaNodeManager.Clock"/>.
/// </summary>
public sealed class NodeManagerWriteDispatchTests : IDisposable
{
private static CancellationToken Ct => TestContext.Current.CancellationToken;
private readonly string _pkiRoot = Path.Combine(Path.GetTempPath(), $"otopcua-dispatch-{Guid.NewGuid():N}");
/// <summary>H5 — a synchronous throw out of the gateway is a Bad result, never an escaped exception.</summary>
[Fact]
public async Task A_gateway_that_throws_synchronously_is_refused_inline()
{
var (host, server) = await BootAsync();
try
{
var nm = server.NodeManager!;
nm.EnsureVariable("eq-1/sp", null, "Sp", "Int32", writable: true, realm: AddressSpaceRealm.Uns);
nm.WriteValue("eq-1/sp", 7, OpcUaStatus.Good, DateTime.UtcNow, AddressSpaceRealm.Uns);
var result = nm.DispatchDeviceWrite(
new ThrowingGateway(), "ns=2;s=eq-1/sp", "eq-1/sp", AddressSpaceRealm.Uns,
optimisticValue: 42, priorValue: 7, priorStatus: StatusCodes.Good, clientUserId: "op");
ServiceResult.IsBad(result).ShouldBeTrue();
result.StatusCode.ShouldBe((StatusCode)StatusCodes.BadDeviceFailure);
result.ToString().ShouldContain("boom");
nm.TryGetVariable("eq-1/sp")!.Value.ShouldBe(7, "nothing was applied, so nothing needs reverting");
}
finally { await host.DisposeAsync(); }
}
/// <summary>L3 — a faulted dispatch task is observed and the optimistic value is reverted with its reason.</summary>
[Fact]
public async Task A_faulted_dispatch_is_observed_and_reverts_the_optimistic_value()
{
var (host, server) = await BootAsync();
try
{
var nm = server.NodeManager!;
nm.EnsureVariable("eq-1/sp", null, "Sp", "Int32", writable: true, realm: AddressSpaceRealm.Uns);
nm.WriteValue("eq-1/sp", 7, OpcUaStatus.Good, DateTime.UtcNow, AddressSpaceRealm.Uns);
var node = nm.TryGetVariable("eq-1/sp")!;
var result = nm.DispatchDeviceWrite(
new FaultingGateway(), "ns=2;s=eq-1/sp", "eq-1/sp", AddressSpaceRealm.Uns,
optimisticValue: 42, priorValue: 7, priorStatus: StatusCodes.Good, clientUserId: "op");
ServiceResult.IsGood(result).ShouldBeTrue("the dispatch itself started; the SDK applies the value optimistically");
node.Value = 42; // what the SDK does on Good
var deadline = DateTime.UtcNow + TimeSpan.FromSeconds(5);
while (DateTime.UtcNow < deadline && !Equals(node.Value, 7)) await Task.Delay(20, Ct);
node.Value.ShouldBe(7, "the faulted dispatch must revert the optimistic value");
node.StatusCode.ShouldBe((StatusCode)StatusCodes.Good);
}
finally { await host.DisposeAsync(); }
}
/// <summary>Positive control for the two above: a gateway that succeeds leaves the optimistic value standing.</summary>
[Fact]
public async Task A_successful_dispatch_leaves_the_optimistic_value()
{
var (host, server) = await BootAsync();
try
{
var nm = server.NodeManager!;
nm.EnsureVariable("eq-1/sp", null, "Sp", "Int32", writable: true, realm: AddressSpaceRealm.Uns);
nm.WriteValue("eq-1/sp", 7, OpcUaStatus.Good, DateTime.UtcNow, AddressSpaceRealm.Uns);
var node = nm.TryGetVariable("eq-1/sp")!;
nm.DispatchDeviceWrite(
new SucceedingGateway(), "ns=2;s=eq-1/sp", "eq-1/sp", AddressSpaceRealm.Uns,
optimisticValue: 42, priorValue: 7, priorStatus: StatusCodes.Good, clientUserId: "op");
node.Value = 42;
await Task.Delay(200, Ct);
node.Value.ShouldBe(42);
}
finally { await host.DisposeAsync(); }
}
/// <summary>H4 — the revert stamps the manager's clock, not the wall clock.</summary>
[Fact]
public async Task The_revert_timestamps_come_from_the_injected_clock()
{
var (host, server) = await BootAsync();
try
{
var nm = server.NodeManager!;
var fixedNow = new DateTime(2030, 1, 2, 3, 4, 5, DateTimeKind.Utc);
nm.Clock = new FixedClock(fixedNow);
nm.EnsureVariable("eq-1/sp", null, "Sp", "Int32", writable: true, realm: AddressSpaceRealm.Uns);
nm.WriteValue("eq-1/sp", 7, OpcUaStatus.Good, new DateTime(2026, 1, 1, 0, 0, 0, DateTimeKind.Utc), AddressSpaceRealm.Uns);
var node = nm.TryGetVariable("eq-1/sp")!;
node.Value = 42;
nm.RevertOptimisticWriteIfNeeded(
"eq-1/sp", new NodeWriteOutcome(false, "rejected"), optimisticValue: 42, priorValue: 7,
priorStatus: StatusCodes.Good, clientUserId: "op", realm: AddressSpaceRealm.Uns);
node.Timestamp.ShouldBe(fixedNow);
OtOpcUaNodeManager.RemainingShelveMs(fixedNow.AddSeconds(30), fixedNow).ShouldBe(30_000);
}
finally { await host.DisposeAsync(); }
}
private async Task<(OpcUaApplicationHost Host, OtOpcUaSdkServer Server)> BootAsync() =>
await OpcUaHostStart.StartAsync(
port => new OpcUaApplicationHost(
new OpcUaApplicationHostOptions
{
ApplicationName = "OtOpcUa.DispatchTest",
ApplicationUri = $"urn:OtOpcUa.DispatchTest:{Guid.NewGuid():N}",
OpcUaPort = port,
PublicHostname = "localhost",
PkiStoreRoot = _pkiRoot,
},
NullLogger<OpcUaApplicationHost>.Instance),
() => new OtOpcUaSdkServer(), Ct);
public void Dispose()
{
if (Directory.Exists(_pkiRoot)) Directory.Delete(_pkiRoot, recursive: true);
}
private sealed class ThrowingGateway : IOpcUaNodeWriteGateway
{
public Task<NodeWriteOutcome> WriteAsync(string nodeId, object? value, AddressSpaceRealm realm, CancellationToken ct) =>
throw new InvalidOperationException("boom");
}
private sealed class FaultingGateway : IOpcUaNodeWriteGateway
{
public Task<NodeWriteOutcome> WriteAsync(string nodeId, object? value, AddressSpaceRealm realm, CancellationToken ct) =>
Task.FromException<NodeWriteOutcome>(new TimeoutException("gateway ask timed out"));
}
private sealed class SucceedingGateway : IOpcUaNodeWriteGateway
{
public Task<NodeWriteOutcome> WriteAsync(string nodeId, object? value, AddressSpaceRealm realm, CancellationToken ct) =>
Task.FromResult(new NodeWriteOutcome(true, null));
}
private sealed class FixedClock(DateTime utcNow) : TimeProvider
{
public override DateTimeOffset GetUtcNow() => new(utcNow, TimeSpan.Zero);
}
}
@@ -0,0 +1,74 @@
using Opc.Ua;
using Shouldly;
using Xunit;
namespace ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests;
/// <summary>Review 01 P6 — the SDK request and transport quotas are configurable (<c>OpcUa:Quotas:*</c>) and
/// default to exactly what used to be hard-coded.</summary>
public sealed class OpcUaApplicationHostQuotaTests
{
[Fact]
public void Default_request_quotas_are_the_former_hard_coded_values()
{
var config = new ServerConfiguration();
OpcUaApplicationHost.ApplyRequestQuotas(config, new OpcUaQuotaOptions());
config.MinRequestThreadCount.ShouldBe(5);
config.MaxRequestThreadCount.ShouldBe(100);
config.MaxQueuedRequestCount.ShouldBe(200);
}
[Fact]
public void Configured_request_quotas_are_applied()
{
var config = new ServerConfiguration();
OpcUaApplicationHost.ApplyRequestQuotas(config, new OpcUaQuotaOptions
{
MinRequestThreadCount = 8, MaxRequestThreadCount = 250, MaxQueuedRequestCount = 1000,
});
config.MinRequestThreadCount.ShouldBe(8);
config.MaxRequestThreadCount.ShouldBe(250);
config.MaxQueuedRequestCount.ShouldBe(1000);
}
[Fact]
public void Unset_transport_quotas_keep_the_sdk_defaults()
{
var sdkDefault = new TransportQuotas();
var built = OpcUaApplicationHost.BuildTransportQuotas(new OpcUaQuotaOptions());
built.OperationTimeout.ShouldBe(sdkDefault.OperationTimeout);
built.MaxStringLength.ShouldBe(sdkDefault.MaxStringLength);
built.MaxByteStringLength.ShouldBe(sdkDefault.MaxByteStringLength);
built.MaxArrayLength.ShouldBe(sdkDefault.MaxArrayLength);
built.MaxMessageSize.ShouldBe(sdkDefault.MaxMessageSize);
built.MaxBufferSize.ShouldBe(sdkDefault.MaxBufferSize);
built.ChannelLifetime.ShouldBe(sdkDefault.ChannelLifetime);
built.SecurityTokenLifetime.ShouldBe(sdkDefault.SecurityTokenLifetime);
}
[Fact]
public void Configured_transport_quotas_replace_only_what_is_set()
{
var sdkDefault = new TransportQuotas();
var built = OpcUaApplicationHost.BuildTransportQuotas(new OpcUaQuotaOptions
{
MaxMessageSize = 8 * 1024 * 1024, OperationTimeoutMs = 45_000, MaxArrayLength = 1_000_000,
});
built.MaxMessageSize.ShouldBe(8 * 1024 * 1024);
built.OperationTimeout.ShouldBe(45_000);
built.MaxArrayLength.ShouldBe(1_000_000);
built.MaxStringLength.ShouldBe(sdkDefault.MaxStringLength);
built.ChannelLifetime.ShouldBe(sdkDefault.ChannelLifetime);
}
[Fact]
public void History_read_batch_cap_default_scales_with_the_machine()
{
OtOpcUaNodeManager.DefaultMaxConcurrentHistoryReadBatches
.ShouldBe(Math.Clamp(Environment.ProcessorCount * 2, 4, 16));
}
}