Joseph Doherty
86ad4d5c8e
Extract the central→site send path in CentralCommunicationActor behind a new ISiteCommandTransport, selected by ScadaBridge:Communication:SiteTransport (Akka | Grpc, default Akka — rollback = flip the flag). CommunicationService's 27 commands, SiteCallAuditActor's 2 parked relays and DebugStreamBridgeActor's subscribe/unsubscribe are untouched; the seam sits below SiteEnvelope. - AkkaSiteTransport: today's per-site ClusterClient path extracted verbatim (the _siteClients lookup + ClusterClient.Send with the reply-to sender preserved, and the "no client ⇒ warn + drop, caller's Ask times out" path). - GrpcSiteTransport: dials the site SiteCommandService (T1B.1 proto client) via SiteCommandDtoMapper, PSK + x-scadabridge-site on the channel through ControlPlaneCredentials, per-command deadlines set EQUAL to today's CommunicationService Ask timeouts (per-command, not per-group: DeploymentState query and TriggerSiteFailover use QueryTimeout; the two parked relays map to QueryTimeout so SiteCallAudit's inner RelayTimeout 10s < 30s ordering holds; WaitForAttribute keeps its dynamic Timeout + IntegrationTimeout). - SitePairChannelProvider: per-site A/B channel pair with sticky failover (flip only on Unavailable — NEVER on DeadlineExceeded, a write/deploy/failover may have run), background failback probe to the preferred node with 1s→60s doubling backoff, PSK invalidation on site removal. Fed by the SAME DB refresh loop (extended to carry GrpcNodeA/GrpcNodeBAddress) — no second poll. Tests: actor-with-substitute-transport (routing, Ask-reply plumbing, per-site lifecycle across refreshes), ResolveDeadline pinned to each command's current Ask timeout, and GrpcSiteTransport/SitePairChannelProvider over dual in-process TestServers (PSK+header+deadline attached, Unavailable failover + stickiness, failback to preferred, no-retry-on-DeadlineExceeded). Proto csproj untouched (no active <Protobuf> item). Full solution builds 0 warnings; Communication.Tests 607 green with Akka default.
108 lines
5.7 KiB
C#
108 lines
5.7 KiB
C#
using Microsoft.Extensions.Logging.Abstractions;
|
|
using Microsoft.Extensions.Options;
|
|
using ZB.MOM.WW.ScadaBridge.Commons.Messages.Artifacts;
|
|
using ZB.MOM.WW.ScadaBridge.Commons.Messages.DataConnection;
|
|
using ZB.MOM.WW.ScadaBridge.Commons.Messages.DebugView;
|
|
using ZB.MOM.WW.ScadaBridge.Commons.Messages.Deployment;
|
|
using ZB.MOM.WW.ScadaBridge.Commons.Messages.InboundApi;
|
|
using ZB.MOM.WW.ScadaBridge.Commons.Messages.Lifecycle;
|
|
using ZB.MOM.WW.ScadaBridge.Commons.Messages.Management;
|
|
using ZB.MOM.WW.ScadaBridge.Commons.Messages.RemoteQuery;
|
|
using ZB.MOM.WW.ScadaBridge.Commons.Types;
|
|
using ZB.MOM.WW.ScadaBridge.Communication.Grpc;
|
|
|
|
namespace ZB.MOM.WW.ScadaBridge.Communication.Tests;
|
|
|
|
/// <summary>
|
|
/// Pins <see cref="GrpcSiteTransport.ResolveDeadline"/> to the EXACT Ask timeout
|
|
/// <see cref="CommunicationService"/> uses for each command today, so flipping the transport cannot
|
|
/// change how long any call waits. Every timeout is given a distinct value so a wrong mapping is
|
|
/// caught, not masked by two defaults that happen to be equal (QueryTimeout == IntegrationTimeout in
|
|
/// production).
|
|
/// </summary>
|
|
public class GrpcSiteTransportDeadlineTests
|
|
{
|
|
private static readonly CommunicationOptions Opts = new()
|
|
{
|
|
DeploymentTimeout = TimeSpan.FromSeconds(120),
|
|
LifecycleTimeout = TimeSpan.FromSeconds(30),
|
|
ArtifactDeploymentTimeout = TimeSpan.FromSeconds(60),
|
|
QueryTimeout = TimeSpan.FromSeconds(31),
|
|
IntegrationTimeout = TimeSpan.FromSeconds(32),
|
|
DebugViewTimeout = TimeSpan.FromSeconds(10)
|
|
};
|
|
|
|
private static readonly GrpcSiteTransport Transport = new(
|
|
new SitePairChannelProvider(
|
|
new NoKeyProvider(), Options.Create(Opts), NullLogger<SitePairChannelProvider>.Instance),
|
|
Opts,
|
|
NullLogger<GrpcSiteTransport>.Instance);
|
|
|
|
private static readonly DateTimeOffset T = DateTimeOffset.UtcNow;
|
|
|
|
public static IEnumerable<object[]> Cases()
|
|
{
|
|
// Lifecycle group — note it is NOT one deadline class.
|
|
yield return [new RefreshDeploymentCommand("d", "i", "h", "by", T, "u", "t"), Opts.DeploymentTimeout];
|
|
yield return [new EnableInstanceCommand("c", "i", T), Opts.LifecycleTimeout];
|
|
yield return [new DisableInstanceCommand("c", "i", T), Opts.LifecycleTimeout];
|
|
yield return [new DeleteInstanceCommand("c", "i", T), Opts.LifecycleTimeout];
|
|
// DeploymentStateQuery uses QueryTimeout in CommunicationService, NOT LifecycleTimeout.
|
|
yield return [new DeploymentStateQueryRequest("c", "i", T), Opts.QueryTimeout];
|
|
yield return [new DeployArtifactsCommand("d", null, null, null, null, null, null, T), Opts.ArtifactDeploymentTimeout];
|
|
|
|
// OPC UA group — all QueryTimeout.
|
|
yield return [new BrowseNodeCommand("conn", null, null, null), Opts.QueryTimeout];
|
|
yield return [new SearchAddressSpaceCommand("conn", "q", 1, 1, null), Opts.QueryTimeout];
|
|
yield return [new ReadTagValuesCommand("conn", []), Opts.QueryTimeout];
|
|
yield return [new VerifyEndpointCommand("conn", "OpcUa", "{}", null), Opts.QueryTimeout];
|
|
yield return [new TrustServerCertCommand("conn", "ZGVy", "AA", null), Opts.QueryTimeout];
|
|
yield return [new ListServerCertsCommand(null), Opts.QueryTimeout];
|
|
yield return [new RemoveServerCertCommand("AA", null), Opts.QueryTimeout];
|
|
yield return [new WriteTagRequest("c", "conn", "tag", 1, T), Opts.QueryTimeout];
|
|
|
|
// Query group.
|
|
yield return [new EventLogQueryRequest("c", "s", null, null, null, null, null, null, null, 10, T), Opts.QueryTimeout];
|
|
yield return [new DebugSnapshotRequest("i", "c"), Opts.QueryTimeout];
|
|
yield return [new SubscribeDebugViewRequest("i", "c"), Opts.DebugViewTimeout];
|
|
yield return [new UnsubscribeDebugViewRequest("i", "c"), Opts.DebugViewTimeout];
|
|
|
|
// Parked group — the two relays keep QueryTimeout (30s) so SiteCallAudit's inner
|
|
// RelayTimeout (10s) still expires first.
|
|
yield return [new ParkedMessageQueryRequest("c", "s", 1, 10, T), Opts.QueryTimeout];
|
|
yield return [new ParkedMessageRetryRequest("c", "s", "m", T), Opts.QueryTimeout];
|
|
yield return [new ParkedMessageDiscardRequest("c", "s", "m", T), Opts.QueryTimeout];
|
|
yield return [new RetryParkedOperation("c", new TrackedOperationId(Guid.NewGuid())), Opts.QueryTimeout];
|
|
yield return [new DiscardParkedOperation("c", new TrackedOperationId(Guid.NewGuid())), Opts.QueryTimeout];
|
|
|
|
// Route group.
|
|
yield return [new RouteToCallRequest("c", "i", "m", null, T), Opts.IntegrationTimeout];
|
|
yield return [new RouteToGetAttributesRequest("c", "i", [], T), Opts.IntegrationTimeout];
|
|
yield return [new RouteToSetAttributesRequest("c", "i", new Dictionary<string, string>(), T), Opts.IntegrationTimeout];
|
|
|
|
// Failover — QueryTimeout in CommunicationService, NOT LifecycleTimeout.
|
|
yield return [new TriggerSiteFailover("c", "s"), Opts.QueryTimeout];
|
|
}
|
|
|
|
[Theory]
|
|
[MemberData(nameof(Cases))]
|
|
public void ResolveDeadline_MatchesTodaysAskTimeout(object command, TimeSpan expected)
|
|
{
|
|
Assert.Equal(expected, Transport.ResolveDeadline(command));
|
|
}
|
|
|
|
[Fact]
|
|
public void WaitForAttribute_UsesItsDynamicTimeoutPlusIntegrationSlack()
|
|
{
|
|
// CommunicationService.RouteToWaitForAttributeAsync uses request.Timeout + IntegrationTimeout.
|
|
var wait = new RouteToWaitForAttributeRequest("c", "i", "attr", "10", TimeSpan.FromSeconds(45), T);
|
|
Assert.Equal(TimeSpan.FromSeconds(45) + Opts.IntegrationTimeout, Transport.ResolveDeadline(wait));
|
|
}
|
|
|
|
private sealed class NoKeyProvider : ISitePskProvider
|
|
{
|
|
public ValueTask<string> GetAsync(string siteId, CancellationToken ct) => new("k");
|
|
public void Invalidate(string siteId) { }
|
|
}
|
|
}
|