dohertj2/lmxopcua
1
0
Fork 0
Code Issues 32 Pull Requests Actions Packages Projects Releases Wiki Activity

fix(runtime): Primary verdict published from host start, cache boot never adopts malformed bytes, strict drain-gate predicate, HostAbortedException passthrough, seed only Up members

Browse Source 
Review R1:
- M1 (#609): DriverHostActor.StopChild releases the instance's resilience state before evicting the hub.
- m2 (#597): DriverHostActor publishes ShouldServiceAsPrimary() to IDriverPrimaryRoleSource at start and on
  cluster membership changes (deduped), so the native-ack gate denies in a pair's boot window.
- m4 (#601): ApplyCachedArtifact returns false on empty/malformed bytes; the cached revision is not adopted,
  Direct mode goes Stale (no RUNNING FROM CACHE), FetchAndCache lands Steady-with-no-revision and a
  re-dispatch of the same revision really applies.
- m3 (#603): LocalDbReplicationPolicy splits into CapturesChanges (PeerAddress||ApiKey, registration) and
  SharesRowsWithPeer (ApiKey && (PeerAddress||SyncListenPort>0), alarm drain gate).
- m5 (#618): Program.cs catch-all filters HostAbortedException (design-time tooling).
- m6 (Cluster-002): ClusterRoleInfo seeds only Up/WeaklyUp members.

Claude-Session: https://claude.ai/code/session_01Ef2ozxS7QPanGT8uoATbTe
This commit is contained in:
Joseph Doherty
2026-10-08 07:47:34 -04:00
parent 83e9ffd27a
commit e3dcb180c3
17 changed files with 453 additions and 65 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/DriverHostActor.cs
+128 -11
View File
@@ -788,9 +788,91 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS
// Same rationale for the ScriptedAlarm host — the bootstrap-restore path Tells it
// ApplyScriptedAlarms through the same PushDesiredSubscriptions pass.
SpawnScriptedAlarmHost();
StartPrimaryVerdictPublishing();
Bootstrap();
}
/// <summary>
/// Review R1 m2 (Gitea #597): the published Primary data-plane verdict (<see cref="IDriverPrimaryRoleSource"/>
/// — the native-ack gate and the Sparkplug primary-host claim) must mirror <see cref="ShouldServiceAsPrimary"/>
/// from the moment this host starts, not only from the first redundancy snapshot. Before this, the view sat
/// on its single-node seed (allow) through the whole boot window, so on a pair BOTH nodes accepted a native
/// ack while this host — which evaluates the member count live — dropped the upstream route on the
/// non-Primary: the acked → unacked revert #597 targets. So publish once now, and again whenever cluster
/// membership changes (a driver peer coming Up turns an unknown role from allow into deny before any
/// snapshot names this node). A non-cluster ActorRefProvider (legacy/test harnesses) just keeps the
/// initial publish.
/// </summary>
private void StartPrimaryVerdictPublishing()
{
if (_primaryRoleSource is null) return;
PublishPrimaryVerdict();
try
{
Akka.Cluster.Cluster.Get(Context.System).Subscribe(
Self, Akka.Cluster.ClusterEvent.SubscriptionInitialStateMode.InitialStateAsSnapshot,
typeof(Akka.Cluster.ClusterEvent.IMemberEvent));
_subscribedToMemberEvents = true;
}
catch (Exception)
{
// Non-cluster ActorRefProvider — no membership to follow; DriverMemberCount() reads 0 there too.
}
}
/// <summary>A cluster member changed status: re-publish the Primary verdict now and once more shortly after.
/// The deferred pass covers the cluster read-view (what <see cref="DriverMemberCount"/> reads) updating on its
/// own subscriber AFTER this event was delivered here — the two are not ordered.</summary>
private void OnDriverMembershipChanged(Akka.Cluster.ClusterEvent.IMemberEvent _)
{
PublishPrimaryVerdictIfChanged();
// One keyed timer: a burst of member events collapses into a single deferred pass.
Timers.StartSingleTimer("republish-primary-verdict", RepublishPrimaryVerdict.Instance, MembershipRepublishDelay);
}
/// <summary>Publishes <see cref="ShouldServiceAsPrimary"/> to the primary-role source (no-op when none is wired).
/// The snapshot path publishes on every snapshot, as the other verdicts do.</summary>
private void PublishPrimaryVerdict()
{
if (_primaryRoleSource is null) return;
var verdict = ShouldServiceAsPrimary();
_lastPublishedPrimaryVerdict = verdict;
_primaryRoleSource.PublishPrimaryRole(verdict);
}
/// <summary>The membership path publishes only a CHANGED verdict — member events are frequent and carry no
/// role information, so republishing an unchanged verdict on each would be noise.</summary>
private void PublishPrimaryVerdictIfChanged()
{
if (_primaryRoleSource is null || ShouldServiceAsPrimary() == _lastPublishedPrimaryVerdict) return;
PublishPrimaryVerdict();
}
/// <inheritdoc />
protected override void PostStop()
{
if (_subscribedToMemberEvents)
{
try { Akka.Cluster.Cluster.Get(Context.System).Unsubscribe(Self); }
catch (Exception) { /* system shutting down */ }
}
base.PostStop();
}
/// <summary>Self-message: re-evaluate and re-publish the Primary verdict (see <see cref="OnDriverMembershipChanged"/>).</summary>
internal sealed class RepublishPrimaryVerdict
{
/// <summary>The shared instance.</summary>
public static readonly RepublishPrimaryVerdict Instance = new();
private RepublishPrimaryVerdict() { }
}
/// <summary>How long after a membership event the Primary verdict is re-published a second time.</summary>
private static readonly TimeSpan MembershipRepublishDelay = TimeSpan.FromSeconds(1);
private bool _subscribedToMemberEvents;
private bool? _lastPublishedPrimaryVerdict;
/// <summary>
/// Spawns the single <see cref="VirtualTagHostActor"/> child that owns the Equipment-namespace
/// VirtualTagActors and bridges their results onto the OPC UA publish actor. A test-supplied
@@ -990,12 +1072,24 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS
var deploymentId = DeploymentId.Parse(cached.DeploymentId);
var revision = RevisionHash.Parse(cached.RevisionHash);
_currentRevision = revision;
BecomeSteady();
// Restore drivers + address space + subscriptions from the cached bytes — no SQL read, no
// re-ack (the deployment is already Applied). Same in-hand-bytes apply as the Direct-mode
// cache boot, but this is the steady-state config source, not an outage fallback.
ApplyCachedArtifact(deploymentId, cached.Artifact);
if (!ApplyCachedArtifact(deploymentId, cached.Artifact))
{
// #601 / review R1 m4: unusable cached bytes applied nothing, so they must not become the
// current revision — that would ack a re-dispatch of the same revision as already applied
// without ever applying it. Steady-with-no-revision, exactly as for an empty cache.
_log.Warning(
"DriverHost {Node}: FetchAndCache boot — the cached artifact for deployment {Id} (rev {Rev}) " +
"is unreadable; entering Steady (no revision). The next dispatch will fetch.",
_localNode, deploymentId, revision);
BecomeSteady();
return;
}
_currentRevision = revision;
BecomeSteady();
_log.Info(
"DriverHost {Node}: FetchAndCache boot — restored served state for deployment {Id} " +
@@ -1059,14 +1153,24 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS
var deploymentId = DeploymentId.Parse(cached.DeploymentId);
var revision = RevisionHash.Parse(cached.RevisionHash);
// #601 / review R1 m4: unusable cached bytes applied nothing — the node is NOT serving a
// configuration, so it must not claim RUNNING FROM CACHE, adopt the revision (a re-dispatch of
// it would then be acked without being applied) or skip Stale's ConfigDb retry loop.
if (!ApplyCachedArtifact(deploymentId, cached.Artifact))
{
_log.Error(
"DriverHost {Node}: the cached artifact for deployment {Id} (rev {Rev}) is unreadable; " +
"not booting from it — entering Stale with no configuration.",
_localNode, deploymentId, revision);
return false;
}
// Steady, not Stale: this node is serving a real configuration. The retry-db timer that
// Stale would have started does not run here, so recovery rides on the next dispatch —
// matching how a normally-booted node behaves.
_currentRevision = revision;
BecomeSteady();
ApplyCachedArtifact(deploymentId, cached.Artifact);
_isRunningFromCache = true;
_log.Warning(
"DriverHost {Node}: RUNNING FROM CACHE — central ConfigDb is unreachable, so this node " +
@@ -1094,16 +1198,16 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS
/// <c>UpsertNodeDeploymentState</c> and <c>SendAck</c> for the same reason: both write to or
/// depend on central.
/// </remarks>
private void ApplyCachedArtifact(DeploymentId deploymentId, byte[] blob)
/// <returns><see langword="false"/> when the bytes were empty or malformed and nothing was applied — the
/// caller must not adopt the cached revision (#601, review R1 m4).</returns>
private bool ApplyCachedArtifact(DeploymentId deploymentId, byte[] blob)
{
var correlation = CorrelationId.NewId();
// #601: null = the cached bytes were empty or malformed — no answer, not an empty configuration.
// Keep last-known-good: no address-space rebuild and no subscription push from unusable bytes.
// #601: null = the cached bytes were empty or malformed — no answer, not an empty configuration.
// Keep last-known-good: no address-space rebuild and no subscription push from unusable bytes.
if (ReconcileDriversFromBlob(deploymentId, blob) is null)
return;
return false;
// Hand the cached blob to the rebuild so it materialises the client-facing address space from
// it directly. Passing only the deploymentId would drive a ConfigDb read — unreachable here by
@@ -1113,6 +1217,7 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS
new ZB.MOM.WW.OtOpcUa.Runtime.OpcUa.OpcUaPublishActor.RebuildAddressSpace(correlation, deploymentId, blob));
PushDesiredSubscriptionsFromArtifact(deploymentId, blob);
return true;
}
/// <summary>
@@ -1170,6 +1275,9 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS
Receive<RouteNodeWrite>(HandleRouteNodeWrite);
Receive<RouteNativeAlarmAck>(HandleRouteNativeAlarmAck);
Receive<RedundancyStateChanged>(OnRedundancyStateChanged);
Receive<Akka.Cluster.ClusterEvent.IMemberEvent>(OnDriverMembershipChanged);
Receive<RepublishPrimaryVerdict>(_ => PublishPrimaryVerdictIfChanged());
Receive<Akka.Cluster.ClusterEvent.CurrentClusterState>(_ => PublishPrimaryVerdictIfChanged());
// The redundancy-state topic also carries OpcUaProbeResult (OpcUaPublishActor peer-probes).
// We don't consume it here — drop it so it doesn't dead-letter (matches PeerProbeSupervisor).
Receive<ZB.MOM.WW.OtOpcUa.Runtime.Health.PeerOpcUaProbeActor.OpcUaProbeResult>(_ => { });
@@ -1228,6 +1336,9 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS
Receive<RouteNodeWrite>(HandleRouteNodeWrite);
Receive<RouteNativeAlarmAck>(HandleRouteNativeAlarmAck);
Receive<RedundancyStateChanged>(OnRedundancyStateChanged);
Receive<Akka.Cluster.ClusterEvent.IMemberEvent>(OnDriverMembershipChanged);
Receive<RepublishPrimaryVerdict>(_ => PublishPrimaryVerdictIfChanged());
Receive<Akka.Cluster.ClusterEvent.CurrentClusterState>(_ => PublishPrimaryVerdictIfChanged());
// The redundancy-state topic also carries OpcUaProbeResult (OpcUaPublishActor peer-probes).
// We don't consume it here — drop it so it doesn't dead-letter (matches PeerProbeSupervisor).
Receive<ZB.MOM.WW.OtOpcUa.Runtime.Health.PeerOpcUaProbeActor.OpcUaProbeResult>(_ => { });
@@ -1975,7 +2086,7 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS
// drain verdict published immediately above. A driver reading this claims an exclusive identity
// on a shared external system (Gitea #511: the Sparkplug host id), where two claimants is the
// harm, so an unknown role with a driver peer present must deny.
_primaryRoleSource?.PublishPrimaryRole(ShouldServiceAsPrimary());
PublishPrimaryVerdict();
}
private void Stale()
@@ -1998,6 +2109,9 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS
Receive<ReprovisionDeferred>(m => ReprovisionChild(m.Request, m.Child));
Receive<ReconnectDriver>(HandleReconnectDriver);
Receive<RedundancyStateChanged>(OnRedundancyStateChanged);
Receive<Akka.Cluster.ClusterEvent.IMemberEvent>(OnDriverMembershipChanged);
Receive<RepublishPrimaryVerdict>(_ => PublishPrimaryVerdictIfChanged());
Receive<Akka.Cluster.ClusterEvent.CurrentClusterState>(_ => PublishPrimaryVerdictIfChanged());
// The redundancy-state topic also carries OpcUaProbeResult (OpcUaPublishActor peer-probes).
// We don't consume it here — drop it so it doesn't dead-letter (matches PeerProbeSupervisor).
Receive<ZB.MOM.WW.OtOpcUa.Runtime.Health.PeerOpcUaProbeActor.OpcUaProbeResult>(_ => { });
@@ -3233,7 +3347,10 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS
Context.Stop(entry.Actor);
_children.Remove(driverInstanceId);
// #609: the instance left the plan (StopChild is only reached via plan.ToStop), so its cached
// Health/Resilience snapshots must not replay to new telemetry subscribers.
// Health/Resilience snapshots must not replay to new telemetry subscribers. Release the shared
// resilience state FIRST: the resilience publisher re-emits the tracker's snapshots every tick, so
// evicting the hub alone was undone within one interval.
_invokerFactory.Release(driverInstanceId);
_telemetryHub?.Evict(driverInstanceId);
_log.Info("DriverHost {Node}: stopped driver child {Id}", _localNode, driverInstanceId);
}