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
This commit is contained in:
+202
-23
@@ -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,
|
/// <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.
|
/// so honouring it here would invent semantics. Disable the nodes, not the cluster row.
|
||||||
/// </para>
|
/// </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>
|
/// </summary>
|
||||||
public sealed class ConfigPublishCoordinator : ReceiveActor, IWithTimers
|
public sealed class ConfigPublishCoordinator : ReceiveActor, IWithTimers, IWithUnboundedStash
|
||||||
{
|
{
|
||||||
public const string DeploymentsTopic = "deployments";
|
public const string DeploymentsTopic = "deployments";
|
||||||
public const string DeploymentAcksTopic = "deployment-acks";
|
public const string DeploymentAcksTopic = "deployment-acks";
|
||||||
@@ -61,8 +73,18 @@ public sealed class ConfigPublishCoordinator : ReceiveActor, IWithTimers
|
|||||||
/// apply-outcome timeout is derived from.</summary>
|
/// apply-outcome timeout is derived from.</summary>
|
||||||
public static readonly TimeSpan DefaultApplyDeadline = DeployTimings.ApplyDeadline;
|
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 IDbContextFactory<OtOpcUaConfigDbContext> _dbFactory;
|
||||||
private readonly TimeSpan _applyDeadline;
|
private readonly TimeSpan _applyDeadline;
|
||||||
|
private readonly TimeSpan _loadRetryMin;
|
||||||
|
private readonly TimeSpan _loadRetryMax;
|
||||||
|
private readonly ConfigPublishCoordinatorReadiness? _readiness;
|
||||||
|
private int _failedLoads;
|
||||||
|
|
||||||
/// <summary>
|
/// <summary>
|
||||||
/// Where central→node commands are handed off (per-cluster mesh Phase 2), or
|
/// 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>
|
/// <summary>Gets the timer scheduler for managing apply deadlines.</summary>
|
||||||
public ITimerScheduler Timers { get; set; } = null!;
|
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>
|
/// <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="dbFactory">The database context factory for accessing configuration data.</param>
|
||||||
/// <param name="applyDeadline">The timeout for waiting for apply acknowledgments from cluster nodes.</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">
|
/// <param name="meshRouter">
|
||||||
/// Central comm actor the dispatch is handed to, or <see langword="null"/> to publish on the
|
/// Central comm actor the dispatch is handed to, or <see langword="null"/> to publish on the
|
||||||
/// DistributedPubSub mediator directly (the original behaviour).
|
/// DistributedPubSub mediator directly (the original behaviour).
|
||||||
/// </param>
|
/// </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(
|
public static Props Props(
|
||||||
IDbContextFactory<OtOpcUaConfigDbContext> dbFactory,
|
IDbContextFactory<OtOpcUaConfigDbContext> dbFactory,
|
||||||
TimeSpan? applyDeadline = null,
|
TimeSpan? applyDeadline = null,
|
||||||
IActorRef? meshRouter = null) =>
|
IActorRef? meshRouter = null,
|
||||||
Akka.Actor.Props.Create(() =>
|
ConfigPublishCoordinatorReadiness? readiness = null,
|
||||||
new ConfigPublishCoordinator(dbFactory, applyDeadline ?? DefaultApplyDeadline, meshRouter));
|
TimeSpan? loadRetryMin = null,
|
||||||
|
TimeSpan? loadRetryMax = null) =>
|
||||||
|
Akka.Actor.Props.Create(() => new ConfigPublishCoordinator(
|
||||||
|
dbFactory, applyDeadline ?? DefaultApplyDeadline, meshRouter, readiness,
|
||||||
|
loadRetryMin ?? DefaultLoadRetryMin, loadRetryMax ?? DefaultLoadRetryMax));
|
||||||
|
|
||||||
/// <summary>
|
/// <summary>
|
||||||
/// Initializes a new instance of the <see cref="ConfigPublishCoordinator"/> class.
|
/// 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="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="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="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(
|
public ConfigPublishCoordinator(
|
||||||
IDbContextFactory<OtOpcUaConfigDbContext> dbFactory,
|
IDbContextFactory<OtOpcUaConfigDbContext> dbFactory,
|
||||||
TimeSpan applyDeadline,
|
TimeSpan applyDeadline,
|
||||||
IActorRef? meshRouter = null)
|
IActorRef? meshRouter = null,
|
||||||
|
ConfigPublishCoordinatorReadiness? readiness = null,
|
||||||
|
TimeSpan? loadRetryMin = null,
|
||||||
|
TimeSpan? loadRetryMax = null)
|
||||||
{
|
{
|
||||||
_dbFactory = dbFactory;
|
_dbFactory = dbFactory;
|
||||||
_applyDeadline = applyDeadline;
|
_applyDeadline = applyDeadline;
|
||||||
_meshRouter = meshRouter;
|
_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<DispatchDeployment>(HandleDispatch);
|
||||||
Receive<ApplyAck>(HandleAck);
|
Receive<ApplyAck>(HandleAck);
|
||||||
Receive<DeadlineElapsed>(HandleDeadline);
|
Receive<DeadlineElapsed>(HandleDeadline);
|
||||||
Receive<SubscribeAck>(_ => { /* DPS subscribe confirmation */ });
|
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 />
|
/// <inheritdoc />
|
||||||
protected override void PreStart()
|
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
|
// Subscribe to per-node ApplyAck broadcasts so DriverHostActors on remote members can
|
||||||
// route their ACKs to whichever node currently hosts this singleton.
|
// route their ACKs to whichever node currently hosts this singleton.
|
||||||
DistributedPubSub.Get(Context.System).Mediator.Tell(new Subscribe(DeploymentAcksTopic, Self));
|
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
|
// 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.
|
// 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)
|
.Where(d => d.Status == DeploymentStatus.Dispatching || d.Status == DeploymentStatus.AwaitingApplyAcks)
|
||||||
.OrderByDescending(d => d.CreatedAtUtc)
|
.OrderByDescending(d => d.CreatedAtUtc)
|
||||||
.Select(d => new { d.DeploymentId, d.RevisionHash, 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;
|
if (inflight is null) return;
|
||||||
|
|
||||||
_current = new DeploymentId(inflight.DeploymentId);
|
_current = new DeploymentId(inflight.DeploymentId);
|
||||||
var nodeStates = db.NodeDeploymentStates
|
var nodeStates = inflight.NodeStates;
|
||||||
.Where(x => x.DeploymentId == inflight.DeploymentId)
|
|
||||||
.AsNoTracking()
|
|
||||||
.ToList();
|
|
||||||
|
|
||||||
// Resume the deadline timer using the remaining time. The deadline runs from when the
|
// 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
|
// 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); " +
|
"before any node was expected (coordinator restart or singleton hand-over mid-dispatch); " +
|
||||||
"re-dispatching with {Remaining} of the apply deadline left",
|
"re-dispatching with {Remaining} of the apply deadline left",
|
||||||
_current, remaining);
|
_current, remaining);
|
||||||
Self.Tell(new DispatchDeployment(
|
HandleDispatch(new DispatchDeployment(
|
||||||
_current.Value, RevisionHash.Parse(inflight.RevisionHash), CorrelationId.NewId()));
|
_current.Value, RevisionHash.Parse(inflight.RevisionHash), CorrelationId.NewId()));
|
||||||
}
|
}
|
||||||
else
|
else
|
||||||
{
|
{
|
||||||
Self.Tell(new DeadlineElapsed(_current.Value));
|
HandleDeadline(new DeadlineElapsed(_current.Value));
|
||||||
}
|
}
|
||||||
return;
|
return;
|
||||||
}
|
}
|
||||||
@@ -222,18 +376,25 @@ public sealed class ConfigPublishCoordinator : ReceiveActor, IWithTimers
|
|||||||
|
|
||||||
if (remaining <= TimeSpan.Zero)
|
if (remaining <= TimeSpan.Zero)
|
||||||
{
|
{
|
||||||
Self.Tell(new DeadlineElapsed(_current.Value));
|
HandleDeadline(new DeadlineElapsed(_current.Value));
|
||||||
}
|
return;
|
||||||
else
|
|
||||||
{
|
|
||||||
Timers.StartSingleTimer(DeadlineTimerKey, new DeadlineElapsed(_current.Value), remaining);
|
|
||||||
}
|
}
|
||||||
|
|
||||||
|
Timers.StartSingleTimer(DeadlineTimerKey, new DeadlineElapsed(_current.Value), remaining);
|
||||||
_log.Info("Coordinator recovered in-flight deployment {Id} ({Acked}/{Total} acks landed)",
|
_log.Info("Coordinator recovered in-flight deployment {Id} ({Acked}/{Total} acks landed)",
|
||||||
_current, _acks.Count, _expectedAcks.Count);
|
_current, _acks.Count, _expectedAcks.Count);
|
||||||
}
|
}
|
||||||
|
|
||||||
private void HandleDispatch(DispatchDeployment msg)
|
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
|
// 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
|
// 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.
|
// 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 DeadlineTimerKey = "apply-deadline";
|
||||||
|
private const string LoadRetryTimerKey = "recovery-load-retry";
|
||||||
|
|
||||||
public sealed record DeadlineElapsed(DeploymentId DeploymentId);
|
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>
|
/// <summary>
|
||||||
/// Stages a status change WITHOUT materialising the row: the row carries the whole <c>ArtifactBlob</c>
|
/// 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
|
/// (varbinary(max), multi-MB on a real fleet), which no status transition reads (fix round 1). A stub
|
||||||
|
|||||||
+50
@@ -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
|
// coordinator. Deliberately NOT an ActorSelection.ResolveOne here: that blocks
|
||||||
// inside a props factory and races the spawn it is waiting for.
|
// inside a props factory and races the spawn it is waiting for.
|
||||||
var meshRouter = registry.Get<CentralCommunicationActorKey>();
|
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);
|
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),
|
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
|
checks
|
||||||
|
|||||||
@@ -176,7 +176,12 @@ else
|
|||||||
builder.Services.Configure<AclOptions>(builder.Configuration.GetSection(AclOptions.SectionName));
|
builder.Services.Configure<AclOptions>(builder.Configuration.GetSection(AclOptions.SectionName));
|
||||||
|
|
||||||
if (hasAdmin)
|
if (hasAdmin)
|
||||||
|
{
|
||||||
builder.Services.AddOtOpcUaConfigDb(builder.Configuration);
|
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
|
else
|
||||||
Log.Information("Driver-only node — no ConfigDb registered; configuration via ConfigSource:Mode=FetchAndCache");
|
Log.Information("Driver-only node — no ConfigDb registered; configuration via ConfigSource:Mode=FetchAndCache");
|
||||||
|
|
||||||
|
|||||||
+185
@@ -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();
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
+3
-3
@@ -136,9 +136,9 @@ public sealed class ConfigPublishCoordinatorRecoveryTests : ControlPlaneActorTes
|
|||||||
var coordinator = Sys.ActorOf(ConfigPublishCoordinator.Props(faulting, NoTimeout));
|
var coordinator = Sys.ActorOf(ConfigPublishCoordinator.Props(faulting, NoTimeout));
|
||||||
var admin = Sys.ActorOf(AdminOperationsActor.Props(adminFactory, coordinator, Enumerable.Empty<IDriverProbe>()));
|
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
|
// Arm the fault AFTER PreStart has run. Since lmxopcua#868 the recovery read is asynchronous and
|
||||||
// the actor would be stopped, not restarted). Identify is processed after creation, so a
|
// retried, so a fault that lands on it instead of on HandleDispatch is also recovered (the dispatch is
|
||||||
// resolved selection proves PreStart completed.
|
// queued until the retried load succeeds) — either way the deployment must complete.
|
||||||
await Sys.ActorSelection(coordinator.Path).ResolveOne(PresenceBudget, TestContext.Current.CancellationToken);
|
await Sys.ActorSelection(coordinator.Path).ResolveOne(PresenceBudget, TestContext.Current.CancellationToken);
|
||||||
faulting.FailNext();
|
faulting.FailNext();
|
||||||
admin.Tell(new StartDeployment("joe", CorrelationId.NewId()));
|
admin.Tell(new StartDeployment("joe", CorrelationId.NewId()));
|
||||||
|
|||||||
+65
@@ -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]);
|
||||||
|
}
|
||||||
|
}
|
||||||
Reference in New Issue
Block a user