From 8e97fc94fd73112e8b3562a8a27fec7db77c580d Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Wed, 11 Nov 2026 12:20:33 -0500 Subject: [PATCH 1/2] fix(heidenhain): Tool/Number, Tool/Index, Tool/AxisCode publish their declared integer types (#867) HeidenhainSession.Leaf's `case Lsv2Tool t` switch expression mixed long, int and double? arms, so C# unified the natural type to double? and Tool/Number (Int64), Tool/Index and Tool/AxisCode (Int32) were published as Double. Every arm in that switch (and in the alarm-list, axes and feature-acquire switches in the same file) is now boxed to object explicitly so no numeric arm can widen another. New HeidenhainLeafRuntimeTypeTests asserts value.GetType() for every catalog value leaf against its declared DataType (scalar + array), plus PLC/DATA binding coercion for Int16/Int32/Int64/Boolean. Verified red on the old code (exactly the three Tool leaves) and green with the fix. Claude-Session: https://claude.ai/code/session_01GT67AD99zQQqpLCVm3p7xk --- .../HeidenhainSession.cs | 32 +++-- .../HeidenhainLeafRuntimeTypeTests.cs | 133 ++++++++++++++++++ 2 files changed, 151 insertions(+), 14 deletions(-) create mode 100644 tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Heidenhain.Tests/HeidenhainLeafRuntimeTypeTests.cs diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Heidenhain/HeidenhainSession.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Heidenhain/HeidenhainSession.cs index 0c2b831ed..e354caa35 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Heidenhain/HeidenhainSession.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Heidenhain/HeidenhainSession.cs @@ -256,14 +256,15 @@ internal sealed class HeidenhainSession : IAsyncDisposable await PrepareTelemetryAsync(wire, ct).ConfigureAwait(false); value = feature switch { - "Execution" => Lsv2Data.U16(await _client.ReadRuntimeAsync(23, ct).ConfigureAwait(false)), - "ProgramState" => Lsv2Data.U16(await _client.ReadRuntimeAsync(26, ct).ConfigureAwait(false)), + // Each arm boxed explicitly so no numeric arm can widen another (lmxopcua#867). + "Execution" => (object)Lsv2Data.U16(await _client.ReadRuntimeAsync(23, ct).ConfigureAwait(false)), + "ProgramState" => (object)Lsv2Data.U16(await _client.ReadRuntimeAsync(26, ct).ConfigureAwait(false)), "Program" => Lsv2Data.Program(await _client.ReadRuntimeAsync(24, ct).ConfigureAwait(false)), "Overrides" => Lsv2Data.Overrides(await _client.ReadRuntimeAsync(25, ct).ConfigureAwait(false)), "Axes" => Lsv2Data.Axes(await _client.ReadRuntimeAsync(22, ct).ConfigureAwait(false)), "Tool" => Lsv2Data.Tool(await _client.ReadRuntimeAsync(51, ct).ConfigureAwait(false)), HeidenhainCatalog.AlarmsFeature => await AlarmsAsync(ct).ConfigureAwait(false), - _ => await TimerAsync(feature, ct).ConfigureAwait(false), + _ => (object)await TimerAsync(feature, ct).ConfigureAwait(false), }; } @@ -444,24 +445,27 @@ internal sealed class HeidenhainSession : IAsyncDisposable case double[] o when address.StartsWith("Overrides/", StringComparison.Ordinal): return o[name switch { "Feed" => 0, "Spindle" => 1, _ => 2 }]; case Lsv2Axes a: - return name == "Names" ? a.Names : a.Positions; + return name == "Names" ? (object)a.Names : (object)a.Positions; case Lsv2Tool t: + // lmxopcua#867: every arm is cast to object. Without it C# unifies long/int/double? to double?, so + // Tool/Number (Int64) and Tool/Index + Tool/AxisCode (Int32) were published as Double. A null + // geometry still boxes to null (BadNotSupported upstream). return name switch { - "Number" => t.Number, - "Index" => t.Index, - "AxisCode" => t.Axis, - "Length" => t.Length, - _ => t.Radius, + "Number" => (object)t.Number, + "Index" => (object)t.Index, + "AxisCode" => (object)t.Axis, + "Length" => (object?)t.Length, + _ => (object?)t.Radius, }; case List list: return name switch { - "Count" => list.Count, - "Messages" => list.Select(x => x.Message).ToArray(), - "Classes" => list.Select(x => x.ErrorClass).ToArray(), - "Groups" => list.Select(x => x.Group).ToArray(), - _ => list.Select(x => x.Number).ToArray(), + "Count" => (object)list.Count, + "Messages" => (object)list.Select(x => x.Message).ToArray(), + "Classes" => (object)list.Select(x => x.ErrorClass).ToArray(), + "Groups" => (object)list.Select(x => x.Group).ToArray(), + _ => (object)list.Select(x => x.Number).ToArray(), }; case int i: return i; diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Heidenhain.Tests/HeidenhainLeafRuntimeTypeTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Heidenhain.Tests/HeidenhainLeafRuntimeTypeTests.cs new file mode 100644 index 000000000..7c10bd2cb --- /dev/null +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Heidenhain.Tests/HeidenhainLeafRuntimeTypeTests.cs @@ -0,0 +1,133 @@ +using System.Buffers.Binary; +using System.Text; +using Shouldly; +using Xunit; +using ZB.MOM.WW.OtOpcUa.Core.Abstractions; +using ZB.MOM.WW.OtOpcUa.Driver.Heidenhain.Tests.Simulator; + +namespace ZB.MOM.WW.OtOpcUa.Driver.Heidenhain.Tests; + +/// +/// lmxopcua#867: the published value's RUNTIME type must be the CLR type of the leaf's declared data type. +/// Shouldly equality cannot catch this (7.0 equals 7L), so every assertion here compares GetType(). +/// +[Trait("Category", "Unit")] +public sealed class HeidenhainLeafRuntimeTypeTests +{ + private static CancellationToken Ct => TestContext.Current.CancellationToken; + + private static Type ClrType(DriverDataType type, bool isArray) + { + var scalar = type switch + { + DriverDataType.Boolean => typeof(bool), + DriverDataType.Int16 => typeof(short), + DriverDataType.Int32 => typeof(int), + DriverDataType.Int64 => typeof(long), + DriverDataType.Float64 => typeof(double), + DriverDataType.String => typeof(string), + _ => throw new ArgumentOutOfRangeException(nameof(type), type, null), + }; + return isArray ? scalar.MakeArrayType() : scalar; + } + + /// Every fixed value leaf (capability strings excluded), with its declared type. + public static TheoryData CatalogValueLeaves() + { + var data = new TheoryData(); + foreach (var e in HeidenhainCatalog.Entries.Where(e => !e.IsCapability)) data.Add(e.Address); + return data; + } + + private static SimulatedControl FullyPopulatedControl() + { + var c = new SimulatedControl(); + c.Runtime[22] = () => SimReply.Ok("S_RI", [0, 1, .. Encoding.ASCII.GetBytes("1.5\0X\0")]); + c.Runtime[23] = () => SimReply.Ok("S_RI", [0, 4]); + c.Runtime[26] = () => SimReply.Ok("S_RI", [0, 7]); + c.Runtime[24] = () => SimReply.Ok("S_RI", [0, 0, 0, 5, .. "A\0B\0"u8.ToArray()]); + c.Runtime[25] = () => SimReply.Ok("S_RI", new byte[12]); + var alarm = new byte[10]; + BinaryPrimitives.WriteInt16BigEndian(alarm, 3); + BinaryPrimitives.WriteInt16BigEndian(alarm.AsSpan(2), 4); + BinaryPrimitives.WriteInt32BigEndian(alarm.AsSpan(4), 12); + alarm[8] = 65; + c.Runtime[27] = () => SimReply.Ok("S_RI", alarm); + c.Runtime[28] = () => SimReply.Refuse(57); + // 24-byte tool reply: number 49, index 1, axis 2, length 100.5, radius 3.25 (float64le). + var tool = new byte[24]; + BinaryPrimitives.WriteUInt32BigEndian(tool, 49); + BinaryPrimitives.WriteUInt16BigEndian(tool.AsSpan(4), 1); + BinaryPrimitives.WriteUInt16BigEndian(tool.AsSpan(6), 2); + BinaryPrimitives.WriteDoubleLittleEndian(tool.AsSpan(8), 100.5); + BinaryPrimitives.WriteDoubleLittleEndian(tool.AsSpan(16), 3.25); + c.Runtime[51] = () => SimReply.Ok("S_RI", tool); + for (var s = 30; s <= 34; s++) c.Runtime[s] = () => SimReply.Ok("S_RI", [0, 0, 1, 0]); + return c; + } + + [Theory] + [MemberData(nameof(CatalogValueLeaves))] + public async Task Catalog_leaf_publishes_the_clr_type_of_its_declared_data_type(string address) + { + HeidenhainCatalog.TryGet(address, out var entry).ShouldBeTrue(); + var control = FullyPopulatedControl(); + await using var sim = Lsv2Simulator.WithControl(control); + var client = new Lsv2.Lsv2Client("127.0.0.1", sim.Port, TimeSpan.FromSeconds(1)); + var tag = TestTags.Def(address); + var settings = new HeidenhainSessionSettings(true, true, false, 64, true, false, false); + await using var session = new HeidenhainSession(client, settings, (_, _) => ValueTask.FromResult("")); + await client.OpenAsync(Ct); + await session.InitializeAsync(Ct); + + var sample = (await session.ReadBatchAsync([tag], Ct))[0]; + + sample.Status.ShouldBe(HeidenhainStatus.Good, address); + sample.Value.ShouldNotBeNull(address); + sample.Value.GetType().ShouldBe(ClrType(entry!.DataType, entry.IsArray), address); + control.Violations.ShouldBeEmpty(); + } + + [Fact] + public void Every_integer_catalog_leaf_is_covered() + { + // Pins the theory's reach: the integer leaves #867 broke are in the data set. + var leaves = CatalogValueLeaves().Select(r => r.Data).ToList(); + leaves.ShouldContain("Tool/Number"); + leaves.ShouldContain("Tool/Index"); + leaves.ShouldContain("Tool/AxisCode"); + leaves.ShouldContain("Alarms/Count"); + leaves.ShouldContain("Program/Line"); + leaves.ShouldContain("Status/ExecutionState"); + leaves.Count(l => l.StartsWith("Timers/", StringComparison.Ordinal)).ShouldBe(HeidenhainCatalog.TimerNames.Count); + } + + [Theory] + [InlineData(DriverDataType.Int16, 42)] + [InlineData(DriverDataType.Int32, 42)] + [InlineData(DriverDataType.Int64, 42)] + [InlineData(DriverDataType.Int16, 42L)] + [InlineData(DriverDataType.Int32, 42L)] + [InlineData(DriverDataType.Int64, 42L)] + [InlineData(DriverDataType.Int16, 42.0)] + [InlineData(DriverDataType.Int32, 42.0)] + [InlineData(DriverDataType.Int64, 42.0)] + [InlineData(DriverDataType.Boolean, true)] + [InlineData(DriverDataType.Boolean, false)] + public void Binding_coercion_publishes_the_clr_type_of_the_declared_data_type(DriverDataType type, object native) + { + HeidenhainBindingCodec.Coerce(native, type).GetType().ShouldBe(ClrType(type, false)); + } + + [Theory] + [InlineData(HeidenhainPlcArea.Byte, new byte[] { 7 }, DriverDataType.Int16)] + [InlineData(HeidenhainPlcArea.Byte, new byte[] { 7 }, DriverDataType.Int32)] + [InlineData(HeidenhainPlcArea.Word, new byte[] { 7, 0 }, DriverDataType.Int64)] + [InlineData(HeidenhainPlcArea.Dword, new byte[] { 7, 0, 0, 0 }, DriverDataType.Int32)] + [InlineData(HeidenhainPlcArea.InputWord, new byte[] { 7, 0 }, DriverDataType.Int32)] + [InlineData(HeidenhainPlcArea.Marker, new byte[] { 1 }, DriverDataType.Boolean)] + public void Plc_binding_decode_publishes_the_clr_type_of_the_declared_data_type(HeidenhainPlcArea area, byte[] bytes, DriverDataType type) + { + HeidenhainBindingCodec.DecodePlc(area, bytes, type).GetType().ShouldBe(ClrType(type, false)); + } +} From 02503a4a9f6092c6e4b08fd83f6656806b857e0b Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Wed, 11 Nov 2026 12:28:39 -0500 Subject: [PATCH 2/2] fix(controlplane): ConfigPublishCoordinator survives an unreachable ConfigDb at singleton start (#868) PreStart ran the in-flight recovery query synchronously. With the ConfigDb unreachable when the singleton started, the throw became an ActorInitializationException, the default strategy stopped the actor, the ClusterSingletonManager never re-created it, and every later deploy hung. - PreStart now only subscribes and kicks off an async recovery load piped back to the actor. A failure retries with exponential backoff (1 s doubling to 30 s, forever) and never stops the actor. - Until the load succeeds the coordinator is in a Loading behaviour that stashes every message (DispatchDeployment logged as queued, ApplyAck too); they are replayed in order after the recovered state is applied. Recovery decisions (re-dispatch / time out) now run inline before the replay instead of via self-tells, and a duplicate dispatch of the in-flight deployment with a seeded expected set is ignored so queued acks are never reset. - New ConfigPublishCoordinatorReadiness (admin-node DI singleton) + "config-publish" ready-tier health check: Degraded while loading (with the attempt count and last error), Healthy when ready or not hosted here. Tests: ConfigPublishCoordinatorDbOutageTests (DB down at start -> actor alive, Loading reported, deploy queued, DB returns -> load succeeds, queued deploy dispatched and sealed; queued ack applied to the recovered in-flight set; NotHosted after stop) and ConfigPublishCoordinatorHealthCheckTests. Claude-Session: https://claude.ai/code/session_01GT67AD99zQQqpLCVm3p7xk --- .../Coordinators/ConfigPublishCoordinator.cs | 225 ++++++++++++++++-- .../ConfigPublishCoordinatorReadiness.cs | 50 ++++ .../ServiceCollectionExtensions.cs | 6 +- .../ConfigPublishCoordinatorHealthCheck.cs | 45 ++++ .../Health/HealthEndpoints.cs | 9 + src/Server/ZB.MOM.WW.OtOpcUa.Host/Program.cs | 5 + .../ConfigPublishCoordinatorDbOutageTests.cs | 185 ++++++++++++++ .../ConfigPublishCoordinatorRecoveryTests.cs | 6 +- ...onfigPublishCoordinatorHealthCheckTests.cs | 65 +++++ 9 files changed, 569 insertions(+), 27 deletions(-) create mode 100644 src/Server/ZB.MOM.WW.OtOpcUa.ControlPlane/Coordinators/ConfigPublishCoordinatorReadiness.cs create mode 100644 src/Server/ZB.MOM.WW.OtOpcUa.Host/Health/ConfigPublishCoordinatorHealthCheck.cs create mode 100644 tests/Server/ZB.MOM.WW.OtOpcUa.ControlPlane.Tests/ConfigPublishCoordinatorDbOutageTests.cs create mode 100644 tests/Server/ZB.MOM.WW.OtOpcUa.Host.Tests/Health/ConfigPublishCoordinatorHealthCheckTests.cs diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.ControlPlane/Coordinators/ConfigPublishCoordinator.cs b/src/Server/ZB.MOM.WW.OtOpcUa.ControlPlane/Coordinators/ConfigPublishCoordinator.cs index 01143dd4d..c96b682a9 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.ControlPlane/Coordinators/ConfigPublishCoordinator.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.ControlPlane/Coordinators/ConfigPublishCoordinator.cs @@ -52,8 +52,20 @@ namespace ZB.MOM.WW.OtOpcUa.ControlPlane.Coordinators; /// ServerCluster.Enabled is not consulted — nothing else in the codebase consults it, /// so honouring it here would invent semantics. Disable the nodes, not the cluster row. /// +/// +/// Recovery load is asynchronous and retried (lmxopcua#868). The in-flight recovery read used to run +/// synchronously in PreStart. When the ConfigDb was unreachable as the singleton started, the throw +/// became an ActorInitializationException, the default strategy STOPPED the actor, and the +/// ClusterSingletonManager never re-creates a singleton that terminated on its own — so every later +/// deployment was told to a dead singleton and hung. The read now runs off the actor thread, a failure +/// retries with exponential backoff ( doubling to +/// , forever), and until it succeeds every message — deployments and acks +/// alike — is STASHED and replayed in order once the recovered state is in place, so a deploy arriving +/// during the outage is queued, never dropped. reports the +/// loading state (and the last error) to the node's health endpoint as Degraded. +/// /// -public sealed class ConfigPublishCoordinator : ReceiveActor, IWithTimers +public sealed class ConfigPublishCoordinator : ReceiveActor, IWithTimers, IWithUnboundedStash { public const string DeploymentsTopic = "deployments"; public const string DeploymentAcksTopic = "deployment-acks"; @@ -61,8 +73,18 @@ public sealed class ConfigPublishCoordinator : ReceiveActor, IWithTimers /// apply-outcome timeout is derived from. public static readonly TimeSpan DefaultApplyDeadline = DeployTimings.ApplyDeadline; + /// First retry delay after a failed recovery load (lmxopcua#868). + public static readonly TimeSpan DefaultLoadRetryMin = TimeSpan.FromSeconds(1); + + /// Ceiling of the recovery-load retry backoff (lmxopcua#868). + public static readonly TimeSpan DefaultLoadRetryMax = TimeSpan.FromSeconds(30); + private readonly IDbContextFactory _dbFactory; private readonly TimeSpan _applyDeadline; + private readonly TimeSpan _loadRetryMin; + private readonly TimeSpan _loadRetryMax; + private readonly ConfigPublishCoordinatorReadiness? _readiness; + private int _failedLoads; /// /// Where central→node commands are handed off (per-cluster mesh Phase 2), or @@ -115,20 +137,30 @@ public sealed class ConfigPublishCoordinator : ReceiveActor, IWithTimers /// Gets the timer scheduler for managing apply deadlines. public ITimerScheduler Timers { get; set; } = null!; + /// Holds every message that arrives before the recovery load completes (lmxopcua#868). + public IStash Stash { get; set; } = null!; + /// Creates actor props for the ConfigPublishCoordinator with the specified configuration. /// The database context factory for accessing configuration data. /// The timeout for waiting for apply acknowledgments from cluster nodes. - /// The Akka.NET used to spawn this actor. /// /// Central comm actor the dispatch is handed to, or to publish on the /// DistributedPubSub mediator directly (the original behaviour). /// + /// Where the recovery-load state is reported for health, or . + /// First recovery-load retry delay; = . + /// Recovery-load retry ceiling; = . + /// The Akka.NET used to spawn this actor. public static Props Props( IDbContextFactory dbFactory, TimeSpan? applyDeadline = null, - IActorRef? meshRouter = null) => - Akka.Actor.Props.Create(() => - new ConfigPublishCoordinator(dbFactory, applyDeadline ?? DefaultApplyDeadline, meshRouter)); + IActorRef? meshRouter = null, + ConfigPublishCoordinatorReadiness? readiness = null, + TimeSpan? loadRetryMin = null, + TimeSpan? loadRetryMax = null) => + Akka.Actor.Props.Create(() => new ConfigPublishCoordinator( + dbFactory, applyDeadline ?? DefaultApplyDeadline, meshRouter, readiness, + loadRetryMin ?? DefaultLoadRetryMin, loadRetryMax ?? DefaultLoadRetryMax)); /// /// Initializes a new instance of the class. @@ -136,45 +168,167 @@ public sealed class ConfigPublishCoordinator : ReceiveActor, IWithTimers /// The database context factory for persisting deployment state. /// The timeout for waiting for per-node apply acknowledgments. /// Central comm actor, or for the mediator. + /// Where the recovery-load state is reported, or . + /// First recovery-load retry delay. + /// Recovery-load retry ceiling. public ConfigPublishCoordinator( IDbContextFactory dbFactory, TimeSpan applyDeadline, - IActorRef? meshRouter = null) + IActorRef? meshRouter = null, + ConfigPublishCoordinatorReadiness? readiness = null, + TimeSpan? loadRetryMin = null, + TimeSpan? loadRetryMax = null) { _dbFactory = dbFactory; _applyDeadline = applyDeadline; _meshRouter = meshRouter; + _readiness = readiness; + _loadRetryMin = loadRetryMin ?? DefaultLoadRetryMin; + _loadRetryMax = loadRetryMax ?? DefaultLoadRetryMax; + Loading(); + } + + /// + /// The state before the recovery load has succeeded: only the load's own outcome and retry tick are + /// handled; everything else is stashed and replayed, in arrival order, once is in force. + /// + private void Loading() + { + Receive(HandleRecoveryLoaded); + Receive(HandleRecoveryLoadFailed); + Receive(_ => StartRecoveryLoad()); + Receive(_ => { /* DPS subscribe confirmation */ }); + Receive(msg => + { + _log.Info( + "DispatchDeployment {Id} queued: the coordinator has not yet loaded its recovery state from the " + + "ConfigDb ({Failed} failed attempt(s) so far); it is dispatched as soon as the load succeeds", + msg.DeploymentId, _failedLoads); + Stash.Stash(); + }); + ReceiveAny(_ => Stash.Stash()); + } + + /// The normal operating state, entered once the recovery load has been applied. + private void Ready() + { Receive(HandleDispatch); Receive(HandleAck); Receive(HandleDeadline); Receive(_ => { /* DPS subscribe confirmation */ }); + // A late load outcome / retry tick cannot normally arrive once Ready; ignored, never stashed. + Receive(_ => { }); + Receive(_ => { }); + Receive(_ => { }); } /// protected override void PreStart() { - // On startup recover any deployment that was mid-flight when a prior singleton instance - // died. We re-derive _expectedAcks from NodeDeploymentState, replay the ACKs - // that already landed in the DB, and resume the deadline timer. // Subscribe to per-node ApplyAck broadcasts so DriverHostActors on remote members can // route their ACKs to whichever node currently hosts this singleton. DistributedPubSub.Get(Context.System).Mediator.Tell(new Subscribe(DeploymentAcksTopic, Self)); - using var db = _dbFactory.CreateDbContext(); + // lmxopcua#868: NO database access here. A throw from PreStart stops the actor for good (the + // singleton manager does not re-create it), so the recovery read runs asynchronously and retries. + _readiness?.MarkLoading(_failedLoads, lastError: null); + StartRecoveryLoad(); + } + + /// + protected override void PostStop() + { + // A restart re-enters PreStart (Loading again); a stop or hand-over leaves this node not hosting it. + _readiness?.MarkNotHosted(); + base.PostStop(); + } + + /// + /// Reads the deployment that was mid-flight when a prior singleton instance died, off the actor thread, + /// and pipes the outcome back. Touches no actor state. + /// + private void StartRecoveryLoad() + { + LoadRecoveryAsync(_dbFactory).PipeTo( + Self, + success: loaded => loaded, + failure: ex => new RecoveryLoadFailed(ex is AggregateException { InnerException: { } inner } ? inner : ex)); + } + + private static async Task LoadRecoveryAsync(IDbContextFactory dbFactory) + { + // Yield first so a factory or provider that throws synchronously still faults the TASK (piped back as + // RecoveryLoadFailed) rather than the caller on the actor thread. + await Task.Yield(); + await using var db = await dbFactory.CreateDbContextAsync().ConfigureAwait(false); // Projected: the row carries the whole ArtifactBlob (varbinary(max)), which this recovery never // reads — and this query runs on every singleton start and hand-over. - var inflight = db.Deployments + var inflight = await db.Deployments .Where(d => d.Status == DeploymentStatus.Dispatching || d.Status == DeploymentStatus.AwaitingApplyAcks) .OrderByDescending(d => d.CreatedAtUtc) .Select(d => new { d.DeploymentId, d.RevisionHash, d.CreatedAtUtc }) - .FirstOrDefault(); + .FirstOrDefaultAsync() + .ConfigureAwait(false); + if (inflight is null) return new RecoveryLoaded(null); + + var nodeStates = await db.NodeDeploymentStates + .Where(x => x.DeploymentId == inflight.DeploymentId) + .AsNoTracking() + .Select(x => new RecoveredNodeState(x.NodeId, x.Status, x.FailureReason)) + .ToListAsync() + .ConfigureAwait(false); + return new RecoveryLoaded(new RecoveredDeployment( + inflight.DeploymentId, inflight.RevisionHash, inflight.CreatedAtUtc, nodeStates)); + } + + private void HandleRecoveryLoadFailed(RecoveryLoadFailed msg) + { + _failedLoads++; + // Exponential backoff, capped: min * 2^(n-1), with the shift bounded so it never overflows. + var factor = 1L << Math.Min(_failedLoads - 1, 20); + var delay = TimeSpan.FromTicks(Math.Min(_loadRetryMax.Ticks, _loadRetryMin.Ticks * factor)); + _readiness?.MarkLoading(_failedLoads, msg.Error.Message); + _log.Warning(msg.Error, + "ConfigPublishCoordinator could not load its recovery state from the ConfigDb (attempt {Attempt}); " + + "retrying in {Delay}. Deployments and acks are queued until it succeeds", + _failedLoads, delay); + Timers.StartSingleTimer(LoadRetryTimerKey, RetryRecoveryLoad.Instance, delay); + } + + /// + /// Applies the recovered state, switches to and replays every stashed message. The + /// recovery DECISIONS (re-dispatch, time out) run inline, before the replay, so a queued message for the + /// same deployment is judged against the recovered state rather than racing a self-sent one. + /// + private void HandleRecoveryLoaded(RecoveryLoaded msg) + { + Timers.Cancel(LoadRetryTimerKey); + if (_failedLoads > 0) + _log.Info("ConfigPublishCoordinator loaded its recovery state after {Failed} failed attempt(s)", _failedLoads); + _failedLoads = 0; + Become(Ready); + _readiness?.MarkReady(); + try + { + ApplyRecovery(msg.Inflight); + } + finally + { + Stash.UnstashAll(); + } + } + + /// + /// Recovers the deployment that was mid-flight when a prior singleton instance died: re-derives + /// _expectedAcks from NodeDeploymentState, replays the acks that already landed, and resumes + /// the deadline timer. + /// + private void ApplyRecovery(RecoveredDeployment? inflight) + { if (inflight is null) return; _current = new DeploymentId(inflight.DeploymentId); - var nodeStates = db.NodeDeploymentStates - .Where(x => x.DeploymentId == inflight.DeploymentId) - .AsNoTracking() - .ToList(); + var nodeStates = inflight.NodeStates; // Resume the deadline timer using the remaining time. The deadline runs from when the // deployment was first marked AwaitingApplyAcks (Deployment.CreatedAtUtc is a close enough @@ -201,12 +355,12 @@ public sealed class ConfigPublishCoordinator : ReceiveActor, IWithTimers "before any node was expected (coordinator restart or singleton hand-over mid-dispatch); " + "re-dispatching with {Remaining} of the apply deadline left", _current, remaining); - Self.Tell(new DispatchDeployment( + HandleDispatch(new DispatchDeployment( _current.Value, RevisionHash.Parse(inflight.RevisionHash), CorrelationId.NewId())); } else { - Self.Tell(new DeadlineElapsed(_current.Value)); + HandleDeadline(new DeadlineElapsed(_current.Value)); } return; } @@ -222,18 +376,25 @@ public sealed class ConfigPublishCoordinator : ReceiveActor, IWithTimers if (remaining <= TimeSpan.Zero) { - Self.Tell(new DeadlineElapsed(_current.Value)); - } - else - { - Timers.StartSingleTimer(DeadlineTimerKey, new DeadlineElapsed(_current.Value), remaining); + HandleDeadline(new DeadlineElapsed(_current.Value)); + return; } + + Timers.StartSingleTimer(DeadlineTimerKey, new DeadlineElapsed(_current.Value), remaining); _log.Info("Coordinator recovered in-flight deployment {Id} ({Acked}/{Total} acks landed)", _current, _acks.Count, _expectedAcks.Count); } private void HandleDispatch(DispatchDeployment msg) { + // A duplicate of the deployment already in flight with a seeded expected set (a dispatch queued while the + // recovery load ran, for the very row that load re-drove) must not reset the acks already gathered. + if (_current == msg.DeploymentId && _expectedAcks.Count > 0) + { + _log.Debug("Ignoring duplicate DispatchDeployment {Id}: already in flight", msg.DeploymentId); + return; + } + // Terminal-state guard (fix round 1): a late or re-delivered dispatch for a row that is already // Sealed / PartiallyFailed / TimedOut — e.g. one AdminOps closed as a lost dispatch while this message // was still queued — must never flip it back to AwaitingApplyAcks or seed an expected set for it. @@ -519,9 +680,27 @@ public sealed class ConfigPublishCoordinator : ReceiveActor, IWithTimers } private const string DeadlineTimerKey = "apply-deadline"; + private const string LoadRetryTimerKey = "recovery-load-retry"; public sealed record DeadlineElapsed(DeploymentId DeploymentId); + /// The recovery read's result: the newest in-flight deployment, or null when none. + private sealed record RecoveryLoaded(RecoveredDeployment? Inflight); + + /// The recovery read failed (ConfigDb unreachable, ...); retried with backoff. + private sealed record RecoveryLoadFailed(Exception Error); + + /// The backoff timer's tick: run the recovery read again. + private sealed class RetryRecoveryLoad + { + public static readonly RetryRecoveryLoad Instance = new(); + } + + private sealed record RecoveredDeployment( + Guid DeploymentId, string RevisionHash, DateTime CreatedAtUtc, IReadOnlyList NodeStates); + + private sealed record RecoveredNodeState(string NodeId, NodeDeploymentStatus Status, string? FailureReason); + /// /// Stages a status change WITHOUT materialising the row: the row carries the whole ArtifactBlob /// (varbinary(max), multi-MB on a real fleet), which no status transition reads (fix round 1). A stub diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.ControlPlane/Coordinators/ConfigPublishCoordinatorReadiness.cs b/src/Server/ZB.MOM.WW.OtOpcUa.ControlPlane/Coordinators/ConfigPublishCoordinatorReadiness.cs new file mode 100644 index 000000000..0074a8eee --- /dev/null +++ b/src/Server/ZB.MOM.WW.OtOpcUa.ControlPlane/Coordinators/ConfigPublishCoordinatorReadiness.cs @@ -0,0 +1,50 @@ +namespace ZB.MOM.WW.OtOpcUa.ControlPlane.Coordinators; + +/// Where the singleton stands on this node (lmxopcua#868). +public enum ConfigPublishCoordinatorState +{ + /// The singleton is not running on this node (the partner hosts it, or it has not started yet). + NotHosted, + + /// Running here, but its recovery state has not loaded from the ConfigDb yet; deployments are queued. + Loading, + + /// Running here with its recovery state loaded; deployments are dispatched. + Ready, +} + +/// One immutable observation of the coordinator's state. +/// The state. +/// Consecutive failed recovery loads (0 unless ). +/// The last load failure's message, or null. +/// When the state was entered. +public sealed record ConfigPublishCoordinatorSnapshot( + ConfigPublishCoordinatorState State, int FailedLoadAttempts, string? LastError, DateTime SinceUtc); + +/// +/// Thread-safe state bridge from the actor to the node's health endpoint +/// (one DI singleton per admin node). The actor writes; the health check reads. +/// +public sealed class ConfigPublishCoordinatorReadiness +{ + private volatile ConfigPublishCoordinatorSnapshot _current = + new(ConfigPublishCoordinatorState.NotHosted, 0, null, DateTime.UtcNow); + + /// The latest observation. + public ConfigPublishCoordinatorSnapshot Current => _current; + + /// The coordinator is loading (or retrying the load of) its recovery state. + /// Failed attempts so far. + /// The last failure's message, or null. + public void MarkLoading(int failedAttempts, string? lastError) + { + var since = _current.State == ConfigPublishCoordinatorState.Loading ? _current.SinceUtc : DateTime.UtcNow; + _current = new(ConfigPublishCoordinatorState.Loading, failedAttempts, lastError, since); + } + + /// The recovery state is loaded. + public void MarkReady() => _current = new(ConfigPublishCoordinatorState.Ready, 0, null, DateTime.UtcNow); + + /// The singleton stopped on this node. + public void MarkNotHosted() => _current = new(ConfigPublishCoordinatorState.NotHosted, 0, null, DateTime.UtcNow); +} diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.ControlPlane/ServiceCollectionExtensions.cs b/src/Server/ZB.MOM.WW.OtOpcUa.ControlPlane/ServiceCollectionExtensions.cs index 9a9f07423..048613739 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.ControlPlane/ServiceCollectionExtensions.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.ControlPlane/ServiceCollectionExtensions.cs @@ -112,7 +112,11 @@ public static class ServiceCollectionExtensions // coordinator. Deliberately NOT an ActorSelection.ResolveOne here: that blocks // inside a props factory and races the spawn it is waiting for. var meshRouter = registry.Get(); - return ConfigPublishCoordinator.Props(dbFactory, applyDeadline: DeployTimings.ApplyDeadline, meshRouter: meshRouter); + // lmxopcua#868: optional — through IServiceProvider for GetService's null-for-absent contract + // (the Akka resolver makes no such promise), so harnesses that never register it still boot. + var readiness = resolver.GetService()?.GetService(); + return ConfigPublishCoordinator.Props( + dbFactory, applyDeadline: DeployTimings.ApplyDeadline, meshRouter: meshRouter, readiness: readiness); }, singletonOptions); diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.Host/Health/ConfigPublishCoordinatorHealthCheck.cs b/src/Server/ZB.MOM.WW.OtOpcUa.Host/Health/ConfigPublishCoordinatorHealthCheck.cs new file mode 100644 index 000000000..579eb660d --- /dev/null +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Host/Health/ConfigPublishCoordinatorHealthCheck.cs @@ -0,0 +1,45 @@ +using Microsoft.Extensions.Diagnostics.HealthChecks; +using ZB.MOM.WW.OtOpcUa.ControlPlane.Coordinators; + +namespace ZB.MOM.WW.OtOpcUa.Host.Health; + +/// +/// The deploy coordinator's readiness (lmxopcua#868): Degraded while the +/// singleton on this node has not yet loaded its recovery state from the +/// ConfigDb (deployments are queued, not dispatched), Healthy otherwise — including on the admin node that does +/// not host the singleton. Never Unhealthy: the node still serves (the configdb probe owns "DB down"). +/// +/// The coordinator's state bridge; null on graphs that never register one. +public sealed class ConfigPublishCoordinatorHealthCheck(ConfigPublishCoordinatorReadiness? readiness) : IHealthCheck +{ + /// The check's registration name. + public const string Name = "config-publish"; + + /// + public Task CheckHealthAsync(HealthCheckContext context, CancellationToken cancellationToken = default) + { + if (readiness is null) + return Task.FromResult(HealthCheckResult.Healthy("Deploy coordinator readiness is not tracked on this node.")); + + var s = readiness.Current; + var data = new Dictionary(StringComparer.Ordinal) + { + ["state"] = s.State.ToString(), + ["sinceUtc"] = s.SinceUtc.ToString("O"), + ["failedLoadAttempts"] = s.FailedLoadAttempts, + }; + if (s.LastError is not null) data["lastError"] = s.LastError; + + return Task.FromResult(s.State switch + { + ConfigPublishCoordinatorState.Loading => HealthCheckResult.Degraded( + s.FailedLoadAttempts == 0 + ? "Deploy coordinator is loading its recovery state from the ConfigDb; deployments are queued until it completes." + : $"Deploy coordinator cannot load its recovery state from the ConfigDb ({s.FailedLoadAttempts} failed attempt(s), " + + $"last: {s.LastError}); retrying with backoff. Deployments are queued until it succeeds.", + data: data), + ConfigPublishCoordinatorState.Ready => HealthCheckResult.Healthy("Deploy coordinator ready.", data), + _ => HealthCheckResult.Healthy("Deploy coordinator singleton is not hosted on this node.", data), + }); + } +} diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.Host/Health/HealthEndpoints.cs b/src/Server/ZB.MOM.WW.OtOpcUa.Host/Health/HealthEndpoints.cs index e7fe87e91..24fec2e01 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.Host/Health/HealthEndpoints.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Host/Health/HealthEndpoints.cs @@ -87,6 +87,15 @@ public static class HealthEndpoints { ProbeQuery = static (db, ct) => db.Deployments.AsNoTracking().Take(1).ToListAsync(ct), }); + + // lmxopcua#868: Degraded (still 200) while the deploy coordinator singleton on this node is retrying + // its ConfigDb recovery load and queueing deployments; Healthy otherwise. Ready tier only. + checks.Add(new HealthCheckRegistration( + ConfigPublishCoordinatorHealthCheck.Name, + sp => new ConfigPublishCoordinatorHealthCheck( + sp.GetService()), + failureStatus: null, + tags: new[] { ZbHealthTags.Ready })); } checks diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.Host/Program.cs b/src/Server/ZB.MOM.WW.OtOpcUa.Host/Program.cs index 132ead6f6..0d5d88ab6 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.Host/Program.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Host/Program.cs @@ -176,7 +176,12 @@ else builder.Services.Configure(builder.Configuration.GetSection(AclOptions.SectionName)); if (hasAdmin) +{ builder.Services.AddOtOpcUaConfigDb(builder.Configuration); + // lmxopcua#868: the deploy coordinator singleton reports its ConfigDb recovery-load state here; the + // config-publish health check (AddOtOpcUaHealth) reads it — Degraded while deployments are queued. + builder.Services.AddSingleton(); +} else Log.Information("Driver-only node — no ConfigDb registered; configuration via ConfigSource:Mode=FetchAndCache"); diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.ControlPlane.Tests/ConfigPublishCoordinatorDbOutageTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.ControlPlane.Tests/ConfigPublishCoordinatorDbOutageTests.cs new file mode 100644 index 000000000..0e702382a --- /dev/null +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.ControlPlane.Tests/ConfigPublishCoordinatorDbOutageTests.cs @@ -0,0 +1,185 @@ +using Akka.Actor; +using Microsoft.EntityFrameworkCore; +using Shouldly; +using Xunit; +using ZB.MOM.WW.OtOpcUa.Commons.Messages.Deploy; +using ZB.MOM.WW.OtOpcUa.Commons.Types; +using ZB.MOM.WW.OtOpcUa.Configuration; +using ZB.MOM.WW.OtOpcUa.Configuration.Entities; +using ZB.MOM.WW.OtOpcUa.Configuration.Enums; +using ZB.MOM.WW.OtOpcUa.ControlPlane.Coordinators; +using ZB.MOM.WW.OtOpcUa.ControlPlane.Tests.Harness; + +namespace ZB.MOM.WW.OtOpcUa.ControlPlane.Tests; + +/// +/// lmxopcua#868: the coordinator's recovery read used to run synchronously in PreStart. With the +/// ConfigDb unreachable when the singleton started, the actor died on initialization, was never re-created, +/// and every deployment hung. It now loads asynchronously with backoff, queues what arrives meanwhile and +/// reports until the load succeeds. +/// +public sealed class ConfigPublishCoordinatorDbOutageTests : ControlPlaneActorTestBase +{ + private static readonly RevisionHash TestRevision = RevisionHash.Parse(new string('d', 64)); + private static readonly TimeSpan NoTimeout = TimeSpan.FromMinutes(5); + private static readonly TimeSpan RetryMin = TimeSpan.FromMilliseconds(50); + private static readonly TimeSpan RetryMax = TimeSpan.FromMilliseconds(200); + private const string Node = "site-a-1:4053"; + + [Fact] + public void Db_unreachable_at_start_does_not_kill_the_coordinator_and_a_deploy_sent_during_the_outage_completes_once_it_returns() + { + var dbName = Guid.NewGuid().ToString("N"); + var direct = NewInMemoryDbFactory(dbName); + var outage = new OutageDbFactory(NewInMemoryDbFactory(dbName)) { Down = true }; + SeedNodes(direct, Node); + var deploymentId = SeedDeployment(direct, DeploymentStatus.Dispatching, DateTime.UtcNow); + var readiness = new ConfigPublishCoordinatorReadiness(); + + var coordinator = Sys.ActorOf(ConfigPublishCoordinator.Props( + outage, NoTimeout, readiness: readiness, loadRetryMin: RetryMin, loadRetryMax: RetryMax)); + Watch(coordinator); + + // Several failed loads: the actor keeps retrying (it is alive) and reports Loading with the error. + AwaitAssert(() => + { + outage.Attempts.ShouldBeGreaterThanOrEqualTo(3); + readiness.Current.State.ShouldBe(ConfigPublishCoordinatorState.Loading); + readiness.Current.FailedLoadAttempts.ShouldBeGreaterThanOrEqualTo(2); + readiness.Current.LastError.ShouldNotBeNull(); + readiness.Current.LastError.ShouldContain("unreachable"); + }, duration: PresenceBudget, cancellationToken: TestContext.Current.CancellationToken); + + // A deployment dispatched during the outage is queued, not processed and not dropped. + coordinator.Tell(new DispatchDeployment(deploymentId, TestRevision, CorrelationId.NewId())); + ExpectNoMsg(TimeSpan.FromMilliseconds(300), TestContext.Current.CancellationToken); + using (var db = direct.CreateDbContext()) + { + db.Deployments.Single().Status.ShouldBe(DeploymentStatus.Dispatching); + db.NodeDeploymentStates.ShouldBeEmpty(); + } + + // The DB comes back: the load succeeds, the queued dispatch runs, and the node's ack seals it. + outage.Down = false; + AwaitAssert(() => + { + readiness.Current.State.ShouldBe(ConfigPublishCoordinatorState.Ready); + using var db = direct.CreateDbContext(); + db.Deployments.Single().Status.ShouldBe(DeploymentStatus.AwaitingApplyAcks); + db.NodeDeploymentStates.Select(s => s.NodeId).ShouldBe(new[] { Node }); + }, duration: PresenceBudget, cancellationToken: TestContext.Current.CancellationToken); + + coordinator.Tell(new ApplyAck(deploymentId, NodeId.Parse(Node), ApplyAckOutcome.Applied, null, CorrelationId.NewId())); + AwaitAssert(() => + { + using var db = direct.CreateDbContext(); + db.Deployments.Single().Status.ShouldBe(DeploymentStatus.Sealed); + }, duration: PresenceBudget, cancellationToken: TestContext.Current.CancellationToken); + + ExpectNoMsg(TimeSpan.FromMilliseconds(100), TestContext.Current.CancellationToken); // never Terminated + } + + [Fact] + public void In_flight_deployment_is_recovered_after_the_outage_and_an_ack_queued_during_it_is_not_lost() + { + var dbName = Guid.NewGuid().ToString("N"); + var direct = NewInMemoryDbFactory(dbName); + var outage = new OutageDbFactory(NewInMemoryDbFactory(dbName)) { Down = true }; + SeedNodes(direct, Node, "site-a-2:4053"); + var deploymentId = SeedDeployment(direct, DeploymentStatus.AwaitingApplyAcks, DateTime.UtcNow); + using (var db = direct.CreateDbContext()) + { + db.NodeDeploymentStates.Add(new NodeDeploymentState + { NodeId = Node, DeploymentId = deploymentId.Value, Status = NodeDeploymentStatus.Applying }); + db.NodeDeploymentStates.Add(new NodeDeploymentState + { NodeId = "site-a-2:4053", DeploymentId = deploymentId.Value, Status = NodeDeploymentStatus.Applied, AppliedAtUtc = DateTime.UtcNow }); + db.SaveChanges(); + } + + var readiness = new ConfigPublishCoordinatorReadiness(); + var coordinator = Sys.ActorOf(ConfigPublishCoordinator.Props( + outage, NoTimeout, readiness: readiness, loadRetryMin: RetryMin, loadRetryMax: RetryMax)); + AwaitAssert(() => outage.Attempts.ShouldBeGreaterThanOrEqualTo(2), + duration: PresenceBudget, cancellationToken: TestContext.Current.CancellationToken); + + // The last outstanding ack arrives while the coordinator is still loading: it is queued, then applied + // against the recovered expected set (site-a-2 already Applied in the DB) and seals the deployment. + coordinator.Tell(new ApplyAck(deploymentId, NodeId.Parse(Node), ApplyAckOutcome.Applied, null, CorrelationId.NewId())); + outage.Down = false; + + AwaitAssert(() => + { + readiness.Current.State.ShouldBe(ConfigPublishCoordinatorState.Ready); + using var db = direct.CreateDbContext(); + db.Deployments.Single().Status.ShouldBe(DeploymentStatus.Sealed); + }, duration: PresenceBudget, cancellationToken: TestContext.Current.CancellationToken); + } + + [Fact] + public void Readiness_is_NotHosted_after_the_coordinator_stops() + { + var readiness = new ConfigPublishCoordinatorReadiness(); + var coordinator = Sys.ActorOf(ConfigPublishCoordinator.Props(NewInMemoryDbFactory(), NoTimeout, readiness: readiness)); + AwaitAssert(() => readiness.Current.State.ShouldBe(ConfigPublishCoordinatorState.Ready), + duration: PresenceBudget, cancellationToken: TestContext.Current.CancellationToken); + + Watch(coordinator); + Sys.Stop(coordinator); + ExpectTerminated(coordinator, cancellationToken: TestContext.Current.CancellationToken); + readiness.Current.State.ShouldBe(ConfigPublishCoordinatorState.NotHosted); + } + + private static DeploymentId SeedDeployment( + IDbContextFactory dbFactory, DeploymentStatus status, DateTime createdAtUtc) + { + var id = DeploymentId.NewId(); + using var db = dbFactory.CreateDbContext(); + db.Deployments.Add(new Deployment + { + DeploymentId = id.Value, + RevisionHash = TestRevision.Value, + Status = status, + CreatedBy = "test", + CreatedAtUtc = createdAtUtc, + }); + db.SaveChanges(); + return id; + } + + private static void SeedNodes(IDbContextFactory dbFactory, params string[] nodeIds) + { + using var db = dbFactory.CreateDbContext(); + db.ServerClusters.Add(new ServerCluster + { + ClusterId = "SITE-A", Name = "Site A", Enterprise = "zb", Site = "site-a", + NodeCount = (byte)nodeIds.Length, RedundancyMode = RedundancyMode.None, CreatedBy = "test", + }); + foreach (var nodeId in nodeIds) + { + db.ClusterNodes.Add(new ClusterNode + { + NodeId = nodeId, ClusterId = "SITE-A", Host = nodeId.Split(':')[0], + ApplicationUri = $"urn:OtOpcUa:{nodeId}", Enabled = true, CreatedBy = "test", + }); + } + db.SaveChanges(); + } + + /// A factory that throws while — a ConfigDb that is unreachable. + private sealed class OutageDbFactory(IDbContextFactory inner) : IDbContextFactory + { + private volatile bool _down; + private int _attempts; + + public bool Down { get => _down; set => _down = value; } + + public int Attempts => Volatile.Read(ref _attempts); + + public OtOpcUaConfigDbContext CreateDbContext() + { + Interlocked.Increment(ref _attempts); + if (_down) throw new InvalidOperationException("ConfigDb unreachable (simulated outage)"); + return inner.CreateDbContext(); + } + } +} diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.ControlPlane.Tests/ConfigPublishCoordinatorRecoveryTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.ControlPlane.Tests/ConfigPublishCoordinatorRecoveryTests.cs index bd73313ad..91a947c8d 100644 --- a/tests/Server/ZB.MOM.WW.OtOpcUa.ControlPlane.Tests/ConfigPublishCoordinatorRecoveryTests.cs +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.ControlPlane.Tests/ConfigPublishCoordinatorRecoveryTests.cs @@ -136,9 +136,9 @@ public sealed class ConfigPublishCoordinatorRecoveryTests : ControlPlaneActorTes var coordinator = Sys.ActorOf(ConfigPublishCoordinator.Props(faulting, NoTimeout)); var admin = Sys.ActorOf(AdminOperationsActor.Props(adminFactory, coordinator, Enumerable.Empty())); - // Arm the fault AFTER PreStart has run (it would otherwise fault the recovery read itself and - // the actor would be stopped, not restarted). Identify is processed after creation, so a - // resolved selection proves PreStart completed. + // Arm the fault AFTER PreStart has run. Since lmxopcua#868 the recovery read is asynchronous and + // retried, so a fault that lands on it instead of on HandleDispatch is also recovered (the dispatch is + // queued until the retried load succeeds) — either way the deployment must complete. await Sys.ActorSelection(coordinator.Path).ResolveOne(PresenceBudget, TestContext.Current.CancellationToken); faulting.FailNext(); admin.Tell(new StartDeployment("joe", CorrelationId.NewId())); diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.Host.Tests/Health/ConfigPublishCoordinatorHealthCheckTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.Host.Tests/Health/ConfigPublishCoordinatorHealthCheckTests.cs new file mode 100644 index 000000000..a9202432c --- /dev/null +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.Host.Tests/Health/ConfigPublishCoordinatorHealthCheckTests.cs @@ -0,0 +1,65 @@ +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Diagnostics.HealthChecks; +using Microsoft.Extensions.Options; +using Shouldly; +using Xunit; +using ZB.MOM.WW.Health; +using ZB.MOM.WW.OtOpcUa.ControlPlane.Coordinators; +using ZB.MOM.WW.OtOpcUa.Host.Health; + +namespace ZB.MOM.WW.OtOpcUa.Host.Tests.Health; + +/// lmxopcua#868: the deploy coordinator reports Degraded (never Unhealthy) while it cannot load from the ConfigDb. +public sealed class ConfigPublishCoordinatorHealthCheckTests +{ + private static async Task Check(ConfigPublishCoordinatorReadiness? readiness) => + await new ConfigPublishCoordinatorHealthCheck(readiness).CheckHealthAsync(new HealthCheckContext(), CancellationToken.None); + + [Fact] + public async Task Loading_after_failed_attempts_is_degraded_and_names_the_error() + { + var r = new ConfigPublishCoordinatorReadiness(); + r.MarkLoading(3, "connection refused"); + var result = await Check(r); + result.Status.ShouldBe(HealthStatus.Degraded); + result.Description.ShouldNotBeNull(); + result.Description.ShouldContain("connection refused"); + result.Description.ShouldContain("queued"); + } + + [Fact] + public async Task Ready_is_healthy() + { + var r = new ConfigPublishCoordinatorReadiness(); + r.MarkLoading(2, "x"); + r.MarkReady(); + (await Check(r)).Status.ShouldBe(HealthStatus.Healthy); + } + + [Fact] + public async Task Not_hosted_and_untracked_are_healthy() + { + (await Check(new ConfigPublishCoordinatorReadiness())).Status.ShouldBe(HealthStatus.Healthy); + (await Check(null)).Status.ShouldBe(HealthStatus.Healthy); + } + + [Theory] + [InlineData(true, true)] + [InlineData(false, false)] + public void It_is_registered_on_the_ready_tier_only_on_admin_nodes(bool hasAdmin, bool registered) + { + var registrations = new ServiceCollection() + .AddOtOpcUaHealth(hasAdmin: hasAdmin, hasDriver: true) + .BuildServiceProvider() + .GetRequiredService>() + .Value.Registrations + .Where(x => x.Name == ConfigPublishCoordinatorHealthCheck.Name) + .ToList(); + if (!registered) + { + registrations.ShouldBeEmpty(); + return; + } + registrations.ShouldHaveSingleItem().Tags.ShouldBe([ZbHealthTags.Ready]); + } +}