fix(comms): fail known-dead sends immediately instead of burning Ask timeouts

This commit is contained in:
Joseph Doherty
2026-08-14 19:49:00 -04:00
parent ee193cd2bb
commit c5e66ed4e4
4 changed files with 120 additions and 12 deletions
@@ -24,8 +24,10 @@ public interface ISiteCommandTransport
/// Routes <paramref name="envelope"/>'s message to its site. Any reply the site produces is /// Routes <paramref name="envelope"/>'s message to its site. Any reply the site produces is
/// delivered to <paramref name="replyTo"/> — for an <c>Ask</c> that is the temporary ask actor /// delivered to <paramref name="replyTo"/> — for an <c>Ask</c> that is the temporary ask actor
/// (completing the caller's task); for a <c>Tell</c>-with-sender (the debug bridge) that is the /// (completing the caller's task); for a <c>Tell</c>-with-sender (the debug bridge) that is the
/// originating actor. A message with no route (an unknown site) is warned and dropped so the /// originating actor. A message with no route (an unknown/unconfigured site) fails fast: a
/// caller's <c>Ask</c> times out — central never buffers for an unreachable site. /// <see cref="Status.Failure"/> is delivered to <paramref name="replyTo"/> immediately instead of
/// letting the caller's <c>Ask</c> burn its full timeout — central still never buffers for an
/// unreachable site.
/// </summary> /// </summary>
/// <param name="envelope">The site-addressed command envelope.</param> /// <param name="envelope">The site-addressed command envelope.</param>
/// <param name="replyTo">Where a reply (or a <see cref="Status.Failure"/>) is delivered.</param> /// <param name="replyTo">Where a reply (or a <see cref="Status.Failure"/>) is delivered.</param>
@@ -22,9 +22,10 @@ namespace ZB.MOM.WW.ScadaBridge.Communication.Grpc;
/// <remarks> /// <remarks>
/// <para> /// <para>
/// <b>Per-call deadlines match today's Ask timeouts exactly</b> — see <see cref="ResolveDeadline"/>. /// <b>Per-call deadlines match today's Ask timeouts exactly</b> — see <see cref="ResolveDeadline"/>.
/// Behaviour is otherwise unchanged from the Akka path: an unknown/unconfigured site is warned and /// An unknown/unconfigured site now fails fast: it surfaces to the caller as a
/// dropped (the caller's Ask times out), and a transport fault surfaces to the caller as a /// <see cref="Status.Failure"/> immediately (WP1.6) instead of being warned-and-dropped to let the
/// <see cref="Status.Failure"/>, which the S&amp;F/audit layers already treat as transient. /// caller's Ask burn its full timeout. A transport fault takes the same <see cref="Status.Failure"/>
/// path, which the S&amp;F/audit layers already treat as transient.
/// </para> /// </para>
/// <para> /// <para>
/// <b>Cross-node retry is the channel provider's job</b> and happens only on /// <b>Cross-node retry is the channel provider's job</b> and happens only on
@@ -79,13 +80,21 @@ public sealed class GrpcSiteTransport : ISiteCommandTransport
replyTo.Tell(reply, ActorRefs.NoSender); 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 // WP1.6: a known-dead send (no configured channel for this site) used to warn-and-drop,
// Ask times out. Central never buffers. // 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( _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); envelope.SiteId, envelope.Message.GetType().Name);
if (!fireAndForget && !replyTo.IsNobody())
{
replyTo.Tell(new Status.Failure(ex), ActorRefs.NoSender);
}
} }
catch (Exception ex) catch (Exception ex)
{ {
@@ -8,9 +8,11 @@ namespace ZB.MOM.WW.ScadaBridge.Communication.Grpc;
/// <summary> /// <summary>
/// Raised when a site has no usable gRPC channel — an unknown site, or a site with neither /// Raised when a site has no usable gRPC channel — an unknown site, or a site with neither
/// <c>GrpcNodeAAddress</c> nor <c>GrpcNodeBAddress</c> configured. The gRPC transport treats this /// <c>GrpcNodeAAddress</c> nor <c>GrpcNodeBAddress</c> configured. <see cref="GrpcSiteTransport"/>
/// the way the removed Akka path treated "no ClusterClient for site": warn and drop, so the caller's Ask /// treats this as a known-dead send: it fails the caller's <c>Ask</c> fast with an
/// times out (central never buffers). /// <c>Akka.Actor.Status.Failure</c> 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).
/// </summary> /// </summary>
public sealed class SiteChannelUnavailableException(string siteId) public sealed class SiteChannelUnavailableException(string siteId)
: Exception($"No gRPC channel is configured for site '{siteId}'.") : Exception($"No gRPC channel is configured for site '{siteId}'.")
@@ -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;
/// <summary>
/// WP1.6: a send to a site with no configured gRPC channel must fail the caller's Ask immediately
/// (<see cref="Status.Failure"/> wrapping <see cref="SiteChannelUnavailableException"/>) instead of
/// warning-and-dropping the message and leaving the Ask to expire at its full timeout.
/// </summary>
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<SitePairChannelProvider>.Instance),
Opts,
NullLogger<GrpcSiteTransport>.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<Status.Failure>(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<SiteChannelUnavailableException>(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<SiteChannelUnavailableException>(async () =>
await askActor.Ask<object>(
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}");
}
/// <summary>Relays an Ask into the transport, exactly as <c>CentralCommunicationActor</c> does.</summary>
private sealed class EchoingAskTarget : ReceiveActor
{
public EchoingAskTarget(GrpcSiteTransport transport)
{
Receive<SiteEnvelope>(env => transport.Send(env, Sender));
}
}
private sealed class NoKeyProvider : ISitePskProvider
{
public ValueTask<string> GetAsync(string siteId, CancellationToken ct) => new("k");
public void Invalidate(string siteId) { }
}
}