/* ========================================================================
* 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/
* ======================================================================*/
// CA2000: system-under-test disposables are created per test and released at teardown;
// there is no cross-test resource leak. Suppressed file-level for the suite.
#pragma warning disable CA2000 // Dispose objects before losing scope
// CA2007: tests run without a SynchronizationContext; ConfigureAwait(false)
// adds noise without a behavioural benefit. Disabled file-level for the suite.
#pragma warning disable CA2007
using System;
using System.Collections.Generic;
using System.Threading;
using System.Threading.Tasks;
using Crdt;
using Crdt.Transport;
using Moq;
using NUnit.Framework;
using Opc.Ua.Redundancy;
using Opc.Ua.Redundancy.Server;
using Opc.Ua.Tests;
namespace Opc.Ua.Server.Tests.Redundancy
{
///
/// Tests for : it wires a
/// synchronizer to every opted-in node manager and (as writer) seeds the
/// node manager's address space into the shared store.
///
[TestFixture]
[Category("Distributed")]
[Parallelizable(ParallelScope.All)]
public class DistributedAddressSpaceStartupTaskTests
{
private const ushort NamespaceIndex = 1;
///
/// Verifies that startup seeds only the namespaces owned by synchronous and asynchronous node managers.
///
[TestCase(false)]
[TestCase(true)]
public async Task WiresSynchronizerAndSeedsOptedInNodeManagerAsync(
bool useAsyncNodeManager)
{
ITelemetryContext telemetry = NUnitTelemetryContext.Create();
var messageContext = ServiceMessageContext.CreateEmpty(telemetry);
messageContext.NamespaceUris.GetIndexOrAppend("urn:test:wire");
ushort foreignNamespaceIndex = messageContext.NamespaceUris.GetIndexOrAppend("urn:test:foreign");
var systemContext = new SystemContext(telemetry)
{
NamespaceUris = messageContext.NamespaceUris,
ServerUris = messageContext.ServerUris,
EncodeableFactory = messageContext.Factory
};
// A node manager that opts in and exposes a one-node address space.
var addressSpace = new DictionaryAddressSpace(systemContext);
var nodeId = new NodeId("seeded", NamespaceIndex);
await addressSpace.AddOrUpdateNodeAsync(new BaseDataVariableState(null)
{
NodeId = nodeId,
BrowseName = new QualifiedName("Seeded", NamespaceIndex),
DisplayName = new LocalizedText("Seeded"),
DataType = DataTypeIds.Double,
ValueRank = ValueRanks.Scalar,
Value = new Variant(1.0)
}).ConfigureAwait(false);
var standardNodeId = new NodeId(22u);
await addressSpace.AddOrUpdateNodeAsync(new BaseObjectState(null)
{
NodeId = standardNodeId,
BrowseName = new QualifiedName("Standard"),
DisplayName = new LocalizedText("Standard")
}).ConfigureAwait(false);
var foreignNodeId = new NodeId("foreign", foreignNamespaceIndex);
await addressSpace.AddOrUpdateNodeAsync(new BaseObjectState(null)
{
NodeId = foreignNodeId,
BrowseName = new QualifiedName("Foreign", foreignNamespaceIndex),
DisplayName = new LocalizedText("Foreign")
}).ConfigureAwait(false);
ILocalAddressSpaceSource source;
if (useAsyncNodeManager)
{
var nodeManager = new Mock();
nodeManager.SetupGet(m => m.NamespaceUris).Returns(["urn:test:wire"]);
nodeManager.As()
.Setup(s => s.CreateLocalAddressSpace())
.Returns(addressSpace);
source = nodeManager.As().Object;
}
else
{
var nodeManager = new Mock();
nodeManager.SetupGet(m => m.NamespaceUris).Returns(["urn:test:wire"]);
nodeManager.As()
.Setup(s => s.CreateLocalAddressSpace())
.Returns(addressSpace);
source = nodeManager.As().Object;
}
var masterNodeManager = new Mock();
masterNodeManager.Setup(m => m.NodeManagers).Returns([]);
var server = new Mock();
server.Setup(s => s.Telemetry).Returns(telemetry);
server.Setup(s => s.MessageContext).Returns(messageContext);
server.Setup(s => s.NamespaceUris).Returns(messageContext.NamespaceUris);
server.Setup(s => s.NodeManager).Returns(masterNodeManager.Object);
server.Setup(s => s.DefaultSystemContext).Returns(new ServerSystemContext(server.Object));
server.Setup(s => s.FindNodeManagers())
.Returns([source]);
using var kv = new InMemorySharedKeyValueStore();
var election = new StaticLeaderElection(true);
var task = new DistributedAddressSpaceStartupTask(kv, election);
await task.OnServerStartedAsync(server.Object).ConfigureAwait(false);
// The writer must have seeded the node into the shared store, and
// registered a default store in the task-owned registry.
var verifyStore = new InMemoryNodeStateStore(kv, messageContext);
IStoredNode? stored = await verifyStore.TryGetNodeAsync(nodeId).ConfigureAwait(false);
IStoredNode? standard = await verifyStore.TryGetNodeAsync(standardNodeId).ConfigureAwait(false);
IStoredNode? foreign = await verifyStore.TryGetNodeAsync(foreignNodeId).ConfigureAwait(false);
Assert.That(stored, Is.Not.Null, "writer should have seeded the opted-in node manager's address space");
Assert.That(standard, Is.Null, "replica-local standard nodes must not be distributed");
Assert.That(foreign, Is.Null, "a node manager must not distribute nodes owned by another namespace");
Assert.That(task.NodeStateStoreRegistry, Is.Not.Null);
Assert.That(task.NodeStateStoreRegistry!.Resolve(nodeId), Is.Not.Null, "a default node state store should be registered");
await task.DisposeAsync().ConfigureAwait(false);
}
[Test]
public async Task UsesExplicitOwnershipForCustomPartitionAsync()
{
ITelemetryContext telemetry = NUnitTelemetryContext.Create();
var messageContext = ServiceMessageContext.CreateEmpty(telemetry);
ushort ownedNamespaceIndex =
messageContext.NamespaceUris.GetIndexOrAppend("urn:test:owned");
ushort foreignNamespaceIndex =
messageContext.NamespaceUris.GetIndexOrAppend("urn:test:foreign");
var systemContext = new SystemContext(telemetry)
{
NamespaceUris = messageContext.NamespaceUris,
ServerUris = messageContext.ServerUris,
EncodeableFactory = messageContext.Factory
};
var addressSpace = new DictionaryAddressSpace(systemContext);
var ownedNodeId = new NodeId("owned", ownedNamespaceIndex);
var foreignNodeId = new NodeId("foreign", foreignNamespaceIndex);
await addressSpace.AddOrUpdateNodeAsync(new BaseObjectState(null)
{
NodeId = ownedNodeId,
BrowseName = new QualifiedName("Owned", ownedNamespaceIndex),
DisplayName = new LocalizedText("Owned")
}).ConfigureAwait(false);
await addressSpace.AddOrUpdateNodeAsync(new BaseObjectState(null)
{
NodeId = foreignNodeId,
BrowseName = new QualifiedName("Foreign", foreignNamespaceIndex),
DisplayName = new LocalizedText("Foreign")
}).ConfigureAwait(false);
var nodeManager = new Mock();
nodeManager.SetupGet(m => m.NamespaceUris).Returns((IEnumerable)null!);
nodeManager.As()
.Setup(s => s.CreateLocalAddressSpace())
.Returns(addressSpace);
nodeManager.As()
.SetupGet(o => o.PartitionId)
.Returns("custom-partition");
nodeManager.As()
.Setup(o => o.OwnsNode(It.IsAny()))
.Returns(nodeId => nodeId == ownedNodeId);
var masterNodeManager = new Mock();
masterNodeManager.Setup(m => m.NodeManagers).Returns([]);
var server = new Mock();
server.Setup(s => s.Telemetry).Returns(telemetry);
server.Setup(s => s.MessageContext).Returns(messageContext);
server.Setup(s => s.NamespaceUris).Returns(messageContext.NamespaceUris);
server.Setup(s => s.NodeManager).Returns(masterNodeManager.Object);
server.Setup(s => s.DefaultSystemContext).Returns(new ServerSystemContext(server.Object));
server.Setup(s => s.FindNodeManagers())
.Returns([nodeManager.As().Object]);
using var kv = new InMemorySharedKeyValueStore();
var election = new StaticLeaderElection(true);
var task = new DistributedAddressSpaceStartupTask(kv, election);
await task.OnServerStartedAsync(server.Object).ConfigureAwait(false);
var verifyStore = new InMemoryNodeStateStore(kv, messageContext);
Assert.That(
await verifyStore.TryGetNodeAsync(ownedNodeId).ConfigureAwait(false),
Is.Not.Null);
Assert.That(
await verifyStore.TryGetNodeAsync(foreignNodeId).ConfigureAwait(false),
Is.Null);
await task.DisposeAsync().ConfigureAwait(false);
}
///
/// Verifies that startup excludes replica-local built-in address spaces from replication.
///
[Test]
public async Task DoesNotReplicateReplicaLocalBuiltInAddressSpacesAsync()
{
ITelemetryContext telemetry = NUnitTelemetryContext.Create();
var messageContext = ServiceMessageContext.CreateEmpty(telemetry);
var systemContext = new SystemContext(telemetry)
{
NamespaceUris = messageContext.NamespaceUris,
ServerUris = messageContext.ServerUris,
EncodeableFactory = messageContext.Factory
};
var addressSpace = new DictionaryAddressSpace(systemContext);
await addressSpace.AddOrUpdateNodeAsync(new BaseObjectState(null)
{
NodeId = ObjectIds.Server,
BrowseName = new QualifiedName(BrowseNames.Server),
DisplayName = new LocalizedText("Server")
}).ConfigureAwait(false);
var diagnosticsNodeManager = new Mock();
Mock diagnosticsSource =
diagnosticsNodeManager.As();
diagnosticsSource
.Setup(value => value.CreateLocalAddressSpace())
.Returns(addressSpace);
var coreNodeManager = new Mock();
Mock coreSource =
coreNodeManager.As();
coreSource
.Setup(value => value.CreateLocalAddressSpace())
.Returns(addressSpace);
var server = new Mock();
server.Setup(value => value.Telemetry).Returns(telemetry);
server.Setup(value => value.MessageContext).Returns(messageContext);
server.Setup(value => value.NamespaceUris).Returns(messageContext.NamespaceUris);
server.Setup(value => value.DefaultSystemContext).Returns(new ServerSystemContext(server.Object));
server
.Setup(value => value.FindNodeManagers())
.Returns([diagnosticsSource.Object, coreSource.Object]);
using var keyValueStore = new InMemorySharedKeyValueStore();
await using var startup = new DistributedAddressSpaceStartupTask(
keyValueStore,
new StaticLeaderElection(true));
await startup.OnServerStartedAsync(server.Object).ConfigureAwait(false);
using var store = new InMemoryNodeStateStore(keyValueStore, messageContext);
Assert.That(
await store.TryGetNodeAsync(ObjectIds.Server).ConfigureAwait(false),
Is.Null,
"Replica-local core, diagnostics, and configuration nodes must never enter the shared address space.");
diagnosticsSource.Verify(value => value.CreateLocalAddressSpace(), Times.Never);
coreSource.Verify(value => value.CreateLocalAddressSpace(), Times.Never);
}
///
/// Verifies that the distributed address-space startup task rejects null constructor dependencies.
///
[Test]
public async Task DuplicateNamespaceClaimsRequireExplicitOwnershipAsync()
{
ITelemetryContext telemetry = NUnitTelemetryContext.Create();
var messageContext = ServiceMessageContext.CreateEmpty(telemetry);
messageContext.NamespaceUris.GetIndexOrAppend("urn:test:shared");
var systemContext = new SystemContext(telemetry)
{
NamespaceUris = messageContext.NamespaceUris,
ServerUris = messageContext.ServerUris,
EncodeableFactory = messageContext.Factory
};
var firstSpace = new DictionaryAddressSpace(systemContext);
var secondSpace = new DictionaryAddressSpace(systemContext);
var firstManager = new Mock();
firstManager.SetupGet(m => m.NamespaceUris).Returns(["urn:test:shared"]);
firstManager.As()
.Setup(s => s.CreateLocalAddressSpace())
.Returns(firstSpace);
var secondManager = new Mock();
secondManager.SetupGet(m => m.NamespaceUris).Returns(["urn:test:shared"]);
secondManager.As()
.Setup(s => s.CreateLocalAddressSpace())
.Returns(secondSpace);
var masterNodeManager = new Mock();
masterNodeManager.Setup(m => m.NodeManagers).Returns([]);
var server = new Mock();
server.Setup(s => s.Telemetry).Returns(telemetry);
server.Setup(s => s.MessageContext).Returns(messageContext);
server.Setup(s => s.NamespaceUris).Returns(messageContext.NamespaceUris);
server.Setup(s => s.NodeManager).Returns(masterNodeManager.Object);
server.Setup(s => s.DefaultSystemContext).Returns(new ServerSystemContext(server.Object));
server.Setup(s => s.FindNodeManagers())
.Returns(
[
firstManager.As().Object,
secondManager.As().Object
]);
using var kv = new InMemorySharedKeyValueStore();
var election = new Mock();
var task = new DistributedAddressSpaceStartupTask(kv, election.Object);
try
{
await Assert.ThatAsync(
async () => await task.OnServerStartedAsync(server.Object).ConfigureAwait(false),
Throws.TypeOf()).ConfigureAwait(false);
}
finally
{
await task.DisposeAsync().ConfigureAwait(false);
}
election.Verify(
e => e.TryAcquireOrRenewAsync(It.IsAny()),
Times.Never);
election.Verify(e => e.Start(), Times.Never);
}
[Test]
public async Task ExplicitOwnershipCannotClaimNamespaceZeroAsync()
{
ITelemetryContext telemetry = NUnitTelemetryContext.Create();
var messageContext = ServiceMessageContext.CreateEmpty(telemetry);
var systemContext = new SystemContext(telemetry)
{
NamespaceUris = messageContext.NamespaceUris,
ServerUris = messageContext.ServerUris,
EncodeableFactory = messageContext.Factory
};
var addressSpace = new DictionaryAddressSpace(systemContext);
await addressSpace.AddOrUpdateNodeAsync(new BaseObjectState(null)
{
NodeId = new NodeId(22u),
BrowseName = new QualifiedName("Standard"),
DisplayName = new LocalizedText("Standard")
}).ConfigureAwait(false);
var nodeManager = new Mock();
nodeManager.SetupGet(m => m.NamespaceUris).Returns((IEnumerable)null!);
nodeManager.As()
.Setup(s => s.CreateLocalAddressSpace())
.Returns(addressSpace);
nodeManager.As()
.SetupGet(o => o.PartitionId)
.Returns("invalid-standard");
nodeManager.As()
.Setup(o => o.OwnsNode(It.IsAny()))
.Returns(true);
var masterNodeManager = new Mock();
masterNodeManager.Setup(m => m.NodeManagers).Returns([]);
var server = new Mock();
server.Setup(s => s.Telemetry).Returns(telemetry);
server.Setup(s => s.MessageContext).Returns(messageContext);
server.Setup(s => s.NamespaceUris).Returns(messageContext.NamespaceUris);
server.Setup(s => s.NodeManager).Returns(masterNodeManager.Object);
server.Setup(s => s.DefaultSystemContext).Returns(new ServerSystemContext(server.Object));
server.Setup(s => s.FindNodeManagers())
.Returns([nodeManager.As().Object]);
using var kv = new InMemorySharedKeyValueStore();
var election = new Mock();
var task = new DistributedAddressSpaceStartupTask(kv, election.Object);
try
{
await Assert.ThatAsync(
async () => await task.OnServerStartedAsync(server.Object).ConfigureAwait(false),
Throws.TypeOf()).ConfigureAwait(false);
}
finally
{
await task.DisposeAsync().ConfigureAwait(false);
}
election.Verify(
e => e.TryAcquireOrRenewAsync(It.IsAny()),
Times.Never);
}
///
/// Rejects an unsupported CRDT writer before creating address spaces, acquiring a lease, or writing markers.
///
[Test]
public async Task BareCrdtFailsBeforeOwnershipOrElectionAsync()
{
await using var network = new InMemoryNetwork();
await using var crdt = new ReplicatedSharedKeyValueStore(
ReplicaId.New(), network.CreateTransport(), TimeProvider.System, CrdtReaderOptions.Default);
var election = new Mock();
var server = new Mock(MockBehavior.Strict);
await using var startup = new DistributedAddressSpaceStartupTask(crdt, election.Object);
await Assert.ThatAsync(
async () => await startup.OnServerStartedAsync(server.Object).ConfigureAwait(false),
Throws.InvalidOperationException.With.Message.Contains(InMemoryNodeStateStore.SequenceKey))
.ConfigureAwait(false);
election.Verify(e => e.TryAcquireOrRenewAsync(It.IsAny()), Times.Never);
election.Verify(e => e.Start(), Times.Never);
server.VerifyNoOtherCalls();
Assert.That(startup.NodeStateStoreRegistry, Is.Null);
int count = 0;
await foreach (KeyValuePair _ in crdt.ScanAsync(string.Empty).ConfigureAwait(false))
{
count++;
}
Assert.That(count, Is.Zero);
}
///
/// Rejects a configured lease routed to CRDT even when the sequence key is correctly coordinated.
///
[Test]
public async Task MisroutedLeaseFailsBeforeStartupMutationAsync()
{
await using var network = new InMemoryNetwork();
await using var crdt = new ReplicatedSharedKeyValueStore(
ReplicaId.New(), network.CreateTransport(), TimeProvider.System, CrdtReaderOptions.Default);
await using var coordinator = new RaftSharedKeyValueStore(
DefaultRaftConsensus.CreateSingleNode(), ownsConsensus: true);
await using var hybrid = new HybridSharedKeyValueStore(
crdt, coordinator, [InMemoryNodeStateStore.SequenceKey]);
var election = new Mock();
var server = new Mock(MockBehavior.Strict);
await using var startup = new DistributedAddressSpaceStartupTask(
hybrid, election.Object, protector: null, leaseKey: "custom/leader");
await Assert.ThatAsync(
async () => await startup.OnServerStartedAsync(server.Object).ConfigureAwait(false),
Throws.InvalidOperationException.With.Message.Contains("custom/leader")).ConfigureAwait(false);
election.Verify(e => e.TryAcquireOrRenewAsync(It.IsAny()), Times.Never);
election.Verify(e => e.Start(), Times.Never);
server.VerifyNoOtherCalls();
Assert.That(startup.NodeStateStoreRegistry, Is.Null);
}
///
/// Rejects replicated startup before accessing the address space when no fixed identity policy was configured.
///
[Test]
public async Task ReplicatedStartupRequiresNodeIdentityConfigurationAsync()
{
await using var coordinator = new RaftSharedKeyValueStore(
DefaultRaftConsensus.CreateSingleNode(), ownsConsensus: true);
var election = new Mock();
var server = new Mock(MockBehavior.Strict);
await using var startup = new DistributedAddressSpaceStartupTask(coordinator, election.Object);
await Assert.ThatAsync(
async () => await startup.OnServerStartedAsync(server.Object).ConfigureAwait(false),
Throws.TypeOf()
.With.Property(nameof(ServiceResultException.StatusCode))
.EqualTo(StatusCodes.BadConfigurationError)).ConfigureAwait(false);
election.Verify(value => value.TryAcquireOrRenewAsync(It.IsAny()), Times.Never);
server.VerifyNoOtherCalls();
Assert.That(startup.NodeStateStoreRegistry, Is.Null);
}
///
/// Wires a hosted registry and a directly constructed store to the same strong coordinator and real CRDT bulk.
///
[Test]
public async Task HostedRegistryAndDirectStoreShareHybridCoordinationAsync()
{
ITelemetryContext telemetry = NUnitTelemetryContext.Create();
var context = ServiceMessageContext.CreateEmpty(telemetry);
context.NamespaceUris.GetIndexOrAppend("urn:test:hybrid-application");
context.NamespaceUris.GetIndexOrAppend("urn:test:shared-hybrid");
var identity = new ReplicaNodeIdFactory("shared-hybrid", ["urn:test:shared-hybrid"]);
var server = new Mock();
server.As().Setup(s => s.NodeIdFactory).Returns(identity);
server.Setup(s => s.Telemetry).Returns(telemetry);
server.Setup(s => s.MessageContext).Returns(context);
server.Setup(s => s.NamespaceUris).Returns(context.NamespaceUris);
server.Setup(s => s.DefaultSystemContext).Returns(new ServerSystemContext(server.Object));
server.Setup(s => s.FindNodeManagers()).Returns([]);
await using var network = new InMemoryNetwork();
await using var crdt = new ReplicatedSharedKeyValueStore(
ReplicaId.New(), network.CreateTransport(), TimeProvider.System, CrdtReaderOptions.Default);
await using var coordinator = new RaftSharedKeyValueStore(
DefaultRaftConsensus.CreateSingleNode(), ownsConsensus: true);
using var protector = new AesCbcHmacRecordProtector(new byte[32]);
await identity.InitializeNewStoreAsync(coordinator, protector).ConfigureAwait(false);
await using var hybrid = new HybridSharedKeyValueStore(crdt, coordinator);
await using var startup = new DistributedAddressSpaceStartupTask(
hybrid, new StaticLeaderElection(true), protector);
await startup.OnServerStartedAsync(server.Object).ConfigureAwait(false);
var hostedId = new NodeId("hosted", 2);
var directId = new NodeId("direct", 2);
Assert.That(startup.NodeStateStoreRegistry, Is.Not.Null);
INodeStateStore? hosted = startup.NodeStateStoreRegistry!.Resolve(hostedId);
Assert.That(hosted, Is.Not.Null);
await hosted!.UpsertNodeAsync(new StoredNode(hostedId, new ByteString(new byte[] { 1 })))
.ConfigureAwait(false);
using var direct = new InMemoryNodeStateStore(hybrid, context, protector);
await direct.UpsertNodeAsync(new StoredNode(directId, new ByteString(new byte[] { 2 })))
.ConfigureAwait(false);
var changes = new List();
await foreach (NodeStateChange change in direct.ReadDeltaLogAsync(0).ConfigureAwait(false))
{
changes.Add(change);
}
Assert.That(changes, Has.Count.EqualTo(2));
Assert.That(changes[0].Sequence, Is.EqualTo(1));
Assert.That(changes[1].Sequence, Is.EqualTo(2));
(bool hostedStored, _) = await crdt.TryGetAsync("n/" + hostedId).ConfigureAwait(false);
(bool directStored, _) = await crdt.TryGetAsync("n/" + directId).ConfigureAwait(false);
(bool misplacedCounter, _) = await crdt.TryGetAsync(InMemoryNodeStateStore.SequenceKey)
.ConfigureAwait(false);
Assert.That(hostedStored, Is.True);
Assert.That(directStored, Is.True);
Assert.That(misplacedCounter, Is.False);
}
[Test]
public void ConstructorThrowsOnNullArguments()
{
using var kv = new InMemorySharedKeyValueStore();
var election = new StaticLeaderElection(true);
Assert.That(() => new DistributedAddressSpaceStartupTask(null!, election), Throws.ArgumentNullException);
Assert.That(() => new DistributedAddressSpaceStartupTask(kv, null!), Throws.ArgumentNullException);
}
}
}