f9f1b8fcee
Gate record: docs/plans/2026-07-20-localdb-phase2-live-gate.md. Checks 1, 2, 5, 6 pass. Checks 3 and 4 are NOT satisfied — the defect they were meant to confirm turned out to be the opposite of what the plan assumed. Three defects crash-looped every driver node before check 1 could even run: 1. An empty ServerHistorian:ApiKey kills the host. ServerHistorianOptions- Validator exists to turn exactly that class of failure into a named OptionsValidationException, but its documented fail tier explicitly excluded ApiKey on the reasoning that a keyless client "degrades — the gateway rejects calls". It does not: the client validates its own options at construction, so the process dies during Akka startup and never makes a call. 2. UseTls disagreeing with the endpoint scheme kills the host too, in both directions (both messages confirmed in the shipped client assembly). Moving an endpoint from https to http without clearing UseTls is an ordinary migration slip. 3. Plaintext h2c was UNREACHABLE. HistorianGatewayClientAdapter forwarded the TLS-only options unconditionally, and AllowUntrustedServerCertificate defaults to false, so it always sent RequireCertificateValidation=true — which the client rejects outright when UseTls=false. Every http:// deployment crashed, though the scheme is documented as the supported way to select h2c, and the only workaround was to assert a certificate posture for a connection that has no certificate. The fourth was the blocker, and it is Phase 2's own: 4. The drain gate deferred to a Primary that cannot deliver. Redundancy roles are elected CLUSTER-WIDE; the alarm queue is PAIR-LOCAL. On the rig the elected driver Primary is central-1 — it carries the driver Akka role, replicates nobody's LocalDb and does not even run the alarm historian — so every driver node logged "Historian drain suspended", including the two site-b nodes that have no peer at all. Nothing drained anywhere, where before Phase 2 it drained fine. The cost is not a duplicate; it is the buffer growing to the capacity wall and evicting the audit trail it exists to protect. Fixed in three layers: a separate ShouldDrainAlarmHistory policy (unknown role drains; the two gates now deliberately disagree, and a test pins that); peer- host matching in DriverHostActor so a node stands down only for a Primary holding its rows; and AddAlarmHistorian short-circuiting the gate when replication is unconfigured — testing BOTH Replication:PeerAddress and SyncListenPort, since only the dialing half sets the former while both halves share the queue. Every one of these follows from the asymmetry: a false allow costs a duplicate row, which at-least-once delivery already accepts and payload-hash ids collapse; a false deny loses data silently. A third vacuous test, caught by the same delete-the-guard discipline: the role-view tests stayed green with the guard removed, because AwaitAssert polls until an assertion passes and the assertion was "reads open" — which is the SEEDED value, satisfied at the first poll before the actor processed anything. They now assert the sequence of published values through a recording view; the control then goes red for exactly the cases that matter. Migration evidence: 11 legacy rows across two deliberately overlapping files converged to exactly 9 identical rows on both nodes, proving D-6's payload-hash identity on real nodes rather than in a fixture. Open design fork, recorded in the gate doc rather than decided here: a pair cannot currently identify its own Primary, so both halves drain. Safe in every topology — nothing loses data — but the gate's de-duplication benefit is unrealised until roles are scoped per pair. Claude-Session: https://claude.ai/code/session_01GASWkNEi68FSCtvr6rLoEW
440 lines
28 KiB
C#
440 lines
28 KiB
C#
using Akka.Actor;
|
|
using Akka.Hosting;
|
|
using Microsoft.EntityFrameworkCore;
|
|
using Microsoft.Extensions.Configuration;
|
|
using Microsoft.Extensions.DependencyInjection;
|
|
using Microsoft.Extensions.DependencyInjection.Extensions;
|
|
using Microsoft.Extensions.Logging;
|
|
using Microsoft.Extensions.Logging.Abstractions;
|
|
using ZB.MOM.WW.OtOpcUa.Commons.Engines;
|
|
using ZB.MOM.WW.OtOpcUa.Commons.Interfaces;
|
|
using ZB.MOM.WW.OtOpcUa.Commons.OpcUa;
|
|
using ZB.MOM.WW.OtOpcUa.Configuration;
|
|
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
|
|
using ZB.MOM.WW.OtOpcUa.Core.AlarmHistorian;
|
|
using ZB.MOM.WW.OtOpcUa.Core.Scripting;
|
|
using ZB.MOM.WW.OtOpcUa.Core.VirtualTags;
|
|
using ZB.MOM.WW.OtOpcUa.OpcUaServer;
|
|
using ZB.MOM.WW.OtOpcUa.Runtime.DeploymentCache;
|
|
using ZB.MOM.WW.OtOpcUa.Runtime.Drivers;
|
|
using ZB.MOM.WW.OtOpcUa.Runtime.Health;
|
|
using ZB.MOM.WW.OtOpcUa.Runtime.Historian;
|
|
using ZB.MOM.WW.OtOpcUa.Runtime.OpcUa;
|
|
using ZB.MOM.WW.OtOpcUa.Runtime.Redundancy;
|
|
using ZB.MOM.WW.OtOpcUa.Runtime.VirtualTags;
|
|
using ZB.MOM.WW.LocalDb;
|
|
|
|
namespace ZB.MOM.WW.OtOpcUa.Runtime;
|
|
|
|
public static class ServiceCollectionExtensions
|
|
{
|
|
public const string DriverRole = "driver";
|
|
|
|
public const string DriverHostActorName = "driver-host";
|
|
public const string DbHealthProbeActorName = "db-health";
|
|
public const string HistorianAdapterActorName = "historian-adapter";
|
|
public const string DependencyMuxActorName = "dependency-mux";
|
|
public const string OpcUaPublishActorName = "opcua-publish";
|
|
public const string PeerProbeSupervisorName = "peer-probe-supervisor";
|
|
public const string ContinuousHistorizationRecorderActorName = "continuous-historization-recorder";
|
|
|
|
/// <summary>
|
|
/// Registers shared runtime services. Currently binds <see cref="IAlarmHistorianSink"/>
|
|
/// to <see cref="NullAlarmHistorianSink"/> as the default; production deployments
|
|
/// override this with <c>LocalDbStoreAndForwardSink</c> wrapping the HistorianGateway alarm writer.
|
|
/// Call this BEFORE <c>AddAkka</c>.
|
|
/// </summary>
|
|
/// <param name="services">The service collection to register with.</param>
|
|
/// <returns>The same <paramref name="services"/> instance for chaining.</returns>
|
|
public static IServiceCollection AddOtOpcUaRuntime(this IServiceCollection services)
|
|
{
|
|
services.TryAddSingleton<IAlarmHistorianSink>(NullAlarmHistorianSink.Instance);
|
|
services.TryAddSingleton<IHistorianDataSource>(NullHistorianDataSource.Instance);
|
|
// Historian tag provisioning. Null default (no-op) so the AddressSpaceApplier resolves a real
|
|
// IHistorianProvisioning only when AddHistorianProvisioning registers the gateway-backed one
|
|
// (ServerHistorian:Enabled). TryAddSingleton so the gateway registration wins last.
|
|
services.TryAddSingleton<IHistorianProvisioning>(NullHistorianProvisioning.Instance);
|
|
// VirtualTag historization sink. Null default — the durable AVEVA sink is infra-gated (there is
|
|
// no live-data historian write RPC). TryAddSingleton so a deployment that bound a real
|
|
// IHistoryWriter earlier wins.
|
|
services.TryAddSingleton<IHistoryWriter>(NullHistoryWriter.Instance);
|
|
services.TryAddSingleton<IDriverFactory>(NullDriverFactory.Instance);
|
|
services.TryAddSingleton<IOpcUaAddressSpaceSink>(NullOpcUaAddressSpaceSink.Instance);
|
|
services.TryAddSingleton<IServiceLevelPublisher>(NullServiceLevelPublisher.Instance);
|
|
services.TryAddSingleton<IDriverHealthPublisher, AkkaDriverHealthPublisher>();
|
|
return services;
|
|
}
|
|
|
|
/// <summary>
|
|
/// Config-gated durable alarm-historian sink. When the <c>AlarmHistorian</c> section has
|
|
/// <c>Enabled=true</c>, registers a <see cref="LocalDbStoreAndForwardSink"/> (draining via the
|
|
/// <paramref name="writerFactory"/>-supplied writer) as the <see cref="IAlarmHistorianSink"/>,
|
|
/// overriding the <see cref="NullAlarmHistorianSink"/> default. Otherwise a no-op (Null stays).
|
|
/// The writer is injected so the durable downstream (the HistorianGateway alarm writer) can be
|
|
/// supplied by the Host, which is the only project that references it.
|
|
/// </summary>
|
|
/// <remarks>
|
|
/// <para>
|
|
/// The queue lives in the node's consolidated <see cref="ILocalDb"/>, which replicates it
|
|
/// to the redundant pair peer — so undelivered alarm history survives losing the node
|
|
/// holding it. That is also why the drain is gated on <see cref="IRedundancyRoleView"/>:
|
|
/// the Secondary holds a full replica of the Primary's queue, and an ungated drain there
|
|
/// would re-deliver every event. Both services are resolved from the provider, so a
|
|
/// deployment that enables this section without registering LocalDb fails loudly at
|
|
/// resolution rather than silently buffering to nowhere.
|
|
/// </para>
|
|
/// </remarks>
|
|
/// <param name="services">The service collection to register with.</param>
|
|
/// <param name="configuration">The configuration carrying the <c>AlarmHistorian</c> section.</param>
|
|
/// <param name="writerFactory">
|
|
/// Factory the Host supplies to build the concrete <see cref="IAlarmHistorianWriter"/>
|
|
/// (the HistorianGateway alarm writer) from the bound options + the resolving provider.
|
|
/// </param>
|
|
/// <returns>The same <paramref name="services"/> instance for chaining.</returns>
|
|
public static IServiceCollection AddAlarmHistorian(
|
|
this IServiceCollection services,
|
|
IConfiguration configuration,
|
|
Func<AlarmHistorianOptions, IServiceProvider, IAlarmHistorianWriter> writerFactory)
|
|
{
|
|
var opts = configuration.GetSection(AlarmHistorianOptions.SectionName).Get<AlarmHistorianOptions>();
|
|
if (opts is not { Enabled: true }) return services; // leave the Null default from AddOtOpcUaRuntime
|
|
|
|
foreach (var warning in opts.Validate())
|
|
Serilog.Log.Logger.ForContext<LocalDbStoreAndForwardSink>().Warning("Historian config: {HistorianConfigWarning}", warning);
|
|
|
|
// The view is a plain singleton so it exists even on nodes whose DriverHostActor never
|
|
// publishes to it — an unpublished view reads as "no peer, drain", which is the correct
|
|
// posture for a deployment that runs no redundancy at all.
|
|
services.TryAddSingleton<IRedundancyRoleView, RedundancyRoleView>();
|
|
|
|
// THE GATE ONLY APPLIES TO A REPLICATED QUEUE. Standing down is only ever safe because some
|
|
// other node holds the same rows and will send them instead — and the only node that does is
|
|
// this node's LocalDb replication peer. Without replication configured, these rows exist here
|
|
// and nowhere else, so deferring to anyone means they are never delivered by anyone.
|
|
//
|
|
// This is not hypothetical. The redundancy role is a CLUSTER-WIDE election
|
|
// (RedundancyStateActor keys on Akka's RoleLeader("driver")), while the queue is PAIR-LOCAL.
|
|
// On the docker-dev rig the elected driver Primary is a central node — which carries the
|
|
// driver Akka role, replicates nobody's LocalDb, and does not even run the alarm historian —
|
|
// so every site node dutifully suspended its drain in favour of a node that could not
|
|
// possibly deliver its events. Scoping the gate to "is my queue actually shared?" is what
|
|
// keeps the two scopes from disagreeing.
|
|
// BOTH halves of a pair share the queue, but only one of them dials: the initiator sets
|
|
// Replication:PeerAddress, its partner only sets SyncListenPort and waits. Testing the dial
|
|
// side alone would leave the listening half permanently ungated — one drainer per pair by
|
|
// accident rather than by role, and the wrong one whenever the roles swap.
|
|
var replicated =
|
|
!string.IsNullOrWhiteSpace(configuration["LocalDb:Replication:PeerAddress"])
|
|
|| !string.IsNullOrWhiteSpace(configuration["LocalDb:SyncListenPort"]);
|
|
|
|
if (!replicated)
|
|
{
|
|
Serilog.Log.Logger.ForContext<LocalDbStoreAndForwardSink>().Information(
|
|
"Alarm historian: LocalDb replication is not configured, so this node's queue is not "
|
|
+ "shared with any peer and the Primary drain gate does not apply — this node always "
|
|
+ "drains its own alarm queue.");
|
|
}
|
|
|
|
services.AddSingleton<IAlarmHistorianSink>(sp =>
|
|
{
|
|
// LocalDbStoreAndForwardSink takes a Serilog ILogger (not Microsoft.Extensions.Logging).
|
|
// Resolve it off the host's configured static logger so the drain worker's WARN/INFO
|
|
// lines land in the same sinks as the rest of the process.
|
|
var roleView = sp.GetRequiredService<IRedundancyRoleView>();
|
|
var sink = new LocalDbStoreAndForwardSink(
|
|
sp.GetRequiredService<ILocalDb>(),
|
|
writerFactory(opts, sp),
|
|
Serilog.Log.Logger.ForContext<LocalDbStoreAndForwardSink>(),
|
|
batchSize: opts.BatchSize,
|
|
capacity: opts.Capacity,
|
|
deadLetterRetention: TimeSpan.FromDays(opts.DeadLetterRetentionDays),
|
|
maxAttempts: opts.MaxAttempts,
|
|
drainGate: () => !replicated || roleView.ShouldDrainAlarmHistory);
|
|
sink.StartDrainLoop(TimeSpan.FromSeconds(opts.DrainIntervalSeconds));
|
|
return sink;
|
|
});
|
|
return services;
|
|
}
|
|
|
|
/// <summary>
|
|
/// Config-gated server-side HistoryRead backend. When the <c>ServerHistorian</c> section has
|
|
/// <c>Enabled=true</c>, registers the <paramref name="dataSourceFactory"/>-supplied
|
|
/// <see cref="IHistorianDataSource"/> (the read-only HistorianGateway-backed data source) overriding
|
|
/// the <see cref="NullHistorianDataSource"/> default from <see cref="AddOtOpcUaRuntime"/>. Otherwise
|
|
/// a no-op (the Null default stays and the node manager's HistoryRead returns
|
|
/// <c>GoodNoData</c>-empty). The data source is injected so the gateway-backed client can be supplied
|
|
/// by the Host, which is the only project that references the driver.
|
|
/// </summary>
|
|
/// <param name="services">The service collection to register with.</param>
|
|
/// <param name="configuration">The configuration carrying the <c>ServerHistorian</c> section.</param>
|
|
/// <param name="dataSourceFactory">
|
|
/// Factory the Host supplies to build the concrete read <see cref="IHistorianDataSource"/>
|
|
/// (the gateway-backed data source) from the bound options + the resolving provider.
|
|
/// </param>
|
|
/// <returns>The same <paramref name="services"/> instance for chaining.</returns>
|
|
public static IServiceCollection AddServerHistorian(
|
|
this IServiceCollection services,
|
|
IConfiguration configuration,
|
|
Func<ServerHistorianOptions, IServiceProvider, IHistorianDataSource> dataSourceFactory)
|
|
{
|
|
var opts = configuration.GetSection(ServerHistorianOptions.SectionName).Get<ServerHistorianOptions>();
|
|
if (opts is not { Enabled: true }) return services; // leave the Null default from AddOtOpcUaRuntime
|
|
|
|
foreach (var warning in opts.Validate())
|
|
Serilog.Log.Logger.ForContext<IHistorianDataSource>().Warning("ServerHistorian config: {ServerHistorianConfigWarning}", warning);
|
|
|
|
// Last-registration-wins over the TryAddSingleton Null default seeded by AddOtOpcUaRuntime.
|
|
services.AddSingleton<IHistorianDataSource>(sp => dataSourceFactory(opts, sp));
|
|
return services;
|
|
}
|
|
|
|
/// <summary>
|
|
/// Config-gated historian tag provisioning. When the <c>ServerHistorian</c> section has
|
|
/// <c>Enabled=true</c>, registers the <paramref name="provisioningFactory"/>-supplied
|
|
/// <see cref="IHistorianProvisioning"/> (the gateway-backed <c>GatewayTagProvisioner</c> that calls
|
|
/// the gateway's <c>EnsureTags</c>) overriding the <see cref="NullHistorianProvisioning"/> default from
|
|
/// <see cref="AddOtOpcUaRuntime"/>. Otherwise a no-op (the Null default stays and deploying historized
|
|
/// tags provisions nothing). The provisioner is consumed by the <c>AddressSpaceApplier</c>, which fires
|
|
/// a non-blocking <see cref="IHistorianProvisioning.EnsureTagsAsync"/> for added historized tags on
|
|
/// every deploy. Gated on the <b>same</b> <c>ServerHistorian:Enabled</c> flag as the read path
|
|
/// (<see cref="AddServerHistorian"/>) — provisioning targets the same single gateway. The provisioner
|
|
/// is injected so the gateway-backed client can be supplied by the Host, which is the only project that
|
|
/// references the driver.
|
|
/// </summary>
|
|
/// <param name="services">The service collection to register with.</param>
|
|
/// <param name="configuration">The configuration carrying the <c>ServerHistorian</c> section.</param>
|
|
/// <param name="provisioningFactory">
|
|
/// Factory the Host supplies to build the concrete <see cref="IHistorianProvisioning"/>
|
|
/// (the gateway-backed provisioner) from the bound options + the resolving provider.
|
|
/// </param>
|
|
/// <returns>The same <paramref name="services"/> instance for chaining.</returns>
|
|
public static IServiceCollection AddHistorianProvisioning(
|
|
this IServiceCollection services,
|
|
IConfiguration configuration,
|
|
Func<ServerHistorianOptions, IServiceProvider, IHistorianProvisioning> provisioningFactory)
|
|
{
|
|
var opts = configuration.GetSection(ServerHistorianOptions.SectionName).Get<ServerHistorianOptions>();
|
|
if (opts is not { Enabled: true }) return services; // leave the Null default from AddOtOpcUaRuntime
|
|
|
|
// Last-registration-wins over the TryAddSingleton Null default seeded by AddOtOpcUaRuntime.
|
|
services.AddSingleton<IHistorianProvisioning>(sp => provisioningFactory(opts, sp));
|
|
return services;
|
|
}
|
|
|
|
/// <summary>
|
|
/// Spawns the per-node driver-role actors on the host's <see cref="ActorSystem"/>:
|
|
/// <see cref="DriverHostActor"/> (one per node), <see cref="DbHealthProbeActor"/>
|
|
/// (consumed by the health endpoint + redundancy calc), and
|
|
/// <see cref="HistorianAdapterActor"/> wrapping the registered <see cref="IAlarmHistorianSink"/>.
|
|
///
|
|
/// Mirror of <c>WithOtOpcUaControlPlaneSingletons</c> for the driver role. Both must
|
|
/// be registered on the same <see cref="AkkaConfigurationBuilder"/> as the cluster
|
|
/// bootstrap so the actors share the host's ActorSystem.
|
|
///
|
|
/// Wire from the fused Host's Program.cs when the node carries the <c>driver</c> role:
|
|
/// <code>
|
|
/// services.AddOtOpcUaRuntime();
|
|
/// services.AddAkka("otopcua", (ab, sp) => { ab.WithOtOpcUaClusterBootstrap(sp); if (hasDriver) ab.WithOtOpcUaRuntimeActors(); });
|
|
/// </code>
|
|
/// </summary>
|
|
/// <param name="builder">The Akka configuration builder.</param>
|
|
/// <returns>The same <paramref name="builder"/> instance for chaining.</returns>
|
|
public static AkkaConfigurationBuilder WithOtOpcUaRuntimeActors(this AkkaConfigurationBuilder builder)
|
|
{
|
|
// Production cluster HOCON (akka.conf) carries this dispatcher block, but consumers that
|
|
// bootstrap their own HOCON (e.g. ServiceCollectionExtensionsTests) wouldn't pick it up
|
|
// — OpcUaPublishActor.Props pins itself to opcua-synchronized-dispatcher and Akka throws
|
|
// ConfigurationException if it doesn't exist. Prepend a fallback so the runtime extension
|
|
// is self-contained.
|
|
builder.AddHocon(@"
|
|
opcua-synchronized-dispatcher {
|
|
type = ""PinnedDispatcher""
|
|
executor = ""thread-pool-executor""
|
|
throughput = 1
|
|
}
|
|
", HoconAddMode.Prepend);
|
|
|
|
builder.WithActors((system, registry, resolver) =>
|
|
{
|
|
var dbFactory = resolver.GetService<IDbContextFactory<OtOpcUaConfigDbContext>>();
|
|
var roleInfo = resolver.GetService<IClusterRoleInfo>();
|
|
// 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;
|
|
var healthPublisher = resolver.GetService<IDriverHealthPublisher>() ?? NullDriverHealthPublisher.Instance;
|
|
// Node-local deployment-artifact cache. Registered by the Host's AddOtOpcUaLocalDb on
|
|
// driver-role nodes only; deliberately left null elsewhere (admin-only graphs, test
|
|
// harnesses) rather than given a null-object, so DriverHostActor skips caching outright
|
|
// instead of pretending to cache into a sink that drops everything.
|
|
var deploymentArtifactCache = resolver.GetService<IDeploymentArtifactCache>();
|
|
// Where this actor publishes its Primary-gate verdict for the alarm store-and-forward
|
|
// drain, which runs on a timer and so cannot read RedundancyStateChanged itself.
|
|
// Registered by AddAlarmHistorian; absent when no durable sink is configured, in which
|
|
// case there is nothing downstream to inform.
|
|
var redundancyRoleView = resolver.GetService<IRedundancyRoleView>();
|
|
// Host of this node's LocalDb replication partner: the ONLY node that holds a copy of this
|
|
// node's alarm queue, and so the only node it may stand down in favour of. Null when this
|
|
// node dials nobody, which correctly means "never stand down".
|
|
var replicationPeerHost =
|
|
Uri.TryCreate(
|
|
resolver.GetService<IConfiguration>()?["LocalDb:Replication:PeerAddress"],
|
|
UriKind.Absolute,
|
|
out var peerUri)
|
|
? peerUri.Host
|
|
: null;
|
|
// Root script logger backs the ScriptedAlarm host's engine + script logging. Registered in
|
|
// Host DI inside the hasDriver block; may be absent in some role configs / test harnesses,
|
|
// in which case the DriverHostActor gracefully skips spawning the ScriptedAlarm host.
|
|
var scriptRootLogger = resolver.GetService<ScriptRootLogger>();
|
|
// Production evaluator is the Host's RoslynVirtualTagEvaluator (registered as
|
|
// IVirtualTagEvaluator); fall back to the null evaluator for test harnesses that don't
|
|
// register one (VirtualTagActor children then evaluate to nothing).
|
|
var virtualTagEvaluator = resolver.GetService<IVirtualTagEvaluator>();
|
|
if (virtualTagEvaluator is null)
|
|
{
|
|
loggerFactory.CreateLogger("ZB.MOM.WW.OtOpcUa.Runtime.ServiceCollectionExtensions")
|
|
.LogWarning("IVirtualTagEvaluator not registered; Equipment VirtualTags will evaluate to NoChange (no live values). Expected only in test harnesses — driver-role nodes should register RoslynVirtualTagEvaluator.");
|
|
virtualTagEvaluator = NullVirtualTagEvaluator.Instance;
|
|
}
|
|
|
|
// VirtualTag historization sink threaded to the spawned VirtualTagHostActor. Null default
|
|
// (durable AVEVA sink is infra-gated); a deployment binding a real IHistoryWriter overrides.
|
|
var historyWriter = resolver.GetService<IHistoryWriter>() ?? NullHistoryWriter.Instance;
|
|
|
|
var dbHealth = system.ActorOf(
|
|
DbHealthProbeActor.Props(dbFactory),
|
|
DbHealthProbeActorName);
|
|
registry.Register<DbHealthProbeActorKey>(dbHealth);
|
|
|
|
// Dependency mux must be spawned before DriverHostActor so the host can forward
|
|
// AttributeValuePublished into it from the very first driver spawn.
|
|
var mux = system.ActorOf(DependencyMuxActor.Props(), DependencyMuxActorName);
|
|
registry.Register<DependencyMuxActorKey>(mux);
|
|
|
|
// Continuous-historization recorder — gated on ContinuousHistorization:Enabled AND the
|
|
// gateway-backed IHistorianValueWriter + the durable IHistorizationOutbox being registered
|
|
// (the Host registers both ONLY when historization is enabled and the ServerHistorian gateway
|
|
// is configured). The recorder taps the dependency mux's value fan-out, so it is spawned after
|
|
// (and fed) the same `mux` ref the DriverHostActor uses. It is spawned BEFORE the applier so
|
|
// the applier's historized-ref subscription sink can wrap this recorder's IActorRef and feed it
|
|
// the add/remove delta of historized refs on every deploy (closing the T18 ref-feed gap).
|
|
IActorRef? continuousRecorder = null;
|
|
var continuousOptions = resolver.GetService<ContinuousHistorizationOptions>();
|
|
if (continuousOptions is { Enabled: true })
|
|
{
|
|
var valueWriter = resolver.GetService<IHistorianValueWriter>();
|
|
var outbox = resolver.GetService<IHistorizationOutbox>();
|
|
if (valueWriter is not null && outbox is not null)
|
|
{
|
|
// Initial ref set is EMPTY: the deployed address space (and thus the historized-ref
|
|
// set) is built later at deploy time, not here. The applier's per-deploy add/remove
|
|
// feed populates the recorder's interest from that point on.
|
|
continuousRecorder = system.ActorOf(
|
|
ContinuousHistorizationRecorder.Props(
|
|
dependencyMux: mux,
|
|
writer: valueWriter,
|
|
outbox: outbox,
|
|
historizedRefs: Array.Empty<string>(),
|
|
drainBatchSize: continuousOptions.DrainBatchSize,
|
|
drainInterval: TimeSpan.FromSeconds(continuousOptions.DrainIntervalSeconds),
|
|
minBackoff: TimeSpan.FromSeconds(continuousOptions.MinBackoffSeconds),
|
|
maxBackoff: TimeSpan.FromSeconds(continuousOptions.MaxBackoffSeconds)),
|
|
ContinuousHistorizationRecorderActorName);
|
|
registry.Register<ContinuousHistorizationRecorderKey>(continuousRecorder);
|
|
}
|
|
else
|
|
{
|
|
loggerFactory.CreateLogger("ZB.MOM.WW.OtOpcUa.Runtime.ServiceCollectionExtensions")
|
|
.LogWarning("ContinuousHistorization is enabled but IHistorianValueWriter and/or IHistorizationOutbox are not registered; the recorder will not be spawned. Expected only in misconfigured deployments or test harnesses.");
|
|
}
|
|
}
|
|
|
|
// Historized-ref subscription sink fed by the applier on every deploy. When the recorder was
|
|
// spawned, an adapter wraps its IActorRef (a non-blocking Tell of the add/remove delta);
|
|
// otherwise the Null no-op sink, so the applier behaves identically when historization is off.
|
|
IHistorizedTagSubscriptionSink historizedSubscriptions = continuousRecorder is not null
|
|
? new ActorHistorizedTagSubscriptionSink(continuousRecorder)
|
|
: NullHistorizedTagSubscriptionSink.Instance;
|
|
|
|
// Historian tag provisioner fed to the applier so deploying historized tags auto-ensures
|
|
// them in the historian (EnsureTags). The Host registers the gateway-backed provisioner via
|
|
// AddHistorianProvisioning when ServerHistorian:Enabled; otherwise this resolves the no-op
|
|
// NullHistorianProvisioning default seeded by AddOtOpcUaRuntime (so the applier's hook is inert).
|
|
var provisioning = resolver.GetService<IHistorianProvisioning>();
|
|
|
|
// OPC UA publish actor — pinned dispatcher, owns the address-space side of the
|
|
// pipeline. AddressSpaceApplier is constructed here so the actor + applier share the
|
|
// same sink reference (when DeferredAddressSpaceSink swaps later, both see it).
|
|
var applier = new AddressSpaceApplier(
|
|
addressSpaceSink,
|
|
loggerFactory.CreateLogger<AddressSpaceApplier>(),
|
|
provisioning: provisioning,
|
|
historizedSubscriptions: historizedSubscriptions);
|
|
var publishActor = system.ActorOf(
|
|
OpcUaPublishActor.Props(
|
|
sink: addressSpaceSink,
|
|
serviceLevel: serviceLevel,
|
|
localNode: roleInfo.LocalNode,
|
|
dbFactory: dbFactory,
|
|
applier: applier,
|
|
dbHealthProbe: dbHealth),
|
|
OpcUaPublishActorName);
|
|
registry.Register<OpcUaPublishActorKey>(publishActor);
|
|
|
|
// Per-node peer-probe supervisor — keeps one OPC UA TCP probe per OTHER non-Detached
|
|
// driver node, so every node is continuously probed by all its peers. OpcUaPublishActor
|
|
// consumes the resulting probe verdicts to learn this node's own reachability.
|
|
var peerProbes = system.ActorOf(
|
|
PeerProbeSupervisor.Props(roleInfo.LocalNode),
|
|
PeerProbeSupervisorName);
|
|
registry.Register<PeerProbeSupervisorKey>(peerProbes);
|
|
|
|
var driverHost = system.ActorOf(
|
|
DriverHostActor.Props(dbFactory, roleInfo.LocalNode, coordinator: null,
|
|
driverFactory: driverFactory, localRoles: roleInfo.LocalRoles,
|
|
dependencyMux: mux,
|
|
opcUaPublishActor: publishActor,
|
|
healthPublisher: healthPublisher,
|
|
virtualTagEvaluator: virtualTagEvaluator,
|
|
historyWriter: historyWriter,
|
|
loggerFactory: loggerFactory,
|
|
scriptRootLogger: scriptRootLogger,
|
|
invokerFactory: invokerFactory,
|
|
deploymentArtifactCache: deploymentArtifactCache,
|
|
redundancyRoleView: redundancyRoleView,
|
|
replicationPeerHost: replicationPeerHost),
|
|
DriverHostActorName);
|
|
registry.Register<DriverHostActorKey>(driverHost);
|
|
|
|
var historian = system.ActorOf(
|
|
HistorianAdapterActor.Props(historianSink, roleInfo.LocalNode),
|
|
HistorianAdapterActorName);
|
|
registry.Register<HistorianAdapterActorKey>(historian);
|
|
});
|
|
|
|
return builder;
|
|
}
|
|
}
|
|
|
|
/// <summary>Marker key types used by <c>Akka.Hosting</c> to resolve runtime actors from the registry.</summary>
|
|
public sealed class DriverHostActorKey { }
|
|
public sealed class DbHealthProbeActorKey { }
|
|
public sealed class HistorianAdapterActorKey { }
|
|
public sealed class DependencyMuxActorKey { }
|
|
public sealed class OpcUaPublishActorKey { }
|
|
|
|
/// <summary>Marker key for the per-node ContinuousHistorizationRecorder (spawned only when
|
|
/// <c>ContinuousHistorization:Enabled=true</c> and the gateway value-writer + outbox are registered).</summary>
|
|
public sealed class ContinuousHistorizationRecorderKey { }
|
|
|
|
/// <summary>Marker key for the per-node PeerProbeSupervisor.</summary>
|
|
public sealed class PeerProbeSupervisorKey { }
|