From 79d6b0a4b23b4732c979fbc904e7a13b23ff4811 Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Thu, 22 Oct 2026 06:36:12 -0400 Subject: [PATCH] =?UTF-8?q?fix(runtime):=20lifecycle=20housekeeping=20?= =?UTF-8?q?=E2=80=94=20reprovision=20stamp,=20Stale=20drops,=20off-thread?= =?UTF-8?q?=20DB=20probe,=20tombstone=20sweep=20(A4)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - DriverHostActor.StopChild removes the driver's _lastReprovisionUtc stamp (and its deferred-reprovision timer): the map grew per removed id and a re-added driver inherited a backoff it never earned (02/L2). - Stale now registers explicit drops for AttributeValueBatchPublished, AttributeAlarmPublished, ArtifactForApply and ApplyWedged instead of letting them hit Unhandled (02/H5). - DbHealthProbeActor runs its SELECT on the thread pool and pipes the outcome back (single-flight), with a 2 s command timeout on relational providers, so a hung SQL Server no longer pins the actor thread and the cached GetStatus answer stays reachable (02/P7). - TelemetryLocalHub eviction tombstones are timestamped and swept past a 10 min TTL on the next Evict; a TimeProvider ctor is the test seam (02/L4). - Tests: ValuePublishes.Single helper replaces the per-value AttributeValuePublished in CalculationDependencyFlowTests (02/U4, test half; the production receive is owned by Lane E). - R05/8 audit: PollGroupEngine.Unsubscribe has no production caller; every driver already uses UnsubscribeAsync (no change). Refs #731 Claude-Session: https://claude.ai/code/session_01Ub9hGmYLQvn9XETzw7CMsT --- .../Drivers/DriverHostActor.cs | 23 ++++++ .../Health/DbHealthProbeActor.cs | 58 +++++++++++++-- .../Telemetry/TelemetryLocalHub.cs | 53 ++++++++++++-- .../Drivers/CalculationDependencyFlowTests.cs | 12 +-- .../DriverHostActorReprovisionStampTests.cs | 63 ++++++++++++++++ .../Drivers/DriverHostActorStaleDropTests.cs | 66 +++++++++++++++++ .../Harness/V3ArtifactSeeder.cs | 32 +++++++- .../Harness/ValuePublishes.cs | 23 ++++++ .../Health/DbHealthProbeOffThreadTests.cs | 73 +++++++++++++++++++ .../Health/HealthProbeActorTests.cs | 14 ++-- .../TelemetryLocalHubTombstoneSweepTests.cs | 69 ++++++++++++++++++ 11 files changed, 460 insertions(+), 26 deletions(-) create mode 100644 tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverHostActorReprovisionStampTests.cs create mode 100644 tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverHostActorStaleDropTests.cs create mode 100644 tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Harness/ValuePublishes.cs create mode 100644 tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Health/DbHealthProbeOffThreadTests.cs create mode 100644 tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Telemetry/TelemetryLocalHubTombstoneSweepTests.cs 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 2d1bec44..0405be1a 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/DriverHostActor.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/DriverHostActor.cs @@ -2489,6 +2489,24 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS // A child connectivity transition while the host is Stale has no live address space to annotate — drop // it (the post-recovery rebuild re-materialises conditions, and the child re-announces on its next entry). Receive(_ => { }); + // Likewise a child's value / alarm burst: Stale is entered from Bootstrap's catch before any child is + // spawned, so none should arrive — but the orphan-replay BeginDirectApply in that same try block has + // already started a pipe-back before the catch can run, so "cannot happen" rests on ordering. Explicit + // drops, so a burst that does land is a logged drop rather than an Unhandled dead-letter. + Receive(m => + _log.Debug("DriverHost {Node}: dropping a {Count}-item value burst from {Id} while Stale (no address space)", + _localNode, m.Items.Count, m.DriverInstanceId)); + Receive(m => + _log.Debug("DriverHost {Node}: dropping a native alarm transition ({Cond}) from {Id} while Stale", + _localNode, m.Args.ConditionId, m.DriverInstanceId)); + // A late artifact pipe-back from the apply that was in flight when the host fell Stale: that apply is + // lost (the coordinator's deadline and the boot-time orphan replay recover it). Warning, not Debug — + // an apply outcome is being discarded. + Receive(m => + _log.Warning("DriverHost {Node}: dropping the artifact for {Id} (rev {Rev}) that arrived while Stale; the apply is abandoned", + _localNode, m.DeploymentId, m.Revision)); + Receive(m => + _log.Debug("DriverHost {Node}: late apply-wedge tick for {Id} while Stale; dropping", _localNode, m.DeploymentId)); // Likewise the 03/S15 recovery audit: a timer armed before the host went Stale has nothing to report // on (no latch was set while Stale, and the rebuild re-establishes every node's status anyway). Receive(_ => { }); @@ -3771,6 +3789,11 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS // id lifts it (the tracker via _invokerFactory.Create, the hub via Readmit). _invokerFactory.Release(driverInstanceId); _telemetryHub?.Evict(driverInstanceId); + // The reprovision rate-limit stamp belongs to the incarnation history that just ended. Kept, it grew the + // map by one entry per id ever reprovisioned-then-removed AND made a later re-add of the same id inherit a + // backoff it never earned (its first crash was deferred as a "crash loop"). + _lastReprovisionUtc.Remove(driverInstanceId); + Timers.Cancel(ReprovisionTimerKey(driverInstanceId)); _log.Info("DriverHost {Node}: stopped driver child {Id}", _localNode, driverInstanceId); } diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Health/DbHealthProbeActor.cs b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Health/DbHealthProbeActor.cs index 5ea6c44f..8134c914 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Health/DbHealthProbeActor.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Health/DbHealthProbeActor.cs @@ -7,21 +7,37 @@ namespace ZB.MOM.WW.OtOpcUa.Runtime.Health; /// /// Single-flight cached health probe against the ConfigDb. Reads cached state via -/// Ask<DbHealthStatus>; a single SELECT 1 runs at most every RefreshInterval. +/// Ask<DbHealthStatus>; a single SELECT runs at most every RefreshInterval. /// Consumed by both the host's /health/ready endpoint and /// RedundancyStateActor's stale calc. /// +/// +/// The probe runs OFF the mailbox (a thread-pool task that pipes its outcome back), with a bounded +/// command timeout. It used to run synchronously on the actor thread, relying on SqlClient's implicit +/// 15 s connect / 30 s command timeouts — so a hanging SQL Server pinned a dispatcher thread for up to +/// 30 s per tick, later ticks queued behind it, and the cached answer was +/// unreachable for exactly as long as the database was. Single-flight: a tick that lands while a probe +/// is still running is skipped rather than stacked. +/// public sealed class DbHealthProbeActor : ReceiveActor, IWithTimers { public static readonly TimeSpan RefreshInterval = TimeSpan.FromSeconds(5); + /// Bound on the probe's SQL command, relational providers only (the in-memory provider the + /// actor tests run against has no command timeout). + public static readonly TimeSpan CommandTimeout = TimeSpan.FromSeconds(2); + public sealed class GetStatus { public static readonly GetStatus Instance = new(); private GetStatus() { } } public sealed record DbHealthStatus(bool Reachable, DateTime AsOfUtc, string? LastError); public sealed class Tick { public static readonly Tick Instance = new(); private Tick() { } } + /// Piped back by the off-thread probe with its outcome. + private sealed record ProbeCompleted(DbHealthStatus Status); + private readonly IDbContextFactory _dbFactory; private readonly ILoggingAdapter _log = Context.GetLogger(); private DbHealthStatus _last = new(false, DateTime.MinValue, "not probed yet"); + private bool _probeInFlight; /// Gets or sets the timer scheduler for periodic health probes. public ITimerScheduler Timers { get; set; } = null!; @@ -38,28 +54,54 @@ public sealed class DbHealthProbeActor : ReceiveActor, IWithTimers { _dbFactory = dbFactory; Receive(_ => Sender.Tell(_last)); - Receive(_ => RunProbe()); + Receive(_ => StartProbe()); + Receive(msg => + { + _probeInFlight = false; + _last = msg.Status; + if (!msg.Status.Reachable) + _log.Warning("DbHealthProbe: probe failed: {Error}", msg.Status.LastError); + }); } /// protected override void PreStart() { - RunProbe(); + StartProbe(); Timers.StartPeriodicTimer("probe", Tick.Instance, RefreshInterval); } - private void RunProbe() + private void StartProbe() + { + if (_probeInFlight) + { + _log.Debug("DbHealthProbe: previous probe still running; skipping this tick"); + return; + } + + _probeInFlight = true; + var factory = _dbFactory; + // Never touches actor state; the outcome comes back as a message. + Task.Run(() => Probe(factory)).PipeTo( + Self, + success: status => new ProbeCompleted(status), + failure: ex => new ProbeCompleted(new DbHealthStatus(false, DateTime.UtcNow, ex.Message))); + } + + private static DbHealthStatus Probe(IDbContextFactory factory) { try { - using var db = _dbFactory.CreateDbContext(); + using var db = factory.CreateDbContext(); + // Local to this lease — the pool resets it on return (see DriverHostActor.LoadArtifactBlob). + if (db.Database.IsRelational()) + db.Database.SetCommandTimeout(CommandTimeout); _ = db.Deployments.AsNoTracking().Take(1).ToList(); - _last = new DbHealthStatus(true, DateTime.UtcNow, null); + return new DbHealthStatus(true, DateTime.UtcNow, null); } catch (Exception ex) { - _last = new DbHealthStatus(false, DateTime.UtcNow, ex.Message); - _log.Warning(ex, "DbHealthProbe: probe failed"); + return new DbHealthStatus(false, DateTime.UtcNow, ex.Message); } } } 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 57d11973..1e58c54b 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Telemetry/TelemetryLocalHub.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Telemetry/TelemetryLocalHub.cs @@ -27,7 +27,27 @@ namespace ZB.MOM.WW.OtOpcUa.Runtime.Telemetry; /// public sealed class TelemetryLocalHub : ITelemetryLocalHub { + /// + /// How long an eviction tombstone (see ) is kept. It only has to outlive the + /// removed child's last late write — an in-flight capability call that completes after + /// StopChild, bounded by the longest capability timeout (seconds) plus the child's own stop — + /// so ten minutes is a wide margin. Without a sweep a driver removed for good kept its id forever. + /// + public static readonly TimeSpan EvictionTombstoneTtl = TimeSpan.FromMinutes(10); + private readonly object _gate = new(); + private readonly TimeProvider _time; + + /// Initializes a hub on the system clock. + public TelemetryLocalHub() : this(TimeProvider.System) { } + + /// Initializes a hub on — the tombstone-sweep test seam. + /// The clock the eviction tombstones are stamped and swept by. + public TelemetryLocalHub(TimeProvider time) + { + ArgumentNullException.ThrowIfNull(time); + _time = time; + } // Plain Dictionary, NOT ConcurrentDictionary: every access is already under _gate, and // ConcurrentDictionary.Values would materialize a fresh List (and take its own internal lock) on @@ -42,9 +62,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. 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); + // Instances evicted and not since readmitted, stamped with the eviction time. 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. Swept past EvictionTombstoneTtl on the next Evict (never on the Emit hot path). Under _gate. + private readonly Dictionary _evicted = new(StringComparer.Ordinal); /// public void Emit(TelemetryItem item) @@ -55,9 +76,9 @@ public sealed class TelemetryLocalHub : ITelemetryLocalHub { switch (item) { - case TelemetryItem.Health h when _evicted.Contains(h.E.DriverInstanceId): + case TelemetryItem.Health h when _evicted.ContainsKey(h.E.DriverInstanceId): return; - case TelemetryItem.Resilience r when _evicted.Contains(r.E.DriverInstanceId): + case TelemetryItem.Resilience r when _evicted.ContainsKey(r.E.DriverInstanceId): return; case TelemetryItem.Health h: _healthCache[h.E.DriverInstanceId] = h; @@ -80,7 +101,9 @@ public sealed class TelemetryLocalHub : ITelemetryLocalHub lock (_gate) { - _evicted.Add(driverInstanceId); + var now = _time.GetUtcNow(); + SweepExpiredTombstones(now); + _evicted[driverInstanceId] = now; _healthCache.TryRemove(driverInstanceId, out _); foreach (var key in _resilienceCache.Keys) if (string.Equals(key.InstanceId, driverInstanceId, StringComparison.Ordinal)) @@ -97,6 +120,24 @@ public sealed class TelemetryLocalHub : ITelemetryLocalHub _evicted.Remove(driverInstanceId); } + /// Drops tombstones older than . Called under + /// from only — the set is bounded by distinct removed ids, so a linear pass per + /// eviction is cheap, and Emit never pays for it. + private void SweepExpiredTombstones(DateTimeOffset now) + { + if (_evicted.Count == 0) return; + List? expired = null; + foreach (var (id, evictedAt) in _evicted) + { + if (now - evictedAt >= EvictionTombstoneTtl) + (expired ??= new List()).Add(id); + } + + if (expired is null) return; + foreach (var id in expired) + _evicted.Remove(id); + } + /// public ITelemetrySubscription Subscribe(int boundedCapacity) { diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/CalculationDependencyFlowTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/CalculationDependencyFlowTests.cs index 008004d5..d342023c 100644 --- a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/CalculationDependencyFlowTests.cs +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/CalculationDependencyFlowTests.cs @@ -64,7 +64,7 @@ public sealed class CalculationDependencyFlowTests : RuntimeActorTestBase { published.Enqueue(e); var quality = e.Snapshot.StatusCode == 0u ? OpcUaStatus.Good : OpcUaStatus.Bad; - mux.Tell(new DriverInstanceActor.AttributeValuePublished( + mux.Tell(ValuePublishes.Single( "calc-drv", e.FullReference, e.Snapshot.Value, quality, Ts)); }; @@ -75,7 +75,7 @@ public sealed class CalculationDependencyFlowTests : RuntimeActorTestBase // compute dedups) until the computed value has flowed all the way through. AwaitAssert(() => { - mux.Tell(new DriverInstanceActor.AttributeValuePublished( + mux.Tell(ValuePublishes.Single( "src-drv", "src/a", 21.0, OpcUaStatus.Good, Ts)); // Leg 1 — dep flow: src/a=21 → A computes 42 and publishes on calc/engine/out. @@ -118,14 +118,14 @@ public sealed class CalculationDependencyFlowTests : RuntimeActorTestBase { published.Enqueue(e); var quality = e.Snapshot.StatusCode == 0u ? OpcUaStatus.Good : OpcUaStatus.Bad; - mux.Tell(new DriverInstanceActor.AttributeValuePublished("calc-drv", e.FullReference, e.Snapshot.Value, quality, Ts)); + mux.Tell(ValuePublishes.Single("calc-drv", e.FullReference, e.Snapshot.Value, quality, Ts)); }; var adapter = Sys.ActorOf(DependencyConsumerMuxAdapter.Props(driver, mux), "dep-adapter2"); // Leg 1 — before redeploy: src/a=2 → keep=20, remove=7. AwaitAssert(() => { - mux.Tell(new DriverInstanceActor.AttributeValuePublished("src-drv", "src/a", 2.0, OpcUaStatus.Good, Ts)); + mux.Tell(ValuePublishes.Single("src-drv", "src/a", 2.0, OpcUaStatus.Good, Ts)); published.ShouldContain(e => e.FullReference == "calc/keep" && Equals(e.Snapshot.Value, 20.0)); published.ShouldContain(e => e.FullReference == "calc/remove" && Equals(e.Snapshot.Value, 7.0)); }, PresenceBudget, TimeSpan.FromMilliseconds(100)); @@ -150,8 +150,8 @@ public sealed class CalculationDependencyFlowTests : RuntimeActorTestBase // new result; the REMOVED tag never publishes a value derived from the post-redeploy input. AwaitAssert(() => { - mux.Tell(new DriverInstanceActor.AttributeValuePublished("src-drv", "src/b", 50.0, OpcUaStatus.Good, Ts)); - mux.Tell(new DriverInstanceActor.AttributeValuePublished("src-drv", "src/a", 3.0, OpcUaStatus.Good, Ts)); + mux.Tell(ValuePublishes.Single("src-drv", "src/b", 50.0, OpcUaStatus.Good, Ts)); + mux.Tell(ValuePublishes.Single("src-drv", "src/a", 3.0, OpcUaStatus.Good, Ts)); // (a) new dep B flows through the RE-REGISTERED interest → new tag computes: 50 + 100 = 150. published.ShouldContain(e => e.FullReference == "calc/add" && Equals(e.Snapshot.Value, 150.0)); diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverHostActorReprovisionStampTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverHostActorReprovisionStampTests.cs new file mode 100644 index 00000000..509e7ea9 --- /dev/null +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverHostActorReprovisionStampTests.cs @@ -0,0 +1,63 @@ +using Akka.Actor; +using Shouldly; +using Xunit; +using ZB.MOM.WW.OtOpcUa.Commons.Types; +using ZB.MOM.WW.OtOpcUa.Runtime.Tests.Harness; + +namespace ZB.MOM.WW.OtOpcUa.Runtime.Tests.Drivers; + +/// +/// Review 02 L2: DriverHostActor._lastReprovisionUtc was never pruned — a driver removed by a deploy +/// after a supervision crash kept its stamp, so the map grew with every id ever reprovisioned-then-removed, +/// and (the observable half) a driver removed and later re-added inherited the OLD stamp: its first crash +/// after the re-add was deferred by the backoff as if it were crash-looping, although the new incarnation had +/// never crashed. StopChild now drops the stamp. +/// +public sealed class DriverHostActorReprovisionStampTests : RuntimeActorTestBase +{ + private static readonly RevisionHash RevA = RevisionHash.Parse(new string('a', 64)); + private static readonly RevisionHash RevB = RevisionHash.Parse(new string('b', 64)); + private static readonly RevisionHash RevC = RevisionHash.Parse(new string('c', 64)); + private static readonly NodeId TestNode = NodeId.Parse("node-1:4053"); + private const string DriverId = "DI-1"; + private const string SpeedPath = "Plant/Modbus/dev1/speed"; + + [Fact] + public void A_driver_removed_and_re_added_does_not_inherit_the_old_incarnations_reprovision_stamp() + { + var db = NewInMemoryDbFactory(); + // Far longer than the presence budget below: an inherited stamp would defer the re-added driver's + // reprovision past it, which is the failure this test reports. + var backoff = TimeSpan.FromSeconds(60); + var factory = new RecordingSubscribableDriverFactory("Modbus"); + var withDriver = new[] { (DriverId, "Modbus", "dev1", "speed", "Float", false) }; + var none = Array.Empty<(string, string, string, string, string?)>(); + + var revA = V3ArtifactSeeder.SeedV3Deployment(db, RevA, withDriver, none, enabled: true); + var (host, _, _, coordinator) = V3ArtifactSeeder.SpawnHostAndApplyWithCoordinator( + this, db, TestNode, revA, RevA, factory, reprovisionBackoff: backoff); + AwaitAssert(() => factory.Created.Count.ShouldBe(1), PresenceBudget); + AwaitAssert(() => factory.Created[0].SubscribeCount.ShouldBe(1), PresenceBudget); + + // Crash 1 on the first incarnation: reprovisioned immediately, which stamps the id. + factory.Created[0].FireMalformedDataChange(SpeedPath); + AwaitAssert(() => factory.Created[0].InitializeCount.ShouldBe(2), PresenceBudget); + + // Deploy B removes the driver (StopChild), deploy C re-adds it: a brand-new driver object. + var revB = V3ArtifactSeeder.SeedV3Deployment( + db, RevB, Array.Empty<(string, string, string, string, string, bool)>(), none, enabled: true); + V3ArtifactSeeder.Dispatch(host, coordinator, revB, RevB); + var revC = V3ArtifactSeeder.SeedV3Deployment(db, RevC, withDriver, none, enabled: true); + V3ArtifactSeeder.Dispatch(host, coordinator, revC, RevC); + AwaitAssert(() => factory.Created.Count.ShouldBe(2), PresenceBudget); + var reAdded = factory.Created[1]; + AwaitAssert(() => reAdded.SubscribeCount.ShouldBe(1), PresenceBudget); + + // Its FIRST crash must reprovision immediately — it has no history of its own. + reAdded.FireMalformedDataChange(SpeedPath); + AwaitAssert( + () => reAdded.InitializeCount.ShouldBe( + 2, "the re-added driver's first reprovision was deferred by the REMOVED incarnation's stamp"), + duration: TimeSpan.FromSeconds(5)); + } +} diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverHostActorStaleDropTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverHostActorStaleDropTests.cs new file mode 100644 index 00000000..442f3909 --- /dev/null +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverHostActorStaleDropTests.cs @@ -0,0 +1,66 @@ +using Akka.Actor; +using Akka.Event; +using Microsoft.EntityFrameworkCore; +using Shouldly; +using Xunit; +using ZB.MOM.WW.OtOpcUa.Commons.OpcUa; +using ZB.MOM.WW.OtOpcUa.Commons.Types; +using ZB.MOM.WW.OtOpcUa.Configuration; +using ZB.MOM.WW.OtOpcUa.Core.Abstractions; +using ZB.MOM.WW.OtOpcUa.Runtime.Drivers; +using ZB.MOM.WW.OtOpcUa.Runtime.Tests.Harness; + +namespace ZB.MOM.WW.OtOpcUa.Runtime.Tests.Drivers; + +/// +/// Review 02 H5: the Stale behaviour registered no receive for a driver child's value/alarm bursts (nor +/// for a late ArtifactForApply / ApplyWedged pipe-back), so any that landed there hit +/// Unhandled — dead-letter noise and a dropped message with no log naming why. They are now explicit +/// drops. The two private pipe-back records cannot be constructed from a test; the two public child +/// messages are pinned here, and the private pair by the same registration pattern (compile-time). +/// +public sealed class DriverHostActorStaleDropTests : RuntimeActorTestBase +{ + private static readonly NodeId TestNode = NodeId.Parse("stale-drop-test"); + + [Fact] + public void Child_value_and_alarm_bursts_landing_in_Stale_are_dropped_not_unhandled() + { + var unhandled = CreateTestProbe(); + Sys.EventStream.Subscribe(unhandled.Ref, typeof(UnhandledMessage)); + + // ThrowingDbFactory ⇒ Bootstrap's catch ⇒ Become(Stale). + var host = Sys.ActorOf(DriverHostActor.Props( + new ThrowingDbFactory(), TestNode, CreateTestProbe().Ref, + localRoles: new HashSet { "driver" })); + // Barrier: Stale answers RouteNodeWrite synchronously, so a reply proves the host is in Stale. + var asker = CreateTestProbe(); + host.Tell(new DriverHostActor.RouteNodeWrite("eq-x/speed", 0.0, AddressSpaceRealm.Uns), asker.Ref); + asker.ExpectMsg(PresenceBudget).Reason!.ShouldContain("stale"); + + // Positive control: an unknown message IS reported as unhandled by this host. + host.Tell("not a message the host knows"); + unhandled.ExpectMsg(m => ReferenceEquals(m.Recipient, host) || m.Recipient.Equals(host), PresenceBudget); + + host.Tell(new DriverInstanceActor.AttributeValueBatchPublished( + "DI-1", new[] { new DriverInstanceActor.AttributeValueItem("Plant/x/y", 1.0, OpcUaStatus.Good, DateTime.UtcNow) })); + host.Tell(new DriverInstanceActor.AttributeAlarmPublished( + "DI-1", new AlarmEventArgs(new Handle(), "Plant/x/alarm", "Plant/x/alarm", "AlarmCondition", "test", + AlarmSeverity.High, DateTime.UtcNow))); + // Barrier again, then the absence window. + host.Tell(new DriverHostActor.RouteNodeWrite("eq-x/speed", 0.0, AddressSpaceRealm.Uns), asker.Ref); + asker.ExpectMsg(PresenceBudget); + + unhandled.ExpectNoMsg(TimeSpan.FromMilliseconds(500)); + } + + private sealed class Handle : IAlarmSubscriptionHandle + { + public string DiagnosticId => "stale-test"; + } + + private sealed class ThrowingDbFactory : IDbContextFactory + { + public OtOpcUaConfigDbContext CreateDbContext() => throw new InvalidOperationException("db unreachable"); + } +} diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Harness/V3ArtifactSeeder.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Harness/V3ArtifactSeeder.cs index e928b1b2..20c2a62f 100644 --- a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Harness/V3ArtifactSeeder.cs +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Harness/V3ArtifactSeeder.cs @@ -201,7 +201,37 @@ public static class V3ArtifactSeeder return (actor, mux, coordinator); } - private static void Dispatch(IActorRef actor, TestProbe coordinator, DeploymentId deploymentId, RevisionHash rev) + /// The same arrangement as , also returning the coordinator probe so a + /// test can FURTHER deployments to the same host (remove a driver, re-add it). + /// The calling TestKit, supplying the ActorSystem + probes. + /// The factory whose deployment was seeded by . + /// The host's local node id. + /// The deployment to dispatch. + /// The revision hash the deployment was sealed under. + /// Optional driver factory forwarded to DriverHostActor.Props. + /// Optional per-driver supervision-reprovision rate limit. + /// The host actor plus the publish, mux and coordinator probes wired into it. + public static (IActorRef Actor, TestProbe Publish, TestProbe Mux, TestProbe Coordinator) SpawnHostAndApplyWithCoordinator( + TestKitBase kit, IDbContextFactory db, + NodeId localNode, DeploymentId deploymentId, RevisionHash rev, + IDriverFactory? driverFactory = null, + TimeSpan? reprovisionBackoff = null) + { + var publish = kit.CreateCleanOutcomePublishProbe(); + var (actor, mux, coordinator) = Spawn(kit, db, localNode, publish.Ref, driverFactory, reprovisionBackoff); + + Dispatch(actor, coordinator, deploymentId, rev); + publish.ExpectMsg(RuntimeActorTestBase.PresenceBudget); + + return (actor, publish, mux, coordinator); + } + + /// Dispatches to and waits for its Applied ack. + /// The host. + /// The coordinator probe the host acks to. + /// The deployment to dispatch. + /// Its revision hash. + public static void Dispatch(IActorRef actor, TestProbe coordinator, DeploymentId deploymentId, RevisionHash rev) { actor.Tell(new DispatchDeployment(deploymentId, rev, CorrelationId.NewId())); coordinator.ExpectMsg(RuntimeActorTestBase.PresenceBudget) diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Harness/ValuePublishes.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Harness/ValuePublishes.cs new file mode 100644 index 00000000..d0582eff --- /dev/null +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Harness/ValuePublishes.cs @@ -0,0 +1,23 @@ +using ZB.MOM.WW.OtOpcUa.Runtime.Drivers; + +namespace ZB.MOM.WW.OtOpcUa.Runtime.Tests.Harness; + +/// +/// Builds the ONE message shape a driver child sends for its values — a +/// — for tests that push a single value into the +/// dependency mux by hand. The per-value has no +/// production sender (02/U4); tests should not keep one alive either. +/// +public static class ValuePublishes +{ + /// A one-item batch for . + /// The publishing driver's id. + /// The value's reference (RawPath). + /// The value. + /// The full OPC UA status code. + /// The source timestamp. + /// The batch message. + public static DriverInstanceActor.AttributeValueBatchPublished Single( + string driverInstanceId, string fullReference, object? value, uint statusCode, DateTime timestampUtc) => + new(driverInstanceId, new[] { new DriverInstanceActor.AttributeValueItem(fullReference, value, statusCode, timestampUtc) }); +} diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Health/DbHealthProbeOffThreadTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Health/DbHealthProbeOffThreadTests.cs new file mode 100644 index 00000000..c10e24f2 --- /dev/null +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Health/DbHealthProbeOffThreadTests.cs @@ -0,0 +1,73 @@ +using Akka.Actor; +using Microsoft.EntityFrameworkCore; +using Shouldly; +using Xunit; +using ZB.MOM.WW.OtOpcUa.Configuration; +using ZB.MOM.WW.OtOpcUa.Runtime.Health; +using ZB.MOM.WW.OtOpcUa.Runtime.Tests.Harness; + +namespace ZB.MOM.WW.OtOpcUa.Runtime.Tests.Health; + +/// +/// Review 02 P7: ran its SELECT synchronously on the actor thread, so a +/// hanging SQL connect/command held the dispatcher thread for the implicit 15 s / 30 s SqlClient timeouts and +/// queued every later tick (and every GetStatus Ask) behind it. The probe now runs off the mailbox +/// and pipes its outcome back, so the cached answer stays available while a probe hangs. +/// +public sealed class DbHealthProbeOffThreadTests : RuntimeActorTestBase +{ + [Fact] + public async Task GetStatus_answers_from_cache_while_a_probe_is_hung() + { + var factory = new HangingDbFactory(); + try + { + var actor = Sys.ActorOf(DbHealthProbeActor.Props(factory)); + AwaitCondition(() => factory.Entered > 0, PresenceBudget); + + var status = await actor.Ask( + DbHealthProbeActor.GetStatus.Instance, TimeSpan.FromSeconds(3)); + + status.Reachable.ShouldBeFalse(); + status.LastError.ShouldBe("not probed yet"); + factory.Entered.ShouldBe(1, "a second tick must not stack a second probe behind a hung one"); + } + finally + { + factory.Release(); + } + } + + [Fact] + public async Task A_probe_that_completes_updates_the_cached_status() + { + // Positive control for the off-thread path: the piped-back outcome lands in the cache. + var db = NewInMemoryDbFactory(); + var actor = Sys.ActorOf(DbHealthProbeActor.Props(db)); + + await AwaitAssertAsync(async () => + { + var status = await actor.Ask( + DbHealthProbeActor.GetStatus.Instance, TimeSpan.FromSeconds(3)); + status.Reachable.ShouldBeTrue(); + }, PresenceBudget); + } + + /// Blocks every CreateDbContext until released — a connect that never answers. + private sealed class HangingDbFactory : IDbContextFactory + { + private readonly ManualResetEventSlim _release = new(false); + private int _entered; + + public int Entered => Volatile.Read(ref _entered); + + public void Release() => _release.Set(); + + public OtOpcUaConfigDbContext CreateDbContext() + { + Interlocked.Increment(ref _entered); + _release.Wait(); + throw new InvalidOperationException("released without a database"); + } + } +} diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Health/HealthProbeActorTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Health/HealthProbeActorTests.cs index cf5863cd..0dacfebf 100644 --- a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Health/HealthProbeActorTests.cs +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Health/HealthProbeActorTests.cs @@ -15,18 +15,22 @@ namespace ZB.MOM.WW.OtOpcUa.Runtime.Tests.Health; public sealed class HealthProbeActorTests : RuntimeActorTestBase { - /// Verifies that the DB health probe actor returns reachable status against an in-memory database. + /// Verifies that the DB health probe actor returns reachable status against an in-memory database. + /// The probe runs off the mailbox (02/P7), so the first Ask may still see "not probed yet" — poll. [Fact] public async Task DbHealthProbeActor_returns_reachable_against_in_memory_db() { var db = NewInMemoryDbFactory(); var actor = Sys.ActorOf(DbHealthProbeActor.Props(db)); - var status = await actor.Ask( - DbHealthProbeActor.GetStatus.Instance, PresenceBudget); + await AwaitAssertAsync(async () => + { + var status = await actor.Ask( + DbHealthProbeActor.GetStatus.Instance, PresenceBudget); - status.Reachable.ShouldBeTrue(); - status.LastError.ShouldBeNull(); + status.Reachable.ShouldBeTrue(); + status.LastError.ShouldBeNull(); + }, PresenceBudget); } /// Verifies that the peer OPC UA probe actor reports Ok true against a live listener. diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Telemetry/TelemetryLocalHubTombstoneSweepTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Telemetry/TelemetryLocalHubTombstoneSweepTests.cs new file mode 100644 index 00000000..6e0b8c64 --- /dev/null +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Telemetry/TelemetryLocalHubTombstoneSweepTests.cs @@ -0,0 +1,69 @@ +using Shouldly; +using Xunit; +using ZB.MOM.WW.OtOpcUa.Commons.Messages.Drivers; +using ZB.MOM.WW.OtOpcUa.Runtime.Telemetry; + +namespace ZB.MOM.WW.OtOpcUa.Runtime.Tests.Telemetry; + +/// +/// Review 02 L4: 's tombstone set only ever shrank on a re-spawn of the +/// SAME id, so a driver removed for good kept its id forever. The tombstone is load-bearing (#609 — it holds +/// back the removed child's late writes), so it cannot simply be cleared; it is now timestamped and swept +/// once older than , which outlives any in-flight +/// capability call by a wide margin. +/// +public sealed class TelemetryLocalHubTombstoneSweepTests +{ + private static DriverHealthChanged Health(string instanceId) => + new("cluster-a", instanceId, "Healthy", null, null, 0, DateTime.UtcNow); + + private static List Drain(ITelemetrySubscription sub) + { + var items = new List(); + while (sub.Reader.TryRead(out var item)) items.Add(item); + return items; + } + + [Fact] + public void A_tombstone_holds_late_writes_until_the_ttl_then_is_swept() + { + var clock = new ManualTime { Now = new DateTimeOffset(2026, 10, 22, 12, 0, 0, TimeSpan.Zero) }; + var hub = new TelemetryLocalHub(clock); + hub.Emit(new TelemetryItem.Health(Health("drv-a"))); + hub.Evict("drv-a"); + + // Inside the TTL: the late write is dropped (the #609 contract) — and another eviction's sweep does + // not lift a tombstone that is still young. + clock.Now += TelemetryLocalHub.EvictionTombstoneTtl / 2; + hub.Evict("drv-b"); + hub.Emit(new TelemetryItem.Health(Health("drv-a"))); + using (var sub = hub.Subscribe(16)) + Drain(sub).ShouldBeEmpty("a late write inside the TTL must still be held back"); + + // Past the TTL: the next eviction sweeps drv-a's tombstone and its writes land again. + clock.Now += TelemetryLocalHub.EvictionTombstoneTtl; + hub.Evict("drv-c"); + hub.Emit(new TelemetryItem.Health(Health("drv-a"))); + using (var sub = hub.Subscribe(16)) + Drain(sub).OfType().Single().E.DriverInstanceId.ShouldBe("drv-a"); + } + + [Fact] + public void Readmit_lifts_a_tombstone_before_its_ttl() + { + var clock = new ManualTime { Now = DateTimeOffset.UtcNow }; + var hub = new TelemetryLocalHub(clock); + hub.Evict("drv-a"); + hub.Readmit("drv-a"); + + hub.Emit(new TelemetryItem.Health(Health("drv-a"))); + using var sub = hub.Subscribe(16); + Drain(sub).Count.ShouldBe(1); + } + + private sealed class ManualTime : TimeProvider + { + public DateTimeOffset Now { get; set; } + public override DateTimeOffset GetUtcNow() => Now; + } +}