refactor(fleet): #542 delete the dead fleet-status chain (broadcaster, SignalR bridge, hub, messages)

#542 filed FleetStatusBroadcaster as blind to site nodes since the Phase 6 per-cluster mesh
split: it derives "the fleet" from central's own Akka gossip ring, which after the split is
just central's own pair. Scoping the cross-mesh rewrite found the deeper truth — the whole
chain is dead end to end, so this deletes it rather than reviving it.

  - Zero consumers. FleetStatusBroadcaster published FleetStatusChanged on the DPS
    "fleet-status" topic; FleetStatusSignalRBridge forwarded it to FleetStatusHub
    ("/hubs/fleet-status"); nothing anywhere connected. The AdminUI bans self-connecting
    HubConnections, no browser JS subscribes, and /fleet (Fleet.razor) reads
    NodeDeploymentState from the ConfigDb plus IClusterRoleInfo on a 5 s poll instead.
  - Its second input never existed. DriverHostStatusHeartbeat had no production sender —
    only the broadcaster's own test told it — so CurrentRevision was always null and the
    30 s staleness -> Degraded flip could never fire.
  - Post-Phase-6 DPS is pair-local, so even the central-only view it did produce was an
    accident of topology rather than a design.

The fleet visibility that actually works is unchanged and already cross-mesh-correct:
/fleet over NodeDeploymentState, and the four Phase 5 per-node telemetry dials.

Precedent for deleting inert machinery rather than activating it: #522 (driver stability
tiers), #507 (discovered-node injection), #525 (GetMemoryFootprint).

Deleted: FleetStatusBroadcaster (+ its nested DriverHostStatusHeartbeat), its admin
singleton registration and FleetStatusBroadcasterKey, FleetStatusSignalRBridge and its
key, FleetStatusHub and its route mapping, and the FleetStatusChanged / FleetNodeStatus /
FleetNodeHealth messages. Both Fleet/ directories survive — ControlPlane keeps
ClusterNodeAddressReconciler, Commons keeps GetDiagnostics. Docs swept where they claimed
a fleet-status channel exists (Telemetry.md's Phase 5 non-migrated channels,
Configuration.md, CLAUDE.md); completed-program plan docs under docs/plans and docs/v2 are
left as the historical record they are.

Build 0 errors, 868 warnings — byte-identical to master's rebuild baseline.
ControlPlane.Tests 231/231, AdminUI.Tests 1109/1109.

Claude-Session: https://claude.ai/code/session_014WNM4vjoVksyyBraTXSZE1
This commit is contained in:
Joseph Doherty
2026-08-06 17:21:30 -04:00
parent 97854a074e
commit f1c188114b
15 changed files with 38 additions and 357 deletions
+6 -4
View File
@@ -509,11 +509,13 @@ touching it:
- **Deferred, with reasons (do not read "seven observability topics" from the program sketch as the - **Deferred, with reasons (do not read "seven observability topics" from the program sketch as the
delivered scope):** `redundancy-state` stays on DPS in every mode (pair-local, built from delivered scope):** `redundancy-state` stays on DPS in every mode (pair-local, built from
`Cluster.State`, drives ServiceLevel + the Primary gate — genuinely mesh-bound by design, not an `Cluster.State`, drives ServiceLevel + the Primary gate — genuinely mesh-bound by design, not an
observability panel); `fleet-status` is central-internal (`FleetStatusBroadcaster` builds it from observability panel); `deployment-acks` already rides the Phase 2 gRPC command transport
the admin node's own membership events, `Fleet.razor` polls the Config DB and ignores the feed —
never a node→central stream); `deployment-acks` already rides the Phase 2 gRPC command transport
(`MeshTransport:Mode=Grpc`, `CentralCommandService.ReportApply`) as a command-plane reply, not (`MeshTransport:Mode=Grpc`, `CentralCommandService.ReportApply`) as a command-plane reply, not
telemetry. telemetry. A third, `fleet-status`, was **DELETED** rather than deferred (2026-08-06, Gitea #542):
`FleetStatusBroadcaster` → DPS → SignalR bridge → `FleetStatusHub` had **zero clients end to end**
(no browser JS, no `HubConnection` — `/fleet` polls `NodeDeploymentState` in the Config DB), its
`DriverHostStatusHeartbeat` input had no production sender, and Phase 6 left its "fleet" as just
central's own pair. Fleet visibility is `/fleet` + the four per-node telemetry dials above.
- **The node ALWAYS hosts and ALWAYS double-publishes, in both modes.** Every publish seam emits - **The node ALWAYS hosts and ALWAYS double-publishes, in both modes.** Every publish seam emits
into a node-local `ITelemetryLocalHub` **and** still publishes DPS, unconditionally; the gRPC into a node-local `ITelemetryLocalHub` **and** still publishes DPS, unconditionally; the gRPC
server binds whenever `Telemetry:GrpcListenPort > 0`, independent of `Telemetry:Mode`. Only server binds whenever `Telemetry:GrpcListenPort > 0`, independent of `Telemetry:Mode`. Only
+5 -4
View File
@@ -273,11 +273,12 @@ discipline as `MeshTransport:Mode` and `ConfigSource:Mode`.
otherwise) — enforced by `TelemetryOptionsValidator`. **`Grpc` mode on either side requires a otherwise) — enforced by `TelemetryOptionsValidator`. **`Grpc` mode on either side requires a
non-empty `ApiKey`** — fail-closed, same as `ConfigServe`/`LocalDb:Replication`. non-empty `ApiKey`** — fail-closed, same as `ConfigServe`/`LocalDb:Replication`.
**Three DPS observability topics stay out of scope for this transport, on purpose:** **Two DPS observability topics stay out of scope for this transport, on purpose:**
`redundancy-state` (pair-local, built from `Cluster.State`, stays on DPS in both modes — see `redundancy-state` (pair-local, built from `Cluster.State`, stays on DPS in both modes — see
[`Redundancy.md`](Redundancy.md#command-transport-centralnode)), `fleet-status` (central-internal; [`Redundancy.md`](Redundancy.md#command-transport-centralnode)) and `deployment-acks` (already rides
`Fleet.razor` polls the Config DB, not any stream), and `deployment-acks` (already rides the Phase 2 the Phase 2 gRPC command transport, `CentralCommandService.ReportApply`, as a command-plane reply).
gRPC command transport, `CentralCommandService.ReportApply`, as a command-plane reply). See A third, `fleet-status`, was **deleted** rather than deferred (2026-08-06, Gitea #542 — it had no
client end to end; `Fleet.razor` polls the Config DB). See
[`docs/Telemetry.md`](Telemetry.md) for the full architecture, the reconnect/generation-stamped [`docs/Telemetry.md`](Telemetry.md) for the full architecture, the reconnect/generation-stamped
dialer story, and the rationale for the node-hosts/central-dials inversion. dialer story, and the rationale for the node-hosts/central-dials inversion.
+13 -8
View File
@@ -78,7 +78,6 @@ What actually changes per mode, precisely:
`IDriverStatusSnapshotStore`, `IDriverResilienceStatusStore`). `IDriverStatusSnapshotStore`, `IDriverResilienceStatusStore`).
- **The AdminUI panels themselves are untouched.** `/alerts`, `/script-log`, and the `/hosts` - **The AdminUI panels themselves are untouched.** `/alerts`, `/script-log`, and the `/hosts`
driver table all read from the same sinks in both modes — only the sinks' upstream feed swaps. driver table all read from the same sinks in both modes — only the sinks' upstream feed swaps.
The `fleet-status` bridge (out of Phase 5 scope — see below) stays on DPS in both modes.
Because hosting is unconditional and both ingest paths are always compiled in, a fleet can be Because hosting is unconditional and both ingest paths are always compiled in, a fleet can be
flipped node-by-node and central-by-central with no coordination window where telemetry is lost — flipped node-by-node and central-by-central with no coordination window where telemetry is lost —
@@ -130,10 +129,10 @@ Three things to know before touching it:
and no reader — the only reference in the repo is its DbSet declaration — while the live data reaches and no reader — the only reference in the repo is its DbSet declaration — while the live data reaches
the AdminUI over the `driver-resilience-status` channel above. Keep-or-delete is Gitea **#524**. the AdminUI over the `driver-resilience-status` channel above. Keep-or-delete is Gitea **#524**.
## The three deferred channels (NOT migrated in Phase 5 — do not read this as "seven done") ## The three non-migrated channels (NOT migrated in Phase 5 — do not read this as "seven done")
The program sketch originally named seven observability topics for Phase 5. Three were scoped out, The program sketch originally named seven observability topics for Phase 5. Three were scoped out,
each for a distinct, settled reason: each for a distinct, settled reason — and one of those three has since been deleted outright:
- **`redundancy-state`** — bidirectional, built directly from `Cluster.State`, and **pair-local**: - **`redundancy-state`** — bidirectional, built directly from `Cluster.State`, and **pair-local**:
it drives ServiceLevel and the Primary data-plane gate, consumed in-process by it drives ServiceLevel and the Primary data-plane gate, consumed in-process by
@@ -143,11 +142,17 @@ each for a distinct, settled reason:
in-mesh. Central's *display* of each pair's redundancy state is a Phase 6 cross-mesh concern in-mesh. Central's *display* of each pair's redundancy state is a Phase 6 cross-mesh concern
(a future observability channel, once central no longer shares gossip with any site pair), not a (a future observability channel, once central no longer shares gossip with any site pair), not a
Phase 5 telemetry-panel migration. Phase 5 telemetry-panel migration.
- **`fleet-status`** — **central-internal**, not a node→central stream at all. The admin singleton - **`fleet-status`** — **DELETED, 2026-08-06, Gitea #542.** There is no `fleet-status` channel any
`FleetStatusBroadcaster` builds it from the admin node's own cluster membership/reachability/ more, on DPS or anywhere else. It was central-internal rather than a node→central stream: the
leader events, and `Fleet.razor` **polls the Config DB** and ignores the feed entirely. There is admin singleton `FleetStatusBroadcaster` built it from the admin node's own cluster membership/
nothing here for a per-node gRPC stream to carry. Revisit in Phase 6, once central genuinely reachability/leader events and published to a SignalR hub that **no client anywhere ever
loses gossip visibility of site nodes and needs another way to know a pair's membership. connected to** — `Fleet.razor` polls `NodeDeploymentState` in the Config DB and always ignored
the feed. Its second input, `DriverHostStatusHeartbeat`, had no production sender either. Phase 6
then made the central-only view wrong as well (central's gossip ring is its own pair). The whole
chain — broadcaster, singleton registration, DPS topic, SignalR bridge, `FleetStatusHub`, and the
`FleetStatusChanged`/`FleetNodeStatus` messages — was removed rather than rewritten cross-mesh.
Fleet visibility that does work today: `/fleet` over `NodeDeploymentState`, and the four Phase 5
telemetry channels above, which are per-node dials and therefore already mesh-independent.
- **`deployment-acks`** — already migrated, but onto a different transport: it rides the Phase 2 - **`deployment-acks`** — already migrated, but onto a different transport: it rides the Phase 2
gRPC command transport (`MeshTransport:Mode=Grpc`, `CentralCommandService.ReportApply`) as a gRPC command transport (`MeshTransport:Mode=Grpc`, `CentralCommandService.ReportApply`) as a
command-plane reply, not an observability broadcast. It was never a Phase 5 candidate. command-plane reply, not an observability broadcast. It was never a Phase 5 candidate.
@@ -13,7 +13,7 @@ public enum StartDeploymentOutcome
/// <summary> /// <summary>
/// Reply from the <c>AdminOperationsActor</c> singleton. <c>Accepted</c> means the snapshot /// Reply from the <c>AdminOperationsActor</c> singleton. <c>Accepted</c> means the snapshot
/// was sealed and a <c>Deployment</c> row was created; the in-flight deployment can be /// was sealed and a <c>Deployment</c> row was created; the in-flight deployment can be
/// tracked through fleet-status broadcasts. /// tracked through the per-node <c>NodeDeploymentState</c> rows the AdminUI's /fleet page reads.
/// </summary> /// </summary>
public sealed record StartDeploymentResult( public sealed record StartDeploymentResult(
StartDeploymentOutcome Outcome, StartDeploymentOutcome Outcome,
@@ -1,21 +0,0 @@
using ZB.MOM.WW.OtOpcUa.Commons.Types;
namespace ZB.MOM.WW.OtOpcUa.Commons.Messages.Fleet;
public enum FleetNodeHealth { Healthy, Degraded, Unreachable }
public sealed record FleetNodeStatus(
NodeId NodeId,
FleetNodeHealth Health,
RevisionHash? CurrentRevision,
DateTime LastSeenUtc);
/// <summary>
/// Periodic fleet-wide status broadcast pushed by <c>FleetStatusBroadcaster</c> to admin UI
/// subscribers via SignalR.
/// </summary>
public sealed record FleetStatusChanged(
IReadOnlyList<FleetNodeStatus> Nodes,
DeploymentId? CurrentDeployment,
DateTime AsOfUtc,
CorrelationId CorrelationId);
@@ -9,8 +9,7 @@ namespace ZB.MOM.WW.OtOpcUa.AdminUI.Hubs;
/// <summary> /// <summary>
/// Akka actor that subscribes to the <c>alerts</c> DistributedPubSub topic and forwards each /// 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"/>. /// <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 /// One bridge per admin node — hub fan-out is per-node, so no cluster-singleton is needed.
/// per-node, no cluster-singleton needed.
/// </summary> /// </summary>
public sealed class AlertSignalRBridge : ReceiveActor public sealed class AlertSignalRBridge : ReceiveActor
{ {
@@ -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";
}
@@ -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);
}
}
}
@@ -13,7 +13,6 @@ public static class HubRouteBuilderExtensions
/// <returns>The same endpoint route builder, for chaining.</returns> /// <returns>The same endpoint route builder, for chaining.</returns>
public static IEndpointRouteBuilder MapOtOpcUaHubs(this IEndpointRouteBuilder app) public static IEndpointRouteBuilder MapOtOpcUaHubs(this IEndpointRouteBuilder app)
{ {
app.MapHub<FleetStatusHub>(FleetStatusHub.Endpoint);
app.MapHub<AlertHub>(AlertHub.Endpoint); app.MapHub<AlertHub>(AlertHub.Endpoint);
app.MapHub<ScriptLogHub>(ScriptLogHub.Endpoint); app.MapHub<ScriptLogHub>(ScriptLogHub.Endpoint);
app.MapHub<DriverStatusHub>(DriverStatusHub.Endpoint); app.MapHub<DriverStatusHub>(DriverStatusHub.Endpoint);
@@ -14,7 +14,6 @@ namespace ZB.MOM.WW.OtOpcUa.AdminUI.Hubs;
public static class HubServiceCollectionExtensions public static class HubServiceCollectionExtensions
{ {
public const string FleetStatusSignalRBridgeName = "fleet-status-signalr-bridge";
public const string AlertSignalRBridgeName = "alert-signalr-bridge"; public const string AlertSignalRBridgeName = "alert-signalr-bridge";
public const string ScriptLogSignalRBridgeName = "script-log-signalr-bridge"; public const string ScriptLogSignalRBridgeName = "script-log-signalr-bridge";
public const string DriverStatusSignalRBridgeName = "driver-status-signalr-bridge"; public const string DriverStatusSignalRBridgeName = "driver-status-signalr-bridge";
@@ -44,8 +43,8 @@ public static class HubServiceCollectionExtensions
/// <summary> /// <summary>
/// Spawns the SignalR bridge actors that forward DPS messages to browser-facing SignalR /// Spawns the SignalR bridge actors that forward DPS messages to browser-facing SignalR
/// hubs: <c>fleet-status</c> → <see cref="FleetStatusHub"/>, <c>alerts</c> → /// hubs: <c>alerts</c> → <see cref="AlertHub"/>, <c>script-logs</c> →
/// <see cref="AlertHub"/>, <c>script-logs</c> → <see cref="ScriptLogHub"/>. /// <see cref="ScriptLogHub"/>, <c>driver-health</c> → <see cref="DriverStatusHub"/>.
/// ///
/// Call inside the admin-role configurator on the shared <see cref="AkkaConfigurationBuilder"/>: /// Call inside the admin-role configurator on the shared <see cref="AkkaConfigurationBuilder"/>:
/// <code> /// <code>
@@ -62,12 +61,6 @@ public static class HubServiceCollectionExtensions
{ {
builder.WithActors((system, registry, resolver) => 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 // 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 // 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. // 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> /// <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 AlertSignalRBridgeKey { }
public sealed class ScriptLogSignalRBridgeKey { } public sealed class ScriptLogSignalRBridgeKey { }
public sealed class DriverStatusSignalRBridgeKey { } public sealed class DriverStatusSignalRBridgeKey { }
@@ -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() { }
}
}
@@ -26,12 +26,11 @@ public static class ServiceCollectionExtensions
public const string ConfigPublishSingletonName = "config-publish"; public const string ConfigPublishSingletonName = "config-publish";
public const string AdminOperationsSingletonName = "admin-operations"; public const string AdminOperationsSingletonName = "admin-operations";
public const string AuditWriterSingletonName = "audit-writer"; public const string AuditWriterSingletonName = "audit-writer";
public const string FleetStatusSingletonName = "fleet-status";
public const string RedundancyStateSingletonName = "redundancy-state"; public const string RedundancyStateSingletonName = "redundancy-state";
public const string ClusterNodeAddressReconcilerSingletonName = "cluster-node-address-reconciler"; public const string ClusterNodeAddressReconcilerSingletonName = "cluster-node-address-reconciler";
/// <summary> /// <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. /// 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 /// Must be called against the same builder used by <c>AkkaHostedService</c> so the singletons
/// share the host's ActorSystem. /// share the host's ActorSystem.
@@ -139,10 +138,11 @@ public static class ServiceCollectionExtensions
}, },
singletonOptions); singletonOptions);
builder.WithSingleton<FleetStatusBroadcasterKey>( // Gitea #542 (2026-08-06): the fleet-status singleton is NO LONGER registered here — the
FleetStatusSingletonName, // whole FleetStatusBroadcaster → DPS → SignalR bridge → FleetStatusHub chain was DELETED.
(system, registry, resolver) => FleetStatusBroadcaster.Props(), // It had zero clients end-to-end (no browser JS, no HubConnection — /fleet reads
singletonOptions); // 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. // 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. // It was an admin-scoped, fleet-wide singleton — one Primary elected across the whole mesh.
@@ -253,7 +253,6 @@ public static class ServiceCollectionExtensions
public sealed class ConfigPublishCoordinatorKey { } public sealed class ConfigPublishCoordinatorKey { }
public sealed class AdminOperationsActorKey { } public sealed class AdminOperationsActorKey { }
public sealed class AuditWriterActorKey { } public sealed class AuditWriterActorKey { }
public sealed class FleetStatusBroadcasterKey { }
public sealed class RedundancyStateActorKey { } public sealed class RedundancyStateActorKey { }
public sealed class ClusterNodeAddressReconcilerKey { } public sealed class ClusterNodeAddressReconcilerKey { }
public sealed class CentralCommunicationActorKey { } public sealed class CentralCommunicationActorKey { }
@@ -16,12 +16,11 @@ namespace ZB.MOM.WW.OtOpcUa.AdminUI.Tests.Hubs;
/// Verifies the Phase 5 dark switch in <see cref="HubServiceCollectionExtensions.WithOtOpcUaSignalRBridges"/>: /// Verifies the Phase 5 dark switch in <see cref="HubServiceCollectionExtensions.WithOtOpcUaSignalRBridges"/>:
/// <c>TelemetryDial:Mode = Dps</c> (default) spawns the four telemetry DPS bridge actors and NOT the /// <c>TelemetryDial:Mode = Dps</c> (default) spawns the four telemetry DPS bridge actors and NOT the
/// gRPC dial supervisor; <c>Grpc</c> spawns the dial supervisor and NONE of the four telemetry /// gRPC dial supervisor; <c>Grpc</c> spawns the dial supervisor and NONE of the four telemetry
/// bridges. The fleet-status bridge (deferred / out of Phase 5 scope) is spawned in BOTH modes. /// bridges. The gate is proven on a real <see cref="ActorRegistry"/> from a started Akka host.
/// The gate is proven on a real <see cref="ActorRegistry"/> from a started Akka host.
/// </summary> /// </summary>
public sealed class TelemetryModeWiringTests public sealed class TelemetryModeWiringTests
{ {
/// <summary>Dps mode: the four telemetry bridges + fleet bridge are registered; the dial supervisor is not.</summary> /// <summary>Dps mode: the four telemetry bridges are registered; the dial supervisor is not.</summary>
[Fact] [Fact]
public async Task Dps_mode_spawns_the_four_telemetry_bridges_and_not_the_dial_supervisor() public async Task Dps_mode_spawns_the_four_telemetry_bridges_and_not_the_dial_supervisor()
{ {
@@ -31,7 +30,6 @@ public sealed class TelemetryModeWiringTests
{ {
var registry = host.Services.GetRequiredService<ActorRegistry>(); var registry = host.Services.GetRequiredService<ActorRegistry>();
registry.TryGet<FleetStatusSignalRBridgeKey>(out _).ShouldBeTrue();
registry.TryGet<AlertSignalRBridgeKey>(out _).ShouldBeTrue(); registry.TryGet<AlertSignalRBridgeKey>(out _).ShouldBeTrue();
registry.TryGet<ScriptLogSignalRBridgeKey>(out _).ShouldBeTrue(); registry.TryGet<ScriptLogSignalRBridgeKey>(out _).ShouldBeTrue();
registry.TryGet<DriverStatusSignalRBridgeKey>(out _).ShouldBeTrue(); registry.TryGet<DriverStatusSignalRBridgeKey>(out _).ShouldBeTrue();
@@ -45,7 +43,7 @@ public sealed class TelemetryModeWiringTests
} }
} }
/// <summary>Grpc mode: the dial supervisor + fleet bridge are registered; none of the four telemetry bridges are.</summary> /// <summary>Grpc mode: the dial supervisor is registered; none of the four telemetry bridges are.</summary>
[Fact] [Fact]
public async Task Grpc_mode_spawns_the_dial_supervisor_and_none_of_the_four_telemetry_bridges() public async Task Grpc_mode_spawns_the_dial_supervisor_and_none_of_the_four_telemetry_bridges()
{ {
@@ -55,7 +53,6 @@ public sealed class TelemetryModeWiringTests
{ {
var registry = host.Services.GetRequiredService<ActorRegistry>(); var registry = host.Services.GetRequiredService<ActorRegistry>();
registry.TryGet<FleetStatusSignalRBridgeKey>(out _).ShouldBeTrue();
registry.TryGet<TelemetryDialSupervisorKey>(out _).ShouldBeTrue(); registry.TryGet<TelemetryDialSupervisorKey>(out _).ShouldBeTrue();
registry.TryGet<AlertSignalRBridgeKey>(out _).ShouldBeFalse(); registry.TryGet<AlertSignalRBridgeKey>(out _).ShouldBeFalse();
@@ -1,47 +0,0 @@
using Akka.Actor;
using Shouldly;
using Xunit;
using ZB.MOM.WW.OtOpcUa.Commons.Messages.Fleet;
using ZB.MOM.WW.OtOpcUa.Commons.Types;
using ZB.MOM.WW.OtOpcUa.ControlPlane.Fleet;
using ZB.MOM.WW.OtOpcUa.ControlPlane.Tests.Harness;
namespace ZB.MOM.WW.OtOpcUa.ControlPlane.Tests;
public sealed class FleetStatusBroadcasterTests : ControlPlaneActorTestBase
{
/// <summary>Verifies that the self member's join appears in the initial fleet status snapshot.</summary>
[Fact]
public void Self_member_up_lands_in_snapshot()
{
var probe = CreateTestProbe();
Sys.ActorOf(FleetStatusBroadcaster.Props(broadcast: msg => probe.Ref.Tell(msg)));
// The cluster self-joined in the harness — InitialStateAsEvents replays MemberUp for self.
var snapshot = probe.ExpectMsg<FleetStatusChanged>(TimeSpan.FromSeconds(5));
snapshot.Nodes.Count.ShouldBeGreaterThanOrEqualTo(1);
snapshot.Nodes.Any(n => n.Health == FleetNodeHealth.Healthy).ShouldBeTrue();
}
/// <summary>Verifies that heartbeat messages update the node revision in the next broadcast.</summary>
[Fact]
public void Heartbeat_updates_revision_in_next_snapshot()
{
var probe = CreateTestProbe();
var actor = Sys.ActorOf(FleetStatusBroadcaster.Props(broadcast: msg => probe.Ref.Tell(msg)));
// Drain the initial MemberUp snapshot.
probe.ExpectMsg<FleetStatusChanged>(TimeSpan.FromSeconds(5));
// Send a heartbeat for a node we know about (the self node from the cluster).
var selfHost = Akka.Cluster.Cluster.Get(Sys).SelfAddress.Host!;
var rev = RevisionHash.Parse(new string('c', 64));
actor.Tell(new FleetStatusBroadcaster.DriverHostStatusHeartbeat(NodeId.Parse(selfHost), rev));
// Wait for next periodic broadcast and assert the revision propagated.
var next = probe.FishForMessage<FleetStatusChanged>(
isMessage: m => m.Nodes.Any(n => n.CurrentRevision == rev),
max: TimeSpan.FromSeconds(10));
next.Nodes.Single(n => n.NodeId.Value == selfHost).CurrentRevision.ShouldBe(rev);
}
}
@@ -69,7 +69,7 @@ public sealed class RedundancyStateSingletonRehomeTests
registry.TryGet<RedundancyStateActorKey>(out _).ShouldBeFalse( registry.TryGet<RedundancyStateActorKey>(out _).ShouldBeFalse(
"the redundancy singleton was re-homed off the admin node onto every driver node (Phase 6)"); "the redundancy singleton was re-homed off the admin node onto every driver node (Phase 6)");
registry.TryGet<FleetStatusBroadcasterKey>(out _).ShouldBeTrue( registry.TryGet<AuditWriterActorKey>(out _).ShouldBeTrue(
"the remaining admin singletons must still register"); "the remaining admin singletons must still register");
} }