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

refactor(comm): retire ClusterClient naming after the gRPC cutover

Browse Source 
Phase 4 of the ClusterClient -> gRPC migration deleted the Akka transports
but left the naming behind: `ClusterClientSiteAuditClient` was transport-
agnostic and worked unchanged, so it survived the deletion under a name
that now describes a transport the repo no longer has. Same for a scatter
of doc-comments still framing gRPC as "the new transport" beside an Akka
one that is gone.

Renames it to `SiteCommunicationAuditClient` (and its test file) and
rewrites the stale comments to describe the single transport that exists.
Also tightens CLAUDE.md: drops the self-describing directory listing and
the 27-component enumeration in favour of the non-obvious parts only.

Behaviour-neutral: names and prose only. Recorded as the Phase 4
follow-up in docs/plans/2026-07-22-clusterclient-to-grpc-plan.md.
This commit is contained in:
Joseph Doherty
2026-07-27 15:40:01 -04:00
parent b3dc17a5fe
commit 63c16d6912
57 changed files with 193 additions and 244 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/ZB.MOM.WW.ScadaBridge.Communication/Actors/CentralCommunicationActor.cs
+19 -19
View File
@@ -89,7 +89,7 @@ public class CentralCommunicationActor : ReceiveActor
/// Default Ask timeout for routing audit ingest commands to the
/// Effective Ask timeout for audit ingest routing. Defaults to
/// <see cref="Grpc.SiteStreamGrpcServer.AuditIngestAskTimeout"/> (30 s) — the two
/// audit-ingest transports (gRPC vs ClusterClient) now share one source of truth
/// audit-ingest entry points (the site stream server and the control plane) share one source of truth
/// for the timeout. Overridable via the constructor so tests can exercise the
/// timeout/fault path without waiting 30 s. When the window is exceeded the Ask
/// faults and that fault is piped back to the caller as a
@@ -167,8 +167,8 @@ public class CentralCommunicationActor : ReceiveActor
});
// Notification Outbox ingest: a site forwards a buffered NotificationSubmit to the
// central cluster via ClusterClient. Forward to the outbox proxy so the original
// Sender (the site's ClusterClient path) is preserved and the NotificationSubmitAck
// central cluster. Forward to the outbox proxy so the original
// Sender (the site's transport path) is preserved and the NotificationSubmitAck
// routes straight back to the site.
Receive<NotificationSubmit>(HandleNotificationSubmit);
@@ -186,9 +186,9 @@ public class CentralCommunicationActor : ReceiveActor
});
// Audit Log site→central ingest: a site forwards a batch of audit
// events to the central cluster via ClusterClient. Ask the ingest proxy
// events to the central cluster. Ask the ingest proxy
// and pipe the IngestAuditEventsReply back to the original Sender (the
// site's ClusterClient path) so the site can flip its rows to Forwarded.
// site's transport path) so the site can flip its rows to Forwarded.
Receive<IngestAuditEventsCommand>(HandleIngestAuditEvents);
// Audit Log combined-telemetry ingest: routes to the same proxy
@@ -196,9 +196,9 @@ public class CentralCommunicationActor : ReceiveActor
Receive<IngestCachedTelemetryCommand>(HandleIngestCachedTelemetry);
// Startup reconciliation: a site node forwards its local deployed inventory on
// startup via ClusterClient. Resolve the scoped ReconcileService, diff the
// startup. Resolve the scoped ReconcileService, diff the
// inventory against central's expected set, and pipe the ReconcileSiteResponse
// (gap fetch tokens + orphans) straight back to the site node's ClusterClient.
// (gap fetch tokens + orphans) straight back to the site node.
Receive<ReconcileSiteRequest>(HandleReconcileSiteRequest);
}
@@ -254,7 +254,7 @@ public class CentralCommunicationActor : ReceiveActor
}
// Capture Sender before the async/PipeTo — Akka resets Sender between
// dispatches. The reply is piped straight back to the site's ClusterClient.
// dispatches. The reply is piped straight back to the calling site node.
// On an Ask timeout or a faulted reply, PipeTo delivers a Status.Failure to
// replyTo: the fault propagates to the caller rather than being swallowed.
// The site's own Ask through this path then faults, and the site drain loop
@@ -285,10 +285,10 @@ public class CentralCommunicationActor : ReceiveActor
}
/// <summary>
/// Startup reconciliation (site→central over ClusterClient): resolve the scoped
/// Startup reconciliation (site→central): resolve the scoped
/// <see cref="ReconcileService"/> in a DI scope, diff the node's reported inventory
/// against central's expected set, and pipe the <see cref="ReconcileSiteResponse"/>
/// back to the site node's ClusterClient path. The actor stays thin — all the diff
/// back to the site node's transport path. The actor stays thin — all the diff
/// and staging logic lives in the service. Mirrors the DB-access pattern used by
/// <see cref="LoadSiteAddressesFromDb"/> (Task.Run + CreateScope + PipeTo) and the
/// Sender-preservation pattern of <see cref="HandleIngestAuditEvents"/>.
@@ -331,7 +331,7 @@ public class CentralCommunicationActor : ReceiveActor
MarkHeartbeatLocally(heartbeat);
// Fan the heartbeat out to the peer central node so BOTH aggregators mark
// it, regardless of which central node the site's ClusterClient delivered
// it, regardless of which central node the site delivered
// to. Without this, a heartbeat that only ever reaches one node leaves the
// other node's aggregator blind to that site's liveness after a failover
// (arch review 02, Low). MarkHeartbeat is idempotent (timestamp overwrite),
@@ -358,7 +358,7 @@ public class CentralCommunicationActor : ReceiveActor
}
/// <summary>
/// Handles a report delivered directly from a site (via ClusterClient):
/// Handles a report delivered directly from a site:
/// process locally, then fan out to the peer central node so its
/// aggregator stays in sync.
/// </summary>
@@ -407,7 +407,7 @@ public class CentralCommunicationActor : ReceiveActor
// HandleConnectionStateChanged removed — no production
// caller emitted ConnectionStateChanged, so the workflow ran only in tests.
// Disconnect detection is owned by the transport layers (gRPC keepalive +
// ClusterClient/Ask timeout).
// Ask timeout).
private void HandleSiteEnvelope(SiteEnvelope envelope)
{
@@ -491,8 +491,8 @@ public class CentralCommunicationActor : ReceiveActor
private void HandleSiteAddressCacheLoaded(SiteAddressCacheLoaded msg)
{
// Per-transport per-site resource reconciliation (create/stop ClusterClients, or
// build/drop gRPC channel pairs). Runs on the actor thread each refresh tick.
// Per-site transport resource reconciliation (build/drop gRPC channel
// pairs). Runs on the actor thread each refresh tick.
_transport.ReconcileSites(msg);
// Self-healing eviction: a site deleted from configuration would otherwise
@@ -525,7 +525,7 @@ public class CentralCommunicationActor : ReceiveActor
// Subscribe to the peer-replication topic so we receive health reports
// delivered to the other central node and keep our local aggregator
// in sync (ClusterClient load-balances reports across nodes).
// in sync (a site may deliver its report to either central node).
// Tolerant of non-clustered hosts (TestKit) where the extension is absent.
try
{
@@ -581,7 +581,7 @@ public record RefreshSiteAddresses;
/// discipline. The producer wraps the constructed buckets with
/// <c>List&lt;T&gt;.AsReadOnly()</c> before piping to Self.
/// </summary>
/// <param name="SiteContacts">Akka ClusterClient contact addresses per site (from NodeA/NodeBAddress).</param>
/// <param name="SiteContacts">Akka remote addresses per site (from NodeA/NodeBAddress).</param>
/// <param name="KnownSiteIds">Every configured site id, address-bearing or not, for aggregator pruning.</param>
/// <param name="GrpcContacts">
/// gRPC endpoint pairs per site (from GrpcNodeA/GrpcNodeBAddress) — the streaming path's columns,
@@ -603,8 +603,8 @@ public readonly record struct SiteGrpcEndpoints(string? NodeA, string? NodeB);
/// <summary>
/// Peer-replication envelope for a site heartbeat, fanned out over the same
/// DistributedPubSub topic as <see cref="SiteHealthReportReplica"/> so both
/// central aggregators mark heartbeats regardless of which node the site's
/// ClusterClient delivered to. Only ever travels central↔central (same assembly
/// central aggregators mark heartbeats regardless of which node the site
/// delivered to. Only ever travels central↔central (same assembly
/// version, rolled together), so the default reflective serializer is safe.
/// </summary>
public sealed record SiteHeartbeatReplica(HeartbeatMessage Heartbeat);