Joseph Doherty
3be8a1217e
The drain gate's 'is my queue-sharing peer Primary?' used only the host of
LocalDb:Replication:PeerAddress, which the passive half (ApiKey +
SyncListenPort) does not set - so as Secondary it kept draining beside its
Primary partner and every alarm event reached the historian twice.
DriverHostActor takes a sharesAlarmQueueWithLocal predicate used when no
peer host is known; WithOtOpcUaRuntimeActors builds it
(BuildAlarmQueuePeerPredicate) for a SharesRowsWithPeer node with no
PeerAddress as 'a member of my own cluster-{ClusterId} mesh other than
me'. A node sharing no rows, or with no mesh role, gets none and keeps
draining. AlarmHistorian.md / Redundancy.md / pair-replication runbook
updated.
Claude-Session: https://claude.ai/code/session_01Ef2ozxS7QPanGT8uoATbTe
655 lines
44 KiB
C#
655 lines
44 KiB
C#
using Akka.Actor;
|
||
using Akka.Event;
|
||
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 Microsoft.Extensions.Options;
|
||
using ZB.MOM.WW.OtOpcUa.Cluster;
|
||
using ZB.MOM.WW.OtOpcUa.Commons.Messages.Mesh;
|
||
using ZB.MOM.WW.OtOpcUa.Runtime.Communication;
|
||
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.Authorization;
|
||
using ZB.MOM.WW.OtOpcUa.Core.AlarmHistorian;
|
||
using ZB.MOM.WW.OtOpcUa.Core.ScriptedAlarms;
|
||
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.Telemetry;
|
||
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 = RoleParser.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>();
|
||
// Per-cluster mesh Phase 5: the node-local live-telemetry fan-out hub. Feeds this node's OWN
|
||
// telemetry (never DPS) to the Phase-5 gRPC streaming service central dials in on. AddOtOpcUaRuntime
|
||
// runs only inside Program.cs's hasDriver block, so this lands on driver-role nodes only.
|
||
services.TryAddSingleton<ITelemetryLocalHub, TelemetryLocalHub>();
|
||
|
||
// Gitea #511: the Primary data-plane verdict, published for drivers that assert a standing
|
||
// exclusive identity on a shared external system (today the MQTT driver's Sparkplug host id).
|
||
//
|
||
// Registered HERE, unconditionally on every driver-role node, and NOT alongside the alarm
|
||
// drain's IRedundancyRoleView — which AddAlarmHistorian registers only when that section is
|
||
// enabled. Piggybacking on it would leave a redundant pair with no alarm historian configured
|
||
// with no primary-role source at all, and the driver's documented no-source fallback is
|
||
// "claim" — so both nodes of that pair would claim the same host id and their Wills would
|
||
// fight over the retained STATE. Exactly the failure #511 exists to prevent, reached by a
|
||
// registration accident in an unrelated section.
|
||
//
|
||
// Both interfaces are forwarded to ONE RedundancyRoleView instance, so a node that has both
|
||
// publishes two verdicts into the same object rather than into two that disagree.
|
||
services.TryAddSingleton<RedundancyRoleView>();
|
||
services.TryAddSingleton<IDriverPrimaryRoleSource>(sp => sp.GetRequiredService<RedundancyRoleView>());
|
||
|
||
// Gitea #537: the continuous-historization capture verdict, registered here — unconditionally on
|
||
// every driver-role node — for the same reason as the line above, and NOT alongside the alarm
|
||
// drain's IRedundancyRoleView. AddAlarmHistorian registers that one only when its own section is
|
||
// enabled, so piggybacking would leave a redundant pair that historizes values but not alarms
|
||
// with no gate at all, and both nodes would write every sample twice with nothing to say so.
|
||
services.TryAddSingleton<IHistorizationRoleView>(sp => sp.GetRequiredService<RedundancyRoleView>());
|
||
|
||
// The node→central ack transport under MeshTransport:Mode = Grpc (mesh gRPC command plane,
|
||
// Task 1B.2). Registered unconditionally but resolved only by the Grpc arm of
|
||
// WithOtOpcUaRuntimeActors — the mode is not knowable here (this runs before configuration is
|
||
// read into options), and a singleton nobody resolves is never constructed, so a Dps node
|
||
// opens no channel.
|
||
//
|
||
// Registered in the CONTAINER rather than new-ed up beside the actor so its per-endpoint
|
||
// GrpcChannel cache is disposed at host shutdown; an actor-scoped instance would leak the
|
||
// channels on every restart.
|
||
//
|
||
// Built by a factory, not by type: MS.DI does not honour optional constructor parameters, so
|
||
// the clientFactory test seam would make type-based activation fail outright. The logger is
|
||
// resolved leniently for the same reason DriverHostActor's is — a harness that skipped logging
|
||
// should not fail to build the graph.
|
||
services.TryAddSingleton<IApplyAckSender>(sp => new GrpcApplyAckSender(
|
||
sp.GetRequiredService<IOptions<MeshTransportOptions>>(),
|
||
sp.GetRequiredService<IOptions<AkkaClusterOptions>>(),
|
||
sp.GetService<ILoggerFactory>()?.CreateLogger<GrpcApplyAckSender>()
|
||
?? NullLogger<GrpcApplyAckSender>.Instance));
|
||
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. Forwarded to the same instance
|
||
// AddOtOpcUaRuntime registered for IDriverPrimaryRoleSource: one object, two verdicts.
|
||
services.TryAddSingleton<RedundancyRoleView>();
|
||
services.TryAddSingleton<IRedundancyRoleView>(sp => sp.GetRequiredService<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 carries the shared ApiKey plus SyncListenPort and waits.
|
||
// The predicate is LocalDbReplicationPolicy.SharesRowsWithPeer: ApiKey AND (PeerAddress OR
|
||
// SyncListenPort > 0) - an authenticated sync path in either direction. Both auth directions
|
||
// fail closed, so a path without the key, or a key with no path (a fleet-wide secret on an
|
||
// unpaired node), shares nothing and must keep draining. Deliberately stricter than the
|
||
// registration predicate (CapturesChanges) - see the policy's remarks.
|
||
var replicated = LocalDbReplicationPolicy.SharesRowsWithPeer(configuration);
|
||
|
||
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>
|
||
/// Review R6 MINOR-4 — "does this snapshot node hold my replicated alarm queue?" for a node that does not
|
||
/// dial its peer. Both halves of a LocalDb pair hold the same rows, but only the dialling half knows its
|
||
/// partner (<c>LocalDb:Replication:PeerAddress</c>); the passive half (<c>ApiKey</c> + <c>SyncListenPort</c>)
|
||
/// knew nobody, so as Secondary it never stood its drain down and both nodes delivered every alarm event.
|
||
/// Its partner is its per-cluster mesh partner, which a cluster-scoped role identifies.
|
||
/// </summary>
|
||
/// <param name="configuration">The host configuration (the LocalDb replication keys).</param>
|
||
/// <param name="roleInfo">This node's cluster role info.</param>
|
||
/// <param name="replicationPeerHost">The dialled peer's host, when this node dials one.</param>
|
||
/// <returns>
|
||
/// <c>null</c> when the node dials its peer (the host match covers it), when it shares its queue with no
|
||
/// peer (<see cref="LocalDbReplicationPolicy.SharesRowsWithPeer"/> false — it must never stand down), or
|
||
/// when it carries no cluster-scoped role (a legacy node cannot prove which node is its partner, and
|
||
/// standing down for the wrong Primary would strand its rows); otherwise "is a member of my own mesh other
|
||
/// than me".
|
||
/// </returns>
|
||
internal static Func<Commons.Types.NodeId, bool>? BuildAlarmQueuePeerPredicate(
|
||
IConfiguration? configuration, IClusterRoleInfo? roleInfo, string? replicationPeerHost)
|
||
{
|
||
if (replicationPeerHost is not null || configuration is null || roleInfo is null) return null;
|
||
if (!LocalDbReplicationPolicy.SharesRowsWithPeer(configuration)) return null;
|
||
if (roleInfo.ClusterRole is not { } meshRole) return null;
|
||
var local = roleInfo.LocalNode;
|
||
return nodeId => nodeId != local && roleInfo.MembersWithRole(meshRole).Contains(nodeId);
|
||
}
|
||
|
||
/// <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. Keep this block identical to the akka.conf one.
|
||
//
|
||
// The OPC UA SDK's single-threaded session/subscription requirement is guaranteed by the
|
||
// PinnedDispatcher alone (a dedicated thread per actor instance), not by throughput.
|
||
// throughput only sets how many queued messages drain per scheduling run, so 100 removes a
|
||
// mailbox re-schedule round-trip per message on the hottest actor in the process with zero
|
||
// change to concurrency.
|
||
builder.AddHocon(@"
|
||
opcua-synchronized-dispatcher {
|
||
type = ""PinnedDispatcher""
|
||
executor = ""thread-pool-executor""
|
||
throughput = 100
|
||
}
|
||
", 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>();
|
||
// Per-cluster mesh Phase 4: scripted-alarm condition-state store. Registered by the Host's
|
||
// AddOtOpcUaLocalDb on driver-role nodes (the replicated LocalDb store), so condition state
|
||
// persists with no ConfigDb; null elsewhere (admin-only graphs, test harnesses) → the actor
|
||
// skips spawning the alarm host (the ConfigDb-backed EF fallback was retired in Phase 4 Task 9).
|
||
var alarmStateStore = resolver.GetService<IAlarmStateStore>();
|
||
// Per-cluster mesh Phase 3: ConfigSource:Mode selects where this driver reads config.
|
||
// FetchAndCache pulls the artifact from central over gRPC and reads only the LocalDb cache;
|
||
// Direct (default) reads central SQL as before. The fetcher is resolved only in FetchAndCache
|
||
// mode (registered by the Host under hasDriver); its absence there is a misconfiguration the
|
||
// actor fails the apply on rather than fetching nothing forever.
|
||
var configSourceOptions =
|
||
resolver.GetService<IOptions<ConfigSourceOptions>>()?.Value ?? new ConfigSourceOptions();
|
||
var fetchAndCacheMode = string.Equals(
|
||
configSourceOptions.Mode, ConfigSourceOptions.ModeFetchAndCache, StringComparison.OrdinalIgnoreCase);
|
||
var artifactFetcher = fetchAndCacheMode ? resolver.GetService<IDeploymentArtifactFetcher>() : null;
|
||
// 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 — unpaired (never stand down), or the PASSIVE half of a pair, which is
|
||
// told its partner through BuildAlarmQueuePeerPredicate instead (review R6 MINOR-4).
|
||
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;
|
||
|
||
// Per-cluster mesh Phase 4: a driver-only node has NO ConfigDb (LocalDb is its config
|
||
// store), so dbFactory is null and there is nothing to probe. Skip spawning the probe
|
||
// entirely — spawning it with a null factory would NRE on the first SELECT 1 — and do not
|
||
// register its marker key. OpcUaPublishActor then runs in dbLess mode (DbReachable retired,
|
||
// treated constant-true) so a healthy DB-less node still publishes full ServiceLevel.
|
||
IActorRef? dbHealth = null;
|
||
if (dbFactory is not null)
|
||
{
|
||
dbHealth = system.ActorOf(
|
||
DbHealthProbeActor.Props(dbFactory),
|
||
DbHealthProbeActorName);
|
||
registry.Register<DbHealthProbeActorKey>(dbHealth);
|
||
}
|
||
|
||
// Dependency mux must be spawned before DriverHostActor so the host can forward
|
||
// AttributeValueBatchPublished into it from the very first driver spawn.
|
||
var mux = system.ActorOf(DependencyMuxActor.Props(), DependencyMuxActorName);
|
||
registry.Register<DependencyMuxActorKey>(mux);
|
||
|
||
// Per-cluster mesh Phase 2: the node end of the central↔node command boundary. Spawned
|
||
// BEFORE DriverHostActor because it is the host's ackRouter under Grpc mode. It has no
|
||
// dependency of its own on the host — inbound commands travel via the node's
|
||
// EventStream — so this ordering costs nothing.
|
||
var meshOptions = resolver.GetService<IOptions<MeshTransportOptions>>().Value;
|
||
var useGrpc = string.Equals(
|
||
meshOptions.Mode, MeshTransportOptions.ModeGrpc, StringComparison.OrdinalIgnoreCase);
|
||
|
||
// The node→central ack transport. Built ONLY under Grpc: under Dps the ack goes out on
|
||
// the deployment-acks topic from DriverHostActor and never reaches this actor at all.
|
||
IApplyAckSender? ackSender = null;
|
||
if (useGrpc)
|
||
{
|
||
ackSender = resolver.GetService<IApplyAckSender>();
|
||
if (ackSender is null)
|
||
{
|
||
// Reachable only in a graph that skipped AddOtOpcUaRuntime (a test harness), never
|
||
// in the Host. Said here rather than left to the actor's per-ack drop warning,
|
||
// because the cause is a missing registration and the symptom would otherwise
|
||
// appear once per deployment with no hint of why.
|
||
system.Log.Warning(
|
||
"MeshTransport:Mode is {Mode} but no {Sender} is registered — every ApplyAck "
|
||
+ "this node produces will be dropped. AddOtOpcUaRuntime registers it.",
|
||
meshOptions.Mode, nameof(IApplyAckSender));
|
||
}
|
||
}
|
||
|
||
var nodeComm = system.ActorOf(
|
||
NodeCommunicationActor.Props(ackSender),
|
||
MeshPaths.NodeCommunicationName);
|
||
|
||
// The registry entry is the whole reachability story: NodeCommandGrpcService resolves
|
||
// this key to hand an inbound command over. Nothing registers with a receptionist any
|
||
// more — there is no Akka association from central into this mesh to register into.
|
||
registry.Register<NodeCommunicationActorKey>(nodeComm);
|
||
|
||
// 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),
|
||
// Gitea #537: read the verdict PER VALUE off the shared role view rather
|
||
// than capturing a bool, so a promotion resumes capture on the next sample.
|
||
// The view is registered unconditionally on driver nodes; a null one (test
|
||
// graphs) captures unconditionally, which is the right posture there.
|
||
captureGate: resolver.GetService<IHistorizationRoleView>() is { } histRoleView
|
||
? () => histRoleView.ShouldRecordHistorizedValues
|
||
: null,
|
||
maxWriteAttempts: continuousOptions.MaxWriteAttempts,
|
||
// Bounded-parallel per-tag drain: the groups in one pass are independent
|
||
// gateway round-trips, so a sequential pass costs `tags × latency`.
|
||
drainConcurrency: continuousOptions.DrainConcurrency),
|
||
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);
|
||
// #520 stage 4 — the sink that receives each applied artifact's API keys + ACL grants, so OPC
|
||
// UA session activation can authenticate. Optional: GetService (not GetRequiredService), because
|
||
// a node with no driver role never registers it and test harnesses build this graph without it;
|
||
// the actor treats null as "nothing consumes authorization inputs" and skips the decode.
|
||
var dataPlaneAuthority = resolver.GetService<IDataPlaneAuthoritySink>();
|
||
var publishActor = system.ActorOf(
|
||
OpcUaPublishActor.Props(
|
||
sink: addressSpaceSink,
|
||
serviceLevel: serviceLevel,
|
||
localNode: roleInfo.LocalNode,
|
||
dbFactory: dbFactory,
|
||
applier: applier,
|
||
dbHealthProbe: dbHealth,
|
||
dbLess: dbFactory is null,
|
||
dataPlaneAuthority: dataPlaneAuthority),
|
||
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);
|
||
|
||
// Gitea #537: "does this snapshot node record the same historized refs as me?" — true only
|
||
// for a partner in this node's own per-cluster mesh, which is exactly the set that serves
|
||
// this node's address space. A node carrying NO cluster-scoped role (the legacy,
|
||
// pre-Phase-6 shape) cannot establish that, so the predicate stays null and this node never
|
||
// stands down: on a fleet-wide election the elected Primary is usually running a DIFFERENT
|
||
// cluster's configuration, and deferring to it would silently end this cluster's
|
||
// historization altogether — the same scope mismatch that once suspended the alarm drain on
|
||
// every node at once.
|
||
Func<Commons.Types.NodeId, bool>? historizationScopePeer =
|
||
roleInfo.ClusterRole is { } meshRole
|
||
? nodeId => roleInfo.MembersWithRole(meshRole).Contains(nodeId)
|
||
: null;
|
||
|
||
var driverHost = system.ActorOf(
|
||
// ackRouter: under Grpc the node comm actor relays ApplyAcks across the mesh
|
||
// boundary through the gRPC sender it was built with; under Dps it stays null and
|
||
// the host publishes on the deployment-acks topic exactly as before. Routed to DPS
|
||
// instead, a Grpc node's ack would never reach the sender at all and every
|
||
// deployment to it would time out while it applied perfectly.
|
||
// dbFactory is nullable (Phase 4): a driver-only node passes null here — ConfigDb is
|
||
// gated on hasAdmin and driver-only nodes are forced to FetchAndCache. DriverHostActor
|
||
// guards every ConfigDb-touching path on it (UpsertNodeDeploymentState no-ops; central
|
||
// persists acks from the ApplyAck) and serves scripted-alarm state from the LocalDb
|
||
// alarmStateStore (there is no ConfigDb-backed EF fallback; the host skips the alarm
|
||
// host when no store was wired).
|
||
DriverHostActor.Props(dbFactory, roleInfo.LocalNode,
|
||
ackRouter: useGrpc ? nodeComm : 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,
|
||
fetchAndCacheMode: fetchAndCacheMode,
|
||
artifactFetcher: artifactFetcher,
|
||
alarmStateStore: alarmStateStore,
|
||
// Phase 5: the node-local live-telemetry hub (registered in AddOtOpcUaRuntime). The host
|
||
// emits its Primary-gated native-alarm transitions into it and threads it into the
|
||
// VirtualTag + ScriptedAlarm hosts it spawns.
|
||
telemetryHub: resolver.GetService<ITelemetryLocalHub>(),
|
||
// Gitea #511: where this actor publishes its Primary DATA-PLANE verdict for drivers
|
||
// that claim an exclusive external identity. Registered unconditionally by
|
||
// AddOtOpcUaRuntime, unlike redundancyRoleView above — see the comment there.
|
||
primaryRoleSource: resolver.GetService<IDriverPrimaryRoleSource>(),
|
||
// Gitea #537: where this actor publishes the continuous-historization CAPTURE
|
||
// verdict, and how it decides which peer is worth standing down for.
|
||
historizationRoleView: resolver.GetService<IHistorizationRoleView>(),
|
||
sharesHistorizationScopeWithLocal: historizationScopePeer,
|
||
// Review R6 MINOR-4: the passive half of a LocalDb pair has no PeerAddress, so it is told
|
||
// its queue-sharing peer as its mesh partner instead — otherwise it never stands down.
|
||
sharesAlarmQueueWithLocal: BuildAlarmQueuePeerPredicate(
|
||
resolver.GetService<IConfiguration>(), roleInfo, 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 { }
|
||
public sealed class NodeCommunicationActorKey { }
|
||
|
||
/// <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 { }
|