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

Resolve blocking I/O finding and complete Historian lifecycle test coverage

Browse Source 
Move subscribe/unsubscribe I/O outside lock(Lock) in SyncAddressSpace to avoid
blocking all OPC UA operations during rebuilds. Replace blocking ReadAsync calls
for alarm priority/description in dispatch loop with cached subscription values.
Extract IHistorianConnectionFactory so EnsureConnected can be tested without the
SDK runtime — adds 5 connection lifecycle tests (failure, timeout, reconnect,
state resilience, dispose-after-failure). All stability review findings and test
coverage gaps are now fully resolved.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
This commit is contained in:
Joseph Doherty
2026-04-07 16:16:03 -04:00
parent 95ad9c6866
commit 9e1a180ce3
8 changed files with 346 additions and 108 deletions
Download Patch File Download Diff File Expand all files Collapse all files

77
src/ZB.MOM.WW.LmxOpcUa.Host/Historian/HistorianDataSource.cs
View File

@@ -21,6 +21,7 @@ namespace ZB.MOM.WW.LmxOpcUa.Host.Historian
private readonly HistorianConfiguration _config;
private readonly object _connectionLock = new object();
private readonly object _eventConnectionLock = new object();
private readonly IHistorianConnectionFactory _factory;
private HistorianAccess? _connection;
private HistorianAccess? _eventConnection;
private bool _disposed;
@@ -30,8 +31,15 @@ namespace ZB.MOM.WW.LmxOpcUa.Host.Historian
/// </summary>
/// <param name="config">The Historian SDK connection settings used for runtime history lookups.</param>
public HistorianDataSource(HistorianConfiguration config)
: this(config, new SdkHistorianConnectionFactory()) { }
/// <summary>
/// Initializes a Historian reader with a custom connection factory for testing.
/// </summary>
internal HistorianDataSource(HistorianConfiguration config, IHistorianConnectionFactory factory)
{
_config = config;
_factory = factory;
}
private void EnsureConnected()
@@ -39,33 +47,29 @@ namespace ZB.MOM.WW.LmxOpcUa.Host.Historian
if (_disposed)
throw new ObjectDisposedException(nameof(HistorianDataSource));
// Fast path: already connected (no lock needed)
if (Volatile.Read(ref _connection) != null)
return;
// Create and wait for connection outside the lock so concurrent history
// requests are not serialized behind a slow Historian handshake.
var conn = _factory.CreateAndConnect(_config, HistorianConnectionType.Process);
lock (_connectionLock)
{
if (_connection != null)
return;
var conn = new HistorianAccess();
var args = new HistorianConnectionArgs
if (_disposed)
{
ServerName = _config.ServerName,
TcpPort = (ushort)_config.Port,
IntegratedSecurity = _config.IntegratedSecurity,
ConnectionType = HistorianConnectionType.Process,
ReadOnly = true,
PacketTimeout = (uint)(_config.CommandTimeoutSeconds * 1000)
};
if (!_config.IntegratedSecurity)
{
args.UserName = _config.UserName ?? string.Empty;
args.Password = _config.Password ?? string.Empty;
conn.CloseConnection(out _);
conn.Dispose();
throw new ObjectDisposedException(nameof(HistorianDataSource));
}
if (!conn.OpenConnection(args, out var error))
if (_connection != null)
{
// Another thread connected while we were waiting
conn.CloseConnection(out _);
conn.Dispose();
throw new InvalidOperationException(
$"Failed to open Historian SDK connection to {_config.ServerName}:{_config.Port}: {error.ErrorCode}");
return;
}
_connection = conn;
@@ -100,33 +104,25 @@ namespace ZB.MOM.WW.LmxOpcUa.Host.Historian
if (_disposed)
throw new ObjectDisposedException(nameof(HistorianDataSource));
if (Volatile.Read(ref _eventConnection) != null)
return;
var conn = _factory.CreateAndConnect(_config, HistorianConnectionType.Event);
lock (_eventConnectionLock)
{
if (_eventConnection != null)
return;
var conn = new HistorianAccess();
var args = new HistorianConnectionArgs
if (_disposed)
{
ServerName = _config.ServerName,
TcpPort = (ushort)_config.Port,
IntegratedSecurity = _config.IntegratedSecurity,
ConnectionType = HistorianConnectionType.Event,
ReadOnly = true,
PacketTimeout = (uint)(_config.CommandTimeoutSeconds * 1000)
};
if (!_config.IntegratedSecurity)
{
args.UserName = _config.UserName ?? string.Empty;
args.Password = _config.Password ?? string.Empty;
conn.CloseConnection(out _);
conn.Dispose();
throw new ObjectDisposedException(nameof(HistorianDataSource));
}
if (!conn.OpenConnection(args, out var error))
if (_eventConnection != null)
{
conn.CloseConnection(out _);
conn.Dispose();
throw new InvalidOperationException(
$"Failed to open Historian SDK event connection to {_config.ServerName}:{_config.Port}: {error.ErrorCode}");
return;
}
_eventConnection = conn;
@@ -157,6 +153,7 @@ namespace ZB.MOM.WW.LmxOpcUa.Host.Historian
}
}
/// <summary>
/// Reads raw historical values for a tag from the Historian.
/// </summary>

81
src/ZB.MOM.WW.LmxOpcUa.Host/Historian/IHistorianConnectionFactory.cs Normal file
View File

@@ -0,0 +1,81 @@
using System;
using System.Threading;
using ArchestrA;
using ZB.MOM.WW.LmxOpcUa.Host.Configuration;
namespace ZB.MOM.WW.LmxOpcUa.Host.Historian
{
/// <summary>
/// Creates and opens Historian SDK connections. Extracted so tests can inject
/// fakes that control connection success, failure, and timeout behavior.
/// </summary>
internal interface IHistorianConnectionFactory
{
/// <summary>
/// Creates a new Historian SDK connection, opens it, and waits until it is ready.
/// Throws on connection failure or timeout.
/// </summary>
HistorianAccess CreateAndConnect(HistorianConfiguration config, HistorianConnectionType type);
}
/// <summary>
/// Production implementation that creates real Historian SDK connections.
/// </summary>
internal sealed class SdkHistorianConnectionFactory : IHistorianConnectionFactory
{
public HistorianAccess CreateAndConnect(HistorianConfiguration config, HistorianConnectionType type)
{
var conn = new HistorianAccess();
var args = new HistorianConnectionArgs
{
ServerName = config.ServerName,
TcpPort = (ushort)config.Port,
IntegratedSecurity = config.IntegratedSecurity,
UseArchestrAUser = config.IntegratedSecurity,
ConnectionType = type,
ReadOnly = true,
PacketTimeout = (uint)(config.CommandTimeoutSeconds * 1000)
};
if (!config.IntegratedSecurity)
{
args.UserName = config.UserName ?? string.Empty;
args.Password = config.Password ?? string.Empty;
}
if (!conn.OpenConnection(args, out var error))
{
conn.Dispose();
throw new InvalidOperationException(
$"Failed to open Historian SDK connection to {config.ServerName}:{config.Port}: {error.ErrorCode}");
}
// The SDK connects asynchronously — poll until the connection is ready
var timeoutMs = config.CommandTimeoutSeconds * 1000;
var elapsed = 0;
while (elapsed < timeoutMs)
{
var status = new HistorianConnectionStatus();
conn.GetConnectionStatus(ref status);
if (status.ConnectedToServer)
return conn;
if (status.ErrorOccurred)
{
conn.Dispose();
throw new InvalidOperationException(
$"Historian SDK connection failed: {status.Error}");
}
Thread.Sleep(250);
elapsed += 250;
}
conn.Dispose();
throw new TimeoutException(
$"Historian SDK connection to {config.ServerName}:{config.Port} timed out after {config.CommandTimeoutSeconds}s");
}
}
}

118
src/ZB.MOM.WW.LmxOpcUa.Host/OpcUa/LmxNodeManager.cs
View File

@@ -25,6 +25,9 @@ namespace ZB.MOM.WW.LmxOpcUa.Host.OpcUa
// Alarm tracking: maps InAlarm tag reference → alarm source info
private readonly Dictionary<string, AlarmInfo> _alarmInAlarmTags = new(StringComparer.OrdinalIgnoreCase);
// Reverse lookups: priority/description tag reference → alarm info for cache updates
private readonly Dictionary<string, AlarmInfo> _alarmPriorityTags = new(StringComparer.OrdinalIgnoreCase);
private readonly Dictionary<string, AlarmInfo> _alarmDescTags = new(StringComparer.OrdinalIgnoreCase);
private readonly bool _alarmTrackingEnabled;
private readonly bool _anonymousCanWrite;
private readonly AutoResetEvent _dataChangeSignal = new(false);
@@ -172,6 +175,8 @@ namespace ZB.MOM.WW.LmxOpcUa.Host.OpcUa
_tagMetadata.Clear();
_alarmInAlarmTags.Clear();
_alarmAckedTags.Clear();
_alarmPriorityTags.Clear();
_alarmDescTags.Clear();
_nodeMap.Clear();
_gobjectToTagRefs.Clear();
VariableNodeCount = 0;
@@ -357,6 +362,10 @@ namespace ZB.MOM.WW.LmxOpcUa.Host.OpcUa
};
_alarmInAlarmTags[inAlarmTagRef] = alarmInfo;
_alarmAckedTags[alarmInfo.AckedTagReference] = alarmInfo;
if (!string.IsNullOrEmpty(alarmInfo.PriorityTagReference))
_alarmPriorityTags[alarmInfo.PriorityTagReference] = alarmInfo;
if (!string.IsNullOrEmpty(alarmInfo.DescAttrNameTagReference))
_alarmDescTags[alarmInfo.DescAttrNameTagReference] = alarmInfo;
hasAlarms = true;
}
@@ -530,6 +539,9 @@ namespace ZB.MOM.WW.LmxOpcUa.Host.OpcUa
/// <param name="attributes">The latest Galaxy attribute snapshot to compare against the currently published variables.</param>
public void SyncAddressSpace(List<GalaxyObjectInfo> hierarchy, List<GalaxyAttributeInfo> attributes)
{
var tagsToUnsubscribe = new List<string>();
var tagsToResubscribe = new List<string>();
lock (Lock)
{
if (_lastHierarchy == null || _lastAttributes == null)
@@ -565,29 +577,22 @@ namespace ZB.MOM.WW.LmxOpcUa.Host.OpcUa
if (_subscriptionRefCounts.TryGetValue(tagRef, out var count))
affectedSubscriptions[tagRef] = count;
// Tear down changed subtrees
TearDownGobjects(changedIds);
// Tear down changed subtrees (collects tags for deferred unsubscription)
TearDownGobjects(changedIds, tagsToUnsubscribe);
// Rebuild changed subtrees from new data
var changedHierarchy = hierarchy.Where(h => changedIds.Contains(h.GobjectId)).ToList();
var changedAttributes = attributes.Where(a => changedIds.Contains(a.GobjectId)).ToList();
BuildSubtree(changedHierarchy, changedAttributes);
// Restore subscriptions for surviving tags
// Restore subscription bookkeeping for surviving tags
foreach (var kvp in affectedSubscriptions)
{
if (!_tagToVariableNode.ContainsKey(kvp.Key))
continue;
try
{
_mxAccessClient.SubscribeAsync(kvp.Key, (_, _) => { }).GetAwaiter().GetResult();
_subscriptionRefCounts[kvp.Key] = kvp.Value;
}
catch (Exception ex)
{
Log.Warning(ex, "Failed to restore subscription for {TagRef} after sync", kvp.Key);
}
_subscriptionRefCounts[kvp.Key] = kvp.Value;
tagsToResubscribe.Add(kvp.Key);
}
_lastHierarchy = new List<GalaxyObjectInfo>(hierarchy);
@@ -596,9 +601,18 @@ namespace ZB.MOM.WW.LmxOpcUa.Host.OpcUa
Log.Information("Incremental sync complete: {Objects} objects, {Variables} variables, {Alarms} alarms",
ObjectNodeCount, VariableNodeCount, _alarmInAlarmTags.Count);
}
// Perform subscribe/unsubscribe I/O outside Lock so read/write/browse operations are not blocked
foreach (var tag in tagsToUnsubscribe)
try { _mxAccessClient.UnsubscribeAsync(tag).GetAwaiter().GetResult(); }
catch (Exception ex) { Log.Warning(ex, "Failed to unsubscribe {Tag} after sync", tag); }
foreach (var tag in tagsToResubscribe)
try { _mxAccessClient.SubscribeAsync(tag, (_, _) => { }).GetAwaiter().GetResult(); }
catch (Exception ex) { Log.Warning(ex, "Failed to restore subscription for {Tag} after sync", tag); }
}
private void TearDownGobjects(HashSet<int> gobjectIds)
private void TearDownGobjects(HashSet<int> gobjectIds, List<string> tagsToUnsubscribe)
{
foreach (var id in gobjectIds)
{
@@ -607,18 +621,10 @@ namespace ZB.MOM.WW.LmxOpcUa.Host.OpcUa
{
foreach (var tagRef in tagRefs.ToList())
{
// Unsubscribe if actively subscribed
// Defer unsubscribe to outside lock
if (_subscriptionRefCounts.ContainsKey(tagRef))
{
try
{
_mxAccessClient.UnsubscribeAsync(tagRef).GetAwaiter().GetResult();
}
catch
{
/* ignore */
}
tagsToUnsubscribe.Add(tagRef);
_subscriptionRefCounts.Remove(tagRef);
}
@@ -630,20 +636,17 @@ namespace ZB.MOM.WW.LmxOpcUa.Host.OpcUa
foreach (var alarmKey in alarmKeysToRemove)
{
var info = _alarmInAlarmTags[alarmKey];
// Unsubscribe alarm auto-subscriptions
// Defer alarm tag unsubscription to outside lock
foreach (var alarmTag in new[]
{ alarmKey, info.PriorityTagReference, info.DescAttrNameTagReference })
if (!string.IsNullOrEmpty(alarmTag))
try
{
_mxAccessClient.UnsubscribeAsync(alarmTag).GetAwaiter().GetResult();
}
catch
{
/* ignore */
}
tagsToUnsubscribe.Add(alarmTag);
_alarmInAlarmTags.Remove(alarmKey);
if (!string.IsNullOrEmpty(info.PriorityTagReference))
_alarmPriorityTags.Remove(info.PriorityTagReference);
if (!string.IsNullOrEmpty(info.DescAttrNameTagReference))
_alarmDescTags.Remove(info.DescAttrNameTagReference);
if (!string.IsNullOrEmpty(info.AckedTagReference))
_alarmAckedTags.Remove(info.AckedTagReference);
}
@@ -871,6 +874,10 @@ namespace ZB.MOM.WW.LmxOpcUa.Host.OpcUa
};
_alarmInAlarmTags[inAlarmTagRef] = alarmInfo;
_alarmAckedTags[alarmInfo.AckedTagReference] = alarmInfo;
if (!string.IsNullOrEmpty(alarmInfo.PriorityTagReference))
_alarmPriorityTags[alarmInfo.PriorityTagReference] = alarmInfo;
if (!string.IsNullOrEmpty(alarmInfo.DescAttrNameTagReference))
_alarmDescTags[alarmInfo.DescAttrNameTagReference] = alarmInfo;
hasAlarms = true;
}
@@ -2075,6 +2082,23 @@ namespace ZB.MOM.WW.LmxOpcUa.Host.OpcUa
alarmInfo = null;
}
// Cache alarm priority/description values as they arrive via subscription
if (_alarmPriorityTags.TryGetValue(address, out var priorityInfo))
{
if (vtq.Value is int ipCache)
priorityInfo.CachedSeverity =
(ushort)Math.Min(Math.Max(ipCache, 1), 1000);
else if (vtq.Value is short spCache)
priorityInfo.CachedSeverity =
(ushort)Math.Min(Math.Max((int)spCache, 1), 1000);
}
if (_alarmDescTags.TryGetValue(address, out var descInfo))
{
if (vtq.Value is string descCache && !string.IsNullOrEmpty(descCache))
descInfo.CachedMessage = descCache;
}
// Check for Acked transitions — skip if state hasn't changed
if (_alarmAckedTags.TryGetValue(address, out ackedAlarmInfo))
{
@@ -2095,31 +2119,11 @@ namespace ZB.MOM.WW.LmxOpcUa.Host.OpcUa
if (newInAlarm)
{
try
{
var pVtq = _mxAccessClient.ReadAsync(alarmInfo.PriorityTagReference).GetAwaiter()
.GetResult();
if (pVtq.Value is int ip)
severity = (ushort)Math.Min(Math.Max(ip, 1), 1000);
else if (pVtq.Value is short sp)
severity = (ushort)Math.Min(Math.Max((int)sp, 1), 1000);
}
catch
{
// Keep the previously cached severity when refresh reads fail.
}
try
{
var dVtq = _mxAccessClient.ReadAsync(alarmInfo.DescAttrNameTagReference).GetAwaiter()
.GetResult();
if (dVtq.Value is string desc && !string.IsNullOrEmpty(desc))
message = desc;
}
catch
{
// Keep the previously cached message when refresh reads fail.
}
// Use cached values from subscription data changes instead of blocking reads
severity = alarmInfo.CachedSeverity > 0 ? alarmInfo.CachedSeverity : (ushort?)null;
message = !string.IsNullOrEmpty(alarmInfo.CachedMessage)
? alarmInfo.CachedMessage
: null;
}
pendingAlarmEvents.Add((address, alarmInfo, newInAlarm, severity, message));

4
src/ZB.MOM.WW.LmxOpcUa.Host/ZB.MOM.WW.LmxOpcUa.Host.csproj
View File

@@ -53,6 +53,10 @@
<HintPath>..\..\lib\aahClientManaged.dll</HintPath>
<EmbedInteropTypes>false</EmbedInteropTypes>
</Reference>
<Reference Include="aahClientCommon">
<HintPath>..\..\lib\aahClientCommon.dll</HintPath>
<EmbedInteropTypes>false</EmbedInteropTypes>
</Reference>
</ItemGroup>
<ItemGroup>