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

feat(runtime): F10 OpcUaPublishActor sink seams + redundancy-driven ServiceLevel

Browse Source 
OpcUaPublishActor now routes through pluggable seams instead of just
incrementing a counter:

- IOpcUaAddressSpaceSink (Commons.OpcUa) — WriteValue / WriteAlarmState
  / RebuildAddressSpace. OpcUaQuality enum moved here from the actor's
  nested type so producers don't have to reference the actor itself.
- IServiceLevelPublisher — Publish(byte). NullServiceLevelPublisher
  retains the last level for inspection.
- The actor subscribes to the redundancy-state DPS topic in PreStart
  and maps the local node's NodeRedundancyState to a coarse
  ServiceLevel (Primary+leader=240, Primary=200, Secondary=100,
  Detached=0). This keeps the local SDK's ServiceLevel node honest
  without round-tripping back through the admin-singleton calculator.
- ServiceLevelChanged dedupes identical levels so the SDK doesn't see
  redundant writes.
- Sink + publisher exceptions are caught and logged; the actor never
  crashes its own dispatcher.
- PropsForTests gets optional sink/publisher/localNode params and
  skips the DPS subscribe so unit tests stay on a vanilla TestKit
  cluster.

Production binding to a real SDK NodeManager + Variable nodes is the
remaining residual — split as F10b. Task 60 still blocked on F10b.

Tests: Runtime 40 -> 46 (+6):
- AttributeValueUpdate routes to sink
- AlarmStateUpdate routes to sink
- RebuildAddressSpace calls sink.Rebuild
- ServiceLevelChanged dedupes
- RedundancyStateChanged for primary-leader publishes 240
- RedundancyStateChanged for secondary publishes 100

All 6 v2 test suites green: 132 tests passing.
This commit is contained in:
Joseph Doherty
2026-05-26 09:10:55 -04:00
parent 14fb2b05ed
commit a1325299ce
5 changed files with 340 additions and 40 deletions
Download Patch File Download Diff File Expand all files Collapse all files

37
src/Core/ZB.MOM.WW.OtOpcUa.Commons/OpcUa/IOpcUaAddressSpaceSink.cs Normal file
View File

@@ -0,0 +1,37 @@
namespace ZB.MOM.WW.OtOpcUa.Commons.OpcUa;
/// <summary>
/// Abstraction over the OPC UA SDK's address space. <c>OpcUaPublishActor</c> consumes this
/// so the Runtime project doesn't reference <c>Opc.Ua.Server</c> directly — production
/// binds a real SDK-backed sink in the fused Host's wiring, dev/Mac binds the
/// <see cref="NullOpcUaAddressSpaceSink"/> no-op.
/// </summary>
public interface IOpcUaAddressSpaceSink
{
/// <summary>Write a Variable node's current value + quality + source timestamp.</summary>
void WriteValue(string nodeId, object? value, OpcUaQuality quality, DateTime sourceTimestampUtc);
/// <summary>Write an alarm-condition Variable's active/acknowledged state.</summary>
void WriteAlarmState(string alarmNodeId, bool active, bool acknowledged, DateTime sourceTimestampUtc);
/// <summary>
/// Tear down + repopulate the address space. Called by <c>OpcUaPublishActor</c> after a
/// successful deployment apply so the node manager reflects the new config. Idempotent.
/// </summary>
void RebuildAddressSpace();
}
/// <summary>OPC UA status code projection — Good / Uncertain / Bad. Real SDK has finer-grained
/// codes; the engine actors only need this 3-state classification.</summary>
public enum OpcUaQuality { Good, Uncertain, Bad }
/// <summary>No-op sink. Bound by default so the actors are safe to run in dev / Mac /
/// integration tests without a real SDK behind them.</summary>
public sealed class NullOpcUaAddressSpaceSink : IOpcUaAddressSpaceSink
{
public static readonly NullOpcUaAddressSpaceSink Instance = new();
private NullOpcUaAddressSpaceSink() { }
public void WriteValue(string nodeId, object? value, OpcUaQuality quality, DateTime sourceTimestampUtc) { }
public void WriteAlarmState(string alarmNodeId, bool active, bool acknowledged, DateTime sourceTimestampUtc) { }
public void RebuildAddressSpace() { }
}

22
src/Core/ZB.MOM.WW.OtOpcUa.Commons/OpcUa/IServiceLevelPublisher.cs Normal file
View File

@@ -0,0 +1,22 @@
namespace ZB.MOM.WW.OtOpcUa.Commons.OpcUa;
/// <summary>
/// Writes the OPC UA Server object's <c>ServiceLevel</c> Variable (0–255). Production binds
/// a sink that pokes the SDK's ServiceLevel node; tests + dev mode bind
/// <see cref="NullServiceLevelPublisher"/> which just records the most recently set level
/// for inspection.
/// </summary>
public interface IServiceLevelPublisher
{
void Publish(byte serviceLevel);
}
/// <summary>No-op default that retains the last-written ServiceLevel in
/// <see cref="LastPublished"/>. Used by dev mode + verified by tests.</summary>
public sealed class NullServiceLevelPublisher : IServiceLevelPublisher
{
public static readonly NullServiceLevelPublisher Instance = new();
private NullServiceLevelPublisher() { }
public byte LastPublished { get; private set; }
public void Publish(byte serviceLevel) => LastPublished = serviceLevel;
}

182
src/Server/ZB.MOM.WW.OtOpcUa.Runtime/OpcUa/OpcUaPublishActor.cs
View File

@@ -1,70 +1,178 @@
using Akka.Actor;
using Akka.Cluster.Tools.PublishSubscribe;
using Akka.Event;
using ZB.MOM.WW.OtOpcUa.Commons.Messages.Redundancy;
using ZB.MOM.WW.OtOpcUa.Commons.OpcUa;
using ZB.MOM.WW.OtOpcUa.Commons.Types;
namespace ZB.MOM.WW.OtOpcUa.Runtime.OpcUa;
/// <summary>
/// Single-threaded bridge between Akka messages and the OPC UA SDK address space. Hosted on
/// the pinned <c>opcua-synchronized-dispatcher</c> (Task 19 HOCON) so the OPC UA SDK sees
/// only one thread per actor instance — its session/subscription locks expect strict
/// single-threaded access.
/// Single-threaded bridge between Akka messages and the OPC UA SDK address space. Hosted on
/// the pinned <c>opcua-synchronized-dispatcher</c> (Task 19 HOCON) so the OPC UA SDK sees
/// only one thread per actor instance — its session/subscription locks expect strict
/// single-threaded access.
///
/// Engine wiring (call into <c>OpcUaApplicationHost</c> address-space writes, manage
/// <c>ServiceLevel</c> + <c>ServerUriArray</c> nodes, subscribe to the <c>redundancy-state</c>
/// DistributedPubSub topic) is staged for follow-up F10. This skeleton compiles + exposes the
/// message contracts so producers (DriverInstance, VirtualTag, ScriptedAlarm) can target it.
/// Address-space writes route through <see cref="IOpcUaAddressSpaceSink"/>; ServiceLevel
/// writes route through <see cref="IServiceLevelPublisher"/>. Production binds SDK-backed
/// implementations; dev/Mac/tests bind the Null* defaults so the actor stays decoupled from
/// <c>Opc.Ua.Server</c>. The remaining piece is wiring those bindings to a real
/// <c>StandardServer</c> address space — tracked as F10b.
/// </summary>
public sealed class OpcUaPublishActor : ReceiveActor
{
public const string DispatcherId = "opcua-synchronized-dispatcher";
public const string RedundancyStateTopic = "redundancy-state";
public sealed record AttributeValueUpdate(string NodeId, object? Value, OpcUaQuality Quality, DateTime TimestampUtc);
public sealed record AlarmStateUpdate(string AlarmNodeId, bool Active, bool Acknowledged, DateTime TimestampUtc);
public sealed record RebuildAddressSpace(CorrelationId Correlation);
public sealed record ServiceLevelChanged(byte ServiceLevel);
public enum OpcUaQuality { Good, Uncertain, Bad }
private readonly IOpcUaAddressSpaceSink _sink;
private readonly IServiceLevelPublisher _serviceLevel;
private readonly bool _subscribeRedundancyTopic;
private readonly NodeId? _localNode;
private readonly ILoggingAdapter _log = Context.GetLogger();
private int _writes;
/// <summary>
/// Returns Props pre-configured to use the <c>opcua-synchronized-dispatcher</c>. Caller can
/// still override by chaining <c>.WithDispatcher(otherId)</c> for unit tests.
/// </summary>
public static Props Props() =>
Akka.Actor.Props.Create(() => new OpcUaPublishActor()).WithDispatcher(DispatcherId);
/// <summary>Test-only Props that omits the pinned dispatcher requirement.</summary>
public static Props PropsForTests() =>
Akka.Actor.Props.Create(() => new OpcUaPublishActor());
private byte _lastServiceLevel;
public int WriteCount => _writes;
public byte LastServiceLevel => _lastServiceLevel;
public OpcUaPublishActor()
/// <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.</summary>
public static Props Props(
IOpcUaAddressSpaceSink? sink = null,
IServiceLevelPublisher? serviceLevel = null,
NodeId? localNode = null) =>
Akka.Actor.Props.Create(() => new OpcUaPublishActor(
sink ?? NullOpcUaAddressSpaceSink.Instance,
serviceLevel ?? NullServiceLevelPublisher.Instance,
subscribeRedundancyTopic: true,
localNode)).WithDispatcher(DispatcherId);
/// <summary>Test-only Props that omits the pinned-dispatcher requirement and skips the
/// DPS subscribe so unit tests can spin up the actor on a vanilla TestKit cluster.</summary>
public static Props PropsForTests(
IOpcUaAddressSpaceSink? sink = null,
IServiceLevelPublisher? serviceLevel = null,
bool subscribeRedundancyTopic = false,
NodeId? localNode = null) =>
Akka.Actor.Props.Create(() => new OpcUaPublishActor(
sink ?? NullOpcUaAddressSpaceSink.Instance,
serviceLevel ?? NullServiceLevelPublisher.Instance,
subscribeRedundancyTopic,
localNode));
public OpcUaPublishActor(
IOpcUaAddressSpaceSink sink,
IServiceLevelPublisher serviceLevel,
bool subscribeRedundancyTopic,
NodeId? localNode)
{
Receive<AttributeValueUpdate>(msg =>
_sink = sink;
_serviceLevel = serviceLevel;
_subscribeRedundancyTopic = subscribeRedundancyTopic;
_localNode = localNode;
Receive<AttributeValueUpdate>(HandleAttributeUpdate);
Receive<AlarmStateUpdate>(HandleAlarmUpdate);
Receive<RebuildAddressSpace>(HandleRebuild);
Receive<ServiceLevelChanged>(HandleServiceLevelChanged);
Receive<RedundancyStateChanged>(HandleRedundancyStateChanged);
Receive<SubscribeAck>(_ => { /* PubSub ack */ });
}
protected override void PreStart()
{
if (_subscribeRedundancyTopic)
{
// F10: call into OpcUaApplicationHost to write the address-space node.
DistributedPubSub.Get(Context.System).Mediator.Tell(new Subscribe(RedundancyStateTopic, Self));
}
}
private void HandleAttributeUpdate(AttributeValueUpdate msg)
{
try
{
_sink.WriteValue(msg.NodeId, msg.Value, msg.Quality, msg.TimestampUtc);
Interlocked.Increment(ref _writes);
_log.Debug("OpcUaPublish: queued AttributeValueUpdate for {Node} ({Quality}) (write staged for F10)",
msg.NodeId, msg.Quality);
});
Receive<AlarmStateUpdate>(msg =>
}
catch (Exception ex)
{
_log.Warning(ex, "OpcUaPublish: sink.WriteValue threw for {Node}", msg.NodeId);
}
}
private void HandleAlarmUpdate(AlarmStateUpdate msg)
{
try
{
_sink.WriteAlarmState(msg.AlarmNodeId, msg.Active, msg.Acknowledged, msg.TimestampUtc);
Interlocked.Increment(ref _writes);
_log.Debug("OpcUaPublish: queued AlarmStateUpdate for {Node} (active={Active})",
msg.AlarmNodeId, msg.Active);
});
Receive<RebuildAddressSpace>(msg =>
}
catch (Exception ex)
{
_log.Info("OpcUaPublish: address-space rebuild requested (correlation={Correlation}); F10 wires the SDK call",
_log.Warning(ex, "OpcUaPublish: sink.WriteAlarmState threw for {Node}", msg.AlarmNodeId);
}
}
private void HandleRebuild(RebuildAddressSpace msg)
{
try
{
_sink.RebuildAddressSpace();
_log.Info("OpcUaPublish: address-space rebuilt (correlation={Correlation})", msg.Correlation);
}
catch (Exception ex)
{
_log.Error(ex, "OpcUaPublish: sink.RebuildAddressSpace threw (correlation={Correlation})",
msg.Correlation);
});
Receive<ServiceLevelChanged>(msg =>
}
}
private void HandleServiceLevelChanged(ServiceLevelChanged msg)
{
if (msg.ServiceLevel == _lastServiceLevel) return;
_lastServiceLevel = msg.ServiceLevel;
try
{
_log.Debug("OpcUaPublish: ServiceLevel={Level} (write staged for F10)", msg.ServiceLevel);
});
_serviceLevel.Publish(msg.ServiceLevel);
_log.Debug("OpcUaPublish: ServiceLevel={Level}", msg.ServiceLevel);
}
catch (Exception ex)
{
_log.Warning(ex, "OpcUaPublish: ServiceLevel publisher threw at level {Level}", msg.ServiceLevel);
}
}
/// <summary>
/// Compute a coarse ServiceLevel from the cluster snapshot and forward to the
/// <see cref="IServiceLevelPublisher"/>. This is a placeholder for F10b's full health
/// aggregation — for now we surface "primary-leader → 240, secondary → 100, detached → 0"
/// so the local SDK at least reflects role state. The full <see cref="ServiceLevelCalculator"/>
/// path (with DB-reachable, OPC UA probe inputs) lives in <c>RedundancyStateActor</c> on
/// admin nodes; this driver-side mirror exists so each node's own SDK exposes a sensible
/// ServiceLevel without round-tripping back through the admin singleton.
/// </summary>
private void HandleRedundancyStateChanged(RedundancyStateChanged msg)
{
if (_localNode is null) return;
var local = msg.Nodes.FirstOrDefault(n => n.NodeId == _localNode.Value);
if (local is null) return;
byte level = local.Role switch
{
RedundancyRole.Primary when local.IsRoleLeaderForDriver => 240,
RedundancyRole.Primary => 200,
RedundancyRole.Secondary => 100,
RedundancyRole.Detached => 0,
_ => 0,
};
Self.Tell(new ServiceLevelChanged(level));
}
}