fix(telemetry): release a removed driver's resilience tracker state so the publisher stops re-emitting it (#609)

Review R1 M1: StopChild evicted the hub cache, but DriverResilienceStatusPublisherService re-emits the tracker every tick. IDriverCapabilityInvokerFactory.Release (default no-op) now invalidates the pipeline builder + status tracker; the host calls it before Evict. Test runs a publisher tick after removal, with a neighbour instance as positive control. Claude-Session: https://claude.ai/code/session_01Ef2ozxS7QPanGT8uoATbTe
2026-10-08 07:47:13 -04:00
parent b176469b43
commit 8a9bf23048
4 changed files with 98 additions and 0 deletions
@@ -255,6 +255,11 @@ driver node**:
  reconnect loses whatever transitions happened while disconnected, same as a DPS resubscribe would.
  The cache is evicted per instance (`ITelemetryLocalHub.Evict`) when `DriverHostActor.StopChild` removes a
  driver from the plan (#609), so a decommissioned instance's last state is not replayed to new subscribers.
+  `StopChild` first releases the instance's resilience state (`IDriverCapabilityInvokerFactory.Release` →
+  pipeline-builder + `DriverResilienceStatusTracker.Invalidate`): `DriverResilienceStatusPublisherService`
+  re-emits the whole tracker every 5 s, so evicting only the hub was undone within one tick (review R1 M1).
+  Residual: a Health emit already in flight from the stopping child can land after the eviction and
+  re-cache that one snapshot — a narrow race, not a recurring replay.
 - **Connection indicator** — the `/alerts` and `/script-log` live/disconnected pill is driven by
  **aggregate** stream health across all dialers: connected when **at least one** node stream is
  up, disconnected when all are down. This matches today's DPS `SubscribeAck`/`PostStop` pill
@@ -28,6 +28,16 @@ public interface IDriverCapabilityInvokerFactory
    /// </param>
    /// <returns>A per-instance <see cref="IDriverCapabilityInvoker"/>.</returns>
    IDriverCapabilityInvoker Create(string driverInstanceId, string driverType, string? resilienceConfigJson = null);
+
+    /// <summary>
+    ///     Release every piece of shared resilience state held for <paramref name="driverInstanceId"/> — called
+    ///     by the host when the instance leaves the deployed plan (#609). A concrete factory drops the
+    ///     instance's cached pipelines and its status-tracker snapshots, so the periodic resilience publisher
+    ///     stops re-emitting (and re-priming new telemetry subscribers with) a decommissioned driver's last
+    ///     breaker/failure counters. Default: no-op (a factory that holds no state).
+    /// </summary>
+    /// <param name="driverInstanceId">The driver instance that was removed.</param>
+    void Release(string driverInstanceId) { }
 }

 /// <summary>
@@ -90,4 +90,16 @@ public sealed class DriverCapabilityInvokerFactory : IDriverCapabilityInvokerFac
            statusTracker: _statusTracker,
            optionsGeneration: generation);
    }
+
+    /// <inheritdoc />
+    /// <remarks>
+    ///     #609: without this the tracker kept a removed instance's snapshots forever (only <see cref="Create"/>
+    ///     invalidated, and a removed instance is never re-created), and the resilience publisher re-emitted
+    ///     them every tick — undoing the host's telemetry-hub eviction within one interval.
+    /// </remarks>
+    public void Release(string driverInstanceId)
+    {
+        _builder.Invalidate(driverInstanceId);
+        _statusTracker.Invalidate(driverInstanceId);
+    }
 }
@@ -227,6 +227,77 @@ public sealed class DriverHostActorUnreadableArtifactTests : RuntimeActorTestBas
        replayed.ShouldContain("other-node-driver");
    }

+    /// <summary>
+    ///     #609 (review R1 M1) — the Resilience half of the eviction must SURVIVE the next resilience-publisher
+    ///     tick. <c>DriverResilienceStatusPublisherService</c> re-emits <c>tracker.Snapshot()</c> into the hub
+    ///     every interval, so evicting only the hub cache was undone within one tick: the host must also
+    ///     release the instance's tracker state. The tick below is the publisher's own body
+    ///     (<c>BuildMessages</c> → <c>hub.Emit</c>) minus the DPS publish, run against the REAL factory +
+    ///     tracker the Host wires. The neighbour instance is the positive control: it is still emitted by
+    ///     the same tick, so an absent drv-1 is not an empty tick.
+    /// </summary>
+    [Fact]
+    public void Removing_a_driver_instance_releases_its_resilience_state_so_the_next_publisher_tick_omits_it()
+    {
+        var db = NewInMemoryDbFactory();
+        var factory = new SubscribingDriverFactory("Modbus");
+        var hub = new ZB.MOM.WW.OtOpcUa.Runtime.Telemetry.TelemetryLocalHub();
+        var tracker = new ZB.MOM.WW.OtOpcUa.Core.Resilience.DriverResilienceStatusTracker();
+        var invokerFactory = new ZB.MOM.WW.OtOpcUa.Core.Resilience.DriverCapabilityInvokerFactory(
+            new ZB.MOM.WW.OtOpcUa.Core.Resilience.DriverResiliencePipelineBuilder(statusTracker: tracker),
+            tracker,
+            _ => ZB.MOM.WW.OtOpcUa.Core.Abstractions.DriverTier.A);
+        var coordinator = CreateTestProbe();
+        var actor = Sys.ActorOf(DriverHostActor.Props(
+            db, TestNode, coordinator.Ref,
+            driverFactory: factory,
+            localRoles: new HashSet<string> { "driver" },
+            invokerFactory: invokerFactory,
+            telemetryHub: hub));
+
+        actor.Tell(new DispatchDeployment(SeedV3Deployment(db, RevA), RevA, CorrelationId.NewId()));
+        coordinator.ExpectMsg<ApplyAck>(Timeout).Outcome.ShouldBe(ApplyAckOutcome.Applied);
+
+        // Resilience counters for the running driver and for an instance this host does not own.
+        tracker.RecordBreakerOpen("drv-1", "dev1", DateTime.UtcNow);
+        tracker.RecordFailure("neighbour-drv", "dev9", DateTime.UtcNow);
+        PublisherTick(tracker, hub);
+        ReplayedResilienceIds(hub).ShouldContain("drv-1"); // control: drv-1 is primed before the removal
+
+        actor.Tell(new DispatchDeployment(SeedDriverlessDeployment(db, RevB), RevB, CorrelationId.NewId()));
+        coordinator.ExpectMsg<ApplyAck>(Timeout).Outcome.ShouldBe(ApplyAckOutcome.Applied);
+        AwaitAssert(
+            () => AskDiagnostics(actor).Drivers.Select(d => d.Name).ShouldNotContain("drv-1"),
+            duration: Timeout);
+
+        // The publisher's next tick, after the removal.
+        PublisherTick(tracker, hub);
+
+        var replayed = ReplayedResilienceIds(hub);
+        replayed.ShouldNotContain("drv-1");
+        replayed.ShouldContain("neighbour-drv");
+
+        static void PublisherTick(
+            ZB.MOM.WW.OtOpcUa.Core.Resilience.DriverResilienceStatusTracker t,
+            ZB.MOM.WW.OtOpcUa.Runtime.Telemetry.TelemetryLocalHub h)
+        {
+            var now = DateTime.UtcNow;
+            foreach (var e in t.Snapshot())
+                h.Emit(new ZB.MOM.WW.OtOpcUa.Runtime.Telemetry.TelemetryItem.Resilience(new(
+                    e.DriverInstanceId, e.HostName, e.Snapshot.IsBreakerOpen, e.Snapshot.ConsecutiveFailures,
+                    e.Snapshot.CurrentInFlight, e.Snapshot.LastBreakerOpenUtc, e.Snapshot.LastSampledUtc, now)));
+        }
+
+        static List<string> ReplayedResilienceIds(ZB.MOM.WW.OtOpcUa.Runtime.Telemetry.TelemetryLocalHub h)
+        {
+            using var sub = h.Subscribe(64);
+            var ids = new List<string>();
+            while (sub.Reader.TryRead(out var i))
+                if (i is ZB.MOM.WW.OtOpcUa.Runtime.Telemetry.TelemetryItem.Resilience r) ids.Add(r.E.DriverInstanceId);
+            return ids;
+        }
+    }
+
    /// <summary>Spawns the host with the subscribing factory, dispatches <paramref name="deploymentId"/> and
    /// waits for the Applied ACK so the child + the initial SubscribeBulk pass are in place.</summary>
    private (IActorRef Actor, Akka.TestKit.TestProbe Coordinator) SpawnHostAndApply(