From 53247a056af563e7832a9506adc2aafc8f003502 Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Thu, 8 Oct 2026 06:53:45 -0400 Subject: [PATCH] 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 --- docs/Telemetry.md | 2 + .../Drivers/DriverHostActor.cs | 37 ++++++++++++-- .../Telemetry/ITelemetryLocalHub.cs | 7 +++ .../Telemetry/TelemetryLocalHub.cs | 19 +++++-- ...erResilienceStatusPublisherServiceTests.cs | 2 + .../DriverHostActorBootFromCacheTests.cs | 26 ++++++++++ .../DriverHostActorUnreadableArtifactTests.cs | 49 +++++++++++++++++++ .../Scripting/DpsScriptLogPublisherTests.cs | 2 + .../Telemetry/PublishSeamEmitsToHubTests.cs | 2 + .../Telemetry/TelemetryLocalHubTests.cs | 22 +++++++++ 10 files changed, 162 insertions(+), 6 deletions(-) diff --git a/docs/Telemetry.md b/docs/Telemetry.md index 5caae91a..79d00c5a 100644 --- a/docs/Telemetry.md +++ b/docs/Telemetry.md @@ -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 diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/DriverHostActor.cs b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/DriverHostActor.cs index d27c437b..21a6d84e 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/DriverHostActor.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/DriverHostActor.cs @@ -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 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); } diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Telemetry/ITelemetryLocalHub.cs b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Telemetry/ITelemetryLocalHub.cs index 036b6f3c..d2af09db 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Telemetry/ITelemetryLocalHub.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Telemetry/ITelemetryLocalHub.cs @@ -84,4 +84,11 @@ public interface ITelemetryLocalHub /// instances (a handful), so a normal large capacity leaves ample headroom. /// ITelemetrySubscription Subscribe(int boundedCapacity); + + /// + /// Drops the cached Health and Resilience snapshots of a driver instance that has been removed from + /// this node, so a later no longer replays its last-known state. Live + /// subscribers are not told; the instance simply stops emitting. Unknown ids are a no-op. + /// + void Evict(string driverInstanceId); } diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Telemetry/TelemetryLocalHub.cs b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Telemetry/TelemetryLocalHub.cs index aa70c434..8d8fb40b 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Telemetry/TelemetryLocalHub.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Telemetry/TelemetryLocalHub.cs @@ -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 _healthCache = new(); private readonly ConcurrentDictionary<(string InstanceId, string HostName), TelemetryItem.Resilience> _resilienceCache = new(); @@ -66,6 +65,20 @@ public sealed class TelemetryLocalHub : ITelemetryLocalHub } } + /// + 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 _); + } + } + /// public ITelemetrySubscription Subscribe(int boundedCapacity) { diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.Host.IntegrationTests/DriverResilienceStatusPublisherServiceTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.Host.IntegrationTests/DriverResilienceStatusPublisherServiceTests.cs index 180eea82..f9e195ac 100644 --- a/tests/Server/ZB.MOM.WW.OtOpcUa.Host.IntegrationTests/DriverResilienceStatusPublisherServiceTests.cs +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.Host.IntegrationTests/DriverResilienceStatusPublisherServiceTests.cs @@ -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(); } } diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverHostActorBootFromCacheTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverHostActorBootFromCacheTests.cs index dbd1323e..ecfb2311 100644 --- a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverHostActorBootFromCacheTests.cs +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverHostActorBootFromCacheTests.cs @@ -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 { "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() { 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 a3f5f1bc..f1cd00cd 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 @@ -178,6 +178,55 @@ public sealed class DriverHostActorUnreadableArtifactTests : RuntimeActorTestBas AskDiagnostics(actor).CurrentRevision.ShouldBe(RevB); } + /// + /// #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. + /// + [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 { "driver" }, + telemetryHub: hub)); + + actor.Tell(new DispatchDeployment(SeedV3Deployment(db, RevA), RevA, CorrelationId.NewId())); + coordinator.ExpectMsg(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(); + 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(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(); + 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"); + } + /// 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( diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Scripting/DpsScriptLogPublisherTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Scripting/DpsScriptLogPublisherTests.cs index 0157cb9b..04bd633f 100644 --- a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Scripting/DpsScriptLogPublisherTests.cs +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Scripting/DpsScriptLogPublisherTests.cs @@ -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(); } } diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Telemetry/PublishSeamEmitsToHubTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Telemetry/PublishSeamEmitsToHubTests.cs index 61591d2f..16be3cf8 100644 --- a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Telemetry/PublishSeamEmitsToHubTests.cs +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Telemetry/PublishSeamEmitsToHubTests.cs @@ -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(); } } diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Telemetry/TelemetryLocalHubTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Telemetry/TelemetryLocalHubTests.cs index 23eab83c..86141791 100644 --- a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Telemetry/TelemetryLocalHubTests.cs +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Telemetry/TelemetryLocalHubTests.cs @@ -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().Single().E.DriverInstanceId.ShouldBe("drv-b"); + items.OfType().Single().E.DriverInstanceId.ShouldBe("drv-b"); + } + [Fact] public void New_subscriber_is_primed_with_cached_Health_snapshot() {