/* ======================================================================== * Copyright (c) 2005-2025 The OPC Foundation, Inc. All rights reserved. * * OPC Foundation MIT License 1.00 * * Permission is hereby granted, free of charge, to any person * obtaining a copy of this software and associated documentation * files (the "Software"), to deal in the Software without * restriction, including without limitation the rights to use, * copy, modify, merge, publish, distribute, sublicense, and/or sell * copies of the Software, and to permit persons to whom the * Software is furnished to do so, subject to the following * conditions: * * The above copyright notice and this permission notice shall be * included in all copies or substantial portions of the Software. * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, * EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES * OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND * NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT * HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, * WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING * FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR * OTHER DEALINGS IN THE SOFTWARE. * * The complete license agreement can be found here: * http://opcfoundation.org/License/MIT/1.00/ * ======================================================================*/ using System; using System.Threading; using System.Threading.Tasks; using NUnit.Framework; using Opc.Ua.Client; using Opc.Ua.Client.Redundancy; using Opc.Ua.Client.TestFramework; using Opc.Ua.Redundancy; using Opc.Ua.Redundancy.Client; using Opc.Ua.Server.TestFramework; using Quickstarts.ReferenceServer; namespace Opc.Ua.Sessions.Tests { /// /// End-to-end integration test for the transparent client-redundancy facade /// (). Two client replicas share one store and /// coordinate leadership against a live server; the test drives real browse/read /// traffic through the facade and forces a leader change, proving that the follower /// blocks until promoted and that the same facade reference keeps serving after the /// underlying leader session is swapped. /// [TestFixture] [Category("Client")] [Category("ManagedSession")] [Category("Distributed")] [Category("ClientRedundancy")] [SetCulture("en-us")] [SetUICulture("en-us")] [NonParallelizable] public sealed class RedundantClientSessionFailoverIntegrationTests : ClientTestFramework { [OneTimeSetUp] public override Task OneTimeSetUpAsync() { SupportsExternalServerUrl = true; SingleSession = false; return OneTimeSetUpCoreAsync(securityNone: true); } [OneTimeTearDown] public override Task OneTimeTearDownAsync() { return base.OneTimeTearDownAsync(); } [SetUp] public override Task SetUpAsync() { return base.SetUpAsync(); } [TearDown] public override Task TearDownAsync() { return base.TearDownAsync(); } [Test] [Order(100)] [CancelAfter(120_000)] public async Task FacadeBlocksUntilLeaderThenSurvivesHandoffAsync(CancellationToken ct) { var timeout = TimeSpan.FromSeconds(30); using var store = new InMemorySharedKeyValueStore(); var electionA = new ControllableLeaderElection(); var electionB = new ControllableLeaderElection(); ServerFixture? fixture = null; RedundantClientSession? replicaA = null; RedundantClientSession? replicaB = null; Task? blockedOnB = null; try { (fixture, Uri url) = await StartServerAsync().ConfigureAwait(false); ConfiguredEndpoint endpoint = await ClientFixture .GetEndpointAsync(url, SecurityPolicies.None) .ConfigureAwait(false); replicaA = BuildReplica("replica-A", endpoint, electionA, store); replicaB = BuildReplica("replica-B", endpoint, electionB, store); await replicaA.StartAsync(ct).ConfigureAwait(false); await replicaB.StartAsync(ct).ConfigureAwait(false); // No replica is leader yet: synchronous members fault and async calls block. ServiceResultException? syncFault = Assert.Throws( () => _ = replicaB.SessionName); Assert.That(syncFault!.StatusCode, Is.EqualTo(StatusCodes.BadInvalidState)); #pragma warning disable CA2025 // Do not pass 'IDisposable' instances into unawaited tasks blockedOnB = ReadServerStateAsync(replicaB, ct); #pragma warning restore CA2025 // Do not pass 'IDisposable' instances into unawaited tasks await Task.Delay(500, ct).ConfigureAwait(false); Assert.That( blockedOnB.IsCompleted, Is.False, "an async call on a follower must block until this replica is promoted."); // Promote A: it creates its session and the facade starts serving. electionA.SetLeader(true); DataValue stateA = await ReadServerStateAsync(replicaA, ct) .WaitAsync(timeout, ct).ConfigureAwait(false); Assert.That(StatusCode.IsGood(stateA.StatusCode), Is.True); Assert.That(replicaA.IsLeader, Is.True); BrowseResponse browseA = await BrowseObjectsAsync(replicaA, ct) .WaitAsync(timeout, ct).ConfigureAwait(false); Assert.That(browseA.Results.Count, Is.GreaterThan(0)); Assert.That(browseA.Results[0].References.Count, Is.GreaterThan(0)); // Hand off leadership A -> B. B's previously-blocked read must now complete // against the freshly swapped-in session, through the same facade reference. electionA.SetLeader(false); electionB.SetLeader(true); DataValue stateB = await blockedOnB.WaitAsync(timeout, ct).ConfigureAwait(false); Assert.That(StatusCode.IsGood(stateB.StatusCode), Is.True); Assert.That(replicaB.IsLeader, Is.True); Assert.That(replicaA.IsLeader, Is.False); // The same B facade keeps serving fresh traffic after the swap. BrowseResponse browseB = await BrowseObjectsAsync(replicaB, ct) .WaitAsync(timeout, ct).ConfigureAwait(false); Assert.That(browseB.Results.Count, Is.GreaterThan(0)); DataValue stateBAgain = await ReadServerStateAsync(replicaB, ct) .WaitAsync(timeout, ct).ConfigureAwait(false); Assert.That(StatusCode.IsGood(stateBAgain.StatusCode), Is.True); } finally { if (replicaA != null) { await replicaA.DisposeAsync().ConfigureAwait(false); } if (blockedOnB != null) { await blockedOnB.WaitAsync(TimeSpan.FromSeconds(5), CancellationToken.None).ConfigureAwait(false); } if (replicaB != null) { await replicaB.DisposeAsync().ConfigureAwait(false); } if (fixture != null) { await fixture.StopAsync().ConfigureAwait(false); } } } private RedundantClientSession BuildReplica( string nodeId, ConfiguredEndpoint endpoint, ILeaderElection election, ISharedKeyValueStore store) { return new RedundantClientSessionBuilder(Telemetry) .WithNodeId(nodeId) .WithStandbyMode(ClientStandbyMode.Cold) .UseSession(token => new ValueTask( new ManagedSessionBuilder(ClientFixture.Config, Telemetry) .UseEndpoint(endpoint) .WithSessionName(nodeId) .ConnectAsync(token))) .UseRedundancy(election, store, NullRecordProtector.Instance) .Build(); } private async Task<(ServerFixture Fixture, Uri Url)> StartServerAsync() { var fixture = new ServerFixture(telemetry => new ReferenceServer(telemetry)) { UriScheme = Utils.UriSchemeOpcTcp, SecurityNone = true, AutoAccept = true, OperationLimits = true }; await fixture.StartAsync().ConfigureAwait(false); return (fixture, new Uri($"{Utils.UriSchemeOpcTcp}://localhost:{fixture.Port}")); } private static async Task ReadServerStateAsync(TSession session, CancellationToken ct) where TSession : ISession { ReadValueId[] nodesToRead = [ new ReadValueId { NodeId = VariableIds.Server_ServerStatus_State, AttributeId = Attributes.Value } ]; ReadResponse response = await session .ReadAsync(null, 0, TimestampsToReturn.Both, nodesToRead, ct) .ConfigureAwait(false); return response.Results[0]; } private static async Task BrowseObjectsAsync(TSession session, CancellationToken ct) where TSession : ISession { var description = new BrowseDescription { NodeId = ObjectIds.ObjectsFolder, BrowseDirection = BrowseDirection.Forward, ReferenceTypeId = ReferenceTypeIds.HierarchicalReferences, IncludeSubtypes = true, ResultMask = (uint)BrowseResultMask.All }; BrowseDescription[] nodesToBrowse = [description]; return await session .BrowseAsync(null, null, 0u, nodesToBrowse, ct) .ConfigureAwait(false); } /// /// A leader election whose leadership can be flipped from the test so a handoff is /// deterministic (the shipped StaticLeaderElection is fixed at construction). /// private sealed class ControllableLeaderElection : ILeaderElection { public bool IsLeader => m_isLeader; public event Action? LeadershipChanged; public ValueTask TryAcquireOrRenewAsync(CancellationToken ct = default) { return new ValueTask(m_isLeader); } public void Start() { LeadershipChanged?.Invoke(m_isLeader); } public void SetLeader(bool isLeader) { m_isLeader = isLeader; LeadershipChanged?.Invoke(isLeader); } public ValueTask DisposeAsync() { return default; } private volatile bool m_isLeader; } } }