UA-.NETStandard/tests/Opc.Ua.Server.Tests/SubscriptionTests.cs

/* ========================================================================
 * Copyright (c) 2005-2026 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.Concurrent;
using System.Collections.Generic;
using System.Reflection;
using System.Threading;
using System.Threading.Tasks;
using Microsoft.Extensions.Time.Testing;
using Moq;
using NUnit.Framework;
using Opc.Ua.Security.Certificates;
using Opc.Ua.Tests;

namespace Opc.Ua.Server.Tests
{
    [TestFixture]
    [Category("Subscription")]
    [Parallelizable]
    public class SubscriptionTests
    {
        private Mock<IServerInternal> m_serverMock;
        private Mock<ISession> m_sessionMock;
        private Mock<IDiagnosticsNodeManager> m_diagnosticsNodeManagerMock;
        private Mock<IMasterNodeManager> m_nodeManagerMock;
        private Mock<IMonitoredItemQueueFactory> m_queueFactoryMock;
        private ITelemetryContext m_telemetry;

        [SetUp]
        public void SetUp()
        {
            m_telemetry = NUnitTelemetryContext.Create();
            m_serverMock = new Mock<IServerInternal>();
            m_sessionMock = new Mock<ISession>();
            m_diagnosticsNodeManagerMock = new Mock<IDiagnosticsNodeManager>();
            m_nodeManagerMock = new Mock<IMasterNodeManager>();
            m_queueFactoryMock = new Mock<IMonitoredItemQueueFactory>();

            m_serverMock.Setup(s => s.Telemetry).Returns(m_telemetry);
            m_serverMock.Setup(s => s.DiagnosticsNodeManager).Returns(m_diagnosticsNodeManagerMock.Object);
            m_serverMock.Setup(s => s.NodeManager).Returns(m_nodeManagerMock.Object);
            m_serverMock.Setup(s => s.MonitoredItemQueueFactory).Returns(m_queueFactoryMock.Object);
            var serverDiagnostics = new ServerDiagnosticsSummaryDataType();
            m_serverMock
                .Setup(s => s.UpdateServerDiagnostics(
                    It.IsAny<Action<ServerDiagnosticsSummaryDataType>>()))
                .Callback<Action<ServerDiagnosticsSummaryDataType>>(
                    update => update(serverDiagnostics));

            var namespaceUris = new NamespaceTable();
            m_serverMock.Setup(s => s.NamespaceUris).Returns(namespaceUris);
            m_serverMock.Setup(s => s.ServerUris).Returns(new StringTable());
            m_serverMock.Setup(s => s.TypeTree).Returns(new TypeTable(namespaceUris));
            m_serverMock.Setup(s => s.Factory).Returns(new Mock<IEncodeableFactory>().Object);

            // ServerSystemContext requires invoked server mock to have properties setup
            m_serverMock.Setup(s => s.DefaultSystemContext).Returns(new ServerSystemContext(m_serverMock.Object));

            var identity = new UserIdentity(new AnonymousIdentityToken());
            m_sessionMock.Setup(s => s.Id).Returns(new NodeId(Guid.NewGuid()));
            m_sessionMock.Setup(s => s.Identity).Returns(identity);
            m_sessionMock.Setup(s => s.IdentityToken).Returns(identity.TokenHandler);
            var sessionDiagnostics = new SessionDiagnosticsDataType
            {
                ClientDescription = new ApplicationDescription
                {
                    ApplicationUri = "urn:localhost:opcfoundation.org:SubscriptionTests"
                }
            };
            m_sessionMock
                .Setup(s => s.UpdateDiagnostics(
                    It.IsAny<Action<SessionDiagnosticsDataType>>()))
                .Callback<Action<SessionDiagnosticsDataType>>(
                    update => update(sessionDiagnostics));
            m_sessionMock.Setup(s => s.ClientApplicationUri)
                .Returns(() => sessionDiagnostics.ClientDescription?.ApplicationUri);

            m_diagnosticsNodeManagerMock
                .Setup(d => d.CreateSubscriptionDiagnosticsAsync(
                    It.IsAny<ServerSystemContext>(),
                    It.IsAny<SubscriptionDiagnosticsDataType>(),
                    It.IsAny<NodeValueSimpleEventHandler>()))
                .ReturnsAsync(new NodeId(1));
        }

        private Subscription CreateSubscription(
            double publishingInterval = 1000,
            uint maxNotificationsPerPublish = 0,
            TimeProvider? timeProvider = null)
        {
            return new Subscription(
                m_serverMock.Object,
                m_sessionMock.Object,
                subscriptionId: 1,
                publishingInterval: publishingInterval,
                maxLifetimeCount: 10,
                maxKeepAliveCount: 5,
                maxNotificationsPerPublish: maxNotificationsPerPublish,
                priority: 0,
                publishingEnabled: true,
                maxMessageCount: 10,
                timeProvider: timeProvider);
        }

        [Test]
        [Category("NodeManagerLifecycle")]
        public void HasMonitoredItemsThrowsForNullNodeManager()
        {
            using Subscription subscription = CreateSubscription();

            ArgumentNullException exception = Assert.Throws<ArgumentNullException>(
                () => subscription.HasMonitoredItems(null!));

            Assert.That(exception.ParamName, Is.EqualTo("nodeManager"));
        }

        [Test]
        [Category("NodeManagerLifecycle")]
        public void CreateMonitoredItemsDoesNotRejectClosingSession()
        {
            using ServerInternalData server = CreateServerInternalData();
            m_sessionMock.SetupGet(session => session.IsClosing).Returns(true);
            using var subscription = new Subscription(
                server,
                m_sessionMock.Object,
                subscriptionId: 1,
                publishingInterval: 1000,
                maxLifetimeCount: 10,
                maxKeepAliveCount: 5,
                maxNotificationsPerPublish: 0,
                priority: 0,
                publishingEnabled: true,
                maxMessageCount: 10);
            var context = new OperationContext(
                m_sessionMock.Object,
                DiagnosticsMasks.None);

            // The subscription no longer gates monitored item operations on the
            // owning session's closing state; ownership and deletion are the only
            // guards. A closing session that still owns the subscription is allowed.
            Assert.DoesNotThrowAsync(
                async () => await subscription
                    .CreateMonitoredItemsAsync(
                        context,
                        TimestampsToReturn.Both,
                        [],
                        CancellationToken.None)
                    .ConfigureAwait(false));
        }

        /// <summary>
        /// Verifies that creation cleans up unattached items when deletion, transfer, closure, or cancellation takes
        /// ownership.
        /// </summary>
        [TestCase("delete")]
        [TestCase("transfer")]
        [TestCase("close")]
        [TestCase("cancel")]
        public async Task UnattachedMonitoredItemsAreDeletedWhenCreationLosesOwnershipAsync(string outcome)
        {
            using Subscription subscription = CreateSubscription();
            using var cancellation = new CancellationTokenSource();
            using var context = new OperationContext(m_sessionMock.Object, DiagnosticsMasks.None);
            var item = new Mock<IMonitoredItem>();
            item.SetupGet(i => i.MonitoringMode).Returns(MonitoringMode.Reporting);
            item.SetupGet(value => value.Id).Returns(77);
            var created = new MonitoredItemCreateResult { MonitoredItemId = 77, RevisedSamplingInterval = 1000 };
            item.Setup(value => value.GetCreateResult(out created)).Returns(ServiceResult.Good);
            var entered = new TaskCompletionSource<bool>(TaskCreationOptions.RunContinuationsAsynchronously);
            var release = new TaskCompletionSource<bool>(TaskCreationOptions.RunContinuationsAsynchronously);
            bool registered = false;
            m_nodeManagerMock.Setup(manager => manager.CreateMonitoredItemsAsync(
                    It.IsAny<OperationContext>(), It.IsAny<uint>(), It.IsAny<double>(),
                    It.IsAny<TimestampsToReturn>(), It.IsAny<ArrayOf<MonitoredItemCreateRequest>>(),
                    It.IsAny<IList<ServiceResult>>(), It.IsAny<IList<MonitoringFilterResult>>(),
                    It.IsAny<IList<IMonitoredItem>>(), It.IsAny<bool>(), It.IsAny<CancellationToken>()))
                .Returns(async (
                    OperationContext _, uint _, double _, TimestampsToReturn _,
                    ArrayOf<MonitoredItemCreateRequest> _, IList<ServiceResult> errors,
                    IList<MonitoringFilterResult> _, IList<IMonitoredItem> items, bool _, CancellationToken ct) =>
                {
                    items[0] = item.Object;
                    errors[0] = ServiceResult.Good;
                    registered = true;
                    entered.TrySetResult(true);
                    await release.Task.ConfigureAwait(false);
                    ct.ThrowIfCancellationRequested();
                });
            m_nodeManagerMock.Setup(manager => manager.DeleteMonitoredItemsAsync(
                    It.IsAny<OperationContext>(), 1, It.IsAny<IList<IMonitoredItem>>(),
                    It.IsAny<IList<ServiceResult>>(), It.IsAny<CancellationToken>()))
                .Returns((
                    OperationContext _, uint _, IList<IMonitoredItem> items, IList<ServiceResult> errors,
                    CancellationToken ct) =>
                {
                    ct.ThrowIfCancellationRequested();
                    Assert.That(items, Has.Count.EqualTo(1));
                    Assert.That(items[0], Is.SameAs(item.Object));
                    registered = false;
                    errors[0] = ServiceResult.Good;
                    return default;
                });
            Task<CreateMonitoredItemsResponse> creating = subscription.CreateMonitoredItemsAsync(
                context, TimestampsToReturn.Both, [new MonitoredItemCreateRequest()], cancellation.Token).AsTask();
            try
            {
                await entered.Task.WaitAsync(TimeSpan.FromSeconds(5)).ConfigureAwait(false);
                switch (outcome)
                {
                    case "delete":
                        await subscription.DeleteAsync(context).ConfigureAwait(false);
                        break;
                    case "transfer":
                        Assert.That(((ISubscriptionPublishPipeline)subscription).TryBeginTransfer(m_sessionMock.Object),
                            Is.True);
                        break;
                    case "close":
                        Assert.That(((ISubscriptionPublishPipeline)subscription).SessionClosed(m_sessionMock.Object),
                            Is.True);
                        break;
                    default:
                        cancellation.Cancel();
                        break;
                }
            }
            finally
            {
                release.TrySetResult(true);
            }
            if (outcome == "cancel")
            {
                Assert.CatchAsync<OperationCanceledException>(async () => await creating.ConfigureAwait(false));
            }
            else
            {
                ServiceResultException error = Assert.ThrowsAsync<ServiceResultException>(
                    async () => await creating.ConfigureAwait(false));
                Assert.That(error.StatusCode, Is.EqualTo(StatusCodes.BadSubscriptionIdInvalid));
            }
            Assert.That(registered, Is.False);
            Assert.That(subscription.Diagnostics.MonitoredItemCount, Is.Zero);
            item.Verify(value => value.Dispose(), Times.Once);
            item.VerifySet(value => value.SubscriptionCallback = subscription, Times.Never);
        }

        /// <summary>
        /// Verifies that a durable snapshot owns notification data independently of recycled retransmission messages.
        /// </summary>
        [Test]
        public void DurableSubscriptionSnapshotSurvivesRetransmissionQueueRecycling()
        {
            using Subscription subscription = CreateSubscription();
            var queue = (SentMessageQueue)typeof(Subscription)
                .GetField("m_messageQueue", BindingFlags.Instance | BindingFlags.NonPublic)!.GetValue(subscription)!;
            var item = new MonitoredItemNotification { ClientHandle = 77, Value = new DataValue(42) };
            var message = new UnpooledNotificationMessage
            {
                SequenceNumber = 9,
                NotificationData = [new ExtensionObject(new DataChangeNotification { MonitoredItems = [item] })]
            };
            queue.Enqueue([message], [], out _, out _);
            IStoredSubscription snapshot = subscription.ToStorableSubscription();
            Assert.That(snapshot.SentMessages[0], Is.Not.SameAs(message));
            queue.Clear();
            Assert.That(message.ReuseCount, Is.EqualTo(1));
            Assert.That(message.IsEmpty, Is.True);
            var messageContext = ServiceMessageContext.Create(m_telemetry);
            using var encoder = new BinaryEncoder(messageContext);
            snapshot.SentMessages[0].Encode(encoder);
            using var decoder = new BinaryDecoder(encoder.CloseAndReturnBuffer()!, messageContext);
            var decoded = new NotificationMessage();
            decoded.Decode(decoder);
            Assert.That(decoded.SequenceNumber, Is.EqualTo(9u));
            Assert.That(decoded.NotificationData[0].TryGetValue(out DataChangeNotification? data), Is.True);
            Assert.That(data!.MonitoredItems[0].ClientHandle, Is.EqualTo(77u));
            Assert.That(data.MonitoredItems[0].Value.WrappedValue.GetInt32(), Is.EqualTo(42));
        }

        /// <summary>
        /// Records recycling and resets like a pooled message, but stays out of the process-wide pool so a parallel
        /// test cannot rent and repopulate it before the assertions read it.
        /// </summary>
        private sealed class UnpooledNotificationMessage : NotificationMessage
        {
            public int ReuseCount { get; private set; }

            protected override void ReuseCore()
            {
                ReuseCount++;
                ResetForReuse();
            }
        }

        private ServerInternalData CreateServerInternalData()
        {
            var configuration = new ApplicationConfiguration
            {
                ApplicationUri = "urn:opcfoundation.org:Tests:Subscription",
                ServerConfiguration = new ServerConfiguration
                {
                    BaseAddresses = []
                }
            };
            var server = new ServerInternalData(
                new ServerProperties(),
                configuration,
                ServiceMessageContext.Create(m_telemetry));
            var masterNodeManager = new Mock<IMasterNodeManager>();
            masterNodeManager
                .SetupGet(manager => manager.DiagnosticsNodeManager)
                .Returns(m_diagnosticsNodeManagerMock.Object);
            masterNodeManager
                .SetupGet(manager => manager.ConfigurationNodeManager)
                .Returns((IConfigurationNodeManager)null!);
            masterNodeManager
                .SetupGet(manager => manager.CoreNodeManager)
                .Returns((ICoreNodeManager)null!);
            server.SetNodeManager(masterNodeManager.Object);
            server.SetMonitoredItemQueueFactory(m_queueFactoryMock.Object);
            return server;
        }

        [Test]
        [Category("NodeManagerLifecycle")]
        public async Task HasMonitoredItemsReturnsTrueForDifferentAdaptersOverSameSynchronousManagerAsync()
        {
            var synchronousNodeManager = new Mock<INodeManager>();
            var adapterA = new AsyncNodeManagerAdapter(synchronousNodeManager.Object);
            var adapterB = new AsyncNodeManagerAdapter(synchronousNodeManager.Object);
            var differentAdapter = new AsyncNodeManagerAdapter(new Mock<INodeManager>().Object);
            using var queueFactory = new MonitoredItemQueueFactory(m_telemetry);
            m_serverMock.Setup(s => s.MonitoredItemQueueFactory).Returns(queueFactory);

            using Subscription subscription = CreateSubscription();
            var itemToMonitor = new ReadValueId
            {
                NodeId = new NodeId(1),
                AttributeId = Attributes.Value
            };
            var monitoredItem = new MonitoredItem(
                m_serverMock.Object,
                adapterA,
                new object(),
                subscription.Id,
                id: 1,
                itemToMonitor,
                DiagnosticsMasks.None,
                TimestampsToReturn.Both,
                MonitoringMode.Reporting,
                clientHandle: 1,
                originalFilter: null,
                filterToUse: null,
                range: null,
                samplingInterval: 0,
                queueSize: 1,
                discardOldest: true,
                sourceSamplingInterval: 0);
            var masterNodeManager = new Mock<IMasterNodeManager>();
            masterNodeManager
                .Setup(n => n.CreateMonitoredItemsAsync(
                    It.IsAny<OperationContext>(),
                    It.IsAny<uint>(),
                    It.IsAny<double>(),
                    It.IsAny<TimestampsToReturn>(),
                    It.IsAny<ArrayOf<MonitoredItemCreateRequest>>(),
                    It.IsAny<IList<ServiceResult>>(),
                    It.IsAny<IList<MonitoringFilterResult>>(),
                    It.IsAny<IList<IMonitoredItem>>(),
                    It.IsAny<bool>(),
                    It.IsAny<CancellationToken>()))
                .Callback<
                    OperationContext,
                    uint,
                    double,
                    TimestampsToReturn,
                    ArrayOf<MonitoredItemCreateRequest>,
                    IList<ServiceResult>,
                    IList<MonitoringFilterResult>,
                    IList<IMonitoredItem>,
                    bool,
                    CancellationToken>((
                    _,
                    _,
                    _,
                    _,
                    _,
                    errors,
                    filterResults,
                    monitoredItems,
                    _,
                    _) =>
                {
                    errors[0] = ServiceResult.Good;
                    filterResults[0] = null!;
                    monitoredItems[0] = monitoredItem;
                })
                .Returns(default(ValueTask));
            m_serverMock.Setup(s => s.NodeManager).Returns(masterNodeManager.Object);

            var request = new MonitoredItemCreateRequest
            {
                ItemToMonitor = itemToMonitor,
                MonitoringMode = MonitoringMode.Reporting,
                RequestedParameters = new MonitoringParameters
                {
                    ClientHandle = 1,
                    SamplingInterval = 0,
                    QueueSize = 1,
                    DiscardOldest = true
                }
            };
            var context = new OperationContext(m_sessionMock.Object, DiagnosticsMasks.None);

            CreateMonitoredItemsResponse response = await subscription.CreateMonitoredItemsAsync(
                context,
                TimestampsToReturn.Both,
                [request],
                CancellationToken.None).ConfigureAwait(false);

            Assert.That(adapterA, Is.Not.SameAs(adapterB));
            Assert.That(adapterA.SyncNodeManager, Is.SameAs(synchronousNodeManager.Object));
            Assert.That(adapterB.SyncNodeManager, Is.SameAs(synchronousNodeManager.Object));
            Assert.That(response.Results, Has.Count.EqualTo(1));
            Assert.That(response.Results[0].StatusCode, Is.EqualTo(StatusCodes.Good));
            Assert.That(subscription.HasMonitoredItems(adapterA), Is.True);
            Assert.That(subscription.HasMonitoredItems(adapterB), Is.True);
            Assert.That(subscription.HasMonitoredItems(differentAdapter), Is.False);
        }

        [Test]
        public void TransferIsRejectedWhenTargetUsesADifferentTokenTypeWithTheSameIdentifier()
        {
            // A user name may be spelled exactly like a certificate subject. The
            // two are different principals, so the subscription must not move
            // between them (OPC 10000-4 transfer keeps the same ClientUserId).
            const string identifier = "CN=SubscriptionTransferTests";
            var userNameToken = new UserNameIdentityTokenHandler(identifier, [1, 2, 3]);
            SetSessionIdentity(m_sessionMock, userNameToken);
            using Subscription subscription = CreateSubscription();

            using Certificate certificate = DefaultCertificateFactory.Instance
                .CreateCertificate(identifier)
                .SetRSAKeySize(CertificateFactory.DefaultKeySize)
                .CreateForRSA();
            var x509Token = new X509IdentityTokenHandler(new X509IdentityToken
            {
                CertificateData = certificate.RawData.ToByteString()
            });
            var targetSession = new Mock<ISession>();
            targetSession.Setup(s => s.Id).Returns(new NodeId(Guid.NewGuid()));
            SetSessionIdentity(targetSession, x509Token);

            Assert.That(subscription.IsTransferIdentityCompatible(targetSession.Object), Is.False);
        }

        [Test]
        public void TransferIsAllowedWhenTargetPresentsTheSameUserNameIdentity()
        {
            var ownerToken = new UserNameIdentityTokenHandler("alice", [1, 2, 3]);
            SetSessionIdentity(m_sessionMock, ownerToken);
            using Subscription subscription = CreateSubscription();

            var targetToken = new UserNameIdentityTokenHandler("alice", [4, 5, 6]);
            var targetSession = new Mock<ISession>();
            targetSession.Setup(s => s.Id).Returns(new NodeId(Guid.NewGuid()));
            SetSessionIdentity(targetSession, targetToken);

            Assert.That(subscription.IsTransferIdentityCompatible(targetSession.Object), Is.True);
        }

        private static SessionDiagnosticsDataType CreateSessionDiagnostics()
        {
            return new SessionDiagnosticsDataType
            {
                ClientDescription = new ApplicationDescription
                {
                    ApplicationUri = "urn:localhost:opcfoundation.org:SubscriptionTests"
                }
            };
        }

        private static void SetSessionIdentity(
            Mock<ISession> session,
            IUserIdentityTokenHandler tokenHandler)
        {
            var identity = new UserIdentity(tokenHandler);
            session.Setup(s => s.Identity).Returns(identity);
            session.Setup(s => s.IdentityToken).Returns(tokenHandler);
            var diagnostics = new SessionDiagnosticsDataType
            {
                ClientDescription = new ApplicationDescription
                {
                    ApplicationUri = "urn:localhost:opcfoundation.org:SubscriptionTests"
                }
            };
            session
                .Setup(s => s.UpdateDiagnostics(
                    It.IsAny<Action<SessionDiagnosticsDataType>>()))
                .Callback<Action<SessionDiagnosticsDataType>>(
                    update => update(diagnostics));
            session.Setup(s => s.ClientApplicationUri)
                .Returns(() => diagnostics.ClientDescription?.ApplicationUri);
        }

        private static void SetExpiryTime(Subscription subscription, long expiryTime)
        {
            FieldInfo field = typeof(Subscription).GetField("m_publishTimerExpiry", BindingFlags.NonPublic | BindingFlags.Instance)
                ?? throw new InvalidOperationException("Field m_publishTimerExpiry not found");
            field.SetValue(subscription, expiryTime);
        }

        private static T GetPrivateField<T>(
            object instance,
            string fieldName)
        {
            FieldInfo field = instance.GetType().GetField(
                fieldName,
                BindingFlags.Instance | BindingFlags.NonPublic)
                ?? throw new InvalidOperationException(
                    $"Field {fieldName} not found.");
            return (T)(field.GetValue(instance) ??
                throw new InvalidOperationException(
                    $"Field {fieldName} is null."));
        }

        private static void SetPrivateField<T>(
            object instance,
            string fieldName,
            T value)
        {
            FieldInfo field = instance.GetType().GetField(
                fieldName,
                BindingFlags.Instance | BindingFlags.NonPublic)
                ?? throw new InvalidOperationException(
                    $"Field {fieldName} not found.");
            field.SetValue(instance, value);
        }

        private static void ExpireOnNextPublishTimer(Subscription subscription)
        {
            uint maxLifetimeCount = GetPrivateField<uint>(
                subscription,
                "m_maxLifetimeCount");
            SetPrivateField(
                subscription,
                "m_lifetimeCounter",
                maxLifetimeCount - 1);
            SetPrivateField(subscription, "m_waitingForPublish", true);
            SetExpiryTime(
                subscription,
                TimeProvider.System.GetTimestampMilliseconds() - 100);
        }

        private static SessionPublishQueue GetPublishQueue(
            SubscriptionManager manager,
            NodeId sessionId)
        {
            NodeIdDictionary<SessionPublishQueue> publishQueues =
                GetPrivateField<NodeIdDictionary<SessionPublishQueue>>(
                    manager,
                    "m_publishQueues");
            return publishQueues[sessionId];
        }

        private readonly record struct TransferFixture(
            SubscriptionManager Manager,
            Subscription Subscription,
            OperationContext SourceContext,
            OperationContext DestinationContext,
            Mock<ISession> DestinationSession);

        private async Task<TransferFixture> CreateTransferSubscriptionAsync()
        {
            var configuration = new ApplicationConfiguration
            {
                ServerConfiguration = new ServerConfiguration()
            };
            var manager = new SubscriptionManager(
                m_serverMock.Object,
                configuration);
            m_serverMock.SetupGet(server => server.SubscriptionManager).Returns(manager);

            var identity = new UserIdentity("transfer-user", [1, 2, 3]);
            m_sessionMock.SetupGet(session => session.EffectiveIdentity).Returns(identity);
            m_sessionMock.SetupGet(session => session.Identity).Returns(identity);
            m_sessionMock.SetupGet(session => session.IdentityToken).Returns(identity.TokenHandler);
            m_sessionMock.SetupGet(session => session.IsClosing).Returns(false);
            var destinationSession = new Mock<ISession>();
            destinationSession.SetupGet(session => session.Id)
                .Returns(new NodeId(Guid.NewGuid()));
            destinationSession.SetupGet(session => session.EffectiveIdentity).Returns(identity);
            destinationSession.SetupGet(session => session.Identity).Returns(identity);
            destinationSession.SetupGet(session => session.IdentityToken)
                .Returns(identity.TokenHandler);
            SessionDiagnosticsDataType destinationDiagnostics = CreateSessionDiagnostics();
            destinationSession.SetupGet(session => session.ClientApplicationUri)
                .Returns(() => destinationDiagnostics.ClientDescription?.ApplicationUri);
            destinationSession
                .Setup(session => session.UpdateDiagnostics(
                    It.IsAny<Action<SessionDiagnosticsDataType>>()))
                .Callback<Action<SessionDiagnosticsDataType>>(
                    update => update(destinationDiagnostics));
            var sourceContext = new OperationContext(
                m_sessionMock.Object,
                DiagnosticsMasks.None);
            var destinationContext = new OperationContext(
                destinationSession.Object,
                DiagnosticsMasks.None);

            CreateSubscriptionResponse created = await manager.CreateSubscriptionAsync(
                sourceContext,
                requestedPublishingInterval: 1000,
                requestedLifetimeCount: 30,
                requestedMaxKeepAliveCount: 10,
                maxNotificationsPerPublish: 0,
                publishingEnabled: true,
                priority: 0).ConfigureAwait(false);
            if (!manager.TryGetSubscription(
                    created.SubscriptionId,
                    out ISubscription? subscription))
            {
                manager.Dispose();
                throw new InvalidOperationException("Created subscription was not registered.");
            }

            return new TransferFixture(
                manager,
                (Subscription)subscription,
                sourceContext,
                destinationContext,
                destinationSession);
        }

        private static async ValueTask DeleteSubscriptionAndSignalAsync(
            SubscriptionManager manager,
            uint subscriptionId,
            TaskCompletionSource<bool> deletionCompleted,
            CancellationToken cancellationToken)
        {
            await manager.DeleteSubscriptionAsync(
                    null!,
                    subscriptionId,
                    cancellationToken)
                .ConfigureAwait(false);
            deletionCompleted.TrySetResult(true);
        }

        private static void ResetKeepAlive(Subscription subscription)
        {
            FieldInfo field = typeof(Subscription).GetField("m_keepAliveCounter", BindingFlags.NonPublic | BindingFlags.Instance)
                ?? throw new InvalidOperationException("Field m_keepAliveCounter not found");
            field.SetValue(subscription, (uint)0);
        }

        private static void AddMonitoredItem(Subscription subscription, IMonitoredItem item)
        {
            // Subscription has:
            // private readonly Dictionary<uint, LinkedListNode<IMonitoredItem>> m_monitoredItems;
            // private readonly LinkedList<IMonitoredItem> m_itemsToCheck;

            FieldInfo monitoredItemsField = typeof(Subscription).GetField("m_monitoredItems", BindingFlags.NonPublic | BindingFlags.Instance)
                ?? throw new InvalidOperationException("Field m_monitoredItems not found");
            FieldInfo itemsToCheckField = typeof(Subscription).GetField("m_itemsToCheck", BindingFlags.NonPublic | BindingFlags.Instance)
                ?? throw new InvalidOperationException("Field m_itemsToCheck not found");

            var monitoredItems = (System.Collections.IDictionary)monitoredItemsField.GetValue(subscription)!;
            var itemsToCheck = (LinkedList<IMonitoredItem>)itemsToCheckField.GetValue(subscription)!;

            // Add to itemsToCheck first to get the node
            LinkedListNode<IMonitoredItem> node = itemsToCheck!.AddLast(item);
            // Add to dictionary
            monitoredItems!.Add(item.Id, node);
        }

        private static void AddTriggerLink(Subscription subscription, uint triggeringId, ITriggeredMonitoredItem triggeredItem)
        {
            // private readonly Dictionary<uint, List<ITriggeredMonitoredItem>> m_itemsToTrigger;
            FieldInfo itemsToTriggerField = typeof(Subscription).GetField("m_itemsToTrigger", BindingFlags.NonPublic | BindingFlags.Instance)
                ?? throw new InvalidOperationException("Field m_itemsToTrigger not found");
            var itemsToTrigger = (System.Collections.IDictionary)itemsToTriggerField.GetValue(subscription)!;

            if (!itemsToTrigger!.Contains(triggeringId))
            {
                itemsToTrigger.Add(triggeringId, new List<ITriggeredMonitoredItem>());
            }
            var list = (List<ITriggeredMonitoredItem>)itemsToTrigger[triggeringId]!;
            list!.Add(triggeredItem);
        }

        private static int GetItemsToPublishCount(Subscription subscription)
        {
            FieldInfo itemsToPublishField = typeof(Subscription).GetField("m_itemsToPublish", BindingFlags.NonPublic | BindingFlags.Instance)
                ?? throw new InvalidOperationException("Field m_itemsToPublish not found");
            var itemsToPublish = (LinkedList<IMonitoredItem>)itemsToPublishField.GetValue(subscription)!;
            return itemsToPublish!.Count;
        }

        [Test]
        public void PublishTimerExpired_NotExpired_ReturnsIdle()
        {
            using Subscription subscription = CreateSubscription(1000);
            ResetKeepAlive(subscription);

            // Set expiry in far future
            SetExpiryTime(subscription, TimeProvider.System.GetTimestampMilliseconds() + 100000);

            PublishingState state = subscription.PublishTimerExpired();

            Assert.That(state, Is.EqualTo(PublishingState.Idle));
        }

        [Test]
        public void PublishTimerExpired_Expired_ReturnsNotificationsAvailable_ForKeepAlive()
        {
            using Subscription subscription = CreateSubscription(1000);
            // Don't reset keepalive, it should be maxKeepAliveCount initially.

            SetExpiryTime(subscription, TimeProvider.System.GetTimestampMilliseconds() - 100);

            PublishingState state = subscription.PublishTimerExpired();

            Assert.That(state, Is.EqualTo(PublishingState.NotificationsAvailable));
        }

        [Test]
        public void PublishTimerExpired_Expired_ItemsReady_ReturnsNotificationsAvailable()
        {
            using Subscription subscription = CreateSubscription(1000);
            ResetKeepAlive(subscription);
            SetExpiryTime(subscription, TimeProvider.System.GetTimestampMilliseconds() - 100);

            // Mock Monitored Item
            var itemMock = new Mock<IMonitoredItem>();
            itemMock.SetupGet(i => i.MonitoringMode).Returns(MonitoringMode.Reporting);
            itemMock.Setup(i => i.Id).Returns(1);
            itemMock.Setup(i => i.IsReadyToPublish).Returns(true);

            AddMonitoredItem(subscription, itemMock.Object);

            PublishingState state = subscription.PublishTimerExpired();

            Assert.That(state, Is.EqualTo(PublishingState.NotificationsAvailable));
            Assert.That(GetItemsToPublishCount(subscription), Is.EqualTo(1));
        }

        [Test]
        public void PublishTimerExpired_Expired_ItemsNotReady_ReturnsIdle()
        {
            using Subscription subscription = CreateSubscription(1000);
            ResetKeepAlive(subscription);
            SetExpiryTime(subscription, TimeProvider.System.GetTimestampMilliseconds() - 100);

            // Mock Monitored Item
            var itemMock = new Mock<IMonitoredItem>();
            itemMock.SetupGet(i => i.MonitoringMode).Returns(MonitoringMode.Reporting);
            itemMock.Setup(i => i.Id).Returns(1);
            itemMock.Setup(i => i.IsReadyToPublish).Returns(false);

            AddMonitoredItem(subscription, itemMock.Object);

            PublishingState state = subscription.PublishTimerExpired();

            Assert.That(state, Is.EqualTo(PublishingState.Idle));
            Assert.That(GetItemsToPublishCount(subscription), Is.Zero);
        }

        [Test]
        public void PublishTimerExpired_Expired_IncrementsKeepAlive()
        {
            using Subscription subscription = CreateSubscription(1000);
            ResetKeepAlive(subscription);
            SetExpiryTime(subscription, TimeProvider.System.GetTimestampMilliseconds() - 100);

            Assert.That(subscription.Diagnostics.CurrentKeepAliveCount, Is.Zero);

            subscription.PublishTimerExpired();

            Assert.That(subscription.Diagnostics.CurrentKeepAliveCount, Is.EqualTo(1));
        }

        [TestCase(false)]
        [TestCase(true)]
        public async Task PublishTimerKeepsQueuedMessagesReadyUntilDrainedAsync(bool eventNotifications)
        {
            var clock = new FakeTimeProvider();
            using Subscription subscription = CreateSubscription(
                publishingInterval: 100,
                maxNotificationsPerPublish: 1,
                timeProvider: clock);
            var limits = new List<uint>();
            IMonitoredItem item = eventNotifications
                ? CreateEventMonitoredItem(1, 3, limits).Object
                : CreateDataChangeMonitoredItem(1, 3, limits).Object;
            await RegisterMonitoredItemsAsync(subscription, item).ConfigureAwait(false);
            using var context = new OperationContext(m_sessionMock.Object, DiagnosticsMasks.None);

            for (uint sequence = 1; sequence <= 3; sequence++)
            {
                clock.Advance(TimeSpan.FromMilliseconds(101));
                Assert.That(subscription.PublishTimerExpired(), Is.EqualTo(PublishingState.NotificationsAvailable),
                    "Queued notification messages must not wait for keep-alive expiry.");
                NotificationMessage message = subscription.Publish(context, out _, out bool moreNotifications)!;
                Assert.Multiple(() =>
                {
                    Assert.That(message!.SequenceNumber, Is.EqualTo(sequence));
                    Assert.That(message.NotificationData, Has.Count.EqualTo(1));
                    Assert.That(moreNotifications, Is.EqualTo(sequence < 3));
                    Assert.That(GetItemsToPublishCount(subscription), Is.Zero);
                });
            }

            clock.Advance(TimeSpan.FromMilliseconds(101));
            Assert.That(subscription.PublishTimerExpired(), Is.EqualTo(PublishingState.Idle),
                "Messages retained only for retransmission must not count as queued notifications.");
        }

        [Test]
        public void PublishTimerExpired_Triggering_CorrectlyTriggersAndPublishes()
        {
            using Subscription subscription = CreateSubscription(1000);
            ResetKeepAlive(subscription);
            SetExpiryTime(subscription, TimeProvider.System.GetTimestampMilliseconds() - 100);

            // Item A: Triggering item. Ready to publish, Ready to trigger.
            var itemAMock = new Mock<IMonitoredItem>();
            itemAMock.SetupGet(i => i.MonitoringMode).Returns(MonitoringMode.Reporting);
            itemAMock.Setup(i => i.Id).Returns(1);
            itemAMock.Setup(i => i.IsReadyToPublish).Returns(true);
            itemAMock.SetupProperty(i => i.IsReadyToTrigger, true); // Use property behavior so it can be set to false by Subscription

            // Item B: Triggered item. Initially NOT ready to publish.
            // B must implement ITriggeredMonitoredItem as well.
            var itemBMock = new Mock<IMonitoredItem>();
            itemBMock.SetupGet(i => i.MonitoringMode).Returns(MonitoringMode.Reporting);
            itemBMock.As<ITriggeredMonitoredItem>();
            Mock<ITriggeredMonitoredItem> triggeredItemB = itemBMock.As<ITriggeredMonitoredItem>();

            itemBMock.Setup(i => i.Id).Returns(2);
            triggeredItemB.Setup(i => i.Id).Returns(2);

            // "State" of ready
            bool bIsReady = false;
            itemBMock.Setup(i => i.IsReadyToPublish).Returns(() => bIsReady);

            // SetTriggered updates state
            triggeredItemB.Setup(i => i.SetTriggered()).Returns(() =>
            {
                bIsReady = true;
                return true; // True indicates it has something to publish
            });

            // Add both items
            AddMonitoredItem(subscription, itemAMock.Object);
            AddMonitoredItem(subscription, itemBMock.Object);

            // Add trigger link A -> B
            AddTriggerLink(subscription, 1, triggeredItemB.Object);

            PublishingState state = subscription.PublishTimerExpired();

            Assert.That(state, Is.EqualTo(PublishingState.NotificationsAvailable));

            // Both items should be in publish queue
            Assert.That(GetItemsToPublishCount(subscription), Is.EqualTo(2), "Both items should be ready to publish");

            // Verify trigger was called
            triggeredItemB.Verify(i => i.SetTriggered(), Times.Once);

            // Verify IsReadyToTrigger on A was reset to false
            Assert.That(itemAMock.Object.IsReadyToTrigger, Is.False, "IsReadyToTrigger should be reset");
        }

        private static void AddMonitoredItemToPublish(Subscription subscription, IMonitoredItem item)
        {
            FieldInfo itemsToPublishField = typeof(Subscription).GetField("m_itemsToPublish", BindingFlags.NonPublic | BindingFlags.Instance)
                ?? throw new InvalidOperationException("Field m_itemsToPublish not found");
            var itemsToPublish = (LinkedList<IMonitoredItem>)itemsToPublishField.GetValue(subscription)!;
            itemsToPublish!.AddLast(item);
        }

        [Test]
        public void Publish_MultipleTimes_WithMaxMessageCount()
        {
            using var subscription = new Subscription(m_serverMock.Object, m_sessionMock.Object, 1, 100, 1000, 10, 1, 0, true, 2);
            var itemMock = new Mock<IDataChangeMonitoredItem2>();
            itemMock.SetupGet(i => i.MonitoringMode).Returns(MonitoringMode.Reporting);

            var values = new List<MonitoredItemNotification>
            {
                new() { Value = new DataValue(1) },
                new() { Value = new DataValue(2) },
                new() { Value = new DataValue(3) }
            };

            int counter = 0;
            itemMock.Setup(i => i.Publish(
                It.IsAny<OperationContext>(),
                It.IsAny<Queue<MonitoredItemNotification>>(),
                It.IsAny<Queue<DiagnosticInfo>>(),
                It.IsAny<uint>(),
                It.IsAny<Microsoft.Extensions.Logging.ILogger>()))
                .Returns<OperationContext, Queue<MonitoredItemNotification>, Queue<DiagnosticInfo>, uint, Microsoft.Extensions.Logging.ILogger>(
                (ctx, nq, dq, max, logger) =>
                {
                    if (counter < values.Count)
                    {
                        nq.Enqueue(values[counter++]);
                        dq.Enqueue(new DiagnosticInfo());
                        itemMock.SetupGet(x => x.IsReadyToPublish).Returns(counter < values.Count);
                        return counter < values.Count;
                    }
                    return false;
                });
            itemMock.SetupGet(i => i.Id).Returns(1);
            itemMock.SetupGet(i => i.IsReadyToPublish).Returns(true);
            itemMock.SetupGet(i => i.AttributeId).Returns(Attributes.Value);
            itemMock.SetupGet(i => i.MonitoredItemType).Returns(MonitoredItemTypeMask.DataChange);

            AddMonitoredItem(subscription, itemMock.Object);
            SetExpiryTime(subscription, TimeProvider.System.GetTimestampMilliseconds() - 100);
            PublishingState state = subscription.PublishTimerExpired();

            AddMonitoredItemToPublish(subscription, itemMock.Object);

            var messages = new List<NotificationMessage>();

            // First publish
            var ctx1 = new OperationContext(m_sessionMock.Object, new DiagnosticsMasks());
            NotificationMessage message = subscription.Publish(ctx1, out ArrayOf<uint> availableSequenceNumbers, out bool moreNotifications1)!;
            messages.Add(message!);

            // Should be more because we generated multiple notifications and limit the max per publish to 1 for tests.
            Assert.That(moreNotifications1, Is.True);

            // Second publish
            NotificationMessage message2 = subscription.Publish(ctx1, out availableSequenceNumbers, out bool moreNotifications2)!;

            // third publish
            NotificationMessage message3 = subscription.Publish(ctx1, out availableSequenceNumbers, out bool moreNotifications3)!;

            Assert.That(message2, Is.Not.Null);
            Assert.That(message3, Is.Not.Null);
            Assert.That(moreNotifications2, Is.True);
            Assert.That(moreNotifications3, Is.False);

            // The third message evicted the unacknowledged first one from the
            // two-message retransmission queue.
            Assert.That(availableSequenceNumbers.Count, Is.EqualTo(2));
            Assert.That(subscription.Diagnostics.UnacknowledgedMessageCount, Is.EqualTo(2u));
            Assert.That(subscription.Diagnostics.DiscardedMessageCount, Is.EqualTo(1u));
        }

        [Test]
        public async Task PublishWithZeroLimitDrainsEventAndDataChangeQueuesAsync()
        {
            var timeProvider = new FakeTimeProvider(
                new DateTimeOffset(2026, 1, 1, 0, 0, 0, TimeSpan.Zero));
            using Subscription subscription = CreateSubscription(
                publishingInterval: 100,
                maxNotificationsPerPublish: 0,
                timeProvider);
            var publishLimits = new List<uint>();
            Mock<IEventMonitoredItem> eventItem = CreateEventMonitoredItem(
                id: 1,
                notificationCount: 2,
                publishLimits);
            Mock<IDataChangeMonitoredItem> dataChangeItem = CreateDataChangeMonitoredItem(
                id: 2,
                notificationCount: 2,
                publishLimits);
            await RegisterMonitoredItemsAsync(
                subscription,
                eventItem.Object,
                dataChangeItem.Object).ConfigureAwait(false);

            timeProvider.Advance(TimeSpan.FromMilliseconds(101));
            Assert.That(
                subscription.PublishTimerExpired(),
                Is.EqualTo(PublishingState.NotificationsAvailable));

            var context = new OperationContext(m_sessionMock.Object, new DiagnosticsMasks());
            NotificationMessage message = subscription.Publish(
                context,
                out _,
                out bool moreNotifications)!;

            Assert.That(message!.NotificationData, Has.Count.EqualTo(2));
            var eventNotification = (EventNotificationList)ExtensionObject.ToEncodeable(
                message.NotificationData[0])!;
            var dataChangeNotification = (DataChangeNotification)ExtensionObject.ToEncodeable(
                message.NotificationData[1])!;
            Assert.Multiple(() =>
            {
                Assert.That(moreNotifications, Is.False);
                Assert.That(eventNotification!.Events, Has.Count.EqualTo(2));
                Assert.That(dataChangeNotification!.MonitoredItems, Has.Count.EqualTo(2));
                Assert.That(dataChangeNotification.DiagnosticInfos, Has.Count.EqualTo(2));
                Assert.That(
                    publishLimits,
                    Is.EqualTo([uint.MaxValue, uint.MaxValue]));
            });
        }

        [TestCase(0u, 1)]
        [TestCase(1u, 1)]
        [TestCase(10u, 1)]
        [TestCase(0u, 25)]
        [TestCase(2u, 25)]
        public async Task PublishPreservesEveryReadyValueAcrossMessageBudgetsAsync(
            uint maxMessageCount, int valuesPerItem)
        {
            var clock = new FakeTimeProvider();
            using var subscription = new Subscription(
                m_serverMock.Object, m_sessionMock.Object, 1, 100, 1000, 10, 10, 0, true, maxMessageCount, clock);
            int itemCount = valuesPerItem == 1 ? 151 : 6;
            var items = new IMonitoredItem[itemCount];
            var expected = new List<(uint Handle, int Value)>();
            var limits = new List<uint>();
            for (int index = 0; index < itemCount; index++)
            {
                uint handle = (uint)index + 1;
                items[index] = index % 2 == 0
                    ? CreateDataChangeMonitoredItem(handle, valuesPerItem, limits).Object
                    : CreateEventMonitoredItem(handle, valuesPerItem, limits).Object;
                for (int value = 0; value < valuesPerItem; value++)
                {
                    expected.Add((handle, value));
                }
            }
            await RegisterMonitoredItemsAsync(subscription, items).ConfigureAwait(false);
            clock.Advance(TimeSpan.FromMilliseconds(101));
            Assert.That(subscription.PublishTimerExpired(), Is.EqualTo(PublishingState.NotificationsAvailable));
            using var context = new OperationContext(m_sessionMock.Object, DiagnosticsMasks.None);
            var received = new List<(uint Handle, int Value)>();
            uint expectedSequenceNumber = 1;
            bool moreNotifications;
            do
            {
                NotificationMessage message = subscription.Publish(context, out _, out moreNotifications)!;
                Assert.That(message, Is.Not.Null);
                Assert.That(message.NotificationData, Is.Not.Empty, "Ready values must not become a keepalive.");
                Assert.That(message.SequenceNumber, Is.EqualTo(expectedSequenceNumber++));
                int count = 0;
                foreach (ExtensionObject notification in message.NotificationData)
                {
                    if (notification.TryGetValue(out DataChangeNotification? data))
                    {
                        Assert.That(data.DiagnosticInfos, Has.Count.EqualTo(data.MonitoredItems.Count));
                        foreach (MonitoredItemNotification item in data.MonitoredItems)
                        {
                            Assert.That(item.Value.WrappedValue.TryGetValue(out int value), Is.True);
                            received.Add((item.ClientHandle, value));
                            count++;
                        }
                    }
                    else
                    {
                        Assert.That(notification.TryGetValue(out EventNotificationList? events), Is.True);
                        foreach (EventFieldList item in events!.Events)
                        {
                            Assert.That(item.EventFields[0].TryGetValue(out int value), Is.True);
                            received.Add((item.ClientHandle, value));
                            count++;
                        }
                    }
                }
                Assert.That(count, Is.InRange(1, 10));
                Assert.That(moreNotifications, Is.EqualTo(received.Count < expected.Count));
                Assert.That(subscription.Acknowledge(context, message.SequenceNumber), Is.Null);
                Assert.That(expectedSequenceNumber, Is.LessThanOrEqualTo((uint)expected.Count + 1));
            }
            while (moreNotifications);

            Assert.That(received, Is.EquivalentTo(expected));
            Assert.That(subscription.Diagnostics.NotificationsCount, Is.EqualTo(expected.Count));
        }

        [TestCase("missing", true)]
        [TestCase("enabled", true)]
        [TestCase("disabled", false)]
        public async Task RestoredPublishingDefaultsToEnabledUnlessExplicitlyDisabledAsync(
            string storedState, bool expected)
        {
            var stored = new StoredSubscription
            {
                Id = 41,
                IsDurable = true,
                PublishingInterval = 1000,
                MaxKeepaliveCount = 10,
                MaxLifetimeCount = 30,
                MaxMessageCount = 10,
                SequenceNumber = 1,
                SentMessages = [],
                MonitoredItems = []
            };
            if (storedState != "missing")
            {
                stored.PublishingEnabled = expected;
            }

            using Subscription restored = await Subscription.RestoreAsync(m_serverMock.Object, stored)
                .ConfigureAwait(false);

            Assert.That(stored.PublishingEnabled, Is.EqualTo(expected));
            Assert.That(restored.Diagnostics.PublishingEnabled, Is.EqualTo(expected));
            Assert.That(((IStoredSubscriptionState)restored.ToStorableSubscription()).PublishingEnabled,
                Is.EqualTo(expected));
        }

        [TestCase("caller-cancelled")]
        [TestCase("diagnostics-cancelled")]
        [TestCase("diagnostics-stopping")]
        public async Task DeleteCleansDetachedItemsDespiteDiagnosticsCancellationOrStopAsync(string failure)
        {
            using Subscription subscription = CreateSubscription();
            using var context = new OperationContext(m_sessionMock.Object, DiagnosticsMasks.None);
            using var cancellation = new CancellationTokenSource();
            var calls = new List<string>();
            Mock<IDataChangeMonitoredItem> item = CreateDataChangeMonitoredItem(7, 1, []);
            item.Setup(value => value.Dispose()).Callback(() => calls.Add("dispose"));
            await RegisterMonitoredItemsAsync(subscription, item.Object).ConfigureAwait(false);
            m_nodeManagerMock.Setup(manager => manager.DeleteMonitoredItemsAsync(
                    It.IsAny<OperationContext>(), subscription.Id, It.IsAny<IList<IMonitoredItem>>(),
                    It.IsAny<IList<ServiceResult>>(), It.IsAny<CancellationToken>()))
                .Returns((OperationContext _, uint _, IList<IMonitoredItem> items,
                    IList<ServiceResult> errors, CancellationToken ct) =>
                {
                    Assert.That(ct.CanBeCanceled, Is.False);
                    Assert.That(items, Has.Count.EqualTo(1));
                    Assert.That(items[0], Is.SameAs(item.Object));
                    errors[0] = ServiceResult.Good;
                    calls.Add("unregister");
                    return default;
                });
            m_diagnosticsNodeManagerMock.Setup(manager => manager.DeleteSubscriptionDiagnosticsAsync(
                    It.IsAny<ServerSystemContext>(), It.IsAny<NodeId>(), It.IsAny<CancellationToken>()))
                .Returns((ServerSystemContext _, NodeId _, CancellationToken ct) =>
                {
                    calls.Add("diagnostics");
                    ct.ThrowIfCancellationRequested();
                    return failure switch
                    {
                        "diagnostics-cancelled" => throw new OperationCanceledException(),
                        "diagnostics-stopping" => throw new ServiceResultException(StatusCodes.BadServerHalted),
                        _ => default
                    };
                });
            cancellation.Cancel();

            if (failure == "caller-cancelled")
            {
                await subscription.DeleteAsync(context, cancellation.Token).ConfigureAwait(false);
            }
            else if (failure == "diagnostics-cancelled")
            {
                Assert.ThrowsAsync<OperationCanceledException>(
                    async () => await subscription.DeleteAsync(context).ConfigureAwait(false));
            }
            else
            {
                ServiceResultException exception = Assert.ThrowsAsync<ServiceResultException>(
                    async () => await subscription.DeleteAsync(context).ConfigureAwait(false));
                Assert.That(exception.StatusCode, Is.EqualTo(StatusCodes.BadServerHalted));
            }

            Assert.That(subscription.MonitoredItemCount, Is.Zero);
            Assert.That(calls, Is.EqualTo(s_deleteCleanupOrder));
            item.Verify(value => value.Dispose(), Times.Once);
        }

        [Test]
        public async Task PublishWithFiniteLimitBuildsAndDrainsQueuedMessagesAsync()
        {
            var timeProvider = new FakeTimeProvider(
                new DateTimeOffset(2026, 1, 1, 0, 0, 0, TimeSpan.Zero));
            using Subscription subscription = CreateSubscription(
                publishingInterval: 100,
                maxNotificationsPerPublish: 2,
                timeProvider);
            var publishLimits = new List<uint>();
            Mock<IEventMonitoredItem> firstEventItem = CreateEventMonitoredItem(
                id: 1,
                notificationCount: 3,
                publishLimits);
            Mock<IEventMonitoredItem> secondEventItem = CreateEventMonitoredItem(
                id: 2,
                notificationCount: 1,
                publishLimits);
            Mock<IDataChangeMonitoredItem> dataChangeItem = CreateDataChangeMonitoredItem(
                id: 3,
                notificationCount: 2,
                publishLimits);
            await RegisterMonitoredItemsAsync(
                subscription,
                firstEventItem.Object,
                secondEventItem.Object,
                dataChangeItem.Object).ConfigureAwait(false);

            timeProvider.Advance(TimeSpan.FromMilliseconds(101));
            Assert.That(
                subscription.PublishTimerExpired(),
                Is.EqualTo(PublishingState.NotificationsAvailable));

            var context = new OperationContext(m_sessionMock.Object, new DiagnosticsMasks());
            NotificationMessage firstMessage = subscription.Publish(
                context,
                out _,
                out bool moreAfterFirst)!;
            NotificationMessage secondMessage = subscription.Publish(
                context,
                out _,
                out bool moreAfterSecond)!;
            NotificationMessage thirdMessage = subscription.Publish(
                context,
                out _,
                out bool moreAfterThird)!;

            Assert.Multiple(() =>
            {
                Assert.That(firstMessage!.NotificationData, Has.Count.EqualTo(1));
                Assert.That(secondMessage!.NotificationData, Has.Count.EqualTo(1));
                Assert.That(thirdMessage!.NotificationData, Has.Count.EqualTo(1));
            });
            var firstEvents = (EventNotificationList)ExtensionObject.ToEncodeable(
                firstMessage!.NotificationData[0])!;
            var secondEvents = (EventNotificationList)ExtensionObject.ToEncodeable(
                secondMessage!.NotificationData[0])!;
            var dataChanges = (DataChangeNotification)ExtensionObject.ToEncodeable(
                thirdMessage!.NotificationData[0])!;
            Assert.Multiple(() =>
            {
                Assert.That(moreAfterFirst, Is.True);
                Assert.That(moreAfterSecond, Is.True);
                Assert.That(moreAfterThird, Is.False);
                Assert.That(firstEvents!.Events, Has.Count.EqualTo(2));
                Assert.That(secondEvents!.Events, Has.Count.EqualTo(2));
                Assert.That(dataChanges!.MonitoredItems, Has.Count.EqualTo(2));
                Assert.That(publishLimits, Is.EqualTo(new uint[] { 6, 6, 6 }));
            });
        }

        [TestCase(0, 0L, 0L)]
        [TestCase(0, 25L, 25L)]
        [TestCase(10, 0L, 10L)]
        [TestCase(10, 10L, 10L)]
        [TestCase(10, 25L, 10L)]
        [TestCase(0, uint.MaxValue, uint.MaxValue)]
        [TestCase(10, uint.MaxValue, 10L)]
        [TestCase(int.MaxValue, uint.MaxValue, int.MaxValue)]
        public async Task CreateSubscriptionWithNotificationLimitsUsesEffectiveLimitAsync(
            int serverLimit,
            long requestedLimit,
            long expectedLimit)
        {
            var configuration = new ApplicationConfiguration
            {
                ServerConfiguration = new ServerConfiguration
                {
                    MaxNotificationsPerPublish = serverLimit
                }
            };
            using var manager = new SubscriptionManager(
                m_serverMock.Object,
                configuration);

            var context = new OperationContext(m_sessionMock.Object, new DiagnosticsMasks());
            CreateSubscriptionResponse response = await manager.CreateSubscriptionAsync(
                context,
                requestedPublishingInterval: 1000,
                requestedLifetimeCount: 30,
                requestedMaxKeepAliveCount: 10,
                maxNotificationsPerPublish: (uint)requestedLimit,
                publishingEnabled: true,
                priority: 0).ConfigureAwait(false);

            Assert.That(
                manager.TryGetSubscription(response.SubscriptionId, out ISubscription? subscription),
                Is.True);
            Assert.That(
                subscription!.Diagnostics.MaxNotificationsPerPublish,
                Is.EqualTo((uint)expectedLimit));
        }

        [Test]
        public async Task ShutdownCancelsPublishWorkerBeforeWaitingForManagerSemaphoreAsync()
        {
            var configuration = new ApplicationConfiguration
            {
                ServerConfiguration = new ServerConfiguration()
            };
            using var manager = new SubscriptionManager(
                m_serverMock.Object,
                configuration);
            await manager.StartupAsync().ConfigureAwait(false);

            // Shutdown clears the manager's task field after draining the worker.
            Task publishWorker = GetPrivateField<Task>(
                manager,
                "m_publishWorkerTask");
            Assert.That(publishWorker.IsCompleted, Is.False);
            SemaphoreSlim semaphore = GetPrivateField<SemaphoreSlim>(
                manager,
                "m_semaphoreSlim");
            await semaphore.WaitAsync().ConfigureAwait(false);
            Task shutdown = manager.ShutdownAsync().AsTask();
            try
            {
                await publishWorker.WaitAsync(TimeSpan.FromSeconds(5)).ConfigureAwait(false);
                Assert.That(shutdown.IsCompleted, Is.False);
            }
            finally
            {
                semaphore.Release();
            }

            await shutdown.ConfigureAwait(false);
        }

        [Test]
        public void CreateSubscriptionRejectsAnImplementationNotDerivedFromSubscription()
        {
            var impostor = new Mock<ISubscription>();
            using var manager = new NonPipelineSubscriptionManager(
                m_serverMock.Object,
                new ApplicationConfiguration
                {
                    ServerConfiguration = new ServerConfiguration()
                },
                impostor.Object);
            var context = new OperationContext(m_sessionMock.Object, new DiagnosticsMasks());

            ServiceResultException exception = Assert.ThrowsAsync<ServiceResultException>(
                async () => await manager.CreateSubscriptionAsync(
                    context,
                    requestedPublishingInterval: 1000,
                    requestedLifetimeCount: 30,
                    requestedMaxKeepAliveCount: 10,
                    maxNotificationsPerPublish: 0,
                    publishingEnabled: true,
                    priority: 0).ConfigureAwait(false));

            Assert.That(exception.StatusCode, Is.EqualTo(StatusCodes.BadInternalError));
            impostor.Verify(subscription => subscription.Dispose(), Times.Once);
        }

        /// <summary>
        /// Returns a fixed ISubscription that does not derive from Subscription, so the
        /// manager's pipeline admission check has something to reject.
        /// </summary>
        private sealed class NonPipelineSubscriptionManager : SubscriptionManager
        {
            public NonPipelineSubscriptionManager(
                IServerInternal server,
                ApplicationConfiguration configuration,
                ISubscription created)
                : base(server, configuration)
            {
                m_created = created;
            }

            protected override ISubscription CreateSubscription(
                OperationContext context,
                uint subscriptionId,
                double publishingInterval,
                uint lifetimeCount,
                uint keepAliveCount,
                uint maxNotificationsPerPublish,
                byte priority,
                bool publishingEnabled)
            {
                return m_created;
            }

            private readonly ISubscription m_created;
        }

        [Test]
        public void RestoreTransferClaimRemovesCurrentClaimWhenRestoreEntryAlreadyExists()
        {
            using Subscription subscription = CreateSubscription();
            using var queue = new SessionPublishQueue(
                m_serverMock.Object,
                m_sessionMock.Object,
                maxPublishRequests: 10);
            queue.Add(subscription);

            Assert.That(
                queue.TryClaimForTransfer(
                    subscription,
                    m_sessionMock.Object,
                    out SessionPublishQueue.SubscriptionTransferClaim? claim),
                Is.True);
            Assert.That(claim, Is.Not.Null);

            var collidingSubscription = new Mock<ISubscriptionPublishPipeline>();
            collidingSubscription.SetupGet(sub => sub.Id).Returns(subscription.Id);
            queue.Add(collidingSubscription.Object);

            Assert.That(queue.RestoreTransferClaim(claim!), Is.False);

            subscription.AbortTransfer(m_sessionMock.Object);
            queue.Remove(collidingSubscription.Object, removeQueuedRequests: false);
            queue.Add(subscription);

            Assert.That(
                queue.TryClaimForTransfer(
                    subscription,
                    m_sessionMock.Object,
                    out SessionPublishQueue.SubscriptionTransferClaim? retryClaim),
                Is.True,
                "The failed restore must not leave a stale claim that blocks future transfers.");
            Assert.That(retryClaim, Is.Not.Null);
            queue.CompleteTransferClaim(retryClaim!);
            subscription.AbortTransfer(m_sessionMock.Object);
        }

        [TestCase(0, 0L, 0L)]
        [TestCase(0, 25L, 25L)]
        [TestCase(10, 0L, 10L)]
        [TestCase(10, 10L, 10L)]
        [TestCase(10, 25L, 10L)]
        [TestCase(0, uint.MaxValue, uint.MaxValue)]
        [TestCase(10, uint.MaxValue, 10L)]
        [TestCase(int.MaxValue, uint.MaxValue, int.MaxValue)]
        public async Task ModifySubscriptionWithNotificationLimitsUsesEffectiveLimitAsync(
            int serverLimit,
            long requestedLimit,
            long expectedLimit)
        {
            var configuration = new ApplicationConfiguration
            {
                ServerConfiguration = new ServerConfiguration
                {
                    MaxNotificationsPerPublish = serverLimit
                }
            };
            using var manager = new SubscriptionManager(
                m_serverMock.Object,
                configuration);

            var context = new OperationContext(m_sessionMock.Object, new DiagnosticsMasks());
            CreateSubscriptionResponse response = await manager.CreateSubscriptionAsync(
                context,
                requestedPublishingInterval: 1000,
                requestedLifetimeCount: 30,
                requestedMaxKeepAliveCount: 10,
                maxNotificationsPerPublish: 1,
                publishingEnabled: true,
                priority: 0).ConfigureAwait(false);

            manager.ModifySubscription(
                context,
                response.SubscriptionId,
                requestedPublishingInterval: 1000,
                requestedLifetimeCount: 30,
                requestedMaxKeepAliveCount: 10,
                maxNotificationsPerPublish: (uint)requestedLimit,
                priority: 0,
                revisedPublishingInterval: out _,
                revisedLifetimeCount: out _,
                revisedMaxKeepAliveCount: out _);

            Assert.That(
                manager.TryGetSubscription(response.SubscriptionId, out ISubscription? subscription),
                Is.True);
            Assert.That(
                subscription!.Diagnostics.MaxNotificationsPerPublish,
                Is.EqualTo((uint)expectedLimit));
        }

        [Test]
        public async Task TransferSerializesWithClosingSourceSessionAsync()
        {
            var configuration = new ApplicationConfiguration
            {
                ServerConfiguration = new ServerConfiguration()
            };
            using var manager = new SubscriptionManager(
                m_serverMock.Object,
                configuration);
            var identity = new UserIdentity("transfer-user", [1, 2, 3]);
            bool sourceClosing = false;
            m_sessionMock.SetupGet(session => session.EffectiveIdentity).Returns(identity);
            m_sessionMock.SetupGet(session => session.Identity).Returns(identity);
            m_sessionMock.SetupGet(session => session.IdentityToken).Returns(identity.TokenHandler);
            m_sessionMock.SetupGet(session => session.IsClosing).Returns(() => sourceClosing);
            var destinationSession = new Mock<ISession>();
            destinationSession.SetupGet(session => session.Id).Returns(new NodeId(Guid.NewGuid()));
            destinationSession.SetupGet(session => session.EffectiveIdentity).Returns(identity);
            destinationSession.SetupGet(session => session.Identity).Returns(identity);
            destinationSession.SetupGet(session => session.IdentityToken)
                .Returns(identity.TokenHandler);
            SessionDiagnosticsDataType destinationDiagnostics = CreateSessionDiagnostics();
            destinationSession.SetupGet(session => session.ClientApplicationUri)
                .Returns(() => destinationDiagnostics.ClientDescription?.ApplicationUri);
            destinationSession
                .Setup(session => session.UpdateDiagnostics(
                    It.IsAny<Action<SessionDiagnosticsDataType>>()))
                .Callback<Action<SessionDiagnosticsDataType>>(
                    update => update(destinationDiagnostics));
            var sourceContext = new OperationContext(
                m_sessionMock.Object,
                DiagnosticsMasks.None);
            var destinationContext = new OperationContext(
                destinationSession.Object,
                DiagnosticsMasks.None);
            CreateSubscriptionResponse created = await manager.CreateSubscriptionAsync(
                sourceContext,
                requestedPublishingInterval: 1000,
                requestedLifetimeCount: 30,
                requestedMaxKeepAliveCount: 10,
                maxNotificationsPerPublish: 0,
                publishingEnabled: true,
                priority: 0).ConfigureAwait(false);
            Assert.That(
                manager.TryGetSubscription(created.SubscriptionId, out ISubscription? subscription),
                Is.True);

            var transferEntered = new TaskCompletionSource<bool>(
                TaskCreationOptions.RunContinuationsAsynchronously);
            var releaseTransfer = new TaskCompletionSource<bool>(
                TaskCreationOptions.RunContinuationsAsynchronously);
            m_nodeManagerMock
                .Setup(manager => manager.TransferMonitoredItemsAsync(
                    It.IsAny<OperationContext>(),
                    It.IsAny<bool>(),
                    It.IsAny<IList<IMonitoredItem>>(),
                    It.IsAny<IList<ServiceResult>>(),
                    It.IsAny<MonitoredItemTransferOptions>(),
                    It.IsAny<CancellationToken>()))
                .Callback(() => transferEntered.TrySetResult(true))
                .Returns(() => new ValueTask(releaseTransfer.Task));
            sourceClosing = true;

            Task<TransferSubscriptionsResponse> transferTask = manager
                .TransferSubscriptionsAsync(
                    destinationContext,
                    [created.SubscriptionId],
                    sendInitialValues: false)
                .AsTask();
            Task entered = await Task.WhenAny(
                transferEntered.Task,
                Task.Delay(TimeSpan.FromSeconds(5))).ConfigureAwait(false);
            if (!ReferenceEquals(entered, transferEntered.Task))
            {
                releaseTransfer.TrySetResult(true);
                Assert.Fail("Transfer did not reach the monitored-item barrier.");
            }

            Task closeTask = manager
                .SessionClosingAsync(
                    sourceContext,
                    m_sessionMock.Object.Id,
                    deleteSubscriptions: false,
                    CancellationToken.None)
                .AsTask();
            bool closeWaitedForTransfer = !closeTask.IsCompleted;
            releaseTransfer.TrySetResult(true);
            TransferSubscriptionsResponse transferred = await transferTask.ConfigureAwait(false);
            await closeTask.ConfigureAwait(false);
            ConcurrentDictionary<uint, ISubscriptionPublishPipeline> abandonedSubscriptions =
                GetPrivateField<ConcurrentDictionary<uint, ISubscriptionPublishPipeline>>(
                    manager,
                    "m_abandonedSubscriptions");

            Assert.Multiple(() =>
            {
                Assert.That(closeWaitedForTransfer, Is.True);
                Assert.That(transferred.Results, Has.Count.EqualTo(1));
                Assert.That(transferred.Results[0].StatusCode, Is.EqualTo(StatusCodes.Good));
                Assert.That(subscription!.Session, Is.SameAs(destinationSession.Object));
                Assert.That(abandonedSubscriptions, Is.Empty);
            });
            Assert.DoesNotThrow(() => subscription!.ResendData(destinationContext));
        }

        [Test]
        public async Task TransferClaimsSourceBeforeCallbacksAndBlocksStalePublishAsync()
        {
            TransferFixture fixture = await CreateTransferSubscriptionAsync().ConfigureAwait(false);
            using SubscriptionManager manager = fixture.Manager;
            Subscription subscription = fixture.Subscription;
            SessionPublishQueue sourceQueue = GetPublishQueue(
                manager,
                fixture.SourceContext.SessionId);
            SetPrivateField(subscription, "m_waitingForPublish", false);
            SetPrivateField(subscription, "m_lifetimeCounter", 0u);
            SetExpiryTime(
                subscription,
                TimeProvider.System.GetTimestampMilliseconds() - 100);
            m_sessionMock
                .Setup(session => session.IsSecureChannelValid("source-channel"))
                .Returns(true);
            IReadOnlyList<SessionPublishQueue.QueuedSubscription> staleSnapshot =
                sourceQueue.CapturePublishTimerSnapshot();
            using var publishCancellation = new CancellationTokenSource();
            Task<ISubscriptionPublishPipeline> sourcePublish = sourceQueue.PublishAsync(
                "source-channel",
                DateTime.MaxValue,
                requeue: false,
                parkSink: null,
                publishCancellation.Token);
            var transferEntered = new TaskCompletionSource<bool>(
                TaskCreationOptions.RunContinuationsAsynchronously);
            var releaseTransfer = new TaskCompletionSource<bool>(
                TaskCreationOptions.RunContinuationsAsynchronously);
            m_nodeManagerMock
                .Setup(nodeManager => nodeManager.TransferMonitoredItemsAsync(
                    It.IsAny<OperationContext>(),
                    It.IsAny<bool>(),
                    It.IsAny<IList<IMonitoredItem>>(),
                    It.IsAny<IList<ServiceResult>>(),
                    It.IsAny<MonitoredItemTransferOptions>(),
                    It.IsAny<CancellationToken>()))
                .Callback(() => transferEntered.TrySetResult(true))
                .Returns(() => new ValueTask(releaseTransfer.Task));

            Task<TransferSubscriptionsResponse> transferTask = manager
                .TransferSubscriptionsAsync(
                    fixture.DestinationContext,
                    [subscription.Id],
                    sendInitialValues: true)
                .AsTask();
            Task entered = await Task.WhenAny(
                transferEntered.Task,
                Task.Delay(TimeSpan.FromSeconds(5))).ConfigureAwait(false);
            if (!ReferenceEquals(entered, transferEntered.Task))
            {
                releaseTransfer.TrySetResult(true);
                Assert.Fail("Transfer did not reach the monitored-item callback barrier.");
            }

            try
            {
                sourceQueue.PublishTimerExpired(staleSnapshot);

                Assert.Multiple(() =>
                {
                    Assert.That(
                        sourceQueue.CapturePublishTimerSnapshot(),
                        Is.Empty,
                        "The source entry must be claimed before callbacks run.");
                    Assert.That(
                        sourcePublish.IsCompleted,
                        Is.False,
                        "A stale source timer must not assign the claimed subscription.");
                    ServiceResultException error = Assert.Throws<ServiceResultException>(
                        () => subscription.Publish(
                            fixture.SourceContext,
                            out _,
                            out _))!;
                    Assert.That(
                        error!.StatusCode,
                        Is.EqualTo(StatusCodes.BadSubscriptionIdInvalid));
                });
            }
            finally
            {
                releaseTransfer.TrySetResult(true);
            }

            TransferSubscriptionsResponse transferred = await transferTask.ConfigureAwait(false);
            publishCancellation.Cancel();
            try
            {
                ISubscriptionPublishPipeline assigned = await sourcePublish.ConfigureAwait(false);
                Assert.That(assigned, Is.Not.SameAs(subscription));
            }
            catch (OperationCanceledException) when (publishCancellation.IsCancellationRequested)
            {
                // The old session's parked Publish is cancelled or completed by
                // the transferred-status notification, never with the subscription.
            }

            Assert.Multiple(() =>
            {
                Assert.That(transferred.Results, Has.Count.EqualTo(1));
                Assert.That(transferred.Results[0].StatusCode, Is.EqualTo(StatusCodes.Good));
                Assert.That(
                    subscription.Session,
                    Is.SameAs(fixture.DestinationSession.Object));
            });
            Assert.DoesNotThrow(
                () => subscription.ResendData(fixture.DestinationContext));
        }

        [Test]
        public async Task TransferInitialValueIsPublishedOnlyByDestinationAsync()
        {
            TransferFixture fixture = await CreateTransferSubscriptionAsync().ConfigureAwait(false);
            using SubscriptionManager manager = fixture.Manager;
            Subscription subscription = fixture.Subscription;
            using var queueFactory = new MonitoredItemQueueFactory(m_telemetry);
            m_serverMock.Setup(server => server.MonitoredItemQueueFactory).Returns(queueFactory);
            var itemOwner = new Mock<IAsyncNodeManager>();
            using var monitoredItem = new MonitoredItem(
                m_serverMock.Object,
                itemOwner.Object,
                new object(),
                subscription.Id,
                id: 10,
                new ReadValueId
                {
                    NodeId = new NodeId("TransferValue", 2),
                    AttributeId = Attributes.Value
                },
                DiagnosticsMasks.None,
                TimestampsToReturn.Both,
                MonitoringMode.Reporting,
                clientHandle: 11,
                originalFilter: null,
                filterToUse: null,
                range: null,
                samplingInterval: 1000,
                queueSize: 1,
                discardOldest: true,
                sourceSamplingInterval: 1000);
            await RegisterMonitoredItemsAsync(subscription, monitoredItem).ConfigureAwait(false);
            monitoredItem.QueueValue(new DataValue(new Variant(1234)), null);
            var initialNotifications = new Queue<MonitoredItemNotification>();
            monitoredItem.Publish(
                new OperationContext(monitoredItem),
                initialNotifications,
                new Queue<DiagnosticInfo>(),
                1,
                m_telemetry.CreateLogger<SubscriptionTests>());
            Assert.That(initialNotifications, Has.Count.EqualTo(1));

            var transferEntered = new TaskCompletionSource<bool>(
                TaskCreationOptions.RunContinuationsAsynchronously);
            var releaseTransfer = new TaskCompletionSource<bool>(
                TaskCreationOptions.RunContinuationsAsynchronously);
            itemOwner
                .Setup(nodeManager => nodeManager.TransferMonitoredItemsAsync(
                    It.IsAny<OperationContext>(),
                    It.IsAny<bool>(),
                    It.IsAny<IList<IMonitoredItem>>(),
                    It.IsAny<IList<bool>>(),
                    It.IsAny<IList<ServiceResult>>(),
                    It.IsAny<MonitoredItemTransferOptions>(),
                    It.IsAny<CancellationToken>()))
                .Returns<
                    OperationContext,
                    bool,
                    IList<IMonitoredItem>,
                    IList<bool>,
                    IList<ServiceResult>,
                    MonitoredItemTransferOptions,
                    CancellationToken>(async (_, sendInitialValues, monitoredItems, processedItems, errors, transferOptions, cancellationToken) =>
                    {
                        for (int ii = 0; ii < monitoredItems.Count; ii++)
                        {
                            if (processedItems[ii])
                            {
                                continue;
                            }

                            processedItems[ii] = true;
                            errors[ii] = ServiceResult.Good;
                            if (sendInitialValues && !transferOptions.DeferInitialValues)
                            {
                                monitoredItems[ii].SetupResendDataTrigger();
                            }
                        }
                        transferEntered.TrySetResult(true);
                        await releaseTransfer.Task.WaitAsync(cancellationToken).ConfigureAwait(false);
                    });
            var configurationNodeManager = new Mock<IConfigurationNodeManager>();
            configurationNodeManager
                .SetupGet(nodeManager => nodeManager.NamespaceUris)
                .Returns([]);
            var coreNodeManager = new Mock<ICoreNodeManager>();
            coreNodeManager
                .SetupGet(nodeManager => nodeManager.NamespaceUris)
                .Returns([]);
            var factory = new Mock<IMainNodeManagerFactory>();
            factory
                .Setup(nodeManagerFactory => nodeManagerFactory.CreateConfigurationNodeManager())
                .Returns(configurationNodeManager.Object);
            factory
                .Setup(nodeManagerFactory => nodeManagerFactory.CreateCoreNodeManager(It.IsAny<ushort>()))
                .Returns(coreNodeManager.Object);
            m_serverMock.Setup(server => server.MainNodeManagerFactory).Returns(factory.Object);
            using var masterNodeManager = new MasterNodeManager(
                m_serverMock.Object,
                new ApplicationConfiguration { ServerConfiguration = new ServerConfiguration() },
                null,
                itemOwner.Object);
            m_serverMock.Setup(server => server.NodeManager).Returns(masterNodeManager);

            Task<TransferSubscriptionsResponse> transferTask = manager
                .TransferSubscriptionsAsync(
                    fixture.DestinationContext,
                    [subscription.Id],
                    sendInitialValues: true)
                .AsTask();
            Task entered = await Task.WhenAny(
                transferEntered.Task,
                Task.Delay(TimeSpan.FromSeconds(5))).ConfigureAwait(false);
            if (!ReferenceEquals(entered, transferEntered.Task))
            {
                releaseTransfer.TrySetResult(true);
                Assert.Fail("Transfer did not reach the monitored-item callback barrier.");
            }

            try
            {
                itemOwner.Verify(
                    nodeManager => nodeManager.TransferMonitoredItemsAsync(
                        It.IsAny<OperationContext>(),
                        true,
                        It.IsAny<IList<IMonitoredItem>>(),
                        It.IsAny<IList<bool>>(),
                        It.IsAny<IList<ServiceResult>>(),
                        It.Is<MonitoredItemTransferOptions>(options => options.DeferInitialValues),
                        It.IsAny<CancellationToken>()),
                    Times.Once);
                Assert.That(monitoredItem.IsResendData, Is.False);
                ServiceResultException error = Assert.Throws<ServiceResultException>(
                    () => subscription.Publish(
                        fixture.SourceContext,
                        out _,
                        out _));
                Assert.That(error.StatusCode, Is.EqualTo(StatusCodes.BadSubscriptionIdInvalid));
            }
            finally
            {
                releaseTransfer.TrySetResult(true);
            }

            TransferSubscriptionsResponse transferred = await transferTask.ConfigureAwait(false);
            Assert.That(transferred.Results[0].StatusCode, Is.EqualTo(StatusCodes.Good));
            Assert.That(monitoredItem.IsResendData, Is.True);

            SessionPublishQueue destinationQueue = GetPublishQueue(
                manager,
                fixture.DestinationContext.SessionId);
            SetPrivateField(subscription, "m_waitingForPublish", false);
            SetPrivateField(subscription, "m_lifetimeCounter", 0u);
            SetExpiryTime(
                subscription,
                TimeProvider.System.GetTimestampMilliseconds() - 100);
            destinationQueue.PublishTimerExpired(
                destinationQueue.CapturePublishTimerSnapshot());
            ISubscription readySubscription = await destinationQueue.PublishAsync(
                "destination-channel",
                DateTime.MaxValue,
                requeue: false,
                parkSink: null,
                CancellationToken.None).ConfigureAwait(false);
            NotificationMessage message = readySubscription.Publish(
                fixture.DestinationContext,
                out _,
                out _)!;

            Assert.Multiple(() =>
            {
                Assert.That(readySubscription, Is.SameAs(subscription));
                Assert.That(message, Is.Not.Null);
                Assert.That(message!.NotificationData, Is.Not.Empty);
                Assert.That(monitoredItem.IsResendData, Is.False);
            });
        }

        [Test]
        public async Task TransferFallbackAppliesInitialValueExactlyOnceAsync()
        {
            TransferFixture fixture = await CreateTransferSubscriptionAsync().ConfigureAwait(false);
            using SubscriptionManager manager = fixture.Manager;
            Subscription subscription = fixture.Subscription;
            using var queueFactory = new MonitoredItemQueueFactory(m_telemetry);
            m_serverMock.Setup(server => server.MonitoredItemQueueFactory).Returns(queueFactory);
            var itemOwner = new Mock<IAsyncNodeManager>();
            using var monitoredItem = new MonitoredItem(
                m_serverMock.Object,
                itemOwner.Object,
                new object(),
                subscription.Id,
                id: 11,
                new ReadValueId
                {
                    NodeId = new NodeId("FallbackTransferValue", 2),
                    AttributeId = Attributes.Value
                },
                DiagnosticsMasks.None,
                TimestampsToReturn.Both,
                MonitoringMode.Reporting,
                clientHandle: 12,
                originalFilter: null,
                filterToUse: null,
                range: null,
                samplingInterval: 1000,
                queueSize: 1,
                discardOldest: true,
                sourceSamplingInterval: 1000);
            await RegisterMonitoredItemsAsync(subscription, monitoredItem).ConfigureAwait(false);
            monitoredItem.QueueValue(new DataValue(new Variant(5678)), null);
            var initialNotifications = new Queue<MonitoredItemNotification>();
            monitoredItem.Publish(
                new OperationContext(monitoredItem),
                initialNotifications,
                new Queue<DiagnosticInfo>(),
                1,
                m_telemetry.CreateLogger<SubscriptionTests>());
            Assert.That(initialNotifications, Has.Count.EqualTo(1));
            int setupResendCalls = 0;
            m_nodeManagerMock
                .Setup(nodeManager => nodeManager.TransferMonitoredItemsAsync(
                    It.IsAny<OperationContext>(),
                    true,
                    It.Is<IList<IMonitoredItem>>(items => items.Contains(monitoredItem)),
                    It.IsAny<IList<ServiceResult>>(),
                    It.Is<MonitoredItemTransferOptions>(options => !options.DeferInitialValues),
                    It.IsAny<CancellationToken>()))
                .Callback<OperationContext, bool, IList<IMonitoredItem>, IList<ServiceResult>, MonitoredItemTransferOptions, CancellationToken>(
                    (_, sendInitialValues, monitoredItems, errors, transferOptions, _) =>
                    {
                        Assert.That(transferOptions.DeferInitialValues, Is.False);
                        for (int ii = 0; ii < monitoredItems.Count; ii++)
                        {
                            errors[ii] = ServiceResult.Good;
                            if (sendInitialValues)
                            {
                                setupResendCalls++;
                                monitoredItems[ii].SetupResendDataTrigger();
                            }
                        }
                    })
                .Returns(default(ValueTask));

            TransferSubscriptionsResponse transferred = await manager
                .TransferSubscriptionsAsync(
                    fixture.DestinationContext,
                    [subscription.Id],
                    sendInitialValues: true)
                .ConfigureAwait(false);

            Assert.That(transferred.Results[0].StatusCode, Is.EqualTo(StatusCodes.Good));
            Assert.That(setupResendCalls, Is.EqualTo(1));
            SessionPublishQueue destinationQueue = GetPublishQueue(
                manager,
                fixture.DestinationContext.SessionId);
            SetPrivateField(subscription, "m_waitingForPublish", false);
            SetPrivateField(subscription, "m_lifetimeCounter", 0u);
            SetExpiryTime(
                subscription,
                TimeProvider.System.GetTimestampMilliseconds() - 100);
            destinationQueue.PublishTimerExpired(
                destinationQueue.CapturePublishTimerSnapshot());
            ISubscription readySubscription = await destinationQueue.PublishAsync(
                "destination-channel",
                DateTime.MaxValue,
                requeue: false,
                parkSink: null,
                CancellationToken.None).ConfigureAwait(false);
            NotificationMessage firstMessage = readySubscription.Publish(
                fixture.DestinationContext,
                out _,
                out _)!;
            NotificationMessage secondMessage = readySubscription.Publish(
                fixture.DestinationContext,
                out _,
                out _)!;

            Assert.Multiple(() =>
            {
                Assert.That(firstMessage!.NotificationData, Is.Not.Empty);
                Assert.That(secondMessage?.NotificationData ?? [], Is.Empty);
                Assert.That(monitoredItem.IsResendData, Is.False);
            });
        }

        [Test]
        public async Task SourcePublishTimerSnapshotDoesNotExpireTransferredSubscriptionAsync()
        {
            TransferFixture fixture = await CreateTransferSubscriptionAsync().ConfigureAwait(false);
            using SubscriptionManager manager = fixture.Manager;
            Subscription subscription = fixture.Subscription;
            ExpireOnNextPublishTimer(subscription);
            SessionPublishQueue sourceQueue = GetPublishQueue(
                manager,
                fixture.SourceContext.SessionId);
            var snapshotCaptured = new TaskCompletionSource<bool>(
                TaskCreationOptions.RunContinuationsAsynchronously);
            var releaseSnapshot = new TaskCompletionSource<bool>(
                TaskCreationOptions.RunContinuationsAsynchronously);
            var staleTimer = Task.Run(
                async () =>
                {
                    IReadOnlyList<SessionPublishQueue.QueuedSubscription> snapshot =
                        sourceQueue.CapturePublishTimerSnapshot();
                    snapshotCaptured.TrySetResult(true);
                    await releaseSnapshot.Task.ConfigureAwait(false);
                    sourceQueue.PublishTimerExpired(snapshot);
                });

            Task captured = await Task.WhenAny(
                snapshotCaptured.Task,
                Task.Delay(TimeSpan.FromSeconds(5))).ConfigureAwait(false);
            if (!ReferenceEquals(captured, snapshotCaptured.Task))
            {
                releaseSnapshot.TrySetResult(true);
                Assert.Fail("Publish timer did not capture the source queue.");
            }

            TransferSubscriptionsResponse transferred;
            try
            {
                transferred = await manager.TransferSubscriptionsAsync(
                    fixture.DestinationContext,
                    [subscription.Id],
                    sendInitialValues: false).ConfigureAwait(false);
            }
            finally
            {
                releaseSnapshot.TrySetResult(true);
            }
            await staleTimer.ConfigureAwait(false);

            Assert.Multiple(() =>
            {
                Assert.That(transferred.Results, Has.Count.EqualTo(1));
                Assert.That(transferred.Results[0].StatusCode, Is.EqualTo(StatusCodes.Good));
                Assert.That(
                    subscription.Session,
                    Is.SameAs(fixture.DestinationSession.Object));
                Assert.That(
                    manager.TryGetSubscription(subscription.Id, out _),
                    Is.True);
            });
            Assert.DoesNotThrow(
                () => subscription.ResendData(fixture.DestinationContext));
        }

        [Test]
        public async Task FirstPublishAfterTransferOfAbandonedSubscriptionReturnsQueuedDataAsync()
        {
            TransferFixture fixture = await CreateTransferSubscriptionAsync().ConfigureAwait(false);
            using SubscriptionManager manager = fixture.Manager;
            Subscription subscription = fixture.Subscription;
            using var queueFactory = new MonitoredItemQueueFactory(m_telemetry);
            m_serverMock.Setup(server => server.MonitoredItemQueueFactory).Returns(queueFactory);
            var itemOwner = new Mock<IAsyncNodeManager>();
            using var monitoredItem = new MonitoredItem(
                m_serverMock.Object,
                itemOwner.Object,
                new object(),
                subscription.Id,
                id: 12,
                new ReadValueId
                {
                    NodeId = new NodeId("AbandonedTransferValue", 2),
                    AttributeId = Attributes.Value
                },
                DiagnosticsMasks.None,
                TimestampsToReturn.Both,
                MonitoringMode.Reporting,
                clientHandle: 13,
                originalFilter: null,
                filterToUse: null,
                range: null,
                samplingInterval: 1000,
                queueSize: 10,
                discardOldest: true,
                sourceSamplingInterval: 1000);
            using var linkedItem = new MonitoredItem(
                m_serverMock.Object,
                itemOwner.Object,
                new object(),
                subscription.Id,
                id: 14,
                new ReadValueId
                {
                    NodeId = new NodeId("AbandonedTransferLinkedValue", 2),
                    AttributeId = Attributes.Value
                },
                DiagnosticsMasks.None,
                TimestampsToReturn.Both,
                MonitoringMode.Sampling,
                clientHandle: 15,
                originalFilter: null,
                filterToUse: null,
                range: null,
                samplingInterval: 1000,
                queueSize: 10,
                discardOldest: true,
                sourceSamplingInterval: 1000);
            await RegisterMonitoredItemsAsync(subscription, monitoredItem, linkedItem).ConfigureAwait(false);
            subscription.SetTriggering(
                fixture.SourceContext,
                monitoredItem.Id,
                [linkedItem.Id],
                [],
                out ArrayOf<StatusCode> addResults,
                out _,
                out _,
                out _);
            Assert.That(addResults[0], Is.EqualTo(StatusCodes.Good));
            m_nodeManagerMock
                .Setup(nodeManager => nodeManager.TransferMonitoredItemsAsync(
                    It.IsAny<OperationContext>(),
                    It.IsAny<bool>(),
                    It.IsAny<IList<IMonitoredItem>>(),
                    It.IsAny<IList<ServiceResult>>(),
                    It.IsAny<MonitoredItemTransferOptions>(),
                    It.IsAny<CancellationToken>()))
                .Callback<OperationContext, bool, IList<IMonitoredItem>, IList<ServiceResult>, MonitoredItemTransferOptions, CancellationToken>(
                    (_, _, monitoredItems, errors, _, _) =>
                    {
                        for (int ii = 0; ii < monitoredItems.Count; ii++)
                        {
                            errors[ii] = ServiceResult.Good;
                        }
                    })
                .Returns(default(ValueTask));

            // The client closes its Session without deleting the (durable) subscription.
            m_sessionMock.SetupGet(session => session.IsClosing).Returns(true);
            await manager.SessionClosingAsync(
                    fixture.SourceContext,
                    fixture.SourceContext.SessionId,
                    deleteSubscriptions: false,
                    CancellationToken.None)
                .ConfigureAwait(false);

            // Values are sampled while no Session owns the subscription, and the publish
            // timer keeps running until the keep-alive is due.
            // The triggering item only becomes ready to trigger with its second value.
            linkedItem.QueueValue(new DataValue(new Variant(7)), null);
            monitoredItem.QueueValue(new DataValue(new Variant(41)), null);
            monitoredItem.QueueValue(new DataValue(new Variant(42)), null);
            uint maxKeepAliveCount = GetPrivateField<uint>(subscription, "m_maxKeepAliveCount");
            for (uint ii = 0; ii < maxKeepAliveCount; ii++)
            {
                SetExpiryTime(
                    subscription,
                    TimeProvider.System.GetTimestampMilliseconds() - 100);
                manager.ProcessAbandonedPublishTimers(
                    manager.CaptureAbandonedPublishTimerSnapshot());
            }
            Assert.That(GetPrivateField<bool>(subscription, "m_waitingForPublish"), Is.True);

            TransferSubscriptionsResponse transferred = await manager.TransferSubscriptionsAsync(
                fixture.DestinationContext,
                [subscription.Id],
                sendInitialValues: false).ConfigureAwait(false);
            Assert.That(transferred.Results[0].StatusCode, Is.EqualTo(StatusCodes.Good));

            // OPC 10000-4 §5.14.1.1: a keep-alive is only sent when no notifications are
            // available; the values buffered while the client was away, including the value
            // of the linked Sampling item (§5.13.5), must be returned.
            NotificationMessage message = subscription.Publish(
                fixture.DestinationContext,
                out _,
                out _)!;

            Assert.That(message, Is.Not.Null);
            Assert.That(message.NotificationData, Has.Count.EqualTo(1));
            Assert.That(
                message.NotificationData[0].TryGetValue(out DataChangeNotification? dataChange),
                Is.True);
            var clientHandles = new List<uint>();
            foreach (MonitoredItemNotification notification in dataChange!.MonitoredItems)
            {
                clientHandles.Add(notification.ClientHandle);
            }
            Assert.That(clientHandles, Does.Contain(13u), "Triggering item values");
            Assert.That(clientHandles, Does.Contain(15u), "Linked Sampling item value");
        }

        [Test]
        public async Task AbandonedTimerSnapshotDoesNotExpireTransferredSubscriptionAsync()
        {
            TransferFixture fixture = await CreateTransferSubscriptionAsync().ConfigureAwait(false);
            using SubscriptionManager manager = fixture.Manager;
            Subscription subscription = fixture.Subscription;
            m_sessionMock.SetupGet(session => session.IsClosing).Returns(true);
            await manager.SessionClosingAsync(
                    fixture.SourceContext,
                    fixture.SourceContext.SessionId,
                    deleteSubscriptions: false,
                    CancellationToken.None)
                .ConfigureAwait(false);
            ExpireOnNextPublishTimer(subscription);

            var snapshotCaptured = new TaskCompletionSource<bool>(
                TaskCreationOptions.RunContinuationsAsynchronously);
            var releaseSnapshot = new TaskCompletionSource<bool>(
                TaskCreationOptions.RunContinuationsAsynchronously);
            var staleTimer = Task.Run(
                async () =>
                {
                    IReadOnlyList<ISubscriptionPublishPipeline> snapshot =
                        manager.CaptureAbandonedPublishTimerSnapshot();
                    snapshotCaptured.TrySetResult(true);
                    await releaseSnapshot.Task.ConfigureAwait(false);
                    manager.ProcessAbandonedPublishTimers(snapshot);
                });

            Task captured = await Task.WhenAny(
                snapshotCaptured.Task,
                Task.Delay(TimeSpan.FromSeconds(5))).ConfigureAwait(false);
            if (!ReferenceEquals(captured, snapshotCaptured.Task))
            {
                releaseSnapshot.TrySetResult(true);
                Assert.Fail("Publish timer did not capture the abandoned subscription.");
            }

            TransferSubscriptionsResponse transferred;
            try
            {
                transferred = await manager.TransferSubscriptionsAsync(
                    fixture.DestinationContext,
                    [subscription.Id],
                    sendInitialValues: false).ConfigureAwait(false);
            }
            finally
            {
                releaseSnapshot.TrySetResult(true);
            }
            await staleTimer.ConfigureAwait(false);
            ConcurrentDictionary<uint, ISubscriptionPublishPipeline> abandonedSubscriptions =
                GetPrivateField<ConcurrentDictionary<uint, ISubscriptionPublishPipeline>>(
                    manager,
                    "m_abandonedSubscriptions");

            Assert.Multiple(() =>
            {
                Assert.That(transferred.Results, Has.Count.EqualTo(1));
                Assert.That(transferred.Results[0].StatusCode, Is.EqualTo(StatusCodes.Good));
                Assert.That(
                    subscription.Session,
                    Is.SameAs(fixture.DestinationSession.Object));
                Assert.That(abandonedSubscriptions, Is.Empty);
                Assert.That(
                    manager.TryGetSubscription(subscription.Id, out _),
                    Is.True);
            });
            Assert.DoesNotThrow(
                () => subscription.ResendData(fixture.DestinationContext));
        }

        [TestCase(false)]
        [TestCase(true)]
        public async Task CurrentOwnerPublishTimerStillExpiresSubscriptionAsync(
            bool abandonBeforeExpiration)
        {
            TransferFixture fixture = await CreateTransferSubscriptionAsync().ConfigureAwait(false);
            using SubscriptionManager manager = fixture.Manager;
            Subscription subscription = fixture.Subscription;
            SessionPublishQueue sourceQueue = GetPublishQueue(
                manager,
                fixture.SourceContext.SessionId);
            if (abandonBeforeExpiration)
            {
                m_sessionMock.SetupGet(session => session.IsClosing).Returns(true);
                await manager.SessionClosingAsync(
                        fixture.SourceContext,
                        fixture.SourceContext.SessionId,
                        deleteSubscriptions: false,
                        CancellationToken.None)
                    .ConfigureAwait(false);
            }
            var deletionCompleted = new TaskCompletionSource<bool>(
                TaskCreationOptions.RunContinuationsAsynchronously);
            m_serverMock
                .Setup(server => server.DeleteSubscriptionAsync(
                    subscription.Id,
                    It.IsAny<CancellationToken>()))
                .Returns((uint subscriptionId, CancellationToken cancellationToken) =>
                    DeleteSubscriptionAndSignalAsync(
                        manager,
                        subscriptionId,
                        deletionCompleted,
                        cancellationToken));
            ExpireOnNextPublishTimer(subscription);

            if (abandonBeforeExpiration)
            {
                manager.ProcessAbandonedPublishTimers(
                    manager.CaptureAbandonedPublishTimerSnapshot());
            }
            else
            {
                sourceQueue.PublishTimerExpired(
                    sourceQueue.CapturePublishTimerSnapshot());
            }

            ServiceResultException exception = Assert.Throws<ServiceResultException>(
                () => subscription.ResendData(fixture.SourceContext));
            Assert.That(
                exception.StatusCode,
                Is.EqualTo(StatusCodes.BadSubscriptionIdInvalid));
            Task deleted = await Task.WhenAny(
                deletionCompleted.Task,
                Task.Delay(TimeSpan.FromSeconds(5))).ConfigureAwait(false);
            Assert.That(
                deleted,
                Is.SameAs(deletionCompleted.Task),
                "Expired subscription cleanup did not complete.");
            Assert.That(
                manager.TryGetSubscription(subscription.Id, out _),
                Is.False);
        }

        [TestCase(false)]
        [TestCase(true)]
        public async Task FailedTransferRestoresClaimedExpirationSourceAsync(
            bool abandonBeforeTransfer)
        {
            TransferFixture fixture = await CreateTransferSubscriptionAsync().ConfigureAwait(false);
            using SubscriptionManager manager = fixture.Manager;
            Subscription subscription = fixture.Subscription;
            SessionPublishQueue sourceQueue = GetPublishQueue(
                manager,
                fixture.SourceContext.SessionId);
            if (abandonBeforeTransfer)
            {
                m_sessionMock.SetupGet(session => session.IsClosing).Returns(true);
                await manager.SessionClosingAsync(
                        fixture.SourceContext,
                        fixture.SourceContext.SessionId,
                        deleteSubscriptions: false,
                        CancellationToken.None)
                    .ConfigureAwait(false);
            }
            int transferCalls = 0;
            m_nodeManagerMock
                .Setup(nodeManager => nodeManager.TransferMonitoredItemsAsync(
                    It.IsAny<OperationContext>(),
                    It.IsAny<bool>(),
                    It.IsAny<IList<IMonitoredItem>>(),
                    It.IsAny<IList<ServiceResult>>(),
                    It.IsAny<MonitoredItemTransferOptions>(),
                    It.IsAny<CancellationToken>()))
                .Returns(
                    () => Interlocked.Increment(ref transferCalls) == 1
                        ? new ValueTask(
                            Task.FromException(
                                new ServiceResultException(
                                    StatusCodes.BadUnexpectedError)))
                        : new ValueTask());

            TransferSubscriptionsResponse failed = await manager
                .TransferSubscriptionsAsync(
                    fixture.DestinationContext,
                    [subscription.Id],
                    sendInitialValues: false)
                .ConfigureAwait(false);

            Assert.Multiple(() =>
            {
                Assert.That(failed.Results, Has.Count.EqualTo(1));
                Assert.That(
                    ServiceResult.IsBad(failed.Results[0].StatusCode),
                    Is.True);
                if (abandonBeforeTransfer)
                {
                    Assert.That(subscription.Session, Is.Null);
                    ConcurrentDictionary<uint, ISubscriptionPublishPipeline> abandonedSubscriptions =
                        GetPrivateField<ConcurrentDictionary<uint, ISubscriptionPublishPipeline>>(
                            manager,
                            "m_abandonedSubscriptions");
                    Assert.That(
                        abandonedSubscriptions.TryGetValue(
                            subscription.Id,
                            out ISubscriptionPublishPipeline? restoredSubscription),
                        Is.True);
                    Assert.That(restoredSubscription, Is.SameAs(subscription));
                }
                else
                {
                    IReadOnlyList<SessionPublishQueue.QueuedSubscription> restoredSource =
                        sourceQueue.CapturePublishTimerSnapshot();
                    Assert.That(subscription.Session, Is.SameAs(m_sessionMock.Object));
                    Assert.That(restoredSource, Has.Count.EqualTo(1));
                    Assert.That(
                        restoredSource[0].Subscription,
                        Is.SameAs(subscription));
                }
            });

            TransferSubscriptionsResponse retried = await manager
                .TransferSubscriptionsAsync(
                    fixture.DestinationContext,
                    [subscription.Id],
                    sendInitialValues: false)
                .ConfigureAwait(false);

            Assert.That(retried.Results, Has.Count.EqualTo(1));
            Assert.That(retried.Results[0].StatusCode, Is.EqualTo(StatusCodes.Good));
            Assert.DoesNotThrow(
                () => subscription.ResendData(fixture.DestinationContext));
        }

        [Test]
        public async Task FailedTransferPreservesDestinationQueuedPublishRequestsAsync()
        {
            TransferFixture fixture = await CreateTransferSubscriptionAsync().ConfigureAwait(false);
            using SubscriptionManager manager = fixture.Manager;
            Subscription subscription = fixture.Subscription;
            fixture.DestinationSession
                .Setup(session => session.IsSecureChannelValid("destination-channel"))
                .Returns(true);
            CreateSubscriptionResponse destinationCreated = await manager
                .CreateSubscriptionAsync(
                    fixture.DestinationContext,
                    requestedPublishingInterval: 1000,
                    requestedLifetimeCount: 30,
                    requestedMaxKeepAliveCount: 10,
                    maxNotificationsPerPublish: 0,
                    publishingEnabled: true,
                    priority: 0)
                .ConfigureAwait(false);
            Assert.That(
                manager.TryGetSubscription(
                    destinationCreated.SubscriptionId,
                    out ISubscription? destinationSubscription),
                Is.True);
            Assert.That(destinationSubscription, Is.Not.Null);
            SessionPublishQueue destinationQueue = GetPublishQueue(
                manager,
                fixture.DestinationContext.SessionId);
            Task<ISubscriptionPublishPipeline> destinationPublish = destinationQueue.PublishAsync(
                "destination-channel",
                DateTime.MaxValue,
                requeue: false,
                parkSink: null,
                CancellationToken.None);
            Assert.That(destinationPublish.IsCompleted, Is.False);
            m_nodeManagerMock
                .Setup(nodeManager => nodeManager.TransferMonitoredItemsAsync(
                    It.IsAny<OperationContext>(),
                    It.IsAny<bool>(),
                    It.IsAny<IList<IMonitoredItem>>(),
                    It.IsAny<IList<ServiceResult>>(),
                    It.IsAny<MonitoredItemTransferOptions>(),
                    It.IsAny<CancellationToken>()))
                .Returns(new ValueTask(Task.FromException(
                    new ServiceResultException(StatusCodes.BadUnexpectedError))));

            TransferSubscriptionsResponse failed = await manager
                .TransferSubscriptionsAsync(
                    fixture.DestinationContext,
                    [subscription.Id],
                    sendInitialValues: false)
                .ConfigureAwait(false);

            Assert.Multiple(() =>
            {
                Assert.That(failed.Results, Has.Count.EqualTo(1));
                Assert.That(ServiceResult.IsBad(failed.Results[0].StatusCode), Is.True);
                Assert.That(destinationPublish.IsCompleted, Is.False);
            });

            destinationQueue.PublishCompleted(
                destinationSubscription!.AsPipeline(),
                moreNotifications: true);

            ISubscription publishedSubscription = await destinationPublish.ConfigureAwait(false);
            Assert.That(publishedSubscription, Is.SameAs(destinationSubscription));
        }

        [Test]
        public void TryClaimForTransferFailsWhenTheClaimingSessionIsNotTheOwner()
        {
            using Subscription subscription = CreateSubscription();
            using var queue = new SessionPublishQueue(
                m_serverMock.Object,
                m_sessionMock.Object,
                maxPublishRequests: 10);
            queue.Add(subscription);
            var staleSession = new Mock<ISession>();
            staleSession.Setup(session => session.Id).Returns(new NodeId(Guid.NewGuid()));

            bool claimed = queue.TryClaimForTransfer(
                subscription,
                staleSession.Object,
                out SessionPublishQueue.SubscriptionTransferClaim? claim);

            Assert.Multiple(() =>
            {
                Assert.That(claimed, Is.False);
                Assert.That(claim, Is.Null);
                Assert.That(
                    queue.ContainsSubscription(subscription),
                    Is.True,
                    "A refused claim must leave the subscription publishable by its owner.");
            });

            Assert.That(
                queue.TryClaimForTransfer(
                    subscription,
                    m_sessionMock.Object,
                    out SessionPublishQueue.SubscriptionTransferClaim? ownerClaim),
                Is.True,
                "The refused claim must not block the real owner from starting a transfer.");
            queue.CompleteTransferClaim(ownerClaim!);
            subscription.AbortTransfer(m_sessionMock.Object);
        }

        [Test]
        public void RestoreTransferClaimFailsWhenTheClaimWasAlreadyRestored()
        {
            using Subscription subscription = CreateSubscription();
            using var queue = new SessionPublishQueue(
                m_serverMock.Object,
                m_sessionMock.Object,
                maxPublishRequests: 10);
            queue.Add(subscription);
            Assert.That(
                queue.TryClaimForTransfer(
                    subscription,
                    m_sessionMock.Object,
                    out SessionPublishQueue.SubscriptionTransferClaim? claim),
                Is.True);

            Assert.That(queue.RestoreTransferClaim(claim!), Is.True);

            Assert.Multiple(() =>
            {
                Assert.That(
                    queue.RestoreTransferClaim(claim!),
                    Is.False,
                    "A claim may only be restored once, otherwise a later queue entry is overwritten.");
                Assert.That(queue.ContainsSubscription(subscription), Is.True);
            });
            subscription.AbortTransfer(m_sessionMock.Object);
        }

        [Test]
        public void TryRemoveForTransferRemovesOnlyTheExactQueuedEntry()
        {
            using Subscription subscription = CreateSubscription();
            using var queue = new SessionPublishQueue(
                m_serverMock.Object,
                m_sessionMock.Object,
                maxPublishRequests: 10);
            queue.Add(subscription);
            var impostor = new Mock<ISubscriptionPublishPipeline>();
            impostor.Setup(candidate => candidate.Id).Returns(subscription.Id);

            Assert.Multiple(() =>
            {
                Assert.That(
                    queue.TryRemoveForTransfer(impostor.Object),
                    Is.False,
                    "A different subscription instance with the same id must not remove the entry.");
                Assert.That(queue.ContainsSubscription(subscription), Is.True);
            });

            Assert.That(queue.TryRemoveForTransfer(subscription), Is.True);

            Assert.Multiple(() =>
            {
                Assert.That(queue.ContainsSubscription(subscription), Is.False);
                Assert.That(
                    queue.TryRemoveForTransfer(subscription),
                    Is.False,
                    "A subscription that is no longer queued cannot be removed a second time.");
            });
        }

        [Test]
        public async Task PublishCompletedKeepsNotificationsOfAClaimedSubscriptionAsync()
        {
            using Subscription subscription = CreateSubscription();
            using var queue = new SessionPublishQueue(
                m_serverMock.Object,
                m_sessionMock.Object,
                maxPublishRequests: 10);
            queue.Add(subscription);
            Assert.That(
                queue.TryClaimForTransfer(
                    subscription,
                    m_sessionMock.Object,
                    out SessionPublishQueue.SubscriptionTransferClaim? claim),
                Is.True);

            queue.PublishCompleted(subscription, moreNotifications: true);

            Assert.Multiple(() =>
            {
                Assert.That(claim!.Entry.Publishing, Is.False);
                Assert.That(
                    claim.Entry.ReadyToPublish,
                    Is.True,
                    "A notification raised while the entry is claimed must survive an abandoned transfer.");
            });

            Assert.That(queue.RestoreTransferClaim(claim!), Is.True);
            Task<ISubscriptionPublishPipeline> publish = queue.PublishAsync(
                "channel1",
                DateTime.MaxValue,
                requeue: false,
                parkSink: null,
                CancellationToken.None);

            Assert.That(publish.Status, Is.EqualTo(TaskStatus.RanToCompletion));
            Assert.That(await publish.ConfigureAwait(false), Is.SameAs(subscription));
            subscription.AbortTransfer(m_sessionMock.Object);
        }

        [Test]
        public void TryBeginTransferFailsWhileATransferIsAlreadyInProgress()
        {
            using Subscription subscription = CreateSubscription();

            Assert.That(subscription.TryBeginTransfer(m_sessionMock.Object), Is.True);

            Assert.That(
                subscription.TryBeginTransfer(m_sessionMock.Object),
                Is.False,
                "Only one transfer may reserve a subscription at a time.");

            subscription.AbortTransfer(m_sessionMock.Object);
            Assert.That(
                subscription.TryBeginTransfer(m_sessionMock.Object),
                Is.True,
                "An aborted transfer must release the reservation.");
            subscription.AbortTransfer(m_sessionMock.Object);
        }

        [Test]
        public void PublishTimerIsIdleWhileATransferIsInProgress()
        {
            using Subscription subscription = CreateSubscription();
            ExpireOnNextPublishTimer(subscription);
            Assert.That(subscription.TryBeginTransfer(m_sessionMock.Object), Is.True);

            PublishingState state = subscription.PublishTimerExpired();

            Assert.That(
                state,
                Is.EqualTo(PublishingState.Idle),
                "A reserved subscription must not expire on the source session timer.");
            subscription.AbortTransfer(m_sessionMock.Object);
        }

        [Test]
        public void PrepareSessionTransferAsyncRejectsAnUnreservedSubscription()
        {
            using Subscription subscription = CreateSubscription();
            var context = new OperationContext(
                m_sessionMock.Object,
                DiagnosticsMasks.None);

            ServiceResultException exception = Assert.ThrowsAsync<ServiceResultException>(
                async () => await subscription
                        .PrepareSessionTransferAsync(
                            context,
                            m_sessionMock.Object,
                            sendInitialValues: false,
                            CancellationToken.None)
                        .ConfigureAwait(false));

            Assert.That(
                exception.StatusCode,
                Is.EqualTo(StatusCodes.BadSubscriptionIdInvalid));
        }

        [Test]
        public void CompleteTransferRejectsAnUnexpectedOwner()
        {
            using Subscription subscription = CreateSubscription();

            ServiceResultException exception = Assert.Throws<ServiceResultException>(
                () => subscription.CompleteTransfer(m_sessionMock.Object));

            Assert.That(
                exception.StatusCode,
                Is.EqualTo(StatusCodes.BadSubscriptionIdInvalid));
        }

        [Test]
        public async Task PreparedSessionTransferMovesOwnershipToTheDestinationAsync()
        {
            using Subscription subscription = CreateSubscription();
            var destinationSession = new Mock<ISession>();
            destinationSession.Setup(session => session.Id).Returns(new NodeId(Guid.NewGuid()));
            SetSessionIdentity(
                destinationSession,
                new UserNameIdentityTokenHandler("transfer-user", [1, 2, 3]));
            var context = new OperationContext(
                destinationSession.Object,
                DiagnosticsMasks.None);

            Assert.That(subscription.TryBeginTransfer(m_sessionMock.Object), Is.True);
            Subscription.PreparedSessionTransfer prepared = await subscription
                .PrepareSessionTransferAsync(
                    context,
                    m_sessionMock.Object,
                    sendInitialValues: true,
                    CancellationToken.None)
                .ConfigureAwait(false);
            prepared.CommitOwnership();
            prepared.CommitMonitoredItemEffects();
            subscription.CompleteTransfer(destinationSession.Object);
            prepared.Complete();

            Assert.That(subscription.Session, Is.SameAs(destinationSession.Object));
        }

        [Test]
        public async Task FailedTransferRestoresDiagnosticsLinksAndSourcePermissionsAsync()
        {
            m_serverMock.SetupGet(server => server.CoreNodeManager).Returns(Mock.Of<ICoreNodeManager>());
            m_serverMock.SetupGet(server => server.SubscriptionManager).Returns(Mock.Of<ISubscriptionManager>());
            m_serverMock.SetupGet(server => server.MessageContext).Returns(ServiceMessageContext.Create(m_telemetry));
            using var diagnostics = new DiagnosticsNodeManager(m_serverMock.Object,
                new ApplicationConfiguration { ServerConfiguration = new ServerConfiguration() },
                m_telemetry.CreateLogger<DiagnosticsNodeManager>());
            m_serverMock.SetupGet(server => server.DiagnosticsNodeManager).Returns(diagnostics);
            await diagnostics.CreateAddressSpaceAsync(new Dictionary<NodeId, IList<IReference>>())
                .ConfigureAwait(false);

            static ServiceResult Update(ISystemContext context, NodeState node, ref Variant value) => ServiceResult.Good;

            NodeId sourceId = await diagnostics.CreateSessionDiagnosticsAsync(
                diagnostics.SystemContext, new SessionDiagnosticsDataType { SessionName = "source" }, Update,
                new SessionSecurityDiagnosticsDataType(), Update).ConfigureAwait(false);
            NodeId destinationId = await diagnostics.CreateSessionDiagnosticsAsync(
                diagnostics.SystemContext, new SessionDiagnosticsDataType { SessionName = "destination" }, Update,
                new SessionSecurityDiagnosticsDataType(), Update).ConfigureAwait(false);
            m_sessionMock.SetupGet(session => session.Id).Returns(sourceId);
            var destination = new Mock<ISession>();
            destination.SetupGet(session => session.Id).Returns(destinationId);
            using Subscription subscription = CreateSubscription();
            SessionDiagnosticsObjectState sourceNode =
                diagnostics.FindPredefinedNode<SessionDiagnosticsObjectState>(sourceId);
            SessionDiagnosticsObjectState destinationNode =
                diagnostics.FindPredefinedNode<SessionDiagnosticsObjectState>(destinationId);
            var sourceArray = (SubscriptionDiagnosticsArrayState)sourceNode.CreateChild(
                diagnostics.SystemContext, QualifiedName.From(BrowseNames.SubscriptionDiagnosticsArray))!;
            var destinationArray = (SubscriptionDiagnosticsArrayState)destinationNode.CreateChild(
                diagnostics.SystemContext, QualifiedName.From(BrowseNames.SubscriptionDiagnosticsArray))!;
            var references = new List<IReference>();
            sourceArray!.GetReferences(diagnostics.SystemContext, references, ReferenceTypeIds.HasComponent, false);
            Assert.That(references, Has.Count.EqualTo(1));
            var diagnosticsId = ExpandedNodeId.ToNodeId(references[0].TargetId, m_serverMock.Object.NamespaceUris);
            SubscriptionDiagnosticsState diagnosticsNode =
                diagnostics.FindPredefinedNode<SubscriptionDiagnosticsState>(diagnosticsId);
            var transaction = new Mock<IMonitoredItemTransferTransaction>();
            transaction.Setup(value => value.Commit())
                .Throws(new ServiceResultException(StatusCodes.BadUnexpectedError));
            Assert.That(subscription.TryBeginTransfer(m_sessionMock.Object), Is.True);
            var prepared = new Subscription.PreparedSessionTransfer(
                subscription, m_sessionMock.Object, destination.Object, transaction.Object);

            prepared.CommitOwnership();
            Assert.That(sourceArray.ReferenceExists(ReferenceTypeIds.HasComponent, false, diagnosticsId), Is.False);
            Assert.That(destinationArray!.ReferenceExists(ReferenceTypeIds.HasComponent, false, diagnosticsId), Is.True);
            Assert.That(ReadPermissions(m_sessionMock.Object), Is.Zero);
            Assert.That(ReadPermissions(destination.Object), Is.Not.Zero);
            ServiceResultException error = Assert.Throws<ServiceResultException>(prepared.CommitMonitoredItemEffects);
            Assert.That(error.StatusCode, Is.EqualTo(StatusCodes.BadUnexpectedError));

            await prepared.RollbackAsync(CancellationToken.None).ConfigureAwait(false);

            Assert.Multiple(() =>
            {
                Assert.That(subscription.Session, Is.SameAs(m_sessionMock.Object));
                Assert.That(subscription.Diagnostics.SessionId, Is.EqualTo(sourceId));
                Assert.That(sourceArray.ReferenceExists(ReferenceTypeIds.HasComponent, false, diagnosticsId), Is.True);
                Assert.That(destinationArray.ReferenceExists(ReferenceTypeIds.HasComponent, false, diagnosticsId),
                    Is.False);
                Assert.That(diagnosticsNode.ReferenceExists(ReferenceTypeIds.HasComponent, true, sourceArray.NodeId),
                    Is.True);
                Assert.That(diagnosticsNode.ReferenceExists(
                    ReferenceTypeIds.HasComponent, true, destinationArray.NodeId),
                    Is.False);
                Assert.That(ReadPermissions(m_sessionMock.Object), Is.Not.Zero);
                Assert.That(ReadPermissions(destination.Object), Is.Zero);
            });
            subscription.AbortTransfer(m_sessionMock.Object);

            uint ReadPermissions(ISession session)
            {
                ServerSystemContext context = m_serverMock.Object.DefaultSystemContext.Copy(session);
                ArrayOf<RolePermissionType> permissions = default;
                Assert.That(diagnosticsNode.OnReadUserRolePermissions, Is.Not.Null);
                diagnosticsNode.OnReadUserRolePermissions(context, diagnosticsNode, ref permissions);
                Assert.That(permissions, Is.Not.Empty);
                return permissions[0].Permissions & (uint)PermissionType.Read;
            }
        }

        [Test]
        public async Task TransferFailsWhenTheSourceQueueEntryIsAlreadyClaimedAsync()
        {
            TransferFixture fixture = await CreateTransferSubscriptionAsync().ConfigureAwait(false);
            using SubscriptionManager manager = fixture.Manager;
            Subscription subscription = fixture.Subscription;
            SessionPublishQueue sourceQueue = GetPublishQueue(
                manager,
                fixture.SourceContext.SessionId);
            Assert.That(
                sourceQueue.TryClaimForTransfer(
                    subscription,
                    m_sessionMock.Object,
                    out SessionPublishQueue.SubscriptionTransferClaim? claim),
                Is.True);
            var diagnosticsContext = new OperationContext(
                fixture.DestinationSession.Object,
                DiagnosticsMasks.OperationAll);

            TransferSubscriptionsResponse failed = await manager
                .TransferSubscriptionsAsync(
                    diagnosticsContext,
                    [subscription.Id],
                    sendInitialValues: false)
                .ConfigureAwait(false);

            Assert.Multiple(() =>
            {
                Assert.That(failed.Results, Has.Count.EqualTo(1));
                Assert.That(
                    failed.Results[0].StatusCode,
                    Is.EqualTo(StatusCodes.BadSubscriptionIdInvalid));
                Assert.That(failed.DiagnosticInfos, Has.Count.EqualTo(1));
                Assert.That(
                    subscription.Session,
                    Is.SameAs(m_sessionMock.Object),
                    "A refused claim must leave ownership with the source session.");
            });

            Assert.That(sourceQueue.RestoreTransferClaim(claim!), Is.True);
            subscription.AbortTransfer(m_sessionMock.Object);
            TransferSubscriptionsResponse retried = await manager
                .TransferSubscriptionsAsync(
                    fixture.DestinationContext,
                    [subscription.Id],
                    sendInitialValues: false)
                .ConfigureAwait(false);

            Assert.That(retried.Results[0].StatusCode, Is.EqualTo(StatusCodes.Good));
        }

        [Test]
        public async Task TransferFailsWhenTheAbandonedSubscriptionIsAlreadyReservedAsync()
        {
            TransferFixture fixture = await CreateTransferSubscriptionAsync().ConfigureAwait(false);
            using SubscriptionManager manager = fixture.Manager;
            Subscription subscription = fixture.Subscription;
            m_sessionMock.SetupGet(session => session.IsClosing).Returns(true);
            await manager.SessionClosingAsync(
                    fixture.SourceContext,
                    fixture.SourceContext.SessionId,
                    deleteSubscriptions: false,
                    CancellationToken.None)
                .ConfigureAwait(false);
            Assert.That(subscription.Session, Is.Null);
            Assert.That(subscription.TryBeginTransfer(null), Is.True);
            var diagnosticsContext = new OperationContext(
                fixture.DestinationSession.Object,
                DiagnosticsMasks.OperationAll);

            TransferSubscriptionsResponse failed = await manager
                .TransferSubscriptionsAsync(
                    diagnosticsContext,
                    [subscription.Id],
                    sendInitialValues: false)
                .ConfigureAwait(false);

            ConcurrentDictionary<uint, ISubscriptionPublishPipeline> abandonedSubscriptions =
                GetPrivateField<ConcurrentDictionary<uint, ISubscriptionPublishPipeline>>(
                    manager,
                    "m_abandonedSubscriptions");
            Assert.Multiple(() =>
            {
                Assert.That(failed.Results, Has.Count.EqualTo(1));
                Assert.That(
                    failed.Results[0].StatusCode,
                    Is.EqualTo(StatusCodes.BadSubscriptionIdInvalid));
                Assert.That(failed.DiagnosticInfos, Has.Count.EqualTo(1));
                Assert.That(
                    abandonedSubscriptions.TryGetValue(
                        subscription.Id,
                        out ISubscriptionPublishPipeline? retainedSubscription),
                    Is.True,
                    "A refused reservation must leave the subscription abandoned, not lost.");
                Assert.That(retainedSubscription, Is.SameAs(subscription));
            });

            subscription.AbortTransfer(null);
            TransferSubscriptionsResponse retried = await manager
                .TransferSubscriptionsAsync(
                    fixture.DestinationContext,
                    [subscription.Id],
                    sendInitialValues: false)
                .ConfigureAwait(false);

            Assert.That(retried.Results[0].StatusCode, Is.EqualTo(StatusCodes.Good));
        }

        [Test]
        public async Task FailedOwnershipCommitRollsBackThePreparedTransferAsync()
        {
            TransferFixture fixture = await CreateTransferSubscriptionAsync().ConfigureAwait(false);
            using SubscriptionManager manager = fixture.Manager;
            Subscription subscription = fixture.Subscription;
            SessionPublishQueue sourceQueue = GetPublishQueue(
                manager,
                fixture.SourceContext.SessionId);
            int transferCalls = 0;
            m_nodeManagerMock
                .Setup(nodeManager => nodeManager.TransferMonitoredItemsAsync(
                    It.IsAny<OperationContext>(),
                    It.IsAny<bool>(),
                    It.IsAny<IList<IMonitoredItem>>(),
                    It.IsAny<IList<ServiceResult>>(),
                    It.IsAny<MonitoredItemTransferOptions>(),
                    It.IsAny<CancellationToken>()))
                .Callback(() =>
                {
                    // Drop the reservation while the monitored items are handed over so the
                    // ownership commit that follows preparation has to fail and roll back.
                    if (Interlocked.Increment(ref transferCalls) == 1)
                    {
                        subscription.AbortTransfer(m_sessionMock.Object);
                    }
                })
                .Returns(default(ValueTask));

            TransferSubscriptionsResponse failed = await manager
                .TransferSubscriptionsAsync(
                    fixture.DestinationContext,
                    [subscription.Id],
                    sendInitialValues: false)
                .ConfigureAwait(false);

            IReadOnlyList<SessionPublishQueue.QueuedSubscription> restoredSource =
                sourceQueue.CapturePublishTimerSnapshot();
            Assert.Multiple(() =>
            {
                Assert.That(failed.Results, Has.Count.EqualTo(1));
                Assert.That(ServiceResult.IsBad(failed.Results[0].StatusCode), Is.True);
                Assert.That(
                    subscription.Session,
                    Is.SameAs(m_sessionMock.Object),
                    "A failed ownership commit must leave the source session as the owner.");
                Assert.That(restoredSource, Has.Count.EqualTo(1));
                Assert.That(restoredSource[0].Subscription, Is.SameAs(subscription));
            });

            TransferSubscriptionsResponse retried = await manager
                .TransferSubscriptionsAsync(
                    fixture.DestinationContext,
                    [subscription.Id],
                    sendInitialValues: false)
                .ConfigureAwait(false);

            Assert.Multiple(() =>
            {
                Assert.That(retried.Results[0].StatusCode, Is.EqualTo(StatusCodes.Good));
                Assert.That(
                    subscription.Session,
                    Is.SameAs(fixture.DestinationSession.Object));
            });
        }

        [Test]
        public async Task FailedSourceQueueRestoreAggregatesTheTransferErrorAsync()
        {
            TransferFixture fixture = await CreateTransferSubscriptionAsync().ConfigureAwait(false);
            using SubscriptionManager manager = fixture.Manager;
            Subscription subscription = fixture.Subscription;
            SessionPublishQueue sourceQueue = GetPublishQueue(
                manager,
                fixture.SourceContext.SessionId);
            var collidingSubscription = new Mock<ISubscriptionPublishPipeline>();
            collidingSubscription.Setup(candidate => candidate.Id).Returns(subscription.Id);
            int transferCalls = 0;
            m_nodeManagerMock
                .Setup(nodeManager => nodeManager.TransferMonitoredItemsAsync(
                    It.IsAny<OperationContext>(),
                    It.IsAny<bool>(),
                    It.IsAny<IList<IMonitoredItem>>(),
                    It.IsAny<IList<ServiceResult>>(),
                    It.IsAny<MonitoredItemTransferOptions>(),
                    It.IsAny<CancellationToken>()))
                .Returns(() =>
                {
                    if (Interlocked.Increment(ref transferCalls) != 1)
                    {
                        return default;
                    }

                    // Occupy the claimed slot so restoring the source queue entry has to
                    // fail, which is the only way the rollback itself can report an error.
                    sourceQueue.Add(collidingSubscription.Object);
                    return new ValueTask(
                        Task.FromException(
                            new ServiceResultException(StatusCodes.BadUnexpectedError)));
                });

            TransferSubscriptionsResponse failed = await manager
                .TransferSubscriptionsAsync(
                    fixture.DestinationContext,
                    [subscription.Id],
                    sendInitialValues: false)
                .ConfigureAwait(false);

            Assert.Multiple(() =>
            {
                Assert.That(failed.Results, Has.Count.EqualTo(1));
                Assert.That(ServiceResult.IsBad(failed.Results[0].StatusCode), Is.True);
                Assert.That(
                    subscription.Session,
                    Is.SameAs(m_sessionMock.Object),
                    "A rollback that cannot restore the queue must still keep source ownership.");
                Assert.That(
                    sourceQueue.ContainsSubscription(subscription),
                    Is.False);
            });

            sourceQueue.Remove(collidingSubscription.Object, removeQueuedRequests: false);
            sourceQueue.Add(subscription);
            TransferSubscriptionsResponse retried = await manager
                .TransferSubscriptionsAsync(
                    fixture.DestinationContext,
                    [subscription.Id],
                    sendInitialValues: false)
                .ConfigureAwait(false);

            Assert.That(
                retried.Results[0].StatusCode,
                Is.EqualTo(StatusCodes.Good),
                "A reported rollback failure must not leave a stale claim behind.");
        }

        private async Task RegisterMonitoredItemsAsync(
            Subscription subscription,
            params IMonitoredItem[] monitoredItems)
        {
            m_nodeManagerMock
                .Setup(n => n.CreateMonitoredItemsAsync(
                    It.IsAny<OperationContext>(),
                    subscription.Id,
                    It.IsAny<double>(),
                    It.IsAny<TimestampsToReturn>(),
                    It.IsAny<ArrayOf<MonitoredItemCreateRequest>>(),
                    It.IsAny<IList<ServiceResult>>(),
                    It.IsAny<IList<MonitoringFilterResult>>(),
                    It.IsAny<IList<IMonitoredItem>>(),
                    It.IsAny<bool>(),
                    It.IsAny<CancellationToken>()))
                .Callback<
                    OperationContext,
                    uint,
                    double,
                    TimestampsToReturn,
                    ArrayOf<MonitoredItemCreateRequest>,
                    IList<ServiceResult>,
                    IList<MonitoringFilterResult>,
                    IList<IMonitoredItem>,
                    bool,
                    CancellationToken>(
                    (_, _, _, _, _, _, _, createdItems, _, _) =>
                    {
                        for (int ii = 0; ii < monitoredItems.Length; ii++)
                        {
                            createdItems[ii] = monitoredItems[ii];
                        }
                    })
                .Returns(default(ValueTask));

            var requests = new MonitoredItemCreateRequest[monitoredItems.Length];
            for (int ii = 0; ii < monitoredItems.Length; ii++)
            {
                requests[ii] = new MonitoredItemCreateRequest
                {
                    MonitoringMode = MonitoringMode.Reporting
                };
            }
            ArrayOf<MonitoredItemCreateRequest> itemsToCreate = [.. requests];

            var context = new OperationContext(m_sessionMock.Object, new DiagnosticsMasks());
            CreateMonitoredItemsResponse response = await subscription.CreateMonitoredItemsAsync(
                context,
                TimestampsToReturn.Both,
                itemsToCreate).ConfigureAwait(false);

            Assert.That(response.Results, Has.Count.EqualTo(monitoredItems.Length));
        }

        private static Mock<IEventMonitoredItem> CreateEventMonitoredItem(
            uint id,
            int notificationCount,
            List<uint> publishLimits)
        {
            var item = new Mock<IEventMonitoredItem>();
            item.SetupGet(i => i.MonitoringMode).Returns(MonitoringMode.Reporting);
            var pending = new Queue<EventFieldList>();
            for (int index = 0; index < notificationCount; index++)
            {
                pending.Enqueue(new EventFieldList { ClientHandle = id, EventFields = [Variant.From(index)] });
            }
            var createResult = new MonitoredItemCreateResult
            {
                MonitoredItemId = id,
                StatusCode = StatusCodes.Good,
                RevisedSamplingInterval = 0,
                RevisedQueueSize = (uint)notificationCount
            };
            item.SetupGet(i => i.Id).Returns(id);
            item.SetupGet(i => i.IsReadyToPublish).Returns(() => pending.Count > 0);
            item.SetupGet(i => i.MonitoredItemType).Returns(MonitoredItemTypeMask.Events);
            item.Setup(i => i.GetCreateResult(out createResult)).Returns(ServiceResult.Good);
            item.Setup(i => i.Publish(
                    It.IsAny<OperationContext>(),
                    It.IsAny<Queue<EventFieldList>>(),
                    It.IsAny<uint>()))
                .Returns<OperationContext, Queue<EventFieldList>, uint>(
                    (_, notifications, maxNotificationsPerPublish) =>
                    {
                        publishLimits.Add(maxNotificationsPerPublish);
                        for (uint published = 0; pending.Count > 0 && published < maxNotificationsPerPublish; published++)
                        {
                            notifications.Enqueue(pending.Dequeue());
                        }
                        return pending.Count > 0;
                    });
            return item;
        }

        private static Mock<IDataChangeMonitoredItem> CreateDataChangeMonitoredItem(
            uint id,
            int notificationCount,
            List<uint> publishLimits)
        {
            var item = new Mock<IDataChangeMonitoredItem>();
            item.SetupGet(i => i.MonitoringMode).Returns(MonitoringMode.Reporting);
            var pending = new Queue<MonitoredItemNotification>();
            for (int index = 0; index < notificationCount; index++)
            {
                pending.Enqueue(new MonitoredItemNotification { ClientHandle = id, Value = new DataValue(index) });
            }
            var createResult = new MonitoredItemCreateResult
            {
                MonitoredItemId = id,
                StatusCode = StatusCodes.Good,
                RevisedSamplingInterval = 0,
                RevisedQueueSize = (uint)notificationCount
            };
            item.SetupGet(i => i.Id).Returns(id);
            item.SetupGet(i => i.IsReadyToPublish).Returns(() => pending.Count > 0);
            item.SetupGet(i => i.MonitoredItemType).Returns(MonitoredItemTypeMask.DataChange);
            item.Setup(i => i.GetCreateResult(out createResult)).Returns(ServiceResult.Good);
            item.Setup(i => i.Publish(
                    It.IsAny<OperationContext>(),
                    It.IsAny<Queue<MonitoredItemNotification>>(),
                    It.IsAny<Queue<DiagnosticInfo>>(),
                    It.IsAny<uint>(),
                    It.IsAny<Microsoft.Extensions.Logging.ILogger>()))
                .Returns<
                    OperationContext,
                    Queue<MonitoredItemNotification>,
                    Queue<DiagnosticInfo>,
                    uint,
                    Microsoft.Extensions.Logging.ILogger>(
                    (_, notifications, diagnostics, maxNotificationsPerPublish, _) =>
                    {
                        publishLimits.Add(maxNotificationsPerPublish);
                        for (uint published = 0; pending.Count > 0 && published < maxNotificationsPerPublish; published++)
                        {
                            notifications.Enqueue(pending.Dequeue());
                            diagnostics.Enqueue(new DiagnosticInfo());
                        }
                        return pending.Count > 0;
                    });
            return item;
        }

        private static readonly string[] s_deleteCleanupOrder = ["unregister", "dispose", "diagnostics"];
    }
}