From 97dc715dac2937d8122b05fd993aa7c775c2d278 Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Mon, 9 Nov 2026 15:46:04 -0500 Subject: [PATCH] fix(runtime): derive every write bound and Ask from one write timeout (lmxopcua#777) 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 --- .../DriverCapabilityInvokerFactory.cs | 19 +- .../Configuration/DriverWriteBoundsBinding.cs | 101 ++++++++ .../Drivers/DriverFactoryBootstrap.cs | 6 +- .../OpcUa/OtOpcUaServerHostedService.cs | 11 +- src/Server/ZB.MOM.WW.OtOpcUa.Host/Program.cs | 12 + .../ZB.MOM.WW.OtOpcUa.Host/appsettings.json | 4 + .../Drivers/ActorNodeWriteGateway.cs | 9 +- .../Drivers/DriverHostActor.cs | 62 ++++- .../Drivers/DriverInstanceActor.cs | 103 +++++--- .../Drivers/DriverWriteBounds.cs | 105 ++++++++ .../ServiceCollectionExtensions.cs | 6 +- .../DriverCapabilityInvokerFactoryTests.cs | 29 +++ .../DriverWriteBoundsBindingTests.cs | 84 ++++++ .../Drivers/DriverHostActorWriteAskTests.cs | 239 ++++++++++++++++++ ...DriverInstanceActorConcurrentWriteTests.cs | 43 ++++ .../Drivers/DriverWriteBoundsTests.cs | 48 ++++ 16 files changed, 824 insertions(+), 57 deletions(-) create mode 100644 src/Server/ZB.MOM.WW.OtOpcUa.Host/Configuration/DriverWriteBoundsBinding.cs create mode 100644 src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/DriverWriteBounds.cs create mode 100644 tests/Server/ZB.MOM.WW.OtOpcUa.Host.Tests/Configuration/DriverWriteBoundsBindingTests.cs create mode 100644 tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverHostActorWriteAskTests.cs create mode 100644 tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverWriteBoundsTests.cs diff --git a/src/Core/ZB.MOM.WW.OtOpcUa.Core/Resilience/DriverCapabilityInvokerFactory.cs b/src/Core/ZB.MOM.WW.OtOpcUa.Core/Resilience/DriverCapabilityInvokerFactory.cs index 2c2dfb512..e15cca157 100644 --- a/src/Core/ZB.MOM.WW.OtOpcUa.Core/Resilience/DriverCapabilityInvokerFactory.cs +++ b/src/Core/ZB.MOM.WW.OtOpcUa.Core/Resilience/DriverCapabilityInvokerFactory.cs @@ -33,6 +33,7 @@ public sealed class DriverCapabilityInvokerFactory : IDriverCapabilityInvokerFac private readonly DriverResilienceStatusTracker _statusTracker; private readonly Func _tierResolver; private readonly ILogger? _logger; + private readonly TimeSpan? _writeTimeoutCeiling; private long _generation; /// Construct the factory over the shared resilience infrastructure. @@ -41,11 +42,16 @@ public sealed class DriverCapabilityInvokerFactory : IDriverCapabilityInvokerFac /// Resolves a driver type name to its (e.g. DriverFactoryRegistry.GetTier). /// Optional logger; when non-null, a ResilienceConfig parse diagnostic (malformed JSON, /// unknown capability, misapplied Tier-C recycle) is logged as a warning at time. + /// The node's operator-write timeout (DriverWrites:Timeout), 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 time: that policy timeout + /// never takes effect. Null skips the check. public DriverCapabilityInvokerFactory( DriverResiliencePipelineBuilder builder, DriverResilienceStatusTracker statusTracker, Func 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; } /// @@ -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 diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.Host/Configuration/DriverWriteBoundsBinding.cs b/src/Server/ZB.MOM.WW.OtOpcUa.Host/Configuration/DriverWriteBoundsBinding.cs new file mode 100644 index 000000000..07de69125 --- /dev/null +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Host/Configuration/DriverWriteBoundsBinding.cs @@ -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; + +/// +/// Binds DriverWrites:Timeout into and checks it against the resilience +/// tiers' Write timeouts (lmxopcua#777). One place, so Program.cs and its unit test bind the same way. +/// +/// +/// Matches : 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. +/// +public static class DriverWriteBoundsBinding +{ + /// The configuration key of the write timeout. + public const string TimeoutKey = DriverWriteBounds.SectionName + ":Timeout"; + + /// Binds, clamps and validates the write timeout. + /// The host configuration. + /// One line per clamped value or tier inconsistency, for the caller to log as a Warning. + /// The write bounds; when the key is absent. + public static DriverWriteBounds Bind(IConfiguration configuration, out IReadOnlyList warnings) + { + ArgumentNullException.ThrowIfNull(configuration); + var found = new List(); + 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; + } + + /// + /// Reports every resilience tier whose DEFAULT Write timeout exceeds ' 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 ResilienceConfig overrides arrive + /// with a deployment, not at startup; checks those at spawn.) + /// + /// The write bounds in force. + /// One warning per inconsistent tier; empty when every tier fits. + public static IReadOnlyList CheckTierDefaults(DriverWriteBounds bounds) + { + ArgumentNullException.ThrowIfNull(bounds); + var found = new List(); + foreach (var tier in Enum.GetValues()) + { + 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; + } + + /// One line naming every derived bound, for the startup log. + /// The write bounds in force. + /// The description. + 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"; +} diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.Host/Drivers/DriverFactoryBootstrap.cs b/src/Server/ZB.MOM.WW.OtOpcUa.Host/Drivers/DriverFactoryBootstrap.cs index 7cb8d869f..5ad5a462b 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.Host/Drivers/DriverFactoryBootstrap.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Host/Drivers/DriverFactoryBootstrap.cs @@ -109,7 +109,11 @@ public static class DriverFactoryBootstrap sp.GetRequiredService(), sp.GetRequiredService(), registry.GetTier, - logger: sp.GetService()?.CreateLogger("ZB.MOM.WW.OtOpcUa.Core.Resilience.DriverCapabilityInvokerFactory")); + logger: sp.GetService()?.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.Default).Timeout); }); // The tracker's operator-facing reader: a periodic DPS publisher that mirrors each snapshot to the diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.Host/OpcUa/OtOpcUaServerHostedService.cs b/src/Server/ZB.MOM.WW.OtOpcUa.Host/OpcUa/OtOpcUaServerHostedService.cs index 2e614594d..5fefe1fa3 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.Host/OpcUa/OtOpcUaServerHostedService.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Host/OpcUa/OtOpcUaServerHostedService.cs @@ -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. /// The native-ack pre-check snapshot DriverHostActor publishes /// (lmxopcua#760). Optional: registered on every driver-role node; null wires no pre-check. + /// The node's configured write bounds (DriverWrites:Timeout); the inbound + /// write gateway's Ask is derived from them (lmxopcua#777). Optional: null ⇒ + /// . public OtOpcUaServerHostedService( IOptions 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(); @@ -323,7 +329,8 @@ public sealed class OtOpcUaServerHostedService : IHostedService, IAsyncDisposabl // resolves the write to "writes unavailable". _server.SetNodeWriteGateway(new ActorNodeWriteGateway( resolveDriverHost: () => _actorRegistry.TryGet(out var driverHost) ? driverHost : null, - logger: _loggerFactory.CreateLogger())); + logger: _loggerFactory.CreateLogger(), + 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). diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.Host/Program.cs b/src/Server/ZB.MOM.WW.OtOpcUa.Host/Program.cs index c3c105abd..1238af06a 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.Host/Program.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Host/Program.cs @@ -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(sp => new RoslynVirtualTagEvaluator( sp.GetRequiredService().CreateLogger(), diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.Host/appsettings.json b/src/Server/ZB.MOM.WW.OtOpcUa.Host/appsettings.json index 095531f53..fddf96ebf 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.Host/appsettings.json +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Host/appsettings.json @@ -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", diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/ActorNodeWriteGateway.cs b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/ActorNodeWriteGateway.cs index 7c6cb37ea..a24629a9e 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/ActorNodeWriteGateway.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/ActorNodeWriteGateway.cs @@ -22,8 +22,6 @@ namespace ZB.MOM.WW.OtOpcUa.Runtime.Drivers; /// public sealed class ActorNodeWriteGateway : IOpcUaNodeWriteGateway { - /// Default Ask timeout — matches the legacy inline lambda in the hosted service. - private static readonly TimeSpan DefaultAskTimeout = TimeSpan.FromSeconds(10); private readonly Func _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). /// Logger for dropped/rejected/timed-out writes. - /// Ask timeout; defaults to 10s (the legacy lambda's value). + /// Ask timeout; defaults to of + /// . 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. public ActorNodeWriteGateway(Func 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; } /// diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/DriverHostActor.cs b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/DriverHostActor.cs index f8d152bf3..c99471208 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/DriverHostActor.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/DriverHostActor.cs @@ -479,6 +479,14 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS /// Where the synchronous native-ack pre-check snapshot is published (lmxopcua#760); null ⇒ none. private readonly NativeAlarmAckPreflightView? _ackPreflightView; + /// The write bounds every child is spawned with and this host's write Ask is derived from + /// (lmxopcua#777). + private readonly DriverWriteBounds _writeBounds; + + /// The clock the write bounds run on, shared with every child so a write's issue stamp + /// () means the same thing on both sides. + private readonly TimeProvider _writeClock; + /// Whether any redundancy snapshot has reached this host — the HistoryUpdate gate stays /// closed until one has (see ). 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. /// The node-wide Scripting run-guard options handed to the scripted-alarm /// engine. Null ⇒ the compiled defaults. + /// Where the native-ack pre-check snapshot is published. Null ⇒ none. + /// The write timeout every child is spawned with, and from which this host's write + /// Ask is derived (). Null ⇒ . + /// The clock the write bounds (and the host's write Ask) run on, shared with every + /// child. Null ⇒ . /// The Akka.NET used to spawn this actor. public static Props Props( IDbContextFactory? 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)); /// Initializes a new DriverHostActor with the specified dependencies. /// Database context factory for configuration database access. @@ -861,6 +876,10 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS /// engine. Null ⇒ the compiled defaults. /// 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). + /// The write timeout every child is spawned with, and from which this host's write + /// Ask is derived (). Null ⇒ . + /// The clock the write bounds (and the host's write Ask) run on, shared with every + /// child. Null ⇒ . public DriverHostActor( IDbContextFactory? 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( - 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)); } diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/DriverInstanceActor.cs b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/DriverInstanceActor.cs index d06efe9a6..606ac015e 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/DriverInstanceActor.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/DriverInstanceActor.cs @@ -61,7 +61,15 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers /// The driver instance that was restarted. /// The type name of the exception that tripped supervision, for the host's log. public sealed record ReprovisionRequested(string DriverInstanceId, string Reason); - public sealed record WriteAttribute(string TagId, object Value); + /// An operator write of to the tag . + /// The tag's driver wire reference (RawPath). + /// The value to write. + /// When the asker issued the write, as a timestamp of the actor's write clock + /// (); null = when the actor receives it. + /// 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). + public sealed record WriteAttribute(string TagId, object Value, long? IssuedAt = null); public sealed record WriteAttributeResult(bool Success, string? Reason); /// /// Sent by when an OPC UA client Acknowledges a NATIVE Part 9 @@ -216,10 +224,13 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers /// /// How long a queued write may wait for a slot before it is failed BadTimeout 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: 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 (), so a write started inside the budget is always answered + /// before the host reports it failed. /// - public static readonly TimeSpan DefaultQueuedWriteBudget = TimeSpan.FromSeconds(8); + public static readonly TimeSpan DefaultQueuedWriteBudget = DriverWriteBounds.Default.QueueBudget; /// /// Pessimistic bound on one attempt. The invoker's tier timeout is @@ -236,23 +247,19 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers /// queue. /// /// - /// 5 s, so both bounds (5 s cancel, 7 s abandon) answer inside the host's 8 s write Ask - /// (DriverHostActor): 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. + /// . The host's write Ask is DERIVED from the same value + /// (, 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. /// - public static readonly TimeSpan DefaultWriteTimeout = TimeSpan.FromSeconds(5); + public static readonly TimeSpan DefaultWriteTimeout = DriverWriteBounds.DefaultTimeout; /// - /// How long after a write's token is cancelled () 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 . /// /// The write timeout in force. /// The grace between cancelling the write's token and abandoning the write. - 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); /// Self-sent by a write task once it has replied, releasing its slot and its tag. /// The tag the write held. @@ -292,8 +299,8 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers /// Writes waiting for a slot (or for their tag), in arrival order, stamped with arrival time. /// Failed fast on every exit from Connected, so a queued write never reaches a driver that is - /// reconnecting; failed BadTimeout at dequeue once older than . - private readonly LinkedList<(WriteAttribute Write, IActorRef ReplyTo, DateTime EnqueuedUtc)> _pendingWrites = new(); + /// reconnecting; failed BadTimeout once older than since issued. + private readonly LinkedList<(WriteAttribute Write, IActorRef ReplyTo, long IssuedAt)> _pendingWrites = new(); /// 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. @@ -453,7 +460,7 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers /// Optional cap on concurrent writes against the driver; defaults to /// . Writes beyond it queue in arrival order. /// Optional bound on how long a queued write may wait before it is failed - /// BadTimeout; defaults to . + /// BadTimeout; defaults to of the write timeout in force. /// Optional pessimistic bound on one initialise attempt; defaults to /// . /// Optional bound on one write round-trip (its token is cancelled when it elapses; @@ -533,7 +540,7 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers /// Cap on concurrent writes; defaults to /// when null or non-positive. /// Bound on a queued write's wait; defaults to - /// when null or non-positive. + /// of the write timeout when null or non-positive. /// Pessimistic bound on one initialise attempt; defaults to /// when null or non-positive. /// 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(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(_ => 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. /// A persistently failing subscribe backs off: after the n-th consecutive failure the reconcile /// skips 2^(n-1) − 1 ticks, capped at , 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. /// 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 } } + /// + /// Fails a write that has waited longer than 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 (), 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. + /// + /// True when the write was expired (and answered). + 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; + } + /// Replies to every queued write with ; called on each exit from /// Connected. In-flight writes are left to finish — they already hold the device and reply with its /// real outcome. diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/DriverWriteBounds.cs b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/DriverWriteBounds.cs new file mode 100644 index 000000000..e2c585417 --- /dev/null +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/DriverWriteBounds.cs @@ -0,0 +1,105 @@ +namespace ZB.MOM.WW.OtOpcUa.Runtime.Drivers; + +/// +/// Every bound on an operator write's path, derived from ONE configured value: the write timeout +/// (DriverWrites:Timeout). lmxopcua#777. +/// +/// +/// +/// A write passes three bounds on its way to the device, and they used to be configured independently: +/// the gateway's Ask of (10 s), the host's Ask of the +/// 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. +/// +/// +/// Here every bound is derived from , innermost first, so each outer bound covers +/// the worst case of everything inside it: +/// +/// — the write's cancellation token fires. +/// = timeout + — a write that ignored its +/// token is abandoned (answers BadTimeout). +/// = — 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. +/// = queue + run — the child answers every write within this of its +/// arrival. +/// = reply + , and +/// = host Ask + margin. +/// +/// The Asks are safety nets for an actor that never replies, not a second write timeout. +/// +/// +/// The Phase 6.1 resilience tier's per-capability Write timeout (DriverResilienceOptions) runs +/// INSIDE : the actor's token is the outer cap, so a tier Write timeout above +/// never takes effect. The Host warns about that at startup (tier defaults) and at +/// each driver spawn (per-instance ResilienceConfig overrides). +/// +/// +public sealed record DriverWriteBounds +{ + /// The configuration section the write timeout is bound from. + public const string SectionName = "DriverWrites"; + + /// + /// 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. + /// + public static readonly TimeSpan DefaultTimeout = TimeSpan.FromSeconds(10); + + /// Largest accepted write timeout; a larger configured value is clamped (with a startup warning). + public static readonly TimeSpan MaxTimeout = TimeSpan.FromMinutes(5); + + /// + /// Headroom each Ask keeps over the bound it wraps, for mailbox and dispatch latency between the reply being + /// sent and the Ask observing it. + /// + public static readonly TimeSpan ReplyMargin = TimeSpan.FromSeconds(1); + + /// The write bounds derived from . + public static DriverWriteBounds Default { get; } = new(DefaultTimeout); + + /// Creates the bounds for a write timeout. + /// The write timeout; must be positive. + /// The timeout is zero or negative. + public DriverWriteBounds(TimeSpan timeout) + { + ArgumentOutOfRangeException.ThrowIfLessThanOrEqual(timeout, TimeSpan.Zero); + Timeout = timeout; + } + + /// When a write's cancellation token is cancelled (the write answers "write timeout"). + public TimeSpan Timeout { get; } + + /// + /// How long after 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. + /// + public TimeSpan AbandonGrace => AbandonGraceFor(Timeout); + + /// The latest a running write is answered after it starts: + . + public TimeSpan RunBound => Timeout + AbandonGrace; + + /// + /// The longest a write may wait for a slot before it is failed BadTimeout instead of being written: + /// one , i.e. long enough for an earlier write to the same tag to finish. + /// + public TimeSpan QueueBudget => RunBound; + + /// The latest the driver child answers a write after it arrives: + . + public TimeSpan ReplyBound => QueueBudget + RunBound; + + /// 's Ask of the child: + . + public TimeSpan HostAsk => ReplyBound + ReplyMargin; + + /// 's Ask of the host: + . + public TimeSpan GatewayAsk => HostAsk + ReplyMargin; + + /// See . + /// The write timeout in force. + /// The grace between cancelling the write's token and abandoning the write. + public static TimeSpan AbandonGraceFor(TimeSpan timeout) => + TimeSpan.FromTicks(Math.Min(timeout.Ticks / 2, TimeSpan.FromSeconds(2).Ticks)); +} diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/ServiceCollectionExtensions.cs b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/ServiceCollectionExtensions.cs index 498a0eb93..2f3239b89 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/ServiceCollectionExtensions.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/ServiceCollectionExtensions.cs @@ -660,7 +660,11 @@ public static class ServiceCollectionExtensions // engine. Unbound in test graphs ⇒ null ⇒ the compiled defaults. scripting: resolver.GetService(), // Where the native-ack pre-check snapshot is published (lmxopcua#760). - ackPreflightView: resolver.GetService()), + ackPreflightView: resolver.GetService(), + // 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()), DriverHostActorName); registry.Register(driverHost); diff --git a/tests/Core/ZB.MOM.WW.OtOpcUa.Core.Tests/Resilience/DriverCapabilityInvokerFactoryTests.cs b/tests/Core/ZB.MOM.WW.OtOpcUa.Core.Tests/Resilience/DriverCapabilityInvokerFactoryTests.cs index 13b7019b9..3d2940f95 100644 --- a/tests/Core/ZB.MOM.WW.OtOpcUa.Core.Tests/Resilience/DriverCapabilityInvokerFactoryTests.cs +++ b/tests/Core/ZB.MOM.WW.OtOpcUa.Core.Tests/Resilience/DriverCapabilityInvokerFactoryTests.cs @@ -200,6 +200,35 @@ public sealed class DriverCapabilityInvokerFactoryTests logger.Entries.ShouldContain(e => e.Level == LogLevel.Warning && e.Message.Contains("resilience config", StringComparison.OrdinalIgnoreCase)); } + /// + /// 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. + /// + [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(); + } + /// /// 01/S-6 production-wiring proof: the exact DriverHostActor.SpawnChild path /// (factory.Create(id, type, hostileJson)) must yield a WORKING invoker whose capability diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.Host.Tests/Configuration/DriverWriteBoundsBindingTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.Host.Tests/Configuration/DriverWriteBoundsBindingTests.cs new file mode 100644 index 000000000..618b14b71 --- /dev/null +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.Host.Tests/Configuration/DriverWriteBoundsBindingTests.cs @@ -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; + +/// +/// lmxopcua#777: DriverWrites:Timeout is the one value every write bound derives from, and it is checked +/// against the resilience tiers' Write timeouts at startup. +/// +public sealed class DriverWriteBoundsBindingTests +{ + private static IConfiguration Config(Dictionary values) => + new ConfigurationBuilder().AddInMemoryCollection(values).Build(); + + /// The shipped default is consistent with every tier — no tier's Write policy is capped below itself. + [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()) + { + 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); + } + + /// + /// 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. + /// + [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"); + } +} diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverHostActorWriteAskTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverHostActorWriteAskTests.cs new file mode 100644 index 000000000..2ace8c577 --- /dev/null +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverHostActorWriteAskTests.cs @@ -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; + +/// +/// lmxopcua#777: 's write Ask is derived from the configured write timeout +/// () 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. +/// +/// Drives a real apply (a real 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. +/// +/// +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"; + + /// The value the gated driver holds; every other value is written at once. + private const double HeldValue = 99.0; + + /// + /// 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.) + /// + [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(Timeout); + result.Success.ShouldBeTrue(result.Reason); + } + + /// + /// 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. + /// + [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(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); + } + + /// + /// 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 , and it + /// does complete (cancelled, which the host reports as "write timeout") at it, for a child that never answers. + /// + [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( + 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(); + } + + /// 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. + 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(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 { "driver" }, + writeBounds: bounds, + writeClock: clock)); + + host.Tell(new DispatchDeployment(deploymentId, RevA, CorrelationId.NewId())); + coordinator.ExpectMsg(Timeout).Outcome.ShouldBe(ApplyAckOutcome.Applied); + return host; + } + + /// One enabled Modbus driver "drv-1" with one ReadWrite tag, RawPath Plant/Modbus/dev1/speed. + private static DeploymentId SeedDeployment(IDbContextFactory 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(), + Tags = new[] + { + new { TagId = "t-speed", DeviceId = "dev-1", TagGroupId = (string?)null, Name = "speed", DataType = "Double", AccessLevel = 1, TagConfig = "{}" }, + }, + UnsAreas = Array.Empty(), + UnsLines = Array.Empty(), + Equipment = Array.Empty(), + UnsTagReferences = Array.Empty(), + }); + + 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 SupportedTypes => new[] { "Modbus" }; + } + + /// Writes only once released (honouring its token unless + /// ); every other value is written at once. + 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> WriteAsync(IReadOnlyList 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); + } +} diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverInstanceActorConcurrentWriteTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverInstanceActorConcurrentWriteTests.cs index 87f082a47..ee1d99706 100644 --- a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverInstanceActorConcurrentWriteTests.cs +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverInstanceActorConcurrentWriteTests.cs @@ -307,6 +307,49 @@ public sealed class DriverInstanceActorConcurrentWriteTests : RuntimeActorTestBa private void AwaitWriteBoundsArmed(ManualTimeProvider clock) => AwaitCondition(() => clock.ArmedTimers == 2, PresenceBudget, TimeSpan.FromMilliseconds(20)); + /// + /// lmxopcua#777: the queue budget runs from when the HOST issued the write + /// (), 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.) + /// + [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( + 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"); + } + + /// Positive control for the test above: a write issued just INSIDE the budget is written. + [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( + 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() { diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverWriteBoundsTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverWriteBoundsTests.cs new file mode 100644 index 000000000..76ae0f5e1 --- /dev/null +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverWriteBoundsTests.cs @@ -0,0 +1,48 @@ +using Shouldly; +using Xunit; +using ZB.MOM.WW.OtOpcUa.Runtime.Drivers; + +namespace ZB.MOM.WW.OtOpcUa.Runtime.Tests.Drivers; + +/// +/// 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.) +/// +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(() => new DriverWriteBounds(TimeSpan.FromSeconds(seconds))); +}