fix(runtime): malformed artifact bytes on boot/restore are no answer (#601); evict telemetry snapshots on driver removal (#609)

Claude-Session: https://claude.ai/code/session_01Ef2ozxS7QPanGT8uoATbTe
2026-10-08 06:53:45 -04:00
parent 3baa332d6b
commit 53247a056a
10 changed files with 162 additions and 6 deletions
@@ -253,6 +253,8 @@ driver node**:
  `IDriverStatusSnapshotStore` / `IDriverResilienceStatusStore` re-prime immediately rather than
  sitting stale until the next natural event. `alerts`/`script-logs` simply tolerate the gap — a
  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.
 - **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
@@ -1,4 +1,5 @@
 using System.Diagnostics;
+using System.Text.Json;
 using Akka.Actor;
 using Akka.Cluster.Tools.PublishSubscribe;
 using Akka.Event;
@@ -1097,7 +1098,12 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS
    {
        var correlation = CorrelationId.NewId();

-        ReconcileDriversFromBlob(deploymentId, blob);
+        // #601: null = the cached bytes were empty or malformed — no answer, not an empty configuration.
+        // Keep last-known-good: no address-space rebuild and no subscription push from unusable bytes.
+        // #601: null = the cached bytes were empty or malformed — no answer, not an empty configuration.
+        // Keep last-known-good: no address-space rebuild and no subscription push from unusable bytes.
+        if (ReconcileDriversFromBlob(deploymentId, blob) is null)
+            return;

        // Hand the cached blob to the rebuild so it materialises the client-facing address space from
        // it directly. Passing only the deploymentId would drive a ConfigDb read — unreachable here by
@@ -2483,7 +2489,25 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS
            return null;
        }

-        ApplyDriverSpecs(DeploymentArtifact.ParseDriverInstances(blob, _localNode.Value));
+        // #601 (S-10): non-empty but malformed bytes are "no answer" too. The lenient static
+        // ParseDriverInstances swallows the JsonException into an empty list, and an empty list plans every
+        // running child for StopChild. Open throws instead, so a corrupt cache row / restored blob leaves
+        // the running drivers alone, exactly like the zero-length case above.
+        IReadOnlyList<DriverInstanceSpec> specs;
+        try
+        {
+            using var parsed = DeploymentArtifact.Open(blob);
+            specs = parsed.ParseDriverInstances(_localNode.Value);
+        }
+        catch (JsonException ex)
+        {
+            _log.Error(ex,
+                "DriverHost {Node}: artifact for {Id} is malformed ({Bytes} bytes); skipping reconcile and keeping the {Count} running driver(s)",
+                _localNode, deploymentId, blob.Length, _children.Count);
+            return null;
+        }
+
+        ApplyDriverSpecs(specs);
        return blob;
    }

@@ -2772,10 +2796,14 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS
            return;
        }

+        // #601 (S-10): Open, not the lenient static ParseComposition — that wrapper swallows malformed JSON
+        // into an EMPTY composition, which would hand every child an empty desired set and drop its live
+        // subscription. A malformed artifact is no answer: leave the subscriptions as they are.
        AddressSpaceComposition composition;
        try
        {
-            composition = DeploymentArtifact.ParseComposition(blob, _localNode.Value);
+            using var parsed = DeploymentArtifact.Open(blob);
+            composition = parsed.ParseComposition(_localNode.Value);
        }
        catch (Exception ex)
        {
@@ -3204,6 +3232,9 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS
        StopDependencyAdapter(driverInstanceId);
        Context.Stop(entry.Actor);
        _children.Remove(driverInstanceId);
+        // #609: the instance left the plan (StopChild is only reached via plan.ToStop), so its cached
+        // Health/Resilience snapshots must not replay to new telemetry subscribers.
+        _telemetryHub?.Evict(driverInstanceId);
        _log.Info("DriverHost {Node}: stopped driver child {Id}", _localNode, driverInstanceId);
    }

@@ -84,4 +84,11 @@ public interface ITelemetryLocalHub
    ///     instances (a handful), so a normal large capacity leaves ample headroom.
    /// </param>
    ITelemetrySubscription Subscribe(int boundedCapacity);
+
+    /// <summary>
+    ///     Drops the cached Health and Resilience snapshots of a driver instance that has been removed from
+    ///     this node, so a later <see cref="Subscribe"/> no longer replays its last-known state. Live
+    ///     subscribers are not told; the instance simply stops emitting. Unknown ids are a no-op.
+    /// </summary>
+    void Evict(string driverInstanceId);
 }
@@ -37,9 +37,8 @@ public sealed class TelemetryLocalHub : ITelemetryLocalHub

    // Snapshot last-value caches. Only ever mutated under _gate (kept ConcurrentDictionary so a
    // Subscribe reading them under the gate is trivially safe even against any future lock-free read).
-    // TODO(mesh-phase5+): evict snapshot cache entries on driver-instance removal (needs a
-    // driver-lifecycle hook); pre-production, accepted for now. Until then a decommissioned driver
-    // instance's last-known Health/Resilience state replays to new subscribers forever.
+    // Entries are evicted via Evict when DriverHostActor removes the instance (#609), so a
+    // decommissioned driver's last-known state does not replay to new subscribers.
    private readonly ConcurrentDictionary<string, TelemetryItem.Health> _healthCache = new();
    private readonly ConcurrentDictionary<(string InstanceId, string HostName), TelemetryItem.Resilience> _resilienceCache = new();

@@ -66,6 +65,20 @@ public sealed class TelemetryLocalHub : ITelemetryLocalHub
        }
    }

+    /// <inheritdoc />
+    public void Evict(string driverInstanceId)
+    {
+        ArgumentNullException.ThrowIfNull(driverInstanceId);
+
+        lock (_gate)
+        {
+            _healthCache.TryRemove(driverInstanceId, out _);
+            foreach (var key in _resilienceCache.Keys)
+                if (string.Equals(key.InstanceId, driverInstanceId, StringComparison.Ordinal))
+                    _resilienceCache.TryRemove(key, out _);
+        }
+    }
+
    /// <inheritdoc />
    public ITelemetrySubscription Subscribe(int boundedCapacity)
    {
@@ -133,6 +133,8 @@ public sealed class DriverResilienceStatusPublisherServiceTests
            lock (Items) Items.Add(item);
        }

+        public void Evict(string driverInstanceId) { }
+
        public ITelemetrySubscription Subscribe(int boundedCapacity) => throw new NotSupportedException();
    }
 }
@@ -80,6 +80,32 @@ public sealed class DriverHostActorBootFromCacheTests : RuntimeActorTestBase
        rebuild.Artifact.ShouldBe(blob);
    }

+    [Theory]
+    [InlineData("not json at all")]
+    [InlineData("{\"DriverInstances\": [")]               // truncated JSON
+    [InlineData("\u001f\u008b\u0008garbage-after-gzip-magic")] // gzip magic, undecompressible body
+    public void BootFromCache_withMalformedNonEmptyBytes_isNoAnswer_notAnEmptyConfiguration(string text)
+    {
+        // #601 (S-10): the lenient parse turned malformed bytes into "zero driver specs", which plans every
+        // running child for StopChild and then rebuilds the address space from the garbage. Malformed bytes
+        // must be treated like the zero-length case: nothing reconciled, nothing rebuilt.
+        var cached = new CachedDeploymentArtifact(
+            DeploymentId.NewId().ToString(), CachedRev.Value,
+            System.Text.Encoding.Latin1.GetBytes(text), DateTimeOffset.UtcNow.AddHours(-1));
+        var publishProbe = CreateTestProbe();
+
+        var actor = Sys.ActorOf(DriverHostActor.Props(
+            new ThrowingDbFactory(), TestNode, ackRouter: null,
+            localRoles: new HashSet<string> { "driver" },
+            opcUaPublishActor: publishProbe.Ref,
+            deploymentArtifactCache: new StubArtifactCache(cached)));
+
+        // Barrier: diagnostics are served only after PreStart's boot work ran, so the absence below is
+        // asserted after the boot path had its chance to Tell (positive control is the valid-bytes test).
+        AskDiagnostics(actor);
+        publishProbe.ExpectNoMsg(TimeSpan.FromMilliseconds(500));
+    }
+
    [Fact]
    public void CentralUnreachableAndCacheEmpty_StaysStaleWithNoConfiguration()
    {
@@ -178,6 +178,55 @@ public sealed class DriverHostActorUnreadableArtifactTests : RuntimeActorTestBas
        AskDiagnostics(actor).CurrentRevision.ShouldBe(RevB);
    }

+    /// <summary>
+    ///     #609 — removing a driver instance from the plan evicts its cached telemetry snapshots, so a new
+    ///     subscriber is no longer primed with a decommissioned driver's last Health/Resilience. The
+    ///     surviving instance's snapshot is the positive control: eviction is per-instance, not a wipe.
+    /// </summary>
+    [Fact]
+    public void Removing_a_driver_instance_evicts_its_telemetry_snapshots_but_not_a_neighbours()
+    {
+        var db = NewInMemoryDbFactory();
+        var factory = new SubscribingDriverFactory("Modbus");
+        var hub = new ZB.MOM.WW.OtOpcUa.Runtime.Telemetry.TelemetryLocalHub();
+        var coordinator = CreateTestProbe();
+        var actor = Sys.ActorOf(DriverHostActor.Props(
+            db, TestNode, coordinator.Ref,
+            driverFactory: factory,
+            localRoles: new HashSet<string> { "driver" },
+            telemetryHub: hub));
+
+        actor.Tell(new DispatchDeployment(SeedV3Deployment(db, RevA), RevA, CorrelationId.NewId()));
+        coordinator.ExpectMsg<ApplyAck>(Timeout).Outcome.ShouldBe(ApplyAckOutcome.Applied);
+
+        hub.Emit(new ZB.MOM.WW.OtOpcUa.Runtime.Telemetry.TelemetryItem.Health(new(
+            "c1", "drv-1", "Healthy", null, null, 0, DateTime.UtcNow)));
+        hub.Emit(new ZB.MOM.WW.OtOpcUa.Runtime.Telemetry.TelemetryItem.Health(new(
+            "c1", "other-node-driver", "Healthy", null, null, 0, DateTime.UtcNow)));
+
+        // Control: before the removal, drv-1 replays.
+        using (var before = hub.Subscribe(16))
+        {
+            var seen = new List<string>();
+            while (before.Reader.TryRead(out var i))
+                if (i is ZB.MOM.WW.OtOpcUa.Runtime.Telemetry.TelemetryItem.Health h) seen.Add(h.E.DriverInstanceId);
+            seen.ShouldContain("drv-1");
+        }
+
+        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);
+
+        using var after = hub.Subscribe(16);
+        var replayed = new List<string>();
+        while (after.Reader.TryRead(out var i))
+            if (i is ZB.MOM.WW.OtOpcUa.Runtime.Telemetry.TelemetryItem.Health h) replayed.Add(h.E.DriverInstanceId);
+        replayed.ShouldNotContain("drv-1");
+        replayed.ShouldContain("other-node-driver");
+    }
+
    /// <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(
@@ -80,6 +80,8 @@ public sealed class DpsScriptLogPublisherTests : RuntimeActorTestBase
    {
        public void Emit(TelemetryItem item) { }

+        public void Evict(string driverInstanceId) { }
+
        public ITelemetrySubscription Subscribe(int boundedCapacity) => throw new NotSupportedException();
    }
 }
@@ -312,6 +312,8 @@ public sealed class PublishSeamEmitsToHubTests : RuntimeActorTestBase

        public void Emit(TelemetryItem item) => Items.Add(item);

+        public void Evict(string driverInstanceId) { }
+
        public ITelemetrySubscription Subscribe(int boundedCapacity) => throw new NotSupportedException();
    }
 }
@@ -36,6 +36,28 @@ public sealed class TelemetryLocalHubTests
        return items;
    }

+    [Fact]
+    public void Evict_drops_the_instances_Health_and_every_Resilience_snapshot_but_keeps_others()
+    {
+        var hub = new TelemetryLocalHub();
+        hub.Emit(new TelemetryItem.Health(Health("drv-a")));
+        hub.Emit(new TelemetryItem.Resilience(Resilience("drv-a", "host-1")));
+        hub.Emit(new TelemetryItem.Resilience(Resilience("drv-a", "host-2")));
+        hub.Emit(new TelemetryItem.Health(Health("drv-b")));
+        hub.Emit(new TelemetryItem.Resilience(Resilience("drv-b", "host-1")));
+
+        hub.Evict("drv-a");
+        hub.Evict("never-existed"); // unknown id is a no-op
+
+        using var sub = hub.Subscribe(boundedCapacity: 16);
+        var items = Drain(sub);
+
+        // Positive control: drv-b still replays; drv-a does not.
+        items.Count.ShouldBe(2);
+        items.OfType<TelemetryItem.Health>().Single().E.DriverInstanceId.ShouldBe("drv-b");
+        items.OfType<TelemetryItem.Resilience>().Single().E.DriverInstanceId.ShouldBe("drv-b");
+    }
+
    [Fact]
    public void New_subscriber_is_primed_with_cached_Health_snapshot()
    {