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

feat(comm): Phase 4 — delete Akka ClusterClient site↔central transport, gRPC-only

Browse Source 
ClusterClient→gRPC migration Phase 4 (docs/plans/2026-07-22-clusterclient-to-grpc-plan.md).
Phases 2/3 proved both directions on gRPC; this removes the Akka transport underneath.

Deleted:
- AkkaCentralTransport, AkkaSiteTransport (+ their dedicated tests)
- ISiteClientFactory + DefaultSiteClientFactory; CentralCommunicationActor legacy
  ctor + SelectTransport (Host now builds GrpcSiteTransport and injects it)
- ClusterClient creation + both ClusterClientReceptionist.RegisterService calls in
  AkkaHostedService; the RegisterCentralClient message + receive block
- CommunicationOptions.CentralContactPoints; the CentralTransport/SiteTransport
  coexistence flags; the CentralTransportMode/SiteTransportKind enums

gRPC is now the only site↔central transport (site→central CentralControlService via
GrpcCentralTransport; central→site SiteCommandService via GrpcSiteTransport), both
built unconditionally by the Host. NoOpCentralTransport is the fail-loud null-default
so TestKit command-dispatch suites still construct the site actor without a wired
transport; production always injects GrpcCentralTransport.

Config: CentralGrpcEndpoints is now unconditional — CommunicationOptionsValidator
rejects blank entries (role-agnostic), and StartupValidator requires a Site node to
list >=1 endpoint (fail-fast, mirrors GrpcPsk). Rig configs moved
CentralContactPoints -> CentralGrpcEndpoints (docker x6, docker-env2 x2, Host default,
deploy/wonder-app-vd03). Kept Akka.Cluster.Tools (ClusterSingleton still used).

Tests: build 0/0; Communication.Tests 640, Host.Tests 421 green. Removed the
ClusterClient.Send per-site-routing tests (covered by the transport suites), swapped
the ISiteClientFactory-based ctors to a substitute ISiteCommandTransport, converted
the audit-push integration relay to an in-process bridge transport.

Docs: Component-Communication/Host/StoreAndForward, components/Communication,
topology-guide, grpc_streams (SUPERSEDED note), the frame-size known-issue (retired
amendment), and CLAUDE.md transport decisions.

Not included: the dead IntegrationCallRequest path (#32) is a separate user-owned
behavioral decision — SiteEnvelope routing is transport-agnostic so it still compiles.
This commit is contained in:
Joseph Doherty
2026-07-23 12:54:32 -04:00
parent 3a8ddb7087
commit 7fd5cb2b56
42 changed files with 479 additions and 1295 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/Actors/AkkaCentralTransportTests.cs
-67
View File
@@ -1,67 +0,0 @@
using Akka.Actor;
using Akka.Cluster.Tools.Client;
using Akka.TestKit.Xunit2;
using ZB.MOM.WW.ScadaBridge.Commons.Messages.Notification;
using ZB.MOM.WW.ScadaBridge.Communication.Actors;
namespace ZB.MOM.WW.ScadaBridge.Communication.Tests.Actors;
/// <summary>
/// T1A.3: the regression-prone bit of the Akka transport — that a forwarded
/// <see cref="ClusterClient.Send"/> carries the caller's sender, so central's reply routes
/// straight back to the waiting Ask rather than to the site communication actor — plus the
/// no-ClusterClient-yet fallback that keeps the S&amp;F layer treating the send as transient.
/// </summary>
public class AkkaCentralTransportTests : TestKit
{
[Fact]
public void SubmitNotification_ForwardsToClusterClient_WithReplyToAsSender()
{
var transport = new AkkaCentralTransport();
var clusterClient = CreateTestProbe();
var replyTo = CreateTestProbe();
transport.SetCentralClient(clusterClient.Ref);
var submit = new NotificationSubmit(
"notif-1", "Operators", "S", "B", "site1", null, null, DateTimeOffset.UtcNow);
transport.SubmitNotification(submit, replyTo.Ref);
// The ClusterClient receives a Send addressed to the central actor...
var send = clusterClient.ExpectMsg<ClusterClient.Send>();
Assert.Equal("/user/central-communication", send.Path);
Assert.IsType<NotificationSubmit>(send.Message);
// ...and replying to it lands at replyTo, proving the sender was forwarded (not the
// transport / actor). This is the routing the waiting Ask relies on.
clusterClient.Reply(new NotificationSubmitAck("notif-1", Accepted: true, Error: null));
replyTo.ExpectMsg<NotificationSubmitAck>(ack => ack.NotificationId == "notif-1" && ack.Accepted);
}
[Fact]
public void SubmitNotification_WithNoClusterClient_RepliesNonAcceptedToReplyTo()
{
var transport = new AkkaCentralTransport();
var replyTo = CreateTestProbe();
transport.SubmitNotification(
new NotificationSubmit("notif-2", "Operators", "S", "B", "site1", null, null, DateTimeOffset.UtcNow),
replyTo.Ref);
replyTo.ExpectMsg<NotificationSubmitAck>(ack => ack.NotificationId == "notif-2" && !ack.Accepted);
}
[Fact]
public void IngestAuditEvents_WithNoClusterClient_FaultsTheReplyTo()
{
// The audit drain treats a faulted Ask as transient and keeps rows Pending — so the
// no-client path must be a Status.Failure, not a silent drop.
var transport = new AkkaCentralTransport();
var replyTo = CreateTestProbe();
transport.IngestAuditEvents(
new Commons.Messages.Audit.IngestAuditEventsCommand(new List<ZB.MOM.WW.Audit.AuditEvent>()),
replyTo.Ref);
replyTo.ExpectMsg<Status.Failure>();
}
}
tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/CentralCommunicationActorAuditTests.cs
+2 -2
View File
@@ -36,9 +36,9 @@ public class CentralCommunicationActorAuditTests : TestKit
services.AddScoped(_ => mockRepo);
var sp = services.BuildServiceProvider();
var mockFactory = Substitute.For<ISiteClientFactory>();
var transport = Substitute.For<ISiteCommandTransport>();
return Sys.ActorOf(Props.Create(() =>
new CentralCommunicationActor(sp, mockFactory, auditIngestAskTimeout)));
new CentralCommunicationActor(sp, transport, auditIngestAskTimeout)));
}
// C3 (Task 2.5): canonical ZB.MOM.WW.Audit.AuditEvent via the shared factory.
tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/CentralCommunicationActorClientLifecycleTests.cs
+10 -72
View File
@@ -1,47 +1,25 @@
using System.Collections.Immutable;
using Akka.Actor;
using Akka.Cluster.Tools.Client;
using Akka.Configuration;
using Akka.TestKit.Xunit2;
using Microsoft.Extensions.DependencyInjection;
using NSubstitute;
using Xunit;
using ZB.MOM.WW.ScadaBridge.Commons.Entities.Sites;
using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Repositories;
using ZB.MOM.WW.ScadaBridge.Communication;
using ZB.MOM.WW.ScadaBridge.Communication.Actors;
using ZB.MOM.WW.ScadaBridge.HealthMonitoring;
namespace ZB.MOM.WW.ScadaBridge.Communication.Tests;
/// <summary>
/// Transport-agnostic refresh behaviour of <see cref="CentralCommunicationActor"/>: the periodic
/// DB refresh evicts deleted sites from the health aggregator. The two former tests here —
/// per-site ClusterClient recreate-without-restart-loop and factory-throw resilience — asserted
/// Akka <c>AkkaSiteTransport</c> internals (InvalidActorNameException, "No ClusterClient for site")
/// and were removed with that transport in the ClusterClient→gRPC migration's Phase 4; the
/// equivalent gRPC channel reconcile is covered by the <c>GrpcSiteTransport</c> suites.
/// </summary>
public class CentralCommunicationActorClientLifecycleTests : TestKit
{
private static readonly Config TestConfig = ConfigurationFactory.ParseString(@"
akka.actor.provider = cluster
akka.remote.dot-netty.tcp.port = 0
akka.remote.dot-netty.tcp.hostname = localhost")
.WithFallback(ClusterClientReceptionist.DefaultConfig());
public CentralCommunicationActorClientLifecycleTests() : base(TestConfig) { }
// Empty ServiceProvider with a no-op ISiteRepository so the actor's periodic
// db-refresh (fired at PreStart) resolves and returns no sites, keeping the
// logs clean; the tests drive SiteAddressCacheLoaded directly.
private static IServiceProvider EmptyProvider()
{
var siteRepo = Substitute.For<ISiteRepository>();
siteRepo.GetAllSitesAsync(Arg.Any<CancellationToken>()).Returns(new List<Site>());
var services = new ServiceCollection();
services.AddScoped(_ => siteRepo);
return services.BuildServiceProvider();
}
private static SiteAddressCacheLoaded Load(string siteId, params string[] addrs) =>
new(new Dictionary<string, IReadOnlyList<string>>
{ [siteId] = addrs.ToList().AsReadOnly() },
new[] { siteId },
new Dictionary<string, SiteGrpcEndpoints>());
[Fact]
public void PeriodicRefresh_PrunesDeletedSites_FromHealthAggregator()
{
@@ -57,8 +35,9 @@ public class CentralCommunicationActorClientLifecycleTests : TestKit
services.AddSingleton(aggregator);
var provider = services.BuildServiceProvider();
var transport = Substitute.For<ISiteCommandTransport>();
var actor = Sys.ActorOf(Props.Create(() => new CentralCommunicationActor(
provider, new DefaultSiteClientFactory(), null)));
provider, transport, (TimeSpan?)null)));
// Trigger the refresh (also fires at PreStart, but drive it explicitly so
// the assertion is deterministic). The load runs on a detached task and
@@ -70,45 +49,4 @@ public class CentralCommunicationActorClientLifecycleTests : TestKit
Arg.Is<IReadOnlyCollection<string>>(ids => ids.Contains("site-a"))),
TimeSpan.FromSeconds(3));
}
[Fact]
public void AddressEdit_RecreatesClient_WithoutRestartLoop()
{
var actor = Sys.ActorOf(Props.Create(() => new CentralCommunicationActor(
EmptyProvider(), new DefaultSiteClientFactory(), null)));
// First load creates the client; second load (edited NodeA address) stops
// the old one and creates a replacement in the same message handling.
// Pre-fix this throws InvalidActorNameException and restarts the actor.
EventFilter.Exception<InvalidActorNameException>().Expect(0, () =>
{
actor.Tell(Load("site-a", "akka.tcp://scadabridge@node-a:8081"));
actor.Tell(Load("site-a", "akka.tcp://scadabridge@node-a-edited:8081"));
actor.Tell(Load("site-a", "akka.tcp://scadabridge@node-a:8081")); // and back — third generation
});
// The actor must still be alive and routing (not crash-looping):
// an envelope for an unknown site produces the "No ClusterClient" warning,
// proving the Receive pipeline is healthy.
EventFilter.Warning(contains: "No ClusterClient for site").ExpectOne(() =>
actor.Tell(new SiteEnvelope("unknown-site", new object())));
}
[Fact]
public void FactoryThrow_SkipsSite_DoesNotCrashActor()
{
var throwingFactory = Substitute.For<ISiteClientFactory>();
throwingFactory.Create(Arg.Any<ActorSystem>(), Arg.Any<string>(), Arg.Any<ImmutableHashSet<ActorPath>>())
.Returns(_ => throw new InvalidOperationException("boom"));
var actor = Sys.ActorOf(Props.Create(() => new CentralCommunicationActor(
EmptyProvider(), throwingFactory, null)));
EventFilter.Error(contains: "Failed to create ClusterClient").ExpectOne(() =>
actor.Tell(Load("site-a", "akka.tcp://scadabridge@node-a:8081")));
// Actor survived — a subsequent envelope for that (unrouted) site still
// produces the healthy "No ClusterClient" warning rather than a dead actor.
EventFilter.Warning(contains: "No ClusterClient for site").ExpectOne(() =>
actor.Tell(new SiteEnvelope("site-a", new object())));
}
}
tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/CentralCommunicationActorReconcileTests.cs
+2 -2
View File
@@ -62,8 +62,8 @@ public class CentralCommunicationActorReconcileTests : TestKit
services.AddScoped<ReconcileService>();
var sp = services.BuildServiceProvider();
var factory = Substitute.For<ISiteClientFactory>();
var actor = Sys.ActorOf(Props.Create(() => new CentralCommunicationActor(sp, factory, null)));
var transport = Substitute.For<ISiteCommandTransport>();
var actor = Sys.ActorOf(Props.Create(() => new CentralCommunicationActor(sp, transport, (TimeSpan?)null)));
// Node B is missing inst-B entirely → it should come back as a gap item.
actor.Tell(new ReconcileSiteRequest(
tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/CentralCommunicationActorTests.cs
+22 -186
View File
@@ -1,6 +1,4 @@
using System.Collections.Immutable;
using Akka.Actor;
using Akka.Cluster.Tools.Client;
using Akka.TestKit.Xunit2;
using Microsoft.Extensions.DependencyInjection;
using NSubstitute;
@@ -18,15 +16,19 @@ using Akka.TestKit;
namespace ZB.MOM.WW.ScadaBridge.Communication.Tests;
/// <summary>
/// Tests for CentralCommunicationActor with per-site ClusterClient routing.
/// WP-4: Message routing via ClusterClient instances created per site.
/// WP-5: Connection failure and failover handling.
/// Tests for <see cref="CentralCommunicationActor"/> behaviour that is independent of the
/// central→site command transport: heartbeat/replica aggregation, notification-outbox proxying,
/// and the DB-refresh failure path. The per-site routing itself (envelope → transport → the right
/// site) is the transport's job and is covered by
/// <see cref="CentralCommunicationActorTransportTests"/> and the <c>GrpcSiteTransport</c> suites;
/// the old ClusterClient-routing tests here were removed with the Akka transport in the
/// ClusterClient→gRPC migration's Phase 4.
/// </summary>
public class CentralCommunicationActorTests : TestKit
{
public CentralCommunicationActorTests() : base(@"akka.loglevel = DEBUG") { }
private (IActorRef actor, ISiteRepository mockRepo, Dictionary<string, TestProbe> siteProbes) CreateActorWithMockRepo(
private (IActorRef actor, ISiteRepository mockRepo) CreateActorWithMockRepo(
IEnumerable<Site>? sites = null)
{
var mockRepo = Substitute.For<ISiteRepository>();
@@ -37,93 +39,11 @@ public class CentralCommunicationActorTests : TestKit
services.AddScoped(_ => mockRepo);
var sp = services.BuildServiceProvider();
var siteProbes = new Dictionary<string, TestProbe>();
var mockFactory = Substitute.For<ISiteClientFactory>();
mockFactory.Create(Arg.Any<ActorSystem>(), Arg.Any<string>(), Arg.Any<ImmutableHashSet<ActorPath>>())
.Returns(callInfo =>
{
var siteId = callInfo.ArgAt<string>(1);
var probe = CreateTestProbe();
siteProbes[siteId] = probe;
return probe.Ref;
});
var actor = Sys.ActorOf(Props.Create(() => new CentralCommunicationActor(sp, mockFactory)));
return (actor, mockRepo, siteProbes);
var transport = Substitute.For<ISiteCommandTransport>();
var actor = Sys.ActorOf(Props.Create(() => new CentralCommunicationActor(sp, transport, (TimeSpan?)null)));
return (actor, mockRepo);
}
private (IActorRef actor, ISiteRepository mockRepo, Dictionary<string, TestProbe> siteProbes, ISiteClientFactory mockFactory) CreateActorWithFactory(
IEnumerable<Site>? sites = null)
{
var mockRepo = Substitute.For<ISiteRepository>();
mockRepo.GetAllSitesAsync(Arg.Any<CancellationToken>())
.Returns(sites?.ToList() ?? new List<Site>());
var services = new ServiceCollection();
services.AddScoped(_ => mockRepo);
var sp = services.BuildServiceProvider();
var siteProbes = new Dictionary<string, TestProbe>();
var mockFactory = Substitute.For<ISiteClientFactory>();
mockFactory.Create(Arg.Any<ActorSystem>(), Arg.Any<string>(), Arg.Any<ImmutableHashSet<ActorPath>>())
.Returns(callInfo =>
{
var siteId = callInfo.ArgAt<string>(1);
var probe = CreateTestProbe();
siteProbes[siteId] = probe;
return probe.Ref;
});
var actor = Sys.ActorOf(Props.Create(() => new CentralCommunicationActor(sp, mockFactory)));
return (actor, mockRepo, siteProbes, mockFactory);
}
private Site CreateSite(string identifier, string? nodeAAddress, string? nodeBAddress = null) =>
new("Test Site", identifier) { NodeAAddress = nodeAAddress, NodeBAddress = nodeBAddress };
[Fact]
public void ClusterClientRouting_RefreshDeploymentCommand_RoutesToSite()
{
var site = CreateSite("site1", "akka.tcp://scadabridge@host:8082");
var (actor, _, siteProbes) = CreateActorWithMockRepo(new[] { site });
Thread.Sleep(1000);
var command = new RefreshDeploymentCommand(
"dep1", "inst1", "rev1", "admin", DateTimeOffset.UtcNow,
"https://central:9000", "tok1");
actor.Tell(new SiteEnvelope("site1", command));
var msg = siteProbes["site1"].ExpectMsg<ClusterClient.Send>();
Assert.Equal("/user/site-communication", msg.Path);
Assert.IsType<RefreshDeploymentCommand>(msg.Message);
Assert.Equal("dep1", ((RefreshDeploymentCommand)msg.Message).DeploymentId);
}
[Fact]
public void UnconfiguredSite_MessageIsDropped()
{
var (actor, _, _) = CreateActorWithMockRepo();
// Wait for auto-refresh
Thread.Sleep(1000);
var command = new RefreshDeploymentCommand(
"dep1", "inst1", "hash1", "admin", DateTimeOffset.UtcNow,
"https://central:9000", "tok1");
actor.Tell(new SiteEnvelope("unknown-site", command));
ExpectNoMsg(TimeSpan.FromMilliseconds(200));
}
// Communication-016: the prior `ConnectionLost_DebugStreamsKilled` test was
// removed alongside the dead HandleConnectionStateChanged handler. No
// production code ever emitted ConnectionStateChanged, so the test was
// exercising a workflow that never ran. Disconnect detection is owned by
// the gRPC keepalive (DebugStreamBridgeActor self-terminates) and by the
// Ask-timeout path at the CommunicationService layer (deploy callers see
// a failure).
[Fact]
public void Heartbeat_BumpsAggregatorTimestamp()
{
@@ -138,9 +58,9 @@ public class CentralCommunicationActorTests : TestKit
services.AddSingleton(aggregator);
var sp = services.BuildServiceProvider();
var siteClientFactory = Substitute.For<ISiteClientFactory>();
var transport = Substitute.For<ISiteCommandTransport>();
var centralActor = Sys.ActorOf(
Props.Create(() => new CentralCommunicationActor(sp, siteClientFactory)));
Props.Create(() => new CentralCommunicationActor(sp, transport, (TimeSpan?)null)));
var timestamp = DateTimeOffset.UtcNow;
centralActor.Tell(new HeartbeatMessage("site1", "host1", true, timestamp));
@@ -165,9 +85,9 @@ public class CentralCommunicationActorTests : TestKit
services.AddSingleton(aggregator);
var sp = services.BuildServiceProvider();
var siteClientFactory = Substitute.For<ISiteClientFactory>();
var transport = Substitute.For<ISiteCommandTransport>();
var centralActor = Sys.ActorOf(
Props.Create(() => new CentralCommunicationActor(sp, siteClientFactory)));
Props.Create(() => new CentralCommunicationActor(sp, transport, (TimeSpan?)null)));
var ts = DateTimeOffset.UtcNow;
centralActor.Tell(new SiteHeartbeatReplica(new HeartbeatMessage("site-1", "host-a", true, ts)));
@@ -175,41 +95,6 @@ public class CentralCommunicationActorTests : TestKit
AwaitAssert(() => aggregator.Received(1).MarkHeartbeat("site-1", ts));
}
[Fact]
public void RefreshSiteAddresses_UpdatesCache()
{
var site1 = CreateSite("site1", "akka.tcp://scadabridge@host1:8082");
var (actor, mockRepo, siteProbes) = CreateActorWithMockRepo(new[] { site1 });
// Wait for initial load
Thread.Sleep(1000);
// Verify routing to site1 works
var cmd1 = new RefreshDeploymentCommand(
"dep1", "inst1", "hash1", "admin", DateTimeOffset.UtcNow,
"https://central:9000", "tok1");
actor.Tell(new SiteEnvelope("site1", cmd1));
var msg1 = siteProbes["site1"].ExpectMsg<ClusterClient.Send>();
Assert.Equal("dep1", ((RefreshDeploymentCommand)msg1.Message).DeploymentId);
// Update mock repo to return both sites
var site2 = CreateSite("site2", "akka.tcp://scadabridge@host2:8082");
mockRepo.GetAllSitesAsync(Arg.Any<CancellationToken>())
.Returns(new List<Site> { site1, site2 });
// Refresh again
actor.Tell(new RefreshSiteAddresses());
Thread.Sleep(1000);
// Verify routing to site2 now works
var cmd2 = new RefreshDeploymentCommand(
"dep2", "inst2", "hash2", "admin", DateTimeOffset.UtcNow,
"https://central:9000", "tok2");
actor.Tell(new SiteEnvelope("site2", cmd2));
var msg2 = siteProbes["site2"].ExpectMsg<ClusterClient.Send>();
Assert.Equal("dep2", ((RefreshDeploymentCommand)msg2.Message).DeploymentId);
}
[Fact]
public void LoadSiteAddressesFailure_IsLoggedNotSilentlySwallowed()
{
@@ -226,56 +111,26 @@ public class CentralCommunicationActorTests : TestKit
services.AddScoped(_ => mockRepo);
var sp = services.BuildServiceProvider();
var mockFactory = Substitute.For<ISiteClientFactory>();
var transport = Substitute.For<ISiteCommandTransport>();
// The fix logs a Warning carrying the InvalidOperationException as the cause.
EventFilter.Warning(contains: "Failed to load site addresses from the database").ExpectOne(() =>
{
Sys.ActorOf(Props.Create(() => new CentralCommunicationActor(sp, mockFactory)));
Sys.ActorOf(Props.Create(() => new CentralCommunicationActor(sp, transport, (TimeSpan?)null)));
});
}
[Fact]
public void MalformedSiteAddress_DoesNotAbortRefresh_OtherSitesStillRegistered()
{
// Regression test for Communication-009. HandleSiteAddressCacheLoaded calls
// ActorPath.Parse for every site in a single loop. A malformed NodeAAddress
// throws inside that loop; before the fix the whole refresh aborted partway
// through, leaving the cache half-updated (some sites registered, others not).
// The fix wraps the parse in a try/catch that logs and skips the bad site so
// a single garbage row cannot starve every other site of its ClusterClient.
var goodSite = CreateSite("good-site", "akka.tcp://scadabridge@host1:8082");
// A garbage address that ActorPath.Parse rejects.
var badSite = CreateSite("bad-site", "this is not a valid actor path !!!");
// Order the bad site first so a non-resilient loop aborts before reaching good-site.
var (actor, _, siteProbes) = CreateActorWithMockRepo(new[] { badSite, goodSite });
Thread.Sleep(1000);
// good-site must still be registered and routable despite bad-site failing to parse.
var cmd = new RefreshDeploymentCommand(
"dep1", "inst1", "hash1", "admin", DateTimeOffset.UtcNow,
"https://central:9000", "tok1");
actor.Tell(new SiteEnvelope("good-site", cmd));
Assert.True(siteProbes.ContainsKey("good-site"),
"good-site should have a ClusterClient even though bad-site's address is malformed");
var msg = siteProbes["good-site"].ExpectMsg<ClusterClient.Send>();
Assert.Equal("dep1", ((RefreshDeploymentCommand)msg.Message).DeploymentId);
}
private NotificationSubmit CreateSubmit(string id = "notif1") =>
new(id, "ops-list", "Subject", "Body", "site1", "inst1", "script.cs", DateTimeOffset.UtcNow);
[Fact]
public void NotificationSubmit_ForwardedToOutboxProxy_AckRoutesBackToSite()
{
var (actor, _, _) = CreateActorWithMockRepo();
var (actor, _) = CreateActorWithMockRepo();
var outboxProbe = CreateTestProbe();
actor.Tell(new RegisterNotificationOutbox(outboxProbe.Ref));
// A second probe stands in for the site's ClusterClient (the original Sender).
// A second probe stands in for the site (the original Sender).
var siteProbe = CreateTestProbe();
var submit = CreateSubmit();
actor.Tell(submit, siteProbe.Ref);
@@ -290,7 +145,7 @@ public class CentralCommunicationActorTests : TestKit
[Fact]
public void NotificationStatusQuery_ForwardedToOutboxProxy_ResponseRoutesBackToSite()
{
var (actor, _, _) = CreateActorWithMockRepo();
var (actor, _) = CreateActorWithMockRepo();
var outboxProbe = CreateTestProbe();
actor.Tell(new RegisterNotificationOutbox(outboxProbe.Ref));
@@ -308,7 +163,7 @@ public class CentralCommunicationActorTests : TestKit
[Fact]
public void NotificationSubmit_NoOutboxConfigured_RepliesNonAccepted()
{
var (actor, _, _) = CreateActorWithMockRepo();
var (actor, _) = CreateActorWithMockRepo();
// No RegisterNotificationOutbox sent — the proxy is null.
var submit = CreateSubmit();
@@ -323,7 +178,7 @@ public class CentralCommunicationActorTests : TestKit
[Fact]
public void NotificationStatusQuery_NoOutboxConfigured_RepliesNotFound()
{
var (actor, _, _) = CreateActorWithMockRepo();
var (actor, _) = CreateActorWithMockRepo();
// No RegisterNotificationOutbox sent — the proxy is null.
var query = new NotificationStatusQuery("corr1", "notif1");
@@ -333,23 +188,4 @@ public class CentralCommunicationActorTests : TestKit
Assert.Equal("corr1", response.CorrelationId);
Assert.False(response.Found);
}
[Fact]
public void BothContactPoints_UsedInSingleClient()
{
var site = CreateSite("site1",
"akka.tcp://scadabridge@host1:8082",
"akka.tcp://scadabridge@host2:8082");
var (actor, _, siteProbes, mockFactory) = CreateActorWithFactory(new[] { site });
// Wait for auto-refresh
Thread.Sleep(1000);
// Verify the factory was called with 2 contact paths
mockFactory.Received(1).Create(
Arg.Any<ActorSystem>(),
Arg.Is("site1"),
Arg.Is<ImmutableHashSet<ActorPath>>(paths => paths.Count == 2));
}
}
tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/CommunicationOptionsValidatorTests.cs
+10 -21
View File
@@ -105,26 +105,28 @@ public class CommunicationOptionsValidatorTests
Assert.Contains("LiveAlarmCachePublishCoalesce", result.FailureMessage);
}
// ── T1A.3: gRPC central transport endpoints (required only when selected) ────
// ── gRPC central transport endpoints ─────────────────────────────────────────
// gRPC (CentralControlService) is the only site→central transport after the
// ClusterClient→gRPC migration's Phase 4. This role-agnostic validator only rejects BLANK
// entries; an EMPTY list is valid (central nodes legitimately declare none). The role-aware
// "a Site must list at least one endpoint" rule lives in StartupValidator, tested there.
[Fact]
public void GrpcTransport_WithNoEndpoints_IsRejected()
public void EmptyEndpoints_IsValid()
{
// A central node hosts CentralControlService; it does not dial it, so it declares none.
var result = Validate(new CommunicationOptions
{
CentralTransport = CentralTransportMode.Grpc,
CentralGrpcEndpoints = new List<string>(),
});
Assert.True(result.Failed);
Assert.Contains("CentralGrpcEndpoints", result.FailureMessage);
Assert.True(result.Succeeded, result.FailureMessage);
}
[Fact]
public void GrpcTransport_WithBlankEndpoint_IsRejected()
public void BlankEndpoint_IsRejected()
{
var result = Validate(new CommunicationOptions
{
CentralTransport = CentralTransportMode.Grpc,
CentralGrpcEndpoints = new List<string> { " " },
});
Assert.True(result.Failed);
@@ -132,11 +134,10 @@ public class CommunicationOptionsValidatorTests
}
[Fact]
public void GrpcTransport_WithEndpoints_IsValid()
public void Endpoints_IsValid()
{
var result = Validate(new CommunicationOptions
{
CentralTransport = CentralTransportMode.Grpc,
CentralGrpcEndpoints = new List<string>
{
"http://scadabridge-central-a:8083",
@@ -145,16 +146,4 @@ public class CommunicationOptionsValidatorTests
});
Assert.True(result.Succeeded, result.FailureMessage);
}
[Fact]
public void AkkaTransport_IgnoresMissingGrpcEndpoints()
{
// The default transport must not be forced to declare gRPC endpoints it never dials.
var result = Validate(new CommunicationOptions
{
CentralTransport = CentralTransportMode.Akka,
CentralGrpcEndpoints = new List<string>(),
});
Assert.True(result.Succeeded, result.FailureMessage);
}
}
tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/CoordinatorSupervisionTests.cs
+5 -5
View File
@@ -12,7 +12,7 @@ namespace ZB.MOM.WW.ScadaBridge.Communication.Tests;
/// explicit <c>Resume</c> supervision strategy per the CLAUDE.md decision
/// ("Resume for coordinator actors"). A child fault under the default
/// (Restart) strategy would wipe a child's in-memory state; the long-lived
/// coordinators own per-site ClusterClients and must not silently discard
/// coordinators own per-site transport resources and must not silently discard
/// their children on a transient fault.
/// </summary>
public class CoordinatorSupervisionTests : TestKit
@@ -23,8 +23,8 @@ public class CoordinatorSupervisionTests : TestKit
/// </summary>
private sealed class CentralCommunicationActorProbe : CentralCommunicationActor
{
public CentralCommunicationActorProbe(IServiceProvider sp, ISiteClientFactory factory)
: base(sp, factory) { }
public CentralCommunicationActorProbe(IServiceProvider sp, ISiteCommandTransport transport)
: base(sp, transport) { }
public SupervisorStrategy GetSupervisorStrategy() => SupervisorStrategy();
}
@@ -54,10 +54,10 @@ public class CoordinatorSupervisionTests : TestKit
public void CentralCommunicationActor_SupervisorStrategy_IsResume()
{
var sp = EmptyServiceProvider();
var factory = Substitute.For<ISiteClientFactory>();
var transport = Substitute.For<ISiteCommandTransport>();
var actorRef = new Akka.TestKit.TestActorRef<CentralCommunicationActorProbe>(
Sys, Props.Create(() => new CentralCommunicationActorProbe(sp, factory)));
Sys, Props.Create(() => new CentralCommunicationActorProbe(sp, transport)));
var strategy = actorRef.UnderlyingActor.GetSupervisorStrategy();
tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/DefaultSiteClientFactoryTests.cs
-54
View File
@@ -1,54 +0,0 @@
using System.Collections.Immutable;
using Akka.Actor;
using Akka.Cluster.Tools.Client;
using Akka.Configuration;
using Akka.TestKit.Xunit2;
using Xunit;
using ZB.MOM.WW.ScadaBridge.Communication.Actors;
namespace ZB.MOM.WW.ScadaBridge.Communication.Tests;
public class DefaultSiteClientFactoryTests : TestKit
{
private static readonly Config TestConfig = ConfigurationFactory.ParseString(@"
akka.actor.provider = cluster
akka.remote.dot-netty.tcp.port = 0
akka.remote.dot-netty.tcp.hostname = localhost")
.WithFallback(ClusterClientReceptionist.DefaultConfig());
public DefaultSiteClientFactoryTests() : base(TestConfig) { }
private static ImmutableHashSet<ActorPath> Contacts() =>
ImmutableHashSet.Create(ActorPath.Parse("akka.tcp://other@localhost:2552/system/receptionist"));
[Fact]
public void Create_TwiceForSameSite_DoesNotCollide()
{
var factory = new DefaultSiteClientFactory();
var first = factory.Create(Sys, "site-a", Contacts());
var second = factory.Create(Sys, "site-a", Contacts()); // pre-fix: InvalidActorNameException
Assert.NotEqual(first.Path, second.Path);
}
[Theory]
[InlineData("site/with/slashes")]
[InlineData("site with spaces")]
[InlineData("näme#!")]
public void Create_WithUnsafeSiteId_SanitizesAndSucceeds(string siteId)
{
var factory = new DefaultSiteClientFactory();
var client = factory.Create(Sys, siteId, Contacts()); // pre-fix: InvalidActorNameException
Assert.NotNull(client);
}
[Theory]
[InlineData("plant-01", "plant-01")]
[InlineData("a/b c", "a_b_c")]
[InlineData("", "site")]
public void SanitizeForActorName_ProducesValidPathElement(string input, string expected)
{
Assert.Equal(expected, DefaultSiteClientFactory.SanitizeForActorName(input));
Assert.True(ActorPath.IsValidPathElement(
DefaultSiteClientFactory.SanitizeForActorName(input)));
}
}
tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/HealthReportAckTests.cs
+11 -11
View File
@@ -14,9 +14,9 @@ namespace ZB.MOM.WW.ScadaBridge.Communication.Tests;
/// <summary>
/// Review 01 [Medium]: site→central health reports were fire-and-forget, so a lost
/// report was invisible to the sender. These tests pin the new end-to-end ack:
/// the site actor replies not-accepted when it has no central ClusterClient, and the
/// central actor processes + acks a report it receives.
/// report was invisible to the sender. These tests pin the end-to-end ack: the site actor
/// always replies to the sender (even when the transport cannot forward — a fail-loud failure,
/// not silence), and the central actor processes + acks a report it receives.
/// </summary>
public class HealthReportAckTests : TestKit
{
@@ -29,17 +29,17 @@ public class HealthReportAckTests : TestKit
0, 0, new Dictionary<string, int>(), 0, 0, 0, 0);
[Fact]
public void SiteCommunicationActor_NoCentralClient_RepliesNotAccepted()
public void SiteCommunicationActor_NoTransport_RepliesFailure_NotSilence()
{
// With no transport injected the actor falls back to the fail-loud
// NoOpCentralTransport, which answers a Status.Failure so the health transport's
// Ask sees a transient failure rather than hanging. (Production always injects the
// gRPC GrpcCentralTransport; this pins the wiring-guard fallback.)
var dmProbe = CreateTestProbe();
var siteComm = Sys.ActorOf(Props.Create(() =>
new SiteCommunicationActor("site-a", _options, dmProbe.Ref)));
// No RegisterCentralClient sent => _centralClient is null.
siteComm.Tell(SampleReport(seq: 7));
var ack = ExpectMsg<SiteHealthReportAck>();
Assert.False(ack.Accepted);
Assert.Equal(7, ack.SequenceNumber);
Assert.Equal("site-a", ack.SiteId);
ExpectMsg<Status.Failure>();
}
[Fact]
@@ -54,8 +54,8 @@ public class HealthReportAckTests : TestKit
services.AddSingleton(aggregator);
var sp = services.BuildServiceProvider();
var siteClientFactory = Substitute.For<ISiteClientFactory>();
var actor = Sys.ActorOf(Props.Create(() => new CentralCommunicationActor(sp, siteClientFactory)));
var transport = Substitute.For<ISiteCommandTransport>();
var actor = Sys.ActorOf(Props.Create(() => new CentralCommunicationActor(sp, transport, (TimeSpan?)null)));
actor.Tell(SampleReport(seq: 3));
var ack = ExpectMsg<SiteHealthReportAck>();
tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/SiteCommunicationActorTests.cs
+17 -106
View File
@@ -1,6 +1,6 @@
using Akka.Actor;
using Akka.Cluster.Tools.Client;
using Akka.TestKit.Xunit2;
using NSubstitute;
using ZB.MOM.WW.ScadaBridge.Commons.Messages.DataConnection;
using ZB.MOM.WW.ScadaBridge.Commons.Messages.Deployment;
using ZB.MOM.WW.ScadaBridge.Commons.Messages.Health;
@@ -109,100 +109,11 @@ public class SiteCommunicationActorTests : TestKit
handlerProbe.ExpectMsg<IntegrationCallRequest>(msg => msg.CorrelationId == "corr1");
}
[Fact]
public void NotificationSubmit_WithCentralClient_ForwardedToCentralAndAckRoutedBack()
{
// The site forwards a buffered notification to central over the ClusterClient
// command/control transport; the central ack must route back to the original
// sender (the S&F forwarder's Ask), not to the SiteCommunicationActor.
var dmProbe = CreateTestProbe();
var centralClientProbe = CreateTestProbe();
var siteActor = Sys.ActorOf(Props.Create(() =>
new SiteCommunicationActor("site1", _options, dmProbe.Ref)));
siteActor.Tell(new RegisterCentralClient(centralClientProbe.Ref));
var submit = new NotificationSubmit(
"notif-1", "Operators", "Subj", "Body", "site1", "inst1", "alarmScript",
DateTimeOffset.UtcNow);
siteActor.Tell(submit);
// Central client (acting as ClusterClient) receives a ClusterClient.Send wrapping
// the NotificationSubmit, addressed to the central communication actor. Fish past
// any periodic HeartbeatMessage the actor's timer may interleave.
var send = centralClientProbe.FishForMessage<ClusterClient.Send>(
s => s.Message is NotificationSubmit);
Assert.Equal("/user/central-communication", send.Path);
var forwarded = Assert.IsType<NotificationSubmit>(send.Message);
Assert.Equal("notif-1", forwarded.NotificationId);
// The ack is sent to the ClusterClient.Send's Sender — replying as that probe
// must land back at the test actor (the original Tell sender).
centralClientProbe.Reply(new NotificationSubmitAck("notif-1", Accepted: true, Error: null));
ExpectMsg<NotificationSubmitAck>(ack => ack.NotificationId == "notif-1" && ack.Accepted);
}
[Fact]
public void NotificationSubmit_WithoutCentralClient_RepliesWithNonAccepted()
{
// No ClusterClient registered yet: the submit cannot be forwarded, so the actor
// replies with a non-accepted ack and the S&F forwarder treats it as transient.
var dmProbe = CreateTestProbe();
var siteActor = Sys.ActorOf(Props.Create(() =>
new SiteCommunicationActor("site1", _options, dmProbe.Ref)));
var submit = new NotificationSubmit(
"notif-2", "Operators", "Subj", "Body", "site1", null, null,
DateTimeOffset.UtcNow);
siteActor.Tell(submit);
ExpectMsg<NotificationSubmitAck>(ack => ack.NotificationId == "notif-2" && !ack.Accepted);
}
[Fact]
public void NotificationStatusQuery_WithCentralClient_ForwardedToCentralAndResponseRoutedBack()
{
// Notify.Status(id) issues a NotificationStatusQuery; the site actor forwards it
// to central over the ClusterClient command/control transport and the central
// response must route back to the original sender (the helper's Ask).
var dmProbe = CreateTestProbe();
var centralClientProbe = CreateTestProbe();
var siteActor = Sys.ActorOf(Props.Create(() =>
new SiteCommunicationActor("site1", _options, dmProbe.Ref)));
siteActor.Tell(new RegisterCentralClient(centralClientProbe.Ref));
var query = new NotificationStatusQuery("corr-99", "notif-1");
siteActor.Tell(query);
var send = centralClientProbe.FishForMessage<ClusterClient.Send>(
s => s.Message is NotificationStatusQuery);
Assert.Equal("/user/central-communication", send.Path);
var forwarded = Assert.IsType<NotificationStatusQuery>(send.Message);
Assert.Equal("notif-1", forwarded.NotificationId);
// The response is sent to the ClusterClient.Send's Sender — replying as that
// probe must land back at the test actor (the original Tell sender).
centralClientProbe.Reply(new NotificationStatusResponse(
"corr-99", Found: true, Status: "Delivered", RetryCount: 0,
LastError: null, DeliveredAt: DateTimeOffset.UtcNow));
ExpectMsg<NotificationStatusResponse>(r => r.CorrelationId == "corr-99" && r.Found);
}
[Fact]
public void NotificationStatusQuery_WithoutCentralClient_RepliesWithNotFound()
{
// No ClusterClient registered yet: the query cannot reach central, so the actor
// replies Found: false. Notify.Status then falls back to the site S&F buffer.
var dmProbe = CreateTestProbe();
var siteActor = Sys.ActorOf(Props.Create(() =>
new SiteCommunicationActor("site1", _options, dmProbe.Ref)));
siteActor.Tell(new NotificationStatusQuery("corr-100", "notif-2"));
ExpectMsg<NotificationStatusResponse>(
r => r.CorrelationId == "corr-100" && !r.Found);
}
// The site→central send-forwarding tests that used to live here (NotificationSubmit /
// NotificationStatusQuery, with and without a registered central ClusterClient) were removed
// with the Akka transport in the ClusterClient→gRPC migration's Phase 4: the actor no longer
// owns the wire, it delegates to an injected ICentralTransport, and those seven sends (plus the
// reply routing and the fail-loud fallback) are now covered by SiteCommunicationActorTransportTests.
[Fact]
public void EventLogQuery_WithoutHandler_ReturnsFailure()
@@ -355,22 +266,22 @@ public class SiteCommunicationActorTests : TestKit
// leader check in production); tests inject a stub so they do not need
// to bring up a full cluster in the TestKit ActorSystem.
var dmProbe = CreateTestProbe();
var centralClientProbe = CreateTestProbe();
var transport = Substitute.For<ICentralTransport>();
var fastHeartbeatOptions = new CommunicationOptions
{
TransportHeartbeatInterval = TimeSpan.FromMilliseconds(50)
ApplicationHeartbeatInterval = TimeSpan.FromMilliseconds(100)
};
var siteActor = Sys.ActorOf(Props.Create(() =>
new SiteCommunicationActor("site1", fastHeartbeatOptions, dmProbe.Ref, () => isActive)));
Sys.ActorOf(Props.Create(() =>
new SiteCommunicationActor("site1", fastHeartbeatOptions, dmProbe.Ref, () => isActive, null, transport)));
siteActor.Tell(new RegisterCentralClient(centralClientProbe.Ref));
var send = centralClientProbe.FishForMessage<ClusterClient.Send>(
s => s.Message is HeartbeatMessage, TimeSpan.FromSeconds(3));
var heartbeat = Assert.IsType<HeartbeatMessage>(send.Message);
Assert.Equal(isActive, heartbeat.IsActive);
Assert.Equal("site1", heartbeat.SiteId);
// The heartbeat now rides the injected ICentralTransport (SendHeartbeat), not a
// ClusterClient. The captured message must carry the injected active/standby flag.
AwaitAssert(
() => transport.Received().SendHeartbeat(
Arg.Is<HeartbeatMessage>(h => h.IsActive == isActive && h.SiteId == "site1"),
Arg.Any<IActorRef>()),
TimeSpan.FromSeconds(3));
}
// ---- Central→site failover relay (Task 10) ----------------------------------