From 8a9bf23048f11eebb76ec6e079477f84165978e8 Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Thu, 8 Oct 2026 07:47:13 -0400 Subject: [PATCH] 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 --- docs/Telemetry.md | 5 ++ .../IDriverCapabilityInvokerFactory.cs | 10 +++ .../DriverCapabilityInvokerFactory.cs | 12 ++++ .../DriverHostActorUnreadableArtifactTests.cs | 71 +++++++++++++++++++ 4 files changed, 98 insertions(+) diff --git a/docs/Telemetry.md b/docs/Telemetry.md index 79d00c5a..b4430173 100644 --- a/docs/Telemetry.md +++ b/docs/Telemetry.md @@ -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 diff --git a/src/Core/ZB.MOM.WW.OtOpcUa.Core.Abstractions/IDriverCapabilityInvokerFactory.cs b/src/Core/ZB.MOM.WW.OtOpcUa.Core.Abstractions/IDriverCapabilityInvokerFactory.cs index 6c28bdae..c27ea515 100644 --- a/src/Core/ZB.MOM.WW.OtOpcUa.Core.Abstractions/IDriverCapabilityInvokerFactory.cs +++ b/src/Core/ZB.MOM.WW.OtOpcUa.Core.Abstractions/IDriverCapabilityInvokerFactory.cs @@ -28,6 +28,16 @@ public interface IDriverCapabilityInvokerFactory /// /// A per-instance . IDriverCapabilityInvoker Create(string driverInstanceId, string driverType, string? resilienceConfigJson = null); + + /// + /// Release every piece of shared resilience state held for — 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). + /// + /// The driver instance that was removed. + void Release(string driverInstanceId) { } } /// diff --git a/src/Core/ZB.MOM.WW.OtOpcUa.Core/Resilience/DriverCapabilityInvokerFactory.cs b/src/Core/ZB.MOM.WW.OtOpcUa.Core/Resilience/DriverCapabilityInvokerFactory.cs index e0a0584c..44ffb21b 100644 --- a/src/Core/ZB.MOM.WW.OtOpcUa.Core/Resilience/DriverCapabilityInvokerFactory.cs +++ b/src/Core/ZB.MOM.WW.OtOpcUa.Core/Resilience/DriverCapabilityInvokerFactory.cs @@ -90,4 +90,16 @@ public sealed class DriverCapabilityInvokerFactory : IDriverCapabilityInvokerFac statusTracker: _statusTracker, optionsGeneration: generation); } + + /// + /// + /// #609: without this the tracker kept a removed instance's snapshots forever (only + /// 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. + /// + public void Release(string driverInstanceId) + { + _builder.Invalidate(driverInstanceId); + _statusTracker.Invalidate(driverInstanceId); + } } diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverHostActorUnreadableArtifactTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverHostActorUnreadableArtifactTests.cs index f1cd00cd..ee918813 100644 --- a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverHostActorUnreadableArtifactTests.cs +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverHostActorUnreadableArtifactTests.cs @@ -227,6 +227,77 @@ public sealed class DriverHostActorUnreadableArtifactTests : RuntimeActorTestBas replayed.ShouldContain("other-node-driver"); } + /// + /// #609 (review R1 M1) — the Resilience half of the eviction must SURVIVE the next resilience-publisher + /// tick. DriverResilienceStatusPublisherService re-emits tracker.Snapshot() 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 + /// (BuildMessages → hub.Emit) 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. + /// + [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 { "driver" }, + invokerFactory: invokerFactory, + telemetryHub: hub)); + + actor.Tell(new DispatchDeployment(SeedV3Deployment(db, RevA), RevA, CorrelationId.NewId())); + coordinator.ExpectMsg(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(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 ReplayedResilienceIds(ZB.MOM.WW.OtOpcUa.Runtime.Telemetry.TelemetryLocalHub h) + { + using var sub = h.Subscribe(64); + var ids = new List(); + 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; + } + } + /// Spawns the host with the subscribing factory, dispatches and /// waits for the Applied ACK so the child + the initial SubscribeBulk pass are in place. private (IActorRef Actor, Akka.TestKit.TestProbe Coordinator) SpawnHostAndApply(