dohertj2/lmxopcua
1
0
Fork 0
Code Issues 43 Pull Requests Actions Packages Projects Releases Wiki Activity

Merge lane B: OPC UA address space remediation (#732)

Browse Source 
Claude-Session: https://claude.ai/code/session_01Ub9hGmYLQvn9XETzw7CMsT
This commit is contained in:
Joseph Doherty
2026-10-22 16:01:44 -04:00
parent cb9a48ee96 ff4e210a27
commit 0ad460c5c4
32 changed files with 4508 additions and 1688 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/AddressSpaceApplier.cs
+154 -149
View File
@@ -93,37 +93,33 @@ public sealed class AddressSpaceApplier
}
var ts = DateTime.UtcNow;
var removedCount = 0;
// A removed equipment is a FOLDER: it gets no terminal condition write (review 01 L4 — the write the
// applier used to make against the equipment id never found a condition and minted a placeholder
// variable that shadowed the folder's SDK lookup entry). Its subtree teardown is in ApplyPureRemove.
var removedCount = plan.RemovedEquipment.Count;
// Swallowed removal-pass condition-write failures are tallied here (not lost to a per-node Warning)
// so a degraded removal is operator-visible via AddressSpaceApplyOutcome.FailedNodes (archreview 01/S-1).
var failedNodes = 0;
foreach (var eq in plan.RemovedEquipment)
{
if (!SafeWriteAlarmCondition(eq.EquipmentId, RemovedConditionState, ts, AddressSpaceRealm.Uns)) failedNodes++;
removedCount++;
}
foreach (var alarm in plan.RemovedAlarms)
{
if (!SafeWriteAlarmCondition(alarm.ScriptedAlarmId, RemovedConditionState, ts, AddressSpaceRealm.Uns)) failedNodes++;
removedCount++;
}
// Removed equipment tags / VirtualTags are plain variable nodes (no Part 9 condition to write
// before tear-down), but they ARE real removals — count them so AddressSpaceApplyOutcome.RemovedNodes
// is accurate on a removed-tag-only deploy, which now reaches the rebuild path below. v3 Batch 4:
// removed raw containers/tags + UNS reference variables are likewise real removals.
removedCount += plan.RemovedEquipmentTags.Count + plan.RemovedEquipmentVirtualTags.Count +
// Removed VirtualTags are plain variable nodes (no Part 9 condition to write before tear-down), but
// they ARE real removals — count them so AddressSpaceApplyOutcome.RemovedNodes is accurate on a
// removed-vtag-only deploy. v3 Batch 4: removed raw containers/tags + UNS reference variables are
// likewise real removals.
removedCount += plan.RemovedEquipmentVirtualTags.Count +
plan.RemovedRawContainers.Count + plan.RemovedRawTags.Count + plan.RemovedUnsReferenceVariables.Count;
var changedCount =
plan.ChangedEquipment.Count + plan.ChangedDrivers.Count + plan.ChangedAlarms.Count +
plan.ChangedEquipmentTags.Count +
plan.ChangedEquipmentVirtualTags.Count +
plan.ChangedRawContainers.Count + plan.ChangedRawTags.Count + plan.ChangedUnsReferenceVariables.Count +
// A UNS Area/Line rename is an in-place change to an existing folder node.
plan.RenamedFolders.Count;
var addedCount =
plan.AddedEquipment.Count + plan.AddedDrivers.Count + plan.AddedAlarms.Count +
plan.AddedEquipmentTags.Count +
plan.AddedEquipmentVirtualTags.Count +
plan.AddedRawContainers.Count + plan.AddedRawTags.Count + plan.AddedUnsReferenceVariables.Count;
@@ -145,7 +141,7 @@ public sealed class AddressSpaceApplier
// order even when an id is reused across the remove + add sets (the recreated node is a fresh
// NodeState, correct — it is a different tag). Any remove false/throw ⇒ full-rebuild fallback.
// • Rebuild ⇒ the default-closed safety valve for any node-affecting change (ChangedEquipment /
// ChangedAlarms / non-surgical ChangedEquipmentTags / node-relevant ChangedEquipmentVirtualTags)
// ChangedAlarms / node-relevant ChangedEquipmentVirtualTags / non-surgical raw-tag deltas)
// — the classifier's rule 2, which also catches any future plan field that makes a changed
// record unequal, so the worst outcome of a misclassification is today's full rebuild.
// ChangedDrivers is node-inert (routes through DriverHostActor's spawn plan) — the classifier leaves
@@ -156,7 +152,6 @@ public sealed class AddressSpaceApplier
// afterward by the publish actor's Materialise passes + announce).
var hasRemovePass = kind is AddressSpaceChangeKind.PureRemove or AddressSpaceChangeKind.AddRemoveMix;
var surgicalTagDeltas = plan.ChangedEquipmentTags.Where(AddressSpaceChangeClassifier.TagDeltaIsSurgicalEligible).ToList();
// UNS Area / Line renames are surgically applicable (in-place DisplayName swap)
// when no structural rebuild fires. When a rebuild DOES fire (any add/remove/structural change),
// MaterialiseHierarchy re-creates every folder with the new display names, so the renames are
@@ -176,7 +171,7 @@ public sealed class AddressSpaceApplier
rebuildFailed = !SafeRebuild();
rebuilt = true;
}
else if (surgicalTagDeltas.Count > 0 || renamedFolders.Count > 0 || hasRemovePass ||
else if (renamedFolders.Count > 0 || hasRemovePass ||
surgicalRawTagDeltas.Count > 0 || unsRefDeltas.Count > 0)
{
if (_sink is ISurgicalAddressSpaceSink surgical)
@@ -194,25 +189,6 @@ public sealed class AddressSpaceApplier
}
if (!ok) { allApplied = false; break; }
}
foreach (var d in allApplied ? surgicalTagDeltas : Enumerable.Empty<AddressSpacePlan.EquipmentTagDelta>())
{
// Compute the node id + writable + historian + shape EXACTLY as MaterialiseEquipmentTags
// would so the in-place update matches what a rebuild would have produced. Array tags are
// forced read-only (same as EnsureVariable: the driver write path doesn't handle arrays).
var nodeId = UnsEquipmentVar(d.Current.EquipmentId, d.Current.FolderPath, d.Current.Name);
var writable = d.Current.Writable && !d.Current.IsArray;
var historian = d.Current.IsHistorized
? (string.IsNullOrWhiteSpace(d.Current.HistorianTagname) ? d.Current.FullName : d.Current.HistorianTagname)
: null;
bool ok;
try { ok = surgical.UpdateTagAttributes(nodeId, writable, historian, d.Current.DataType, d.Current.IsArray, d.Current.ArrayLength, AddressSpaceRealm.Uns); }
catch (Exception ex)
{
_logger.LogError(ex, "AddressSpaceApplier: surgical UpdateTagAttributes threw for {Node}", nodeId);
ok = false;
}
if (!ok) { allApplied = false; break; }
}
// v3 RAW tags — the same in-place attribute update, in the RAW realm. Adding or dropping ONE
// UnsTagReference flips its backing raw tag to Changed purely through ReferencingEquipmentPaths,
// and before this path existed that tore down every node on the server for an edit that creates
@@ -243,6 +219,9 @@ public sealed class AddressSpaceApplier
ok = false;
}
if (!ok) { allApplied = false; break; }
RecordReshape(AddressSpaceRealm.Raw, d.Current.NodeId,
d.Previous.DataType, d.Previous.IsArray, d.Previous.ArrayLength,
d.Current.DataType, d.Current.IsArray, d.Current.ArrayLength);
}
// v3 UNS reference variables — TWO shapes, split on whether the delta moves the node's
// IDENTITY (2026-08-15 review I2).
@@ -294,6 +273,9 @@ public sealed class AddressSpaceApplier
updated = false;
}
if (!updated) { allApplied = false; break; }
RecordReshape(AddressSpaceRealm.Uns, d.Current.NodeId,
d.Previous.DataType, d.Previous.BackingIsArray, d.Previous.BackingArrayLength,
d.Current.DataType, d.Current.BackingIsArray, d.Current.BackingArrayLength);
continue;
}
@@ -320,8 +302,8 @@ public sealed class AddressSpaceApplier
}
_logger.LogInformation(
"AddressSpaceApplier: applied plan (kind={Kind}, added={Added}, removed={Removed}, changed={Changed}, surgicalTags={Surgical}, surgicalRawTags={SurgicalRaw}, unsRefDeltas={UnsRefs}, renamedFolders={Renamed}, rebuild={Rebuild})",
kind, addedCount, removedCount, changedCount, rebuilt ? 0 : surgicalTagDeltas.Count,
"AddressSpaceApplier: applied plan (kind={Kind}, added={Added}, removed={Removed}, changed={Changed}, surgicalRawTags={SurgicalRaw}, unsRefDeltas={UnsRefs}, renamedFolders={Renamed}, rebuild={Rebuild})",
kind, addedCount, removedCount, changedCount,
rebuilt ? 0 : surgicalRawTagDeltas.Count, rebuilt ? 0 : unsRefDeltas.Count,
rebuilt ? 0 : renamedFolders.Count, rebuilt);
@@ -342,10 +324,9 @@ public sealed class AddressSpaceApplier
/// Apply a PureRemove plan by tearing down ONLY the affected nodes IN PLACE (no full
/// rebuild). Removed equipment "own" their child tags/vtags/alarms, so any removed child whose
/// <c>EquipmentId</c> is itself in <see cref="AddressSpacePlan.RemovedEquipment"/> is SUBSUMED by the
/// subtree removal and skipped individually. For surviving-equipment removals: a removed value
/// subtree removal and skipped individually. For surviving-equipment removals: a removed VirtualTag
/// variable gets a terminal Bad <c>WriteValue</c> then <see cref="ISurgicalAddressSpaceSink.RemoveVariableNode"/>;
/// a removed alarm-bearing tag (previously it got no terminal condition write) and a
/// removed scripted alarm get a terminal <see cref="RemovedConditionState"/> write then
/// a removed scripted alarm gets a terminal <see cref="RemovedConditionState"/> write then
/// <see cref="ISurgicalAddressSpaceSink.RemoveAlarmConditionNode"/>; each removed equipment gets one
/// <see cref="ISurgicalAddressSpaceSink.RemoveEquipmentSubtree"/>. The terminal write is what an
/// in-flight MonitoredItem observes (a final Bad); after the node is gone, re-subscription gets
@@ -363,25 +344,6 @@ public sealed class AddressSpaceApplier
plan.RemovedEquipment.Select(e => e.EquipmentId), StringComparer.Ordinal);
bool NotSubsumed(string equipmentId) => !removedEquipmentIds.Contains(equipmentId);
// Removed equipment tags — value variables OR alarm-bearing conditions — for SURVIVING equipment.
foreach (var tag in plan.RemovedEquipmentTags.Where(t => NotSubsumed(t.EquipmentId)))
{
var nodeId = UnsEquipmentVar(tag.EquipmentId, tag.FolderPath, tag.Name);
if (tag.Alarm is not null)
{
// Alarm-bearing tag → a Part 9 condition node. Terminal RemovedConditionState (today-gap
// closed), then remove the condition in place.
if (!SafeWriteAlarmCondition(nodeId, RemovedConditionState, ts, AddressSpaceRealm.Uns)) failedNodes++;
if (!SafeRemoveAlarmCondition(surgical, nodeId, AddressSpaceRealm.Uns)) return false;
}
else
{
// Plain value variable → terminal Bad so an in-flight MonitoredItem sees the removal.
SafeWriteValue(nodeId, null, OpcUaStatus.Bad, ts, AddressSpaceRealm.Uns);
if (!SafeRemoveVariable(surgical, nodeId, AddressSpaceRealm.Uns)) return false;
}
}
// Removed VirtualTags (always plain value variables) for surviving equipment.
foreach (var v in plan.RemovedEquipmentVirtualTags.Where(t => NotSubsumed(t.EquipmentId)))
{
@@ -550,7 +512,6 @@ public sealed class AddressSpaceApplier
ArgumentNullException.ThrowIfNull(plan);
var ids = new HashSet<string>(StringComparer.Ordinal);
foreach (var t in plan.AddedEquipmentTags) ids.Add(MaterialiseParent(t.EquipmentId, t.FolderPath));
foreach (var v in plan.AddedEquipmentVirtualTags) ids.Add(MaterialiseParent(v.EquipmentId, v.FolderPath));
foreach (var a in plan.AddedAlarms) ids.Add(a.EquipmentId);
// v3 Batch 4 / review I5 — an added UNS reference variable is materialised directly under its
@@ -1074,96 +1035,9 @@ public sealed class AddressSpaceApplier
return failed;
}
/// <summary>
/// Materialise Equipment-namespace tags from a composition snapshot.
/// For each <see cref="EquipmentTagPlan"/>,
/// ensure its optional <c>FolderPath</c> sub-folder under the existing equipment folder, then
/// ensure a Variable (NodeId = <c>FullName</c>, the driver-side ref) inside it. Variables
/// start BadWaitingForInitialData; the driver fills live values in a later milestone.
/// Idempotent. This is a sink-based pass, NOT a reuse of <c>EquipmentNodeWalker</c>: no
/// sink-backed <c>IAddressSpaceBuilder</c> adapter exists, and the walker re-creates the
/// whole Area/Line/Equipment tree with browse-path NodeIds, incompatible with this path's
/// logical-Id NodeIds and the already-materialised equipment folders. This pass adds ONLY
/// variables (and any per-tag sub-folder); <see cref="MaterialiseHierarchy"/> owns the
/// equipment folders and this pass never re-creates them. The sink's <c>EnsureVariable</c>
/// takes a plain <c>string dataType</c> (not a DriverAttributeInfo).
/// <para><b>Subset overload (2026-08-15 surgical-deploys).</b> <paramref name="only"/> narrows WHAT is
/// materialised to a plan's added tags; <paramref name="composition"/> stays the FULL composition for
/// context. A <c>null</c> subset means "the whole composition" (the rebuild path).</para>
/// </summary>
/// <param name="composition">The FULL composition — the context even when a subset is passed.</param>
/// <param name="only">Materialise only these tags; <c>null</c> ⇒ the whole composition's.</param>
/// <returns>The count of swallowed sink failures in this pass (0 on a clean apply) — the publish actor
/// sums it into the degraded-apply surface (archreview 01/S-1).</returns>
public int MaterialiseEquipmentTags(
AddressSpaceComposition composition,
IReadOnlyList<EquipmentTagPlan>? only = null)
{
ArgumentNullException.ThrowIfNull(composition);
var equipmentTags = only ?? composition.EquipmentTags;
if (equipmentTags.Count == 0) return 0;
var failed = 0;
// Sub-folders first — a tag's FolderPath becomes one folder UNDER its equipment folder
// (deduped per distinct equipment+path). Tags with no FolderPath hang directly under the
// equipment folder, which MaterialiseHierarchy already created (never re-create
// the equipment folder here).
var foldersCreated = new HashSet<string>(StringComparer.Ordinal);
foreach (var tag in equipmentTags)
{
if (string.IsNullOrWhiteSpace(tag.FolderPath)) continue;
var folderNodeId = UnsSubFolder(tag.EquipmentId, tag.FolderPath);
if (!foldersCreated.Add(folderNodeId)) continue;
if (!SafeEnsureFolder(folderNodeId, parentNodeId: tag.EquipmentId, displayName: tag.FolderPath, realm: AddressSpaceRealm.Uns)) failed++;
}
// Variables: NodeId is FOLDER-SCOPED ("<parent>/<Name>"), NOT the raw FullName — a driver
// ref (e.g. a Modbus register) is not unique across identical machines, so FullName-as-NodeId
// would collide in the sink (EnsureVariable keys on NodeId) and drop all but one machine's
// signal. The driver-side FullName lives on EquipmentTagPlan for the later values milestone to
// route by. Parent is the FolderPath sub-folder when set, else the equipment folder directly.
// Per-variable idempotency relies on the sink's own EnsureVariable.
foreach (var tag in equipmentTags)
{
var parent = string.IsNullOrWhiteSpace(tag.FolderPath)
? tag.EquipmentId
: UnsSubFolder(tag.EquipmentId, tag.FolderPath);
var nodeId = UnsEquipmentVar(tag.EquipmentId, tag.FolderPath, tag.Name);
if (tag.Alarm is not null)
{
// Native alarm tag → a real Part 9 condition node (reuses the scripted-alarm path),
// NOT a value variable. Parent is the sub-folder when set, else the equipment folder.
if (!SafeMaterialiseAlarmCondition(nodeId, parent, tag.Name, tag.Alarm.AlarmType, tag.Alarm.Severity, isNative: true, realm: AddressSpaceRealm.Uns)) failed++;
}
else
{
// Phase C: a historized tag materialises Historizing + HistoryRead. Resolve the effective
// historian tagname HERE (default-vs-override): a null/blank override falls back to the
// driver-side FullName; null means the tag is not historized at all.
string? historianTagname = tag.IsHistorized
? (string.IsNullOrWhiteSpace(tag.HistorianTagname) ? tag.FullName : tag.HistorianTagname)
: null;
// Retired equipment-tag path (structurally empty in v3): array tags stay forced read-only
// here. The live array-write capability is carried on RawTagPlan /
// UnsReferenceVariable only — see RawNodeWritable.
var writable = tag.Writable && !tag.IsArray;
if (!SafeEnsureVariable(nodeId, parent, tag.Name, tag.DataType, writable, AddressSpaceRealm.Uns, historianTagname, tag.IsArray, tag.ArrayLength)) failed++;
}
}
_logger.LogInformation(
"AddressSpaceApplier: equipment tags materialised (tags={Tags}, equipment={Equipment}, subset={Subset}, failed={Failed})",
equipmentTags.Count,
equipmentTags.Select(t => t.EquipmentId).Distinct(StringComparer.Ordinal).Count(),
only is not null,
failed);
return failed;
}
/// <summary>
/// Materialise Equipment-namespace VirtualTags from a composition snapshot — the VirtualTag
/// analogue of <see cref="MaterialiseEquipmentTags"/>. For each <see cref="EquipmentVirtualTagPlan"/>,
/// computed-signal pass. For each <see cref="EquipmentVirtualTagPlan"/>,
/// ensure its optional <c>FolderPath</c> sub-folder under the existing equipment folder (in
/// practice <c>FolderPath</c> is empty for VirtualTags, so this is usually a no-op), then ensure
/// a Variable inside it. Like the tag pass, the variable's NodeId is FOLDER-SCOPED
@@ -1285,10 +1159,53 @@ 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);
_recording?.Ensured.Add((realm, nodeId));
return true;
}
catch (Exception ex) { _logger.LogWarning(ex, "AddressSpaceApplier: EnsureVariable threw for {Node}", nodeId); return false; }
}
/// <summary>Records an in-place UpdateTagAttributes that changed the node's SHAPE (DataType / array-ness /
/// array length): the node manager drops such a node's value to BadWaitingForInitialData.</summary>
private void RecordReshape(
AddressSpaceRealm realm, string nodeId,
string previousDataType, bool previousIsArray, uint? previousArrayLength,
string dataType, bool isArray, uint? arrayLength)
{
if (_recording is null) return;
if (string.Equals(previousDataType, dataType, StringComparison.Ordinal)
&& previousIsArray == isArray
&& (!isArray || previousArrayLength == arrayLength))
{
return;
}
_recording.Reshaped[(realm, nodeId)] = new VariableShape(dataType, isArray, arrayLength);
}
/// <summary>The recording in progress, or null. See <see cref="BeginRecording"/>.</summary>
private MaterialisationRecording? _recording;
/// <summary>
/// Starts recording what the following apply and materialise passes do to VARIABLES, until
/// <see cref="EndRecording"/>: every variable ensured, and every variable an in-place attribute update
/// re-shaped (which resets its value). The publish actor scopes the replay of writes it deferred while a plan
/// was in flight to these — a variable the apply kept took the live write already, and rewriting it would
/// revert a client write made in the window. Single-caller (the publish actor's thread), like the apply.
/// </summary>
/// <returns>The live recording.</returns>
public MaterialisationRecording BeginRecording()
{
_recording = new MaterialisationRecording();
return _recording;
}
/// <summary>Stops the recording <see cref="BeginRecording"/> started (the returned recording is kept).</summary>
public void EndRecording() => _recording = 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
@@ -1401,3 +1318,91 @@ public sealed record AddressSpaceApplyOutcome(
bool RebuildCalled,
bool RebuildFailed = false,
int FailedNodes = 0);
/// <summary>What one apply did to variables (see <see cref="AddressSpaceApplier.BeginRecording"/>).</summary>
public sealed class MaterialisationRecording
{
/// <summary>Every variable an <c>EnsureVariable</c> was issued for, as (realm, NodeId).</summary>
public HashSet<(AddressSpaceRealm, string)> Ensured { get; } = new();
/// <summary>Every variable an in-place attribute update RE-SHAPED (its value reset), with its new shape.</summary>
public Dictionary<(AddressSpaceRealm, string), VariableShape> Reshaped { get; } = new();
}
/// <summary>A variable's value shape: its Tag.DataType name, and its array-ness / 1-D length.</summary>
/// <param name="DataType">The Tag.DataType name ("Int32", "Float", …; unknown ⇒ BaseDataType, any value fits).</param>
/// <param name="IsArray">Whether the node is a one-dimensional array.</param>
/// <param name="ArrayLength">The fixed array length, when known (null / 0 ⇒ any length).</param>
public sealed record VariableShape(string DataType, bool IsArray, uint? ArrayLength)
{
/// <summary>
/// Fits <paramref name="value"/> to this shape for a write onto a re-shaped node: unchanged when it already
/// fits, converted (invariant culture) when a lossless-enough conversion exists, otherwise refused — an
/// old-typed value must never be served under the new DataType (the reset exists to prevent exactly that).
/// A null value always fits (a status annotation carries none).
/// </summary>
/// <param name="value">The value to fit.</param>
/// <param name="fitted">The fitted value, when this returns true.</param>
/// <returns>True when the value fits (as is, or converted).</returns>
public bool TryFit(object? value, out object? fitted)
{
fitted = null;
if (value is null) return true;
var element = ElementType(DataType);
if (!IsArray)
{
if (value is Array) return false;
return TryConvert(value, element, out fitted);
}
if (value is not Array array || value is string) return false;
if (ArrayLength is > 0 && array.Length != ArrayLength) return false;
if (element is null) { fitted = value; return true; }
var result = Array.CreateInstance(element, array.Length);
for (var i = 0; i < array.Length; i++)
{
if (!TryConvert(array.GetValue(i), element, out var item) || item is null) return false;
result.SetValue(item, i);
}
fitted = result;
return true;
}
private static bool TryConvert(object? value, Type? target, out object? converted)
{
converted = value;
if (value is null || target is null || target.IsInstanceOfType(value)) return true;
if (value is not IConvertible) return false;
try
{
converted = Convert.ChangeType(value, target, System.Globalization.CultureInfo.InvariantCulture);
return true;
}
catch (Exception ex) when (ex is InvalidCastException or FormatException or OverflowException)
{
converted = null;
return false;
}
}
/// <summary>The CLR type the node manager serves for a Tag.DataType name (its ResolveBuiltInDataType set);
/// null ⇒ BaseDataType, where any value fits.</summary>
private static Type? ElementType(string dataType) => dataType switch
{
"Boolean" => typeof(bool),
"SByte" => typeof(sbyte),
"Byte" => typeof(byte),
"Int16" => typeof(short),
"UInt16" => typeof(ushort),
"Int32" => typeof(int),
"UInt32" => typeof(uint),
"Int64" => typeof(long),
"UInt64" => typeof(ulong),
"Float" => typeof(float),
"Double" => typeof(double),
"String" => typeof(string),
"DateTime" => typeof(DateTime),
_ => null,
};
}
src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/AddressSpaceChangeClassifier.cs
+5 -37
View File
@@ -8,7 +8,7 @@ public enum AddressSpaceChangeKind
/// <summary><see cref="AddressSpacePlan.IsEmpty"/> — nothing to do.</summary>
Empty,
/// <summary>Only surgical-eligible changed equipment/raw tags, surgical UNS-reference deltas,
/// <summary>Only surgical-eligible changed raw tags, surgical UNS-reference deltas,
/// node-irrelevant vtag changes, folder renames, and/or node-inert driver deltas — the existing F10b
/// in-place path. No adds, no removes.</summary>
AttributeOnly,
@@ -27,7 +27,7 @@ public enum AddressSpaceChangeKind
AddRemoveMix,
/// <summary>Anything with a node-affecting change (ChangedEquipment, ChangedAlarms, ChangedRawContainers,
/// non-surgical ChangedEquipmentTags, alarm-bearing/identity-changing ChangedRawTags, node-relevant
/// alarm-bearing/identity-changing ChangedRawTags, node-relevant
/// ChangedEquipmentVirtualTags) — the unclassifiable-safe default. Full rebuild.</summary>
Rebuild,
}
@@ -53,8 +53,7 @@ public static class AddressSpaceChangeClassifier
if (plan.IsEmpty) return AddressSpaceChangeKind.Empty;
// 2 — any node-affecting CHANGE forces a full rebuild (default-closed). ChangedEquipment /
// ChangedAlarms / ChangedRawContainers are always node-affecting; a changed equipment tag is
// node-affecting UNLESS it is surgical-eligible, a changed raw tag UNLESS it is
// ChangedAlarms / ChangedRawContainers are always node-affecting; a changed raw tag UNLESS it is
// RawTagDeltaIsSurgicalEligible, and a changed vtag UNLESS it is node-irrelevant. Evaluated BEFORE the
// add/remove split so a non-surgical change alongside pure adds still rebuilds.
//
@@ -94,7 +93,6 @@ public static class AddressSpaceChangeClassifier
plan.ChangedAlarms.Count > 0 ||
plan.ChangedRawContainers.Count > 0 ||
plan.ChangedRawTags.Any(d => !RawTagDeltaIsSurgicalEligible(d)) ||
plan.ChangedEquipmentTags.Any(d => !TagDeltaIsSurgicalEligible(d)) ||
plan.ChangedEquipmentVirtualTags.Any(d => !VtagDeltaIsNodeIrrelevant(d)))
{
return AddressSpaceChangeKind.Rebuild;
@@ -105,12 +103,12 @@ public static class AddressSpaceChangeClassifier
// add/remove split below and leave a driver-only plan classified AttributeOnly.
var hasAdds =
plan.AddedEquipment.Count + plan.AddedAlarms.Count +
plan.AddedEquipmentTags.Count + plan.AddedEquipmentVirtualTags.Count +
plan.AddedEquipmentVirtualTags.Count +
plan.AddedRawContainers.Count + plan.AddedRawTags.Count +
plan.AddedUnsReferenceVariables.Count > 0;
var hasRemoves =
plan.RemovedEquipment.Count + plan.RemovedAlarms.Count +
plan.RemovedEquipmentTags.Count + plan.RemovedEquipmentVirtualTags.Count +
plan.RemovedEquipmentVirtualTags.Count +
plan.RemovedRawContainers.Count + plan.RemovedRawTags.Count +
plan.RemovedUnsReferenceVariables.Count > 0;
@@ -141,36 +139,6 @@ public static class AddressSpaceChangeClassifier
Historize = d.Current.Historize,
}).Equals(d.Current);
// F10b: a CHANGED equipment tag whose ONLY differences are Writable / IsHistorized /
// HistorianTagname / DataType / IsArray / ArrayLength (a plain value variable — no alarm condition node)
// can be updated IN PLACE on the existing node via ISurgicalAddressSpaceSink.UpdateTagAttributes,
// avoiding a full rebuild (preserving subscriptions). The presentation-shape fields (DataType / IsArray /
// ArrayLength) join the whitelist now that the surgical sink swaps DataType + ValueRank + ArrayDimensions
// in place (and raises a GeneralModelChangeEvent). FullName / DriverInstanceId / Name / identity / alarm
// differences still fall through to a rebuild — FullName/DriverInstanceId re-route the node to a different
// driver point, Name re-derives the NodeId, and an alarm flip turns the node into a Part 9 condition. The
// override-unequal default also covers any future field.
// REACH (live-verified): the shape path only fires for drivers whose TagConfig carries a stable
// top-level "FullName" (Galaxy = tag_name.AttributeName; OpcUaClient = the node id) — there a DataType/array
// edit leaves FullName untouched ⇒ surgical. For structured-TagConfig protocol drivers (Modbus/S7/AbCip/…)
// TagConfigIntent.Parse falls back to the RAW TagConfig blob as FullName, so a DataType/
// array edit also mutates that blob ⇒ FullName differs ⇒ this returns false ⇒ full rebuild. That is the
// correct safe default (a protocol driver's subscription needs re-spawning for a new shape anyway).
/// <summary>Whether a changed equipment tag can be updated in place on its existing node instead of rebuilding.</summary>
/// <param name="d">The equipment-tag delta to classify.</param>
/// <returns><see langword="true"/> when the tag has no alarm and differs only in the in-place-updatable attributes; otherwise <see langword="false"/>.</returns>
internal static bool TagDeltaIsSurgicalEligible(AddressSpacePlan.EquipmentTagDelta d) =>
d.Previous.Alarm is null && d.Current.Alarm is null &&
(d.Previous with
{
Writable = d.Current.Writable,
IsHistorized = d.Current.IsHistorized,
HistorianTagname = d.Current.HistorianTagname,
DataType = d.Current.DataType,
IsArray = d.Current.IsArray,
ArrayLength = d.Current.ArrayLength,
}).Equals(d.Current);
// v3 surgical raw-tag eligibility: a NON-alarm raw tag whose only differences are the in-place-updatable
// attributes (Writable / IsHistorized / HistorianTagname / DataType / IsArray / ArrayLength / ArrayWritable)
// and/or its ReferencingEquipmentPaths. ArrayWritable only feeds the node's writability — the
src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/AddressSpacePlan.cs
+7 -30
View File
@@ -23,28 +23,15 @@ public sealed record AddressSpacePlan(
IReadOnlyList<ScriptedAlarmPlan> RemovedAlarms,
IReadOnlyList<AddressSpacePlan.AlarmDelta> ChangedAlarms)
{
/// <summary>
/// Equipment-namespace tag diff sets, keyed by <see cref="EquipmentTagPlan.TagId"/>. Added as
/// init-only members (defaulting empty) rather than positional parameters so existing
/// <c>AddressSpacePlan</c> construction sites compile unchanged — consistent with how
/// <see cref="AddressSpaceComposition.EquipmentTags"/> was added. Without these, an
/// incremental deploy that changes ONLY equipment tags produced an empty plan and
/// <c>OpcUaPublishActor.HandleRebuild</c> short-circuited before materialising them.
/// </summary>
public IReadOnlyList<EquipmentTagPlan> AddedEquipmentTags { get; init; } = Array.Empty<EquipmentTagPlan>();
/// <inheritdoc cref="AddedEquipmentTags"/>
public IReadOnlyList<EquipmentTagPlan> RemovedEquipmentTags { get; init; } = Array.Empty<EquipmentTagPlan>();
/// <inheritdoc cref="AddedEquipmentTags"/>
public IReadOnlyList<EquipmentTagDelta> ChangedEquipmentTags { get; init; } = Array.Empty<EquipmentTagDelta>();
/// <summary>
/// Equipment-namespace VirtualTag diff sets, keyed by <see cref="EquipmentVirtualTagPlan.VirtualTagId"/>.
/// The value-side analogue of <see cref="AddedEquipmentTags"/>: a VirtualTag carries an
/// A VirtualTag carries an
/// <c>Expression</c> evaluated over <c>DependencyRefs</c>, so a deploy that changes ONLY
/// VirtualTags (e.g. a new computed signal or an edited formula) must still produce a
/// non-empty plan and drive a rebuild — without these the diff was blind to VirtualTags and
/// such a deploy silently no-op'd. Added as init-only members (defaulting empty) for the same
/// compile-compatibility reason as <see cref="AddedEquipmentTags"/>.
/// such a deploy silently no-op'd. Added as init-only members (defaulting empty) so existing
/// construction sites compile unchanged. (The retired v3 equipment-tag diff sets — review 01 U2 — are gone:
/// device values live in the Raw subtree and reach equipment as <see cref="UnsReferenceVariable"/>s.)
/// </summary>
public IReadOnlyList<EquipmentVirtualTagPlan> AddedEquipmentVirtualTags { get; init; } = Array.Empty<EquipmentVirtualTagPlan>();
/// <inheritdoc cref="AddedEquipmentVirtualTags"/>
@@ -65,7 +52,7 @@ public sealed record AddressSpacePlan(
/// uses), preserving every client's subscriptions; a sink lacking the surgical capability or a
/// missing folder falls back to a full rebuild. Added as an init-only member (defaulting empty) so
/// every existing <c>AddressSpacePlan</c> construction site compiles unchanged — consistent with
/// the EquipmentTag / EquipmentVirtualTag diff sets above.
/// the EquipmentVirtualTag diff sets above.
/// </summary>
public IReadOnlyList<FolderRename> RenamedFolders { get; init; } = Array.Empty<FolderRename>();
@@ -74,7 +61,7 @@ public sealed record AddressSpacePlan(
/// keyed by <see cref="RawContainerNode.NodeId"/> (the RawPath). A rename of any container changes
/// its RawPath (and every descendant's), so it manifests as remove(old)+add(new) — an OPC UA NodeId
/// is immutable identity. Init-only, defaulting empty, so existing construction sites compile
/// unchanged (consistent with the EquipmentTag diff sets above).
/// unchanged.
/// </summary>
public IReadOnlyList<RawContainerNode> AddedRawContainers { get; init; } = Array.Empty<RawContainerNode>();
/// <inheritdoc cref="AddedRawContainers"/>
@@ -114,7 +101,6 @@ public sealed record AddressSpacePlan(
AddedEquipment.Count == 0 && RemovedEquipment.Count == 0 && ChangedEquipment.Count == 0 &&
AddedDrivers.Count == 0 && RemovedDrivers.Count == 0 && ChangedDrivers.Count == 0 &&
AddedAlarms.Count == 0 && RemovedAlarms.Count == 0 && ChangedAlarms.Count == 0 &&
AddedEquipmentTags.Count == 0 && RemovedEquipmentTags.Count == 0 && ChangedEquipmentTags.Count == 0 &&
AddedEquipmentVirtualTags.Count == 0 && RemovedEquipmentVirtualTags.Count == 0 && ChangedEquipmentVirtualTags.Count == 0 &&
AddedRawContainers.Count == 0 && RemovedRawContainers.Count == 0 && ChangedRawContainers.Count == 0 &&
AddedRawTags.Count == 0 && RemovedRawTags.Count == 0 && ChangedRawTags.Count == 0 &&
@@ -124,7 +110,6 @@ public sealed record AddressSpacePlan(
public sealed record EquipmentDelta(EquipmentNode Previous, EquipmentNode Current);
public sealed record DriverDelta(DriverInstancePlan Previous, DriverInstancePlan Current);
public sealed record AlarmDelta(ScriptedAlarmPlan Previous, ScriptedAlarmPlan Current);
public sealed record EquipmentTagDelta(EquipmentTagPlan Previous, EquipmentTagPlan Current);
public sealed record EquipmentVirtualTagDelta(EquipmentVirtualTagPlan Previous, EquipmentVirtualTagPlan Current);
/// <summary>A Raw container node present in both snapshots (same RawPath) with ≥1 differing field.</summary>
@@ -177,12 +162,7 @@ public static class AddressSpacePlanner
a => a.ScriptedAlarmId,
(a, b) => new AddressSpacePlan.AlarmDelta(a, b));
var (addedEqTags, removedEqTags, changedEqTags) = DiffById(
previous.EquipmentTags, next.EquipmentTags,
t => t.TagId,
(a, b) => new AddressSpacePlan.EquipmentTagDelta(a, b));
// VirtualTags diff by VirtualTagId, mirroring the EquipmentTags pass. EquipmentVirtualTagPlan
// VirtualTags diff by VirtualTagId. EquipmentVirtualTagPlan
// overrides record equality to compare ALL fields by value — scalars (Expression/DataType/
// Name/FolderPath) plus DependencyRefs element-wise (SequenceEqual). So a no-op redeploy (fresh
// list instances, identical contents) correctly diffs to empty; only a real content change is
@@ -232,9 +212,6 @@ public static class AddressSpacePlanner
addedDrv, removedDrv, changedDrv,
addedAlarm, removedAlarm, changedAlarm)
{
AddedEquipmentTags = addedEqTags,
RemovedEquipmentTags = removedEqTags,
ChangedEquipmentTags = changedEqTags,
AddedEquipmentVirtualTags = addedVTags,
RemovedEquipmentVirtualTags = removedVTags,
ChangedEquipmentVirtualTags = changedVTags,
src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/OpcUaApplicationHost.cs
+85 -4
View File
@@ -105,6 +105,55 @@ public sealed class OpcUaApplicationHostOptions
/// is always element 0.
/// </summary>
public IList<string> PeerApplicationUris { get; set; } = new List<string>();
/// <summary>
/// SDK request-processing and transport quotas (<c>OpcUa:Quotas:*</c>, review 01 P6). The defaults are
/// exactly the values that used to be hard-coded (5 / 100 / 200 request threads and queue) and the SDK's
/// own transport defaults, so an absent section changes nothing.
/// </summary>
public OpcUaQuotaOptions Quotas { get; set; } = new();
}
/// <summary>
/// The <c>OpcUa:Quotas</c> section: the SDK's request-thread / queue ceilings and its transport quotas.
/// Request quotas carry the former hard-coded defaults; a transport quota left <c>null</c> keeps the SDK
/// default (<see cref="TransportQuotas"/>'s constructor values).
/// </summary>
public sealed class OpcUaQuotaOptions
{
/// <summary>Minimum SDK request-processing threads (default 5).</summary>
public int MinRequestThreadCount { get; set; } = 5;
/// <summary>Maximum SDK request-processing threads (default 100).</summary>
public int MaxRequestThreadCount { get; set; } = 100;
/// <summary>Requests the SDK queues before refusing with <c>BadTooManyOperations</c> (default 200) — the
/// burst ceiling review 01 P6 flagged.</summary>
public int MaxQueuedRequestCount { get; set; } = 200;
/// <summary>Per-operation timeout in milliseconds; <c>null</c> keeps the SDK default.</summary>
public int? OperationTimeoutMs { get; set; }
/// <summary>Largest string the transport decodes; <c>null</c> keeps the SDK default.</summary>
public int? MaxStringLength { get; set; }
/// <summary>Largest byte string the transport decodes; <c>null</c> keeps the SDK default.</summary>
public int? MaxByteStringLength { get; set; }
/// <summary>Largest array the transport decodes; <c>null</c> keeps the SDK default.</summary>
public int? MaxArrayLength { get; set; }
/// <summary>Largest message the transport accepts, in bytes; <c>null</c> keeps the SDK default.</summary>
public int? MaxMessageSize { get; set; }
/// <summary>Transport buffer size in bytes; <c>null</c> keeps the SDK default.</summary>
public int? MaxBufferSize { get; set; }
/// <summary>Secure-channel lifetime in milliseconds; <c>null</c> keeps the SDK default.</summary>
public int? ChannelLifetimeMs { get; set; }
/// <summary>Security-token lifetime in milliseconds; <c>null</c> keeps the SDK default.</summary>
public int? SecurityTokenLifetimeMs { get; set; }
}
/// <summary>
@@ -435,10 +484,8 @@ public sealed class OpcUaApplicationHost : IAsyncDisposable
var serverConfig = new ServerConfiguration
{
BaseAddresses = { $"opc.tcp://{_options.PublicHostname}:{_options.OpcUaPort}/OtOpcUa" },
MinRequestThreadCount = 5,
MaxRequestThreadCount = 100,
MaxQueuedRequestCount = 200,
};
ApplyRequestQuotas(serverConfig, _options.Quotas);
foreach (var policy in BuildSecurityPolicies(_options.EnabledSecurityProfiles))
{
@@ -469,7 +516,7 @@ public sealed class OpcUaApplicationHost : IAsyncDisposable
RejectedCertificateStore = new CertificateTrustList { StoreType = "Directory", StorePath = Path.Combine(_options.PkiStoreRoot, "rejected") },
AutoAcceptUntrustedCertificates = _options.AutoAcceptUntrustedClientCertificates,
},
TransportQuotas = new TransportQuotas(),
TransportQuotas = BuildTransportQuotas(_options.Quotas),
ClientConfiguration = new ClientConfiguration(),
TraceConfiguration = new TraceConfiguration(),
};
@@ -479,6 +526,40 @@ public sealed class OpcUaApplicationHost : IAsyncDisposable
return config;
}
/// <summary>
/// Applies the request-processing quotas (<c>OpcUa:Quotas:*</c>, review 01 P6 — they were hard-coded at
/// 5 / 100 / 200) to the SDK server configuration.
/// </summary>
/// <param name="serverConfig">The configuration to set.</param>
/// <param name="quotas">The configured quotas.</param>
internal static void ApplyRequestQuotas(ServerConfiguration serverConfig, OpcUaQuotaOptions quotas)
{
serverConfig.MinRequestThreadCount = quotas.MinRequestThreadCount;
serverConfig.MaxRequestThreadCount = quotas.MaxRequestThreadCount;
serverConfig.MaxQueuedRequestCount = quotas.MaxQueuedRequestCount;
}
/// <summary>
/// Builds the SDK transport quotas from <c>OpcUa:Quotas:*</c>: every member starts at the SDK default and
/// only a configured (non-null) value replaces it, so an absent section is byte-identical to the former
/// <c>new TransportQuotas()</c>.
/// </summary>
/// <param name="quotas">The configured quotas.</param>
/// <returns>The transport quotas to install.</returns>
internal static TransportQuotas BuildTransportQuotas(OpcUaQuotaOptions quotas)
{
var t = new TransportQuotas();
if (quotas.OperationTimeoutMs is { } operationTimeout) t.OperationTimeout = operationTimeout;
if (quotas.MaxStringLength is { } maxString) t.MaxStringLength = maxString;
if (quotas.MaxByteStringLength is { } maxByteString) t.MaxByteStringLength = maxByteString;
if (quotas.MaxArrayLength is { } maxArray) t.MaxArrayLength = maxArray;
if (quotas.MaxMessageSize is { } maxMessage) t.MaxMessageSize = maxMessage;
if (quotas.MaxBufferSize is { } maxBuffer) t.MaxBufferSize = maxBuffer;
if (quotas.ChannelLifetimeMs is { } channelLifetime) t.ChannelLifetime = channelLifetime;
if (quotas.SecurityTokenLifetimeMs is { } tokenLifetime) t.SecurityTokenLifetime = tokenLifetime;
return t;
}
/// <summary>
/// Maps each configured <see cref="OpcUaSecurityProfile"/> to a SDK
/// <see cref="ServerSecurityPolicy"/>. Duplicate profiles are silently de-duped because
src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/OtOpcUaNodeManager.cs
+738 -240
View File
File diff suppressed because it is too large Load Diff
src/Server/ZB.MOM.WW.OtOpcUa.Runtime/OpcUa/OpcUaPublishActor.cs
+643 -174
View File
@@ -86,19 +86,17 @@ public sealed class OpcUaPublishActor : ReceiveActor, IWithTimers
/// <param name="Realm">The namespace realm the condition lives in (<see cref="AddressSpaceRealm.Raw"/> for native).</param>
public sealed record AlarmQualityUpdate(string AlarmNodeId, uint StatusCode, DateTime TimestampUtc, AddressSpaceRealm Realm);
/// <summary>
/// Triggers an address-space rebuild. <paramref name="DeploymentId"/> is the deployment
/// just applied by the host; the rebuild loads THAT artifact so materialisation matches the
/// applied config + the SubscribeBulk pass. It is null only for legacy/dev callers, which
/// fall back to the latest sealed deployment (lags a not-yet-sealed apply by one revision).
/// Triggers an address-space rebuild of the artifact carried in <paramref name="Artifact"/>.
/// <paramref name="DeploymentId"/> is the deployment just applied by the host (echoed in the reply and
/// the logs). Review 02 P5: the ConfigDb fallback for a message carrying no bytes is gone — it had no
/// production producer and would have blocked the pinned OPC UA thread on central SQL — so a message
/// without bytes is "no answer" and holds the served address space.
/// </summary>
/// <param name="Correlation">Correlation id for tracing this rebuild.</param>
/// <param name="DeploymentId">The applied deployment whose artifact to materialise, or null for the latest-sealed fallback.</param>
/// <param name="DeploymentId">The applied deployment whose artifact this is, or null for a legacy caller.</param>
/// <param name="Artifact">
/// The artifact bytes already in hand, used INSTEAD of loading them from the ConfigDb. This is
/// what makes boot-from-cache actually serve its address space: on a central-SQL outage the
/// host has the cached blob but <see cref="DeploymentId"/> alone would drive a ConfigDb read
/// that cannot succeed, leaving the rebuild a no-op and clients browsing an empty server. Null
/// on the normal path, where loading from the ConfigDb by id is correct.
/// The artifact bytes. Every producer (the deploy path, boot-from-cache, FetchAndCache) has them in
/// hand; null or empty is treated as no answer.
/// </param>
/// <param name="ReportOutcome">
/// when <see langword="true"/>, the actor replies to the sender with a
@@ -128,11 +126,110 @@ public sealed class OpcUaPublishActor : ReceiveActor, IWithTimers
public sealed record ServiceLevelChanged(byte ServiceLevel);
/// <summary>
/// Review 02 P5 — a rebuild whose artifact parse, cluster-scoped composition and plan were computed
/// OFF the pinned OPC UA thread (a <c>Task.Run</c> piped back to <c>Self</c>), ready for the sink passes,
/// which are the only part that must run on this actor's thread. Carries the <see cref="Request"/> and
/// the original <see cref="Requester"/> so the <see cref="RebuildCompleted"/> reply reaches whoever asked.
/// </summary>
/// <para><see cref="Incarnation"/> stamps the actor instance that started the plan: a <c>PipeTo(Self)</c>
/// outlives a restart (the ActorRef survives it), and a plan computed against a previous instance's
/// <see cref="_lastApplied"/> must not be applied by the next one. <see cref="AuthorityError"/> carries a
/// data-plane-authority parse failure, which never fails the address-space apply (review fix I2).</para>
/// </summary>
private sealed record PlanReady(
object Incarnation, RebuildAddressSpace Request, IActorRef Requester, byte[] Artifact,
DataPlaneAuthorityProjection? Authority, Exception? AuthorityError,
AddressSpaceComposition Composition, AddressSpacePlan Plan, IReadOnlyList<string> Inconsistencies);
/// <summary>The off-thread parse / plan step threw (unreadable bytes, a composer fault).</summary>
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>
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>
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>
internal const int DefaultDeferredWriteCapacity = 250_000;
/// <summary>Identity of THIS actor instance (see <see cref="PlanReady.Incarnation"/>).</summary>
private readonly object _incarnation = new();
private readonly TestHooks? _hooks;
/// <summary>
/// Review fix C1 — the latest value / status / alarm write per (realm, id) received while a rebuild's plan is
/// in flight (or queued), replayed through the sink after each apply. Writes are non-creating (review 01 L1),
/// so a driver's first publish for a node the in-flight plan ADDS used to be skipped — and a change-driven
/// driver (or a poll group whose last-value cache already holds it) never sends it again, leaving the node
/// <c>BadWaitingForInitialData</c> indefinitely. On a full rebuild the same holds for every node: the
/// re-subscribe's first values land on old nodes the apply then discards.
/// <para>Writes are still applied IMMEDIATELY (existing nodes keep updating during the plan); the buffer
/// only remembers each key's latest so the replay can land it on a node the apply created. Replaying onto a
/// node that already holds it is inert: the same value/status/timestamp is filtered by every monitored
/// item's data-change filter, and an alarm write is delta-gated against the node's current state. That is
/// also why alarms coalesce to the latest state per id rather than replaying every transition: replaying an
/// older transition onto a live condition would move it backwards and fire spurious events.</para>
/// </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>Variables an in-place attribute update RE-SHAPED (review N4): the node manager reset their value to
/// BadWaitingForInitialData, so the window's value is replayed — but only once fitted to the new shape
/// (<see cref="VariableShape.TryFit"/>); a value that does not fit is dropped and the node left waiting.
/// The raw node and its UNS mirror are re-shaped by the same edit, so both get the same treatment.</item>
/// <item>Conditions: those in the applied composition but not the previous one; after a full rebuild ALSO
/// every kept SCRIPTED condition (review N3) — the node manager carries only NATIVE conditions' state over
/// a rebuild, so a scripted one is recreated fresh and needs the window's state. A kept native condition
/// is restored from the carry-over (which already holds the live write and any SDK ack), so it is not
/// replayed. Accepted cost on the scripted path: an SDK ack on the discarded instance in the window is
/// lost with it; the engine's own ack emission (a scripted ack routes to the engine) restores it.</item>
/// <item>Known gaps (degraded path only, review m1 / m2): a degraded non-rebuild apply does not advance
/// <c>_lastApplied</c>, so the next apply counts its conditions as added again and may replay onto them;
/// and the empty-plan repair of a degraded composition scopes no condition (the composition is
/// unchanged), so a condition the repair newly places gets no window state — writes to it between the
/// two applies were already dropped before this replay existed.</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,
IReadOnlyDictionary<(AddressSpaceRealm, string), VariableShape> Reshaped,
IReadOnlySet<(AddressSpaceRealm, string)> Conditions)
{
private static readonly IReadOnlySet<(AddressSpaceRealm, string)> None = new HashSet<(AddressSpaceRealm, string)>();
public static readonly ReplayScope Nothing =
new(false, None, new Dictionary<(AddressSpaceRealm, string), VariableShape>(), 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
/// are applied at once AND remembered in <see cref="_deferred"/> for replay after the apply.</summary>
private bool _planInFlight;
private readonly Queue<(RebuildAddressSpace Request, IActorRef Requester)> _pendingRebuilds = new();
private readonly IOpcUaAddressSpaceSink _sink;
private readonly IServiceLevelPublisher _serviceLevel;
private readonly bool _subscribeRedundancyTopic;
private readonly NodeId? _localNode;
private readonly IDbContextFactory<OtOpcUaConfigDbContext>? _dbFactory;
private readonly AddressSpaceApplier? _applier;
/// <summary>Receives the API keys + ACL grants carried by each applied artifact, so
/// OPC UA session activation can authenticate. Null when nothing consumes it (tests, and any node
@@ -184,9 +281,9 @@ public sealed class OpcUaPublishActor : ReceiveActor, IWithTimers
/// <summary>Production Props — pins the OPC UA dispatcher + subscribes to the
/// <c>redundancy-state</c> DPS topic so cluster transitions drive the local ServiceLevel
/// publish path. When <paramref name="dbFactory"/> + <paramref name="applier"/> are supplied,
/// <see cref="RebuildAddressSpace"/> reads the latest deployment artifact + drives the
/// applier through the sink. When <paramref name="dbHealthProbe"/> is supplied the local
/// publish path. When <paramref name="applier"/> is supplied, <see cref="RebuildAddressSpace"/>
/// diffs + applies the artifact bytes it carries through the sink (<paramref name="dbFactory"/> is
/// accepted for wiring compatibility only — review 02 P5 removed the ConfigDb fallback). When <paramref name="dbHealthProbe"/> is supplied the local
/// ServiceLevel is computed via <see cref="ServiceLevelCalculator"/> from real DB-health +
/// staleness + role-leader inputs; otherwise the legacy role-only switch is used.</summary>
/// <param name="sink">The OPC UA address space sink.</param>
@@ -293,6 +390,28 @@ public sealed class OpcUaPublishActor : ReceiveActor, IWithTimers
dataPlaneAuthority,
utcNow));
/// <summary>Test-only Props exposing the deferred-write bound and the plan-step seams.</summary>
internal static Props PropsForTests(
IOpcUaAddressSpaceSink sink,
AddressSpaceApplier? applier,
TestHooks? hooks,
int deferredWriteCapacity = DefaultDeferredWriteCapacity,
NodeId? localNode = null,
IDataPlaneAuthoritySink? dataPlaneAuthority = null) =>
// 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)));
/// <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>
/// <param name="serviceLevel">The service level publisher.</param>
@@ -329,13 +448,41 @@ public sealed class OpcUaPublishActor : ReceiveActor, IWithTimers
bool dbLess = false,
IDataPlaneAuthoritySink? dataPlaneAuthority = 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;
_sink = sink;
_dataPlaneAuthority = dataPlaneAuthority;
_serviceLevel = serviceLevel;
_subscribeRedundancyTopic = subscribeRedundancyTopic;
_localNode = localNode;
_dbFactory = dbFactory;
// dbFactory is accepted for wiring compatibility only (review 02 P5): the bytes-less ConfigDb read the
// rebuild used to fall back to is gone — every producer carries the artifact bytes in the message.
_ = dbFactory;
_applier = applier;
_dbHealthProbe = dbHealthProbe;
_dbLess = dbLess;
@@ -350,6 +497,8 @@ public sealed class OpcUaPublishActor : ReceiveActor, IWithTimers
Receive<AlarmStateUpdate>(HandleAlarmUpdate);
Receive<AlarmQualityUpdate>(HandleAlarmQualityUpdate);
Receive<RebuildAddressSpace>(HandleRebuild);
Receive<PlanReady>(HandlePlanReady);
Receive<PlanFailed>(HandlePlanFailed);
Receive<ServiceLevelChanged>(HandleServiceLevelChanged);
Receive<RedundancyStateChanged>(HandleRedundancyStateChanged);
Receive<DbHealthProbeActor.DbHealthStatus>(HandleDbHealthStatus);
@@ -358,6 +507,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()
{
@@ -380,6 +547,7 @@ public sealed class OpcUaPublishActor : ReceiveActor, IWithTimers
{
try
{
if (PlanPipelineBusy) DeferValue(new ValueWrite(msg.NodeId, msg.Value, msg.StatusCode, msg.TimestampUtc, msg.Realm));
_sink.WriteValue(msg.NodeId, msg.Value, msg.StatusCode, msg.TimestampUtc, msg.Realm);
Interlocked.Increment(ref _writes);
OtOpcUaTelemetry.OpcUaSinkWrite.Add(1, new KeyValuePair<string, object?>("kind", "value"));
@@ -401,6 +569,8 @@ public sealed class OpcUaPublishActor : ReceiveActor, IWithTimers
/// a whole batch because one item was bad.</para></summary>
private void HandleAttributeBatchUpdate(AttributeValueBatchUpdate msg)
{
if (PlanPipelineBusy)
foreach (var item in msg.Items) DeferValue(item);
try
{
_sink.WriteValues(msg.Items);
@@ -431,6 +601,8 @@ public sealed class OpcUaPublishActor : ReceiveActor, IWithTimers
/// value write on the same counter.</summary>
private void HandleAttributeStatusBatchUpdate(AttributeStatusBatchUpdate msg)
{
if (PlanPipelineBusy)
foreach (var item in msg.Items) DeferStatus(item);
try
{
_sink.WriteStatuses(msg.Items);
@@ -454,6 +626,7 @@ public sealed class OpcUaPublishActor : ReceiveActor, IWithTimers
private void HandleAlarmUpdate(AlarmStateUpdate msg)
{
if (PlanPipelineBusy) DeferAlarm(msg);
try
{
_sink.WriteAlarmCondition(msg.AlarmNodeId, msg.State, msg.TimestampUtc, msg.Realm);
@@ -468,6 +641,7 @@ public sealed class OpcUaPublishActor : ReceiveActor, IWithTimers
private void HandleAlarmQualityUpdate(AlarmQualityUpdate msg)
{
if (PlanPipelineBusy) DeferAlarmQuality(msg);
try
{
_sink.WriteAlarmQuality(msg.AlarmNodeId, msg.StatusCode, msg.TimestampUtc, msg.Realm);
@@ -480,114 +654,467 @@ public sealed class OpcUaPublishActor : ReceiveActor, IWithTimers
}
}
private void HandleRebuild(RebuildAddressSpace msg)
private void HandleRebuild(RebuildAddressSpace msg) => StartRebuild(msg, Sender);
/// <summary>
/// The entry half of a rebuild, on the actor thread. The raw-sink fallback (no applier) and the
/// no-artifact case complete synchronously; otherwise the artifact parse, the cluster-scoped composition
/// and the plan — the O(artifact) work that used to serialise every live value batch and (because the
/// node manager's lock is the SDK's) every client Publish behind it — run on the thread pool and come
/// back as a <see cref="PlanReady"/>. Only the sink materialisation (<see cref="ApplyPlan"/>) runs here.
/// </summary>
private void StartRebuild(RebuildAddressSpace msg, IActorRef requester)
{
// Captured up front: the pipeline below is synchronous on this actor's thread today, but the reply
// must reach whoever asked even if a future edit introduces an await into it.
var requester = Sender;
(bool Degraded, string? Reason) outcome;
try
// Strict arrival order for EVERY rebuild, including the ones that complete synchronously (no applier, no
// bytes): a bytes-less request arriving mid-plan used to answer before the plan ahead of it, so replies went
// out of order. HandleRebuild is the only entry that can find the pipeline busy; the drain calls this only
// when it is idle.
if (_planInFlight)
{
outcome = HandleRebuildCore(msg);
}
finally
{
// close the apply on EVERY exit, early returns and a throwing pass included: the node
// manager reports here the native conditions a rebuild carried over (after the notifier wiring the
// passes did) and discards any captured state no materialise claimed. Skipping it on a throw would
// leave that state for a later surgical re-add of the same id to resurrect. No-op when nothing is
// pending; never throws out of the handler.
try { _sink.CompleteAddressSpaceApply(); }
catch (Exception ex)
{
_log.Error(ex, "OpcUaPublish: sink.CompleteAddressSpaceApply threw (correlation={Correlation})", msg.Correlation);
}
_pendingRebuilds.Enqueue((msg, requester));
return;
}
// the outcome goes out only after the apply (including its completion step) is done.
if (msg.ReportOutcome)
requester.Tell(new RebuildCompleted(msg.Correlation, msg.DeploymentId, outcome.Degraded, outcome.Reason));
}
/// <summary>The rebuild pipeline. Returns whether the apply was degraded, and why.</summary>
private (bool Degraded, string? Reason) HandleRebuildCore(RebuildAddressSpace msg)
{
using var span = OtOpcUaTelemetry.StartAddressSpaceRebuildSpan();
span?.SetTag("otopcua.correlation_id", msg.Correlation.ToString());
// Two modes. With an applier wired, do a REAL diff-and-apply pass — driven either by the
// artifact bytes already in hand (a per-cluster-mesh driver-only / FetchAndCache node, whose
// ConfigDb is admin-only now so _dbFactory is null, hands its fetched/cached bytes in
// msg.Artifact) or, when no bytes are carried, by loading the artifact from the ConfigDb (the
// Direct path, which requires _dbFactory). Only a node with NO applier at all (the F10b/dev
// seam) falls back to the raw-sink rebuild — and that fallback WIPES the served address space
// without re-materialising, so it must NEVER be reached by a real driver node that has an
// applier but simply has no ConfigDb (that node diff-and-applies from its in-hand bytes below).
// Only a node with NO applier at all (the F10b/dev seam) falls back to the raw-sink rebuild — and that
// fallback WIPES the served address space without re-materialising, so it must NEVER be reached by a
// real driver node that has an applier (that node diff-and-applies from its in-hand bytes below).
if (_applier is null)
{
// Every rebuild is traced, whichever path completes it: the applier path opens its span in ApplyPlan
// (the actor-thread half), the two synchronous completions below open their own.
using var span = OtOpcUaTelemetry.StartAddressSpaceRebuildSpan();
span?.SetTag("otopcua.correlation_id", msg.Correlation.ToString());
(bool Degraded, string? Reason) outcome;
try
{
_sink.RebuildAddressSpace();
OtOpcUaTelemetry.OpcUaSinkWrite.Add(1, new KeyValuePair<string, object?>("kind", "rebuild"));
outcome = (false, null);
}
catch (Exception ex)
{
_log.Error(ex, "OpcUaPublish: sink.RebuildAddressSpace threw (correlation={Correlation})",
msg.Correlation);
return (true, $"address-space rebuild threw: {ex.Message}");
_log.Error(ex, "OpcUaPublish: sink.RebuildAddressSpace threw (correlation={Correlation})", msg.Correlation);
outcome = (true, $"address-space rebuild threw: {ex.Message}");
}
return (false, null);
FinishRebuild(msg, requester, outcome);
return;
}
// An artifact we could not obtain is NOT an empty configuration. Parsing zero bytes yields an empty
// composition, which the planner diffs against the live one as a PureRemove and the applier then
// faithfully tears down. Nothing legitimate produces a zero-length blob (an operator who really deletes
// everything still deploys a JSON document with empty arrays), so the only honest reading of "no bytes"
// is "no answer" — hold the last-known-good address space and let the next deploy (or the next boot)
// supply a real one. Leaving _lastApplied intact means the retry diffs against what is materialised.
if (msg.Artifact is null or { Length: 0 })
{
using var span = OtOpcUaTelemetry.StartAddressSpaceRebuildSpan();
span?.SetTag("otopcua.correlation_id", msg.Correlation.ToString());
if (_hasAppliedComposition)
_log.Warning(
"OpcUaPublish: no usable artifact for deployment {Id} (correlation={Correlation}) — KEEPING the currently served address space rather than tearing it down",
msg.DeploymentId, msg.Correlation);
else
_log.Debug(
"OpcUaPublish: no artifact to materialise yet (deployment={Id}, correlation={Correlation})",
msg.DeploymentId, msg.Correlation);
FinishRebuild(msg, requester, (true, "no usable artifact reached the address-space apply; the previously served address space was kept"));
return;
}
_planInFlight = true;
var artifact = msg.Artifact;
var baseComposition = _lastApplied;
var nodeId = _localNode?.Value;
var wantAuthority = _dataPlaneAuthority is not null;
var incarnation = _incarnation;
var hooks = _hooks;
Task.Run(() =>
{
hooks?.BeforePlan?.Invoke();
// Open() throws on bytes that are not valid JSON, where the static wrappers swallowed to an EMPTY
// result — and an empty composition is not a benign no-op here (see the no-bytes guard above). The
// throw lands in PlanFailed: logged at Error, _lastApplied untouched, last-known-good still served.
using var parsed = DeploymentArtifact.Open(artifact);
// The authority parse is isolated from the address-space plan (review fix I2): before the plan moved
// off-thread a throw here was contained by FeedDataPlaneAuthority's own catch; it must not now
// become PlanFailed and degrade an apply whose topology is perfectly readable.
DataPlaneAuthorityProjection? authority = null;
Exception? authorityError = null;
if (wantAuthority)
{
try
{
authority = hooks?.ParseAuthority is { } parse
? parse(parsed, nodeId ?? string.Empty)
: parsed.ParseDataPlaneAuthority(nodeId ?? string.Empty);
}
catch (Exception ex)
{
authorityError = ex;
}
}
var inconsistencies = new List<string>();
var composition = nodeId is not null
? parsed.ParseComposition(nodeId, inconsistencies.Add)
: parsed.ParseComposition();
var plan = AddressSpacePlanner.Compute(baseComposition, composition);
return new PlanReady(incarnation, msg, requester, artifact, authority, authorityError, composition, plan, inconsistencies);
})
.PipeTo(Self, failure: ex => new PlanFailed(incarnation, msg, requester, ex));
}
private void HandlePlanReady(PlanReady ready)
{
if (IsFromAnotherIncarnation(ready.Incarnation, ready.Request, ready.Requester)) return;
if (ready.AuthorityError is { } authorityError)
_log.Error(authorityError,
"OpcUaPublish: data-plane authority parse threw (correlation={Correlation}) — the address space is applied; the previously installed keys and grants stay in force",
ready.Request.Correlation);
(bool Degraded, string? Reason) outcome;
try
{
outcome = ApplyPlan(ready);
}
finally
{
_planInFlight = false;
}
FinishRebuild(ready.Request, ready.Requester, outcome);
AfterPlanSettled();
}
private void HandlePlanFailed(PlanFailed failed)
{
if (IsFromAnotherIncarnation(failed.Incarnation, failed.Request, failed.Requester)) return;
_planInFlight = false;
_log.Error(failed.Error, "OpcUaPublish: rebuild pipeline threw (correlation={Correlation})", failed.Request.Correlation);
FinishRebuild(failed.Request, failed.Requester, (true, $"address-space apply threw: {failed.Error.Message}"));
AfterPlanSettled();
}
/// <summary>A plan piped back to a RESTARTED instance (the ActorRef survives a restart) was computed against the
/// previous instance's <see cref="_lastApplied"/>: never applied. Its requester is still owed an answer.</summary>
private bool IsFromAnotherIncarnation(object incarnation, RebuildAddressSpace request, IActorRef requester)
{
if (ReferenceEquals(incarnation, _incarnation)) return false;
_log.Warning(
"OpcUaPublish: ignoring a rebuild plan started by a previous incarnation of this actor (correlation={Correlation})",
request.Correlation);
if (request.ReportOutcome)
requester.Tell(new RebuildCompleted(request.Correlation, request.DeploymentId, true,
"the address-space publisher restarted while this rebuild was being planned; it was not applied"));
return true;
}
/// <summary>After an apply (or a failed plan) has been answered — its deferred writes already replayed by
/// <see cref="FinishRebuild"/> — start what queued behind it, and drop the buffer once nothing is in flight or
/// queued (a queued plan may add nodes the buffer names, so it is replayed again after that apply).</summary>
private void AfterPlanSettled()
{
StartNextPendingRebuild();
if (!PlanPipelineBusy) _deferred.Clear();
}
/// <summary>Starts queued rebuilds in arrival order until one goes off-thread (or the queue empties) — the
/// synchronous completions (no bytes) answer in their turn.</summary>
private void StartNextPendingRebuild()
{
while (!_planInFlight && _pendingRebuilds.TryDequeue(out var next))
StartRebuild(next.Request, next.Requester);
}
/// <summary>True while a plan is computing or a rebuild is queued behind one.</summary>
private bool PlanPipelineBusy => _planInFlight || _pendingRebuilds.Count > 0;
private void DeferValue(in ValueWrite write)
{
if (!_deferred.Value(write)) WarnDeferredOverflow();
}
private void DeferStatus(in StatusWrite write)
{
if (!_deferred.Status(write)) WarnDeferredOverflow();
}
private void DeferAlarm(AlarmStateUpdate msg)
{
if (!_deferred.Alarm(msg)) WarnDeferredOverflow();
}
private void DeferAlarmQuality(AlarmQualityUpdate msg)
{
if (!_deferred.AlarmQuality(msg)) WarnDeferredOverflow();
}
private void WarnDeferredOverflow()
{
if (_deferred.Dropped == 1)
_log.Warning(
"OpcUaPublish: more than {Capacity} distinct nodes written while a rebuild plan is in flight — later new nodes are not held for replay (their first value lands only if the driver republishes it)",
_deferred.Capacity);
}
/// <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(ReplayScope scope)
{
if (_deferred.IsEmpty) return;
var (allValues, allStatuses) = _deferred.VariableWrites(
key => scope.AllVariables || scope.Variables.Contains(key) || scope.Reshaped.ContainsKey(key));
var values = new List<ValueWrite>(allValues.Count);
var statuses = new List<StatusWrite>(allStatuses.Count);
var unfit = 0;
foreach (var v in allValues)
{
if (!scope.Reshaped.TryGetValue((v.Realm, v.NodeId), out var shape)) { values.Add(v); continue; }
if (shape.TryFit(v.Value, out var fitted)) { values.Add(v with { Value = fitted }); continue; }
unfit++;
_log.Debug(
"OpcUaPublish: a value deferred for re-shaped {Realm} node {Node} does not fit its new shape ({DataType}, array={IsArray}) — not replayed; the node waits for the driver's next value",
v.Realm, v.NodeId, shape.DataType, shape.IsArray);
}
foreach (var st in allStatuses)
{
// A status-only annotation onto a re-shaped (value-less) node would serve a null value under it; the
// node's honest state is BadWaitingForInitialData until a value of the new shape arrives.
if (scope.Reshaped.ContainsKey((st.Realm, st.NodeId))) { unfit++; continue; }
statuses.Add(st);
}
if (values.Count > 0)
{
try { _sink.WriteValues(values); }
catch (Exception ex) { _log.Warning(ex, "OpcUaPublish: replaying {Count} deferred value writes threw", values.Count); }
}
if (statuses.Count > 0)
{
try { _sink.WriteStatuses(statuses); }
catch (Exception ex) { _log.Warning(ex, "OpcUaPublish: replaying {Count} deferred status writes threw", statuses.Count); }
}
var alarms = 0;
foreach (var alarm in _deferred.Alarms)
{
if (!scope.Conditions.Contains((alarm.Realm, alarm.AlarmNodeId))) continue;
alarms++;
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); }
}
var qualities = 0;
foreach (var quality in _deferred.AlarmQualities)
{
if (!scope.Conditions.Contains((quality.Realm, quality.AlarmNodeId))) continue;
qualities++;
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, {Alarms} alarm-state and {Qualities} alarm-quality writes (of {Held} held; {Unfit} not fitting a re-shaped node) deferred during a rebuild plan",
values.Count, statuses.Count, alarms, qualities, _deferred.Count, unfit);
}
/// <summary>
/// The latest write per (realm, id) received while a plan is in flight — see <see cref="_deferred"/>. A value
/// followed by a status-only write for the same node composes to the value under the new status (exactly what
/// the two live writes leave on the node). Bounded by <see cref="Capacity"/> distinct keys across all kinds;
/// beyond it a NEW key is not held (an existing key still updates), counted in <see cref="Dropped"/>.
/// </summary>
private sealed class DeferredWrites(int capacity)
{
private struct VariableEntry
{
public bool HasValue;
public object? Value;
public uint Status;
public DateTime TimestampUtc;
}
private readonly Dictionary<(AddressSpaceRealm, string), VariableEntry> _variables = new();
private readonly Dictionary<(AddressSpaceRealm, string), AlarmStateUpdate> _alarms = new();
private readonly Dictionary<(AddressSpaceRealm, string), AlarmQualityUpdate> _qualities = new();
public int Capacity { get; } = capacity;
public long Dropped { get; private set; }
public int Count => _variables.Count + _alarms.Count + _qualities.Count;
public bool IsEmpty => Count == 0;
public IEnumerable<AlarmStateUpdate> Alarms => _alarms.Values;
public IEnumerable<AlarmQualityUpdate> AlarmQualities => _qualities.Values;
public bool Value(in ValueWrite write)
{
if (string.IsNullOrEmpty(write.NodeId)) return true;
var key = (write.Realm, write.NodeId);
if (!_variables.ContainsKey(key) && !HasRoom()) return false;
_variables[key] = new VariableEntry
{
HasValue = true, Value = write.Value, Status = write.StatusCode, TimestampUtc = write.SourceTimestampUtc,
};
return true;
}
public bool Status(in StatusWrite write)
{
if (string.IsNullOrEmpty(write.NodeId)) return true;
var key = (write.Realm, write.NodeId);
if (_variables.TryGetValue(key, out var entry))
{
entry.Status = write.StatusCode;
entry.TimestampUtc = write.SourceTimestampUtc;
_variables[key] = entry;
return true;
}
if (!HasRoom()) return false;
_variables[key] = new VariableEntry { Status = write.StatusCode, TimestampUtc = write.SourceTimestampUtc };
return true;
}
/// <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(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));
}
return (values, statuses);
}
public void Clear()
{
_variables.Clear();
_alarms.Clear();
_qualities.Clear();
Dropped = 0;
}
private bool Put<T>(Dictionary<(AddressSpaceRealm, string), T> map, (AddressSpaceRealm, string) key, T value)
{
if (string.IsNullOrEmpty(key.Item2)) return true;
if (!map.ContainsKey(key) && !HasRoom()) return false;
map[key] = value;
return true;
}
private bool HasRoom()
{
if (Count < Capacity) return true;
Dropped++;
return false;
}
}
/// <summary>
/// Closes an apply on EVERY exit — early returns and a throwing pass included: the node manager reports
/// here the native conditions a rebuild carried over (after the notifier wiring the passes did) and
/// discards any captured state no materialise claimed. Skipping it on a throw would leave that state for a
/// later surgical re-add of the same id to resurrect. No-op when nothing is pending; never throws. The
/// outcome goes out only after the apply (including its completion step) is done.
/// </summary>
private void FinishRebuild(RebuildAddressSpace msg, IActorRef requester, (bool Degraded, string? Reason) outcome)
{
try { _sink.CompleteAddressSpaceApply(); }
catch (Exception ex)
{
_log.Error(ex, "OpcUaPublish: sink.CompleteAddressSpaceApply threw (correlation={Correlation})", msg.Correlation);
}
// 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(_replayScope);
_replayScope = ReplayScope.Nothing;
if (msg.ReportOutcome)
requester.Tell(new RebuildCompleted(msg.Correlation, msg.DeploymentId, outcome.Degraded, outcome.Reason));
}
/// <summary>The actor-thread half of a rebuild: install the artifact's authority, apply the plan through the
/// sink passes, announce. Returns whether the apply was degraded, and why.</summary>
private (bool Degraded, string? Reason) ApplyPlan(PlanReady ready)
{
var previous = _lastApplied;
var recording = _applier!.BeginRecording();
try
{
var result = ApplyPlanCore(ready, out var rebuilt);
// What this apply created or reset (see _replayScope). A throwing pass still scopes to what it did.
var conditions = AddedConditions(previous, ready.Composition);
if (rebuilt)
{
// Review N3: a full rebuild recreates every SCRIPTED condition fresh — the node manager carries only
// NATIVE conditions' state over (OtOpcUaNodeManager.RebuildAddressSpace). The engine's re-assert
// (or any transition) handled in the window landed on the discarded instance, so every kept scripted
// condition must be replayed onto its new one. Native kept conditions stay "added only".
foreach (var a in ready.Composition.EquipmentScriptedAlarms)
if (a.Enabled) conditions.Add((AddressSpaceRealm.Uns, a.ScriptedAlarmId));
}
_replayScope = new ReplayScope(rebuilt, recording.Ensured, recording.Reshaped, conditions);
return result;
}
finally
{
_applier.EndRecording();
}
}
/// <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());
try
{
// An in-hand artifact (boot-from-cache / FetchAndCache) wins: the host already has the
// cached/fetched bytes, and the ConfigDb read the DeploymentId path would do is exactly what
// is unreachable during the outage this exists to survive — and is simply absent on a
// driver-only node with no ConfigDb. Otherwise, when a ConfigDb is present (Direct mode),
// prefer the artifact of the deployment the host just applied — at apply time it is not yet
// Sealed, so LoadLatestArtifact would return the PREVIOUS revision and materialise a stale
// composition (variables that don't match the SubscribeBulk refs); fall back to latest-sealed
// only for legacy callers that carry neither. With NO bytes AND no ConfigDb there is nothing
// to load, so yield Array.Empty<byte>() — the guard below reads that as "no answer" and
// abandons the rebuild, holding last-known-good rather than wiping the address space.
var artifact = msg.Artifact
?? (_dbFactory is not null
? (msg.DeploymentId is { } depId ? LoadArtifact(depId) : LoadLatestArtifact())
: Array.Empty<byte>());
// An artifact we could not obtain is NOT an empty configuration. Parsing zero
// bytes yields an empty composition, which the planner diffs against the live one as a
// PureRemove and the applier then faithfully tears down: a transient ConfigDb blip empties the
// served address space. Nothing legitimate produces a zero-length blob (an operator who really
// deletes everything still deploys a JSON document with empty arrays), so the only honest
// reading of "no bytes" is "no answer" — hold the last-known-good address space and let the
// next deploy (or the next boot) supply a real one. Returning here also leaves _lastApplied
// intact, so the retry diffs against what is actually materialised.
if (artifact is null or { Length: 0 })
{
if (_hasAppliedComposition)
_log.Warning(
"OpcUaPublish: no usable artifact for deployment {Id} (correlation={Correlation}) — KEEPING the currently served address space rather than tearing it down",
msg.DeploymentId, msg.Correlation);
else
_log.Debug(
"OpcUaPublish: no artifact to materialise yet (deployment={Id}, correlation={Correlation})",
msg.DeploymentId, msg.Correlation);
return (true, "no usable artifact reached the address-space apply; the previously served address space was kept");
}
// Parse the artifact ONCE for this rebuild. Every consumer below (data-plane authority, then the
// composition) reads this one JsonDocument instead of rebuilding it per question — the rebuild
// used to pay a full parse-and-throw-away per static call. Disposed when the handler returns.
//
// Open() throws on bytes that are not valid JSON, where the static wrappers swallowed to an
// EMPTY result — and an empty composition is not a benign no-op here: the planner diffs it as a
// PureRemove and the applier tears the served address space down. So the throw is the honest
// outcome, and it lands in this method's catch below: logged at Error, _lastApplied untouched,
// last-known-good still served. Same "no answer ≠ empty configuration" rule as the guard
// immediately above, applied to bytes that arrived unreadable rather than absent.
using var parsed = DeploymentArtifact.Open(artifact);
foreach (var inconsistency in ready.Inconsistencies)
_log.Warning("OpcUaPublish {Node}: cross-cluster binding — {Message}", _localNode, inconsistency);
// install this artifact's API keys + ACL grants BEFORE the address-space work
// below, and deliberately ahead of the `plan.IsEmpty` early-return: an artifact that revokes a
@@ -595,13 +1122,10 @@ public sealed class OpcUaPublishActor : ReceiveActor, IWithTimers
// feeding this after that return would leave a revoked key authenticating until some unrelated
// structural change happened to redeploy. Authorization and topology change on their own
// schedules.
FeedDataPlaneAuthority(parsed, artifact);
if (ready.Authority is not null) FeedDataPlaneAuthority(ready.Authority, ready.Artifact);
var composition = _localNode is { } ln
? parsed.ParseComposition(ln.Value,
inconsistency => _log.Warning("OpcUaPublish {Node}: cross-cluster binding — {Message}", ln, inconsistency))
: parsed.ParseComposition();
var plan = AddressSpacePlanner.Compute(_lastApplied, composition);
var composition = ready.Composition;
var plan = ready.Plan;
if (plan.IsEmpty)
{
@@ -610,7 +1134,7 @@ public sealed class OpcUaPublishActor : ReceiveActor, IWithTimers
// The composition is unchanged but the last apply of it was a degraded
// rebuild: re-ensure every node (each Ensure* no-ops on a node that stands) so the
// re-deploy actually repairs what that rebuild could not place.
var repairFailed = _applier.MaterialiseHierarchy(composition)
var repairFailed = _applier!.MaterialiseHierarchy(composition)
+ MaterialiseNodes(plan, composition, fullComposition: true);
if (repairFailed > 0)
{
@@ -632,7 +1156,8 @@ public sealed class OpcUaPublishActor : ReceiveActor, IWithTimers
return (false, null);
}
var outcome = _applier.Apply(plan);
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
@@ -725,8 +1250,8 @@ public sealed class OpcUaPublishActor : ReceiveActor, IWithTimers
}
/// <summary>
/// Run the five node-materialising applier passes — Raw subtree, UNS references, scripted alarms,
/// equipment tags, equipment VirtualTags — either over the FULL composition (after a rebuild) or over
/// Run the four node-materialising applier passes — Raw subtree, UNS references, scripted alarms,
/// equipment VirtualTags — either over the FULL composition (after a rebuild) or over
/// just this plan's delta (surgical deploys). <c>MaterialiseHierarchy</c> is NOT one of them:
/// the caller always runs it in full, because parents must exist before any add.
/// </summary>
@@ -809,21 +1334,7 @@ public sealed class OpcUaPublishActor : ReceiveActor, IWithTimers
if (addedAlarms.Count > 0) failedNodes += applier.MaterialiseScriptedAlarms(composition, addedAlarms);
}
// Equipment-namespace tags get their own pass: ensures each signal's Variable (and any
// FolderPath sub-folder) exists under its already-materialised equipment folder so
// clients can browse them. Live values are pushed by DriverHostActor.ForwardBatchToMux after
// each subscription cycle; variables show BadWaitingForInitialData only until the first
// publish interval fires.
if (fullComposition)
{
failedNodes += applier.MaterialiseEquipmentTags(composition);
}
else if (plan.AddedEquipmentTags.Count > 0)
{
failedNodes += applier.MaterialiseEquipmentTags(composition, plan.AddedEquipmentTags);
}
// Equipment-namespace VirtualTags get their own pass right after the equipment tags:
// Equipment-namespace VirtualTags get their own pass right after the scripted alarms:
// ensures each computed signal's Variable (and any FolderPath sub-folder) exists under its
// equipment folder with a folder-scoped NodeId. VirtualTagHostActor.OnResult pushes live
// values once the first dependency update arrives; until then variables show BadWaitingForInitialData.
@@ -881,17 +1392,15 @@ public sealed class OpcUaPublishActor : ReceiveActor, IWithTimers
/// the safe direction for availability but means a revocation carried by an artifact this node
/// could not decode has not taken effect. It is logged at Error for exactly that reason.
/// </remarks>
/// <param name="parsed">The rebuild's single parse of the artifact, read for the keys + grants.</param>
/// <param name="projection">The keys + grants decoded (off-thread) from the rebuild's single parse of the artifact.</param>
/// <param name="artifactBytes">The RAW artifact bytes — hashed for the revision id, which must be
/// SHA-256 of the exact blob (see below); never re-parsed here.</param>
private void FeedDataPlaneAuthority(ParsedDeploymentArtifact parsed, byte[] artifactBytes)
private void FeedDataPlaneAuthority(DataPlaneAuthorityProjection projection, byte[] artifactBytes)
{
if (_dataPlaneAuthority is null) return;
try
{
var nodeId = _localNode?.Value ?? string.Empty;
var projection = parsed.ParseDataPlaneAuthority(nodeId);
// The revision hash is SHA-256 of the artifact bytes — the same derivation ConfigComposer uses,
// which is what lets a node recognise a re-applied revision without the hash riding the wire.
// Hashed off the RAW bytes for exactly that reason: a re-serialisation of the parsed document
@@ -936,46 +1445,6 @@ public sealed class OpcUaPublishActor : ReceiveActor, IWithTimers
}
}
/// <summary>Read a specific deployment's artifact blob from ConfigDb (the one just applied,
/// which may not be Sealed yet). Empty array on any failure — <see cref="HandleRebuild"/> treats that as
/// "no answer" and abandons the rebuild, leaving the served address space untouched.</summary>
private byte[] LoadArtifact(DeploymentId deploymentId)
{
try
{
using var db = _dbFactory!.CreateDbContext();
return db.Deployments.AsNoTracking()
.Where(d => d.DeploymentId == deploymentId.Value)
.Select(d => d.ArtifactBlob)
.FirstOrDefault() ?? Array.Empty<byte>();
}
catch (Exception ex)
{
_log.Warning(ex, "OpcUaPublish: failed to load artifact for deployment {Id}; the rebuild is abandoned", deploymentId);
return Array.Empty<byte>();
}
}
/// <summary>Read the most recent <c>Sealed</c> deployment's artifact blob from ConfigDb.
/// Empty array on any failure — see <see cref="LoadArtifact"/> for how that is handled.</summary>
private byte[] LoadLatestArtifact()
{
try
{
using var db = _dbFactory!.CreateDbContext();
return db.Deployments.AsNoTracking()
.Where(d => d.Status == Configuration.Enums.DeploymentStatus.Sealed)
.OrderByDescending(d => d.SealedAtUtc)
.Select(d => d.ArtifactBlob)
.FirstOrDefault() ?? Array.Empty<byte>();
}
catch (Exception ex)
{
_log.Warning(ex, "OpcUaPublish: failed to load latest deployment artifact; the rebuild is abandoned");
return Array.Empty<byte>();
}
}
private void HandleServiceLevelChanged(ServiceLevelChanged msg)
{
// Always publish the FIRST computed level, even if it equals the byte-default 0. Otherwise a