Merge #867 Heidenhain integer leaves keep their declared type; #868 ConfigPublishCoordinator survives ConfigDb outage at start

Claude-Session: https://claude.ai/code/session_01GT67AD99zQQqpLCVm3p7xk
This commit is contained in:
Joseph Doherty
2026-11-11 12:29:00 -05:00
11 changed files with 720 additions and 41 deletions
@@ -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<Lsv2Alarm> 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;
@@ -52,8 +52,20 @@ namespace ZB.MOM.WW.OtOpcUa.ControlPlane.Coordinators;
/// <c>ServerCluster.Enabled</c> is <b>not</b> consulted — nothing else in the codebase consults it,
/// so honouring it here would invent semantics. Disable the nodes, not the cluster row.
/// </para>
/// <para>
/// <b>Recovery load is asynchronous and retried (lmxopcua#868).</b> The in-flight recovery read used to run
/// synchronously in <c>PreStart</c>. When the ConfigDb was unreachable as the singleton started, the throw
/// became an <c>ActorInitializationException</c>, the default strategy STOPPED the actor, and the
/// <c>ClusterSingletonManager</c> 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 (<see cref="DefaultLoadRetryMin"/> doubling to
/// <see cref="DefaultLoadRetryMax"/>, 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. <see cref="ConfigPublishCoordinatorReadiness"/> reports the
/// loading state (and the last error) to the node's health endpoint as Degraded.
/// </para>
/// </summary>
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.</summary>
public static readonly TimeSpan DefaultApplyDeadline = DeployTimings.ApplyDeadline;
/// <summary>First retry delay after a failed recovery load (lmxopcua#868).</summary>
public static readonly TimeSpan DefaultLoadRetryMin = TimeSpan.FromSeconds(1);
/// <summary>Ceiling of the recovery-load retry backoff (lmxopcua#868).</summary>
public static readonly TimeSpan DefaultLoadRetryMax = TimeSpan.FromSeconds(30);
private readonly IDbContextFactory<OtOpcUaConfigDbContext> _dbFactory;
private readonly TimeSpan _applyDeadline;
private readonly TimeSpan _loadRetryMin;
private readonly TimeSpan _loadRetryMax;
private readonly ConfigPublishCoordinatorReadiness? _readiness;
private int _failedLoads;
/// <summary>
/// Where central→node commands are handed off (per-cluster mesh Phase 2), or
@@ -115,20 +137,30 @@ public sealed class ConfigPublishCoordinator : ReceiveActor, IWithTimers
/// <summary>Gets the timer scheduler for managing apply deadlines.</summary>
public ITimerScheduler Timers { get; set; } = null!;
/// <summary>Holds every message that arrives before the recovery load completes (lmxopcua#868).</summary>
public IStash Stash { get; set; } = null!;
/// <summary>Creates actor props for the ConfigPublishCoordinator with the specified configuration.</summary>
/// <param name="dbFactory">The database context factory for accessing configuration data.</param>
/// <param name="applyDeadline">The timeout for waiting for apply acknowledgments from cluster nodes.</param>
/// <returns>The Akka.NET <see cref="Akka.Actor.Props"/> used to spawn this actor.</returns>
/// <param name="meshRouter">
/// Central comm actor the dispatch is handed to, or <see langword="null"/> to publish on the
/// DistributedPubSub mediator directly (the original behaviour).
/// </param>
/// <param name="readiness">Where the recovery-load state is reported for health, or <see langword="null"/>.</param>
/// <param name="loadRetryMin">First recovery-load retry delay; <see langword="null"/> = <see cref="DefaultLoadRetryMin"/>.</param>
/// <param name="loadRetryMax">Recovery-load retry ceiling; <see langword="null"/> = <see cref="DefaultLoadRetryMax"/>.</param>
/// <returns>The Akka.NET <see cref="Akka.Actor.Props"/> used to spawn this actor.</returns>
public static Props Props(
IDbContextFactory<OtOpcUaConfigDbContext> 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));
/// <summary>
/// Initializes a new instance of the <see cref="ConfigPublishCoordinator"/> class.
@@ -136,45 +168,167 @@ public sealed class ConfigPublishCoordinator : ReceiveActor, IWithTimers
/// <param name="dbFactory">The database context factory for persisting deployment state.</param>
/// <param name="applyDeadline">The timeout for waiting for per-node apply acknowledgments.</param>
/// <param name="meshRouter">Central comm actor, or <see langword="null"/> for the mediator.</param>
/// <param name="readiness">Where the recovery-load state is reported, or <see langword="null"/>.</param>
/// <param name="loadRetryMin">First recovery-load retry delay.</param>
/// <param name="loadRetryMax">Recovery-load retry ceiling.</param>
public ConfigPublishCoordinator(
IDbContextFactory<OtOpcUaConfigDbContext> 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();
}
/// <summary>
/// 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 <see cref="Ready"/> is in force.
/// </summary>
private void Loading()
{
Receive<RecoveryLoaded>(HandleRecoveryLoaded);
Receive<RecoveryLoadFailed>(HandleRecoveryLoadFailed);
Receive<RetryRecoveryLoad>(_ => StartRecoveryLoad());
Receive<SubscribeAck>(_ => { /* DPS subscribe confirmation */ });
Receive<DispatchDeployment>(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());
}
/// <summary>The normal operating state, entered once the recovery load has been applied.</summary>
private void Ready()
{
Receive<DispatchDeployment>(HandleDispatch);
Receive<ApplyAck>(HandleAck);
Receive<DeadlineElapsed>(HandleDeadline);
Receive<SubscribeAck>(_ => { /* DPS subscribe confirmation */ });
// A late load outcome / retry tick cannot normally arrive once Ready; ignored, never stashed.
Receive<RecoveryLoaded>(_ => { });
Receive<RecoveryLoadFailed>(_ => { });
Receive<RetryRecoveryLoad>(_ => { });
}
/// <inheritdoc />
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();
}
/// <inheritdoc />
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();
}
/// <summary>
/// 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.
/// </summary>
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<RecoveryLoaded> LoadRecoveryAsync(IDbContextFactory<OtOpcUaConfigDbContext> 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);
}
/// <summary>
/// Applies the recovered state, switches to <see cref="Ready"/> 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.
/// </summary>
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();
}
}
/// <summary>
/// Recovers the deployment that was mid-flight when a prior singleton instance died: re-derives
/// <c>_expectedAcks</c> from <c>NodeDeploymentState</c>, replays the acks that already landed, and resumes
/// the deadline timer.
/// </summary>
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);
/// <summary>The recovery read's result: the newest in-flight deployment, or null when none.</summary>
private sealed record RecoveryLoaded(RecoveredDeployment? Inflight);
/// <summary>The recovery read failed (ConfigDb unreachable, ...); retried with backoff.</summary>
private sealed record RecoveryLoadFailed(Exception Error);
/// <summary>The backoff timer's tick: run the recovery read again.</summary>
private sealed class RetryRecoveryLoad
{
public static readonly RetryRecoveryLoad Instance = new();
}
private sealed record RecoveredDeployment(
Guid DeploymentId, string RevisionHash, DateTime CreatedAtUtc, IReadOnlyList<RecoveredNodeState> NodeStates);
private sealed record RecoveredNodeState(string NodeId, NodeDeploymentStatus Status, string? FailureReason);
/// <summary>
/// Stages a status change WITHOUT materialising the row: the row carries the whole <c>ArtifactBlob</c>
/// (varbinary(max), multi-MB on a real fleet), which no status transition reads (fix round 1). A stub
@@ -0,0 +1,50 @@
namespace ZB.MOM.WW.OtOpcUa.ControlPlane.Coordinators;
/// <summary>Where the <see cref="ConfigPublishCoordinator"/> singleton stands on this node (lmxopcua#868).</summary>
public enum ConfigPublishCoordinatorState
{
/// <summary>The singleton is not running on this node (the partner hosts it, or it has not started yet).</summary>
NotHosted,
/// <summary>Running here, but its recovery state has not loaded from the ConfigDb yet; deployments are queued.</summary>
Loading,
/// <summary>Running here with its recovery state loaded; deployments are dispatched.</summary>
Ready,
}
/// <summary>One immutable observation of the coordinator's state.</summary>
/// <param name="State">The state.</param>
/// <param name="FailedLoadAttempts">Consecutive failed recovery loads (0 unless <see cref="ConfigPublishCoordinatorState.Loading"/>).</param>
/// <param name="LastError">The last load failure's message, or null.</param>
/// <param name="SinceUtc">When the state was entered.</param>
public sealed record ConfigPublishCoordinatorSnapshot(
ConfigPublishCoordinatorState State, int FailedLoadAttempts, string? LastError, DateTime SinceUtc);
/// <summary>
/// Thread-safe state bridge from the <see cref="ConfigPublishCoordinator"/> actor to the node's health endpoint
/// (one DI singleton per admin node). The actor writes; the health check reads.
/// </summary>
public sealed class ConfigPublishCoordinatorReadiness
{
private volatile ConfigPublishCoordinatorSnapshot _current =
new(ConfigPublishCoordinatorState.NotHosted, 0, null, DateTime.UtcNow);
/// <summary>The latest observation.</summary>
public ConfigPublishCoordinatorSnapshot Current => _current;
/// <summary>The coordinator is loading (or retrying the load of) its recovery state.</summary>
/// <param name="failedAttempts">Failed attempts so far.</param>
/// <param name="lastError">The last failure's message, or null.</param>
public void MarkLoading(int failedAttempts, string? lastError)
{
var since = _current.State == ConfigPublishCoordinatorState.Loading ? _current.SinceUtc : DateTime.UtcNow;
_current = new(ConfigPublishCoordinatorState.Loading, failedAttempts, lastError, since);
}
/// <summary>The recovery state is loaded.</summary>
public void MarkReady() => _current = new(ConfigPublishCoordinatorState.Ready, 0, null, DateTime.UtcNow);
/// <summary>The singleton stopped on this node.</summary>
public void MarkNotHosted() => _current = new(ConfigPublishCoordinatorState.NotHosted, 0, null, DateTime.UtcNow);
}
@@ -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<CentralCommunicationActorKey>();
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<IServiceProvider>()?.GetService<ConfigPublishCoordinatorReadiness>();
return ConfigPublishCoordinator.Props(
dbFactory, applyDeadline: DeployTimings.ApplyDeadline, meshRouter: meshRouter, readiness: readiness);
},
singletonOptions);
@@ -0,0 +1,45 @@
using Microsoft.Extensions.Diagnostics.HealthChecks;
using ZB.MOM.WW.OtOpcUa.ControlPlane.Coordinators;
namespace ZB.MOM.WW.OtOpcUa.Host.Health;
/// <summary>
/// The deploy coordinator's readiness (lmxopcua#868): <b>Degraded</b> while the
/// <see cref="ConfigPublishCoordinator"/> 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").
/// </summary>
/// <param name="readiness">The coordinator's state bridge; null on graphs that never register one.</param>
public sealed class ConfigPublishCoordinatorHealthCheck(ConfigPublishCoordinatorReadiness? readiness) : IHealthCheck
{
/// <summary>The check's registration name.</summary>
public const string Name = "config-publish";
/// <inheritdoc />
public Task<HealthCheckResult> 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<string, object>(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),
});
}
}
@@ -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<ZB.MOM.WW.OtOpcUa.ControlPlane.Coordinators.ConfigPublishCoordinatorReadiness>()),
failureStatus: null,
tags: new[] { ZbHealthTags.Ready }));
}
checks
@@ -176,7 +176,12 @@ else
builder.Services.Configure<AclOptions>(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<ZB.MOM.WW.OtOpcUa.ControlPlane.Coordinators.ConfigPublishCoordinatorReadiness>();
}
else
Log.Information("Driver-only node — no ConfigDb registered; configuration via ConfigSource:Mode=FetchAndCache");
@@ -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;
/// <summary>
/// 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 <c>GetType()</c>.
/// </summary>
[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;
}
/// <summary>Every fixed value leaf (capability strings excluded), with its declared type.</summary>
public static TheoryData<string> CatalogValueLeaves()
{
var data = new TheoryData<string>();
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));
}
}
@@ -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;
/// <summary>
/// lmxopcua#868: the coordinator's recovery read used to run synchronously in <c>PreStart</c>. 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 <see cref="ConfigPublishCoordinatorState.Loading"/> until the load succeeds.
/// </summary>
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<OtOpcUaConfigDbContext> 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<OtOpcUaConfigDbContext> 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();
}
/// <summary>A factory that throws while <see cref="Down"/> — a ConfigDb that is unreachable.</summary>
private sealed class OutageDbFactory(IDbContextFactory<OtOpcUaConfigDbContext> inner) : IDbContextFactory<OtOpcUaConfigDbContext>
{
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();
}
}
}
@@ -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<IDriverProbe>()));
// 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()));
@@ -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;
/// <summary>lmxopcua#868: the deploy coordinator reports Degraded (never Unhealthy) while it cannot load from the ConfigDb.</summary>
public sealed class ConfigPublishCoordinatorHealthCheckTests
{
private static async Task<HealthCheckResult> 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<IOptions<HealthCheckServiceOptions>>()
.Value.Registrations
.Where(x => x.Name == ConfigPublishCoordinatorHealthCheck.Name)
.ToList();
if (!registered)
{
registrations.ShouldBeEmpty();
return;
}
registrations.ShouldHaveSingleItem().Tags.ShouldBe([ZbHealthTags.Ready]);
}
}