dohertj2/lmxopcua
1
0
Fork 0
Code Issues 43 Pull Requests Actions Packages Projects Releases Wiki Activity

fix(controlplane): close a lost dispatch from the in-flight check; guard terminal rows; blob-free status writes

Browse Source 
I7: a Dispatching row committed after the coordinator's PreStart whose Tell never arrived had no owner and
refused every later deploy. The in-flight check (now fully off the admin singleton's thread, serialised by the
actor-local compose flag alone) closes a Dispatching row with zero NodeDeploymentState rows that is past the
apply deadline as TimedOut (optimistic on RowVersion) and proceeds.

HandleDispatch ignores a dispatch for a Sealed/PartiallyFailed/TimedOut row, so a late or duplicated dispatch
cannot flip it back. Coordinator status/reason writes attach a key+RowVersion stub instead of loading the
multi-MB ArtifactBlob; proven on the real SQL Server (blob untouched, seal lands).

Refs #733
Claude-Session: https://claude.ai/code/session_01Ub9hGmYLQvn9XETzw7CMsT
This commit is contained in:
Joseph Doherty
2026-10-22 13:46:54 -04:00
parent f7e7ba0263
commit 9466b8b242
3 changed files with 332 additions and 59 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/Server/ZB.MOM.WW.OtOpcUa.ControlPlane/AdminOperations/AdminOperationsActor.cs
+127 -45
View File
@@ -11,6 +11,7 @@ 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.Configuration.Validation;
using ZB.MOM.WW.OtOpcUa.ControlPlane.Coordinators;
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
using ZB.MOM.WW.OtOpcUa.Core.Scripting;
@@ -28,6 +29,12 @@ public sealed class AdminOperationsActor : ReceiveActor
private readonly IReadOnlyDictionary<string, IDriverProbe> _probesByType;
private readonly TagConfigValidationMode _tagConfigValidationMode;
/// <summary>
/// The coordinator's apply deadline. A <c>Dispatching</c> row with no expected-ack rows that is older
/// than this is a lost dispatch, closed by the in-flight check (fix round 1 / I7).
/// </summary>
private readonly TimeSpan _applyDeadline;
/// <summary>
/// Where central→node commands are handed off (per-cluster mesh), or
/// <see langword="null"/> to publish on the mediator directly.
@@ -48,15 +55,17 @@ public sealed class AdminOperationsActor : ReceiveActor
/// Central comm actor commands are handed to, or <see langword="null"/> to publish on the
/// DistributedPubSub mediator directly (the behaviour before the per-cluster mesh).
/// </param>
/// <param name="applyDeadline">The coordinator's apply deadline; defaults to <see cref="ConfigPublishCoordinator.DefaultApplyDeadline"/>.</param>
/// <returns>Props configured to create an AdminOperationsActor.</returns>
public static Props Props(
IDbContextFactory<OtOpcUaConfigDbContext> dbFactory,
IActorRef coordinator,
IEnumerable<IDriverProbe> probes,
TagConfigValidationMode tagConfigValidationMode = TagConfigValidationMode.Warn,
IActorRef? meshRouter = null) =>
IActorRef? meshRouter = null,
TimeSpan? applyDeadline = null) =>
Akka.Actor.Props.Create(() =>
new AdminOperationsActor(dbFactory, coordinator, probes, tagConfigValidationMode, meshRouter));
new AdminOperationsActor(dbFactory, coordinator, probes, tagConfigValidationMode, meshRouter, applyDeadline));
/// <summary>Initializes a new instance of the AdminOperationsActor.</summary>
/// <param name="dbFactory">Factory for creating config database contexts.</param>
@@ -64,13 +73,16 @@ public sealed class AdminOperationsActor : ReceiveActor
/// <param name="probes">Driver probes registered in DI; keyed by DriverType (case-insensitive).</param>
/// <param name="tagConfigValidationMode">Deploy-gate TagConfig strictness mode (Warn default | Error opt-in).</param>
/// <param name="meshRouter">Central comm actor, or <see langword="null"/> for the mediator.</param>
/// <param name="applyDeadline">The coordinator's apply deadline; defaults to <see cref="ConfigPublishCoordinator.DefaultApplyDeadline"/>.</param>
public AdminOperationsActor(
IDbContextFactory<OtOpcUaConfigDbContext> dbFactory,
IActorRef coordinator,
IEnumerable<IDriverProbe> probes,
TagConfigValidationMode tagConfigValidationMode = TagConfigValidationMode.Warn,
IActorRef? meshRouter = null)
IActorRef? meshRouter = null,
TimeSpan? applyDeadline = null)
{
_applyDeadline = applyDeadline ?? ConfigPublishCoordinator.DefaultApplyDeadline;
_dbFactory = dbFactory;
_coordinator = coordinator;
_probesByType = probes.ToDictionary(p => p.DriverType, StringComparer.OrdinalIgnoreCase);
@@ -201,16 +213,17 @@ public sealed class AdminOperationsActor : ReceiveActor
}
/// <summary>
/// Serialised part of <c>StartDeployment</c>: the in-flight check. Everything after it — the
/// snapshot read, validation, gzip and the row write — runs on the thread pool and comes back as a
/// <see cref="DeployComposed"/> / <see cref="DeployComposeFailed"/> message, so a large compose no
/// longer suspends the mailbox (R03:3.1) and alarm acks keep flowing while it runs.
/// <c>StartDeployment</c>. The only serialised part is the actor-local <see cref="_composeInFlight"/>
/// flag; the DB in-flight check, the snapshot read, validation, gzip and the row write all run on the
/// thread pool and come back as a <see cref="DeployComposed"/> / <see cref="DeployComposeFailed"/>
/// message, so a slow or blocked ConfigDb never suspends the mailbox (R03:3.1; fix round 1 moved the
/// in-flight query off the actor thread too) and alarm acks keep flowing while it runs.
/// </summary>
/// <remarks>
/// <see cref="_composeInFlight"/> is what keeps the check correct once the compose is off-actor: the
/// Deployment row is written at the END of the compose, so a second request arriving mid-compose
/// would pass a DB-only check and two artifacts would race for one coordinator. The flag is cleared
/// when the outcome message lands, on both branches.
/// <see cref="_composeInFlight"/> is what keeps the check correct once it is off-actor: it is set
/// BEFORE the off-actor work starts and cleared when the outcome message lands (both branches), so a
/// second request is refused in memory for the whole check-plus-compose and two artifacts can never
/// race for one coordinator.
/// </remarks>
private void HandleStartDeployment(StartDeployment msg)
{
@@ -226,49 +239,118 @@ public sealed class AdminOperationsActor : ReceiveActor
return;
}
try
{
// Refuse if any deployment is already in flight — keeps the coordinator's state
// unambiguous. The UI is expected to wait for the in-flight one to seal or fail. One indexed
// projection (IX_Deployment_Status), synchronous on purpose: it IS the serialised part.
using var db = _dbFactory.CreateDbContext();
var inflight = db.Deployments
.Where(d => d.Status == DeploymentStatus.Dispatching || d.Status == DeploymentStatus.AwaitingApplyAcks)
.Select(d => d.DeploymentId)
.FirstOrDefault();
if (inflight != Guid.Empty)
{
replyTo.Tell(new StartDeploymentResult(
StartDeploymentOutcome.AnotherDeploymentInFlight,
DeploymentId: new DeploymentId(inflight),
RevisionHash: null,
Message: $"Deployment {inflight:N} is still in flight.",
msg.CorrelationId));
return;
}
}
catch (Exception ex)
{
_log.Error(ex, "StartDeployment failed for {CreatedBy} (correlation {CorrelationId})", msg.CreatedBy, msg.CorrelationId);
replyTo.Tell(new StartDeploymentResult(
StartDeploymentOutcome.Rejected,
DeploymentId: null,
RevisionHash: null,
Message: ServerFault("StartDeployment", msg.CorrelationId),
msg.CorrelationId));
return;
}
_composeInFlight = msg;
// Captured here: Context/Sender are not available on the thread-pool thread.
var sourceNode = Akka.Cluster.Cluster.Get(Context.System).SelfAddress.Host ?? "unknown";
var self = Self;
Task.Run(() => ComposeDeploymentAsync(msg, sourceNode))
Task.Run(async () =>
{
var refusal = await CheckInFlightAsync(msg).ConfigureAwait(false);
return refusal ?? await ComposeDeploymentAsync(msg, sourceNode).ConfigureAwait(false);
})
.PipeTo(self,
success: composed => new DeployComposed(msg, replyTo, composed.Result, composed.Dispatch),
failure: ex => new DeployComposeFailed(msg, replyTo, ex));
}
/// <summary>
/// The DB half of the in-flight check, off the actor thread. Refuses while a deployment is in flight —
/// keeping the coordinator's state unambiguous — EXCEPT a lost dispatch, which it closes first.
/// </summary>
/// <param name="msg">The request.</param>
/// <returns>The refusal, or null to proceed.</returns>
/// <remarks>
/// <para>
/// <b>Lost dispatch (fix round 1 / I7).</b> A <c>Dispatching</c> row with ZERO
/// <c>NodeDeploymentState</c> rows was never taken by the coordinator (its <c>HandleDispatch</c>
/// seeds the expected set and flips the status in one <c>SaveChanges</c>). The coordinator's PreStart
/// recovers such a row — but only one that already existed when it started. A row committed AFTER
/// PreStart whose <c>Tell</c> never arrived (this actor died between the compose's commit and
/// <see cref="HandleDeployComposed"/>) has no owner at all, and refused every later deploy forever.
/// Once it is past the apply deadline nothing can still be legitimately dispatching it, so it is
/// closed <c>TimedOut</c> with a reason and the new deploy proceeds. A late <c>DispatchDeployment</c>
/// for it is ignored by the coordinator's terminal-state guard.
/// </para>
/// <para>
/// One indexed projection (<c>IX_Deployment_Status</c>) plus, only for a Dispatching row past the
/// deadline, a count and a tracked update.
/// </para>
/// </remarks>
private async Task<ComposeOutcome?> CheckInFlightAsync(StartDeployment msg)
{
await using var db = await _dbFactory.CreateDbContextAsync().ConfigureAwait(false);
var inflight = await db.Deployments
.AsNoTracking()
.Where(d => d.Status == DeploymentStatus.Dispatching || d.Status == DeploymentStatus.AwaitingApplyAcks)
.OrderByDescending(d => d.CreatedAtUtc)
.Select(d => new { d.DeploymentId, d.Status, d.CreatedAtUtc })
.ToListAsync()
.ConfigureAwait(false);
foreach (var row in inflight)
{
if (row.Status == DeploymentStatus.Dispatching
&& DateTime.UtcNow - row.CreatedAtUtc > _applyDeadline
&& !await db.NodeDeploymentStates.AnyAsync(s => s.DeploymentId == row.DeploymentId).ConfigureAwait(false)
&& await CloseLostDispatchAsync(db, row.DeploymentId).ConfigureAwait(false))
{
continue;
}
return new ComposeOutcome(
new StartDeploymentResult(
StartDeploymentOutcome.AnotherDeploymentInFlight,
DeploymentId: new DeploymentId(row.DeploymentId),
RevisionHash: null,
Message: $"Deployment {row.DeploymentId:N} is still in flight.",
msg.CorrelationId),
Dispatch: null);
}
return null;
}
/// <summary>Closes a lost dispatch as TimedOut, unless it moved on meanwhile (optimistic, on its RowVersion).</summary>
/// <param name="db">The context.</param>
/// <param name="deploymentId">The row.</param>
/// <returns><see langword="true"/> when the row is now closed.</returns>
private async Task<bool> CloseLostDispatchAsync(OtOpcUaConfigDbContext db, Guid deploymentId)
{
var stamp = await db.Deployments
.AsNoTracking()
.Where(d => d.DeploymentId == deploymentId && d.Status == DeploymentStatus.Dispatching)
.Select(d => new { d.RowVersion })
.FirstOrDefaultAsync()
.ConfigureAwait(false);
if (stamp is null) return true; // no longer Dispatching — not in flight as far as this check goes
// Stub update: the row's ArtifactBlob is never read just to change its status.
var stub = new Deployment
{
DeploymentId = deploymentId, RevisionHash = string.Empty, CreatedBy = string.Empty,
RowVersion = stamp.RowVersion,
Status = DeploymentStatus.TimedOut,
FailureReason =
"Dispatch was lost before any node was expected to ack (the coordinator was never told, or never " +
"took it) and the apply deadline has passed; closed by the next StartDeployment. Redeploy.",
};
var entry = db.Deployments.Attach(stub);
entry.Property(d => d.Status).IsModified = true;
entry.Property(d => d.FailureReason).IsModified = true;
try
{
await db.SaveChangesAsync().ConfigureAwait(false);
}
catch (DbUpdateConcurrencyException)
{
// The coordinator took it after all (status flipped, RowVersion moved): it is genuinely in flight.
return false;
}
_log.Warning(
"StartDeployment: closed lost dispatch {DeploymentId} as TimedOut — Dispatching with no expected-ack rows " +
"past the {Deadline} apply deadline (the coordinator never took it)", deploymentId, _applyDeadline);
return true;
}
private void HandleDeployComposed(DeployComposed msg)
{
_composeInFlight = null;