Wires the Phase 6.3 Stream B pure-logic pieces (ServiceLevelCalculator,
RecoveryStateManager, ApplyLeaseRegistry) + Stream A topology loader
(RedundancyCoordinator) into one orchestrator the runtime + OPC UA node
surface consume. The actual OPC UA variable-node plumbing (mapping
ServiceLevel Byte + ServerUriArray String[] onto the Opc.Ua.Server stack)
is narrower follow-up on top of this — the publisher emits change events
the OPC UA layer subscribes to.
Server.Redundancy additions:
- PeerReachability record + PeerReachabilityTracker — thread-safe
per-peer-NodeId holder of the latest (HttpHealthy, UaHealthy) tuple. Probe
loops (Stream B.1/B.2 runtime follow-up) write via Update; the publisher
reads via Get. PeerReachability.FullyHealthy / Unknown sentinels for the
two most-common states.
- RedundancyStatePublisher — pure orchestrator, no background timer, no OPC
UA stack dep. ComputeAndPublish reads the 6 inputs + calls the calculator:
* role (from coordinator.Current.SelfRole)
* selfHealthy (caller-supplied Func<bool>)
* peerHttpHealthy + peerUaHealthy (aggregate across all peers in
coordinator.Current.Peers)
* applyInProgress (ApplyLeaseRegistry.IsApplyInProgress)
* recoveryDwellMet (RecoveryStateManager.IsDwellMet)
* topologyValid (coordinator.IsTopologyValid)
* operatorMaintenance (caller-supplied Func<bool>)
Before-coordinator-init returns NoData=1 so clients never see an
authoritative value from an un-bootstrapped server.
OnStateChanged event fires edge-triggered when the byte changes;
OnServerUriArrayChanged fires edge-triggered when the topology's self-first
peer-sorted URI array content changes.
- ServiceLevelSnapshot record — per-tick output with Value + Band +
Topology. The OPC UA layer's ServiceLevel Byte node subscribes to
OnStateChanged; the ServerUriArray node subscribes to OnServerUriArrayChanged.
Tests (8 new RedundancyStatePublisherTests, all pass):
- Before-init returns NoData (Value=1, Band=NoData).
- Authoritative-Primary when healthy + peer fully reachable.
- Isolated-Primary (230) retains authority when peer unreachable — matches
decision #154 non-promotion semantics.
- Mid-apply band dominates: open lease → Value=200 even with peer healthy.
- Self-unhealthy → NoData regardless of other inputs.
- OnStateChanged fires only on value transitions (edge-triggered).
- OnServerUriArrayChanged fires once per topology content change; repeat
ticks with same topology don't re-emit.
- Standalone cluster treats healthy as AuthoritativePrimary=255.
Microsoft.EntityFrameworkCore.InMemory 10.0.0 added to Server.Tests for the
coordinator-backed publisher tests.
Full solution dotnet test: 1186 passing (was 1178, +8). Pre-existing
Client.CLI Subscribe flake unchanged.
Closes the core of release blocker #3 — the pure-logic + orchestration
layer now exists + is unit-tested. Remaining Stream C surfaces: OPC UA
ServiceLevel Byte variable wiring (binds to OnStateChanged), ServerUriArray
String[] wiring (binds to OnServerUriArrayChanged), RedundancySupport
static from RedundancyMode. Those touch the OPC UA stack directly + land
as Stream C.2 follow-up.
Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
214 lines
7.7 KiB
C#
214 lines
7.7 KiB
C#
using Microsoft.EntityFrameworkCore;
|
|
using Microsoft.Extensions.Logging.Abstractions;
|
|
using Shouldly;
|
|
using Xunit;
|
|
using ZB.MOM.WW.OtOpcUa.Configuration;
|
|
using ZB.MOM.WW.OtOpcUa.Configuration.Entities;
|
|
using ZB.MOM.WW.OtOpcUa.Configuration.Enums;
|
|
using ZB.MOM.WW.OtOpcUa.Server.Redundancy;
|
|
|
|
namespace ZB.MOM.WW.OtOpcUa.Server.Tests;
|
|
|
|
[Trait("Category", "Unit")]
|
|
public sealed class RedundancyStatePublisherTests : IDisposable
|
|
{
|
|
private readonly OtOpcUaConfigDbContext _db;
|
|
private readonly IDbContextFactory<OtOpcUaConfigDbContext> _dbFactory;
|
|
|
|
public RedundancyStatePublisherTests()
|
|
{
|
|
var options = new DbContextOptionsBuilder<OtOpcUaConfigDbContext>()
|
|
.UseInMemoryDatabase($"redundancy-publisher-{Guid.NewGuid():N}")
|
|
.Options;
|
|
_db = new OtOpcUaConfigDbContext(options);
|
|
_dbFactory = new DbContextFactory(options);
|
|
}
|
|
|
|
public void Dispose() => _db.Dispose();
|
|
|
|
private sealed class DbContextFactory(DbContextOptions<OtOpcUaConfigDbContext> options)
|
|
: IDbContextFactory<OtOpcUaConfigDbContext>
|
|
{
|
|
public OtOpcUaConfigDbContext CreateDbContext() => new(options);
|
|
}
|
|
|
|
private async Task<RedundancyCoordinator> SeedAndInitialize(string selfNodeId, params (string id, RedundancyRole role, string appUri)[] nodes)
|
|
{
|
|
var cluster = new ServerCluster
|
|
{
|
|
ClusterId = "c1",
|
|
Name = "Warsaw-West",
|
|
Enterprise = "zb",
|
|
Site = "warsaw-west",
|
|
RedundancyMode = nodes.Length == 1 ? RedundancyMode.None : RedundancyMode.Warm,
|
|
CreatedBy = "test",
|
|
};
|
|
_db.ServerClusters.Add(cluster);
|
|
foreach (var (id, role, appUri) in nodes)
|
|
{
|
|
_db.ClusterNodes.Add(new ClusterNode
|
|
{
|
|
NodeId = id,
|
|
ClusterId = "c1",
|
|
RedundancyRole = role,
|
|
Host = id.ToLowerInvariant(),
|
|
ApplicationUri = appUri,
|
|
CreatedBy = "test",
|
|
});
|
|
}
|
|
await _db.SaveChangesAsync();
|
|
|
|
var coordinator = new RedundancyCoordinator(_dbFactory, NullLogger<RedundancyCoordinator>.Instance, selfNodeId, "c1");
|
|
await coordinator.InitializeAsync(CancellationToken.None);
|
|
return coordinator;
|
|
}
|
|
|
|
[Fact]
|
|
public async Task BeforeInit_Publishes_NoData()
|
|
{
|
|
// Coordinator not initialized — current topology is null.
|
|
var coordinator = new RedundancyCoordinator(_dbFactory, NullLogger<RedundancyCoordinator>.Instance, "A", "c1");
|
|
var publisher = new RedundancyStatePublisher(
|
|
coordinator, new ApplyLeaseRegistry(), new RecoveryStateManager(), new PeerReachabilityTracker());
|
|
|
|
var snap = publisher.ComputeAndPublish();
|
|
|
|
snap.Band.ShouldBe(ServiceLevelBand.NoData);
|
|
snap.Value.ShouldBe((byte)1);
|
|
await Task.Yield();
|
|
}
|
|
|
|
[Fact]
|
|
public async Task AuthoritativePrimary_WhenHealthyAndPeerReachable()
|
|
{
|
|
var coordinator = await SeedAndInitialize("A",
|
|
("A", RedundancyRole.Primary, "urn:A"),
|
|
("B", RedundancyRole.Secondary, "urn:B"));
|
|
var peers = new PeerReachabilityTracker();
|
|
peers.Update("B", PeerReachability.FullyHealthy);
|
|
|
|
var publisher = new RedundancyStatePublisher(
|
|
coordinator, new ApplyLeaseRegistry(), new RecoveryStateManager(), peers);
|
|
|
|
var snap = publisher.ComputeAndPublish();
|
|
|
|
snap.Value.ShouldBe((byte)255);
|
|
snap.Band.ShouldBe(ServiceLevelBand.AuthoritativePrimary);
|
|
}
|
|
|
|
[Fact]
|
|
public async Task IsolatedPrimary_WhenPeerUnreachable_RetainsAuthority()
|
|
{
|
|
var coordinator = await SeedAndInitialize("A",
|
|
("A", RedundancyRole.Primary, "urn:A"),
|
|
("B", RedundancyRole.Secondary, "urn:B"));
|
|
var peers = new PeerReachabilityTracker();
|
|
peers.Update("B", PeerReachability.Unknown);
|
|
|
|
var publisher = new RedundancyStatePublisher(
|
|
coordinator, new ApplyLeaseRegistry(), new RecoveryStateManager(), peers);
|
|
|
|
var snap = publisher.ComputeAndPublish();
|
|
|
|
snap.Value.ShouldBe((byte)230);
|
|
}
|
|
|
|
[Fact]
|
|
public async Task MidApply_WhenLeaseOpen_Dominates()
|
|
{
|
|
var coordinator = await SeedAndInitialize("A",
|
|
("A", RedundancyRole.Primary, "urn:A"),
|
|
("B", RedundancyRole.Secondary, "urn:B"));
|
|
var leases = new ApplyLeaseRegistry();
|
|
var peers = new PeerReachabilityTracker();
|
|
peers.Update("B", PeerReachability.FullyHealthy);
|
|
|
|
await using var lease = leases.BeginApplyLease(1, Guid.NewGuid());
|
|
var publisher = new RedundancyStatePublisher(
|
|
coordinator, leases, new RecoveryStateManager(), peers);
|
|
|
|
var snap = publisher.ComputeAndPublish();
|
|
|
|
snap.Value.ShouldBe((byte)200);
|
|
}
|
|
|
|
[Fact]
|
|
public async Task SelfUnhealthy_Returns_NoData()
|
|
{
|
|
var coordinator = await SeedAndInitialize("A",
|
|
("A", RedundancyRole.Primary, "urn:A"),
|
|
("B", RedundancyRole.Secondary, "urn:B"));
|
|
var peers = new PeerReachabilityTracker();
|
|
peers.Update("B", PeerReachability.FullyHealthy);
|
|
|
|
var publisher = new RedundancyStatePublisher(
|
|
coordinator, new ApplyLeaseRegistry(), new RecoveryStateManager(), peers,
|
|
selfHealthy: () => false);
|
|
|
|
var snap = publisher.ComputeAndPublish();
|
|
|
|
snap.Value.ShouldBe((byte)1);
|
|
}
|
|
|
|
[Fact]
|
|
public async Task OnStateChanged_FiresOnly_OnValueChange()
|
|
{
|
|
var coordinator = await SeedAndInitialize("A",
|
|
("A", RedundancyRole.Primary, "urn:A"),
|
|
("B", RedundancyRole.Secondary, "urn:B"));
|
|
var peers = new PeerReachabilityTracker();
|
|
peers.Update("B", PeerReachability.FullyHealthy);
|
|
|
|
var publisher = new RedundancyStatePublisher(
|
|
coordinator, new ApplyLeaseRegistry(), new RecoveryStateManager(), peers);
|
|
|
|
var emitCount = 0;
|
|
byte? lastEmitted = null;
|
|
publisher.OnStateChanged += snap => { emitCount++; lastEmitted = snap.Value; };
|
|
|
|
publisher.ComputeAndPublish(); // first tick — emits 255 since _lastByte was seeded at 255; no change
|
|
peers.Update("B", PeerReachability.Unknown);
|
|
publisher.ComputeAndPublish(); // 255 → 230 transition — emits
|
|
publisher.ComputeAndPublish(); // still 230 — no emit
|
|
|
|
emitCount.ShouldBe(1);
|
|
lastEmitted.ShouldBe((byte)230);
|
|
}
|
|
|
|
[Fact]
|
|
public async Task OnServerUriArrayChanged_FiresOnce_PerTopology()
|
|
{
|
|
var coordinator = await SeedAndInitialize("A",
|
|
("A", RedundancyRole.Primary, "urn:A"),
|
|
("B", RedundancyRole.Secondary, "urn:B"));
|
|
var peers = new PeerReachabilityTracker();
|
|
peers.Update("B", PeerReachability.FullyHealthy);
|
|
|
|
var publisher = new RedundancyStatePublisher(
|
|
coordinator, new ApplyLeaseRegistry(), new RecoveryStateManager(), peers);
|
|
|
|
var emits = new List<IReadOnlyList<string>>();
|
|
publisher.OnServerUriArrayChanged += arr => emits.Add(arr);
|
|
|
|
publisher.ComputeAndPublish();
|
|
publisher.ComputeAndPublish();
|
|
publisher.ComputeAndPublish();
|
|
|
|
emits.Count.ShouldBe(1, "ServerUriArray event is edge-triggered on topology content change");
|
|
emits[0].ShouldBe(["urn:A", "urn:B"]);
|
|
}
|
|
|
|
[Fact]
|
|
public async Task Standalone_Cluster_IsAuthoritative_When_Healthy()
|
|
{
|
|
var coordinator = await SeedAndInitialize("A",
|
|
("A", RedundancyRole.Standalone, "urn:A"));
|
|
var publisher = new RedundancyStatePublisher(
|
|
coordinator, new ApplyLeaseRegistry(), new RecoveryStateManager(), new PeerReachabilityTracker());
|
|
|
|
var snap = publisher.ComputeAndPublish();
|
|
|
|
snap.Value.ShouldBe((byte)255);
|
|
}
|
|
}
|