fix(runtime): a removed driver's late writes no longer resurrect its resilience/telemetry state (#609)

DriverHostActor.StopChild released the tracker (Invalidate) and evicted the hub synchronously, but Context.Stop is asynchronous: the stopping child's in-flight SubscribeBulk completed through its capability invoker AFTER the release, and its RecordCallStart/RecordSuccess/RecordCallComplete re-created the tracker entry, which DriverResilienceStatusPublisherService then re-emitted every tick forever. Confirmed with a temporary stack capture in the tracker (removed): every post-Release write came from DriverInstanceActor.HandleSubscribeAsync -> CapabilityInvoker.ExecuteAsync; the test failed whenever it landed before the second publisher tick. - DriverResilienceStatusTracker.Retire (Release) drops the id's snapshots and ignores every later Record* for it; one guarded write path re-checks after the upsert so a write racing Retire cannot survive it. Reinstate (called by DriverCapabilityInvokerFactory.Create) lifts it, so the same id re-added in a later deployment records again. - TelemetryLocalHub.Evict now also holds the id (late Health/Resilience dropped, neither cached nor fanned out); ITelemetryLocalHub.Readmit (default no-op), called at the top of SpawnChild, lifts it. - The removal test is now deterministic: an invoker decorator holds the child's subscribe inside the real invoker's bracket across the removal and signals when its late writes are done (positive control) — red 5/5 without the fix, green 20/20 with it. New unit tests for the tracker guard (incl. a Retire race), the factory Release/Create pair, the hub hold/readmit, and a re-add-same-id host test. Claude-Session: https://claude.ai/code/session_01Ef2ozxS7QPanGT8uoATbTe
2026-10-08 09:07:11 -04:00
parent 5c0623ad2b
commit 4a10ea1e1c
10 changed files with 418 additions and 42 deletions
@@ -243,10 +243,15 @@ public sealed class DriverHostActorUnreadableArtifactTests : RuntimeActorTestBas
        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);
+        // The child's subscribe is HELD across the removal so the tracker writes its capability invoker makes
+        // around it land AFTER StopChild's Release — the late writer that resurrected the entry (#609). Without
+        // the gate this test only failed when the child's in-flight SubscribeBulk happened to finish between the
+        // removal and the second tick (~4/6 runs); the gate forces that interleaving on every run.
+        var invokerFactory = new GatedSubscribeInvokerFactory(
+            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,
@@ -257,6 +262,7 @@ public sealed class DriverHostActorUnreadableArtifactTests : RuntimeActorTestBas

        actor.Tell(new DispatchDeployment(SeedV3Deployment(db, RevA), RevA, CorrelationId.NewId()));
        coordinator.ExpectMsg<ApplyAck>(Timeout).Outcome.ShouldBe(ApplyAckOutcome.Applied);
+        invokerFactory.SubscribeEntered.Wait(Timeout).ShouldBeTrue("the child never reached its subscribe");

        // Resilience counters for the running driver and for an instance this host does not own.
        tracker.RecordBreakerOpen("drv-1", "dev1", DateTime.UtcNow);
@@ -270,31 +276,141 @@ public sealed class DriverHostActorUnreadableArtifactTests : RuntimeActorTestBas
            () => AskDiagnostics(actor).Drivers.Select(d => d.Name).ShouldNotContain("drv-1"),
            duration: Timeout);

-        // The publisher's next tick, after the removal.
+        // Let the removed child's held subscribe run to completion through the REAL invoker (RecordCallStart /
+        // RecordSuccess / RecordCallComplete). Positive control: the late write has demonstrably happened before
+        // the tick below, so the absence asserted after it is not "nothing has arrived yet".
+        invokerFactory.OpenGate();
+        invokerFactory.SubscribeCompleted.Wait(Timeout).ShouldBeTrue("the held subscribe never completed");
+
+        // The publisher's next tick, after the removal AND after the late writer.
        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)));
-        }
+    /// <summary>
+    ///     #609 — the release a removal makes must not outlive the removal: when the SAME instance id comes back
+    ///     in a later deployment, its fresh invoker records resilience state again (the tracker un-retires the id
+    ///     on <c>Create</c>).
+    /// </summary>
+    [Fact]
+    public void A_driver_instance_re_added_after_removal_records_resilience_state_again()
+    {
+        var db = NewInMemoryDbFactory();
+        var factory = new SubscribingDriverFactory("Modbus");
+        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: new ZB.MOM.WW.OtOpcUa.Runtime.Telemetry.TelemetryLocalHub()));

-        static List<string> ReplayedResilienceIds(ZB.MOM.WW.OtOpcUa.Runtime.Telemetry.TelemetryLocalHub h)
+        actor.Tell(new DispatchDeployment(SeedV3Deployment(db, RevA), RevA, CorrelationId.NewId()));
+        coordinator.ExpectMsg<ApplyAck>(Timeout).Outcome.ShouldBe(ApplyAckOutcome.Applied);
+        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);
+
+        var revC = RevisionHash.Parse(new string('c', 64));
+        actor.Tell(new DispatchDeployment(SeedV3Deployment(db, revC), revC, CorrelationId.NewId()));
+        coordinator.ExpectMsg<ApplyAck>(Timeout).Outcome.ShouldBe(ApplyAckOutcome.Applied);
+        AwaitAssert(
+            () => AskDiagnostics(actor).Drivers.Select(d => d.Name).ShouldContain("drv-1"),
+            duration: Timeout);
+
+        // An explicit counter for the re-added instance is retained, not swallowed as a late write.
+        tracker.RecordBreakerOpen("drv-1", "dev1", DateTime.UtcNow);
+        tracker.TryGet("drv-1", "dev1").ShouldNotBeNull().IsBreakerOpen.ShouldBeTrue();
+    }
+
+    private 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)));
+    }
+
+    private 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>
+    ///     Wraps a real invoker factory so every <see cref="DriverCapability.Subscribe"/> call is HELD before it
+    ///     reaches the real invoker until <see cref="OpenGate"/> — forcing a removed child's subscribe (and the
+    ///     tracker writes the real invoker makes around it) to run after the host's Release. Every other call
+    ///     passes straight through.
+    /// </summary>
+    private sealed class GatedSubscribeInvokerFactory(IDriverCapabilityInvokerFactory inner) : IDriverCapabilityInvokerFactory
+    {
+        private readonly TaskCompletionSource _gate = new(TaskCreationOptions.RunContinuationsAsynchronously);
+
+        /// <summary>Set once a subscribe has reached the gate.</summary>
+        public ManualResetEventSlim SubscribeEntered { get; } = new();
+
+        /// <summary>Set once a held subscribe has returned from the REAL invoker, i.e. its tracker writes are done.</summary>
+        public ManualResetEventSlim SubscribeCompleted { get; } = new();
+
+        public void OpenGate() => _gate.TrySetResult();
+
+        public IDriverCapabilityInvoker Create(string driverInstanceId, string driverType, string? resilienceConfigJson = null) =>
+            new Gated(this, inner.Create(driverInstanceId, driverType, resilienceConfigJson));
+
+        public void Release(string driverInstanceId) => inner.Release(driverInstanceId);
+
+        private sealed class Gated(GatedSubscribeInvokerFactory owner, IDriverCapabilityInvoker real) : IDriverCapabilityInvoker
        {
-            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;
+            public async ValueTask<TResult> ExecuteAsync<TResult>(
+                DriverCapability capability, string hostName,
+                Func<CancellationToken, ValueTask<TResult>> callSite, CancellationToken cancellationToken)
+            {
+                if (capability != DriverCapability.Subscribe)
+                    return await real.ExecuteAsync(capability, hostName, callSite, cancellationToken);
+
+                // The gate sits INSIDE the call site, i.e. inside the real invoker's RecordCallStart … RecordSuccess /
+                // RecordCallComplete bracket, and every await here is ConfigureAwait(false): the real invoker resumes
+                // off the actor's context too, so its closing writes run on the pool after the child has stopped —
+                // exactly the production shape. (Gating in front of the real invoker on the actor's context does
+                // not work: Akka drops a stopped actor's pending continuations, so nothing late ever runs.)
+                try
+                {
+                    return await real.ExecuteAsync(capability, hostName, async ct =>
+                    {
+                        owner.SubscribeEntered.Set();
+                        await owner._gate.Task.ConfigureAwait(false);
+                        return await callSite(CancellationToken.None).ConfigureAwait(false);
+                    }, CancellationToken.None).ConfigureAwait(false);
+                }
+                finally { owner.SubscribeCompleted.Set(); }
+            }
+
+            public ValueTask ExecuteAsync(
+                DriverCapability capability, string hostName,
+                Func<CancellationToken, ValueTask> callSite, CancellationToken cancellationToken) =>
+                real.ExecuteAsync(capability, hostName, callSite, cancellationToken);
+
+            public ValueTask<TResult> ExecuteWriteAsync<TResult>(
+                string hostName, bool isIdempotent,
+                Func<CancellationToken, ValueTask<TResult>> callSite, CancellationToken cancellationToken) =>
+                real.ExecuteWriteAsync(hostName, isIdempotent, callSite, cancellationToken);
        }
    }

@@ -58,6 +58,55 @@ public sealed class TelemetryLocalHubTests
        items.OfType<TelemetryItem.Resilience>().Single().E.DriverInstanceId.ShouldBe("drv-b");
    }

+    /// <summary>
+    ///     #609 — the host evicts a removed instance while its child is still stopping, and that child can
+    ///     still publish (e.g. a health snapshot when an in-flight connect/subscribe completes). A snapshot for
+    ///     an evicted instance must not re-prime the cache — nor reach live subscribers, where it would
+    ///     re-populate central's view of a driver that no longer exists. A neighbour emitted at the same time is
+    ///     the positive control.
+    /// </summary>
+    [Fact]
+    public void A_late_snapshot_for_an_evicted_instance_is_neither_cached_nor_delivered()
+    {
+        var hub = new TelemetryLocalHub();
+        hub.Emit(new TelemetryItem.Health(Health("drv-a")));
+        hub.Evict("drv-a");
+        using var live = hub.Subscribe(boundedCapacity: 16);
+
+        hub.Emit(new TelemetryItem.Health(Health("drv-a")));
+        hub.Emit(new TelemetryItem.Resilience(Resilience("drv-a", "host-1")));
+        hub.Emit(new TelemetryItem.Health(Health("drv-b")));
+
+        Drain(live).Select(Id).ShouldBe(new[] { "drv-b" });
+        using var primed = hub.Subscribe(boundedCapacity: 16);
+        Drain(primed).Select(Id).ShouldBe(new[] { "drv-b" });
+    }
+
+    /// <summary>
+    ///     #609 — eviction does not outlive the removal: the same instance id re-spawned in a later deployment
+    ///     is <c>Readmit</c>ted and its snapshots are cached again.
+    /// </summary>
+    [Fact]
+    public void Readmit_lets_an_evicted_instance_publish_again()
+    {
+        var hub = new TelemetryLocalHub();
+        hub.Evict("drv-a");
+        hub.Readmit("drv-a");
+
+        hub.Emit(new TelemetryItem.Health(Health("drv-a")));
+        hub.Emit(new TelemetryItem.Resilience(Resilience("drv-a", "host-1")));
+
+        using var sub = hub.Subscribe(boundedCapacity: 16);
+        Drain(sub).Select(Id).ShouldBe(new[] { "drv-a", "drv-a" });
+    }
+
+    private static string Id(TelemetryItem item) => item switch
+    {
+        TelemetryItem.Health h => h.E.DriverInstanceId,
+        TelemetryItem.Resilience r => r.E.DriverInstanceId,
+        _ => "<other>",
+    };
+
    [Fact]
    public void New_subscriber_is_primed_with_cached_Health_snapshot()
    {