perf(otopcua): one SetDesiredSubscriptions per driver per redeploy (follow-up D)

2026-06-26 14:30:16 -04:00
parent cde16063d9
commit 05c820795a
2 changed files with 203 additions and 0 deletions
@@ -595,6 +595,184 @@ public sealed class DriverHostActorDiscoveryTests : RuntimeActorTestBase
        factory.DiscoverCount.ShouldBe(1);
    }

+    /// <summary>Follow-up D (one-send invariant — SURVIVOR path): a CACHED (FixedTree-discovered) driver whose
+    /// plan SURVIVES a redeploy must receive EXACTLY ONE <see cref="DriverInstanceActor.SetDesiredSubscriptions"/>
+    /// for that pass — the authored∪discovered UNION the re-inject tail sends — NOT the old TWO (a bulk
+    /// authored-only send that dropped the whole handle, then the tail union that re-subscribed it). Observed via
+    /// the shared driver's <c>SubscribeCount</c> (each non-empty SetDesiredSubscriptions ⇒ exactly one
+    /// SubscribeAsync — no de-dup in <see cref="DriverInstanceActor"/>): the count rises by EXACTLY 1 across the
+    /// redeploy and the final subscribed set is the union. (Pre-task: the bulk loop ALSO sent the authored-only
+    /// set first ⇒ the count rose by 2 and the set transiently dropped "ft-ref-1".)</summary>
+    [Fact]
+    public void Cached_driver_survivor_redeploy_sends_exactly_one_union_subscription()
+    {
+        var db = NewInMemoryDbFactory();
+        var factory = new SubscribingDriverFactory("Modbus");
+        var deploymentId = SeedDeploymentWithEquipmentTags(db, RevA,
+            (Equip: "EQ-1", Driver: "d1", FullName: "40001", Folder: (string?)null, Name: "speed"));
+
+        var (actor, publish, coordinator) = SpawnHostAndApply(db, deploymentId, factory);
+
+        actor.Tell(new DriverInstanceActor.DiscoveredNodesReady("d1", new[]
+        {
+            new DiscoveredNode(
+                FolderPathSegments: new[] { "FOCAS", "10.0.0.5:8193", "Identity" },
+                BrowseName: "Model", DisplayName: "Model", FullReference: "ft-ref-1",
+                DataType: DriverDataType.Float64, IsArray: false, ArrayDim: null,
+                Writable: false, IsHistorized: false),
+        }));
+
+        // First injection: the union subscribe (authored "40001" + discovered "ft-ref-1") lands and the cache is
+        // populated (_discoveredByDriver[d1] = { EQ-1: plan }).
+        publish.ExpectMsg<OpcUaPublishActor.MaterialiseDiscoveredNodes>(Timeout);
+        AwaitAssert(() =>
+        {
+            var refs = factory.LastSubscribedRefs;
+            refs.ShouldNotBeNull();
+            refs!.ShouldContain("40001");
+            refs.ShouldContain("ft-ref-1");
+        }, duration: Timeout);
+        // Let the first-injection traffic settle, then snapshot the subscribe count as the redeploy baseline.
+        publish.ExpectNoMsg(TimeSpan.FromMilliseconds(300));
+        var countBeforeRedeploy = factory.SubscribeCount;
+
+        // Redeploy the SAME composition (new revision so it applies; d1 → EQ-1 unchanged ⇒ the cached plan
+        // SURVIVES the re-inject tail ⇒ re-applied as a single union send).
+        var deploymentId2 = SeedDeploymentWithEquipmentTags(db, RevB,
+            (Equip: "EQ-1", Driver: "d1", FullName: "40001", Folder: (string?)null, Name: "speed"));
+        actor.Tell(new DispatchDeployment(deploymentId2, RevB, CorrelationId.NewId()));
+        coordinator.ExpectMsg<ApplyAck>(Timeout).Outcome.ShouldBe(ApplyAckOutcome.Applied);
+        publish.ExpectMsg<OpcUaPublishActor.RebuildAddressSpace>(Timeout);
+        publish.ExpectMsg<OpcUaPublishActor.MaterialiseDiscoveredNodes>(Timeout); // tail survivor re-materialise
+
+        // EXACTLY ONE SetDesiredSubscriptions this redeploy: the count rises by 1 and the set is the union (the
+        // combined condition is UNSATISFIABLE under the old double-send — at count+1 the set was authored-only
+        // (no "ft-ref-1"), at count+2 the count overshoots — so this fails RED before the fix).
+        AwaitAssert(() =>
+        {
+            var refs = factory.LastSubscribedRefs;
+            refs.ShouldNotBeNull();
+            refs!.ShouldContain("40001");
+            refs.ShouldContain("ft-ref-1");
+            factory.SubscribeCount.ShouldBe(countBeforeRedeploy + 1);
+        }, duration: Timeout);
+        // Settle + re-confirm the count did NOT creep to +2 (the retired bulk authored-only + tail union double-send).
+        publish.ExpectNoMsg(TimeSpan.FromMilliseconds(300));
+        factory.SubscribeCount.ShouldBe(countBeforeRedeploy + 1);
+    }
+
+    /// <summary>Follow-up D (one-send invariant — DROPPED path): a CACHED driver whose plan is FULLY DROPPED by a
+    /// config-unchanged rebind (the inner map empties ⇒ the driver is removed from <c>_discoveredByDriver</c>)
+    /// must still receive EXACTLY ONE <see cref="DriverInstanceActor.SetDesiredSubscriptions"/> — the AUTHORED-ONLY
+    /// fallback — so its authored subscriptions are not lost now that the bulk loop SKIPS cached drivers. The
+    /// re-inject tail no longer re-applies a (now-empty) plan for it, so the fallback is the only send. Observed
+    /// via <c>SubscribeCount</c> (+1) and the subscribed set ("40001" only, NOT the dropped "ft-ref-1"). (It also
+    /// gets a <see cref="DriverInstanceActor.TriggerRediscovery"/> — a different message type the non-discovery
+    /// child no-ops, so it adds no subscribe.) Guards that the bulk-skip didn't reduce this path to ZERO sends.</summary>
+    [Fact]
+    public void Cached_driver_fully_dropped_redeploy_sends_exactly_one_authored_only_fallback()
+    {
+        var db = NewInMemoryDbFactory();
+        var factory = new SubscribingDriverFactory("Modbus");
+        var deploymentId = SeedDeploymentWithEquipmentTags(db, RevA,
+            (Equip: "EQ-1", Driver: "d1", FullName: "40001", Folder: (string?)null, Name: "speed"));
+
+        var (actor, publish, coordinator) = SpawnHostAndApply(db, deploymentId, factory);
+
+        actor.Tell(new DriverInstanceActor.DiscoveredNodesReady("d1", new[]
+        {
+            new DiscoveredNode(
+                FolderPathSegments: new[] { "FOCAS", "10.0.0.5:8193", "Identity" },
+                BrowseName: "Model", DisplayName: "Model", FullReference: "ft-ref-1",
+                DataType: DriverDataType.Float64, IsArray: false, ArrayDim: null,
+                Writable: false, IsHistorized: false),
+        }));
+
+        publish.ExpectMsg<OpcUaPublishActor.MaterialiseDiscoveredNodes>(Timeout);
+        AwaitAssert(() =>
+        {
+            var refs = factory.LastSubscribedRefs;
+            refs.ShouldNotBeNull();
+            refs!.ShouldContain("40001");
+            refs.ShouldContain("ft-ref-1");
+        }, duration: Timeout);
+        publish.ExpectNoMsg(TimeSpan.FromMilliseconds(300));
+        var countBeforeRedeploy = factory.SubscribeCount;
+
+        // Redeploy REBINDING d1 EQ-1 → EQ-2 (same FullName; DriverConfig "{}" unchanged ⇒ child NOT restarted).
+        // The cached EQ-1-scoped plan is dropped by the rebind guard ⇒ the inner map empties ⇒ d1 is removed from
+        // _discoveredByDriver ⇒ NO survivor re-apply. The fallback must send the authored-only set so "40001"
+        // stays subscribed this pass.
+        var deploymentId2 = SeedDeploymentWithEquipmentTags(db, RevB,
+            (Equip: "EQ-2", Driver: "d1", FullName: "40001", Folder: (string?)null, Name: "speed"));
+        actor.Tell(new DispatchDeployment(deploymentId2, RevB, CorrelationId.NewId()));
+        coordinator.ExpectMsg<ApplyAck>(Timeout).Outcome.ShouldBe(ApplyAckOutcome.Applied);
+        publish.ExpectMsg<OpcUaPublishActor.RebuildAddressSpace>(Timeout);
+        // No MaterialiseDiscoveredNodes — the plan was dropped, not re-grafted — so no further publish traffic.
+        publish.ExpectNoMsg(TimeSpan.FromMilliseconds(500));
+
+        // EXACTLY ONE SetDesiredSubscriptions this redeploy: the authored-only fallback. The count rises by 1 and
+        // the set is "40001" only (the dropped FixedTree ref is gone).
+        AwaitAssert(() =>
+        {
+            var refs = factory.LastSubscribedRefs;
+            refs.ShouldNotBeNull();
+            refs!.ShouldContain("40001");
+            refs.ShouldNotContain("ft-ref-1");
+            refs.Count.ShouldBe(1);
+            factory.SubscribeCount.ShouldBe(countBeforeRedeploy + 1);
+        }, duration: Timeout);
+        publish.ExpectNoMsg(TimeSpan.FromMilliseconds(300));
+        factory.SubscribeCount.ShouldBe(countBeforeRedeploy + 1);
+    }
+
+    /// <summary>Follow-up D (one-send invariant — NON-CACHED path): a driver that was NEVER cached (no FixedTree
+    /// discovered) is unaffected by the cached-driver bulk-loop skip — it still gets EXACTLY ONE bulk
+    /// authored-only <see cref="DriverInstanceActor.SetDesiredSubscriptions"/> per redeploy (the re-inject tail
+    /// never runs for it). Guards that the skip didn't accidentally suppress (or double) a non-cached driver's
+    /// send. Observed via <c>SubscribeCount</c> (+1) and the subscribed set ("40001" only).</summary>
+    [Fact]
+    public void Non_cached_driver_redeploy_sends_exactly_one_authored_only_subscription()
+    {
+        var db = NewInMemoryDbFactory();
+        var factory = new SubscribingDriverFactory("Modbus");
+        var deploymentId = SeedDeploymentWithEquipmentTags(db, RevA,
+            (Equip: "EQ-1", Driver: "d1", FullName: "40001", Folder: (string?)null, Name: "speed"));
+
+        var (actor, publish, coordinator) = SpawnHostAndApply(db, deploymentId, factory);
+
+        // No DiscoveredNodesReady ⇒ d1 is never cached. Wait for the initial bulk subscribe to settle, then
+        // snapshot the count as the redeploy baseline.
+        AwaitAssert(() =>
+        {
+            var refs = factory.LastSubscribedRefs;
+            refs.ShouldNotBeNull();
+            refs!.ShouldContain("40001");
+        }, duration: Timeout);
+        publish.ExpectNoMsg(TimeSpan.FromMilliseconds(300));
+        var countBeforeRedeploy = factory.SubscribeCount;
+
+        // Redeploy SAME composition (new rev). d1 is NOT in _discoveredByDriver ⇒ the bulk loop sends it once and
+        // the re-inject tail skips it.
+        var deploymentId2 = SeedDeploymentWithEquipmentTags(db, RevB,
+            (Equip: "EQ-1", Driver: "d1", FullName: "40001", Folder: (string?)null, Name: "speed"));
+        actor.Tell(new DispatchDeployment(deploymentId2, RevB, CorrelationId.NewId()));
+        coordinator.ExpectMsg<ApplyAck>(Timeout).Outcome.ShouldBe(ApplyAckOutcome.Applied);
+        publish.ExpectMsg<OpcUaPublishActor.RebuildAddressSpace>(Timeout);
+        publish.ExpectNoMsg(TimeSpan.FromMilliseconds(500)); // no materialise — d1 was never cached
+
+        AwaitAssert(() =>
+        {
+            var refs = factory.LastSubscribedRefs;
+            refs.ShouldNotBeNull();
+            refs!.ShouldContain("40001");
+            refs.Count.ShouldBe(1);
+            factory.SubscribeCount.ShouldBe(countBeforeRedeploy + 1);
+        }, duration: Timeout);
+        publish.ExpectNoMsg(TimeSpan.FromMilliseconds(300));
+        factory.SubscribeCount.ShouldBe(countBeforeRedeploy + 1);
+    }
+
    /// <summary>Spawns the host with the subscribing driver factory + a publish probe, dispatches the
    /// deployment, and waits for the Applied ACK so the apply (and thus <c>_lastComposition</c> + the live
    /// child + the initial SubscribeBulk pass) has completed before the test injects discovered nodes. A