test(opcua): over-the-wire subscription-survival on pure-add deploy (R2-07 T5)

2026-07-13 12:01:23 -04:00
parent e6280cb1da
commit 376233b92c
2 changed files with 207 additions and 2 deletions
@@ -45,10 +45,11 @@
    {
      "id": "T5",
      "subject": "Phase 1: over-the-wire SubscriptionSurvivalTests \u2014 subscribe, pure-add, item survives + new node browsable",
-      "status": "pending",
+      "status": "deferred-live",
      "blockedBy": [
        "T4b"
-      ]
+      ],
+      "note": "authored + compiles (dotnet build green); NOT run \u2014 *.IntegrationTests suite leaks ~16GB/run per the memory constraint. MUST run in the serial heavy integration pass (single-test, boots one in-process server + client)."
    },
    {
      "id": "T6",
@@ -0,0 +1,204 @@
+using System.Net;
+using System.Net.Sockets;
+using Microsoft.Extensions.Logging.Abstractions;
+using Opc.Ua;
+using Opc.Ua.Client;
+using Shouldly;
+using Xunit;
+
+namespace ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests;
+
+/// <summary>
+///     R2-07 03/P1 — the HEADLINE proof: an over-the-wire subscription on an existing tag SURVIVES a
+///     pure-add deploy (one new tag), instead of being killed by the full-rebuild teardown that pre-R2-07
+///     every topology change forced. Boots the real <see cref="OtOpcUaSdkServer"/>, materialises tag A,
+///     opens a real client subscription + monitored item on A, then drives a PURE-ADD apply adding tag B
+///     (classifier ⇒ PureAdd ⇒ no rebuild ⇒ the idempotent Materialise passes create B and leave A's
+///     NodeState untouched), and asserts:
+///     <list type="bullet">
+///         <item>A's monitored item still delivers a subsequent value write (the item is alive — no
+///             <c>BadNodeIdUnknown</c>, no dead subscription);</item>
+///         <item>B is browsable + readable (the add landed).</item>
+///     </list>
+///     Written against the master (pre-T2) behaviour this FAILS (the rebuild recreates A as a new NodeState
+///     and the monitored item goes dead); it passes once the classifier routing (T2) + the publish-actor
+///     announce (T4b) land.
+///     <para><b>Run note:</b> this is a heavy in-process server+client integration test — run it in the
+///     serial integration pass (it is NOT part of the lightweight unit sweep). The <see cref="OtOpcUaNodeManager"/>
+///     namespace URI is <see cref="OtOpcUaNodeManager.DefaultNamespaceUri"/>; the client resolves the
+///     folder-scoped variable NodeIds against it.</para>
+/// </summary>
+public sealed class SubscriptionSurvivalTests
+{
+    private const string ServerUri = "urn:OtOpcUa.SubscriptionSurvival";
+
+    /// <summary>A subscription on tag A survives a pure-add deploy adding tag B; B becomes readable.</summary>
+    [Fact]
+    public async Task Subscription_on_existing_tag_survives_pure_add_of_another_tag()
+    {
+        var pkiRoot = Path.Combine(Path.GetTempPath(), $"otopcua-subsurvive-{Guid.NewGuid():N}");
+        var port = AllocateFreePort();
+        var ct = TestContext.Current.CancellationToken;
+
+        try
+        {
+            var options = new OpcUaApplicationHostOptions
+            {
+                ApplicationName = ServerUri,
+                ApplicationUri = ServerUri,
+                OpcUaPort = port,
+                PublicHostname = "127.0.0.1",
+                PkiStoreRoot = pkiRoot,
+                EnabledSecurityProfiles = new List<OpcUaSecurityProfile> { OpcUaSecurityProfile.None },
+                AutoAcceptUntrustedClientCertificates = true,
+            };
+            var server = new OtOpcUaSdkServer();
+            await using var host = new OpcUaApplicationHost(options, NullLogger<OpcUaApplicationHost>.Instance);
+            await host.StartAsync(server, ct);
+
+            var nm = server.NodeManager!;
+            var sink = new SdkAddressSpaceSink(nm);
+            var applier = new AddressSpaceApplier(sink, NullLogger<AddressSpaceApplier>.Instance);
+
+            // --- Materialise tag A (equipment eq-1, tag "A") and seed it a Good value. ---
+            var withA = CompositionWith(
+                new EquipmentTagPlan("tag-a", "eq-1", "drv", FolderPath: "", Name: "A", DataType: "Int32", FullName: "1", Writable: false, Alarm: null));
+            applier.MaterialiseHierarchy(withA);
+            applier.MaterialiseEquipmentTags(withA);
+            var nodeIdAString = Commons.OpcUa.EquipmentNodeIds.Variable("eq-1", "", "A");
+            sink.WriteValue(nodeIdAString, 41, Commons.OpcUa.OpcUaQuality.Good, DateTime.UtcNow);
+
+            // --- Open a client session + subscription + monitored item on A. ---
+            using var session = await OpenSessionAsync($"opc.tcp://127.0.0.1:{port}/OtOpcUa", ct);
+            var ns = (ushort)session.NamespaceUris.GetIndex(OtOpcUaNodeManager.DefaultNamespaceUri);
+            ns.ShouldBeGreaterThan((ushort)0);
+            var nodeIdA = new NodeId(nodeIdAString, ns);
+
+            var received = new List<DataValue>();
+            var gate = new object();
+            var subscription = new Subscription(session.DefaultSubscription) { PublishingInterval = 100 };
+            session.AddSubscription(subscription);
+            await subscription.CreateAsync(ct);
+
+            var monitored = new MonitoredItem(subscription.DefaultItem)
+            {
+                StartNodeId = nodeIdA,
+                AttributeId = Attributes.Value,
+                SamplingInterval = 50,
+                QueueSize = 10,
+            };
+            monitored.Notification += (item, _) =>
+            {
+                foreach (var v in item.DequeueValues())
+                    lock (gate) received.Add(v);
+            };
+            subscription.AddItem(monitored);
+            await subscription.ApplyChangesAsync(ct);
+
+            // Baseline: the initial value (41) is delivered.
+            await WaitUntilAsync(() => { lock (gate) return received.Count >= 1; }, TimeSpan.FromSeconds(5));
+
+            // --- Drive a PURE-ADD apply adding tag B (previous={A}, next={A,B}). ---
+            var withAB = CompositionWith(
+                new EquipmentTagPlan("tag-a", "eq-1", "drv", FolderPath: "", Name: "A", DataType: "Int32", FullName: "1", Writable: false, Alarm: null),
+                new EquipmentTagPlan("tag-b", "eq-1", "drv", FolderPath: "", Name: "B", DataType: "Int32", FullName: "2", Writable: false, Alarm: null));
+            var plan = AddressSpacePlanner.Compute(withA, withAB);
+            plan.AddedEquipmentTags.Count.ShouldBe(1);
+
+            var outcome = applier.Apply(plan);
+            outcome.RebuildCalled.ShouldBeFalse(); // PureAdd — no teardown
+
+            // Materialise passes (as the publish actor runs them) then announce — over the FULL composition.
+            applier.MaterialiseHierarchy(withAB);
+            applier.MaterialiseEquipmentTags(withAB);
+            applier.AnnounceAddedNodes(plan);
+
+            // --- A's monitored item must still be alive: a new write to A is delivered. ---
+            lock (gate) received.Clear();
+            sink.WriteValue(nodeIdAString, 42, Commons.OpcUa.OpcUaQuality.Good, DateTime.UtcNow);
+
+            await WaitUntilAsync(() => { lock (gate) return received.Any(v => Equals(v.Value, 42)); }, TimeSpan.FromSeconds(5));
+            lock (gate)
+            {
+                received.ShouldContain(v => Equals(v.Value, 42));
+                received.ShouldNotContain(v => v.StatusCode == StatusCodes.BadNodeIdUnknown);
+            }
+
+            // --- B is browsable + readable (the add landed). ---
+            var nodeIdB = new NodeId(Commons.OpcUa.EquipmentNodeIds.Variable("eq-1", "", "B"), ns);
+            sink.WriteValue(nodeIdB.Identifier.ToString()!, 7, Commons.OpcUa.OpcUaQuality.Good, DateTime.UtcNow);
+            var bValue = await session.ReadValueAsync(nodeIdB, ct);
+            bValue.StatusCode.ShouldNotBe(StatusCodes.BadNodeIdUnknown);
+            bValue.Value.ShouldBe(7);
+        }
+        finally
+        {
+            if (Directory.Exists(pkiRoot)) Directory.Delete(pkiRoot, recursive: true);
+        }
+    }
+
+    private static AddressSpaceComposition CompositionWith(params EquipmentTagPlan[] tags) =>
+        new(
+            UnsAreas: Array.Empty<UnsAreaProjection>(),
+            UnsLines: Array.Empty<UnsLineProjection>(),
+            EquipmentNodes: new[] { new EquipmentNode("eq-1", "Equipment 1", "") },
+            DriverInstancePlans: Array.Empty<DriverInstancePlan>(),
+            ScriptedAlarmPlans: Array.Empty<ScriptedAlarmPlan>())
+        {
+            EquipmentTags = tags,
+        };
+
+    private static async Task<ISession> OpenSessionAsync(string endpointUrl, CancellationToken ct)
+    {
+        var appConfig = new ApplicationConfiguration
+        {
+            ApplicationName = "OtOpcUa.SubscriptionSurvivalClient",
+            ApplicationUri = $"urn:OtOpcUa.SubscriptionSurvivalClient.{Guid.NewGuid():N}",
+            ApplicationType = ApplicationType.Client,
+            SecurityConfiguration = new SecurityConfiguration
+            {
+                ApplicationCertificate = new CertificateIdentifier(),
+                AutoAcceptUntrustedCertificates = true,
+            },
+            ClientConfiguration = new ClientConfiguration { DefaultSessionTimeout = 60_000 },
+        };
+        await appConfig.ValidateAsync(ApplicationType.Client, ct);
+        appConfig.CertificateValidator.CertificateValidation += (_, e) => e.Accept = true;
+
+        var endpoint = await CoreClientUtils.SelectEndpointAsync(
+            appConfig, endpointUrl, false, DefaultTelemetry.Create(_ => { }), ct);
+        var endpointConfiguration = EndpointConfiguration.Create(appConfig);
+        var configuredEndpoint = new ConfiguredEndpoint(null, endpoint, endpointConfiguration);
+
+        return await new DefaultSessionFactory(DefaultTelemetry.Create(_ => { })).CreateAsync(
+            appConfig,
+            configuredEndpoint,
+            updateBeforeConnect: false,
+            checkDomain: false,
+            sessionName: "SubscriptionSurvivalTests",
+            sessionTimeout: 60_000,
+            identity: new UserIdentity(new AnonymousIdentityToken()),
+            preferredLocales: null,
+            ct: ct);
+    }
+
+    private static async Task WaitUntilAsync(Func<bool> condition, TimeSpan timeout)
+    {
+        var deadline = DateTime.UtcNow + timeout;
+        while (DateTime.UtcNow < deadline)
+        {
+            if (condition()) return;
+            await Task.Delay(50);
+        }
+        condition().ShouldBeTrue("condition not met within timeout");
+    }
+
+    private static int AllocateFreePort()
+    {
+        var listener = new TcpListener(IPAddress.Loopback, 0);
+        listener.Start();
+        var port = ((IPEndPoint)listener.LocalEndpoint).Port;
+        listener.Stop();
+        return port;
+    }
+}