diff --git a/src/Core/ZB.MOM.WW.OtOpcUa.Cluster/Redundancy/ServiceLevelCalculator.cs b/src/Core/ZB.MOM.WW.OtOpcUa.Cluster/Redundancy/ServiceLevelCalculator.cs
index 2be0b2612..fe9c5b4dc 100644
--- a/src/Core/ZB.MOM.WW.OtOpcUa.Cluster/Redundancy/ServiceLevelCalculator.cs
+++ b/src/Core/ZB.MOM.WW.OtOpcUa.Cluster/Redundancy/ServiceLevelCalculator.cs
@@ -62,4 +62,44 @@ public static class ServiceLevelCalculator
return (byte)Math.Clamp(basis + (h.IsDriverPrimary ? 10 : 0), 0, 255);
}
+
+ /// Warm Standby, healthy (Gitea #717 spec §4.3): Degraded band.
+ public const byte WarmStandbyHealthy = 90;
+
+ /// Warm Standby, stale or DB unreachable (spec §4.3): Degraded band.
+ public const byte WarmStandbyDegraded = 50;
+
+ /// Warm Active, critically degraded (spec §4.3): the lowest level a data-serving Active node publishes.
+ public const byte WarmActiveCritical = 100;
+
+ /// Warm Active, stale config with the DB reachable (spec §4.3).
+ public const byte WarmActiveStale = 200;
+
+ /// Warm Active, healthy (spec §4.3): 240 + the 10-point Primary bonus.
+ public const byte WarmActiveHealthy = 250;
+
+ ///
+ /// The Warm ServiceLevel bands (Gitea #717 spec §4.3). is unchanged and still serves
+ /// Unset and None; this is used only under OpcUa:Redundancy:Mode=Warm.
+ ///
+ ///
+ /// Invariant: every Standby level (≤ ) is below every level a data-serving
+ /// Active node publishes (≥ ), so a client that picks the highest
+ /// ServiceLevel never prefers a Standby. A member that is not Up/Joining, or whose own endpoint failed, is 0
+ /// whatever its role (unchanged from , lmxopcua#762).
+ ///
+ /// The node's health facts; is ignored — the
+ /// Warm verdict replaces it.
+ /// The Warm role verdict (spec §4.1).
+ /// The ServiceLevel byte.
+ public static byte ComputeWarm(NodeHealthInputs h, bool isActive)
+ {
+ if (h.MemberState is not (MemberStatus.Up or MemberStatus.Joining)) return 0;
+ if (!h.EndpointListening) return 0;
+
+ var healthy = h.DbReachable && h.OpcUaProbeOk && !h.Stale;
+ if (!isActive) return healthy ? WarmStandbyHealthy : WarmStandbyDegraded;
+ if (healthy) return WarmActiveHealthy;
+ return h.DbReachable && h.Stale ? WarmActiveStale : WarmActiveCritical;
+ }
}
diff --git a/src/Core/ZB.MOM.WW.OtOpcUa.Commons/Messages/Redundancy/RedundancyStateChanged.cs b/src/Core/ZB.MOM.WW.OtOpcUa.Commons/Messages/Redundancy/RedundancyStateChanged.cs
index 6b2163875..722733fb6 100644
--- a/src/Core/ZB.MOM.WW.OtOpcUa.Commons/Messages/Redundancy/RedundancyStateChanged.cs
+++ b/src/Core/ZB.MOM.WW.OtOpcUa.Commons/Messages/Redundancy/RedundancyStateChanged.cs
@@ -6,6 +6,74 @@ namespace ZB.MOM.WW.OtOpcUa.Commons.Messages.Redundancy;
/// Broadcast whenever the cluster's redundancy topology changes (node up/down, role-leader
/// change, partition heal). Subscribers compute their local OPC UA ServiceLevel from this.
///
+/// One entry per cluster member.
+/// A random id per publish (diagnostics only; it carries no order).
+/// The publisher's ordered revision (Gitea #717, spec §4.2). null from a pre-#717
+/// publisher during a rolling upgrade: such a snapshot still feeds the existing gates, but can never make a Warm
+/// node Active.
public sealed record RedundancyStateChanged(
IReadOnlyList Nodes,
- CorrelationId CorrelationId);
+ CorrelationId CorrelationId,
+ SnapshotRevision? Revision = null);
+
+///
+/// The order of snapshots (Gitea #717, spec §4.2), compared
+/// lexicographically. Within one cluster incarnation the singleton always lives on the oldest member, and every
+/// surviving member has a larger up-number than a departed oldest one, so is
+/// monotonic across handovers; orders a singleton restarted in the same process;
+/// orders the publishes of one incarnation.
+///
+/// The Member.UpNumber of the node hosting the singleton.
+/// Environment.TickCount64 at the singleton's PreStart.
+/// Incremented on every publish, heartbeats included.
+public sealed record SnapshotRevision(long PublisherUpNumber, long IncarnationTick, long Sequence)
+ : IComparable
+{
+ ///
+ public int CompareTo(SnapshotRevision? other)
+ {
+ if (other is null) return 1;
+ var c = PublisherUpNumber.CompareTo(other.PublisherUpNumber);
+ if (c != 0) return c;
+ c = IncarnationTick.CompareTo(other.IncarnationTick);
+ return c != 0 ? c : Sequence.CompareTo(other.Sequence);
+ }
+
+ ///
+ public override string ToString() => $"{PublisherUpNumber}.{IncarnationTick}.{Sequence}";
+}
+
+/// Why a snapshot was not accepted by a .
+public enum SnapshotRejection
+{
+ /// Accepted.
+ None,
+
+ /// Its revision is not greater than the last accepted one (a late publish of an older incarnation).
+ OlderRevision,
+
+ /// It carries no revision, while revisioned snapshots are being ordered.
+ Unrevisioned,
+}
+
+///
+/// The subscriber-side order rule of spec §4.2: accept a snapshot only if its revision is greater than the last
+/// accepted one. Not thread-safe; owned by one actor.
+///
+public sealed class SnapshotRevisionGate
+{
+ /// The last accepted revision; null until a revisioned snapshot was accepted.
+ public SnapshotRevision? LastAccepted { get; private set; }
+
+ /// Decide one snapshot. An unrevisioned snapshot is reported
+ /// and never advances the order (the caller decides whether the existing gates still use it).
+ /// The snapshot.
+ /// when accepted.
+ public SnapshotRejection Offer(RedundancyStateChanged msg)
+ {
+ if (msg.Revision is not { } revision) return SnapshotRejection.Unrevisioned;
+ if (LastAccepted is { } last && revision.CompareTo(last) <= 0) return SnapshotRejection.OlderRevision;
+ LastAccepted = revision;
+ return SnapshotRejection.None;
+ }
+}
diff --git a/src/Core/ZB.MOM.WW.OtOpcUa.Commons/Observability/OtOpcUaTelemetry.cs b/src/Core/ZB.MOM.WW.OtOpcUa.Commons/Observability/OtOpcUaTelemetry.cs
index 1bce3a838..54f5d5316 100644
--- a/src/Core/ZB.MOM.WW.OtOpcUa.Commons/Observability/OtOpcUaTelemetry.cs
+++ b/src/Core/ZB.MOM.WW.OtOpcUa.Commons/Observability/OtOpcUaTelemetry.cs
@@ -190,6 +190,44 @@ public static class OtOpcUaTelemetry
Meter.CreateCounter("otopcua.opcua.condition_events.superseded", unit: "{delivery}",
description: "Deferred condition events withheld from a Server-object item whose ConditionRefresh already replayed a newer state.");
+ // ---------------- Warm redundancy (Gitea #717, spec §7) ----------------
+
+ private static int _warmRole = -1;
+
+ /// Records this process's Warm role for : 1 Active, 0 Standby, −1 not Warm.
+ /// The role value.
+ public static void SetWarmRole(int value) => Volatile.Write(ref _warmRole, value);
+
+ /// The Warm role of this node: 1 = Active, 0 = Standby, −1 = not Warm.
+ public static readonly ObservableGauge WarmRole =
+ Meter.CreateObservableGauge("otopcua.redundancy.warm_role", () => Volatile.Read(ref _warmRole), unit: "{role}",
+ description: "Warm redundancy role of this node: 1 Active, 0 Standby, -1 not Warm.");
+
+ /// Warm role transitions (direction=promote|demote, reason=membership|snapshot|endpoint|boot|unreachable-primary|agreed).
+ public static readonly Counter WarmTransition =
+ Meter.CreateCounter("otopcua.redundancy.warm_transition", unit: "{transition}",
+ description: "Warm role transitions (direction=promote|demote, reason).");
+
+ /// Redundancy snapshots not accepted by a subscriber (reason=older_revision|unrevisioned).
+ public static readonly Counter SnapshotRejected =
+ Meter.CreateCounter("otopcua.redundancy.snapshot_rejected", unit: "{snapshot}",
+ description: "Redundancy snapshots dropped by a subscriber (reason=older_revision|unrevisioned).");
+
+ /// The artifact's ServerCluster.RedundancyMode disagrees with the local OpcUa:Redundancy:Mode.
+ public static readonly Counter RedundancyModeMismatch =
+ Meter.CreateCounter("otopcua.redundancy.mode_mismatch", unit: "{apply}",
+ description: "Applied artifacts whose cluster RedundancyMode disagrees with the local OpcUa:Redundancy:Mode.");
+
+ /// Ordinary-tag notifications rewritten BadOutOfService at a Warm Standby's Publish/Republish (D4).
+ public static readonly Counter StandbyProjectedNotifications =
+ Meter.CreateCounter("otopcua.opcua.standby.projected_notifications", unit: "{notification}",
+ description: "Ordinary-tag notifications rewritten BadOutOfService at a Warm standby's first delivery.");
+
+ /// New alarm condition events not delivered on a Warm Standby (still stamped).
+ public static readonly Counter StandbySuppressedAlarmEvents =
+ Meter.CreateCounter("otopcua.opcua.standby.suppressed_alarm_events", unit: "{event}",
+ description: "New alarm condition events suppressed on a Warm standby.");
+
// ---------------- Convenience helpers ----------------
///
diff --git a/src/Core/ZB.MOM.WW.OtOpcUa.Commons/OpcUa/IStandbyProjection.cs b/src/Core/ZB.MOM.WW.OtOpcUa.Commons/OpcUa/IStandbyProjection.cs
new file mode 100644
index 000000000..26f5a3462
--- /dev/null
+++ b/src/Core/ZB.MOM.WW.OtOpcUa.Commons/OpcUa/IStandbyProjection.cs
@@ -0,0 +1,112 @@
+namespace ZB.MOM.WW.OtOpcUa.Commons.OpcUa;
+
+///
+/// The data-plane side of a Warm role transition (Gitea #717, spec §4.7). OpcUaPublishActor owns the
+/// verdict and drives these steps in a fixed order; the node manager implements them. Every member is
+/// non-blocking and idempotent. Under and
+/// nothing calls it.
+///
+public interface IStandbyProjection
+{
+ /// Opens (false) or closes (true) the per-tag read projection: while projected, every
+ /// ordinary-tag Value read and sample answers BadOutOfService/null, and writes and Part 9 condition
+ /// methods are refused BadInvalidState. The stored value keeps updating from acquisition.
+ /// Whether ordinary-tag values are projected.
+ void SetDataProjection(bool projected);
+
+ /// Switches the first-delivery decision (spec §4.6, D4): a Publish response delivered for the first time
+ /// while is recorded as Standby and its ordinary-tag notifications are rewritten.
+ /// Whether first deliveries are decided Standby.
+ void SetStandbyDelivery(bool standby);
+
+ /// Suppresses (true) or restores delivery of NEW alarm condition events (spec §4.5). Condition
+ /// state keeps being stamped, so a later refresh carries current, ackable state.
+ /// Whether new alarm events are suppressed.
+ void SetAlarmSuppression(bool suppressed);
+
+ /// Queues each live, non-stood-down ordinary-tag data item one value, ignoring its filters: the current
+ /// stored value (promotion) or BadOutOfService/null (demotion).
+ /// Whether to queue the projected value instead of the stored one.
+ /// The number of items queued to.
+ int QueueCurrentToLiveItems(bool projected);
+
+ /// Runs a ConditionRefresh on every live event item (Server-object, folder and direct), through the
+ /// existing refresh paths and their ACL decisions (spec §4.5 promotion refresh).
+ /// The number of event items refreshed.
+ int RefreshAllEventItems();
+}
+
+///
+/// Late-binding (the pattern): the
+/// actor gets it at construction, the node manager only exists after SDK start. The latest requested
+/// projection / delivery / suppression state is retained and replayed on bind (the #719 lesson); the two
+/// queue/refresh actions are not replayed — before bind there are no items to act on.
+///
+public sealed class DeferredStandbyProjection : IStandbyProjection
+{
+ private readonly object _gate = new();
+ private IStandbyProjection? _inner;
+ private bool? _projected;
+ private bool? _standby;
+ private bool? _suppressed;
+
+ /// Bind (and replay the retained state), or unbind with null.
+ /// The live implementation, or null to unbind.
+ public void SetInner(IStandbyProjection? inner)
+ {
+ lock (_gate)
+ {
+ _inner = inner;
+ if (inner is null) return;
+ if (_projected is { } p) inner.SetDataProjection(p);
+ if (_standby is { } s) inner.SetStandbyDelivery(s);
+ if (_suppressed is { } a) inner.SetAlarmSuppression(a);
+ }
+ }
+
+ ///
+ public void SetDataProjection(bool projected)
+ {
+ lock (_gate)
+ {
+ _projected = projected;
+ _inner?.SetDataProjection(projected);
+ }
+ }
+
+ ///
+ public void SetStandbyDelivery(bool standby)
+ {
+ lock (_gate)
+ {
+ _standby = standby;
+ _inner?.SetStandbyDelivery(standby);
+ }
+ }
+
+ ///
+ public void SetAlarmSuppression(bool suppressed)
+ {
+ lock (_gate)
+ {
+ _suppressed = suppressed;
+ _inner?.SetAlarmSuppression(suppressed);
+ }
+ }
+
+ ///
+ public int QueueCurrentToLiveItems(bool projected)
+ {
+ IStandbyProjection? inner;
+ lock (_gate) inner = _inner;
+ return inner?.QueueCurrentToLiveItems(projected) ?? 0;
+ }
+
+ ///
+ public int RefreshAllEventItems()
+ {
+ IStandbyProjection? inner;
+ lock (_gate) inner = _inner;
+ return inner?.RefreshAllEventItems() ?? 0;
+ }
+}
diff --git a/src/Core/ZB.MOM.WW.OtOpcUa.Commons/OpcUa/WarmRedundancySettings.cs b/src/Core/ZB.MOM.WW.OtOpcUa.Commons/OpcUa/WarmRedundancySettings.cs
new file mode 100644
index 000000000..327722f05
--- /dev/null
+++ b/src/Core/ZB.MOM.WW.OtOpcUa.Commons/OpcUa/WarmRedundancySettings.cs
@@ -0,0 +1,36 @@
+namespace ZB.MOM.WW.OtOpcUa.Commons.OpcUa;
+
+///
+/// The parsed OpcUa:Redundancy:Mode (Gitea #717). Startup-static: changing it requires a restart.
+///
+public enum OpcUaRedundancyMode
+{
+ /// Not opted in (the default): byte-for-byte the pre-#717 server. Nothing in the redundancy metadata,
+ /// ServiceLevel bands, projection or alarm delivery changes.
+ Unset = 0,
+
+ /// Standalone: advertises RedundancySupport=None with no ServerUriArray; never projects.
+ None = 1,
+
+ /// A 2-node pair with a degraded standby (spec §4): non-transparent Warm metadata, Standby bands below
+ /// every Active level, ordinary-tag values projected BadOutOfService on the Standby, new alarm events
+ /// suppressed there, and writes / Part 9 methods refused BadInvalidState there.
+ Warm = 2,
+}
+
+///
+/// The runtime half of OpcUa:Redundancy (Gitea #717): what the redundancy actors need. The Host builds it
+/// from the validated OpcUaApplicationHostOptions and registers it as a singleton; a node that registers
+/// none (a harness, an admin-only node) is .
+///
+/// The redundancy mode.
+/// Warm only: let the survivor go Active while the Primary is
+/// unreachable but not yet downed (spec §4.1, owner answer Q2). Default on.
+public sealed record WarmRedundancySettings(OpcUaRedundancyMode Mode, bool PromoteWhilePrimaryUnreachable = true)
+{
+ /// The not-opted-in default.
+ public static WarmRedundancySettings Unset { get; } = new(OpcUaRedundancyMode.Unset);
+
+ /// True when this node runs the Warm role contract.
+ public bool IsWarm => Mode == OpcUaRedundancyMode.Warm;
+}
diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.ControlPlane/Redundancy/RedundancyStateActor.cs b/src/Server/ZB.MOM.WW.OtOpcUa.ControlPlane/Redundancy/RedundancyStateActor.cs
index dd0ff9cb6..e93ad966e 100644
--- a/src/Server/ZB.MOM.WW.OtOpcUa.ControlPlane/Redundancy/RedundancyStateActor.cs
+++ b/src/Server/ZB.MOM.WW.OtOpcUa.ControlPlane/Redundancy/RedundancyStateActor.cs
@@ -37,6 +37,11 @@ public sealed class RedundancyStateActor : ReceiveActor, IWithTimers
private readonly Action
-internal sealed class EquipmentVariableState(NodeState? parent, object ownerLock) : BaseDataVariableState(parent)
+internal sealed class EquipmentVariableState(NodeState? parent, object ownerLock, WarmStandbyState? standby = null)
+ : BaseDataVariableState(parent)
{
+ private readonly WarmStandbyState _standby = standby ?? WarmStandbyState.Off;
+
// Every field below is accessed under ownerLock. Tokens identify client attempts; the mutation
// revision detects a driver publication that re-enters the callback before its optimistic commit.
private long _nextWriteToken;
@@ -82,6 +85,16 @@ internal sealed class EquipmentVariableState(NodeState? parent, object ownerLock
QualifiedName dataEncoding, ref Variant value, ref DateTimeUtc sourceTimestamp)
{
// Subscription samplers read the node directly, bypassing the manager's service lock.
- lock (ownerLock) return base.ReadValueAttribute(context, indexRange, dataEncoding, ref value, ref sourceTimestamp);
+ ServiceResult result;
+ lock (ownerLock) result = base.ReadValueAttribute(context, indexRange, dataEncoding, ref value, ref sourceTimestamp);
+
+ // Gitea #717 spec §4.4 (T0b): on a Warm Standby every value path — Read (IndexRange and DataEncoding included),
+ // the initial sample, the change fan-out, ResendData, AggregateFilter input — answers BadOutOfService/null. The
+ // stored Value is untouched and keeps updating from acquisition. The base read runs first, so an IndexRange or
+ // encoding error is still reported as such. Off (Unset/None/Active): returned unchanged.
+ if (ServiceResult.IsBad(result) || !_standby.DataProjected) return result;
+ value = Variant.Null;
+ sourceTimestamp = DateTimeUtc.MinValue;
+ return StatusCodes.BadOutOfService;
}
}
diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/OpcUaApplicationHost.cs b/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/OpcUaApplicationHost.cs
index 893462bad..2f945ec62 100644
--- a/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/OpcUaApplicationHost.cs
+++ b/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/OpcUaApplicationHost.cs
@@ -4,6 +4,7 @@ using Opc.Ua;
using Opc.Ua.Configuration;
using Opc.Ua.Identity;
using Opc.Ua.Server;
+using ZB.MOM.WW.OtOpcUa.Commons.OpcUa;
using ZB.MOM.WW.OtOpcUa.Core.Authorization;
using ZB.MOM.WW.OtOpcUa.OpcUaServer.Security;
@@ -174,6 +175,83 @@ public sealed class OpcUaApplicationHostOptions
/// Replaces the SDK's own lockout, which also refused Anonymous; see .
///
public OpcUaCredentialLockoutOptions CredentialLockout { get; set; } = new();
+
+ ///
+ /// Warm non-transparent redundancy (OpcUa:Redundancy:*, Gitea #717). The default Mode=Unset is
+ /// byte-for-byte today's server. Startup-static.
+ ///
+ public OpcUaRedundancyOptions Redundancy { get; set; } = new();
+}
+
+/// The OpcUa:Redundancy section (Gitea #717, spec §6).
+public sealed class OpcUaRedundancyOptions
+{
+ ///
+ /// Unset (default) / None / Warm, case-insensitive. Hot is not advertisable
+ /// (spec §3) and, like any other value, fails host start. Bound as a string so a bad value is a readable
+ /// validation failure rather than a binder exception.
+ ///
+ public string Mode { get; set; } = nameof(OpcUaRedundancyMode.Unset);
+
+ ///
+ /// Warm only: let the survivor go Active while the Primary is unreachable but not yet downed (spec §4.1, owner
+ /// answer Q2). Default true.
+ ///
+ public bool PromoteWhilePrimaryUnreachable { get; set; } = true;
+
+ /// Parses .
+ /// The parsed mode.
+ /// false for Hot, numerics or any unknown value.
+ public bool TryParseMode(out OpcUaRedundancyMode mode)
+ {
+ var text = (Mode ?? string.Empty).Trim();
+ if (text.Length == 0) text = nameof(OpcUaRedundancyMode.Unset);
+ if (!text.All(char.IsLetter)) { mode = default; return false; }
+ return Enum.TryParse(text, ignoreCase: true, out mode) && Enum.IsDefined(mode);
+ }
+
+ /// The parsed mode; throws on an invalid one (the validator rejects it first).
+ public OpcUaRedundancyMode ParsedMode => TryParseMode(out var m)
+ ? m
+ : throw new InvalidOperationException(
+ $"OpcUa:Redundancy:Mode '{Mode}' is not supported; use Unset, None or Warm (Hot is not advertisable).");
+
+ /// The runtime settings the redundancy actors read.
+ /// The settings.
+ public WarmRedundancySettings ToSettings() => new(ParsedMode, PromoteWhilePrimaryUnreachable);
+
+ ///
+ /// The rules of spec §3 that need only configuration: Warm needs at least one peer, all URIs distinct
+ /// and the self URI not listed as a peer; None needs no peers. Returns the failure messages (empty when
+ /// valid). The effective self URI is checked again after an ApplicationConfigPath load.
+ ///
+ /// The self URI to check against.
+ /// The configured peer URIs.
+ /// The failures.
+ public IReadOnlyList Check(string? selfUri, IEnumerable? peers)
+ {
+ var failures = new List();
+ if (!TryParseMode(out var mode))
+ {
+ failures.Add($"OpcUa:Redundancy:Mode '{Mode}' is not supported; use Unset, None or Warm (Hot is not advertisable: continuous acquisition alone does not meet Part 4 Hot)");
+ return failures;
+ }
+
+ var list = (peers ?? Array.Empty()).Where(p => !string.IsNullOrWhiteSpace(p)).Select(p => p.Trim()).ToList();
+ if (mode == OpcUaRedundancyMode.None && list.Count > 0)
+ failures.Add("OpcUa:PeerApplicationUris must be empty when OpcUa:Redundancy:Mode is None (a standalone server has no peers)");
+ if (mode == OpcUaRedundancyMode.Warm)
+ {
+ if (list.Count == 0)
+ failures.Add("OpcUa:Redundancy:Mode Warm requires at least one OpcUa:PeerApplicationUris entry");
+ if (list.Distinct(StringComparer.Ordinal).Count() != list.Count)
+ failures.Add("OpcUa:PeerApplicationUris must not contain duplicates under OpcUa:Redundancy:Mode Warm");
+ if (!string.IsNullOrWhiteSpace(selfUri) && list.Contains(selfUri.Trim(), StringComparer.Ordinal))
+ failures.Add($"OpcUa:PeerApplicationUris must not list this server's own ApplicationUri '{selfUri}' under OpcUa:Redundancy:Mode Warm (each pair node needs a unique ApplicationUri)");
+ }
+
+ return failures;
+ }
}
///
@@ -289,7 +367,8 @@ public sealed class OpcUaApplicationHost : IAsyncDisposable
try
{
- _ = await BuildConfigurationAsync(cancellationToken);
+ var configuration = await BuildConfigurationAsync(cancellationToken);
+ ConfigureRedundancy(sdkServer, configuration.ApplicationUri ?? _options.ApplicationUri);
await EnsureApplicationCertificateAsync(cancellationToken).ConfigureAwait(false);
AttachUserAuthenticator(sdkServer);
await _application.StartAsync(server, cancellationToken).ConfigureAwait(false);
@@ -305,6 +384,31 @@ public sealed class OpcUaApplicationHost : IAsyncDisposable
_options.PublicHostname, _options.OpcUaPort);
}
+ ///
+ /// Gitea #717 (spec §3): validate the redundancy mode against the EFFECTIVE self URI — the loaded
+ /// ApplicationConfigPath XML wins over OpcUa:ApplicationUri (c7787) — log the boot banner and hand
+ /// the startup-static metadata to the SDK server, which installs it in OnNodeManagerStartedAsync.
+ /// Unset hands nothing over, so the SDK's graph is untouched.
+ ///
+ /// The SDK server about to start.
+ /// The effective ApplicationConfiguration.ApplicationUri.
+ /// The redundancy configuration is invalid for the effective URI.
+ private void ConfigureRedundancy(OtOpcUaSdkServer server, string effectiveSelfUri)
+ {
+ var redundancy = _options.Redundancy ?? new OpcUaRedundancyOptions();
+ var failures = redundancy.Check(effectiveSelfUri, _options.PeerApplicationUris);
+ if (failures.Count > 0)
+ throw new InvalidOperationException("OPC UA redundancy configuration is invalid: " + string.Join("; ", failures));
+
+ var mode = redundancy.ParsedMode;
+ var peers = (_options.PeerApplicationUris ?? new List())
+ .Where(p => !string.IsNullOrWhiteSpace(p)).Select(p => p.Trim()).ToArray();
+ _logger.LogInformation(
+ "OPC UA redundancy: Mode={Mode}, effective ApplicationUri={SelfUri}, peers=[{Peers}], PromoteWhilePrimaryUnreachable={Promote}",
+ mode, effectiveSelfUri, string.Join(", ", peers), redundancy.PromoteWhilePrimaryUnreachable);
+ server.ConfigureRedundancy(mode, effectiveSelfUri, peers);
+ }
+
private void AttachUserAuthenticator(OtOpcUaSdkServer server)
{
foreach (var type in new[] { UserTokenType.Anonymous, UserTokenType.UserName })
diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/OtOpcUaNodeManager.cs b/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/OtOpcUaNodeManager.cs
index 7978efbc0..9214e1e29 100644
--- a/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/OtOpcUaNodeManager.cs
+++ b/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/OtOpcUaNodeManager.cs
@@ -161,6 +161,34 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2, INodeManager3
private FolderState? _root;
private readonly ILogger _logger;
+ /// Initializes a new instance of the class with the OPC UA server and configuration.
+ /// The OPC UA server instance.
+ /// The application configuration.
+ /// Logger for the manager's swallowed-fault and skipped-wiring diagnostics;
+ /// null (tests) falls back to .
+ /// The Warm standby flags (Gitea #717); null (every non-Warm server) is
+ /// , which leaves every data path exactly as it was.
+ internal OtOpcUaNodeManager(
+ IServerInternal server, ApplicationConfiguration configuration, ILogger? logger,
+ WarmStandbyState? standby)
+ : this(server, configuration, logger)
+ {
+ _standby = standby ?? WarmStandbyState.Off;
+ if (_standby.WarmMode)
+ {
+ // Spec §4.6 item 1: ordinary-tag Value items are built by OrdinaryTagMonitoredItem, which marks each
+ // notification. Warm only — Unset/None keep the SDK's own manager.
+ m_monitoredItemManager.Dispose();
+ m_monitoredItemManager = new OrdinaryTagItemManager(this.ToAsyncNodeManager(), server);
+ }
+ }
+
+ /// The Warm standby flags (Gitea #717). unless the server is Warm.
+ private readonly WarmStandbyState _standby = WarmStandbyState.Off;
+
+ /// Test observation: the Warm standby flags.
+ internal WarmStandbyState Standby => _standby;
+
/// Initializes a new instance of the class with the OPC UA server and configuration.
/// The OPC UA server instance.
/// The application configuration.
@@ -1434,6 +1462,18 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2, INodeManager3
// the InstanceStateSnapshot captures the just-projected values.
alarm.ClearChangeMasks(SystemContext, includeChildren: true);
+ // Gitea #717 spec §4.5: on a Warm Standby only the STAMP half above runs — the condition node is current and a
+ // later refresh carries a valid, ackable EventId — and the DELIVER half below does not. Suppressing here, before
+ // the snapshot and every SDK field projection, covers the Server-object, folder and direct paths, ACL on or
+ // off, by construction; overflow, filtered retain and the #838 deferral queue never see the event. The
+ // decision is made at report time: an event reported while Active is delivered even if that ends after
+ // demotion.
+ if (_standby.AlarmsSuppressed)
+ {
+ ZB.MOM.WW.OtOpcUa.Commons.Observability.OtOpcUaTelemetry.StandbySuppressedAlarmEvents.Add(1);
+ return;
+ }
+
try
{
// InstanceStateSnapshot is the IFilterTarget — a frozen copy of the condition's fields at fire
@@ -2991,6 +3031,10 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2, INodeManager3
var gate = EvaluateAlarmCommandGate(identity, condition, operation);
if (gate is not null) return gate;
+ // Gitea #717 owner answer Q4: a Warm Standby refuses every Part 9 condition method before it changes any state
+ // (nothing routed, nothing applied, no event). Unset/None keep today's acceptance.
+ if (_standby.RefusesMutations) return WarmStandbyState.StandbyRefusal($"call {operation}");
+
var cmd = new AlarmCommand(
AlarmId: condition.NodeId.IdentifierAsString,
Operation: operation,
@@ -3032,6 +3076,9 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2, INodeManager3
var gate = EvaluateAlarmCommandGate(identity, condition, "Acknowledge");
if (gate is not null) return gate;
+ // Gitea #717 owner answer Q4: refused on a Warm Standby, ahead of (and whatever) the Primary gate below says.
+ if (_standby.RefusesMutations) return WarmStandbyState.StandbyRefusal("acknowledge the alarm");
+
// PRIMARY GATE, after authorization (a caller who may not ack learns nothing about the
// node's role). Only the Primary routes a native ack upstream (DriverHostActor applies the same verdict
// to the route), so on any other node the ack is refused outright rather than applied locally and then
@@ -3215,6 +3262,11 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2, INodeManager3
?? EvaluateEquipmentWriteGate(identity, gatewayWired);
if (gate is not null) return gate;
+ // Gitea #717 owner answer Q1: a Warm Standby refuses the write synchronously, AFTER authorization (a caller who
+ // may not write learns nothing about the node's role) and before anything reaches the driver. Nothing is
+ // applied, so nothing is reverted. Unset/None never set the flag and keep the #5 optimistic path.
+ if (_standby.RefusesMutations) return WarmStandbyState.StandbyRefusal("write the value");
+
// Item A (synchronous structural fail-fast): reject a structurally-invalid write INLINE — return Bad
// synchronously so the SDK never applies it (no optimistic-Good-then-revert round-trip + no needless
// device dispatch). Runs AFTER the authz gate (so we never leak structure detail to an unauthorised
@@ -4120,7 +4172,7 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2, INodeManager3
// The SDK exposes the flags separately (no CurrentReadWrite composite): ReadWrite is
// CurrentRead | CurrentWrite, with the HistoryRead bit OR-ed in when historized.
var access = ComposeAccessLevel(writable, historized, HistoryWritable(_historianDataUpdater));
- var variable = new EquipmentVariableState(parent, Lock)
+ var variable = new EquipmentVariableState(parent, Lock, _standby)
{
NodeId = new NodeId(variableNodeId, ns),
BrowseName = new QualifiedName(variableNodeId, ns),
@@ -5725,6 +5777,88 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2, INodeManager3
lock (Lock) return MonitoredNodes.ContainsKey(nodeId);
}
+ ///
+ /// Gitea #717 spec §4.7 promotion step 2 / demotion step 5: queues one value to every live, non-stood-down
+ /// ordinary-tag Value item, ignoring its filters (the queue pattern of
+ /// — never detach or Disabled, which drops the queue). Promotion queues the current stored value read
+ /// through the item's own IndexRange/DataEncoding; demotion queues BadOutOfService/null.
+ ///
+ /// Queue the projected value (demotion) instead of the stored one (promotion).
+ /// The number of items queued to.
+ internal int QueueCurrentToLiveItems(bool projected)
+ {
+ var queued = 0;
+ foreach (var (item, tracked) in _liveItems)
+ {
+ if (tracked.Suppressed || tracked.Handle.Node is not EquipmentVariableState variable) continue;
+ if (item is not IDataChangeMonitoredItem2 { AttributeId: Attributes.Value } data) continue;
+ var now = UtcNow;
+ DataValue value;
+ if (projected)
+ {
+ value = new DataValue(Variant.Null, StatusCodes.BadOutOfService, DateTimeUtc.MinValue, now);
+ }
+ else
+ {
+ var stored = new DataValue();
+ ServiceResult read;
+ lock (Lock) read = variable.ReadAttribute(SystemContext, Attributes.Value, data.IndexRange, data.DataEncoding, ref stored);
+ value = ServiceResult.IsBad(read)
+ ? new DataValue(Variant.Null, read.StatusCode, DateTimeUtc.MinValue, now)
+ : new DataValue(stored.WrappedValue, stored.StatusCode, stored.SourceTimestamp, now);
+ }
+
+ try
+ {
+ item.QueueValue(value, StatusCode.IsBad(value.StatusCode) ? new ServiceResult(value.StatusCode) : null, ignoreFilters: true);
+ queued++;
+ }
+ catch (Exception ex)
+ {
+ // A disabled item refuses the value; it re-samples the (projected or not) node when re-enabled.
+ _logger.LogDebug(ex, "OtOpcUaNodeManager: could not queue a Warm transition value to a monitored item");
+ }
+ }
+
+ return queued;
+ }
+
+ ///
+ /// Gitea #717 spec §4.5 promotion refresh: a ConditionRefresh2 on every live event item — Server-object items
+ /// and the direct/folder items of this manager's notifiers — through the SDK's own subscription path, so each
+ /// client sees RefreshStart, the retained conditions (through this manager's ACL decision and the #854/#855
+ /// cut), then RefreshEnd. Must not be called under Lock.
+ ///
+ /// The number of event items refreshed.
+ internal int RefreshAllEventItems()
+ {
+ var items = new List(_allEventsItems.Keys);
+ foreach (var (item, tracked) in _eventItems)
+ if (!tracked.Suppressed) items.Add(item);
+ if (items.Count == 0) return 0;
+
+ var subscriptions = new Dictionary();
+ foreach (var subscription in Server.SubscriptionManager.GetSubscriptions()) subscriptions[subscription.Id] = subscription;
+
+ var refreshed = 0;
+ foreach (var item in items)
+ {
+ if (!subscriptions.TryGetValue(item.SubscriptionId, out var subscription)) continue;
+ try
+ {
+ subscription.ConditionRefresh2Async(item.Id).AsTask().GetAwaiter().GetResult();
+ refreshed++;
+ }
+ catch (Exception ex)
+ {
+ // A subscription deleted mid-walk, or an item whose mode refuses a refresh: that client re-refreshes itself.
+ _logger.LogDebug(ex, "OtOpcUaNodeManager: Warm promotion refresh of event item {ItemId} failed", item.Id);
+ }
+ }
+
+ return refreshed;
+ }
+
/// Test observability: how many data-change items the revalidation map tracks.
internal int LiveItemCount => _liveItems.Count;
diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/OtOpcUaSdkServer.cs b/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/OtOpcUaSdkServer.cs
index ceef01ed2..86bd5837f 100644
--- a/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/OtOpcUaSdkServer.cs
+++ b/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/OtOpcUaSdkServer.cs
@@ -315,6 +315,160 @@ public sealed class OtOpcUaSdkServer : StandardServer
///
public const byte StartupServiceLevel = 0;
+ private OpcUaRedundancyMode _redundancyMode = OpcUaRedundancyMode.Unset;
+ private string _selfUri = string.Empty;
+ private string[] _peerUris = [];
+ private WarmStandbyState _standby = WarmStandbyState.Off;
+
+ /// The redundancy mode configured before start (Gitea #717).
+ public OpcUaRedundancyMode RedundancyMode => _redundancyMode;
+
+ /// The Warm standby flags shared with the node manager.
+ internal WarmStandbyState Standby => _standby;
+
+ /// The initial Server.ServiceLevel under Warm (spec §4.3: the NoData band, replacing the
+ /// SDK's 255 and this server's #885 value 0 until the first computed level).
+ public const byte WarmBootServiceLevel = 1;
+
+ ///
+ /// Sets the startup-static redundancy metadata (Gitea #717 spec §3) before StartAsync.
+ /// (the default when this is never called) leaves the SDK graph and
+ /// every data path exactly as they are today.
+ ///
+ /// The validated mode.
+ /// The effective ApplicationConfiguration.ApplicationUri.
+ /// The peer URIs, in configured order.
+ public void ConfigureRedundancy(OpcUaRedundancyMode mode, string effectiveSelfUri, IReadOnlyList peerUris)
+ {
+ if (_otOpcUaNodeManager is not null)
+ throw new InvalidOperationException("Redundancy must be configured before the server starts.");
+ _redundancyMode = mode;
+ _selfUri = effectiveSelfUri;
+ _peerUris = peerUris.ToArray();
+ _standby = new WarmStandbyState(mode == OpcUaRedundancyMode.Warm);
+ }
+
+ ///
+ /// The data-plane transition steps a Warm OpcUaPublishActor drives (spec §4.7), or null when the server
+ /// is not Warm or has not started.
+ ///
+ /// The projection, or null.
+ public IStandbyProjection? CreateStandbyProjection() =>
+ _standby.WarmMode && _otOpcUaNodeManager is { } nm ? new SdkStandbyProjection(nm, _standby) : null;
+
+ /// Test seam: ids of the live subscriptions, for the decision record's dead-subscription sweep.
+ private IReadOnlySet LiveSubscriptionIds()
+ {
+ var ids = new HashSet();
+ try
+ {
+ if (CurrentInstance?.SubscriptionManager is { } manager)
+ foreach (var subscription in manager.GetSubscriptions()) ids.Add(subscription.Id);
+ }
+ catch (Exception)
+ {
+ // Shutdown race: an empty set only drops records, which re-decide (fail-safe) at their next delivery.
+ }
+
+ return ids;
+ }
+
+ private long _publishesSinceSweep;
+
+ ///
+ public override async ValueTask PublishAsync(SecureChannelContext secureChannelContext,
+ RequestHeader? requestHeader, ArrayOf subscriptionAcknowledgements,
+ RequestLifetime requestLifetime)
+ {
+ var response = await base.PublishAsync(secureChannelContext, requestHeader, subscriptionAcknowledgements, requestLifetime)
+ .ConfigureAwait(false);
+ if (_standby.Delivery is not { } record || response?.NotificationMessage is not { } message) return response!;
+
+ if (message.NotificationData.Count > 0)
+ {
+ response.NotificationMessage = Decide(record, response.SubscriptionId, message);
+ var available = new List(response.AvailableSequenceNumbers.Count);
+ for (var i = 0; i < response.AvailableSequenceNumbers.Count; i++) available.Add(response.AvailableSequenceNumbers[i]);
+ record.Prune(response.SubscriptionId, available, message.SequenceNumber);
+ }
+
+ if (Interlocked.Increment(ref _publishesSinceSweep) % 1024 == 0) record.SweepDead(LiveSubscriptionIds());
+ return response;
+ }
+
+ ///
+ public override async ValueTask RepublishAsync(SecureChannelContext secureChannelContext,
+ RequestHeader? requestHeader, uint subscriptionId, uint retransmitSequenceNumber, RequestLifetime requestLifetime)
+ {
+ var response = await base.RepublishAsync(secureChannelContext, requestHeader, subscriptionId, retransmitSequenceNumber,
+ requestLifetime).ConfigureAwait(false);
+ if (_standby.Delivery is { } record && response?.NotificationMessage is { NotificationData.Count: > 0 } message)
+ response.NotificationMessage = Decide(record, subscriptionId, message);
+ return response!;
+ }
+
+ /// The recorded (or, first time, current) decision for one delivery, applied to the SDK's fresh clone.
+ private NotificationMessage Decide(FirstDeliveryRecord record, uint subscriptionId, NotificationMessage message)
+ {
+ if (!record.Decide(subscriptionId, message.SequenceNumber)) return message;
+ var projected = StandbyNotificationRewriter.Project(message, out var rewritten);
+ if (rewritten > 0)
+ {
+ ZB.MOM.WW.OtOpcUa.Commons.Observability.OtOpcUaTelemetry.StandbyProjectedNotifications.Add(rewritten);
+ }
+
+ return projected;
+ }
+
+ ///
+ /// Installs the standard redundancy metadata (Gitea #717 spec §3, T0a): under Warm
+ /// Server.ServerRedundancy becomes a NonTransparentRedundancyType whose read-only
+ /// ServerUriArray (i=11314) is [self, peers…], RedundancySupport is Warm and
+ /// ServiceLevel boots at ; under NoneRedundancySupport is
+ /// None and the SDK's orphan i=11314 is deleted, so it is absent by NodeId as well as by browse.
+ /// Unset touches nothing.
+ ///
+ private async ValueTask InstallRedundancyMetadataAsync(IServerInternal server, CancellationToken cancellationToken)
+ {
+ if (_redundancyMode == OpcUaRedundancyMode.Unset) return;
+ var serverObject = server.ServerObject
+ ?? throw new InvalidOperationException("The Server object is unavailable during server startup.");
+ var context = server.DefaultSystemContext;
+ var redundancy = serverObject.ServerRedundancy
+ ?? throw new InvalidOperationException("Server.ServerRedundancy is unavailable during server startup.");
+
+ if (_redundancyMode == OpcUaRedundancyMode.None)
+ {
+ redundancy.RedundancySupport!.Value = RedundancySupport.None;
+ redundancy.RedundancySupport.ClearChangeMasks(context, false);
+ // T0a finding: the SDK's predefined i=11314 survives as an orphan (unreferenced, but Good [] by NodeId).
+ if (server.DiagnosticsNodeManager is AsyncCustomNodeManager diagnostics)
+ await diagnostics.DeleteNodeAsync(context, VariableIds.Server_ServerRedundancy_ServerUriArray, cancellationToken)
+ .ConfigureAwait(false);
+ return;
+ }
+
+ var replacer = server.DiagnosticsNodeManager as IPredefinedNodeSubtypeReplacer
+ ?? throw new InvalidOperationException("The diagnostics node manager cannot replace predefined node subtypes.");
+ var replaced = (NonTransparentRedundancyState)await replacer.ReplacePredefinedInstanceSubtypeAsync(context, redundancy,
+ context.CreateInstanceOfNonTransparentRedundancyType(),
+ new Dictionary
+ {
+ [new QualifiedName(BrowseNames.ServerUriArray, 0)] = VariableIds.Server_ServerRedundancy_ServerUriArray,
+ },
+ node => serverObject.ServerRedundancy = (ServerRedundancyState)node, cancellationToken).ConfigureAwait(false);
+ replaced.RedundancySupport!.Value = RedundancySupport.Warm;
+ replaced.ServerUriArray!.Value = (ArrayOf)new[] { _selfUri }.Concat(_peerUris).ToArray();
+ replaced.ServerUriArray.AccessLevel = replaced.ServerUriArray.UserAccessLevel = AccessLevels.CurrentRead;
+ replaced.ClearChangeMasks(context, true);
+ if (serverObject.ServiceLevel is { } serviceLevel)
+ {
+ serviceLevel.Value = WarmBootServiceLevel;
+ serviceLevel.Timestamp = DateTime.UtcNow;
+ serviceLevel.ClearChangeMasks(context, includeChildren: false);
+ }
+ }
+
///
protected override async ValueTask OnNodeManagerStartedAsync(
IServerInternal server, CancellationToken cancellationToken = default)
@@ -329,6 +483,7 @@ public sealed class OtOpcUaSdkServer : StandardServer
serviceLevel.Timestamp = DateTime.UtcNow;
serviceLevel.ClearChangeMasks(server.DefaultSystemContext, includeChildren: false);
}
+ await InstallRedundancyMetadataAsync(server, cancellationToken).ConfigureAwait(false);
var auditType = server.DiagnosticsNodeManager?
.FindPredefinedNode(ObjectTypeIds.AuditWriteUpdateEventType)
?? throw new InvalidOperationException("AuditWriteUpdateEventType is unavailable during server startup.");
@@ -360,7 +515,7 @@ public sealed class OtOpcUaSdkServer : StandardServer
protected override MasterNodeManager CreateMasterNodeManager(
IServerInternal server, ApplicationConfiguration configuration)
{
- _otOpcUaNodeManager = new OtOpcUaNodeManager(server, configuration, _nodeManagerLogger);
+ _otOpcUaNodeManager = new OtOpcUaNodeManager(server, configuration, _nodeManagerLogger, _standby);
return new MasterNodeManager(server, configuration, dynamicNamespaceUri: null, _otOpcUaNodeManager);
}
}
diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/WarmStandby.cs b/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/WarmStandby.cs
new file mode 100644
index 000000000..af53baaf6
--- /dev/null
+++ b/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/WarmStandby.cs
@@ -0,0 +1,388 @@
+using Microsoft.Extensions.Logging;
+using Opc.Ua;
+using Opc.Ua.Server;
+
+namespace ZB.MOM.WW.OtOpcUa.OpcUaServer;
+
+///
+/// The node-manager-wide Warm standby flags (Gitea #717, spec §4.4–§4.6). One instance per server. Under
+/// Unset/None is false and nothing ever sets a flag, so every reader sees
+/// today's behaviour. Under Warm the node starts Standby (all flags set) until the redundancy layer
+/// promotes it.
+///
+internal sealed class WarmStandbyState
+{
+ private volatile bool _dataProjected;
+ private volatile bool _alarmsSuppressed;
+
+ /// Initializes the state.
+ /// Whether this server runs the Warm role contract.
+ public WarmStandbyState(bool warmMode)
+ {
+ WarmMode = warmMode;
+ _dataProjected = warmMode;
+ _alarmsSuppressed = warmMode;
+ Delivery = warmMode ? new FirstDeliveryRecord(standby: true) : null;
+ }
+
+ /// The not-Warm state (no projection, no suppression, no decision record).
+ public static WarmStandbyState Off { get; } = new(false);
+
+ /// Whether this server runs the Warm role contract.
+ public bool WarmMode { get; }
+
+ /// Ordinary-tag reads/samples project BadOutOfService; writes and Part 9 methods are refused.
+ public bool DataProjected
+ {
+ get => _dataProjected;
+ set => _dataProjected = WarmMode && value;
+ }
+
+ /// New alarm condition events are not delivered (still stamped).
+ public bool AlarmsSuppressed
+ {
+ get => _alarmsSuppressed;
+ set => _alarmsSuppressed = WarmMode && value;
+ }
+
+ /// Whether a Warm Standby must refuse a write or a Part 9 method (owner answers Q1/Q4).
+ public bool RefusesMutations => WarmMode && _dataProjected;
+
+ /// The first-delivery decision record; null when not Warm.
+ public FirstDeliveryRecord? Delivery { get; }
+
+ /// The refusal returned to a write or a Part 9 method on a Warm Standby.
+ public static ServiceResult StandbyRefusal(string what) => new(StatusCodes.BadInvalidState,
+ new LocalizedText($"This server is the Warm redundancy standby; {what} on the active server (the one with the higher ServiceLevel)."));
+}
+
+///
+/// The ordinary-tag marker (spec §4.6 item 1): set on , a helper
+/// reference the binary encoder never writes and the SDK's per-delivery CoreUtils.Clone copies
+/// (pinned by WarmStandbySdkFieldContractTests). Valid only for the notification whose
+/// and it captured, so a stale marker left on a pooled payload
+/// (2.0.1's Initialize() does not clear the helper) fails .
+///
+/// Derives from only because that is the helper's declared type.
+internal sealed class OrdinaryTagMarker : NotificationMessage
+{
+ /// Initializes the marker.
+ /// The notification's client handle when it was built.
+ /// The notification's final value when it was built.
+ public OrdinaryTagMarker(uint clientHandle, DataValue value)
+ {
+ ClientHandle = clientHandle;
+ Value = value;
+ }
+
+ /// The client handle captured at build time.
+ public uint ClientHandle { get; }
+
+ /// The value captured at build time.
+ public DataValue Value { get; }
+
+ /// Whether is the notification this marker was made for.
+ /// A delivered notification.
+ /// True for a match on client handle and value.
+ public bool Matches(MonitoredItemNotification n) => n.ClientHandle == ClientHandle && Equals(n.Value, Value);
+
+ /// Whether carries a valid ordinary-tag marker.
+ /// A delivered notification.
+ /// True when it is an ordinary-tag notification.
+ public static bool IsOrdinaryTag(MonitoredItemNotification n) => n.Message is OrdinaryTagMarker m && m.Matches(n);
+}
+
+///
+/// The per-message first-delivery decision (spec §4.6 item 2, D4), keyed by
+/// (SubscriptionId, SequenceNumber) — never by object: SDK 2.0.1 hands every Publish and Republish a fresh
+/// clone of the cached message. The first delivery of a sequence number records Active or Standby; every later
+/// delivery (Republish included) applies the recorded decision. One lock orders a decision against a role flip;
+/// it is never held across an await or a projection.
+///
+internal sealed class FirstDeliveryRecord
+{
+ /// Recorded messages kept per subscription beyond what AvailableSequenceNumbers prunes.
+ internal const int MaxPerSubscription = 1024;
+
+ /// Subscriptions kept before the least-recently-used one is dropped (dead subscriptions).
+ internal const int MaxSubscriptions = 8192;
+
+ private readonly object _gate = new();
+ private readonly Dictionary _subscriptions = new();
+ private bool _standby;
+ private long _touch;
+
+ /// Initializes the record.
+ /// The initial delivery state (a Warm node boots Standby).
+ public FirstDeliveryRecord(bool standby) => _standby = standby;
+
+ /// Test seam: runs inside the lock after the decision is read and before it is recorded.
+ internal Action? DecisionBarrier { get; set; }
+
+ /// The current first-delivery state.
+ public bool Standby
+ {
+ get { lock (_gate) return _standby; }
+ }
+
+ /// Flips the first-delivery state (the D4 boundary). Waits for a decision in progress.
+ /// Whether later first deliveries are Standby.
+ public void SetStandby(bool standby)
+ {
+ lock (_gate) _standby = standby;
+ }
+
+ /// The decision for one delivery: the recorded one, or (first delivery) the current state, recorded.
+ /// The subscription.
+ /// The message's sequence number.
+ /// True when the delivery must be projected (Standby).
+ public bool Decide(uint subscriptionId, uint sequenceNumber)
+ {
+ lock (_gate)
+ {
+ if (!_subscriptions.TryGetValue(subscriptionId, out var sub))
+ {
+ if (_subscriptions.Count >= MaxSubscriptions) EvictLeastRecentlyUsed();
+ _subscriptions[subscriptionId] = sub = new SubscriptionRecord();
+ }
+
+ sub.Touched = ++_touch;
+ if (sub.Decisions.TryGetValue(sequenceNumber, out var recorded)) return recorded;
+ var standby = _standby;
+ DecisionBarrier?.Invoke();
+ sub.Decisions[sequenceNumber] = standby;
+ sub.Order.Enqueue(sequenceNumber);
+ while (sub.Order.Count > MaxPerSubscription && sub.Order.TryDequeue(out var oldest)) sub.Decisions.Remove(oldest);
+ return standby;
+ }
+ }
+
+ /// Drops the decisions of messages the subscription no longer retains (they can never be republished).
+ /// The subscription the Publish response belongs to.
+ /// The response's AvailableSequenceNumbers.
+ /// The sequence number just delivered (kept: it becomes available for Republish).
+ public void Prune(uint subscriptionId, IReadOnlyCollection available, uint justDelivered)
+ {
+ lock (_gate)
+ {
+ if (!_subscriptions.TryGetValue(subscriptionId, out var sub)) return;
+ var keep = new HashSet(available) { justDelivered };
+ var kept = new Queue();
+ while (sub.Order.TryDequeue(out var seq))
+ {
+ if (keep.Contains(seq)) kept.Enqueue(seq);
+ else sub.Decisions.Remove(seq);
+ }
+
+ sub.Order.Clear();
+ foreach (var seq in kept) sub.Order.Enqueue(seq);
+ }
+ }
+
+ /// Drops every record of subscriptions not in .
+ /// The ids of live subscriptions.
+ public void SweepDead(IReadOnlySet live)
+ {
+ lock (_gate)
+ {
+ foreach (var id in _subscriptions.Keys.Where(id => !live.Contains(id)).ToList()) _subscriptions.Remove(id);
+ }
+ }
+
+ /// Recorded decisions (test observation).
+ internal int Count
+ {
+ get { lock (_gate) return _subscriptions.Values.Sum(s => s.Decisions.Count); }
+ }
+
+ /// Recorded subscriptions (test observation).
+ internal int SubscriptionCount
+ {
+ get { lock (_gate) return _subscriptions.Count; }
+ }
+
+ private void EvictLeastRecentlyUsed()
+ {
+ var victim = _subscriptions.MinBy(kv => kv.Value.Touched).Key;
+ _subscriptions.Remove(victim);
+ }
+
+ private sealed class SubscriptionRecord
+ {
+ public readonly Dictionary Decisions = new();
+ public readonly Queue Order = new();
+ public long Touched;
+ }
+}
+
+///
+/// Rewrites a Standby delivery (spec §4.6): every marked ordinary-tag notification whose status is not Bad becomes
+/// BadOutOfService/null with its server timestamp kept. Builds new objects; the message handed in (the SDK's
+/// fresh per-delivery clone) and, a fortiori, the SDK's cached copy are never mutated.
+///
+internal static class StandbyNotificationRewriter
+{
+ /// Projects , or returns it unchanged when nothing needs rewriting.
+ /// A delivered message.
+ /// The number of notifications rewritten.
+ /// The message to deliver.
+ public static NotificationMessage Project(NotificationMessage message, out int rewritten)
+ {
+ rewritten = 0;
+ if (message is null || message.NotificationData.Count == 0) return message!;
+
+ var data = new List(message.NotificationData.Count);
+ var changed = false;
+ for (var i = 0; i < message.NotificationData.Count; i++)
+ {
+ var eo = message.NotificationData[i];
+ if (eo.TryGetValue(out var dcn) && dcn is not null)
+ {
+ List? items = null;
+ for (var j = 0; j < dcn.MonitoredItems.Count; j++)
+ {
+ var n = dcn.MonitoredItems[j];
+ if (OrdinaryTagMarker.IsOrdinaryTag(n) && !StatusCode.IsBad(n.Value.StatusCode))
+ {
+ items ??= CopyPrefix(dcn, j);
+ items.Add(new MonitoredItemNotification
+ {
+ ClientHandle = n.ClientHandle,
+ Value = new DataValue(Variant.Null, StatusCodes.BadOutOfService, DateTimeUtc.MinValue, n.Value.ServerTimestamp),
+ });
+ rewritten++;
+ }
+ else
+ {
+ items?.Add(n);
+ }
+ }
+
+ if (items is not null)
+ {
+ changed = true;
+ data.Add(new ExtensionObject(new DataChangeNotification { MonitoredItems = items, DiagnosticInfos = dcn.DiagnosticInfos }));
+ continue;
+ }
+ }
+
+ data.Add(eo);
+ }
+
+ if (!changed) return message;
+ return new NotificationMessage { SequenceNumber = message.SequenceNumber, PublishTime = message.PublishTime, NotificationData = data };
+ }
+
+ private static List CopyPrefix(DataChangeNotification dcn, int count)
+ {
+ var list = new List(dcn.MonitoredItems.Count);
+ for (var k = 0; k < count; k++) list.Add(dcn.MonitoredItems[k]);
+ return list;
+ }
+}
+
+///
+/// A sampled data-change item for an ordinary-tag Value (spec §4.6 item 1): after the SDK's own per-value
+/// publish (so the timestamp filter has run) it marks each notification it produced with an
+/// . Everything else is the SDK's .
+///
+internal sealed class OrdinaryTagMonitoredItem : MonitoredItem
+{
+ /// Initializes the item exactly as does.
+ public OrdinaryTagMonitoredItem(IServerInternal server, IAsyncNodeManager nodeManager, NodeHandle handle, uint subscriptionId,
+ uint id, MonitoredItemCreateRequest request, DiagnosticsMasks diagnosticsMasks, TimestampsToReturn timestampsToReturn,
+ MonitoringFilter? filter, Opc.Ua.Range? range, double samplingInterval, uint queueSize, bool durable)
+ : base(server, nodeManager, handle, subscriptionId, id, request.ItemToMonitor, diagnosticsMasks, timestampsToReturn,
+ request.MonitoringMode, request.RequestedParameters.ClientHandle, filter, filter, range, samplingInterval, queueSize,
+ request.RequestedParameters.DiscardOldest, 0, durable)
+ {
+ }
+
+ ///
+ public override bool Publish(OperationContext context, Queue notifications,
+ Queue diagnostics, uint maxNotificationsPerPublish, ILogger logger)
+ {
+ var local = new Queue();
+ var localDiagnostics = new Queue();
+ var more = base.Publish(context, local, localDiagnostics, maxNotificationsPerPublish, logger);
+ while (local.Count > 0)
+ {
+ var n = local.Dequeue();
+ n.Message = new OrdinaryTagMarker(n.ClientHandle, n.Value);
+ notifications.Enqueue(n);
+ if (localDiagnostics.Count > 0) diagnostics.Enqueue(localDiagnostics.Dequeue());
+ }
+
+ while (localDiagnostics.Count > 0) diagnostics.Enqueue(localDiagnostics.Dequeue());
+ return more;
+ }
+}
+
+///
+/// Installs for ordinary-tag ()
+/// Value items only. Re-implements on the SDK's
+/// public, non-sealed manager, so CustomNodeManager2's is MonitoredNodeMonitoredItemManager checks and
+/// its internal lifecycle interface keep working (T0c; the 2.0.1 MonitoredItemFactory hook exists only on
+/// AsyncCustomNodeManager). Installed only under Warm; Unset/None keep the SDK's manager.
+///
+internal sealed class OrdinaryTagItemManager(IAsyncNodeManager nodeManager, IServerInternal server)
+ : MonitoredNodeMonitoredItemManager(nodeManager, server), IMonitoredItemManager
+{
+ ///
+ ISampledDataChangeMonitoredItem IMonitoredItemManager.CreateMonitoredItem(IServerInternal server, IAsyncNodeManager nodeManager,
+ ServerSystemContext context, NodeHandle handle, uint subscriptionId, double publishingInterval,
+ DiagnosticsMasks diagnosticsMasks, TimestampsToReturn timestampsToReturn, MonitoredItemCreateRequest itemToCreate,
+ Opc.Ua.Range euRange, MonitoringFilter filterToUse, double samplingInterval, uint revisedQueueSize, bool createDurable,
+ MonitoredItemIdFactory monitoredItemIdFactory, Func addNodeToComponentCache)
+ {
+ if (handle.Node is not EquipmentVariableState || itemToCreate.ItemToMonitor.AttributeId != Attributes.Value)
+ {
+ return CreateMonitoredItem(server, nodeManager, context, handle, subscriptionId, publishingInterval, diagnosticsMasks,
+ timestampsToReturn, itemToCreate, euRange, filterToUse, samplingInterval, revisedQueueSize, createDurable,
+ monitoredItemIdFactory, addNodeToComponentCache);
+ }
+
+ // Mirrors MonitoredNodeMonitoredItemManager.CreateMonitoredItem in exact tagged 2.0.1, swapping only the item type.
+ var cachedNode = addNodeToComponentCache(context, handle, handle.Node);
+ if (!MonitoredNodes.TryGetValue(handle.Node.NodeId, out var monitoredNode))
+ {
+ MonitoredNodes[handle.Node.NodeId] = monitoredNode = new MonitoredNode2(nodeManager, server, cachedNode,
+ nodeManager.IsMultipleEventConsumerNode(cachedNode.NodeId));
+ }
+
+ handle.Node = monitoredNode.Node;
+ handle.MonitoredNode = monitoredNode;
+ uint id;
+ do
+ {
+ id = monitoredItemIdFactory.GetNextId();
+ }
+ while (!MonitoredItems.TryAdd(id, null!));
+
+ var item = new OrdinaryTagMonitoredItem(server, nodeManager, handle, subscriptionId, id, itemToCreate, diagnosticsMasks,
+ timestampsToReturn, filterToUse, euRange, samplingInterval, revisedQueueSize, createDurable);
+ monitoredNode.Add((IDataChangeMonitoredItem2)item);
+ MonitoredItems[id] = item;
+ return item;
+ }
+}
+
+/// The a Warm server hands the redundancy layer.
+internal sealed class SdkStandbyProjection(OtOpcUaNodeManager nodeManager, WarmStandbyState state)
+ : ZB.MOM.WW.OtOpcUa.Commons.OpcUa.IStandbyProjection
+{
+ ///
+ public void SetDataProjection(bool projected) => state.DataProjected = projected;
+
+ ///
+ public void SetStandbyDelivery(bool standby) => state.Delivery?.SetStandby(standby);
+
+ ///
+ public void SetAlarmSuppression(bool suppressed) => state.AlarmsSuppressed = suppressed;
+
+ ///
+ public int QueueCurrentToLiveItems(bool projected) => nodeManager.QueueCurrentToLiveItems(projected);
+
+ ///
+ public int RefreshAllEventItems() => nodeManager.RefreshAllEventItems();
+}
diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/DeploymentArtifact.cs b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/DeploymentArtifact.cs
index 882d3d408..40cb890c5 100644
--- a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/DeploymentArtifact.cs
+++ b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/DeploymentArtifact.cs
@@ -258,6 +258,52 @@ public static class DeploymentArtifact
}
}
+ ///
+ /// Gitea #717 spec §3: the ServerCluster.RedundancyMode of this node's cluster as the artifact carries it
+ /// (None/Warm/Hot), for the deploy-time cross-check against OpcUa:Redundancy:Mode.
+ /// A single-cluster artifact answers for its only cluster; a multi-cluster one for the cluster of this node's
+ /// ClusterNode row. Null when the artifact does not say (empty, malformed, no matching cluster).
+ ///
+ /// The deployment artifact.
+ /// This node's host:port id.
+ /// The mode name, or null.
+ public static string? ReadClusterRedundancyMode(ReadOnlySpan blob, string nodeId)
+ {
+ if (blob.IsEmpty) return null;
+ try
+ {
+ using var parsed = Open(blob.ToArray());
+ var root = parsed.Root;
+ if (!root.TryGetProperty("Clusters", out var clusters) || clusters.ValueKind != JsonValueKind.Array) return null;
+ var scope = parsed.ResolveClusterScope(nodeId);
+ JsonElement? match = null;
+ foreach (var cluster in clusters.EnumerateArray())
+ {
+ if (cluster.ValueKind != JsonValueKind.Object) continue;
+ if (scope.Mode == ClusterFilterMode.None) { match = cluster; break; }
+ if (scope.Mode == ClusterFilterMode.ScopeTo
+ && string.Equals(cluster.TryGetProperty("ClusterId", out var cid) && cid.ValueKind == JsonValueKind.String ? cid.GetString() : null, scope.ClusterId, StringComparison.OrdinalIgnoreCase))
+ {
+ match = cluster;
+ break;
+ }
+ }
+
+ if (match is not { } m || !m.TryGetProperty("RedundancyMode", out var mode)) return null;
+ return mode.ValueKind switch
+ {
+ JsonValueKind.String => mode.GetString(),
+ JsonValueKind.Number when mode.TryGetInt32(out var n) && Enum.IsDefined(typeof(Configuration.Enums.RedundancyMode), n)
+ => ((Configuration.Enums.RedundancyMode)n).ToString(),
+ _ => null,
+ };
+ }
+ catch (JsonException)
+ {
+ return null;
+ }
+ }
+
///
/// Resolve how a node should scope a deployment artifact to its own ClusterId. Single-cluster
/// (or legacy) artifacts resolve to so every existing
diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/DriverHostActor.cs b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/DriverHostActor.cs
index 352ef3b7a..7affe2891 100644
--- a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/DriverHostActor.cs
+++ b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/DriverHostActor.cs
@@ -789,7 +789,8 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS
DriverWriteBounds? writeBounds = null,
TimeProvider? writeClock = null,
IScriptRunner? scriptedAlarmRunner = null,
- TimeSpan? childShutdownTimeout = null) =>
+ TimeSpan? childShutdownTimeout = null,
+ bool enforceSnapshotOrdering = false) =>
// WARNING: this forwarding list is POSITIONAL, and Props.Create compiles it into an
// expression tree. Six IActorRef? parameters and several interface-typed ones mean a
// mis-ordered argument is usually type-compatible and therefore compiles clean, then binds
@@ -803,7 +804,8 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS
alarmStateStore, telemetryHub, primaryRoleSource, historizationRoleView,
sharesHistorizationScopeWithLocal, applyWedgeTimeout, reprovisionBackoff, commsRecoveryAuditDelay,
sharesAlarmQueueWithLocal, applyOutcomeTimeout, captureWhilePrimaryUnreachable, historyUpdateRoleView,
- scripting, ackPreflightView, writeBounds, writeClock, scriptedAlarmRunner, childShutdownTimeout));
+ scripting, ackPreflightView, writeBounds, writeClock, scriptedAlarmRunner, childShutdownTimeout,
+ enforceSnapshotOrdering));
/// Initializes a new DriverHostActor with the specified dependencies.
/// Database context factory for configuration database access.
@@ -938,8 +940,10 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS
DriverWriteBounds? writeBounds = null,
TimeProvider? writeClock = null,
IScriptRunner? scriptedAlarmRunner = null,
- TimeSpan? childShutdownTimeout = null)
+ TimeSpan? childShutdownTimeout = null,
+ bool enforceSnapshotOrdering = false)
{
+ _snapshotOrdering = enforceSnapshotOrdering ? new SnapshotRevisionGate() : null;
_childShutdownTimeout = childShutdownTimeout;
_scriptedAlarmRunner = scriptedAlarmRunner;
_writeBounds = writeBounds ?? DriverWriteBounds.Default;
@@ -2552,8 +2556,21 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS
/// the cached role unchanged; when this node has a real driver peer (count > 1) that omission is the
/// 03/S5 identity-mismatch shape, logged ONCE at Warning. Mirrors
/// ScriptedAlarmHostActor.OnRedundancyStateChanged / OpcUaPublishActor.
+ /// Gitea #717: the snapshot order gate under Warm; null (no ordering, today's behaviour) otherwise.
+ private readonly SnapshotRevisionGate? _snapshotOrdering;
+
private void OnRedundancyStateChanged(RedundancyStateChanged msg)
{
+ // Gitea #717 spec §4.2, under OpcUa:Redundancy:Mode=Warm only (Unset/None keep today's unconditional replace):
+ // a snapshot older than the last accepted one — a late heartbeat of an old singleton incarnation — is dropped,
+ // so it cannot overwrite the new incarnation's roles. An unrevisioned snapshot (pre-#717 publisher) still feeds
+ // the gates, as before.
+ if (_snapshotOrdering?.Offer(msg) == SnapshotRejection.OlderRevision)
+ {
+ OtOpcUaTelemetry.SnapshotRejected.Add(1, new KeyValuePair("reason", "older_revision"));
+ return;
+ }
+
_redundancySnapshotReceived = true;
var local = msg.Nodes.FirstOrDefault(n => n.NodeId == _localNode);
if (local is not null)
diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/OpcUa/OpcUaPublishActor.cs b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/OpcUa/OpcUaPublishActor.cs
index f90d229c4..d87737dff 100644
--- a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/OpcUa/OpcUaPublishActor.cs
+++ b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/OpcUa/OpcUaPublishActor.cs
@@ -14,6 +14,7 @@ using ZB.MOM.WW.OtOpcUa.Core.Authorization;
using ZB.MOM.WW.OtOpcUa.OpcUaServer;
using ZB.MOM.WW.OtOpcUa.Runtime.Drivers;
using ZB.MOM.WW.OtOpcUa.Runtime.Health;
+using ZB.MOM.WW.OtOpcUa.Runtime.Redundancy;
namespace ZB.MOM.WW.OtOpcUa.Runtime.OpcUa;
@@ -326,7 +327,9 @@ public sealed class OpcUaPublishActor : ReceiveActor, IWithTimers
bool dbLess = false,
IDataPlaneAuthoritySink? dataPlaneAuthority = null,
Func? utcNow = null,
- IOpcUaEndpointState? endpointState = null) =>
+ IOpcUaEndpointState? endpointState = null,
+ WarmRedundancySettings? redundancy = null,
+ IStandbyProjection? standbyProjection = null) =>
Akka.Actor.Props.Create(() => new OpcUaPublishActor(
sink ?? NullOpcUaAddressSpaceSink.Instance,
serviceLevel ?? NullServiceLevelPublisher.Instance,
@@ -341,7 +344,9 @@ public sealed class OpcUaPublishActor : ReceiveActor, IWithTimers
dbLess,
dataPlaneAuthority,
utcNow,
- endpointState)).WithDispatcher(DispatcherId);
+ endpointState,
+ redundancy,
+ standbyProjection)).WithDispatcher(DispatcherId);
/// Test-only Props that omits the pinned-dispatcher requirement and skips the
/// DPS subscribe so unit tests can spin up the actor on a vanilla TestKit cluster.
@@ -385,7 +390,9 @@ public sealed class OpcUaPublishActor : ReceiveActor, IWithTimers
bool dbLess = false,
IDataPlaneAuthoritySink? dataPlaneAuthority = null,
Func? utcNow = null,
- IOpcUaEndpointState? endpointState = null) =>
+ IOpcUaEndpointState? endpointState = null,
+ WarmRedundancySettings? redundancy = null,
+ IStandbyProjection? standbyProjection = null) =>
Akka.Actor.Props.Create(() => new OpcUaPublishActor(
sink ?? NullOpcUaAddressSpaceSink.Instance,
serviceLevel ?? NullServiceLevelPublisher.Instance,
@@ -400,7 +407,9 @@ public sealed class OpcUaPublishActor : ReceiveActor, IWithTimers
dbLess,
dataPlaneAuthority,
utcNow,
- endpointState));
+ endpointState,
+ redundancy,
+ standbyProjection));
/// Test-only Props exposing the deferred-write bound and the plan-step seams.
internal static Props PropsForTests(
@@ -461,10 +470,12 @@ public sealed class OpcUaPublishActor : ReceiveActor, IWithTimers
bool dbLess = false,
IDataPlaneAuthoritySink? dataPlaneAuthority = null,
Func? utcNow = null,
- IOpcUaEndpointState? endpointState = null)
+ IOpcUaEndpointState? endpointState = null,
+ WarmRedundancySettings? redundancy = null,
+ IStandbyProjection? standbyProjection = null)
: this(sink, serviceLevel, subscribeRedundancyTopic, localNode, dbFactory, applier, dbHealthProbe, staleWindow,
probeFreshnessWindow, healthTickInterval, dbLess, dataPlaneAuthority, utcNow, DefaultDeferredWriteCapacity, hooks: null,
- endpointState)
+ endpointState, redundancy, standbyProjection)
{
}
@@ -488,8 +499,12 @@ public sealed class OpcUaPublishActor : ReceiveActor, IWithTimers
Func? utcNow,
int deferredWriteCapacity,
TestHooks? hooks,
- IOpcUaEndpointState? endpointState)
+ IOpcUaEndpointState? endpointState,
+ WarmRedundancySettings? redundancy = null,
+ IStandbyProjection? standbyProjection = null)
{
+ _redundancy = redundancy ?? WarmRedundancySettings.Unset;
+ _standbyProjection = standbyProjection;
_deferred = new DeferredWrites(Math.Max(1, deferredWriteCapacity));
_hooks = hooks;
_sink = sink;
@@ -523,6 +538,12 @@ public sealed class OpcUaPublishActor : ReceiveActor, IWithTimers
Receive(_ => OnHealthTick());
Receive(HandlePeerProbe);
Receive(_ => { /* PubSub ack */ });
+ // Gitea #717 spec §4.1 / owner answer Q2: under Warm the verdict re-derives on membership and reachability, not
+ // only on snapshots. Nothing subscribes these under Unset/None.
+ Receive(_ => RecomputeServiceLevel());
+ Receive(_ => RecomputeServiceLevel());
+ Receive(_ => RecomputeServiceLevel());
+ Receive(_ => RecomputeServiceLevel());
}
///
@@ -543,6 +564,18 @@ public sealed class OpcUaPublishActor : ReceiveActor, IWithTimers
base.PreRestart(reason, message);
}
+ ///
+ protected override void PostStop()
+ {
+ if (_redundancy.IsWarm)
+ {
+ try { _cluster.Unsubscribe(Self); }
+ catch (Exception) { /* the cluster is shutting down with us */ }
+ }
+
+ base.PostStop();
+ }
+
///
protected override void PreStart()
{
@@ -551,6 +584,20 @@ public sealed class OpcUaPublishActor : ReceiveActor, IWithTimers
DistributedPubSub.Get(Context.System).Mediator.Tell(new Subscribe(RedundancyStateTopic, Self));
}
+ if (_redundancy.IsWarm)
+ {
+ OtOpcUaTelemetry.SetWarmRole(0);
+ try
+ {
+ _cluster.Subscribe(Self, ClusterEvent.InitialStateAsEvents,
+ typeof(ClusterEvent.IMemberEvent), typeof(ClusterEvent.IReachabilityEvent));
+ }
+ catch (Exception ex)
+ {
+ _log.Warning(ex, "OpcUaPublish: Warm redundancy could not subscribe to cluster membership; the verdict re-derives on snapshots only");
+ }
+ }
+
// DB-backed nodes refresh their cached probe status; DB-less nodes still need periodic
// snapshot/probe-verdict age checks even when no new messages arrive. Legacy nodes keep
// their role-only behavior without a timer.
@@ -672,7 +719,11 @@ public sealed class OpcUaPublishActor : ReceiveActor, IWithTimers
}
}
- private void HandleRebuild(RebuildAddressSpace msg) => StartRebuild(msg, Sender);
+ private void HandleRebuild(RebuildAddressSpace msg)
+ {
+ if (msg.Artifact is { Length: > 0 } artifact) CheckRedundancyMode(artifact);
+ StartRebuild(msg, Sender);
+ }
///
/// The entry half of a rebuild, on the actor thread. The raw-sink fallback (no applier) and the
@@ -1465,26 +1516,34 @@ public sealed class OpcUaPublishActor : ReceiveActor, IWithTimers
}
}
- private void HandleServiceLevelChanged(ServiceLevelChanged msg)
+ private void HandleServiceLevelChanged(ServiceLevelChanged msg) => TryPublishServiceLevel(msg.ServiceLevel);
+
+ /// The dedup/publish/metric step behind ; synchronous so a Warm
+ /// transition can order it against the projection steps (spec §4.7).
+ /// The level to publish.
+ /// True when the level is published (or already was).
+ private bool TryPublishServiceLevel(byte level)
{
// Always publish the FIRST computed level, even if it equals the byte-default 0. Otherwise a
// node starting Detached/role-less (first level = 0) would be dedup'd away, leaving the SDK's
// built-in default (255 = full service) standing — a degraded node wrongly advertising 255.
- if (_publishedAtLeastOnce && msg.ServiceLevel == _lastServiceLevel) return;
+ if (_publishedAtLeastOnce && level == _lastServiceLevel) return true;
try
{
- _serviceLevel.Publish(msg.ServiceLevel);
- _lastServiceLevel = msg.ServiceLevel;
+ _serviceLevel.Publish(level);
+ _lastServiceLevel = level;
_publishedAtLeastOnce = true;
OtOpcUaTelemetry.ServiceLevelChange.Add(1,
- new KeyValuePair("level", msg.ServiceLevel));
- _log.Debug("OpcUaPublish: ServiceLevel={Level}", msg.ServiceLevel);
+ new KeyValuePair("level", level));
+ _log.Debug("OpcUaPublish: ServiceLevel={Level}", level);
+ return true;
}
catch (Exception ex)
{
// A failed write may be partially applied; even a previous successful value must retry.
_publishedAtLeastOnce = false;
- _log.Warning(ex, "OpcUaPublish: ServiceLevel publisher threw at level {Level}", msg.ServiceLevel);
+ _log.Warning(ex, "OpcUaPublish: ServiceLevel publisher threw at level {Level}", level);
+ return false;
}
}
@@ -1494,6 +1553,28 @@ public sealed class OpcUaPublishActor : ReceiveActor, IWithTimers
/// wired) or via the legacy role-only seam (back-compat / bootstrap).
private void HandleRedundancyStateChanged(RedundancyStateChanged msg)
{
+ if (_redundancy.IsWarm)
+ {
+ // Gitea #717 spec §4.2: under Warm a snapshot older than the last accepted one (a late heartbeat from an old
+ // singleton incarnation) is dropped outright. An unrevisioned one (a pre-#717 publisher mid-upgrade) still
+ // feeds the health inputs but can never make this node Active. Unset/None keep today's unconditional
+ // replace below.
+ switch (_revisionGate.Offer(msg))
+ {
+ case SnapshotRejection.OlderRevision:
+ OtOpcUaTelemetry.SnapshotRejected.Add(1, new KeyValuePair("reason", "older_revision"));
+ _log.Debug("OpcUaPublish: dropped redundancy snapshot {Revision} (last accepted {Last})",
+ msg.Revision, _revisionGate.LastAccepted);
+ return;
+ case SnapshotRejection.Unrevisioned:
+ OtOpcUaTelemetry.SnapshotRejected.Add(1, new KeyValuePair("reason", "unrevisioned"));
+ break;
+ default:
+ _acceptedSnapshot = msg;
+ break;
+ }
+ }
+
_lastSnapshot = msg;
RecomputeServiceLevel();
}
@@ -1576,6 +1657,11 @@ public sealed class OpcUaPublishActor : ReceiveActor, IWithTimers
private void RecomputeServiceLevel()
{
if (_localNode is null || _lastSnapshot is null) return;
+ if (_redundancy.IsWarm)
+ {
+ RecomputeWarm();
+ return;
+ }
var entry = _lastSnapshot.Nodes.FirstOrDefault(n => n.NodeId == _localNode.Value);
@@ -1635,6 +1721,197 @@ public sealed class OpcUaPublishActor : ReceiveActor, IWithTimers
Self.Tell(new ServiceLevelChanged(ServiceLevelCalculator.Compute(inputs)));
}
+ // ---------------------------------------------------------------- Warm redundancy (Gitea #717)
+
+ private sealed class WarmRetry { public static readonly WarmRetry Instance = new(); private WarmRetry() { } }
+
+ private readonly WarmRedundancySettings _redundancy;
+ private readonly IStandbyProjection? _standbyProjection;
+ private readonly SnapshotRevisionGate _revisionGate = new();
+ /// The latest ACCEPTED revisioned snapshot — the only one the Warm verdict reads (spec §4.2).
+ private RedundancyStateChanged? _acceptedSnapshot;
+ /// The role the data plane is in. A Warm node manager boots Standby, so this starts Standby.
+ private WarmRole _warmRole = WarmRole.Standby;
+ /// A promotion whose Active ServiceLevel publish failed (spec §4.7): the data plane is Active and the
+ /// level is retried; the Standby level stays visible meanwhile, never the reverse.
+ private bool _promotionLevelPending;
+ private DateTime? _disagreeSinceUtc;
+ private bool _disagreementWarned;
+
+ /// How long membership and snapshot may disagree before a Warning (two heartbeats, spec §4.1).
+ internal static readonly TimeSpan WarmDisagreementWarning = TimeSpan.FromSeconds(20);
+
+ /// The current Warm role (test observation).
+ internal WarmRole CurrentWarmRole => _warmRole;
+
+ /// A promoted node whose Active ServiceLevel publish is being retried (test observation).
+ internal bool PromotionLevelPending => _promotionLevelPending;
+
+ ///
+ /// The Warm ServiceLevel + role path (spec §4.1, §4.3, §4.7): the health facts the non-Warm paths use, the
+ /// Warm verdict, then a transition in the fixed step order when the verdict changed, else just the band.
+ ///
+ private void RecomputeWarm()
+ {
+ var self = _localNode!.Value;
+ var entry = _lastSnapshot!.Nodes.FirstOrDefault(n => n.NodeId == self);
+ var now = _utcNow();
+ var memberState = SafeSelfStatus();
+ var listening = EndpointListening();
+
+ bool dbReachable;
+ bool stale;
+ if (_dbLess || _dbHealthProbe is null || _lastDbHealth is null)
+ {
+ dbReachable = true;
+ stale = entry is null || (now - entry.AsOfUtc) > _staleWindow;
+ }
+ else
+ {
+ dbReachable = _lastDbHealth.Reachable;
+ stale = !_lastDbHealth.Reachable
+ || (now - _lastDbHealth.AsOfUtc) > _staleWindow
+ || entry is null || (now - entry.AsOfUtc) > _staleWindow;
+ }
+
+ var verdict = EvaluateWarmVerdict(self, memberState, listening);
+ TrackDisagreement(verdict, now);
+
+ var detached = entry is null || entry.Role == RedundancyRole.Detached;
+ var inputs = new NodeHealthInputs(memberState, dbReachable, OpcUaProbeOk(), stale, IsDriverPrimary: false, listening);
+ var target = detached ? WarmRole.Standby : verdict.Role;
+ var level = detached ? (byte)0 : ServiceLevelCalculator.ComputeWarm(inputs, target == WarmRole.Active);
+ ApplyWarmTransition(target, level, verdict.Reason);
+ }
+
+ private WarmVerdict EvaluateWarmVerdict(NodeId self, MemberStatus memberState, bool listening)
+ {
+ NodeId? oldest = null, oldestReachable = null;
+ var primaryUnreachable = false;
+ var snapshotPrimary = _acceptedSnapshot?.Nodes.FirstOrDefault(n => n.Role == RedundancyRole.Primary)?.NodeId;
+ try
+ {
+ var state = _cluster.State;
+ var unreachable = state.Unreachable.ToHashSet();
+ oldest = WarmRoleAuthority.OldestUpMember(state.Members, "driver");
+ oldestReachable = WarmRoleAuthority.OldestUpMember(state.Members, "driver", unreachable);
+ primaryUnreachable = snapshotPrimary is { } sp && unreachable.Any(m => WarmRoleAuthority.ToNodeId(m) == sp);
+ }
+ catch (Exception ex)
+ {
+ _log.Debug(ex, "OpcUaPublish: cluster state unavailable for the Warm verdict; treating membership as unknown");
+ }
+
+ return WarmRoleAuthority.Evaluate(new WarmRoleInputs(
+ self, memberState == MemberStatus.Up, listening, oldest, oldestReachable,
+ HasAcceptedSnapshot: _acceptedSnapshot is not null, snapshotPrimary, primaryUnreachable,
+ _redundancy.PromoteWhilePrimaryUnreachable));
+ }
+
+ private void TrackDisagreement(WarmVerdict verdict, DateTime now)
+ {
+ if (!verdict.Disagreement)
+ {
+ _disagreeSinceUtc = null;
+ _disagreementWarned = false;
+ return;
+ }
+
+ _disagreeSinceUtc ??= now;
+ if (_disagreementWarned || now - _disagreeSinceUtc.Value <= WarmDisagreementWarning) return;
+ _disagreementWarned = true;
+ _log.Warning(
+ "OpcUaPublish: Warm redundancy — cluster membership and the redundancy snapshot (revision {Revision}) have disagreed about this node being Primary for more than {Seconds}s; the node stays {Role}",
+ _revisionGate.LastAccepted, WarmDisagreementWarning.TotalSeconds, _warmRole);
+ }
+
+ ///
+ /// Moves the data plane to in the spec §4.7 order and publishes
+ /// . Invariant: an Active-band ServiceLevel is never visible while the projection is
+ /// closed — a promotion publishes the level LAST and, if that publish fails, retries it; a demotion publishes
+ /// the Standby level FIRST and degrades the data whether or not that publish succeeded (failing safe).
+ ///
+ private void ApplyWarmTransition(WarmRole target, byte level, string reason)
+ {
+ if (target == _warmRole)
+ {
+ if (!TryPublishServiceLevel(level) && target == WarmRole.Active) ScheduleWarmRetry();
+ else if (target == WarmRole.Active) _promotionLevelPending = false;
+ return;
+ }
+
+ var projection = _standbyProjection;
+ if (target == WarmRole.Active)
+ {
+ projection?.SetDataProjection(false); // 1. open the read projection
+ projection?.QueueCurrentToLiveItems(false); // 2. current Good value to every live data item
+ projection?.SetStandbyDelivery(false); // 3. first deliveries decided Active
+ projection?.SetAlarmSuppression(false); // 4. new alarm events delivered
+ projection?.RefreshAllEventItems(); // 5. promotion refresh on every live event item
+ _warmRole = WarmRole.Active;
+ if (TryPublishServiceLevel(level)) // 6. Active ServiceLevel, last
+ {
+ _promotionLevelPending = false;
+ }
+ else
+ {
+ _promotionLevelPending = true;
+ ScheduleWarmRetry();
+ }
+ }
+ else
+ {
+ TryPublishServiceLevel(level); // 1. Standby ServiceLevel, first (continue on failure)
+ projection?.SetStandbyDelivery(true); // 2. the D4 boundary
+ projection?.SetAlarmSuppression(true); // 3. suppress new alarm events
+ projection?.SetDataProjection(true); // 4. close the read projection
+ projection?.QueueCurrentToLiveItems(true); // 5. BadOutOfService to every live data item
+ _warmRole = WarmRole.Standby;
+ _promotionLevelPending = false;
+ }
+
+ OtOpcUaTelemetry.SetWarmRole(_warmRole == WarmRole.Active ? 1 : 0);
+ OtOpcUaTelemetry.WarmTransition.Add(1,
+ new KeyValuePair("direction", target == WarmRole.Active ? "promote" : "demote"),
+ new KeyValuePair("reason", reason));
+ _log.Info(
+ "OpcUaPublish: Warm redundancy {Direction} to {Role} (reason={Reason}, revision={Revision}, ServiceLevel={Level})",
+ target == WarmRole.Active ? "promotion" : "demotion", _warmRole, reason, _revisionGate.LastAccepted, level);
+ }
+
+ ///
+ /// Gitea #717 spec §3 deploy-time cross-check: under None or Warm, an artifact whose cluster
+ /// RedundancyMode disagrees with the local mode (Hot included, owner answer Q5) logs an Error and
+ /// counts otopcua.redundancy.mode_mismatch. The node keeps its local mode — the only value available
+ /// before SDK start. Unset has opted out and is not checked.
+ ///
+ /// The applied artifact bytes.
+ private void CheckRedundancyMode(byte[] artifact)
+ {
+ if (_redundancy.Mode == OpcUaRedundancyMode.Unset || _localNode is null) return;
+ string? clusterMode;
+ try
+ {
+ clusterMode = DeploymentArtifact.ReadClusterRedundancyMode(artifact, _localNode.Value.Value);
+ }
+ catch (Exception ex)
+ {
+ _log.Debug(ex, "OpcUaPublish: could not read the artifact's cluster RedundancyMode");
+ return;
+ }
+
+ if (clusterMode is null) return;
+ if (string.Equals(clusterMode, _redundancy.Mode.ToString(), StringComparison.OrdinalIgnoreCase)) return;
+ OtOpcUaTelemetry.RedundancyModeMismatch.Add(1,
+ new KeyValuePair("local", _redundancy.Mode.ToString()),
+ new KeyValuePair("cluster", clusterMode));
+ _log.Error(
+ "OpcUaPublish: the deployed ServerCluster.RedundancyMode for this node's cluster is {ClusterMode} but OpcUa:Redundancy:Mode is {LocalMode}; the node keeps running as {LocalMode} (local configuration is the authority — fix whichever is wrong)",
+ clusterMode, _redundancy.Mode, _redundancy.Mode);
+ }
+
+ private void ScheduleWarmRetry() => Timers.StartSingleTimer("warm-retry", WarmRetry.Instance, TimeSpan.FromSeconds(1));
+
/// The legacy role-only ServiceLevel switch (primary-leader → 240, primary → 200,
/// secondary → 100, _ → 0). Preserved as the back-compat / bootstrap seam.
private static byte LegacyRoleOnly(NodeRedundancyState entry) => entry.Role switch
diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Redundancy/WarmRoleAuthority.cs b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Redundancy/WarmRoleAuthority.cs
new file mode 100644
index 000000000..6a96fbaaf
--- /dev/null
+++ b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Redundancy/WarmRoleAuthority.cs
@@ -0,0 +1,109 @@
+using Akka.Cluster;
+using ZB.MOM.WW.OtOpcUa.Commons.Types;
+
+namespace ZB.MOM.WW.OtOpcUa.Runtime.Redundancy;
+
+/// The Warm role (Gitea #717 spec §4.1).
+public enum WarmRole
+{
+ /// Degraded standby: projected data, suppressed new alarm events, Degraded-band ServiceLevel.
+ Standby,
+
+ /// Serves data and alarms; Healthy-band ServiceLevel.
+ Active,
+}
+
+/// The facts the Warm verdict is a pure function of.
+/// This node's canonical host:port id.
+/// This node's own member status is Up.
+/// This node's OPC UA endpoint did not fail to start (#762).
+/// The oldest Up driver member in this node's OWN cluster state (the singleton
+/// placement rule), or null when there is none.
+/// The same, ignoring members this node sees as unreachable.
+/// A revisioned snapshot has been accepted (spec §4.2). Unrevisioned snapshots
+/// never count.
+/// The Primary named by the latest accepted revisioned snapshot, or null.
+/// This node sees as unreachable.
+/// The owner-answer-Q2 option (default on).
+public readonly record struct WarmRoleInputs(
+ NodeId Self,
+ bool SelfUp,
+ bool EndpointListening,
+ NodeId? OldestUpDriver,
+ NodeId? OldestReachableUpDriver,
+ bool HasAcceptedSnapshot,
+ NodeId? SnapshotPrimary,
+ bool SnapshotPrimaryUnreachable,
+ bool PromoteWhilePrimaryUnreachable);
+
+/// The verdict and why.
+/// Active or Standby.
+/// A short reason for the transition metric/log: boot, endpoint,
+/// membership, snapshot, unreachable-primary or agreed.
+/// Membership (2) and snapshot (3) disagree about whether this node is Primary.
+public readonly record struct WarmVerdict(WarmRole Role, string Reason, bool Disagreement);
+
+///
+/// The Warm role verdict (Gitea #717 spec §4.1), a pure function so every row is unit-testable. Active iff the
+/// node is Up and listening, its own cluster state names it the oldest Up driver member, and the latest accepted
+/// revisioned snapshot names it Primary. Anything else — no snapshot yet, an unrevisioned snapshot, a
+/// disagreement — is Standby. With the survivor of a
+/// crashed Primary goes Active once that Primary is unreachable, before auto-down removes it.
+///
+///
+/// This is a separate output from PrimaryGatePolicy, whose lone-node allow is unchanged for writes and
+/// acks (D7).
+///
+public static class WarmRoleAuthority
+{
+ /// Evaluates the verdict.
+ /// The inputs.
+ /// The verdict.
+ public static WarmVerdict Evaluate(in WarmRoleInputs i)
+ {
+ if (!i.SelfUp) return new(WarmRole.Standby, "membership", false);
+ if (!i.EndpointListening) return new(WarmRole.Standby, "endpoint", false);
+ if (!i.HasAcceptedSnapshot) return new(WarmRole.Standby, "boot", false);
+
+ var membershipSaysSelf = i.OldestUpDriver == i.Self;
+ var snapshotSaysSelf = i.SnapshotPrimary == i.Self;
+ var disagreement = membershipSaysSelf != snapshotSaysSelf;
+
+ if (membershipSaysSelf && snapshotSaysSelf) return new(WarmRole.Active, "agreed", false);
+
+ // Owner answer Q2: a SIGKILLed Primary stays an Up-but-unreachable member until auto-down removes it, and the
+ // singleton usually died with it. Evaluate (2) and (3) with unreachable members excluded.
+ if (i.PromoteWhilePrimaryUnreachable
+ && i.OldestReachableUpDriver == i.Self
+ && (snapshotSaysSelf || (i.SnapshotPrimary is not null && i.SnapshotPrimaryUnreachable)))
+ {
+ return new(WarmRole.Active, "unreachable-primary", disagreement);
+ }
+
+ return new(WarmRole.Standby, membershipSaysSelf ? "snapshot" : "membership", disagreement);
+ }
+
+ /// The oldest Up member carrying (the singleton placement rule), optionally
+ /// ignoring members.
+ /// The current members.
+ /// The role.
+ /// Members to ignore, or null.
+ /// Its canonical node id, or null.
+ public static NodeId? OldestUpMember(IEnumerable members, string role, IReadOnlySet? unreachable = null)
+ {
+ Member? oldest = null;
+ foreach (var m in members)
+ {
+ if (m.Status != MemberStatus.Up || !m.HasRole(role)) continue;
+ if (unreachable is not null && unreachable.Contains(m)) continue;
+ if (oldest is null || m.IsOlderThan(oldest)) oldest = m;
+ }
+
+ return oldest is null ? null : ToNodeId(oldest);
+ }
+
+ /// The canonical host:port id of a member (the RedundancyStateActor format).
+ /// The member.
+ /// The node id.
+ public static NodeId ToNodeId(Member m) => NodeId.Parse($"{m.Address.Host ?? string.Empty}:{m.Address.Port ?? 0}");
+}
diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/ServiceCollectionExtensions.cs b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/ServiceCollectionExtensions.cs
index 8acc830be..54cb60809 100644
--- a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/ServiceCollectionExtensions.cs
+++ b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/ServiceCollectionExtensions.cs
@@ -566,6 +566,9 @@ public static class ServiceCollectionExtensions
// Optional: a harness that builds this graph without the hosted service registers none, and both
// consumers then behave as if the endpoint were listening (the pre-#762 behaviour).
var endpointState = resolver.GetService();
+ // Gitea #717: OpcUa:Redundancy (registered by the Host from the validated OpcUa options). Absent in harnesses ⇒
+ // Unset ⇒ today's behaviour. The standby projection is late-bound like the ServiceLevel publisher.
+ var redundancy = resolver.GetService() ?? WarmRedundancySettings.Unset;
var publishActor = system.ActorOf(
OpcUaPublishActor.Props(
sink: addressSpaceSink,
@@ -576,7 +579,9 @@ public static class ServiceCollectionExtensions
dbHealthProbe: dbHealth,
dbLess: dbFactory is null,
dataPlaneAuthority: dataPlaneAuthority,
- endpointState: endpointState),
+ endpointState: endpointState,
+ redundancy: redundancy,
+ standbyProjection: redundancy.IsWarm ? resolver.GetService() : null),
OpcUaPublishActorName);
registry.Register(publishActor);
@@ -670,7 +675,9 @@ public static class ServiceCollectionExtensions
// lmxopcua#780: the out-of-process runner when Scripting:Runner:Seams lists ScriptedAlarm. Null under
// the InProcess default, when the seam is not listed, and in test graphs that register no provider.
scriptedAlarmRunner: resolver.GetService()
- ?.ForSeam(ZB.MOM.WW.OtOpcUa.Core.Scripting.ScriptSeam.ScriptedAlarm)),
+ ?.ForSeam(ZB.MOM.WW.OtOpcUa.Core.Scripting.ScriptSeam.ScriptedAlarm),
+ // Gitea #717 spec §4.2: drop older redundancy snapshots under Warm only.
+ enforceSnapshotOrdering: redundancy.IsWarm),
DriverHostActorName);
registry.Register(driverHost);
diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.ControlPlane.Tests/WarmRedundancyRevisionTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.ControlPlane.Tests/WarmRedundancyRevisionTests.cs
new file mode 100644
index 000000000..b80af2c5a
--- /dev/null
+++ b/tests/Server/ZB.MOM.WW.OtOpcUa.ControlPlane.Tests/WarmRedundancyRevisionTests.cs
@@ -0,0 +1,154 @@
+using Akka.Actor;
+using Akka.Cluster;
+using Shouldly;
+using Xunit;
+using ZB.MOM.WW.OtOpcUa.Cluster.Redundancy;
+using ZB.MOM.WW.OtOpcUa.Commons.Messages.Redundancy;
+using ZB.MOM.WW.OtOpcUa.Commons.Types;
+using ZB.MOM.WW.OtOpcUa.ControlPlane.Redundancy;
+using ZB.MOM.WW.OtOpcUa.ControlPlane.Tests.Harness;
+
+namespace ZB.MOM.WW.OtOpcUa.ControlPlane.Tests;
+
+///
+/// Gitea #717 spec §4.2 (ordered snapshots) and §4.3 (Warm bands): the revision the singleton stamps, its order,
+/// the subscriber gate, the wire shape of the extended message, and the Warm ServiceLevel truth table.
+///
+public sealed class WarmRedundancyRevisionTests : ControlPlaneActorTestBase
+{
+ [Fact]
+ public void The_singleton_stamps_an_increasing_revision_on_every_publish_heartbeats_included()
+ {
+ var probe = CreateTestProbe();
+ Sys.ActorOf(RedundancyStateActor.Props(msg => probe.Ref.Tell(msg), TimeSpan.FromMilliseconds(200)));
+ var first = probe.ExpectMsg(TimeSpan.FromSeconds(5), cancellationToken: TestContext.Current.CancellationToken);
+ var second = probe.ExpectMsg(TimeSpan.FromSeconds(5), cancellationToken: TestContext.Current.CancellationToken);
+
+ first.Revision.ShouldNotBeNull();
+ second.Revision.ShouldNotBeNull();
+ second.Revision!.CompareTo(first.Revision).ShouldBeGreaterThan(0);
+ second.Revision.Sequence.ShouldBe(first.Revision!.Sequence + 1);
+ first.Revision.PublisherUpNumber.ShouldBeGreaterThan(0L, "the publisher's Akka up-number");
+ }
+
+ /// Akka.NET keeps Member.UpNumber internal; an upgrade that renames it must fail here, not degrade
+ /// every revision to up-number 0 silently.
+ [Fact]
+ public void Akka_member_up_number_is_readable()
+ {
+ AwaitCondition(() => Akka.Cluster.Cluster.Get(Sys).SelfMember.Status == MemberStatus.Up, TimeSpan.FromSeconds(10),
+ cancellationToken: TestContext.Current.CancellationToken);
+ RedundancyStateActor.ReadUpNumber(Akka.Cluster.Cluster.Get(Sys).SelfMember).ShouldNotBeNull().ShouldBeGreaterThan(0);
+ }
+
+ [Fact]
+ public void A_singleton_restarted_in_the_same_process_outranks_its_predecessor()
+ {
+ var a = RedundancyStateActor.NextIncarnationTick();
+ var b = RedundancyStateActor.NextIncarnationTick();
+ b.ShouldBeGreaterThan(a, "strictly increasing even within one tick");
+ new SnapshotRevision(1, b, 1).CompareTo(new SnapshotRevision(1, a, 500)).ShouldBeGreaterThan(0);
+ }
+
+ [Fact]
+ public void Revisions_order_by_up_number_then_incarnation_then_sequence()
+ {
+ var baseline = new SnapshotRevision(2, 100, 50);
+ new SnapshotRevision(3, 1, 1).CompareTo(baseline).ShouldBeGreaterThan(0, "a handover to a younger member outranks");
+ new SnapshotRevision(1, 999, 999).CompareTo(baseline).ShouldBeLessThan(0, "a departed older member's late publish loses");
+ new SnapshotRevision(2, 101, 1).CompareTo(baseline).ShouldBeGreaterThan(0);
+ new SnapshotRevision(2, 100, 51).CompareTo(baseline).ShouldBeGreaterThan(0);
+ new SnapshotRevision(2, 100, 50).CompareTo(baseline).ShouldBe(0);
+ }
+
+ [Fact]
+ public void The_gate_accepts_only_newer_revisions_and_never_advances_on_an_unrevisioned_snapshot()
+ {
+ var gate = new SnapshotRevisionGate();
+ gate.Offer(Snap(null)).ShouldBe(SnapshotRejection.Unrevisioned);
+ gate.LastAccepted.ShouldBeNull();
+ gate.Offer(Snap(new(2, 1, 5))).ShouldBe(SnapshotRejection.None);
+ gate.Offer(Snap(new(1, 9, 9))).ShouldBe(SnapshotRejection.OlderRevision, "an old incarnation's late heartbeat");
+ gate.Offer(Snap(new(2, 1, 5))).ShouldBe(SnapshotRejection.OlderRevision, "a duplicate");
+ gate.Offer(Snap(new(2, 1, 6))).ShouldBe(SnapshotRejection.None);
+ gate.LastAccepted.ShouldBe(new SnapshotRevision(2, 1, 6));
+ }
+
+ [Fact]
+ public void The_extended_message_round_trips_and_a_pre_717_payload_decodes_with_no_revision()
+ {
+ var msg = Snap(new SnapshotRevision(3, 42, 7));
+ var serializer = Sys.Serialization.FindSerializerFor(msg);
+ var manifest = Akka.Serialization.Serialization.ManifestFor(serializer, msg);
+ var round = (RedundancyStateChanged)Sys.Serialization.Deserialize(serializer.ToBinary(msg), serializer.Identifier, manifest);
+ round.Revision.ShouldBe(msg.Revision);
+ round.Nodes.Single().NodeId.ShouldBe(msg.Nodes.Single().NodeId);
+
+ // A pre-#717 publisher's payload: the same message without the field.
+ var legacy = Snap(null);
+ var legacyBytes = serializer.ToBinary(legacy);
+ var text = System.Text.Encoding.UTF8.GetString(legacyBytes);
+ var stripped = System.Text.Encoding.UTF8.GetBytes(text.Replace(",\"Revision\":null", string.Empty, StringComparison.Ordinal));
+ var decoded = (RedundancyStateChanged)Sys.Serialization.Deserialize(stripped, serializer.Identifier, manifest);
+ decoded.Revision.ShouldBeNull();
+ decoded.Nodes.Count.ShouldBe(1);
+ }
+
+ private static RedundancyStateChanged Snap(SnapshotRevision? revision) =>
+ new(new[] { new NodeRedundancyState(NodeId.Parse("a:1"), RedundancyRole.Primary, true, true, DateTime.UtcNow) },
+ CorrelationId.NewId(), revision);
+}
+
+/// Gitea #717 spec §4.3: the Warm ServiceLevel bands (pure; no actor system).
+public sealed class WarmServiceLevelBandTests
+{
+
+ public static IEnumerable