Files
lmxopcua/src/Server/ZB.MOM.WW.OtOpcUa.Host/Drivers/DriverFactoryBootstrap.cs
T
Joseph Doherty 97dc715dac
v2-ci / unit-tests (push) Has been cancelled
fix(runtime): derive every write bound and Ask from one write timeout (lmxopcua#777)
Three write bounds were configured independently: the inbound-write
gateway's 10 s Ask, DriverHostActor's fixed 8 s Ask of the child, and the
child's own write timeout (5 s + up to 2 s grace) with a fixed 8 s queue
budget. An outer Ask that fires first reports the write failed and the node
reverts while the child is still driving the device, so any write timeout
above ~6 s (or a write dequeued late in its 8 s budget) let the device take
a value the client was told failed.

DriverWriteBounds (Runtime) derives them all from DriverWrites:Timeout:
  token cancel   = Timeout
  abandon        = Timeout + min(Timeout/2, 2 s)            (RunBound)
  queue budget   = RunBound
  child replies  <= queue budget + RunBound                  (ReplyBound)
  host Ask       = ReplyBound + 1 s
  gateway Ask    = host Ask + 1 s
Default Timeout is 10 s (the Tier C Write timeout, so no tier default is
capped below its own policy); host Ask 25 s, gateway Ask 26 s. Tier A/B
writes are still cancelled first by their 2 s / 4 s tier timeouts.

- DriverHostActor takes writeBounds + writeClock, spawns children with them,
  and runs its Ask deadline as a token on the write clock (Infinite Akka
  timeout), so host and child bounds share one clock.
- WriteAttribute gains IssuedAt (write-clock timestamp, stamped by the
  host): the queue budget counts mailbox time too, and a write past it is
  failed BadTimeout and never started - including on arrival. Stale queued
  writes are expired across the whole queue, not only up to a free slot.
- ActorNodeWriteGateway's default Ask = Default.GatewayAsk; the hosted
  service passes the configured one.
- Startup validation (Host DriverWriteBoundsBinding, same warn-and-clamp
  style as ScriptingOptionsBinding): unparsable / non-positive -> default,
  above 5 min -> clamped, and each tier whose default Write timeout exceeds
  the write timeout is named in a Warning (its policy would never take
  effect). The derived bounds are logged at Information. Per-instance
  ResilienceConfig Write overrides arrive with a deployment, so
  DriverCapabilityInvokerFactory warns at spawn (writeTimeoutCeiling).
- appsettings.json documents DriverWrites:Timeout.

Tests (manual clock):
- DriverHostActorWriteAskTests: a held write under a 9 s timeout survives
  an 8.5 s step and succeeds; a token-ignoring write is answered by the
  child's BadTimeout at 9 s + grace, not by the Ask; the Infinite-timeout
  Ask ends exactly at its clock deadline.
- DriverInstanceActorConcurrentWriteTests: a write issued longer ago than
  the queue budget is failed and never reaches the driver; positive control
  at exactly the budget is written.
- DriverWriteBoundsTests: every outer bound covers the inner ones for 0.5 s
  to 300 s timeouts.
- DriverWriteBoundsBindingTests: default fits every tier; 3 s timeout is
  caught at startup with Tier B and C named (not A); bad values fall back
  with a warning; over-max clamps.
- DriverCapabilityInvokerFactoryTests: spawn warns when a Write override
  exceeds the ceiling; no warning within it or without a ceiling.
Mutation checks: a fixed 8 s host Ask on the same clock fails both 9 s host
tests; stamping arrival time instead of IssuedAt fails the issued-at test;
dropping the tier check / factory check fails their tests.

Runtime.Tests 961 passed / 3 skipped at AKKA_TEST_TIMEFACTOR 3 and 1;
Host.Tests 232/232 at 3 and 1; Core.Tests 361 passed / 1 skipped.
Solution Release build: 0 warnings.

ContinuousHistorizationRecorderTypedValueTests (historian) deliberately
left for the historian lane.

Claude-Session: https://claude.ai/code/session_01GT67AEHd2Acv1YPwGfXEoV
2026-11-09 15:46:30 -05:00

241 lines
16 KiB
C#

using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.DependencyInjection.Extensions;
using Microsoft.Extensions.Logging;
using Microsoft.Extensions.Options;
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
using ZB.MOM.WW.OtOpcUa.Core.Hosting;
using ZB.MOM.WW.OtOpcUa.Core.Resilience;
using ZB.MOM.WW.Secrets.Abstractions;
using ZB.MOM.WW.OtOpcUa.Host.Configuration;
using ZB.MOM.WW.OtOpcUa.Runtime.Redundancy;
namespace ZB.MOM.WW.OtOpcUa.Host.Drivers;
// Probe type aliases — keep the using list concise.
using ModbusProbe = Driver.Modbus.ModbusDriverProbe;
using AbCipProbe = Driver.AbCip.AbCipDriverProbe;
using AbLegacyProbe = Driver.AbLegacy.AbLegacyDriverProbe;
using S7Probe = Driver.S7.S7DriverProbe;
using TwinCATProbe = Driver.TwinCAT.TwinCATDriverProbe;
using FocasProbe = Driver.FOCAS.FocasDriverProbe;
using OpcUaProbe = Driver.OpcUaClient.OpcUaClientDriverProbe;
using GalaxyProbe = Driver.Galaxy.GalaxyDriverProbe;
using CalculationProbe = Driver.Calculation.CalculationDriverProbe;
using SqlProbe = Driver.Sql.SqlDriverProbe;
using MqttProbe = Driver.Mqtt.MqttDriverProbe;
using MTConnectProbe = Driver.MTConnect.MTConnectDriverProbe;
/// <summary>
/// Wires every cross-platform driver assembly's <c>Register(registry, loggerFactory)</c>
/// extension into a single <see cref="DriverFactoryRegistry"/> singleton and binds the
/// v2 <see cref="IDriverFactory"/> abstraction to a <see cref="DriverFactoryRegistryAdapter"/>
/// over it. Replaces the F7 seam's <c>NullDriverFactory</c> default so deploys actually
/// materialise real <see cref="IDriver"/> instances on driver-role nodes.
///
/// The factory registry is skipped on admin-only nodes — they never run drivers, so it doesn't
/// need to exist (Program.cs guards via the <c>hasDriver</c> flag). The driver <em>probe</em>
/// set is the exception: it backs the AdminUI Test Connect button and so must also be wired on
/// admin nodes — see <see cref="AddOtOpcUaDriverProbes"/>.
/// </summary>
public static class DriverFactoryBootstrap
{
/// <summary>
/// Register the cross-platform driver factories + bind <see cref="IDriverFactory"/>.
/// Must be called BEFORE <c>services.AddAkka</c> so the runtime extension can resolve
/// <see cref="IDriverFactory"/> from DI when spawning <c>DriverHostActor</c>.
/// </summary>
/// <param name="services">The service collection to register driver factories with.</param>
/// <returns>The same service collection, for chaining.</returns>
public static IServiceCollection AddOtOpcUaDriverFactories(this IServiceCollection services)
{
// The Primary data-plane verdict the MQTT driver's Sparkplug host claim is gated on.
//
// Registered HERE as well as in AddOtOpcUaRuntime, and both sides TryAdd the same concrete
// RedundancyRoleView, so whichever extension runs first creates it and the other resolves the
// SAME instance. That is deliberately not a defensive duplicate: the resolve below is
// GetRequiredService (a null would silently fall the driver back to "claim unconditionally",
// which on a redundant pair is the dual-claim this mechanism exists to prevent), so without
// this line the registration becomes order-dependent on two extensions a caller is free to
// invoke in either order — and any graph that wires driver factories without the full runtime
// fails at resolution. Two Host integration suites do exactly that, and caught it.
services.TryAddSingleton<RedundancyRoleView>();
services.TryAddSingleton<IDriverPrimaryRoleSource>(sp => sp.GetRequiredService<RedundancyRoleView>());
services.AddSingleton<DriverFactoryRegistry>(sp =>
{
var registry = new DriverFactoryRegistry();
var loggerFactory = sp.GetService<ILoggerFactory>();
// The calc driver needs the root script logger so a script failure fans out onto the
// script-logs topic; resolve it here (null on nodes without the script pipeline wired).
var scriptRoot = sp.GetService<ZB.MOM.WW.OtOpcUa.Core.Scripting.ScriptRootLogger>();
// The node-wide Scripting section (bound + clamped in Program.cs); null in test graphs ⇒ defaults.
var scripting = sp.GetService<ZB.MOM.WW.OtOpcUa.Core.Scripting.ScriptingOptions>();
// The Galaxy driver resolves its Gateway.ApiKeySecretRef secret: arm through the
// shared ISecretResolver (registered unconditionally by AddZbSecrets on every node).
var secretResolver = sp.GetRequiredService<ISecretResolver>();
// The MQTT driver's Sparkplug primary-host claim is Primary-gated, and unlike
// every other gated operation it is a STANDING assertion on the driver's own connection —
// there is no inbound message for DriverHostActor to intercept — so the verdict has to
// reach the driver itself. Registered unconditionally on driver-role nodes by
// AddOtOpcUaRuntime. GetRequiredService, not GetService: a null here silently falls the
// driver back to "claim unconditionally", which on a redundant pair is the dual-claim this
// whole mechanism exists to prevent, and it would fail as a plant outage rather than as a
// startup error.
var primaryRoleSource = sp.GetRequiredService<IDriverPrimaryRoleSource>();
// Final review I2: the native AB transport's process-wide pool logs options conflicts and close
// failures; without this it writes to NullLogger in production.
ConfigureAbTransportLogging(sp);
Register(registry, loggerFactory, secretResolver, primaryRoleSource, ResolveAbCipDefaultTransport(sp),
scriptRoot, scripting);
return registry;
});
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,
logger: sp.GetService<ILoggerFactory>()?.CreateLogger("ZB.MOM.WW.OtOpcUa.Core.Resilience.DriverCapabilityInvokerFactory"),
// lmxopcua#777: warn at spawn when a per-instance ResilienceConfig Write timeout exceeds the
// node's write timeout (it would never take effect). Unregistered ⇒ the default bounds.
writeTimeoutCeiling: (sp.GetService<ZB.MOM.WW.OtOpcUa.Runtime.Drivers.DriverWriteBounds>()
?? ZB.MOM.WW.OtOpcUa.Runtime.Drivers.DriverWriteBounds.Default).Timeout);
});
// The tracker's operator-facing reader: a periodic DPS publisher that mirrors each snapshot to the
// AdminUI resilience panel (closes the "tracker fed by everything, read by nothing" gap).
// Lives on driver nodes exactly where the tracker exists. The lazy Func<ActorSystem> is idempotent
// with the one Program.cs registers for the alarm-command router.
services.TryAddSingleton<Func<Akka.Actor.ActorSystem>>(sp => () => sp.GetRequiredService<Akka.Actor.ActorSystem>());
services.AddHostedService<DriverResilienceStatusPublisherService>();
// 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();
return services;
}
/// <summary>
/// Register one <see cref="IDriverProbe"/> per driver type. These back the AdminUI's
/// "Test Connect" button: the <c>admin-operations</c> cluster singleton resolves
/// <see cref="IEnumerable{T}"/> of <see cref="IDriverProbe"/> and dispatches by DriverType.
/// <para>
/// That singleton is role-pinned to <c>admin</c>, so this MUST be wired on admin nodes —
/// including admin-only nodes that lack the <c>driver</c> role (e.g. the MAIN cluster's
/// admin-a/admin-b). Probes are lightweight (cheap connect, no persistent state) and don't
/// need the driver-factory registry, so they register independently of
/// <see cref="AddOtOpcUaDriverFactories"/>.
/// </para>
/// <para>
/// Uses <c>TryAddEnumerable</c> so a fused admin,driver node — which reaches this from both
/// the driver-factory path and the admin path — registers each probe exactly once. A
/// duplicate would make the singleton's <c>ToDictionary(p =&gt; p.DriverType)</c> throw.
/// </para>
/// </summary>
/// <param name="services">The service collection to register driver probes with.</param>
/// <returns>The same service collection, for chaining.</returns>
public static IServiceCollection AddOtOpcUaDriverProbes(this IServiceCollection services)
{
services.TryAddEnumerable(ServiceDescriptor.Singleton<IDriverProbe, ModbusProbe>());
// Factory-built so the probe resolves an absent "transport" key to the SAME node default the driver
// factory uses (Drivers:AbCip:DefaultTransport, final review I1) — Test Connect probes the stack the
// driver would run. The generic factory overload keeps AbCipProbe as the implementation type, which
// TryAddEnumerable de-dups on.
services.TryAddEnumerable(ServiceDescriptor.Singleton<IDriverProbe, AbCipProbe>(
sp =>
{
ConfigureAbTransportLogging(sp); // admin-only nodes reach the pool through the probes alone
return new AbCipProbe(tagFactory: null, defaultTransport: ResolveAbCipDefaultTransport(sp));
}));
services.TryAddEnumerable(ServiceDescriptor.Singleton<IDriverProbe, AbLegacyProbe>(sp =>
{
ConfigureAbTransportLogging(sp);
return new AbLegacyProbe();
}));
services.TryAddEnumerable(ServiceDescriptor.Singleton<IDriverProbe, S7Probe>());
services.TryAddEnumerable(ServiceDescriptor.Singleton<IDriverProbe, TwinCATProbe>());
services.TryAddEnumerable(ServiceDescriptor.Singleton<IDriverProbe, FocasProbe>());
services.TryAddEnumerable(ServiceDescriptor.Singleton<IDriverProbe, OpcUaProbe>());
services.TryAddEnumerable(ServiceDescriptor.Singleton<IDriverProbe, GalaxyProbe>());
services.TryAddEnumerable(ServiceDescriptor.Singleton<IDriverProbe, CalculationProbe>());
services.TryAddEnumerable(ServiceDescriptor.Singleton<IDriverProbe, SqlProbe>());
services.TryAddEnumerable(ServiceDescriptor.Singleton<IDriverProbe, MqttProbe>());
services.TryAddEnumerable(ServiceDescriptor.Singleton<IDriverProbe, MTConnectProbe>());
return services;
}
/// <summary>
/// Gives the native AB transport's process-wide <c>DeviceSessionPool.Shared</c> the host logger factory
/// (final review I2). Called from every DI factory that can put the pool to use on this node (the driver
/// registry, and the AbCip/AbLegacy probes for admin-only nodes); idempotent. A container without an
/// <see cref="ILoggerFactory"/> leaves the pool on its null logger.
/// </summary>
/// <param name="sp">The service provider.</param>
internal static void ConfigureAbTransportLogging(IServiceProvider sp)
{
if (sp.GetService<ILoggerFactory>() is { } loggerFactory)
Driver.Ab.Transport.DeviceSessionPool.ConfigureSharedLogger(loggerFactory);
}
/// <summary>
/// The node-level AbCip default transport (<c>Drivers:AbCip:DefaultTransport</c>, bound + validated in
/// Program.cs). A container without the options bound (test graphs) gets the compiled default.
/// </summary>
/// <param name="sp">The service provider.</param>
/// <returns>The transport an AbCip driver with no <c>transport</c> key runs on.</returns>
internal static Driver.AbCip.AbCipTransport ResolveAbCipDefaultTransport(IServiceProvider sp) =>
sp.GetService<IOptions<AbCipDriverDefaultsOptions>>()?.Value.ResolveDefaultTransport()
?? Driver.AbCip.AbCipTransportResolver.Default;
/// <summary>
/// Invoke every cross-platform driver's <c>Register</c> extension. New driver assemblies
/// get added here — one line per type. ShouldStub() in <c>DriverInstanceActor</c> still
/// handles platform/role-dependent stubbing (e.g. Galaxy on macOS), so registering a
/// factory here doesn't mean it always runs in production.
/// </summary>
private static void Register(
DriverFactoryRegistry registry,
ILoggerFactory? loggerFactory,
ISecretResolver secretResolver,
IDriverPrimaryRoleSource primaryRoleSource,
Driver.AbCip.AbCipTransport abCipDefaultTransport,
ZB.MOM.WW.OtOpcUa.Core.Scripting.ScriptRootLogger? scriptRoot = null,
ZB.MOM.WW.OtOpcUa.Core.Scripting.ScriptingOptions? scripting = null)
{
Driver.AbCip.AbCipDriverFactoryExtensions.Register(registry, loggerFactory, abCipDefaultTransport);
Driver.AbLegacy.AbLegacyDriverFactoryExtensions.Register(registry, loggerFactory);
Driver.Calculation.CalculationDriverFactoryExtensions.Register(registry, loggerFactory, scriptRoot, scripting);
Driver.FOCAS.FocasDriverFactoryExtensions.Register(registry, loggerFactory);
Driver.Galaxy.GalaxyDriverFactoryExtensions.Register(registry, secretResolver, loggerFactory);
Driver.Modbus.ModbusDriverFactoryExtensions.Register(registry, loggerFactory);
Driver.Mqtt.MqttDriverFactoryExtensions.Register(registry, loggerFactory, primaryRoleSource);
// Tier A (the Register default): fully managed — HttpClient + System.Xml.Linq, no native SDK,
// no COM. Tier C is the only tier that arms process-level recycle (MemoryRecycle hard-breach /
// ScheduledRecycleScheduler), which would be wrong here: the driver is in-process and killing
// it kills every OPC UA session. The long-lived /sample stream does not argue for a slower
// tier either — SubscribeAsync returns synchronously after starting the pump on a background
// task, so the Tier A 5s Subscribe budget never covers the stream's lifetime.
Driver.MTConnect.MTConnectDriverFactoryExtensions.Register(registry, loggerFactory);
Driver.OpcUaClient.OpcUaClientDriverFactoryExtensions.Register(registry, loggerFactory, secretResolver);
Driver.S7.S7DriverFactoryExtensions.Register(registry, loggerFactory);
Driver.Sql.SqlDriverFactoryExtensions.Register(registry, loggerFactory);
Driver.TwinCAT.TwinCATDriverFactoryExtensions.Register(registry, loggerFactory);
}
}