From 4a10ea1e1cc62f7b5fa64d44e0c1c9d10b7ac9ef Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Thu, 8 Oct 2026 09:07:11 -0400 Subject: [PATCH] fix(runtime): a removed driver's late writes no longer resurrect its resilience/telemetry state (#609) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 --- docs/Telemetry.md | 13 +- .../DriverCapabilityInvokerFactory.cs | 10 +- .../DriverResilienceStatusTracker.cs | 75 +++++++-- .../Drivers/DriverHostActor.cs | 9 +- .../Telemetry/ITelemetryLocalHub.cs | 10 ++ .../Telemetry/TelemetryLocalHub.cs | 18 ++ .../DriverCapabilityInvokerFactoryTests.cs | 41 +++++ .../DriverResilienceStatusTrackerTests.cs | 77 +++++++++ .../DriverHostActorUnreadableArtifactTests.cs | 158 +++++++++++++++--- .../Telemetry/TelemetryLocalHubTests.cs | 49 ++++++ 10 files changed, 418 insertions(+), 42 deletions(-) diff --git a/docs/Telemetry.md b/docs/Telemetry.md index f653a071..255c8588 100644 --- a/docs/Telemetry.md +++ b/docs/Telemetry.md @@ -256,10 +256,17 @@ driver node**: 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` + pipeline-builder + `DriverResilienceStatusTracker.Retire`): `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. + ⚠️ **Both releases HOLD the id, not just clear it.** `Context.Stop` is asynchronous, so the stopping child + can still finish a capability call it had in flight — observed: its `SubscribeBulk`, whose + `RecordCallStart`/`RecordSuccess`/`RecordCallComplete` ran after the release and re-created the tracker + entry, which the publisher then re-emitted every tick for the life of the process — and can still publish a + Health snapshot. So the tracker ignores every `Record*` for a retired id, and the hub drops every later + Health/Resilience item for an evicted id (neither cached nor fanned out), until the same id is spawned + again: `IDriverCapabilityInvokerFactory.Create` reinstates it in the tracker and `SpawnChild` calls + `ITelemetryLocalHub.Readmit`. Residual: a late write from the *old* child that lands after a re-add of the + same id is indistinguishable from the new child's and is kept — the same window a respawn always had. - **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/Resilience/DriverCapabilityInvokerFactory.cs b/src/Core/ZB.MOM.WW.OtOpcUa.Core/Resilience/DriverCapabilityInvokerFactory.cs index 44ffb21b..7634ed2c 100644 --- a/src/Core/ZB.MOM.WW.OtOpcUa.Core/Resilience/DriverCapabilityInvokerFactory.cs +++ b/src/Core/ZB.MOM.WW.OtOpcUa.Core/Resilience/DriverCapabilityInvokerFactory.cs @@ -80,6 +80,9 @@ public sealed class DriverCapabilityInvokerFactory : IDriverCapabilityInvokerFac // The tracker's snapshots describe the breakers inside the pipelines just discarded, so they go // too — a retained "Breaker OPEN" from a pre-respawn outage could never self-correct (only the // discarded pipeline's OnClosed stamps the close) and would show a false OPEN chip forever. + // Reinstate first: an instance id removed by an earlier deployment (Release ⇒ Retire) and now re-added + // must record again (#609). + _statusTracker.Reinstate(driverInstanceId); _statusTracker.Invalidate(driverInstanceId); return new CapabilityInvoker( @@ -96,10 +99,15 @@ public sealed class DriverCapabilityInvokerFactory : IDriverCapabilityInvokerFac /// #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. + /// The tracker retires the instance rather than merely invalidating it: the host stops the + /// child asynchronously, so a capability call it still had in flight completes through the released + /// invoker afterwards, and its call-start/success/complete writes re-created the entry (observed: the + /// child's in-flight SubscribeBulk). A retired id ignores those writes until a later + /// for the same id reinstates it. /// public void Release(string driverInstanceId) { _builder.Invalidate(driverInstanceId); - _statusTracker.Invalidate(driverInstanceId); + _statusTracker.Retire(driverInstanceId); } } diff --git a/src/Core/ZB.MOM.WW.OtOpcUa.Core/Resilience/DriverResilienceStatusTracker.cs b/src/Core/ZB.MOM.WW.OtOpcUa.Core/Resilience/DriverResilienceStatusTracker.cs index eb1296f7..050cd3b2 100644 --- a/src/Core/ZB.MOM.WW.OtOpcUa.Core/Resilience/DriverResilienceStatusTracker.cs +++ b/src/Core/ZB.MOM.WW.OtOpcUa.Core/Resilience/DriverResilienceStatusTracker.cs @@ -22,14 +22,38 @@ public sealed class DriverResilienceStatusTracker { private readonly ConcurrentDictionary _status = new(); + // Instances retired by Retire (removed from the deployed plan) and not since Reinstated (#609). Every + // Record* for one of these is ignored: the removed instance's child stops ASYNCHRONOUSLY, so a capability + // call it still had in flight completes through its invoker after the host's release and would otherwise + // re-create the entry — which the resilience publisher then re-emits every tick, forever. One entry per + // distinct removed id, so the set stays as small as the driver churn that fills it. + private readonly ConcurrentDictionary _retired = new(StringComparer.Ordinal); + + /// + /// The single write path for every Record*. A write for a retired instance is dropped; the + /// post-write re-check closes the race with a concurrent — a writer that passed the + /// first check just before the retire and lands its upsert just after the retire's sweep removes its own + /// write again, so no interleaving resurrects the instance. + /// + private void Upsert( + string driverInstanceId, + string hostName, + Func add, + Func update) + { + if (_retired.ContainsKey(driverInstanceId)) return; + var key = new StatusKey(driverInstanceId, hostName); + _status.AddOrUpdate(key, add, update); + if (_retired.ContainsKey(driverInstanceId)) _status.TryRemove(key, out _); + } + /// Record a Polly pipeline failure for . /// The driver instance identifier. /// The host name. /// The UTC timestamp of the failure event. public void RecordFailure(string driverInstanceId, string hostName, DateTime utcNow) { - var key = new StatusKey(driverInstanceId, hostName); - _status.AddOrUpdate(key, + Upsert(driverInstanceId, hostName, _ => new ResilienceStatusSnapshot { ConsecutiveFailures = 1, LastSampledUtc = utcNow }, (_, existing) => existing with { @@ -44,8 +68,7 @@ public sealed class DriverResilienceStatusTracker /// The UTC timestamp of the success event. public void RecordSuccess(string driverInstanceId, string hostName, DateTime utcNow) { - var key = new StatusKey(driverInstanceId, hostName); - _status.AddOrUpdate(key, + Upsert(driverInstanceId, hostName, _ => new ResilienceStatusSnapshot { ConsecutiveFailures = 0, LastSampledUtc = utcNow }, (_, existing) => existing with { @@ -60,8 +83,7 @@ public sealed class DriverResilienceStatusTracker /// The UTC timestamp of the breaker open event. public void RecordBreakerOpen(string driverInstanceId, string hostName, DateTime utcNow) { - var key = new StatusKey(driverInstanceId, hostName); - _status.AddOrUpdate(key, + Upsert(driverInstanceId, hostName, _ => new ResilienceStatusSnapshot { LastBreakerOpenUtc = utcNow, LastSampledUtc = utcNow }, (_, existing) => existing with { LastBreakerOpenUtc = utcNow, LastSampledUtc = utcNow }); } @@ -77,8 +99,7 @@ public sealed class DriverResilienceStatusTracker /// The UTC timestamp of the breaker close event. public void RecordBreakerClose(string driverInstanceId, string hostName, DateTime utcNow) { - var key = new StatusKey(driverInstanceId, hostName); - _status.AddOrUpdate(key, + Upsert(driverInstanceId, hostName, _ => new ResilienceStatusSnapshot { LastBreakerClosedUtc = utcNow, LastSampledUtc = utcNow }, (_, existing) => existing with { LastBreakerClosedUtc = utcNow, LastSampledUtc = utcNow }); } @@ -104,14 +125,39 @@ public sealed class DriverResilienceStatusTracker return removed; } + /// + /// Drop every snapshot for a driver instance that has been REMOVED from the deployed plan, and ignore + /// every later write for it until (#609). Called by + /// DriverCapabilityInvokerFactory.Release. Unlike (a respawn, whose new + /// invoker records straight away), the removed instance's child is still stopping — a call it had in + /// flight finishes after this returns, and without the hold its RecordCallStart / RecordSuccess / + /// RecordCallComplete re-create the entry the publisher then re-emits forever. + /// + /// The removed driver instance. + /// The number of snapshots removed. + public int Retire(string driverInstanceId) + { + // Retire BEFORE sweeping: a write racing the sweep either lands first (and is swept) or re-checks after + // its upsert and removes itself (see Upsert). + _retired[driverInstanceId] = 0; + return Invalidate(driverInstanceId); + } + + /// + /// Lift an earlier so the instance records again — the same driver instance id + /// re-added in a later deployment. Called by DriverCapabilityInvokerFactory.Create; a no-op for + /// an id that was never retired. + /// + /// The driver instance being (re)created. + public void Reinstate(string driverInstanceId) => _retired.TryRemove(driverInstanceId, out _); + /// Record a process recycle event (Tier C only). /// The driver instance identifier. /// The host name. /// The UTC timestamp of the recycle event. public void RecordRecycle(string driverInstanceId, string hostName, DateTime utcNow) { - var key = new StatusKey(driverInstanceId, hostName); - _status.AddOrUpdate(key, + Upsert(driverInstanceId, hostName, _ => new ResilienceStatusSnapshot { LastRecycleUtc = utcNow, LastSampledUtc = utcNow }, (_, existing) => existing with { LastRecycleUtc = utcNow, LastSampledUtc = utcNow }); } @@ -124,8 +170,7 @@ public sealed class DriverResilienceStatusTracker /// The UTC timestamp of the footprint update. public void RecordFootprint(string driverInstanceId, string hostName, long baselineBytes, long currentBytes, DateTime utcNow) { - var key = new StatusKey(driverInstanceId, hostName); - _status.AddOrUpdate(key, + Upsert(driverInstanceId, hostName, _ => new ResilienceStatusSnapshot { BaselineFootprintBytes = baselineBytes, @@ -150,8 +195,7 @@ public sealed class DriverResilienceStatusTracker /// The host name. public void RecordCallStart(string driverInstanceId, string hostName) { - var key = new StatusKey(driverInstanceId, hostName); - _status.AddOrUpdate(key, + Upsert(driverInstanceId, hostName, _ => new ResilienceStatusSnapshot { CurrentInFlight = 1 }, (_, existing) => existing with { CurrentInFlight = existing.CurrentInFlight + 1 }); } @@ -161,8 +205,7 @@ public sealed class DriverResilienceStatusTracker /// The host name. public void RecordCallComplete(string driverInstanceId, string hostName) { - var key = new StatusKey(driverInstanceId, hostName); - _status.AddOrUpdate(key, + Upsert(driverInstanceId, hostName, _ => new ResilienceStatusSnapshot { CurrentInFlight = 0 }, // start-without-complete shouldn't happen; clamp to 0 (_, existing) => existing with { CurrentInFlight = Math.Max(0, existing.CurrentInFlight - 1) }); } 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 33ed440a..d3e974e0 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/DriverHostActor.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/DriverHostActor.cs @@ -3198,6 +3198,10 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS private void SpawnChild(DriverInstanceSpec spec) { + // #609: an instance id removed by an earlier deployment (StopChild ⇒ Evict, which also holds back late + // items) may come back; lift the hold BEFORE the child exists, since it publishes health as it starts. + // The resilience tracker's equivalent is lifted by _invokerFactory.Create below. + _telemetryHub?.Readmit(spec.DriverInstanceId); var stub = DriverInstanceActor.ShouldStub(spec.DriverType, _localRoles); IDriver? driver = null; if (!stub) @@ -3388,7 +3392,10 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS // #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. Release the shared // resilience state FIRST: the resilience publisher re-emits the tracker's snapshots every tick, so - // evicting the hub alone was undone within one interval. + // evicting the hub alone was undone within one interval. Context.Stop is asynchronous, so the child can + // still finish an in-flight capability call (observed: its SubscribeBulk) and publish health AFTER these + // two calls; both therefore HOLD the id — late writes are dropped — until a later SpawnChild of the same + // id lifts it (the tracker via _invokerFactory.Create, the hub via Readmit). _invokerFactory.Release(driverInstanceId); _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 d2af09db..c6f559a3 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Telemetry/ITelemetryLocalHub.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Telemetry/ITelemetryLocalHub.cs @@ -89,6 +89,16 @@ public interface ITelemetryLocalHub /// 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. + /// The eviction also holds (#609): a removed instance's child stops asynchronously and can + /// still publish after this call (a health snapshot when an in-flight connect or subscribe finishes), + /// so until every later Health/Resilience item for the instance is dropped — not + /// cached and not fanned out. /// void Evict(string driverInstanceId); + + /// + /// Lifts an earlier so a driver instance spawned again under the same id (a later + /// deployment re-adding it) publishes normally. A no-op for an id that was never evicted. + /// + void Readmit(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 8d8fb40b..e1a8d7ee 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Telemetry/TelemetryLocalHub.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Telemetry/TelemetryLocalHub.cs @@ -42,6 +42,10 @@ public sealed class TelemetryLocalHub : ITelemetryLocalHub private readonly ConcurrentDictionary _healthCache = new(); private readonly ConcurrentDictionary<(string InstanceId, string HostName), TelemetryItem.Resilience> _resilienceCache = new(); + // Instances evicted and not since readmitted (#609). Their late Health/Resilience items are dropped, so a + // removed driver's still-stopping child cannot re-prime the cache it was just evicted from. Under _gate. + private readonly HashSet _evicted = new(StringComparer.Ordinal); + /// public void Emit(TelemetryItem item) { @@ -51,6 +55,10 @@ public sealed class TelemetryLocalHub : ITelemetryLocalHub { switch (item) { + case TelemetryItem.Health h when _evicted.Contains(h.E.DriverInstanceId): + return; + case TelemetryItem.Resilience r when _evicted.Contains(r.E.DriverInstanceId): + return; case TelemetryItem.Health h: _healthCache[h.E.DriverInstanceId] = h; break; @@ -72,6 +80,7 @@ public sealed class TelemetryLocalHub : ITelemetryLocalHub lock (_gate) { + _evicted.Add(driverInstanceId); _healthCache.TryRemove(driverInstanceId, out _); foreach (var key in _resilienceCache.Keys) if (string.Equals(key.InstanceId, driverInstanceId, StringComparison.Ordinal)) @@ -79,6 +88,15 @@ public sealed class TelemetryLocalHub : ITelemetryLocalHub } } + /// + public void Readmit(string driverInstanceId) + { + ArgumentNullException.ThrowIfNull(driverInstanceId); + + lock (_gate) + _evicted.Remove(driverInstanceId); + } + /// public ITelemetrySubscription Subscribe(int boundedCapacity) { diff --git a/tests/Core/ZB.MOM.WW.OtOpcUa.Core.Tests/Resilience/DriverCapabilityInvokerFactoryTests.cs b/tests/Core/ZB.MOM.WW.OtOpcUa.Core.Tests/Resilience/DriverCapabilityInvokerFactoryTests.cs index f4246677..13b7019b 100644 --- a/tests/Core/ZB.MOM.WW.OtOpcUa.Core.Tests/Resilience/DriverCapabilityInvokerFactoryTests.cs +++ b/tests/Core/ZB.MOM.WW.OtOpcUa.Core.Tests/Resilience/DriverCapabilityInvokerFactoryTests.cs @@ -45,6 +45,47 @@ public sealed class DriverCapabilityInvokerFactoryTests tracker.TryGet("d2", "d2")!.IsBreakerOpen.ShouldBeTrue("other instances must be untouched"); } + /// + /// #609 — a call the removed instance still had in flight finishes through its (now released) invoker + /// AFTER Release; its RecordCallStart/RecordSuccess/RecordCallComplete must not re-create the + /// tracker entry, or the resilience publisher re-emits the removed driver on every tick. The neighbour's + /// call through its own invoker is the positive control — the same path still records. + /// + [Fact] + public async Task A_call_through_a_released_invoker_records_nothing() + { + var tracker = new DriverResilienceStatusTracker(); + var factory = new DriverCapabilityInvokerFactory( + new DriverResiliencePipelineBuilder(statusTracker: tracker), tracker, _ => DriverTier.A); + var released = factory.Create("gone", "Modbus"); + var live = factory.Create("live", "Modbus"); + + factory.Release("gone"); + await released.ExecuteAsync(DriverCapability.Subscribe, "gone", _ => ValueTask.FromResult(1), CancellationToken.None); + await live.ExecuteAsync(DriverCapability.Subscribe, "live", _ => ValueTask.FromResult(1), CancellationToken.None); + + tracker.Snapshot().Select(e => e.DriverInstanceId).ShouldBe(new[] { "live" }); + } + + /// + /// #609 — re-adding the same instance id in a later deployment (a fresh Create after + /// Release) records again: the release must not outlive the removal. + /// + [Fact] + public async Task Create_after_Release_records_again_for_the_same_instance() + { + var tracker = new DriverResilienceStatusTracker(); + var factory = new DriverCapabilityInvokerFactory( + new DriverResiliencePipelineBuilder(statusTracker: tracker), tracker, _ => DriverTier.A); + factory.Create("d1", "Modbus"); + factory.Release("d1"); + + var readded = factory.Create("d1", "Modbus"); + await readded.ExecuteAsync(DriverCapability.Subscribe, "d1", _ => ValueTask.FromResult(1), CancellationToken.None); + + tracker.TryGet("d1", "d1").ShouldNotBeNull(); + } + /// /// End-to-end proof the per-instance override REACHES execution: a ResilienceConfig that sets /// Read's retryCount to 0 must suppress the tier-A Read retry — a failing Read is attempted once, diff --git a/tests/Core/ZB.MOM.WW.OtOpcUa.Core.Tests/Resilience/DriverResilienceStatusTrackerTests.cs b/tests/Core/ZB.MOM.WW.OtOpcUa.Core.Tests/Resilience/DriverResilienceStatusTrackerTests.cs index ba901a7e..5a53e905 100644 --- a/tests/Core/ZB.MOM.WW.OtOpcUa.Core.Tests/Resilience/DriverResilienceStatusTrackerTests.cs +++ b/tests/Core/ZB.MOM.WW.OtOpcUa.Core.Tests/Resilience/DriverResilienceStatusTrackerTests.cs @@ -172,4 +172,81 @@ public sealed class DriverResilienceStatusTrackerTests tracker.TryGet("drv", "host")!.ConsecutiveFailures.ShouldBe(500); } + + /// + /// #609 — Retire (a removed instance) drops the instance's snapshots AND ignores every write + /// that arrives after it. A removed driver's child stops asynchronously, so a capability call it still + /// had in flight finishes through its invoker after the host released it; each of those writes used to + /// re-create the entry, and the resilience publisher re-emitted the removed driver every tick, forever. + /// The neighbour is the positive control: the same writes against a live instance still land. + /// + [Fact] + public void Retire_drops_the_instance_and_ignores_every_late_write_for_it() + { + var tracker = new DriverResilienceStatusTracker(); + tracker.RecordBreakerOpen("gone", "host-a", Now); + tracker.RecordFailure("gone", "host-b", Now); + + tracker.Retire("gone").ShouldBe(2); + + foreach (var id in new[] { "gone", "live" }) + { + tracker.RecordCallStart(id, "host-a"); + tracker.RecordSuccess(id, "host-a", Now); + tracker.RecordCallComplete(id, "host-a"); + tracker.RecordFailure(id, "host-a", Now); + tracker.RecordBreakerOpen(id, "host-a", Now); + tracker.RecordBreakerClose(id, "host-a", Now); + tracker.RecordRecycle(id, "host-a", Now); + tracker.RecordFootprint(id, "host-a", 1, 2, Now); + } + + tracker.Snapshot().Select(e => e.DriverInstanceId).ShouldBe(new[] { "live" }); + tracker.TryGet("gone", "host-a").ShouldBeNull(); + tracker.TryGet("gone", "host-b").ShouldBeNull(); + } + + /// + /// #609 — a retired id is not retired forever: the same instance id re-added in a later deployment is + /// Reinstated (by the invoker factory's Create) and records again. + /// + [Fact] + public void Reinstate_lets_a_retired_instance_record_again() + { + var tracker = new DriverResilienceStatusTracker(); + tracker.Retire("drv"); + tracker.RecordFailure("drv", "host", Now); + tracker.TryGet("drv", "host").ShouldBeNull(); // control: still retired + + tracker.Reinstate("drv"); + tracker.RecordFailure("drv", "host", Now); + + tracker.TryGet("drv", "host")!.ConsecutiveFailures.ShouldBe(1); + } + + /// + /// #609 — a write racing Retire never survives it. Writers hammer the instance while it is + /// retired; whatever interleaving occurs, nothing is left once both sides finish. + /// + [Fact] + public async Task A_write_racing_Retire_never_resurrects_the_instance() + { + for (var round = 0; round < 50; round++) + { + var tracker = new DriverResilienceStatusTracker(); + tracker.RecordFailure("drv", "host", Now); + using var go = new ManualResetEventSlim(); + var writers = Enumerable.Range(0, 4).Select(_ => Task.Run(() => + { + go.Wait(); + for (var i = 0; i < 200; i++) tracker.RecordFailure("drv", "host", Now); + }, TestContext.Current.CancellationToken)).ToArray(); + + go.Set(); + tracker.Retire("drv"); + await Task.WhenAll(writers); + + tracker.TryGet("drv", "host").ShouldBeNull($"round {round}"); + } + } } 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 ee918813..c8d16e32 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 @@ -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(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))); - } + /// + /// #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 Create). + /// + [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 { "driver" }, + invokerFactory: invokerFactory, + telemetryHub: new ZB.MOM.WW.OtOpcUa.Runtime.Telemetry.TelemetryLocalHub())); - static List ReplayedResilienceIds(ZB.MOM.WW.OtOpcUa.Runtime.Telemetry.TelemetryLocalHub h) + actor.Tell(new DispatchDeployment(SeedV3Deployment(db, RevA), RevA, CorrelationId.NewId())); + coordinator.ExpectMsg(Timeout).Outcome.ShouldBe(ApplyAckOutcome.Applied); + 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); + + var revC = RevisionHash.Parse(new string('c', 64)); + actor.Tell(new DispatchDeployment(SeedV3Deployment(db, revC), revC, CorrelationId.NewId())); + coordinator.ExpectMsg(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 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; + } + + /// + /// Wraps a real invoker factory so every call is HELD before it + /// reaches the real invoker until — 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. + /// + private sealed class GatedSubscribeInvokerFactory(IDriverCapabilityInvokerFactory inner) : IDriverCapabilityInvokerFactory + { + private readonly TaskCompletionSource _gate = new(TaskCreationOptions.RunContinuationsAsynchronously); + + /// Set once a subscribe has reached the gate. + public ManualResetEventSlim SubscribeEntered { get; } = new(); + + /// Set once a held subscribe has returned from the REAL invoker, i.e. its tracker writes are done. + 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(); - 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 ExecuteAsync( + DriverCapability capability, string hostName, + Func> 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 callSite, CancellationToken cancellationToken) => + real.ExecuteAsync(capability, hostName, callSite, cancellationToken); + + public ValueTask ExecuteWriteAsync( + string hostName, bool isIdempotent, + Func> callSite, CancellationToken cancellationToken) => + real.ExecuteWriteAsync(hostName, isIdempotent, callSite, cancellationToken); } } 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 86141791..e2d5ac24 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 @@ -58,6 +58,55 @@ public sealed class TelemetryLocalHubTests items.OfType().Single().E.DriverInstanceId.ShouldBe("drv-b"); } + /// + /// #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. + /// + [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" }); + } + + /// + /// #609 — eviction does not outlive the removal: the same instance id re-spawned in a later deployment + /// is Readmitted and its snapshots are cached again. + /// + [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, + _ => "", + }; + [Fact] public void New_subscriber_is_primed_with_cached_Health_snapshot() {