diff --git a/src/Core/ZB.MOM.WW.OtOpcUa.Core.Abstractions/IDriverCapabilityInvoker.cs b/src/Core/ZB.MOM.WW.OtOpcUa.Core.Abstractions/IDriverCapabilityInvoker.cs new file mode 100644 index 00000000..348113a5 --- /dev/null +++ b/src/Core/ZB.MOM.WW.OtOpcUa.Core.Abstractions/IDriverCapabilityInvoker.cs @@ -0,0 +1,99 @@ +namespace ZB.MOM.WW.OtOpcUa.Core.Abstractions; + +/// +/// Abstraction over the Phase 6.1 resilience pipeline that wraps a single driver instance's +/// capability calls (retry / circuit-breaker / bulkhead / tracker telemetry). The Runtime +/// dispatch layer (DriverInstanceActor) consumes this interface instead of the concrete +/// CapabilityInvoker so the actor assembly does NOT pull in +/// ZB.MOM.WW.OtOpcUa.Core (which drags in Polly + driver hosting) — the same boundary +/// rationale as . The fused Host binds a real invoker (built by +/// ) per spawned driver; when none is supplied the +/// dispatch layer uses , a genuine pass-through. +/// +/// +/// The method shapes mirror CapabilityInvoker exactly so the concrete invoker satisfies +/// this interface without adapters. Callers pass the resolved hostName per call (a +/// multi-host driver resolves it from the tag reference via ; +/// a single-host driver passes its ) so per-host breaker +/// isolation + bulkhead-depth accounting key on the right target. +/// +public interface IDriverCapabilityInvoker +{ + /// Execute a value-returning capability call through the per-capability pipeline. + /// Return type of the underlying driver call. + /// The driver capability being executed. + /// The resolved host name for pipeline keying + status tracking. + /// The async function to execute (the raw driver-capability call). + /// Cancellation token for the operation. + /// The result of the underlying driver call. + ValueTask ExecuteAsync( + DriverCapability capability, + string hostName, + Func> callSite, + CancellationToken cancellationToken); + + /// Execute a void-returning capability call through the per-capability pipeline. + /// The driver capability being executed. + /// The resolved host name for pipeline keying + status tracking. + /// The async function to execute (the raw driver-capability call). + /// Cancellation token for the operation. + /// A task that represents the asynchronous operation. + ValueTask ExecuteAsync( + DriverCapability capability, + string hostName, + Func callSite, + CancellationToken cancellationToken); + + /// + /// Execute a call honoring idempotence semantics — when + /// is false, retries are disabled regardless of the + /// configured policy; when true, the call runs through the Write pipeline which may + /// retry if the tier configuration permits. + /// + /// Return type of the underlying driver write call. + /// The resolved host name for pipeline keying + status tracking. + /// Whether the write operation is safe to retry. + /// The async function to execute (the raw driver write call). + /// Cancellation token for the operation. + /// The result of the underlying driver write call. + ValueTask ExecuteWriteAsync( + string hostName, + bool isIdempotent, + Func> callSite, + CancellationToken cancellationToken); +} + +/// +/// Pass-through that invokes the call site directly with +/// NO pipeline — no timeout, retry, breaker, or telemetry. Bound when the dispatch layer is +/// constructed without a real invoker (unit tests, the F7 skeleton path, admin-only nodes that +/// never run drivers). Keeps behavior byte-for-byte identical to a raw driver call so existing +/// dispatch-layer tests are unaffected by the resilience wiring. +/// +public sealed class NullDriverCapabilityInvoker : IDriverCapabilityInvoker +{ + /// The shared singleton instance. + public static readonly NullDriverCapabilityInvoker Instance = new(); + private NullDriverCapabilityInvoker() { } + + /// + public ValueTask ExecuteAsync( + DriverCapability capability, + string hostName, + Func> callSite, + CancellationToken cancellationToken) => callSite(cancellationToken); + + /// + public ValueTask ExecuteAsync( + DriverCapability capability, + string hostName, + Func callSite, + CancellationToken cancellationToken) => callSite(cancellationToken); + + /// + public ValueTask ExecuteWriteAsync( + string hostName, + bool isIdempotent, + Func> callSite, + CancellationToken cancellationToken) => callSite(cancellationToken); +} diff --git a/src/Core/ZB.MOM.WW.OtOpcUa.Core.Abstractions/IDriverCapabilityInvokerFactory.cs b/src/Core/ZB.MOM.WW.OtOpcUa.Core.Abstractions/IDriverCapabilityInvokerFactory.cs new file mode 100644 index 00000000..2160f74b --- /dev/null +++ b/src/Core/ZB.MOM.WW.OtOpcUa.Core.Abstractions/IDriverCapabilityInvokerFactory.cs @@ -0,0 +1,39 @@ +namespace ZB.MOM.WW.OtOpcUa.Core.Abstractions; + +/// +/// Builds a per-driver-instance . The Runtime +/// DriverHostActor consumes this factory (resolved from DI, like ) +/// to attach a resilience invoker to each spawned DriverInstanceActor WITHOUT the actor +/// assembly referencing ZB.MOM.WW.OtOpcUa.Core (which drags in Polly). The fused Host +/// binds a concrete factory closing over the process-singleton pipeline builder + status tracker + +/// driver-tier resolver; when unbound, yields the +/// pass-through invoker so dispatch behaves exactly as before. +/// +public interface IDriverCapabilityInvokerFactory +{ + /// + /// Create the resilience invoker for one driver instance. The returned invoker is keyed on + /// and applies the tier policy resolved from + /// . + /// + /// The driver instance identifier (pipeline + tracker key). + /// The driver type name, used to resolve the resilience tier defaults. + /// A per-instance . + IDriverCapabilityInvoker Create(string driverInstanceId, string driverType); +} + +/// +/// Returns for every driver instance. Bound +/// when the fused Host hasn't registered a real resilience factory (Mac dev path, smoke tests, +/// admin-only nodes). Dispatch then routes calls straight through with no pipeline. +/// +public sealed class NullDriverCapabilityInvokerFactory : IDriverCapabilityInvokerFactory +{ + /// The shared singleton instance. + public static readonly NullDriverCapabilityInvokerFactory Instance = new(); + private NullDriverCapabilityInvokerFactory() { } + + /// + public IDriverCapabilityInvoker Create(string driverInstanceId, string driverType) => + NullDriverCapabilityInvoker.Instance; +} diff --git a/src/Core/ZB.MOM.WW.OtOpcUa.Core/OpcUa/GenericDriverNodeManager.cs b/src/Core/ZB.MOM.WW.OtOpcUa.Core/OpcUa/GenericDriverNodeManager.cs index c6118261..e7a9f19d 100644 --- a/src/Core/ZB.MOM.WW.OtOpcUa.Core/OpcUa/GenericDriverNodeManager.cs +++ b/src/Core/ZB.MOM.WW.OtOpcUa.Core/OpcUa/GenericDriverNodeManager.cs @@ -68,7 +68,7 @@ public class GenericDriverNodeManager(IDriver driver) : IDisposable throw new NotSupportedException($"Driver '{Driver.DriverInstanceId}' does not implement ITagDiscovery."); var capturing = new CapturingBuilder(builder, _alarmSinks); -#pragma warning disable OTOPCUA0001 // RESILIENCE-DISPATCH-GAP (07/C-1): dispatch bypasses CapabilityInvoker — Phase 6.1 pipeline built but never wired into GenericDriverNodeManager; tracked, NOT intentional. +#pragma warning disable OTOPCUA0001 // NOT a production dispatch path: this Core scaffolding node-manager is exercised only by GenericDriverNodeManagerTests — the live server materialises the address space via the Server project's own OpcUaAddressSpaceBuilder path. No CapabilityInvoker by design (verified 07/#10: zero production references). If this class is ever wired into live dispatch, route this call through IDriverCapabilityInvoker. await discovery.DiscoverAsync(capturing, ct); #pragma warning restore OTOPCUA0001 diff --git a/src/Core/ZB.MOM.WW.OtOpcUa.Core/Resilience/CapabilityInvoker.cs b/src/Core/ZB.MOM.WW.OtOpcUa.Core/Resilience/CapabilityInvoker.cs index e4e42922..6c55d4b1 100644 --- a/src/Core/ZB.MOM.WW.OtOpcUa.Core/Resilience/CapabilityInvoker.cs +++ b/src/Core/ZB.MOM.WW.OtOpcUa.Core/Resilience/CapabilityInvoker.cs @@ -7,16 +7,21 @@ namespace ZB.MOM.WW.OtOpcUa.Core.Resilience; /// /// Executes driver-capability calls through a shared Polly pipeline. One invoker per /// (DriverInstance, IDriver) pair; the underlying -/// is process-singleton so all invokers share its cache. +/// is process-singleton so all invokers share its cache. Implements +/// so the Polly-free Runtime dispatch layer can consume it +/// through the abstraction (bound per-driver by ). /// /// -/// The server's dispatch -/// layer routes every capability call (IReadable.ReadAsync, IWritable.WriteAsync, -/// ITagDiscovery.DiscoverAsync, ISubscribable.SubscribeAsync/UnsubscribeAsync, -/// IHostConnectivityProbe probe loop, IAlarmSource.SubscribeAlarmsAsync/AcknowledgeAsync, -/// and all four IHistoryProvider reads) through this invoker. +/// The Runtime dispatch layer (DriverInstanceActor) routes its capability calls +/// (IWritable.WriteAsync, ITagDiscovery.DiscoverAsync, +/// ISubscribable.SubscribeAsync/UnsubscribeAsync, +/// IAlarmSource.SubscribeAlarmsAsync/AcknowledgeAsync) through this invoker via the +/// seam. The UnwrappedCapabilityCallAnalyzer +/// (OTOPCUA0001) enforces that every guarded call is wrapped in an +/// / lambda (the interface's methods are registered wrapper +/// homes too, so an interface-typed invoker call suppresses the diagnostic). /// -public sealed class CapabilityInvoker +public sealed class CapabilityInvoker : IDriverCapabilityInvoker { private readonly DriverResiliencePipelineBuilder _builder; private readonly string _driverInstanceId; diff --git a/src/Core/ZB.MOM.WW.OtOpcUa.Core/Resilience/DriverCapabilityInvokerFactory.cs b/src/Core/ZB.MOM.WW.OtOpcUa.Core/Resilience/DriverCapabilityInvokerFactory.cs new file mode 100644 index 00000000..6c7f0039 --- /dev/null +++ b/src/Core/ZB.MOM.WW.OtOpcUa.Core/Resilience/DriverCapabilityInvokerFactory.cs @@ -0,0 +1,58 @@ +using ZB.MOM.WW.OtOpcUa.Core.Abstractions; + +namespace ZB.MOM.WW.OtOpcUa.Core.Resilience; + +/// +/// Concrete that builds a per-driver-instance +/// over the process-singleton pipeline builder + status tracker. +/// The fused Host registers this (it references Core + Polly); the Runtime dispatch layer +/// resolves the abstraction from DI and never sees this type. +/// +/// +/// Resilience policy is currently the tier default for the driver type — resolved +/// via the injected tier resolver (backed by DriverFactoryRegistry.GetTier). The +/// per-instance DriverInstance.ResilienceConfig JSON override is NOT yet applied because +/// the deployment artifact (DriverInstanceSpec) does not carry it; plumbing that column +/// through the composer/artifact is a tracked follow-up. Until then every instance of a given +/// driver type gets that type's tier defaults. +/// +/// Options are snapshotted once per (tier is fixed for a driver type), +/// so the invoker's per-call options accessor is allocation-free on the hot path. +/// +public sealed class DriverCapabilityInvokerFactory : IDriverCapabilityInvokerFactory +{ + private readonly DriverResiliencePipelineBuilder _builder; + private readonly DriverResilienceStatusTracker _statusTracker; + private readonly Func _tierResolver; + + /// Construct the factory over the shared resilience infrastructure. + /// Process-singleton Polly pipeline builder (shared pipeline cache). + /// Process-singleton resilience status tracker feeding Admin /hosts. + /// Resolves a driver type name to its (e.g. DriverFactoryRegistry.GetTier). + public DriverCapabilityInvokerFactory( + DriverResiliencePipelineBuilder builder, + DriverResilienceStatusTracker statusTracker, + Func tierResolver) + { + ArgumentNullException.ThrowIfNull(builder); + ArgumentNullException.ThrowIfNull(statusTracker); + ArgumentNullException.ThrowIfNull(tierResolver); + _builder = builder; + _statusTracker = statusTracker; + _tierResolver = tierResolver; + } + + /// + public IDriverCapabilityInvoker Create(string driverInstanceId, string driverType) + { + var tier = _tierResolver(driverType); + // No per-instance ResilienceConfig in the artifact yet — tier defaults only (see remarks). + var options = DriverResilienceOptionsParser.ParseOrDefaults(tier, resilienceConfigJson: null, out _); + return new CapabilityInvoker( + _builder, + driverInstanceId, + optionsAccessor: () => options, + driverType: driverType, + statusTracker: _statusTracker); + } +} diff --git a/src/Core/ZB.MOM.WW.OtOpcUa.Core/Resilience/DriverResiliencePipelineBuilder.cs b/src/Core/ZB.MOM.WW.OtOpcUa.Core/Resilience/DriverResiliencePipelineBuilder.cs index 03f36ef3..b1190f86 100644 --- a/src/Core/ZB.MOM.WW.OtOpcUa.Core/Resilience/DriverResiliencePipelineBuilder.cs +++ b/src/Core/ZB.MOM.WW.OtOpcUa.Core/Resilience/DriverResiliencePipelineBuilder.cs @@ -1,4 +1,5 @@ using System.Collections.Concurrent; +using Microsoft.Extensions.Logging; using Polly; using Polly.CircuitBreaker; using Polly.Retry; @@ -25,6 +26,7 @@ public sealed class DriverResiliencePipelineBuilder private readonly ConcurrentDictionary _pipelines = new(); private readonly TimeProvider _timeProvider; private readonly DriverResilienceStatusTracker? _statusTracker; + private readonly ILogger? _logger; /// Construct with the ambient clock (use in prod). /// Clock source for pipeline timeouts + breaker sampling. Defaults to system. @@ -33,12 +35,19 @@ public sealed class DriverResiliencePipelineBuilder /// LastBreakerOpenUtc, breaker-close resets failures. Feeds Admin /hosts + /// the Polly bulkhead-depth column. Absent tracker means no telemetry (unit tests + /// deployments that don't care about resilience observability). + /// When non-null, retry / circuit-breaker-open / breaker-close events are + /// logged (instance + host + capability + attempt/exception). This is the operator-facing + /// observability surface for the resilience pipeline until the Admin /hosts reader + /// (Phase 6.1 Stream E.2/E.3) ships — without it the pipeline runs silently. Absent logger = + /// no resilience logging (unit tests). public DriverResiliencePipelineBuilder( TimeProvider? timeProvider = null, - DriverResilienceStatusTracker? statusTracker = null) + DriverResilienceStatusTracker? statusTracker = null, + ILogger? logger = null) { _timeProvider = timeProvider ?? TimeProvider.System; _statusTracker = statusTracker; + _logger = logger; } /// @@ -66,8 +75,8 @@ public sealed class DriverResiliencePipelineBuilder var key = new PipelineKey(driverInstanceId, hostName, capability); return _pipelines.GetOrAdd(key, static (k, state) => Build( - k.DriverInstanceId, k.HostName, state.capability, state.options, state.timeProvider, state.tracker), - (capability, options, timeProvider: _timeProvider, tracker: _statusTracker)); + k.DriverInstanceId, k.HostName, state.capability, state.options, state.timeProvider, state.tracker, state.logger), + (capability, options, timeProvider: _timeProvider, tracker: _statusTracker, logger: _logger)); } /// Drop cached pipelines for one driver instance (e.g. on ResilienceConfig change). Test + Admin-reload use. @@ -93,7 +102,8 @@ public sealed class DriverResiliencePipelineBuilder DriverCapability capability, DriverResilienceOptions options, TimeProvider timeProvider, - DriverResilienceStatusTracker? tracker) + DriverResilienceStatusTracker? tracker, + ILogger? logger) { var policy = options.Resolve(capability); var builder = new ResiliencePipelineBuilder { TimeProvider = timeProvider }; @@ -114,11 +124,14 @@ public sealed class DriverResiliencePipelineBuilder MaxDelay = TimeSpan.FromSeconds(5), ShouldHandle = new PredicateBuilder().Handle(ex => ex is not OperationCanceledException), }; - if (tracker is not null) + if (tracker is not null || logger is not null) { retryOptions.OnRetry = args => { - tracker.RecordFailure(driverInstanceId, hostName, timeProvider.GetUtcNow().UtcDateTime); + tracker?.RecordFailure(driverInstanceId, hostName, timeProvider.GetUtcNow().UtcDateTime); + logger?.LogWarning( + "Driver resilience retry: instance={DriverInstanceId} host={HostName} capability={Capability} attempt={Attempt}: {Reason}", + driverInstanceId, hostName, capability, args.AttemptNumber, args.Outcome.Exception?.Message ?? "unknown"); return default; }; } @@ -135,18 +148,24 @@ public sealed class DriverResiliencePipelineBuilder BreakDuration = TimeSpan.FromSeconds(15), ShouldHandle = new PredicateBuilder().Handle(ex => ex is not OperationCanceledException), }; - if (tracker is not null) + if (tracker is not null || logger is not null) { breakerOptions.OnOpened = args => { - tracker.RecordBreakerOpen(driverInstanceId, hostName, timeProvider.GetUtcNow().UtcDateTime); + tracker?.RecordBreakerOpen(driverInstanceId, hostName, timeProvider.GetUtcNow().UtcDateTime); + logger?.LogWarning( + "Driver resilience circuit-breaker OPENED: instance={DriverInstanceId} host={HostName} capability={Capability} breakDuration={BreakDuration}: {Reason}", + driverInstanceId, hostName, capability, args.BreakDuration, args.Outcome.Exception?.Message ?? "unknown"); return default; }; breakerOptions.OnClosed = args => { // Closing the breaker means the target recovered — reset the consecutive- // failure counter so Admin UI stops flashing red for this host. - tracker.RecordSuccess(driverInstanceId, hostName, timeProvider.GetUtcNow().UtcDateTime); + tracker?.RecordSuccess(driverInstanceId, hostName, timeProvider.GetUtcNow().UtcDateTime); + logger?.LogInformation( + "Driver resilience circuit-breaker closed (recovered): instance={DriverInstanceId} host={HostName} capability={Capability}", + driverInstanceId, hostName, capability); return default; }; } 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 0619af3e..b2123bc7 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.Host/Drivers/DriverFactoryBootstrap.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Host/Drivers/DriverFactoryBootstrap.cs @@ -3,6 +3,7 @@ using Microsoft.Extensions.DependencyInjection.Extensions; using Microsoft.Extensions.Logging; using ZB.MOM.WW.OtOpcUa.Core.Abstractions; using ZB.MOM.WW.OtOpcUa.Core.Hosting; +using ZB.MOM.WW.OtOpcUa.Core.Resilience; namespace ZB.MOM.WW.OtOpcUa.Host.Drivers; @@ -49,6 +50,25 @@ public static class DriverFactoryBootstrap services.AddSingleton(sp => new DriverFactoryRegistryAdapter(sp.GetRequiredService())); + // Phase 6.1 resilience infrastructure — the process-singleton tracker + Polly pipeline builder, + // and the per-driver-instance invoker factory the DriverHostActor uses to wrap every capability + // call. Registered here (driver nodes) because the factory needs DriverFactoryRegistry.GetTier; + // admin-only nodes skip this and the runtime falls back to the pass-through factory. + services.AddSingleton(); + services.AddSingleton(sp => + new DriverResiliencePipelineBuilder( + timeProvider: null, // TimeProvider.System + statusTracker: sp.GetRequiredService(), + logger: sp.GetService()?.CreateLogger("ZB.MOM.WW.OtOpcUa.Core.Resilience.DriverResiliencePipelineBuilder"))); + services.AddSingleton(sp => + { + var registry = sp.GetRequiredService(); + return new DriverCapabilityInvokerFactory( + sp.GetRequiredService(), + sp.GetRequiredService(), + registry.GetTier); + }); + // Driver nodes also carry the probe set so a fused admin,driver node has it; the admin-only // case is covered by Program.cs calling AddOtOpcUaDriverProbes() in the hasAdmin block. services.AddOtOpcUaDriverProbes(); 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 1df6bb83..2f258d98 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/DriverHostActor.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/DriverHostActor.cs @@ -61,6 +61,12 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers private readonly CommonsNodeId _localNode; private readonly IActorRef? _coordinatorOverride; private readonly IDriverFactory _driverFactory; + + /// Builds the per-driver-instance attached to each + /// spawned . Resolved from DI like ; + /// defaults to (pass-through) when the fused Host + /// hasn't bound the real resilience factory. + private readonly IDriverCapabilityInvokerFactory _invokerFactory; private readonly IReadOnlySet _localRoles; private readonly IActorRef? _dependencyMux; private readonly IActorRef? _opcUaPublishActor; @@ -281,6 +287,9 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers /// ScriptedAlarm host instead of spawning a real child, so /// tests can intercept the message. Null /// in production (the real host is spawned). + /// Optional Phase 6.1 resilience-invoker factory used to attach an + /// to each spawned driver; defaults to + /// (pass-through) when not supplied. /// The Akka.NET used to spawn this actor. public static Props Props( IDbContextFactory dbFactory, @@ -296,11 +305,12 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers IActorRef? virtualTagHostOverride = null, ILoggerFactory? loggerFactory = null, ScriptRootLogger? scriptRootLogger = null, - IActorRef? scriptedAlarmHostOverride = null) => + IActorRef? scriptedAlarmHostOverride = null, + IDriverCapabilityInvokerFactory? invokerFactory = null) => Akka.Actor.Props.Create(() => new DriverHostActor( dbFactory, localNode, coordinator, driverFactory, localRoles, dependencyMux, opcUaPublishActor, healthPublisher, virtualTagEvaluator, historyWriter, virtualTagHostOverride, - loggerFactory, scriptRootLogger, scriptedAlarmHostOverride)); + loggerFactory, scriptRootLogger, scriptedAlarmHostOverride, invokerFactory)); /// Initializes a new DriverHostActor with the specified dependencies. /// Database context factory for configuration database access. @@ -323,6 +333,9 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers /// host; when null the host is left unspawned. /// Test seam: when supplied, used as the ScriptedAlarm host /// instead of spawning a real child. + /// Phase 6.1 resilience-invoker factory used to attach an + /// to each spawned driver; defaults to + /// (pass-through) when null. public DriverHostActor( IDbContextFactory dbFactory, CommonsNodeId localNode, @@ -337,12 +350,14 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers IActorRef? virtualTagHostOverride = null, ILoggerFactory? loggerFactory = null, ScriptRootLogger? scriptRootLogger = null, - IActorRef? scriptedAlarmHostOverride = null) + IActorRef? scriptedAlarmHostOverride = null, + IDriverCapabilityInvokerFactory? invokerFactory = null) { _dbFactory = dbFactory; _localNode = localNode; _coordinatorOverride = coordinator; _driverFactory = driverFactory ?? NullDriverFactory.Instance; + _invokerFactory = invokerFactory ?? NullDriverCapabilityInvokerFactory.Instance; _localRoles = localRoles ?? new HashSet(StringComparer.Ordinal); _dependencyMux = dependencyMux; _opcUaPublishActor = opcUaPublishActor; @@ -1609,11 +1624,16 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers } else { + // Attach the per-instance Phase 6.1 resilience invoker so this driver's capability calls + // route through the retry/breaker/bulkhead/telemetry pipeline. The pass-through factory + // yields a no-op invoker on nodes without the real resilience factory bound. + var invoker = _invokerFactory.Create(spec.DriverInstanceId, spec.DriverType); child = Context.ActorOf( DriverInstanceActor.Props( driver!, healthPublisher: _healthPublisher, - clusterId: clusterId), + clusterId: clusterId, + invoker: invoker), 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 ef7fc99c..49de5c74 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/DriverInstanceActor.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/DriverInstanceActor.cs @@ -137,6 +137,15 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers public static readonly TimeSpan HealthPollInterval = TimeSpan.FromSeconds(30); private readonly IDriver _driver; + + /// Phase 6.1 resilience seam: every guarded driver-capability call this actor makes is + /// routed through this invoker (retry / breaker / bulkhead / tracker telemetry). Defaults to + /// (a genuine pass-through) when none is supplied — so + /// unit tests + the F7 skeleton path behave exactly as a raw driver call. The fused Host builds a + /// real per-instance invoker via and injects it at + /// spawn. Consumed through the abstraction because Runtime + /// is deliberately Polly-free (it does not reference ZB.MOM.WW.OtOpcUa.Core). + private readonly IDriverCapabilityInvoker _invoker; private readonly string _driverInstanceId; private readonly string _clusterId; private readonly IDriverHealthPublisher _healthPublisher; @@ -212,6 +221,8 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers /// Optional interval between post-connect re-discovery passes; defaults to 2 seconds. /// Optional cap on re-discovery passes; defaults to 15. /// Optional per-pass timeout for ; defaults to 30 seconds. + /// Optional Phase 6.1 resilience invoker wrapping this driver's capability + /// calls; defaults to (pass-through) when not supplied. /// A instance configured to create the wrapped . public static Props Props( IDriver driver, @@ -221,7 +232,8 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers string? clusterId = null, TimeSpan? rediscoverInterval = null, int rediscoverMaxAttempts = DefaultRediscoverMaxAttempts, - TimeSpan? rediscoverDiscoverTimeout = null) => + TimeSpan? rediscoverDiscoverTimeout = null, + IDriverCapabilityInvoker? invoker = null) => Akka.Actor.Props.Create(() => new DriverInstanceActor( driver, reconnectInterval ?? DefaultReconnectInterval, @@ -230,7 +242,8 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers clusterId ?? string.Empty, rediscoverInterval, rediscoverMaxAttempts, - rediscoverDiscoverTimeout)); + rediscoverDiscoverTimeout, + invoker)); /// /// Returns true when the driver should boot in DEV-STUB mode based on host platform and @@ -262,6 +275,8 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers /// Interval between post-connect re-discovery passes; defaults to 2 seconds. /// Cap on the number of re-discovery passes; defaults to 15. /// Per-pass timeout for ; defaults to 30 seconds. + /// Phase 6.1 resilience invoker wrapping this driver's capability calls; + /// defaults to (pass-through) when null. public DriverInstanceActor( IDriver driver, TimeSpan reconnectInterval, @@ -270,9 +285,11 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers string? clusterId = null, TimeSpan? rediscoverInterval = null, int rediscoverMaxAttempts = DefaultRediscoverMaxAttempts, - TimeSpan? rediscoverDiscoverTimeout = null) + TimeSpan? rediscoverDiscoverTimeout = null, + IDriverCapabilityInvoker? invoker = null) { _driver = driver; + _invoker = invoker ?? NullDriverCapabilityInvoker.Instance; _driverInstanceId = driver.DriverInstanceId; _clusterId = clusterId ?? string.Empty; _healthPublisher = healthPublisher ?? NullDriverHealthPublisher.Instance; @@ -572,9 +589,16 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers using var cts = new CancellationTokenSource(TimeSpan.FromSeconds(5)); try { -#pragma warning disable OTOPCUA0001 // RESILIENCE-DISPATCH-GAP (07/C-1): dispatch bypasses CapabilityInvoker — Phase 6.1 pipeline built but never wired into DriverInstanceActor; tracked, NOT intentional. - var results = await writable.WriteAsync(request, cts.Token); -#pragma warning restore OTOPCUA0001 + // Route through the resilience invoker (Write pipeline: timeout, no retry by default, + // per-host breaker). An OPC UA attribute set is idempotent (writing value X twice == once), + // so a configured Write retry is safe to apply; the Write tier default is RetryCount=0, so + // this stays behavior-neutral until an operator opts in via ResilienceConfig. The outer cts + // is retained as a hard backstop above the (typically tighter) tier timeout. + var results = await _invoker.ExecuteWriteAsync( + ResolveHost(msg.TagId), + isIdempotent: true, + async ct => await writable.WriteAsync(request, ct), + cts.Token); if (results is { Count: 1 } && IsGoodStatus(results[0].StatusCode)) { replyTo.Tell(new WriteAttributeResult(true, null)); @@ -622,9 +646,12 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers using var cts = new CancellationTokenSource(TimeSpan.FromSeconds(5)); try { -#pragma warning disable OTOPCUA0001 // RESILIENCE-DISPATCH-GAP (07/C-1): dispatch bypasses CapabilityInvoker — Phase 6.1 pipeline built but never wired into DriverInstanceActor; tracked, NOT intentional. - await src.AcknowledgeAsync(request, cts.Token); -#pragma warning restore OTOPCUA0001 + // Ack is a write-shaped operation — the AlarmAcknowledge tier policy never retries. + await _invoker.ExecuteAsync( + DriverCapability.AlarmAcknowledge, + ResolveHost(msg.ConditionId), + async ct => await src.AcknowledgeAsync(request, ct), + cts.Token); _log.Info("DriverInstance {Id}: acknowledged native condition {Cond} by {User}", _driverInstanceId, msg.ConditionId, msg.OperatorUser); } @@ -661,10 +688,13 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers subscribable.OnDataChange += _dataChangeHandler; using var cts = new CancellationTokenSource(TimeSpan.FromSeconds(10)); -#pragma warning disable OTOPCUA0001 // RESILIENCE-DISPATCH-GAP (07/C-1): dispatch bypasses CapabilityInvoker — Phase 6.1 pipeline built but never wired into DriverInstanceActor; tracked, NOT intentional. - _subscriptionHandle = await subscribable - .SubscribeAsync(msg.FullReferences, msg.PublishingInterval, cts.Token); -#pragma warning restore OTOPCUA0001 + // A bulk subscribe spans potentially many hosts, so key the pipeline on the driver + // instance (single-host fallback) rather than any one reference's host. + _subscriptionHandle = await _invoker.ExecuteAsync( + DriverCapability.Subscribe, + _driverInstanceId, + async ct => await subscribable.SubscribeAsync(msg.FullReferences, msg.PublishingInterval, ct), + cts.Token); replyTo.Tell(new SubscriptionEstablished(_subscriptionHandle.DiagnosticId, msg.FullReferences.Count)); _log.Info("DriverInstance {Id}: subscribed to {Count} refs ({Diag})", @@ -688,9 +718,12 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers try { using var cts = new CancellationTokenSource(TimeSpan.FromSeconds(5)); -#pragma warning disable OTOPCUA0001 // RESILIENCE-DISPATCH-GAP (07/C-1): dispatch bypasses CapabilityInvoker — Phase 6.1 pipeline built but never wired into DriverInstanceActor; tracked, NOT intentional. - await subscribable.UnsubscribeAsync(_subscriptionHandle, cts.Token); -#pragma warning restore OTOPCUA0001 + var handle = _subscriptionHandle; // non-null per the guard above; capture before the async lambda + await _invoker.ExecuteAsync( + DriverCapability.Subscribe, + _driverInstanceId, + async ct => await subscribable.UnsubscribeAsync(handle, ct), + cts.Token); } catch (Exception ex) { @@ -767,9 +800,11 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers using var cts = new CancellationTokenSource(TimeSpan.FromSeconds(10)); try { -#pragma warning disable OTOPCUA0001 // RESILIENCE-DISPATCH-GAP (07/C-1): dispatch bypasses CapabilityInvoker — Phase 6.1 pipeline built but never wired into DriverInstanceActor; tracked, NOT intentional. - _alarmSubscriptionHandle = await src.SubscribeAlarmsAsync(refs, cts.Token); -#pragma warning restore OTOPCUA0001 + _alarmSubscriptionHandle = await _invoker.ExecuteAsync( + DriverCapability.AlarmSubscribe, + _driverInstanceId, + async ct => await src.SubscribeAlarmsAsync(refs, ct), + cts.Token); _log.Info("DriverInstance {Id}: native-alarm subscription established for {Count} alarm ref(s) ({Diag})", _driverInstanceId, refs.Count, _alarmSubscriptionHandle.DiagnosticId); } @@ -824,13 +859,17 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers // Bound the browse — ReceiveAsync suspends the mailbox for the whole handler, so an unbounded // DiscoverAsync would block DisconnectObserved / ForceReconnect / writes / health-poll behind it. using var cts = new CancellationTokenSource(_rediscoverDiscoverTimeout); - // NO ConfigureAwait(false): a genuinely-async DiscoverAsync (Galaxy / OpcUaClient / TwinCAT) must - // resume on the actor task scheduler so the Context.Parent.Tell + Timers calls below run with a - // live ActorContext. ConfigureAwait(false) would resume off-context and throw - // NotSupportedException("no active ActorContext") — see the same warning on HandleSubscribeAsync. -#pragma warning disable OTOPCUA0001 // RESILIENCE-DISPATCH-GAP (07/C-1): dispatch bypasses CapabilityInvoker — Phase 6.1 pipeline built but never wired into DriverInstanceActor; tracked, NOT intentional. - await discovery.DiscoverAsync(builder, cts.Token); -#pragma warning restore OTOPCUA0001 + // NO ConfigureAwait(false) on this outer await: a genuinely-async DiscoverAsync (Galaxy / + // OpcUaClient / TwinCAT) must resume on the actor task scheduler so the Context.Parent.Tell + + // Timers calls below run with a live ActorContext. ConfigureAwait(false) would resume + // off-context and throw NotSupportedException("no active ActorContext"). The invoker's own + // internal ConfigureAwait(false) does NOT propagate to this caller's await — the actor + // continuation still resumes on the captured actor scheduler. (Discover retries per tier.) + await _invoker.ExecuteAsync( + DriverCapability.Discover, + _driverInstanceId, + async ct => await discovery.DiscoverAsync(builder, ct), + cts.Token); nodes = builder.Nodes.ToArray(); // immutable snapshot — never hand the builder's live list across actors } catch (Exception ex) @@ -910,6 +949,14 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers private static bool IsGoodStatus(uint statusCode) => (statusCode >> 30) == 0; + /// Resolve the resilience-pipeline host key for a single driver-side reference. Multi-host + /// drivers (e.g. Modbus across N PLCs) implement so one dead host + /// trips only its own breaker; single-host drivers fall back to the driver-instance id. Used for the + /// single-reference capability calls (write, alarm ack); bulk/whole-driver calls key on the instance + /// id directly. + private string ResolveHost(string reference) => + _driver is IPerCallHostResolver resolver ? resolver.ResolveHost(reference) : _driverInstanceId; + /// Records a transition into a Faulted / error state for the 5-minute sliding counter. private void RecordFault() { diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/ServiceCollectionExtensions.cs b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/ServiceCollectionExtensions.cs index 6872052e..662f3df9 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/ServiceCollectionExtensions.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/ServiceCollectionExtensions.cs @@ -213,6 +213,11 @@ public static class ServiceCollectionExtensions // Fallback to Null* if AddOtOpcUaRuntime wasn't called (e.g., test harnesses). var historianSink = resolver.GetService() ?? NullAlarmHistorianSink.Instance; var driverFactory = resolver.GetService() ?? NullDriverFactory.Instance; + // Phase 6.1 resilience-invoker factory — bound by the fused Host (which references Core + + // Polly); pass-through on nodes/harnesses that didn't register it. Resolved through the + // Core.Abstractions seam so this Polly-free assembly stays that way. + var invokerFactory = resolver.GetService() + ?? NullDriverCapabilityInvokerFactory.Instance; var addressSpaceSink = resolver.GetService() ?? NullOpcUaAddressSpaceSink.Instance; var serviceLevel = resolver.GetService() ?? NullServiceLevelPublisher.Instance; var loggerFactory = resolver.GetService() ?? NullLoggerFactory.Instance; @@ -333,7 +338,8 @@ public static class ServiceCollectionExtensions virtualTagEvaluator: virtualTagEvaluator, historyWriter: historyWriter, loggerFactory: loggerFactory, - scriptRootLogger: scriptRootLogger), + scriptRootLogger: scriptRootLogger, + invokerFactory: invokerFactory), DriverHostActorName); registry.Register(driverHost); diff --git a/src/Tooling/ZB.MOM.WW.OtOpcUa.Analyzers/UnwrappedCapabilityCallAnalyzer.cs b/src/Tooling/ZB.MOM.WW.OtOpcUa.Analyzers/UnwrappedCapabilityCallAnalyzer.cs index 82620cb4..644e0f86 100644 --- a/src/Tooling/ZB.MOM.WW.OtOpcUa.Analyzers/UnwrappedCapabilityCallAnalyzer.cs +++ b/src/Tooling/ZB.MOM.WW.OtOpcUa.Analyzers/UnwrappedCapabilityCallAnalyzer.cs @@ -91,6 +91,12 @@ public sealed class UnwrappedCapabilityCallAnalyzer : DiagnosticAnalyzer [ ("ZB.MOM.WW.OtOpcUa.Core.Resilience.CapabilityInvoker", "ExecuteAsync"), ("ZB.MOM.WW.OtOpcUa.Core.Resilience.CapabilityInvoker", "ExecuteWriteAsync"), + // The Polly-free Runtime dispatch layer wraps calls through the IDriverCapabilityInvoker + // seam (it cannot reference the concrete CapabilityInvoker in Core). An interface-typed + // invoker call's containing type is the interface, so it must be a wrapper home too — else + // correctly-wrapped dispatch-layer calls would still trip OTOPCUA0001. + ("ZB.MOM.WW.OtOpcUa.Core.Abstractions.IDriverCapabilityInvoker", "ExecuteAsync"), + ("ZB.MOM.WW.OtOpcUa.Core.Abstractions.IDriverCapabilityInvoker", "ExecuteWriteAsync"), ]; private static readonly DiagnosticDescriptor Rule = new( diff --git a/tests/Core/ZB.MOM.WW.OtOpcUa.Core.Tests/Resilience/DriverResiliencePipelineBuilderTests.cs b/tests/Core/ZB.MOM.WW.OtOpcUa.Core.Tests/Resilience/DriverResiliencePipelineBuilderTests.cs index 2a78f3b8..32b6ea73 100644 --- a/tests/Core/ZB.MOM.WW.OtOpcUa.Core.Tests/Resilience/DriverResiliencePipelineBuilderTests.cs +++ b/tests/Core/ZB.MOM.WW.OtOpcUa.Core.Tests/Resilience/DriverResiliencePipelineBuilderTests.cs @@ -1,3 +1,4 @@ +using Microsoft.Extensions.Logging; using Polly.CircuitBreaker; using Polly.Timeout; using Shouldly; @@ -296,4 +297,48 @@ public sealed class DriverResiliencePipelineBuilderTests tracker.TryGet("drv-trk", "dead")!.ConsecutiveFailures.ShouldBeGreaterThan(0); tracker.TryGet("drv-trk", "live").ShouldBeNull(); } + + /// + /// Retry events are logged when a logger is supplied — this is the operator-facing observability + /// surface the arch-review #10 live-verify relies on (the Admin /hosts reader isn't built). + /// Without it the resilience pipeline runs silently and a live retry/breaker transition is invisible. + /// + [Fact] + public async Task Retry_Is_Logged_When_Logger_Supplied() + { + var logger = new CapturingLogger(); + var builder = new DriverResiliencePipelineBuilder(logger: logger); + var pipeline = builder.GetOrCreate("drv-log", "host-9", DriverCapability.Read, TierAOptions); + var attempts = 0; + + await pipeline.ExecuteAsync(async _ => + { + attempts++; + if (attempts < 3) throw new InvalidOperationException("transient"); + await Task.Yield(); + }); + + attempts.ShouldBe(3); + logger.Entries.ShouldContain(e => + e.Level == LogLevel.Warning && + e.Message.Contains("resilience retry") && + e.Message.Contains("host-9")); + } + + /// Minimal in-memory capturing (level, formatted-message) pairs so the + /// resilience logging contract can be asserted without a real logging provider. + private sealed class CapturingLogger : ILogger + { + public List<(LogLevel Level, string Message)> Entries { get; } = new(); + public IDisposable BeginScope(TState state) where TState : notnull => NullScope.Instance; + public bool IsEnabled(LogLevel logLevel) => true; + public void Log(LogLevel logLevel, EventId eventId, TState state, Exception? exception, + Func formatter) => Entries.Add((logLevel, formatter(state, exception))); + + private sealed class NullScope : IDisposable + { + public static readonly NullScope Instance = new(); + public void Dispose() { } + } + } } diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverInstanceActorDiscoveryTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverInstanceActorDiscoveryTests.cs index 7973c920..2e74a13e 100644 --- a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverInstanceActorDiscoveryTests.cs +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverInstanceActorDiscoveryTests.cs @@ -2,6 +2,7 @@ using Akka.Actor; using Shouldly; using Xunit; using ZB.MOM.WW.OtOpcUa.Core.Abstractions; +using ZB.MOM.WW.OtOpcUa.Core.Resilience; using ZB.MOM.WW.OtOpcUa.Runtime.Drivers; using ZB.MOM.WW.OtOpcUa.Runtime.Tests.Harness; @@ -144,6 +145,38 @@ public sealed class DriverInstanceActorDiscoveryTests : RuntimeActorTestBase published.DriverInstanceId.ShouldBe(driver.DriverInstanceId); } + /// + /// Arch-review #10: the SAME async-discovery context-preservation guarantee, but driven through the + /// REAL — whose Polly pipeline is awaited with + /// ConfigureAwait(false) internally. This is the one path the pass-through + /// NullDriverCapabilityInvoker (used by the sibling test) cannot cover: it returns the + /// call-site task directly, so it never exercises the invoker's internal off-context await. If that + /// internal ConfigureAwait(false) leaked to the actor's own await _invoker.ExecuteAsync(…), + /// the post-await Context.Parent.Tell in HandleRediscoverAsync would fault with + /// "no active ActorContext" and no message would arrive. Its arrival proves the actor context survives + /// the real resilience pipeline. + /// + [Fact] + public void Async_discovery_through_real_CapabilityInvoker_preserves_actor_context() + { + var pipelineBuilder = new DriverResiliencePipelineBuilder(); // real Polly pipeline, tier-A defaults + var invoker = new CapabilityInvoker( + pipelineBuilder, + driverInstanceId: "yielding-discoverable", + optionsAccessor: () => new DriverResilienceOptions { Tier = DriverTier.A }, + driverType: "Stub"); + var driver = new YieldingDiscoverableStubDriver(); + var parent = CreateTestProbe(); + var actor = parent.ChildActorOf(DriverInstanceActor.Props( + driver, rediscoverInterval: TimeSpan.FromMilliseconds(20), invoker: invoker)); + + actor.Tell(new DriverInstanceActor.InitializeRequested("{}")); + + var published = parent.ExpectMsg(TimeSpan.FromSeconds(5)); + published.Nodes.Count.ShouldBe(3); + published.DriverInstanceId.ShouldBe(driver.DriverInstanceId); + } + /// /// The attempt cap bounds a discovered set that never stabilises: a driver whose set keeps GROWING /// (1,2,3,…) never repeats its signature, so the loop is stopped only by diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverInstanceActorResilienceWiringTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverInstanceActorResilienceWiringTests.cs new file mode 100644 index 00000000..323f3431 --- /dev/null +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverInstanceActorResilienceWiringTests.cs @@ -0,0 +1,138 @@ +using System.Collections.Concurrent; +using Akka.Actor; +using Shouldly; +using Xunit; +using ZB.MOM.WW.OtOpcUa.Commons.Types; +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; + +/// +/// Runtime guard for the Phase 6.1 resilience wiring (arch-review #10 / RESILIENCE-DISPATCH-GAP): +/// proves the routes each driver-capability call through the +/// injected at RUNTIME, with the correct +/// and resolved host key — not just that the calls compile wrapped +/// (the OTOPCUA0001 analyzer is the compile-time guard; this is the behavioural one). +/// A recording invoker stands in for the real CapabilityInvoker so the test stays in the +/// Polly-free Runtime layer. +/// +public sealed class DriverInstanceActorResilienceWiringTests : RuntimeActorTestBase +{ + /// + /// A bulk subscribe issued on connect (via ResubscribeDesired) routes through the invoker + /// with keyed on the driver-instance id (single-host + /// fallback for a set that may span hosts). The driver's SubscribeAsync only runs from + /// inside the invoker's call-site, so SubscribeCount >= 1 already proves the route. + /// + [Fact] + public async Task Subscribe_routes_through_the_capability_invoker() + { + var invoker = new RecordingInvoker(); + var driver = new SubscribableStubDriver(); + var actor = Sys.ActorOf(DriverInstanceActor.Props(driver, invoker: invoker)); + + actor.Tell(new DriverInstanceActor.SetDesiredSubscriptions( + new[] { "tag-a", "tag-b" }, TimeSpan.FromMilliseconds(100))); + actor.Tell(new DriverInstanceActor.InitializeRequested("{}")); + AwaitCondition(() => driver.SubscribeCount >= 1, TimeSpan.FromSeconds(3)); + + // Barrier: round-trip a Connected-only Ask so the async subscribe handler has fully recorded. + await actor.Ask( + new DriverInstanceActor.Subscribe(new[] { "tag-a" }, TimeSpan.FromMilliseconds(100)), + TimeSpan.FromSeconds(3)); + + invoker.Calls.ShouldContain(c => c.Capability == DriverCapability.Subscribe && c.Host == driver.DriverInstanceId); + } + + /// + /// A write routes through the invoker's ExecuteWriteAsync path (recorded as + /// ) with the host resolved via + /// — proving per-host breaker keying is wired, not the + /// single-host fallback. A write that arrives before Connected fast-fails WITHOUT + /// touching the invoker, so the loop drives to Connected and the recorded call is the real one. + /// + [Fact] + public async Task Write_routes_through_the_invoker_with_resolved_host() + { + var invoker = new RecordingInvoker(); + var driver = new HostResolvingWritableDriver(); // ResolveHost("tag-1") => "plc-7" + var actor = Sys.ActorOf(DriverInstanceActor.Props(driver, invoker: invoker)); + actor.Tell(new DriverInstanceActor.InitializeRequested("{}")); + + // Retry until the write succeeds (i.e. the actor reached Connected). Connecting-state writes + // fast-fail on a synchronous path that never reaches the invoker, so they don't pollute Calls. + DriverInstanceActor.WriteAttributeResult? reply = null; + for (var i = 0; i < 60 && reply is not { Success: true }; i++) + { + reply = await actor.Ask( + new DriverInstanceActor.WriteAttribute("tag-1", 42), TimeSpan.FromSeconds(1)); + if (reply is not { Success: true }) await Task.Delay(25); + } + + reply.ShouldNotBeNull(); + reply!.Success.ShouldBeTrue(); + driver.Writes.Count.ShouldBeGreaterThanOrEqualTo(1); // the invoker's call-site actually ran + invoker.Calls.ShouldContain(c => c.Capability == DriverCapability.Write && c.Host == "plc-7"); + } + + /// Records every capability + resolved-host key routed through it, then delegates to the + /// real call-site so the driver still runs. Stands in for CapabilityInvoker without pulling + /// Polly / Core into the Runtime test. + private sealed class RecordingInvoker : IDriverCapabilityInvoker + { + public ConcurrentQueue<(DriverCapability Capability, string Host)> Calls { get; } = new(); + + public ValueTask ExecuteAsync( + DriverCapability capability, string hostName, + Func> callSite, CancellationToken cancellationToken) + { + Calls.Enqueue((capability, hostName)); + return callSite(cancellationToken); + } + + public ValueTask ExecuteAsync( + DriverCapability capability, string hostName, + Func callSite, CancellationToken cancellationToken) + { + Calls.Enqueue((capability, hostName)); + return callSite(cancellationToken); + } + + public ValueTask ExecuteWriteAsync( + string hostName, bool isIdempotent, + Func> callSite, CancellationToken cancellationToken) + { + Calls.Enqueue((DriverCapability.Write, hostName)); + return callSite(cancellationToken); + } + } + + /// A writable driver that also resolves a per-tag host, so the write path exercises + /// 's ResolveHost (via ). + private sealed class HostResolvingWritableDriver : IDriver, IWritable, IPerCallHostResolver + { + public List Writes { get; } = new(); + + public string DriverInstanceId => "host-resolving-driver"; + public string DriverType => "Stub"; + + public string ResolveHost(string fullReference) => "plc-7"; + + public Task InitializeAsync(string driverConfigJson, CancellationToken cancellationToken) => Task.CompletedTask; + public Task ReinitializeAsync(string driverConfigJson, CancellationToken cancellationToken) => Task.CompletedTask; + public Task ShutdownAsync(CancellationToken cancellationToken) => Task.CompletedTask; + public DriverHealth GetHealth() => new(DriverState.Healthy, DateTime.UtcNow, null); + public long GetMemoryFootprint() => 0; + public Task FlushOptionalCachesAsync(CancellationToken cancellationToken) => Task.CompletedTask; + + public Task> WriteAsync( + IReadOnlyList writes, CancellationToken cancellationToken) + { + Writes.AddRange(writes); + IReadOnlyList results = writes.Select(_ => new WriteResult(0u)).ToList(); + return Task.FromResult(results); + } + } +} diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/ZB.MOM.WW.OtOpcUa.Runtime.Tests.csproj b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/ZB.MOM.WW.OtOpcUa.Runtime.Tests.csproj index 5c48ed9b..14f0e0b2 100644 --- a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/ZB.MOM.WW.OtOpcUa.Runtime.Tests.csproj +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/ZB.MOM.WW.OtOpcUa.Runtime.Tests.csproj @@ -22,6 +22,10 @@ + + diff --git a/tests/Tooling/ZB.MOM.WW.OtOpcUa.Analyzers.Tests/UnwrappedCapabilityCallAnalyzerTests.cs b/tests/Tooling/ZB.MOM.WW.OtOpcUa.Analyzers.Tests/UnwrappedCapabilityCallAnalyzerTests.cs index 0fcaab12..c13a25bb 100644 --- a/tests/Tooling/ZB.MOM.WW.OtOpcUa.Analyzers.Tests/UnwrappedCapabilityCallAnalyzerTests.cs +++ b/tests/Tooling/ZB.MOM.WW.OtOpcUa.Analyzers.Tests/UnwrappedCapabilityCallAnalyzerTests.cs @@ -61,6 +61,11 @@ namespace ZB.MOM.WW.OtOpcUa.Core.Abstractions { => throw new NotSupportedException(); } public enum DriverCapability { Read, Write, Discover, AlarmSubscribe, AlarmAcknowledge } + public interface IDriverCapabilityInvoker { + ValueTask ExecuteAsync(DriverCapability c, string host, Func> call, CancellationToken ct); + ValueTask ExecuteAsync(DriverCapability c, string host, Func call, CancellationToken ct); + ValueTask ExecuteWriteAsync(string host, bool isIdempotent, Func> call, CancellationToken ct); + } } namespace ZB.MOM.WW.OtOpcUa.Core.Resilience { using System; @@ -130,6 +135,36 @@ namespace ZB.MOM.WW.OtOpcUa.Server { diags.ShouldBeEmpty(); } + /// The Polly-free Runtime dispatch layer wraps calls through the + /// IDriverCapabilityInvoker seam (it cannot reference the concrete CapabilityInvoker). + /// An interface-typed invoker call must be recognised as a wrapper home too — arch-review #10. + [Fact] + public async Task Wrapped_InsideDriverCapabilityInvokerInterfaceLambda_PassesCleanly() + { + const string userSrc = """ +using System.Collections.Generic; +using System.Threading; +using System.Threading.Tasks; +using ZB.MOM.WW.OtOpcUa.Core.Abstractions; + +namespace ZB.MOM.WW.OtOpcUa.Server { + public sealed class GoodSeamCaller { + public async Task Sub(ISubscribable driver, IDriverCapabilityInvoker invoker) { + var _ = await invoker.ExecuteAsync(DriverCapability.Read, "h1", + async ct => await driver.SubscribeAsync(new List(), System.TimeSpan.Zero, ct), CancellationToken.None); + } + public async Task Wr(IReadable driver, IDriverCapabilityInvoker invoker) { + // ExecuteWriteAsync interface overload is a wrapper home too (value-returning call wrapped). + var _ = await invoker.ExecuteWriteAsync("h1", true, + async ct => await driver.ReadAsync(new List(), ct), CancellationToken.None); + } + } +} +"""; + var diags = await Compile(userSrc); + diags.ShouldBeEmpty(); + } + /// Verifies that a direct write without wrapper trips the diagnostic. [Fact] public async Task DirectWrite_WithoutWrapper_TripsDiagnostic()