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