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()
{