feat(grpc): host CentralControlService on the central node (T1A.2)
Central now ALSO listens for the seven site→central control messages over gRPC, alongside the existing ClusterClient path. Nothing flips to gRPC yet — sites keep CentralTransport=Akka (T1A.3's job); central simply starts also accepting. - CentralControlGrpcService (Communication.Grpc): decodes each RPC onto the SAME in-process message the ClusterClient path carries, Asks the existing CentralCommunicationActor (zero handler-logic changes), encodes the reply via the T1A.1 mapper. Readiness-gated like SiteStreamGrpcServer.SetReady — Unavailable until AkkaHostedService hands the actor over. Heartbeat stays fire-and-forget (Tell, always-OK, never gated on readiness). Ingest reuses the shared SiteStreamGrpcServer.AuditIngestAskTimeout constant. Fault→status mapping is retry-aware: Unavailable (never dispatched, safe to cross-node retry) vs DeadlineExceeded/Internal (it ran, do not re-send elsewhere). - CentralControlAuthInterceptor (Host): a SEPARATE interceptor class, not a variant constructor on ControlPlaneAuthInterceptor. Central's model is per-site (verify the Bearer token against the key for the site in the required x-scadabridge-site header, via ISitePskProvider) where a site verifies its one own-key — a genuinely different model. Fail-closed on every branch: missing or blank header, unresolvable key, and mismatched token all → PermissionDenied, never pass-through. One public constructor only (the explicit-prefix ctor is internal), pinned by a reflection test — a second public ctor makes Grpc.AspNetCore's GetFactory() throw per-call and silently disables the gate. - Explicit Kestrel h2c listener on new option ScadaBridge:Node:CentralGrpcPort (default 8083, symmetric with sites), mirroring the Site branch. Additive to central's :5000 HTTP/1 surface, which is untouched — gRPC does NOT go through Traefik (HTTP/1 only). Registered by type on AddGrpc; service mapped with MapGrpcService. Port range-validated by NodeOptionsValidator. - Rig: publish the central gRPC port 9013:8083 / 9014:8083 on both central nodes so a later task can exercise it. Tests: CentralControlEndToEndTests (Host.Tests, TestServer + real interceptor + real service over a stub actor) proves auth positives/negatives are distinguishable and covers unary + the ingest bridge shapes; the interceptor is registered BY TYPE, never in DI. CentralControlAuthInterceptorTests pins the per-site gate + one-public-ctor invariant. CentralControlGrpcServiceTests (Communication.Tests, TestKit) covers the readiness gate, fire-and-forget heartbeat, and the DeadlineExceeded-vs-Unavailable status mapping. No active <Protobuf> item. Communication.Tests (356) + Host.Tests (384) green.
This commit is contained in:
Joseph Doherty
+181
@@ -0,0 +1,181 @@
|
||||
using Akka.Actor;
|
||||
using Akka.TestKit;
|
||||
using Akka.TestKit.Xunit2;
|
||||
using Grpc.Core;
|
||||
using Microsoft.Extensions.Logging.Abstractions;
|
||||
using Microsoft.Extensions.Options;
|
||||
using NSubstitute;
|
||||
using ZB.MOM.WW.ScadaBridge.Commons.Messages.Audit;
|
||||
using ZB.MOM.WW.ScadaBridge.Commons.Messages.Notification;
|
||||
using ZB.MOM.WW.ScadaBridge.Communication;
|
||||
using ZB.MOM.WW.ScadaBridge.Communication.Grpc;
|
||||
|
||||
namespace ZB.MOM.WW.ScadaBridge.Communication.Tests.Grpc;
|
||||
|
||||
/// <summary>
|
||||
/// Behaviour of <see cref="CentralControlGrpcService"/> that does not need a real gRPC pipeline:
|
||||
/// the readiness gate, the fire-and-forget heartbeat contract, the ingest-timeout status
|
||||
/// mapping, and the shared audit-ingest timeout constant. The auth interceptor + real transport
|
||||
/// are proven separately in <c>CentralControlEndToEndTests</c> (Host.Tests).
|
||||
/// </summary>
|
||||
public class CentralControlGrpcServiceTests : TestKit
|
||||
{
|
||||
private static ServerCallContext NewContext(CancellationToken ct = default)
|
||||
{
|
||||
var context = Substitute.For<ServerCallContext>();
|
||||
context.CancellationToken.Returns(ct);
|
||||
return context;
|
||||
}
|
||||
|
||||
private CentralControlGrpcService CreateService(CommunicationOptions? options = null)
|
||||
=> new(
|
||||
NullLogger<CentralControlGrpcService>.Instance,
|
||||
Options.Create(options ?? new CommunicationOptions()));
|
||||
|
||||
[Fact]
|
||||
public async Task BeforeSetReady_AUnaryCall_IsUnavailable()
|
||||
{
|
||||
// Nothing was dispatched, so Unavailable is the right status — it tells a site transport
|
||||
// the call never ran and a cross-node retry is safe.
|
||||
var service = CreateService();
|
||||
|
||||
var ex = await Assert.ThrowsAsync<RpcException>(
|
||||
() => service.SubmitNotification(new NotificationSubmitDto(), NewContext()));
|
||||
|
||||
Assert.Equal(StatusCode.Unavailable, ex.StatusCode);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task BeforeSetReady_ANonEmptyIngestBatch_IsUnavailable()
|
||||
{
|
||||
var service = CreateService();
|
||||
var batch = new AuditEventBatch();
|
||||
batch.Events.Add(NewAuditDto());
|
||||
|
||||
var ex = await Assert.ThrowsAsync<RpcException>(
|
||||
() => service.IngestAuditEvents(batch, NewContext()));
|
||||
|
||||
Assert.Equal(StatusCode.Unavailable, ex.StatusCode);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task AfterSetReady_AUnaryCall_ReachesTheActor()
|
||||
{
|
||||
var stub = Sys.ActorOf(Props.Create(() => new StubActor()));
|
||||
var service = CreateService();
|
||||
service.SetReady(stub);
|
||||
|
||||
var ack = await service.SubmitNotification(NewNotificationDto(), NewContext());
|
||||
|
||||
Assert.True(ack.Accepted);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task Heartbeat_BeforeSetReady_SucceedsAndIsDropped()
|
||||
{
|
||||
// Fire-and-forget end-to-end: a heartbeat must never fault the site's timer, so even
|
||||
// with no actor wired the call returns OK rather than Unavailable.
|
||||
var service = CreateService();
|
||||
|
||||
var reply = await service.Heartbeat(NewHeartbeatDto(), NewContext());
|
||||
|
||||
Assert.NotNull(reply);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task Heartbeat_AfterSetReady_TellsTheActor_AndNeverAsks()
|
||||
{
|
||||
// A black-hole actor that never replies would hang an Ask forever; the heartbeat still
|
||||
// returns immediately, proving it is a Tell, not an Ask.
|
||||
var blackHole = Sys.ActorOf(Props.Create(() => new NeverRepliesActor()));
|
||||
var service = CreateService();
|
||||
service.SetReady(blackHole);
|
||||
|
||||
var reply = await service.Heartbeat(NewHeartbeatDto(), NewContext());
|
||||
|
||||
Assert.NotNull(reply);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task WhenTheActorNeverReplies_TheCall_IsDeadlineExceeded_NotUnavailable()
|
||||
{
|
||||
// The message WAS delivered (Unavailable would wrongly invite a duplicate retry on the
|
||||
// peer node); a timeout is DeadlineExceeded, which callers never cross-node-retry.
|
||||
var blackHole = Sys.ActorOf(Props.Create(() => new NeverRepliesActor()));
|
||||
var service = CreateService(new CommunicationOptions
|
||||
{
|
||||
NotificationForwardTimeout = TimeSpan.FromMilliseconds(200),
|
||||
});
|
||||
service.SetReady(blackHole);
|
||||
|
||||
var ex = await Assert.ThrowsAsync<RpcException>(
|
||||
() => service.SubmitNotification(NewNotificationDto(), NewContext()));
|
||||
|
||||
Assert.Equal(StatusCode.DeadlineExceeded, ex.StatusCode);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task EmptyIngestBatch_ShortCircuits_EvenBeforeSetReady()
|
||||
{
|
||||
// An empty batch is a no-op the actor need never see; it must not depend on readiness.
|
||||
var service = CreateService();
|
||||
|
||||
var ack = await service.IngestAuditEvents(new AuditEventBatch(), NewContext());
|
||||
|
||||
Assert.Empty(ack.AcceptedEventIds);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void TheAuditIngestTimeout_IsTheOneSharedConstant()
|
||||
{
|
||||
// The plan calls SiteStreamGrpcServer.AuditIngestAskTimeout "one source of truth" shared
|
||||
// between the two audit-ingest transports; the central service must not re-declare 30s.
|
||||
Assert.Equal(TimeSpan.FromSeconds(30), SiteStreamGrpcServer.AuditIngestAskTimeout);
|
||||
}
|
||||
|
||||
// The mapper reads SiteEnqueuedAt unconditionally, so a DTO that reaches mapping must carry
|
||||
// a timestamp. Only DTOs that get past the readiness/auth gate map, so the negative tests
|
||||
// above can pass a bare DTO.
|
||||
private static NotificationSubmitDto NewNotificationDto() => new()
|
||||
{
|
||||
NotificationId = Guid.NewGuid().ToString(),
|
||||
ListName = "ops",
|
||||
SiteEnqueuedAt = Google.Protobuf.WellKnownTypes.Timestamp.FromDateTime(
|
||||
DateTime.SpecifyKind(new DateTime(2026, 5, 20, 10, 0, 0), DateTimeKind.Utc)),
|
||||
};
|
||||
|
||||
private static HeartbeatDto NewHeartbeatDto() => new()
|
||||
{
|
||||
SiteId = "site-a",
|
||||
NodeHostname = "node-a",
|
||||
IsActive = true,
|
||||
Timestamp = Google.Protobuf.WellKnownTypes.Timestamp.FromDateTime(
|
||||
DateTime.SpecifyKind(new DateTime(2026, 5, 20, 10, 0, 0), DateTimeKind.Utc)),
|
||||
};
|
||||
|
||||
private static AuditEventDto NewAuditDto() => new()
|
||||
{
|
||||
EventId = Guid.NewGuid().ToString(),
|
||||
OccurredAtUtc = Google.Protobuf.WellKnownTypes.Timestamp.FromDateTime(
|
||||
DateTime.SpecifyKind(new DateTime(2026, 5, 20, 10, 0, 0), DateTimeKind.Utc)),
|
||||
Channel = "ApiOutbound",
|
||||
Kind = "ApiCall",
|
||||
Status = "Delivered",
|
||||
SourceSiteId = "site-a",
|
||||
};
|
||||
|
||||
private sealed class StubActor : ReceiveActor
|
||||
{
|
||||
public StubActor()
|
||||
{
|
||||
Receive<NotificationSubmit>(msg =>
|
||||
Sender.Tell(new NotificationSubmitAck(msg.NotificationId, Accepted: true, Error: null)));
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>Swallows every message and never replies, so an Ask against it times out.</summary>
|
||||
private sealed class NeverRepliesActor : ReceiveActor
|
||||
{
|
||||
public NeverRepliesActor() => ReceiveAny(_ => { });
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user