From a8284f499e287c70c89eafc1d34a8129baffbbe1 Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Thu, 22 Oct 2026 15:08:44 -0400 Subject: [PATCH] fix(opcua): scope the deferred-write replay to nodes the apply created (Lane B fix round 2) N1: a full alarm-state snapshot also sets the condition's Quality, so it now supersedes an earlier quality-only write for the same id in the deferred buffer; a quality arriving after the state is kept and replays after it. N2 / M1: the replay is not inert on a node the apply KEPT - an SDK-applied Acknowledge/Confirm/Shelve or an optimistic client Write in the window would be rolled back (with spurious Part 9 events). The replay is now scoped to nodes the apply created: variables the applier ensured during this apply (AddressSpaceApplier.BeginRecordingEnsuredVariables) or every variable after a full rebuild; conditions in the applied composition but not the previous one (also on a rebuild, which carries a kept condition's state over). Option (b), a sink that reports whether a live write landed, needs the Commons DeferredAddressSpaceSink seam outside this lane (follow-up F7). M2: the benchmark remark now matches its Category=Benchmark trait. M3: TestHooks and the seam-carrying constructor are internal; PropsForTests produces the actor through an IIndirectActorProducer. M4: pinned replay-nothing-and-drop-the-buffer after a failed plan. M5: PreRestart answers every queued rebuild requester (degraded). Refs #732 Claude-Session: https://claude.ai/code/session_01Ub9hGmYLQvn9XETzw7CMsT --- .../AddressSpaceApplier.cs | 26 +- .../OpcUa/OpcUaPublishActor.cs | 162 ++++++++++-- .../NodeManagerLockTimeBenchmarkTests.cs | 3 +- .../OpcUaPublishActorDeferredWriteTests.cs | 235 +++++++++++++++++- 4 files changed, 400 insertions(+), 26 deletions(-) diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/AddressSpaceApplier.cs b/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/AddressSpaceApplier.cs index 9d42e409..bb0f2752 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/AddressSpaceApplier.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/AddressSpaceApplier.cs @@ -1153,10 +1153,34 @@ public sealed class AddressSpaceApplier /// true when the variable was ensured; false when the sink threw. private bool SafeEnsureVariable(string nodeId, string? parentNodeId, string displayName, string dataType, bool writable, AddressSpaceRealm realm, string? historianTagname = null, bool isArray = false, uint? arrayLength = null, bool refuseArrayOnScalar = false) { - try { _sink.EnsureVariable(nodeId, parentNodeId, displayName, dataType, writable, realm, historianTagname, isArray, arrayLength, refuseArrayOnScalar); return true; } + try + { + _sink.EnsureVariable(nodeId, parentNodeId, displayName, dataType, writable, realm, historianTagname, isArray, arrayLength, refuseArrayOnScalar); + _ensuredVariables?.Add((realm, nodeId)); + return true; + } catch (Exception ex) { _logger.LogWarning(ex, "AddressSpaceApplier: EnsureVariable threw for {Node}", nodeId); return false; } } + /// While non-null, every variable this applier ensures (realm, NodeId). See . + private HashSet<(AddressSpaceRealm, string)>? _ensuredVariables; + + /// + /// Starts recording every variable the following materialise passes ensure, until + /// . The publish actor scopes the replay of writes it deferred while a + /// plan was in flight to these (a surgical apply ensures only its delta), so a variable the apply kept is not + /// rewritten. Single-caller (the publish actor's thread), like the rest of the apply. + /// + /// The live set the recording fills. + public IReadOnlySet<(AddressSpaceRealm, string)> BeginRecordingEnsuredVariables() + { + _ensuredVariables = new HashSet<(AddressSpaceRealm, string)>(); + return _ensuredVariables; + } + + /// Stops the recording started (the returned set is kept). + public void EndRecordingEnsuredVariables() => _ensuredVariables = null; + // The tag/vtag surgical-eligibility predicates moved to AddressSpaceChangeClassifier so the // whole delta classification is one pure, table-testable unit; the applier references them there (and the // classifier's Classify names the routing). The `with`-expression whitelist technique is preserved there 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 833525a0..94f2f24d 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/OpcUa/OpcUaPublishActor.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/OpcUa/OpcUaPublishActor.cs @@ -146,17 +146,17 @@ public sealed class OpcUaPublishActor : ReceiveActor, IWithTimers private sealed record PlanFailed(object Incarnation, RebuildAddressSpace Request, IActorRef Requester, Exception Error); /// Test seams for the off-thread plan step. Production passes none. - public sealed class TestHooks + internal sealed class TestHooks { /// Runs on the thread pool at the start of every plan step (a test blocks it to hold a plan in flight). - public Action? BeforePlan { get; init; } + internal Action? BeforePlan { get; init; } /// Replaces (a test makes it throw). internal Func? ParseAuthority { get; init; } } /// The default bound on : distinct (realm, id) keys held while a plan is in flight. - public const int DefaultDeferredWriteCapacity = 250_000; + internal const int DefaultDeferredWriteCapacity = 250_000; /// Identity of THIS actor instance (see ). private readonly object _incarnation = new(); @@ -178,6 +178,32 @@ public sealed class OpcUaPublishActor : ReceiveActor, IWithTimers /// private readonly DeferredWrites _deferred; + /// + /// Review fix round 2 (N2 / M1) — WHICH deferred writes the next replay may land: only the ones whose live write + /// could not have, i.e. onto nodes the apply just CREATED. A node the apply kept took the live write already, + /// and replaying it is not inert: a client changed it in the window (an SDK-applied Acknowledge / Confirm / + /// Shelve on a condition, an optimistic Write on a variable) and the replay would roll that back and fire + /// spurious Part 9 events. Set by , consumed by . + /// + /// Variables: after a FULL rebuild every one (each was recreated, its live write landed on the + /// discarded node); otherwise exactly those the apply asked the sink to ensure (the applier records them). + /// Conditions: those in the applied composition but not the previous one — on a rebuild too, because the + /// node manager carries a kept condition's state over to its re-materialised instance. + /// + /// (Precise "did the live write land" would need the sink to report it; the production sink chain runs through + /// Commons' DeferredAddressSpaceSink, outside this lane — follow-up F7.) + /// + private ReplayScope _replayScope = ReplayScope.Nothing; + + private sealed record ReplayScope( + bool AllVariables, + IReadOnlySet<(AddressSpaceRealm, string)> Variables, + IReadOnlySet<(AddressSpaceRealm, string)> Conditions) + { + private static readonly IReadOnlySet<(AddressSpaceRealm, string)> None = new HashSet<(AddressSpaceRealm, string)>(); + public static readonly ReplayScope Nothing = new(false, None, None); + } + /// True while a rebuild's parse / plan step is running off-thread. Rebuilds are applied strictly in /// arrival order — the plan is a diff against , which only the apply advances — so a /// rebuild arriving meanwhile waits in ; value, status and alarm writes do not — they @@ -357,9 +383,19 @@ public sealed class OpcUaPublishActor : ReceiveActor, IWithTimers int deferredWriteCapacity = DefaultDeferredWriteCapacity, NodeId? localNode = null, IDataPlaneAuthoritySink? dataPlaneAuthority = null) => - Akka.Actor.Props.Create(() => new OpcUaPublishActor( + // CreateBy, not Create(expression): Akka's expression path activates through a PUBLIC constructor, and the + // seam-carrying constructor is internal so the test seams stay off the public API (review M3). + Akka.Actor.Props.CreateBy(new TestProducer(() => new OpcUaPublishActor( sink, NullServiceLevelPublisher.Instance, false, localNode, null, applier, null, null, null, null, - false, dataPlaneAuthority, null, deferredWriteCapacity, hooks)); + false, dataPlaneAuthority, null, deferredWriteCapacity, hooks))); + + /// Produces the actor through its internal (seam-carrying) constructor, for . + private sealed class TestProducer(Func create) : IIndirectActorProducer + { + public Type ActorType => typeof(OpcUaPublishActor); + public ActorBase Produce() => create(); + public void Release(ActorBase actor) { } + } /// Initializes a new instance of the class. /// The OPC UA address space sink. @@ -383,8 +419,6 @@ public sealed class OpcUaPublishActor : ReceiveActor, IWithTimers /// ServiceLevel DB-reachability input is retired (DbReachable treated constant-true, only /// staleness demotes). No is wired on such a node. /// The optional sink that receives the API keys and ACL grants carried by each applied artifact. - /// The bound on distinct nodes whose writes are held for replay while a rebuild plan is in flight. - /// Test seams for the off-thread plan step; null in production. public OpcUaPublishActor( IOpcUaAddressSpaceSink sink, IServiceLevelPublisher serviceLevel, @@ -398,9 +432,31 @@ public sealed class OpcUaPublishActor : ReceiveActor, IWithTimers TimeSpan? healthTickInterval = null, bool dbLess = false, IDataPlaneAuthoritySink? dataPlaneAuthority = null, - Func? utcNow = null, - int deferredWriteCapacity = DefaultDeferredWriteCapacity, - TestHooks? hooks = null) + Func? utcNow = null) + : this(sink, serviceLevel, subscribeRedundancyTopic, localNode, dbFactory, applier, dbHealthProbe, staleWindow, + probeFreshnessWindow, healthTickInterval, dbLess, dataPlaneAuthority, utcNow, DefaultDeferredWriteCapacity, hooks: null) + { + } + + /// The full constructor: also the deferred-write bound and the test seams (internal — tests only). + /// The bound on distinct nodes whose writes are held for replay while a rebuild plan is in flight. + /// Test seams for the off-thread plan step; null in production. + internal OpcUaPublishActor( + IOpcUaAddressSpaceSink sink, + IServiceLevelPublisher serviceLevel, + bool subscribeRedundancyTopic, + NodeId? localNode, + IDbContextFactory? dbFactory, + AddressSpaceApplier? applier, + IActorRef? dbHealthProbe, + TimeSpan? staleWindow, + TimeSpan? probeFreshnessWindow, + TimeSpan? healthTickInterval, + bool dbLess, + IDataPlaneAuthoritySink? dataPlaneAuthority, + Func? utcNow, + int deferredWriteCapacity, + TestHooks? hooks) { _deferred = new DeferredWrites(Math.Max(1, deferredWriteCapacity)); _hooks = hooks; @@ -436,6 +492,24 @@ public sealed class OpcUaPublishActor : ReceiveActor, IWithTimers Receive(_ => { /* PubSub ack */ }); } + /// + /// Review M5 — a restart discards ; answer each queued requester (degraded) rather + /// than leave it to the deploy timeout. The in-flight plan's requester is answered when its plan pipes back to + /// the new instance (). + /// + /// + protected override void PreRestart(Exception reason, object message) + { + while (_pendingRebuilds.TryDequeue(out var queued)) + { + if (!queued.Request.ReportOutcome) continue; + queued.Requester.Tell(new RebuildCompleted(queued.Request.Correlation, queued.Request.DeploymentId, true, + "the address-space publisher restarted before this rebuild was applied; it was not applied")); + } + + base.PreRestart(reason, message); + } + /// protected override void PreStart() { @@ -774,10 +848,11 @@ public sealed class OpcUaPublishActor : ReceiveActor, IWithTimers /// Re-applies the deferred writes through the sink: values, then status-only annotations, then alarm /// states and alarm qualities. Each pass isolated like the live handlers; not re-metered (counted on arrival). - private void ReplayDeferredWrites() + private void ReplayDeferredWrites(ReplayScope scope) { if (_deferred.IsEmpty) return; - var (values, statuses) = _deferred.VariableWrites(); + var (values, statuses) = _deferred.VariableWrites( + key => scope.AllVariables || scope.Variables.Contains(key)); if (values.Count > 0) { try { _sink.WriteValues(values); } @@ -792,18 +867,20 @@ public sealed class OpcUaPublishActor : ReceiveActor, IWithTimers foreach (var alarm in _deferred.Alarms) { + if (!scope.Conditions.Contains((alarm.Realm, alarm.AlarmNodeId))) continue; try { _sink.WriteAlarmCondition(alarm.AlarmNodeId, alarm.State, alarm.TimestampUtc, alarm.Realm); } catch (Exception ex) { _log.Warning(ex, "OpcUaPublish: replaying a deferred alarm write threw for {Node}", alarm.AlarmNodeId); } } foreach (var quality in _deferred.AlarmQualities) { + if (!scope.Conditions.Contains((quality.Realm, quality.AlarmNodeId))) continue; try { _sink.WriteAlarmQuality(quality.AlarmNodeId, quality.StatusCode, quality.TimestampUtc, quality.Realm); } catch (Exception ex) { _log.Warning(ex, "OpcUaPublish: replaying a deferred alarm-quality write threw for {Node}", quality.AlarmNodeId); } } - _log.Debug("OpcUaPublish: replayed {Values} value, {Statuses} status and {Alarms} alarm writes deferred during a rebuild plan", - values.Count, statuses.Count, _deferred.AlarmCount); + _log.Debug("OpcUaPublish: replayed {Values} value and {Statuses} status writes (of {Held} held) deferred during a rebuild plan", + values.Count, statuses.Count, _deferred.Count); } /// @@ -830,7 +907,6 @@ public sealed class OpcUaPublishActor : ReceiveActor, IWithTimers public long Dropped { get; private set; } public int Count => _variables.Count + _alarms.Count + _qualities.Count; public bool IsEmpty => Count == 0; - public int AlarmCount => _alarms.Count + _qualities.Count; public IEnumerable Alarms => _alarms.Values; public IEnumerable AlarmQualities => _qualities.Values; @@ -863,16 +939,26 @@ public sealed class OpcUaPublishActor : ReceiveActor, IWithTimers return true; } - public bool Alarm(AlarmStateUpdate msg) => Put(_alarms, (msg.Realm, msg.AlarmNodeId), msg); + /// A full snapshot also sets the condition's Quality (from its StatusCode), so it SUPERSEDES an + /// earlier quality-only write for the same id (review N1): replayed state-then-quality, the stale quality + /// would overwrite the snapshot's. A quality arriving AFTER the state is kept and replays after it. + public bool Alarm(AlarmStateUpdate msg) + { + var key = (msg.Realm, msg.AlarmNodeId); + if (!Put(_alarms, key, msg)) return false; + _qualities.Remove(key); + return true; + } public bool AlarmQuality(AlarmQualityUpdate msg) => Put(_qualities, (msg.Realm, msg.AlarmNodeId), msg); - public (List Values, List Statuses) VariableWrites() + public (List Values, List Statuses) VariableWrites(Func<(AddressSpaceRealm, string), bool> include) { var values = new List(); var statuses = new List(); foreach (var ((realm, id), e) in _variables) { + if (!include((realm, id))) continue; if (e.HasValue) values.Add(new ValueWrite(id, e.Value, e.Status, e.TimestampUtc, realm)); else statuses.Add(new StatusWrite(id, e.Status, e.TimestampUtc, realm)); } @@ -921,7 +1007,8 @@ public sealed class OpcUaPublishActor : ReceiveActor, IWithTimers // Land what arrived while the plan was in flight on the nodes the apply just created — before the answer, // so "applied" means the deferred first values are on the address space too (review fix C1). - ReplayDeferredWrites(); + ReplayDeferredWrites(_replayScope); + _replayScope = ReplayScope.Nothing; if (msg.ReportOutcome) requester.Tell(new RebuildCompleted(msg.Correlation, msg.DeploymentId, outcome.Degraded, outcome.Reason)); @@ -931,6 +1018,44 @@ public sealed class OpcUaPublishActor : ReceiveActor, IWithTimers /// sink passes, announce. Returns whether the apply was degraded, and why. private (bool Degraded, string? Reason) ApplyPlan(PlanReady ready) { + var previous = _lastApplied; + var ensured = _applier!.BeginRecordingEnsuredVariables(); + try + { + var result = ApplyPlanCore(ready, out var rebuilt); + // What this apply created (see _replayScope). A throwing pass still scopes to what it did create. + _replayScope = new ReplayScope(rebuilt, ensured, AddedConditions(previous, ready.Composition)); + return result; + } + finally + { + _applier.EndRecordingEnsuredVariables(); + } + } + + /// The condition ids in that did not materialise: + /// enabled scripted alarms (Uns, by ScriptedAlarmId) and native alarms (Raw, at their RawPath). + private static HashSet<(AddressSpaceRealm, string)> AddedConditions( + AddressSpaceComposition previous, AddressSpaceComposition next) + { + var added = ConditionIds(next); + added.ExceptWith(ConditionIds(previous)); + return added; + + static HashSet<(AddressSpaceRealm, string)> ConditionIds(AddressSpaceComposition c) + { + var ids = new HashSet<(AddressSpaceRealm, string)>(); + foreach (var a in c.EquipmentScriptedAlarms) + if (a.Enabled) ids.Add((AddressSpaceRealm.Uns, a.ScriptedAlarmId)); + foreach (var t in c.RawTags) + if (t.Alarm is not null) ids.Add((AddressSpaceRealm.Raw, t.NodeId)); + return ids; + } + } + + private (bool Degraded, string? Reason) ApplyPlanCore(PlanReady ready, out bool rebuilt) + { + rebuilt = false; var msg = ready.Request; using var span = OtOpcUaTelemetry.StartAddressSpaceRebuildSpan(); span?.SetTag("otopcua.correlation_id", msg.Correlation.ToString()); @@ -981,6 +1106,7 @@ public sealed class OpcUaPublishActor : ReceiveActor, IWithTimers } var outcome = _applier!.Apply(plan); + rebuilt = outcome.RebuildCalled; // Sum swallowed per-node materialise failures across every pass together with Apply's own // removal-pass tally (archreview 01/S-1). A degraded apply used to vanish into per-node diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests/NodeManagerLockTimeBenchmarkTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests/NodeManagerLockTimeBenchmarkTests.cs index 865bef48..7940bd09 100644 --- a/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests/NodeManagerLockTimeBenchmarkTests.cs +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests/NodeManagerLockTimeBenchmarkTests.cs @@ -20,7 +20,8 @@ namespace ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests; /// /// The wall-clock budgets are generous for CI hardware; the before/after numbers are in the commit /// message. P3 is pinned by the bookkeeping's own visit counter, which does not depend on the machine. -/// Kept in the default run on purpose (not a Benchmark category the CI filter would have to learn to exclude): +/// Tagged Category=Benchmark, but no filter excludes that category (CI drops only E2E + LiveIntegration), +/// so it runs by default — on purpose: /// measured after the fix at 5–6 ms (P1) and 58–64 ms (P2) on a loaded dev machine against the 1 s budget — /// ≥ 15× headroom, so it does not flake on slow CI. The flip side: the pre-fix code (75–150 ms / 470–680 ms) also /// fits 1 s at these sizes, so the wall-clock legs guard against gross (order-of-magnitude) regressions only; the diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/OpcUa/OpcUaPublishActorDeferredWriteTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/OpcUa/OpcUaPublishActorDeferredWriteTests.cs index 7b795e3d..82bc6d8a 100644 --- a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/OpcUa/OpcUaPublishActorDeferredWriteTests.cs +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/OpcUa/OpcUaPublishActorDeferredWriteTests.cs @@ -235,6 +235,196 @@ public sealed class OpcUaPublishActorDeferredWriteTests : RuntimeActorTestBase sink.Exists(NewTag).ShouldBeTrue(); } + // ───────────────────── fix round 2: N1 / N2 / M1 / M4 / M5 ───────────────────── + + /// N1 — a full state snapshot carries quality, so it supersedes an EARLIER quality-only write for the + /// same condition (the engine sends a transition with the recovered worst quality and no QualityChanged). + [Fact] + public void A_state_after_a_quality_only_write_supersedes_it_on_replay() + { + var sink = new ExistenceSink(); + using var gate = new ManualResetEventSlim(false); + var actor = Spawn(sink, gate); + + actor.Tell(Rebuild(AlarmArtifact())); + actor.Tell(new OpcUaPublishActor.AlarmQualityUpdate("al-1", OpcUaStatus.BadNoCommunication, DateTime.UtcNow, AddressSpaceRealm.Uns)); + actor.Tell(new OpcUaPublishActor.AlarmStateUpdate("al-1", ActiveUnacked(OpcUaStatus.Good), DateTime.UtcNow, AddressSpaceRealm.Uns)); + AwaitAssert(() => sink.SkippedAlarms.Count.ShouldBe(2), PlanBudget); + gate.Set(); + ExpectMsg(PlanBudget); + + sink.AlarmQualityOf("al-1").ShouldBe(OpcUaStatus.Good, "the later snapshot's quality is what live would leave"); + sink.AlarmStateOf("al-1")!.Active.ShouldBeTrue(); + } + + /// N1 positive control — the reverse order: a quality-only write AFTER the state still lands last. + [Fact] + public void A_quality_only_write_after_a_state_still_lands_after_it_on_replay() + { + var sink = new ExistenceSink(); + using var gate = new ManualResetEventSlim(false); + var actor = Spawn(sink, gate); + + actor.Tell(Rebuild(AlarmArtifact())); + actor.Tell(new OpcUaPublishActor.AlarmStateUpdate("al-1", ActiveUnacked(OpcUaStatus.Good), DateTime.UtcNow, AddressSpaceRealm.Uns)); + actor.Tell(new OpcUaPublishActor.AlarmQualityUpdate("al-1", OpcUaStatus.BadNoCommunication, DateTime.UtcNow, AddressSpaceRealm.Uns)); + AwaitAssert(() => sink.SkippedAlarms.Count.ShouldBe(2), PlanBudget); + gate.Set(); + ExpectMsg(PlanBudget); + + sink.AlarmQualityOf("al-1").ShouldBe(OpcUaStatus.BadNoCommunication); + } + + /// + /// N2 — a condition the plan KEEPS took the live write already; replaying it would roll back an operator's + /// Acknowledge applied in the window (the SDK mutates the node directly) and fire spurious events. Driven over + /// both apply shapes: a surgical pure-add, and a full rebuild (which carries the condition's state over). + /// + [Theory] + [InlineData(false)] + [InlineData(true)] + public void An_out_of_band_ack_on_a_kept_condition_is_not_reverted_by_the_replay(bool rebuild) + { + var sink = new ExistenceSink(); + using var gate = new ManualResetEventSlim(true); + var actor = Spawn(sink, gate); + actor.Tell(Rebuild(AlarmArtifact())); + ExpectMsg(PlanBudget).Degraded.ShouldBeFalse(); + sink.ConditionExists("al-1").ShouldBeTrue(); + + gate.Reset(); + // A surgical pure-add (new raw tags) or a full rebuild (an equipment rename) that keeps al-1. + actor.Tell(Rebuild(rebuild ? AlarmArtifact(equipmentName: "Eq-renamed") : AlarmArtifact(rawTags: 2))); + actor.Tell(new OpcUaPublishActor.AlarmStateUpdate("al-1", ActiveUnacked(OpcUaStatus.Good), DateTime.UtcNow, AddressSpaceRealm.Uns)); + AwaitAssert(() => sink.AlarmWritesTo("al-1").ShouldBe(1), PlanBudget); // the live write landed + sink.AcknowledgeOutOfBand("al-1"); // an operator acks in the window + + gate.Set(); + ExpectMsg(PlanBudget).Degraded.ShouldBeFalse(); + + sink.SawRebuild.ShouldBe(rebuild, "non-vacuity: the intended apply shape ran"); + sink.AlarmStateOf("al-1")!.Acknowledged.ShouldBeTrue("the operator's ack must survive the apply"); + sink.AlarmWritesTo("al-1").ShouldBe(1, "no replayed write (and so no extra Part 9 event) on a kept condition"); + } + + /// M1 — a variable the surgical plan keeps took the live write; replaying it would revert a client + /// Write applied in the window. Positive control in the same apply: the node the plan ADDS gets its value. + [Fact] + public void A_surgical_apply_replays_values_only_onto_the_variables_it_added() + { + var sink = new ExistenceSink(); + using var gate = new ManualResetEventSlim(true); + var actor = Spawn(sink, gate); + actor.Tell(Rebuild(RawTagArtifact(1))); + ExpectMsg(PlanBudget); + const string kept = "Plant/Modbus/dev1/T0"; + + gate.Reset(); + actor.Tell(Rebuild(RawTagArtifact(3))); + var at = DateTime.UtcNow; + actor.Tell(new OpcUaPublishActor.AttributeValueBatchUpdate( + [ + new ValueWrite(kept, 5, OpcUaStatus.Good, at, AddressSpaceRealm.Raw), + new ValueWrite(NewTag, 7, OpcUaStatus.Good, at, AddressSpaceRealm.Raw), + ])); + AwaitAssert(() => sink.ValueOf(kept)!.Value.Value.ShouldBe(5), PlanBudget); + sink.SetValueOutOfBand(kept, 9); // a client Write in the window + + gate.Set(); + ExpectMsg(PlanBudget); + + sink.SawRebuild.ShouldBeFalse("non-vacuity: a surgical apply"); + sink.ValueOf(kept)!.Value.Value.ShouldBe(9, "the kept node's client write is not reverted"); + sink.ValueOf(NewTag)!.Value.Value.ShouldBe(7, "positive control: the added node gets its deferred value"); + } + + /// M1 — a FULL rebuild recreates every variable, so every deferred value replays (the live write landed + /// on a node the rebuild discarded). + [Fact] + public void A_rebuild_replays_every_deferred_value() + { + var sink = new ExistenceSink(); + using var gate = new ManualResetEventSlim(true); + var actor = Spawn(sink, gate); + actor.Tell(Rebuild(AlarmArtifact(rawTags: 1))); + ExpectMsg(PlanBudget); + const string kept = "Plant/Modbus/dev1/T0"; + sink.Exists(kept).ShouldBeTrue(); + + gate.Reset(); + // An equipment rename is a ChangedEquipment delta ⇒ full rebuild; the raw tag keeps its path. + actor.Tell(Rebuild(AlarmArtifact(equipmentName: "Eq-renamed", rawTags: 1))); + var at = DateTime.UtcNow; + actor.Tell(new OpcUaPublishActor.AttributeValueBatchUpdate([new ValueWrite(kept, 5, OpcUaStatus.Good, at, AddressSpaceRealm.Raw)])); + AwaitAssert(() => sink.ValueOf(kept)!.Value.Value.ShouldBe(5), PlanBudget); + + gate.Set(); + ExpectMsg(PlanBudget); + + sink.SawRebuild.ShouldBeTrue("non-vacuity: a full rebuild ran"); + sink.ValueOf(kept)!.Value.Value.ShouldBe(5, "the recreated node gets the value its predecessor took"); + } + + /// M4 — a FAILED plan applies nothing, so nothing is replayed, and the buffer is dropped once the + /// pipeline is idle: a later plan that adds the node does not receive a write from the failed window. + [Fact] + public void A_failed_plan_replays_nothing_and_drops_the_buffer() + { + var sink = new ExistenceSink(); + using var gate = new ManualResetEventSlim(true); + var actor = Spawn(sink, gate); + actor.Tell(Rebuild(RawTagArtifact(2))); + ExpectMsg(PlanBudget); + + gate.Reset(); + actor.Tell(Rebuild("not json"u8.ToArray())); + actor.Tell(new OpcUaPublishActor.AttributeValueBatchUpdate( + [new ValueWrite("Plant/Modbus/dev1/T2", 42, OpcUaStatus.Good, DateTime.UtcNow, AddressSpaceRealm.Raw)])); + AwaitAssert(() => sink.SkippedWrites.ShouldContain("Plant/Modbus/dev1/T2"), PlanBudget); + // Snapshot AFTER the live write (the sink's counter cannot tell a live write after an apply from a replay). + var replayedBefore = sink.ReplayedItems; + gate.Set(); + ExpectMsg(PlanBudget).Degraded.ShouldBeTrue(); + sink.ReplayedItems.ShouldBe(replayedBefore, "nothing was applied, so nothing is replayed"); + + actor.Tell(Rebuild(RawTagArtifact(3))); + ExpectMsg(PlanBudget).Degraded.ShouldBeFalse(); + sink.Exists("Plant/Modbus/dev1/T2").ShouldBeTrue("positive control: the later plan adds it"); + sink.ValueOf("Plant/Modbus/dev1/T2").ShouldBeNull("the failed window's buffer was dropped"); + } + + /// M5 — a restart answers every QUEUED requester (degraded) instead of leaving it to the deploy timeout; + /// the in-flight one is answered by the incarnation check. + [Fact] + public void A_restart_answers_queued_rebuild_requesters() + { + var sink = new ExistenceSink(); + using var gate = new ManualResetEventSlim(false); + var applier = new AddressSpaceApplier(sink, NullLogger.Instance); + var childProps = OpcUaPublishActor.PropsForTests(sink, applier, new OpcUaPublishActor.TestHooks { BeforePlan = () => gate.Wait() }); + var parent = Sys.ActorOf(Props.Create(() => new RestartingParent(childProps))); + parent.Tell(RestartingParent.GetChild.Instance); + var child = ExpectMsg(); + + var inFlight = Rebuild(RawTagArtifact(2)); + var queued = Rebuild(RawTagArtifact(3)); + child.Tell(inFlight); + child.Tell(queued); + child.Tell(Kill.Instance); + + var first = ExpectMsg(PlanBudget); + first.Correlation.ShouldBe(queued.Correlation, "the queued requester is answered at the restart"); + first.Degraded.ShouldBeTrue(); + + gate.Set(); + var second = ExpectMsg(PlanBudget); + second.Correlation.ShouldBe(inFlight.Correlation); + second.Degraded.ShouldBeTrue(); + } + + private static AlarmConditionSnapshot ActiveUnacked(uint quality) => + new(Active: true, Acknowledged: false, Confirmed: true, Enabled: true, AlarmShelvingKind.Unshelved, 500, "high", quality); + // ───────────────────────────── helpers ───────────────────────────── private IActorRef Spawn(ExistenceSink sink, ManualResetEventSlim gate) => @@ -264,17 +454,30 @@ public sealed class OpcUaPublishActorDeferredWriteTests : RuntimeActorTestBase ScriptedAlarms = Array.Empty(), }); - private static byte[] AlarmArtifact() => + private static byte[] AlarmArtifact(string equipmentName = "Eq", int rawTags = 0) => 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 = "eq-1", DriverInstanceId = (string?)null, UnsLineId = "line-1", Name = "Eq", MachineCode = "EQ-1" }, + new { EquipmentId = "eq-1", DriverInstanceId = (string?)null, UnsLineId = "line-1", Name = equipmentName, MachineCode = "EQ-1" }, }, - DriverInstances = Array.Empty(), - Tags = Array.Empty(), + RawFolders = rawTags == 0 + ? Array.Empty() + : new object[] { new { RawFolderId = "rf-1", ParentRawFolderId = (string?)null, Name = "Plant", ClusterId = "c1" } }, + DriverInstances = rawTags == 0 + ? Array.Empty() + : new object[] { new { DriverInstanceId = "drv-1", DriverType = "Modbus", DriverConfig = "{}", Name = "Modbus", RawFolderId = "rf-1", ClusterId = "c1" } }, + Devices = rawTags == 0 + ? Array.Empty() + : new object[] { new { DeviceId = "dev-1", DriverInstanceId = "drv-1", Name = "dev1", DeviceConfig = "{}" } }, + TagGroups = Array.Empty(), + Tags = Enumerable.Range(0, rawTags).Select(i => (object)new + { + TagId = $"t-{i}", DeviceId = "dev-1", TagGroupId = (string?)null, Name = $"T{i}", + DataType = "Float", AccessLevel = 0, TagConfig = "{}", + }).ToArray(), ScriptedAlarms = new[] { new @@ -334,6 +537,22 @@ public sealed class OpcUaPublishActorDeferredWriteTests : RuntimeActorTestBase private readonly Dictionary _alarmQualities = new(); public ConcurrentQueue SkippedWrites { get; } = new(); + private readonly Dictionary _alarmWrites = new(); + private volatile bool _sawRebuild; + + /// Whether any apply called (the full-rebuild shape). + public bool SawRebuild => _sawRebuild; + public int AlarmWritesTo(string id) { lock (_gate) return _alarmWrites.GetValueOrDefault(id); } + public void SetValueOutOfBand(string id, object? value) + { + lock (_gate) { var c = _values.GetValueOrDefault(id); _values[id] = (value, c.Status, c.At); } + } + + /// What an SDK-applied Part 9 Acknowledge does: mutates the condition directly, no sink write. + public void AcknowledgeOutOfBand(string id) + { + lock (_gate) _alarmStates[id] = _alarmStates[id] with { Acknowledged = true }; + } public ConcurrentQueue SkippedAlarms { get; } = new(); public int ReplayedItems; @@ -378,7 +597,10 @@ public sealed class OpcUaPublishActorDeferredWriteTests : RuntimeActorTestBase lock (_gate) { if (!_conditions.Contains(alarmNodeId)) { SkippedAlarms.Enqueue(alarmNodeId); return; } + _alarmWrites[alarmNodeId] = _alarmWrites.GetValueOrDefault(alarmNodeId) + 1; + // The node manager projects the snapshot's StatusCode onto the condition's Quality. _alarmStates[alarmNodeId] = state; + _alarmQualities[alarmNodeId] = state.StatusCode; } } @@ -410,12 +632,13 @@ public sealed class OpcUaPublishActorDeferredWriteTests : RuntimeActorTestBase { lock (_gate) { + // Variables are recreated empty; a condition's state is captured and carried over to its + // re-materialised instance (the node manager's rebuild carry-over). _applyClosed = false; + _sawRebuild = true; _variables.Clear(); _conditions.Clear(); _values.Clear(); - _alarmStates.Clear(); - _alarmQualities.Clear(); } }