fix(runtime): derive every write bound and Ask from one write timeout (lmxopcua#777)
v2-ci / unit-tests (push) Has been cancelled
v2-ci / unit-tests (push) Has been cancelled
Three write bounds were configured independently: the inbound-write gateway's 10 s Ask, DriverHostActor's fixed 8 s Ask of the child, and the child's own write timeout (5 s + up to 2 s grace) with a fixed 8 s queue budget. An outer Ask that fires first reports the write failed and the node reverts while the child is still driving the device, so any write timeout above ~6 s (or a write dequeued late in its 8 s budget) let the device take a value the client was told failed. DriverWriteBounds (Runtime) derives them all from DriverWrites:Timeout: token cancel = Timeout abandon = Timeout + min(Timeout/2, 2 s) (RunBound) queue budget = RunBound child replies <= queue budget + RunBound (ReplyBound) host Ask = ReplyBound + 1 s gateway Ask = host Ask + 1 s Default Timeout is 10 s (the Tier C Write timeout, so no tier default is capped below its own policy); host Ask 25 s, gateway Ask 26 s. Tier A/B writes are still cancelled first by their 2 s / 4 s tier timeouts. - DriverHostActor takes writeBounds + writeClock, spawns children with them, and runs its Ask deadline as a token on the write clock (Infinite Akka timeout), so host and child bounds share one clock. - WriteAttribute gains IssuedAt (write-clock timestamp, stamped by the host): the queue budget counts mailbox time too, and a write past it is failed BadTimeout and never started - including on arrival. Stale queued writes are expired across the whole queue, not only up to a free slot. - ActorNodeWriteGateway's default Ask = Default.GatewayAsk; the hosted service passes the configured one. - Startup validation (Host DriverWriteBoundsBinding, same warn-and-clamp style as ScriptingOptionsBinding): unparsable / non-positive -> default, above 5 min -> clamped, and each tier whose default Write timeout exceeds the write timeout is named in a Warning (its policy would never take effect). The derived bounds are logged at Information. Per-instance ResilienceConfig Write overrides arrive with a deployment, so DriverCapabilityInvokerFactory warns at spawn (writeTimeoutCeiling). - appsettings.json documents DriverWrites:Timeout. Tests (manual clock): - DriverHostActorWriteAskTests: a held write under a 9 s timeout survives an 8.5 s step and succeeds; a token-ignoring write is answered by the child's BadTimeout at 9 s + grace, not by the Ask; the Infinite-timeout Ask ends exactly at its clock deadline. - DriverInstanceActorConcurrentWriteTests: a write issued longer ago than the queue budget is failed and never reaches the driver; positive control at exactly the budget is written. - DriverWriteBoundsTests: every outer bound covers the inner ones for 0.5 s to 300 s timeouts. - DriverWriteBoundsBindingTests: default fits every tier; 3 s timeout is caught at startup with Tier B and C named (not A); bad values fall back with a warning; over-max clamps. - DriverCapabilityInvokerFactoryTests: spawn warns when a Write override exceeds the ceiling; no warning within it or without a ceiling. Mutation checks: a fixed 8 s host Ask on the same clock fails both 9 s host tests; stamping arrival time instead of IssuedAt fails the issued-at test; dropping the tier check / factory check fails their tests. Runtime.Tests 961 passed / 3 skipped at AKKA_TEST_TIMEFACTOR 3 and 1; Host.Tests 232/232 at 3 and 1; Core.Tests 361 passed / 1 skipped. Solution Release build: 0 warnings. ContinuousHistorizationRecorderTypedValueTests (historian) deliberately left for the historian lane. Claude-Session: https://claude.ai/code/session_01GT67AEHd2Acv1YPwGfXEoV
This commit is contained in:
@@ -33,6 +33,7 @@ public sealed class DriverCapabilityInvokerFactory : IDriverCapabilityInvokerFac
|
||||
private readonly DriverResilienceStatusTracker _statusTracker;
|
||||
private readonly Func<string, DriverTier> _tierResolver;
|
||||
private readonly ILogger? _logger;
|
||||
private readonly TimeSpan? _writeTimeoutCeiling;
|
||||
private long _generation;
|
||||
|
||||
/// <summary>Construct the factory over the shared resilience infrastructure.</summary>
|
||||
@@ -41,11 +42,16 @@ public sealed class DriverCapabilityInvokerFactory : IDriverCapabilityInvokerFac
|
||||
/// <param name="tierResolver">Resolves a driver type name to its <see cref="DriverTier"/> (e.g. <c>DriverFactoryRegistry.GetTier</c>).</param>
|
||||
/// <param name="logger">Optional logger; when non-null, a ResilienceConfig parse diagnostic (malformed JSON,
|
||||
/// unknown capability, misapplied Tier-C recycle) is logged as a warning at <see cref="Create"/> time.</param>
|
||||
/// <param name="writeTimeoutCeiling">The node's operator-write timeout (<c>DriverWrites:Timeout</c>), which
|
||||
/// cancels every write before the Write policy's own timeout can (lmxopcua#777). When non-null, a driver whose
|
||||
/// effective Write timeout exceeds it is logged as a warning at <see cref="Create"/> time: that policy timeout
|
||||
/// never takes effect. Null skips the check.</param>
|
||||
public DriverCapabilityInvokerFactory(
|
||||
DriverResiliencePipelineBuilder builder,
|
||||
DriverResilienceStatusTracker statusTracker,
|
||||
Func<string, DriverTier> tierResolver,
|
||||
ILogger? logger = null)
|
||||
ILogger? logger = null,
|
||||
TimeSpan? writeTimeoutCeiling = null)
|
||||
{
|
||||
ArgumentNullException.ThrowIfNull(builder);
|
||||
ArgumentNullException.ThrowIfNull(statusTracker);
|
||||
@@ -54,6 +60,7 @@ public sealed class DriverCapabilityInvokerFactory : IDriverCapabilityInvokerFac
|
||||
_statusTracker = statusTracker;
|
||||
_tierResolver = tierResolver;
|
||||
_logger = logger;
|
||||
_writeTimeoutCeiling = writeTimeoutCeiling;
|
||||
}
|
||||
|
||||
/// <inheritdoc />
|
||||
@@ -68,6 +75,16 @@ public sealed class DriverCapabilityInvokerFactory : IDriverCapabilityInvokerFac
|
||||
driverInstanceId, driverType, diagnostic);
|
||||
}
|
||||
|
||||
var writePolicy = options.Resolve(DriverCapability.Write);
|
||||
if (_writeTimeoutCeiling is { } ceiling && TimeSpan.FromSeconds(writePolicy.TimeoutSeconds) > ceiling)
|
||||
{
|
||||
_logger?.LogWarning(
|
||||
"Driver resilience config for instance={DriverInstanceId} type={DriverType}: Write timeout " +
|
||||
"{WriteTimeoutSeconds}s exceeds DriverWrites:Timeout ({WriteTimeoutCeiling}); writes are cancelled at " +
|
||||
"the latter, so the policy timeout never takes effect",
|
||||
driverInstanceId, driverType, writePolicy.TimeoutSeconds, ceiling);
|
||||
}
|
||||
|
||||
// Stamp a fresh options generation for this invoker. The generation is part of the pipeline-cache
|
||||
// key (01/S-7), so this invoker can only ever read pipelines it built — a lingering old child's
|
||||
// late re-cache lands under its OWN (older) generation and is unreachable here. Correctness rests on
|
||||
|
||||
@@ -0,0 +1,101 @@
|
||||
using System.Globalization;
|
||||
using Microsoft.Extensions.Configuration;
|
||||
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
|
||||
using ZB.MOM.WW.OtOpcUa.Core.Resilience;
|
||||
using ZB.MOM.WW.OtOpcUa.Runtime.Drivers;
|
||||
|
||||
namespace ZB.MOM.WW.OtOpcUa.Host.Configuration;
|
||||
|
||||
/// <summary>
|
||||
/// Binds <c>DriverWrites:Timeout</c> into <see cref="DriverWriteBounds"/> and checks it against the resilience
|
||||
/// tiers' Write timeouts (lmxopcua#777). One place, so <c>Program.cs</c> and its unit test bind the same way.
|
||||
/// </summary>
|
||||
/// <remarks>
|
||||
/// Matches <see cref="ScriptingOptionsBinding"/>: a bad value is clamped to a safe one and reported as a startup
|
||||
/// Warning, never a boot failure. No setting here can make a write outlive an Ask — every Ask is derived from the
|
||||
/// timeout — so an inconsistency only means a configured bound does not take effect, which is worth saying loudly
|
||||
/// but not worth an outage.
|
||||
/// </remarks>
|
||||
public static class DriverWriteBoundsBinding
|
||||
{
|
||||
/// <summary>The configuration key of the write timeout.</summary>
|
||||
public const string TimeoutKey = DriverWriteBounds.SectionName + ":Timeout";
|
||||
|
||||
/// <summary>Binds, clamps and validates the write timeout.</summary>
|
||||
/// <param name="configuration">The host configuration.</param>
|
||||
/// <param name="warnings">One line per clamped value or tier inconsistency, for the caller to log as a Warning.</param>
|
||||
/// <returns>The write bounds; <see cref="DriverWriteBounds.Default"/> when the key is absent.</returns>
|
||||
public static DriverWriteBounds Bind(IConfiguration configuration, out IReadOnlyList<string> warnings)
|
||||
{
|
||||
ArgumentNullException.ThrowIfNull(configuration);
|
||||
var found = new List<string>();
|
||||
var timeout = DriverWriteBounds.DefaultTimeout;
|
||||
var raw = configuration[TimeoutKey];
|
||||
if (!string.IsNullOrWhiteSpace(raw))
|
||||
{
|
||||
if (!TimeSpan.TryParse(raw, CultureInfo.InvariantCulture, out var parsed))
|
||||
{
|
||||
found.Add($"{TimeoutKey} '{raw}' is not a TimeSpan (e.g. 00:00:10); using the default {Format(timeout)}.");
|
||||
}
|
||||
else if (parsed <= TimeSpan.Zero)
|
||||
{
|
||||
found.Add($"{TimeoutKey} {parsed} is not positive; using the default {Format(timeout)}.");
|
||||
}
|
||||
else if (parsed > DriverWriteBounds.MaxTimeout)
|
||||
{
|
||||
timeout = DriverWriteBounds.MaxTimeout;
|
||||
found.Add($"{TimeoutKey} {parsed} exceeds the max {Format(timeout)}; clamped.");
|
||||
}
|
||||
else
|
||||
{
|
||||
timeout = parsed;
|
||||
}
|
||||
}
|
||||
|
||||
var bounds = new DriverWriteBounds(timeout);
|
||||
found.AddRange(CheckTierDefaults(bounds));
|
||||
warnings = found;
|
||||
return bounds;
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Reports every resilience tier whose DEFAULT Write timeout exceeds <paramref name="bounds"/>' timeout. The
|
||||
/// actor cancels a write at that timeout, so such a tier's Write policy never takes effect: its drivers'
|
||||
/// writes are cut off sooner than their tier says. (Per-instance <c>ResilienceConfig</c> overrides arrive
|
||||
/// with a deployment, not at startup; <see cref="DriverCapabilityInvokerFactory"/> checks those at spawn.)
|
||||
/// </summary>
|
||||
/// <param name="bounds">The write bounds in force.</param>
|
||||
/// <returns>One warning per inconsistent tier; empty when every tier fits.</returns>
|
||||
public static IReadOnlyList<string> CheckTierDefaults(DriverWriteBounds bounds)
|
||||
{
|
||||
ArgumentNullException.ThrowIfNull(bounds);
|
||||
var found = new List<string>();
|
||||
foreach (var tier in Enum.GetValues<DriverTier>())
|
||||
{
|
||||
var write = DriverResilienceOptions.GetTierDefaults(tier)[DriverCapability.Write];
|
||||
if (TimeSpan.FromSeconds(write.TimeoutSeconds) > bounds.Timeout)
|
||||
{
|
||||
found.Add(
|
||||
$"Tier {tier} drivers' Write timeout ({write.TimeoutSeconds}s) exceeds {TimeoutKey} " +
|
||||
$"({Format(bounds.Timeout)}): their writes are cancelled at {Format(bounds.Timeout)}, so the tier " +
|
||||
$"policy never takes effect. Raise {TimeoutKey} to at least {write.TimeoutSeconds}s.");
|
||||
}
|
||||
}
|
||||
|
||||
return found;
|
||||
}
|
||||
|
||||
/// <summary>One line naming every derived bound, for the startup log.</summary>
|
||||
/// <param name="bounds">The write bounds in force.</param>
|
||||
/// <returns>The description.</returns>
|
||||
public static string Describe(DriverWriteBounds bounds)
|
||||
{
|
||||
ArgumentNullException.ThrowIfNull(bounds);
|
||||
return $"write timeout {Format(bounds.Timeout)}, token-ignoring writes abandoned at {Format(bounds.RunBound)}, " +
|
||||
$"queue budget {Format(bounds.QueueBudget)}, driver-host Ask {Format(bounds.HostAsk)}, " +
|
||||
$"inbound-write gateway Ask {Format(bounds.GatewayAsk)}";
|
||||
}
|
||||
|
||||
private static string Format(TimeSpan value) =>
|
||||
value.TotalSeconds.ToString("0.###", CultureInfo.InvariantCulture) + "s";
|
||||
}
|
||||
@@ -109,7 +109,11 @@ public static class DriverFactoryBootstrap
|
||||
sp.GetRequiredService<DriverResiliencePipelineBuilder>(),
|
||||
sp.GetRequiredService<DriverResilienceStatusTracker>(),
|
||||
registry.GetTier,
|
||||
logger: sp.GetService<ILoggerFactory>()?.CreateLogger("ZB.MOM.WW.OtOpcUa.Core.Resilience.DriverCapabilityInvokerFactory"));
|
||||
logger: sp.GetService<ILoggerFactory>()?.CreateLogger("ZB.MOM.WW.OtOpcUa.Core.Resilience.DriverCapabilityInvokerFactory"),
|
||||
// lmxopcua#777: warn at spawn when a per-instance ResilienceConfig Write timeout exceeds the
|
||||
// node's write timeout (it would never take effect). Unregistered ⇒ the default bounds.
|
||||
writeTimeoutCeiling: (sp.GetService<ZB.MOM.WW.OtOpcUa.Runtime.Drivers.DriverWriteBounds>()
|
||||
?? ZB.MOM.WW.OtOpcUa.Runtime.Drivers.DriverWriteBounds.Default).Timeout);
|
||||
});
|
||||
|
||||
// The tracker's operator-facing reader: a periodic DPS publisher that mirrors each snapshot to the
|
||||
|
||||
@@ -52,6 +52,7 @@ public sealed class OtOpcUaServerHostedService : IHostedService, IAsyncDisposabl
|
||||
private readonly NativeAlarmAckPreflightView? _ackPreflightView;
|
||||
private readonly OpcUaEndpointState _endpointState;
|
||||
private readonly bool _rethrowStartFailure;
|
||||
private readonly DriverWriteBounds _writeBounds;
|
||||
|
||||
private OpcUaApplicationHost? _appHost;
|
||||
private OtOpcUaSdkServer? _server;
|
||||
@@ -116,6 +117,9 @@ public sealed class OtOpcUaServerHostedService : IHostedService, IAsyncDisposabl
|
||||
/// the health probe instead. Null (harnesses) behaves as fused.</param>
|
||||
/// <param name="ackPreflightView">The native-ack pre-check snapshot <c>DriverHostActor</c> publishes
|
||||
/// (lmxopcua#760). Optional: registered on every driver-role node; null wires no pre-check.</param>
|
||||
/// <param name="writeBounds">The node's configured write bounds (<c>DriverWrites:Timeout</c>); the inbound
|
||||
/// write gateway's Ask is derived from them (lmxopcua#777). Optional: null ⇒
|
||||
/// <see cref="DriverWriteBounds.Default"/>.</param>
|
||||
public OtOpcUaServerHostedService(
|
||||
IOptions<OpcUaApplicationHostOptions> options,
|
||||
DeferredAddressSpaceSink deferredSink,
|
||||
@@ -135,7 +139,8 @@ public sealed class OtOpcUaServerHostedService : IHostedService, IAsyncDisposabl
|
||||
IHistoryUpdateRoleView? historyUpdateRoleView = null,
|
||||
OpcUaEndpointState? endpointState = null,
|
||||
EffectiveRoles? roles = null,
|
||||
NativeAlarmAckPreflightView? ackPreflightView = null)
|
||||
NativeAlarmAckPreflightView? ackPreflightView = null,
|
||||
DriverWriteBounds? writeBounds = null)
|
||||
{
|
||||
ArgumentNullException.ThrowIfNull(aclOptions);
|
||||
ArgumentNullException.ThrowIfNull(time);
|
||||
@@ -161,6 +166,7 @@ public sealed class OtOpcUaServerHostedService : IHostedService, IAsyncDisposabl
|
||||
_historianUpdater = historianUpdater ?? HistorianUpdaterHolder.None;
|
||||
_historyUpdateRoleView = historyUpdateRoleView;
|
||||
_ackPreflightView = ackPreflightView;
|
||||
_writeBounds = writeBounds ?? DriverWriteBounds.Default;
|
||||
_endpointState = endpointState ?? new OpcUaEndpointState();
|
||||
_rethrowStartFailure = roles is { HasDriver: true, HasAdmin: false };
|
||||
_logger = loggerFactory.CreateLogger<OtOpcUaServerHostedService>();
|
||||
@@ -323,7 +329,8 @@ public sealed class OtOpcUaServerHostedService : IHostedService, IAsyncDisposabl
|
||||
// resolves the write to "writes unavailable".
|
||||
_server.SetNodeWriteGateway(new ActorNodeWriteGateway(
|
||||
resolveDriverHost: () => _actorRegistry.TryGet<DriverHostActorKey>(out var driverHost) ? driverHost : null,
|
||||
logger: _loggerFactory.CreateLogger<ActorNodeWriteGateway>()));
|
||||
logger: _loggerFactory.CreateLogger<ActorNodeWriteGateway>(),
|
||||
askTimeout: _writeBounds.GatewayAsk));
|
||||
|
||||
// Wire the server-side read backend resolved from DI — the NullHistorianDataSource default (when
|
||||
// the ServerHistorian section is disabled) or the configured HistorianGateway read client (when enabled).
|
||||
|
||||
@@ -376,6 +376,18 @@ if (hasDriver)
|
||||
Log.Warning("Scripting misconfiguration detected at startup: {Adjustment}", adjustment);
|
||||
builder.Services.AddSingleton(scriptingOptions);
|
||||
|
||||
// lmxopcua#777: the ONE operator-write timeout (DriverWrites:Timeout) every write bound is derived from — the
|
||||
// driver child's cancel + abandon, its queue budget, DriverHostActor's Ask and the inbound-write gateway's Ask.
|
||||
// Checked against the resilience tiers' Write timeouts here; an inconsistency is a startup Warning (a tier
|
||||
// timeout above it never takes effect), never a write that outlives an Ask.
|
||||
var driverWriteBounds = ZB.MOM.WW.OtOpcUa.Host.Configuration.DriverWriteBoundsBinding.Bind(
|
||||
builder.Configuration, out var driverWriteWarnings);
|
||||
foreach (var warning in driverWriteWarnings)
|
||||
Log.Warning("Driver write misconfiguration detected at startup: {DriverWriteWarning}", warning);
|
||||
Log.Information("Driver write bounds: {DriverWriteBounds}",
|
||||
ZB.MOM.WW.OtOpcUa.Host.Configuration.DriverWriteBoundsBinding.Describe(driverWriteBounds));
|
||||
builder.Services.AddSingleton(driverWriteBounds);
|
||||
|
||||
builder.Services.AddSingleton<RoslynVirtualTagEvaluator>(sp =>
|
||||
new RoslynVirtualTagEvaluator(
|
||||
sp.GetRequiredService<ILoggerFactory>().CreateLogger<RoslynVirtualTagEvaluator>(),
|
||||
|
||||
@@ -68,6 +68,10 @@
|
||||
"Capacity": 1000000,
|
||||
"DeadLetterRetentionDays": 30
|
||||
},
|
||||
"DriverWrites": {
|
||||
"_comment": "lmxopcua#777: the ONE operator-write timeout every write bound derives from. A write's token is cancelled at Timeout; a token-ignoring write is abandoned min(Timeout/2, 2s) later; a write may wait that long again for a slot; DriverHostActor's Ask and the inbound-write gateway's Ask sit 1s and 2s outside the sum, so a write is always answered by the driver before any Ask can report it failed. A resilience-tier Write timeout above this never takes effect (warned at startup and at driver spawn). Default 10s = the Tier C Write timeout.",
|
||||
"Timeout": "00:00:10"
|
||||
},
|
||||
"LocalDb": {
|
||||
"_comment": "Node-local SQLite store caching the deployed-configuration artifact, so a driver node can boot from its last-known-good config when central SQL is unreachable. Registered on driver-role nodes only (see LocalDbRegistration). Storage is unconditional; replication to the redundant pair peer is default-OFF.",
|
||||
"Path": "./data/otopcua-localdb.db",
|
||||
|
||||
@@ -22,8 +22,6 @@ namespace ZB.MOM.WW.OtOpcUa.Runtime.Drivers;
|
||||
/// </summary>
|
||||
public sealed class ActorNodeWriteGateway : IOpcUaNodeWriteGateway
|
||||
{
|
||||
/// <summary>Default Ask timeout — matches the legacy inline lambda in the hosted service.</summary>
|
||||
private static readonly TimeSpan DefaultAskTimeout = TimeSpan.FromSeconds(10);
|
||||
|
||||
private readonly Func<IActorRef?> _resolveDriverHost;
|
||||
private readonly TimeSpan _askTimeout;
|
||||
@@ -34,12 +32,15 @@ public sealed class ActorNodeWriteGateway : IOpcUaNodeWriteGateway
|
||||
/// returns null until the actor has registered (StartAsync ordering — the actor registers AFTER the host
|
||||
/// wires this gateway).</param>
|
||||
/// <param name="logger">Logger for dropped/rejected/timed-out writes.</param>
|
||||
/// <param name="askTimeout">Ask timeout; defaults to 10s (the legacy lambda's value).</param>
|
||||
/// <param name="askTimeout">Ask timeout; defaults to <see cref="DriverWriteBounds.GatewayAsk"/> of
|
||||
/// <see cref="DriverWriteBounds.Default"/>. Production passes the GatewayAsk of the node's configured write
|
||||
/// bounds, so this Ask always outlasts the host's (lmxopcua#777) — an outer Ask that fires first reports the
|
||||
/// write failed, and the node reverts, while the device may still take the value.</param>
|
||||
public ActorNodeWriteGateway(Func<IActorRef?> resolveDriverHost, ILogger logger, TimeSpan? askTimeout = null)
|
||||
{
|
||||
_resolveDriverHost = resolveDriverHost ?? throw new ArgumentNullException(nameof(resolveDriverHost));
|
||||
_logger = logger ?? throw new ArgumentNullException(nameof(logger));
|
||||
_askTimeout = askTimeout ?? DefaultAskTimeout;
|
||||
_askTimeout = askTimeout ?? DriverWriteBounds.Default.GatewayAsk;
|
||||
}
|
||||
|
||||
/// <inheritdoc />
|
||||
|
||||
@@ -479,6 +479,14 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS
|
||||
/// <summary>Where the synchronous native-ack pre-check snapshot is published (lmxopcua#760); null ⇒ none.</summary>
|
||||
private readonly NativeAlarmAckPreflightView? _ackPreflightView;
|
||||
|
||||
/// <summary>The write bounds every child is spawned with and this host's write Ask is derived from
|
||||
/// (lmxopcua#777).</summary>
|
||||
private readonly DriverWriteBounds _writeBounds;
|
||||
|
||||
/// <summary>The clock the write bounds run on, shared with every child so a write's issue stamp
|
||||
/// (<see cref="DriverInstanceActor.WriteAttribute.IssuedAt"/>) means the same thing on both sides.</summary>
|
||||
private readonly TimeProvider _writeClock;
|
||||
|
||||
/// <summary>Whether any redundancy snapshot has reached this host — the HistoryUpdate gate stays
|
||||
/// closed until one has (see <see cref="PrimaryGatePolicy.ShouldServeHistoryUpdate"/>).</summary>
|
||||
private bool _redundancySnapshotReceived;
|
||||
@@ -721,6 +729,11 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS
|
||||
/// HistoryUpdate Primary gate. Null on nodes that wire no redundancy manager.</param>
|
||||
/// <param name="scripting">The node-wide <c>Scripting</c> run-guard options handed to the scripted-alarm
|
||||
/// engine. Null ⇒ the compiled defaults.</param>
|
||||
/// <param name="ackPreflightView">Where the native-ack pre-check snapshot is published. Null ⇒ none.</param>
|
||||
/// <param name="writeBounds">The write timeout every child is spawned with, and from which this host's write
|
||||
/// Ask is derived (<see cref="DriverWriteBounds.HostAsk"/>). Null ⇒ <see cref="DriverWriteBounds.Default"/>.</param>
|
||||
/// <param name="writeClock">The clock the write bounds (and the host's write Ask) run on, shared with every
|
||||
/// child. Null ⇒ <see cref="TimeProvider.System"/>.</param>
|
||||
/// <returns>The Akka.NET <see cref="Akka.Actor.Props"/> used to spawn this actor.</returns>
|
||||
public static Props Props(
|
||||
IDbContextFactory<OtOpcUaConfigDbContext>? dbFactory,
|
||||
@@ -756,7 +769,9 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS
|
||||
bool captureWhilePrimaryUnreachable = true,
|
||||
IHistoryUpdateRoleView? historyUpdateRoleView = null,
|
||||
ScriptingOptions? scripting = null,
|
||||
NativeAlarmAckPreflightView? ackPreflightView = null) =>
|
||||
NativeAlarmAckPreflightView? ackPreflightView = null,
|
||||
DriverWriteBounds? writeBounds = null,
|
||||
TimeProvider? writeClock = null) =>
|
||||
// WARNING: this forwarding list is POSITIONAL, and Props.Create compiles it into an
|
||||
// expression tree. Six IActorRef? parameters and several interface-typed ones mean a
|
||||
// mis-ordered argument is usually type-compatible and therefore compiles clean, then binds
|
||||
@@ -770,7 +785,7 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS
|
||||
alarmStateStore, telemetryHub, primaryRoleSource, historizationRoleView,
|
||||
sharesHistorizationScopeWithLocal, applyWedgeTimeout, reprovisionBackoff, commsRecoveryAuditDelay,
|
||||
sharesAlarmQueueWithLocal, applyOutcomeTimeout, captureWhilePrimaryUnreachable, historyUpdateRoleView,
|
||||
scripting, ackPreflightView));
|
||||
scripting, ackPreflightView, writeBounds, writeClock));
|
||||
|
||||
/// <summary>Initializes a new DriverHostActor with the specified dependencies.</summary>
|
||||
/// <param name="dbFactory">Database context factory for configuration database access.</param>
|
||||
@@ -861,6 +876,10 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS
|
||||
/// engine. Null ⇒ the compiled defaults.</param>
|
||||
/// <param name="ackPreflightView">Where the condition → owning-driver native-ack pre-check snapshot is published
|
||||
/// for the node manager's synchronous refusal (lmxopcua#760). Null ⇒ nothing published (no pre-check).</param>
|
||||
/// <param name="writeBounds">The write timeout every child is spawned with, and from which this host's write
|
||||
/// Ask is derived (<see cref="DriverWriteBounds.HostAsk"/>). Null ⇒ <see cref="DriverWriteBounds.Default"/>.</param>
|
||||
/// <param name="writeClock">The clock the write bounds (and the host's write Ask) run on, shared with every
|
||||
/// child. Null ⇒ <see cref="TimeProvider.System"/>.</param>
|
||||
public DriverHostActor(
|
||||
IDbContextFactory<OtOpcUaConfigDbContext>? dbFactory,
|
||||
CommonsNodeId localNode,
|
||||
@@ -895,8 +914,12 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS
|
||||
bool captureWhilePrimaryUnreachable = true,
|
||||
IHistoryUpdateRoleView? historyUpdateRoleView = null,
|
||||
ScriptingOptions? scripting = null,
|
||||
NativeAlarmAckPreflightView? ackPreflightView = null)
|
||||
NativeAlarmAckPreflightView? ackPreflightView = null,
|
||||
DriverWriteBounds? writeBounds = null,
|
||||
TimeProvider? writeClock = null)
|
||||
{
|
||||
_writeBounds = writeBounds ?? DriverWriteBounds.Default;
|
||||
_writeClock = writeClock ?? TimeProvider.System;
|
||||
_ackPreflightView = ackPreflightView;
|
||||
_scripting = scripting;
|
||||
_captureWhilePrimaryUnreachable = captureWhilePrimaryUnreachable;
|
||||
@@ -2237,17 +2260,30 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS
|
||||
return;
|
||||
}
|
||||
|
||||
// Ask the child a bounded write — DriverInstanceActor bounds the backend call by its write timeout
|
||||
// (DefaultWriteTimeout 5 s; a token-ignoring write is abandoned at 7 s), so the 8s Ask is a safety net
|
||||
// for an actor that never replies. Capture Sender before
|
||||
// the continuation: it runs off the actor thread, where raw Sender is unsafe to read.
|
||||
// Ask the child a bounded write. The child answers every write within DriverWriteBounds.ReplyBound of the
|
||||
// IssuedAt stamped here (queue budget + write timeout + abandon grace — it refuses to START a write that
|
||||
// could not finish inside it), and the Ask is DERIVED from the same write timeout with a margin on top
|
||||
// (lmxopcua#777). So the Ask is only a safety net for a child that never replies: it can no longer fire
|
||||
// first and report a write failed — reverting the node — while the child is still driving the device. It
|
||||
// was a fixed 8 s, and any write timeout above ~6 s let a write outlive it.
|
||||
//
|
||||
// The Ask's deadline is a token on the write clock rather than Akka's scheduler, so it runs on the same
|
||||
// clock as the child's bounds (and a test can step both together). Capture Sender before the
|
||||
// continuation: it runs off the actor thread, where raw Sender is unsafe to read.
|
||||
var replyTo = Sender;
|
||||
var askDeadline = new CancellationTokenSource(_writeBounds.HostAsk, _writeClock);
|
||||
entry.Actor.Ask<DriverInstanceActor.WriteAttributeResult>(
|
||||
new DriverInstanceActor.WriteAttribute(target.RawPath, msg.Value!), TimeSpan.FromSeconds(8))
|
||||
new DriverInstanceActor.WriteAttribute(target.RawPath, msg.Value!, _writeClock.GetTimestamp()),
|
||||
System.Threading.Timeout.InfiniteTimeSpan,
|
||||
askDeadline.Token)
|
||||
.ContinueWith(
|
||||
t => t.IsCompletedSuccessfully
|
||||
? new NodeWriteResult(t.Result.Success, t.Result.Reason)
|
||||
: new NodeWriteResult(false, "write timeout"),
|
||||
t =>
|
||||
{
|
||||
askDeadline.Dispose();
|
||||
return t.IsCompletedSuccessfully
|
||||
? new NodeWriteResult(t.Result.Success, t.Result.Reason)
|
||||
: new NodeWriteResult(false, "write timeout");
|
||||
},
|
||||
CancellationToken.None,
|
||||
TaskContinuationOptions.None,
|
||||
TaskScheduler.Default)
|
||||
@@ -3890,7 +3926,9 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS
|
||||
driver!,
|
||||
healthPublisher: _healthPublisher,
|
||||
clusterId: clusterId,
|
||||
invoker: invoker),
|
||||
invoker: invoker,
|
||||
writeTimeout: _writeBounds.Timeout,
|
||||
writeClock: _writeClock),
|
||||
actorName);
|
||||
child.Tell(new DriverInstanceActor.InitializeRequested(spec.DriverConfig));
|
||||
}
|
||||
|
||||
@@ -61,7 +61,15 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers
|
||||
/// <param name="DriverInstanceId">The driver instance that was restarted.</param>
|
||||
/// <param name="Reason">The type name of the exception that tripped supervision, for the host's log.</param>
|
||||
public sealed record ReprovisionRequested(string DriverInstanceId, string Reason);
|
||||
public sealed record WriteAttribute(string TagId, object Value);
|
||||
/// <summary>An operator write of <paramref name="Value"/> to the tag <paramref name="TagId"/>.</summary>
|
||||
/// <param name="TagId">The tag's driver wire reference (RawPath).</param>
|
||||
/// <param name="Value">The value to write.</param>
|
||||
/// <param name="IssuedAt">When the asker issued the write, as a timestamp of the actor's write clock
|
||||
/// (<see cref="TimeProvider.GetTimestamp"/>); null = when the actor receives it. <see cref="DriverHostActor"/>
|
||||
/// stamps it so the time the write spends in this actor's MAILBOX counts against the queue budget too: a write
|
||||
/// delayed there (behind a hanging alarm subscribe, say) and then started could otherwise still be running
|
||||
/// when the host's Ask fires (lmxopcua#777).</param>
|
||||
public sealed record WriteAttribute(string TagId, object Value, long? IssuedAt = null);
|
||||
public sealed record WriteAttributeResult(bool Success, string? Reason);
|
||||
/// <summary>
|
||||
/// Sent by <see cref="DriverHostActor"/> when an OPC UA client Acknowledges a NATIVE Part 9
|
||||
@@ -216,10 +224,13 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers
|
||||
|
||||
/// <summary>
|
||||
/// How long a queued write may wait for a slot before it is failed <c>BadTimeout</c> instead of being
|
||||
/// written. The host's write Ask is 8 s: a write older than that has already been reported as a timeout
|
||||
/// (and its node reverted) upstream, so writing it late would push a stale value to the device.
|
||||
/// written, when no explicit budget is given: <see cref="DriverWriteBounds.QueueBudget"/> of the write
|
||||
/// timeout in force (lmxopcua#777 — it was a fixed 8 s, equal to the host's whole Ask, so a write dequeued
|
||||
/// just inside it could still be running when the Ask fired). The host's Ask covers this budget PLUS a full
|
||||
/// write (<see cref="DriverWriteBounds.HostAsk"/>), so a write started inside the budget is always answered
|
||||
/// before the host reports it failed.
|
||||
/// </summary>
|
||||
public static readonly TimeSpan DefaultQueuedWriteBudget = TimeSpan.FromSeconds(8);
|
||||
public static readonly TimeSpan DefaultQueuedWriteBudget = DriverWriteBounds.Default.QueueBudget;
|
||||
|
||||
/// <summary>
|
||||
/// Pessimistic bound on one <see cref="IDriver.InitializeAsync"/> attempt. The invoker's tier timeout is
|
||||
@@ -236,23 +247,19 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers
|
||||
/// queue.
|
||||
/// </summary>
|
||||
/// <remarks>
|
||||
/// 5 s, so both bounds (5 s cancel, 7 s abandon) answer inside the host's 8 s write Ask
|
||||
/// (<c>DriverHostActor</c>): a write that outlives the Ask has already been reported as a timeout and its
|
||||
/// node reverted upstream, so letting it run on would let the device take a value the client was told
|
||||
/// failed. Until lmxopcua#774 the cancel was a hard-coded 5 s that ignored this value, and the default was
|
||||
/// 10 s — a configured bound above 5 s never took effect.
|
||||
/// <see cref="DriverWriteBounds.DefaultTimeout"/>. The host's write Ask is DERIVED from the same value
|
||||
/// (<see cref="DriverWriteBounds.HostAsk"/>, lmxopcua#777), so any configured write timeout is answered
|
||||
/// inside it; until then the Ask was a fixed 8 s and a write timeout above ~6 s let a write outlive it.
|
||||
/// </remarks>
|
||||
public static readonly TimeSpan DefaultWriteTimeout = TimeSpan.FromSeconds(5);
|
||||
public static readonly TimeSpan DefaultWriteTimeout = DriverWriteBounds.DefaultTimeout;
|
||||
|
||||
/// <summary>
|
||||
/// How long after a write's token is cancelled (<see cref="DefaultWriteTimeout"/>) a write that has not
|
||||
/// returned is abandoned as token-ignoring: half the write timeout, at most 2 s. Long enough for a driver
|
||||
/// that honours its token to unwind and answer the cancellation itself.
|
||||
/// How long after a write's token is cancelled a write that has not returned is abandoned as
|
||||
/// token-ignoring; see <see cref="DriverWriteBounds.AbandonGrace"/>.
|
||||
/// </summary>
|
||||
/// <param name="writeTimeout">The write timeout in force.</param>
|
||||
/// <returns>The grace between cancelling the write's token and abandoning the write.</returns>
|
||||
internal static TimeSpan WriteAbandonGrace(TimeSpan writeTimeout) =>
|
||||
TimeSpan.FromTicks(Math.Min(writeTimeout.Ticks / 2, TimeSpan.FromSeconds(2).Ticks));
|
||||
internal static TimeSpan WriteAbandonGrace(TimeSpan writeTimeout) => DriverWriteBounds.AbandonGraceFor(writeTimeout);
|
||||
|
||||
/// <summary>Self-sent by a write task once it has replied, releasing its slot and its tag.</summary>
|
||||
/// <param name="TagId">The tag the write held.</param>
|
||||
@@ -292,8 +299,8 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers
|
||||
|
||||
/// <summary>Writes waiting for a slot (or for their tag), in arrival order, stamped with arrival time.
|
||||
/// Failed fast on every exit from <c>Connected</c>, so a queued write never reaches a driver that is
|
||||
/// reconnecting; failed <c>BadTimeout</c> at dequeue once older than <see cref="_queuedWriteBudget"/>.</summary>
|
||||
private readonly LinkedList<(WriteAttribute Write, IActorRef ReplyTo, DateTime EnqueuedUtc)> _pendingWrites = new();
|
||||
/// reconnecting; failed <c>BadTimeout</c> once older than <see cref="_queuedWriteBudget"/> since issued.</summary>
|
||||
private readonly LinkedList<(WriteAttribute Write, IActorRef ReplyTo, long IssuedAt)> _pendingWrites = new();
|
||||
|
||||
/// <summary>Tags with a write currently inside the driver. A second write to the same tag waits in the
|
||||
/// queue behind it — two concurrent writes v1, v2 to one tag could otherwise land as v1.</summary>
|
||||
@@ -453,7 +460,7 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers
|
||||
/// <param name="maxWritesInFlight">Optional cap on concurrent writes against the driver; defaults to
|
||||
/// <see cref="DefaultMaxWritesInFlight"/>. Writes beyond it queue in arrival order.</param>
|
||||
/// <param name="queuedWriteBudget">Optional bound on how long a queued write may wait before it is failed
|
||||
/// <c>BadTimeout</c>; defaults to <see cref="DefaultQueuedWriteBudget"/>.</param>
|
||||
/// <c>BadTimeout</c>; defaults to <see cref="DriverWriteBounds.QueueBudget"/> of the write timeout in force.</param>
|
||||
/// <param name="initializeTimeout">Optional pessimistic bound on one initialise attempt; defaults to
|
||||
/// <see cref="DefaultInitializeTimeout"/>.</param>
|
||||
/// <param name="writeTimeout">Optional bound on one write round-trip (its token is cancelled when it elapses;
|
||||
@@ -533,7 +540,7 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers
|
||||
/// <param name="maxWritesInFlight">Cap on concurrent writes; defaults to
|
||||
/// <see cref="DefaultMaxWritesInFlight"/> when null or non-positive.</param>
|
||||
/// <param name="queuedWriteBudget">Bound on a queued write's wait; defaults to
|
||||
/// <see cref="DefaultQueuedWriteBudget"/> when null or non-positive.</param>
|
||||
/// <see cref="DriverWriteBounds.QueueBudget"/> of the write timeout when null or non-positive.</param>
|
||||
/// <param name="initializeTimeout">Pessimistic bound on one initialise attempt; defaults to
|
||||
/// <see cref="DefaultInitializeTimeout"/> when null or non-positive.</param>
|
||||
/// <param name="writeTimeout">Bound on one write round-trip; defaults to
|
||||
@@ -559,7 +566,9 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers
|
||||
_writeClock = writeClock ?? TimeProvider.System;
|
||||
_incarnation = Interlocked.Increment(ref s_initGenerationSource);
|
||||
_maxWritesInFlight = maxWritesInFlight is > 0 ? maxWritesInFlight.Value : DefaultMaxWritesInFlight;
|
||||
_queuedWriteBudget = queuedWriteBudget is { } qb && qb > TimeSpan.Zero ? qb : DefaultQueuedWriteBudget;
|
||||
_queuedWriteBudget = queuedWriteBudget is { } qb && qb > TimeSpan.Zero
|
||||
? qb
|
||||
: new DriverWriteBounds(_writeTimeout).QueueBudget;
|
||||
_initializeTimeout = initializeTimeout is { } it && it > TimeSpan.Zero ? it : DefaultInitializeTimeout;
|
||||
_driver = driver;
|
||||
_invoker = invoker ?? NullDriverCapabilityInvoker.Instance;
|
||||
@@ -646,7 +655,7 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers
|
||||
{
|
||||
Receive<InitializeRequested>(msg => InitializeAsync(msg.DriverConfigJson));
|
||||
// Fast-fail writes while still connecting — without this the inbound WriteAttribute dead-letters
|
||||
// and DriverHostActor.HandleRouteNodeWrite waits its full 8s Ask before reporting a generic
|
||||
// and DriverHostActor.HandleRouteNodeWrite waits its full Ask before reporting a generic
|
||||
// "write timeout". Synchronous Receive: Sender.Tell on the actor thread is safe.
|
||||
Receive<WriteAttribute>(_ =>
|
||||
Sender.Tell(new WriteAttributeResult(false, "driver not connected")));
|
||||
@@ -919,7 +928,7 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers
|
||||
/// non-empty desired set, so it costs nothing on a driver with no alarm tags.
|
||||
/// <para>A persistently failing subscribe backs off: after the n-th consecutive failure the reconcile
|
||||
/// skips 2^(n-1) − 1 ticks, capped at <see cref="MaxAlarmRetrySkipTicks"/>, because a HANGING subscribe
|
||||
/// holds the mailbox for its full 10 s bound (writes queued behind it can miss the host's 8 s Ask). A
|
||||
/// holds the mailbox for its full 10 s bound (writes delayed behind it eat into their queue budget). A
|
||||
/// Connected entry or a deploy's desired-set push still retries immediately — only the timer backs off.</para>
|
||||
/// </summary>
|
||||
private void ReconcileAlarmSubscription()
|
||||
@@ -1151,9 +1160,13 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers
|
||||
return;
|
||||
}
|
||||
|
||||
// Monotonic, so a wall-clock step cannot expire (or extend) a queued write.
|
||||
var issuedAt = msg.IssuedAt ?? _writeClock.GetTimestamp();
|
||||
if (TryExpireWrite(replyTo, issuedAt)) return;
|
||||
|
||||
if (_writesInFlight >= _maxWritesInFlight || _tagsInFlight.Contains(msg.TagId))
|
||||
{
|
||||
_pendingWrites.AddLast((msg, replyTo, DateTime.UtcNow));
|
||||
_pendingWrites.AddLast((msg, replyTo, issuedAt));
|
||||
return;
|
||||
}
|
||||
|
||||
@@ -1258,21 +1271,22 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers
|
||||
private void DrainWriteQueue()
|
||||
{
|
||||
if (_driver is not IWritable writable) return;
|
||||
var now = DateTime.UtcNow;
|
||||
var node = _pendingWrites.First;
|
||||
while (node is not null && _writesInFlight < _maxWritesInFlight)
|
||||
|
||||
// Expire FIRST, over the whole queue, regardless of free slots: a write past its budget must be answered
|
||||
// now, not whenever a slot next frees and the scan happens to reach it — the host's Ask covers the budget
|
||||
// plus one write (DriverWriteBounds.HostAsk), not an unbounded wait behind a full set of slots.
|
||||
for (var node = _pendingWrites.First; node is not null;)
|
||||
{
|
||||
var next = node.Next;
|
||||
var (write, replyTo, enqueuedUtc) = node.Value;
|
||||
if (now - enqueuedUtc > _queuedWriteBudget)
|
||||
{
|
||||
// Already reported as a timeout upstream (the host's Ask is the budget) and the node reverted —
|
||||
// writing it now would push a stale value to the device. BadTimeout, never written.
|
||||
_pendingWrites.Remove(node);
|
||||
replyTo.Tell(new WriteAttributeResult(false,
|
||||
$"StatusCode=0x800A0000 (BadTimeout: queued for {(now - enqueuedUtc).TotalSeconds:F1}s, over the {_queuedWriteBudget.TotalSeconds:F0}s write budget)"));
|
||||
}
|
||||
else if (!_tagsInFlight.Contains(write.TagId))
|
||||
if (TryExpireWrite(node.Value.ReplyTo, node.Value.IssuedAt)) _pendingWrites.Remove(node);
|
||||
node = next;
|
||||
}
|
||||
|
||||
for (var node = _pendingWrites.First; node is not null && _writesInFlight < _maxWritesInFlight;)
|
||||
{
|
||||
var next = node.Next;
|
||||
var (write, replyTo, _) = node.Value;
|
||||
if (!_tagsInFlight.Contains(write.TagId))
|
||||
{
|
||||
_pendingWrites.Remove(node);
|
||||
LaunchWrite(writable, write, replyTo);
|
||||
@@ -1282,6 +1296,23 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Fails a write that has waited longer than <see cref="_queuedWriteBudget"/> since it was issued (in the
|
||||
/// mailbox or the queue). Only a write inside the budget may start: the host's Ask covers the budget plus one
|
||||
/// full write (<see cref="DriverWriteBounds.HostAsk"/>), so a write started later could still be running
|
||||
/// when the Ask reports it failed and its node is reverted — and the device could then take a value the
|
||||
/// client was told failed. BadTimeout, never written.
|
||||
/// </summary>
|
||||
/// <returns>True when the write was expired (and answered).</returns>
|
||||
private bool TryExpireWrite(IActorRef replyTo, long issuedAt)
|
||||
{
|
||||
var waited = _writeClock.GetElapsedTime(issuedAt);
|
||||
if (waited <= _queuedWriteBudget) return false;
|
||||
replyTo.Tell(new WriteAttributeResult(false,
|
||||
$"StatusCode=0x800A0000 (BadTimeout: queued for {waited.TotalSeconds:F1}s, over the {_queuedWriteBudget.TotalSeconds:F0}s write budget)"));
|
||||
return true;
|
||||
}
|
||||
|
||||
/// <summary>Replies to every queued write with <paramref name="reason"/>; called on each exit from
|
||||
/// <c>Connected</c>. In-flight writes are left to finish — they already hold the device and reply with its
|
||||
/// real outcome.</summary>
|
||||
|
||||
@@ -0,0 +1,105 @@
|
||||
namespace ZB.MOM.WW.OtOpcUa.Runtime.Drivers;
|
||||
|
||||
/// <summary>
|
||||
/// Every bound on an operator write's path, derived from ONE configured value: the write timeout
|
||||
/// (<c>DriverWrites:Timeout</c>). lmxopcua#777.
|
||||
/// </summary>
|
||||
/// <remarks>
|
||||
/// <para>
|
||||
/// A write passes three bounds on its way to the device, and they used to be configured independently:
|
||||
/// the gateway's Ask of <see cref="DriverHostActor"/> (10 s), the host's Ask of the
|
||||
/// <see cref="DriverInstanceActor"/> child (a fixed 8 s), and the child's own write timeout plus the
|
||||
/// queued-write budget (5 s + up to 2 s, and 8 s in the queue). An outer Ask that fires first reports the
|
||||
/// write failed and the node manager reverts the node, while the child is still driving the device and the
|
||||
/// device may still take the value — the client is told "failed" for a write that applied. Raising the
|
||||
/// write timeout above ~6 s silently produced exactly that.
|
||||
/// </para>
|
||||
/// <para>
|
||||
/// Here every bound is derived from <see cref="Timeout"/>, innermost first, so each outer bound covers
|
||||
/// the worst case of everything inside it:
|
||||
/// <list type="number">
|
||||
/// <item><see cref="Timeout"/> — the write's cancellation token fires.</item>
|
||||
/// <item><see cref="RunBound"/> = timeout + <see cref="AbandonGrace"/> — a write that ignored its
|
||||
/// token is abandoned (answers <c>BadTimeout</c>).</item>
|
||||
/// <item><see cref="QueueBudget"/> = <see cref="RunBound"/> — the longest a write may wait for a
|
||||
/// slot (or for an earlier write to the same tag) before it is failed instead of started.</item>
|
||||
/// <item><see cref="ReplyBound"/> = queue + run — the child answers every write within this of its
|
||||
/// arrival.</item>
|
||||
/// <item><see cref="HostAsk"/> = reply + <see cref="ReplyMargin"/>, and
|
||||
/// <see cref="GatewayAsk"/> = host Ask + margin.</item>
|
||||
/// </list>
|
||||
/// The Asks are safety nets for an actor that never replies, not a second write timeout.
|
||||
/// </para>
|
||||
/// <para>
|
||||
/// The Phase 6.1 resilience tier's per-capability Write timeout (<c>DriverResilienceOptions</c>) runs
|
||||
/// INSIDE <see cref="Timeout"/>: the actor's token is the outer cap, so a tier Write timeout above
|
||||
/// <see cref="Timeout"/> never takes effect. The Host warns about that at startup (tier defaults) and at
|
||||
/// each driver spawn (per-instance <c>ResilienceConfig</c> overrides).
|
||||
/// </para>
|
||||
/// </remarks>
|
||||
public sealed record DriverWriteBounds
|
||||
{
|
||||
/// <summary>The configuration section the write timeout is bound from.</summary>
|
||||
public const string SectionName = "DriverWrites";
|
||||
|
||||
/// <summary>
|
||||
/// Default write timeout. 10 s — the longest resilience-tier Write timeout (Tier C), so no tier default
|
||||
/// is capped below its own policy. Tier A/B writes are cancelled by their tighter tier timeout first.
|
||||
/// </summary>
|
||||
public static readonly TimeSpan DefaultTimeout = TimeSpan.FromSeconds(10);
|
||||
|
||||
/// <summary>Largest accepted write timeout; a larger configured value is clamped (with a startup warning).</summary>
|
||||
public static readonly TimeSpan MaxTimeout = TimeSpan.FromMinutes(5);
|
||||
|
||||
/// <summary>
|
||||
/// Headroom each Ask keeps over the bound it wraps, for mailbox and dispatch latency between the reply being
|
||||
/// sent and the Ask observing it.
|
||||
/// </summary>
|
||||
public static readonly TimeSpan ReplyMargin = TimeSpan.FromSeconds(1);
|
||||
|
||||
/// <summary>The write bounds derived from <see cref="DefaultTimeout"/>.</summary>
|
||||
public static DriverWriteBounds Default { get; } = new(DefaultTimeout);
|
||||
|
||||
/// <summary>Creates the bounds for a write timeout.</summary>
|
||||
/// <param name="timeout">The write timeout; must be positive.</param>
|
||||
/// <exception cref="ArgumentOutOfRangeException">The timeout is zero or negative.</exception>
|
||||
public DriverWriteBounds(TimeSpan timeout)
|
||||
{
|
||||
ArgumentOutOfRangeException.ThrowIfLessThanOrEqual(timeout, TimeSpan.Zero);
|
||||
Timeout = timeout;
|
||||
}
|
||||
|
||||
/// <summary>When a write's cancellation token is cancelled (the write answers <c>"write timeout"</c>).</summary>
|
||||
public TimeSpan Timeout { get; }
|
||||
|
||||
/// <summary>
|
||||
/// How long after <see cref="Timeout"/> a write that has not returned is abandoned as token-ignoring: half
|
||||
/// the timeout, at most 2 s. Long enough for a driver that honours its token to answer the cancellation
|
||||
/// itself.
|
||||
/// </summary>
|
||||
public TimeSpan AbandonGrace => AbandonGraceFor(Timeout);
|
||||
|
||||
/// <summary>The latest a running write is answered after it starts: <see cref="Timeout"/> + <see cref="AbandonGrace"/>.</summary>
|
||||
public TimeSpan RunBound => Timeout + AbandonGrace;
|
||||
|
||||
/// <summary>
|
||||
/// The longest a write may wait for a slot before it is failed <c>BadTimeout</c> instead of being written:
|
||||
/// one <see cref="RunBound"/>, i.e. long enough for an earlier write to the same tag to finish.
|
||||
/// </summary>
|
||||
public TimeSpan QueueBudget => RunBound;
|
||||
|
||||
/// <summary>The latest the driver child answers a write after it arrives: <see cref="QueueBudget"/> + <see cref="RunBound"/>.</summary>
|
||||
public TimeSpan ReplyBound => QueueBudget + RunBound;
|
||||
|
||||
/// <summary><see cref="DriverHostActor"/>'s Ask of the child: <see cref="ReplyBound"/> + <see cref="ReplyMargin"/>.</summary>
|
||||
public TimeSpan HostAsk => ReplyBound + ReplyMargin;
|
||||
|
||||
/// <summary><see cref="ActorNodeWriteGateway"/>'s Ask of the host: <see cref="HostAsk"/> + <see cref="ReplyMargin"/>.</summary>
|
||||
public TimeSpan GatewayAsk => HostAsk + ReplyMargin;
|
||||
|
||||
/// <summary>See <see cref="AbandonGrace"/>.</summary>
|
||||
/// <param name="timeout">The write timeout in force.</param>
|
||||
/// <returns>The grace between cancelling the write's token and abandoning the write.</returns>
|
||||
public static TimeSpan AbandonGraceFor(TimeSpan timeout) =>
|
||||
TimeSpan.FromTicks(Math.Min(timeout.Ticks / 2, TimeSpan.FromSeconds(2).Ticks));
|
||||
}
|
||||
@@ -660,7 +660,11 @@ public static class ServiceCollectionExtensions
|
||||
// engine. Unbound in test graphs ⇒ null ⇒ the compiled defaults.
|
||||
scripting: resolver.GetService<ScriptingOptions>(),
|
||||
// Where the native-ack pre-check snapshot is published (lmxopcua#760).
|
||||
ackPreflightView: resolver.GetService<NativeAlarmAckPreflightView>()),
|
||||
ackPreflightView: resolver.GetService<NativeAlarmAckPreflightView>(),
|
||||
// The one configured write timeout (DriverWrites:Timeout, bound + validated in Program.cs) the
|
||||
// children's write bounds AND this host's write Ask are derived from (lmxopcua#777). Unbound in
|
||||
// test graphs ⇒ null ⇒ DriverWriteBounds.Default.
|
||||
writeBounds: resolver.GetService<DriverWriteBounds>()),
|
||||
DriverHostActorName);
|
||||
registry.Register<DriverHostActorKey>(driverHost);
|
||||
|
||||
|
||||
+29
@@ -200,6 +200,35 @@ public sealed class DriverCapabilityInvokerFactoryTests
|
||||
logger.Entries.ShouldContain(e => e.Level == LogLevel.Warning && e.Message.Contains("resilience config", StringComparison.OrdinalIgnoreCase));
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// lmxopcua#777: the node's write timeout cancels every write before the Write policy's own timeout can, so
|
||||
/// a per-instance Write timeout above it never takes effect — the spawn says so. Positive control: a Write
|
||||
/// timeout within the ceiling, and no ceiling at all, log nothing about it.
|
||||
/// </summary>
|
||||
[Fact]
|
||||
public void Create_warns_when_the_Write_timeout_exceeds_the_write_timeout_ceiling()
|
||||
{
|
||||
const string config = "{\"capabilityPolicies\":{\"Write\":{\"timeoutSeconds\":30}}}";
|
||||
static bool IsCeilingWarning((LogLevel Level, string Message) e) =>
|
||||
e.Level == LogLevel.Warning && e.Message.Contains("DriverWrites:Timeout", StringComparison.Ordinal);
|
||||
|
||||
var over = new CapturingLogger();
|
||||
new DriverCapabilityInvokerFactory(new DriverResiliencePipelineBuilder(), new DriverResilienceStatusTracker(),
|
||||
_ => DriverTier.A, over, writeTimeoutCeiling: TimeSpan.FromSeconds(10))
|
||||
.Create("d1", "Modbus", config);
|
||||
over.Entries.Any(e => IsCeilingWarning(e) && e.Message.Contains("30s")).ShouldBeTrue();
|
||||
|
||||
var within = new CapturingLogger();
|
||||
new DriverCapabilityInvokerFactory(new DriverResiliencePipelineBuilder(), new DriverResilienceStatusTracker(),
|
||||
_ => DriverTier.A, within, writeTimeoutCeiling: TimeSpan.FromSeconds(30))
|
||||
.Create("d1", "Modbus", config);
|
||||
within.Entries.Any(IsCeilingWarning).ShouldBeFalse();
|
||||
|
||||
var noCeiling = new CapturingLogger();
|
||||
MakeFactory(new DriverResiliencePipelineBuilder(), logger: noCeiling).Create("d1", "Modbus", config);
|
||||
noCeiling.Entries.Any(IsCeilingWarning).ShouldBeFalse();
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// 01/S-6 production-wiring proof: the exact <c>DriverHostActor.SpawnChild</c> path
|
||||
/// (<c>factory.Create(id, type, hostileJson)</c>) must yield a WORKING invoker whose capability
|
||||
|
||||
+84
@@ -0,0 +1,84 @@
|
||||
using Microsoft.Extensions.Configuration;
|
||||
using Shouldly;
|
||||
using Xunit;
|
||||
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
|
||||
using ZB.MOM.WW.OtOpcUa.Core.Resilience;
|
||||
using ZB.MOM.WW.OtOpcUa.Host.Configuration;
|
||||
using ZB.MOM.WW.OtOpcUa.Runtime.Drivers;
|
||||
|
||||
namespace ZB.MOM.WW.OtOpcUa.Host.Tests.Configuration;
|
||||
|
||||
/// <summary>
|
||||
/// lmxopcua#777: <c>DriverWrites:Timeout</c> is the one value every write bound derives from, and it is checked
|
||||
/// against the resilience tiers' Write timeouts at startup.
|
||||
/// </summary>
|
||||
public sealed class DriverWriteBoundsBindingTests
|
||||
{
|
||||
private static IConfiguration Config(Dictionary<string, string?> values) =>
|
||||
new ConfigurationBuilder().AddInMemoryCollection(values).Build();
|
||||
|
||||
/// <summary>The shipped default is consistent with every tier — no tier's Write policy is capped below itself.</summary>
|
||||
[Fact]
|
||||
public void An_absent_key_yields_the_default_which_fits_every_tier()
|
||||
{
|
||||
var bounds = DriverWriteBoundsBinding.Bind(Config(new()), out var warnings);
|
||||
|
||||
warnings.ShouldBeEmpty();
|
||||
bounds.ShouldBe(DriverWriteBounds.Default);
|
||||
foreach (var tier in Enum.GetValues<DriverTier>())
|
||||
{
|
||||
TimeSpan.FromSeconds(DriverResilienceOptions.GetTierDefaults(tier)[DriverCapability.Write].TimeoutSeconds)
|
||||
.ShouldBeLessThanOrEqualTo(bounds.Timeout, $"tier {tier}");
|
||||
}
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void The_configured_timeout_binds_and_every_bound_derives_from_it()
|
||||
{
|
||||
var bounds = DriverWriteBoundsBinding.Bind(Config(new() { ["DriverWrites:Timeout"] = "00:00:12" }), out var warnings);
|
||||
|
||||
warnings.ShouldBeEmpty();
|
||||
bounds.Timeout.ShouldBe(TimeSpan.FromSeconds(12));
|
||||
bounds.HostAsk.ShouldBe(TimeSpan.FromSeconds(12 + 2 + 12 + 2 + 1));
|
||||
bounds.GatewayAsk.ShouldBe(bounds.HostAsk + DriverWriteBounds.ReplyMargin);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// An inconsistent config is caught at startup: a write timeout below a tier's Write timeout means that tier's
|
||||
/// policy never takes effect, and each such tier is named. Tier A (2 s) fits under 3 s and is not.
|
||||
/// </summary>
|
||||
[Fact]
|
||||
public void A_timeout_below_a_tiers_Write_timeout_is_reported_per_tier()
|
||||
{
|
||||
var bounds = DriverWriteBoundsBinding.Bind(Config(new() { ["DriverWrites:Timeout"] = "00:00:03" }), out var warnings);
|
||||
|
||||
bounds.Timeout.ShouldBe(TimeSpan.FromSeconds(3), "an inconsistency is reported, not overridden");
|
||||
warnings.Count.ShouldBe(2);
|
||||
warnings.ShouldContain(w => w.StartsWith("Tier B ", StringComparison.Ordinal) && w.Contains("(4s)"));
|
||||
warnings.ShouldContain(w => w.StartsWith("Tier C ", StringComparison.Ordinal) && w.Contains("(10s)"));
|
||||
warnings.ShouldNotContain(w => w.StartsWith("Tier A ", StringComparison.Ordinal));
|
||||
}
|
||||
|
||||
[Theory]
|
||||
[InlineData("not-a-timespan")]
|
||||
[InlineData("00:00:00")]
|
||||
[InlineData("-00:00:05")]
|
||||
public void An_unusable_value_falls_back_to_the_default_with_a_warning(string raw)
|
||||
{
|
||||
var bounds = DriverWriteBoundsBinding.Bind(Config(new() { ["DriverWrites:Timeout"] = raw }), out var warnings);
|
||||
|
||||
bounds.ShouldBe(DriverWriteBounds.Default);
|
||||
warnings.Count.ShouldBe(1);
|
||||
warnings[0].ShouldContain("DriverWrites:Timeout");
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void A_value_above_the_max_is_clamped_with_a_warning()
|
||||
{
|
||||
var bounds = DriverWriteBoundsBinding.Bind(Config(new() { ["DriverWrites:Timeout"] = "01:00:00" }), out var warnings);
|
||||
|
||||
bounds.Timeout.ShouldBe(DriverWriteBounds.MaxTimeout);
|
||||
warnings.Count.ShouldBe(1);
|
||||
warnings[0].ShouldContain("clamped");
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,239 @@
|
||||
using System.Text.Json;
|
||||
using Akka.Actor;
|
||||
using Microsoft.EntityFrameworkCore;
|
||||
using Shouldly;
|
||||
using Xunit;
|
||||
using ZB.MOM.WW.OtOpcUa.Commons.Messages.Deploy;
|
||||
using ZB.MOM.WW.OtOpcUa.Commons.OpcUa;
|
||||
using ZB.MOM.WW.OtOpcUa.Commons.Types;
|
||||
using ZB.MOM.WW.OtOpcUa.Configuration;
|
||||
using ZB.MOM.WW.OtOpcUa.Configuration.Entities;
|
||||
using ZB.MOM.WW.OtOpcUa.Configuration.Enums;
|
||||
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
|
||||
using ZB.MOM.WW.OtOpcUa.Runtime.Drivers;
|
||||
using ZB.MOM.WW.OtOpcUa.Runtime.Tests.Harness;
|
||||
|
||||
namespace ZB.MOM.WW.OtOpcUa.Runtime.Tests.Drivers;
|
||||
|
||||
/// <summary>
|
||||
/// lmxopcua#777: <see cref="DriverHostActor"/>'s write Ask is derived from the configured write timeout
|
||||
/// (<see cref="DriverWriteBounds.HostAsk"/>) rather than a fixed 8 s, so a write the child is still entitled to run
|
||||
/// is never reported failed by the Ask — the node would be reverted while the device could still take the value.
|
||||
/// <para>
|
||||
/// Drives a real apply (a real <see cref="DriverInstanceActor"/> child), then routes a held write with the
|
||||
/// host and the child on ONE manual clock: the host's Ask and the child's bounds both run on it, so stepping
|
||||
/// the clock shows which one answers first.
|
||||
/// </para>
|
||||
/// </summary>
|
||||
public sealed class DriverHostActorWriteAskTests : RuntimeActorTestBase
|
||||
{
|
||||
private static readonly NodeId TestNode = NodeId.Parse("driver-write-ask-test");
|
||||
private static readonly RevisionHash RevA = RevisionHash.Parse(new string('a', 64));
|
||||
private static readonly TimeSpan Timeout = TimeSpan.FromSeconds(5);
|
||||
private const string RawNodeId = "ns=2;s=Plant/Modbus/dev1/speed";
|
||||
|
||||
/// <summary>The value the gated driver holds; every other value is written at once.</summary>
|
||||
private const double HeldValue = 99.0;
|
||||
|
||||
/// <summary>
|
||||
/// A write still inside a 9 s write timeout is not cut off at 8 s: the host waits for the child, which
|
||||
/// answers with the device's real outcome. (Mutation-checked: a fixed 8 s Ask on the same clock fails this
|
||||
/// with "write timeout" at the 8.5 s step.)
|
||||
/// </summary>
|
||||
[Fact]
|
||||
public void A_write_within_a_9s_write_timeout_is_not_failed_by_the_host_Ask()
|
||||
{
|
||||
var clock = new ManualTimeProvider();
|
||||
var driver = new GatedWritableDriver();
|
||||
var host = SpawnHostAndApply(driver, new DriverWriteBounds(TimeSpan.FromSeconds(9)), clock);
|
||||
|
||||
var asker = RouteHeldWrite(host, driver, clock);
|
||||
clock.Advance(TimeSpan.FromSeconds(8.5));
|
||||
asker.ExpectNoMsg(TimeSpan.FromMilliseconds(300));
|
||||
|
||||
driver.Release();
|
||||
var result = asker.ExpectMsg<DriverHostActor.NodeWriteResult>(Timeout);
|
||||
result.Success.ShouldBeTrue(result.Reason);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// A token-ignoring write under a 9 s write timeout is answered by the CHILD (abandoned BadTimeout at
|
||||
/// 9 s + its grace), not by the host's Ask: the Ask covers the child's whole bound. Under the fixed 8 s Ask
|
||||
/// this read "write timeout" from the Ask at 8 s while the device still had the write.
|
||||
/// </summary>
|
||||
[Fact]
|
||||
public void A_token_ignoring_write_is_answered_by_the_child_before_the_host_Ask()
|
||||
{
|
||||
var clock = new ManualTimeProvider();
|
||||
var driver = new GatedWritableDriver { IgnoreCancellation = true };
|
||||
var bounds = new DriverWriteBounds(TimeSpan.FromSeconds(9));
|
||||
var host = SpawnHostAndApply(driver, bounds, clock);
|
||||
|
||||
var asker = RouteHeldWrite(host, driver, clock);
|
||||
clock.Advance(bounds.Timeout);
|
||||
asker.ExpectNoMsg(TimeSpan.FromMilliseconds(300));
|
||||
clock.Advance(bounds.AbandonGrace);
|
||||
|
||||
var result = asker.ExpectMsg<DriverHostActor.NodeWriteResult>(Timeout);
|
||||
result.Success.ShouldBeFalse();
|
||||
result.Reason!.ShouldContain("800A0000", customMessage: "the child's BadTimeout, not the host Ask's \"write timeout\"");
|
||||
result.Reason!.ShouldContain("within 9000 ms");
|
||||
bounds.RunBound.ShouldBeLessThan(bounds.HostAsk);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// The mechanism the host's Ask relies on: with an Infinite Akka timeout, the Ask's deadline is the write
|
||||
/// clock's token alone — it does not complete a tick before <see cref="DriverWriteBounds.HostAsk"/>, and it
|
||||
/// does complete (cancelled, which the host reports as "write timeout") at it, for a child that never answers.
|
||||
/// </summary>
|
||||
[Fact]
|
||||
public void An_Ask_with_an_Infinite_timeout_ends_exactly_at_its_clock_deadline()
|
||||
{
|
||||
var clock = new ManualTimeProvider();
|
||||
var bounds = new DriverWriteBounds(TimeSpan.FromSeconds(9));
|
||||
var silentChild = CreateTestProbe();
|
||||
|
||||
using var deadline = new CancellationTokenSource(bounds.HostAsk, clock);
|
||||
var ask = silentChild.Ref.Ask<DriverInstanceActor.WriteAttributeResult>(
|
||||
new DriverInstanceActor.WriteAttribute("x", 1), System.Threading.Timeout.InfiniteTimeSpan, deadline.Token);
|
||||
|
||||
clock.Advance(bounds.HostAsk - TimeSpan.FromMilliseconds(1));
|
||||
Thread.Sleep(200);
|
||||
ask.IsCompleted.ShouldBeFalse("the Ask ended before its clock deadline");
|
||||
|
||||
clock.Advance(TimeSpan.FromMilliseconds(1));
|
||||
AwaitCondition(() => ask.IsCompleted, Timeout, TimeSpan.FromMilliseconds(20));
|
||||
ask.IsCompletedSuccessfully.ShouldBeFalse();
|
||||
}
|
||||
|
||||
/// <summary>Routes the held write once the host accepts writes (a warm-up write proves the child is Connected
|
||||
/// and the reverse map is pushed), then waits until it is parked in the driver with every bound armed.</summary>
|
||||
private Akka.TestKit.TestProbe RouteHeldWrite(IActorRef host, GatedWritableDriver driver, ManualTimeProvider clock)
|
||||
{
|
||||
var lastReason = "(no reply)";
|
||||
AwaitAssert(
|
||||
() =>
|
||||
{
|
||||
var warm = CreateTestProbe();
|
||||
host.Tell(new DriverHostActor.RouteNodeWrite(RawNodeId, 1.0, AddressSpaceRealm.Raw), warm.Ref);
|
||||
var r = warm.ExpectMsg<DriverHostActor.NodeWriteResult>(TimeSpan.FromSeconds(1));
|
||||
lastReason = r.Reason ?? "(none)";
|
||||
r.Success.ShouldBeTrue($"host kept rejecting the warm-up write; last reason: {lastReason}");
|
||||
},
|
||||
duration: Timeout,
|
||||
interval: TimeSpan.FromMilliseconds(50));
|
||||
|
||||
var asker = CreateTestProbe();
|
||||
host.Tell(new DriverHostActor.RouteNodeWrite(RawNodeId, HeldValue, AddressSpaceRealm.Raw), asker.Ref);
|
||||
AwaitCondition(() => driver.HeldEntered, Timeout, TimeSpan.FromMilliseconds(20));
|
||||
// The host's Ask deadline + the child's token cancel + the child's abandon bound.
|
||||
AwaitCondition(() => clock.ArmedTimers == 3, Timeout, TimeSpan.FromMilliseconds(20));
|
||||
return asker;
|
||||
}
|
||||
|
||||
private IActorRef SpawnHostAndApply(GatedWritableDriver driver, DriverWriteBounds bounds, ManualTimeProvider clock)
|
||||
{
|
||||
var db = NewInMemoryDbFactory();
|
||||
var deploymentId = SeedDeployment(db);
|
||||
var coordinator = CreateTestProbe();
|
||||
var host = Sys.ActorOf(DriverHostActor.Props(
|
||||
db, TestNode, coordinator.Ref,
|
||||
driverFactory: new SingleDriverFactory(driver),
|
||||
localRoles: new HashSet<string> { "driver" },
|
||||
writeBounds: bounds,
|
||||
writeClock: clock));
|
||||
|
||||
host.Tell(new DispatchDeployment(deploymentId, RevA, CorrelationId.NewId()));
|
||||
coordinator.ExpectMsg<ApplyAck>(Timeout).Outcome.ShouldBe(ApplyAckOutcome.Applied);
|
||||
return host;
|
||||
}
|
||||
|
||||
/// <summary>One enabled Modbus driver "drv-1" with one ReadWrite tag, RawPath <c>Plant/Modbus/dev1/speed</c>.</summary>
|
||||
private static DeploymentId SeedDeployment(IDbContextFactory<OtOpcUaConfigDbContext> db)
|
||||
{
|
||||
var artifact = JsonSerializer.SerializeToUtf8Bytes(new
|
||||
{
|
||||
RawFolders = new[] { new { RawFolderId = "rf-plant", ParentRawFolderId = (string?)null, Name = "Plant", ClusterId = "c1" } },
|
||||
DriverInstances = new[]
|
||||
{
|
||||
new { DriverInstanceId = "drv-1", RawFolderId = "rf-plant", Name = "Modbus", DriverType = "Modbus", DriverConfig = "{}", ClusterId = "c1", Enabled = true },
|
||||
},
|
||||
Devices = new[] { new { DeviceId = "dev-1", DriverInstanceId = "drv-1", Name = "dev1", DeviceConfig = "{}" } },
|
||||
TagGroups = Array.Empty<object>(),
|
||||
Tags = new[]
|
||||
{
|
||||
new { TagId = "t-speed", DeviceId = "dev-1", TagGroupId = (string?)null, Name = "speed", DataType = "Double", AccessLevel = 1, TagConfig = "{}" },
|
||||
},
|
||||
UnsAreas = Array.Empty<object>(),
|
||||
UnsLines = Array.Empty<object>(),
|
||||
Equipment = Array.Empty<object>(),
|
||||
UnsTagReferences = Array.Empty<object>(),
|
||||
});
|
||||
|
||||
var id = DeploymentId.NewId();
|
||||
using var ctx = db.CreateDbContext();
|
||||
ctx.Deployments.Add(new Deployment
|
||||
{
|
||||
DeploymentId = id.Value,
|
||||
RevisionHash = RevA.Value,
|
||||
Status = DeploymentStatus.Sealed,
|
||||
CreatedBy = "test",
|
||||
SealedAtUtc = DateTime.UtcNow,
|
||||
ArtifactBlob = artifact,
|
||||
});
|
||||
ctx.SaveChanges();
|
||||
return id;
|
||||
}
|
||||
|
||||
private sealed class SingleDriverFactory(GatedWritableDriver driver) : IDriverFactory
|
||||
{
|
||||
public IDriver? TryCreate(string driverType, string driverInstanceId, string driverConfigJson) =>
|
||||
driverType == "Modbus" ? driver.Bind(driverInstanceId, driverType) : null;
|
||||
|
||||
public IReadOnlyCollection<string> SupportedTypes => new[] { "Modbus" };
|
||||
}
|
||||
|
||||
/// <summary>Writes <see cref="HeldValue"/> only once released (honouring its token unless
|
||||
/// <see cref="IgnoreCancellation"/>); every other value is written at once.</summary>
|
||||
private sealed class GatedWritableDriver : IDriver, IWritable
|
||||
{
|
||||
private readonly TaskCompletionSource _release = new(TaskCreationOptions.RunContinuationsAsynchronously);
|
||||
private int _heldEntered;
|
||||
|
||||
public bool IgnoreCancellation { get; init; }
|
||||
public bool HeldEntered => Volatile.Read(ref _heldEntered) == 1;
|
||||
public string DriverInstanceId { get; private set; } = string.Empty;
|
||||
public string DriverType { get; private set; } = string.Empty;
|
||||
|
||||
public GatedWritableDriver Bind(string id, string type)
|
||||
{
|
||||
DriverInstanceId = id;
|
||||
DriverType = type;
|
||||
return this;
|
||||
}
|
||||
|
||||
public void Release() => _release.TrySetResult();
|
||||
|
||||
public async Task<IReadOnlyList<WriteResult>> WriteAsync(IReadOnlyList<WriteRequest> writes, CancellationToken cancellationToken)
|
||||
{
|
||||
if (writes.Any(w => w.Value is HeldValue))
|
||||
{
|
||||
Volatile.Write(ref _heldEntered, 1);
|
||||
if (IgnoreCancellation) await _release.Task;
|
||||
else await _release.Task.WaitAsync(cancellationToken);
|
||||
}
|
||||
|
||||
return writes.Select(_ => new WriteResult(OpcUaStatus.Good)).ToList();
|
||||
}
|
||||
|
||||
public Task InitializeAsync(string driverConfigJson, CancellationToken cancellationToken) => Task.CompletedTask;
|
||||
public Task ReinitializeAsync(string driverConfigJson, CancellationToken cancellationToken) => Task.CompletedTask;
|
||||
public Task ShutdownAsync(CancellationToken cancellationToken)
|
||||
{
|
||||
_release.TrySetResult();
|
||||
return Task.CompletedTask;
|
||||
}
|
||||
|
||||
public DriverHealth GetHealth() => new(DriverState.Healthy, DateTime.UtcNow, null);
|
||||
}
|
||||
}
|
||||
+43
@@ -307,6 +307,49 @@ public sealed class DriverInstanceActorConcurrentWriteTests : RuntimeActorTestBa
|
||||
private void AwaitWriteBoundsArmed(ManualTimeProvider clock) =>
|
||||
AwaitCondition(() => clock.ArmedTimers == 2, PresenceBudget, TimeSpan.FromMilliseconds(20));
|
||||
|
||||
/// <summary>
|
||||
/// lmxopcua#777: the queue budget runs from when the HOST issued the write
|
||||
/// (<see cref="DriverInstanceActor.WriteAttribute.IssuedAt"/>), not from when this actor got round to it. A
|
||||
/// write that sat in the mailbox past the budget and was then started could still be running when the host's
|
||||
/// Ask fired — reported failed, node reverted, device still taking the value. It is failed BadTimeout and
|
||||
/// never written. (Mutation-checked: stamping the arrival time instead of IssuedAt writes it.)
|
||||
/// </summary>
|
||||
[Fact]
|
||||
public async Task A_write_issued_longer_ago_than_the_queue_budget_is_failed_not_started()
|
||||
{
|
||||
var clock = new ManualTimeProvider();
|
||||
var bounds = new DriverWriteBounds(TimeSpan.FromSeconds(10));
|
||||
var driver = new SlowWritableDriver();
|
||||
var (actor, _) = Connect(driver, writeTimeout: bounds.Timeout, writeClock: clock);
|
||||
|
||||
var issuedAt = clock.GetTimestamp();
|
||||
clock.Advance(bounds.QueueBudget + TimeSpan.FromMilliseconds(1));
|
||||
var result = await actor.Ask<DriverInstanceActor.WriteAttributeResult>(
|
||||
new DriverInstanceActor.WriteAttribute("tag-a", 1, issuedAt), PresenceBudget);
|
||||
|
||||
result.Success.ShouldBeFalse();
|
||||
result.Reason!.ShouldContain("800A0000", customMessage: "BadTimeout");
|
||||
driver.WritesEntered.ShouldBe(0, "a write past its budget must never reach the device");
|
||||
}
|
||||
|
||||
/// <summary>Positive control for the test above: a write issued just INSIDE the budget is written.</summary>
|
||||
[Fact]
|
||||
public async Task A_write_issued_within_the_queue_budget_is_written()
|
||||
{
|
||||
var clock = new ManualTimeProvider();
|
||||
var bounds = new DriverWriteBounds(TimeSpan.FromSeconds(10));
|
||||
var driver = new SlowWritableDriver();
|
||||
var (actor, _) = Connect(driver, writeTimeout: bounds.Timeout, writeClock: clock);
|
||||
|
||||
var issuedAt = clock.GetTimestamp();
|
||||
clock.Advance(bounds.QueueBudget);
|
||||
var result = await actor.Ask<DriverInstanceActor.WriteAttributeResult>(
|
||||
new DriverInstanceActor.WriteAttribute("tag-a", 1, issuedAt), PresenceBudget);
|
||||
|
||||
result.Success.ShouldBeTrue(result.Reason);
|
||||
driver.Writes.Select(w => w.FullReference).ShouldBe(new[] { "tag-a" });
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task Queued_writes_fail_fast_when_the_driver_leaves_Connected()
|
||||
{
|
||||
|
||||
@@ -0,0 +1,48 @@
|
||||
using Shouldly;
|
||||
using Xunit;
|
||||
using ZB.MOM.WW.OtOpcUa.Runtime.Drivers;
|
||||
|
||||
namespace ZB.MOM.WW.OtOpcUa.Runtime.Tests.Drivers;
|
||||
|
||||
/// <summary>
|
||||
/// lmxopcua#777: every write bound derives from the one write timeout, and each outer bound covers the worst case
|
||||
/// of everything inside it — whatever the timeout. (The fixed 8 s Ask failed this from a 7 s timeout up.)
|
||||
/// </summary>
|
||||
public sealed class DriverWriteBoundsTests
|
||||
{
|
||||
[Theory]
|
||||
[InlineData(0.5)]
|
||||
[InlineData(2)]
|
||||
[InlineData(5)]
|
||||
[InlineData(9)]
|
||||
[InlineData(10)]
|
||||
[InlineData(30)]
|
||||
[InlineData(300)]
|
||||
public void Each_outer_bound_covers_everything_inside_it(double timeoutSeconds)
|
||||
{
|
||||
var b = new DriverWriteBounds(TimeSpan.FromSeconds(timeoutSeconds));
|
||||
|
||||
b.AbandonGrace.ShouldBeGreaterThan(TimeSpan.Zero);
|
||||
b.RunBound.ShouldBe(b.Timeout + b.AbandonGrace);
|
||||
// The child answers a write by queue budget + run bound after it was issued; the Asks sit strictly outside.
|
||||
b.ReplyBound.ShouldBe(b.QueueBudget + b.RunBound);
|
||||
b.HostAsk.ShouldBeGreaterThanOrEqualTo(b.ReplyBound + DriverWriteBounds.ReplyMargin);
|
||||
b.GatewayAsk.ShouldBeGreaterThanOrEqualTo(b.HostAsk + DriverWriteBounds.ReplyMargin);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void The_default_is_10s_and_the_grace_is_half_the_timeout_capped_at_2s()
|
||||
{
|
||||
DriverWriteBounds.Default.Timeout.ShouldBe(TimeSpan.FromSeconds(10));
|
||||
new DriverWriteBounds(TimeSpan.FromSeconds(3)).AbandonGrace.ShouldBe(TimeSpan.FromSeconds(1.5));
|
||||
new DriverWriteBounds(TimeSpan.FromSeconds(9)).AbandonGrace.ShouldBe(TimeSpan.FromSeconds(2));
|
||||
DriverInstanceActor.DefaultWriteTimeout.ShouldBe(DriverWriteBounds.Default.Timeout);
|
||||
DriverInstanceActor.DefaultQueuedWriteBudget.ShouldBe(DriverWriteBounds.Default.QueueBudget);
|
||||
}
|
||||
|
||||
[Theory]
|
||||
[InlineData(0)]
|
||||
[InlineData(-1)]
|
||||
public void A_non_positive_timeout_is_rejected(double seconds) =>
|
||||
Should.Throw<ArgumentOutOfRangeException>(() => new DriverWriteBounds(TimeSpan.FromSeconds(seconds)));
|
||||
}
|
||||
Reference in New Issue
Block a user