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

Merge fix/issue-542-fleetstatus: delete the dead fleet-status chain (#542)

Browse Source 
Claude-Session: https://claude.ai/code/session_014WNM4vjoVksyyBraTXSZE1
This commit is contained in:
Joseph Doherty
2026-08-06 17:28:48 -04:00
parent 1df8470356 b5607d355a
commit d01b717eb9
16 changed files with 40 additions and 359 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/Server/ZB.MOM.WW.OtOpcUa.AdminUI/Hubs/AlertSignalRBridge.cs
+1 -2
View File
@@ -9,8 +9,7 @@ namespace ZB.MOM.WW.OtOpcUa.AdminUI.Hubs;
/// <summary>
/// Akka actor that subscribes to the <c>alerts</c> DistributedPubSub topic and forwards each
/// <see cref="AlarmTransitionEvent"/> to every SignalR client connected to <see cref="AlertHub"/>.
/// Mirrors <c>FleetStatusSignalRBridge</c>'s design — one bridge per admin node, hub fan-out is
/// per-node, no cluster-singleton needed.
/// One bridge per admin node — hub fan-out is per-node, so no cluster-singleton is needed.
/// </summary>
public sealed class AlertSignalRBridge : ReceiveActor
{
src/Server/ZB.MOM.WW.OtOpcUa.AdminUI/Hubs/FleetStatusHub.cs
-19
View File
@@ -1,19 +0,0 @@
using Microsoft.AspNetCore.Authorization;
using Microsoft.AspNetCore.SignalR;
using ZB.MOM.WW.OtOpcUa.Commons.Messages.Fleet;
namespace ZB.MOM.WW.OtOpcUa.AdminUI.Hubs;
/// <summary>
/// Browser-facing fleet-status push channel. Subscribers receive <see cref="FleetStatusChanged"/>
/// snapshots whenever the admin-role <c>FleetStatusBroadcaster</c> publishes a diff.
///
/// Server pushes fleet-status updates to connected clients via <c>FleetStatusSignalRBridge</c>
/// (DistributedPubSub 'fleet-status' → <c>IHubContext&lt;FleetStatusHub&gt;</c>).
/// </summary>
[Authorize]
public sealed class FleetStatusHub : Hub
{
public const string Endpoint = "/hubs/fleet-status";
public const string MethodName = "fleetStatusChanged";
}
src/Server/ZB.MOM.WW.OtOpcUa.AdminUI/Hubs/FleetStatusSignalRBridge.cs
-58
View File
@@ -1,58 +0,0 @@
using Akka.Actor;
using Akka.Cluster.Tools.PublishSubscribe;
using Akka.Event;
using Microsoft.AspNetCore.SignalR;
using ZB.MOM.WW.OtOpcUa.Commons.Messages.Fleet;
namespace ZB.MOM.WW.OtOpcUa.AdminUI.Hubs;
/// <summary>
/// Akka actor that subscribes to the <c>fleet-status</c> DistributedPubSub topic and forwards
/// every <see cref="FleetStatusChanged"/> snapshot to all SignalR clients connected to
/// <see cref="FleetStatusHub"/>. Spawned on admin-role nodes by
/// <c>AddOtOpcUaSignalRBridges</c>.
///
/// The bridge runs locally on each admin node — every node that hosts the hub also forwards
/// snapshots to its own connected clients. That keeps the hub-to-actor wiring simple (no
/// cluster-singleton coordination needed) at the cost of duplicated DPS subscriptions on
/// multi-admin deployments. Acceptable since the messages are small + the SignalR fan-out
/// is per-node anyway.
/// </summary>
public sealed class FleetStatusSignalRBridge : ReceiveActor
{
public const string TopicName = "fleet-status";
private readonly IHubContext<FleetStatusHub> _hub;
private readonly ILoggingAdapter _log = Context.GetLogger();
/// <summary>Creates actor props for a FleetStatusSignalRBridge with the specified hub context.</summary>
/// <param name="hub">The SignalR hub context.</param>
/// <returns>Props for creating a new FleetStatusSignalRBridge actor.</returns>
public static Props Props(IHubContext<FleetStatusHub> hub) =>
Akka.Actor.Props.Create(() => new FleetStatusSignalRBridge(hub));
/// <summary>Initializes a new instance of the FleetStatusSignalRBridge class.</summary>
/// <param name="hub">The SignalR hub context for sending messages to clients.</param>
public FleetStatusSignalRBridge(IHubContext<FleetStatusHub> hub)
{
_hub = hub;
ReceiveAsync<FleetStatusChanged>(ForwardAsync);
Receive<SubscribeAck>(_ => { /* DPS confirmation */ });
}
/// <inheritdoc />
protected override void PreStart() =>
DistributedPubSub.Get(Context.System).Mediator.Tell(new Subscribe(TopicName, Self));
private async Task ForwardAsync(FleetStatusChanged msg)
{
try
{
await _hub.Clients.All.SendAsync(FleetStatusHub.MethodName, msg);
}
catch (Exception ex)
{
_log.Warning(ex, "FleetStatusSignalRBridge: SignalR push failed (count={Count})", msg.Nodes.Count);
}
}
}
src/Server/ZB.MOM.WW.OtOpcUa.AdminUI/Hubs/HubRouteBuilderExtensions.cs
-1
View File
@@ -13,7 +13,6 @@ public static class HubRouteBuilderExtensions
/// <returns>The same endpoint route builder, for chaining.</returns>
public static IEndpointRouteBuilder MapOtOpcUaHubs(this IEndpointRouteBuilder app)
{
app.MapHub<FleetStatusHub>(FleetStatusHub.Endpoint);
app.MapHub<AlertHub>(AlertHub.Endpoint);
app.MapHub<ScriptLogHub>(ScriptLogHub.Endpoint);
app.MapHub<DriverStatusHub>(DriverStatusHub.Endpoint);
src/Server/ZB.MOM.WW.OtOpcUa.AdminUI/Hubs/HubServiceCollectionExtensions.cs
+2 -10
View File
@@ -14,7 +14,6 @@ namespace ZB.MOM.WW.OtOpcUa.AdminUI.Hubs;
public static class HubServiceCollectionExtensions
{
public const string FleetStatusSignalRBridgeName = "fleet-status-signalr-bridge";
public const string AlertSignalRBridgeName = "alert-signalr-bridge";
public const string ScriptLogSignalRBridgeName = "script-log-signalr-bridge";
public const string DriverStatusSignalRBridgeName = "driver-status-signalr-bridge";
@@ -44,8 +43,8 @@ public static class HubServiceCollectionExtensions
/// <summary>
/// Spawns the SignalR bridge actors that forward DPS messages to browser-facing SignalR
/// hubs: <c>fleet-status</c> → <see cref="FleetStatusHub"/>, <c>alerts</c> →
/// <see cref="AlertHub"/>, <c>script-logs</c> → <see cref="ScriptLogHub"/>.
/// hubs: <c>alerts</c> → <see cref="AlertHub"/>, <c>script-logs</c> →
/// <see cref="ScriptLogHub"/>, <c>driver-health</c> → <see cref="DriverStatusHub"/>.
///
/// Call inside the admin-role configurator on the shared <see cref="AkkaConfigurationBuilder"/>:
/// <code>
@@ -62,12 +61,6 @@ public static class HubServiceCollectionExtensions
{
builder.WithActors((system, registry, resolver) =>
{
// Fleet-status always stays on DPS in both modes (deferred / out of Phase 5 scope) — it is
// never gated by the telemetry dark switch.
var fleetHub = resolver.GetService<IHubContext<FleetStatusHub>>();
var fleetBridge = system.ActorOf(FleetStatusSignalRBridge.Props(fleetHub), FleetStatusSignalRBridgeName);
registry.Register<FleetStatusSignalRBridgeKey>(fleetBridge);
// The four telemetry sinks are registered identically in both modes; only the upstream that
// feeds them swaps (DPS bridges vs. the gRPC dial supervisor). Resolve them once, above the
// branch, so both paths feed the SAME singletons the Blazor panels read.
@@ -131,7 +124,6 @@ public static class HubServiceCollectionExtensions
}
/// <summary>Marker keys for <see cref="ActorRegistry"/> lookup of the SignalR bridge actors.</summary>
public sealed class FleetStatusSignalRBridgeKey { }
public sealed class AlertSignalRBridgeKey { }
public sealed class ScriptLogSignalRBridgeKey { }
public sealed class DriverStatusSignalRBridgeKey { }
src/Server/ZB.MOM.WW.OtOpcUa.ControlPlane/Fleet/FleetStatusBroadcaster.cs
-168
View File
@@ -1,168 +0,0 @@
using Akka.Actor;
using Akka.Cluster;
using Akka.Cluster.Tools.PublishSubscribe;
using Akka.Event;
using ZB.MOM.WW.OtOpcUa.Commons.Messages.Fleet;
using ZB.MOM.WW.OtOpcUa.Commons.Types;
namespace ZB.MOM.WW.OtOpcUa.ControlPlane.Fleet;
/// <summary>
/// Admin-role cluster singleton that maintains an in-memory <see cref="FleetStatusChanged"/>
/// snapshot from cluster membership + reachability events plus per-node heartbeats, and pushes
/// diffs on the <c>fleet-status</c> DistributedPubSub topic. The SignalR hub layer
/// subscribes to that topic and forwards to browser clients.
///
/// Heartbeat staleness: a node we haven't heard from in <see cref="HeartbeatTimeout"/> flips
/// from <see cref="FleetNodeHealth.Healthy"/> to <see cref="FleetNodeHealth.Degraded"/>. A
/// cluster <c>Unreachable</c> event flips it straight to <see cref="FleetNodeHealth.Unreachable"/>.
/// </summary>
public sealed class FleetStatusBroadcaster : ReceiveActor, IWithTimers
{
public const string Topic = "fleet-status";
public static readonly TimeSpan BroadcastInterval = TimeSpan.FromSeconds(5);
public static readonly TimeSpan HeartbeatTimeout = TimeSpan.FromSeconds(30);
private readonly Akka.Cluster.Cluster _cluster;
private readonly Action<object> _broadcast;
private readonly ILoggingAdapter _log = Context.GetLogger();
private readonly Dictionary<NodeId, NodeRecord> _nodes = new();
/// <summary>Gets or sets the timer scheduler for this actor.</summary>
public ITimerScheduler Timers { get; set; } = null!;
/// <summary>
/// Per-node heartbeat carrying the node's currently-applied revision hash, sent by
/// <c>DriverHostActor</c> periodically.
/// </summary>
public sealed record DriverHostStatusHeartbeat(NodeId NodeId, RevisionHash? CurrentRevision);
/// <summary>Creates actor props for the fleet status broadcaster.</summary>
/// <param name="broadcast">Optional custom broadcast delegate; defaults to distributed pub-sub mediator.</param>
/// <returns>Props configured to create a FleetStatusBroadcaster.</returns>
public static Props Props(Action<object>? broadcast = null) =>
Akka.Actor.Props.Create(() => new FleetStatusBroadcaster(broadcast));
/// <summary>Initializes a new instance of the FleetStatusBroadcaster actor.</summary>
/// <param name="broadcast">Optional custom broadcast delegate; if null, uses distributed pub-sub mediator.</param>
public FleetStatusBroadcaster(Action<object>? broadcast = null)
{
_cluster = Akka.Cluster.Cluster.Get(Context.System);
_broadcast = broadcast ?? (msg =>
DistributedPubSub.Get(Context.System).Mediator.Tell(new Publish(Topic, msg)));
Receive<ClusterEvent.MemberUp>(e => OnMemberUp(e.Member));
Receive<ClusterEvent.MemberRemoved>(e => OnMemberRemoved(e.Member));
Receive<ClusterEvent.UnreachableMember>(e => OnUnreachable(e.Member));
Receive<ClusterEvent.ReachableMember>(e => OnReachable(e.Member));
Receive<ClusterEvent.LeaderChanged>(_ => PublishSnapshot());
Receive<ClusterEvent.RoleLeaderChanged>(_ => PublishSnapshot());
Receive<DriverHostStatusHeartbeat>(OnHeartbeat);
Receive<Tick>(_ => OnTick());
}
/// <inheritdoc />
protected override void PreStart()
{
_cluster.Subscribe(
Self,
ClusterEvent.InitialStateAsEvents,
typeof(ClusterEvent.IMemberEvent),
typeof(ClusterEvent.LeaderChanged),
typeof(ClusterEvent.RoleLeaderChanged),
typeof(ClusterEvent.ReachabilityEvent));
Timers.StartPeriodicTimer("tick", Tick.Instance, BroadcastInterval);
}
/// <inheritdoc />
protected override void PostStop() => _cluster.Unsubscribe(Self);
private void OnMemberUp(Member m)
{
if (!TryNode(m, out var nodeId)) return;
_nodes[nodeId] = _nodes.TryGetValue(nodeId, out var prev)
? prev with { Health = FleetNodeHealth.Healthy, LastSeenUtc = DateTime.UtcNow }
: new NodeRecord(nodeId, FleetNodeHealth.Healthy, CurrentRevision: null, DateTime.UtcNow);
PublishSnapshot();
}
private void OnMemberRemoved(Member m)
{
if (!TryNode(m, out var nodeId)) return;
_nodes.Remove(nodeId);
PublishSnapshot();
}
private void OnUnreachable(Member m)
{
if (!TryNode(m, out var nodeId) || !_nodes.TryGetValue(nodeId, out var rec)) return;
_nodes[nodeId] = rec with { Health = FleetNodeHealth.Unreachable };
PublishSnapshot();
}
private void OnReachable(Member m)
{
if (!TryNode(m, out var nodeId) || !_nodes.TryGetValue(nodeId, out var rec)) return;
_nodes[nodeId] = rec with { Health = FleetNodeHealth.Healthy, LastSeenUtc = DateTime.UtcNow };
PublishSnapshot();
}
private void OnHeartbeat(DriverHostStatusHeartbeat hb)
{
var rec = _nodes.TryGetValue(hb.NodeId, out var prev)
? prev with
{
CurrentRevision = hb.CurrentRevision,
LastSeenUtc = DateTime.UtcNow,
Health = prev.Health == FleetNodeHealth.Unreachable ? prev.Health : FleetNodeHealth.Healthy,
}
: new NodeRecord(hb.NodeId, FleetNodeHealth.Healthy, hb.CurrentRevision, DateTime.UtcNow);
_nodes[hb.NodeId] = rec;
}
/// <summary>
/// Periodic tick — flips heartbeat-stale nodes to <see cref="FleetNodeHealth.Degraded"/> and
/// rebroadcasts the snapshot regardless so SignalR subscribers always have a recent baseline.
/// </summary>
private void OnTick()
{
var stale = DateTime.UtcNow - HeartbeatTimeout;
var flippedCount = 0;
foreach (var kv in _nodes.ToList())
{
if (kv.Value.LastSeenUtc < stale && kv.Value.Health == FleetNodeHealth.Healthy)
{
_nodes[kv.Key] = kv.Value with { Health = FleetNodeHealth.Degraded };
flippedCount++;
}
}
if (flippedCount > 0) _log.Debug("FleetStatusBroadcaster flipped {Count} nodes to Degraded", flippedCount);
PublishSnapshot();
}
private void PublishSnapshot()
{
var statuses = _nodes.Values
.Select(r => new FleetNodeStatus(r.NodeId, r.Health, r.CurrentRevision, r.LastSeenUtc))
.ToList();
var msg = new FleetStatusChanged(statuses, CurrentDeployment: null, DateTime.UtcNow, CorrelationId.NewId());
_broadcast(msg);
}
private static bool TryNode(Member m, out NodeId nodeId)
{
nodeId = default;
var host = m.Address.Host;
if (string.IsNullOrWhiteSpace(host)) return false;
nodeId = NodeId.Parse(host);
return true;
}
private sealed record NodeRecord(NodeId NodeId, FleetNodeHealth Health, RevisionHash? CurrentRevision, DateTime LastSeenUtc);
public sealed class Tick
{
public static readonly Tick Instance = new();
private Tick() { }
}
}
src/Server/ZB.MOM.WW.OtOpcUa.ControlPlane/ServiceCollectionExtensions.cs
+6 -8
View File
@@ -26,12 +26,11 @@ public static class ServiceCollectionExtensions
public const string ConfigPublishSingletonName = "config-publish";
public const string AdminOperationsSingletonName = "admin-operations";
public const string AuditWriterSingletonName = "audit-writer";
public const string FleetStatusSingletonName = "fleet-status";
public const string RedundancyStateSingletonName = "redundancy-state";
public const string ClusterNodeAddressReconcilerSingletonName = "cluster-node-address-reconciler";
/// <summary>
/// Registers the admin-role cluster singletons on the AkkaConfigurationBuilder — five with
/// Registers the admin-role cluster singletons on the AkkaConfigurationBuilder — three with
/// registry proxies, plus the address reconciler, which is proxy-less because nothing addresses it.
/// Must be called against the same builder used by <c>AkkaHostedService</c> so the singletons
/// share the host's ActorSystem.
@@ -51,7 +50,6 @@ public static class ServiceCollectionExtensions
public static AkkaConfigurationBuilder WithOtOpcUaControlPlaneSingletons(this AkkaConfigurationBuilder builder)
{
var singletonOptions = new ClusterSingletonOptions { Role = AdminRole };
var proxyOptions = new ClusterSingletonOptions { Role = AdminRole };
// Per-cluster mesh Phase 2: the central end of the central↔node command boundary.
//
@@ -139,10 +137,11 @@ public static class ServiceCollectionExtensions
},
singletonOptions);
builder.WithSingleton<FleetStatusBroadcasterKey>(
FleetStatusSingletonName,
(system, registry, resolver) => FleetStatusBroadcaster.Props(),
singletonOptions);
// Gitea #542 (2026-08-06): the fleet-status singleton is NO LONGER registered here — the
// whole FleetStatusBroadcaster → DPS → SignalR bridge → FleetStatusHub chain was DELETED.
// It had zero clients end-to-end (no browser JS, no HubConnection — /fleet reads
// NodeDeploymentState from the ConfigDb on a poll), its DriverHostStatusHeartbeat input had
// no production sender, and after mesh Phase 6 its "fleet" was only central's own pair.
// Per-cluster mesh Phase 6: the redundancy-state singleton is NO LONGER registered here.
// It was an admin-scoped, fleet-wide singleton — one Primary elected across the whole mesh.
@@ -253,7 +252,6 @@ public static class ServiceCollectionExtensions
public sealed class ConfigPublishCoordinatorKey { }
public sealed class AdminOperationsActorKey { }
public sealed class AuditWriterActorKey { }
public sealed class FleetStatusBroadcasterKey { }
public sealed class RedundancyStateActorKey { }
public sealed class ClusterNodeAddressReconcilerKey { }
public sealed class CentralCommunicationActorKey { }