diff --git a/CLAUDE.md b/CLAUDE.md index c4f46c2a..3051ac43 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -184,6 +184,11 @@ is still default-closed; what changed is which deltas it whitelists. re-diffs the same composition and re-ensures what failed, at the cost of re-emitting that apply's removes (a re-remove returns false and ratchets to a rebuild, which is the correct, bounded resync). A **rebuild** still advances even with failures. +- **…and the deploy ack now SAYS so (#601).** The Applied ack waits for `OpcUaPublishActor`'s + `RebuildCompleted` (timeout = shared `DeployTimings.ApplyDeadline` − 15 s = 105 s ⇒ degraded "outcome not reported in time", never clean) and carries an additive `Degraded`/`DegradedReason` + on both ack transports; central seals but records `NodeDeploymentState.FailureReason = "Applied degraded: …"` + (`/fleet` "Applied (degraded)"), and a re-dispatch of a degraded revision re-applies instead of short-circuiting. + See `docs/Redundancy.md` §"What an `Applied` ack means". - **Any surgical `false`/throw still ratchets to `SafeRebuild()`** — that one-way fallback is what makes every whitelist above safe. - **Announcements:** `AnnounceAddedNodes` announces surgically-added **Raw** containers/tags diff --git a/docs/Configuration.md b/docs/Configuration.md index 58cf17c3..5e254b77 100644 --- a/docs/Configuration.md +++ b/docs/Configuration.md @@ -175,6 +175,8 @@ The checked-in `appsettings*.json` files are deliberately thin: they carry only **`Grpc` mode: central dials each node, a node posts its ack to central.** `CentralCommunicationActor` fans one `NodeCommandService.Execute` call per node — the target set is every enabled, non-maintenance `ClusterNode` row that carries a `GrpcPort` (dialled at `Host`+`GrpcPort`), rebuilt every `ContactRefreshSeconds`. There is no receptionist and no single fleet-wide (or per-cluster) client object: it is a plain per-node dial. A node posts its `ApplyAck` with `GrpcApplyAckSender`, which walks `CentralGrpcEndpoints` in order and stops at the first success — an availability list, not a fan-out list, so central normally sees one ack per node per dispatch (the walk as a whole is at-least-once on a slow-then-retried endpoint; `ConfigPublishCoordinator.HandleAck` absorbs a duplicate idempotently). Both directions are **fire-and-forget, with no retry**: a failed dispatch is logged (naming the node and the gRPC status) and dropped in the dispatch second — the deploy path's 2-minute apply deadline and a node's boot-time replay (`DriverHostActor.Bootstrap()` re-applying an orphan `Applying` row) are the only recovery mechanisms, exactly as they were under `ClusterClient`. +The ack itself is sent only once the node's address-space apply outcome is known, and carries an additive `degraded` flag (`ApplyAckMessage` fields 6/7; trailing `ApplyAck.Degraded`/`DegradedReason` on DPS) — an old node's ack decodes clean, exactly as before. See [Redundancy.md § "What an `Applied` ack means"](Redundancy.md#what-an-applied-ack-means-the-degraded-apply-amendment-gitea-601) (Gitea #601). + | Key | Type | Default | Meaning | |---|---|---|---| | `GrpcPsk` | string? | `null` (unset) | Node-side only, required under `Grpc` on a driver-role node: this node's pre-shared key for its own `Cluster`, presented as the bearer on outbound acks and required on inbound `NodeCommandService` calls. Production form is `${secret:OT-MESH-PSK-}`; supply a literal only on a rig, via `MeshTransport__GrpcPsk` — never commit one. | diff --git a/docs/Redundancy.md b/docs/Redundancy.md index d031b520..e1f5d67c 100644 --- a/docs/Redundancy.md +++ b/docs/Redundancy.md @@ -585,6 +585,54 @@ Two properties are load-bearing and easy to break: 2-minute apply deadline and a node's boot-time replay (`DriverHostActor.Bootstrap()` re-applying an orphan `Applying` row) are the only recovery mechanisms, unchanged from the DPS/`ClusterClient` era. +### What an `Applied` ack means: the degraded-apply amendment (Gitea #601) + +Until #601 the node sent `ApplyAck(Applied)` the moment its revision advanced — *before* +`OpcUaPublishActor` had even dequeued the address-space rebuild — so an apply that swallowed sink faults +(the degraded non-rebuild apply that deliberately does not advance `_lastApplied`) still sealed the +deployment green fleet-wide. Now: + +- **The Applied ack waits for the address-space outcome.** `DriverHostActor` sends + `RebuildAddressSpace(ReportOutcome: true)`; `OpcUaPublishActor` replies `RebuildCompleted(Degraded, Reason)` + on every path (clean / sink faults / failed rebuild / pipeline throw / no usable artifact — an empty plan + is clean), and only then is the ack sent. It is a plain `Tell` + reply message — no `Ask`, no `await`, so + nothing resumes off the actor thread. +- **A missing outcome is degraded, never clean.** If no reply arrives within + `DriverHostActor.DefaultApplyOutcomeTimeout` the ack goes out `Applied` + degraded, with a reason starting + *"address-space apply outcome not reported in time"* — distinct from a real sink failure (which carries + `failedNodes=` / "rebuild failed" / "threw"), so an operator can tell a slow or lost apply from a broken one. + The timeout is **derived**, not a free constant: `DeployTimings.ApplyDeadline` (Commons, 2 min — the same + value `ConfigPublishCoordinator.DefaultApplyDeadline` reads) minus `DeployTimings.ApplyOutcomeSafetyMargin` + (15 s) = **105 s**, so a slow-but-clean rebuild of a large address space is not labelled degraded and the + ack still reaches central before its deadline. Change the deadline there and both sides move together. + Accepted edge: the timer starts after the artifact read + driver reconcile, so if those alone exceed the + 15 s margin a never-answered apply can still time the deployment out. A node with no publish actor wired + acks clean at once, as before. +- **The ack carries an ADDITIVE degraded flag on both transports.** `ApplyAck` gained trailing + `Degraded = false, DegradedReason = null` (an old node's DPS JSON decodes clean); `ApplyAckMessage` gained + `bool degraded = 6; optional string degraded_reason = 7;` (an old node's gRPC ack decodes clean). The + outcome stays `APPLIED` — a degraded node *did* apply the revision. +- **Central records it without a status-model change.** `ConfigPublishCoordinator.PersistNodeAck` keeps the + node's `NodeDeploymentState.Status = Applied` and writes `FailureReason = "Applied degraded: "` + (`ConfigPublishCoordinator.IsDegraded` = Applied **and** a reason). The deployment still **seals** — + every node runs the revision, and "latest sealed" drives drift detection and the publish actor's fallback, + so leaving it unsealed would show the fleet a revision behind what it runs — but `Deployment.FailureReason` + names the degraded nodes and the seal logs at Warning. Recovered from the rows on singleton failover. A + degraded ack alongside a `Failed` one is still `PartiallyFailed`. `/fleet` renders the node as + **Applied (degraded)** (caution chip); `/deployments` renders **Sealed (degraded)** with the reason as a tooltip. +- **Re-deploying a degraded revision repairs it.** The node's idempotent "already at this revision" short- + circuit is skipped while the current revision's apply is degraded or its outcome is still pending — the + dispatch re-applies instead, the publish actor re-diffs the un-advanced composition and re-ensures the + missing nodes, and the ack reports the new outcome. A degraded **rebuild** is the other shape: it DOES + advance `_lastApplied`, so the identical re-deploy diffs to an empty plan — the publish actor remembers + that the last advance was degraded and, on that empty plan, re-ensures the whole composition (idempotent, + no second rebuild), reporting clean only if that pass is clean. +- **A node with a ConfigDb also writes the reason on its own row** (`ApplyAck.DegradedReasonPrefix`), because + central only records it when it *accepts* the ack — a timed-out deployment's late ack is discarded. +- **Mixed-version fleets.** An old node never sets the flag, so its acks are treated exactly as before + (clean). A new node acking an old central: the old coordinator ignores the unknown field and seals as it + always did — no worse than today. + Phase 2 changes no redundancy semantics: the Primary gate, ServiceLevel, and the singleton layout are untouched, and `redundancy-state` itself stays on DPS (it is bidirectional and built from `Cluster.State`, which makes it genuinely mesh-bound — the hardest remaining dependency before the diff --git a/src/Core/ZB.MOM.WW.OtOpcUa.Commons/Messages/Deploy/ApplyAck.cs b/src/Core/ZB.MOM.WW.OtOpcUa.Commons/Messages/Deploy/ApplyAck.cs index 10ff6f7d..2431bd41 100644 --- a/src/Core/ZB.MOM.WW.OtOpcUa.Commons/Messages/Deploy/ApplyAck.cs +++ b/src/Core/ZB.MOM.WW.OtOpcUa.Commons/Messages/Deploy/ApplyAck.cs @@ -7,9 +7,32 @@ public enum ApplyAckOutcome { Applied, Failed } /// /// Per-node acknowledgment returned by DriverHostActor to the dispatching coordinator. /// +/// The deployment being acknowledged. +/// The acknowledging node. +/// Whether the node applied the deployment. +/// Why the apply failed; null on . +/// The dispatch's correlation id. +/// +/// Gitea #601: when the node applied the deployment (the revision advanced) but +/// its OPC UA address-space apply was DEGRADED — swallowed per-node sink faults, a failed rebuild, or no +/// outcome reported in time — so the served address space may be partial. Only meaningful with +/// . Trailing and defaulted so an ack from a node built before +/// this field existed (DPS JSON or gRPC without field 6) decodes exactly as it always did: clean. +/// +/// A short, operator-readable description of the degradation, or null. public sealed record ApplyAck( DeploymentId DeploymentId, NodeId NodeId, ApplyAckOutcome Outcome, string? FailureReason, - CorrelationId CorrelationId); + CorrelationId CorrelationId, + bool Degraded = false, + string? DegradedReason = null) +{ + /// + /// Gitea #601 — prefix of NodeDeploymentState.FailureReason on an Applied row whose + /// address-space apply was degraded. Shared by central (which writes it from the ack) and a node with + /// a ConfigDb (which writes its own row), so both spell the row identically. + /// + public const string DegradedReasonPrefix = "Applied degraded: "; +} diff --git a/src/Core/ZB.MOM.WW.OtOpcUa.Commons/Messages/Deploy/DeployTimings.cs b/src/Core/ZB.MOM.WW.OtOpcUa.Commons/Messages/Deploy/DeployTimings.cs new file mode 100644 index 00000000..1f0866de --- /dev/null +++ b/src/Core/ZB.MOM.WW.OtOpcUa.Commons/Messages/Deploy/DeployTimings.cs @@ -0,0 +1,25 @@ +namespace ZB.MOM.WW.OtOpcUa.Commons.Messages.Deploy; + +/// +/// The deploy-path timing contract shared by central (ConfigPublishCoordinator) and the node +/// (DriverHostActor). One place, so the node's ack-hold timeout can never drift past the deadline +/// central times the deployment out on. +/// +public static class DeployTimings +{ + /// How long central waits for every expected node's ack before sealing the deployment TimedOut + /// (ConfigPublishCoordinator.DefaultApplyDeadline). + public static readonly TimeSpan ApplyDeadline = TimeSpan.FromMinutes(2); + + /// Gitea #601 — how far inside a node's held Applied ack is released + /// when the address-space outcome has not arrived, leaving room for transport latency. + public static readonly TimeSpan ApplyOutcomeSafetyMargin = TimeSpan.FromSeconds(15); + + /// Gitea #601 — the node's default apply-outcome timeout: − + /// (105 s). Long enough that a slow-but-clean rebuild of a large + /// address space is not labelled degraded, short enough to ack before central gives up. + /// The timer is armed after the artifact read and driver reconcile, so if those alone took + /// longer than the margin the ack can still land after the deadline — accepted: they are seconds in + /// practice, and lengthening the margin would label more slow-but-clean rebuilds degraded. + public static TimeSpan ApplyOutcomeTimeout => ApplyDeadline - ApplyOutcomeSafetyMargin; +} diff --git a/src/Core/ZB.MOM.WW.OtOpcUa.Commons/Messages/Mesh/MeshTransportDtoMapper.cs b/src/Core/ZB.MOM.WW.OtOpcUa.Commons/Messages/Mesh/MeshTransportDtoMapper.cs index f3d78255..4fe43816 100644 --- a/src/Core/ZB.MOM.WW.OtOpcUa.Commons/Messages/Mesh/MeshTransportDtoMapper.cs +++ b/src/Core/ZB.MOM.WW.OtOpcUa.Commons/Messages/Mesh/MeshTransportDtoMapper.cs @@ -216,6 +216,11 @@ public static class MeshTransportDtoMapper if (message.FailureReason is not null) proto.FailureReason = message.FailureReason; + // #601 — additive fields; a clean ack leaves both unset, byte-identical to a pre-#601 encode. + proto.Degraded = message.Degraded; + if (message.DegradedReason is not null) + proto.DegradedReason = message.DegradedReason; + return proto; } @@ -249,7 +254,10 @@ public static class MeshTransportDtoMapper NodeId.Parse(proto.NodeId), outcome, proto.HasFailureReason ? proto.FailureReason : null, - CorrelationId.Parse(proto.CorrelationId)); + CorrelationId.Parse(proto.CorrelationId), + // #601 — absent on an old node's ack, which decodes as false/null: today's clean ack. + Degraded: proto.Degraded, + DegradedReason: proto.HasDegradedReason ? proto.DegradedReason : null); } // ---------------------------------------------------------------- Envelope diff --git a/src/Core/ZB.MOM.WW.OtOpcUa.Commons/Protos/mesh_transport.proto b/src/Core/ZB.MOM.WW.OtOpcUa.Commons/Protos/mesh_transport.proto index 1dd09580..848d45d2 100644 --- a/src/Core/ZB.MOM.WW.OtOpcUa.Commons/Protos/mesh_transport.proto +++ b/src/Core/ZB.MOM.WW.OtOpcUa.Commons/Protos/mesh_transport.proto @@ -105,6 +105,11 @@ message ApplyAckMessage { ApplyOutcome outcome = 3; optional string failure_reason = 4; // nullable in the record — presence distinguishes null from "" string correlation_id = 5; // CorrelationId (Guid) in its "N" hex form + // Gitea #601 — ADDITIVE. True when the node applied the deployment but its address-space apply was + // degraded (sink faults, failed rebuild, or no outcome in time). An old node never sets it, so it decodes + // as false: exactly today's clean ack. Only meaningful with APPLY_OUTCOME_APPLIED. + bool degraded = 6; + optional string degraded_reason = 7; // short operator-readable reason; presence = set } // Mirrors ZB.MOM.WW.OtOpcUa.Commons.Messages.Deploy.ApplyAckOutcome, shifted by one so 0 stays diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.AdminUI/Components/Pages/Deployments.razor b/src/Server/ZB.MOM.WW.OtOpcUa.AdminUI/Components/Pages/Deployments.razor index 69fd15ba..01df5ffd 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.AdminUI/Components/Pages/Deployments.razor +++ b/src/Server/ZB.MOM.WW.OtOpcUa.AdminUI/Components/Pages/Deployments.razor @@ -57,7 +57,8 @@ @Short(d.DeploymentId) @DisplayText.Abbreviate(d.RevisionHash, 12) - @d.Status + @* #601: Sealed + a reason = sealed with a degraded address-space apply on some node(s). *@ + @(d.Status == DeploymentStatus.Sealed && !string.IsNullOrEmpty(d.FailureReason) ? "Sealed (degraded)" : d.Status.ToString()) @d.CreatedBy @d.CreatedAtUtc.ToString("u") @(d.SealedAtUtc?.ToString("u") ?? "—") @@ -77,7 +78,8 @@ DeploymentStatus Status, string CreatedBy, DateTime CreatedAtUtc, - DateTime? SealedAtUtc); + DateTime? SealedAtUtc, + string? FailureReason); private IReadOnlyList _deployments = Array.Empty(); private bool _busy; @@ -102,7 +104,8 @@ d.Status, d.CreatedBy, d.CreatedAtUtc, - d.SealedAtUtc)) + d.SealedAtUtc, + d.FailureReason)) .Take(50) .ToListAsync(); diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.AdminUI/Components/Pages/Fleet.razor b/src/Server/ZB.MOM.WW.OtOpcUa.AdminUI/Components/Pages/Fleet.razor index 6e91c695..76cd5db1 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.AdminUI/Components/Pages/Fleet.razor +++ b/src/Server/ZB.MOM.WW.OtOpcUa.AdminUI/Components/Pages/Fleet.razor @@ -88,7 +88,7 @@ else @role } - @StatusLabel(r.Status) + @StatusLabel(r.Status, r.FailureReason) @(r.AppliedAtUtc?.ToString("u") ?? "—") @(r.StartedAtUtc?.ToString("u") ?? "—") @(string.IsNullOrEmpty(r.FailureReason) ? "" : DisplayText.Abbreviate(r.FailureReason, 80)) @@ -184,15 +184,22 @@ else } } - private static string StatusChipClass(NodeDeploymentStatus? status) => status switch - { - NodeDeploymentStatus.Applied => "chip-ok", - NodeDeploymentStatus.Applying => "chip-caution", - NodeDeploymentStatus.Failed => "chip-alert", - _ => "chip-idle", - }; + // #601 — an Applied row carrying a reason is a DEGRADED apply (the node runs the revision but its address + // space may be partial); it must not render as the same green chip as a clean apply. + private static bool IsDegraded(NodeDeploymentStatus? status, string? failureReason) => + status is { } s && ZB.MOM.WW.OtOpcUa.ControlPlane.Coordinators.ConfigPublishCoordinator.IsDegraded(s, failureReason); - private static string StatusLabel(NodeDeploymentStatus? status) => status?.ToString() ?? "waiting"; + private static string StatusChipClass(NodeDeploymentStatus? status, string? failureReason) => + IsDegraded(status, failureReason) ? "chip-caution" : status switch + { + NodeDeploymentStatus.Applied => "chip-ok", + NodeDeploymentStatus.Applying => "chip-caution", + NodeDeploymentStatus.Failed => "chip-alert", + _ => "chip-idle", + }; + + private static string StatusLabel(NodeDeploymentStatus? status, string? failureReason) => + IsDegraded(status, failureReason) ? "Applied (degraded)" : status?.ToString() ?? "waiting"; public async ValueTask DisposeAsync() { diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.ControlPlane/Coordinators/ConfigPublishCoordinator.cs b/src/Server/ZB.MOM.WW.OtOpcUa.ControlPlane/Coordinators/ConfigPublishCoordinator.cs index b52beb87..32b31116 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.ControlPlane/Coordinators/ConfigPublishCoordinator.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.ControlPlane/Coordinators/ConfigPublishCoordinator.cs @@ -57,7 +57,9 @@ public sealed class ConfigPublishCoordinator : ReceiveActor, IWithTimers { public const string DeploymentsTopic = "deployments"; public const string DeploymentAcksTopic = "deployment-acks"; - public static readonly TimeSpan DefaultApplyDeadline = TimeSpan.FromMinutes(2); + /// Central's apply deadline — the shared , which the node's + /// apply-outcome timeout is derived from (Gitea #601). + public static readonly TimeSpan DefaultApplyDeadline = DeployTimings.ApplyDeadline; private readonly IDbContextFactory _dbFactory; private readonly TimeSpan _applyDeadline; @@ -83,6 +85,30 @@ public sealed class ConfigPublishCoordinator : ReceiveActor, IWithTimers private static readonly IEqualityComparer NodeIdComparer = new CaseInsensitiveNodeIdComparer(); private readonly Dictionary _acks = new(NodeIdComparer); + /// Gitea #601 — the expected nodes whose Applied ack was DEGRADED (address-space apply partial). + /// Recovered from the DB on failover via . + private readonly HashSet _degraded = new(NodeIdComparer); + + /// + /// Gitea #601 — prefix of NodeDeploymentState.FailureReason on an Applied row whose apply + /// was degraded. The status model is deliberately unchanged (no new enum value, no migration): a + /// degraded node DID apply the revision, so its status stays + /// and the degradation rides the existing reason column, which /fleet already renders. + /// + public const string DegradedReasonPrefix = ApplyAck.DegradedReasonPrefix; + + /// + /// Gitea #601 — whether a NodeDeploymentState row records a degraded apply: status + /// with a non-empty FailureReason. A clean Applied row + /// never carries a reason (neither this coordinator nor DriverHostActor writes one), so an + /// old node's row reads as clean exactly as before. + /// + /// The row's status. + /// The row's failure reason. + /// for an Applied-but-degraded row. + public static bool IsDegraded(NodeDeploymentStatus status, string? failureReason) => + status == NodeDeploymentStatus.Applied && !string.IsNullOrEmpty(failureReason); + private DeploymentId? _current; private HashSet _expectedAcks = new(NodeIdComparer); @@ -149,9 +175,12 @@ public sealed class ConfigPublishCoordinator : ReceiveActor, IWithTimers _expectedAcks = nodeStates.Select(s => NodeId.Parse(s.NodeId)).ToHashSet(NodeIdComparer); foreach (var s in nodeStates.Where(s => s.Status != NodeDeploymentStatus.Applying)) + { _acks[NodeId.Parse(s.NodeId)] = s.Status == NodeDeploymentStatus.Applied ? ApplyAckOutcome.Applied : ApplyAckOutcome.Failed; + if (IsDegraded(s.Status, s.FailureReason)) _degraded.Add(NodeId.Parse(s.NodeId)); + } // Resume the deadline timer using the remaining time. The deadline runs from when the // deployment was first marked AwaitingApplyAcks (Deployment.CreatedAtUtc is a close enough @@ -174,6 +203,7 @@ public sealed class ConfigPublishCoordinator : ReceiveActor, IWithTimers { _current = msg.DeploymentId; _acks.Clear(); + _degraded.Clear(); // Seed NodeDeploymentState rows so a failover knows which nodes were expected to ack. using (var db = _dbFactory.CreateDbContext()) @@ -237,6 +267,9 @@ public sealed class ConfigPublishCoordinator : ReceiveActor, IWithTimers } _acks[msg.NodeId] = msg.Outcome; + // #601 — an old node never sets Degraded, so its ack takes the clean branch exactly as before. + if (IsDegradedAck(msg)) _degraded.Add(msg.NodeId); + else _degraded.Remove(msg.NodeId); PersistNodeAck(msg); if (_acks.Count < _expectedAcks.Count) return; @@ -258,15 +291,51 @@ public sealed class ConfigPublishCoordinator : ReceiveActor, IWithTimers ? NodeDeploymentStatus.Applied : NodeDeploymentStatus.Failed; row.AppliedAtUtc = DateTime.UtcNow; - row.FailureReason = msg.FailureReason; + // #601 — an Applied+degraded ack keeps status Applied and records WHY in the reason column, so the row + // reads as degraded (IsDegraded) rather than clean. A clean or Failed ack writes what it always did. + row.FailureReason = IsDegradedAck(msg) + ? Truncate(DegradedReasonPrefix + (string.IsNullOrWhiteSpace(msg.DegradedReason) + ? "address-space apply degraded (no reason given)" + : msg.DegradedReason)) + : msg.FailureReason; db.SaveChanges(); } + private static bool IsDegradedAck(ApplyAck msg) => msg.Outcome == ApplyAckOutcome.Applied && msg.Degraded; + + // FailureReason is nvarchar(2048) on both tables (OtOpcUaConfigDbContext); a reason built from a node's + // free text must never fail the SaveChanges that records the ack. + private const int FailureReasonMaxLength = 2048; + + private static string Truncate(string s) => + s.Length <= FailureReasonMaxLength ? s : s[..(FailureReasonMaxLength - 1)] + "…"; + private void SealDeployment() { if (_current is null) return; using var db = _dbFactory.CreateDbContext(); UpdateDeploymentStatus(db, _current.Value, DeploymentStatus.Sealed, sealNow: true); + if (_degraded.Count > 0) + { + // #601 — still Sealed: every node DID apply this revision and is running it, and "latest sealed" is + // what drift detection and the publish actor's fallback key on, so a degraded deploy that was not + // sealed would make the fleet look one revision behind what it actually runs. Sealed is not the + // same as clean, though: the deployment's reason names the degraded nodes (per-node detail is on + // each NodeDeploymentState row) and the seal logs at Warning. + var nodes = _degraded.Select(n => n.Value).Order(StringComparer.OrdinalIgnoreCase).ToList(); + var d = db.Deployments.FirstOrDefault(x => x.DeploymentId == _current.Value.Value); + if (d is not null) + d.FailureReason = Truncate( + $"Sealed with a degraded address-space apply on {nodes.Count} node(s): {string.Join(", ", nodes)}. " + + "See each node's NodeDeploymentState.FailureReason; re-deploying the same revision re-applies it."); + db.SaveChanges(); + _log.Warning( + "Deployment {Id} sealed DEGRADED (acks={Count}) — the address-space apply was degraded on: {Nodes}", + _current.Value, _acks.Count, string.Join(", ", nodes)); + ResetForNext(); + return; + } + db.SaveChanges(); _log.Info("Deployment {Id} sealed (acks={Count})", _current.Value, _acks.Count); ResetForNext(); @@ -318,6 +387,7 @@ public sealed class ConfigPublishCoordinator : ReceiveActor, IWithTimers _current = null; _expectedAcks.Clear(); _acks.Clear(); + _degraded.Clear(); } private const string DeadlineTimerKey = "apply-deadline"; 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 b287ca3c..c131024d 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/DriverHostActor.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/DriverHostActor.cs @@ -84,6 +84,25 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS /// public static readonly TimeSpan DefaultApplyWedgeTimeout = TimeSpan.FromMinutes(5); + /// + /// Gitea #601 — how long an Applied ack waits for OpcUaPublishActor's address-space apply + /// outcome () before it is sent anyway. + /// + /// + /// Derived from central's apply deadline (, 2 min — the same + /// constant ConfigPublishCoordinator.DefaultApplyDeadline reads) minus + /// (15 s) = 105 s: long enough that a + /// slow-but-clean rebuild of a large address space is not labelled degraded, short enough that the ack + /// still lands before central times the deployment out. An ack sent on this timeout is marked + /// degraded with a reason starting , never clean — + /// the revision did advance, but nobody has confirmed the address space matches it. + /// + public static readonly TimeSpan DefaultApplyOutcomeTimeout = DeployTimings.ApplyOutcomeTimeout; + + /// Gitea #601 — the reason prefix of a degraded ack sent because the outcome TIMED OUT, so an + /// operator can tell "not reported in time" from a real sink failure. + public const string OutcomeNotReportedReasonPrefix = "address-space apply outcome not reported in time"; + /// Minimum interval between two supervision reprovisions of the SAME driver child (archreview /// review F3). Akka's default decider restarts a faulting child immediately and without limit, and the /// reprovision hook makes every restart do ShutdownAsync + InitializeAsync + SubscribeAsync — so an @@ -244,6 +263,21 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS private IActorRef? _scriptedAlarmHost; private RevisionHash? _currentRevision; + + /// + /// Gitea #601 — the last address-space apply of reported DEGRADED (or no + /// outcome in time). A re-dispatch of that same revision then RE-APPLIES instead of taking the + /// idempotent "already at this revision" short-circuit: the short-circuit would ack a clean Applied + /// for an address space known to be partial, and the re-apply is what repairs it (a degraded + /// non-rebuild apply leaves the publish actor's composition un-advanced, so it re-diffs the missing + /// nodes). Reset by every clean outcome and by every boot path that restores a revision. + /// + private bool _currentRevisionDegraded; + + /// Gitea #601 — Applied acks held until the address-space apply reports its outcome, keyed by + /// the apply's correlation id (echoed back in ). + private readonly Dictionary _pendingAcks = new(); + private DeploymentId? _applyingDeploymentId; private readonly Dictionary _children = new(StringComparer.Ordinal); @@ -431,6 +465,10 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS /// private readonly Func? _sharesHistorizationScopeWithLocal; + /// Gitea #601 — how long an Applied ack waits for the address-space outcome (see + /// ). + private readonly TimeSpan _applyOutcomeTimeout; + /// This host's -wedge watchdog delay (see /// ). private readonly TimeSpan _applyWedgeTimeout; @@ -595,6 +633,9 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS /// Archreview 03/S15: how long after a driver reconnects the /// recovery audit waits before reporting refs still latched BadNoCommunication. Defaults to /// ; tests shorten it. + /// Gitea #601: how long an Applied ack waits for the address-space + /// apply outcome before it is sent anyway, marked degraded ("outcome unknown"). Defaults to + /// ; tests shorten it. /// The Akka.NET used to spawn this actor. public static Props Props( IDbContextFactory? dbFactory, @@ -626,7 +667,8 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS TimeSpan? applyWedgeTimeout = null, TimeSpan? reprovisionBackoff = null, TimeSpan? commsRecoveryAuditDelay = null, - Func? sharesAlarmQueueWithLocal = null) => + Func? sharesAlarmQueueWithLocal = null, + TimeSpan? applyOutcomeTimeout = null) => // WARNING: this forwarding list is POSITIONAL, and Props.Create compiles it into an // expression tree. Six IActorRef? parameters and several interface-typed ones mean a // mis-ordered argument is usually type-compatible and therefore compiles clean, then binds @@ -639,7 +681,7 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS deploymentArtifactCache, redundancyRoleView, replicationPeerHost, fetchAndCacheMode, artifactFetcher, alarmStateStore, telemetryHub, primaryRoleSource, historizationRoleView, sharesHistorizationScopeWithLocal, applyWedgeTimeout, reprovisionBackoff, commsRecoveryAuditDelay, - sharesAlarmQueueWithLocal)); + sharesAlarmQueueWithLocal, applyOutcomeTimeout)); /// Initializes a new DriverHostActor with the specified dependencies. /// Database context factory for configuration database access. @@ -737,8 +779,10 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS TimeSpan? applyWedgeTimeout = null, TimeSpan? reprovisionBackoff = null, TimeSpan? commsRecoveryAuditDelay = null, - Func? sharesAlarmQueueWithLocal = null) + Func? sharesAlarmQueueWithLocal = null, + TimeSpan? applyOutcomeTimeout = null) { + _applyOutcomeTimeout = applyOutcomeTimeout ?? DefaultApplyOutcomeTimeout; _sharesAlarmQueueWithLocal = sharesAlarmQueueWithLocal; _applyWedgeTimeout = applyWedgeTimeout ?? DefaultApplyWedgeTimeout; _reprovisionBackoff = reprovisionBackoff ?? DefaultReprovisionBackoff; @@ -1288,6 +1332,10 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS private void Steady() { Receive(HandleDispatchFromSteady); + // #601 — the held Applied ack is released by the address-space outcome (or its timeout); both can + // land in any behaviour, so all three register them. + Receive(HandleRebuildCompleted); + Receive(HandleApplyOutcomeTimedOut); Receive(HandleArtifactCached); Receive(HandleArtifactCacheFailed); // A wedge tick already in the mailbox when the apply finished — dropped at Debug, see the handler. @@ -1348,6 +1396,8 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS // Direct ConfigDb read (BeginFetchAndCacheApply / BeginDirectApply ran in Steady, then // Become(Applying)); completing the apply transitions back to Steady. Receive(HandleArtifactForApply); + Receive(HandleRebuildCompleted); + Receive(HandleApplyOutcomeTimedOut); // 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); @@ -2180,6 +2230,10 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS // The redundancy-state topic also carries OpcUaProbeResult (OpcUaPublishActor peer-probes). // We don't consume it here — drop it so it doesn't dead-letter (matches PeerProbeSupervisor). Receive(_ => { }); + // #601 — an apply that completed just before the host went Stale still owes its ack; the + // coordinator is not behind the ConfigDb, so release it exactly as in Steady. + Receive(HandleRebuildCompleted); + Receive(HandleApplyOutcomeTimedOut); // An inbound operator write can't be serviced while the config DB is unreachable — fast-fail so the // node-manager's bounded Ask gets an immediate clear status instead of dead-lettering into a timeout. Receive(_ => @@ -2226,7 +2280,12 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS private void HandleDispatchFromSteady(DispatchDeployment msg) { - if (_currentRevision is { } cur && cur == msg.RevisionHash) + // #601: the short-circuit claims a CLEAN Applied, so it is only honest while the current revision's + // address space is known-good — not after a degraded apply, and not while its outcome is still + // pending. Either way the dispatch re-applies instead, which is also what repairs a degraded apply. + var currentRevisionUnconfirmed = _currentRevisionDegraded + || _pendingAcks.Values.Any(p => p.Revision == msg.RevisionHash); + if (_currentRevision is { } cur && cur == msg.RevisionHash && !currentRevisionUnconfirmed) { // Idempotent — already at this rev. Ack and stay Steady. _log.Debug("DriverHost {Node}: dispatch {Id} matches current rev {Rev}; immediate ACK", @@ -2235,6 +2294,11 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS return; } + if (_currentRevision is { } same && same == msg.RevisionHash) + _log.Info( + "DriverHost {Node}: dispatch {Id} matches current rev {Rev}, but its address-space apply was degraded or is unconfirmed — re-applying rather than acking clean", + _localNode, msg.DeploymentId, msg.RevisionHash); + if (_fetchAndCacheMode) { BeginFetchAndCacheApply(msg.DeploymentId, msg.RevisionHash, msg.CorrelationId); @@ -2394,7 +2458,13 @@ 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); + // Gitea #601: the Applied ack is NOT sent here any more. It used to go out right now, before the + // address-space apply below had even been dequeued, so a degraded apply (swallowed sink faults, + // _lastApplied not advanced) still sealed the deployment green fleet-wide. The ack is now held + // until OpcUaPublishActor replies RebuildCompleted (HandleRebuildCompleted) or the outcome + // timeout fires (sent degraded, "outcome unknown"). With no publish actor there is no address + // space to wait for, so the ack goes out clean immediately — exactly the old behaviour. + AwaitAddressSpaceOutcomeThenAck(msg.DeploymentId, msg.Revision, msg.Correlation); // 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 @@ -2406,8 +2476,11 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS // 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. + // ReportOutcome: the reply (to Self — this Tell runs inside a receive, so the implicit sender is + // this actor) releases the held ack. Plain Tell + reply message, no Ask/await, so nothing here + // resumes off the actor thread. _opcUaPublishActor?.Tell(new ZB.MOM.WW.OtOpcUa.Runtime.OpcUa.OpcUaPublishActor.RebuildAddressSpace( - msg.Correlation, msg.DeploymentId, appliedBlob)); + msg.Correlation, msg.DeploymentId, appliedBlob, ReportOutcome: true)); // 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); @@ -2427,6 +2500,9 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS } catch (Exception ex) { + // #601: a throw after the Applied ack was put on hold — the Failed ack below is this apply's one + // answer, so the held Applied must never follow it when the address-space outcome lands. + DropPendingAck(msg.Correlation); UpsertNodeDeploymentState(msg.DeploymentId, NodeDeploymentStatus.Failed, ex.Message); SendAck(msg.DeploymentId, ApplyAckOutcome.Failed, ex.Message, msg.Correlation); OtOpcUaTelemetry.DeploymentApplied.Add(1, new KeyValuePair("outcome", "reject")); @@ -2447,6 +2523,99 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS } } + /// Gitea #601 — an Applied ack held until the address-space apply reports its outcome. + private sealed record PendingAck(DeploymentId DeploymentId, RevisionHash Revision, CorrelationId Correlation); + + /// Gitea #601 — self-timer: the address-space outcome for this correlation did not arrive in time. + private sealed record ApplyOutcomeTimedOut(CorrelationId Correlation); + + private static string ApplyOutcomeTimerKey(CorrelationId correlation) => $"apply-outcome-{correlation}"; + + /// + /// Gitea #601 — holds the Applied ack for an apply whose revision has just advanced until the + /// address-space apply outcome is known, arming as the backstop. + /// With no publish actor wired (tests, a node with no OPC UA data plane) there is no outcome to wait + /// for, so the ack is sent clean at once — byte-for-byte the pre-#601 behaviour. + /// + private void AwaitAddressSpaceOutcomeThenAck(DeploymentId deploymentId, RevisionHash revision, CorrelationId correlation) + { + if (_opcUaPublishActor is null) + { + _currentRevisionDegraded = false; + SendAck(deploymentId, ApplyAckOutcome.Applied, failureReason: null, correlation); + return; + } + + if (_pendingAcks.Remove(correlation, out var superseded)) + _log.Warning( + "DriverHost {Node}: a second apply reused correlation {Correlation} while deployment {Old}'s ack was still waiting for its address-space outcome — that ack is superseded", + _localNode, correlation, superseded.DeploymentId); + _pendingAcks[correlation] = new PendingAck(deploymentId, revision, correlation); + Timers.StartSingleTimer(ApplyOutcomeTimerKey(correlation), new ApplyOutcomeTimedOut(correlation), _applyOutcomeTimeout); + } + + /// Gitea #601 — forgets a held ack (its apply failed after all, and acks Failed instead). + private void DropPendingAck(CorrelationId correlation) + { + if (_pendingAcks.Remove(correlation)) + Timers.Cancel(ApplyOutcomeTimerKey(correlation)); + } + + /// + /// Gitea #601 — the address-space apply reported its outcome: release the held ack, clean or + /// degraded. Registered in every behaviour (Steady, Applying, Stale) because the reply can land in + /// any of them — the host returns to Steady before the publish actor has even dequeued the rebuild. + /// + private void HandleRebuildCompleted(OpcUa.OpcUaPublishActor.RebuildCompleted msg) + { + if (!_pendingAcks.Remove(msg.Correlation, out var pending)) + { + _log.Debug("DriverHost {Node}: address-space outcome for {Correlation} has no held ack (already timed out or superseded); dropping", + _localNode, msg.Correlation); + return; + } + + Timers.Cancel(ApplyOutcomeTimerKey(msg.Correlation)); + ReleasePendingAck(pending, msg.Degraded, msg.DegradedReason); + } + + /// + /// Gitea #601 — the address-space outcome did not arrive within . + /// The ack is sent anyway so the coordinator does not time the deployment out on a node that DID + /// advance its revision — but degraded, never clean: nobody has confirmed the address space. + /// + private void HandleApplyOutcomeTimedOut(ApplyOutcomeTimedOut msg) + { + if (!_pendingAcks.Remove(msg.Correlation, out var pending)) return; + + ReleasePendingAck(pending, degraded: true, + $"{OutcomeNotReportedReasonPrefix} ({_applyOutcomeTimeout.TotalSeconds:0}s) — no sink failure was observed; the apply may still be running (slow) or its outcome was lost"); + } + + private void ReleasePendingAck(PendingAck pending, bool degraded, string? reason) + { + // Only the CURRENT revision's flag is ours to set — an older apply's late outcome must not mark a + // newer revision degraded (or clean). + if (_currentRevision is { } cur && cur == pending.Revision) + _currentRevisionDegraded = degraded; + + if (degraded) + { + // #601 review — the apply wrote this node's own row Applied/null before the outcome was known. Central + // adds the reason from the ack only when it ACCEPTS the ack (a timed-out or superseded deployment's ack + // is discarded), so a node with a ConfigDb records it itself; a driver-only node no-ops here. + UpsertNodeDeploymentState(pending.DeploymentId, NodeDeploymentStatus.Applied, + ApplyAck.DegradedReasonPrefix + (reason ?? "address-space apply degraded")); + OtOpcUaTelemetry.DeploymentApplied.Add(1, new KeyValuePair("outcome", "degraded")); + _log.Warning( + "DriverHost {Node}: deployment {Id} (rev {Rev}) applied DEGRADED — {Reason}. Acking Applied+degraded; re-dispatching the same revision re-applies it", + _localNode, pending.DeploymentId, pending.Revision, reason); + } + + SendAck(pending.DeploymentId, ApplyAckOutcome.Applied, failureReason: null, pending.Correlation, + degraded, degraded ? reason ?? "address-space apply degraded" : null); + } + /// /// Self-timer telling this host it has been in for /// without finishing. @@ -3677,9 +3846,11 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS } } - private void SendAck(DeploymentId deploymentId, ApplyAckOutcome outcome, string? failureReason, CorrelationId correlation) + private void SendAck( + DeploymentId deploymentId, ApplyAckOutcome outcome, string? failureReason, CorrelationId correlation, + bool degraded = false, string? degradedReason = null) { - var ack = new ApplyAck(deploymentId, _localNode, outcome, failureReason, correlation); + var ack = new ApplyAck(deploymentId, _localNode, outcome, failureReason, correlation, degraded, degradedReason); if (_ackRouter is not null) { // Either a direct coordinator handle (tests) or, under MeshTransport:Mode = Grpc, the 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 883f8c6f..dfd1de73 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/OpcUa/OpcUaPublishActor.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/OpcUa/OpcUaPublishActor.cs @@ -100,8 +100,31 @@ public sealed class OpcUaPublishActor : ReceiveActor, IWithTimers /// that cannot succeed, leaving the rebuild a no-op and clients browsing an empty server. Null /// on the normal path, where loading from the ConfigDb by id is correct. /// + /// + /// Gitea #601 — when , the actor replies to the sender with a + /// once the apply outcome is known (clean or degraded), on EVERY path + /// through the handler. DriverHostActor sets it on the deploy path and sends the deployment's + /// ack only after that reply, so a degraded apply can no longer seal a deployment green. Off for the + /// boot/restore paths, which re-ack nothing. + /// public sealed record RebuildAddressSpace( - CorrelationId Correlation, DeploymentId? DeploymentId = null, byte[]? Artifact = null); + CorrelationId Correlation, DeploymentId? DeploymentId = null, byte[]? Artifact = null, + bool ReportOutcome = false); + + /// + /// Gitea #601 — the reply to a carrying + /// : whether the address-space apply was clean. + /// + /// The rebuild's correlation id, echoed. + /// The rebuild's deployment id, echoed (null for a legacy caller). + /// + /// when the apply swallowed per-node sink faults, its rebuild failed, the + /// pipeline threw, or there was no usable artifact — i.e. the served address space may be partial or + /// stale. An empty plan (nothing changed) is clean. + /// + /// A short operator-readable reason; null when clean. + public sealed record RebuildCompleted( + CorrelationId Correlation, DeploymentId? DeploymentId, bool Degraded, string? DegradedReason); public sealed record ServiceLevelChanged(byte ServiceLevel); @@ -134,6 +157,15 @@ public sealed class OpcUaPublishActor : ReceiveActor, IWithTimers /// protecting. Distinguishes "the artifact went missing under a running server" (issue #485 — warn, and /// hold what is materialised) from "nothing has been deployed here yet" (a quiet no-op at boot). private bool _hasAppliedComposition; + /// + /// Gitea #601 review — was advanced by a DEGRADED rebuild (a rebuild advances + /// even with swallowed failures, see ). A re-deploy of the same revision + /// then diffs to an EMPTY plan; without this flag that empty plan reported clean, so the "re-deploy + /// re-applies" repair path sealed green over an address space still known to be partial. While set, an + /// empty plan re-materialises the whole composition (idempotent ensures) and reports clean only when + /// that pass is itself clean. + /// + private bool _lastAppliedDegraded; private DbHealthProbeActor.DbHealthStatus? _lastDbHealth; private RedundancyStateChanged? _lastSnapshot; private (bool Ok, DateTime At)? _probeAboutMe; @@ -438,9 +470,13 @@ public sealed class OpcUaPublishActor : ReceiveActor, IWithTimers private void HandleRebuild(RebuildAddressSpace msg) { + // Captured up front: the pipeline below is synchronous on this actor's thread today, but the reply + // must reach whoever asked even if a future edit introduces an await into it. + var requester = Sender; + (bool Degraded, string? Reason) outcome; try { - HandleRebuildCore(msg); + outcome = HandleRebuildCore(msg); } finally { @@ -455,9 +491,14 @@ public sealed class OpcUaPublishActor : ReceiveActor, IWithTimers _log.Error(ex, "OpcUaPublish: sink.CompleteAddressSpaceApply threw (correlation={Correlation})", msg.Correlation); } } + + // Gitea #601 — the outcome goes out only after the apply (including #641's completion step) is done. + if (msg.ReportOutcome) + requester.Tell(new RebuildCompleted(msg.Correlation, msg.DeploymentId, outcome.Degraded, outcome.Reason)); } - private void HandleRebuildCore(RebuildAddressSpace msg) + /// The rebuild pipeline. Returns whether the apply was degraded, and why (Gitea #601). + private (bool Degraded, string? Reason) HandleRebuildCore(RebuildAddressSpace msg) { using var span = OtOpcUaTelemetry.StartAddressSpaceRebuildSpan(); span?.SetTag("otopcua.correlation_id", msg.Correlation.ToString()); @@ -481,8 +522,9 @@ public sealed class OpcUaPublishActor : ReceiveActor, IWithTimers { _log.Error(ex, "OpcUaPublish: sink.RebuildAddressSpace threw (correlation={Correlation})", msg.Correlation); + return (true, $"address-space rebuild threw: {ex.Message}"); } - return; + return (false, null); } try @@ -520,7 +562,7 @@ public sealed class OpcUaPublishActor : ReceiveActor, IWithTimers _log.Debug( "OpcUaPublish: no artifact to materialise yet (deployment={Id}, correlation={Correlation})", msg.DeploymentId, msg.Correlation); - return; + return (true, "no usable artifact reached the address-space apply; the previously served address space was kept"); } // Parse the artifact ONCE for this rebuild. Every consumer below (data-plane authority, then the @@ -551,9 +593,31 @@ public sealed class OpcUaPublishActor : ReceiveActor, IWithTimers if (plan.IsEmpty) { + if (_lastAppliedDegraded) + { + // #601 review — the composition is unchanged but the last apply of it was a degraded + // rebuild: re-ensure every node (each Ensure* no-ops on a node that stands) so the + // re-deploy actually repairs what that rebuild could not place. + var repairFailed = _applier.MaterialiseHierarchy(composition) + + MaterialiseNodes(plan, composition, fullComposition: true); + if (repairFailed > 0) + { + OtOpcUaTelemetry.OpcUaApplyFailed.Add(1, new KeyValuePair("kind", "nodes")); + _log.Error( + "OpcUaPublish: re-apply of a degraded composition is STILL degraded (correlation={Correlation}, failedNodes={FailedNodes})", + msg.Correlation, repairFailed); + return (true, $"re-apply of a previously degraded address space is still degraded (failedNodes={repairFailed})"); + } + + _lastAppliedDegraded = false; + _log.Info("OpcUaPublish: re-apply repaired the previously degraded address space (correlation={Correlation})", + msg.Correlation); + return (false, null); + } + _log.Debug("OpcUaPublish: rebuild requested but plan is empty (correlation={Correlation})", msg.Correlation); - return; + return (false, null); } var outcome = _applier.Apply(plan); @@ -596,6 +660,9 @@ public sealed class OpcUaPublishActor : ReceiveActor, IWithTimers { _lastApplied = composition; _hasAppliedComposition = true; + // #601 review — advanced despite failures only on a rebuild; remember it so an identical + // re-deploy repairs instead of short-circuiting clean on the empty plan. + _lastAppliedDegraded = outcome.RebuildFailed || failedNodes > 0; } else { @@ -629,16 +696,19 @@ public sealed class OpcUaPublishActor : ReceiveActor, IWithTimers "OpcUaPublish: address-space apply DEGRADED (correlation={Correlation}, rebuildFailed={RebuildFailed}, failedNodes={FailedNodes}, added={Added}, removed={Removed}, changed={Changed}) — the running address space may be stale or partial", msg.Correlation, outcome.RebuildFailed, failedNodes, outcome.AddedNodes, outcome.RemovedNodes, outcome.ChangedNodes); + return (true, outcome.RebuildFailed + ? $"address-space rebuild failed (failedNodes={failedNodes})" + : $"address-space apply degraded (failedNodes={failedNodes}); not recorded as applied, the next deploy of this revision re-diffs and repairs it"); } - else - { - _log.Info("OpcUaPublish: applied rebuild (correlation={Correlation}, added={Added}, removed={Removed}, changed={Changed}, rebuild={Rebuild})", - msg.Correlation, outcome.AddedNodes, outcome.RemovedNodes, outcome.ChangedNodes, outcome.RebuildCalled); - } + + _log.Info("OpcUaPublish: applied rebuild (correlation={Correlation}, added={Added}, removed={Removed}, changed={Changed}, rebuild={Rebuild})", + msg.Correlation, outcome.AddedNodes, outcome.RemovedNodes, outcome.ChangedNodes, outcome.RebuildCalled); + return (false, null); } catch (Exception ex) { _log.Error(ex, "OpcUaPublish: rebuild pipeline threw (correlation={Correlation})", msg.Correlation); + return (true, $"address-space apply threw: {ex.Message}"); } } diff --git a/tests/Core/ZB.MOM.WW.OtOpcUa.Commons.Tests/MeshTransportDtoMapperTests.cs b/tests/Core/ZB.MOM.WW.OtOpcUa.Commons.Tests/MeshTransportDtoMapperTests.cs index aa2adcee..ce19d0da 100644 --- a/tests/Core/ZB.MOM.WW.OtOpcUa.Commons.Tests/MeshTransportDtoMapperTests.cs +++ b/tests/Core/ZB.MOM.WW.OtOpcUa.Commons.Tests/MeshTransportDtoMapperTests.cs @@ -201,6 +201,61 @@ public class MeshTransportDtoMapperTests back.FailureReason.ShouldBeEmpty(); } + [Fact] + public void ApplyAck_Degraded_RoundTrips_WithItsReason() + { + // #601: an Applied ack whose address-space apply was DEGRADED must reach central as degraded — + // a node that loses the flag on the wire reports a clean apply it did not have. + var domain = new ApplyAck( + new DeploymentId(DeploymentGuid), + new NodeId("site-a-1"), + ApplyAckOutcome.Applied, + null, + new CorrelationId(CorrelationGuid), + Degraded: true, + DegradedReason: "failedNodes=3"); + + var back = MeshTransportDtoMapper.FromProto(Wire(MeshTransportDtoMapper.ToProto(domain))); + + back.ShouldBe(domain); + back.Degraded.ShouldBeTrue(); + back.DegradedReason.ShouldBe("failedNodes=3"); + } + + [Fact] + public void ApplyAck_Clean_EncodesNoDegradedFields() + { + var proto = MeshTransportDtoMapper.ToProto(new ApplyAck( + new DeploymentId(DeploymentGuid), new NodeId("site-a-1"), ApplyAckOutcome.Applied, null, + new CorrelationId(CorrelationGuid))); + + proto.Degraded.ShouldBeFalse(); + proto.HasDegradedReason.ShouldBeFalse(); + } + + [Fact] + public void ApplyAck_FromAnOldNode_WithoutTheDegradedFields_DecodesAsTodaysCleanAck() + { + // #601 compatibility: a node built before the degraded fields existed sends exactly fields 1-5. + // Its ack must decode exactly as it always did — not degraded, no reason. + var oldNodeBytes = new ApplyAckMessage + { + DeploymentId = DeploymentGuid.ToString("N"), + NodeId = "site-a-1", + Outcome = ApplyOutcome.Applied, + CorrelationId = CorrelationGuid.ToString("N"), + }.ToByteArray(); + + var back = MeshTransportDtoMapper.FromProto(ApplyAckMessage.Parser.ParseFrom(oldNodeBytes)); + + back.Outcome.ShouldBe(ApplyAckOutcome.Applied); + back.Degraded.ShouldBeFalse(); + back.DegradedReason.ShouldBeNull(); + back.ShouldBe(new ApplyAck( + new DeploymentId(DeploymentGuid), new NodeId("site-a-1"), ApplyAckOutcome.Applied, null, + new CorrelationId(CorrelationGuid))); + } + [Fact] public void ApplyAck_UnspecifiedOutcome_Throws_RatherThanDecodingToAMeaningfulValue() { diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.ControlPlane.Tests/ConfigPublishCoordinatorDegradedAckTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.ControlPlane.Tests/ConfigPublishCoordinatorDegradedAckTests.cs new file mode 100644 index 00000000..f67424b1 --- /dev/null +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.ControlPlane.Tests/ConfigPublishCoordinatorDegradedAckTests.cs @@ -0,0 +1,220 @@ +using System.Text; +using Akka.Actor; +using Microsoft.EntityFrameworkCore; +using Newtonsoft.Json.Linq; +using Shouldly; +using Xunit; +using ZB.MOM.WW.OtOpcUa.Commons.Messages.Deploy; +using ZB.MOM.WW.OtOpcUa.Commons.Types; +using ZB.MOM.WW.OtOpcUa.Configuration; +using ZB.MOM.WW.OtOpcUa.Configuration.Entities; +using ZB.MOM.WW.OtOpcUa.Configuration.Enums; +using ZB.MOM.WW.OtOpcUa.ControlPlane.Coordinators; +using ZB.MOM.WW.OtOpcUa.ControlPlane.Tests.Harness; + +namespace ZB.MOM.WW.OtOpcUa.ControlPlane.Tests; + +/// +/// Gitea #601 — central's side of the honest deploy ack. A node that applied a deployment but whose +/// address-space apply was DEGRADED acks with +/// . The deployment still SEALS (the revision is what the node now runs, and +/// "latest sealed" drives drift + the publish actor's fallback) but never looks clean: the node's +/// NodeDeploymentState row keeps status Applied with its degraded reason in FailureReason, +/// and the Deployment row's FailureReason names the degraded nodes. An ack from an old node +/// (no degraded field) is handled exactly as before. +/// +public sealed class ConfigPublishCoordinatorDegradedAckTests : ControlPlaneActorTestBase +{ + private static readonly RevisionHash TestRevision = RevisionHash.Parse(new string('c', 64)); + private static readonly TimeSpan NoTimeout = TimeSpan.FromMinutes(5); + private const string NodeA = "site-a-1:4053"; + private const string NodeB = "site-a-2:4053"; + + [Fact] + public void A_degraded_ack_seals_but_is_recorded_as_degraded_on_the_node_and_the_deployment() + { + var dbFactory = NewInMemoryDbFactory(); + var deploymentId = SeedDispatchingDeployment(dbFactory); + SeedNodes(dbFactory, NodeA, NodeB); + var actor = Sys.ActorOf(ConfigPublishCoordinator.Props(dbFactory, NoTimeout)); + actor.Tell(new DispatchDeployment(deploymentId, TestRevision, CorrelationId.NewId())); + AwaitStatus(dbFactory, DeploymentStatus.AwaitingApplyAcks); + + actor.Tell(new ApplyAck(deploymentId, NodeId.Parse(NodeA), ApplyAckOutcome.Applied, null, + CorrelationId.NewId())); + actor.Tell(new ApplyAck(deploymentId, NodeId.Parse(NodeB), ApplyAckOutcome.Applied, null, + CorrelationId.NewId(), Degraded: true, DegradedReason: "address-space apply degraded (failedNodes=3)")); + + AwaitStatus(dbFactory, DeploymentStatus.Sealed); + using var db = dbFactory.CreateDbContext(); + + var deployment = db.Deployments.Single(); + deployment.FailureReason.ShouldNotBeNull(); + deployment.FailureReason.ShouldContain("degraded"); + deployment.FailureReason.ShouldContain(NodeB); + deployment.FailureReason.ShouldNotContain(NodeA); + + var rowB = db.NodeDeploymentStates.Single(s => s.NodeId == NodeB); + rowB.Status.ShouldBe(NodeDeploymentStatus.Applied); + rowB.FailureReason.ShouldNotBeNull(); + rowB.FailureReason.ShouldStartWith(ConfigPublishCoordinator.DegradedReasonPrefix); + rowB.FailureReason.ShouldContain("failedNodes=3"); + ConfigPublishCoordinator.IsDegraded(rowB.Status, rowB.FailureReason).ShouldBeTrue(); + + var rowA = db.NodeDeploymentStates.Single(s => s.NodeId == NodeA); + rowA.Status.ShouldBe(NodeDeploymentStatus.Applied); + rowA.FailureReason.ShouldBeNull(); + ConfigPublishCoordinator.IsDegraded(rowA.Status, rowA.FailureReason).ShouldBeFalse(); + } + + [Fact] + public void Old_node_acks_without_the_degraded_field_seal_clean_exactly_as_before() + { + // The five-argument constructor IS the pre-#601 ack shape: Degraded defaults to false. + var dbFactory = NewInMemoryDbFactory(); + var deploymentId = SeedDispatchingDeployment(dbFactory); + SeedNodes(dbFactory, NodeA, NodeB); + var actor = Sys.ActorOf(ConfigPublishCoordinator.Props(dbFactory, NoTimeout)); + actor.Tell(new DispatchDeployment(deploymentId, TestRevision, CorrelationId.NewId())); + AwaitStatus(dbFactory, DeploymentStatus.AwaitingApplyAcks); + + actor.Tell(new ApplyAck(deploymentId, NodeId.Parse(NodeA), ApplyAckOutcome.Applied, null, CorrelationId.NewId())); + actor.Tell(new ApplyAck(deploymentId, NodeId.Parse(NodeB), ApplyAckOutcome.Applied, null, CorrelationId.NewId())); + + AwaitStatus(dbFactory, DeploymentStatus.Sealed); + using var db = dbFactory.CreateDbContext(); + db.Deployments.Single().FailureReason.ShouldBeNull(); + db.NodeDeploymentStates.All(s => s.Status == NodeDeploymentStatus.Applied && s.FailureReason == null) + .ShouldBeTrue(); + } + + [Fact] + public void A_degraded_ack_alongside_a_failed_one_is_still_PartiallyFailed() + { + var dbFactory = NewInMemoryDbFactory(); + var deploymentId = SeedDispatchingDeployment(dbFactory); + SeedNodes(dbFactory, NodeA, NodeB); + var actor = Sys.ActorOf(ConfigPublishCoordinator.Props(dbFactory, NoTimeout)); + actor.Tell(new DispatchDeployment(deploymentId, TestRevision, CorrelationId.NewId())); + AwaitStatus(dbFactory, DeploymentStatus.AwaitingApplyAcks); + + actor.Tell(new ApplyAck(deploymentId, NodeId.Parse(NodeA), ApplyAckOutcome.Failed, "boom", CorrelationId.NewId())); + actor.Tell(new ApplyAck(deploymentId, NodeId.Parse(NodeB), ApplyAckOutcome.Applied, null, + CorrelationId.NewId(), Degraded: true, DegradedReason: "x")); + + AwaitStatus(dbFactory, DeploymentStatus.PartiallyFailed); + using var db = dbFactory.CreateDbContext(); + db.NodeDeploymentStates.Single(s => s.NodeId == NodeA).FailureReason.ShouldBe("boom"); + db.NodeDeploymentStates.Single(s => s.NodeId == NodeA).Status.ShouldBe(NodeDeploymentStatus.Failed); + } + + [Fact] + public void A_recovered_coordinator_remembers_a_degraded_ack_that_landed_before_failover() + { + // Node B's degraded ack was persisted by a previous singleton instance; this instance recovers the + // in-flight deployment from the DB and must still seal it as degraded when node A acks clean. + var dbFactory = NewInMemoryDbFactory(); + var deploymentId = SeedDispatchingDeployment(dbFactory, DeploymentStatus.AwaitingApplyAcks); + SeedNodes(dbFactory, NodeA, NodeB); + using (var db = dbFactory.CreateDbContext()) + { + db.NodeDeploymentStates.Add(new NodeDeploymentState + { + NodeId = NodeA, DeploymentId = deploymentId.Value, Status = NodeDeploymentStatus.Applying, + }); + db.NodeDeploymentStates.Add(new NodeDeploymentState + { + NodeId = NodeB, DeploymentId = deploymentId.Value, Status = NodeDeploymentStatus.Applied, + FailureReason = ConfigPublishCoordinator.DegradedReasonPrefix + "failedNodes=1", + AppliedAtUtc = DateTime.UtcNow, + }); + db.SaveChanges(); + } + + var actor = Sys.ActorOf(ConfigPublishCoordinator.Props(dbFactory, NoTimeout)); + actor.Tell(new ApplyAck(deploymentId, NodeId.Parse(NodeA), ApplyAckOutcome.Applied, null, CorrelationId.NewId())); + + AwaitStatus(dbFactory, DeploymentStatus.Sealed); + using var verify = dbFactory.CreateDbContext(); + var reason = verify.Deployments.Single().FailureReason; + reason.ShouldNotBeNull(); + reason.ShouldContain(NodeB); + } + + [Fact] + public void An_old_nodes_DPS_ack_without_the_degraded_fields_deserializes_as_a_clean_ack() + { + // The DPS ack path rides Akka's default serializer for the record. Serialize a NEW ack, strip the two + // #601 members from the payload (what an old node sends), and decode it with this build. + var ack = new ApplyAck(DeploymentId.NewId(), NodeId.Parse(NodeA), ApplyAckOutcome.Applied, null, + CorrelationId.NewId(), Degraded: true, DegradedReason: "will be stripped"); + var serializer = Sys.Serialization.FindSerializerFor(ack); + var json = JObject.Parse(Encoding.UTF8.GetString(serializer.ToBinary(ack))); + json.Remove("Degraded").ShouldBeTrue(); + json.Remove("DegradedReason").ShouldBeTrue(); + + var back = (ApplyAck)serializer.FromBinary( + Encoding.UTF8.GetBytes(json.ToString(Newtonsoft.Json.Formatting.None)), typeof(ApplyAck)); + + back.Outcome.ShouldBe(ApplyAckOutcome.Applied); + back.Degraded.ShouldBeFalse(); + back.DegradedReason.ShouldBeNull(); + back.DeploymentId.ShouldBe(ack.DeploymentId); + + // ...and the new fields survive the same serializer when present. + var full = (ApplyAck)serializer.FromBinary(serializer.ToBinary(ack), typeof(ApplyAck)); + full.ShouldBe(ack); + } + + private void AwaitStatus(IDbContextFactory dbFactory, DeploymentStatus expected) => + AwaitAssert(() => + { + using var db = dbFactory.CreateDbContext(); + db.Deployments.Single().Status.ShouldBe(expected); + }, duration: TimeSpan.FromSeconds(5)); + + private static void SeedNodes(IDbContextFactory dbFactory, params string[] nodes) + { + using var db = dbFactory.CreateDbContext(); + db.ServerClusters.Add(new ServerCluster + { + ClusterId = "SITE-A", + Name = "Site A", + Enterprise = "zb", + Site = "site-a", + NodeCount = 2, + RedundancyMode = RedundancyMode.Warm, + CreatedBy = "test", + }); + foreach (var nodeId in nodes) + { + db.ClusterNodes.Add(new ClusterNode + { + NodeId = nodeId, + ClusterId = "SITE-A", + Host = nodeId.Split(':')[0], + ApplicationUri = $"urn:OtOpcUa:{nodeId}", + Enabled = true, + CreatedBy = "test", + }); + } + db.SaveChanges(); + } + + private static DeploymentId SeedDispatchingDeployment( + IDbContextFactory dbFactory, + DeploymentStatus status = DeploymentStatus.Dispatching) + { + var id = DeploymentId.NewId(); + using var db = dbFactory.CreateDbContext(); + db.Deployments.Add(new Deployment + { + DeploymentId = id.Value, + RevisionHash = TestRevision.Value, + Status = status, + CreatedBy = "test", + }); + db.SaveChanges(); + return id; + } +} diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverHostActorBootFromCacheTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverHostActorBootFromCacheTests.cs index 11da96f3..bdc671f2 100644 --- a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverHostActorBootFromCacheTests.cs +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverHostActorBootFromCacheTests.cs @@ -67,7 +67,7 @@ public sealed class DriverHostActorBootFromCacheTests : RuntimeActorTestBase var blob = EmptyArtifact(); var cached = new CachedDeploymentArtifact( DeploymentId.NewId().ToString(), CachedRev.Value, blob, DateTimeOffset.UtcNow.AddHours(-1)); - var publishProbe = CreateTestProbe(); + var publishProbe = this.CreateCleanOutcomePublishProbe(); Sys.ActorOf(DriverHostActor.Props( new ThrowingDbFactory(), TestNode, ackRouter: null, @@ -92,7 +92,7 @@ public sealed class DriverHostActorBootFromCacheTests : RuntimeActorTestBase var cached = new CachedDeploymentArtifact( DeploymentId.NewId().ToString(), CachedRev.Value, System.Text.Encoding.Latin1.GetBytes(text), DateTimeOffset.UtcNow.AddHours(-1)); - var publishProbe = CreateTestProbe(); + var publishProbe = this.CreateCleanOutcomePublishProbe(); var actor = Sys.ActorOf(DriverHostActor.Props( new ThrowingDbFactory(), TestNode, ackRouter: null, diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverHostActorDbLessTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverHostActorDbLessTests.cs index dc3504f9..fbf049f3 100644 --- a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverHostActorDbLessTests.cs +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverHostActorDbLessTests.cs @@ -112,7 +112,7 @@ public sealed class DriverHostActorDbLessTests : RuntimeActorTestBase var cache = new RecordingArtifactCache(); var fetcher = new RecordingFetcher(ArtifactBytes()); var deploymentId = DeploymentId.NewId(); - var publish = CreateTestProbe(); + var publish = this.CreateCleanOutcomePublishProbe(); var coordinator = CreateTestProbe(); var actor = Sys.ActorOf(DriverHostActor.Props( @@ -141,7 +141,7 @@ public sealed class DriverHostActorDbLessTests : RuntimeActorTestBase [Fact] public void DbLessNode_WithLocalDbStore_SpawnsScriptedAlarmHost() { - var publish = CreateTestProbe(); + var publish = this.CreateCleanOutcomePublishProbe(); var actor = Sys.ActorOf(DriverHostActor.Props( dbFactory: null, TestNode, @@ -162,7 +162,7 @@ public sealed class DriverHostActorDbLessTests : RuntimeActorTestBase [Fact] public void DbLessNode_WithNoStoreAndNoDbFactory_SkipsScriptedAlarmHost() { - var publish = CreateTestProbe(); + var publish = this.CreateCleanOutcomePublishProbe(); var actor = Sys.ActorOf(DriverHostActor.Props( dbFactory: null, TestNode, @@ -185,7 +185,7 @@ public sealed class DriverHostActorDbLessTests : RuntimeActorTestBase [Fact] public void DbBackedNode_WithNoStore_SkipsScriptedAlarmHost() { - var publish = CreateTestProbe(); + var publish = this.CreateCleanOutcomePublishProbe(); var actor = Sys.ActorOf(DriverHostActor.Props( dbFactory: NewInMemoryDbFactory(), TestNode, diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverHostActorDegradedAckTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverHostActorDegradedAckTests.cs new file mode 100644 index 00000000..e79f0ec2 --- /dev/null +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverHostActorDegradedAckTests.cs @@ -0,0 +1,244 @@ +using Akka.Actor; +using Microsoft.EntityFrameworkCore; +using Microsoft.Extensions.Logging.Abstractions; +using Shouldly; +using Xunit; +using ZB.MOM.WW.OtOpcUa.Commons.Messages.Deploy; +using ZB.MOM.WW.OtOpcUa.Commons.OpcUa; +using ZB.MOM.WW.OtOpcUa.Commons.Types; +using ZB.MOM.WW.OtOpcUa.Configuration; +using ZB.MOM.WW.OtOpcUa.OpcUaServer; +using ZB.MOM.WW.OtOpcUa.Runtime.Drivers; +using ZB.MOM.WW.OtOpcUa.Runtime.OpcUa; +using ZB.MOM.WW.OtOpcUa.Runtime.Tests.Harness; + +namespace ZB.MOM.WW.OtOpcUa.Runtime.Tests.Drivers; + +/// +/// Gitea #601 (archreview 01/S-9) — the deploy ack must be honest about the address-space apply. It used +/// to be sent the moment the revision advanced, before OpcUaPublishActor had even dequeued the +/// rebuild, so a degraded apply (swallowed sink faults; _lastApplied deliberately not advanced) +/// still acked a clean and the deployment sealed green fleet-wide. +/// Now the Applied ack waits for the publish actor's and +/// carries its verdict; a missing verdict times out to degraded, never clean. +/// +[Collection("opcua-publish-meters")] +public sealed class DriverHostActorDegradedAckTests : RuntimeActorTestBase +{ + private static readonly NodeId TestNode = NodeId.Parse("driver-test"); + private static readonly RevisionHash RevA = RevisionHash.Parse(new string('a', 64)); + + [Fact] + public void The_Applied_ack_waits_for_the_address_space_outcome_and_a_clean_one_acks_clean() + { + var (db, deploymentId) = SeedOneTagDeployment(); + var coordinator = CreateTestProbe(); + var publish = CreateTestProbe(); + var actor = Sys.ActorOf(DriverHostActor.Props( + db, TestNode, coordinator.Ref, localRoles: new HashSet { "driver" }, + opcUaPublishActor: publish.Ref)); + + actor.Tell(new DispatchDeployment(deploymentId, RevA, CorrelationId.NewId())); + + var rebuild = publish.FishForMessage(_ => true, PresenceBudget); + rebuild.ReportOutcome.ShouldBeTrue(); + // The ack is HELD until the outcome lands — the pre-#601 code had already sent it by now. + coordinator.ExpectNoMsg(TimeSpan.FromMilliseconds(300)); + + publish.Reply(new OpcUaPublishActor.RebuildCompleted(rebuild.Correlation, rebuild.DeploymentId, false, null)); + + var ack = coordinator.ExpectMsg(PresenceBudget); + ack.Outcome.ShouldBe(ApplyAckOutcome.Applied); + ack.Degraded.ShouldBeFalse(); + ack.DegradedReason.ShouldBeNull(); + } + + [Fact] + public void A_degraded_address_space_outcome_acks_Applied_but_degraded_with_the_reason() + { + var (db, deploymentId) = SeedOneTagDeployment(); + var coordinator = CreateTestProbe(); + var publish = CreateTestProbe(); + var actor = Sys.ActorOf(DriverHostActor.Props( + db, TestNode, coordinator.Ref, localRoles: new HashSet { "driver" }, + opcUaPublishActor: publish.Ref)); + + actor.Tell(new DispatchDeployment(deploymentId, RevA, CorrelationId.NewId())); + var rebuild = publish.FishForMessage(_ => true, PresenceBudget); + publish.Reply(new OpcUaPublishActor.RebuildCompleted( + rebuild.Correlation, rebuild.DeploymentId, Degraded: true, DegradedReason: "failedNodes=2")); + + var ack = coordinator.ExpectMsg(PresenceBudget); + ack.Outcome.ShouldBe(ApplyAckOutcome.Applied); + ack.Degraded.ShouldBeTrue(); + ack.DegradedReason.ShouldBe("failedNodes=2"); + ack.DeploymentId.ShouldBe(deploymentId); + } + + [Fact] + public void A_real_publish_actor_whose_sink_faults_produces_a_degraded_ack_end_to_end() + { + // No hand-written verdict: the real OpcUaPublishActor + AddressSpaceApplier over a sink that faults on + // every node. This is the production shape of the defect — a degraded apply that sealed green. + var (db, deploymentId) = SeedOneTagDeployment(); + var coordinator = CreateTestProbe(); + var sink = new FaultingSink(); + var publish = Sys.ActorOf(OpcUaPublishActor.PropsForTests( + sink: sink, applier: new AddressSpaceApplier(sink, NullLogger.Instance))); + var actor = Sys.ActorOf(DriverHostActor.Props( + db, TestNode, coordinator.Ref, localRoles: new HashSet { "driver" }, + opcUaPublishActor: publish)); + + actor.Tell(new DispatchDeployment(deploymentId, RevA, CorrelationId.NewId())); + + var ack = coordinator.ExpectMsg(PresenceBudget); + ack.Outcome.ShouldBe(ApplyAckOutcome.Applied); + ack.Degraded.ShouldBeTrue(); + ack.DegradedReason.ShouldNotBeNull(); + ack.DegradedReason.ShouldContain("failedNodes="); + } + + [Fact] + public void A_real_publish_actor_with_a_healthy_sink_produces_a_clean_ack_end_to_end() + { + // The control for the test above: same wiring, a sink that does not fault ⇒ clean. + var (db, deploymentId) = SeedOneTagDeployment(); + var coordinator = CreateTestProbe(); + var sink = new FaultingSink(fault: false); + var publish = Sys.ActorOf(OpcUaPublishActor.PropsForTests( + sink: sink, applier: new AddressSpaceApplier(sink, NullLogger.Instance))); + var actor = Sys.ActorOf(DriverHostActor.Props( + db, TestNode, coordinator.Ref, localRoles: new HashSet { "driver" }, + opcUaPublishActor: publish)); + + actor.Tell(new DispatchDeployment(deploymentId, RevA, CorrelationId.NewId())); + + var ack = coordinator.ExpectMsg(PresenceBudget); + ack.Outcome.ShouldBe(ApplyAckOutcome.Applied); + ack.Degraded.ShouldBeFalse(); + } + + [Fact] + public void No_outcome_within_the_timeout_acks_degraded_never_clean() + { + var (db, deploymentId) = SeedOneTagDeployment(); + var coordinator = CreateTestProbe(); + var publish = CreateTestProbe(); // never replies — a hung / lost apply + var actor = Sys.ActorOf(DriverHostActor.Props( + db, TestNode, coordinator.Ref, localRoles: new HashSet { "driver" }, + opcUaPublishActor: publish.Ref, applyOutcomeTimeout: TimeSpan.FromMilliseconds(300))); + + actor.Tell(new DispatchDeployment(deploymentId, RevA, CorrelationId.NewId())); + + var ack = coordinator.ExpectMsg(PresenceBudget); + ack.Outcome.ShouldBe(ApplyAckOutcome.Applied); + ack.Degraded.ShouldBeTrue(); + ack.DegradedReason.ShouldNotBeNull(); + // Distinct from a real sink failure: it says the outcome was not reported in time. + ack.DegradedReason.ShouldStartWith(DriverHostActor.OutcomeNotReportedReasonPrefix); + ack.DegradedReason.ShouldNotContain("failedNodes="); + + // A late outcome after the timeout must not produce a second ack. + var rebuild = publish.FishForMessage(_ => true, PresenceBudget); + publish.Reply(new OpcUaPublishActor.RebuildCompleted(rebuild.Correlation, rebuild.DeploymentId, false, null)); + coordinator.ExpectNoMsg(TimeSpan.FromMilliseconds(300)); + } + + [Fact] + public void Redispatching_a_degraded_revision_reapplies_instead_of_acking_clean() + { + var (db, deploymentId) = SeedOneTagDeployment(); + var coordinator = CreateTestProbe(); + var publish = CreateTestProbe(); + var actor = Sys.ActorOf(DriverHostActor.Props( + db, TestNode, coordinator.Ref, localRoles: new HashSet { "driver" }, + opcUaPublishActor: publish.Ref)); + + actor.Tell(new DispatchDeployment(deploymentId, RevA, CorrelationId.NewId())); + var first = publish.FishForMessage(_ => true, PresenceBudget); + publish.Reply(new OpcUaPublishActor.RebuildCompleted(first.Correlation, first.DeploymentId, true, "failedNodes=1")); + coordinator.ExpectMsg(PresenceBudget).Degraded.ShouldBeTrue(); + + // Same revision again: the old idempotent short-circuit would ack a CLEAN Applied right here. It must + // re-apply instead (a second rebuild request), and ack whatever THAT apply reports. + actor.Tell(new DispatchDeployment(deploymentId, RevA, CorrelationId.NewId())); + var second = publish.FishForMessage(_ => true, PresenceBudget); + coordinator.ExpectNoMsg(TimeSpan.FromMilliseconds(200)); + publish.Reply(new OpcUaPublishActor.RebuildCompleted(second.Correlation, second.DeploymentId, false, null)); + coordinator.ExpectMsg(PresenceBudget).Degraded.ShouldBeFalse(); + + // Now clean: a third dispatch of the same revision takes the short-circuit again (no rebuild). + actor.Tell(new DispatchDeployment(deploymentId, RevA, CorrelationId.NewId())); + var third = coordinator.ExpectMsg(PresenceBudget); + third.Degraded.ShouldBeFalse(); + publish.ExpectNoMsg(TimeSpan.FromMilliseconds(200)); + } + + [Fact] + public void A_degraded_apply_records_its_reason_on_the_local_NodeDeploymentState_row() + { + // #601 review — a node WITH a ConfigDb writes its own row Applied/null before the outcome is known. + // Central later adds the reason from the ack, but only when it accepts that ack (a stale/timed-out + // deployment's ack is discarded), so the node must not leave its own row reading clean. + var (db, deploymentId) = SeedOneTagDeployment(); + var coordinator = CreateTestProbe(); + var publish = CreateTestProbe(); + var actor = Sys.ActorOf(DriverHostActor.Props( + db, TestNode, coordinator.Ref, localRoles: new HashSet { "driver" }, + opcUaPublishActor: publish.Ref)); + + actor.Tell(new DispatchDeployment(deploymentId, RevA, CorrelationId.NewId())); + var rebuild = publish.FishForMessage(_ => true, PresenceBudget); + publish.Reply(new OpcUaPublishActor.RebuildCompleted(rebuild.Correlation, rebuild.DeploymentId, true, "failedNodes=3")); + coordinator.ExpectMsg(PresenceBudget).Degraded.ShouldBeTrue(); + + AwaitAssert(() => + { + using var ctx = db.CreateDbContext(); + var row = ctx.NodeDeploymentStates.Single(r => r.NodeId == TestNode.Value && r.DeploymentId == deploymentId.Value); + row.Status.ShouldBe(ZB.MOM.WW.OtOpcUa.Configuration.Enums.NodeDeploymentStatus.Applied); + row.FailureReason.ShouldBe(ApplyAck.DegradedReasonPrefix + "failedNodes=3"); + }, PresenceBudget); + } + + [Fact] + public void The_default_outcome_timeout_is_derived_from_centrals_apply_deadline_minus_a_margin() + { + // A slow-but-clean rebuild must not be labelled degraded at 60 s, yet the held ack must still + // reach central before its apply deadline. Both sides read ONE shared constant. + DeployTimings.ApplyDeadline.ShouldBe(TimeSpan.FromMinutes(2)); + DriverHostActor.DefaultApplyOutcomeTimeout.ShouldBe(DeployTimings.ApplyDeadline - DeployTimings.ApplyOutcomeSafetyMargin); + DeployTimings.ApplyOutcomeSafetyMargin.ShouldBe(TimeSpan.FromSeconds(15)); + DriverHostActor.DefaultApplyOutcomeTimeout.ShouldBeGreaterThan(TimeSpan.FromSeconds(60)); + DriverHostActor.DefaultApplyOutcomeTimeout.ShouldBeLessThan(DeployTimings.ApplyDeadline); + } + + private (IDbContextFactory Db, DeploymentId Id) SeedOneTagDeployment() + { + var db = NewInMemoryDbFactory(); + var id = V3ArtifactSeeder.SeedV3Deployment(db, RevA, + rawTags: new[] { (Driver: "drv-1", DriverName: "Modbus", Device: "dev1", Tag: "speed", DataType: "Double", Writable: false) }, + unsRefs: new[] { (Area: "filling", Line: "line1", Equip: "station1", BackingTag: "speed", Effective: (string?)null) }); + return (db, id); + } + + /// A sink that throws from every node-creating call when fault is set. + private sealed class FaultingSink(bool fault = true) : IOpcUaAddressSpaceSink + { + private void MaybeFault() + { + if (fault) throw new InvalidOperationException("simulated sink fault"); + } + + public void WriteValue(string nodeId, object? value, uint statusCode, DateTime sourceTimestampUtc, AddressSpaceRealm realm = AddressSpaceRealm.Uns) { } + public void WriteAlarmQuality(string alarmNodeId, uint statusCode, DateTime sourceTimestampUtc, AddressSpaceRealm realm) { } + public void WriteAlarmCondition(string alarmNodeId, AlarmConditionSnapshot state, DateTime occurredUtc, AddressSpaceRealm realm = AddressSpaceRealm.Uns) { } + public void MaterialiseAlarmCondition(string alarmNodeId, string equipmentNodeId, string displayName, string alarmType, int severity, AddressSpaceRealm realm, bool isNative = false) => MaybeFault(); + public void EnsureFolder(string folderNodeId, string? parentNodeId, string displayName, AddressSpaceRealm realm = AddressSpaceRealm.Uns) => MaybeFault(); + public void EnsureVariable(string variableNodeId, string? parentFolderNodeId, string displayName, string dataType, bool writable, AddressSpaceRealm realm, string? historianTagname = null, bool isArray = false, uint? arrayLength = null, bool refuseArrayOnScalar = false) => MaybeFault(); + public void RebuildAddressSpace() { } + public void RaiseNodesAddedModelChange(string affectedNodeId, AddressSpaceRealm realm = AddressSpaceRealm.Uns) { } + public void WireAlarmNotifiers(string alarmNodeId, AddressSpaceRealm alarmRealm, IReadOnlyList notifierFolderNodeIds, AddressSpaceRealm notifierFolderRealm) { } + public void AddReference(string sourceNodeId, AddressSpaceRealm sourceRealm, string targetNodeId, AddressSpaceRealm targetRealm, string referenceType = "Organizes") { } + } +} diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverHostActorFetchAndCacheBootstrapTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverHostActorFetchAndCacheBootstrapTests.cs index f7e113b6..b24bf7b8 100644 --- a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverHostActorFetchAndCacheBootstrapTests.cs +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverHostActorFetchAndCacheBootstrapTests.cs @@ -42,7 +42,7 @@ public sealed class DriverHostActorFetchAndCacheBootstrapTests : RuntimeActorTes var cached = new CachedDeploymentArtifact( DeploymentId.NewId().ToString(), CachedRev.Value, blob, DateTimeOffset.UtcNow.AddHours(-1)); var cache = new StubArtifactCache(cached); - var publishProbe = CreateTestProbe(); + var publishProbe = this.CreateCleanOutcomePublishProbe(); var actor = Sys.ActorOf(DriverHostActor.Props( new ThrowingDbFactory(), TestNode, ackRouter: null, diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverHostActorFetchAndCacheUnreadableTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverHostActorFetchAndCacheUnreadableTests.cs index f7d2b8d7..a1963a56 100644 --- a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverHostActorFetchAndCacheUnreadableTests.cs +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverHostActorFetchAndCacheUnreadableTests.cs @@ -42,7 +42,7 @@ public sealed class DriverHostActorFetchAndCacheUnreadableTests : RuntimeActorTe [RevA.Value] = Artifact(), [RevB.Value] = null, }); - var publishProbe = CreateTestProbe(); + var publishProbe = this.CreateCleanOutcomePublishProbe(); var coordinator = CreateTestProbe(); var actor = Sys.ActorOf(DriverHostActor.Props( @@ -72,7 +72,7 @@ public sealed class DriverHostActorFetchAndCacheUnreadableTests : RuntimeActorTe // empty blob handed to it directly: ReconcileDriversFromBlob's #485 guard fails the apply. var db = NewInMemoryDbFactory(); var fetcher = new MapFetcher(new Dictionary { [RevA.Value] = Array.Empty() }); - var publishProbe = CreateTestProbe(); + var publishProbe = this.CreateCleanOutcomePublishProbe(); var coordinator = CreateTestProbe(); var actor = Sys.ActorOf(DriverHostActor.Props( @@ -95,7 +95,7 @@ public sealed class DriverHostActorFetchAndCacheUnreadableTests : RuntimeActorTe // A truncated/corrupt cached artifact surfaces as GetCurrentUnkeyedAsync == null (the cache's // integrity check is deliberately indistinguishable from a miss). Boot must land // Steady-no-revision and apply NOTHING — never a partial address space. - var publishProbe = CreateTestProbe(); + var publishProbe = this.CreateCleanOutcomePublishProbe(); var actor = Sys.ActorOf(DriverHostActor.Props( new ThrowingDbFactory(), TestNode, ackRouter: null, diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverHostActorHistoryWriterTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverHostActorHistoryWriterTests.cs index 76869edd..ab17f6ad 100644 --- a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverHostActorHistoryWriterTests.cs +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverHostActorHistoryWriterTests.cs @@ -53,7 +53,7 @@ public sealed class DriverHostActorHistoryWriterTests : RuntimeActorTestBase // A publish actor MUST be wired for SpawnVirtualTagHost to spawn the real host (its ctor // requires a non-null sink). No virtualTagHostOverride ⇒ the real VirtualTagHostActor is // spawned with _historyWriter threaded in. - var publish = CreateTestProbe(); + var publish = this.CreateCleanOutcomePublishProbe(); var writer = new CapturingHistoryWriter(); var actor = Sys.ActorOf(DriverHostActor.Props( diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverHostActorNativeAlarmTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverHostActorNativeAlarmTests.cs index 36701991..ffee6204 100644 --- a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverHostActorNativeAlarmTests.cs +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverHostActorNativeAlarmTests.cs @@ -556,7 +556,7 @@ public sealed class DriverHostActorNativeAlarmTests : RuntimeActorTestBase IDbContextFactory db, DeploymentId deploymentId) { var coordinator = CreateTestProbe(); - var publish = CreateTestProbe(); + var publish = this.CreateCleanOutcomePublishProbe(); var mux = CreateTestProbe(); var vtHost = CreateTestProbe(); diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverHostActorPrimaryGateTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverHostActorPrimaryGateTests.cs index e16e1b7c..11d0aaa9 100644 --- a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverHostActorPrimaryGateTests.cs +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverHostActorPrimaryGateTests.cs @@ -251,7 +251,7 @@ public sealed class DriverHostActorPrimaryGateTests : RuntimeActorTestBase IDbContextFactory db, DeploymentId dep, int driverMemberCount) { var coordinator = CreateTestProbe(); - var publish = CreateTestProbe(); + var publish = this.CreateCleanOutcomePublishProbe(); var mux = CreateTestProbe(); var vtHost = CreateTestProbe(); diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverHostActorTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverHostActorTests.cs index a712a339..eda42d74 100644 --- a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverHostActorTests.cs +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverHostActorTests.cs @@ -226,7 +226,7 @@ public sealed class DriverHostActorTests : RuntimeActorTestBase seeded.Length.ShouldBeGreaterThan(0); var coordinator = CreateTestProbe(); - var publish = CreateTestProbe(); + var publish = this.CreateCleanOutcomePublishProbe(); var actor = Sys.ActorOf(DriverHostActor.Props( db, TestNode, coordinator.Ref, localRoles: new HashSet { "driver" }, diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Harness/PublishProbes.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Harness/PublishProbes.cs new file mode 100644 index 00000000..dae22a2e --- /dev/null +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Harness/PublishProbes.cs @@ -0,0 +1,34 @@ +using Akka.Actor; +using Akka.TestKit; +using ZB.MOM.WW.OtOpcUa.Runtime.OpcUa; + +namespace ZB.MOM.WW.OtOpcUa.Runtime.Tests.Harness; + +/// +/// Gitea #601 — since the deploy ack waits for 's +/// reply, a bare standing in for the +/// publish actor would hold every Applied ack until the outcome timeout (and then ack it degraded). Tests +/// that are not ABOUT the ack's honesty use this probe instead: it still records every message exactly +/// like a plain probe (the auto-pilot runs in addition to queueing), and answers each +/// that asks for an outcome with a CLEAN one — the +/// reply the real publish actor gives a clean apply. +/// +public static class PublishProbes +{ + /// Creates a publish-actor probe that reports a clean address-space outcome for every rebuild + /// carrying . + /// The calling TestKit. + /// The probe. + public static TestProbe CreateCleanOutcomePublishProbe(this TestKitBase kit) + { + var probe = kit.CreateTestProbe(); + probe.SetAutoPilot(new DelegateAutoPilot((sender, message) => + { + if (message is OpcUaPublishActor.RebuildAddressSpace { ReportOutcome: true } rebuild) + sender.Tell(new OpcUaPublishActor.RebuildCompleted( + rebuild.Correlation, rebuild.DeploymentId, Degraded: false, DegradedReason: null)); + return AutoPilot.KeepRunning; + })); + return probe; + } +} diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Harness/V3ArtifactSeeder.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Harness/V3ArtifactSeeder.cs index da329e79..e928b1b2 100644 --- a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Harness/V3ArtifactSeeder.cs +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Harness/V3ArtifactSeeder.cs @@ -151,7 +151,7 @@ public static class V3ArtifactSeeder TimeSpan? reprovisionBackoff = null, TimeSpan? commsRecoveryAuditDelay = null) { - var publish = kit.CreateTestProbe(); + var publish = kit.CreateCleanOutcomePublishProbe(); var (actor, mux, coordinator) = Spawn(kit, db, localNode, publish.Ref, driverFactory, reprovisionBackoff, commsRecoveryAuditDelay); diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/OpcUa/OpcUaPublishActorApplyOutcomeTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/OpcUa/OpcUaPublishActorApplyOutcomeTests.cs new file mode 100644 index 00000000..f3e0602e --- /dev/null +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/OpcUa/OpcUaPublishActorApplyOutcomeTests.cs @@ -0,0 +1,200 @@ +using System.Text.Json; +using Akka.Actor; +using Microsoft.Extensions.Logging.Abstractions; +using Shouldly; +using Xunit; +using ZB.MOM.WW.OtOpcUa.Commons.OpcUa; +using ZB.MOM.WW.OtOpcUa.Commons.Types; +using ZB.MOM.WW.OtOpcUa.OpcUaServer; +using ZB.MOM.WW.OtOpcUa.Runtime.OpcUa; +using ZB.MOM.WW.OtOpcUa.Runtime.Tests.Harness; + +namespace ZB.MOM.WW.OtOpcUa.Runtime.Tests.OpcUa; + +/// +/// Gitea #601 — the deploy ack is sent only once the address-space apply outcome is known, so the +/// publish actor must REPORT that outcome when asked (): +/// clean for a clean apply, degraded for swallowed per-node sink faults or a pipeline throw, and nothing +/// at all when the caller did not ask (boot/restore paths re-ack nothing). +/// +/// +/// In the opcua-publish-meters collection because the degraded cases increment +/// otopcua.opcua.apply.failed, which the sibling classes assert stays at zero. +/// +[Collection("opcua-publish-meters")] +public sealed class OpcUaPublishActorApplyOutcomeTests : RuntimeActorTestBase +{ + [Fact] + public void A_clean_apply_reports_a_clean_outcome_to_the_requester() + { + var sink = new FaultingSink(throwOnEnsureFolder: false); + var applier = new AddressSpaceApplier(sink, NullLogger.Instance); + var actor = Sys.ActorOf(OpcUaPublishActor.PropsForTests(sink: sink, applier: applier)); + var correlation = CorrelationId.NewId(); + var deployment = DeploymentId.NewId(); + + actor.Tell(new OpcUaPublishActor.RebuildAddressSpace( + correlation, deployment, BuildEquipmentArtifact("eq-1"), ReportOutcome: true)); + + var done = ExpectMsg(PresenceBudget); + done.Correlation.ShouldBe(correlation); + done.DeploymentId.ShouldBe(deployment); + done.Degraded.ShouldBeFalse(); + done.DegradedReason.ShouldBeNull(); + } + + [Fact] + public void A_degraded_non_rebuild_apply_reports_degraded_with_the_failed_node_count() + { + // Every EnsureFolder throws — the applier swallows each and counts it, the classic degraded apply + // that leaves _lastApplied un-advanced and used to seal the deployment green. + var sink = new FaultingSink(throwOnEnsureFolder: true); + var applier = new AddressSpaceApplier(sink, NullLogger.Instance); + var actor = Sys.ActorOf(OpcUaPublishActor.PropsForTests(sink: sink, applier: applier)); + + actor.Tell(new OpcUaPublishActor.RebuildAddressSpace( + CorrelationId.NewId(), DeploymentId.NewId(), BuildEquipmentArtifact("eq-1"), ReportOutcome: true)); + + var done = ExpectMsg(PresenceBudget); + done.Degraded.ShouldBeTrue(); + done.DegradedReason.ShouldNotBeNull(); + done.DegradedReason.ShouldContain("failedNodes="); + } + + [Fact] + public void A_pipeline_that_throws_reports_degraded() + { + var sink = new FaultingSink(throwOnEnsureFolder: false); + var applier = new AddressSpaceApplier(sink, NullLogger.Instance); + var actor = Sys.ActorOf(OpcUaPublishActor.PropsForTests(sink: sink, applier: applier)); + + // Non-empty bytes that are not JSON: DeploymentArtifact.Open throws inside the rebuild pipeline. + actor.Tell(new OpcUaPublishActor.RebuildAddressSpace( + CorrelationId.NewId(), DeploymentId.NewId(), "not json"u8.ToArray(), ReportOutcome: true)); + + var done = ExpectMsg(PresenceBudget); + done.Degraded.ShouldBeTrue(); + done.DegradedReason.ShouldNotBeNull(); + } + + [Fact] + public void An_unchanged_composition_reports_clean() + { + var sink = new FaultingSink(throwOnEnsureFolder: false); + var applier = new AddressSpaceApplier(sink, NullLogger.Instance); + var actor = Sys.ActorOf(OpcUaPublishActor.PropsForTests(sink: sink, applier: applier)); + var artifact = BuildEquipmentArtifact("eq-1"); + + actor.Tell(new OpcUaPublishActor.RebuildAddressSpace(CorrelationId.NewId(), DeploymentId.NewId(), artifact, ReportOutcome: true)); + ExpectMsg(PresenceBudget).Degraded.ShouldBeFalse(); + + // Same composition again ⇒ empty plan ⇒ nothing to apply, nothing degraded. + actor.Tell(new OpcUaPublishActor.RebuildAddressSpace(CorrelationId.NewId(), DeploymentId.NewId(), artifact, ReportOutcome: true)); + ExpectMsg(PresenceBudget).Degraded.ShouldBeFalse(); + } + + [Fact] + public void Without_ReportOutcome_nothing_is_replied() + { + var sink = new FaultingSink(throwOnEnsureFolder: true); + var applier = new AddressSpaceApplier(sink, NullLogger.Instance); + var actor = Sys.ActorOf(OpcUaPublishActor.PropsForTests(sink: sink, applier: applier)); + + actor.Tell(new OpcUaPublishActor.RebuildAddressSpace( + CorrelationId.NewId(), DeploymentId.NewId(), BuildEquipmentArtifact("eq-1"))); + + // Positive control first: a follow-up REQUESTED outcome must arrive, which proves the first + // message was fully processed before the absence assertion is read. + actor.Tell(new OpcUaPublishActor.RebuildAddressSpace( + CorrelationId.NewId(), DeploymentId.NewId(), BuildEquipmentArtifact("eq-2"), ReportOutcome: true)); + var only = ExpectMsg(PresenceBudget); + only.Degraded.ShouldBeTrue(); + ExpectNoMsg(TimeSpan.FromMilliseconds(200)); + } + + [Fact] + public void Redeploying_the_same_composition_after_a_degraded_REBUILD_repairs_it_instead_of_reporting_clean() + { + // A degraded REBUILD advances _lastApplied (deliberately — see HandleRebuildCore), so a re-deploy of + // the same revision diffs to an EMPTY plan. Reporting that empty plan as clean would let the + // "re-deploy re-applies" repair path (#601) seal green over an address space still known partial. + var sink = new FaultingSink(throwOnEnsureFolder: false); + var applier = new AddressSpaceApplier(sink, NullLogger.Instance); + var actor = Sys.ActorOf(OpcUaPublishActor.PropsForTests(sink: sink, applier: applier)); + + actor.Tell(new OpcUaPublishActor.RebuildAddressSpace( + CorrelationId.NewId(), DeploymentId.NewId(), BuildEquipmentArtifact("eq-1", "First"), ReportOutcome: true)); + ExpectMsg(PresenceBudget).Degraded.ShouldBeFalse(); + + // A changed equipment row routes to a full rebuild; its re-materialise faults. + sink.ThrowOnEnsureFolder = true; + var changed = BuildEquipmentArtifact("eq-1", "Second"); + actor.Tell(new OpcUaPublishActor.RebuildAddressSpace(CorrelationId.NewId(), DeploymentId.NewId(), changed, ReportOutcome: true)); + var degraded = ExpectMsg(PresenceBudget); + degraded.Degraded.ShouldBeTrue(); + sink.RebuildCount.ShouldBe(1); + + // Same revision again, sink still faulting: must NOT claim clean. + actor.Tell(new OpcUaPublishActor.RebuildAddressSpace(CorrelationId.NewId(), DeploymentId.NewId(), changed, ReportOutcome: true)); + ExpectMsg(PresenceBudget).Degraded.ShouldBeTrue(); + + // Fault gone: the same revision re-materialises the whole composition and only THEN reports clean. + sink.ThrowOnEnsureFolder = false; + var foldersBefore = sink.EnsureFolderCount; + actor.Tell(new OpcUaPublishActor.RebuildAddressSpace(CorrelationId.NewId(), DeploymentId.NewId(), changed, ReportOutcome: true)); + ExpectMsg(PresenceBudget).Degraded.ShouldBeFalse(); + sink.EnsureFolderCount.ShouldBeGreaterThan(foldersBefore); + + // And now it is genuinely clean, so a further identical deploy is the ordinary empty no-op. + var foldersAfterRepair = sink.EnsureFolderCount; + actor.Tell(new OpcUaPublishActor.RebuildAddressSpace(CorrelationId.NewId(), DeploymentId.NewId(), changed, ReportOutcome: true)); + ExpectMsg(PresenceBudget).Degraded.ShouldBeFalse(); + sink.EnsureFolderCount.ShouldBe(foldersAfterRepair); + } + + private static byte[] BuildEquipmentArtifact(string equipmentId) => BuildEquipmentArtifact(equipmentId, equipmentId); + + private static byte[] BuildEquipmentArtifact(string equipmentId, string name) => + JsonSerializer.SerializeToUtf8Bytes(new + { + UnsAreas = new[] { new { UnsAreaId = "area-1", ClusterId = "c1", Name = "Area" } }, + UnsLines = new[] { new { UnsLineId = "line-1", UnsAreaId = "area-1", Name = "Line" } }, + Equipment = new[] + { + new + { + EquipmentId = equipmentId, + DriverInstanceId = (string?)null, + UnsLineId = "line-1", + Name = name, + MachineCode = equipmentId.ToUpperInvariant(), + }, + }, + DriverInstances = Array.Empty(), + Namespaces = Array.Empty(), + Tags = Array.Empty(), + ScriptedAlarms = Array.Empty(), + }); + + /// A sink that optionally throws from EnsureFolder (a swallowed per-node fault). + private sealed class FaultingSink(bool throwOnEnsureFolder) : IOpcUaAddressSpaceSink + { + public bool ThrowOnEnsureFolder { get; set; } = throwOnEnsureFolder; + public int EnsureFolderCount; + public int RebuildCount; + public void WriteValue(string nodeId, object? value, uint statusCode, DateTime sourceTimestampUtc, AddressSpaceRealm realm = AddressSpaceRealm.Uns) { } + public void WriteAlarmQuality(string alarmNodeId, uint statusCode, DateTime sourceTimestampUtc, AddressSpaceRealm realm) { } + public void WriteAlarmCondition(string alarmNodeId, AlarmConditionSnapshot state, DateTime occurredUtc, AddressSpaceRealm realm = AddressSpaceRealm.Uns) { } + public void MaterialiseAlarmCondition(string alarmNodeId, string equipmentNodeId, string displayName, string alarmType, int severity, AddressSpaceRealm realm, bool isNative = false) { } + public void EnsureFolder(string folderNodeId, string? parentNodeId, string displayName, AddressSpaceRealm realm = AddressSpaceRealm.Uns) + { + EnsureFolderCount++; + if (ThrowOnEnsureFolder) throw new InvalidOperationException("simulated sink fault"); + } + public void EnsureVariable(string variableNodeId, string? parentFolderNodeId, string displayName, string dataType, bool writable, AddressSpaceRealm realm, string? historianTagname = null, bool isArray = false, uint? arrayLength = null, bool refuseArrayOnScalar = false) { } + public void RebuildAddressSpace() => RebuildCount++; + public void RaiseNodesAddedModelChange(string affectedNodeId, AddressSpaceRealm realm = AddressSpaceRealm.Uns) { } + public void WireAlarmNotifiers(string alarmNodeId, AddressSpaceRealm alarmRealm, IReadOnlyList notifierFolderNodeIds, AddressSpaceRealm notifierFolderRealm) { } + public void AddReference(string sourceNodeId, AddressSpaceRealm sourceRealm, string targetNodeId, AddressSpaceRealm targetRealm, string referenceType = "Organizes") { } + } +} diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/OpcUa/OpcUaPublishActorRebuildTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/OpcUa/OpcUaPublishActorRebuildTests.cs index 93880e0c..2e0d239d 100644 --- a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/OpcUa/OpcUaPublishActorRebuildTests.cs +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/OpcUa/OpcUaPublishActorRebuildTests.cs @@ -880,7 +880,10 @@ public sealed class OpcUaPublishActorRebuildTests : RuntimeActorTestBase /// _lastApplied. A rebuild already re-materialised the whole composition, and the applier's /// one-way rebuild ratchet is the resync mechanism for anything it could not place — holding the /// composition back here would make every subsequent deploy re-diff the entire address space as added. - /// Proven by the third apply of the same composition being an empty-plan no-op. + /// Proven by the third apply of the same composition NOT rebuilding again (it diffs to an empty plan). + /// Gitea #601 review: that empty plan is no longer a silent no-op, though — the previous apply was a + /// degraded rebuild, so the identical re-deploy re-ensures the whole composition (idempotently, no + /// rebuild) and thereby repairs the node the rebuild could not place. /// [Fact] public void Rebuild_with_failures_still_advances_last_applied() @@ -904,13 +907,24 @@ public sealed class OpcUaPublishActorRebuildTests : RuntimeActorTestBase var afterSecond = sink.Calls.Count; // Re-deploy the SAME composition. _lastApplied advanced despite the failure, so this diffs to an - // empty plan and adds no sink calls at all. + // empty plan — no second rebuild (an un-advanced composition would re-diff ChangedEquipment and + // rebuild again) — but the degraded rebuild is remembered, so the failed variable is re-ensured. sink.FailEnsureVariableFor = null; var dep3 = SeedUnsRawDeployment(db, equipmentName: "station1-renamed", referencedTag: "a"); actor.Tell(new OpcUaPublishActor.RebuildAddressSpace(CorrelationId.NewId(), new DeploymentId(dep3))); - Thread.Sleep(300); + AwaitAssert( + () => sink.Calls.Skip(afterSecond).ShouldContain("EV:Plant/Modbus/dev1/b"), + duration: PresenceBudget); + sink.RebuildCalls.ShouldBe(1); - sink.Calls.Count.ShouldBe(afterSecond); + // Repaired ⇒ genuinely clean now, so a FOURTH identical deploy is the ordinary empty-plan no-op. + Thread.Sleep(200); + var afterRepair = sink.Calls.Count; + var dep4 = SeedUnsRawDeployment(db, equipmentName: "station1-renamed", referencedTag: "a"); + actor.Tell(new OpcUaPublishActor.RebuildAddressSpace(CorrelationId.NewId(), new DeploymentId(dep4))); + Thread.Sleep(300); + sink.Calls.Count.ShouldBe(afterRepair); + sink.RebuildCalls.ShouldBe(1); } /// Seal a deployment whose raw tree is Plant/Modbus/dev1 carrying one tag per name in diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Telemetry/PublishSeamEmitsToHubTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Telemetry/PublishSeamEmitsToHubTests.cs index 16be3cf8..4cfaf6c7 100644 --- a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Telemetry/PublishSeamEmitsToHubTests.cs +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Telemetry/PublishSeamEmitsToHubTests.cs @@ -232,7 +232,7 @@ public sealed class PublishSeamEmitsToHubTests : RuntimeActorTestBase IDbContextFactory db, DeploymentId deploymentId, ITelemetryLocalHub hub) { var coordinator = CreateTestProbe(); - var publish = CreateTestProbe(); + var publish = this.CreateCleanOutcomePublishProbe(); var mux = CreateTestProbe(); var vtHost = CreateTestProbe();