test(opcua): subscription-survival on pure-remove + live-verify note (R2-07 T12)
This commit is contained in:
Joseph Doherty
@@ -104,10 +104,11 @@
|
||||
{
|
||||
"id": "T12",
|
||||
"subject": "Phase 2: remove-side subscription-survival integration test + live-/run remove gate (kind=PureRemove rebuild=False) \u2014 Phase 2 shippable boundary",
|
||||
"status": "pending",
|
||||
"status": "deferred-live",
|
||||
"blockedBy": [
|
||||
"T11"
|
||||
]
|
||||
],
|
||||
"note": "remove-side SubscriptionSurvivalTests authored + compiles (dotnet build green); NOT run \u2014 *.IntegrationTests memory constraint. Live docker-dev pure-remove gate (survivor subscription uninterrupted; central log kind=PureRemove, rebuild=False; removed node reads BadNodeIdUnknown) MUST run in the serial heavy pass \u2014 F10b inertness risk."
|
||||
},
|
||||
{
|
||||
"id": "T13",
|
||||
|
||||
@@ -178,7 +178,7 @@ what the unit leg cannot; (c) migrate the 3 xunit v2 holdouts once Akka ships an
|
||||
|
||||
| **R2-11** | 01/C-1, 01/P-1 + 05/CONV-2, UNDER-1, UNDER-6 | `r2/11-tagconfig-consolidation` (off `46fedda3`) | first `2e14fe1f` … (24 task commits) | **Code-complete, offline-verified.** **Part 1 (C-1/P-1):** single `TagConfigIntent.Parse` + `DeviceConfigIntent` in Commons replace the four byte-parity TagConfig parse copies (composer/artifact/walker/DraftValidator) — composer now parses once per tag (P-1); a permanent `Runtime.Tests/TagConfigCorpusParityTests` (30-blob corpus) guards compose≡decode; `grep "MUST parse identically" src/` = 0. **Part 2 (CONV-2):** shared strict-capable `TagConfigJson` readers in Core.Abstractions replace the six copy-pasted `ReadEnum`s; all six driver parsers gain `Inspect()` warnings + honour the `writable` key (FOCAS **forced read-only**, 05/UNDER-1); FOCAS equipment tags run the capability-matrix pre-flight (05/UNDER-6); Modbus probe parses the factory DTO shape (`timeoutMs` no longer dropped). New `EquipmentTagConfigInspector` in ControlPlane + `AdminOperationsActor` deploy-gate wiring with `Deployment:TagConfigValidationMode` (**Warn default | Error opt-in**), proven through the actor (F10b consuming test). AdminUI: `writable` checkbox in the six typed editors. **Phase-C follow-up (deferred, recorded here):** flip the runtime parsers to strict (typo'd enum ⇒ `BadNodeIdUnknown` instead of wrong-width `Good`) once fleets have run `Error` mode clean — behavior change deliberately not shipped in this plan. **Deferred (T22/T24):** docker-dev `/run` of one typed editor's Writable checkbox + whole-solution serial sweep. |
|
||||
|
||||
| **R2-07 (Phase 1)** | 03/P1 (action #7, surgical pure-adds) | `r2/07-surgical-pure-adds` (off `1a698cbb`) | `bb226f45`… (T1–T6) | **Phase 1 code-complete, offline-verified.** New pure `AddressSpaceChangeClassifier` (Empty/AttributeOnly/PureAdd/PureRemove/AddRemoveMix/Rebuild; first-match-wins, default-closed Rebuild; the F10b tag/vtag surgical-eligibility predicates moved here as `internal static`). `AddressSpaceApplier.Apply` routes by kind: **PureAdd/AttributeOnly skip `RebuildAddressSpace`** (the idempotent Materialise passes create exactly the added nodes; existing subscriptions survive) — PureRemove/AddRemoveMix still full-rebuild until Phases 2/3. Adds are announced via a new `ComputeAddAnnouncements`+`AnnounceAddedNodes` (dedup affected-parent set → existing guard-covered `RaiseNodesAddedModelChange`), fired by `OpcUaPublishActor` after the 4 Materialise passes gated on `!outcome.RebuildCalled`. `MaterialiseAlarmCondition` made idempotent (skip-if-present-and-same-kind ⇒ keeps the condition NodeState alive across a pure-add). Apply log carries `kind={Kind}`. **NO new sink members in Phase 1.** OpcUaServer.Tests 321/321, Runtime.Tests 398/398 (incl. flipped add-rebuild pins → pure-add-no-rebuild + new ordering/rebuild-skips-announce publish-actor tests). **Deferred-live (T5, T6):** the over-the-wire `SubscriptionSurvivalTests` (authored + compiles; NOT run — `*.IntegrationTests` memory constraint) and the docker-dev live-`/run` gate (subscribe survives a +1-tag deploy; central log `kind=PureAdd, rebuild=False`; new node browsable/Good) — MUST run in the serial heavy pass (F10b inertness risk). Phases 2 (pure-removes) + 3 (mixed) still pending. |
|
||||
| **R2-07 (Phases 1+2)** | 03/P1 (action #7, surgical pure-adds/removes) | `r2/07-surgical-pure-adds` (off `1a698cbb`) | `bb226f45`… (T1–T12) | **Phases 1+2 code-complete, offline-verified.** New pure `AddressSpaceChangeClassifier` (Empty/AttributeOnly/PureAdd/PureRemove/AddRemoveMix/Rebuild; first-match-wins, default-closed Rebuild; the F10b tag/vtag surgical-eligibility predicates moved here as `internal static`). `AddressSpaceApplier.Apply` routes by kind. **Phase 1 (PureAdd/AttributeOnly):** skip `RebuildAddressSpace` (the idempotent Materialise passes create exactly the added nodes; existing subscriptions survive); adds announced via `ComputeAddAnnouncements`+`AnnounceAddedNodes` (dedup affected-parent → existing guard-covered `RaiseNodesAddedModelChange`), fired by `OpcUaPublishActor` after the 4 passes gated on `!outcome.RebuildCalled`; `MaterialiseAlarmCondition` made idempotent (skip-if-present-and-same-kind keeps the condition NodeState alive). **NO new sink members in Phase 1.** **Phase 2 (PureRemove):** 3 new `ISurgicalAddressSpaceSink` members (`RemoveVariableNode`/`RemoveAlarmConditionNode`/`RemoveEquipmentSubtree`) — guard-first with the **observed negative control** (`DeferredSinkForwardingReflectionTests` fails member-named until `DeferredAddressSpaceSink` forwards) + `SdkAddressSpaceSink` forwards + `OtOpcUaNodeManager` in-place teardown (scoped map cleanup, notifier demotion via `RemoveRootNotifier`, `BuildNodesRemovedModelChange` verb=NodeDeleted reported outside `Lock`). Applier `PureRemove` arm: terminal Bad / `RemovedConditionState` writes (closing the removed-alarm-tag today-gap) → subsumption partition (removed-equipment children folded into the subtree remove) → in-place removes → any false/throw ⇒ full-rebuild ratchet; non-surgical sink ⇒ rebuild. `AddRemoveMix` still full-rebuild until Phase 3. **Guard stays green (Commons.Tests 17/17), OpcUaServer.Tests 334/334, Runtime.Tests 398/398.** **Deferred-live (T5, T6, T12):** `SubscriptionSurvivalTests` add + remove cases (authored + compiles; NOT run — `*.IntegrationTests` memory constraint) and the docker-dev live-`/run` add + remove gates (`kind=PureAdd/PureRemove, rebuild=False`; survivor subscription uninterrupted; removed node → `BadNodeIdUnknown`) — MUST run in the serial heavy pass (F10b inertness risk). Phase 3 (mixed add+remove) still pending. |
|
||||
|
||||
**Report anchor drifts to fold into the next report refresh** (from the R2-02 verification pass): (1)
|
||||
`01-core-composition.md` U-6 anchor `DriverResilienceOptions.cs:23-25` → the two bulkhead properties sat at
|
||||
|
||||
+88
@@ -137,6 +137,94 @@ public sealed class SubscriptionSurvivalTests
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>R2-07 Phase 2 — a subscription on tag A survives a PURE-REMOVE deploy that removes tag B; B's
|
||||
/// monitored item observes a final Bad and the node then disappears (a re-read returns BadNodeIdUnknown),
|
||||
/// while A stays alive.</summary>
|
||||
[Fact]
|
||||
public async Task Subscription_on_surviving_tag_survives_pure_remove_of_another_tag()
|
||||
{
|
||||
var pkiRoot = Path.Combine(Path.GetTempPath(), $"otopcua-subsurvive-rm-{Guid.NewGuid():N}");
|
||||
var port = AllocateFreePort();
|
||||
var ct = TestContext.Current.CancellationToken;
|
||||
|
||||
try
|
||||
{
|
||||
var options = new OpcUaApplicationHostOptions
|
||||
{
|
||||
ApplicationName = ServerUri,
|
||||
ApplicationUri = ServerUri + ".Remove",
|
||||
OpcUaPort = port,
|
||||
PublicHostname = "127.0.0.1",
|
||||
PkiStoreRoot = pkiRoot,
|
||||
EnabledSecurityProfiles = new List<OpcUaSecurityProfile> { OpcUaSecurityProfile.None },
|
||||
AutoAcceptUntrustedClientCertificates = true,
|
||||
};
|
||||
var server = new OtOpcUaSdkServer();
|
||||
await using var host = new OpcUaApplicationHost(options, NullLogger<OpcUaApplicationHost>.Instance);
|
||||
await host.StartAsync(server, ct);
|
||||
|
||||
var nm = server.NodeManager!;
|
||||
var sink = new SdkAddressSpaceSink(nm);
|
||||
var applier = new AddressSpaceApplier(sink, NullLogger<AddressSpaceApplier>.Instance);
|
||||
|
||||
// Materialise A + B and seed both Good.
|
||||
var withAB = CompositionWith(
|
||||
new EquipmentTagPlan("tag-a", "eq-1", "drv", FolderPath: "", Name: "A", DataType: "Int32", FullName: "1", Writable: false, Alarm: null),
|
||||
new EquipmentTagPlan("tag-b", "eq-1", "drv", FolderPath: "", Name: "B", DataType: "Int32", FullName: "2", Writable: false, Alarm: null));
|
||||
applier.MaterialiseHierarchy(withAB);
|
||||
applier.MaterialiseEquipmentTags(withAB);
|
||||
var nodeIdAString = Commons.OpcUa.EquipmentNodeIds.Variable("eq-1", "", "A");
|
||||
var nodeIdBString = Commons.OpcUa.EquipmentNodeIds.Variable("eq-1", "", "B");
|
||||
sink.WriteValue(nodeIdAString, 10, Commons.OpcUa.OpcUaQuality.Good, DateTime.UtcNow);
|
||||
sink.WriteValue(nodeIdBString, 20, Commons.OpcUa.OpcUaQuality.Good, DateTime.UtcNow);
|
||||
|
||||
using var session = await OpenSessionAsync($"opc.tcp://127.0.0.1:{port}/OtOpcUa", ct);
|
||||
var ns = (ushort)session.NamespaceUris.GetIndex(OtOpcUaNodeManager.DefaultNamespaceUri);
|
||||
var nodeIdA = new NodeId(nodeIdAString, ns);
|
||||
var nodeIdB = new NodeId(nodeIdBString, ns);
|
||||
|
||||
var receivedA = new List<DataValue>();
|
||||
var receivedB = new List<DataValue>();
|
||||
var gate = new object();
|
||||
var subscription = new Subscription(session.DefaultSubscription) { PublishingInterval = 100 };
|
||||
session.AddSubscription(subscription);
|
||||
await subscription.CreateAsync(ct);
|
||||
var miA = new MonitoredItem(subscription.DefaultItem) { StartNodeId = nodeIdA, AttributeId = Attributes.Value, SamplingInterval = 50, QueueSize = 10 };
|
||||
var miB = new MonitoredItem(subscription.DefaultItem) { StartNodeId = nodeIdB, AttributeId = Attributes.Value, SamplingInterval = 50, QueueSize = 10 };
|
||||
miA.Notification += (item, _) => { foreach (var v in item.DequeueValues()) lock (gate) receivedA.Add(v); };
|
||||
miB.Notification += (item, _) => { foreach (var v in item.DequeueValues()) lock (gate) receivedB.Add(v); };
|
||||
subscription.AddItem(miA);
|
||||
subscription.AddItem(miB);
|
||||
await subscription.ApplyChangesAsync(ct);
|
||||
await WaitUntilAsync(() => { lock (gate) return receivedA.Count >= 1 && receivedB.Count >= 1; }, TimeSpan.FromSeconds(5));
|
||||
|
||||
// Drive a PURE-REMOVE apply removing B (previous={A,B}, next={A}).
|
||||
var withA = CompositionWith(
|
||||
new EquipmentTagPlan("tag-a", "eq-1", "drv", FolderPath: "", Name: "A", DataType: "Int32", FullName: "1", Writable: false, Alarm: null));
|
||||
var plan = AddressSpacePlanner.Compute(withAB, withA);
|
||||
plan.RemovedEquipmentTags.Count.ShouldBe(1);
|
||||
|
||||
lock (gate) { receivedA.Clear(); receivedB.Clear(); }
|
||||
var outcome = applier.Apply(plan);
|
||||
outcome.RebuildCalled.ShouldBeFalse(); // PureRemove — no full rebuild
|
||||
|
||||
// B's monitored item observes a final Bad; the node is then gone from the address space.
|
||||
await WaitUntilAsync(() => { lock (gate) return receivedB.Any(v => StatusCode.IsBad(v.StatusCode)); }, TimeSpan.FromSeconds(5));
|
||||
|
||||
// A stays alive: a fresh write to A is delivered.
|
||||
sink.WriteValue(nodeIdAString, 11, Commons.OpcUa.OpcUaQuality.Good, DateTime.UtcNow);
|
||||
await WaitUntilAsync(() => { lock (gate) return receivedA.Any(v => Equals(v.Value, 11)); }, TimeSpan.FromSeconds(5));
|
||||
|
||||
// B is gone: a re-read returns BadNodeIdUnknown.
|
||||
var bRead = await session.ReadValueAsync(nodeIdB, ct);
|
||||
bRead.StatusCode.ShouldBe((StatusCode)StatusCodes.BadNodeIdUnknown);
|
||||
}
|
||||
finally
|
||||
{
|
||||
if (Directory.Exists(pkiRoot)) Directory.Delete(pkiRoot, recursive: true);
|
||||
}
|
||||
}
|
||||
|
||||
private static AddressSpaceComposition CompositionWith(params EquipmentTagPlan[] tags) =>
|
||||
new(
|
||||
UnsAreas: Array.Empty<UnsAreaProjection>(),
|
||||
|
||||
Reference in New Issue
Block a user