From 5b92eb0ff67883ceff4370639d3529bef4bf204c Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Sun, 16 Aug 2026 04:48:02 -0400 Subject: [PATCH] perf(runtime): direct-mode artifact read piped off the host mailbox --- .../Drivers/DriverHostActor.cs | 335 +++++++++--------- ...iverHostActorDirectApplyOffMailboxTests.cs | 152 ++++++++ 2 files changed, 323 insertions(+), 164 deletions(-) create mode 100644 tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverHostActorDirectApplyOffMailboxTests.cs diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/DriverHostActor.cs b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/DriverHostActor.cs index ed207cf7..9c5417c5 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/DriverHostActor.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/DriverHostActor.cs @@ -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(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(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(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); } /// /// Per-cluster mesh Phase 3 (FetchAndCache). Kicks off a gRPC artifact fetch and pipes - /// the result back to as a — the fetch is I/O + /// the result back to as an — 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 - /// , mirroring . + /// , the continuation both modes share. /// /// - /// This is the ONLY path that arms the watchdog, deliberately. - /// enters and leaves 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 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.) /// 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)); } /// - /// Completes a FetchAndCache 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 (Direct). Kicks off the central-SQL artifact read on a pool + /// thread and pipes the bytes back to as an , + /// the same shape has always had for its gRPC fetch. The + /// apply itself completes on the actor thread in . /// - private void HandleFetchedForApply(FetchedForApply msg) + /// + /// + /// The read used to run INLINE in the receive. Deployment.ArtifactBlob 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. + /// + /// + /// The bytes come back as a MESSAGE, never as an escaping fault (#491): the task closes over + /// , and and touches + /// neither nor Context, 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. + /// + /// + 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)); + } + + /// + /// 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. + /// + 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("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("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("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); } - /// Pipe-back carrier for a FetchAndCache fetch result (null bytes = no answer). - private sealed record FetchedForApply( - DeploymentId DeploymentId, RevisionHash Revision, CorrelationId Correlation, byte[]? Bytes); - - private void ApplyAndAck(DeploymentId deploymentId, RevisionHash revision, CorrelationId correlation) + /// Where an apply's artifact bytes came from — the ONLY difference between the two apply + /// paths, and therefore the only thing branches on. + private enum ArtifactSource { - _applyingDeploymentId = deploymentId; - Become(Applying); + /// Direct mode: read from Deployment.ArtifactBlob in central SQL. + 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("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("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("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(); - } + /// FetchAndCache mode: fetched from central over gRPC. + CentralFetch, } + /// Pipe-back carrier for an apply's artifact bytes (null bytes = no answer). + private sealed record ArtifactForApply( + DeploymentId DeploymentId, + RevisionHash Revision, + CorrelationId Correlation, + byte[]? Bytes, + ArtifactSource Source); + /// /// Read the deployment artifact + reconcile the set of running /// children. Spawn missing, ApplyDelta on config change, stop removed/disabled drivers. @@ -2053,7 +2048,7 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS /// /// /// The bootstrap-restore entry point (). The apply paths use - /// , which keeps its parse for the rest of the apply. + /// , which keeps its parse for the rest of the apply. /// /// /// 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 /// /// The ConfigDb read half of , shared with the Direct apply path - /// () so both read the artifact exactly once and log the + /// () so both read the artifact exactly once and log the /// same way when they cannot. /// + /// + /// Called on the actor thread by the PreStart restore path and on a POOL thread by + /// , so it must stay free of / Context + /// (#491). It touches only , and + /// , all of which are safe off the actor thread — the logger publishes to the + /// EventStream and captures nothing from the cell. + /// /// The artifact bytes (possibly empty, when the row is missing), or /// when there is no ConfigDb to read or the read failed. 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); /// - /// 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 - /// and threads it through the rest of the apply, so a - /// Direct apply builds exactly one JsonDocument in this actor. - /// - /// The artifact bytes are not valid JSON. Unlike the - /// lenient blob wrappers (malformed ⇒ no specs ⇒ every running child planned for StopChild) this - /// surfaces, and 's catch turns it into a Failed ACK with the children left - /// running — the honest reading of bytes that cannot be understood. - private ReconciledArtifact ReconcileDriversForApply(DeploymentId deploymentId) - { - var blob = LoadArtifactBlob(deploymentId); - return blob is null ? default : ReconcileDriversFromParsed(deploymentId, blob); - } - - /// - /// The parse-once reconcile shared by both apply paths ( via - /// , and 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 + /// () owns and disposes the returned + /// and threads it through the rest of the apply, so an + /// apply builds exactly one JsonDocument in this actor. /// /// The reconciled blob + its parsed artifact, or a default (both null) when the blob was - /// empty (#485). + /// empty or absent (#485). + /// The artifact bytes are not valid JSON. Unlike the + /// lenient blob wrappers (malformed ⇒ no specs ⇒ every running child planned for StopChild) this + /// surfaces, and 's catch turns it into a Failed ACK with the + /// children left running — the honest reading of bytes that cannot be understood. private ReconciledArtifact ReconcileDriversFromParsed(DeploymentId deploymentId, byte[]? blob) { blob ??= Array.Empty(); @@ -2227,7 +2219,7 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS /// /// Never throws. 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 's catch and send a second, + /// escaping here would unwind into 's catch and send a second, /// contradictory Failed ACK for a deployment the fleet already believes is live. /// /// @@ -2250,7 +2242,7 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS /// Stale is included on purpose. This used to claim Stale could not be reached after a /// store had started, because it is only ever entered from . That is /// not a property of the Stale entry — starts stores of its own, via - /// the orphan-replay and via , and both + /// the orphan-replay and via , 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. /// + /// + /// Its two ConfigDb reads ( and + /// ) stay SYNCHRONOUS deliberately: this runs in + /// , 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. + /// 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). /// /// - /// The bootstrap-restore entry point only. Neither apply path reaches it any more: both - /// and already hold the apply's single - /// parse and call directly, so this ConfigDb read - /// exists for , which restores from a deployment id recovered out of + /// The bootstrap-restore entry point only. Neither apply path reaches it any more: + /// already holds the apply's single parse and calls + /// directly, so this ConfigDb read exists for + /// , which restores from a deployment id recovered out of /// NodeDeploymentState and may hold no blob of its own. /// 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); diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverHostActorDirectApplyOffMailboxTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverHostActorDirectApplyOffMailboxTests.cs new file mode 100644 index 00000000..1855a594 --- /dev/null +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverHostActorDirectApplyOffMailboxTests.cs @@ -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; + +/// +/// Perf roadmap Task 26 — the Direct-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. +/// +/// +/// +/// The FetchAndCache path has always piped its artifact fetch back to Self for +/// exactly this reason. Direct 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. +/// +/// +/// The load-bearing assertion is therefore that the host answers an unrelated 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. +/// +/// +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 { "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(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(TimeSpan.FromSeconds(3)); + + // Releasing the read completes the apply exactly as before — same Applied ACK, same revision. + db.Open(); + coordinator.ExpectMsg(TimeSpan.FromSeconds(10)).Outcome.ShouldBe(ApplyAckOutcome.Applied); + + actor.Tell(new GetDiagnostics(CorrelationId.NewId()), coordinator.Ref); + coordinator.ExpectMsg(TimeSpan.FromSeconds(5)).CurrentRevision.ShouldBe(RevA); + } + finally + { + // Never leave the gate closed: a parked read would hang ActorSystem shutdown. + db.Open(); + } + } + + /// Seeds a sealed, parseable deployment and returns its id. + private static DeploymentId SeedDeployment(IDbContextFactory db, RevisionHash rev) + { + var artifact = JsonSerializer.SerializeToUtf8Bytes(new { DriverInstances = Array.Empty() }); + 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; + } + + /// + /// 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. + /// + private sealed class GatedConfigDbFactory(IDbContextFactory inner) + : IDbContextFactory + { + private readonly ManualResetEventSlim _gate = new(initialState: true); + private int _allowance = int.MaxValue; + private int _blocked; + private int _unblocked; + + /// How many callers have reached the closed gate. + public int Blocked => Volatile.Read(ref _blocked); + + /// How many of those have been let through again. + public int Unblocked => Volatile.Read(ref _unblocked); + + /// Closes the gate, after letting more contexts past. + public void CloseAfter(int allowThrough) + { + Volatile.Write(ref _allowance, allowThrough); + _gate.Reset(); + } + + /// Opens the gate for good — also the test-teardown safety net. + 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(); + } + } +}