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 79cd29a8..d9767ea5 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/DriverHostActor.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/DriverHostActor.cs @@ -1651,13 +1651,17 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers span?.SetTag("otopcua.correlation_id", msg.Correlation.ToString()); var sw = Stopwatch.StartNew(); + // 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 { // #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. - var appliedBlob = ReconcileDriversFromBlob(msg.DeploymentId, msg.Bytes); - if (appliedBlob is null) + 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"; @@ -1676,11 +1680,14 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers 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. _opcUaPublishActor?.Tell(new ZB.MOM.WW.OtOpcUa.Runtime.OpcUa.OpcUaPublishActor.RebuildAddressSpace( msg.Correlation, msg.DeploymentId, appliedBlob)); - PushDesiredSubscriptionsFromArtifact(msg.DeploymentId, appliedBlob); + PushDesiredSubscriptionsFromParsed(msg.DeploymentId, parsed); // Cache the fetched bytes as this node's last-known-good (idempotent store; skips empty). - CacheAppliedArtifact(msg.DeploymentId, msg.Revision, appliedBlob); + CacheAppliedArtifact(msg.DeploymentId, msg.Revision, appliedBlob, parsed); _isRunningFromCache = false; OtOpcUaTelemetry.DeploymentApplied.Add(1, new KeyValuePair("outcome", "ack")); _log.Info("DriverHost {Node}: applied fetched deployment {Id} (rev {Rev}, children={Count})", @@ -1696,6 +1703,7 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers } finally { + parsed?.Dispose(); OtOpcUaTelemetry.DeploymentApplyDurationSec.Record(sw.Elapsed.TotalSeconds); _applyingDeploymentId = null; Become(Steady); @@ -1720,9 +1728,13 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers // 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 = ReconcileDrivers(deploymentId); + 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 @@ -1732,7 +1744,7 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers // 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) + 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"; @@ -1761,11 +1773,16 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers // 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. - PushDesiredSubscriptionsFromArtifact(deploymentId, appliedBlob); - CacheAppliedArtifact(deploymentId, revision, appliedBlob); + 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 @@ -1785,6 +1802,7 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers } finally { + parsed?.Dispose(); OtOpcUaTelemetry.DeploymentApplyDurationSec.Record(sw.Elapsed.TotalSeconds); _applyingDeploymentId = null; Become(Steady); @@ -1803,18 +1821,36 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers /// loaded at all. /// /// - /// Returning the blob rather than swallowing it is load-bearing for the artifact cache. This - /// method catches its own DB failures, logs a warning and returns WITHOUT rethrowing — so - /// proceeds to its success path, ACKs Applied and logs success - /// having spawned zero drivers. A cache write that trusted "the apply succeeded" would then - /// persist an empty artifact as this node's last-known-good configuration, and the node would - /// later boot from it into an empty address space. The caller distinguishes the two cases by - /// this return value. + /// + /// The bootstrap-restore entry point (). The apply paths use + /// , 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 + /// method catches its own DB failures, logs a warning and returns WITHOUT rethrowing — so the + /// caller proceeds to its success path having spawned zero drivers. A cache write that trusted + /// "the apply succeeded" would then persist an empty artifact as this node's last-known-good + /// configuration, and the node would later boot from it into an empty address space. The caller + /// distinguishes the two cases by this return value. + /// /// private byte[]? ReconcileDrivers(DeploymentId deploymentId) + { + var blob = LoadArtifactBlob(deploymentId); + return blob is null ? null : ReconcileDriversFromBlob(deploymentId, blob); + } + + /// + /// The ConfigDb read half of , shared with the Direct apply path + /// () so both read the artifact exactly once and log the + /// same way when they cannot. + /// + /// 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) { // Phase 4 belt-and-suspenders: this ConfigDb read is Direct-only (the FetchAndCache apply uses - // ReconcileDriversFromBlob with bytes in hand). A null factory can't read, so return null — the + // ReconcileDriversFromParsed with bytes in hand). A null factory can't read, so return null — the // caller treats that as "artifact could not be read" (#486) and keeps last-known-good. if (_dbFactory is null) { @@ -1822,11 +1858,10 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers return null; } - byte[] blob; try { using var db = _dbFactory.CreateDbContext(); - blob = db.Deployments.AsNoTracking() + return db.Deployments.AsNoTracking() .Where(d => d.DeploymentId == deploymentId.Value) .Select(d => d.ArtifactBlob) .FirstOrDefault() ?? Array.Empty(); @@ -1837,35 +1872,104 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers _localNode, deploymentId); return null; } - - return ReconcileDriversFromBlob(deploymentId, blob); } /// - /// The blob-processing half of , split out so the - /// FetchAndCache apply path (which already holds the fetched bytes) and the - /// boot-from-cache path can reconcile without a ConfigDb read. + /// An apply's artifact: the bytes that were reconciled plus the ONE + /// every later consumer of that apply reads. Both are + /// when nothing was applied (#485/#486). + /// + /// The reconciled artifact bytes, or null when there was no usable artifact. + /// The parsed artifact, owned by the apply handler — dispose before it returns. + 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 reconciled blob + its parsed artifact, or a default (both null) when the blob was + /// empty (#485). + private ReconciledArtifact ReconcileDriversFromParsed(DeploymentId deploymentId, byte[]? blob) + { + blob ??= Array.Empty(); + if (blob.Length == 0) + { + WarnEmptyArtifactSkippingReconcile(deploymentId); + return default; + } + + // Opened here, NOT inside the try below: DeploymentArtifact.Open throws on malformed bytes and + // that throw must reach the apply handler's catch (Failed ACK), not be mistaken for a reconcile + // failure. Ownership transfers to the caller on success; a throw from the reconcile itself + // disposes it here, since the caller never saw it. + var parsed = DeploymentArtifact.Open(blob); + try + { + ApplyDriverSpecs(parsed.ParseDriverInstances(_localNode.Value)); + } + catch + { + parsed.Dispose(); + throw; + } + + return new ReconciledArtifact(blob, parsed); + } + + /// + /// The blob-processing half of , split out so the boot-from-cache + /// and bootstrap-restore paths can reconcile from bytes with the lenient + /// "malformed ⇒ no specs" parse those paths have always had. The apply paths use + /// instead, so they parse once and keep the parse. /// /// The reconciled blob, or when it was empty (#485). private byte[]? ReconcileDriversFromBlob(DeploymentId deploymentId, byte[]? blob) { blob ??= Array.Empty(); - - // Issue #485 — no bytes is no answer, not "a configuration with no drivers". Parsing zero bytes - // yields zero specs, and DriverSpawnPlanner then plans EVERY running child for StopChild: a missing - // deployment row (or a half-written one) silently takes this node's entire field I/O down while the - // apply still ACKs Applied. Nothing legitimate produces a zero-length blob — deleting the last - // driver still deploys a JSON document with empty arrays — so treat it exactly like the load - // failure above and leave the running children alone. if (blob.Length == 0) { - _log.Warning( - "DriverHost {Node}: artifact for {Id} read back empty; skipping reconcile and keeping the {Count} running driver(s)", - _localNode, deploymentId, _children.Count); + WarnEmptyArtifactSkippingReconcile(deploymentId); return null; } - var specs = DeploymentArtifact.ParseDriverInstances(blob, _localNode.Value); + ApplyDriverSpecs(DeploymentArtifact.ParseDriverInstances(blob, _localNode.Value)); + return blob; + } + + /// + /// Issue #485 — no bytes is no answer, not "a configuration with no drivers". Parsing zero bytes + /// yields zero specs, and DriverSpawnPlanner then plans EVERY running child for StopChild: a + /// missing deployment row (or a half-written one) silently takes this node's entire field I/O down + /// while the apply still ACKs Applied. Nothing legitimate produces a zero-length blob — deleting the + /// last driver still deploys a JSON document with empty arrays — so treat it exactly like a load + /// failure and leave the running children alone. + /// + private void WarnEmptyArtifactSkippingReconcile(DeploymentId deploymentId) => + _log.Warning( + "DriverHost {Node}: artifact for {Id} read back empty; skipping reconcile and keeping the {Count} running driver(s)", + _localNode, deploymentId, _children.Count); + + /// Spawn / delta / stop the driver children to match . + /// The driver-instance specs this node should be running. + private void ApplyDriverSpecs(IReadOnlyList specs) + { var snapshots = _children.ToDictionary( kv => kv.Key, kv => new DriverChildSnapshot(kv.Value.DriverType, kv.Value.LastConfigJson, kv.Value.ResilienceConfig), @@ -1883,8 +1987,6 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers // sequenced AFTER ReinitializeAsync rebuilds DependencyRefs — a synchronous re-register HERE would // re-read the STALE ref set (ApplyDelta runs asynchronously on the child), so it is deliberately NOT // done inline. - - return blob; } /// @@ -1912,7 +2014,14 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers /// dead-letters after the Become(Steady) in the enclosing finally. /// /// - private void CacheAppliedArtifact(DeploymentId deploymentId, RevisionHash revision, byte[]? blob) + /// The deployment whose artifact is being cached. + /// The revision hash to store alongside the bytes. + /// The applied artifact BYTES — what actually gets stored. + /// The apply's already-parsed artifact, when the caller has one, so the cluster + /// scope is read off that parse instead of re-parsing the same bytes. Null on the boot/restore paths, + /// which fall back to the one-shot static resolve. + private void CacheAppliedArtifact( + DeploymentId deploymentId, RevisionHash revision, byte[]? blob, ParsedDeploymentArtifact? parsed = null) { if (_deploymentArtifactCache is null) return; @@ -1931,7 +2040,9 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers // The node's ClusterId is carried inside the artifact (ClusterNode rows), not in config // or on IClusterRoleInfo. A single-cluster or unscoped artifact has none, so it shares // one sentinel key — the pair still converges on a single pointer row either way. - var scope = DeploymentArtifact.ResolveClusterScope(blob, _localNode.Value); + var scope = parsed is not null + ? parsed.ResolveClusterScope(_localNode.Value) + : DeploymentArtifact.ResolveClusterScope(blob, _localNode.Value); var clusterId = scope.Mode == ClusterFilterMode.ScopeTo && !string.IsNullOrWhiteSpace(scope.ClusterId) ? scope.ClusterId : SingleClusterCacheKey; @@ -1998,15 +2109,15 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers /// /// /// The bootstrap-restore entry point only. Neither apply path reaches it any more: both - /// and already hold the artifact bytes - /// and call directly, so this ConfigDb read exists - /// for , which restores from a deployment id recovered out of + /// 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 /// NodeDeploymentState and may hold no blob of its own. /// private void PushDesiredSubscriptions(DeploymentId deploymentId) { // Phase 4 belt-and-suspenders: a ConfigDb read, so it can only be the restore path on a node that - // has one (every apply path calls PushDesiredSubscriptionsFromArtifact with bytes in hand). + // has one (every apply path calls PushDesiredSubscriptionsFromParsed with its parse in hand). // No ConfigDb ⇒ nothing to load. if (_dbFactory is null) { @@ -2065,6 +2176,44 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers return; } + ApplyDesiredSubscriptions(composition); + } + + /// + /// The apply paths' SubscribeBulk entry point: reads the composition off the apply's SINGLE parse + /// instead of re-parsing the same bytes a second time. + /// + /// + /// No empty-artifact guard here, unlike : a + /// only exists for a non-empty blob (the apply's reconcile + /// already applied that guard and never opened one otherwise). + /// + /// The deployment being applied — diagnostics only. + /// The apply's parsed artifact. + private void PushDesiredSubscriptionsFromParsed(DeploymentId deploymentId, ParsedDeploymentArtifact parsed) + { + AddressSpaceComposition composition; + try + { + composition = parsed.ParseComposition(_localNode.Value); + } + catch (Exception ex) + { + _log.Warning(ex, "DriverHost {Node}: failed to parse composition for SubscribeBulk ({Id})", _localNode, deploymentId); + return; + } + + ApplyDesiredSubscriptions(composition); + } + + /// + /// Hand each running driver child its desired value + alarm subscription set for + /// , rebuild the live-value/write routing maps, and forward the + /// Equipment VirtualTag + ScriptedAlarm plans to their hosts. + /// + /// The composition this node should serve. + private void ApplyDesiredSubscriptions(AddressSpaceComposition composition) + { // v3 Batch 4: the driver subscribes + publishes by RawPath (the wire-ref == RawPath == RawTagPlan.NodeId // per the v3 identity contract). Value-subscription set = the non-alarm raw tags' RawPaths; alarm-bearing // raw tags are Part 9 conditions, fed via the native alarm event stream (ForwardNativeAlarm), so they are diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/OpcUa/OpcUaPublishActor.cs b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/OpcUa/OpcUaPublishActor.cs index 890a0282..dd15cac4 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/OpcUa/OpcUaPublishActor.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/OpcUa/OpcUaPublishActor.cs @@ -462,18 +462,30 @@ public sealed class OpcUaPublishActor : ReceiveActor, IWithTimers return; } + // Parse the artifact ONCE for this rebuild. Every consumer below (data-plane authority, then the + // composition) reads this one JsonDocument instead of rebuilding it per question — the rebuild + // used to pay a full parse-and-throw-away per static call. Disposed when the handler returns. + // + // Open() throws on bytes that are not valid JSON, where the static wrappers swallowed to an + // EMPTY result — and an empty composition is not a benign no-op here: the planner diffs it as a + // PureRemove and the applier tears the served address space down. So the throw is the honest + // outcome, and it lands in this method's catch below: logged at Error, _lastApplied untouched, + // last-known-good still served. Same "no answer ≠ empty configuration" rule as the #485 guard + // immediately above, applied to bytes that arrived unreadable rather than absent. + using var parsed = DeploymentArtifact.Open(artifact); + // #520 stage 4 — install this artifact's API keys + ACL grants BEFORE the address-space work // below, and deliberately ahead of the `plan.IsEmpty` early-return: an artifact that revokes a // key or edits a grant without touching a single node produces an EMPTY address-space plan, so // feeding this after that return would leave a revoked key authenticating until some unrelated // structural change happened to redeploy. Authorization and topology change on their own // schedules. - FeedDataPlaneAuthority(artifact); + FeedDataPlaneAuthority(parsed, artifact); var composition = _localNode is { } ln - ? DeploymentArtifact.ParseComposition(artifact, ln.Value, + ? parsed.ParseComposition(ln.Value, inconsistency => _log.Warning("OpcUaPublish {Node}: cross-cluster binding — {Message}", ln, inconsistency)) - : DeploymentArtifact.ParseComposition(artifact); + : parsed.ParseComposition(); var plan = AddressSpacePlanner.Compute(_lastApplied, composition); if (plan.IsEmpty) @@ -560,17 +572,22 @@ public sealed class OpcUaPublishActor : ReceiveActor, IWithTimers /// the safe direction for availability but means a revocation carried by an artifact this node /// could not decode has not taken effect. It is logged at Error for exactly that reason. /// - private void FeedDataPlaneAuthority(byte[] artifact) + /// The rebuild's single parse of the artifact, read for the keys + grants. + /// The RAW artifact bytes — hashed for the revision id, which must be + /// SHA-256 of the exact blob (see below); never re-parsed here. + private void FeedDataPlaneAuthority(ParsedDeploymentArtifact parsed, byte[] artifactBytes) { if (_dataPlaneAuthority is null) return; try { var nodeId = _localNode?.Value ?? string.Empty; - var projection = DeploymentArtifact.ParseDataPlaneAuthority(artifact, nodeId); + var projection = parsed.ParseDataPlaneAuthority(nodeId); // The revision hash is SHA-256 of the artifact bytes — the same derivation ConfigComposer uses, // which is what lets a node recognise a re-applied revision without the hash riding the wire. - var revision = Convert.ToHexStringLower(System.Security.Cryptography.SHA256.HashData(artifact)); + // Hashed off the RAW bytes for exactly that reason: a re-serialisation of the parsed document + // would be a different byte sequence and would not match the deployment's RevisionHash. + var revision = Convert.ToHexStringLower(System.Security.Cryptography.SHA256.HashData(artifactBytes)); var snapshot = _dataPlaneAuthority.Apply( projection.ClusterId, revision, projection.Keys, projection.Grants, projection.ScopePaths);