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

src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/AddressSpaceApplier.cs

@@ -1153,10 +1153,34 @@ public sealed class AddressSpaceApplier
     /// <returns><c>true</c> when the variable was ensured; <c>false</c> when the sink threw.</returns>
     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; }
     }
 
+    /// <summary>While non-null, every variable this applier ensures (realm, NodeId). See <see cref="BeginRecordingEnsuredVariables"/>.</summary>
+    private HashSet<(AddressSpaceRealm, string)>? _ensuredVariables;
+
+    /// <summary>
+    ///     Starts recording every variable the following materialise passes ensure, until
+    ///     <see cref="EndRecordingEnsuredVariables"/>. 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.
+    /// </summary>
+    /// <returns>The live set the recording fills.</returns>
+    public IReadOnlySet<(AddressSpaceRealm, string)> BeginRecordingEnsuredVariables()
+    {
+        _ensuredVariables = new HashSet<(AddressSpaceRealm, string)>();
+        return _ensuredVariables;
+    }
+
+    /// <summary>Stops the recording <see cref="BeginRecordingEnsuredVariables"/> started (the returned set is kept).</summary>
+    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

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);
 
     /// <summary>Test seams for the off-thread plan step. Production passes none.</summary>
-    public sealed class TestHooks
+    internal sealed class TestHooks
     {
         /// <summary>Runs on the thread pool at the start of every plan step (a test blocks it to hold a plan in flight).</summary>
-        public Action? BeforePlan { get; init; }
+        internal Action? BeforePlan { get; init; }
 
         /// <summary>Replaces <see cref="ParsedDeploymentArtifact.ParseDataPlaneAuthority"/> (a test makes it throw).</summary>
         internal Func<ParsedDeploymentArtifact, string, DataPlaneAuthorityProjection>? ParseAuthority { get; init; }
     }
 
     /// <summary>The default bound on <see cref="_deferred"/>: distinct (realm, id) keys held while a plan is in flight.</summary>
-    public const int DefaultDeferredWriteCapacity = 250_000;
+    internal const int DefaultDeferredWriteCapacity = 250_000;
 
     /// <summary>Identity of THIS actor instance (see <see cref="PlanReady.Incarnation"/>).</summary>
     private readonly object _incarnation = new();
@@ -178,6 +178,32 @@ public sealed class OpcUaPublishActor : ReceiveActor, IWithTimers
     /// </summary>
     private readonly DeferredWrites _deferred;
 
+    /// <summary>
+    ///     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 <see cref="ApplyPlan"/>, consumed by <see cref="FinishRebuild"/>.
+    ///     <list type="bullet">
+    ///         <item>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).</item>
+    ///         <item>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.</item>
+    ///     </list>
+    ///     (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.)
+    /// </summary>
+    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);
+    }
+
     /// <summary>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 <see cref="_lastApplied"/>, which only the apply advances — so a
     /// rebuild arriving meanwhile waits in <see cref="_pendingRebuilds"/>; 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)));
+
+    /// <summary>Produces the actor through its internal (seam-carrying) constructor, for <see cref="PropsForTests(IOpcUaAddressSpaceSink, AddressSpaceApplier?, TestHooks?, int, NodeId?, IDataPlaneAuthoritySink?)"/>.</summary>
+    private sealed class TestProducer(Func<OpcUaPublishActor> create) : IIndirectActorProducer
+    {
+        public Type ActorType => typeof(OpcUaPublishActor);
+        public ActorBase Produce() => create();
+        public void Release(ActorBase actor) { }
+    }
 
     /// <summary>Initializes a new instance of the <see cref="OpcUaPublishActor"/> class.</summary>
     /// <param name="sink">The OPC UA address space sink.</param>
@@ -383,8 +419,6 @@ public sealed class OpcUaPublishActor : ReceiveActor, IWithTimers
     /// ServiceLevel DB-reachability input is retired (<c>DbReachable</c> treated constant-true, only
     /// staleness demotes). No <paramref name="dbHealthProbe"/> is wired on such a node.</param>
     /// <param name="dataPlaneAuthority">The optional sink that receives the API keys and ACL grants carried by each applied artifact.</param>
-    /// <param name="deferredWriteCapacity">The bound on distinct nodes whose writes are held for replay while a rebuild plan is in flight.</param>
-    /// <param name="hooks">Test seams for the off-thread plan step; null in production.</param>
     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<DateTime>? utcNow = null,
-        int deferredWriteCapacity = DefaultDeferredWriteCapacity,
-        TestHooks? hooks = null)
+        Func<DateTime>? utcNow = null)
+        : this(sink, serviceLevel, subscribeRedundancyTopic, localNode, dbFactory, applier, dbHealthProbe, staleWindow,
+            probeFreshnessWindow, healthTickInterval, dbLess, dataPlaneAuthority, utcNow, DefaultDeferredWriteCapacity, hooks: null)
+    {
+    }
+
+    /// <summary>The full constructor: also the deferred-write bound and the test seams (internal — tests only).</summary>
+    /// <param name="deferredWriteCapacity">The bound on distinct nodes whose writes are held for replay while a rebuild plan is in flight.</param>
+    /// <param name="hooks">Test seams for the off-thread plan step; null in production.</param>
+    internal OpcUaPublishActor(
+        IOpcUaAddressSpaceSink sink,
+        IServiceLevelPublisher serviceLevel,
+        bool subscribeRedundancyTopic,
+        NodeId? localNode,
+        IDbContextFactory<OtOpcUaConfigDbContext>? dbFactory,
+        AddressSpaceApplier? applier,
+        IActorRef? dbHealthProbe,
+        TimeSpan? staleWindow,
+        TimeSpan? probeFreshnessWindow,
+        TimeSpan? healthTickInterval,
+        bool dbLess,
+        IDataPlaneAuthoritySink? dataPlaneAuthority,
+        Func<DateTime>? 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<SubscribeAck>(_ => { /* PubSub ack */ });
     }
 
+    /// <summary>
+    ///     Review M5 — a restart discards <see cref="_pendingRebuilds"/>; 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 (<see cref="IsFromAnotherIncarnation"/>).
+    /// </summary>
+    /// <inheritdoc />
+    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);
+    }
+
     /// <inheritdoc />
     protected override void PreStart()
     {
@@ -774,10 +848,11 @@ public sealed class OpcUaPublishActor : ReceiveActor, IWithTimers
 
     /// <summary>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).</summary>
-    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);
     }
 
     /// <summary>
@@ -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<AlarmStateUpdate> Alarms => _alarms.Values;
         public IEnumerable<AlarmQualityUpdate> 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);
+        /// <summary>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.</summary>
+        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<ValueWrite> Values, List<StatusWrite> Statuses) VariableWrites()
+        public (List<ValueWrite> Values, List<StatusWrite> Statuses) VariableWrites(Func<(AddressSpaceRealm, string), bool> include)
         {
             var values = new List<ValueWrite>();
             var statuses = new List<StatusWrite>();
             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.</summary>
     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();
+        }
+    }
+
+    /// <summary>The condition ids in <paramref name="next"/> that <paramref name="previous"/> did not materialise:
+    /// enabled scripted alarms (Uns, by ScriptedAlarmId) and native alarms (Raw, at their RawPath).</summary>
+    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