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

fix(controlplane): keep the admin singleton's mailbox free during Test Connect and deploy compose

Browse Source 
ReceiveAsync suspended the mailbox for the whole of every handler, so one Test Connect against a
black-holed PLC (60 s clamp) or one large deploy compose stalled every other admin operation
fleet-wide — AdminUI alarm acks/shelves, driver restarts, the next deploy.

- TestDriverConnect runs the probe on the thread pool and pipes a TestDriverConnectResult
  straight to the requester (every path yields a result, never Status.Failure).
- StartDeployment serialises only the in-flight check (DB row + an actor-local compose-in-flight
  flag, because the Deployment row is written at the END of the compose); the snapshot read,
  validation, gzip and row write run off-actor and return as DeployComposed /
  DeployComposeFailed, where the coordinator is told and the requester answered.

Tests: an alarm ack completes in < 1 s while a 60 s probe is pending and while a compose is
parked on a gated DbContext (positive controls: both still complete afterwards); a second
StartDeployment during an in-flight compose is refused.

Refs #733

Claude-Session: https://claude.ai/code/session_01Ub9hGmYLQvn9XETzw7CMsT
This commit is contained in:
Joseph Doherty
2026-10-22 06:41:35 -04:00
parent 73e74e270d
commit ae84253773
2 changed files with 432 additions and 181 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/Server/ZB.MOM.WW.OtOpcUa.ControlPlane/AdminOperations/AdminOperationsActor.cs
+273 -181
View File
@@ -36,6 +36,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>
@@ -74,8 +77,12 @@ public sealed class AdminOperationsActor : ReceiveActor
_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);
@@ -181,19 +188,42 @@ public sealed class AdminOperationsActor : ReceiveActor
}
}
private async Task HandleStartDeploymentAsync(StartDeployment msg)
/// <summary>
/// Serialised part of <c>StartDeployment</c>: the in-flight check. Everything after it — the
/// snapshot read, validation, gzip and the row write — runs on the thread pool and comes back as a
/// <see cref="DeployComposed"/> / <see cref="DeployComposeFailed"/> message, so a large compose no
/// longer suspends the mailbox (R03:3.1) and alarm acks keep flowing while it runs.
/// </summary>
/// <remarks>
/// <see cref="_composeInFlight"/> is what keeps the check correct once the compose is off-actor: the
/// Deployment row is written at the END of the compose, so a second request arriving mid-compose
/// would pass a DB-only check and two artifacts would race for one coordinator. The flag is cleared
/// when the outcome message lands, on both branches.
/// </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;
}
try
{
await using var db = await _dbFactory.CreateDbContextAsync();
// 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
// unambiguous. The UI is expected to wait for the in-flight one to seal or fail. One indexed
// projection (IX_Deployment_Status), synchronous on purpose: it IS the serialised part.
using var db = _dbFactory.CreateDbContext();
var inflight = db.Deployments
.Where(d => d.Status == DeploymentStatus.Dispatching || d.Status == DeploymentStatus.AwaitingApplyAcks)
.Select(d => d.DeploymentId)
.FirstOrDefaultAsync();
.FirstOrDefault();
if (inflight != Guid.Empty)
{
replyTo.Tell(new StartDeploymentResult(
@@ -204,172 +234,6 @@ public sealed class AdminOperationsActor : ReceiveActor
msg.CorrelationId));
return;
}
// 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);
replyTo.Tell(new StartDeploymentResult(
StartDeploymentOutcome.Rejected,
DeploymentId: null,
RevisionHash: null,
Message: summary,
msg.CorrelationId));
return;
}
// 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)
// Projected from the snapshot already in hand — Script.SourceCode used to be read three times.
var compiled = draft.Scripts
.Select(s => s.SourceCode)
.Where(src => !PassthroughScript.TryMatch(src, out _))
.Distinct(StringComparer.Ordinal)
.Count();
if (compiled > 0)
{
var estMiB = compiled * PerScriptMiB;
_log.Warning(
"StartDeployment: {Compiled} script(s) will compile (~{EstMiB:F0} MiB RSS per node); ensure node mem_limit covers it",
compiled, estMiB);
advisory = $"{compiled} script(s) will compile (~{estMiB:F0} MiB/node)";
}
}
catch (Exception ex)
{
// Guardrail is advisory-only — a failed estimate must never block a valid deploy.
_log.Warning(ex, "StartDeployment: script compile-cost estimate failed; advisory skipped");
advisory = null;
}
// Composed from the validated snapshot, never from a second read: an edit that lands between
// validation and here is simply not in this artifact (R03:3.2).
var artifact = ConfigComposer.Compose(draft);
var deploymentId = DeploymentId.NewId();
var revHash = RevisionHash.Parse(artifact.RevisionHash);
db.Deployments.Add(new Deployment
{
DeploymentId = deploymentId.Value,
RevisionHash = artifact.RevisionHash,
Status = DeploymentStatus.Dispatching,
CreatedBy = msg.CreatedBy,
ArtifactBlob = artifact.Blob,
});
// Marker ConfigEdit row so the audit timeline shows the deployment snapshot.
db.ConfigEdits.Add(new ConfigEdit
{
EntityType = "Deployment",
EntityId = deploymentId.Value,
FieldsJson = $"{{\"revisionHash\":\"{artifact.RevisionHash}\",\"sizeBytes\":{artifact.Blob.Length}}}",
EditedBy = msg.CreatedBy,
SourceNode = Akka.Cluster.Cluster.Get(Context.System).SelfAddress.Host ?? "unknown",
});
await db.SaveChangesAsync();
_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(
StartDeploymentOutcome.Accepted,
deploymentId,
revHash,
Message: acceptedMessage,
msg.CorrelationId));
}
catch (Exception ex)
{
@@ -380,10 +244,233 @@ public sealed class AdminOperationsActor : ReceiveActor
RevisionHash: null,
Message: ex.Message,
msg.CorrelationId));
return;
}
_composeInFlight = msg;
// Captured here: Context/Sender are not available on the thread-pool thread.
var sourceNode = Akka.Cluster.Cluster.Get(Context.System).SelfAddress.Host ?? "unknown";
var self = Self;
Task.Run(() => ComposeDeploymentAsync(msg, sourceNode))
.PipeTo(self,
success: composed => new DeployComposed(msg, replyTo, composed.Result, composed.Dispatch),
failure: ex => new DeployComposeFailed(msg, replyTo, ex));
}
private async Task HandleTestDriverConnectAsync(TestDriverConnect msg)
private void HandleDeployComposed(DeployComposed msg)
{
_composeInFlight = null;
// The row is committed (the compose task completed), so the coordinator can be told now; it reads
// the row back by id when it seeds the expected-ack set.
if (msg.Dispatch is { } dispatch) _coordinator.Tell(dispatch);
msg.ReplyTo.Tell(msg.Result);
}
private void HandleDeployComposeFailed(DeployComposeFailed msg)
{
_composeInFlight = null;
_log.Error(msg.Error, "StartDeployment failed for {CreatedBy}", msg.Request.CreatedBy);
msg.ReplyTo.Tell(new StartDeploymentResult(
StartDeploymentOutcome.Rejected,
DeploymentId: null,
RevisionHash: null,
Message: msg.Error.Message,
msg.Request.CorrelationId));
}
/// <summary>Outcome of the off-actor compose: the reply to send, and the dispatch to issue when Accepted.</summary>
private sealed record ComposeOutcome(StartDeploymentResult Result, DispatchDeployment? Dispatch);
private sealed record DeployComposed(
StartDeployment Request, IActorRef ReplyTo, StartDeploymentResult Result, DispatchDeployment? Dispatch);
private sealed record DeployComposeFailed(StartDeployment Request, IActorRef ReplyTo, Exception Error);
/// <summary>
/// The deploy compose, off the actor thread. Touches no actor state: everything it needs is in
/// <paramref name="msg"/>, <paramref name="sourceNode"/> and the readonly fields captured at
/// construction (<see cref="_dbFactory"/>, <see cref="_tagConfigValidationMode"/>, <see cref="_log"/> —
/// Akka's logging adapter is thread-safe).
/// </summary>
private async Task<ComposeOutcome> ComposeDeploymentAsync(StartDeployment msg, string sourceNode)
{
await using var db = await _dbFactory.CreateDbContextAsync();
// Full pre-seal gate: run the complete DraftValidator (one pass, all rules) and reject
// on ANY validation error. The config is now canonical — the company overlay loader emits
// canonical EquipmentIds and the seed is clean — so every rule (UNS segments, EquipmentId
// derivation, cross-cluster/namespace binding, driver-namespace compat, signal collisions,
// …) gates the deploy. A green build in this repo does not prove the config is valid; this
// is the last guard before a bad address space (or a non-derived EquipmentId) ships.
// NOTE: a BadDuplicateExternalIdentifier rejection from the reservation pre-flight can
// (rarely) be a false positive if an external-id reservation release has not yet been
// committed/visible when the snapshot is read — operators seeing a spurious one should
// check ExternalIdReservation state before re-submitting.
var draft = await DraftSnapshotFactory.FromConfigDbAsync(db);
var errors = DraftValidator.Validate(draft).ToList();
// Cluster-topology guard. The SQL
// CK_ServerCluster_RedundancyMode_NodeCount CHECK enforces the (NodeCount, RedundancyMode)
// pair on the row itself, but it cannot see the per-node ClusterNode.Enabled flag — an
// operator can disable a node (effective enabled-count = 1) while leaving RedundancyMode at
// Hot/Warm and the constraint stays green, which would boot the runtime into an
// InvalidTopology band. ValidateClusterTopology catches that drift; it is not part of
// DraftValidator.Validate, so it runs separately here over the snapshot's cluster/node rows.
// Errors fold into the same reject summary alongside the snapshot rules so a deploy failing
// either check is rejected with both sets of messages. ClusterId-ordered for a deterministic summary.
// R03:3.2 — both come off the SAME snapshot read as every other table (one SNAPSHOT
// transaction), not a second query that could observe a later edit.
var clusters = draft.ServerClusters.OrderBy(c => c.ClusterId, StringComparer.Ordinal).ToList();
var nodesByCluster = draft.ClusterNodes
.GroupBy(n => n.ClusterId, StringComparer.Ordinal)
.ToDictionary(g => g.Key, g => g.ToList(), StringComparer.Ordinal);
foreach (var cluster in clusters)
{
var nodes = nodesByCluster.TryGetValue(cluster.ClusterId, out var ns)
? (IReadOnlyList<ClusterNode>)ns
: [];
errors.AddRange(DraftValidator.ValidateClusterTopology(cluster, nodes));
}
// Deploy-time TagConfig strictness inspection over equipment tags. The
// driver-typed EquipmentTagConfigInspector surfaces every blob the driver's own reader would
// REJECT — a present-but-invalid enum value, an absent or malformed address field, a
// structurally-unparseable TagConfig — each of which makes the driver skip the tag so it reads
// BadNodeIdUnknown forever. (The runtime used to silently default a typo'd enum;
// it now rejects the tag, so these warnings describe rejections.) In Warn mode
// (default) these are non-blocking (logged + appended to the accepted result Message); in Error
// mode they fold into the reject list so a typo'd config cannot be re-deployed until fixed.
// Running servers are untouched either way — the gate only sees re-deploys.
// v3: a raw Tag's driver is resolved via its Device (Tag no longer carries EquipmentId /
// DriverInstanceId). Inspect EVERY raw tag's TagConfig against its device's driver type.
var driverTypeById = draft.DriverInstances
.GroupBy(d => d.DriverInstanceId, StringComparer.Ordinal)
.ToDictionary(g => g.Key, g => g.First().DriverType, StringComparer.Ordinal);
var driverByDevice = draft.Devices
.GroupBy(d => d.DeviceId, StringComparer.Ordinal)
.ToDictionary(g => g.Key, g => g.First().DriverInstanceId, StringComparer.Ordinal);
var deviceConfigById = draft.Devices
.GroupBy(d => d.DeviceId, StringComparer.Ordinal)
.ToDictionary(g => g.Key, g => g.First().DeviceConfig, StringComparer.Ordinal);
var tagConfigWarnings = new List<(string TagId, string Text)>();
foreach (var t in draft.Tags)
{
if (!driverByDevice.TryGetValue(t.DeviceId, out var driverInstanceId)) continue;
if (!driverTypeById.TryGetValue(driverInstanceId, out var driverType)) continue;
foreach (var w in EquipmentTagConfigInspector.Inspect(
driverType, t.TagConfig, deviceConfigById.GetValueOrDefault(t.DeviceId)))
tagConfigWarnings.Add((t.TagId, $"tag '{t.TagId}': {w}"));
}
if (tagConfigWarnings.Count > 0)
{
if (_tagConfigValidationMode == TagConfigValidationMode.Error)
errors.AddRange(tagConfigWarnings.Select(x => new ValidationError("TagConfigInvalid", x.Text, x.TagId)));
else
foreach (var x in tagConfigWarnings)
_log.Warning("StartDeployment: TagConfig warning (Warn mode, non-blocking): {Warning}", x.Text);
}
if (errors.Count > 0)
{
var summary = string.Join("; ", errors.Select(e => $"[{e.Code}] {e.Message}"));
_log.Warning("StartDeployment rejected ({Count} validation error(s)): {Summary}",
errors.Count, summary);
return new ComposeOutcome(
new StartDeploymentResult(
StartDeploymentOutcome.Rejected,
DeploymentId: null,
RevisionHash: null,
Message: summary,
msg.CorrelationId),
Dispatch: null);
}
// Warn-only compile-cost guardrail (NEVER blocks). Each genuinely-compiled (non-passthrough)
// script costs ~1.66 MiB RSS per node to materialize via Roslyn (measured post-A0; design doc
// 2026-06-07). Passthrough "mirror" scripts compile to ~nothing, so they are excluded; distinct
// sources are counted (the compile cache keys on source, so duplicates collapse to one unit).
// This only surfaces the estimate to the operator — the DraftValidator gate above is the hard
// reject; this advisory rides in the Accepted Message so the UI can show it.
// advisory is declared outside the inner try so the Accepted reply below can still read it even
// when the guardrail throws. A failed estimate must NEVER block a valid deploy — the caller's
// failure branch is reserved for genuine seal/save failures.
string? advisory = null;
try
{
const double PerScriptMiB = 1.66; // measured post-A0 per-script RSS (design doc 2026-06-07)
// Projected from the snapshot already in hand — Script.SourceCode used to be read three times.
var compiled = draft.Scripts
.Select(s => s.SourceCode)
.Where(src => !PassthroughScript.TryMatch(src, out _))
.Distinct(StringComparer.Ordinal)
.Count();
if (compiled > 0)
{
var estMiB = compiled * PerScriptMiB;
_log.Warning(
"StartDeployment: {Compiled} script(s) will compile (~{EstMiB:F0} MiB RSS per node); ensure node mem_limit covers it",
compiled, estMiB);
advisory = $"{compiled} script(s) will compile (~{estMiB:F0} MiB/node)";
}
}
catch (Exception ex)
{
// Guardrail is advisory-only — a failed estimate must never block a valid deploy.
_log.Warning(ex, "StartDeployment: script compile-cost estimate failed; advisory skipped");
advisory = null;
}
// Composed from the validated snapshot, never from a second read: an edit that lands between
// validation and here is simply not in this artifact (R03:3.2).
var artifact = ConfigComposer.Compose(draft);
var deploymentId = DeploymentId.NewId();
var revHash = RevisionHash.Parse(artifact.RevisionHash);
db.Deployments.Add(new Deployment
{
DeploymentId = deploymentId.Value,
RevisionHash = artifact.RevisionHash,
Status = DeploymentStatus.Dispatching,
CreatedBy = msg.CreatedBy,
ArtifactBlob = artifact.Blob,
});
// Marker ConfigEdit row so the audit timeline shows the deployment snapshot.
db.ConfigEdits.Add(new ConfigEdit
{
EntityType = "Deployment",
EntityId = deploymentId.Value,
FieldsJson = $"{{\"revisionHash\":\"{artifact.RevisionHash}\",\"sizeBytes\":{artifact.Blob.Length}}}",
EditedBy = msg.CreatedBy,
SourceNode = sourceNode,
});
await db.SaveChangesAsync();
// Accepted message = the compile-cost advisory (if any) + any Warn-mode TagConfig warnings, so
// the operator sees the non-blocking strictness findings on a successful deploy.
var messageParts = new List<string>();
if (advisory is not null) messageParts.Add(advisory);
messageParts.AddRange(tagConfigWarnings.Select(x => x.Text));
var acceptedMessage = messageParts.Count > 0 ? string.Join("; ", messageParts) : null;
return new ComposeOutcome(
new StartDeploymentResult(
StartDeploymentOutcome.Accepted,
deploymentId,
revHash,
Message: acceptedMessage,
msg.CorrelationId),
new DispatchDeployment(deploymentId, revHash, msg.CorrelationId));
}
/// <summary>
/// Test Connect runs on the thread pool and pipes its reply straight to the requester: a probe
/// against a black-holed PLC (60 s clamp) must not suspend the singleton's mailbox (R03:3.1). Every
/// path inside the task yields a <see cref="TestDriverConnectResult"/>, so the requester never
/// sees a <c>Status.Failure</c>.
/// </summary>
private void HandleTestDriverConnect(TestDriverConnect msg)
{
var replyTo = Sender;
if (!_probesByType.TryGetValue(msg.DriverType, out var probe))
@@ -396,6 +483,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);
@@ -405,28 +497,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(
return new TestDriverConnectResult(
false,
ex.Message,
null,
msg.CorrelationId));
msg.CorrelationId);
}
}
tests/Server/ZB.MOM.WW.OtOpcUa.ControlPlane.Tests/AdminOperationsActorMailboxTests.cs
+159
View File
@@ -0,0 +1,159 @@
using Akka.Actor;
using Microsoft.EntityFrameworkCore;
using Shouldly;
using Xunit;
using ZB.MOM.WW.OtOpcUa.Commons.Messages.Admin;
using ZB.MOM.WW.OtOpcUa.Commons.Messages.Deploy;
using ZB.MOM.WW.OtOpcUa.Commons.Types;
using ZB.MOM.WW.OtOpcUa.Configuration;
using ZB.MOM.WW.OtOpcUa.ControlPlane.AdminOperations;
using ZB.MOM.WW.OtOpcUa.ControlPlane.Tests.Harness;
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
namespace ZB.MOM.WW.OtOpcUa.ControlPlane.Tests;
/// <summary>
/// R03:3.1 — the admin singleton's mailbox used to be suspended for the whole of every
/// <c>ReceiveAsync</c>: one Test Connect against a black-holed PLC (60 s clamp) or one large deploy
/// compose stalled every other admin operation fleet-wide, alarm acks included. Both now run off the
/// actor thread; only the "is one in flight" check stays serialised.
/// </summary>
public sealed class AdminOperationsActorMailboxTests : ControlPlaneActorTestBase
{
private static readonly TimeSpan AckBudget = TimeSpan.FromSeconds(1);
/// <summary>An alarm ack completes in under a second while a 60 s Test Connect is still pending.</summary>
[Fact]
public void Alarm_ack_completes_while_a_60s_test_connect_is_pending()
{
var dbFactory = NewInMemoryDbFactory();
var coordinator = CreateTestProbe("coord");
var probe = new BlockingProbe();
var actor = Sys.ActorOf(AdminOperationsActor.Props(dbFactory, coordinator.Ref, new IDriverProbe[] { probe }));
var connectClient = CreateTestProbe("connect-client");
connectClient.Send(actor, new TestDriverConnect("Blocking", "{}", TimeoutSeconds: 60, Guid.NewGuid()));
probe.Started.Wait(PresenceBudget).ShouldBeTrue("the probe must be in flight before the ack is sent");
actor.Tell(new AcknowledgeAlarmCommand("alarm-1", "op", null, CorrelationId.NewId()));
var ack = ExpectMsg<AcknowledgeAlarmResult>(AckBudget);
ack.Ok.ShouldBeTrue();
// Positive control: the probe reply still lands, with the right sender, once the probe returns.
connectClient.ExpectNoMsg(TimeSpan.FromMilliseconds(200));
probe.Release(new DriverProbeResult(true, null, TimeSpan.FromMilliseconds(5)));
var result = connectClient.ExpectMsg<TestDriverConnectResult>(PresenceBudget);
result.Ok.ShouldBeTrue();
result.LatencyMs.ShouldNotBeNull();
}
/// <summary>A probe that times out still answers its own sender with the clamped timeout message.</summary>
[Fact]
public void Test_connect_timeout_is_reported_to_its_sender()
{
var dbFactory = NewInMemoryDbFactory();
var coordinator = CreateTestProbe("coord");
var probe = new BlockingProbe(); // never released: the CTS fires at the 1 s clamp floor
var actor = Sys.ActorOf(AdminOperationsActor.Props(dbFactory, coordinator.Ref, new IDriverProbe[] { probe }));
actor.Tell(new TestDriverConnect("Blocking", "{}", TimeoutSeconds: 1, Guid.NewGuid()));
var result = ExpectMsg<TestDriverConnectResult>(PresenceBudget);
result.Ok.ShouldBeFalse();
result.Message.ShouldNotBeNull();
result.Message.ShouldContain("timed out");
}
/// <summary>
/// The deploy compose (snapshot + validate + gzip + row write) runs off the actor: an alarm ack
/// completes while the compose is parked on a gated DB context, and the deploy still completes
/// (Accepted + dispatched) once the gate opens.
/// </summary>
[Fact]
public void Alarm_ack_completes_while_a_deploy_compose_is_in_flight()
{
var gated = new GatedDbFactory(NewInMemoryDbFactory());
var coordinator = CreateTestProbe("coord");
var actor = Sys.ActorOf(AdminOperationsActor.Props(gated, coordinator.Ref, Enumerable.Empty<IDriverProbe>()));
var deployer = CreateTestProbe("deployer");
gated.Hold();
deployer.Send(actor, new StartDeployment("joe", CorrelationId.NewId()));
gated.Entered.Wait(PresenceBudget).ShouldBeTrue("the compose must be parked before the ack is sent");
actor.Tell(new AcknowledgeAlarmCommand("alarm-2", "op", null, CorrelationId.NewId()));
ExpectMsg<AcknowledgeAlarmResult>(AckBudget).Ok.ShouldBeTrue();
// Positive control: the deploy completes once released.
deployer.ExpectNoMsg(TimeSpan.FromMilliseconds(200));
gated.Release();
deployer.ExpectMsg<StartDeploymentResult>(PresenceBudget).Outcome.ShouldBe(StartDeploymentOutcome.Accepted);
coordinator.ExpectMsg<DispatchDeployment>(PresenceBudget);
}
/// <summary>
/// Only the in-flight check is serialised — and it must still catch a second request that arrives
/// while the first compose has not yet written its row (the DB alone would let both through).
/// </summary>
[Fact]
public void Second_StartDeployment_during_an_in_flight_compose_is_refused()
{
var gated = new GatedDbFactory(NewInMemoryDbFactory());
var coordinator = CreateTestProbe("coord");
var actor = Sys.ActorOf(AdminOperationsActor.Props(gated, coordinator.Ref, Enumerable.Empty<IDriverProbe>()));
var first = CreateTestProbe("first");
var second = CreateTestProbe("second");
gated.Hold();
first.Send(actor, new StartDeployment("joe", CorrelationId.NewId()));
gated.Entered.Wait(PresenceBudget).ShouldBeTrue();
second.Send(actor, new StartDeployment("kim", CorrelationId.NewId()));
second.ExpectMsg<StartDeploymentResult>(AckBudget).Outcome.ShouldBe(StartDeploymentOutcome.AnotherDeploymentInFlight);
gated.Release();
first.ExpectMsg<StartDeploymentResult>(PresenceBudget).Outcome.ShouldBe(StartDeploymentOutcome.Accepted);
coordinator.ExpectMsg<DispatchDeployment>(PresenceBudget);
}
private sealed class BlockingProbe : IDriverProbe
{
private readonly TaskCompletionSource<DriverProbeResult> _tcs =
new(TaskCreationOptions.RunContinuationsAsynchronously);
public ManualResetEventSlim Started { get; } = new();
public string DriverType => "Blocking";
public async Task<DriverProbeResult> ProbeAsync(string configJson, TimeSpan timeout, CancellationToken ct)
{
Started.Set();
return await _tcs.Task.WaitAsync(ct);
}
public void Release(DriverProbeResult result) => _tcs.TrySetResult(result);
}
/// <summary>
/// Passes synchronous context creation through (the serialised in-flight check) and parks the
/// asynchronous one (the off-actor compose) on a gate while <see cref="Hold"/> is in effect.
/// </summary>
private sealed class GatedDbFactory(IDbContextFactory<OtOpcUaConfigDbContext> inner)
: IDbContextFactory<OtOpcUaConfigDbContext>
{
private volatile TaskCompletionSource? _gate;
public ManualResetEventSlim Entered { get; } = new();
public void Hold() => _gate = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
public void Release() => _gate?.TrySetResult();
public OtOpcUaConfigDbContext CreateDbContext() => inner.CreateDbContext();
public async Task<OtOpcUaConfigDbContext> CreateDbContextAsync(CancellationToken cancellationToken = default)
{
Entered.Set();
if (_gate is { } gate) await gate.Task.WaitAsync(cancellationToken);
return inner.CreateDbContext();
}
}
}