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 d87737dff..edf98c618 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/OpcUa/OpcUaPublishActor.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/OpcUa/OpcUaPublishActor.cs @@ -540,9 +540,21 @@ public sealed class OpcUaPublishActor : ReceiveActor, IWithTimers 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(e => + { + TrackMember(e); + RecomputeServiceLevel(); + }); + Receive(e => + { + TrackReachability(e); + RecomputeServiceLevel(); + }); + Receive(state => + { + ResetWarmView(state); + RecomputeServiceLevel(); + }); Receive(_ => RecomputeServiceLevel()); } @@ -1730,6 +1742,61 @@ public sealed class OpcUaPublishActor : ReceiveActor, IWithTimers private readonly SnapshotRevisionGate _revisionGate = new(); /// The latest ACCEPTED revisioned snapshot — the only one the Warm verdict reads (spec §4.2). private RedundancyStateChanged? _acceptedSnapshot; + + // Gitea #717 T13 live gate: the Warm verdict reads membership from THIS actor's own event subscription, not from + // Cluster.State / Cluster.SelfMember. Those come from Akka's ClusterReadView, a separate subscriber to the same + // events, so when this actor handles an event the read view may not show it yet. Live, that held a crash survivor + // Standby for 4.3 s after its own UnreachableMember event (promotion waited for the next health tick), and let a + // gracefully leaving Primary keep serving Good ~240 ms after the survivor was already Active. + private readonly Dictionary _warmMembers = []; + private readonly HashSet
_warmUnreachable = []; + + // Gitea #717 T13: nodes seen removed from the cluster (and not back since). A snapshot naming one is as gone as an + // unreachable one, so the survivor of a crash stays Active through auto-down's removal (WarmRoleAuthority.SnapshotPrimaryGone). + private readonly HashSet _removedMembers = []; + + private void TrackMember(ClusterEvent.IMemberEvent e) + { + var node = WarmRoleAuthority.ToNodeId(e.Member); + if (e is ClusterEvent.MemberRemoved) + { + _warmMembers.Remove(e.Member.Address); + _warmUnreachable.Remove(e.Member.Address); + _removedMembers.Add(node); + return; + } + + _warmMembers[e.Member.Address] = e.Member; + if (e is ClusterEvent.MemberJoined or ClusterEvent.MemberUp) _removedMembers.Remove(node); + } + + private void TrackReachability(ClusterEvent.IReachabilityEvent e) + { + if (e is ClusterEvent.UnreachableMember u) _warmUnreachable.Add(u.Member.Address); + else if (e is ClusterEvent.ReachableMember r) _warmUnreachable.Remove(r.Member.Address); + } + + private void ResetWarmView(ClusterEvent.CurrentClusterState state) + { + _warmMembers.Clear(); + _warmUnreachable.Clear(); + foreach (var m in state.Members) _warmMembers[m.Address] = m; + foreach (var m in state.Unreachable) _warmUnreachable.Add(m.Address); + } + + /// The Warm view's own member status, falling back to the read view before the first event arrives. + private MemberStatus WarmSelfStatus() + { + try + { + return _warmMembers.TryGetValue(_cluster.SelfAddress, out var self) ? self.Status : SafeSelfStatus(); + } + catch (Exception ex) + { + _log.Debug(ex, "OpcUaPublish: self address unavailable for the Warm view; using the read view"); + return SafeSelfStatus(); + } + } /// 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 @@ -1756,7 +1823,7 @@ public sealed class OpcUaPublishActor : ReceiveActor, IWithTimers var self = _localNode!.Value; var entry = _lastSnapshot!.Nodes.FirstOrDefault(n => n.NodeId == self); var now = _utcNow(); - var memberState = SafeSelfStatus(); + var memberState = WarmSelfStatus(); var listening = EndpointListening(); bool dbReachable; @@ -1791,11 +1858,23 @@ public sealed class OpcUaPublishActor : ReceiveActor, IWithTimers 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); + IReadOnlyCollection members; + HashSet unreachable; + if (_warmMembers.Count > 0) + { + members = _warmMembers.Values; + unreachable = _warmMembers.Values.Where(m => _warmUnreachable.Contains(m.Address)).ToHashSet(); + } + else + { + var state = _cluster.State; + members = state.Members; + unreachable = state.Unreachable.ToHashSet(); + } + + oldest = WarmRoleAuthority.OldestUpMember(members, "driver"); + oldestReachable = WarmRoleAuthority.OldestUpMember(members, "driver", unreachable); + primaryUnreachable = WarmRoleAuthority.SnapshotPrimaryGone(snapshotPrimary, members, unreachable, _removedMembers); } catch (Exception ex) { diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Redundancy/WarmRoleAuthority.cs b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Redundancy/WarmRoleAuthority.cs index 6a96fbaaf..df7c46763 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Redundancy/WarmRoleAuthority.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Redundancy/WarmRoleAuthority.cs @@ -102,6 +102,36 @@ public static class WarmRoleAuthority return oldest is null ? null : ToNodeId(oldest); } + /// + /// Whether the snapshot's Primary is gone from this node's view, for : + /// unreachable, marked Down, or removed from the cluster after this node saw it as a member. + /// + /// + /// The "removed" arm is the end of the same crash path (T13 live gate, 2026-11-13). Auto-down removes the + /// unreachable Primary about 15 s after the survivor promoted, so it leaves the unreachable set. But the + /// survivor's own singleton does not publish a snapshot naming it until the hand-over retries time out, about + /// 15 s later. Without this arm the survivor demoted for that whole window (BadOutOfService, ServiceLevel 50) + /// and then promoted again. A member that is merely Leaving or Exiting is still present and reachable, so a + /// graceful hand-over is unaffected, and a node this one never saw as a member is not assumed gone. + /// + /// The Primary named by the accepted snapshot, or null. + /// The current members (this node's own cluster state). + /// The members this node sees as unreachable. + /// Nodes this node has seen removed from the cluster (and not rejoin since). + /// True when a named Primary is unreachable, Down, or removed. + public static bool SnapshotPrimaryGone(NodeId? snapshotPrimary, IEnumerable members, IReadOnlySet unreachable, + IReadOnlySet removed) + { + if (snapshotPrimary is not { } sp) return false; + foreach (var m in members) + { + if (ToNodeId(m) != sp) continue; + return m.Status is MemberStatus.Down or MemberStatus.Removed || unreachable.Contains(m); + } + + return removed.Contains(sp); + } + /// The canonical host:port id of a member (the RedundancyStateActor format). /// The member. /// The node id. diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/OpcUa/OpcUaPublishActorWarmTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/OpcUa/OpcUaPublishActorWarmTests.cs index 639c7270a..1c83c8f81 100644 --- a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/OpcUa/OpcUaPublishActorWarmTests.cs +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/OpcUa/OpcUaPublishActorWarmTests.cs @@ -157,6 +157,68 @@ public sealed class OpcUaPublishActorWarmTests : RuntimeActorTestBase AssertNoActiveLevelOverProjectedData(); } + /// + /// T13 live gate (2026-11-13), two findings on the crash path, driven here with synthetic cluster events that + /// Akka's read view (Cluster.State) never sees: + /// (1) the verdict reacts to the event it is handling, not to the read view, which can lag it (live: the survivor + /// stayed Standby 4.3 s after its own UnreachableMember event); + /// (2) when auto-down removes the unreachable Primary, the survivor stays Active (live: it demoted for ~15 s until + /// its own singleton published). + /// + [Fact] + public void The_crash_survivor_promotes_on_its_own_unreachable_event_and_stays_Active_through_the_removal() + { + var actor = Start(new WarmRedundancySettings(OpcUaRedundancyMode.Warm)); + var other = MemberAt(Other); + actor.Tell(new Akka.Cluster.ClusterEvent.MemberUp(other)); + actor.Tell(Snapshot(Other)); + AwaitLogEndsWith("SL:90"); + + actor.Tell(new Akka.Cluster.ClusterEvent.UnreachableMember(other)); + AwaitLogEndsWith(Promotion); + + actor.Tell(new Akka.Cluster.ClusterEvent.MemberRemoved(MemberAt(Other, Akka.Cluster.MemberStatus.Removed), Akka.Cluster.MemberStatus.Down)); + actor.Tell(Snapshot(Other, upNumber: 1)); // the same accepted revision line, still naming the departed Primary + ExpectNoMsg(TimeSpan.FromMilliseconds(500), TestContext.Current.CancellationToken); + Steps.ShouldNotContain("delivery:True", "a Primary removed by auto-down is still gone; the survivor stays Active"); + + // Positive control: the departed Primary comes back Up and reachable, so the snapshot naming it demotes. + actor.Tell(new Akka.Cluster.ClusterEvent.MemberUp(other)); + actor.Tell(new Akka.Cluster.ClusterEvent.ReachableMember(other)); + actor.Tell(Snapshot(Other, upNumber: 1)); + AwaitLogEndsWith(Demotion); + } + + private static int _memberUid = 1000; + + // Akka's Member.Create factories are internal; same reflection idiom as DriverHostActorHistorizationRoleViewTests. + private static readonly System.Reflection.MethodInfo MemberCreate = typeof(Akka.Cluster.Member).GetMethod( + "Create", + System.Reflection.BindingFlags.Static | System.Reflection.BindingFlags.NonPublic | System.Reflection.BindingFlags.Public, + binder: null, + types: + [ + typeof(Akka.Cluster.UniqueAddress), typeof(int), typeof(Akka.Cluster.MemberStatus), + typeof(System.Collections.Immutable.ImmutableHashSet), typeof(Akka.Util.AppVersion), + ], + modifiers: null) + ?? throw new InvalidOperationException("Akka.Cluster.Member.Create(5-arg) not found; Akka API changed"); + + private static Akka.Cluster.Member MemberAt(NodeId node, Akka.Cluster.MemberStatus status = Akka.Cluster.MemberStatus.Up) + { + var text = node.ToString(); + var colon = text.LastIndexOf(':'); + return (Akka.Cluster.Member)MemberCreate.Invoke(null, + [ + new Akka.Cluster.UniqueAddress( + new Address("akka.tcp", "otopcua", text[..colon], int.Parse(text[(colon + 1)..])), 7), + Interlocked.Increment(ref _memberUid), + status, + System.Collections.Immutable.ImmutableHashSet.Create("driver"), + Akka.Util.AppVersion.Zero, + ])!; + } + /// The default-off pin at the actor: Unset never touches the projection and keeps today's levels. [Fact] public void Unset_never_drives_the_projection_and_keeps_todays_levels() diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Redundancy/WarmRoleAuthorityTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Redundancy/WarmRoleAuthorityTests.cs index 3615d4fa8..4d48e47cd 100644 --- a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Redundancy/WarmRoleAuthorityTests.cs +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Redundancy/WarmRoleAuthorityTests.cs @@ -1,3 +1,8 @@ +using System.Collections.Immutable; +using System.Reflection; +using Akka.Actor; +using Akka.Cluster; +using Akka.Util; using Shouldly; using Xunit; using ZB.MOM.WW.OtOpcUa.Commons.OpcUa; @@ -61,6 +66,70 @@ public sealed class WarmRoleAuthorityTests .ShouldBe(WarmRole.Standby, "a node that never accepted a snapshot stays Standby"); } + /// + /// T13 live gate (2026-11-13): auto-down removes the crashed Primary ~15 s after the survivor promoted, while the + /// survivor's own singleton has not yet published a snapshot naming it. A Primary that is Down or no longer a + /// member is as gone as an unreachable one, so the survivor stays Active through the removal. A Primary that is + /// present, reachable and merely Leaving/Exiting (a graceful hand-over) is not gone. + /// + [Fact] + public void A_snapshot_Primary_that_auto_down_removed_is_still_gone_and_the_survivor_stays_Active() + { + var self = MemberAt(Self); + var peer = MemberAt(Peer); + var none = new HashSet(); + var seenRemoved = new HashSet { Peer }; + var noneRemoved = new HashSet(); + + WarmRoleAuthority.SnapshotPrimaryGone(Peer, [self, peer], new HashSet { peer }, noneRemoved).ShouldBeTrue("unreachable"); + WarmRoleAuthority.SnapshotPrimaryGone(Peer, [self, MemberAt(Peer, MemberStatus.Down)], none, noneRemoved).ShouldBeTrue("downed"); + WarmRoleAuthority.SnapshotPrimaryGone(Peer, [self], none, seenRemoved).ShouldBeTrue("removed by auto-down"); + + WarmRoleAuthority.SnapshotPrimaryGone(Peer, [self], none, noneRemoved).ShouldBeFalse("never seen as a member: not assumed gone"); + WarmRoleAuthority.SnapshotPrimaryGone(Peer, [self, peer], none, seenRemoved).ShouldBeFalse("back as a reachable Up member"); + WarmRoleAuthority.SnapshotPrimaryGone(Peer, [self, peer], none, noneRemoved).ShouldBeFalse("a reachable Up Primary is never overridden"); + WarmRoleAuthority.SnapshotPrimaryGone(Peer, [self, MemberAt(Peer, MemberStatus.Leaving)], none, noneRemoved).ShouldBeFalse("graceful leave"); + WarmRoleAuthority.SnapshotPrimaryGone(Peer, [self, MemberAt(Peer, MemberStatus.Exiting)], none, noneRemoved).ShouldBeFalse("graceful exit"); + WarmRoleAuthority.SnapshotPrimaryGone(null, [self], none, seenRemoved).ShouldBeFalse("no snapshot Primary"); + + // The live sequence on the survivor: promoted while the Primary was unreachable, then the Primary is removed. + var removed = In(oldest: Self, oldestReachable: Self, snapshotPrimary: Peer, + primaryUnreachable: WarmRoleAuthority.SnapshotPrimaryGone(Peer, [self], none, seenRemoved)); + WarmRoleAuthority.Evaluate(removed).Role.ShouldBe(WarmRole.Active); + WarmRoleAuthority.Evaluate(removed with { PromoteWhilePrimaryUnreachable = false }).Role.ShouldBe(WarmRole.Standby); + } + + private static int _memberUid; + + // Akka's Member.Create factories are internal; same reflection idiom as DriverHostActorHistorizationRoleViewTests. + private static readonly MethodInfo MemberCreate = typeof(Member).GetMethod( + "Create", + BindingFlags.Static | BindingFlags.NonPublic | BindingFlags.Public, + binder: null, + types: + [ + typeof(UniqueAddress), typeof(int), typeof(MemberStatus), + typeof(ImmutableHashSet), typeof(AppVersion), + ], + modifiers: null) + ?? throw new InvalidOperationException("Akka.Cluster.Member.Create(5-arg) not found; Akka API changed"); + + private static Member MemberAt(NodeId node, MemberStatus status = MemberStatus.Up) + { + var text = node.ToString(); + var colon = text.LastIndexOf(':'); + return (Member)MemberCreate.Invoke(null, + [ + new UniqueAddress( + new Address("akka.tcp", "otopcua", text[..colon], int.Parse(text[(colon + 1)..])), + Interlocked.Increment(ref _memberUid)), + 1, + status, + ImmutableHashSet.Create("driver"), + AppVersion.Zero, + ])!; + } + [Fact] public void A_late_bind_replays_the_retained_state_but_not_the_actions() {