From c5e66ed4e4788152f5c95be2846f61a7df23a5ca Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Fri, 14 Aug 2026 19:49:00 -0400 Subject: [PATCH] fix(comms): fail known-dead sends immediately instead of burning Ask timeouts --- .../Actors/ISiteCommandTransport.cs | 6 +- .../Grpc/GrpcSiteTransport.cs | 23 +++-- .../Grpc/SitePairChannelProvider.cs | 8 +- .../GrpcSiteTransportFailFastTests.cs | 95 +++++++++++++++++++ 4 files changed, 120 insertions(+), 12 deletions(-) create mode 100644 tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/GrpcSiteTransportFailFastTests.cs diff --git a/src/ZB.MOM.WW.ScadaBridge.Communication/Actors/ISiteCommandTransport.cs b/src/ZB.MOM.WW.ScadaBridge.Communication/Actors/ISiteCommandTransport.cs index 89545aa6..1b3efc0f 100644 --- a/src/ZB.MOM.WW.ScadaBridge.Communication/Actors/ISiteCommandTransport.cs +++ b/src/ZB.MOM.WW.ScadaBridge.Communication/Actors/ISiteCommandTransport.cs @@ -24,8 +24,10 @@ public interface ISiteCommandTransport /// Routes 's message to its site. Any reply the site produces is /// delivered to — for an Ask that is the temporary ask actor /// (completing the caller's task); for a Tell-with-sender (the debug bridge) that is the - /// originating actor. A message with no route (an unknown site) is warned and dropped so the - /// caller's Ask times out — central never buffers for an unreachable site. + /// originating actor. A message with no route (an unknown/unconfigured site) fails fast: a + /// is delivered to immediately instead of + /// letting the caller's Ask burn its full timeout — central still never buffers for an + /// unreachable site. /// /// The site-addressed command envelope. /// Where a reply (or a ) is delivered. diff --git a/src/ZB.MOM.WW.ScadaBridge.Communication/Grpc/GrpcSiteTransport.cs b/src/ZB.MOM.WW.ScadaBridge.Communication/Grpc/GrpcSiteTransport.cs index 33322fe3..7d3bffad 100644 --- a/src/ZB.MOM.WW.ScadaBridge.Communication/Grpc/GrpcSiteTransport.cs +++ b/src/ZB.MOM.WW.ScadaBridge.Communication/Grpc/GrpcSiteTransport.cs @@ -22,9 +22,10 @@ namespace ZB.MOM.WW.ScadaBridge.Communication.Grpc; /// /// /// Per-call deadlines match today's Ask timeouts exactly — see . -/// Behaviour is otherwise unchanged from the Akka path: an unknown/unconfigured site is warned and -/// dropped (the caller's Ask times out), and a transport fault surfaces to the caller as a -/// , which the S&F/audit layers already treat as transient. +/// An unknown/unconfigured site now fails fast: it surfaces to the caller as a +/// immediately (WP1.6) instead of being warned-and-dropped to let the +/// caller's Ask burn its full timeout. A transport fault takes the same +/// path, which the S&F/audit layers already treat as transient. /// /// /// Cross-node retry is the channel provider's job and happens only on @@ -79,13 +80,21 @@ public sealed class GrpcSiteTransport : ISiteCommandTransport replyTo.Tell(reply, ActorRefs.NoSender); } } - catch (SiteChannelUnavailableException) + catch (SiteChannelUnavailableException ex) { - // Parity with the removed Akka "no ClusterClient for site" path: warn and drop, so the caller's - // Ask times out. Central never buffers. + // WP1.6: a known-dead send (no configured channel for this site) used to warn-and-drop, + // leaving the caller's Ask to burn its full timeout for a failure we already know about. + // Fail fast instead — same Status.Failure completion path a transport fault takes below — + // so S&F/audit treat it as transient immediately rather than tens of seconds later. + // Central still never buffers for an unreachable site; only the completion timing changed. _logger.LogWarning( - "No gRPC channel for site {SiteId}; dropping {Message} (caller's Ask will time out)", + "No gRPC channel for site {SiteId}; failing {Message} immediately instead of dropping it", envelope.SiteId, envelope.Message.GetType().Name); + + if (!fireAndForget && !replyTo.IsNobody()) + { + replyTo.Tell(new Status.Failure(ex), ActorRefs.NoSender); + } } catch (Exception ex) { diff --git a/src/ZB.MOM.WW.ScadaBridge.Communication/Grpc/SitePairChannelProvider.cs b/src/ZB.MOM.WW.ScadaBridge.Communication/Grpc/SitePairChannelProvider.cs index f2cf359e..b8128f9f 100644 --- a/src/ZB.MOM.WW.ScadaBridge.Communication/Grpc/SitePairChannelProvider.cs +++ b/src/ZB.MOM.WW.ScadaBridge.Communication/Grpc/SitePairChannelProvider.cs @@ -8,9 +8,11 @@ namespace ZB.MOM.WW.ScadaBridge.Communication.Grpc; /// /// Raised when a site has no usable gRPC channel — an unknown site, or a site with neither -/// GrpcNodeAAddress nor GrpcNodeBAddress configured. The gRPC transport treats this -/// the way the removed Akka path treated "no ClusterClient for site": warn and drop, so the caller's Ask -/// times out (central never buffers). +/// GrpcNodeAAddress nor GrpcNodeBAddress configured. +/// treats this as a known-dead send: it fails the caller's Ask fast with an +/// Akka.Actor.Status.Failure wrapping this exception (WP1.6) instead of warning and dropping +/// the message, so the caller does not burn its full Ask timeout on a failure already known at send +/// time (central still never buffers). /// public sealed class SiteChannelUnavailableException(string siteId) : Exception($"No gRPC channel is configured for site '{siteId}'.") diff --git a/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/GrpcSiteTransportFailFastTests.cs b/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/GrpcSiteTransportFailFastTests.cs new file mode 100644 index 00000000..fb934e0c --- /dev/null +++ b/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/GrpcSiteTransportFailFastTests.cs @@ -0,0 +1,95 @@ +using System.Diagnostics; +using Akka.Actor; +using Akka.TestKit.Xunit2; +using Microsoft.Extensions.Logging.Abstractions; +using Microsoft.Extensions.Options; +using ZB.MOM.WW.ScadaBridge.Commons.Messages.Lifecycle; +using ZB.MOM.WW.ScadaBridge.Communication.Grpc; + +namespace ZB.MOM.WW.ScadaBridge.Communication.Tests; + +/// +/// WP1.6: a send to a site with no configured gRPC channel must fail the caller's Ask immediately +/// ( wrapping ) instead of +/// warning-and-dropping the message and leaving the Ask to expire at its full timeout. +/// +public class GrpcSiteTransportFailFastTests : TestKit +{ + private static readonly CommunicationOptions Opts = new() + { + DeploymentTimeout = TimeSpan.FromSeconds(120), + LifecycleTimeout = TimeSpan.FromSeconds(30), + ArtifactDeploymentTimeout = TimeSpan.FromSeconds(60), + QueryTimeout = TimeSpan.FromSeconds(30), + IntegrationTimeout = TimeSpan.FromSeconds(30), + DebugViewTimeout = TimeSpan.FromSeconds(10) + }; + + private static GrpcSiteTransport BuildTransport() => new( + new SitePairChannelProvider( + new NoKeyProvider(), Options.Create(Opts), NullLogger.Instance), + Opts, + NullLogger.Instance); + + [Fact] + public void Send_ToUnconfiguredSite_FaultsTheAskFastInsteadOfTimingOut() + { + // No ReconcileSites/UpdateSite call was ever made for this site — exactly the "unknown + // site" shape SitePairChannelProvider throws SiteChannelUnavailableException for. + var transport = BuildTransport(); + var probe = CreateTestProbe(); + var envelope = new SiteEnvelope( + "unconfigured-site", + new EnableInstanceCommand("cmd-1", "Site1.Pump1", DateTimeOffset.UtcNow)); + + var stopwatch = Stopwatch.StartNew(); + transport.Send(envelope, probe.Ref); + + // The old warn-and-drop behavior would leave the caller silent until its own Ask timeout + // (30s in production). Well under 1s here proves the fail-fast path, not a lucky race. + var failure = probe.ExpectMsg(TimeSpan.FromSeconds(1)); + stopwatch.Stop(); + + Assert.True(stopwatch.Elapsed < TimeSpan.FromSeconds(1), + $"expected the Ask to fault well under 1s; took {stopwatch.Elapsed}"); + var cause = Assert.IsType(failure.Cause); + Assert.Equal("unconfigured-site", cause.SiteId); + Assert.Contains("unconfigured-site", cause.Message); + } + + [Fact] + public async Task Ask_AgainstUnconfiguredSite_CompletesFaultedFastViaTellStatusFailure() + { + // Mirrors how CommunicationService actually calls this seam: Ask a temp actor, transport + // Tells it back. Proves the fail-fast reply satisfies a real Akka Ask, not just a TestProbe. + var transport = BuildTransport(); + var askActor = Sys.ActorOf(Props.Create(() => new EchoingAskTarget(transport))); + + var stopwatch = Stopwatch.StartNew(); + var ex = await Assert.ThrowsAsync(async () => + await askActor.Ask( + new SiteEnvelope("unconfigured-site", new EnableInstanceCommand( + "cmd-1", "Site1.Pump1", DateTimeOffset.UtcNow)), + TimeSpan.FromSeconds(30))); + stopwatch.Stop(); + + Assert.Equal("unconfigured-site", ex.SiteId); + Assert.True(stopwatch.Elapsed < TimeSpan.FromSeconds(1), + $"expected the Ask to fault well under the 30s timeout; took {stopwatch.Elapsed}"); + } + + /// Relays an Ask into the transport, exactly as CentralCommunicationActor does. + private sealed class EchoingAskTarget : ReceiveActor + { + public EchoingAskTarget(GrpcSiteTransport transport) + { + Receive(env => transport.Send(env, Sender)); + } + } + + private sealed class NoKeyProvider : ISitePskProvider + { + public ValueTask GetAsync(string siteId, CancellationToken ct) => new("k"); + public void Invalidate(string siteId) { } + } +}