Joseph Doherty
33b15f10a4
Introduce ICentralTransport as the site->central choke point inside SiteCommunicationActor. The seven site->central sends (notification submit/ status, audit + cached-telemetry ingest, reconcile, health, heartbeat) now delegate to an injected transport instead of owning ClusterClient.Send inline. - AkkaCentralTransport: verbatim extraction of today's ClusterClient.Send path, including the exact sender-forwarding that routes central's reply straight back to the waiting Ask. Default when no transport is injected -> behaviour unchanged, existing SiteCommunicationActorTests pass as-is. - GrpcCentralTransport + CentralChannelProvider: dial CentralControlService with sticky failover + background failback (1s-doubling-cap-60s), PSK + x-scadabridge-site via ControlPlaneCredentials, per-call deadlines mirroring today's Ask timeouts. Cross-node retry ONLY on provably-unsent connect failures; never on DeadlineExceeded. Heartbeat stays fire-and-forget. - StaticSitePskProvider: site's single own-key provider (fail-closed). - CommunicationOptions: CentralTransport flag (default Akka) + CentralGrpcEndpoints (validator: required when transport=Grpc). Host selects the impl; the ClusterClient is created only on the Akka path. Tests: actor-with-fake-transport (7 delegations + fault routing + heartbeat no-fault), AkkaCentralTransport sender-forwarding, GrpcCentralTransport over in-process TestServer (failover flip, sticky, failback, PSK+header, deadline, no-retry-on-deadline), validator. Communication.Tests 371 green, Host.Tests 391 green; the three above-seam suites pass unmodified.
68 lines
2.8 KiB
C#
68 lines
2.8 KiB
C#
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&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>();
|
|
}
|
|
}
|