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

fix(communication): delete dead SiteAddressCacheLoaded.SiteContacts + legacy NodeA/B parse (review 2026-10-21 03/D-1)

Browse Source
This commit is contained in:
Joseph Doherty
2026-10-22 06:13:25 -04:00
parent 004f3bd155
commit 0fd9bcbce6
1 changed files with 8 additions and 46 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/ZB.MOM.WW.ScadaBridge.Communication/Actors/CentralCommunicationActor.cs
+8 -46
View File
@@ -370,51 +370,22 @@ public class CentralCommunicationActor : ReceiveActor
var repo = scope.ServiceProvider.GetRequiredService<ISiteRepository>();
var sites = await repo.GetAllSitesAsync(ct).ConfigureAwait(false);
var contacts = new Dictionary<string, List<string>>();
// Parallel gRPC-endpoint cache fed by the SAME DB read (the streaming path's
// GrpcNodeA/BAddress columns, NOT the Akka NodeA/BAddress ones). No second poll loop —
// the gRPC transport rides this one, and the Akka transport simply ignores the field.
// gRPC-endpoint cache (the GrpcNodeA/BAddress columns, NOT the legacy Akka
// NodeA/BAddress ones, which no longer drive cross-cluster dialling).
var grpcContacts = new Dictionary<string, SiteGrpcEndpoints>();
foreach (var site in sites)
{
var addrs = new List<string>();
if (!string.IsNullOrWhiteSpace(site.NodeAAddress))
{
var addr = site.NodeAAddress;
// Strip actor path suffix if present (legacy format)
var idx = addr.IndexOf("/user/");
if (idx > 0) addr = addr.Substring(0, idx);
addrs.Add(addr);
}
if (!string.IsNullOrWhiteSpace(site.NodeBAddress))
{
var addr = site.NodeBAddress;
var idx = addr.IndexOf("/user/");
if (idx > 0) addr = addr.Substring(0, idx);
addrs.Add(addr);
}
if (addrs.Count > 0)
contacts[site.SiteIdentifier] = addrs;
var grpcA = string.IsNullOrWhiteSpace(site.GrpcNodeAAddress) ? null : site.GrpcNodeAAddress;
var grpcB = string.IsNullOrWhiteSpace(site.GrpcNodeBAddress) ? null : site.GrpcNodeBAddress;
if (grpcA is not null || grpcB is not null)
grpcContacts[site.SiteIdentifier] = new SiteGrpcEndpoints(grpcA, grpcB);
}
// Freeze the cross-task payload before piping to
// Self. The message record exposes read-only types (
// IReadOnlyDictionary / IReadOnlyList) so the Akka.NET message-
// immutability convention is enforced by type, not just convention.
var frozen = contacts.ToDictionary(
kvp => kvp.Key,
kvp => (IReadOnlyList<string>)kvp.Value.AsReadOnly());
// Carry the full configured-site identifier set (not just the
// address-bearing subset in `frozen`) so the aggregator prunes only
// gRPC-address-bearing subset) so the aggregator prunes only
// genuinely-deleted sites and never an addressless-but-configured one.
var knownSiteIds = sites.Select(s => s.SiteIdentifier).ToList();
return new SiteAddressCacheLoaded(frozen, knownSiteIds, grpcContacts);
return new SiteAddressCacheLoaded(knownSiteIds, grpcContacts);
}).PipeTo(self);
}
@@ -502,22 +473,13 @@ public record RefreshSiteAddresses;
/// <summary>
/// Message carrying the loaded site contact data from the database. Per-transport resource
/// reconciliation happens on the actor thread in HandleSiteAddressCacheLoaded, which hands this
/// straight to <see cref="ISiteCommandTransport.ReconcileSites"/>.
///
/// The payload is exposed as <see cref="IReadOnlyDictionary{TKey,TValue}"/>
/// of <see cref="IReadOnlyList{T}"/> so the Akka.NET "messages are immutable"
/// convention is enforced at the type level rather than relying on producer
/// discipline. The producer wraps the constructed buckets with
/// <c>List&lt;T&gt;.AsReadOnly()</c> before piping to Self.
/// straight to <see cref="ISiteCommandTransport.ReconcileSites"/>. In-process only (never on the
/// wire); the payload is exposed through read-only collection types so the Akka.NET "messages are
/// immutable" convention is enforced at the type level.
/// </summary>
/// <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,
/// consumed by the gRPC transport and ignored by the Akka one.
/// </param>
/// <param name="GrpcContacts">gRPC endpoint pairs per site (from GrpcNodeA/GrpcNodeBAddress).</param>
public sealed record SiteAddressCacheLoaded(
IReadOnlyDictionary<string, IReadOnlyList<string>> SiteContacts,
IReadOnlyCollection<string> KnownSiteIds,
IReadOnlyDictionary<string, SiteGrpcEndpoints> GrpcContacts);