Joseph Doherty
241318e7e5
v2-ci / unit-tests (push) Has been cancelled
Item 5: DraftValidator now emits SqlQueryNotDeclared (always blocking) for a Query tag whose query is absent from the merged driver config's queries map, via the driver's own tag parse. Item 3: StartDeployment appends a non-blocking warning to the accepted message when the Tag entity's AccessLevel and the Sql blob's writable flag disagree. Claude-Session: https://claude.ai/code/session_01GT67AEHd2Acv1YPwGfXEoV
688 lines
36 KiB
C#
688 lines
36 KiB
C#
using Akka.Actor;
|
|
using Akka.Cluster.Tools.PublishSubscribe;
|
|
using Akka.Event;
|
|
using Microsoft.EntityFrameworkCore;
|
|
using ZB.MOM.WW.OtOpcUa.Commons.Messages.Admin;
|
|
using ZB.MOM.WW.OtOpcUa.Commons.Messages.Deploy;
|
|
using ZB.MOM.WW.OtOpcUa.Commons.Messages.Mesh;
|
|
using ZB.MOM.WW.OtOpcUa.Commons.OpcUa;
|
|
using ZB.MOM.WW.OtOpcUa.Commons.Types;
|
|
using ZB.MOM.WW.OtOpcUa.Configuration;
|
|
using ZB.MOM.WW.OtOpcUa.Configuration.Entities;
|
|
using ZB.MOM.WW.OtOpcUa.Configuration.Enums;
|
|
using ZB.MOM.WW.OtOpcUa.Configuration.Validation;
|
|
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
|
|
using ZB.MOM.WW.OtOpcUa.Core.Scripting;
|
|
|
|
namespace ZB.MOM.WW.OtOpcUa.ControlPlane.AdminOperations;
|
|
|
|
/// <summary>
|
|
/// Cluster-singleton admin operations actor. Owns the "snapshot the live-edit state and start
|
|
/// a deployment" workflow plus (eventually) all mutating live-edit ops invoked by the admin UI.
|
|
/// Routed to via <see cref="Commons.Interfaces.IAdminOperationsClient"/> from anywhere in the cluster.
|
|
/// </summary>
|
|
public sealed class AdminOperationsActor : ReceiveActor
|
|
{
|
|
private readonly IDbContextFactory<OtOpcUaConfigDbContext> _dbFactory;
|
|
private readonly IActorRef _coordinator;
|
|
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.
|
|
/// </summary>
|
|
private readonly IActorRef? _meshRouter;
|
|
|
|
private readonly ILoggingAdapter _log = Context.GetLogger();
|
|
|
|
/// <summary>The StartDeployment whose compose is running off-actor, or null. See <see cref="HandleStartDeployment"/>.</summary>
|
|
private StartDeployment? _composeInFlight;
|
|
|
|
/// <summary>Creates actor props for the admin operations actor.</summary>
|
|
/// <param name="dbFactory">Factory for creating config database contexts.</param>
|
|
/// <param name="coordinator">Reference to the deployment coordinator actor.</param>
|
|
/// <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 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="DeployTimings.ApplyDeadline"/> (the coordinator's default source).</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,
|
|
TimeSpan? applyDeadline = null) =>
|
|
Akka.Actor.Props.Create(() =>
|
|
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>
|
|
/// <param name="coordinator">Reference to the deployment coordinator actor.</param>
|
|
/// <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="DeployTimings.ApplyDeadline"/> (the coordinator's default source).</param>
|
|
public AdminOperationsActor(
|
|
IDbContextFactory<OtOpcUaConfigDbContext> dbFactory,
|
|
IActorRef coordinator,
|
|
IEnumerable<IDriverProbe> probes,
|
|
TagConfigValidationMode tagConfigValidationMode = TagConfigValidationMode.Warn,
|
|
IActorRef? meshRouter = null,
|
|
TimeSpan? applyDeadline = null)
|
|
{
|
|
// The same source the coordinator's default reads (ConfigPublishCoordinator.DefaultApplyDeadline IS
|
|
// DeployTimings.ApplyDeadline), so the lost-dispatch closure and the coordinator's own deadline agree.
|
|
_applyDeadline = applyDeadline ?? DeployTimings.ApplyDeadline;
|
|
_dbFactory = dbFactory;
|
|
_coordinator = coordinator;
|
|
_probesByType = probes.ToDictionary(p => p.DriverType, StringComparer.OrdinalIgnoreCase);
|
|
_tagConfigValidationMode = tagConfigValidationMode;
|
|
_meshRouter = meshRouter;
|
|
|
|
// Synchronous receives: the compose and the probe run off-actor and come back as messages, so
|
|
// the mailbox is never suspended for the length of a DB snapshot or a PLC connect timeout.
|
|
Receive<StartDeployment>(HandleStartDeployment);
|
|
Receive<DeployComposed>(HandleDeployComposed);
|
|
Receive<DeployComposeFailed>(HandleDeployComposeFailed);
|
|
Receive<TestDriverConnect>(HandleTestDriverConnect);
|
|
ReceiveAsync<RestartDriver>(HandleRestartDriverAsync);
|
|
ReceiveAsync<ReconnectDriver>(HandleReconnectDriverAsync);
|
|
Receive<AcknowledgeAlarmCommand>(HandleAcknowledgeAlarm);
|
|
Receive<ShelveAlarmCommand>(HandleShelveAlarm);
|
|
}
|
|
|
|
/// <summary>
|
|
/// The reply text for an unexpected server-side failure (R03:4.1): a fixed sentence plus the request's
|
|
/// correlation id, never <c>ex.Message</c> — a SqlException or UriFormatException message carries
|
|
/// connection-string fragments and host names, and the reply reaches any ConfigEditor browser session.
|
|
/// The detail is in the central log under the same correlation id.
|
|
/// </summary>
|
|
/// <param name="operation">The admin operation that failed.</param>
|
|
/// <param name="correlationId">The request's correlation token, as logged alongside the exception.</param>
|
|
/// <returns>The operator-facing message.</returns>
|
|
internal static string ServerFault(string operation, object correlationId) =>
|
|
$"{operation} failed on the server; see the central node log for correlation {correlationId}.";
|
|
|
|
/// <summary>
|
|
/// Hands a central→node command to the mesh transport router, or publishes it directly when
|
|
/// no router is wired.
|
|
/// </summary>
|
|
/// <remarks>
|
|
/// The fallback keeps ~12 existing tests (and any node wired before per-cluster mesh)
|
|
/// on exactly the pre-mesh behaviour. The DistributedPubSub branch behind it is
|
|
/// <b>retained</b>, not scheduled for deletion — an earlier note here promised Phase 6 would
|
|
/// remove it, which was deliberately reversed (<c>docs/Configuration.md</c> § "The compiled
|
|
/// transport-mode defaults were deliberately NOT flipped"): DPS is still the correct and
|
|
/// default transport for a single-mesh topology; gRPC is what crosses a mesh boundary.
|
|
/// </remarks>
|
|
/// <param name="topic">The legacy DistributedPubSub topic.</param>
|
|
/// <param name="message">The command.</param>
|
|
private void Dispatch(string topic, object message)
|
|
{
|
|
if (_meshRouter is not null) _meshRouter.Tell(new MeshCommand(topic, message));
|
|
else DistributedPubSub.Get(Context.System).Mediator.Tell(new Publish(topic, message));
|
|
}
|
|
|
|
/// <summary>
|
|
/// AdminUI Acknowledge path. Maps the control-plane command to a
|
|
/// <see cref="AlarmCommand"/> (<c>Operation = "Acknowledge"</c>) and publishes it onto the
|
|
/// cluster <c>alarm-commands</c> topic. The broadcast lands on every driver node's
|
|
/// <c>ScriptedAlarmHostActor</c>; only the node owning the alarm acts (ownership filter), so the
|
|
/// admin singleton needs no knowledge of placement. Synchronous — the publish is fire-and-forget
|
|
/// via the mediator, so there is no awaitable work and no DB write.
|
|
/// </summary>
|
|
private void HandleAcknowledgeAlarm(AcknowledgeAlarmCommand msg)
|
|
{
|
|
var replyTo = Sender;
|
|
try
|
|
{
|
|
var alarmCmd = new AlarmCommand(
|
|
AlarmId: msg.AlarmId,
|
|
Operation: "Acknowledge",
|
|
User: msg.User,
|
|
Comment: msg.Comment,
|
|
UnshelveAtUtc: null);
|
|
|
|
Dispatch(AlarmCommandsTopic.Name, alarmCmd);
|
|
|
|
_log.Info("AdminOps: Acknowledge published for alarm {AlarmId} by {User}", msg.AlarmId, msg.User);
|
|
replyTo.Tell(new AcknowledgeAlarmResult(true, null, msg.CorrelationId));
|
|
}
|
|
catch (Exception ex)
|
|
{
|
|
_log.Error(ex, "AdminOps: Acknowledge failed for alarm {AlarmId} (correlation {CorrelationId})", msg.AlarmId, msg.CorrelationId);
|
|
replyTo.Tell(new AcknowledgeAlarmResult(false, ServerFault("Acknowledge", msg.CorrelationId), msg.CorrelationId));
|
|
}
|
|
}
|
|
|
|
/// <summary>
|
|
/// AdminUI Shelve / Unshelve path. Maps <see cref="ShelveAlarmCommand.Kind"/> to the matching
|
|
/// <see cref="AlarmCommand"/> operation (<c>OneShotShelve</c> / <c>TimedShelve</c> /
|
|
/// <c>Unshelve</c>), threading <see cref="ShelveAlarmCommand.UnshelveAtUtc"/> for the timed kind,
|
|
/// and publishes onto the cluster <c>alarm-commands</c> topic. Ownership filtering happens on the
|
|
/// owning node exactly as for Acknowledge.
|
|
/// </summary>
|
|
private void HandleShelveAlarm(ShelveAlarmCommand msg)
|
|
{
|
|
var replyTo = Sender;
|
|
try
|
|
{
|
|
var (operation, unshelveAt) = msg.Kind switch
|
|
{
|
|
ShelveKind.OneShot => ("OneShotShelve", (DateTime?)null),
|
|
ShelveKind.Timed => ("TimedShelve", msg.UnshelveAtUtc),
|
|
ShelveKind.Unshelve => ("Unshelve", (DateTime?)null),
|
|
_ => throw new ArgumentOutOfRangeException(nameof(msg), msg.Kind, "Unknown shelve kind."),
|
|
};
|
|
|
|
// TimedShelve requires an unshelve instant — the engine rejects it otherwise. Guard here so
|
|
// the AdminUI gets an immediate, attributable failure instead of a silently-dropped command.
|
|
if (msg.Kind == ShelveKind.Timed && unshelveAt is null)
|
|
{
|
|
replyTo.Tell(new ShelveAlarmResult(false, "TimedShelve requires UnshelveAtUtc.", msg.CorrelationId));
|
|
return;
|
|
}
|
|
|
|
var alarmCmd = new AlarmCommand(
|
|
AlarmId: msg.AlarmId,
|
|
Operation: operation,
|
|
User: msg.User,
|
|
Comment: msg.Comment,
|
|
UnshelveAtUtc: unshelveAt);
|
|
|
|
Dispatch(AlarmCommandsTopic.Name, alarmCmd);
|
|
|
|
_log.Info("AdminOps: {Operation} published for alarm {AlarmId} by {User}", operation, msg.AlarmId, msg.User);
|
|
replyTo.Tell(new ShelveAlarmResult(true, null, msg.CorrelationId));
|
|
}
|
|
catch (Exception ex)
|
|
{
|
|
_log.Error(ex, "AdminOps: Shelve ({Kind}) failed for alarm {AlarmId} (correlation {CorrelationId})", msg.Kind, msg.AlarmId, msg.CorrelationId);
|
|
replyTo.Tell(new ShelveAlarmResult(false, ServerFault("Shelve", msg.CorrelationId), msg.CorrelationId));
|
|
}
|
|
}
|
|
|
|
/// <summary>
|
|
/// <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 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)
|
|
{
|
|
var replyTo = Sender;
|
|
if (_composeInFlight is { } inFlightRequest)
|
|
{
|
|
replyTo.Tell(new StartDeploymentResult(
|
|
StartDeploymentOutcome.AnotherDeploymentInFlight,
|
|
DeploymentId: null,
|
|
RevisionHash: null,
|
|
Message: $"A deployment requested by {inFlightRequest.CreatedBy} is still being composed.",
|
|
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(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;
|
|
// The row is committed (the compose task completed), so the coordinator can be told now; it reads
|
|
// the row back by id when it seeds the expected-ack set.
|
|
if (msg.Dispatch is { } dispatch) _coordinator.Tell(dispatch);
|
|
msg.ReplyTo.Tell(msg.Result);
|
|
}
|
|
|
|
private void HandleDeployComposeFailed(DeployComposeFailed msg)
|
|
{
|
|
_composeInFlight = null;
|
|
_log.Error(msg.Error, "StartDeployment failed for {CreatedBy} (correlation {CorrelationId})", msg.Request.CreatedBy, msg.Request.CorrelationId);
|
|
msg.ReplyTo.Tell(new StartDeploymentResult(
|
|
StartDeploymentOutcome.Rejected,
|
|
DeploymentId: null,
|
|
RevisionHash: null,
|
|
Message: ServerFault("StartDeployment", msg.Request.CorrelationId),
|
|
msg.Request.CorrelationId));
|
|
}
|
|
|
|
/// <summary>Outcome of the off-actor compose: the reply to send, and the dispatch to issue when Accepted.</summary>
|
|
private sealed record ComposeOutcome(StartDeploymentResult Result, DispatchDeployment? Dispatch);
|
|
|
|
private sealed record DeployComposed(
|
|
StartDeployment Request, IActorRef ReplyTo, StartDeploymentResult Result, DispatchDeployment? Dispatch);
|
|
|
|
private sealed record DeployComposeFailed(StartDeployment Request, IActorRef ReplyTo, Exception Error);
|
|
|
|
/// <summary>
|
|
/// The deploy compose, off the actor thread. Touches no actor state: everything it needs is in
|
|
/// <paramref name="msg"/>, <paramref name="sourceNode"/> and the readonly fields captured at
|
|
/// construction (<see cref="_dbFactory"/>, <see cref="_tagConfigValidationMode"/>, <see cref="_log"/> —
|
|
/// Akka's logging adapter is thread-safe).
|
|
/// </summary>
|
|
private async Task<ComposeOutcome> ComposeDeploymentAsync(StartDeployment msg, string sourceNode)
|
|
{
|
|
await using var db = await _dbFactory.CreateDbContextAsync();
|
|
|
|
// Full pre-seal gate: run the complete DraftValidator (one pass, all rules) and reject
|
|
// on ANY validation error. The config is now canonical — the company overlay loader emits
|
|
// canonical EquipmentIds and the seed is clean — so every rule (UNS segments, EquipmentId
|
|
// derivation, cross-cluster/namespace binding, driver-namespace compat, signal collisions,
|
|
// …) gates the deploy. A green build in this repo does not prove the config is valid; this
|
|
// is the last guard before a bad address space (or a non-derived EquipmentId) ships.
|
|
// NOTE: a BadDuplicateExternalIdentifier rejection from the reservation pre-flight can
|
|
// (rarely) be a false positive if an external-id reservation release has not yet been
|
|
// committed/visible when the snapshot is read — operators seeing a spurious one should
|
|
// check ExternalIdReservation state before re-submitting.
|
|
var draft = await DraftSnapshotFactory.FromConfigDbAsync(db);
|
|
var errors = DraftValidator.Validate(draft).ToList();
|
|
|
|
// Cluster-topology guard. The SQL
|
|
// CK_ServerCluster_RedundancyMode_NodeCount CHECK enforces the (NodeCount, RedundancyMode)
|
|
// pair on the row itself, but it cannot see the per-node ClusterNode.Enabled flag — an
|
|
// operator can disable a node (effective enabled-count = 1) while leaving RedundancyMode at
|
|
// Hot/Warm and the constraint stays green, which would boot the runtime into an
|
|
// InvalidTopology band. ValidateClusterTopology catches that drift; it is not part of
|
|
// DraftValidator.Validate, so it runs separately here over the snapshot's cluster/node rows.
|
|
// Errors fold into the same reject summary alongside the snapshot rules so a deploy failing
|
|
// either check is rejected with both sets of messages. ClusterId-ordered for a deterministic summary.
|
|
// R03:3.2 — both come off the SAME snapshot read as every other table (one SNAPSHOT
|
|
// transaction), not a second query that could observe a later edit.
|
|
var clusters = draft.ServerClusters.OrderBy(c => c.ClusterId, StringComparer.Ordinal).ToList();
|
|
var nodesByCluster = draft.ClusterNodes
|
|
.GroupBy(n => n.ClusterId, StringComparer.Ordinal)
|
|
.ToDictionary(g => g.Key, g => g.ToList(), StringComparer.Ordinal);
|
|
foreach (var cluster in clusters)
|
|
{
|
|
var nodes = nodesByCluster.TryGetValue(cluster.ClusterId, out var ns)
|
|
? (IReadOnlyList<ClusterNode>)ns
|
|
: [];
|
|
errors.AddRange(DraftValidator.ValidateClusterTopology(cluster, nodes));
|
|
}
|
|
|
|
// Deploy-time TagConfig strictness inspection over equipment tags. The
|
|
// driver-typed EquipmentTagConfigInspector surfaces every blob the driver's own reader would
|
|
// REJECT — a present-but-invalid enum value, an absent or malformed address field, a
|
|
// structurally-unparseable TagConfig — each of which makes the driver skip the tag so it reads
|
|
// BadNodeIdUnknown forever. (The runtime used to silently default a typo'd enum;
|
|
// it now rejects the tag, so these warnings describe rejections.) In Warn mode
|
|
// (default) these are non-blocking (logged + appended to the accepted result Message); in Error
|
|
// mode they fold into the reject list so a typo'd config cannot be re-deployed until fixed.
|
|
// Running servers are untouched either way — the gate only sees re-deploys.
|
|
// v3: a raw Tag's driver is resolved via its Device (Tag no longer carries EquipmentId /
|
|
// DriverInstanceId). Inspect EVERY raw tag's TagConfig against its device's driver type.
|
|
var driverTypeById = draft.DriverInstances
|
|
.GroupBy(d => d.DriverInstanceId, StringComparer.Ordinal)
|
|
.ToDictionary(g => g.Key, g => g.First().DriverType, StringComparer.Ordinal);
|
|
var driverByDevice = draft.Devices
|
|
.GroupBy(d => d.DeviceId, StringComparer.Ordinal)
|
|
.ToDictionary(g => g.Key, g => g.First().DriverInstanceId, StringComparer.Ordinal);
|
|
var deviceConfigById = draft.Devices
|
|
.GroupBy(d => d.DeviceId, StringComparer.Ordinal)
|
|
.ToDictionary(g => g.Key, g => g.First().DeviceConfig, StringComparer.Ordinal);
|
|
var tagConfigWarnings = new List<(string TagId, string Text)>();
|
|
foreach (var t in draft.Tags)
|
|
{
|
|
if (!driverByDevice.TryGetValue(t.DeviceId, out var driverInstanceId)) continue;
|
|
if (!driverTypeById.TryGetValue(driverInstanceId, out var driverType)) continue;
|
|
foreach (var w in EquipmentTagConfigInspector.Inspect(
|
|
driverType, t.TagConfig, deviceConfigById.GetValueOrDefault(t.DeviceId)))
|
|
tagConfigWarnings.Add((t.TagId, $"tag '{t.TagId}': {w}"));
|
|
}
|
|
if (tagConfigWarnings.Count > 0)
|
|
{
|
|
if (_tagConfigValidationMode == TagConfigValidationMode.Error)
|
|
errors.AddRange(tagConfigWarnings.Select(x => new ValidationError("TagConfigInvalid", x.Text, x.TagId)));
|
|
else
|
|
foreach (var x in tagConfigWarnings)
|
|
_log.Warning("StartDeployment: TagConfig warning (Warn mode, non-blocking): {Warning}", x.Text);
|
|
}
|
|
|
|
// lmxopcua#781 item 3: always non-blocking (independent of TagConfigValidationMode) — the Tag
|
|
// entity's Writable flag and the Sql blob's "writable" disagree.
|
|
var writableWarnings = SqlTagDeployChecks.WritableMismatchWarnings(draft);
|
|
foreach (var w in writableWarnings)
|
|
_log.Warning("StartDeployment: Sql writable mismatch (non-blocking): {Warning}", w);
|
|
|
|
if (errors.Count > 0)
|
|
{
|
|
var summary = string.Join("; ", errors.Select(e => $"[{e.Code}] {e.Message}"));
|
|
_log.Warning("StartDeployment rejected ({Count} validation error(s)): {Summary}",
|
|
errors.Count, summary);
|
|
return new ComposeOutcome(
|
|
new StartDeploymentResult(
|
|
StartDeploymentOutcome.Rejected,
|
|
DeploymentId: null,
|
|
RevisionHash: null,
|
|
Message: summary,
|
|
msg.CorrelationId),
|
|
Dispatch: null);
|
|
}
|
|
|
|
// Warn-only compile-cost guardrail (NEVER blocks). Each genuinely-compiled (non-passthrough)
|
|
// script costs ~1.66 MiB RSS per node to materialize via Roslyn (measured post-A0; design doc
|
|
// 2026-06-07). Passthrough "mirror" scripts compile to ~nothing, so they are excluded; distinct
|
|
// sources are counted (the compile cache keys on source, so duplicates collapse to one unit).
|
|
// This only surfaces the estimate to the operator — the DraftValidator gate above is the hard
|
|
// reject; this advisory rides in the Accepted Message so the UI can show it.
|
|
// advisory is declared outside the inner try so the Accepted reply below can still read it even
|
|
// when the guardrail throws. A failed estimate must NEVER block a valid deploy — the caller's
|
|
// failure branch is reserved for genuine seal/save failures.
|
|
string? advisory = null;
|
|
try
|
|
{
|
|
const double PerScriptMiB = 1.66; // measured post-A0 per-script RSS (design doc 2026-06-07)
|
|
// Projected from the snapshot already in hand — Script.SourceCode used to be read three times.
|
|
var compiled = draft.Scripts
|
|
.Select(s => s.SourceCode)
|
|
.Where(src => !PassthroughScript.TryMatch(src, out _))
|
|
.Distinct(StringComparer.Ordinal)
|
|
.Count();
|
|
if (compiled > 0)
|
|
{
|
|
var estMiB = compiled * PerScriptMiB;
|
|
_log.Warning(
|
|
"StartDeployment: {Compiled} script(s) will compile (~{EstMiB:F0} MiB RSS per node); ensure node mem_limit covers it",
|
|
compiled, estMiB);
|
|
advisory = $"{compiled} script(s) will compile (~{estMiB:F0} MiB/node)";
|
|
}
|
|
}
|
|
catch (Exception ex)
|
|
{
|
|
// Guardrail is advisory-only — a failed estimate must never block a valid deploy.
|
|
_log.Warning(ex, "StartDeployment: script compile-cost estimate failed; advisory skipped");
|
|
advisory = null;
|
|
}
|
|
|
|
// Composed from the validated snapshot, never from a second read: an edit that lands between
|
|
// validation and here is simply not in this artifact (R03:3.2).
|
|
var artifact = ConfigComposer.Compose(draft);
|
|
var deploymentId = DeploymentId.NewId();
|
|
var revHash = RevisionHash.Parse(artifact.RevisionHash);
|
|
|
|
db.Deployments.Add(new Deployment
|
|
{
|
|
DeploymentId = deploymentId.Value,
|
|
RevisionHash = artifact.RevisionHash,
|
|
Status = DeploymentStatus.Dispatching,
|
|
CreatedBy = msg.CreatedBy,
|
|
ArtifactBlob = artifact.Blob,
|
|
});
|
|
|
|
// Marker ConfigEdit row so the audit timeline shows the deployment snapshot.
|
|
db.ConfigEdits.Add(new ConfigEdit
|
|
{
|
|
EntityType = "Deployment",
|
|
EntityId = deploymentId.Value,
|
|
FieldsJson = $"{{\"revisionHash\":\"{artifact.RevisionHash}\",\"sizeBytes\":{artifact.Blob.Length}}}",
|
|
EditedBy = msg.CreatedBy,
|
|
SourceNode = sourceNode,
|
|
});
|
|
|
|
await db.SaveChangesAsync();
|
|
|
|
// Accepted message = the compile-cost advisory (if any) + any Warn-mode TagConfig warnings, so
|
|
// the operator sees the non-blocking strictness findings on a successful deploy.
|
|
var messageParts = new List<string>();
|
|
if (advisory is not null) messageParts.Add(advisory);
|
|
messageParts.AddRange(tagConfigWarnings.Select(x => x.Text));
|
|
messageParts.AddRange(writableWarnings);
|
|
var acceptedMessage = messageParts.Count > 0 ? string.Join("; ", messageParts) : null;
|
|
|
|
return new ComposeOutcome(
|
|
new StartDeploymentResult(
|
|
StartDeploymentOutcome.Accepted,
|
|
deploymentId,
|
|
revHash,
|
|
Message: acceptedMessage,
|
|
msg.CorrelationId),
|
|
new DispatchDeployment(deploymentId, revHash, msg.CorrelationId));
|
|
}
|
|
|
|
/// <summary>
|
|
/// Test Connect runs on the thread pool and pipes its reply straight to the requester: a probe
|
|
/// against a black-holed PLC (60 s clamp) must not suspend the singleton's mailbox (R03:3.1). Every
|
|
/// path inside the task yields a <see cref="TestDriverConnectResult"/>, so the requester never
|
|
/// sees a <c>Status.Failure</c>.
|
|
/// </summary>
|
|
private void HandleTestDriverConnect(TestDriverConnect msg)
|
|
{
|
|
var replyTo = Sender;
|
|
if (!_probesByType.TryGetValue(msg.DriverType, out var probe))
|
|
{
|
|
replyTo.Tell(new TestDriverConnectResult(
|
|
false,
|
|
$"No probe registered for driver type '{msg.DriverType}'.",
|
|
null,
|
|
msg.CorrelationId));
|
|
return;
|
|
}
|
|
|
|
Task.Run(() => ProbeAsync(probe, msg)).PipeTo(replyTo);
|
|
}
|
|
|
|
private async Task<TestDriverConnectResult> ProbeAsync(IDriverProbe probe, TestDriverConnect msg)
|
|
{
|
|
var clampedSec = Math.Clamp(msg.TimeoutSeconds, 1, 60);
|
|
var timeout = TimeSpan.FromSeconds(clampedSec);
|
|
using var cts = new CancellationTokenSource(timeout);
|
|
|
|
var sw = System.Diagnostics.Stopwatch.StartNew();
|
|
try
|
|
{
|
|
var result = await probe.ProbeAsync(msg.ConfigJson, timeout, cts.Token);
|
|
sw.Stop();
|
|
return new TestDriverConnectResult(
|
|
result.Ok,
|
|
result.Message,
|
|
result.Ok ? sw.Elapsed.TotalMilliseconds : (double?)null,
|
|
msg.CorrelationId);
|
|
}
|
|
catch (OperationCanceledException)
|
|
{
|
|
return new TestDriverConnectResult(
|
|
false,
|
|
$"Probe timed out after {clampedSec}s.",
|
|
null,
|
|
msg.CorrelationId);
|
|
}
|
|
catch (Exception ex)
|
|
{
|
|
_log.Error(ex, "Probe for {DriverType} threw (correlation {CorrelationId})", msg.DriverType, msg.CorrelationId);
|
|
return new TestDriverConnectResult(
|
|
false,
|
|
ServerFault("Test Connect", msg.CorrelationId),
|
|
null,
|
|
msg.CorrelationId);
|
|
}
|
|
}
|
|
|
|
private async Task HandleRestartDriverAsync(RestartDriver msg)
|
|
{
|
|
var replyTo = Sender;
|
|
try
|
|
{
|
|
// Broadcast to every DriverHostActor on every node via the driver-control DPS topic.
|
|
// Only the host that owns the instance will act; others ignore it (id not found in _children).
|
|
Dispatch(DriverControlTopic.Name, msg);
|
|
|
|
await using var db = await _dbFactory.CreateDbContextAsync();
|
|
db.ConfigEdits.Add(new ConfigEdit
|
|
{
|
|
EntityType = "DriverInstance",
|
|
EntityId = Guid.TryParse(msg.DriverInstanceId, out var guid) ? guid : Guid.Empty,
|
|
FieldsJson = $"{{\"op\":\"restart\",\"driverInstanceId\":{System.Text.Json.JsonSerializer.Serialize(msg.DriverInstanceId)}}}",
|
|
EditedBy = msg.ActorByUserName,
|
|
SourceNode = Akka.Cluster.Cluster.Get(Context.System).SelfAddress.Host ?? "unknown",
|
|
});
|
|
await db.SaveChangesAsync();
|
|
|
|
_log.Info("AdminOps: RestartDriver dispatched for {DriverInstanceId} by {User}",
|
|
msg.DriverInstanceId, msg.ActorByUserName);
|
|
replyTo.Tell(new RestartDriverResult(true, null, msg.CorrelationId));
|
|
}
|
|
catch (Exception ex)
|
|
{
|
|
_log.Error(ex, "AdminOps: RestartDriver failed for {DriverInstanceId} (correlation {CorrelationId})", msg.DriverInstanceId, msg.CorrelationId);
|
|
replyTo.Tell(new RestartDriverResult(false, ServerFault("RestartDriver", msg.CorrelationId), msg.CorrelationId));
|
|
}
|
|
}
|
|
|
|
private async Task HandleReconnectDriverAsync(ReconnectDriver msg)
|
|
{
|
|
var replyTo = Sender;
|
|
try
|
|
{
|
|
// Broadcast to every DriverHostActor; only the one owning the instance reacts.
|
|
Dispatch(DriverControlTopic.Name, msg);
|
|
|
|
await using var db = await _dbFactory.CreateDbContextAsync();
|
|
db.ConfigEdits.Add(new ConfigEdit
|
|
{
|
|
EntityType = "DriverInstance",
|
|
EntityId = Guid.TryParse(msg.DriverInstanceId, out var guid) ? guid : Guid.Empty,
|
|
FieldsJson = $"{{\"op\":\"reconnect\",\"driverInstanceId\":{System.Text.Json.JsonSerializer.Serialize(msg.DriverInstanceId)}}}",
|
|
EditedBy = msg.ActorByUserName,
|
|
SourceNode = Akka.Cluster.Cluster.Get(Context.System).SelfAddress.Host ?? "unknown",
|
|
});
|
|
await db.SaveChangesAsync();
|
|
|
|
_log.Info("AdminOps: ReconnectDriver dispatched for {DriverInstanceId} by {User}",
|
|
msg.DriverInstanceId, msg.ActorByUserName);
|
|
replyTo.Tell(new ReconnectDriverResult(true, null, msg.CorrelationId));
|
|
}
|
|
catch (Exception ex)
|
|
{
|
|
_log.Error(ex, "AdminOps: ReconnectDriver failed for {DriverInstanceId} (correlation {CorrelationId})", msg.DriverInstanceId, msg.CorrelationId);
|
|
replyTo.Tell(new ReconnectDriverResult(false, ServerFault("ReconnectDriver", msg.CorrelationId), msg.CorrelationId));
|
|
}
|
|
}
|
|
}
|