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

perf(runtime): direct-mode artifact read piped off the host mailbox

Browse Source
This commit is contained in:
Joseph Doherty
2026-08-16 04:48:02 -04:00
parent ca87329a83
commit 5b92eb0ff6
2 changed files with 323 additions and 164 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/DriverHostActor.cs
+171 -164
View File
@@ -853,7 +853,7 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS
case NodeDeploymentStatus.Applying:
_log.Warning("DriverHost {Node}: found orphan Applying row for deployment {Id}; replaying",
_localNode, latest.DeploymentId);
if (revision is not null) ApplyAndAck(new DeploymentId(latest.DeploymentId), revision.Value, CorrelationId.NewId());
if (revision is not null) BeginDirectApply(new DeploymentId(latest.DeploymentId), revision.Value, CorrelationId.NewId());
else BecomeSteady();
break;
case NodeDeploymentStatus.Failed:
@@ -1115,9 +1115,10 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS
_localNode, msg.DeploymentId, _applyingDeploymentId);
Stash.Stash();
});
// The FetchAndCache fetch pipes its result back here (BeginFetchAndCacheApply ran in Steady,
// then Become(Applying)); completing the apply transitions back to Steady.
Receive<FetchedForApply>(HandleFetchedForApply);
// Both apply paths pipe their artifact bytes back here — the FetchAndCache gRPC fetch and the
// Direct ConfigDb read (BeginFetchAndCacheApply / BeginDirectApply ran in Steady, then
// Become(Applying)); completing the apply transitions back to Steady.
Receive<ArtifactForApply>(HandleArtifactForApply);
// The cache write is piped back too, and it outlives the apply that started it by design — so
// its completion can land in EITHER state and both must handle it or it dead-letters.
Receive<ArtifactCached>(HandleArtifactCached);
@@ -1703,7 +1704,7 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS
_log.Warning("DriverHost {Node}: ignoring DispatchDeployment while Stale (DB unreachable)", _localNode);
});
// A cache store started before this node fell Stale still pipes back here (see
// CacheAppliedArtifact: Bootstrap's orphan-replay ApplyAndAck and RestoreApplied both start
// CacheAppliedArtifact: Bootstrap's orphan-replay BeginDirectApply and RestoreApplied both start
// stores, and either can be followed by the ConfigDb failure that lands us in Stale). Without
// these the pipe-back dead-letters and a failed cache write goes unreported in exactly the
// situation — central unreachable — where the local cache is the thing keeping the node alive.
@@ -1775,23 +1776,22 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS
return;
}
ApplyAndAck(msg.DeploymentId, msg.RevisionHash, msg.CorrelationId);
BeginDirectApply(msg.DeploymentId, msg.RevisionHash, msg.CorrelationId);
}
/// <summary>
/// Per-cluster mesh Phase 3 (<c>FetchAndCache</c>). Kicks off a gRPC artifact fetch and pipes
/// the result back to <see cref="Self"/> as a <see cref="FetchedForApply"/> — the fetch is I/O
/// the result back to <see cref="Self"/> as an <see cref="ArtifactForApply"/> — the fetch is I/O
/// and MUST NOT block the mailbox for the fetch deadline, so it never runs as an awaited call
/// inside a receive. The apply itself completes synchronously on the actor thread in
/// <see cref="HandleFetchedForApply"/>, mirroring <see cref="ApplyAndAck"/>.
/// <see cref="HandleArtifactForApply"/>, the continuation both modes share.
/// </summary>
/// <remarks>
/// This is the ONLY path that arms the <see cref="ApplyWedgeTimerKey"/> watchdog, deliberately.
/// <see cref="ApplyAndAck"/> enters and leaves <see cref="Applying"/> inside a single mailbox
/// turn, so it cannot be "stuck in Applying" in any sense a watchdog could report: if it hung,
/// it would be hanging the actor thread, and a watchdog delivered as a mailbox message could not
/// be dequeued to say so. The fetch is the only unbounded wait during which the host is still
/// alive and dequeuing — which is exactly what makes a silent wedge possible there.
/// Both begin-paths arm the <see cref="ApplyWedgeTimerKey"/> watchdog, because both leave the
/// host alive and dequeuing while it waits for bytes it does not have yet — which is exactly
/// what makes a silent wedge possible. (Before the Direct read moved off the mailbox this was
/// the only such path: an inline read could not be "stuck in Applying" in any sense a watchdog
/// could report, since a mailbox message cannot be dequeued by an actor thread that is hung.)
/// </remarks>
private void BeginFetchAndCacheApply(DeploymentId deploymentId, RevisionHash revision, CorrelationId correlation)
{
@@ -1799,7 +1799,7 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS
Become(Applying);
Timers.StartSingleTimer(ApplyWedgeTimerKey, new ApplyWedged(deploymentId), _applyWedgeTimeout);
// Persist Applying row (idempotent on PK) — same crash-boundary marker as ApplyAndAck.
// Persist Applying row (idempotent on PK) — same crash-boundary marker as the Direct apply.
UpsertNodeDeploymentState(deploymentId, NodeDeploymentStatus.Applying, failureReason: null);
if (_artifactFetcher is null)
@@ -1823,21 +1823,70 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS
_artifactFetcher.FetchAsync(deploymentId.ToString(), revision.Value, CancellationToken.None)
.PipeTo(
Self,
success: bytes => new FetchedForApply(deploymentId, revision, correlation, bytes),
failure: _ => new FetchedForApply(deploymentId, revision, correlation, Bytes: null));
success: bytes => new ArtifactForApply(
deploymentId, revision, correlation, bytes, ArtifactSource.CentralFetch),
failure: _ => new ArtifactForApply(
deploymentId, revision, correlation, Bytes: null, ArtifactSource.CentralFetch));
}
/// <summary>
/// Completes a <c>FetchAndCache</c> apply once the fetched bytes are in hand. Null bytes are an
/// apply FAILURE (keep last-known-good, do not advance the revision — #485); non-null bytes are
/// applied from hand (no ConfigDb read) and cached.
/// Perf roadmap Task 26 (<c>Direct</c>). Kicks off the central-SQL artifact read on a pool
/// thread and pipes the bytes back to <see cref="Self"/> as an <see cref="ArtifactForApply"/>,
/// the same shape <see cref="BeginFetchAndCacheApply"/> has always had for its gRPC fetch. The
/// apply itself completes on the actor thread in <see cref="HandleArtifactForApply"/>.
/// </summary>
private void HandleFetchedForApply(FetchedForApply msg)
/// <remarks>
/// <para>
/// The read used to run INLINE in the receive. <c>Deployment.ArtifactBlob</c> is a
/// potentially multi-MB gzip blob in central SQL, so for the whole of that round trip the
/// host dequeued nothing: no driver value batch, no diagnostics request, no driver command,
/// no next dispatch. Only the byte SOURCE differs between the two modes, and only the
/// source ever should — everything downstream of "bytes in hand" is one continuation.
/// </para>
/// <para>
/// The bytes come back as a MESSAGE, never as an escaping fault (#491): the task closes over
/// <see cref="_dbFactory"/>, <see cref="_localNode"/> and <see cref="_log"/> and touches
/// neither <see cref="Self"/> nor <c>Context</c>, <see cref="LoadArtifactBlob"/> already
/// swallows its own DB failures into a null return, and the failure mapper turns anything it
/// cannot (an OOM on an oversized blob, a factory that throws) into the same null-bytes
/// apply failure. Supervision never sees a throw from here.
/// </para>
/// </remarks>
private void BeginDirectApply(DeploymentId deploymentId, RevisionHash revision, CorrelationId correlation)
{
_applyingDeploymentId = deploymentId;
Become(Applying);
Timers.StartSingleTimer(ApplyWedgeTimerKey, new ApplyWedged(deploymentId), _applyWedgeTimeout);
// Persist Applying row (idempotent on PK) — the crash-boundary marker, and deliberately still
// on the actor thread: one bounded row write, and the apply must be recorded as in flight
// before the read that could outlive this process starts. See UpsertNodeDeploymentState.
UpsertNodeDeploymentState(deploymentId, NodeDeploymentStatus.Applying, failureReason: null);
_log.Debug("DriverHost {Node}: reading artifact for {Id} (rev {Rev}) from the ConfigDb, off the mailbox",
_localNode, deploymentId, revision);
Task.Run(() => LoadArtifactBlob(deploymentId))
.PipeTo(
Self,
success: bytes => new ArtifactForApply(
deploymentId, revision, correlation, bytes, ArtifactSource.ConfigDb),
failure: _ => new ArtifactForApply(
deploymentId, revision, correlation, Bytes: null, ArtifactSource.ConfigDb));
}
/// <summary>
/// Completes an apply once the artifact bytes are in hand — the ONE continuation both modes
/// share, whether the bytes were read from the ConfigDb or fetched from central. Null bytes are
/// an apply FAILURE (keep last-known-good, do not advance the revision — #485/#486); non-null
/// bytes are applied from hand and cached.
/// </summary>
private void HandleArtifactForApply(ArtifactForApply msg)
{
// Ignore a result for anything but the in-flight apply (a stale/duplicate pipe-back).
if (_applyingDeploymentId is null || msg.DeploymentId != _applyingDeploymentId.Value)
{
_log.Debug("DriverHost {Node}: dropping stale fetch result for {Id} (in-flight={Cur})",
_log.Debug("DriverHost {Node}: dropping stale artifact result for {Id} (in-flight={Cur})",
_localNode, msg.DeploymentId, _applyingDeploymentId);
return;
}
@@ -1853,21 +1902,27 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS
ParsedDeploymentArtifact? parsed = null;
try
{
// #485: null bytes = "no answer" (every endpoint unreachable / NotFound / SHA mismatch).
// Fail the apply and stay on the current revision so a re-dispatch actually retries; keep
// serving the last-known-good address space, drivers and subscriptions throughout.
// #485/#486: null bytes = "no answer" — the ConfigDb was unreachable or the row is
// missing/empty (Direct), or every endpoint was unreachable / NotFound / SHA mismatch
// (FetchAndCache). Either way NOTHING was applied, so fail the apply and stay on the
// current revision: HandleDispatchFromSteady short-circuits on a revision match, so
// claiming Applied here would make the node unhealable by re-dispatching that revision.
// The last-known-good address space, drivers and subscriptions keep serving throughout.
var (appliedBlob, parsedArtifact) = ReconcileDriversFromParsed(msg.DeploymentId, msg.Bytes);
parsed = parsedArtifact;
if (appliedBlob is null || parsed is null)
{
const string reason =
"the deployment artifact could not be fetched from central (all endpoints unreachable, NotFound, or SHA-256 mismatch); nothing was applied";
var reason = msg.Source == ArtifactSource.ConfigDb
? "the deployment artifact could not be read (ConfigDb unreachable, or the row is missing/empty); nothing was applied"
: "the deployment artifact could not be fetched from central (all endpoints unreachable, NotFound, or SHA-256 mismatch); nothing was applied";
UpsertNodeDeploymentState(msg.DeploymentId, NodeDeploymentStatus.Failed, reason);
SendAck(msg.DeploymentId, ApplyAckOutcome.Failed, reason, msg.Correlation);
OtOpcUaTelemetry.DeploymentApplied.Add(1, new KeyValuePair<string, object?>("outcome", "reject"));
span?.SetStatus(ActivityStatusCode.Error, reason);
_log.Error(
"DriverHost {Node}: FetchAndCache apply of {Id} FAILED — {Reason}. Staying on revision {Rev} and continuing to serve it; re-dispatch once central is reachable",
msg.Source == ArtifactSource.ConfigDb
? "DriverHost {Node}: apply of {Id} FAILED — {Reason}. Staying on revision {Rev} and continuing to serve it; re-dispatch once the ConfigDb is readable"
: "DriverHost {Node}: FetchAndCache apply of {Id} FAILED — {Reason}. Staying on revision {Rev} and continuing to serve it; re-dispatch once central is reachable",
_localNode, msg.DeploymentId, reason, _currentRevision);
return;
}
@@ -1875,19 +1930,34 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS
_currentRevision = msg.Revision;
UpsertNodeDeploymentState(msg.DeploymentId, NodeDeploymentStatus.Applied, failureReason: null);
SendAck(msg.DeploymentId, ApplyAckOutcome.Applied, failureReason: null, msg.Correlation);
// Rebuild the address space + push subscriptions FROM the fetched bytes so OpcUaPublishActor
// never reads central SQL (the whole point of FetchAndCache).
// The message carries the raw BYTES, not `parsed`: the publish actor is a different actor with
// its own mailbox, and ParsedDeploymentArtifact is a disposable, non-thread-safe object owned
// by this handler. It opens its own (single) parse on the other side of the mailbox.
// Rebuild the address space + push subscriptions FROM the bytes this apply already has, so
// nothing downstream re-reads the artifact: under FetchAndCache OpcUaPublishActor has no
// central SQL to read at all, and under Direct a bytes-less RebuildAddressSpace would make
// it re-read the same immutable row while the SubscribeBulk pass read it a THIRD time.
// Consequence: OpcUaPublishActor.LoadArtifact / LoadLatestArtifact are on no apply path;
// they remain only for a legacy/dev caller that Tells RebuildAddressSpace without bytes.
//
// What rides along is the raw BYTES, deliberately — not `parsed`. The publish actor is a
// different actor with its own mailbox, and ParsedDeploymentArtifact is a disposable,
// non-thread-safe object owned by this handler (disposed in the finally below, likely before
// the other actor even dequeues). It opens its own single parse on the far side.
_opcUaPublishActor?.Tell(new ZB.MOM.WW.OtOpcUa.Runtime.OpcUa.OpcUaPublishActor.RebuildAddressSpace(
msg.Correlation, msg.DeploymentId, appliedBlob));
// SubscribeBulk pass: hand each driver its desired tag references so live values flow into
// the just-rebuilt address space instead of staying BadWaitingForInitialData.
PushDesiredSubscriptionsFromParsed(msg.DeploymentId, parsed);
// Cache the fetched bytes as this node's last-known-good (idempotent store; skips empty).
// Cache the applied bytes as this node's last-known-good (idempotent store; skips empty).
CacheAppliedArtifact(msg.DeploymentId, msg.Revision, appliedBlob, parsed);
// Reaching here means the artifact came from central, so the node is no longer serving a
// cache-sourced configuration. Note this only clears on a real apply: a dispatch of the
// revision already booted from cache short-circuits in HandleDispatchFromSteady without
// touching the ConfigDb, and the flag correctly stays set.
_isRunningFromCache = false;
OtOpcUaTelemetry.DeploymentApplied.Add(1, new KeyValuePair<string, object?>("outcome", "ack"));
_log.Info("DriverHost {Node}: applied fetched deployment {Id} (rev {Rev}, children={Count})",
_log.Info(
msg.Source == ArtifactSource.ConfigDb
? "DriverHost {Node}: applied deployment {Id} (rev {Rev}, children={Count})"
: "DriverHost {Node}: applied fetched deployment {Id} (rev {Rev}, children={Count})",
_localNode, msg.DeploymentId, msg.Revision, _children.Count);
}
catch (Exception ex)
@@ -1896,7 +1966,12 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS
SendAck(msg.DeploymentId, ApplyAckOutcome.Failed, ex.Message, msg.Correlation);
OtOpcUaTelemetry.DeploymentApplied.Add(1, new KeyValuePair<string, object?>("outcome", "reject"));
span?.SetStatus(ActivityStatusCode.Error, ex.Message);
_log.Error(ex, "DriverHost {Node}: FetchAndCache apply of {Id} failed", _localNode, msg.DeploymentId);
_log.Error(
ex,
msg.Source == ArtifactSource.ConfigDb
? "DriverHost {Node}: apply of {Id} failed"
: "DriverHost {Node}: FetchAndCache apply of {Id} failed",
_localNode, msg.DeploymentId);
}
finally
{
@@ -1935,110 +2010,30 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS
_log.Warning(
"DriverHost {Node}: deployment {Id} appears WEDGED — still applying after {Timeout} " +
"(the artifact fetch has neither answered nor failed); {Deferred} further dispatch(es) are " +
"deferred behind it. The node keeps serving revision {Rev}.",
"(the artifact read/fetch has neither answered nor failed); {Deferred} further dispatch(es) " +
"are deferred behind it. The node keeps serving revision {Rev}.",
_localNode, msg.DeploymentId, _applyWedgeTimeout, _deferredDispatches.Count, _currentRevision);
}
/// <summary>Pipe-back carrier for a <c>FetchAndCache</c> fetch result (null bytes = no answer).</summary>
private sealed record FetchedForApply(
DeploymentId DeploymentId, RevisionHash Revision, CorrelationId Correlation, byte[]? Bytes);
private void ApplyAndAck(DeploymentId deploymentId, RevisionHash revision, CorrelationId correlation)
/// <summary>Where an apply's artifact bytes came from — the ONLY difference between the two apply
/// paths, and therefore the only thing <see cref="HandleArtifactForApply"/> branches on.</summary>
private enum ArtifactSource
{
_applyingDeploymentId = deploymentId;
Become(Applying);
/// <summary><c>Direct</c> mode: read from <c>Deployment.ArtifactBlob</c> in central SQL.</summary>
ConfigDb,
using var span = OtOpcUaTelemetry.StartDeployApplySpan(deploymentId.ToString());
span?.SetTag("otopcua.node_id", _localNode.ToString());
span?.SetTag("otopcua.revision", revision.ToString());
span?.SetTag("otopcua.correlation_id", correlation.ToString());
var sw = Stopwatch.StartNew();
// Persist Applying row (idempotent on PK).
UpsertNodeDeploymentState(deploymentId, NodeDeploymentStatus.Applying, failureReason: null);
// Owned by this handler, disposed in the finally: ONE JsonDocument for the whole apply. Declared
// out here (rather than a `using` inside the try) so the catch/finally still see it.
ParsedDeploymentArtifact? parsed = null;
try
{
var (appliedBlob, parsedArtifact) = ReconcileDriversForApply(deploymentId);
parsed = parsedArtifact;
// Issue #486 — a null blob means the artifact could not be read (unreachable ConfigDb, or the
// empty bytes a missing row yields), so NOTHING was applied. Reporting Applied here used to
// advance _currentRevision too, and HandleDispatchFromSteady short-circuits on a revision
// match — so the node claimed a configuration it never applied AND could never be healed by
// re-dispatching that revision. Failing the apply keeps the revision where it was, which is
// what makes the retry land instead of being waved through. The node keeps serving its
// last-known-good address space, drivers and subscriptions throughout (issue #485) — this is
// about telling the truth, not about tearing anything down.
if (appliedBlob is null || parsed is null)
{
const string reason =
"the deployment artifact could not be read (ConfigDb unreachable, or the row is missing/empty); nothing was applied";
UpsertNodeDeploymentState(deploymentId, NodeDeploymentStatus.Failed, reason);
SendAck(deploymentId, ApplyAckOutcome.Failed, reason, correlation);
OtOpcUaTelemetry.DeploymentApplied.Add(1, new KeyValuePair<string, object?>("outcome", "reject"));
span?.SetStatus(ActivityStatusCode.Error, reason);
_log.Error(
"DriverHost {Node}: apply of {Id} FAILED — {Reason}. Staying on revision {Rev} and continuing to serve it; re-dispatch once the ConfigDb is readable",
_localNode, deploymentId, reason, _currentRevision);
return;
}
_currentRevision = revision;
UpsertNodeDeploymentState(deploymentId, NodeDeploymentStatus.Applied, failureReason: null);
SendAck(deploymentId, ApplyAckOutcome.Applied, failureReason: null, correlation);
// Trigger the OPC UA address-space rebuild so the local SDK reflects the new
// composition. The publish actor handles the load-compose-diff-apply pipeline; we
// just forward the same correlation id so the audit trail joins up.
//
// The already-read bytes ride along. ReconcileDrivers just read this exact blob out of central
// SQL and handed it back (it is non-null here — the null case returned above), so a bytes-less
// RebuildAddressSpace would make the publish actor re-read the same immutable row, and the
// SubscribeBulk pass would read it a THIRD time: three round-trips per apply for one blob that
// cannot have changed between them. Threading it is also the shape the FetchAndCache apply has
// always had (HandleFetchedForApply), for the harder reason that it has no ConfigDb to re-read.
// Consequence: OpcUaPublishActor.LoadArtifact / LoadLatestArtifact are no longer on any apply
// path — they remain only for a legacy/dev caller that Tells RebuildAddressSpace without bytes.
//
// What rides along is the BYTES, deliberately — not this apply's `parsed`. The publish actor is
// a different actor with its own mailbox, and ParsedDeploymentArtifact is a disposable,
// non-thread-safe object owned by this handler (disposed in the finally below, likely before the
// other actor even dequeues). It opens its own single parse on the far side.
_opcUaPublishActor?.Tell(new ZB.MOM.WW.OtOpcUa.Runtime.OpcUa.OpcUaPublishActor.RebuildAddressSpace(correlation, deploymentId, appliedBlob));
// SubscribeBulk pass: hand each driver its desired tag references so live values flow into
// the just-rebuilt address space instead of staying BadWaitingForInitialData.
PushDesiredSubscriptionsFromParsed(deploymentId, parsed);
CacheAppliedArtifact(deploymentId, revision, appliedBlob, parsed);
// Reaching here means the artifact came from central, so the node is no longer serving a
// cache-sourced configuration. Note this only clears on a real apply: a dispatch of the
// revision already booted from cache short-circuits in HandleDispatchFromSteady without
// touching the ConfigDb, and the flag correctly stays set.
_isRunningFromCache = false;
OtOpcUaTelemetry.DeploymentApplied.Add(1, new KeyValuePair<string, object?>("outcome", "ack"));
_log.Info("DriverHost {Node}: applied deployment {Id} (rev {Rev}, children={Count})",
_localNode, deploymentId, revision, _children.Count);
}
catch (Exception ex)
{
UpsertNodeDeploymentState(deploymentId, NodeDeploymentStatus.Failed, ex.Message);
SendAck(deploymentId, ApplyAckOutcome.Failed, ex.Message, correlation);
OtOpcUaTelemetry.DeploymentApplied.Add(1, new KeyValuePair<string, object?>("outcome", "reject"));
span?.SetStatus(ActivityStatusCode.Error, ex.Message);
_log.Error(ex, "DriverHost {Node}: apply of {Id} failed", _localNode, deploymentId);
}
finally
{
parsed?.Dispose();
OtOpcUaTelemetry.DeploymentApplyDurationSec.Record(sw.Elapsed.TotalSeconds);
_applyingDeploymentId = null;
BecomeSteady();
}
/// <summary><c>FetchAndCache</c> mode: fetched from central over gRPC.</summary>
CentralFetch,
}
/// <summary>Pipe-back carrier for an apply's artifact bytes (null bytes = no answer).</summary>
private sealed record ArtifactForApply(
DeploymentId DeploymentId,
RevisionHash Revision,
CorrelationId Correlation,
byte[]? Bytes,
ArtifactSource Source);
/// <summary>
/// Read the deployment artifact + reconcile the set of running <see cref="DriverInstanceActor"/>
/// children. Spawn missing, ApplyDelta on config change, stop removed/disabled drivers.
@@ -2053,7 +2048,7 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS
/// <remarks>
/// <para>
/// The <b>bootstrap-restore</b> entry point (<see cref="RestoreApplied"/>). The apply paths use
/// <see cref="ReconcileDriversForApply"/>, which keeps its parse for the rest of the apply.
/// <see cref="ReconcileDriversFromParsed"/>, which keeps its parse for the rest of the apply.
/// </para>
/// <para>
/// Returning the blob rather than swallowing it is load-bearing for the artifact cache. This
@@ -2072,9 +2067,16 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS
/// <summary>
/// The ConfigDb read half of <see cref="ReconcileDrivers"/>, shared with the Direct apply path
/// (<see cref="ReconcileDriversForApply"/>) so both read the artifact exactly once and log the
/// (<see cref="BeginDirectApply"/>) so both read the artifact exactly once and log the
/// same way when they cannot.
/// </summary>
/// <remarks>
/// Called on the actor thread by the PreStart restore path and on a POOL thread by
/// <see cref="BeginDirectApply"/>, so it must stay free of <see cref="Self"/> / <c>Context</c>
/// (#491). It touches only <see cref="_dbFactory"/>, <see cref="_localNode"/> and
/// <see cref="_log"/>, all of which are safe off the actor thread — the logger publishes to the
/// EventStream and captures nothing from the cell.
/// </remarks>
/// <returns>The artifact bytes (possibly empty, when the row is missing), or <see langword="null"/>
/// when there is no ConfigDb to read or the read failed.</returns>
private byte[]? LoadArtifactBlob(DeploymentId deploymentId)
@@ -2114,28 +2116,18 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS
private readonly record struct ReconciledArtifact(byte[]? Blob, ParsedDeploymentArtifact? Parsed);
/// <summary>
/// The Direct apply path's reconcile: read the artifact from the ConfigDb, parse it ONCE, and
/// reconcile the driver children from that parse. The caller owns (and disposes) the returned
/// <see cref="ParsedDeploymentArtifact"/> and threads it through the rest of the apply, so a
/// Direct apply builds exactly one <c>JsonDocument</c> in this actor.
/// </summary>
/// <exception cref="System.Text.Json.JsonException">The artifact bytes are not valid JSON. Unlike the
/// lenient blob wrappers (malformed ⇒ no specs ⇒ every running child planned for StopChild) this
/// surfaces, and <see cref="ApplyAndAck"/>'s catch turns it into a Failed ACK with the children left
/// running — the honest reading of bytes that cannot be understood.</exception>
private ReconciledArtifact ReconcileDriversForApply(DeploymentId deploymentId)
{
var blob = LoadArtifactBlob(deploymentId);
return blob is null ? default : ReconcileDriversFromParsed(deploymentId, blob);
}
/// <summary>
/// The parse-once reconcile shared by both apply paths (<see cref="ApplyAndAck"/> via
/// <see cref="ReconcileDriversForApply"/>, and <see cref="HandleFetchedForApply"/> with the
/// fetched bytes already in hand).
/// The parse-once reconcile every apply runs, with the artifact bytes already in hand: parse
/// ONCE and reconcile the driver children from that parse. The caller
/// (<see cref="HandleArtifactForApply"/>) owns and disposes the returned
/// <see cref="ParsedDeploymentArtifact"/> and threads it through the rest of the apply, so an
/// apply builds exactly one <c>JsonDocument</c> in this actor.
/// </summary>
/// <returns>The reconciled blob + its parsed artifact, or a default (both null) when the blob was
/// empty (#485).</returns>
/// empty or absent (#485).</returns>
/// <exception cref="System.Text.Json.JsonException">The artifact bytes are not valid JSON. Unlike the
/// lenient blob wrappers (malformed ⇒ no specs ⇒ every running child planned for StopChild) this
/// surfaces, and <see cref="HandleArtifactForApply"/>'s catch turns it into a Failed ACK with the
/// children left running — the honest reading of bytes that cannot be understood.</exception>
private ReconciledArtifact ReconcileDriversFromParsed(DeploymentId deploymentId, byte[]? blob)
{
blob ??= Array.Empty<byte>();
@@ -2227,7 +2219,7 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS
/// <para>
/// <b>Never throws.</b> By the time this runs the deployment is already recorded Applied
/// in central SQL and an Applied ACK has been sent to the coordinator. An exception
/// escaping here would unwind into <see cref="ApplyAndAck"/>'s catch and send a second,
/// escaping here would unwind into <see cref="HandleArtifactForApply"/>'s catch and send a second,
/// contradictory Failed ACK for a deployment the fleet already believes is live.
/// </para>
/// <para>
@@ -2250,7 +2242,7 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS
/// <b>Stale is included on purpose.</b> This used to claim Stale could not be reached after a
/// store had started, because it is only ever entered from <see cref="Bootstrap"/>. That is
/// not a property of the Stale entry — <see cref="Bootstrap"/> starts stores of its own, via
/// the orphan-replay <see cref="ApplyAndAck"/> and via <see cref="RestoreApplied"/>, and both
/// the orphan-replay <see cref="BeginDirectApply"/> and via <see cref="RestoreApplied"/>, and both
/// run inside the very try block whose catch Becomes Stale. Today no statement between those
/// calls and the catch happens to throw, so the old claim held by accident of ordering rather
/// than by design; a restart can orphan an in-flight pipe-back the same way. Registering the
@@ -2368,6 +2360,12 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS
/// drivers, rebuilds the address space from the applied artifact, and re-pushes SubscribeBulk.
/// No re-ack: the deployment is already Applied.
/// </summary>
/// <remarks>
/// Its two ConfigDb reads (<see cref="ReconcileDrivers"/> and
/// <see cref="PushDesiredSubscriptions"/>) stay SYNCHRONOUS deliberately: this runs in
/// <see cref="PreStart"/>, before the host has processed a single message, so there is no live
/// traffic behind them to block — unlike the apply path, which moved its read off the mailbox.
/// </remarks>
private void RestoreApplied(DeploymentId deploymentId, RevisionHash? revision)
{
var correlation = CorrelationId.NewId();
@@ -2378,7 +2376,7 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS
PushDesiredSubscriptions(deploymentId);
// Populate the node-local cache from the artifact this restored node is now serving. The
// cache invariant is "holds what the node currently serves"; without this only a FRESH
// apply (ApplyAndAck) ever wrote it, so a node whose cache was lost — a wiped/fresh volume,
// apply (BeginDirectApply) ever wrote it, so a node whose cache was lost — a wiped/fresh volume,
// a disk failure — would recover its served state here yet stay cache-less, unable to
// boot-from-cache on the NEXT central outage until some future new deploy happened to land.
// Replication does not heal that gap either: a peer's already-acked rows are pruned from its
@@ -2403,10 +2401,10 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS
/// (which clears any stale subscription from a previous deployment).
/// </summary>
/// <remarks>
/// The <b>bootstrap-restore</b> entry point only. Neither apply path reaches it any more: both
/// <see cref="ApplyAndAck"/> and <see cref="HandleFetchedForApply"/> already hold the apply's single
/// parse and call <see cref="PushDesiredSubscriptionsFromParsed"/> directly, so this ConfigDb read
/// exists for <see cref="RestoreApplied"/>, which restores from a deployment id recovered out of
/// The <b>bootstrap-restore</b> entry point only. Neither apply path reaches it any more:
/// <see cref="HandleArtifactForApply"/> already holds the apply's single parse and calls
/// <see cref="PushDesiredSubscriptionsFromParsed"/> directly, so this ConfigDb read exists for
/// <see cref="RestoreApplied"/>, which restores from a deployment id recovered out of
/// NodeDeploymentState and may hold no blob of its own.
/// </remarks>
private void PushDesiredSubscriptions(DeploymentId deploymentId)
@@ -2423,6 +2421,8 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS
byte[] blob;
try
{
// Synchronous by design: the only caller is the PreStart restore path above, where nothing
// is queued behind this read (the apply path reads off the mailbox instead).
using var db = _dbFactory.CreateDbContext();
blob = db.Deployments.AsNoTracking()
.Where(d => d.DeploymentId == deploymentId.Value)
@@ -2647,7 +2647,7 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS
// Hand the Equipment-namespace VirtualTags to the host so it spawns/reconciles a
// VirtualTagActor per plan and streams their evaluated values back onto the just-rebuilt
// address space. Runs on BOTH the fresh-apply path (ApplyAndAck) and the bootstrap-restore
// address space. Runs on BOTH the fresh-apply path (HandleArtifactForApply) and the bootstrap-restore
// path (RestoreApplied) because both call this method, so one send covers both.
// NOTE: the Stale-recovery path (TryRecoverFromStale) does NOT call PushDesiredSubscriptions,
// so — like drivers — VirtualTags remain empty after a Stale recovery until the next
@@ -2672,7 +2672,7 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS
_localNode, composition.EquipmentScriptedAlarms.Count);
}
// Cache the applied composition LAST. Set here (not in ApplyAndAck) so both the fresh-apply and
// Cache the applied composition LAST. Set here (not in HandleArtifactForApply) so both the fresh-apply and
// bootstrap-restore paths — which both route through this method — leave a current composition.
_lastComposition = composition;
@@ -2927,6 +2927,9 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS
try
{
// Synchronous, like the Applying/Applied row writes below: one bounded row read on a timer
// tick, not a multi-MB blob. Recorded follow-up — if the Stale-recovery tick is ever shown
// to stall a busy host, it moves off the mailbox the same way BeginDirectApply's read did.
using var db = _dbFactory.CreateDbContext();
var latestSealed = db.Deployments
.AsNoTracking()
@@ -2973,6 +2976,10 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS
try
{
// Synchronous by design, and a recorded follow-up rather than an oversight: this is a
// bounded per-apply write of ONE row (at most three per deployment), not the multi-MB
// artifact read that Task 26 moved off the mailbox. It is also the apply's crash-boundary
// marker, which is easiest to reason about while it is ordered on the actor thread.
using var db = _dbFactory.CreateDbContext();
var existing = db.NodeDeploymentStates.FirstOrDefault(
x => x.NodeId == _localNode.Value && x.DeploymentId == deploymentId.Value);
tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverHostActorDirectApplyOffMailboxTests.cs
+152
View File
@@ -0,0 +1,152 @@
using System.Text.Json;
using Akka.Actor;
using Microsoft.EntityFrameworkCore;
using Shouldly;
using Xunit;
using ZB.MOM.WW.OtOpcUa.Commons.Interfaces;
using ZB.MOM.WW.OtOpcUa.Commons.Messages.Deploy;
using ZB.MOM.WW.OtOpcUa.Commons.Messages.Fleet;
using ZB.MOM.WW.OtOpcUa.Commons.Types;
using ZB.MOM.WW.OtOpcUa.Configuration;
using ZB.MOM.WW.OtOpcUa.Configuration.Entities;
using ZB.MOM.WW.OtOpcUa.Configuration.Enums;
using ZB.MOM.WW.OtOpcUa.Runtime.Drivers;
using ZB.MOM.WW.OtOpcUa.Runtime.Tests.Harness;
namespace ZB.MOM.WW.OtOpcUa.Runtime.Tests.Drivers;
/// <summary>
/// Perf roadmap Task 26 — the <c>Direct</c>-mode artifact read is I/O (an EF read of a potentially
/// multi-MB gzip blob out of central SQL) and must not run on the actor thread.
/// </summary>
/// <remarks>
/// <para>
/// The <c>FetchAndCache</c> path has always piped its artifact fetch back to <c>Self</c> for
/// exactly this reason. <c>Direct</c> read the blob INLINE inside the receive, so every value
/// batch, diagnostics request and driver command queued behind central SQL for the whole read —
/// on a WAN link or a loaded server, for as long as that read takes.
/// </para>
/// <para>
/// The load-bearing assertion is therefore that the host answers an <b>unrelated</b> message
/// while the read is still in flight, not that the ACK eventually arrives: a blocking read still
/// acks, just late and behind everything else.
/// </para>
/// </remarks>
public sealed class DriverHostActorDirectApplyOffMailboxTests : RuntimeActorTestBase
{
private static readonly NodeId TestNode = NodeId.Parse("driver-test");
private static readonly RevisionHash RevA = RevisionHash.Parse(new string('a', 64));
[Fact]
public void A_slow_direct_artifact_read_does_not_block_the_host_mailbox()
{
var inner = NewInMemoryDbFactory();
var db = new GatedConfigDbFactory(inner);
var deploymentId = SeedDeployment(inner, RevA);
var coordinator = CreateTestProbe();
var actor = Sys.ActorOf(DriverHostActor.Props(
db, TestNode, coordinator.Ref,
localRoles: new HashSet<string> { "driver" }));
try
{
// Barrier: an answered GetDiagnostics proves PreStart (and its bootstrap ConfigDb read) is
// done, so the allowance below counts only the apply's own contexts.
actor.Tell(new GetDiagnostics(CorrelationId.NewId()), coordinator.Ref);
coordinator.ExpectMsg<NodeDiagnosticsSnapshot>(TimeSpan.FromSeconds(5));
// Let ONE more context through — the apply's Applying-row upsert, which is a deliberate
// on-thread write (bounded, one row) — and block the next, which is the artifact read.
// Both misconfigurations of this number fail LOUDLY rather than passing vacuously: too
// small blocks the upsert and the assertion below times out even on a fixed build; too
// large never engages the gate and the presence assertion on Blocked fails.
db.CloseAfter(allowThrough: 1);
actor.Tell(new DispatchDeployment(deploymentId, RevA, CorrelationId.NewId()));
// The read was entered and has NOT completed — the gate is still closed.
AwaitAssert(() => db.Blocked.ShouldBe(1), duration: PresenceBudget);
db.Unblocked.ShouldBe(0);
// THE assertion: with the artifact read still in flight the host is still processing its
// mailbox. Read inline on the actor thread, this times out.
actor.Tell(new GetDiagnostics(CorrelationId.NewId()), coordinator.Ref);
coordinator.ExpectMsg<NodeDiagnosticsSnapshot>(TimeSpan.FromSeconds(3));
// Releasing the read completes the apply exactly as before — same Applied ACK, same revision.
db.Open();
coordinator.ExpectMsg<ApplyAck>(TimeSpan.FromSeconds(10)).Outcome.ShouldBe(ApplyAckOutcome.Applied);
actor.Tell(new GetDiagnostics(CorrelationId.NewId()), coordinator.Ref);
coordinator.ExpectMsg<NodeDiagnosticsSnapshot>(TimeSpan.FromSeconds(5)).CurrentRevision.ShouldBe(RevA);
}
finally
{
// Never leave the gate closed: a parked read would hang ActorSystem shutdown.
db.Open();
}
}
/// <summary>Seeds a sealed, parseable deployment and returns its id.</summary>
private static DeploymentId SeedDeployment(IDbContextFactory<OtOpcUaConfigDbContext> db, RevisionHash rev)
{
var artifact = JsonSerializer.SerializeToUtf8Bytes(new { DriverInstances = Array.Empty<object>() });
var id = DeploymentId.NewId();
using var ctx = db.CreateDbContext();
ctx.Deployments.Add(new Deployment
{
DeploymentId = id.Value,
RevisionHash = rev.Value,
Status = DeploymentStatus.Sealed,
CreatedBy = "test",
SealedAtUtc = DateTime.UtcNow,
ArtifactBlob = artifact,
});
ctx.SaveChanges();
return id;
}
/// <summary>
/// A ConfigDb factory that parks the caller on a gate — the fake shape of a slow central-SQL
/// read. It blocks the calling THREAD (not a task), because that is what an EF read does and it
/// is the only shape that can prove which thread the read runs on.
/// </summary>
private sealed class GatedConfigDbFactory(IDbContextFactory<OtOpcUaConfigDbContext> inner)
: IDbContextFactory<OtOpcUaConfigDbContext>
{
private readonly ManualResetEventSlim _gate = new(initialState: true);
private int _allowance = int.MaxValue;
private int _blocked;
private int _unblocked;
/// <summary>How many callers have reached the closed gate.</summary>
public int Blocked => Volatile.Read(ref _blocked);
/// <summary>How many of those have been let through again.</summary>
public int Unblocked => Volatile.Read(ref _unblocked);
/// <summary>Closes the gate, after letting <paramref name="allowThrough"/> more contexts past.</summary>
public void CloseAfter(int allowThrough)
{
Volatile.Write(ref _allowance, allowThrough);
_gate.Reset();
}
/// <summary>Opens the gate for good — also the test-teardown safety net.</summary>
public void Open() => _gate.Set();
public OtOpcUaConfigDbContext CreateDbContext()
{
if (Interlocked.Decrement(ref _allowance) < 0 && !_gate.IsSet)
{
Interlocked.Increment(ref _blocked);
// Bounded so a mis-wired test fails rather than hanging the whole assembly.
_gate.Wait(TimeSpan.FromSeconds(30));
Interlocked.Increment(ref _unblocked);
}
return inner.CreateDbContext();
}
}
}