/* ========================================================================
* 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.Collections.Generic;
using System.Linq;
using System.Threading;
using System.Threading.Tasks;
using Microsoft.Extensions.Logging;
using Opc.Ua;
using Opc.Ua.Redundancy;
using Opc.Ua.Redundancy.Server;
using Opc.Ua.Server;
using Opc.Ua.Server.Fluent;
using Opc.Ua.Server.Historian;
using Opc.Ua.Server.Hosting;
namespace RedundantServer
{
///
/// Factory that produces instances for DI-hosted servers.
///
public sealed class HaSampleNodeManagerFactory : IAsyncNodeManagerFactory
{
///
/// Shared namespace reserved at the same index before every replica's node managers are created.
///
public const string NamespaceUri = "http://opcfoundation.org/UA/Samples/HighAvailability";
private readonly ILeaderElection m_leaderElection;
private readonly HaSampleReplicaInfo m_replicaInfo;
private readonly IDistributedValueCache? m_valueCache;
private readonly SharedKeyValueHistorianProvider? m_historian;
///
/// Creates a factory using the distributed leader-election service registered by the host.
///
/// The leader-election service that identifies the active writer replica.
/// The local replica identity published by sample variables.
///
/// The distributed value cache registered by UseDistributedAddressSpace, when present. Injected as
/// a sequence so active/active and single-instance topologies (which register no cache) resolve to empty.
///
public HaSampleNodeManagerFactory(
ILeaderElection leaderElection,
HaSampleReplicaInfo replicaInfo,
IEnumerable valueCaches,
IEnumerable historians)
{
m_leaderElection = leaderElection ?? throw new ArgumentNullException(nameof(leaderElection));
m_replicaInfo = replicaInfo ?? throw new ArgumentNullException(nameof(replicaInfo));
m_valueCache = valueCaches?.FirstOrDefault();
m_historian = historians?.FirstOrDefault();
}
///
public ArrayOf NamespacesUris => [NamespaceUri];
///
public ValueTask CreateAsync(
IServerInternal server,
ApplicationConfiguration configuration,
CancellationToken cancellationToken = default)
{
_ = configuration;
_ = cancellationToken;
#pragma warning disable CA2000 // ownership transfers to the server
var manager = new HaSampleNodeManager(
server,
m_leaderElection,
m_replicaInfo,
m_valueCache,
m_historian,
[.. NamespacesUris]);
#pragma warning restore CA2000
return new ValueTask(manager);
}
}
///
/// Carries the local replica identity into sample node managers created by dependency injection.
///
public sealed class HaSampleReplicaInfo
{
///
/// Creates a replica identity descriptor.
///
/// The unique local high-availability node id.
public HaSampleReplicaInfo(string nodeId)
{
NodeId = nodeId;
}
///
/// Gets the unique local high-availability node id.
///
public string NodeId { get; }
}
///
/// Starts the sample producer after distributed services and redundancy metadata are initialized.
///
internal sealed class HaSampleSimulationStartupTask : IServerStartupTask
{
///
public ValueTask OnServerStartedAsync(
IServerContext server,
CancellationToken cancellationToken = default)
{
if (server == null)
{
throw new ArgumentNullException(nameof(server));
}
cancellationToken.ThrowIfCancellationRequested();
foreach (HaSampleNodeManager manager in server.FindNodeManagers())
{
manager.StartSimulation();
}
return default;
}
}
///
/// Minimal address space that participates in distributed replication.
///
public sealed class HaSampleNodeManager : FluentNodeManagerBase
{
private readonly ILeaderElection m_leaderElection;
private readonly HaSampleReplicaInfo m_replicaInfo;
private readonly IDistributedValueCache? m_valueCache;
private readonly SharedKeyValueHistorianProvider? m_historian;
private readonly CancellationTokenSource m_simulationCts = new();
private readonly Lock m_updateLock = new();
private static readonly TimeSpan s_valueFreshness = TimeSpan.FromSeconds(10);
private const string kSampleFolderName = "HighAvailability";
private FolderState? m_folder;
private BaseVariableState? m_counter;
private BaseVariableState? m_activeReplica;
private BaseObjectState? m_historyEvents;
private Task? m_simulationTask;
private int m_counterValue;
///
/// Creates the high-availability sample node manager.
///
/// The server that owns the node manager.
/// The leader-election service that gates sample writes.
/// The local replica identity published by sample variables.
///
/// The distributed value cache used to share the Counter value across the replica set, or null when
/// the topology registers none (active/active and single-instance).
///
///
/// The shared historian used by strong active/passive deployments, or null for other topologies.
///
/// The namespace URIs exposed by this node manager.
public HaSampleNodeManager(
IServerInternal server,
ILeaderElection leaderElection,
HaSampleReplicaInfo replicaInfo,
IDistributedValueCache? valueCache,
SharedKeyValueHistorianProvider? historian,
params string[] namespaceUris)
: base(server, namespaceUris)
{
m_leaderElection = leaderElection ?? throw new ArgumentNullException(nameof(leaderElection));
m_replicaInfo = replicaInfo ?? throw new ArgumentNullException(nameof(replicaInfo));
m_valueCache = valueCache;
m_historian = historian;
}
///
/// Mints the browse name as the identifier, for the handful of nodes
/// this sample publishes by name.
///
///
///
/// Those identifiers are part of the sample's contract - the redundant
/// client addresses ns=N;s=Counter directly - so they cannot be
/// left to the derived form the default factory mints.
///
///
/// Every other node keeps that derived form, and the narrowness
/// matters: a browse name is only unique among its siblings, and the
/// historian hangs an HA Configuration object off each node it
/// historizes, so a blanket rule would hand the Counter's and the
/// HistoryEvents' one the same identifier. An identifier a caller
/// already chose here is kept, on the default factory's own terms.
///
///
public override NodeId New(ISystemContext context, NodeState node)
{
if (node == null)
{
return base.New(context, node!);
}
ushort namespaceIndex = NamespaceIndexes[0];
if (!node.NodeId.IsNull &&
(node.NodeId.NamespaceIndex == namespaceIndex ||
node is not BaseInstanceState { Parent: not null }))
{
return node.NodeId;
}
return IsPublishedByName(node)
? new NodeId(node.BrowseName.Name!, namespaceIndex)
: base.New(context, node);
}
///
/// Whether is the sample folder or one of the
/// nodes directly beneath it - the ones a client spells out.
///
///
/// The folder reaches this before it has been staged, so it has no
/// parent yet and is recognised by its browse name instead.
///
private bool IsPublishedByName(NodeState node)
{
if (string.IsNullOrEmpty(node.BrowseName.Name))
{
return false;
}
NodeState? parent = (node as BaseInstanceState)?.Parent;
return parent == null
? node.BrowseName.Name == kSampleFolderName
: ReferenceEquals(parent, m_folder);
}
///
public override async ValueTask CreateAddressSpaceAsync(
IDictionary> externalReferences,
CancellationToken cancellationToken = default)
{
NodeManagerBuilder builder = CreateFluentBuilder(NamespaceIndexes[0]);
// Unparented, so the staged folder organises itself under the
// Objects folder; the manager used to add that reference and its
// externalReferences counterpart by hand.
INodeBuilder folder = builder.AddFolder(kSampleFolderName);
folder.Node.DisplayName = new LocalizedText("en", "High Availability");
NodeId folderId = folder.Node.NodeId;
// New() names the folder's direct children after their browse
// name, so it has to know which node the folder is.
m_folder = folder.Node;
m_counter = AddSampleVariable(builder, folderId, "Counter", Variant.From(0));
if (m_valueCache != null)
{
// Opt the Counter's read/write callbacks into the distributed
// value cache (extension beyond OPC 10000-4 ยง6.6): the active
// replica caches each new value write-through, and every replica
// serves the last value shared through the store while it is
// fresh (falling back to the local value). This is how the
// Counter is shared across the set and continues after failover;
// monitored items keep reading through the normal pipeline.
m_counter.EnableDistributedValueParticipation(
m_valueCache,
s_valueFreshness,
_ => new ValueTask(ReadLocalCounter()));
}
m_activeReplica = AddSampleVariable(
builder,
folderId,
"ActiveReplica",
Variant.From("unknown"));
m_historyEvents = builder.AddObject("HistoryEvents", folderId).Node;
m_historyEvents.ReferenceTypeId = ReferenceTypeIds.Organizes;
m_historyEvents.DisplayName = new LocalizedText("en", "History Events");
m_historyEvents.EventNotifier = EventNotifiers.SubscribeToEvents;
// Deliberately not staged: this subtree exists to show what a
// factory-assigned identity looks like next to the named ones
// above, so it keeps minting through the NodeIdFactory itself
// rather than through New. Attaching it once the folder has been
// staged leaves those identifiers untouched, and the subtree is
// still registered as part of the folder's.
AddFactoryAssignedNodes(folder.Node, NamespaceIndexes[0]);
if (m_historian != null)
{
#pragma warning disable CA2000 // ownership transfers to the server's historian-builder registry
HistorianBuilder historian = new HistorianBuilder(Server)
.UseProvider(m_historian);
#pragma warning restore CA2000
await historian.HistorizeAsync(
m_counter,
SystemContext,
historyAccessLevel: AccessLevels.HistoryRead | AccessLevels.HistoryWrite,
capabilities: HistorianNodeCapabilities.DataReadWrite,
autoCapture: false,
cancellationToken: cancellationToken).ConfigureAwait(false);
await historian.HistorizeEventsAsync(
m_historyEvents,
SystemContext,
capabilities: HistorianNodeCapabilities.EventReadWrite with
{
EventTypes = [ObjectTypeIds.BaseEventType]
},
cancellationToken: cancellationToken).ConfigureAwait(false);
}
// Registering the staged subtree and re-running the reverse
// reference pass is what publishes the folder's inverse Organizes
// reference into externalReferences[ObjectsFolder].
await RegisterAuthoredNodesAsync(builder, cancellationToken).ConfigureAwait(false);
await CompleteConfigureAsync(externalReferences, cancellationToken)
.ConfigureAwait(false);
await SealConfigurationAsync(builder, cancellationToken).ConfigureAwait(false);
}
private void AddFactoryAssignedNodes(FolderState parent, ushort namespaceIndex)
{
var generated = new BaseObjectState(parent)
{
BrowseName = new QualifiedName("FactoryAssigned", namespaceIndex),
DisplayName = new LocalizedText("Factory-assigned identities"),
ReferenceTypeId = ReferenceTypeIds.HasComponent,
TypeDefinitionId = ObjectTypeIds.BaseObjectType
};
generated.NodeId = NodeIdFactory.New(SystemContext, generated);
parent.AddChild(generated);
var value = new BaseDataVariableState(generated)
{
BrowseName = new QualifiedName("Value", namespaceIndex),
DisplayName = new LocalizedText("Value"),
ReferenceTypeId = ReferenceTypeIds.HasComponent,
TypeDefinitionId = VariableTypeIds.BaseDataVariableType,
DataType = DataTypeIds.Int32,
ValueRank = ValueRanks.Scalar,
AccessLevel = AccessLevels.CurrentRead,
UserAccessLevel = AccessLevels.CurrentRead,
Value = Variant.From(12345),
StatusCode = StatusCodes.Good
};
value.NodeId = NodeIdFactory.New(SystemContext, value);
generated.AddChild(value);
var target = new BaseDataVariableState(generated)
{
BrowseName = new QualifiedName("Target", namespaceIndex),
DisplayName = new LocalizedText("Target"),
ReferenceTypeId = ReferenceTypeIds.HasComponent,
TypeDefinitionId = VariableTypeIds.BaseDataVariableType,
DataType = DataTypeIds.NodeId,
ValueRank = ValueRanks.Scalar,
AccessLevel = AccessLevels.CurrentRead,
UserAccessLevel = AccessLevels.CurrentRead,
Value = Variant.From(value.NodeId),
StatusCode = StatusCodes.Good
};
target.NodeId = NodeIdFactory.New(SystemContext, target);
generated.AddChild(target);
}
///
protected override void Dispose(bool disposing)
{
if (disposing)
{
m_simulationCts.Cancel();
m_simulationCts.Dispose();
}
base.Dispose(disposing);
}
///
protected override IHistorianProvider? GetHistorianProvider(NodeState node)
{
return m_historian ?? base.GetHistorianProvider(node);
}
///
/// Stages one writable sample variable under the sample folder.
///
///
/// CLR type the variable carries; the staged node takes its DataType
/// and ValueRank from it.
///
/// The fluent builder staging the subtree.
/// The folder the variable hangs off.
/// Browse name of the variable.
/// The value a client reads before the
/// simulation has produced one.
private static BaseVariableState AddSampleVariable(
NodeManagerBuilder builder,
NodeId parentId,
string name,
Variant initialValue)
{
BaseVariableState variable = builder
.AddVariable(name, parentId)
.Writable()
.Node;
// The sample organises its variables rather than componentising
// them, which is the shape the redundant client browses.
variable.ReferenceTypeId = ReferenceTypeIds.Organizes;
variable.DisplayName = new LocalizedText("en", name);
variable.Value = initialValue;
variable.StatusCode = StatusCodes.Good;
variable.Timestamp = DateTime.UtcNow;
return variable;
}
///
/// Starts the replica's simulation loop if it has not already been started.
///
internal void StartSimulation()
{
m_simulationTask ??= Task.Run(() => RunSimulationAsync(m_simulationCts.Token));
}
private async Task RunSimulationAsync(CancellationToken cancellationToken)
{
using var timer = new PeriodicTimer(TimeSpan.FromSeconds(1));
bool wasLeader = false;
DateTime lastHeartbeat = DateTime.MinValue;
while (await timer.WaitForNextTickAsync(cancellationToken).ConfigureAwait(false))
{
if (!m_leaderElection.IsLeader)
{
if (wasLeader)
{
// A failover moved the write role away from this replica.
m_logger.ReplicaBecameStandby(m_replicaInfo.NodeId);
}
wasLeader = false;
UpdateActiveReplica("standby");
continue;
}
if (!wasLeader)
{
// This replica has just become the active writer (initial
// election or a failover). Continue the Counter from the
// last value shared through the distributed store instead of
// restarting from the local count.
await SeedCounterFromCacheAsync(cancellationToken).ConfigureAwait(false);
if (m_logger.IsEnabled(LogLevel.Information))
{
m_logger.ReplicaBecameActiveWriter(
m_replicaInfo.NodeId,
Volatile.Read(ref m_counterValue));
}
wasLeader = true;
}
await UpdateCounterAsync(cancellationToken).ConfigureAwait(false);
UpdateActiveReplica(m_replicaInfo.NodeId);
// Periodic heartbeat so the log shows liveness, which replica is
// producing which Counter values, and (across replicas) the
// per-replica divergence a client observes on failover.
DateTime now = DateTime.UtcNow;
if (now - lastHeartbeat >= TimeSpan.FromSeconds(5))
{
lastHeartbeat = now;
if (m_logger.IsEnabled(LogLevel.Information))
{
m_logger.ReplicaActive(
m_replicaInfo.NodeId,
Volatile.Read(ref m_counterValue));
}
}
}
}
private async Task UpdateCounterAsync(CancellationToken cancellationToken)
{
BaseVariableState? counter = m_counter;
if (counter == null)
{
return;
}
int value = Interlocked.Increment(ref m_counterValue);
DateTimeUtc timestamp = DateTimeUtc.Now;
var sample = new DataValue(
Variant.From(value),
StatusCodes.Good,
timestamp,
timestamp);
lock (m_updateLock)
{
counter.Value = value;
counter.Timestamp = timestamp.ToDateTime();
counter.ClearChangeMasks(SystemContext, false);
}
// Share the new value through the distributed store so standby
// replicas can serve it and a promoted replica continues from it.
if (m_valueCache != null)
{
try
{
await m_valueCache
.CacheAsync(
counter.NodeId,
sample,
cancellationToken)
.ConfigureAwait(false);
}
catch (InvalidOperationException)
{
// The distributed store is not ready yet (the server is
// still starting); the value is shared on the next tick.
}
}
bool historyReady = m_historian == null ||
await ArchiveCounterAsync(
counter,
sample,
cancellationToken).ConfigureAwait(false);
BaseObjectState? historyEvents = m_historyEvents;
if (historyEvents != null && historyReady)
{
var e = new BaseEventState(historyEvents);
e.Initialize(
SystemContext,
historyEvents,
EventSeverity.Low,
new LocalizedText(
$"Replica '{m_replicaInfo.NodeId}' archived counter {value}."));
try
{
await historyEvents.ReportEventAsync(
SystemContext,
e,
cancellationToken).ConfigureAwait(false);
}
catch (ServiceResultException exception)
{
m_logger.EventHistoryWriteFailed(
exception,
m_replicaInfo.NodeId);
}
}
}
private async Task ArchiveCounterAsync(
BaseVariableState counter,
DataValue sample,
CancellationToken cancellationToken)
{
SharedKeyValueHistorianProvider? historian = m_historian;
if (historian == null || !m_leaderElection.IsLeader)
{
return false;
}
try
{
using var operationContext = new OperationContext(
new RequestHeader(),
null,
RequestType.HistoryUpdate,
RequestLifetime.None);
var historianContext = new HistorianOperationContext(
SystemContext,
operationContext,
counter,
HistoryUpdateType.Insert);
HistorianUpdateOutcome outcome = await historian.InsertAsync(
historianContext,
counter.NodeId,
[sample],
cancellationToken).ConfigureAwait(false);
StatusCode status = outcome.OperationResults.Count == 1
? outcome.OperationResults[0]
: StatusCodes.BadUnexpectedError;
if (StatusCode.IsBad(status))
{
m_logger.CounterHistoryWriteRejected(
m_replicaInfo.NodeId,
status);
return false;
}
return true;
}
catch (Exception exception) when (
exception is ServiceResultException or
TimeoutException or
InvalidOperationException)
{
m_logger.CounterHistoryWriteFailed(
exception,
m_replicaInfo.NodeId);
return false;
}
}
private async Task SeedCounterFromCacheAsync(CancellationToken cancellationToken)
{
BaseVariableState? counter = m_counter;
if (m_valueCache == null || counter == null)
{
return;
}
try
{
(bool _, DataValue cached) = await m_valueCache
.TryGetAsync(counter.NodeId, s_valueFreshness, cancellationToken)
.ConfigureAwait(false);
if (cached.WrappedValue.TryGetValue(out int last) && last > Volatile.Read(ref m_counterValue))
{
Interlocked.Exchange(ref m_counterValue, last);
lock (m_updateLock)
{
counter.Value = last;
counter.Timestamp = DateTime.UtcNow;
counter.ClearChangeMasks(SystemContext, false);
}
}
}
catch (InvalidOperationException)
{
// The distributed store is not ready yet; start from the local
// value and share on the next tick.
}
}
private DataValue ReadLocalCounter()
{
return new DataValue(
Variant.From(Volatile.Read(ref m_counterValue)),
StatusCodes.Good,
DateTimeUtc.Now);
}
private void UpdateActiveReplica(string value)
{
BaseVariableState? activeReplica = m_activeReplica;
if (activeReplica == null)
{
return;
}
lock (m_updateLock)
{
activeReplica.Value = value;
activeReplica.Timestamp = DateTime.UtcNow;
activeReplica.ClearChangeMasks(SystemContext, false);
}
}
}
///
/// Defines log messages for replica write roles, counter activity, and history capture.
///
internal static partial class HaSampleNodeManagerLog
{
///
/// Logs that a replica has relinquished the active writer role.
///
[LoggerMessage(EventId = RedundantServerEventIds.HaSampleNodeManager + 0, Level = LogLevel.Information,
Message = "HA: replica {ReplicaId} became STANDBY (no longer the active writer).")]
public static partial void ReplicaBecameStandby(this ILogger logger, string replicaId);
///
/// Logs that a replica has become the active writer and reports its counter value.
///
[LoggerMessage(EventId = RedundantServerEventIds.HaSampleNodeManager + 1, Level = LogLevel.Information,
Message = "HA: replica {ReplicaId} became ACTIVE writer (Counter={Counter}).")]
public static partial void ReplicaBecameActiveWriter(this ILogger logger, string replicaId, int counter);
///
/// Logs the active replica's heartbeat and current counter value.
///
[LoggerMessage(EventId = RedundantServerEventIds.HaSampleNodeManager + 2, Level = LogLevel.Information,
Message = "HA: replica {ReplicaId} ACTIVE, Counter={Counter}.")]
public static partial void ReplicaActive(this ILogger logger, string replicaId, int counter);
///
/// Logs a rejected counter history write and its status code.
///
[LoggerMessage(EventId = RedundantServerEventIds.HaSampleNodeManager + 3, Level = LogLevel.Warning,
Message = "HA: replica {ReplicaId} could not archive the Counter sample ({StatusCode}).")]
public static partial void CounterHistoryWriteRejected(
this ILogger logger,
string replicaId,
StatusCode statusCode);
///
/// Logs an exception while archiving a counter sample.
///
[LoggerMessage(EventId = RedundantServerEventIds.HaSampleNodeManager + 4, Level = LogLevel.Warning,
Message = "HA: replica {ReplicaId} failed to archive the Counter sample.")]
public static partial void CounterHistoryWriteFailed(
this ILogger logger,
Exception exception,
string replicaId);
///
/// Logs an exception while archiving a history event.
///
[LoggerMessage(EventId = RedundantServerEventIds.HaSampleNodeManager + 5, Level = LogLevel.Warning,
Message = "HA: replica {ReplicaId} failed to archive the history event.")]
public static partial void EventHistoryWriteFailed(
this ILogger logger,
Exception exception,
string replicaId);
}
}