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

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
+10
View File
@@ -93,6 +93,16 @@ Two postures are deliberate:
- **Unset (the default) means drain.** So does an unpublished view. A deployment - **Unset (the default) means drain.** So does an unpublished view. A deployment
that runs no redundancy never publishes a role, and defaulting closed would that runs no redundancy never publishes a role, and defaulting closed would
silently stop its alarm history forever. silently stop its alarm history forever.
- **Only a shared queue is gated.** Standing down is safe only because a peer
holds the same rows, so the gate applies only when
`LocalDbReplicationPolicy.SharesRowsWithPeer` holds:
`LocalDb:Replication:ApiKey` **AND** (`LocalDb:Replication:PeerAddress` **OR**
`LocalDb:SyncListenPort > 0`) — an authenticated sync path in either
direction. Any other node always drains its own queue. This is deliberately
stricter than the table-registration predicate (`PeerAddress` OR `ApiKey`):
a key with no dial and no listener (e.g. a fleet-wide secret on an unpaired
node) captures changes but shares them with nobody, and gating it would park
its events behind a Primary that does not hold them (review R1 m3).
- **A gate that throws means "not now", never permission.** Draining on a gate - **A gate that throws means "not now", never permission.** Draining on a gate
fault would put a pair into dual delivery exactly when its redundancy signal is fault would put a pair into dual delivery exactly when its redundancy signal is
sick. sick.
+5 -2
View File
@@ -218,7 +218,7 @@ See the [ClusterClient→gRPC migration design](plans/2026-08-06-clusterclient-t
| `GrpcListenPort` | int | `0` | Dedicated **h2c-only** Kestrel port for the artifact gRPC service. `0` = disabled (nothing bound). Must differ from the main HTTP port and from `LocalDb:SyncListenPort` (h2c cannot share a cleartext HTTP/1 port). Mapped only on admin-role nodes. | | `GrpcListenPort` | int | `0` | Dedicated **h2c-only** Kestrel port for the artifact gRPC service. `0` = disabled (nothing bound). Must differ from the main HTTP port and from `LocalDb:SyncListenPort` (h2c cannot share a cleartext HTTP/1 port). Mapped only on admin-role nodes. |
| `ApiKey` | string | `""` | Shared bearer key the serve-side interceptor checks (constant-time, **fail-closed**: an unset key rejects every call). **Supply via env `ConfigServe__ApiKey` — never commit.** | | `ApiKey` | string | `""` | Shared bearer key the serve-side interceptor checks (constant-time, **fail-closed**: an unset key rejects every call). **Supply via env `ConfigServe__ApiKey` — never commit.** |
**It is a dark switch; the serve side is always wired.** Central maps the `DeploymentArtifactService` behind the `ConfigServeAuthInterceptor` (path-scoped, `FixedTimeEquals`, fail-closed) whenever `ConfigServe:GrpcListenPort > 0`, in both modes — so flipping a node to `FetchAndCache` is a config change, not a redeploy. A `FetchAndCache` node that cannot fetch (all endpoints down, `NotFound`, SHA mismatch) **fails the apply and keeps last-known-good** — a null fetch is "no answer," never an empty configuration (the [#485](plans/2026-07-15-v3-batch4-address-space-plan.md) contract). At boot it restores served state from the LocalDb pointer with no SQL read; an empty or unreadable cache lands Steady-with-no-revision (the first dispatch fetches), never Stale. **It is a dark switch; the serve side is always wired.** Central maps the `DeploymentArtifactService` behind the `ConfigServeAuthInterceptor` (path-scoped, `FixedTimeEquals`, fail-closed) whenever `ConfigServe:GrpcListenPort > 0`, in both modes — so flipping a node to `FetchAndCache` is a config change, not a redeploy. A `FetchAndCache` node that cannot fetch (all endpoints down, `NotFound`, SHA mismatch) **fails the apply and keeps last-known-good** — a null fetch is "no answer," never an empty configuration (the [#485](plans/2026-07-15-v3-batch4-address-space-plan.md) contract). At boot it restores served state from the LocalDb pointer with no SQL read; an empty or unreadable cache — including a cached row whose bytes are malformed (review R1 m4) — lands Steady-with-no-revision (the first dispatch fetches, even of the same revision), never Stale.
**The listener is dedicated h2c and coexists with the LocalDb-sync listener.** A fused admin+driver node with **both** `ConfigServe:GrpcListenPort` and `LocalDb:SyncListenPort` set re-binds its existing HTTP surface exactly once and adds both dedicated HTTP/2-only ports on top — see the same Kestrel-takeover caveat under [`LocalDb`](#localdb-pair-local-config-cache--driver-role-only). **The listener is dedicated h2c and coexists with the LocalDb-sync listener.** A fused admin+driver node with **both** `ConfigServe:GrpcListenPort` and `LocalDb:SyncListenPort` set re-binds its existing HTTP surface exactly once and adds both dedicated HTTP/2-only ports on top — see the same Kestrel-takeover caveat under [`LocalDb`](#localdb-pair-local-config-cache--driver-role-only).
@@ -431,7 +431,10 @@ replication stays inert until the sync port + peer are set.
`PeerAddress` **OR** `ApiKey` non-blank — the OR because only the *dialling* half sets `PeerAddress`, so `PeerAddress` **OR** `ApiKey` non-blank — the OR because only the *dialling* half sets `PeerAddress`, so
keying on it alone would strip change capture from every passive node and leave the pair converging in one keying on it alone would strip change capture from every passive node and leave the pair converging in one
direction only. `SyncListenPort` is deliberately **not** part of it (a port with no `ApiKey` is fail-closed direction only. `SyncListenPort` is deliberately **not** part of it (a port with no `ApiKey` is fail-closed
and syncs with nobody). A configured node registers all four tables with `baselineExistingRows: true`, so and syncs with nobody). (The alarm-historian **drain gate** uses a stricter predicate on purpose —
`ApiKey` AND (`PeerAddress` OR `SyncListenPort > 0`), `LocalDbReplicationPolicy.SharesRowsWithPeer` — because
gating a node that shares its queue with nobody strands its alarm history; see `docs/AlarmHistorian.md`
§"The drain gate".) A configured node registers all four tables with `baselineExistingRows: true`, so
turning replication on converges the rows **already in the file** and not merely subsequent writes; an turning replication on converges the rows **already in the file** and not merely subsequent writes; an
unconfigured node **deregisters** them at boot, dropping capture triggers an earlier build installed. ⚠️ The unconfigured node **deregisters** them at boot, dropping capture triggers an earlier build installed. ⚠️ The
sync handshake compares registered-table digests **fail-closed**, so flipping either key is a sync handshake compares registered-table digests **fail-closed**, so flipping either key is a
+6 -3
View File
@@ -526,13 +526,13 @@ Three data-plane surfaces are **Primary-gated** so a warm standby never touches
Before archreview 03/S4 the unknown-role case defaulted to *allow* unconditionally, opening a dual-primary window: a freshly-booted secondary on a multi-node cluster serviced shared-device writes and emitted duplicate fleet-wide alerts for up to the ~10 s `RedundancyStateActor` heartbeat (and indefinitely if the snapshot's NodeId never matched this node). Membership is authoritative much earlier than DPS delivery, so an unknown role on a **multi-driver** cluster now default-**denies**. The driver member count is read live from `Cluster.State.Members` (Up members with the `driver` role); a non-cluster ActorRefProvider yields 0 ⇒ single-node posture. If a snapshot arrives that never mentions this node while a driver peer exists (the 03/S5 identity-mismatch shape), `DriverHostActor` logs a one-time Warning so the silent mismatch is diagnosable. Before archreview 03/S4 the unknown-role case defaulted to *allow* unconditionally, opening a dual-primary window: a freshly-booted secondary on a multi-node cluster serviced shared-device writes and emitted duplicate fleet-wide alerts for up to the ~10 s `RedundancyStateActor` heartbeat (and indefinitely if the snapshot's NodeId never matched this node). Membership is authoritative much earlier than DPS delivery, so an unknown role on a **multi-driver** cluster now default-**denies**. The driver member count is read live from `Cluster.State.Members` (Up members with the `driver` role); a non-cluster ActorRefProvider yields 0 ⇒ single-node posture. If a snapshot arrives that never mentions this node while a driver peer exists (the 03/S5 identity-mismatch shape), `DriverHostActor` logs a one-time Warning so the silent mismatch is diagnosable.
**What a client sees during the boot window (multi-node, role unknown):** an inbound write is **rejected** (not queued — OT actuation seconds late is a surprise). The reject rides the sanctioned optimistic-write self-correction surface: the node reverts to its prior value with a transient Bad-quality blip and a Part 8 `AuditWriteUpdateEvent`; the client's synchronous write call still returned Good (exactly as any failed device write behaves since the #5 write-outcome self-correction). The rejection reason distinguishes the boot window (`"not primary (role unknown)"`) from a steady-state Secondary (`"not primary"`). Acks are dropped (Warning-logged when role-unknown). The alerts emit is skipped (the Primary peer publishes the single fleet copy). The window closes on the first delivered snapshot. **What a client sees during the boot window (multi-node, role unknown):** an inbound write is **rejected** (not queued — OT actuation seconds late is a surprise). The reject rides the sanctioned optimistic-write self-correction surface: the node reverts to its prior value with a transient Bad-quality blip and a Part 8 `AuditWriteUpdateEvent`; the client's synchronous write call still returned Good (exactly as any failed device write behaves since the #5 write-outcome self-correction). The rejection reason distinguishes the boot window (`"not primary (role unknown)"`) from a steady-state Secondary (`"not primary"`). Acks are dropped (Warning-logged when role-unknown) — and since review R1 m2 a native OPC UA Acknowledge is refused up front (`BadInvalidState`) in the same window: `DriverHostActor` publishes its Primary verdict to `IDriverPrimaryRoleSource` (the native-ack gate, and the MQTT Sparkplug primary-host claim) once at start and again on every cluster membership change, not only on redundancy snapshots, so the published verdict denies as soon as a driver peer is Up rather than sitting on the single-node "allow" seed until the first snapshot. The alerts emit is skipped (the Primary peer publishes the single fleet copy). The window closes on the first delivered snapshot.
Under warm/hot redundancy both cluster nodes run `ScriptedAlarmHostActor` and evaluate scripted alarms, keeping each node's address space and engine state warm for instant failover. To avoid duplicate rows on `/alerts` and duplicate historian writes, only the Primary node publishes externally: Under warm/hot redundancy both cluster nodes run `ScriptedAlarmHostActor` and evaluate scripted alarms, keeping each node's address space and engine state warm for instant failover. To avoid duplicate rows on `/alerts` and duplicate historian writes, only the Primary node publishes externally:
- **`alerts` topic emission** — `ScriptedAlarmHostActor` and `DriverHostActor.ForwardNativeAlarm` subscribe to the `redundancy-state` DPS topic and cache the local node's `RedundancyRole`, then gate the cluster `alerts` publish through `PrimaryGatePolicy` (table above). The OPC UA condition-node write and inbound ack/shelve command processing remain **ungated** on both nodes so the secondary is always ready to serve clients after a failover. - **`alerts` topic emission** — `ScriptedAlarmHostActor` and `DriverHostActor.ForwardNativeAlarm` subscribe to the `redundancy-state` DPS topic and cache the local node's `RedundancyRole`, then gate the cluster `alerts` publish through `PrimaryGatePolicy` (table above). The OPC UA condition-node write and inbound ack/shelve command processing remain **ungated** on both nodes so the secondary is always ready to serve clients after a failover.
- **`HistorianAdapterActor` historization** — likewise Primary-gated so alarm historization is exactly-once across all alarm sources. The actor subscribes to the `alerts` DPS topic and translates each `AlarmTransitionEvent` → `AlarmHistorianEvent` before enqueuing it on the sink; scripted alarms therefore historize exactly once regardless of cluster size. - **`HistorianAdapterActor` historization** — likewise Primary-gated so alarm historization is exactly-once across all alarm sources. The actor subscribes to the `alerts` DPS topic and translates each `AlarmTransitionEvent` → `AlarmHistorianEvent` before enqueuing it on the sink; scripted alarms therefore historize exactly once regardless of cluster size.
- **The alarm sink's DRAIN** — a second, independent gate on the same decision. The enqueue gate above is not sufficient any more: since Phase 2 the store-and-forward buffer lives in the replicated LocalDb, so the Secondary holds a full copy of the Primary's queued rows and an ungated drain there would re-deliver every event continuously. `LocalDbStoreAndForwardSink` takes a `drainGate` fed from `IRedundancyRoleView` — a singleton `DriverHostActor` publishes its `PrimaryGatePolicy` verdict to, because the drain runs on a timer and cannot receive `RedundancyStateChanged` itself. A gated-off node reports `HistorianDrainState.NotPrimary`. Delivery is **at-least-once across a failover** by design: rows the old Primary delivered just before dying may not have replicated their deletes yet. See [AlarmHistorian.md](AlarmHistorian.md). - **The alarm sink's DRAIN** — a second, independent gate on the same decision. The enqueue gate above is not sufficient any more: since Phase 2 the store-and-forward buffer lives in the replicated LocalDb, so the Secondary holds a full copy of the Primary's queued rows and an ungated drain there would re-deliver every event continuously. `LocalDbStoreAndForwardSink` takes a `drainGate` fed from `IRedundancyRoleView` — a singleton `DriverHostActor` publishes its `PrimaryGatePolicy` verdict to, because the drain runs on a timer and cannot receive `RedundancyStateChanged` itself. A gated-off node reports `HistorianDrainState.NotPrimary`. The gate applies only to a node whose queue a peer actually holds — `LocalDbReplicationPolicy.SharesRowsWithPeer`: `LocalDb:Replication:ApiKey` AND (`PeerAddress` OR `LocalDb:SyncListenPort` > 0); any other node always drains (review R1 m3). Delivery is **at-least-once across a failover** by design: rows the old Primary delivered just before dying may not have replicated their deletes yet. See [AlarmHistorian.md](AlarmHistorian.md).
- **Continuous-historization value CAPTURE** (Gitea #537) — a third gate on the same role rule, and the - **Continuous-historization value CAPTURE** (Gitea #537) — a third gate on the same role rule, and the
one that gates the *opposite* half of its pipeline. Both pair nodes serve the same address space and see one that gates the *opposite* half of its pipeline. Both pair nodes serve the same address space and see
@@ -622,7 +622,10 @@ and [AlarmHistorian.md](AlarmHistorian.md). Phase 1 caches the **deployed-config
artifact** (chunked). Its job is a single failure mode redundancy does not otherwise cover: a driver artifact** (chunked). Its job is a single failure mode redundancy does not otherwise cover: a driver
node restarting into a **central-SQL-Server outage** would come up with no configuration at all. With node restarting into a **central-SQL-Server outage** would come up with no configuration at all. With
the cache, it **boots from the last artifact it applied** instead, and logs a running-from-cache the cache, it **boots from the last artifact it applied** instead, and logs a running-from-cache
signal. signal. A cached artifact whose bytes are empty or malformed is **no answer** (#601): nothing is applied,
the node does **not** adopt that revision or claim running-from-cache, and it goes **Stale** (retrying the
ConfigDb) — so a later dispatch of that same revision really applies instead of being acked as already
current (review R1 m4). A `FetchAndCache` node lands Steady-with-no-revision instead, as for an empty cache.
- **What replicates.** The cache can *optionally* replicate to the node's redundant pair peer over - **What replicates.** The cache can *optionally* replicate to the node's redundant pair peer over
the LocalDb library's gRPC sync (HLC-stamped, last-writer-wins). Two tables replicate: the LocalDb library's gRPC sync (HLC-stamped, last-writer-wins). Two tables replicate:
@@ -92,7 +92,9 @@ Replication has two knobs, both under `LocalDb`:
> 2026-08-18 `LocalDbSetup.OnReady` installs the change-capture triggers only when the node is > 2026-08-18 `LocalDbSetup.OnReady` installs the change-capture triggers only when the node is
> configured to replicate — the predicate is **`PeerAddress` OR `ApiKey` non-blank** (the OR because > configured to replicate — the predicate is **`PeerAddress` OR `ApiKey` non-blank** (the OR because
> the passive node has no `PeerAddress`; `SyncListenPort` is deliberately not part of it, since a > the passive node has no `PeerAddress`; `SyncListenPort` is deliberately not part of it, since a
> port with no key is fail-closed and syncs with nobody). Consequences at enablement time: > port with no key is fail-closed and syncs with nobody). The alarm-historian drain gate uses the
> stricter `ApiKey` AND (`PeerAddress` OR `SyncListenPort > 0`) — so set the key **and** the port/peer
> together, or the node keeps draining its own queue. Consequences at enablement time:
> >
> - A configured node registers with **`baselineExistingRows: true`**, so the rows **already in the > - A configured node registers with **`baselineExistingRows: true`**, so the rows **already in the
> file** are seeded into the row-version ledger at the LWW floor (HLC `0`) and actually travel. > file** are seeded into the row-version ledger at the LWW floor (HLC `0`) and actually travel.
@@ -76,7 +76,7 @@ public sealed class ClusterRoleInfo : IClusterRoleInfo, IDisposable
} }
} }
SeedFromCurrentState(); SeedFrom(_cluster.State);
_subscriber = system.ActorOf(Props.Create(() => new SubscriberActor(this)), "clusterroleinfo-subscriber"); _subscriber = system.ActorOf(Props.Create(() => new SubscriberActor(this)), "clusterroleinfo-subscriber");
} }
@@ -121,12 +121,19 @@ public sealed class ClusterRoleInfo : IClusterRoleInfo, IDisposable
/// <summary>Occurs when the leader for a role changes.</summary> /// <summary>Occurs when the leader for a role changes.</summary>
public event EventHandler<RoleLeaderChangedEventArgs>? RoleLeaderChanged; public event EventHandler<RoleLeaderChangedEventArgs>? RoleLeaderChanged;
private void SeedFromCurrentState() /// <summary>Seeds the member/leader maps from a cluster-state snapshot (the constructor passes the live
/// <c>Cluster.State</c>; exposed for tests).</summary>
/// <param name="snapshot">The cluster-state snapshot.</param>
internal void SeedFrom(ClusterEvent.CurrentClusterState snapshot)
{ {
var snapshot = _cluster.State;
lock (_lock) lock (_lock)
{ {
foreach (var member in snapshot.Members) // Cluster-002 (review R1 m6): seed only USABLE members. A member already Leaving / Exiting / Down
// (or still Joining) at seed time is not one, and a departing member never fires a later
// MemberLeft / MemberExited / MemberDowned that HandleMemberEvent would drop it on — it would
// linger until MemberRemoved, the very delay the member-event handling exists to avoid. A
// Joining member is added by its MemberUp.
foreach (var member in snapshot.Members.Where(m => m.Status is MemberStatus.Up or MemberStatus.WeaklyUp))
foreach (var role in member.Roles) foreach (var role in member.Roles)
{ {
if (!_membersByRole.TryGetValue(role, out var set)) if (!_membersByRole.TryGetValue(role, out var set))
@@ -46,7 +46,7 @@ public static class LocalDbRegistration
/// Configuration key carrying the dedicated h2c sync listener's port. Zero or absent means /// Configuration key carrying the dedicated h2c sync listener's port. Zero or absent means
/// no listener is bound and the passive sync endpoint is not mapped. /// no listener is bound and the passive sync endpoint is not mapped.
/// </summary> /// </summary>
public const string SyncListenPortKey = "LocalDb:SyncListenPort"; public const string SyncListenPortKey = Runtime.LocalDbReplicationPolicy.SyncListenPortKey;
/// <summary> /// <summary>
/// The port the dedicated cleartext-HTTP/2 sync listener should bind, or <c>0</c> when the /// The port the dedicated cleartext-HTTP/2 sync listener should bind, or <c>0</c> when the
@@ -170,9 +170,9 @@ public static class LocalDbSetup
/// <returns><c>true</c> when this node is configured to replicate.</returns> /// <returns><c>true</c> when this node is configured to replicate.</returns>
public static bool ReplicationIsConfigured(IConfiguration configuration) public static bool ReplicationIsConfigured(IConfiguration configuration)
{ {
// Shared with the alarm-historian drain gate (#603): one predicate, so the nodes that capture // Defined beside the alarm-historian drain gate's stricter predicate (#603, review R1 m3): every node
// changes and the nodes that stand down for a peer can never disagree. // the gate treats as sharing rows also captures, so the two never disagree in the harmful direction.
return Runtime.LocalDbReplicationPolicy.IsConfigured(configuration); return Runtime.LocalDbReplicationPolicy.CapturesChanges(configuration);
} }
/// <summary>Opts every replicated table into change capture, baselining what is already there.</summary> /// <summary>Opts every replicated table into change capture, baselining what is already there.</summary>
+4 -1
View File
@@ -749,7 +749,10 @@ if (LogPipelineDiagnostics.CheckFileSinkGrowth(app.Configuration,
await app.RunAsync(); await app.RunAsync();
} }
catch (Exception ex) // HostAbortedException is how design-time tooling (EF `dotnet ef … --startup-project`, OpenAPI generators —
// HostFactoryResolver) stops Program once it has the built host. It is not a failure: letting it reach the
// Fatal + Environment.Exit(1) below would kill the tool's own process (#618, review R1 m5).
catch (Exception ex) when (ex is not HostAbortedException)
{ {
// Log.Logger is still the silent bootstrap logger when the failure happened before AddZbSerilog // Log.Logger is still the silent bootstrap logger when the failure happened before AddZbSerilog
// ran (e.g. EffectiveRoles.Resolve), so ALSO write to stderr — the one channel that always exists. // ran (e.g. EffectiveRoles.Resolve), so ALSO write to stderr — the one channel that always exists.
@@ -788,9 +788,91 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS
// Same rationale for the ScriptedAlarm host — the bootstrap-restore path Tells it // Same rationale for the ScriptedAlarm host — the bootstrap-restore path Tells it
// ApplyScriptedAlarms through the same PushDesiredSubscriptions pass. // ApplyScriptedAlarms through the same PushDesiredSubscriptions pass.
SpawnScriptedAlarmHost(); SpawnScriptedAlarmHost();
StartPrimaryVerdictPublishing();
Bootstrap(); 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> /// <summary>
/// Spawns the single <see cref="VirtualTagHostActor"/> child that owns the Equipment-namespace /// 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 /// 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 deploymentId = DeploymentId.Parse(cached.DeploymentId);
var revision = RevisionHash.Parse(cached.RevisionHash); var revision = RevisionHash.Parse(cached.RevisionHash);
_currentRevision = revision;
BecomeSteady();
// Restore drivers + address space + subscriptions from the cached bytes — no SQL read, no // 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 // 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. // 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( _log.Info(
"DriverHost {Node}: FetchAndCache boot — restored served state for deployment {Id} " + "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 deploymentId = DeploymentId.Parse(cached.DeploymentId);
var revision = RevisionHash.Parse(cached.RevisionHash); 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 // 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 — // Stale would have started does not run here, so recovery rides on the next dispatch —
// matching how a normally-booted node behaves. // matching how a normally-booted node behaves.
_currentRevision = revision; _currentRevision = revision;
BecomeSteady(); BecomeSteady();
ApplyCachedArtifact(deploymentId, cached.Artifact);
_isRunningFromCache = true; _isRunningFromCache = true;
_log.Warning( _log.Warning(
"DriverHost {Node}: RUNNING FROM CACHE — central ConfigDb is unreachable, so this node " + "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 /// <c>UpsertNodeDeploymentState</c> and <c>SendAck</c> for the same reason: both write to or
/// depend on central. /// depend on central.
/// </remarks> /// </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(); 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. // #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. // Keep last-known-good: no address-space rebuild and no subscription push from unusable bytes.
if (ReconcileDriversFromBlob(deploymentId, blob) is null) 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 // 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 // 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)); new ZB.MOM.WW.OtOpcUa.Runtime.OpcUa.OpcUaPublishActor.RebuildAddressSpace(correlation, deploymentId, blob));
PushDesiredSubscriptionsFromArtifact(deploymentId, blob); PushDesiredSubscriptionsFromArtifact(deploymentId, blob);
return true;
} }
/// <summary> /// <summary>
@@ -1170,6 +1275,9 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS
Receive<RouteNodeWrite>(HandleRouteNodeWrite); Receive<RouteNodeWrite>(HandleRouteNodeWrite);
Receive<RouteNativeAlarmAck>(HandleRouteNativeAlarmAck); Receive<RouteNativeAlarmAck>(HandleRouteNativeAlarmAck);
Receive<RedundancyStateChanged>(OnRedundancyStateChanged); 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). // 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). // 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>(_ => { }); Receive<ZB.MOM.WW.OtOpcUa.Runtime.Health.PeerOpcUaProbeActor.OpcUaProbeResult>(_ => { });
@@ -1228,6 +1336,9 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS
Receive<RouteNodeWrite>(HandleRouteNodeWrite); Receive<RouteNodeWrite>(HandleRouteNodeWrite);
Receive<RouteNativeAlarmAck>(HandleRouteNativeAlarmAck); Receive<RouteNativeAlarmAck>(HandleRouteNativeAlarmAck);
Receive<RedundancyStateChanged>(OnRedundancyStateChanged); 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). // 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). // 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>(_ => { }); 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 // 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 // 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. // harm, so an unknown role with a driver peer present must deny.
_primaryRoleSource?.PublishPrimaryRole(ShouldServiceAsPrimary()); PublishPrimaryVerdict();
} }
private void Stale() private void Stale()
@@ -1998,6 +2109,9 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS
Receive<ReprovisionDeferred>(m => ReprovisionChild(m.Request, m.Child)); Receive<ReprovisionDeferred>(m => ReprovisionChild(m.Request, m.Child));
Receive<ReconnectDriver>(HandleReconnectDriver); Receive<ReconnectDriver>(HandleReconnectDriver);
Receive<RedundancyStateChanged>(OnRedundancyStateChanged); 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). // 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). // 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>(_ => { }); Receive<ZB.MOM.WW.OtOpcUa.Runtime.Health.PeerOpcUaProbeActor.OpcUaProbeResult>(_ => { });
@@ -3233,7 +3347,10 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS
Context.Stop(entry.Actor); Context.Stop(entry.Actor);
_children.Remove(driverInstanceId); _children.Remove(driverInstanceId);
// #609: the instance left the plan (StopChild is only reached via plan.ToStop), so its cached // #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); _telemetryHub?.Evict(driverInstanceId);
_log.Info("DriverHost {Node}: stopped driver child {Id}", _localNode, driverInstanceId); _log.Info("DriverHost {Node}: stopped driver child {Id}", _localNode, driverInstanceId);
} }
@@ -3,26 +3,36 @@ using Microsoft.Extensions.Configuration;
namespace ZB.MOM.WW.OtOpcUa.Runtime; namespace ZB.MOM.WW.OtOpcUa.Runtime;
/// <summary> /// <summary>
/// The ONE definition of "this node's LocalDb is replicated to a peer" (#603). Two consumers must agree /// The definitions of "this node's LocalDb is replicated to a peer" (#603) — one place, two deliberately
/// on it, and used to carry separate predicates that diverged: the Host's table-registration step /// different predicates for the two consumers: the Host's table-registration step
/// (<c>LocalDbSetup.ReplicationIsConfigured</c> - which nodes capture changes) and the alarm-historian /// (<c>LocalDbSetup.ReplicationIsConfigured</c> → <see cref="CapturesChanges"/>: which nodes capture
/// drain gate (<c>AddAlarmHistorian</c> - which nodes stand down in favour of a peer). Lives in Runtime /// changes) and the alarm-historian drain gate (<c>AddAlarmHistorian</c> → <see cref="SharesRowsWithPeer"/>:
/// because Host references Runtime and not the reverse. /// which nodes stand down in favour of a peer). Lives in Runtime because Host references Runtime and not
/// the reverse.
/// </summary> /// </summary>
/// <remarks> /// <remarks>
/// <para> /// <para>
/// <b>Predicate: <c>LocalDb:Replication:PeerAddress</c> OR <c>LocalDb:Replication:ApiKey</c> is /// <b>Why two predicates (review R1 m3).</b> The two answers cost differently when wrong. Registering a
/// non-blank.</b> The OR is required: only the dialling half of a pair sets <c>PeerAddress</c>, while /// node that has no working sync path only costs capture triggers. Gating the drain of a node that
/// the passive half carries the shared <c>ApiKey</c> (plus a listen port) and nothing else. /// shares its queue with nobody parks its alarm events behind a Primary that does not hold them —
/// undelivered until the node is promoted, and drop-oldest at <c>Capacity</c> meanwhile. So
/// registration stays permissive and the gate is strict. Every node the gate treats as sharing also
/// captures, so the two never disagree in the harmful direction.
/// </para> /// </para>
/// <para> /// <para>
/// <b>Why the drain gate takes the same predicate (it previously used PeerAddress OR /// <b>Registration: <c>LocalDb:Replication:PeerAddress</c> OR <c>LocalDb:Replication:ApiKey</c>.</b>
/// <c>SyncListenPort</c>).</b> The gate's question is "does a peer hold the same rows?". The passive /// The OR is required: only the dialling half of a pair sets <c>PeerAddress</c>, while the passive
/// sync endpoint is fail-closed on the key, so a node with a listen port but no <c>ApiKey</c> rejects /// half carries the shared <c>ApiKey</c> (plus a listen port) and nothing else. Keying on
/// every inbound stream and shares its queue with nobody - gating it would park its alarm events /// <c>PeerAddress</c> alone would strip capture from every passive node.
/// behind a Primary that cannot deliver them. Conversely an <c>ApiKey</c>-only passive node IS /// </para>
/// registered and does receive its peer's rows, so it must be gated or both halves drain. The port /// <para>
/// is not evidence of a peer; the key is. /// <b>Drain gate: <c>ApiKey</c> AND (<c>PeerAddress</c> OR <c>LocalDb:SyncListenPort</c> &gt; 0).</b> A
/// peer holds this node's rows only over a working sync path: a dial (<c>PeerAddress</c>) or a
/// listener (<c>SyncListenPort</c>), each authenticated by the key. Both auth directions fail closed,
/// so a path without the key shares nothing, and a key with neither path (e.g. a fleet-wide
/// secret/env <c>ApiKey</c> on a node with no peer) shares nothing either. Every docker-dev shape
/// gives the same answer under both predicates: site-a-1 (PeerAddress + ApiKey + port) and site-a-2
/// (ApiKey + port) are gated; central and site-b (neither) are not.
/// </para> /// </para>
/// </remarks> /// </remarks>
public static class LocalDbReplicationPolicy public static class LocalDbReplicationPolicy
@@ -30,14 +40,31 @@ public static class LocalDbReplicationPolicy
/// <summary>Configuration section holding the LocalDb replication options.</summary> /// <summary>Configuration section holding the LocalDb replication options.</summary>
public const string ReplicationSectionPath = "LocalDb:Replication"; public const string ReplicationSectionPath = "LocalDb:Replication";
/// <summary>Whether this node's LocalDb replicates to a pair peer.</summary> /// <summary>The dedicated passive-sync listener port key (same key as <c>LocalDbRegistration.SyncListenPortKey</c>).</summary>
public const string SyncListenPortKey = "LocalDb:SyncListenPort";
/// <summary>Whether this node registers its replicated tables for change capture (table registration).</summary>
/// <param name="configuration">The application configuration.</param> /// <param name="configuration">The application configuration.</param>
/// <returns><c>true</c> when <c>PeerAddress</c> or <c>ApiKey</c> is set under the replication section.</returns> /// <returns><c>true</c> when <c>PeerAddress</c> or <c>ApiKey</c> is set under the replication section.</returns>
public static bool IsConfigured(IConfiguration configuration) public static bool CapturesChanges(IConfiguration configuration)
{ {
ArgumentNullException.ThrowIfNull(configuration); ArgumentNullException.ThrowIfNull(configuration);
var section = configuration.GetSection(ReplicationSectionPath); var section = configuration.GetSection(ReplicationSectionPath);
return !string.IsNullOrWhiteSpace(section["PeerAddress"]) return !string.IsNullOrWhiteSpace(section["PeerAddress"])
|| !string.IsNullOrWhiteSpace(section["ApiKey"]); || !string.IsNullOrWhiteSpace(section["ApiKey"]);
} }
/// <summary>Whether a pair peer actually holds this node's replicated rows (the alarm drain gate).</summary>
/// <param name="configuration">The application configuration.</param>
/// <returns><c>true</c> when <c>ApiKey</c> is set AND either <c>PeerAddress</c> is set or
/// <c>LocalDb:SyncListenPort</c> is a positive port.</returns>
public static bool SharesRowsWithPeer(IConfiguration configuration)
{
ArgumentNullException.ThrowIfNull(configuration);
var section = configuration.GetSection(ReplicationSectionPath);
if (string.IsNullOrWhiteSpace(section["ApiKey"])) return false;
var dials = !string.IsNullOrWhiteSpace(section["PeerAddress"]);
var listens = int.TryParse(configuration[SyncListenPortKey], out var port) && port > 0;
return dials || listens;
}
} }
@@ -177,11 +177,13 @@ public static class ServiceCollectionExtensions
// possibly deliver its events. Scoping the gate to "is my queue actually shared?" is what // possibly deliver its events. Scoping the gate to "is my queue actually shared?" is what
// keeps the two scopes from disagreeing. // keeps the two scopes from disagreeing.
// BOTH halves of a pair share the queue, but only one of them dials: the initiator sets // 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 and waits. The predicate is // Replication:PeerAddress, its partner carries the shared ApiKey plus SyncListenPort and waits.
// LocalDbReplicationPolicy - the same one LocalDbSetup uses to decide which nodes capture // The predicate is LocalDbReplicationPolicy.SharesRowsWithPeer: ApiKey AND (PeerAddress OR
// changes at all. It deliberately does NOT count SyncListenPort: the passive endpoint is // SyncListenPort > 0) - an authenticated sync path in either direction. Both auth directions
// fail-closed on the key, so a port with no ApiKey shares nothing and must keep draining. // fail closed, so a path without the key, or a key with no path (a fleet-wide secret on an
var replicated = LocalDbReplicationPolicy.IsConfigured(configuration); // 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) if (!replicated)
{ {
@@ -97,4 +97,30 @@ public sealed class ClusterRoleInfoMemberEventTests
return Task.CompletedTask; return Task.CompletedTask;
}); });
} }
/// <summary>
/// Review R1 m6: the initial seed lists only Up / WeaklyUp members. A member already departing at seed
/// time never fires a later Left/Exited/Downed event, so seeding it would list it until MemberRemoved.
/// The Up and WeaklyUp members are the positive control (the seed does list members).
/// </summary>
[Fact]
public async Task Initial_seed_lists_only_up_and_weakly_up_members()
{
await WithInfoAsync(info =>
{
var members = ImmutableSortedSet.Create<Member>(
NewMember(21001, MemberStatus.Up),
NewMember(21002, MemberStatus.WeaklyUp),
NewMember(21003, MemberStatus.Leaving),
NewMember(21004, MemberStatus.Exiting),
NewMember(21005, MemberStatus.Down),
NewMember(21006, MemberStatus.Joining));
info.SeedFrom(new ClusterEvent.CurrentClusterState().Copy(members: members));
info.MembersWithRole("driver").Select(n => n.Value).OrderBy(v => v).ShouldBe(
["127.0.0.1:21001", "127.0.0.1:21002"]);
return Task.CompletedTask;
});
}
} }
@@ -102,8 +102,14 @@ public sealed class DriverHostActorBootFromCacheTests : RuntimeActorTestBase
// Barrier: diagnostics are served only after PreStart's boot work ran, so the absence below is // Barrier: diagnostics are served only after PreStart's boot work ran, so the absence below is
// asserted after the boot path had its chance to Tell (positive control is the valid-bytes test). // asserted after the boot path had its chance to Tell (positive control is the valid-bytes test).
AskDiagnostics(actor); var snapshot = AskDiagnostics(actor);
publishProbe.ExpectNoMsg(TimeSpan.FromMilliseconds(500)); publishProbe.ExpectNoMsg(TimeSpan.FromMilliseconds(500));
// Review R1 m4: nothing was applied, so the node is NOT running from cache and has NOT adopted the
// cached revision (Stale with no configuration, retrying the ConfigDb) — the valid-bytes test above
// is the control that both flip when the bytes are usable.
snapshot.RunningFromCache.ShouldBeFalse();
snapshot.CurrentRevision.ShouldBeNull();
} }
[Fact] [Fact]
@@ -111,6 +111,39 @@ public sealed class DriverHostActorFetchAndCacheBootstrapTests : RuntimeActorTes
fetcher.Calls.ShouldBe(1); fetcher.Calls.ShouldBe(1);
} }
/// <summary>
/// #601 / review R1 m4 — a cache row whose bytes are malformed applied nothing, so its revision must not
/// become current: the boot lands Steady-with-no-revision (as for an empty cache), and a re-dispatch of
/// the SAME revision really applies (fetches) instead of being acked by the "already at this rev"
/// short-circuit. <see cref="WithCachedArtifact_RestoresServedState_WithoutReadingCentralSql"/> is the
/// control: valid bytes do adopt the revision.
/// </summary>
[Fact]
public void WithMalformedCachedBytes_TheRevisionIsNotAdopted_AndARedispatchOfItActuallyApplies()
{
var cached = new CachedDeploymentArtifact(
DeploymentId.NewId().ToString(), CachedRev.Value,
System.Text.Encoding.UTF8.GetBytes("{\"DriverInstances\": ["), DateTimeOffset.UtcNow.AddHours(-1));
var fetcher = new RecordingFetcher(Artifact());
var coordinator = CreateTestProbe();
var actor = Sys.ActorOf(DriverHostActor.Props(
new ThrowingDbFactory(), TestNode, coordinator.Ref,
localRoles: new HashSet<string> { "driver" },
deploymentArtifactCache: new StubArtifactCache(cached),
fetchAndCacheMode: true,
artifactFetcher: fetcher));
var snapshot = AskDiagnostics(actor);
snapshot.CurrentRevision.ShouldBeNull();
snapshot.RunningFromCache.ShouldBeFalse();
actor.Tell(new DispatchDeployment(DeploymentId.Parse(cached.DeploymentId), CachedRev, CorrelationId.NewId()));
coordinator.ExpectMsg<ApplyAck>(TimeSpan.FromSeconds(5)).Outcome.ShouldBe(ApplyAckOutcome.Applied);
fetcher.Calls.ShouldBe(1);
AskDiagnostics(actor).CurrentRevision.ShouldBe(CachedRev);
}
private NodeDiagnosticsSnapshot AskDiagnostics(IActorRef actor) private NodeDiagnosticsSnapshot AskDiagnostics(IActorRef actor)
{ {
var probe = CreateTestProbe(); var probe = CreateTestProbe();
@@ -0,0 +1,109 @@
using Akka.Actor;
using Shouldly;
using Xunit;
using ZB.MOM.WW.OtOpcUa.Commons.Messages.Redundancy;
using ZB.MOM.WW.OtOpcUa.Commons.Types;
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
using ZB.MOM.WW.OtOpcUa.Runtime.Drivers;
using ZB.MOM.WW.OtOpcUa.Runtime.Tests.Harness;
namespace ZB.MOM.WW.OtOpcUa.Runtime.Tests.Drivers;
/// <summary>
/// Review R1 m2 (Gitea #597) — the Primary data-plane verdict published to <see cref="IDriverPrimaryRoleSource"/>
/// (read by the native-ack gate and the Sparkplug primary-host claim) mirrors the host's own live gate from the
/// moment the host starts, not only from the first redundancy snapshot.
/// </summary>
/// <remarks>
/// <c>RedundancyRoleView</c> is seeded with the single-node verdict (allow). Before this fix nothing published to
/// it until a snapshot arrived, so on a pair both nodes accepted a native ack through the whole boot window while
/// <c>DriverHostActor</c> — which evaluates the driver member count live — denied the upstream route on the
/// non-Primary. Assertions are on what was PUBLISHED: the seed already reads <c>true</c>, so a current-value
/// assertion of the single-node case would pass with no publish at all.
/// </remarks>
public sealed class DriverHostActorPrimaryRoleSourceTests : RuntimeActorTestBase
{
private static readonly NodeId TestNode = NodeId.Parse("driver-primary-source-test");
private static readonly TimeSpan Timeout = TimeSpan.FromSeconds(5);
/// <summary>On a pair, an unknown role publishes DENY at start — before any snapshot. The later Primary snapshot
/// is the positive control: the source is wired, and the deny was the verdict, not a missing publish.</summary>
[Fact]
public void A_paired_host_publishes_deny_before_the_first_snapshot()
{
var source = new RecordingPrimaryRoleSource();
var host = SpawnHost(source, () => 2);
AwaitAssert(() => source.Published.ShouldBe([false]), duration: Timeout);
host.Tell(Snapshot(RedundancyRole.Primary));
AwaitAssert(() => source.Published.ShouldBe([false, true]), duration: Timeout);
}
/// <summary>A single-driver node keeps the allow posture — it is published, not merely seeded.</summary>
[Fact]
public void A_single_node_host_publishes_allow_at_start()
{
var source = new RecordingPrimaryRoleSource();
SpawnHost(source, () => 1);
AwaitAssert(() => source.Published.ShouldBe([true]), duration: Timeout);
}
/// <summary>A driver peer coming Up while the role is still unknown flips the published verdict to deny, with no
/// snapshot involved — the membership re-publish path.</summary>
[Fact]
public void A_peer_joining_before_any_snapshot_flips_the_published_verdict_to_deny()
{
var source = new RecordingPrimaryRoleSource();
var members = 1;
var host = SpawnHost(source, () => Volatile.Read(ref members));
AwaitAssert(() => source.Published.ShouldBe([true]), duration: Timeout);
Volatile.Write(ref members, 2);
host.Tell(DriverHostActor.RepublishPrimaryVerdict.Instance);
AwaitAssert(() => source.Published.ShouldBe([true, false]), duration: Timeout);
}
private IActorRef SpawnHost(IDriverPrimaryRoleSource source, Func<int> driverMemberCount) =>
Sys.ActorOf(DriverHostActor.Props(
NewInMemoryDbFactory(),
TestNode,
ackRouter: null,
driverMemberCountProvider: driverMemberCount,
primaryRoleSource: source));
private static RedundancyStateChanged Snapshot(RedundancyRole role) =>
new(
[
new NodeRedundancyState(TestNode, role,
IsClusterLeader: role == RedundancyRole.Primary,
IsDriverPrimary: role == RedundancyRole.Primary,
AsOfUtc: DateTime.UtcNow),
],
CorrelationId.NewId());
/// <summary>Remembers every published verdict, so tests assert on the actor's action, not on the seed.</summary>
private sealed class RecordingPrimaryRoleSource : IDriverPrimaryRoleSource
{
private readonly List<bool> _published = [];
private readonly Lock _gate = new();
/// <summary>Every value published so far, in order.</summary>
public IReadOnlyList<bool> Published
{
get { lock (_gate) return _published.ToArray(); }
}
public bool ShouldActAsPrimary
{
get { lock (_gate) return _published.Count == 0 || _published[^1]; }
}
public void PublishPrimaryRole(bool shouldActAsPrimary)
{
lock (_gate) _published.Add(shouldActAsPrimary);
}
}
}
@@ -109,7 +109,8 @@ public sealed class AlarmHistorianRegistrationTests
/// <remarks> /// <remarks>
/// <para> /// <para>
/// Standing down is only safe because a peer holds the same rows and will send them /// Standing down is only safe because a peer holds the same rows and will send them
/// instead. Without <c>LocalDb:Replication:PeerAddress</c> there is no such peer: the rows /// instead. Without an authenticated sync path (<c>ApiKey</c> plus a dial or a listener —
/// <c>LocalDbReplicationPolicy.SharesRowsWithPeer</c>) there is no such peer: the rows
/// are here and nowhere else, so any deferral means nobody ever delivers them. /// are here and nowhere else, so any deferral means nobody ever delivers them.
/// </para> /// </para>
/// <para> /// <para>
@@ -124,14 +125,18 @@ public sealed class AlarmHistorianRegistrationTests
// Neither key: an unpaired node. Its rows exist nowhere else. // Neither key: an unpaired node. Its rows exist nowhere else.
[InlineData(null, null, null, true)] [InlineData(null, null, null, true)]
// The dialing half of a pair — gated, because its partner holds the same rows. // The dialing half of a pair — gated, because its partner holds the same rows.
[InlineData("http://peer:9001", null, null, false)] [InlineData("http://peer:9001", null, "k", false)]
// The LISTENING half. Same shared queue, but it never dials, so a PeerAddress-only test would // The LISTENING half. Same shared queue, but it never dials, so a PeerAddress-only test would
// wrongly class it as unpaired and leave it permanently ungated. It is identified by the shared // wrongly class it as unpaired and leave it permanently ungated. Key + listen port = a peer.
// ApiKey (the same predicate LocalDbSetup registers tables on), not by its listen port.
[InlineData(null, "9001", "k", false)] [InlineData(null, "9001", "k", false)]
// A listen port with NO key: the fail-closed passive endpoint rejects every inbound stream, so the // A listen port with NO key: the fail-closed passive endpoint rejects every inbound stream, so the
// queue is shared with nobody and the node must keep draining (#603: this used to be gated). // queue is shared with nobody and the node must keep draining (#603: this used to be gated).
[InlineData(null, "9001", null, true)] [InlineData(null, "9001", null, true)]
// A dial with NO key: the peer's fail-closed endpoint rejects it — shares nothing, keeps draining.
[InlineData("http://peer:9001", null, null, true)]
// A key with NEITHER path (review R1 m3: e.g. a fleet-wide secret on an unpaired node) — no listener,
// no dial, so no peer holds these rows. Registered for capture, but must keep draining.
[InlineData(null, null, "k", true)]
public async Task Only_an_unreplicated_node_ignores_the_role_view( public async Task Only_an_unreplicated_node_ignores_the_role_view(
string? peerAddress, string? syncListenPort, string? apiKey, bool expectDrain) string? peerAddress, string? syncListenPort, string? apiKey, bool expectDrain)
{ {
@@ -4,29 +4,64 @@ using Xunit;
namespace ZB.MOM.WW.OtOpcUa.Runtime.Tests.Historian; namespace ZB.MOM.WW.OtOpcUa.Runtime.Tests.Historian;
/// <summary>#603: the single "is this node's LocalDb replicated" predicate shared by table registration and the alarm drain gate.</summary> /// <summary>
/// #603 + review R1 m3: the two "is this node's LocalDb replicated" predicates — permissive for table
/// registration (<see cref="LocalDbReplicationPolicy.CapturesChanges"/>), strict for the alarm drain gate
/// (<see cref="LocalDbReplicationPolicy.SharesRowsWithPeer"/>).
/// </summary>
public sealed class LocalDbReplicationPolicyTests public sealed class LocalDbReplicationPolicyTests
{ {
private const string Peer = "LocalDb:Replication:PeerAddress";
private const string Key = "LocalDb:Replication:ApiKey";
private const string Port = "LocalDb:SyncListenPort";
private static IConfiguration Config(params (string Key, string? Value)[] pairs) => private static IConfiguration Config(params (string Key, string? Value)[] pairs) =>
new ConfigurationBuilder() new ConfigurationBuilder()
.AddInMemoryCollection(pairs.Select(p => new KeyValuePair<string, string?>(p.Key, p.Value))) .AddInMemoryCollection(pairs.Select(p => new KeyValuePair<string, string?>(p.Key, p.Value)))
.Build(); .Build();
/// <summary>PeerAddress OR ApiKey counts; the listen port alone and blank values do not.</summary> /// <summary>Registration: PeerAddress OR ApiKey counts; the listen port alone and blank values do not.</summary>
[Theory] [Theory]
[InlineData("LocalDb:Replication:PeerAddress", "http://peer:9001", true)] [InlineData(Peer, "http://peer:9001", true)]
[InlineData("LocalDb:Replication:ApiKey", "k", true)] [InlineData(Key, "k", true)]
[InlineData("LocalDb:SyncListenPort", "9001", false)] [InlineData(Port, "9001", false)]
[InlineData("LocalDb:Replication:ApiKey", " ", false)] [InlineData(Key, " ", false)]
public void IsConfigured_follows_the_PeerAddress_or_ApiKey_rule(string key, string value, bool expected) public void CapturesChanges_follows_the_PeerAddress_or_ApiKey_rule(string key, string value, bool expected)
{ {
LocalDbReplicationPolicy.IsConfigured(Config((key, value))).ShouldBe(expected); LocalDbReplicationPolicy.CapturesChanges(Config((key, value))).ShouldBe(expected);
} }
/// <summary>No replication keys at all is an unpaired node.</summary> /// <summary>Drain gate: ApiKey AND (PeerAddress OR SyncListenPort &gt; 0).</summary>
[Fact] [Theory]
public void IsConfigured_is_false_for_an_empty_configuration() [InlineData("http://peer:9001", "k", null, true)] // dialling half of a pair
[InlineData(null, "k", "9001", true)] // listening half of a pair
[InlineData("http://peer:9001", "k", "9001", true)] // both
[InlineData(null, "k", null, false)] // key, no path: shares nothing (review R1 m3)
[InlineData(null, "k", "0", false)] // port 0 = no listener
[InlineData(null, "k", "nope", false)] // unparseable port = no listener
[InlineData("http://peer:9001", null, "9001", false)] // paths, no key: both auth directions fail closed
[InlineData(null, " ", "9001", false)] // blank key
public void SharesRowsWithPeer_needs_a_key_and_a_sync_path(string? peer, string? key, string? port, bool expected)
{ {
LocalDbReplicationPolicy.IsConfigured(Config()).ShouldBeFalse(); LocalDbReplicationPolicy.SharesRowsWithPeer(Config((Peer, peer), (Key, key), (Port, port))).ShouldBe(expected);
}
/// <summary>Every node the gate treats as sharing rows also captures — never a gated node with no capture.</summary>
[Theory]
[InlineData("http://peer:9001", "k", null)]
[InlineData(null, "k", "9001")]
public void A_node_sharing_rows_always_captures(string? peer, string? key, string? port)
{
var config = Config((Peer, peer), (Key, key), (Port, port));
LocalDbReplicationPolicy.SharesRowsWithPeer(config).ShouldBeTrue();
LocalDbReplicationPolicy.CapturesChanges(config).ShouldBeTrue();
}
/// <summary>No replication keys at all is an unpaired node under both predicates.</summary>
[Fact]
public void Both_predicates_are_false_for_an_empty_configuration()
{
LocalDbReplicationPolicy.CapturesChanges(Config()).ShouldBeFalse();
LocalDbReplicationPolicy.SharesRowsWithPeer(Config()).ShouldBeFalse();
} }
} }