dohertj2/ScadaBridge
1
0
Fork 0
Code Issues 16 Pull Requests Actions Packages Projects Releases Wiki Activity

feat(cluster): leader-auto-down decision logging + cold-start/flapping/partition/switch-off scenarios (ScadaBridge#106)

Browse Source 
Every downing decision of LeaderAutoDowningProvider is now logged under a
[leader-auto-down] prefix: start-up snapshot (leader, window, already-unreachable),
unreachable observed + its window (Warning), skipped because already Down/Exiting,
pending Down cancelled on reachable-again/removed, timed out on a non-leader and kept
pending, leadership gained/lost, and DOWNING (Warning) before Cluster.Down. Akka's own
AutoDown logged only the final Down. Decision semantics are unchanged.

New real two-node scenarios (IntegrationTests/Cluster/LeaderAutoDownScenarioTests),
built from production BuildHocon with self-first seeds:
- both nodes cold-start together -> one cluster, nobody downed past FD + window;
- lone cold start with the peer dead (the down-if-alone cold-recreate shape) stays Up,
  then the peer cold-starts and joins;
- flapping peer via Akka's trttl throttler Blackhole: unreachable shorter than the
  window, repeated past it in total -> never downed; positive control asserts the cuts
  really produced unreachability;
- partition held past the window -> each side downs the other, both stay Up and
  leader (the accepted dual-active trade);
- switch off: auto-down loads AutoDowning, keep-oldest loads SplitBrainResolverProvider,
  only leader-auto-down loads the in-repo provider.
Unit: flapping never downs; a flapping member that finally stays unreachable is downed
exactly once a full window after its LAST unreachable; each decision log line.

Mutation-proved: no timer cancel on reachable -> flap-then-stay unit fails; reachable
ignored -> both unit flapping tests, the cancel-log test and the real-cluster flapping
test fail; DOWNING log dropped -> log test fails; provider never downs -> partition
test fails; auto-down miswired to the new provider -> switch-off theory fails.

Docker failover drill not run (it would rebuild the shared scadabridge:latest image
and reuse the shared rig volumes while another agent works in the repo); filed as
deploy gate ScadaBridge#116. Default remains auto-down.

Claude-Session: https://claude.ai/code/session_01GT67AEHd2Acv1YPwGfXEoV
This commit is contained in:
Joseph Doherty
2026-11-10 04:35:43 -05:00
parent a13aabc7c3
commit 456c6659e7
6 changed files with 436 additions and 10 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/ZB.MOM.WW.ScadaBridge.Host.Tests/LeaderAutoDownTests.cs
+109
View File
@@ -18,6 +18,12 @@ namespace ZB.MOM.WW.ScadaBridge.Host.Tests;
/// </summary>
public class LeaderAutoDownTests : TestKit
{
/// <summary>INFO so the decision-log assertions can see the Info-level decisions.</summary>
public LeaderAutoDownTests()
: base("akka.loglevel = INFO")
{
}
private static readonly Address MemberA = new("akka.tcp", "sys", "a", 2552);
private static readonly Address MemberB = new("akka.tcp", "sys", "b", 2552);
private static readonly Address MemberC = new("akka.tcp", "sys", "c", 2552);
@@ -212,4 +218,107 @@ public class LeaderAutoDownTests : TestKit
ExpectMsg(new DownCalled(MemberB), TimeSpan.FromSeconds(3));
ExpectNoMsg(TimeSpan.FromSeconds(2));
}
/// <summary>
/// A flapping member — unreachable, reachable again inside the window, repeatedly, for far
/// longer than one window in total — is never downed.
/// </summary>
[Fact]
public void FlappingMember_ReachableAgainInsideEachWindow_IsNeverDowned()
{
var a = AutoDownActor(TimeSpan.FromMilliseconds(1500));
a.Tell(new ClusterEvent.LeaderChanged(MemberA));
var b = CreateMember(MemberB, MemberStatus.Up);
for (var i = 0; i < 5; i++)
{
a.Tell(new ClusterEvent.UnreachableMember(b));
ExpectNoMsg(TimeSpan.FromMilliseconds(700));
a.Tell(new ClusterEvent.ReachableMember(b));
ExpectNoMsg(TimeSpan.FromMilliseconds(100));
}
ExpectNoMsg(TimeSpan.FromSeconds(2));
}
/// <summary>
/// When a flapping member finally stays unreachable, the window restarts from its LAST
/// unreachable — an earlier flap's timer must not fire — and it is downed exactly once.
/// </summary>
[Fact]
public void FlappingMember_ThatStaysUnreachable_IsDownedOneFullWindowAfterLastUnreachable()
{
var a = AutoDownActor(TimeSpan.FromMilliseconds(1500));
a.Tell(new ClusterEvent.LeaderChanged(MemberA));
var b = CreateMember(MemberB, MemberStatus.Up);
a.Tell(new ClusterEvent.UnreachableMember(b));
ExpectNoMsg(TimeSpan.FromMilliseconds(500));
a.Tell(new ClusterEvent.ReachableMember(b));
ExpectNoMsg(TimeSpan.FromMilliseconds(200));
a.Tell(new ClusterEvent.UnreachableMember(b));
ExpectNoMsg(TimeSpan.FromMilliseconds(500));
a.Tell(new ClusterEvent.ReachableMember(b));
ExpectNoMsg(TimeSpan.FromMilliseconds(200));
a.Tell(new ClusterEvent.UnreachableMember(b)); // final: stays unreachable from here
// The first flap's timer would fire ~0.1s from here; the final window ends ~1.5s from here.
ExpectNoMsg(TimeSpan.FromMilliseconds(1100));
ExpectMsg(new DownCalled(MemberB), TimeSpan.FromSeconds(2));
ExpectNoMsg(TimeSpan.FromSeconds(2));
}
/// <summary>The Down decision is logged at Warning, naming the member and the window.</summary>
[Fact]
public void DecisionLog_Down_IsLoggedAtWarning()
{
var a = AutoDownActor(TimeSpan.Zero);
a.Tell(new ClusterEvent.LeaderChanged(MemberA));
EventFilter.Warning(contains: $"[leader-auto-down] DOWNING {MemberB}").ExpectOne(() =>
a.Tell(new ClusterEvent.UnreachableMember(CreateMember(MemberB, MemberStatus.Up))));
ExpectMsg(new DownCalled(MemberB));
}
/// <summary>Observing an unreachable member (and its window) is logged at Warning.</summary>
[Fact]
public void DecisionLog_UnreachableScheduled_IsLoggedAtWarning()
{
var a = AutoDownActor(TimeSpan.FromSeconds(30));
a.Tell(new ClusterEvent.LeaderChanged(MemberA));
EventFilter.Warning(contains: $"{MemberB} is unreachable; it will be downed if still unreachable after").ExpectOne(() =>
a.Tell(new ClusterEvent.UnreachableMember(CreateMember(MemberB, MemberStatus.Up))));
}
/// <summary>A cancelled Down (member reachable again) is logged.</summary>
[Fact]
public void DecisionLog_ReachableAgain_CancelIsLogged()
{
var a = AutoDownActor(TimeSpan.FromSeconds(30));
a.Tell(new ClusterEvent.LeaderChanged(MemberA));
var b = CreateMember(MemberB, MemberStatus.Up);
a.Tell(new ClusterEvent.UnreachableMember(b));
EventFilter.Info(contains: $"{MemberB} became reachable again; its pending Down is cancelled").ExpectOne(() =>
a.Tell(new ClusterEvent.ReachableMember(b)));
}
/// <summary>A timeout on a non-leader (kept pending) is logged, and so is the later Down on becoming leader.</summary>
[Fact]
public void DecisionLog_PendingOnNonLeader_ThenDownOnBecomingLeader_AreLogged()
{
var a = AutoDownActor(TimeSpan.Zero);
a.Tell(new ClusterEvent.LeaderChanged(MemberB));
EventFilter.Info(contains: "is not leader; kept pending").ExpectOne(() =>
a.Tell(new ClusterEvent.UnreachableMember(CreateMember(MemberC, MemberStatus.Up))));
EventFilter.Warning(contains: $"DOWNING {MemberC}: it timed out as unreachable while this node was not leader").ExpectOne(() =>
a.Tell(new ClusterEvent.LeaderChanged(MemberA)));
ExpectMsg(new DownCalled(MemberC));
}
/// <summary>Skipping an already Down member is logged.</summary>
[Fact]
public void DecisionLog_SkippedDownMember_IsLogged()
{
var a = AutoDownActor(TimeSpan.Zero);
a.Tell(new ClusterEvent.LeaderChanged(MemberA));
EventFilter.Info(contains: "but already Down; not scheduling a Down").ExpectOne(() =>
a.Tell(new ClusterEvent.UnreachableMember(CreateMember(MemberB, MemberStatus.Down))));
}
}
tests/ZB.MOM.WW.ScadaBridge.IntegrationTests/Cluster/LeaderAutoDownScenarioTests.cs
+257
View File
@@ -0,0 +1,257 @@
using Akka.Actor;
using Akka.Cluster;
using Akka.Configuration;
using Akka.Remote;
using Akka.Remote.Transport;
using Xunit;
using ZB.MOM.WW.ScadaBridge.ClusterInfrastructure;
using ZB.MOM.WW.ScadaBridge.Host;
using ZB.MOM.WW.ScadaBridge.Host.Actors;
namespace ZB.MOM.WW.ScadaBridge.IntegrationTests.Cluster;
/// <summary>
/// ScadaBridge#106: the operational scenarios of the in-repo <see cref="LeaderAutoDowningProvider"/>
/// beyond the hard-crash clones in <see cref="LeaderAutoDownFailoverTests"/> — cold starts
/// (together, and alone with the peer dead: the shape that makes SBR <c>down-if-alone</c>
/// self-down on a cold recreate), a FLAPPING peer that keeps going unreachable for less than
/// the window, a real partition (the accepted dual-active trade), and the opt-in switch being
/// off (the old providers are what Akka loads).
/// <para>
/// Nodes are built from the production <c>BuildHocon</c> output with the shipped self-first
/// seed ordering, exactly like <see cref="SelfFirstSeedBootstrapTests"/>. The flapping and
/// partition cases add Akka's own throttler transport adapter (<c>trttl</c>, the one the
/// MultiNodeTestRunner uses for <c>Blackhole</c>) so the network can be cut and restored
/// without killing either node.
/// </para>
/// </summary>
public sealed class LeaderAutoDownScenarioTests : IAsyncLifetime
{
private const string Strategy = "leader-auto-down";
private static readonly TimeSpan Heartbeat = TimeSpan.FromMilliseconds(500);
private static readonly TimeSpan Threshold = TimeSpan.FromSeconds(2);
private readonly List<ActorSystem> _systems = new();
private ActorSystem StartNode(
int selfPort, int peerPort, string strategy = Strategy, TimeSpan? stableAfter = null, bool throttled = false)
{
// With the throttler applied the published address scheme becomes akka.trttl.tcp, so the
// seeds must carry it too or the nodes never find each other.
var scheme = throttled ? "akka.trttl.tcp" : "akka.tcp";
var nodeOptions = new NodeOptions { Role = "Central", NodeHostname = "127.0.0.1", RemotingPort = selfPort };
var clusterOptions = new ClusterOptions
{
SeedNodes = new List<string>
{
$"{scheme}://scadabridge@127.0.0.1:{selfPort}",
$"{scheme}://scadabridge@127.0.0.1:{peerPort}",
},
SplitBrainResolverStrategy = strategy,
StableAfter = stableAfter ?? TimeSpan.FromSeconds(3),
HeartbeatInterval = Heartbeat,
FailureDetectionThreshold = Threshold,
MinNrOfMembers = 1,
DownIfAlone = true,
};
// Transport failure detection far above the cluster window: a blackhole must make the
// MEMBER unreachable (what the downing provider reacts to), not tear the association down.
var hocon = AkkaHostedService.BuildHocon(
nodeOptions, clusterOptions, new[] { "Central" },
TimeSpan.FromSeconds(1), TimeSpan.FromSeconds(60));
var overrides = "akka.coordinated-shutdown.run-by-actor-system-terminate = off\n"
+ (throttled ? "akka.remote.dot-netty.tcp.applied-adapters = [trttl]\n" : string.Empty);
var config = ConfigurationFactory.ParseString(overrides)
.WithFallback(ConfigurationFactory.ParseString(hocon));
var system = ActorSystem.Create("scadabridge", config);
_systems.Add(system);
return system;
}
private static Task<bool> Throttle(ActorSystem from, ActorSystem to, ThrottleMode mode) =>
((RemoteActorRefProvider)((ExtendedActorSystem)from).Provider).Transport.ManagementCommand(
new SetThrottle(Akka.Cluster.Cluster.Get(to).SelfAddress, ThrottleTransportAdapter.Direction.Both, mode));
private static void AssertTwoUpNoneUnreachable(ActorSystem node, string because)
{
var cluster = Akka.Cluster.Cluster.Get(node);
Assert.False(cluster.IsTerminated, $"{cluster.SelfAddress}: Cluster extension terminated ({because})");
Assert.Equal(MemberStatus.Up, cluster.SelfMember.Status);
Assert.Equal(2, cluster.State.Members.Count(m => m.Status == MemberStatus.Up));
Assert.Empty(cluster.State.Unreachable);
}
/// <summary>
/// Both nodes cold-start together: they converge on ONE two-member cluster and, once
/// formed, nobody is downed — held for well past failure detection plus the window.
/// </summary>
/// <returns>A task that represents the asynchronous operation.</returns>
[Fact]
public async Task BothNodesColdStartTogether_FormOneCluster_AndNeitherIsDowned()
{
var portA = TwoNodeClusterFixture.GetFreeTcpPort();
var portB = TwoNodeClusterFixture.GetFreeTcpPort();
var nodeA = StartNode(portA, portB);
var nodeB = StartNode(portB, portA);
await TwoNodeClusterFixture.WaitForMembersUp(nodeA, 2, TimeSpan.FromSeconds(60));
await TwoNodeClusterFixture.WaitForMembersUp(nodeB, 2, TimeSpan.FromSeconds(60));
Assert.IsType<LeaderAutoDowningProvider>(Akka.Cluster.Cluster.Get(nodeA).DowningProvider);
await Task.Delay(Threshold + TimeSpan.FromSeconds(3) + TimeSpan.FromSeconds(5));
AssertTwoUpNoneUnreachable(nodeA, "after a joint cold start");
AssertTwoUpNoneUnreachable(nodeB, "after a joint cold start");
}
/// <summary>
/// A lone cold start with the peer DEAD (the cold-recreate shape where SBR keep-oldest
/// <c>down-if-alone</c> downs the lone node) forms a one-node cluster that stays Up past the
/// window; when the peer is later cold-started it joins, and nobody is downed.
/// </summary>
/// <returns>A task that represents the asynchronous operation.</returns>
[Fact]
public async Task LoneColdStart_PeerDead_StaysUp_ThenPeerColdStartsAndJoins()
{
var portA = TwoNodeClusterFixture.GetFreeTcpPort();
var portB = TwoNodeClusterFixture.GetFreeTcpPort(); // nothing listening yet
var nodeA = StartNode(portA, portB);
await TwoNodeClusterFixture.WaitForMembersUp(nodeA, 1, TimeSpan.FromSeconds(30));
await Task.Delay(Threshold + TimeSpan.FromSeconds(3) + TimeSpan.FromSeconds(3));
var clusterA = Akka.Cluster.Cluster.Get(nodeA);
Assert.False(clusterA.IsTerminated, "lone cold-started node downed itself");
Assert.Equal(MemberStatus.Up, clusterA.SelfMember.Status);
var nodeB = StartNode(portB, portA);
await TwoNodeClusterFixture.WaitForMembersUp(nodeA, 2, TimeSpan.FromSeconds(60));
await TwoNodeClusterFixture.WaitForMembersUp(nodeB, 2, TimeSpan.FromSeconds(60));
await Task.Delay(TimeSpan.FromSeconds(5));
AssertTwoUpNoneUnreachable(nodeA, "after the peer joined");
AssertTwoUpNoneUnreachable(nodeB, "after the peer joined");
}
/// <summary>
/// A FLAPPING peer — cut off long enough to be marked unreachable, restored before the
/// window ends, over and over, for far longer than the window in total — is never downed:
/// each return to reachable cancels its pending Down and the next cut starts a fresh window.
/// The test first proves the peer really did go unreachable on (almost) every cut, so the
/// "never downed" outcome is not vacuous.
/// </summary>
/// <returns>A task that represents the asynchronous operation.</returns>
[Fact]
public async Task FlappingPeer_UnreachableShorterThanWindow_IsNeverDowned()
{
var stableAfter = TimeSpan.FromSeconds(8);
var portA = TwoNodeClusterFixture.GetFreeTcpPort();
var portB = TwoNodeClusterFixture.GetFreeTcpPort();
var nodeA = StartNode(portA, portB, stableAfter: stableAfter, throttled: true);
await TwoNodeClusterFixture.WaitForMembersUp(nodeA, 1, TimeSpan.FromSeconds(30));
var nodeB = StartNode(portB, portA, stableAfter: stableAfter, throttled: true);
await TwoNodeClusterFixture.WaitForMembersUp(nodeA, 2, TimeSpan.FromSeconds(30));
await TwoNodeClusterFixture.WaitForMembersUp(nodeB, 2, TimeSpan.FromSeconds(30));
var clusterA = Akka.Cluster.Cluster.Get(nodeA);
const int flaps = 4;
var flapsSeenUnreachable = 0;
var started = DateTime.UtcNow;
for (var i = 0; i < flaps; i++)
{
Assert.True(await Throttle(nodeA, nodeB, Blackhole.Instance), "SetThrottle(Blackhole) rejected");
var seen = false;
var cutUntil = DateTime.UtcNow + TimeSpan.FromSeconds(4.5);
while (DateTime.UtcNow < cutUntil)
{
seen |= clusterA.State.Unreachable.Any();
await Task.Delay(100);
}
Assert.True(await Throttle(nodeA, nodeB, Unthrottled.Instance), "SetThrottle(Unthrottled) rejected");
await TwoNodeClusterFixture.WaitForMembersUp(nodeA, 2, TimeSpan.FromSeconds(15));
await TwoNodeClusterFixture.WaitForMembersUp(nodeB, 2, TimeSpan.FromSeconds(15));
if (seen)
{
flapsSeenUnreachable++;
}
}
// Positive control: the cuts really produced unreachability, and the flapping lasted
// longer than one window — a provider that did not cancel on reachable would have downed.
Assert.True(flapsSeenUnreachable >= flaps - 1,
$"only {flapsSeenUnreachable}/{flaps} cuts made the peer unreachable — the flap never happened");
Assert.True(DateTime.UtcNow - started > stableAfter * 2);
await Task.Delay(stableAfter + TimeSpan.FromSeconds(2));
AssertTwoUpNoneUnreachable(nodeA, "flapping peer");
AssertTwoUpNoneUnreachable(nodeB, "flapping peer");
}
/// <summary>
/// A real partition longer than the window: each side is leader of its own reachable set,
/// so each downs the other and BOTH stay Up — the dual-active outcome the availability-first
/// decision accepts (identical to AutoDowning). Also the positive control for the flapping
/// test: the same cut, held past the window, does down.
/// </summary>
/// <returns>A task that represents the asynchronous operation.</returns>
[Fact]
public async Task PartitionLongerThanWindow_EachSideDownsTheOther_BothStayUp()
{
var stableAfter = TimeSpan.FromSeconds(3);
var portA = TwoNodeClusterFixture.GetFreeTcpPort();
var portB = TwoNodeClusterFixture.GetFreeTcpPort();
var nodeA = StartNode(portA, portB, stableAfter: stableAfter, throttled: true);
await TwoNodeClusterFixture.WaitForMembersUp(nodeA, 1, TimeSpan.FromSeconds(30));
var nodeB = StartNode(portB, portA, stableAfter: stableAfter, throttled: true);
await TwoNodeClusterFixture.WaitForMembersUp(nodeA, 2, TimeSpan.FromSeconds(30));
await TwoNodeClusterFixture.WaitForMembersUp(nodeB, 2, TimeSpan.FromSeconds(30));
var addressA = Akka.Cluster.Cluster.Get(nodeA).SelfAddress;
var addressB = Akka.Cluster.Cluster.Get(nodeB).SelfAddress;
Assert.True(await Throttle(nodeA, nodeB, Blackhole.Instance));
await TwoNodeClusterFixture.WaitForMemberRemoved(nodeA, addressB, TimeSpan.FromSeconds(30));
await TwoNodeClusterFixture.WaitForMemberRemoved(nodeB, addressA, TimeSpan.FromSeconds(30));
foreach (var node in new[] { nodeA, nodeB })
{
var cluster = Akka.Cluster.Cluster.Get(node);
Assert.False(cluster.IsTerminated, $"{cluster.SelfAddress} downed itself during the partition");
Assert.Equal(MemberStatus.Up, cluster.SelfMember.Status);
Assert.Equal(cluster.SelfAddress, cluster.State.Leader);
}
}
/// <summary>
/// The opt-in switch OFF: the default <c>auto-down</c> still loads Akka's own
/// <c>AutoDowning</c> and <c>keep-oldest</c> still loads the SBR provider — the in-repo
/// provider is loaded only for <c>leader-auto-down</c>.
/// </summary>
/// <param name="strategy">The configured strategy.</param>
/// <param name="expectedProvider">The downing provider Akka must have instantiated.</param>
/// <returns>A task that represents the asynchronous operation.</returns>
[Theory]
[InlineData("auto-down", typeof(AutoDowning))]
[InlineData("keep-oldest", typeof(Akka.Cluster.SBR.SplitBrainResolverProvider))]
[InlineData(Strategy, typeof(LeaderAutoDowningProvider))]
public async Task SwitchSelectsProvider_OldPathsUnchangedWhenOff(string strategy, Type expectedProvider)
{
var portA = TwoNodeClusterFixture.GetFreeTcpPort();
var portB = TwoNodeClusterFixture.GetFreeTcpPort();
var nodeA = StartNode(portA, portB, strategy: strategy);
await TwoNodeClusterFixture.WaitForMembersUp(nodeA, 1, TimeSpan.FromSeconds(30));
Assert.IsType(expectedProvider, Akka.Cluster.Cluster.Get(nodeA).DowningProvider);
}
/// <summary>Prepares the fixture before the tests run.</summary>
/// <returns>A task that represents the asynchronous operation.</returns>
public Task InitializeAsync() => Task.CompletedTask;
/// <summary>Releases the resources held by this fixture asynchronously.</summary>
/// <returns>A task that represents the asynchronous operation.</returns>
public async Task DisposeAsync()
{
foreach (var s in _systems)
{
try { await s.Terminate().WaitAsync(TimeSpan.FromSeconds(10)); } catch { /* teardown */ }
}
}
}