[Server] Make subscription transfer between sessions transactional

src/Opc.Ua.Server/Hosting/OpcUaServerHostedService.cs

@@ -40,6 +40,7 @@
 using Opc.Ua.Configuration;
 using Opc.Ua.Identity;
 using Opc.Ua.Schema;
+using Opc.Ua.Security.Certificates;
 using Opc.Ua.Server.AliasNames;
 using Opc.Ua.Server.Historian;
 
@@ -427,7 +428,7 @@ await certificateManager.UpdateAsync(
                 configuration.SecurityConfiguration,
                 configuration.ApplicationUri,
                 ct).ConfigureAwait(false);
-            using var certificates =
+            using CertificateEntryCollection certificates =
                 certificateManager.SnapshotApplicationCertificates();
             return certificates.Count > 0;
         }
@@ -514,7 +515,6 @@ public override void Dispose()
             m_server?.Dispose();
             base.Dispose();
         }
-
     }
 
     /// <summary>
@@ -547,5 +547,4 @@ public static partial void UserTokenPolicyTokenTypeIsConfiguredWithout(
             Message = "Error while stopping OPC UA server.")]
         public static partial void ErrorWhileStoppingOPCUAServer(this ILogger logger, Exception ex);
     }
-
 }

src/Opc.Ua.Server/NodeManager/MasterNodeManager.cs

@@ -5522,6 +5522,7 @@ private async ValueTask<IMonitoredItemTransferTransaction>
 
             var processedItems = new List<bool>(monitoredItems.Count);
             var originalErrors = new ServiceResult[errors.Count];
+            var resendStates = new bool[monitoredItems.Count];
             var effectiveTransferOptions = new MonitoredItemTransferOptions
             {
                 DeferInitialValues = sendInitialValues ||
@@ -5533,6 +5534,7 @@ private async ValueTask<IMonitoredItemTransferTransaction>
             {
                 IMonitoredItem? monitoredItem = monitoredItems[ii];
                 originalErrors[ii] = errors[ii];
+                resendStates[ii] = monitoredItem?.IsResendData ?? false;
                 bool isDetached = monitoredItem is IDetachableMonitoredItem
                 {
                     IsDetached: true
@@ -5588,6 +5590,7 @@ await owner.TransferMonitoredItemsAsync(
                             monitoredItems,
                             errors,
                             originalErrors,
+                            resendStates,
                             participants,
                             effectiveTransferOptions);
                         try
@@ -5624,6 +5627,7 @@ await failedTransaction.RollbackAsync(CancellationToken.None)
                 monitoredItems,
                 errors,
                 originalErrors,
+                resendStates,
                 participants,
                 effectiveTransferOptions);
         }
@@ -5684,6 +5688,7 @@ public MonitoredItemTransferTransaction(
                 IList<IMonitoredItem> monitoredItems,
                 IList<ServiceResult> errors,
                 ServiceResult[] originalErrors,
+                bool[] resendStates,
                 List<MonitoredItemTransferParticipant> participants,
                 MonitoredItemTransferOptions transferOptions)
             {
@@ -5692,6 +5697,7 @@ public MonitoredItemTransferTransaction(
                 m_monitoredItems = monitoredItems;
                 m_errors = errors;
                 m_originalErrors = originalErrors;
+                m_resendStates = resendStates;
                 m_participants = participants;
                 m_transferOptions = transferOptions;
             }
@@ -5766,6 +5772,10 @@ await participant.NodeManager.RollbackMonitoredItemsTransferAsync(
 
                 for (int ii = 0; ii < m_monitoredItems.Count; ii++)
                 {
+                    if (m_monitoredItems[ii] is IMonitoredItemTransferState transferState)
+                    {
+                        transferState.RestoreResendDataTrigger(m_resendStates[ii]);
+                    }
                     m_errors[ii] = m_originalErrors[ii];
                 }
 
@@ -5782,6 +5792,7 @@ await participant.NodeManager.RollbackMonitoredItemsTransferAsync(
             private readonly IList<IMonitoredItem> m_monitoredItems;
             private readonly IList<ServiceResult> m_errors;
             private readonly ServiceResult[] m_originalErrors;
+            private readonly bool[] m_resendStates;
             private readonly List<MonitoredItemTransferParticipant> m_participants;
             private readonly MonitoredItemTransferOptions m_transferOptions;
             private int m_state;

src/Opc.Ua.Server/Server/ServerInternalData.cs

@@ -66,7 +66,7 @@ public class ServerInternalData :
         Historian.IHistorianRegistryProvider,
         ITransportListenerRegistryProvider,
         IServerEndpointRegistryProvider,
-
+        IAsyncDisposable,
         ITimeProviderProvider
     {
         /// <summary>
@@ -138,13 +138,22 @@ public void Dispose()
             GC.SuppressFinalize(this);
         }
 
+        /// <summary>
+        /// Frees managed resources that require asynchronous shutdown.
+        /// </summary>
+        public async ValueTask DisposeAsync()
+        {
+            await DisposeAsyncCore().ConfigureAwait(false);
+            GC.SuppressFinalize(this);
+        }
+
         /// <summary>
         /// An overrideable version of the Dispose.
         /// </summary>
         /// <param name="disposing"><c>true</c> to release both managed and unmanaged resources; <c>false</c> to release only unmanaged resources.</param>
         protected virtual void Dispose(bool disposing)
         {
-            if (disposing)
+            if (disposing && Interlocked.Exchange(ref m_disposed, 1) == 0)
             {
                 m_roleStateBinding?.Dispose();
                 m_roleStateBinding = null;
@@ -163,6 +172,39 @@ protected virtual void Dispose(bool disposing)
             }
         }
 
+        /// <summary>
+        /// An overrideable version of asynchronous dispose.
+        /// </summary>
+        protected virtual async ValueTask DisposeAsyncCore()
+        {
+            if (Interlocked.Exchange(ref m_disposed, 1) != 0)
+            {
+                return;
+            }
+
+            m_roleStateBinding?.Dispose();
+            m_roleStateBinding = null;
+            (RoleManager as IDisposable)?.Dispose();
+            ResourceManager?.Dispose();
+            RequestManager?.Dispose();
+            AggregateManager?.Dispose();
+            ModellingRulesManager?.Dispose();
+            ConformanceUnitsManager?.Dispose();
+            (NodeManager as IDisposable)?.Dispose();
+            SessionManager?.Dispose();
+            if (SubscriptionManager is IAsyncDisposable asyncSubscriptionManager)
+            {
+                await asyncSubscriptionManager.DisposeAsync().ConfigureAwait(false);
+            }
+            else
+            {
+                SubscriptionManager?.Dispose();
+            }
+            MonitoredItemQueueFactory?.Dispose();
+            (AliasNameStoreRegistry as IDisposable)?.Dispose();
+            (HistorianRegistry as IDisposable)?.Dispose();
+        }
+
         /// <summary>
         /// The server-wide registry of OPC UA Part 17 alias-name stores.
         /// Surfaces through the optional
@@ -702,7 +744,8 @@ public async ValueTask CloseSessionAsync(
             {
                 await NodeManager.SessionClosingAsync(context, sessionId, deleteSubscriptions, cancellationToken)
                     .ConfigureAwait(false);
-                await SubscriptionManager.SessionClosingAsync(context, sessionId, deleteSubscriptions, cancellationToken)
+                await SubscriptionManager
+                    .SessionClosingAsync(context, sessionId, deleteSubscriptions, cancellationToken)
                     .ConfigureAwait(false);
             }
             finally
@@ -1289,5 +1332,6 @@ private ServiceResult OnUpdateDiagnostics(
         private RoleStateBinding? m_roleStateBinding;
         private volatile IReadOnlyList<ITransportListener>? m_transportListeners;
         private ArrayOf<EndpointDescription> m_serverEndpoints;
+        private int m_disposed;
     }
 }

src/Opc.Ua.Server/Subscription/MonitoredItem/IMonitoredItem.cs

@@ -346,6 +346,18 @@ public interface ISampledDataChangeMonitoredItem : IDataChangeMonitoredItem2
         void SetSamplingInterval(double samplingInterval);
     }
 
+    /// <summary>
+    /// Restores monitored item transient state when a prepared subscription transfer is rolled back.
+    /// </summary>
+    internal interface IMonitoredItemTransferState
+    {
+        /// <summary>
+        /// Restores the resend-data trigger to the value captured before transfer preparation.
+        /// </summary>
+        /// <param name="resendData">The original resend-data trigger state.</param>
+        void RestoreResendDataTrigger(bool resendData);
+    }
+
     /// <summary>
     /// Defines constants for the monitored item type.
     /// </summary>

src/Opc.Ua.Server/Subscription/MonitoredItem/MonitoredItem.cs

@@ -42,7 +42,8 @@ public class MonitoredItem :
         IEventMonitoredItem,
         ISampledDataChangeMonitoredItem,
         ITriggeredMonitoredItem,
-        IDetachableMonitoredItem
+        IDetachableMonitoredItem,
+        IMonitoredItemTransferState
     {
         /// <summary>
         /// Initializes the object with its node type.
@@ -510,6 +511,14 @@ public void SetupResendDataTrigger()
             }
         }
 
+        void IMonitoredItemTransferState.RestoreResendDataTrigger(bool resendData)
+        {
+            lock (m_lock)
+            {
+                m_resendData = resendData;
+            }
+        }
+
         /// <summary>
         /// Sets a flag indicating that the item has been triggered and should publish.
         /// </summary>
@@ -1591,7 +1600,8 @@ public virtual bool Publish(
                 else
                 {
                     // pull any unprocessed data.
-                    if (m_calculator != null && m_calculator.HasEndTimePassed(DateTime.UtcNow))
+                    if (m_calculator != null &&
+                        m_calculator.HasEndTimePassed(DateTime.UtcNow))
                     {
                         while (m_calculator.TryGetProcessedValue(false, out DataValue processedValue))
                         {