fix(archreview #10): wire CapabilityInvoker into prod dispatch via a Polly-free seam

Remediates RESILIENCE-DISPATCH-GAP (surfaced by the 07/C-1 analyzer): the Phase 6.1
CapabilityInvoker resilience pipeline (retry / breaker / bulkhead / telemetry) was
constructed only in tests — the production dispatch layer called driver-capability
methods directly, bypassing it entirely.

The filed plan (thread CapabilityInvoker directly into DriverInstanceActor) is
INFEASIBLE: Runtime is deliberately Polly-free (references Core.Abstractions, not
Core — the same boundary IDriverFactory documents). So this introduces a seam,
mirroring IDriverFactory exactly:

- Core.Abstractions: IDriverCapabilityInvoker (+ NullDriverCapabilityInvoker
  pass-through) and IDriverCapabilityInvokerFactory (+ null factory).
- Core: CapabilityInvoker now implements the interface; new
  DriverCapabilityInvokerFactory builds a per-instance invoker over the
  process-singleton pipeline builder + status tracker + tier resolver.
- Runtime: DriverInstanceActor takes an IDriverCapabilityInvoker (default =
  pass-through) and routes all 6 dispatch sites (write, alarm-ack, subscribe,
  unsubscribe, alarm-subscribe, discover) through it; per-host key resolved via
  IPerCallHostResolver for single-ref calls, driver-instance key for bulk calls.
  DriverHostActor builds + injects the real invoker per spawned driver.
- Host: DriverFactoryBootstrap registers the tracker, pipeline builder, and
  concrete factory (needs DriverFactoryRegistry.GetTier). Runtime SCE resolves the
  factory from DI like IDriverFactory; pass-through on nodes without it bound.
- Analyzer: IDriverCapabilityInvoker.ExecuteAsync/ExecuteWriteAsync are now
  recognized wrapper homes (interface-typed invoker calls must suppress OTOPCUA0001
  too). All 6 RESILIENCE-DISPATCH-GAP pragmas removed — the analyzer (an error in
  Runtime) is now the standing regression guard.

Also adds retry / breaker-open / breaker-close LOGGING to the pipeline builder —
the operator-facing observability surface (the Admin /hosts reader, Phase 6.1
Stream E.2/E.3, was never built, so the pipeline otherwise runs silently).

Scope notes:
- Tier-DEFAULT policy only: DriverInstanceSpec (the deploy artifact) does not carry
  the per-instance ResilienceConfig JSON, so overrides aren't applied yet (tracked
  follow-up: plumb ResilienceConfig through the composer/artifact).
- GenericDriverNodeManager's call is NOT wired: that class is test-only scaffolding
  (zero production references — the Server has its own address-space path); its
  pragma is re-annotated accordingly, not left as a "tracked gap".

Verification (unit + analyzer; live behavioral gate still pending — see FOLLOWUP-10):
- Negative control: unwrapping any site fails the Runtime build with OTOPCUA0001.
- New Runtime guard (recording invoker): write routes via ExecuteWriteAsync with the
  IPerCallHostResolver host; subscribe via ExecuteAsync — proves runtime routing.
- New analyzer test: the interface is a valid wrapper home.
- New pipeline-builder test: retry events are logged.
- Full solution builds clean (0 errors); Runtime.Tests 357, Core.Tests 238,
  Analyzers.Tests 32 all green; pass-through default keeps existing dispatch tests
  byte-for-byte unchanged.
This commit is contained in:
Joseph Doherty
2026-07-08 21:33:16 -04:00
parent f0082af5b9
commit bacea1a44f
14 changed files with 585 additions and 48 deletions
@@ -0,0 +1,99 @@
namespace ZB.MOM.WW.OtOpcUa.Core.Abstractions;
/// <summary>
/// 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 (<c>DriverInstanceActor</c>) consumes this interface instead of the concrete
/// <c>CapabilityInvoker</c> so the actor assembly does NOT pull in
/// <c>ZB.MOM.WW.OtOpcUa.Core</c> (which drags in Polly + driver hosting) — the same boundary
/// rationale as <see cref="IDriverFactory"/>. The fused Host binds a real invoker (built by
/// <see cref="IDriverCapabilityInvokerFactory"/>) per spawned driver; when none is supplied the
/// dispatch layer uses <see cref="NullDriverCapabilityInvoker"/>, a genuine pass-through.
/// </summary>
/// <remarks>
/// The method shapes mirror <c>CapabilityInvoker</c> exactly so the concrete invoker satisfies
/// this interface without adapters. Callers pass the resolved <c>hostName</c> per call (a
/// multi-host driver resolves it from the tag reference via <see cref="IPerCallHostResolver"/>;
/// a single-host driver passes its <see cref="IDriver.DriverInstanceId"/>) so per-host breaker
/// isolation + bulkhead-depth accounting key on the right target.
/// </remarks>
public interface IDriverCapabilityInvoker
{
/// <summary>Execute a value-returning capability call through the per-capability pipeline.</summary>
/// <typeparam name="TResult">Return type of the underlying driver call.</typeparam>
/// <param name="capability">The driver capability being executed.</param>
/// <param name="hostName">The resolved host name for pipeline keying + status tracking.</param>
/// <param name="callSite">The async function to execute (the raw driver-capability call).</param>
/// <param name="cancellationToken">Cancellation token for the operation.</param>
/// <returns>The result of the underlying driver call.</returns>
ValueTask<TResult> ExecuteAsync<TResult>(
DriverCapability capability,
string hostName,
Func<CancellationToken, ValueTask<TResult>> callSite,
CancellationToken cancellationToken);
/// <summary>Execute a void-returning capability call through the per-capability pipeline.</summary>
/// <param name="capability">The driver capability being executed.</param>
/// <param name="hostName">The resolved host name for pipeline keying + status tracking.</param>
/// <param name="callSite">The async function to execute (the raw driver-capability call).</param>
/// <param name="cancellationToken">Cancellation token for the operation.</param>
/// <returns>A task that represents the asynchronous operation.</returns>
ValueTask ExecuteAsync(
DriverCapability capability,
string hostName,
Func<CancellationToken, ValueTask> callSite,
CancellationToken cancellationToken);
/// <summary>
/// Execute a <see cref="DriverCapability.Write"/> call honoring idempotence semantics — when
/// <paramref name="isIdempotent"/> is <c>false</c>, retries are disabled regardless of the
/// configured policy; when <c>true</c>, the call runs through the Write pipeline which may
/// retry if the tier configuration permits.
/// </summary>
/// <typeparam name="TResult">Return type of the underlying driver write call.</typeparam>
/// <param name="hostName">The resolved host name for pipeline keying + status tracking.</param>
/// <param name="isIdempotent">Whether the write operation is safe to retry.</param>
/// <param name="callSite">The async function to execute (the raw driver write call).</param>
/// <param name="cancellationToken">Cancellation token for the operation.</param>
/// <returns>The result of the underlying driver write call.</returns>
ValueTask<TResult> ExecuteWriteAsync<TResult>(
string hostName,
bool isIdempotent,
Func<CancellationToken, ValueTask<TResult>> callSite,
CancellationToken cancellationToken);
}
/// <summary>
/// Pass-through <see cref="IDriverCapabilityInvoker"/> 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.
/// </summary>
public sealed class NullDriverCapabilityInvoker : IDriverCapabilityInvoker
{
/// <summary>The shared singleton instance.</summary>
public static readonly NullDriverCapabilityInvoker Instance = new();
private NullDriverCapabilityInvoker() { }
/// <inheritdoc />
public ValueTask<TResult> ExecuteAsync<TResult>(
DriverCapability capability,
string hostName,
Func<CancellationToken, ValueTask<TResult>> callSite,
CancellationToken cancellationToken) => callSite(cancellationToken);
/// <inheritdoc />
public ValueTask ExecuteAsync(
DriverCapability capability,
string hostName,
Func<CancellationToken, ValueTask> callSite,
CancellationToken cancellationToken) => callSite(cancellationToken);
/// <inheritdoc />
public ValueTask<TResult> ExecuteWriteAsync<TResult>(
string hostName,
bool isIdempotent,
Func<CancellationToken, ValueTask<TResult>> callSite,
CancellationToken cancellationToken) => callSite(cancellationToken);
}
@@ -0,0 +1,39 @@
namespace ZB.MOM.WW.OtOpcUa.Core.Abstractions;
/// <summary>
/// Builds a per-driver-instance <see cref="IDriverCapabilityInvoker"/>. The Runtime
/// <c>DriverHostActor</c> consumes this factory (resolved from DI, like <see cref="IDriverFactory"/>)
/// to attach a resilience invoker to each spawned <c>DriverInstanceActor</c> WITHOUT the actor
/// assembly referencing <c>ZB.MOM.WW.OtOpcUa.Core</c> (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, <see cref="NullDriverCapabilityInvokerFactory"/> yields the
/// pass-through invoker so dispatch behaves exactly as before.
/// </summary>
public interface IDriverCapabilityInvokerFactory
{
/// <summary>
/// Create the resilience invoker for one driver instance. The returned invoker is keyed on
/// <paramref name="driverInstanceId"/> and applies the tier policy resolved from
/// <paramref name="driverType"/>.
/// </summary>
/// <param name="driverInstanceId">The driver instance identifier (pipeline + tracker key).</param>
/// <param name="driverType">The driver type name, used to resolve the resilience tier defaults.</param>
/// <returns>A per-instance <see cref="IDriverCapabilityInvoker"/>.</returns>
IDriverCapabilityInvoker Create(string driverInstanceId, string driverType);
}
/// <summary>
/// Returns <see cref="NullDriverCapabilityInvoker.Instance"/> 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.
/// </summary>
public sealed class NullDriverCapabilityInvokerFactory : IDriverCapabilityInvokerFactory
{
/// <summary>The shared singleton instance.</summary>
public static readonly NullDriverCapabilityInvokerFactory Instance = new();
private NullDriverCapabilityInvokerFactory() { }
/// <inheritdoc />
public IDriverCapabilityInvoker Create(string driverInstanceId, string driverType) =>
NullDriverCapabilityInvoker.Instance;
}
@@ -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
@@ -7,16 +7,21 @@ namespace ZB.MOM.WW.OtOpcUa.Core.Resilience;
/// <summary>
/// Executes driver-capability calls through a shared Polly pipeline. One invoker per
/// <c>(DriverInstance, IDriver)</c> pair; the underlying <see cref="DriverResiliencePipelineBuilder"/>
/// is process-singleton so all invokers share its cache.
/// is process-singleton so all invokers share its cache. Implements
/// <see cref="IDriverCapabilityInvoker"/> so the Polly-free Runtime dispatch layer can consume it
/// through the abstraction (bound per-driver by <see cref="IDriverCapabilityInvokerFactory"/>).
/// </summary>
/// <remarks>
/// The server's dispatch
/// layer routes every capability call (<c>IReadable.ReadAsync</c>, <c>IWritable.WriteAsync</c>,
/// <c>ITagDiscovery.DiscoverAsync</c>, <c>ISubscribable.SubscribeAsync/UnsubscribeAsync</c>,
/// <c>IHostConnectivityProbe</c> probe loop, <c>IAlarmSource.SubscribeAlarmsAsync/AcknowledgeAsync</c>,
/// and all four <c>IHistoryProvider</c> reads) through this invoker.
/// The Runtime dispatch layer (<c>DriverInstanceActor</c>) routes its capability calls
/// (<c>IWritable.WriteAsync</c>, <c>ITagDiscovery.DiscoverAsync</c>,
/// <c>ISubscribable.SubscribeAsync/UnsubscribeAsync</c>,
/// <c>IAlarmSource.SubscribeAlarmsAsync/AcknowledgeAsync</c>) through this invoker via the
/// <see cref="IDriverCapabilityInvoker"/> seam. The <c>UnwrappedCapabilityCallAnalyzer</c>
/// (OTOPCUA0001) enforces that every guarded call is wrapped in an <see cref="ExecuteAsync{TResult}"/>
/// / <see cref="ExecuteWriteAsync{TResult}"/> lambda (the interface's methods are registered wrapper
/// homes too, so an interface-typed invoker call suppresses the diagnostic).
/// </remarks>
public sealed class CapabilityInvoker
public sealed class CapabilityInvoker : IDriverCapabilityInvoker
{
private readonly DriverResiliencePipelineBuilder _builder;
private readonly string _driverInstanceId;
@@ -0,0 +1,58 @@
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
namespace ZB.MOM.WW.OtOpcUa.Core.Resilience;
/// <summary>
/// Concrete <see cref="IDriverCapabilityInvokerFactory"/> that builds a per-driver-instance
/// <see cref="CapabilityInvoker"/> over the process-singleton pipeline builder + status tracker.
/// The fused Host registers this (it references <c>Core</c> + Polly); the Runtime dispatch layer
/// resolves the abstraction from DI and never sees this type.
/// </summary>
/// <remarks>
/// <para>Resilience policy is currently the <b>tier default</b> for the driver type — resolved
/// via the injected tier resolver (backed by <c>DriverFactoryRegistry.GetTier</c>). The
/// per-instance <c>DriverInstance.ResilienceConfig</c> JSON override is NOT yet applied because
/// the deployment artifact (<c>DriverInstanceSpec</c>) 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.</para>
///
/// <para>Options are snapshotted once per <see cref="Create"/> (tier is fixed for a driver type),
/// so the invoker's per-call options accessor is allocation-free on the hot path.</para>
/// </remarks>
public sealed class DriverCapabilityInvokerFactory : IDriverCapabilityInvokerFactory
{
private readonly DriverResiliencePipelineBuilder _builder;
private readonly DriverResilienceStatusTracker _statusTracker;
private readonly Func<string, DriverTier> _tierResolver;
/// <summary>Construct the factory over the shared resilience infrastructure.</summary>
/// <param name="builder">Process-singleton Polly pipeline builder (shared pipeline cache).</param>
/// <param name="statusTracker">Process-singleton resilience status tracker feeding Admin <c>/hosts</c>.</param>
/// <param name="tierResolver">Resolves a driver type name to its <see cref="DriverTier"/> (e.g. <c>DriverFactoryRegistry.GetTier</c>).</param>
public DriverCapabilityInvokerFactory(
DriverResiliencePipelineBuilder builder,
DriverResilienceStatusTracker statusTracker,
Func<string, DriverTier> tierResolver)
{
ArgumentNullException.ThrowIfNull(builder);
ArgumentNullException.ThrowIfNull(statusTracker);
ArgumentNullException.ThrowIfNull(tierResolver);
_builder = builder;
_statusTracker = statusTracker;
_tierResolver = tierResolver;
}
/// <inheritdoc />
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);
}
}
@@ -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<PipelineKey, ResiliencePipeline> _pipelines = new();
private readonly TimeProvider _timeProvider;
private readonly DriverResilienceStatusTracker? _statusTracker;
private readonly ILogger? _logger;
/// <summary>Construct with the ambient clock (use <see cref="TimeProvider.System"/> in prod).</summary>
/// <param name="timeProvider">Clock source for pipeline timeouts + breaker sampling. Defaults to system.</param>
@@ -33,12 +35,19 @@ public sealed class DriverResiliencePipelineBuilder
/// <c>LastBreakerOpenUtc</c>, breaker-close resets failures. Feeds Admin <c>/hosts</c> +
/// the Polly bulkhead-depth column. Absent tracker means no telemetry (unit tests +
/// deployments that don't care about resilience observability).</param>
/// <param name="logger">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 <c>/hosts</c> reader
/// (Phase 6.1 Stream E.2/E.3) ships — without it the pipeline runs silently. Absent logger =
/// no resilience logging (unit tests).</param>
public DriverResiliencePipelineBuilder(
TimeProvider? timeProvider = null,
DriverResilienceStatusTracker? statusTracker = null)
DriverResilienceStatusTracker? statusTracker = null,
ILogger? logger = null)
{
_timeProvider = timeProvider ?? TimeProvider.System;
_statusTracker = statusTracker;
_logger = logger;
}
/// <summary>
@@ -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));
}
/// <summary>Drop cached pipelines for one driver instance (e.g. on ResilienceConfig change). Test + Admin-reload use.</summary>
@@ -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<Exception>(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<Exception>(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;
};
}
@@ -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<IDriverFactory>(sp =>
new DriverFactoryRegistryAdapter(sp.GetRequiredService<DriverFactoryRegistry>()));
// 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<DriverResilienceStatusTracker>();
services.AddSingleton<DriverResiliencePipelineBuilder>(sp =>
new DriverResiliencePipelineBuilder(
timeProvider: null, // TimeProvider.System
statusTracker: sp.GetRequiredService<DriverResilienceStatusTracker>(),
logger: sp.GetService<ILoggerFactory>()?.CreateLogger("ZB.MOM.WW.OtOpcUa.Core.Resilience.DriverResiliencePipelineBuilder")));
services.AddSingleton<IDriverCapabilityInvokerFactory>(sp =>
{
var registry = sp.GetRequiredService<DriverFactoryRegistry>();
return new DriverCapabilityInvokerFactory(
sp.GetRequiredService<DriverResiliencePipelineBuilder>(),
sp.GetRequiredService<DriverResilienceStatusTracker>(),
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();
@@ -61,6 +61,12 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers
private readonly CommonsNodeId _localNode;
private readonly IActorRef? _coordinatorOverride;
private readonly IDriverFactory _driverFactory;
/// <summary>Builds the per-driver-instance <see cref="IDriverCapabilityInvoker"/> attached to each
/// spawned <see cref="DriverInstanceActor"/>. Resolved from DI like <see cref="_driverFactory"/>;
/// defaults to <see cref="NullDriverCapabilityInvokerFactory"/> (pass-through) when the fused Host
/// hasn't bound the real resilience factory.</summary>
private readonly IDriverCapabilityInvokerFactory _invokerFactory;
private readonly IReadOnlySet<string> _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 <see cref="ScriptedAlarmHostActor"/> child, so
/// tests can intercept the <see cref="ScriptedAlarmHostActor.ApplyScriptedAlarms"/> message. Null
/// in production (the real host is spawned).</param>
/// <param name="invokerFactory">Optional Phase 6.1 resilience-invoker factory used to attach an
/// <see cref="IDriverCapabilityInvoker"/> to each spawned driver; defaults to
/// <see cref="NullDriverCapabilityInvokerFactory"/> (pass-through) when not supplied.</param>
/// <returns>The Akka.NET <see cref="Akka.Actor.Props"/> used to spawn this actor.</returns>
public static Props Props(
IDbContextFactory<OtOpcUaConfigDbContext> 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));
/// <summary>Initializes a new DriverHostActor with the specified dependencies.</summary>
/// <param name="dbFactory">Database context factory for configuration database access.</param>
@@ -323,6 +333,9 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers
/// host; when null the host is left unspawned.</param>
/// <param name="scriptedAlarmHostOverride">Test seam: when supplied, used as the ScriptedAlarm host
/// instead of spawning a real <see cref="ScriptedAlarmHostActor"/> child.</param>
/// <param name="invokerFactory">Phase 6.1 resilience-invoker factory used to attach an
/// <see cref="IDriverCapabilityInvoker"/> to each spawned driver; defaults to
/// <see cref="NullDriverCapabilityInvokerFactory"/> (pass-through) when null.</param>
public DriverHostActor(
IDbContextFactory<OtOpcUaConfigDbContext> 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<string>(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));
}
@@ -137,6 +137,15 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers
public static readonly TimeSpan HealthPollInterval = TimeSpan.FromSeconds(30);
private readonly IDriver _driver;
/// <summary>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
/// <see cref="NullDriverCapabilityInvoker"/> (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 <see cref="IDriverCapabilityInvokerFactory"/> and injects it at
/// spawn. Consumed through the <see cref="IDriverCapabilityInvoker"/> abstraction because Runtime
/// is deliberately Polly-free (it does not reference <c>ZB.MOM.WW.OtOpcUa.Core</c>).</summary>
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
/// <param name="rediscoverInterval">Optional interval between post-connect re-discovery passes; defaults to 2 seconds.</param>
/// <param name="rediscoverMaxAttempts">Optional cap on re-discovery passes; defaults to 15.</param>
/// <param name="rediscoverDiscoverTimeout">Optional per-pass timeout for <see cref="ITagDiscovery.DiscoverAsync"/>; defaults to 30 seconds.</param>
/// <param name="invoker">Optional Phase 6.1 resilience invoker wrapping this driver's capability
/// calls; defaults to <see cref="NullDriverCapabilityInvoker"/> (pass-through) when not supplied.</param>
/// <returns>A <see cref="Akka.Actor.Props"/> instance configured to create the wrapped <see cref="DriverInstanceActor"/>.</returns>
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));
/// <summary>
/// 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
/// <param name="rediscoverInterval">Interval between post-connect re-discovery passes; defaults to 2 seconds.</param>
/// <param name="rediscoverMaxAttempts">Cap on the number of re-discovery passes; defaults to 15.</param>
/// <param name="rediscoverDiscoverTimeout">Per-pass timeout for <see cref="ITagDiscovery.DiscoverAsync"/>; defaults to 30 seconds.</param>
/// <param name="invoker">Phase 6.1 resilience invoker wrapping this driver's capability calls;
/// defaults to <see cref="NullDriverCapabilityInvoker"/> (pass-through) when null.</param>
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;
/// <summary>Resolve the resilience-pipeline host key for a single driver-side reference. Multi-host
/// drivers (e.g. Modbus across N PLCs) implement <see cref="IPerCallHostResolver"/> 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.</summary>
private string ResolveHost(string reference) =>
_driver is IPerCallHostResolver resolver ? resolver.ResolveHost(reference) : _driverInstanceId;
/// <summary>Records a transition into a Faulted / error state for the 5-minute sliding counter.</summary>
private void RecordFault()
{
@@ -213,6 +213,11 @@ public static class ServiceCollectionExtensions
// Fallback to Null* if AddOtOpcUaRuntime wasn't called (e.g., test harnesses).
var historianSink = resolver.GetService<IAlarmHistorianSink>() ?? NullAlarmHistorianSink.Instance;
var driverFactory = resolver.GetService<IDriverFactory>() ?? 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<IDriverCapabilityInvokerFactory>()
?? NullDriverCapabilityInvokerFactory.Instance;
var addressSpaceSink = resolver.GetService<IOpcUaAddressSpaceSink>() ?? NullOpcUaAddressSpaceSink.Instance;
var serviceLevel = resolver.GetService<IServiceLevelPublisher>() ?? NullServiceLevelPublisher.Instance;
var loggerFactory = resolver.GetService<ILoggerFactory>() ?? 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<DriverHostActorKey>(driverHost);
@@ -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(