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
@@ -93,6 +93,16 @@ Two postures are deliberate:
|
||||
- **Unset (the default) means drain.** So does an unpublished view. A deployment
|
||||
that runs no redundancy never publishes a role, and defaulting closed would
|
||||
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
|
||||
fault would put a pair into dual delivery exactly when its redundancy signal is
|
||||
sick.
|
||||
|
||||
@@ -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. |
|
||||
| `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).
|
||||
|
||||
@@ -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
|
||||
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
|
||||
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
|
||||
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
|
||||
|
||||
+6
-3
@@ -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.
|
||||
|
||||
**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:
|
||||
|
||||
- **`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.
|
||||
- **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
|
||||
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
|
||||
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
|
||||
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
|
||||
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
|
||||
> 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
|
||||
> 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
|
||||
> 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");
|
||||
}
|
||||
|
||||
@@ -121,12 +121,19 @@ public sealed class ClusterRoleInfo : IClusterRoleInfo, IDisposable
|
||||
/// <summary>Occurs when the leader for a role changes.</summary>
|
||||
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)
|
||||
{
|
||||
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)
|
||||
{
|
||||
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
|
||||
/// no listener is bound and the passive sync endpoint is not mapped.
|
||||
/// </summary>
|
||||
public const string SyncListenPortKey = "LocalDb:SyncListenPort";
|
||||
public const string SyncListenPortKey = Runtime.LocalDbReplicationPolicy.SyncListenPortKey;
|
||||
|
||||
/// <summary>
|
||||
/// 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>
|
||||
public static bool ReplicationIsConfigured(IConfiguration configuration)
|
||||
{
|
||||
// Shared with the alarm-historian drain gate (#603): one predicate, so the nodes that capture
|
||||
// changes and the nodes that stand down for a peer can never disagree.
|
||||
return Runtime.LocalDbReplicationPolicy.IsConfigured(configuration);
|
||||
// Defined beside the alarm-historian drain gate's stricter predicate (#603, review R1 m3): every node
|
||||
// the gate treats as sharing rows also captures, so the two never disagree in the harmful direction.
|
||||
return Runtime.LocalDbReplicationPolicy.CapturesChanges(configuration);
|
||||
}
|
||||
|
||||
/// <summary>Opts every replicated table into change capture, baselining what is already there.</summary>
|
||||
|
||||
@@ -749,7 +749,10 @@ if (LogPipelineDiagnostics.CheckFileSinkGrowth(app.Configuration,
|
||||
|
||||
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
|
||||
// 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
|
||||
// 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);
|
||||
}
|
||||
|
||||
@@ -3,26 +3,36 @@ using Microsoft.Extensions.Configuration;
|
||||
namespace ZB.MOM.WW.OtOpcUa.Runtime;
|
||||
|
||||
/// <summary>
|
||||
/// The ONE definition of "this node's LocalDb is replicated to a peer" (#603). Two consumers must agree
|
||||
/// on it, and used to carry separate predicates that diverged: the Host's table-registration step
|
||||
/// (<c>LocalDbSetup.ReplicationIsConfigured</c> - which nodes capture changes) and the alarm-historian
|
||||
/// drain gate (<c>AddAlarmHistorian</c> - which nodes stand down in favour of a peer). Lives in Runtime
|
||||
/// because Host references Runtime and not the reverse.
|
||||
/// The definitions of "this node's LocalDb is replicated to a peer" (#603) — one place, two deliberately
|
||||
/// different predicates for the two consumers: the Host's table-registration step
|
||||
/// (<c>LocalDbSetup.ReplicationIsConfigured</c> → <see cref="CapturesChanges"/>: which nodes capture
|
||||
/// changes) and the alarm-historian drain gate (<c>AddAlarmHistorian</c> → <see cref="SharesRowsWithPeer"/>:
|
||||
/// which nodes stand down in favour of a peer). Lives in Runtime because Host references Runtime and not
|
||||
/// the reverse.
|
||||
/// </summary>
|
||||
/// <remarks>
|
||||
/// <para>
|
||||
/// <b>Predicate: <c>LocalDb:Replication:PeerAddress</c> OR <c>LocalDb:Replication:ApiKey</c> is
|
||||
/// non-blank.</b> The OR is required: only the dialling half of a pair sets <c>PeerAddress</c>, while
|
||||
/// the passive half carries the shared <c>ApiKey</c> (plus a listen port) and nothing else.
|
||||
/// <b>Why two predicates (review R1 m3).</b> The two answers cost differently when wrong. Registering a
|
||||
/// node that has no working sync path only costs capture triggers. Gating the drain of a node that
|
||||
/// 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>
|
||||
/// <b>Why the drain gate takes the same predicate (it previously used PeerAddress OR
|
||||
/// <c>SyncListenPort</c>).</b> The gate's question is "does a peer hold the same rows?". The passive
|
||||
/// sync endpoint is fail-closed on the key, so a node with a listen port but no <c>ApiKey</c> rejects
|
||||
/// every inbound stream and shares its queue with nobody - gating it would park its alarm events
|
||||
/// behind a Primary that cannot deliver them. Conversely an <c>ApiKey</c>-only passive node IS
|
||||
/// registered and does receive its peer's rows, so it must be gated or both halves drain. The port
|
||||
/// is not evidence of a peer; the key is.
|
||||
/// <b>Registration: <c>LocalDb:Replication:PeerAddress</c> OR <c>LocalDb:Replication:ApiKey</c>.</b>
|
||||
/// The OR is required: only the dialling half of a pair sets <c>PeerAddress</c>, while the passive
|
||||
/// half carries the shared <c>ApiKey</c> (plus a listen port) and nothing else. Keying on
|
||||
/// <c>PeerAddress</c> alone would strip capture from every passive node.
|
||||
/// </para>
|
||||
/// <para>
|
||||
/// <b>Drain gate: <c>ApiKey</c> AND (<c>PeerAddress</c> OR <c>LocalDb:SyncListenPort</c> > 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>
|
||||
/// </remarks>
|
||||
public static class LocalDbReplicationPolicy
|
||||
@@ -30,14 +40,31 @@ public static class LocalDbReplicationPolicy
|
||||
/// <summary>Configuration section holding the LocalDb replication options.</summary>
|
||||
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>
|
||||
/// <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);
|
||||
var section = configuration.GetSection(ReplicationSectionPath);
|
||||
return !string.IsNullOrWhiteSpace(section["PeerAddress"])
|
||||
|| !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
|
||||
// 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 and waits. The predicate is
|
||||
// LocalDbReplicationPolicy - the same one LocalDbSetup uses to decide which nodes capture
|
||||
// changes at all. It deliberately does NOT count SyncListenPort: the passive endpoint is
|
||||
// fail-closed on the key, so a port with no ApiKey shares nothing and must keep draining.
|
||||
var replicated = LocalDbReplicationPolicy.IsConfigured(configuration);
|
||||
// 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)
|
||||
{
|
||||
|
||||
@@ -97,4 +97,30 @@ public sealed class ClusterRoleInfoMemberEventTests
|
||||
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;
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
+7
-1
@@ -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
|
||||
// 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));
|
||||
|
||||
// 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]
|
||||
|
||||
+33
@@ -111,6 +111,39 @@ public sealed class DriverHostActorFetchAndCacheBootstrapTests : RuntimeActorTes
|
||||
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)
|
||||
{
|
||||
var probe = CreateTestProbe();
|
||||
|
||||
+109
@@ -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);
|
||||
}
|
||||
}
|
||||
}
|
||||
+9
-4
@@ -109,7 +109,8 @@ public sealed class AlarmHistorianRegistrationTests
|
||||
/// <remarks>
|
||||
/// <para>
|
||||
/// 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.
|
||||
/// </para>
|
||||
/// <para>
|
||||
@@ -124,14 +125,18 @@ public sealed class AlarmHistorianRegistrationTests
|
||||
// Neither key: an unpaired node. Its rows exist nowhere else.
|
||||
[InlineData(null, null, null, true)]
|
||||
// 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
|
||||
// wrongly class it as unpaired and leave it permanently ungated. It is identified by the shared
|
||||
// ApiKey (the same predicate LocalDbSetup registers tables on), not by its listen port.
|
||||
// wrongly class it as unpaired and leave it permanently ungated. Key + listen port = a peer.
|
||||
[InlineData(null, "9001", "k", false)]
|
||||
// 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).
|
||||
[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(
|
||||
string? peerAddress, string? syncListenPort, string? apiKey, bool expectDrain)
|
||||
{
|
||||
|
||||
+47
-12
@@ -4,29 +4,64 @@ using Xunit;
|
||||
|
||||
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
|
||||
{
|
||||
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) =>
|
||||
new ConfigurationBuilder()
|
||||
.AddInMemoryCollection(pairs.Select(p => new KeyValuePair<string, string?>(p.Key, p.Value)))
|
||||
.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]
|
||||
[InlineData("LocalDb:Replication:PeerAddress", "http://peer:9001", true)]
|
||||
[InlineData("LocalDb:Replication:ApiKey", "k", true)]
|
||||
[InlineData("LocalDb:SyncListenPort", "9001", false)]
|
||||
[InlineData("LocalDb:Replication:ApiKey", " ", false)]
|
||||
public void IsConfigured_follows_the_PeerAddress_or_ApiKey_rule(string key, string value, bool expected)
|
||||
[InlineData(Peer, "http://peer:9001", true)]
|
||||
[InlineData(Key, "k", true)]
|
||||
[InlineData(Port, "9001", false)]
|
||||
[InlineData(Key, " ", false)]
|
||||
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>
|
||||
[Fact]
|
||||
public void IsConfigured_is_false_for_an_empty_configuration()
|
||||
/// <summary>Drain gate: ApiKey AND (PeerAddress OR SyncListenPort > 0).</summary>
|
||||
[Theory]
|
||||
[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();
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user