fix(opcua): replay writes deferred during an off-thread plan; cache-ref and guard fixes (Lane B fix round 1)

C1 (critical): B6 computes the plan off the pinned thread while B1 made writes non-creating, so a value/status/alarm write for a node the in-flight plan adds was skipped, and change-driven drivers never republish it (BadWaitingForInitialData indefinitely; on a full rebuild every re-subscribe value landed on discarded nodes). OpcUaPublishActor now applies writes immediately AND remembers the latest per (realm, id) while a plan is in flight or queued (value+status compose; alarm state/quality coalesce to the latest, which the node's delta gate makes inert on an existing condition), replays it after each apply before answering, and drops it once the pipeline is idle. Bounded (250k keys by default; beyond it new keys are not held, logged once). I2: the data-plane-authority parse runs in its own try inside the plan task; a throw is logged and never degrades the address-space apply. I3: the SDK's DeleteMonitoredItems releases an item's component-cache reference on a Good delete; a tombstoned item's reference was already released, so the delete took one from the re-created node's entry. Released items are remembered and the orphan delete takes one back first. Event items never held a reference (EventManager + SubscribeToEvents), so the three event-item releases are gone. I4: direct tests for the HistoryUpdate override's per-request updater check and the HistoryReadEvents override's source guard (both reached by restoring the bit after a reset - the B7 race window); fixed the false remark. Minors: the MaterialiseAlarmCondition kind-swap tombstones the dropped instance; a bytes-less rebuild arriving mid-plan queues behind it (replies in order); a PlanReady/PlanFailed from a previous actor incarnation is ignored and its requester answered degraded; the tie-cluster end bound of an uncapped read fits against the backend ceiling (defence in depth - guarded by a sweep test); the lock-time benchmark stays in the default run, headroom noted. Refs #732 Claude-Session: https://claude.ai/code/session_01Ub9hGmYLQvn9XETzw7CMsT
2026-10-22 14:30:11 -04:00
parent 7c7ecff974
commit f966efcfad
9 changed files with 977 additions and 30 deletions
@@ -139,6 +139,10 @@ public sealed class StaleMonitoredNodeWireTests
                "the new item must receive a value written after its creation");
            lock (gate) newValues.ShouldNotContain(v => v.StatusCode == StatusCodes.BadNodeIdUnknown);

+            // Review fix I3: the cache entry the new item holds is ITS reference alone — the old item's was released
+            // by the removal.
+            CacheRefCount(nm, nodeId).ShouldBe(1, "one live item, one component-cache reference");
+
            // The old item was told, and the dead node's fan-out is gone.
            await WaitUntilAsync(() => { lock (gate) return oldValues.Any(v => v.StatusCode == StatusCodes.BadNodeIdUnknown); }, TimeSpan.FromSeconds(5),
                "the item on the removed node must be told BadNodeIdUnknown");
@@ -150,9 +154,18 @@ public sealed class StaleMonitoredNodeWireTests
            (oldItem.Status.Error is null || ServiceResult.IsGood(oldItem.Status.Error)).ShouldBeTrue(
                $"deleting the orphaned item must succeed, got {oldItem.Status.Error}");

+            // The SDK releases a deleted item's cache reference on a Good delete; the old item's was released by the
+            // removal already, so a second release would evict the NEW item's live entry (review fix I3).
+            CacheRefCount(nm, nodeId).ShouldBe(1, "deleting the orphaned item must not take the new item's reference");
+
            // And the new item is unaffected by that delete.
            nm.WriteValue(Tag, 44, OpcUaStatus.Good, DateTime.UtcNow, AddressSpaceRealm.Raw);
            await WaitUntilAsync(() => { lock (gate) return newValues.Any(v => Equals(v.Value, 44)); }, TimeSpan.FromSeconds(5));
+
+            // Positive control: deleting the live item gives its reference back.
+            subscription.RemoveItem(newItem);
+            await subscription.ApplyChangesAsync(ct);
+            CacheRefCount(nm, nodeId).ShouldBe(0);
        }
        finally
        {
@@ -160,6 +173,28 @@ public sealed class StaleMonitoredNodeWireTests
        }
    }

+    /// <summary>The SDK's component-cache reference count for <paramref name="nodeId"/> (private SDK state, read by
+    /// reflection: <c>CustomNodeManager2.m_componentCache</c> → <c>CacheEntry.RefCount</c>); 0 when no entry.</summary>
+    private static int CacheRefCount(OtOpcUaNodeManager nm, NodeId nodeId)
+    {
+        const System.Reflection.BindingFlags Any =
+            System.Reflection.BindingFlags.Instance | System.Reflection.BindingFlags.Public | System.Reflection.BindingFlags.NonPublic;
+        var field = typeof(Opc.Ua.Server.CustomNodeManager2).GetField("m_componentCache", Any)
+            .ShouldNotBeNull("the SDK's component cache field (SDK internals moved?)");
+        lock (nm.Lock)
+        {
+            if (field.GetValue(nm) is not System.Collections.IDictionary cache || !cache.Contains(nodeId)) return 0;
+            var entry = cache[nodeId]!;
+            var member = (object?)entry.GetType().GetField("RefCount", Any) ?? entry.GetType().GetProperty("RefCount", Any);
+            return member switch
+            {
+                System.Reflection.FieldInfo f => (int)f.GetValue(entry)!,
+                System.Reflection.PropertyInfo p => (int)p.GetValue(entry)!,
+                _ => throw new InvalidOperationException("CacheEntry.RefCount not found"),
+            };
+        }
+    }
+
    private static void Materialise(OtOpcUaNodeManager nm)
    {
        nm.EnsureFolder(Device, null, "dev1", AddressSpaceRealm.Raw);
@@ -254,6 +254,38 @@ public sealed class HistoryUpdateTests : IDisposable
        finally { await host.DisposeAsync(); }
    }

+    /// <summary>
+    ///     Review fix I4 — the override's OWN per-request updater check, which the bit retraction above now hides.
+    ///     It is the B7 race window: a request the SDK base admitted on the HistoryWrite bit just before the updater
+    ///     was cleared. Reconstructed by restoring the bit after the clear, so the request reaches the override with
+    ///     no updater; it must answer unsupported on both slots (the override sets results itself) and never reach the
+    ///     old updater.
+    /// </summary>
+    [Fact]
+    public async Task The_override_rechecks_the_updater_per_request_when_the_bit_is_still_set()
+    {
+        var (host, server) = await BootAsync();
+        try
+        {
+            var nm = server.NodeManager!;
+            var updater = new FakeUpdater();
+            nm.HistorianDataUpdater = updater;
+            WireAcl(nm, enabled: false, NodePermissions.HistoryUpdate);
+            var nodeId = Historized(nm, AddressSpaceRealm.Raw, RawPath);
+            nm.HistorianDataUpdater = null;
+            var variable = nm.TryGetVariable(RawPath, AddressSpaceRealm.Raw)!;
+            variable.AccessLevel |= AccessLevels.HistoryWrite;
+            variable.UserAccessLevel |= AccessLevels.HistoryWrite;
+
+            var (results, errors) = Insert(nm, KeyIdentity(OpcUaDataPlaneRoles.HistoryUpdate), nodeId, Value(T1, 1d));
+
+            errors[0].StatusCode.Code.ShouldBe(StatusCodes.BadHistoryOperationUnsupported);
+            results[0].StatusCode.Code.ShouldBe(StatusCodes.BadHistoryOperationUnsupported, "set by the override, not left for the master node manager");
+            updater.UpdateCalls.ShouldBeEmpty();
+        }
+        finally { await host.DisposeAsync(); }
+    }
+
    /// <summary>With an updater wired, both the raw node and its UNS mirror carry HistoryWrite; a non-historized node does not.</summary>
    [Fact]
    public async Task An_updater_grants_history_write_to_raw_and_uns_historized_nodes_only()
@@ -321,6 +321,54 @@ public sealed class NodeManagerHistoryReadBoundsTests(BoundsServerFixture fx) :
        cpWhole.ShouldBeNull();
    }

+    /// <summary>
+    ///     Lane B review fix round 1 (minor): an UNCAPPED read (N = 0) against a backend with a row ceiling L pages at
+    ///     L, and MaxReturnDataValues advertises L — so no page, the tie-cluster slices and their end bound included,
+    ///     may carry more than L values; and the pages still concatenate to the uncapped single-shot answer. Swept
+    ///     over ceilings and tie-cluster sizes up to what one over-fetch read can hold (L − 1).
+    /// </summary>
+    [Fact]
+    public async Task An_uncapped_read_never_returns_a_page_larger_than_the_backend_ceiling()
+    {
+        foreach (var ceiling in new[] { 2, 3, 4, 6 })
+        {
+            for (var ties = 1; ties < ceiling; ties++)
+            {
+                var series = new List<DataValueSnapshot> { Snap("5:00", 0) };
+                for (var i = 0; i < ties; i++) series.Add(Snap("5:02", 10 + i));
+                series.Add(Snap("5:03", 3));
+                for (var i = 0; i < ties; i++) series.Add(Snap("5:05", 50 + i));
+                series.Add(Snap("5:06", 6));
+
+                foreach (var (start, end) in new[] { ("5:01", "5:04"), ("5:00", "5:06"), ("5:01", "5:07"), ("5:06", "5:01") })
+                {
+                    var expected = Render(
+                        (await ReadAsync(new SeriesSource(series, reverse: true, inclusiveForwardEnd: false, resolution: 1), T(start), T(end), 0, true, null)).Values,
+                        T(start), T(end));
+
+                    var capped = new SeriesSource(series, reverse: true, inclusiveForwardEnd: false, resolution: 1, ceiling: ceiling);
+                    var all = new List<DataValue>();
+                    byte[]? cp = null;
+                    var pages = 0;
+                    do
+                    {
+                        var (values, error, next) = await ReadAsync(capped, T(start), T(end), 0, true, cp);
+                        error.StatusCode.Code.ShouldBe(StatusCodes.Good, $"L={ceiling} ties={ties} {start}->{end}");
+                        values.Count.ShouldBeLessThanOrEqualTo(ceiling, $"L={ceiling} ties={ties} {start}->{end} page {pages}");
+                        all.AddRange(values);
+                        cp = next;
+                        (++pages).ShouldBeLessThan(60, "paging must terminate");
+                    }
+                    while (cp is not null);
+
+                    Render(all, T(start), T(end)).ShouldBe(expected, $"L={ceiling} ties={ties} {start}->{end}");
+                }
+            }
+        }
+
+        static DataValueSnapshot Snap(string at, double v) => new(v, StatusCodes.Good, T(at), T(at));
+    }
+
    /// <summary>
    ///     #691 review: no data AND no bound found ⇒ <c>GoodNoData</c> with no values — the answer a tag with no history (and a
    ///     disabled historian) gave before bounds (§6.5.3.2: bounds make the result Good only "if one or both exist"). Only
@@ -249,6 +249,45 @@ public sealed class NodeManagerHistoryReadEventsTests : IDisposable
        await host.DisposeAsync();
    }

+    /// <summary>
+    ///     Review fix I4 — the HistoryReadEvents override's OWN source guard. Every other "unsupported" case here is
+    ///     answered by the SDK base, which admits only notifiers carrying the <c>EventNotifier.HistoryRead</c> bit —
+    ///     and since B7 the bit and the source registration move together. So the guard is reached only by a request
+    ///     the base admitted on the bit just before a source reset took the registration back; reconstructed by
+    ///     restoring the bit after the reset. The override must answer unsupported on both slots and reach no source.
+    /// </summary>
+    [Fact]
+    public async Task The_event_override_refuses_a_notifier_with_the_bit_but_no_registered_source()
+    {
+        var (host, server) = await BootAsync();
+        var nm = server.NodeManager!;
+        var fake = new RecordingHistorianDataSource();
+        nm.HistorianDataSource = fake;
+        nm.EnsureFolder("eq-guard", parentNodeId: null, displayName: "Equipment", realm: AddressSpaceRealm.Uns);
+        nm.MaterialiseAlarmCondition("alarm-g", "eq-guard", "Cond", "OffNormalAlarm", severity: 200, realm: AddressSpaceRealm.Uns);
+        nm.IsEventHistorySource("eq-guard").ShouldBeTrue("non-vacuity: registered while the source was wired");
+
+        nm.HistorianDataSource = null!;
+        var folder = nm.TryGetFolder("eq-guard")!;
+        nm.IsEventHistorySource("eq-guard").ShouldBeFalse();
+        folder.EventNotifier |= EventNotifiers.HistoryRead;
+
+        var details = new ReadEventDetails
+        {
+            StartTime = DateTime.UtcNow.AddHours(-1),
+            EndTime = DateTime.UtcNow,
+            NumValuesPerNode = 10,
+            Filter = SelectFilter("EventId"),
+        };
+        var (results, errors) = InvokeHistoryRead(server, nm, details, folder.NodeId);
+
+        errors[0].StatusCode.Code.ShouldBe(StatusCodes.BadHistoryOperationUnsupported);
+        results[0].StatusCode.Code.ShouldBe(StatusCodes.BadHistoryOperationUnsupported, "set by the override's guard");
+        fake.EventsCalled.ShouldBeFalse();
+
+        await host.DisposeAsync();
+    }
+
    /// <summary>A variable node targeted by a HistoryReadEvents request ⇒ BadHistoryOperationUnsupported;
    /// the source is never invoked.
    /// <para>
@@ -259,8 +298,8 @@ public sealed class NodeManagerHistoryReadEventsTests : IDisposable
    ///     SDK base therefore rejects it and does NOT pass it to our <c>HistoryReadEvents</c> override;
    ///     the Bad result comes from the base's pre-seeding, not from our source-guard. This test pins
    ///     that base-level rejection of variable nodes for event reads.
-    ///     The override's own source-guard (miss in <c>_eventNotifierSources</c>) is exercised by the
-    ///     <c>Events_folder_promoted_without_source_is_unsupported_until_a_source_is_wired</c> test instead.
+    ///     The override's own source-guard (a miss in <c>_eventNotifierSources</c>) is exercised by
+    ///     <c>The_event_override_refuses_a_notifier_with_the_bit_but_no_registered_source</c>.
    /// </para>
    /// </summary>
    [Fact]
@@ -20,6 +20,11 @@ namespace ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests;
 ///     </list>
 ///     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):
+///     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
+///     machine-independent P3 visit counter is the strict pin.
 /// </summary>
 [Trait("Category", "Benchmark")]
 public sealed class NodeManagerLockTimeBenchmarkTests(ITestOutputHelper output) : IDisposable
@@ -93,6 +93,39 @@ public sealed class NodeManagerStaleMonitoredNodeTests : IDisposable
        finally { await host.DisposeAsync(); }
    }

+    /// <summary>
+    ///     Review fix round 1 (minor): the same-id kind swap (scripted ⇄ native) in <c>MaterialiseAlarmCondition</c>
+    ///     drops the old condition instance — one more removal path, so it tombstones like the others. Believed
+    ///     unreachable from a composition (native ids are RawPaths in Raw, scripted ids live in Uns), so it is driven
+    ///     directly.
+    /// </summary>
+    [Fact]
+    public async Task A_condition_kind_swap_tells_and_forgets_items_on_the_dropped_instance()
+    {
+        var (host, server) = await BootAsync();
+        try
+        {
+            var nm = server.NodeManager!;
+            nm.EnsureFolder("eq-1", null, "Equipment", AddressSpaceRealm.Uns);
+            nm.MaterialiseAlarmCondition("al-1", "eq-1", "High", "OffNormalAlarm", 500, AddressSpaceRealm.Uns, isNative: false);
+            var nodeId = new NodeId("al-1", nm.TryGetFolder("eq-1")!.NodeId.NamespaceIndex);
+            var handle = (NodeHandle)nm.GetManagerHandle(nodeId);
+            handle.ShouldNotBeNull("non-vacuity: the condition is materialised");
+            var item = new StubItem(handle, Identity());
+            nm.TrackLiveMonitoredItem(handle, item);
+            nm.Authorizer = Authorizer(permits: false);
+            nm.RevalidateMonitoredItems();
+            item.Statuses.ShouldContain((StatusCode)StatusCodes.BadUserAccessDenied, "non-vacuity: stood down");
+
+            nm.MaterialiseAlarmCondition("al-1", "eq-1", "High", "OffNormalAlarm", 500, AddressSpaceRealm.Uns, isNative: true);
+
+            item.Statuses.ShouldContain((StatusCode)StatusCodes.BadNodeIdUnknown, "the dropped instance's item is told");
+            nm.LiveItemCount.ShouldBe(0);
+            ((NodeHandle)nm.GetManagerHandle(nodeId)).ShouldNotBeNull("positive control: the swapped-in condition exists");
+        }
+        finally { await host.DisposeAsync(); }
+    }
+
    /// <summary>Registers a stub item on <paramref name="rawPath"/> and stands it down through a denying pass.</summary>
    private static async Task<(StubItem Item, NodeId NodeId)> StandDownAsync(OtOpcUaNodeManager nm, string rawPath)
    {
@@ -0,0 +1,438 @@
+using System.Collections.Concurrent;
+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.Core.Authorization;
+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;
+
+/// <summary>
+///     Lane B review fix round 1. C1: the plan runs off the pinned thread (R02:P5) while writes are non-creating
+///     (R01:L1), so a value, status or alarm write for a node the IN-FLIGHT plan adds used to be skipped — and a
+///     change-driven driver never republishes it. The actor now remembers the latest write per node while a plan
+///     is in flight and replays it after the apply. Every test holds the plan open with a gate, so "during the
+///     plan" is deterministic, and runs against a sink that — like the real node manager — drops writes for nodes
+///     that do not exist.
+///     Also here: I2 (an authority parse throw never fails the apply), the bytes-less request queued behind an
+///     in-flight plan, and a plan piped back to a restarted actor.
+/// </summary>
+public sealed class OpcUaPublishActorDeferredWriteTests : RuntimeActorTestBase
+{
+    private const string NewTag = "Plant/Modbus/dev1/T1";
+    private static readonly TimeSpan PlanBudget = TimeSpan.FromSeconds(30);
+
+    [Fact]
+    public void A_value_for_a_node_the_in_flight_plan_adds_lands_on_that_node_after_the_apply()
+    {
+        var sink = new ExistenceSink();
+        using var gate = new ManualResetEventSlim(false);
+        var actor = Spawn(sink, gate);
+
+        actor.Tell(Rebuild(RawTagArtifact(3)));
+        var at = DateTime.UtcNow;
+        actor.Tell(new OpcUaPublishActor.AttributeValueBatchUpdate(
+            [new ValueWrite(NewTag, 7, OpcUaStatus.Good, at, AddressSpaceRealm.Raw)]));
+        AwaitAssert(() => sink.SkippedWrites.ShouldContain(NewTag), PlanBudget);   // non-vacuity: it really missed
+
+        gate.Set();
+        ExpectMsg<OpcUaPublishActor.RebuildCompleted>(PlanBudget).Degraded.ShouldBeFalse();
+
+        sink.Exists(NewTag).ShouldBeTrue("the plan materialised the node");
+        sink.ValueOf(NewTag).ShouldBe((7, OpcUaStatus.Good, at), "the value sent while the plan was in flight must end up on the node");
+    }
+
+    [Fact]
+    public void A_status_after_a_value_composes_onto_the_new_node_in_order()
+    {
+        var sink = new ExistenceSink();
+        using var gate = new ManualResetEventSlim(false);
+        var actor = Spawn(sink, gate);
+
+        actor.Tell(Rebuild(RawTagArtifact(3)));
+        var t1 = DateTime.UtcNow;
+        var t2 = t1.AddSeconds(1);
+        actor.Tell(new OpcUaPublishActor.AttributeValueBatchUpdate([new ValueWrite(NewTag, 7, OpcUaStatus.Good, t1, AddressSpaceRealm.Raw)]));
+        actor.Tell(new OpcUaPublishActor.AttributeStatusBatchUpdate([new StatusWrite(NewTag, OpcUaStatus.BadNoCommunication, t2, AddressSpaceRealm.Raw)]));
+        gate.Set();
+        ExpectMsg<OpcUaPublishActor.RebuildCompleted>(PlanBudget);
+
+        sink.ValueOf(NewTag).ShouldBe((7, OpcUaStatus.BadNoCommunication, t2), "last known value under the later status");
+    }
+
+    [Fact]
+    public void Existing_nodes_keep_updating_while_the_plan_computes()
+    {
+        var sink = new ExistenceSink();
+        using var gate = new ManualResetEventSlim(false);
+        var actor = Spawn(sink, gate);
+        actor.Tell(Rebuild(RawTagArtifact(1)));
+        gate.Set();
+        ExpectMsg<OpcUaPublishActor.RebuildCompleted>(PlanBudget);
+        var existing = "Plant/Modbus/dev1/T0";
+        sink.Exists(existing).ShouldBeTrue();
+
+        gate.Reset();
+        actor.Tell(Rebuild(RawTagArtifact(3)));
+        var at = DateTime.UtcNow;
+        actor.Tell(new OpcUaPublishActor.AttributeValueBatchUpdate([new ValueWrite(existing, 5, OpcUaStatus.Good, at, AddressSpaceRealm.Raw)]));
+
+        // Applied immediately, while the plan is still held open.
+        AwaitAssert(() => sink.ValueOf(existing).ShouldBe((5, OpcUaStatus.Good, at)), PlanBudget);
+        sink.Exists(NewTag).ShouldBeFalse("the plan is still in flight");
+
+        gate.Set();
+        ExpectMsg<OpcUaPublishActor.RebuildCompleted>(PlanBudget);
+        sink.ValueOf(existing).ShouldBe((5, OpcUaStatus.Good, at));
+    }
+
+    [Fact]
+    public void An_alarm_state_sent_during_the_plan_lands_on_the_condition_it_materialises()
+    {
+        var sink = new ExistenceSink();
+        using var gate = new ManualResetEventSlim(false);
+        var actor = Spawn(sink, gate);
+
+        actor.Tell(Rebuild(AlarmArtifact()));
+        var state = new AlarmConditionSnapshot(Active: true, Acknowledged: false, Confirmed: true, Enabled: true, AlarmShelvingKind.Unshelved, 500, "high");
+        actor.Tell(new OpcUaPublishActor.AlarmStateUpdate("al-1", state, DateTime.UtcNow, AddressSpaceRealm.Uns));
+        actor.Tell(new OpcUaPublishActor.AlarmQualityUpdate("al-1", OpcUaStatus.BadNoCommunication, DateTime.UtcNow, AddressSpaceRealm.Uns));
+        AwaitAssert(() => sink.SkippedAlarms.ShouldContain("al-1"), PlanBudget);
+
+        gate.Set();
+        ExpectMsg<OpcUaPublishActor.RebuildCompleted>(PlanBudget).Degraded.ShouldBeFalse();
+
+        sink.ConditionExists("al-1").ShouldBeTrue("the plan materialised the condition");
+        sink.AlarmStateOf("al-1").ShouldBeSameAs(state);
+        sink.AlarmQualityOf("al-1").ShouldBe(OpcUaStatus.BadNoCommunication);
+    }
+
+    [Fact]
+    public void The_deferred_buffer_is_bounded_under_a_flood_and_keeps_the_earliest_keys()
+    {
+        var sink = new ExistenceSink();
+        using var gate = new ManualResetEventSlim(false);
+        var actor = Sys.ActorOf(OpcUaPublishActor.PropsForTests(
+            sink, new AddressSpaceApplier(sink, NullLogger<AddressSpaceApplier>.Instance),
+            new OpcUaPublishActor.TestHooks { BeforePlan = () => gate.Wait() }, deferredWriteCapacity: 2));
+
+        actor.Tell(Rebuild(RawTagArtifact(5)));
+        var at = DateTime.UtcNow;
+        actor.Tell(new OpcUaPublishActor.AttributeValueBatchUpdate(Enumerable.Range(0, 5)
+            .Select(i => new ValueWrite($"Plant/Modbus/dev1/T{i}", i, OpcUaStatus.Good, at, AddressSpaceRealm.Raw)).ToList()));
+        // An existing key still updates when the buffer is full.
+        actor.Tell(new OpcUaPublishActor.AttributeValueBatchUpdate([new ValueWrite("Plant/Modbus/dev1/T0", 100, OpcUaStatus.Good, at, AddressSpaceRealm.Raw)]));
+        AwaitAssert(() => sink.SkippedWrites.Count.ShouldBe(6), PlanBudget);
+
+        gate.Set();
+        ExpectMsg<OpcUaPublishActor.RebuildCompleted>(PlanBudget);
+
+        sink.ValueOf("Plant/Modbus/dev1/T0")!.Value.Value.ShouldBe(100);
+        sink.ValueOf("Plant/Modbus/dev1/T1")!.Value.Value.ShouldBe(1);
+        for (var i = 2; i < 5; i++)
+            sink.ValueOf($"Plant/Modbus/dev1/T{i}").ShouldBeNull("beyond the bound a new key is not held");
+        sink.ReplayedItems.ShouldBe(2);
+    }
+
+    [Fact]
+    public void Writes_after_the_pipeline_drains_are_not_replayed_again()
+    {
+        var sink = new ExistenceSink();
+        using var gate = new ManualResetEventSlim(false);
+        var actor = Spawn(sink, gate);
+        actor.Tell(Rebuild(RawTagArtifact(3)));
+        actor.Tell(new OpcUaPublishActor.AttributeValueBatchUpdate([new ValueWrite(NewTag, 7, OpcUaStatus.Good, DateTime.UtcNow, AddressSpaceRealm.Raw)]));
+        gate.Set();
+        ExpectMsg<OpcUaPublishActor.RebuildCompleted>(PlanBudget);
+        var replayed = sink.ReplayedItems;
+        replayed.ShouldBe(1);
+
+        // A later plan replays only what arrived during IT.
+        actor.Tell(Rebuild(RawTagArtifact(3)));
+        ExpectMsg<OpcUaPublishActor.RebuildCompleted>(PlanBudget);
+        sink.ReplayedItems.ShouldBe(replayed);
+    }
+
+    [Fact]
+    public void A_throwing_authority_parse_does_not_fail_the_apply()
+    {
+        var sink = new ExistenceSink();
+        var authority = new RecordingAuthoritySink();
+        var actor = Sys.ActorOf(OpcUaPublishActor.PropsForTests(
+            sink, new AddressSpaceApplier(sink, NullLogger<AddressSpaceApplier>.Instance),
+            new OpcUaPublishActor.TestHooks { ParseAuthority = (_, _) => throw new InvalidOperationException("bad grants") },
+            dataPlaneAuthority: authority));
+
+        actor.Tell(Rebuild(RawTagArtifact(2)));
+
+        var completed = ExpectMsg<OpcUaPublishActor.RebuildCompleted>(PlanBudget);
+        completed.Degraded.ShouldBeFalse(completed.DegradedReason);
+        sink.Exists("Plant/Modbus/dev1/T1").ShouldBeTrue("the address space is applied");
+        authority.Applies.ShouldBe(0, "nothing is installed from an authority that could not be read");
+
+        // Positive control: the same actor installs an authority that CAN be read.
+        var healthy = Sys.ActorOf(OpcUaPublishActor.PropsForTests(
+            new ExistenceSink(), new AddressSpaceApplier(new ExistenceSink(), NullLogger<AddressSpaceApplier>.Instance),
+            hooks: null, dataPlaneAuthority: authority));
+        healthy.Tell(Rebuild(RawTagArtifact(2)));
+        ExpectMsg<OpcUaPublishActor.RebuildCompleted>(PlanBudget).Degraded.ShouldBeFalse();
+        authority.Applies.ShouldBe(1);
+    }
+
+    [Fact]
+    public void A_bytes_less_rebuild_arriving_mid_plan_answers_after_the_plan_ahead_of_it()
+    {
+        var sink = new ExistenceSink();
+        using var gate = new ManualResetEventSlim(false);
+        var actor = Spawn(sink, gate);
+
+        var first = Rebuild(RawTagArtifact(2));
+        var second = new OpcUaPublishActor.RebuildAddressSpace(CorrelationId.NewId(), DeploymentId.NewId(), Artifact: null, ReportOutcome: true);
+        actor.Tell(first);
+        actor.Tell(second);
+        ExpectNoMsg(TimeSpan.FromMilliseconds(300));   // the bytes-less one must not overtake the held plan
+
+        gate.Set();
+        var a = ExpectMsg<OpcUaPublishActor.RebuildCompleted>(PlanBudget);
+        var b = ExpectMsg<OpcUaPublishActor.RebuildCompleted>(PlanBudget);
+        a.Correlation.ShouldBe(first.Correlation);
+        a.Degraded.ShouldBeFalse();
+        b.Correlation.ShouldBe(second.Correlation);
+        b.Degraded.ShouldBeTrue();
+    }
+
+    [Fact]
+    public void A_plan_piped_back_to_a_restarted_actor_is_not_applied()
+    {
+        var sink = new ExistenceSink();
+        using var gate = new ManualResetEventSlim(false);
+        var applier = new AddressSpaceApplier(sink, NullLogger<AddressSpaceApplier>.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<IActorRef>();
+
+        var stale = Rebuild(RawTagArtifact(3));
+        child.Tell(stale);
+        child.Tell(Kill.Instance);   // the parent's strategy RESTARTS on any failure, ActorKilledException included
+        ExpectNoMsg(TimeSpan.FromMilliseconds(300));
+
+        gate.Set();
+        var answer = ExpectMsg<OpcUaPublishActor.RebuildCompleted>(PlanBudget);
+        answer.Correlation.ShouldBe(stale.Correlation);
+        answer.Degraded.ShouldBeTrue("the requester is still answered");
+        sink.Exists(NewTag).ShouldBeFalse("a plan from the previous incarnation must not be applied");
+
+        // Positive control: the restarted instance applies a fresh rebuild.
+        child.Tell(Rebuild(RawTagArtifact(3)));
+        ExpectMsg<OpcUaPublishActor.RebuildCompleted>(PlanBudget).Degraded.ShouldBeFalse();
+        sink.Exists(NewTag).ShouldBeTrue();
+    }
+
+    // ───────────────────────────── helpers ─────────────────────────────
+
+    private IActorRef Spawn(ExistenceSink sink, ManualResetEventSlim gate) =>
+        Sys.ActorOf(OpcUaPublishActor.PropsForTests(
+            sink, new AddressSpaceApplier(sink, NullLogger<AddressSpaceApplier>.Instance),
+            new OpcUaPublishActor.TestHooks { BeforePlan = () => gate.Wait() }));
+
+    private static OpcUaPublishActor.RebuildAddressSpace Rebuild(byte[] artifact) =>
+        new(CorrelationId.NewId(), DeploymentId.NewId(), artifact, ReportOutcome: true);
+
+    private static byte[] RawTagArtifact(int tagCount) =>
+        JsonSerializer.SerializeToUtf8Bytes(new
+        {
+            UnsAreas = Array.Empty<object>(),
+            UnsLines = Array.Empty<object>(),
+            Equipment = Array.Empty<object>(),
+            RawFolders = new[] { new { RawFolderId = "rf-1", ParentRawFolderId = (string?)null, Name = "Plant", ClusterId = "c1" } },
+            DriverInstances = new[] { new { DriverInstanceId = "drv-1", DriverType = "Modbus", DriverConfig = "{}", Name = "Modbus", RawFolderId = "rf-1", ClusterId = "c1" } },
+            Devices = new[] { new { DeviceId = "dev-1", DriverInstanceId = "drv-1", Name = "dev1", DeviceConfig = "{}" } },
+            TagGroups = Array.Empty<object>(),
+            Tags = Enumerable.Range(0, tagCount).Select(i => new
+            {
+                TagId = $"t-{i}", DeviceId = "dev-1", TagGroupId = (string?)null, Name = $"T{i}",
+                DataType = "Float", AccessLevel = 0, TagConfig = "{}",
+            }).ToArray(),
+            UnsTagReferences = Array.Empty<object>(),
+            ScriptedAlarms = Array.Empty<object>(),
+        });
+
+    private static byte[] AlarmArtifact() =>
+        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" },
+            },
+            DriverInstances = Array.Empty<object>(),
+            Tags = Array.Empty<object>(),
+            ScriptedAlarms = new[]
+            {
+                new
+                {
+                    ScriptedAlarmId = "al-1", EquipmentId = "eq-1", Name = "High", AlarmType = "AlarmCondition",
+                    Severity = 500, MessageTemplate = "high", PredicateScriptId = "s-1", HistorizeToAveva = false,
+                    Retain = true, Enabled = true,
+                },
+            },
+            Scripts = new[] { new { ScriptId = "s-1", SourceCode = "return false;" } },
+        });
+
+    /// <summary>Restarts its one child on any failure, and lets the test reach the child.</summary>
+    private sealed class RestartingParent : ReceiveActor
+    {
+        public sealed class GetChild { public static readonly GetChild Instance = new(); }
+
+        public RestartingParent(Props childProps)
+        {
+            var child = Context.ActorOf(childProps, "publish");
+            Receive<GetChild>(_ => Sender.Tell(child));
+        }
+
+        protected override SupervisorStrategy SupervisorStrategy() =>
+            new OneForOneStrategy(_ => Directive.Restart);
+    }
+
+    /// <summary>An authority sink that counts installs.</summary>
+    private sealed class RecordingAuthoritySink : IDataPlaneAuthoritySink
+    {
+        private int _applies;
+        public int Applies => Volatile.Read(ref _applies);
+
+        public DataPlaneAuthoritySnapshot Apply(
+            string? clusterId, string revision, IReadOnlyList<ZB.MOM.WW.OtOpcUa.Configuration.Entities.ApiKey> keys,
+            IReadOnlyList<ZB.MOM.WW.OtOpcUa.Configuration.Entities.NodeAcl> grants,
+            IReadOnlyDictionary<AclScopeRef, IReadOnlyList<NodeAclPath>> scopePaths)
+        {
+            Interlocked.Increment(ref _applies);
+            return new DataPlaneAuthorityState().Apply(clusterId, revision, keys, grants, scopePaths);
+        }
+    }
+
+    /// <summary>
+    ///     Models the node manager's non-creating writes: a write to a node that was never ensured (or was removed)
+    ///     is skipped and recorded; a full sink rebuild forgets every node. <see cref="ReplayedItems"/> counts the
+    ///     items written after an apply closed and before the next apply starts — the actor's replay, since the
+    ///     tests send nothing between releasing a plan and its answer.
+    /// </summary>
+    private sealed class ExistenceSink : IOpcUaAddressSpaceSink, ISurgicalAddressSpaceSink
+    {
+        private readonly object _gate = new();
+        private readonly HashSet<string> _variables = new();
+        private readonly HashSet<string> _conditions = new();
+        private readonly Dictionary<string, (object? Value, uint Status, DateTime At)> _values = new();
+        private readonly Dictionary<string, AlarmConditionSnapshot> _alarmStates = new();
+        private readonly Dictionary<string, uint> _alarmQualities = new();
+
+        public ConcurrentQueue<string> SkippedWrites { get; } = new();
+        public ConcurrentQueue<string> SkippedAlarms { get; } = new();
+        public int ReplayedItems;
+
+        public bool Exists(string id) { lock (_gate) return _variables.Contains(id); }
+        public bool ConditionExists(string id) { lock (_gate) return _conditions.Contains(id); }
+        public (object? Value, uint Status, DateTime At)? ValueOf(string id) { lock (_gate) return _values.TryGetValue(id, out var v) ? v : null; }
+        public AlarmConditionSnapshot? AlarmStateOf(string id) { lock (_gate) return _alarmStates.GetValueOrDefault(id); }
+        public uint? AlarmQualityOf(string id) { lock (_gate) return _alarmQualities.TryGetValue(id, out var q) ? q : null; }
+
+        public void WriteValue(string nodeId, object? value, uint statusCode, DateTime ts, AddressSpaceRealm realm) =>
+            WriteValues([new ValueWrite(nodeId, value, statusCode, ts, realm)]);
+
+        public void WriteValues(IReadOnlyList<ValueWrite> values)
+        {
+            lock (_gate)
+            {
+                if (_applyClosed) ReplayedItems += values.Count;
+                foreach (var v in values)
+                {
+                    if (!_variables.Contains(v.NodeId)) { SkippedWrites.Enqueue(v.NodeId); continue; }
+                    _values[v.NodeId] = (v.Value, v.StatusCode, v.SourceTimestampUtc);
+                }
+            }
+        }
+
+        public void WriteStatuses(IReadOnlyList<StatusWrite> statuses)
+        {
+            lock (_gate)
+            {
+                if (_applyClosed) ReplayedItems += statuses.Count;
+                foreach (var s in statuses)
+                {
+                    if (!_variables.Contains(s.NodeId)) { SkippedWrites.Enqueue(s.NodeId); continue; }
+                    var current = _values.GetValueOrDefault(s.NodeId);
+                    _values[s.NodeId] = (current.Value, s.StatusCode, s.SourceTimestampUtc);
+                }
+            }
+        }
+
+        public void WriteAlarmCondition(string alarmNodeId, AlarmConditionSnapshot state, DateTime ts, AddressSpaceRealm realm)
+        {
+            lock (_gate)
+            {
+                if (!_conditions.Contains(alarmNodeId)) { SkippedAlarms.Enqueue(alarmNodeId); return; }
+                _alarmStates[alarmNodeId] = state;
+            }
+        }
+
+        public void WriteAlarmQuality(string alarmNodeId, uint statusCode, DateTime ts, AddressSpaceRealm realm)
+        {
+            lock (_gate)
+            {
+                if (!_conditions.Contains(alarmNodeId)) { SkippedAlarms.Enqueue(alarmNodeId); return; }
+                _alarmQualities[alarmNodeId] = statusCode;
+            }
+        }
+
+        public void MaterialiseAlarmCondition(string alarmNodeId, string equipmentNodeId, string displayName, string alarmType, int severity, AddressSpaceRealm realm, bool isNative = false)
+        {
+            lock (_gate) { _applyClosed = false; _conditions.Add(alarmNodeId); }
+        }
+
+        public void EnsureFolder(string folderNodeId, string? parentNodeId, string displayName, AddressSpaceRealm realm)
+        {
+            lock (_gate) _applyClosed = false;
+        }
+
+        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)
+        {
+            lock (_gate) { _applyClosed = false; _variables.Add(variableNodeId); }
+        }
+
+        public void RebuildAddressSpace()
+        {
+            lock (_gate)
+            {
+                _applyClosed = false;
+                _variables.Clear();
+                _conditions.Clear();
+                _values.Clear();
+                _alarmStates.Clear();
+                _alarmQualities.Clear();
+            }
+        }
+
+        private bool _applyClosed;
+
+        public void CompleteAddressSpaceApply()
+        {
+            lock (_gate) _applyClosed = true;
+        }
+
+        public void RaiseNodesAddedModelChange(string affectedNodeId, AddressSpaceRealm realm) { }
+        public void WireAlarmNotifiers(string alarmNodeId, AddressSpaceRealm alarmRealm, IReadOnlyList<string> notifierFolderNodeIds, AddressSpaceRealm notifierFolderRealm) { }
+        public void AddReference(string sourceNodeId, AddressSpaceRealm sourceRealm, string targetNodeId, AddressSpaceRealm targetRealm, string referenceType = "Organizes") { }
+        public bool UpdateTagAttributes(string variableNodeId, bool writable, string? historianTagname, string dataType, bool isArray, uint? arrayLength, AddressSpaceRealm realm) => true;
+        public bool UpdateFolderDisplayName(string folderNodeId, string displayName, AddressSpaceRealm realm) => true;
+        public bool RemoveVariableNode(string variableNodeId, AddressSpaceRealm realm) { lock (_gate) return _variables.Remove(variableNodeId); }
+        public bool RemoveAlarmConditionNode(string alarmNodeId, AddressSpaceRealm realm) { lock (_gate) return _conditions.Remove(alarmNodeId); }
+        public bool RemoveEquipmentSubtree(string equipmentNodeId, AddressSpaceRealm realm) => true;
+    }
+}