Merge lane C: control plane, host, security (#733)

Claude-Session: https://claude.ai/code/session_01Ub9hGmYLQvn9XETzw7CMsT
This commit is contained in:
Joseph Doherty
2026-10-22 14:09:08 -04:00
61 changed files with 6603 additions and 453 deletions
@@ -28,6 +28,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.
@@ -36,6 +42,9 @@ public sealed class AdminOperationsActor : ReceiveActor
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>
@@ -45,15 +54,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="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) =>
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>
@@ -61,27 +72,48 @@ 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="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)
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;
ReceiveAsync<StartDeployment>(HandleStartDeploymentAsync);
ReceiveAsync<TestDriverConnect>(HandleTestDriverConnectAsync);
// 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.
@@ -129,8 +161,8 @@ public sealed class AdminOperationsActor : ReceiveActor
}
catch (Exception ex)
{
_log.Error(ex, "AdminOps: Acknowledge failed for alarm {AlarmId}", msg.AlarmId);
replyTo.Tell(new AcknowledgeAlarmResult(false, ex.Message, msg.CorrelationId));
_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));
}
}
@@ -176,215 +208,364 @@ public sealed class AdminOperationsActor : ReceiveActor
}
catch (Exception ex)
{
_log.Error(ex, "AdminOps: Shelve ({Kind}) failed for alarm {AlarmId}", msg.Kind, msg.AlarmId);
replyTo.Tell(new ShelveAlarmResult(false, ex.Message, msg.CorrelationId));
_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));
}
}
private async Task HandleStartDeploymentAsync(StartDeployment msg)
/// <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 using var db = await _dbFactory.CreateDbContextAsync();
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;
}
// 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.
var inflight = await db.Deployments
.Where(d => d.Status == DeploymentStatus.Dispatching || d.Status == DeploymentStatus.AwaitingApplyAcks)
.Select(d => d.DeploymentId)
.FirstOrDefaultAsync();
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;
}
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);
}
// 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();
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));
}
// 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, but it isn't carried on
// the generation-versioned DraftSnapshot (the cluster/node rows aren't versioned), so it must
// be run separately here against the live rows. Read-only (AsNoTracking); 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.
var clusters = await db.ServerClusters
.AsNoTracking()
.OrderBy(c => c.ClusterId)
.ToListAsync();
var nodesByCluster = (await db.ClusterNodes
.AsNoTracking()
.ToListAsync())
.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));
}
/// <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);
// 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);
}
private sealed record DeployComposed(
StartDeployment Request, IActorRef ReplyTo, StartDeploymentResult Result, DispatchDeployment? Dispatch);
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);
replyTo.Tell(new StartDeploymentResult(
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);
}
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));
return;
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;
}
// 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 query throws (e.g. transient SQL). A failed estimate must NEVER block a
// valid deploy — the outer catch 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)
var scriptSources = await db.Scripts.AsNoTracking().Select(s => s.SourceCode).ToListAsync();
var compiled = scriptSources
.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);
var artifact = await ConfigComposer.SnapshotAndFlattenAsync(db);
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,
});
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,
});
// 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 = Akka.Cluster.Cluster.Get(Context.System).SelfAddress.Host ?? "unknown",
});
await db.SaveChangesAsync();
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));
var acceptedMessage = messageParts.Count > 0 ? string.Join("; ", messageParts) : null;
_coordinator.Tell(new DispatchDeployment(deploymentId, revHash, msg.CorrelationId));
// 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));
var acceptedMessage = messageParts.Count > 0 ? string.Join("; ", messageParts) : null;
replyTo.Tell(new StartDeploymentResult(
return new ComposeOutcome(
new StartDeploymentResult(
StartDeploymentOutcome.Accepted,
deploymentId,
revHash,
Message: acceptedMessage,
msg.CorrelationId));
}
catch (Exception ex)
{
_log.Error(ex, "StartDeployment failed for {CreatedBy}", msg.CreatedBy);
replyTo.Tell(new StartDeploymentResult(
StartDeploymentOutcome.Rejected,
DeploymentId: null,
RevisionHash: null,
Message: ex.Message,
msg.CorrelationId));
}
msg.CorrelationId),
new DispatchDeployment(deploymentId, revHash, msg.CorrelationId));
}
private async Task HandleTestDriverConnectAsync(TestDriverConnect msg)
/// <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))
@@ -397,6 +578,11 @@ public sealed class AdminOperationsActor : ReceiveActor
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);
@@ -406,28 +592,28 @@ public sealed class AdminOperationsActor : ReceiveActor
{
var result = await probe.ProbeAsync(msg.ConfigJson, timeout, cts.Token);
sw.Stop();
replyTo.Tell(new TestDriverConnectResult(
return new TestDriverConnectResult(
result.Ok,
result.Message,
result.Ok ? sw.Elapsed.TotalMilliseconds : (double?)null,
msg.CorrelationId));
msg.CorrelationId);
}
catch (OperationCanceledException)
{
replyTo.Tell(new TestDriverConnectResult(
return new TestDriverConnectResult(
false,
$"Probe timed out after {clampedSec}s.",
null,
msg.CorrelationId));
msg.CorrelationId);
}
catch (Exception ex)
{
_log.Error(ex, "Probe for {DriverType} threw", msg.DriverType);
replyTo.Tell(new TestDriverConnectResult(
_log.Error(ex, "Probe for {DriverType} threw (correlation {CorrelationId})", msg.DriverType, msg.CorrelationId);
return new TestDriverConnectResult(
false,
ex.Message,
ServerFault("Test Connect", msg.CorrelationId),
null,
msg.CorrelationId));
msg.CorrelationId);
}
}
@@ -457,8 +643,8 @@ public sealed class AdminOperationsActor : ReceiveActor
}
catch (Exception ex)
{
_log.Error(ex, "AdminOps: RestartDriver failed for {DriverInstanceId}", msg.DriverInstanceId);
replyTo.Tell(new RestartDriverResult(false, ex.Message, msg.CorrelationId));
_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));
}
}
@@ -487,8 +673,8 @@ public sealed class AdminOperationsActor : ReceiveActor
}
catch (Exception ex)
{
_log.Error(ex, "AdminOps: ReconnectDriver failed for {DriverInstanceId}", msg.DriverInstanceId);
replyTo.Tell(new ReconnectDriverResult(false, ex.Message, msg.CorrelationId));
_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));
}
}
}
@@ -1,17 +1,17 @@
using System.IO.Compression;
using System.Security.Cryptography;
using System.Text.Json;
using Microsoft.EntityFrameworkCore;
using ZB.MOM.WW.OtOpcUa.Configuration;
using ZB.MOM.WW.OtOpcUa.Configuration.Entities;
using ZB.MOM.WW.OtOpcUa.Configuration.Validation;
namespace ZB.MOM.WW.OtOpcUa.ControlPlane.AdminOperations;
/// <summary>
/// Pure snapshot composer: reads the current live-edit state from <see cref="OtOpcUaConfigDbContext"/>
/// and serialises it into a deterministic, gzip-compressed byte[] artifact + SHA-256 hex revision hash.
/// Determinism comes from sorting every collection by its natural key before serialising, so two snapshots
/// over the same DB state always produce the same hash regardless of EF row ordering; compression wraps the
/// Determinism comes from reading every collection ordered by its natural key on the server
/// (<see cref="DraftSnapshotFactory"/>), so two snapshots over the same DB state always produce the same hash
/// — and the same hash as the pre-R03:3.2 composer, which used those same queries; compression wraps the
/// final bytes and reorders nothing. The hash is taken over the COMPRESSED bytes — see <see cref="Pack"/>.
/// </summary>
public static class ConfigComposer
@@ -25,45 +25,71 @@ public static class ConfigComposer
};
/// <summary>Reads the current configuration and returns a deterministic snapshot blob with revision hash.</summary>
/// <remarks>
/// One consistent read (<see cref="DraftSnapshotFactory.FromConfigDbAsync"/>, a SNAPSHOT transaction on
/// SQL Server) handed to <see cref="Compose"/>. The deploy path does not call this: it validates and
/// composes from the one <see cref="DraftSnapshot"/> it already holds, so nothing can land between the
/// validator's read and the composer's. This overload serves the drift probe and tests.
/// </remarks>
/// <param name="db">The configuration database context.</param>
/// <param name="ct">The cancellation token for the operation.</param>
/// <returns>A deterministic <see cref="ConfigArtifact"/> containing the serialised snapshot bytes and its revision hash.</returns>
public static async Task<ConfigArtifact> SnapshotAndFlattenAsync(
OtOpcUaConfigDbContext db, CancellationToken ct = default)
=> Compose(await DraftSnapshotFactory.FromConfigDbAsync(db, ct).ConfigureAwait(false));
/// <summary>
/// Serialises an already-read <see cref="DraftSnapshot"/> into the deterministic artifact. Pure over
/// its input: the same snapshot always yields the same bytes and hash, whatever the DB holds by now.
/// </summary>
/// <remarks>
/// Serialises the collections IN THE ORDER GIVEN. Determinism comes from
/// <see cref="DraftSnapshotFactory.FromConfigDbAsync"/>, which reads every table ordered by its natural key
/// on the server; a snapshot built any other way must be ordered the same way or its hash will differ.
/// </remarks>
/// <param name="snapshot">The rows to compose — the SAME rows the validator was run over.</param>
/// <returns>The compressed artifact blob and its revision hash.</returns>
public static ConfigArtifact Compose(DraftSnapshot snapshot)
{
ArgumentNullException.ThrowIfNull(snapshot);
// v3 greenfield snapshot: the Raw tree (RawFolders → DriverInstances(+RawFolderId) → Devices(+DeviceConfig)
// → TagGroups → Tags(DeviceId/TagGroupId, no EquipmentId/FolderPath/DriverInstanceId)) plus the UNS
// projection (UnsAreas/UnsLines/Equipment/UnsTagReferences) and the retained VirtualTags/ScriptedAlarms/
// PollGroups/NodeAcls/Scripts. The Namespaces snapshot is RETIRED (the two OPC UA namespaces are implicit).
// RevisionHash inputs follow the blob automatically — adding/removing a table shifts the SHA-256.
var snapshot = new
// NO re-sort here (fix round 1): the snapshot's collections arrive ordered by their natural key from
// DraftSnapshotFactory's server-side ORDER BY — the same order, and so the same bytes and RevisionHash,
// as the pre-R03:3.2 composer's own queries. An in-memory ordinal re-sort orders mixed-case keys
// differently from the SQL collation and silently re-hashed every unchanged configuration.
var ordered = new
{
Clusters = await db.ServerClusters.AsNoTracking().OrderBy(x => x.ClusterId).ToListAsync(ct),
Nodes = await db.ClusterNodes.AsNoTracking().OrderBy(x => x.NodeId).ToListAsync(ct),
RawFolders = await db.RawFolders.AsNoTracking().OrderBy(x => x.RawFolderId).ToListAsync(ct),
DriverInstances = await db.DriverInstances.AsNoTracking().OrderBy(x => x.DriverInstanceId).ToListAsync(ct),
Devices = await db.Devices.AsNoTracking().OrderBy(x => x.DeviceId).ToListAsync(ct),
TagGroups = await db.TagGroups.AsNoTracking().OrderBy(x => x.TagGroupId).ToListAsync(ct),
Equipment = await db.Equipment.AsNoTracking().OrderBy(x => x.EquipmentId).ToListAsync(ct),
Tags = await db.Tags.AsNoTracking().OrderBy(x => x.TagId).ToListAsync(ct),
UnsTagReferences = await db.UnsTagReferences.AsNoTracking().OrderBy(x => x.UnsTagReferenceId).ToListAsync(ct),
PollGroups = await db.PollGroups.AsNoTracking().OrderBy(x => x.PollGroupId).ToListAsync(ct),
UnsAreas = await db.UnsAreas.AsNoTracking().OrderBy(x => x.UnsAreaId).ToListAsync(ct),
UnsLines = await db.UnsLines.AsNoTracking().OrderBy(x => x.UnsLineId).ToListAsync(ct),
NodeAcls = await db.NodeAcls.AsNoTracking().OrderBy(x => x.NodeAclId).ToListAsync(ct),
Clusters = snapshot.ServerClusters,
Nodes = snapshot.ClusterNodes,
RawFolders = snapshot.RawFolders,
DriverInstances = snapshot.DriverInstances,
Devices = snapshot.Devices,
TagGroups = snapshot.TagGroups,
Equipment = snapshot.Equipment,
Tags = snapshot.Tags,
UnsTagReferences = snapshot.UnsTagReferences,
PollGroups = snapshot.PollGroups,
UnsAreas = snapshot.UnsAreas,
UnsLines = snapshot.UnsLines,
NodeAcls = snapshot.NodeAcls,
// the data-plane credentials travel with the config, because a driver-only node
// has no ConfigDb to read them from (per-cluster mesh Phase 4) and would otherwise deny every
// UserName token. Only the peppered HMAC of each secret ships — never the secret — and the
// pepper is node configuration that is deliberately NOT in the artifact, so a stolen blob
// still cannot be turned into a working credential. This is also why revoking a key reaches a
// driver-only node only on the next deployment; see the ApiKey entity's remarks.
ApiKeys = await db.ApiKeys.AsNoTracking().OrderBy(x => x.ApiKeyId).ToListAsync(ct),
Scripts = await db.Scripts.AsNoTracking().OrderBy(x => x.ScriptId).ToListAsync(ct),
VirtualTags = await db.VirtualTags.AsNoTracking().OrderBy(x => x.VirtualTagId).ToListAsync(ct),
ScriptedAlarms = await db.ScriptedAlarms.AsNoTracking().OrderBy(x => x.ScriptedAlarmId).ToListAsync(ct),
ApiKeys = snapshot.ApiKeys,
Scripts = snapshot.Scripts,
VirtualTags = snapshot.VirtualTags,
ScriptedAlarms = snapshot.ScriptedAlarms,
};
return Pack(JsonSerializer.SerializeToUtf8Bytes(snapshot, JsonOptions));
return Pack(JsonSerializer.SerializeToUtf8Bytes(ordered, JsonOptions));
}
/// <summary>
@@ -17,8 +17,8 @@ namespace ZB.MOM.WW.OtOpcUa.ControlPlane.Audit;
/// Dedup is two-layer: in-buffer (the <see cref="Dictionary{TKey, TValue}"/> below collapses
/// duplicate EventIds before flush) and at the database via the filtered unique index
/// <c>UX_ConfigAuditLog_EventId</c> (cross-restart safety — a retry of an already-flushed
/// batch hits the constraint and we drop the duplicate insert without losing the rest of
/// the batch).
/// batch hits the constraint; <see cref="SaveBatch"/> drops exactly the already-persisted EventIds
/// and re-saves the rest, so the duplicate is dropped without losing the rest of the batch).
///
/// Implements the shared <see cref="IAuditWriter"/> seam: <see cref="WriteAsync"/> is a
/// best-effort, never-throwing entry point that simply <c>Tell</c>s this actor and returns
@@ -92,23 +92,8 @@ public sealed class AuditWriterActor : ReceiveActor, IWithTimers, IAuditWriter
try
{
using var db = _dbFactory.CreateDbContext();
foreach (var evt in snapshot)
{
db.ConfigAuditLogs.Add(new ConfigAuditLog
{
Timestamp = evt.OccurredAtUtc.UtcDateTime,
Principal = evt.Actor,
EventType = $"{evt.Category}:{evt.Action}",
NodeId = evt.SourceNode,
DetailsJson = evt.DetailsJson,
EventId = evt.EventId,
CorrelationId = evt.CorrelationId,
Outcome = evt.Outcome.ToString(),
});
}
db.SaveChanges();
_log.Debug("AuditWriter flushed {Count} events", snapshot.Count);
var dropped = SaveBatch(snapshot);
_log.Debug("AuditWriter flushed {Count} events ({Dropped} rejected rows dropped)", snapshot.Count - dropped, dropped);
}
catch (Exception ex)
{
@@ -116,6 +101,102 @@ public sealed class AuditWriterActor : ReceiveActor, IWithTimers, IAuditWriter
}
}
private static ConfigAuditLog ToRow(AuditEvent evt) => new()
{
Timestamp = evt.OccurredAtUtc.UtcDateTime,
Principal = evt.Actor,
EventType = $"{evt.Category}:{evt.Action}",
NodeId = evt.SourceNode,
DetailsJson = evt.DetailsJson,
EventId = evt.EventId,
CorrelationId = evt.CorrelationId,
Outcome = evt.Outcome.ToString(),
};
/// <summary>
/// Persists one flush, dropping ONLY the rows the database will not take (R03:2.6, fix round 1 / I4).
/// </summary>
/// <param name="batch">The flush.</param>
/// <returns>How many rows were dropped.</returns>
/// <remarks>
/// <para>
/// <b>Why not detach <c>ex.Entries</c>.</b> SQL Server inserts the batch as one statement batch, so
/// the <see cref="DbUpdateException"/> raised for ONE duplicate <c>EventId</c> names EVERY entry in
/// the batch. Detaching what it names dropped the whole flush — the exact loss the fix was for
/// (pinned against a real SQL Server by <c>On_sql_server_a_duplicate_event_id_costs_only_that_row</c>).
/// </para>
/// <para>
/// So, on a rejection that names entries (a constraint violation; 2601/2627 for the unique index):
/// (1) ask the database which of the batch's EventIds it already holds and drop exactly those —
/// the at-least-once re-delivery the dedup contract expects; (2) re-save the rest as a batch; (3) if
/// that is still rejected (some OTHER per-row constraint), fall back to row-by-row and drop only the
/// rows that fail on their own. Every attempt uses a fresh context and fresh entities, so no state
/// from a failed save leaks into the next one. A failure that names no entry (connection loss,
/// deadlock) propagates — the whole-batch failure it always was. Bounded: at most 2 + N saves.
/// </para>
/// </remarks>
private int SaveBatch(IReadOnlyList<AuditEvent> batch)
{
if (TrySave(batch, out var first)) return 0;
_log.Debug(first, "AuditWriter: a batch of {Count} was rejected; dropping only already-persisted EventIds", batch.Count);
HashSet<Guid> existing;
using (var db = _dbFactory.CreateDbContext())
{
var ids = batch.Select(e => (Guid?)e.EventId).ToList();
existing = db.ConfigAuditLogs
.AsNoTracking()
.Where(r => ids.Contains(r.EventId))
.Select(r => r.EventId!.Value)
.ToHashSet();
}
foreach (var id in existing)
_log.Debug("AuditWriter: EventId {EventId} already persisted (unique index); dropping the duplicate", id);
var remaining = batch.Where(e => !existing.Contains(e.EventId)).ToList();
var dropped = batch.Count - remaining.Count;
if (remaining.Count == 0 || TrySave(remaining, out _)) return dropped;
// Still rejected with no duplicate left to blame: isolate the offending row(s).
foreach (var evt in remaining)
{
if (TrySave([evt], out var rowError)) continue;
dropped++;
if (IsUniqueViolation(rowError!))
_log.Debug("AuditWriter: EventId {EventId} already persisted (unique index); dropping the duplicate", evt.EventId);
else
_log.Warning(rowError, "AuditWriter: the database rejected EventId {EventId}; dropping that row and keeping the batch", evt.EventId);
}
return dropped;
}
/// <summary>Saves <paramref name="events"/> on a fresh context; a rejection that names entries is reported, anything else throws.</summary>
/// <param name="events">The rows to save.</param>
/// <param name="rejection">The row-level rejection, when this returns false.</param>
/// <returns><see langword="true"/> when saved.</returns>
private bool TrySave(IReadOnlyList<AuditEvent> events, out DbUpdateException? rejection)
{
using var db = _dbFactory.CreateDbContext();
foreach (var evt in events) db.ConfigAuditLogs.Add(ToRow(evt));
try
{
db.SaveChanges();
rejection = null;
return true;
}
catch (DbUpdateException ex) when (ex.Entries.Count > 0 || IsUniqueViolation(ex))
{
rejection = ex;
return false;
}
}
/// <summary>SQL Server 2601 (unique index) / 2627 (unique constraint), the dedup contract's expected rejection.</summary>
/// <param name="ex">The update failure.</param>
/// <returns><see langword="true"/> for a duplicate-key rejection.</returns>
private static bool IsUniqueViolation(DbUpdateException ex) =>
ex.InnerException is Microsoft.Data.SqlClient.SqlException { Number: 2601 or 2627 };
/// <inheritdoc />
protected override void PreRestart(Exception reason, object message)
{
@@ -201,6 +201,9 @@ public sealed class CentralCommunicationActor : ReceiveActor, IWithTimers
LoadContactsFromDb();
});
Receive<ContactsLoaded>(HandleContactsLoaded);
Receive<PskRefreshed>(msg => _log.Debug("Mesh PSK refresh completed for {Count} cluster(s)", msg.ClusterCount));
Receive<PskRefreshFailed>(msg => _log.Warning(msg.Error,
"Mesh PSK refresh failed; previously resolved keys stay in service until the next refresh"));
Receive<CommandDispatched>(HandleCommandDispatched);
// A faulted LoadContactsFromDb task is piped here as a Status.Failure. Without this handler
@@ -710,6 +713,20 @@ public sealed class CentralCommunicationActor : ReceiveActor, IWithTimers
_targets = msg.Targets;
// R03:1.3 — release the channels of endpoints that left the dispatch set (host renamed, port
// moved). Only the resolved dialer is asked; nothing is resolved just to prune it.
if (_dialer is { } dialer)
{
try
{
dialer.RetainOnly(_targets.Select(t => t.Endpoint).ToHashSet(StringComparer.Ordinal));
}
catch (Exception ex)
{
_log.Warning(ex, "Mesh dialer failed to release channels for retired endpoints; they stay cached until the next refresh");
}
}
if (_targets.Count == 0)
{
_log.Warning(
@@ -780,7 +797,37 @@ public sealed class CentralCommunicationActor : ReceiveActor, IWithTimers
}
pskProvider.SetKnownClusters(msg.ClusterIds);
pskProvider.InvalidateAll();
// R03:3.4 — refresh the per-cluster keys OFF the request path and swap them in atomically,
// instead of dropping the cache and making the first ack per cluster re-read the store on a
// gRPC thread. A provider that does not pre-resolve (the interface default, and every fake)
// still drops its cache here, synchronously, exactly as before.
CancellationToken ct;
try
{
ct = _lifecycle.Token;
}
catch (ObjectDisposedException)
{
return; // Stopping.
}
var self = Self;
Task refresh;
try
{
refresh = pskProvider.RefreshAsync(msg.ClusterIds, ct);
}
catch (Exception ex)
{
_log.Warning(ex, "Mesh PSK refresh threw synchronously; cached keys are unchanged until the next refresh");
refresh = Task.CompletedTask;
}
if (!refresh.IsCompletedSuccessfully)
{
refresh.PipeTo(self,
success: () => new PskRefreshed(msg.ClusterIds.Count),
failure: ex => new PskRefreshFailed(ex));
}
if (!_gatePublished)
{
@@ -795,6 +842,14 @@ public sealed class CentralCommunicationActor : ReceiveActor, IWithTimers
/// <summary>Internal tick asking for a contact-point refresh.</summary>
public sealed record RefreshContacts;
/// <summary>The asynchronous PSK refresh (R03:3.4) completed.</summary>
/// <param name="ClusterCount">How many clusters were refreshed.</param>
public sealed record PskRefreshed(int ClusterCount);
/// <summary>The asynchronous PSK refresh (R03:3.4) threw; the previous keys stay in service.</summary>
/// <param name="Error">The failure.</param>
public sealed record PskRefreshFailed(Exception Error);
/// <summary>Result of a contact-point load, piped back from the DB read.</summary>
/// <param name="Targets">
/// The gRPC dispatch set: one entry per enabled, non-maintenance row that carries a
@@ -106,6 +106,27 @@ public interface IMeshPskProvider
/// <summary>Drops every cached key.</summary>
void InvalidateAll();
/// <summary>
/// Re-reads the keys for <paramref name="clusterIds"/> from the store OFF the request path and
/// swaps them into the cache atomically (R03:3.4). Until the refresh completes, lookups keep
/// serving the previous key; a cluster whose read FAILS keeps its previous key too, while a cluster whose
/// secret the store reports ABSENT is revoked (no key from then on). Replaces
/// "invalidate on every refresh", which made the first inbound ack per cluster per
/// <c>ContactRefreshSeconds</c> block a gRPC thread on a synchronous, 3 s-bounded store read.
/// </summary>
/// <param name="clusterIds">The clusters central currently knows (the same set given to <see cref="SetKnownClusters"/>).</param>
/// <param name="ct">Cancels the refresh.</param>
/// <returns>Completes when the swap has happened.</returns>
/// <remarks>
/// The default implementation is the pre-R03:3.4 behaviour — drop the cache and let the next lookup
/// re-read — so a fake or a non-caching provider needs nothing more.
/// </remarks>
Task RefreshAsync(IReadOnlyCollection<string> clusterIds, CancellationToken ct)
{
InvalidateAll();
return Task.CompletedTask;
}
}
/// <inheritdoc />
@@ -144,6 +165,16 @@ public sealed class MeshPskProvider : IMeshPskProvider
private readonly ConcurrentDictionary<string, Lazy<string?>> _cache =
new(StringComparer.OrdinalIgnoreCase);
/// <summary>
/// Keys pre-resolved by <see cref="RefreshAsync"/>, consulted after the config map and before the
/// lazy <see cref="_cache"/>. Swapped wholesale (volatile reference to an immutable dictionary),
/// never mutated, so a lookup racing a refresh sees the whole previous snapshot or the whole next
/// one. A cluster missing from it — never refreshed, or whose read failed — falls through to the
/// lazy path exactly as before, which is what keeps "refresh failed" from meaning "no key".
/// </summary>
private volatile IReadOnlyDictionary<string, string> _refreshed =
new Dictionary<string, string>(StringComparer.OrdinalIgnoreCase);
/// <summary>
/// The known-cluster gate's membership set, or <see langword="null"/> when the gate is
/// inactive (never published, or published empty — see <see cref="SetKnownClusters"/>).
@@ -210,6 +241,13 @@ public sealed class MeshPskProvider : IMeshPskProvider
return true;
}
// Pre-resolved by the asynchronous refresh: no store read on this (gRPC request) thread at all.
if (_refreshed.TryGetValue(clusterId, out var refreshed))
{
psk = refreshed;
return true;
}
var entry = _cache.GetOrAdd(
clusterId,
static (id, self) => new Lazy<string?>(
@@ -288,7 +326,124 @@ public sealed class MeshPskProvider : IMeshPskProvider
}
/// <inheritdoc />
public void InvalidateAll() => _cache.Clear();
public void InvalidateAll()
{
_cache.Clear();
_refreshed = new Dictionary<string, string>(StringComparer.OrdinalIgnoreCase);
}
/// <summary>
/// Ceiling on concurrent store reads during one <see cref="RefreshAsync"/>. A fleet of N clusters used to
/// fan out N simultaneous reads every contact refresh; the store (and anything behind it) sees at most
/// this many at once now.
/// </summary>
public const int MaxConcurrentRefreshReads = 8;
/// <inheritdoc />
public async Task RefreshAsync(IReadOnlyCollection<string> clusterIds, CancellationToken ct)
{
ArgumentNullException.ThrowIfNull(clusterIds);
var previous = _refreshed;
var ids = clusterIds.Where(id => !string.IsNullOrWhiteSpace(id)).Distinct(StringComparer.OrdinalIgnoreCase).ToList();
var results = new ConcurrentDictionary<string, StoreRead>(StringComparer.OrdinalIgnoreCase);
// Bounded fan-out; each read bounded like the synchronous path.
await Parallel.ForEachAsync(
ids,
new ParallelOptions { MaxDegreeOfParallelism = MaxConcurrentRefreshReads, CancellationToken = ct },
async (id, token) => results[id] = await ResolveFromStoreAsync(id, token).ConfigureAwait(false))
.ConfigureAwait(false);
// Three outcomes per cluster (fix round 1 / I8):
// - Found → the new key;
// - NotFound → the store answered that the secret does not exist: a REVOCATION — the cluster gets no
// key (and its lazily cached one is evicted below), so a deleted secret stops working
// within one refresh instead of living on for the life of the process;
// - Failed → the store could not be read (threw / timed out / unnameable): keep the PREVIOUS key —
// a transient fault must not open a window in which the cluster has no key at all.
var next = new Dictionary<string, string>(StringComparer.OrdinalIgnoreCase);
var answered = new List<string>();
foreach (var id in ids)
{
var read = results[id];
switch (read.Outcome)
{
case StoreReadOutcome.Found:
next[id] = read.Value!;
answered.Add(id);
break;
case StoreReadOutcome.NotFound:
answered.Add(id);
break;
default:
if (previous.TryGetValue(id, out var kept)) next[id] = kept;
break;
}
}
// The swap. Then drop the lazy entries for every cluster the store ANSWERED for — found (serve the new
// snapshot, not a stale lazy resolution) or not found (revoked: a lazily cached key must die too).
// Entries for clusters whose read failed are left alone.
_refreshed = next;
foreach (var id in answered) _cache.TryRemove(id, out _);
}
private enum StoreReadOutcome { Found, NotFound, Failed }
private readonly record struct StoreRead(StoreReadOutcome Outcome, string? Value);
/// <summary>
/// The asynchronous twin of <see cref="ResolveFromStore"/>: same name derivation, same bound — awaited
/// instead of blocked on, because this one runs on a refresh rather than inside a gRPC interceptor —
/// but it keeps "the secret does not exist" (a blank answer) apart from "the store could not be read".
/// </summary>
/// <param name="clusterId">The application <c>Cluster</c> id.</param>
/// <param name="ct">The caller's token, linked under the store-read bound.</param>
/// <returns>The outcome and, when found, the key.</returns>
private async Task<StoreRead> ResolveFromStoreAsync(string clusterId, CancellationToken ct)
{
SecretName name;
try
{
name = new SecretName(SecretNamePrefix + clusterId);
}
catch (ArgumentException ex)
{
_logger.LogWarning(
ex,
"Cluster {ClusterId} cannot be expressed as a secret name, so its mesh PSK cannot be "
+ "refreshed from the store. Supply it through MeshTransport:ClusterPsks instead, or "
+ "rename the cluster to [a-z0-9._/-].",
clusterId);
return new StoreRead(StoreReadOutcome.Failed, null);
}
using var cts = CancellationTokenSource.CreateLinkedTokenSource(ct);
cts.CancelAfter(StoreReadTimeout);
try
{
var value = await _secretResolver.GetAsync(name, cts.Token).ConfigureAwait(false);
if (!string.IsNullOrWhiteSpace(value)) return new StoreRead(StoreReadOutcome.Found, value);
_logger.LogWarning(
"Mesh PSK secret {SecretName} for cluster {ClusterId} is absent (or blank) in the store; the key is "
+ "REVOKED — commands to that cluster and acks from it are refused until it is restored.",
name.Value,
clusterId);
return new StoreRead(StoreReadOutcome.NotFound, null);
}
catch (Exception ex)
{
_logger.LogWarning(
ex,
"Failed to refresh the mesh PSK for cluster {ClusterId} from secret {SecretName}; the "
+ "previously resolved key (if any) stays in service until the next refresh.",
clusterId,
name.Value);
return new StoreRead(StoreReadOutcome.Failed, null);
}
}
/// <summary>
/// Reads <c>OT-MESH-PSK-{clusterId}</c> from the Secrets store, converting every failure mode
@@ -34,6 +34,15 @@ public interface IMeshCommandDialer
/// <see cref="StatusCode.Unauthenticated"/> rather than dialled unauthenticated.
/// </exception>
Task DispatchAsync(string nodeEndpoint, CommandEnvelope envelope, string clusterId, CancellationToken ct);
/// <summary>
/// Releases any per-endpoint resources for endpoints NOT in <paramref name="nodeEndpoints"/>
/// (R03:1.3). Central calls it on every contact refresh with the current dispatch set, so a
/// renamed host or moved port does not leave the old channel and its idle HTTP/2 connection
/// alive for the life of the process. Default no-op for dialers that cache nothing.
/// </summary>
/// <param name="nodeEndpoints">The endpoints still in the dispatch set.</param>
void RetainOnly(IReadOnlyCollection<string> nodeEndpoints) { }
}
/// <inheritdoc />
@@ -181,6 +190,30 @@ public sealed class MeshGrpcCommandDialer : IMeshCommandDialer, IDisposable
await call.ResponseAsync.ConfigureAwait(false);
}
/// <inheritdoc />
public void RetainOnly(IReadOnlyCollection<string> nodeEndpoints)
{
ArgumentNullException.ThrowIfNull(nodeEndpoints);
if (_disposed) return;
var keep = nodeEndpoints as IReadOnlySet<string> ?? nodeEndpoints.ToHashSet(StringComparer.Ordinal);
foreach (var endpoint in _channels.Keys)
{
if (keep.Contains(endpoint)) continue;
// A dial that already took this channel out of the cache before the removal fails once, as
// a named per-node RpcException the actor logs; the next dial to that endpoint (if it ever
// returns to the set) builds a fresh channel. Bounded to the dials in flight at the instant
// of a contact refresh, the same accepted race Dispose documents.
if (_channels.TryRemove(endpoint, out var channel))
{
channel.Dispose();
_log.LogInformation(
"Mesh dialer released the cached channel for {Endpoint}: it is no longer in the dispatch set",
endpoint);
}
}
}
/// <summary>Marks the dialer disposed and disposes every cached gRPC channel.</summary>
public void Dispose()
{
@@ -161,9 +161,12 @@ public sealed class ConfigPublishCoordinator : ReceiveActor, IWithTimers
DistributedPubSub.Get(Context.System).Mediator.Tell(new Subscribe(DeploymentAcksTopic, Self));
using var db = _dbFactory.CreateDbContext();
// Projected: the row carries the whole ArtifactBlob (varbinary(max)), which this recovery never
// reads — and this query runs on every singleton start and hand-over.
var inflight = db.Deployments
.Where(d => d.Status == DeploymentStatus.Dispatching || d.Status == DeploymentStatus.AwaitingApplyAcks)
.OrderByDescending(d => d.CreatedAtUtc)
.Select(d => new { d.DeploymentId, d.RevisionHash, d.CreatedAtUtc })
.FirstOrDefault();
if (inflight is null) return;
@@ -173,6 +176,41 @@ public sealed class ConfigPublishCoordinator : ReceiveActor, IWithTimers
.AsNoTracking()
.ToList();
// 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
// proxy — we don't track a separate "dispatched at" column).
var elapsed = DateTime.UtcNow - inflight.CreatedAtUtc;
var remaining = _applyDeadline - elapsed;
if (nodeStates.Count == 0)
{
// The expected set and the AwaitingApplyAcks flip are written in ONE SaveChanges by
// HandleDispatch, so a row with no NodeDeploymentState rows is a dispatch that never completed:
// HandleDispatch threw before its SaveChanges (this actor restarted and the message was
// dropped), the process died between AdminOps' row write and the coordinator's Tell, or the
// Tell was lost across a singleton hand-over. Recovering it with an EMPTY expected set used to
// let the deadline take the "everyone acked" race branch (0 == 0) and leave the row in flight
// forever, refusing every later StartDeployment. Inside the deadline the dispatch is simply
// re-driven through the normal handler (bounded: CreatedAtUtc never moves, so a dispatch that
// keeps throwing ends up on the other branch once the deadline passes); past it the row is
// closed as TimedOut with a reason that names the condition.
if (remaining > TimeSpan.Zero)
{
_log.Warning(
"Coordinator recovered deployment {Id} with NO expected-ack rows — the dispatch was lost " +
"before any node was expected (coordinator restart or singleton hand-over mid-dispatch); " +
"re-dispatching with {Remaining} of the apply deadline left",
_current, remaining);
Self.Tell(new DispatchDeployment(
_current.Value, RevisionHash.Parse(inflight.RevisionHash), CorrelationId.NewId()));
}
else
{
Self.Tell(new DeadlineElapsed(_current.Value));
}
return;
}
_expectedAcks = nodeStates.Select(s => NodeId.Parse(s.NodeId)).ToHashSet(NodeIdComparer);
foreach (var s in nodeStates.Where(s => s.Status != NodeDeploymentStatus.Applying))
{
@@ -182,11 +220,6 @@ public sealed class ConfigPublishCoordinator : ReceiveActor, IWithTimers
if (IsDegraded(s.Status, s.FailureReason)) _degraded.Add(NodeId.Parse(s.NodeId));
}
// 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
// proxy — we don't track a separate "dispatched at" column).
var elapsed = DateTime.UtcNow - inflight.CreatedAtUtc;
var remaining = _applyDeadline - elapsed;
if (remaining <= TimeSpan.Zero)
{
Self.Tell(new DeadlineElapsed(_current.Value));
@@ -201,6 +234,33 @@ public sealed class ConfigPublishCoordinator : ReceiveActor, IWithTimers
private void HandleDispatch(DispatchDeployment msg)
{
// Terminal-state guard (fix round 1): a late or re-delivered dispatch for a row that is already
// Sealed / PartiallyFailed / TimedOut — e.g. one AdminOps closed as a lost dispatch while this message
// was still queued — must never flip it back to AwaitingApplyAcks or seed an expected set for it.
//
// The RowVersion read HERE is the one the AwaitingApplyAcks flip below is conditioned on (re-review 1):
// reading a fresh one at flip time left a window in which AdminOps' lost-dispatch closure could commit
// TimedOut between the guard and the flip, and the flip would silently reopen the row. With the guard's
// RowVersion, any change since the guard read makes the flip a concurrency conflict instead.
byte[]? guardRowVersion;
using (var check = _dbFactory.CreateDbContext())
{
var guard = check.Deployments
.AsNoTracking()
.Where(d => d.DeploymentId == msg.DeploymentId.Value)
.Select(d => new { d.Status, d.RowVersion })
.FirstOrDefault();
if (guard?.Status is DeploymentStatus.Sealed or DeploymentStatus.PartiallyFailed or DeploymentStatus.TimedOut)
{
_log.Warning(
"Ignoring DispatchDeployment {Id}: the deployment is already {Status} (a late or duplicated dispatch)",
msg.DeploymentId, guard.Status);
if (_current == msg.DeploymentId) ResetForNext();
return;
}
guardRowVersion = guard?.RowVersion;
}
_current = msg.DeploymentId;
_acks.Clear();
_degraded.Clear();
@@ -208,22 +268,51 @@ public sealed class ConfigPublishCoordinator : ReceiveActor, IWithTimers
// Seed NodeDeploymentState rows so a failover knows which nodes were expected to ack.
using (var db = _dbFactory.CreateDbContext())
{
// Discovery reuses this context rather than opening a second one — it reads the same
// ClusterNode table the NodeDeploymentState rows below are FK-bound to, so a single
// context keeps the expected set and the seeded rows consistent by construction.
_expectedAcks = DiscoverDriverNodes(db);
foreach (var node in _expectedAcks)
// Idempotent on a re-dispatch (PreStart re-drives a lost dispatch through this handler): rows
// already seeded for this deployment ARE the expected set — re-discovering would both duplicate
// the primary key and let a node toggled since the first attempt change who is expected.
var seeded = db.NodeDeploymentStates
.AsNoTracking()
.Where(x => x.DeploymentId == msg.DeploymentId.Value)
.Select(x => x.NodeId)
.ToList();
if (seeded.Count > 0)
{
db.NodeDeploymentStates.Add(new NodeDeploymentState
{
NodeId = node.Value,
DeploymentId = msg.DeploymentId.Value,
Status = NodeDeploymentStatus.Applying,
});
_expectedAcks = seeded.Select(NodeId.Parse).ToHashSet(NodeIdComparer);
}
else
{
// Discovery reuses this context rather than opening a second one — it reads the same
// ClusterNode table the NodeDeploymentState rows below are FK-bound to, so a single
// context keeps the expected set and the seeded rows consistent by construction.
_expectedAcks = DiscoverDriverNodes(db);
foreach (var node in _expectedAcks)
{
db.NodeDeploymentStates.Add(new NodeDeploymentState
{
NodeId = node.Value,
DeploymentId = msg.DeploymentId.Value,
Status = NodeDeploymentStatus.Applying,
});
}
}
UpdateDeploymentStatus(db, msg.DeploymentId, DeploymentStatus.AwaitingApplyAcks, rowVersion: guardRowVersion);
try
{
db.SaveChanges();
}
catch (DbUpdateConcurrencyException)
{
// The row moved after the guard read (on SQL Server: AdminOps closed it as a lost dispatch). The
// expected-set rows and the flip were one SaveChanges, so nothing was written; nothing is
// published and the coordinator is free for the next deployment.
_log.Warning(
"DispatchDeployment {Id} abandoned: the deployment changed state while it was being dispatched " +
"(closed as a lost dispatch?) — not reopening it", msg.DeploymentId);
ResetForNext();
return;
}
UpdateDeploymentStatus(db, msg.DeploymentId, DeploymentStatus.AwaitingApplyAcks);
db.SaveChanges();
}
// Per-cluster mesh Phase 2: hand off to the transport router rather than publishing
@@ -323,11 +412,9 @@ public sealed class ConfigPublishCoordinator : ReceiveActor, IWithTimers
// same as clean, though: the deployment's reason names the degraded nodes (per-node detail is on
// each NodeDeploymentState row) and the seal logs at Warning.
var nodes = _degraded.Select(n => n.Value).Order(StringComparer.OrdinalIgnoreCase).ToList();
var d = db.Deployments.FirstOrDefault(x => x.DeploymentId == _current.Value.Value);
if (d is not null)
d.FailureReason = Truncate(
$"Sealed with a degraded address-space apply on {nodes.Count} node(s): {string.Join(", ", nodes)}. " +
"See each node's NodeDeploymentState.FailureReason; re-deploying the same revision re-applies it.");
SetFailureReason(db, Truncate(
$"Sealed with a degraded address-space apply on {nodes.Count} node(s): {string.Join(", ", nodes)}. " +
"See each node's NodeDeploymentState.FailureReason; re-deploying the same revision re-applies it."));
db.SaveChanges();
_log.Warning(
"Deployment {Id} sealed DEGRADED (acks={Count}) — the address-space apply was degraded on: {Nodes}",
@@ -360,7 +447,9 @@ public sealed class ConfigPublishCoordinator : ReceiveActor, IWithTimers
msg.DeploymentId, _current);
return;
}
if (_acks.Count == _expectedAcks.Count)
// Only a NON-EMPTY expected set can be "fully acked": with nothing expected, 0 == 0 is not the
// everyone-acked race, it is a lost dispatch (see PreStart) that must be closed, not skipped.
if (_expectedAcks.Count > 0 && _acks.Count == _expectedAcks.Count)
{
// Race: every node acked just as the deadline fired. Already sealed/failed elsewhere.
return;
@@ -368,6 +457,19 @@ public sealed class ConfigPublishCoordinator : ReceiveActor, IWithTimers
using var db = _dbFactory.CreateDbContext();
UpdateDeploymentStatus(db, _current.Value, DeploymentStatus.TimedOut);
if (_expectedAcks.Count == 0)
{
SetFailureReason(db, Truncate(
"Dispatch was lost before any node was expected to ack (coordinator restart or singleton " +
"hand-over mid-dispatch) and the apply deadline has passed. Redeploy."));
db.SaveChanges();
_log.Warning(
"Deployment {Id} timed out with NO expected-ack rows: the dispatch was lost before any node was " +
"expected and the {Deadline} apply deadline has passed. Closed as TimedOut — redeploy",
_current.Value, _applyDeadline);
ResetForNext();
return;
}
db.SaveChanges();
// Name the silent nodes, not just the counts. Since the expected set became DB-derived this
// is the failure an operator meets after deploying while a configured node is switched off,
@@ -394,13 +496,63 @@ public sealed class ConfigPublishCoordinator : ReceiveActor, IWithTimers
public sealed record DeadlineElapsed(DeploymentId DeploymentId);
/// <summary>
/// Stages a status change WITHOUT materialising the row: the row carries the whole <c>ArtifactBlob</c>
/// (varbinary(max), multi-MB on a real fleet), which no status transition reads (fix round 1). A stub
/// carrying only the key and the row's current RowVersion is attached and the changed columns marked
/// modified, so the UPDATE writes those columns and keeps the optimistic-concurrency check. The staged
/// stub stays tracked so <see cref="SetFailureReason"/> can add to the same UPDATE.
/// </summary>
private static void UpdateDeploymentStatus(
OtOpcUaConfigDbContext db, DeploymentId id, DeploymentStatus status, bool sealNow = false)
OtOpcUaConfigDbContext db, DeploymentId id, DeploymentStatus status, bool sealNow = false, byte[]? rowVersion = null)
{
var d = db.Deployments.FirstOrDefault(x => x.DeploymentId == id.Value);
if (d is null) return;
d.Status = status;
if (sealNow) d.SealedAtUtc = DateTime.UtcNow;
var entry = TrackStub(db, id, rowVersion);
if (entry is null) return;
entry.Entity.Status = status;
entry.Property(x => x.Status).IsModified = true;
if (sealNow)
{
entry.Entity.SealedAtUtc = DateTime.UtcNow;
entry.Property(x => x.SealedAtUtc).IsModified = true;
}
}
/// <summary>Stages the current deployment's failure reason onto the stub <see cref="UpdateDeploymentStatus"/> staged.</summary>
private void SetFailureReason(OtOpcUaConfigDbContext db, string reason)
{
if (_current is null) return;
var entry = TrackStub(db, _current.Value);
if (entry is null) return;
entry.Entity.FailureReason = reason;
entry.Property(x => x.FailureReason).IsModified = true;
}
/// <summary>The tracked blob-free stub for a deployment, attaching one on first use; null when the row is gone.</summary>
/// <param name="db">The context.</param>
/// <param name="id">The deployment.</param>
/// <param name="rowVersion">
/// The RowVersion the update must be conditioned on, when the caller already read one (the dispatch
/// guard); <see langword="null"/> reads the current one.
/// </param>
private static Microsoft.EntityFrameworkCore.ChangeTracking.EntityEntry<Deployment>? TrackStub(
OtOpcUaConfigDbContext db, DeploymentId id, byte[]? rowVersion = null)
{
var tracked = db.ChangeTracker.Entries<Deployment>().FirstOrDefault(e => e.Entity.DeploymentId == id.Value);
if (tracked is not null) return tracked;
if (rowVersion is null)
{
var stamp = db.Deployments
.AsNoTracking()
.Where(x => x.DeploymentId == id.Value)
.Select(x => new { x.RowVersion })
.FirstOrDefault();
if (stamp is null) return null;
rowVersion = stamp.RowVersion;
}
return db.Deployments.Attach(new Deployment
{
DeploymentId = id.Value, RevisionHash = string.Empty, CreatedBy = string.Empty, RowVersion = rowVersion,
});
}
/// <summary>
@@ -0,0 +1,190 @@
using Akka.Actor;
using Akka.Event;
using Microsoft.EntityFrameworkCore;
using ZB.MOM.WW.OtOpcUa.Configuration;
using ZB.MOM.WW.OtOpcUa.Configuration.Enums;
namespace ZB.MOM.WW.OtOpcUa.ControlPlane.Retention;
/// <summary>
/// Admin-role cluster singleton that periodically empties the <c>ArtifactBlob</c> of old, terminal
/// deployments (R03:1.2). See <see cref="DeploymentRetentionOptions"/> for the policy and
/// <see cref="SweepAsync"/> for the rule, which is pure over a context so it is unit-testable.
/// </summary>
/// <remarks>
/// <para>
/// <b>Empties, never deletes, and never nulls.</b> The column is <c>NOT NULL varbinary(max)</c>, and
/// every reader (<c>DriverHostActor.LoadArtifactBlob</c>, <c>OpcUaPublishActor</c>,
/// <c>DeploymentArtifactService</c>) already treats a zero-length blob as "no artifact" (#485: empty
/// bytes are not an empty configuration — the apply fails and the node keeps last-known-good). An
/// empty blob therefore needs no schema change and no reader change, where a NULL would need both.
/// </para>
/// <para>
/// <b>Never touches an in-flight row</b> (Dispatching / AwaitingApplyAcks): a node may still be
/// fetching it. The newest <see cref="DeploymentRetentionOptions.KeepSealed"/> terminal rows keep
/// their blob whatever their age, so the "latest sealed" every boot path reads is always intact.
/// </para>
/// </remarks>
public sealed class DeploymentRetentionActor : ReceiveActor, IWithTimers
{
private readonly IDbContextFactory<OtOpcUaConfigDbContext> _dbFactory;
private readonly DeploymentRetentionOptions _options;
private readonly ILoggingAdapter _log = Context.GetLogger();
private const string SweepTimerKey = "retention-sweep";
/// <summary>Gets or sets the timer scheduler.</summary>
public ITimerScheduler Timers { get; set; } = null!;
/// <summary>Creates the actor props.</summary>
/// <param name="dbFactory">The config DB factory.</param>
/// <param name="options">The retention policy.</param>
/// <returns>The props.</returns>
public static Props Props(IDbContextFactory<OtOpcUaConfigDbContext> dbFactory, DeploymentRetentionOptions options) =>
Akka.Actor.Props.Create(() => new DeploymentRetentionActor(dbFactory, options));
/// <summary>Initializes a new instance of the <see cref="DeploymentRetentionActor"/> class.</summary>
/// <param name="dbFactory">The config DB factory.</param>
/// <param name="options">The retention policy.</param>
public DeploymentRetentionActor(IDbContextFactory<OtOpcUaConfigDbContext> dbFactory, DeploymentRetentionOptions options)
{
ArgumentNullException.ThrowIfNull(dbFactory);
ArgumentNullException.ThrowIfNull(options);
_dbFactory = dbFactory;
_options = options;
Receive<Sweep>(_ => RunSweep());
Receive<SweepCompleted>(msg =>
{
if (msg.Pruned > 0)
_log.Info("Deployment retention emptied the artifact blob on {Count} old deployment(s)", msg.Pruned);
else
_log.Debug("Deployment retention sweep found nothing to prune");
if (msg.ReplyTo is not null) msg.ReplyTo.Tell(msg);
});
Receive<SweepFailed>(msg =>
{
_log.Warning(msg.Error, "Deployment retention sweep failed; the next sweep retries");
if (msg.ReplyTo is not null) msg.ReplyTo.Tell(msg);
});
}
/// <inheritdoc />
protected override void PreStart()
{
if (!_options.Enabled)
{
_log.Info("Deployment retention is disabled (Deployments:Retention:Enabled=false); artifact blobs are kept forever");
return;
}
var interval = TimeSpan.FromMinutes(Math.Max(1, _options.IntervalMinutes));
var initial = TimeSpan.FromSeconds(Math.Max(0, _options.InitialDelaySeconds));
Timers.StartPeriodicTimer(SweepTimerKey, Sweep.Instance, initial, interval);
}
private void RunSweep()
{
var replyTo = Sender.IsNobody() ? null : Sender;
var self = Self;
var now = DateTime.UtcNow;
Task.Run(async () =>
{
await using var db = await _dbFactory.CreateDbContextAsync().ConfigureAwait(false);
return await SweepAsync(db, _options, now).ConfigureAwait(false);
})
.PipeTo(self,
success: pruned => new SweepCompleted(pruned, replyTo),
failure: ex => new SweepFailed(ex, replyTo));
}
/// <summary>
/// The retention rule. Empties the blob of every terminal deployment that is (a) not one of the
/// newest <see cref="DeploymentRetentionOptions.KeepSealed"/> SEALED rows (at least one — the newest
/// Sealed blob is never pruned), (b) not referenced by a
/// <c>NodeDeploymentState</c> row started or applied within
/// <see cref="DeploymentRetentionOptions.ReferencedDays"/>, and (c) still carrying bytes — at most
/// <see cref="DeploymentRetentionOptions.MaxRowsPerSweep"/> per call.
/// </summary>
/// <param name="db">The config DB.</param>
/// <param name="options">The policy.</param>
/// <param name="utcNow">The sweep's clock.</param>
/// <param name="ct">Cancellation.</param>
/// <returns>How many rows were emptied.</returns>
public static async Task<int> SweepAsync(
OtOpcUaConfigDbContext db, DeploymentRetentionOptions options, DateTime utcNow, CancellationToken ct = default)
{
ArgumentNullException.ThrowIfNull(db);
ArgumentNullException.ThrowIfNull(options);
var keep = Math.Max(1, options.KeepSealed);
var referenceFloor = utcNow.AddDays(-Math.Max(0, options.ReferencedDays));
var batch = Math.Max(1, options.MaxRowsPerSweep);
// Projected: ids only. The candidate scan must not materialise blobs for rows it will keep.
var terminal = await db.Deployments
.AsNoTracking()
.Where(d => d.Status == DeploymentStatus.Sealed
|| d.Status == DeploymentStatus.PartiallyFailed
|| d.Status == DeploymentStatus.TimedOut)
.OrderByDescending(d => d.CreatedAtUtc)
.Select(d => new { d.DeploymentId, d.Status, HasBlob = d.ArtifactBlob.Length > 0 })
.ToListAsync(ct)
.ConfigureAwait(false);
// The keep window counts SEALED rows only (fix round 1 / I3). Counting every terminal row let a burst
// of failed deploys (TimedOut / PartiallyFailed) fill the window by itself and push the fleet's last
// good artifact — the newest Sealed row, which boot, drift detection and the publish fallback all
// read — out of it. `keep` is clamped to ≥ 1, so the newest Sealed row is never a candidate.
var keptSealed = terminal
.Where(d => d.Status == DeploymentStatus.Sealed)
.Take(keep)
.Select(d => d.DeploymentId)
.ToHashSet();
var aged = terminal
.Where(d => d.HasBlob && !keptSealed.Contains(d.DeploymentId))
.Select(d => d.DeploymentId)
.ToList();
if (aged.Count == 0) return 0;
var referenced = (await db.NodeDeploymentStates
.AsNoTracking()
.Where(s => aged.Contains(s.DeploymentId)
&& (s.StartedAtUtc >= referenceFloor
|| (s.AppliedAtUtc != null && s.AppliedAtUtc >= referenceFloor)))
.Select(s => s.DeploymentId)
.Distinct()
.ToListAsync(ct)
.ConfigureAwait(false))
.ToHashSet();
var targets = aged.Where(id => !referenced.Contains(id)).Take(batch).ToList();
if (targets.Count == 0) return 0;
// Tracked load of ONLY the rows being emptied: the update rides the row's real RowVersion on
// both providers, and the blob read is the one-time cost of retiring that row.
var rows = await db.Deployments
.Where(d => targets.Contains(d.DeploymentId))
.ToListAsync(ct)
.ConfigureAwait(false);
foreach (var row in rows) row.ArtifactBlob = [];
await db.SaveChangesAsync(ct).ConfigureAwait(false);
return rows.Count;
}
/// <summary>Asks for a sweep now; the sender (if any) receives the <see cref="SweepCompleted"/> / <see cref="SweepFailed"/>.</summary>
public sealed class Sweep
{
/// <summary>The singleton instance.</summary>
public static readonly Sweep Instance = new();
private Sweep() { }
}
/// <summary>A sweep finished.</summary>
/// <param name="Pruned">How many rows were emptied.</param>
/// <param name="ReplyTo">Who asked, when the sweep was requested rather than timer-driven.</param>
public sealed record SweepCompleted(int Pruned, IActorRef? ReplyTo);
/// <summary>A sweep threw.</summary>
/// <param name="Error">The failure.</param>
/// <param name="ReplyTo">Who asked, when the sweep was requested rather than timer-driven.</param>
public sealed record SweepFailed(Exception Error, IActorRef? ReplyTo);
}
@@ -0,0 +1,55 @@
using Microsoft.Extensions.Configuration;
namespace ZB.MOM.WW.OtOpcUa.ControlPlane.Retention;
/// <summary>
/// <c>Deployments:Retention</c> — how long a sealed deployment keeps its <c>ArtifactBlob</c> (R03:1.2).
/// Every deploy stored the whole gzip'd fleet snapshot forever; nothing pruned it, so a year of routine
/// edits was gigabytes of blobs. The retention sweep EMPTIES the blob on old rows and never deletes a
/// row: the audit record (who deployed what revision, when, and how every node acked) survives.
/// </summary>
public sealed class DeploymentRetentionOptions
{
/// <summary>The configuration section name.</summary>
public const string SectionName = "Deployments:Retention";
/// <summary>Master switch; <see langword="false"/> keeps every blob forever (the previous behaviour).</summary>
public bool Enabled { get; set; } = true;
/// <summary>
/// How many of the most recent SEALED deployments keep their blob regardless of age (clamped to ≥ 1).
/// Failed deploys (PartiallyFailed / TimedOut) do not count toward it, so a burst of them can never push
/// the last good artifact out. A node booting from central SQL reads the LATEST sealed row's blob, so
/// this must stay well above 1; 50 also keeps a comfortable re-deploy history.
/// </summary>
public int KeepSealed { get; set; } = 50;
/// <summary>
/// A deployment any <c>NodeDeploymentState</c> row started or applied within this many days keeps its
/// blob even when older than the <see cref="KeepSealed"/> window — a node that just applied it may
/// still replay it on restart.
/// </summary>
public int ReferencedDays { get; set; } = 30;
/// <summary>How often the sweep runs.</summary>
public int IntervalMinutes { get; set; } = 60;
/// <summary>Delay before the first sweep after the singleton starts, so a boot is not a blob scan.</summary>
public int InitialDelaySeconds { get; set; } = 60;
/// <summary>
/// Rows emptied per sweep. The sweep loads each candidate row (blob included) to update it under its
/// RowVersion, so this bounds one sweep's memory and transaction size; a backlog drains over sweeps.
/// </summary>
public int MaxRowsPerSweep { get; set; } = 200;
/// <summary>Binds the section from configuration, with defaults for anything absent.</summary>
/// <param name="configuration">The app configuration; null yields the defaults.</param>
/// <returns>The bound options.</returns>
public static DeploymentRetentionOptions Bind(IConfiguration? configuration)
{
var options = new DeploymentRetentionOptions();
configuration?.GetSection(SectionName).Bind(options);
return options;
}
}
@@ -7,6 +7,7 @@ using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Options;
using ZB.MOM.WW.OtOpcUa.Configuration;
using ZB.MOM.WW.OtOpcUa.Cluster;
using ZB.MOM.WW.OtOpcUa.Commons.Messages.Deploy;
using ZB.MOM.WW.OtOpcUa.Commons.Messages.Mesh;
using ZB.MOM.WW.OtOpcUa.ControlPlane.AdminOperations;
using ZB.MOM.WW.OtOpcUa.ControlPlane.Audit;
@@ -15,6 +16,7 @@ using ZB.MOM.WW.OtOpcUa.ControlPlane.Coordinators;
using ZB.MOM.WW.OtOpcUa.Commons.Interfaces;
using ZB.MOM.WW.OtOpcUa.ControlPlane.Fleet;
using ZB.MOM.WW.OtOpcUa.ControlPlane.Redundancy;
using ZB.MOM.WW.OtOpcUa.ControlPlane.Retention;
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
namespace ZB.MOM.WW.OtOpcUa.ControlPlane;
@@ -30,6 +32,7 @@ public static class ServiceCollectionExtensions
public const string AuditWriterSingletonName = "audit-writer";
public const string RedundancyStateSingletonName = "redundancy-state";
public const string ClusterNodeAddressReconcilerSingletonName = "cluster-node-address-reconciler";
public const string DeploymentRetentionSingletonName = "deployment-retention";
/// <summary>
/// Registers the admin-role cluster singletons on the AkkaConfigurationBuilder — three with
@@ -109,7 +112,7 @@ public static class ServiceCollectionExtensions
// coordinator. Deliberately NOT an ActorSelection.ResolveOne here: that blocks
// inside a props factory and races the spawn it is waiting for.
var meshRouter = registry.Get<CentralCommunicationActorKey>();
return ConfigPublishCoordinator.Props(dbFactory, applyDeadline: null, meshRouter: meshRouter);
return ConfigPublishCoordinator.Props(dbFactory, applyDeadline: DeployTimings.ApplyDeadline, meshRouter: meshRouter);
},
singletonOptions);
@@ -126,7 +129,10 @@ public static class ServiceCollectionExtensions
resolver.GetService<IConfiguration>()?["Deployment:TagConfigValidationMode"],
ignoreCase: true, out var m) ? m : TagConfigValidationMode.Warn;
var meshRouter = registry.Get<CentralCommunicationActorKey>();
return AdminOperationsActor.Props(dbFactory, coordinator, probes, mode, meshRouter);
// The SAME deadline source as the coordinator above: AdminOps closes a lost dispatch only once it
// is past the coordinator's deadline, so the two must never drift.
return AdminOperationsActor.Props(dbFactory, coordinator, probes, mode, meshRouter,
applyDeadline: DeployTimings.ApplyDeadline);
},
singletonOptions);
@@ -181,6 +187,21 @@ public static class ServiceCollectionExtensions
singletonOptions,
createProxyToo: false);
// Deployment artifact retention (R03:1.2): empties ArtifactBlob on old terminal deployments so the
// Deployment table stops growing by one fleet snapshot per edit. Admin singleton for the same
// reason as the reconciler (one sweeper per fleet, and only central holds the ConfigDb);
// proxy-less because nothing addresses it in production — a test sends Sweep directly.
builder.WithSingleton<DeploymentRetentionKey>(
DeploymentRetentionSingletonName,
(system, registry, resolver) =>
{
var dbFactory = resolver.GetService<IDbContextFactory<OtOpcUaConfigDbContext>>();
var options = DeploymentRetentionOptions.Bind(resolver.GetService<IConfiguration>());
return DeploymentRetentionActor.Props(dbFactory, options);
},
singletonOptions,
createProxyToo: false);
return builder;
}
@@ -256,4 +277,5 @@ public sealed class AdminOperationsActorKey { }
public sealed class AuditWriterActorKey { }
public sealed class RedundancyStateActorKey { }
public sealed class ClusterNodeAddressReconcilerKey { }
public sealed class DeploymentRetentionKey { }
public sealed class CentralCommunicationActorKey { }
@@ -0,0 +1,41 @@
using ZB.MOM.WW.Configuration;
using ZB.MOM.WW.OtOpcUa.Core.Authorization;
namespace ZB.MOM.WW.OtOpcUa.Host.Configuration;
/// <summary>
/// Fail-fast startup validator for <see cref="ApiKeyOptions"/> (R03:2.3). The pepper used to be
/// checked only when the first key was verified — <c>ApiKeySecret.Verify</c> throws on an empty one —
/// so a site node deployed without <c>Security__ApiKeys__Pepper</c> booted green, passed health,
/// applied deployments and denied every machine client with the same opaque "Invalid API key" until
/// someone read the log; a central node with a wrong pepper minted keys that could verify nowhere.
/// Every other security section already fails host start on a bad value; this one now does too.
/// </summary>
/// <remarks>
/// Registered for <c>hasDriver || hasAdmin</c>: a driver node needs the pepper to VERIFY keys, an admin
/// node to MINT them. The minimum length is a floor against a placeholder (<c>"x"</c>) passing as
/// configured, not a strength claim — the pepper is a server-side secret, not a password.
/// </remarks>
public sealed class ApiKeyOptionsValidator : OptionsValidatorBase<ApiKeyOptions>
{
/// <summary>The shortest pepper accepted (characters, after trimming).</summary>
public const int MinPepperLength = 16;
/// <inheritdoc />
protected override void Validate(ValidationBuilder builder, ApiKeyOptions options)
{
var pepper = options.Pepper?.Trim() ?? string.Empty;
builder.RequireThat(pepper.Length > 0,
$"{ApiKeyOptions.SectionName}:Pepper is required on any node with the driver or admin role — " +
"supply it via Security__ApiKeys__Pepper or a ${secret:} reference. An empty pepper would reject " +
"every API key (driver) or mint keys that can never verify (admin).");
builder.RequireThat(pepper.Length == 0 || pepper.Length >= MinPepperLength,
$"{ApiKeyOptions.SectionName}:Pepper must be at least {MinPepperLength} characters.");
builder.RequireThat(!string.IsNullOrWhiteSpace(options.Username),
$"{ApiKeyOptions.SectionName}:Username must not be blank.");
builder.RequireThat(options.StoreTimeoutMs >= 0,
$"{ApiKeyOptions.SectionName}:StoreTimeoutMs must be >= 0 (0 disables the deadline).");
builder.RequireThat(options.StoreCacheSeconds >= 0,
$"{ApiKeyOptions.SectionName}:StoreCacheSeconds must be >= 0 (0 disables the cache).");
}
}
@@ -80,7 +80,10 @@ public static class DataPlaneAuthRegistration
sp.GetRequiredService<IOptions<AkkaClusterOptions>>().Value)),
sp.GetRequiredService<IOptions<ApiKeyOptions>>(),
sp.GetRequiredService<ILoggerFactory>().CreateLogger<BoundedApiKeyStore>(),
sp.GetRequiredService<TimeProvider>()));
sp.GetRequiredService<TimeProvider>(),
// R03:3.3 — the known-id set + miss budget in front of the SQL lookup; same section as
// ApiKeyOptions, Host-owned knobs.
ApiKeyStoreGuardOptions.Bind(sp.GetService<IConfiguration>())));
services.AddSingleton<IOpcUaUserAuthenticator, ApiKeyOpcUaUserAuthenticator>();
// Minting surface for the AdminUI. Same node condition as the store above: both need a ConfigDb.
@@ -11,6 +11,7 @@ using ZB.MOM.WW.LocalDb.Replication;
using ZB.MOM.WW.OtOpcUa.Cluster;
using ZB.MOM.WW.OtOpcUa.Configuration;
using ZB.MOM.WW.OtOpcUa.Host.Configuration;
using ZB.MOM.WW.OtOpcUa.Host.OpcUa;
namespace ZB.MOM.WW.OtOpcUa.Host.Health;
@@ -28,11 +29,40 @@ public static class HealthEndpoints
/// factory, so probing it would throw when the readiness endpoint resolves the factory. Defaults to
/// <c>true</c> so admin/fused callers (including the test harnesses) keep the probe unchanged.
/// </param>
/// <param name="hasDriver">
/// Whether this node carries the <c>driver</c> role and therefore serves an OPC UA endpoint. When
/// <c>true</c> the <see cref="OpcUaEndpointHealthCheck"/> joins the ready tier, and the active tier on a
/// driver-ONLY node (R03:2.1; on a fused node the active tier routes the AdminUI — see the body), reading
/// the <see cref="IOpcUaEndpointState"/> the host registers alongside <c>OtOpcUaServerHostedService</c>.
/// Defaults to <c>false</c> so admin-only callers and the existing harnesses register nothing new.
/// </param>
/// <returns>The same service collection, for chaining.</returns>
public static IServiceCollection AddOtOpcUaHealth(this IServiceCollection services, bool hasAdmin = true)
public static IServiceCollection AddOtOpcUaHealth(this IServiceCollection services, bool hasAdmin = true, bool hasDriver = false)
{
var checks = services.AddHealthChecks();
// A failed SDK start takes the node out of the READY tier on every driver node: it cannot serve.
// The ACTIVE tier depends on the roles (coordinator ruling, fix round 1 / I6):
// - driver-only: Active too — a site node's whole purpose is its OPC UA endpoint, so nothing may
// route to it;
// - fused admin+driver: NOT Active, until the ServiceLevel calculator consumes IOpcUaEndpointState
// (follow-up F-C1). Traefik routes the AdminUI on /health/active, so putting the probe there would
// take the operator's only console out of rotation precisely when they need it to diagnose the
// failed endpoint, while the OPC UA failover (ServiceLevel) still would not move. Once F-C1 lands
// the peer takes Primary and the active tier follows it on its own.
// Factory-registered so the state singleton is resolved per probe.
if (hasDriver)
{
var tags = hasAdmin
? new[] { ZbHealthTags.Ready }
: new[] { ZbHealthTags.Ready, ZbHealthTags.Active };
checks.Add(new HealthCheckRegistration(
OpcUaEndpointHealthCheck.Name,
sp => new OpcUaEndpointHealthCheck(sp.GetRequiredService<IOpcUaEndpointState>()),
failureStatus: null,
tags: tags));
}
// configdb probe is admin-only (per-cluster mesh Phase 4). A driver-only node gets its config
// from central via ConfigSource:Mode=FetchAndCache and never registers the ConfigDb factory, so
// the type-activated DatabaseHealthCheck would throw resolving it. Admin + fused nodes: unchanged.
@@ -0,0 +1,47 @@
using Microsoft.Extensions.Diagnostics.HealthChecks;
using ZB.MOM.WW.OtOpcUa.Host.OpcUa;
namespace ZB.MOM.WW.OtOpcUa.Host.Health;
/// <summary>
/// Ready-tier probe (plus Active on a driver-only node) for the node's OPC UA endpoint (R03:2.1): Healthy only while the SDK is
/// listening. A driver node whose SDK failed to start is otherwise indistinguishable from a healthy
/// one over HTTP — the configdb and akka probes are both green — while every OPC UA client gets
/// connection-refused.
/// </summary>
/// <remarks>
/// Registered on driver-role nodes only (<see cref="HealthEndpoints.AddOtOpcUaHealth"/>); an
/// admin-only node serves no OPC UA endpoint and must not be degraded by one it never starts. On a fused
/// admin+driver node it is Ready-only until ServiceLevel consumes the fact (F-C1), because the Active tier
/// is what Traefik routes the AdminUI on.
/// </remarks>
public sealed class OpcUaEndpointHealthCheck : IHealthCheck
{
/// <summary>The registration name, as it appears in the health JSON.</summary>
public const string Name = "opcua-endpoint";
private readonly IOpcUaEndpointState _state;
/// <summary>Initializes a new instance of the <see cref="OpcUaEndpointHealthCheck"/> class.</summary>
/// <param name="state">The endpoint lifecycle fact the hosted service maintains.</param>
public OpcUaEndpointHealthCheck(IOpcUaEndpointState state)
{
ArgumentNullException.ThrowIfNull(state);
_state = state;
}
/// <inheritdoc />
public Task<HealthCheckResult> CheckHealthAsync(HealthCheckContext context, CancellationToken cancellationToken = default)
{
var data = new Dictionary<string, object> { ["status"] = _state.Status.ToString() };
var result = _state.Status switch
{
OpcUaEndpointStatus.Started => HealthCheckResult.Healthy("OPC UA endpoint is listening", data),
OpcUaEndpointStatus.NotStarted => HealthCheckResult.Unhealthy("OPC UA endpoint has not started yet", data: data),
OpcUaEndpointStatus.Stopped => HealthCheckResult.Unhealthy("OPC UA endpoint is stopped", data: data),
_ => HealthCheckResult.Unhealthy(
"OPC UA SDK failed to start: " + (_state.FailureReason ?? "no reason recorded"), data: data),
};
return Task.FromResult(result);
}
}
@@ -0,0 +1,56 @@
using Microsoft.Extensions.Configuration;
namespace ZB.MOM.WW.OtOpcUa.Host.OpcUa;
/// <summary>
/// The pre-authentication guards <see cref="BoundedApiKeyStore"/> puts in front of the ConfigDb key
/// lookup (R03:3.3). Bound from the same <c>Security:ApiKeys</c> section as <c>ApiKeyOptions</c> but
/// owned by the Host, where the store lives.
/// </summary>
/// <remarks>
/// <para>
/// <b>Why.</b> The negative cache was keyed by key id, so a flood of DISTINCT well-formed
/// <c>otua_&lt;random&gt;_x</c> credentials missed it every time and cost one indexed SQL query per
/// attempt — under the SDK's SessionManager-wide lock — with no credential needed. Two guards close
/// that: a <b>known-id set</b> (every <c>ApiKeyId</c> the ConfigDb holds, refreshed every
/// <c>StoreCacheSeconds</c>) rejects an unknown id in memory, and a <b>miss budget</b> (token
/// bucket) bounds how many lookups that reach the store and MISS are allowed per second when the
/// set is unavailable.
/// </para>
/// <para>
/// The budget is process-wide, not per remote endpoint: the OPC UA SDK's impersonation callback
/// (the <c>IOpcUaUserAuthenticator</c> seam) does not carry the client's address. Per-endpoint
/// accounting needs that seam widened in the OpcUaServer project — a follow-up, not this lane.
/// </para>
/// </remarks>
public sealed class ApiKeyStoreGuardOptions
{
/// <summary>The configuration section the knobs live in (shared with <c>ApiKeyOptions</c>).</summary>
public const string SectionName = "Security:ApiKeys";
/// <summary>
/// Whether the known-id set is consulted. <see langword="false"/> restores a store read per unknown
/// id. Only meaningful when the inner store can list ids (the ConfigDb store can; the artifact
/// store is never wrapped).
/// </summary>
public bool KnownIdSet { get; set; } = true;
/// <summary>
/// How many store MISSES (lookups that reach the inner store and find no key) may burst before
/// further lookups for unknown ids are refused without a store read. <c>0</c> disables the budget.
/// </summary>
public int MissBurst { get; set; } = 100;
/// <summary>How many miss tokens are restored per second, up to <see cref="MissBurst"/>.</summary>
public double MissRefillPerSecond { get; set; } = 20;
/// <summary>Binds the section, with defaults for anything absent.</summary>
/// <param name="configuration">The app configuration; null yields the defaults.</param>
/// <returns>The bound options.</returns>
public static ApiKeyStoreGuardOptions Bind(IConfiguration? configuration)
{
var options = new ApiKeyStoreGuardOptions();
configuration?.GetSection(SectionName).Bind(options);
return options;
}
}
@@ -56,6 +56,33 @@ public sealed class BoundedApiKeyStore : IApiKeyStore
private readonly ConcurrentDictionary<string, Entry<ApiKeyLookup?>> _keys = new(StringComparer.Ordinal);
private readonly ConcurrentDictionary<string, Entry<IReadOnlyList<NodeAcl>>> _grants =
new(StringComparer.Ordinal);
private readonly ApiKeyStoreGuardOptions _guard;
// R03:3.3 — the known-id set: every ApiKeyId the inner store holds, refreshed every StoreCacheSeconds
// (single-flight; a refresh that fails or times out keeps the previous set). Null until the first
// successful load, or when the inner store cannot list ids — then the per-id path runs as before.
// The set and the instant it was loaded are published TOGETHER as one immutable snapshot behind a volatile
// reference (fix round 1): a reader on another thread can never pair a new set with an old stamp, or read
// a torn DateTimeOffset (a 16-byte struct, not atomically readable), as the separate fields allowed.
private volatile KnownIdSnapshot? _knownIds;
private readonly SemaphoreSlim _knownIdsRefresh = new(1, 1);
// Set when the inner store reports it cannot list ids. Per-instance on purpose: the guard options object is
// shared configuration and is never written (it used to be — `_guard.KnownIdSet = false` — which leaked
// one store's capability into every other consumer of the same options).
private volatile bool _listingUnsupported;
/// <summary>The known-id set and the instant it was loaded, published as one unit.</summary>
/// <param name="Ids">The ids the inner store holds.</param>
/// <param name="StampedUtc">When they were read.</param>
private sealed record KnownIdSnapshot(IReadOnlySet<string> Ids, DateTimeOffset StampedUtc);
// R03:3.3 — the miss budget: a token bucket on lookups that REACH the inner store and miss. Guards
// the path the known-id set cannot (set unavailable), and bounds a flood of unknown ids to the
// configured rate instead of one SQL round trip per attempt under the SessionManager lock.
private readonly object _bucketGate = new();
private double _missTokens;
private DateTimeOffset _missRefilledUtc;
/// <summary>Initializes a new instance of the <see cref="BoundedApiKeyStore"/> class.</summary>
/// <param name="inner">The store being bounded.</param>
@@ -66,11 +93,13 @@ public sealed class BoundedApiKeyStore : IApiKeyStore
/// <c>DataPlaneAuthRegistration</c> for why a second clock here would be a hazard rather than a
/// detail.
/// </param>
/// <param name="guard">The pre-authentication guards (R03:3.3); defaults when omitted.</param>
public BoundedApiKeyStore(
IApiKeyStore inner,
IOptions<ApiKeyOptions> options,
ILogger<BoundedApiKeyStore> log,
TimeProvider time)
TimeProvider time,
ApiKeyStoreGuardOptions? guard = null)
{
ArgumentNullException.ThrowIfNull(inner);
ArgumentNullException.ThrowIfNull(options);
@@ -81,8 +110,14 @@ public sealed class BoundedApiKeyStore : IApiKeyStore
_options = options;
_log = log;
_time = time;
_guard = guard ?? new ApiKeyStoreGuardOptions();
_missTokens = Math.Max(0, _guard.MissBurst);
_missRefilledUtc = time.GetUtcNow();
}
/// <summary>The known-id set currently in force, or null when none has loaded. Exposed for tests and wiring guards.</summary>
public IReadOnlySet<string>? KnownKeyIds => _knownIds?.Ids;
/// <summary>The store being bounded.</summary>
/// <remarks>
/// Exposed so the composition root's wiring can be asserted. A decorator registered in one branch
@@ -115,12 +150,103 @@ public sealed class BoundedApiKeyStore : IApiKeyStore
return cached;
}
// R03:3.3, guard 1 — an id the ConfigDb does not hold is refused in memory; the store is never asked.
var known = await KnownIdsAsync(ttl, ct).ConfigureAwait(false);
if (known is not null && !known.Contains(apiKeyId))
{
OtOpcUaTelemetry.ApiKeyStoreBounded.Add(1, new KeyValuePair<string, object?>("outcome", "unknown_id"));
return null;
}
// R03:3.3, guard 2 — a lookup that will reach the store needs a miss token unless the id is known
// to exist (a known id is a real key being verified, never a guess).
if (known is null && !TryTakeMissToken())
{
OtOpcUaTelemetry.ApiKeyStoreBounded.Add(1, new KeyValuePair<string, object?>("outcome", "miss_budget"));
throw new InvalidOperationException(
"ApiKeyStore: the unknown-key miss budget is exhausted (Security:ApiKeys:MissBurst / "
+ "MissRefillPerSecond); refusing this lookup without a store read.");
}
var fresh = await BoundAsync(token => _inner.FindAsync(apiKeyId, utcNow, token), ct)
.ConfigureAwait(false);
if (fresh is not null && known is null) RefundMissToken();
if (ttl > TimeSpan.Zero) Store(_keys, apiKeyId, fresh);
return fresh;
}
/// <summary>
/// The known-id set, refreshed when older than <paramref name="ttl"/>. Single-flight: concurrent
/// callers during a refresh use the previous set (or, before the first load, the per-id path). A
/// refresh that fails or exceeds the deadline logs and keeps the previous set.
/// </summary>
/// <param name="ttl">The refresh interval (the cache TTL); non-positive disables the set.</param>
/// <param name="ct">The caller's token.</param>
/// <returns>The set, or null.</returns>
private async Task<IReadOnlySet<string>?> KnownIdsAsync(TimeSpan ttl, CancellationToken ct)
{
if (!_guard.KnownIdSet || _listingUnsupported || ttl <= TimeSpan.Zero) return null;
var now = _time.GetUtcNow();
var current = _knownIds;
if (current is not null && now - current.StampedUtc < ttl) return current.Ids;
if (!await _knownIdsRefresh.WaitAsync(0, ct).ConfigureAwait(false)) return current?.Ids;
try
{
// Re-check under the gate: another caller may have refreshed while this one waited.
if (_knownIds is { } refreshed && now - refreshed.StampedUtc < ttl) return refreshed.Ids;
var ids = await BoundAsync(token => _inner.ListKeyIdsAsync(token), ct).ConfigureAwait(false);
if (ids is null)
{
// Unsupported by the inner store: stop asking — for this decorator only.
_listingUnsupported = true;
return null;
}
_knownIds = new KnownIdSnapshot(ids, _time.GetUtcNow());
return ids;
}
catch (Exception ex) when (ex is not OperationCanceledException || !ct.IsCancellationRequested)
{
_log.LogWarning(ex,
"ApiKeyStore: refreshing the known key-id set failed; {Mode}",
current is null ? "unknown ids reach the store until it loads" : "the previous set stays in force");
return current?.Ids;
}
finally
{
_knownIdsRefresh.Release();
}
}
/// <summary>Takes one miss token, refilling by elapsed time first; always succeeds when the budget is disabled.</summary>
/// <returns><see langword="false"/> when the budget is exhausted.</returns>
private bool TryTakeMissToken()
{
var burst = _guard.MissBurst;
if (burst <= 0) return true;
lock (_bucketGate)
{
var now = _time.GetUtcNow();
var elapsed = (now - _missRefilledUtc).TotalSeconds;
if (elapsed > 0)
{
_missTokens = Math.Min(burst, _missTokens + elapsed * Math.Max(0, _guard.MissRefillPerSecond));
_missRefilledUtc = now;
}
if (_missTokens < 1) return false;
_missTokens -= 1;
return true;
}
}
/// <summary>A lookup that reached the store and HIT was not a miss: its token goes back.</summary>
private void RefundMissToken()
{
if (_guard.MissBurst <= 0) return;
lock (_bucketGate) _missTokens = Math.Min(_guard.MissBurst, _missTokens + 1);
}
// Keyed by principal — the anonymous activation path, which runs on every unauthenticated connect.
/// <inheritdoc />
public async Task<IReadOnlyList<NodeAcl>> FindGrantsAsync(string principal, CancellationToken ct)
@@ -125,6 +125,16 @@ public sealed class ConfigDbApiKeyStore : IApiKeyStore
return await LoadGrantsAsync(db, principal, ct).ConfigureAwait(false);
}
/// <inheritdoc />
public async Task<IReadOnlySet<string>?> ListKeyIdsAsync(CancellationToken ct)
{
await using var db = await _factory.CreateDbContextAsync(ct).ConfigureAwait(false);
// One projection over the whole (small) table — the unique index on ApiKeyId makes this an index
// scan — so an unknown id can be refused in memory instead of costing a query per attempt.
var ids = await db.ApiKeys.AsNoTracking().Select(k => k.ApiKeyId).ToListAsync(ct).ConfigureAwait(false);
return ids.ToHashSet(StringComparer.Ordinal);
}
/// <summary>Loads a principal's grants, confined to this node's cluster.</summary>
/// <remarks>
/// <b>The full grant set, not just the <c>Cluster</c>-scoped rows</b>. The old
@@ -35,6 +35,16 @@ public interface IApiKeyStore
/// <param name="ct">The cancellation token.</param>
/// <returns>The matching ACL rows; empty when none.</returns>
Task<IReadOnlyList<NodeAcl>> FindGrantsAsync(string principal, CancellationToken ct);
/// <summary>
/// Every <c>ApiKeyId</c> the store holds (enabled or not), for the pre-authentication known-id set
/// (R03:3.3); <see langword="null"/> when the store cannot cheaply list them, in which case callers
/// fall back to a per-id lookup. The set only answers "does this id exist at all" — enabled, expiry
/// and the secret are still decided by <see cref="FindAsync"/>.
/// </summary>
/// <param name="ct">The cancellation token.</param>
/// <returns>The id set, or null when unsupported.</returns>
Task<IReadOnlySet<string>?> ListKeyIdsAsync(CancellationToken ct) => Task.FromResult<IReadOnlySet<string>?>(null);
}
/// <summary>An API key resolved for authentication, with the ACL grants naming its principal.</summary>
@@ -0,0 +1,73 @@
namespace ZB.MOM.WW.OtOpcUa.Host.OpcUa;
/// <summary>Lifecycle of this node's OPC UA endpoint as <see cref="OtOpcUaServerHostedService"/> drives it.</summary>
public enum OpcUaEndpointStatus
{
/// <summary>The hosted service has not run <c>StartAsync</c> yet.</summary>
NotStarted,
/// <summary>The SDK is listening and the address-space / ServiceLevel sinks are bound.</summary>
Started,
/// <summary>The SDK failed to start (port in use, PKI unwritable, bad <c>OpcUa:*</c> override, …).</summary>
Failed,
/// <summary>The hosted service stopped the SDK; the port is released.</summary>
Stopped,
}
/// <summary>
/// The one place the "did the OPC UA SDK actually start?" fact lives (R03:2.1). Before it, a failed
/// start was one Error line: <c>/health/ready</c> stayed 200, the node joined its mesh, could be
/// elected driver Primary and acked deploys as Applied through the Null sink — while clients got
/// connection-refused from the node the fleet called healthy.
/// </summary>
/// <remarks>
/// Consumed by <c>OpcUaEndpointHealthCheck</c> (Ready + Active tiers). The redundancy election and the
/// ServiceLevel calculator do not read it yet — they live outside the Host (<c>Cluster</c> /
/// <c>Runtime</c>); feeding them is the Wave 3 follow-up the lane report names. Until then a fused
/// node whose SDK failed is visible (503) but still Primary-eligible.
/// </remarks>
public interface IOpcUaEndpointState
{
/// <summary>The endpoint's current lifecycle status.</summary>
OpcUaEndpointStatus Status { get; }
/// <summary>Why the start failed, when <see cref="Status"/> is <see cref="OpcUaEndpointStatus.Failed"/>.</summary>
string? FailureReason { get; }
}
/// <summary>Mutable <see cref="IOpcUaEndpointState"/>; written by the hosted service, read by the health check.</summary>
public sealed class OpcUaEndpointState : IOpcUaEndpointState
{
private volatile OpcUaEndpointStatus _status = OpcUaEndpointStatus.NotStarted;
private volatile string? _failureReason;
/// <inheritdoc />
public OpcUaEndpointStatus Status => _status;
/// <inheritdoc />
public string? FailureReason => _failureReason;
/// <summary>Records a successful SDK start.</summary>
public void MarkStarted()
{
_failureReason = null;
_status = OpcUaEndpointStatus.Started;
}
/// <summary>Records a failed SDK start.</summary>
/// <param name="reason">The failure, as the health probe should report it.</param>
public void MarkFailed(string reason)
{
_failureReason = reason;
_status = OpcUaEndpointStatus.Failed;
}
/// <summary>Records that the hosted service stopped the SDK.</summary>
public void MarkStopped()
{
_failureReason = null;
_status = OpcUaEndpointStatus.Stopped;
}
}
@@ -5,6 +5,7 @@ using Microsoft.Extensions.Configuration;
using Microsoft.Extensions.Hosting;
using Microsoft.Extensions.Logging;
using Microsoft.Extensions.Options;
using ZB.MOM.WW.OtOpcUa.Cluster;
using ZB.MOM.WW.OtOpcUa.Commons.OpcUa;
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
using ZB.MOM.WW.OtOpcUa.Core.Authorization;
@@ -48,6 +49,8 @@ public sealed class OtOpcUaServerHostedService : IHostedService, IAsyncDisposabl
private readonly IDriverPrimaryRoleSource? _primaryRoleSource;
private readonly HistorianUpdaterHolder _historianUpdater;
private readonly IHistoryUpdateRoleView? _historyUpdateRoleView;
private readonly OpcUaEndpointState _endpointState;
private readonly bool _rethrowStartFailure;
private OpcUaApplicationHost? _appHost;
private OtOpcUaSdkServer? _server;
@@ -103,6 +106,13 @@ public sealed class OtOpcUaServerHostedService : IHostedService, IAsyncDisposabl
/// <param name="historyUpdateRoleView">The HistoryUpdate verdict <c>DriverHostActor</c> publishes (the
/// data-plane Primary verdict, closed until this node's first redundancy snapshot). Optional: registered on every
/// driver-role node; an admin-only node has none and wires no HistoryUpdate Primary gate (allow).</param>
/// <param name="endpointState">The endpoint lifecycle fact this service maintains for the
/// <c>opcua-endpoint</c> health probe (R03:2.1). Optional: a private instance is kept when none is registered,
/// so a harness that never probes health is unchanged.</param>
/// <param name="roles">The node's resolved roles. A node with the driver role and NO admin role exists only
/// to serve this endpoint, so a failed SDK start RETHROWS there (the host's fatal handler exits non-zero and
/// the supervisor restarts the process); a fused node keeps its AdminUI up and reports the failure through
/// the health probe instead. Null (harnesses) behaves as fused.</param>
public OtOpcUaServerHostedService(
IOptions<OpcUaApplicationHostOptions> options,
DeferredAddressSpaceSink deferredSink,
@@ -119,7 +129,9 @@ public sealed class OtOpcUaServerHostedService : IHostedService, IAsyncDisposabl
ILoggerFactory loggerFactory,
IDriverPrimaryRoleSource? primaryRoleSource = null,
HistorianUpdaterHolder? historianUpdater = null,
IHistoryUpdateRoleView? historyUpdateRoleView = null)
IHistoryUpdateRoleView? historyUpdateRoleView = null,
OpcUaEndpointState? endpointState = null,
EffectiveRoles? roles = null)
{
ArgumentNullException.ThrowIfNull(aclOptions);
ArgumentNullException.ThrowIfNull(time);
@@ -144,6 +156,8 @@ public sealed class OtOpcUaServerHostedService : IHostedService, IAsyncDisposabl
_primaryRoleSource = primaryRoleSource;
_historianUpdater = historianUpdater ?? HistorianUpdaterHolder.None;
_historyUpdateRoleView = historyUpdateRoleView;
_endpointState = endpointState ?? new OpcUaEndpointState();
_rethrowStartFailure = roles is { HasDriver: true, HasAdmin: false };
_logger = loggerFactory.CreateLogger<OtOpcUaServerHostedService>();
}
@@ -166,11 +180,22 @@ public sealed class OtOpcUaServerHostedService : IHostedService, IAsyncDisposabl
}
catch (Exception ex)
{
// R03:2.1 — the failure is a FACT now, not a log line: the opcua-endpoint probe turns Unhealthy
// (ready + active 503) and a driver-only node, which exists only to serve this endpoint, exits
// so its supervisor restarts it. A fused node keeps its AdminUI up so the operator can correct
// the OpcUa:* config without a process bounce — but it is visibly not ready.
_endpointState.MarkFailed(ex.Message);
await ReleaseFailedStartAsync().ConfigureAwait(false);
if (_rethrowStartFailure)
{
_logger.LogCritical(ex,
"OtOpcUaServerHostedService: SDK start failed on a driver-only node — nothing else on this " +
"node justifies staying up; failing host start so the supervisor restarts the process");
throw;
}
_logger.LogError(ex,
"OtOpcUaServerHostedService: SDK start failed; OpcUaPublishActor writes will continue to no-op");
// Don't rethrow — the rest of the host (admin UI, driver actors, etc.) can still boot.
// Operators see the failure via the logs + can correct config without a process bounce
// of the whole binary.
"OtOpcUaServerHostedService: SDK start failed; /health/ready reports Unhealthy and " +
"OpcUaPublishActor writes will continue to no-op until the host is restarted with a corrected OpcUa section");
return;
}
@@ -178,6 +203,7 @@ public sealed class OtOpcUaServerHostedService : IHostedService, IAsyncDisposabl
{
_logger.LogWarning(
"OtOpcUaServerHostedService: SDK reported started but NodeManager is null; sink stays Null");
_endpointState.MarkFailed("SDK reported started but the node manager is null");
return;
}
@@ -352,22 +378,16 @@ public sealed class OtOpcUaServerHostedService : IHostedService, IAsyncDisposabl
{
_logger.LogError(ex,
"OtOpcUaServerHostedService: ServiceLevel binding failed; stopping the OPC UA endpoint while the admin host continues");
_endpointState.MarkFailed("ServiceLevel binding failed: " + ex.Message);
try
{
await StopAsync(cancellationToken).ConfigureAwait(false);
DetachSinks();
}
catch (Exception cleanupError)
{
_logger.LogError(cleanupError, "OtOpcUaServerHostedService: failed to detach OPC UA bindings after startup failure");
}
var released = await FailedServiceLevelBindingCleanup.StopAndDisposeAsync(
async () => await _server.StopAsync(CancellationToken.None).ConfigureAwait(false),
() => _appHost.DisposeAsync(), _logger).ConfigureAwait(false);
if (released)
{
_appHost = null;
_server = null;
}
await ReleaseEndpointAsync().ConfigureAwait(false);
return;
}
@@ -375,9 +395,52 @@ public sealed class OtOpcUaServerHostedService : IHostedService, IAsyncDisposabl
// attached, so the first tick cannot land on a half-wired server.
StartRevalidationTimer();
_endpointState.MarkStarted();
_logger.LogInformation("OtOpcUaServerHostedService: SDK started, address-space + ServiceLevel sinks bound");
}
/// <summary>
/// Releases an SDK whose <c>StartAsync</c> threw. The application instance may hold a half-opened
/// listener (the port-in-use case binds nothing, but a certificate or PKI failure can fire after the
/// transport is up); disposing it is the only way to be sure. The state stays Failed.
/// </summary>
private async Task ReleaseFailedStartAsync()
{
var appHost = _appHost;
_appHost = null;
_server = null;
if (appHost is null) return;
try
{
await appHost.DisposeAsync().ConfigureAwait(false);
}
catch (Exception ex)
{
_logger.LogWarning(ex, "OtOpcUaServerHostedService: disposing the application host after a failed start threw");
}
}
/// <summary>
/// Stops the SDK and disposes the application host, releasing endpoint ownership only when both
/// succeeded (<see cref="FailedServiceLevelBindingCleanup.StopAndDisposeAsync"/>): on a failure the
/// fields are retained so <see cref="DisposeAsync"/> retries at host disposal.
/// </summary>
private async Task ReleaseEndpointAsync()
{
var server = _server;
var appHost = _appHost;
if (server is null && appHost is null) return;
var released = await FailedServiceLevelBindingCleanup.StopAndDisposeAsync(
async () => { if (server is not null) await server.StopAsync(CancellationToken.None).ConfigureAwait(false); },
() => appHost?.DisposeAsync() ?? ValueTask.CompletedTask,
_logger).ConfigureAwait(false);
if (released)
{
_appHost = null;
_server = null;
}
}
/// <summary>
/// The <c>ServerHistorian:HonorReturnBounds</c> kill switch <c>StartAsync</c> applies to the node manager
/// (default <c>true</c> when the section or key is absent). Exposed so the wiring is testable without booting the SDK.
@@ -441,11 +504,30 @@ public sealed class OtOpcUaServerHostedService : IHostedService, IAsyncDisposabl
}
/// <summary>
/// Stops the OPC UA server asynchronously.
/// Stops the OPC UA server asynchronously: detaches every sink, then STOPS the SDK so the port closes.
/// </summary>
/// <remarks>
/// R03:1.4 — the SDK used to stay listening from here until host disposal (Akka coordinated shutdown,
/// the other hosted services' StopAsync), serving the last-written values with an emptied authority:
/// reads succeeded on stale data while writes got "unavailable", and clients saw a server that was
/// up but inert instead of the connection-refused that triggers their failover. <see cref="DisposeAsync"/>
/// stays idempotent with this: ownership is nulled once the stop succeeded.
/// </remarks>
/// <param name="cancellationToken">Cancellation token.</param>
/// <returns>A task that represents the asynchronous operation.</returns>
public Task StopAsync(CancellationToken cancellationToken)
public async Task StopAsync(CancellationToken cancellationToken)
{
DetachSinks();
await ReleaseEndpointAsync().ConfigureAwait(false);
if (_endpointState.Status != OpcUaEndpointStatus.Failed) _endpointState.MarkStopped();
}
/// <summary>
/// Reverts every seam on the node manager to its Null adapter so in-flight work from a stopping actor
/// or a late client call cannot hit a half-disposed server. Pure detach — the SDK keeps running
/// until <see cref="ReleaseEndpointAsync"/>.
/// </summary>
private void DetachSinks()
{
// FIRST, ahead of every detach below (arch-review 03/S19). A tick landing on a node manager whose
// authority and gate have already been cleared is the shape this whole teardown block exists to
@@ -472,7 +554,6 @@ public sealed class OtOpcUaServerHostedService : IHostedService, IAsyncDisposabl
// request rather than let the seams fall back to their role checks — a stopping node should shed
// enforcement in the same order it acquired it.
_server?.SetDataPlaneAuthorizer(null);
return Task.CompletedTask;
}
/// <summary>
@@ -482,10 +563,15 @@ public sealed class OtOpcUaServerHostedService : IHostedService, IAsyncDisposabl
public async ValueTask DisposeAsync()
{
// Before the app host, and idempotent with StopAsync — a host that disposes without stopping must
// not leave a tick armed against a server that is about to go away.
// not leave a tick armed against a server that is about to go away. A StopAsync that released the
// endpoint left both fields null, so this is a no-op after it; a stop that FAILED to release left
// them set, and this is the retry.
_revalidationTimer?.Dispose();
_revalidationTimer = null;
if (_appHost is not null) await _appHost.DisposeAsync();
var appHost = _appHost;
_appHost = null;
_server = null;
if (appHost is not null) await appHost.DisposeAsync();
}
}
+17 -6
View File
@@ -162,10 +162,15 @@ builder.AddZbSerilog(o => o.ServiceName = "otopcua");
// central over the fetch-and-cache path (ConfigSource:Mode=FetchAndCache). Registering it
// unconditionally would throw on a driver-only node, since AddOtOpcUaConfigDb requires the string.
// A fused admin+driver node keeps ConfigDb via its admin role — byte-for-byte unaffected.
// bound unconditionally: a driver node needs the pepper to VERIFY keys, an admin node needs it
// to MINT them, and getting this wrong on either side is a silent authentication failure rather than a
// startup error.
builder.Services.Configure<ApiKeyOptions>(builder.Configuration.GetSection(ApiKeyOptions.SectionName));
// A driver node needs the pepper to VERIFY keys, an admin node needs it to MINT them, and getting this
// wrong on either side used to be a silent authentication failure rather than a startup error (R03:2.3):
// ApiKeyOptionsValidator now fails host start on a blank or placeholder pepper wherever either role is
// present. A node with neither role (dev-only) has no key path and keeps the plain bind.
if (hasDriver || hasAdmin)
builder.Services.AddValidatedOptions<ApiKeyOptions, ApiKeyOptionsValidator>(
builder.Configuration, ApiKeyOptions.SectionName);
else
builder.Services.Configure<ApiKeyOptions>(builder.Configuration.GetSection(ApiKeyOptions.SectionName));
// Security:Acl:Enabled. Default false; see AclOptions for why the flip must be a config
// change rather than a redeploy, and what it swaps rather than adds.
builder.Services.Configure<AclOptions>(builder.Configuration.GetSection(AclOptions.SectionName));
@@ -452,6 +457,11 @@ if (hasDriver)
// mediator (for the inbound alarm-command router) without racing Akka startup — same pattern the
// DpsScriptLogPublisher above uses. TryAdd so a fused admin+driver node registers it exactly once.
builder.Services.TryAddSingleton<Func<ActorSystem>>(sp => () => sp.GetRequiredService<ActorSystem>());
// The "did the SDK start?" fact (R03:2.1): written by the hosted service, read by the opcua-endpoint
// health probe (AddOtOpcUaHealth below, hasDriver). One instance behind both the class and the
// interface, so the writer and the probe cannot disagree.
builder.Services.AddSingleton<OpcUaEndpointState>();
builder.Services.AddSingleton<IOpcUaEndpointState>(sp => sp.GetRequiredService<OpcUaEndpointState>());
builder.Services.AddHostedService<OtOpcUaServerHostedService>();
}
@@ -554,8 +564,9 @@ if (hasAdmin)
builder.Services.AddOtOpcUaSecrets(builder.Configuration);
// hasAdmin gates the configdb probe: a driver-only node registers no ConfigDb factory (see above), so
// probing it would throw when the readiness endpoint resolves the factory.
builder.Services.AddOtOpcUaHealth(hasAdmin);
// probing it would throw when the readiness endpoint resolves the factory. hasDriver adds the
// opcua-endpoint probe (ready + active): a driver node whose SDK failed to start is Unhealthy, not green.
builder.Services.AddOtOpcUaHealth(hasAdmin, hasDriver);
builder.Services.AddOtOpcUaObservability(builder.Configuration);
// gRPC server plumbing shared by three endpoints: the LocalDb passive sync endpoint (driver-role,
@@ -0,0 +1,95 @@
using System.Security.Claims;
using Microsoft.AspNetCore.Components.Authorization;
using Microsoft.AspNetCore.Http;
using ZB.MOM.WW.Auth.AspNetCore;
namespace ZB.MOM.WW.OtOpcUa.Security.Audit;
/// <summary>
/// <see cref="IAuditActorAccessor"/> that reads the Blazor circuit's principal first (R03:2.5).
/// </summary>
/// <remarks>
/// <para>
/// <see cref="HttpAuditActorAccessor"/> reads <c>IHttpContextAccessor</c>, which is right for a
/// minimal-API request but wrong inside a Blazor Server circuit: after the first render there is no
/// current <c>HttpContext</c> on the circuit's threads, so every audit row written from an interactive
/// page would have been stamped <c>"system"</c>. The circuit's identity lives in the scoped
/// <see cref="AuthenticationStateProvider"/>, so that is consulted first and the HTTP context is the
/// fallback for non-circuit scopes (where the provider is not registered at all).
/// </para>
/// <para>
/// <c>CurrentActor</c> is synchronous by contract; the server-side provider's
/// <see cref="AuthenticationStateProvider.GetAuthenticationStateAsync"/> is already completed in a
/// circuit, and an incomplete task is treated as "no circuit identity" rather than blocked on.
/// </para>
/// </remarks>
public sealed class AuthenticationStateAuditActorAccessor : IAuditActorAccessor
{
private readonly IHttpContextAccessor _httpContextAccessor;
private readonly AuthenticationStateProvider? _authenticationState;
/// <summary>Initializes a new instance of the <see cref="AuthenticationStateAuditActorAccessor"/> class.</summary>
/// <param name="httpContextAccessor">The HTTP context accessor (minimal-API fallback).</param>
/// <param name="authenticationState">The circuit's authentication state, when resolved inside a Blazor circuit.</param>
public AuthenticationStateAuditActorAccessor(
IHttpContextAccessor httpContextAccessor,
AuthenticationStateProvider? authenticationState = null)
{
ArgumentNullException.ThrowIfNull(httpContextAccessor);
_httpContextAccessor = httpContextAccessor;
_authenticationState = authenticationState;
}
/// <inheritdoc />
public string? CurrentActor
{
get
{
if (_authenticationState is not null)
{
var actor = TryResolveCircuitActor(_authenticationState);
if (actor is not null) return actor;
}
return ResolveActor(_httpContextAccessor.HttpContext?.User);
}
}
/// <summary>
/// The circuit's actor, or null when this scope has no circuit identity. The server-side provider
/// is registered for EVERY scope on an AdminUI node, but only a circuit calls
/// <c>SetAuthenticationState</c>; anywhere else (a minimal-API request, a hosted service's scope)
/// <c>GetAuthenticationStateAsync</c> throws <see cref="InvalidOperationException"/> synchronously
/// (fix round 1 / I5). That is "no circuit", never an audit failure — the caller falls back to
/// the HTTP context.
/// </summary>
/// <param name="provider">The scoped provider.</param>
/// <returns>The actor, or null.</returns>
private static string? TryResolveCircuitActor(AuthenticationStateProvider provider)
{
Task<AuthenticationState> task;
try
{
task = provider.GetAuthenticationStateAsync();
}
catch (InvalidOperationException)
{
return null;
}
return task.IsCompletedSuccessfully ? ResolveActor(task.Result.User) : null;
}
/// <summary>The claim-resolution rule shared with <see cref="HttpAuditActorAccessor"/>.</summary>
/// <param name="user">The principal, or null.</param>
/// <returns>The actor string, or null when unauthenticated.</returns>
internal static string? ResolveActor(ClaimsPrincipal? user)
{
if (user?.Identity?.IsAuthenticated != true)
return null;
// Prefer the canonical login-name claim; fall back to the Name claim or
// Identity.Name (both of which map to ClaimTypes.Name / ZbClaimTypes.Name).
return user.FindFirst(ZbClaimTypes.Username)?.Value
?? user.FindFirst(ZbClaimTypes.Name)?.Value
?? user.Identity.Name;
}
}
@@ -1,5 +1,4 @@
using Microsoft.AspNetCore.Http;
using ZB.MOM.WW.Auth.AspNetCore;
namespace ZB.MOM.WW.OtOpcUa.Security.Audit;
@@ -17,10 +16,10 @@ namespace ZB.MOM.WW.OtOpcUa.Security.Audit;
/// <see cref="System.Security.Principal.IIdentity.Name"/>, in that order.</item>
/// </list>
/// <para>
/// Registered as <b>scoped</b> in <see cref="ZB.MOM.WW.OtOpcUa.Security.ServiceCollectionExtensions.AddOtOpcUaAuth"/>
/// so that it correctly follows the request scope used by Blazor Server interactive components
/// and minimal-API endpoints. <c>IHttpContextAccessor</c> is registered by
/// <c>AddOtOpcUaAuth</c> via <c>services.AddHttpContextAccessor()</c>.
/// No longer the registered implementation: <see cref="AuthenticationStateAuditActorAccessor"/> is, because
/// a Blazor circuit has no <c>HttpContext</c> after its first render and this accessor would stamp every row
/// written from an interactive page <c>"system"</c>. Kept as the explicit HTTP-only variant for non-circuit
/// hosts and tests.
/// </para>
/// </remarks>
public sealed class HttpAuditActorAccessor : IAuditActorAccessor
@@ -35,19 +34,6 @@ public sealed class HttpAuditActorAccessor : IAuditActorAccessor
}
/// <inheritdoc />
public string? CurrentActor
{
get
{
var user = _httpContextAccessor.HttpContext?.User;
if (user?.Identity?.IsAuthenticated != true)
return null;
// Prefer the canonical login-name claim; fall back to the Name claim or
// Identity.Name (both of which map to ClaimTypes.Name / ZbClaimTypes.Name).
return user.FindFirst(ZbClaimTypes.Username)?.Value
?? user.FindFirst(ZbClaimTypes.Name)?.Value
?? user.Identity.Name;
}
}
public string? CurrentActor =>
AuthenticationStateAuditActorAccessor.ResolveActor(_httpContextAccessor.HttpContext?.User);
}
@@ -0,0 +1,39 @@
using Microsoft.AspNetCore.Http;
using Microsoft.Extensions.Primitives;
using ZB.MOM.WW.OtOpcUa.Security.Jwt;
namespace ZB.MOM.WW.OtOpcUa.Security.Auth;
/// <summary>The authentication scheme names OtOpcUa registers.</summary>
public static class OtOpcUaAuthSchemes
{
/// <summary>
/// The bearer scheme that validates the HS256 tokens <c>/auth/token</c> mints: the first-party
/// <c>AddJwtBearer</c> handler over <see cref="JwtTokenService.BuildValidationParameters"/>.
/// Registered only when <see cref="JwtOptions.IsConfigured"/>.
/// </summary>
public const string Bearer = "Bearer";
/// <summary>
/// The default scheme: a policy scheme that forwards to <see cref="Bearer"/> when the request
/// carries an <c>Authorization: Bearer</c> header and to the cookie scheme otherwise, so one
/// <c>RequireAuthorization()</c> / policy accepts a browser session AND a machine token.
/// </summary>
public const string CookieOrBearer = "OtOpcUa.CookieOrBearer";
/// <summary>Whether the request presents a bearer token (the selector's whole rule).</summary>
/// <param name="context">The request.</param>
/// <returns><see langword="true"/> when an <c>Authorization: Bearer …</c> header is present.</returns>
public static bool HasBearerToken(HttpContext context) =>
TryGetBearerToken(context.Request.Headers.Authorization, out _);
internal static bool TryGetBearerToken(StringValues authorization, out string token)
{
token = string.Empty;
var header = authorization.ToString();
const string prefix = "Bearer ";
if (!header.StartsWith(prefix, StringComparison.OrdinalIgnoreCase)) return false;
token = header[prefix.Length..].Trim();
return token.Length > 0;
}
}
@@ -1,79 +0,0 @@
using System.Net.Http.Json;
using System.Security.Claims;
using Microsoft.AspNetCore.Components.Authorization;
using Microsoft.Extensions.Logging;
namespace ZB.MOM.WW.OtOpcUa.Security.Blazor;
/// <summary>
/// Blazor Server <see cref="AuthenticationStateProvider"/> that snapshots the cookie-backed
/// principal supplied at circuit boot and polls <c>/auth/ping</c> every 60 seconds to detect
/// expiry. Mirrors ScadaLink's CentralUI implementation.
/// </summary>
public sealed class CookieAuthenticationStateProvider : AuthenticationStateProvider, IAsyncDisposable
{
private static readonly TimeSpan PingInterval = TimeSpan.FromSeconds(60);
private readonly HttpClient _http;
private readonly ILogger<CookieAuthenticationStateProvider> _logger;
private readonly CancellationTokenSource _cts = new();
private ClaimsPrincipal _current;
private Task? _pingLoop;
/// <summary>Initializes a new instance of the CookieAuthenticationStateProvider class.</summary>
/// <param name="initial">The initial claims principal from circuit boot.</param>
/// <param name="http">The HTTP client for authentication ping requests.</param>
/// <param name="logger">The logger for diagnostic messages.</param>
public CookieAuthenticationStateProvider(
ClaimsPrincipal initial,
HttpClient http,
ILogger<CookieAuthenticationStateProvider> logger)
{
_current = initial;
_http = http;
_logger = logger;
}
/// <inheritdoc />
public override Task<AuthenticationState> GetAuthenticationStateAsync()
{
_pingLoop ??= Task.Run(() => PingLoopAsync(_cts.Token));
return Task.FromResult(new AuthenticationState(_current));
}
private async Task PingLoopAsync(CancellationToken ct)
{
try
{
while (!ct.IsCancellationRequested)
{
await Task.Delay(PingInterval, ct).ConfigureAwait(false);
var resp = await _http.GetAsync("/auth/ping", ct).ConfigureAwait(false);
if (!resp.IsSuccessStatusCode && _current.Identity?.IsAuthenticated == true)
{
_logger.LogInformation("/auth/ping returned {Code}; notifying circuit", (int)resp.StatusCode);
_current = new ClaimsPrincipal(new ClaimsIdentity());
NotifyAuthenticationStateChanged(
Task.FromResult(new AuthenticationState(_current)));
}
}
}
catch (OperationCanceledException) { /* expected on shutdown */ }
catch (Exception ex)
{
_logger.LogWarning(ex, "Auth ping loop terminated unexpectedly");
}
}
/// <summary>Cancels the background ping loop, awaits its completion, and disposes the cancellation token source.</summary>
/// <returns>A task that represents the asynchronous operation.</returns>
public async ValueTask DisposeAsync()
{
_cts.Cancel();
if (_pingLoop is not null)
{
try { await _pingLoop.ConfigureAwait(false); } catch { /* swallow shutdown errors */ }
}
_cts.Dispose();
}
}
@@ -8,6 +8,7 @@ using Microsoft.AspNetCore.Http;
using Microsoft.AspNetCore.Routing;
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Logging;
using Microsoft.Extensions.Options;
using ZB.MOM.WW.Auth.AspNetCore;
using ZB.MOM.WW.Auth.Abstractions.Roles;
using ZB.MOM.WW.OtOpcUa.Security.Jwt;
@@ -23,6 +24,17 @@ public static class AuthEndpoints
/// <summary>Response for a token issue request.</summary>
public sealed record TokenResponse(string Token);
/// <summary>
/// The <c>/auth/token</c> policy: an authenticated principal from the COOKIE scheme. Naming the scheme
/// makes the authorization middleware authenticate that scheme alone, so an <c>Authorization: Bearer</c>
/// header is never consulted here (with login disabled the auto-login handler sits under the same
/// scheme name, so the dev rig is unchanged).
/// </summary>
internal static readonly AuthorizationPolicy CookieSessionOnly =
new AuthorizationPolicyBuilder(CookieAuthenticationDefaults.AuthenticationScheme)
.RequireAuthenticatedUser()
.Build();
/// <summary>Maps OtOpcUa authentication endpoints to the application route builder.</summary>
/// <param name="app">The endpoint route builder.</param>
/// <returns>The endpoint route builder for chaining.</returns>
@@ -36,7 +48,17 @@ public static class AuthEndpoints
// AddOtOpcUaAuth fallback policy that would otherwise 401 the request.
app.MapPost("/auth/login", (Delegate)LoginAsync).AllowAnonymous().DisableAntiforgery();
app.MapGet("/auth/ping", (Delegate)Ping).AllowAnonymous();
app.MapPost("/auth/token", (Delegate)IssueToken).RequireAuthorization();
// Mapped only when a signing key is configured (JwtOptions.IsConfigured): an unconfigured
// endpoint used to answer HTTP 500 from the JwtTokenService constructor on every call.
if (app.ServiceProvider.GetRequiredService<IOptions<JwtOptions>>().Value.IsConfigured)
// COOKIE scheme only (fix round 1 / I2): the default scheme also accepts a bearer, and a bearer
// that could mint a fresh token would renew itself forever — outliving its own expiry and the
// user's LDAP membership. A token is minted only for an interactive (cookie) session.
app.MapPost("/auth/token", (Delegate)IssueToken).RequireAuthorization(CookieSessionOnly);
else
app.ServiceProvider.GetService<ILoggerFactory>()?
.CreateLogger("ZB.MOM.WW.OtOpcUa.Security.AuthEndpoints")
.LogInformation("Security:Jwt:SigningKey is not configured — /auth/token is not mapped and bearer tokens are not accepted");
app.MapPost("/auth/logout", (Delegate)LogoutAsync).RequireAuthorization();
return app;
}
@@ -73,8 +95,13 @@ public static class AuthEndpoints
{
result = await ldap.AuthenticateAsync(username, password, ct);
}
catch (Exception)
catch (Exception ex) when (ex is not OperationCanceledException)
{
// R03:2.7 — a directory outage used to leave NOTHING in the server log from this path; the
// operator had to infer every 503 from the directory side.
http.RequestServices.GetService<ILoggerFactory>()?
.CreateLogger("ZB.MOM.WW.OtOpcUa.Security.AuthEndpoints")
.LogWarning(ex, "LDAP authentication failed for {User}; answering 503", username);
return Results.StatusCode(StatusCodes.Status503ServiceUnavailable);
}
@@ -15,4 +15,26 @@ public sealed class JwtOptions
/// <summary>Default token expiry. Mirrors ScadaLink (15 min).</summary>
public int ExpiryMinutes { get; set; } = 15;
/// <summary>
/// Whether the JWT feature is on: a non-blank <see cref="SigningKey"/>. When off, <c>/auth/token</c>
/// is not mapped and no bearer scheme is registered, so an admin node that never configured a key
/// keeps booting exactly as before; when on, the key is validated at host start
/// (<c>ValidateOnStart</c>) instead of throwing HTTP 500 on the first token request.
/// </summary>
public bool IsConfigured => !string.IsNullOrWhiteSpace(SigningKey);
/// <summary>The validation rule shared by the startup validator and the tests: why the options are invalid, or null.</summary>
/// <returns>A failure message, or <see langword="null"/> when valid.</returns>
public string? Validate()
{
if (!IsConfigured) return null;
var keyBytes = System.Text.Encoding.UTF8.GetByteCount(SigningKey);
if (keyBytes < MinSigningKeyBytes)
return $"{SectionName}:SigningKey must be at least {MinSigningKeyBytes} bytes (256 bits) UTF-8 for HMAC-SHA256; the configured key is {keyBytes} byte(s).";
if (string.IsNullOrWhiteSpace(Issuer)) return $"{SectionName}:Issuer must not be blank.";
if (string.IsNullOrWhiteSpace(Audience)) return $"{SectionName}:Audience must not be blank.";
if (ExpiryMinutes <= 0) return $"{SectionName}:ExpiryMinutes must be > 0.";
return null;
}
}
@@ -0,0 +1,19 @@
using Microsoft.Extensions.Options;
namespace ZB.MOM.WW.OtOpcUa.Security.Jwt;
/// <summary>
/// Startup validator for <see cref="JwtOptions"/> (R03:2.4): when a signing key is configured it must
/// be usable, and the host must learn that at start (<c>ValidateOnStart</c>) rather than from an HTTP 500
/// on the first <c>/auth/token</c>. An unconfigured section (no key) is valid — the feature is simply off.
/// </summary>
public sealed class JwtOptionsValidator : IValidateOptions<JwtOptions>
{
/// <inheritdoc />
public ValidateOptionsResult Validate(string? name, JwtOptions options)
{
ArgumentNullException.ThrowIfNull(options);
var failure = options.Validate();
return failure is null ? ValidateOptionsResult.Success : ValidateOptionsResult.Fail(failure);
}
}
@@ -25,24 +25,12 @@ public sealed class JwtTokenService
/// <summary>
/// Role claim type used in the JWT payload.
/// <para>
/// <b>Issued-only / no internal JwtBearer scheme:</b> OtOpcUa uses a single Cookie
/// authentication scheme; the JWT is minted by the <c>/auth/token</c> endpoint and
/// consumed externally (e.g. by OPC-UA clients or automation scripts). There is no
/// <c>AddJwtBearer</c> pipeline in OtOpcUa — the cookie stores the
/// <see cref="System.Security.Claims.ClaimsPrincipal"/> directly. Because no internal
/// bearer validation path exists, the short "Role" key is intentionally used here rather
/// than the long <see cref="ClaimTypes.Role"/> URI; external consumers receive exactly the
/// key they expect.
/// </para>
/// <para>
/// <b>If a JwtBearer scheme is ever added:</b> the
/// <see cref="Microsoft.IdentityModel.Tokens.TokenValidationParameters"/> passed to
/// <c>AddJwtBearer</c> MUST set <c>RoleClaimType = JwtTokenService.RoleClaimType</c> (and
/// <c>NameClaimType = JwtTokenService.UsernameClaimType</c>) so that
/// <c>[Authorize(Roles=...)]</c> and <c>ClaimsPrincipal.IsInRole</c> resolve correctly.
/// <see cref="BuildValidationParameters"/> is already wired to do this and MUST be used
/// rather than constructing <see cref="Microsoft.IdentityModel.Tokens.TokenValidationParameters"/>
/// ad hoc.
/// The short "Role" key is what <see cref="Issue"/> mints and what external consumers of the token
/// expect. The internal bearer scheme (<c>AddJwtBearer</c> in <c>AddOtOpcUaAuth</c>) validates with
/// <see cref="BuildValidationParameters"/>, which sets <c>RoleClaimType = RoleClaimType</c> and
/// <c>NameClaimType = UsernameClaimType</c>, and turns <c>MapInboundClaims</c> off, so
/// <c>[Authorize(Roles=...)]</c> and <c>ClaimsPrincipal.IsInRole</c> resolve against this key. Never
/// construct <see cref="Microsoft.IdentityModel.Tokens.TokenValidationParameters"/> ad hoc.
/// </para>
/// </summary>
public const string RoleClaimType = "Role";
@@ -150,5 +138,9 @@ public sealed class JwtTokenService
// if this is ever passed to AddJwtBearer. See RoleClaimType doc comment for rationale.
RoleClaimType = RoleClaimType,
NameClaimType = UsernameClaimType,
// Pinned to the one algorithm Issue signs with: an unsigned (alg:none) token or one signed with a
// different HMAC variant over the same key is refused on both the bearer and TryValidate paths.
ValidAlgorithms = [SecurityAlgorithms.HmacSha256],
RequireSignedTokens = true,
};
}
@@ -1,5 +1,6 @@
using Microsoft.AspNetCore.Authentication;
using Microsoft.AspNetCore.Authentication.Cookies;
using Microsoft.AspNetCore.Authentication.JwtBearer;
using Microsoft.AspNetCore.Authorization;
using Microsoft.AspNetCore.DataProtection;
using Microsoft.AspNetCore.Http;
@@ -19,9 +20,12 @@ using ZB.MOM.WW.OtOpcUa.Security.Ldap;
namespace ZB.MOM.WW.OtOpcUa.Security;
/// <summary>
/// DI registration for OtOpcUa auth. Single Cookie scheme (the JWT lives inside the
/// cookie as its credential payload); no JwtBearer parallel scheme. Matches ScadaBridge
/// structurally — see <c>docs/plans/2026-05-29-auth-alignment-design.md</c>.
/// DI registration for OtOpcUa auth. The cookie scheme carries the browser session; when
/// <c>Security:Jwt:SigningKey</c> is configured a <see cref="OtOpcUaAuthSchemes.Bearer"/> scheme
/// accepts the tokens <c>/auth/token</c> mints (R03:2.4), and the default scheme is a policy scheme
/// that forwards to whichever the request presents — so <c>/api/*</c> and every
/// <c>RequireAuthorization</c> endpoint take a browser session OR a machine token. Matches
/// ScadaBridge structurally — see <c>docs/plans/2026-05-29-auth-alignment-design.md</c>.
/// </summary>
public static class ServiceCollectionExtensions
{
@@ -36,7 +40,14 @@ public static class ServiceCollectionExtensions
/// <returns>The same service collection, for chaining.</returns>
public static IServiceCollection AddOtOpcUaAuth(this IServiceCollection services, IConfiguration configuration)
{
services.AddOptions<JwtOptions>().Bind(configuration.GetSection(JwtOptions.SectionName));
// JWT is opt-in by key: with no SigningKey the feature is off (no /auth/token, no bearer scheme)
// and an admin node boots as before; with one, the key is validated at host START (R03:2.4) —
// the JwtTokenService constructor used to throw on the first /auth/token call (HTTP 500).
services.AddOptions<JwtOptions>()
.Bind(configuration.GetSection(JwtOptions.SectionName))
.ValidateOnStart();
services.AddSingleton<IValidateOptions<JwtOptions>, JwtOptionsValidator>();
var jwtConfigured = configuration.GetSection(JwtOptions.SectionName).Get<JwtOptions>()?.IsConfigured ?? false;
services.AddOptions<OtOpcUaCookieOptions>().Bind(configuration.GetSection(OtOpcUaCookieOptions.SectionName));
services.AddOptions<LdapOptions>().Bind(configuration.GetSection(LdapOptions.SectionName));
services.AddOptions<AuthDisableLoginOptions>().Bind(configuration.GetSection(AuthDisableLoginOptions.SectionName));
@@ -53,7 +64,10 @@ public static class ServiceCollectionExtensions
// as the Actor field when constructing a canonical ZB.MOM.WW.Audit.AuditEvent. Registered
// Scoped so it correctly follows the request scope used by Blazor Server and minimal-API
// endpoints.
services.TryAddScoped<IAuditActorAccessor, HttpAuditActorAccessor>();
// AuthenticationStateAuditActorAccessor reads the Blazor circuit's principal first (no HttpContext
// exists on a circuit's threads after the first render) and falls back to the HTTP context for
// minimal-API scopes, where AuthenticationStateProvider is not registered (R03:2.5).
services.TryAddScoped<IAuditActorAccessor, AuthenticationStateAuditActorAccessor>();
// Singleton — OtOpcUaLdapAuthService is stateless (the shared-library directory client it
// wraps opens/disposes an LdapConnection per call) and must be consumable by the Singleton
@@ -78,7 +92,39 @@ public static class ServiceCollectionExtensions
.GetSection(AuthDisableLoginOptions.SectionName)
.GetValue<bool>(nameof(AuthDisableLoginOptions.DisableLogin));
var authBuilder = services.AddAuthentication(CookieAuthenticationDefaults.AuthenticationScheme);
// The default scheme is the cookie-or-bearer selector; sign-in/out stay on the cookie scheme, which is
// the only one that can issue a session. With JWT off the selector always picks the cookie.
var authBuilder = services.AddAuthentication(o =>
{
o.DefaultScheme = OtOpcUaAuthSchemes.CookieOrBearer;
o.DefaultSignInScheme = CookieAuthenticationDefaults.AuthenticationScheme;
o.DefaultSignOutScheme = CookieAuthenticationDefaults.AuthenticationScheme;
});
authBuilder.AddPolicyScheme(OtOpcUaAuthSchemes.CookieOrBearer, "Cookie or Bearer", o =>
{
o.ForwardDefaultSelector = ctx =>
jwtConfigured && OtOpcUaAuthSchemes.HasBearerToken(ctx)
? OtOpcUaAuthSchemes.Bearer
: CookieAuthenticationDefaults.AuthenticationScheme;
});
if (jwtConfigured)
{
// First-party JwtBearer (R03:2.4, fix round 1 / I1) — the earlier hand-rolled handler is gone.
// Validation parameters come from the SAME JwtTokenService that mints, so issuer/audience/key/
// claim types can never drift; BuildValidationParameters pins ValidAlgorithms = HS256 (an
// alg:none or a different-HMAC token is refused). MapInboundClaims=false keeps the minted
// "Role"/"zb:username" claim keys verbatim, which RoleClaimType/NameClaimType name. A bad token
// is AuthenticateResult.Fail ⇒ 401 + WWW-Authenticate: Bearer, never a 500.
authBuilder.AddJwtBearer(OtOpcUaAuthSchemes.Bearer, "OtOpcUa bearer token", _ => { });
services.AddOptions<JwtBearerOptions>(OtOpcUaAuthSchemes.Bearer)
.Configure<JwtTokenService>((o, jwt) =>
{
o.TokenValidationParameters = jwt.BuildValidationParameters();
o.MapInboundClaims = false;
o.SaveToken = false;
o.IncludeErrorDetails = false;
});
}
if (disableLogin)
{
// DEV/TEST ONLY: replace the cookie handler with an always-succeeding handler registered
@@ -142,8 +188,8 @@ public static class ServiceCollectionExtensions
services.AddAuthorization(o =>
{
o.FallbackPolicy = new AuthorizationPolicyBuilder(
CookieAuthenticationDefaults.AuthenticationScheme)
// Named on the selector, not the cookie scheme, so the fallback accepts a bearer token too.
o.FallbackPolicy = new AuthorizationPolicyBuilder(OtOpcUaAuthSchemes.CookieOrBearer)
.RequireAuthenticatedUser()
.Build();
@@ -10,6 +10,10 @@
</ItemGroup>
<ItemGroup>
<PackageReference Include="Microsoft.AspNetCore.Authentication.JwtBearer"/>
<!-- Direct, to promote JwtBearer's 8.0.1 IdentityModel.Protocols dependencies to the central 8.11.0 pin. -->
<PackageReference Include="Microsoft.IdentityModel.Protocols"/>
<PackageReference Include="Microsoft.IdentityModel.Protocols.OpenIdConnect"/>
<PackageReference Include="Microsoft.IdentityModel.Tokens"/>
<PackageReference Include="System.IdentityModel.Tokens.Jwt"/>
<PackageReference Include="ZB.MOM.WW.Auth.Abstractions"/>