Skip to content

Harden OPC UA layer: reconnect races, handle allocation, interval/security wiring, status semantics #171

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
BrettKinny merged 1 commit into from
Jun 11, 2026
Merged

Harden OPC UA layer: reconnect races, handle allocation, interval/security wiring, status semantics #171

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
10 changes: 9 additions & 1 deletion App/MainWindow.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1251,7 +1251,15 @@ private async Task LoadConfigurationAsync(string filePath)
// old server while (or after) the new connection is attempted.
await DisconnectAsync();

var connected = await _connectionManager.ConnectAsync(config.Server.EndpointUrl, config.Settings.PublishingIntervalMs, credentials);
// Honor the config's security and sampling settings (the connect dialog has
// no UI for these, so the config file is their only source).
var connected = await _connectionManager.ConnectAsync(
config.Server.EndpointUrl,
config.Settings.PublishingIntervalMs,
credentials,
config.Server.SecurityMode,
config.Server.SecurityPolicy,
config.Settings.SamplingIntervalMs);

if (connected)
{
Expand Down
31 changes: 26 additions & 5 deletions OpcUa/ConnectionManager.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,6 +17,11 @@ public sealed class ConnectionManager : IDisposable
private bool _disposed;
private int _isReconnecting;

// Intervals from the most recent ConnectAsync, so subscription restoration
// after a reconnect does not silently fall back to the defaults.
private int _publishingInterval = 250;
private int _samplingInterval = 250;

// Stored event handler references for proper unsubscription
private Action<Models.MonitoredNode>? _valueChangedHandler;
private Action<Models.MonitoredNode>? _variableAddedHandler;
Expand Down Expand Up @@ -115,13 +120,15 @@ public ConnectionManager(Logger logger, bool? allowInsecure = null)
/// <param name="credentials">Authentication credentials (defaults to anonymous).</param>
/// <param name="securityMode">Requested message security mode (e.g. None, Sign, SignAndEncrypt). When null/None, an unsecured endpoint is selected.</param>
/// <param name="securityPolicy">Requested security policy URI or shorthand (e.g. Basic256Sha256). Honored when a matching endpoint exists.</param>
/// <param name="samplingInterval">Sampling interval in milliseconds for monitored items.</param>
/// <returns>True if connection succeeded, false otherwise.</returns>
public async Task<bool> ConnectAsync(
string endpoint,
int publishingInterval = 250,
ConnectionCredentials? credentials = null,
string? securityMode = null,
string? securityPolicy = null)
string? securityPolicy = null,
int samplingInterval = 250)
{
// Async teardown: the synchronous Disconnect() blocks on the OPC UA close
// round-trip (up to the transport timeout against a dead server), which froze
Expand All @@ -130,6 +137,8 @@ public async Task<bool> ConnectAsync(

_lastEndpoint = endpoint;
_credentials = credentials ?? ConnectionCredentials.Anonymous;
_publishingInterval = publishingInterval;
_samplingInterval = samplingInterval;
StateChanged?.Invoke(ConnectionState.Connecting);

try
Expand All @@ -138,7 +147,7 @@ public async Task<bool> ConnectAsync(

if (success)
{
if (!await InitializeSubscriptionAsync(publishingInterval))
if (!await InitializeSubscriptionAsync())
{
// Without a subscription the session is useless for monitoring;
// fail the connect rather than reporting Connected.
Expand Down Expand Up @@ -271,9 +280,20 @@ private async Task RestoreSubscriptionsAsync()
if (!recreated)
{
_logger.Warning("Failed to recreate subscriptions - initializing fresh");
// Last resort: start fresh (will lose monitored nodes)
// Last resort: start fresh. The monitored nodes are gone, so tell the UI -
// otherwise their rows sit at "(reconnecting...)" forever with handles the
// new subscription manager knows nothing about.
var lostHandles = _subscriptionManager.MonitoredVariables
.Select(v => v.ClientHandle)
.ToList();

DisposeSubscription();
await InitializeSubscriptionAsync();

foreach (var handle in lostHandles)
{
VariableRemoved?.Invoke(handle);
}
}
}

Expand Down Expand Up @@ -313,10 +333,11 @@ public Task<bool> UnsubscribeAsync(uint clientHandle)
return _client.WriteValueAsync(nodeId, value);
}

private async Task<bool> InitializeSubscriptionAsync(int publishingInterval = 250)
private async Task<bool> InitializeSubscriptionAsync()
{
_subscriptionManager = new SubscriptionManager(_client, _logger);
_subscriptionManager.PublishingInterval = publishingInterval;
_subscriptionManager.PublishingInterval = _publishingInterval;
_subscriptionManager.SamplingInterval = _samplingInterval;
var initialized = await _subscriptionManager.InitializeAsync();

// Store handler references for proper unsubscription
Expand Down
13 changes: 8 additions & 5 deletions OpcUa/Models/MonitoredNode.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -24,8 +24,11 @@ public class MonitoredNode

public DateTime? Timestamp { get; set; }
public uint StatusCode { get; set; }
public bool IsGood => StatusCode == 0; // StatusCode.Good = 0
public bool IsUncertain => (StatusCode & 0x40000000) != 0;
// OPC UA status severity lives in the top two bits: 00 = Good, 01 = Uncertain,
// 10 = Bad. Good codes with info bits set (e.g. GoodClamped 0x00300000) are
// still Good, so testing for == 0 would misclassify them.
public bool IsGood => (StatusCode & 0xC0000000) == 0;
public bool IsUncertain => (StatusCode & 0xC0000000) == 0x40000000;
public bool IsBad => (StatusCode & 0x80000000) != 0;
public DateTime LastChangeTime { get; set; } = DateTime.MinValue;
public bool RecentlyChanged => (DateTime.Now - LastChangeTime).TotalMilliseconds < 500;
Expand Down Expand Up @@ -72,9 +75,9 @@ public string StatusString
get
{
if (StatusCode == 0) return "Good";
if ((StatusCode & 0x80000000) != 0) return $"Bad (0x{StatusCode:X8})";
if ((StatusCode & 0x40000000) != 0) return $"Uncertain (0x{StatusCode:X8})";
return $"0x{StatusCode:X8}";
if (IsBad) return $"Bad (0x{StatusCode:X8})";
if (IsUncertain) return $"Uncertain (0x{StatusCode:X8})";
return $"Good (0x{StatusCode:X8})";
}
}

Expand Down
65 changes: 50 additions & 15 deletions OpcUa/OpcUaClientWrapper.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -15,6 +15,7 @@ public class OpcUaClientWrapper : IDisposable
private readonly Logger _logger;
private string? _currentEndpoint;
private CancellationTokenSource? _reconnectCts;
private readonly object _reconnectCtsLock = new();
private bool _disposed;
private ApplicationConfiguration? _appConfig;
private ConfiguredEndpoint? _lastConfiguredEndpoint;
Expand Down Expand Up @@ -287,15 +288,23 @@ public void Disconnect()

/// <summary>
/// Cancels and disposes the reconnect cancellation token source, if any.
/// Safe to race with <see cref="ReconnectAsync"/>: the field swap happens
/// under a lock, so two callers can never dispose the same instance twice.
/// </summary>
private void DisposeReconnectCts()
{
if (_reconnectCts != null)
CancellationTokenSource? cts;
lock (_reconnectCtsLock)
{
try { _reconnectCts.Cancel(); } catch (ObjectDisposedException) { }
_reconnectCts.Dispose();
cts = _reconnectCts;
_reconnectCts = null;
}

if (cts != null)
{
try { cts.Cancel(); } catch (ObjectDisposedException) { }
cts.Dispose();
}
}

/// <summary>
Expand All @@ -313,14 +322,23 @@ public async Task<bool> ReconnectAsync()
}

DisposeReconnectCts();
_reconnectCts = new CancellationTokenSource();

// Work with a local reference throughout: a concurrent Disconnect() can null
// and dispose the field at any time, so re-reading it mid-loop would NRE or
// touch a disposed CTS.
var cts = new CancellationTokenSource();
lock (_reconnectCtsLock)
{
_reconnectCts = cts;
}
var token = cts.Token;

// Exponential backoff: 1s, 2s, 4s, 8s
int[] delays = { 1000, 2000, 4000, 8000 };

for (int attempt = 0; attempt < delays.Length; attempt++)
{
if (_reconnectCts.Token.IsCancellationRequested)
if (token.IsCancellationRequested)
return false;

_logger.Info($"Reconnection attempt {attempt + 1}/{delays.Length}...");
Expand All @@ -330,8 +348,8 @@ public async Task<bool> ReconnectAsync()
// Strategy 1: Try to reconnect the existing session (preserves subscriptions automatically)
if (_session != null)
{
var reconnectResult = await TrySessionReconnectAsync();
if (reconnectResult)
var reconnectResult = await TrySessionReconnectAsync(token);
if (reconnectResult && !token.IsCancellationRequested)
{
_logger.Info("Session reconnected successfully (subscriptions preserved)");
Connected?.Invoke();
Expand All @@ -340,9 +358,18 @@ public async Task<bool> ReconnectAsync()
}

// Strategy 2: Recreate session and transfer subscriptions
var recreateResult = await TryRecreateSessionAsync();
var recreateResult = await TryRecreateSessionAsync(token);
if (recreateResult)
{
if (token.IsCancellationRequested)
{
// The user disconnected while the session was being recreated;
// don't resurrect a connection they asked to close.
_logger.Info("Reconnect cancelled after session recreation - closing the new session");
await DisconnectAsync();
return false;
}

_logger.Info("Session recreated successfully (subscriptions transferred)");
Connected?.Invoke();
return true;
Expand All @@ -356,7 +383,7 @@ public async Task<bool> ReconnectAsync()
// Wait before next attempt
try
{
await Task.Delay(delays[attempt], _reconnectCts.Token);
await Task.Delay(delays[attempt], token);
}
catch (OperationCanceledException)
{
Expand All @@ -373,15 +400,15 @@ public async Task<bool> ReconnectAsync()
/// Tries to reconnect the existing session using OPC UA Reconnect service.
/// This preserves the session ID and all subscriptions automatically.
/// </summary>
private async Task<bool> TrySessionReconnectAsync()
private async Task<bool> TrySessionReconnectAsync(CancellationToken cancellationToken)
{
if (_session == null)
return false;

try
{
_logger.Info("Attempting session reconnect...");
await _session.ReconnectAsync(_reconnectCts?.Token ?? CancellationToken.None);
await _session.ReconnectAsync(cancellationToken);
return _session.Connected;
}
catch (ServiceResultException ex)
Expand All @@ -395,7 +422,7 @@ private async Task<bool> TrySessionReconnectAsync()
/// Recreates the session and transfers existing subscriptions to it.
/// Used when direct reconnect fails (e.g., session timed out on server).
/// </summary>
private async Task<bool> TryRecreateSessionAsync()
private async Task<bool> TryRecreateSessionAsync(CancellationToken cancellationToken)
{
if (_lastConfiguredEndpoint == null || string.IsNullOrEmpty(_currentEndpoint))
return false;
Expand Down Expand Up @@ -450,7 +477,7 @@ private async Task<bool> TryRecreateSessionAsync()
{
if (subscriptionsToTransfer != null && subscriptionsToTransfer.Count > 0)
{
var transferred = await TransferSubscriptionsAsync(subscriptionsToTransfer);
var transferred = await TransferSubscriptionsAsync(subscriptionsToTransfer, cancellationToken);
_logger.Info($"Transferred {transferred} of {subscriptionsToTransfer.Count} subscription(s)");
}
}
Expand Down Expand Up @@ -487,7 +514,7 @@ private async Task<bool> TryRecreateSessionAsync()
/// <summary>
/// Transfers subscriptions from a previous session to the current session.
/// </summary>
private async Task<int> TransferSubscriptionsAsync(SubscriptionCollection subscriptions)
private async Task<int> TransferSubscriptionsAsync(SubscriptionCollection subscriptions, CancellationToken cancellationToken)
{
if (_session == null || subscriptions == null)
return 0;
Expand All @@ -500,7 +527,7 @@ private async Task<int> TransferSubscriptionsAsync(SubscriptionCollection subscr
var success = await _session.TransferSubscriptionsAsync(
subscriptions,
sendInitialValues: true,
_reconnectCts?.Token ?? CancellationToken.None);
cancellationToken);

if (success)
{
Expand Down Expand Up @@ -743,6 +770,14 @@ private async Task<EndpointDescription> DiscoverAndSelectEndpointAsync(

_logger.Info($"Selected endpoint: {selectedEndpoint.SecurityMode} / {selectedEndpoint.SecurityPolicyUri}");

if (_credentials.Type != AuthenticationType.Anonymous
&& selectedEndpoint.SecurityMode == MessageSecurityMode.None)
{
_logger.Warning(
"Credentials will be sent over an UNENCRYPTED channel: the selected endpoint uses SecurityMode=None. " +
"Anyone on the network can read the username and password. Prefer a server endpoint with Sign or SignAndEncrypt.");
}

// Update the endpoint URL to use the requested host if different
// (handles cases where server returns localhost but we connected via IP/hostname)
var selectedUri = new Uri(selectedEndpoint.EndpointUrl);
Expand Down
21 changes: 18 additions & 3 deletions OpcUa/SubscriptionManager.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -21,6 +21,7 @@ public class SubscriptionManager : IDisposable, IAsyncDisposable
private readonly Dictionary<uint, uint> _opcHandleToClientHandle = new();
private uint _nextClientHandle = 1;
private int _publishingInterval = 250;
private int _samplingInterval = 250;
private bool _isInitialized;
private readonly object _lock = new();

Expand All @@ -45,6 +46,16 @@ public int PublishingInterval
set => _publishingInterval = Math.Max(100, Math.Min(10000, value));
}

/// <summary>
/// Sampling interval in milliseconds applied to monitored items.
/// 0 requests the server's fastest practical rate.
/// </summary>
public int SamplingInterval
{
get => _samplingInterval;
set => _samplingInterval = Math.Max(0, Math.Min(60000, value));
}

public IReadOnlyCollection<MonitoredNode> MonitoredVariables
{
get
Expand Down Expand Up @@ -107,6 +118,7 @@ public async Task<bool> InitializeAsync()
return null;
}

uint clientHandle;
lock (_lock)
{
// Check if already monitoring this node
Expand All @@ -115,19 +127,22 @@ public async Task<bool> InitializeAsync()
_logger.Warning($"Node {displayName} is already being monitored");
return null;
}

// Allocate the handle under the lock: it keys three dictionaries, and an
// unsynchronized increment lets two concurrent adds collide on one handle.
clientHandle = _nextClientHandle++;
}

try
{
var clientHandle = _nextClientHandle++;

// Create OPC UA monitored item
var monitoredItem = new MonitoredItem(_subscription.DefaultItem)
{
DisplayName = displayName,
StartNodeId = nodeId,
AttributeId = Attributes.Value,
SamplingInterval = 250,
SamplingInterval = _samplingInterval,
QueueSize = 10,
DiscardOldest = true
};
Expand Down Expand Up @@ -589,7 +604,7 @@ public async Task<bool> RecreateSubscriptionsAsync()
DisplayName = displayName,
StartNodeId = nodeId,
AttributeId = Attributes.Value,
SamplingInterval = 250,
SamplingInterval = _samplingInterval,
QueueSize = 10,
DiscardOldest = true
};
Expand Down
Loading