diff --git a/src/ZB.MOM.WW.ScadaBridge.Communication/Grpc/GrpcCentralTransport.cs b/src/ZB.MOM.WW.ScadaBridge.Communication/Grpc/GrpcCentralTransport.cs index f914d7bc..a942dd08 100644 --- a/src/ZB.MOM.WW.ScadaBridge.Communication/Grpc/GrpcCentralTransport.cs +++ b/src/ZB.MOM.WW.ScadaBridge.Communication/Grpc/GrpcCentralTransport.cs @@ -28,7 +28,8 @@ namespace ZB.MOM.WW.ScadaBridge.Communication.Grpc; /// /// Cross-node retry is operation-specific. Read-only status and idempotent notification, /// audit and sequence-guarded health ingestion may retry transient transport failures. Reconcile -/// mints tokens and retries only an explicit server pre-dispatch rejection. Ambiguous reconcile +/// mints tokens and retries only an explicit server pre-dispatch rejection or a request that was +/// provably never sent (connect-phase failure, ScadaBridge#136). Ambiguous reconcile /// failures return OUTCOME_UNKNOWN. Both attempts share one absolute deadline. /// /// @@ -182,12 +183,13 @@ public sealed class GrpcCentralTransport : ICentralTransport catch (RpcException ex) when (IsConnectFailure(ex)) { _channels.ReportUnavailable(index); - if (!retrySafe && !GrpcRetrySafety.WasNotDispatched(ex)) + if (!retrySafe && !GrpcRetrySafety.ProvablyNotApplied(ex)) { replyTo.Tell(new AkkaStatus.Failure(GrpcRetrySafety.OutcomeUnknown(ex))); return; } - // Retry only a read/idempotent ingest or an explicit server pre-dispatch rejection. + // Retry only a read/idempotent ingest, an explicit server pre-dispatch rejection, or a + // connect-phase failure that never sent the request (ScadaBridge#136). var (retryIndex, retryClient) = _channels.Current(); if (retryIndex != index) { @@ -202,7 +204,7 @@ public sealed class GrpcCentralTransport : ICentralTransport _logger.LogWarning(retryEx, "Central control-plane call failed on both endpoints; surfacing as transient."); var reported = retryEx is RpcException rpc && IsConnectFailure(rpc) - && !retrySafe && !GrpcRetrySafety.WasNotDispatched(rpc) + && !retrySafe && !GrpcRetrySafety.ProvablyNotApplied(rpc) ? GrpcRetrySafety.OutcomeUnknown(rpc) : retryEx; replyTo.Tell(new AkkaStatus.Failure(reported)); return; diff --git a/src/ZB.MOM.WW.ScadaBridge.Communication/Grpc/GrpcRetrySafety.cs b/src/ZB.MOM.WW.ScadaBridge.Communication/Grpc/GrpcRetrySafety.cs index 5c58d716..1ee6d01c 100644 --- a/src/ZB.MOM.WW.ScadaBridge.Communication/Grpc/GrpcRetrySafety.cs +++ b/src/ZB.MOM.WW.ScadaBridge.Communication/Grpc/GrpcRetrySafety.cs @@ -1,3 +1,5 @@ +using System.Net.Http; +using System.Net.Sockets; using Grpc.Core; namespace ZB.MOM.WW.ScadaBridge.Communication.Grpc; @@ -20,6 +22,67 @@ public static class GrpcRetrySafety public static bool WasNotDispatched(RpcException error) => error.StatusCode == StatusCode.Unavailable && error.Trailers.Any(e => e.Key == DispatchTrailer && e.Value == "not-started"); + /// + /// Status detail Grpc.Net.Client sets when its load balancer could not connect a subchannel + /// and failed the call at PICK time — before any HTTP request for the call was created. + /// + internal const string SubchannelConnectFailureDetail = "Error connecting to subchannel."; + + /// + /// True only when the client failed while OPENING a connection, so the request provably never + /// left this process (ScadaBridge#136 — the first command to a stopped node of a site pair). + /// Two shapes qualify, both with a + /// in the debug-exception chain: + /// + /// Grpc.Net.Client's subchannel connect failure (): + /// the balancer never handed the call a connection, whatever the socket error. + /// An classified as a name-resolution or connection + /// error whose socket error can only arise before a connection exists (refused, host not + /// found, DNS try-again / no-data). + /// + /// Every other transport failure — a reset, GOAWAY, keepalive timeout or premature end on an + /// established connection — may have delivered the request and stays ambiguous. + /// + /// The RPC failure to inspect. + /// true when the request was never sent. + public static bool WasNeverSent(RpcException error) + { + if (error.StatusCode != StatusCode.Unavailable) + { + return false; + } + + HttpRequestException? http = null; + SocketException? socket = null; + for (var e = error.Status.DebugException; e is not null; e = e.InnerException) + { + http ??= e as HttpRequestException; + socket ??= e as SocketException; + } + + if (socket is null) + { + return false; + } + + if (http is null) + { + return string.Equals(error.Status.Detail, SubchannelConnectFailureDetail, StringComparison.Ordinal); + } + + return http.HttpRequestError is HttpRequestError.NameResolutionError or HttpRequestError.ConnectionError + && socket.SocketErrorCode is SocketError.ConnectionRefused or SocketError.HostNotFound + or SocketError.TryAgain or SocketError.NoData; + } + + /// + /// True when a mutation may be replayed on the peer node: the server rejected it before + /// dispatch (), or the client never sent it (). + /// + /// The RPC failure to inspect. + /// true when no node can have applied the request. + public static bool ProvablyNotApplied(RpcException error) => WasNotDispatched(error) || WasNeverSent(error); + /// Reports an ambiguous mutation without inviting a blind replay. /// The original RPC failure whose trailers are preserved. /// An with Aborted status marking the outcome as unknown. diff --git a/src/ZB.MOM.WW.ScadaBridge.Communication/Grpc/SitePairChannelProvider.cs b/src/ZB.MOM.WW.ScadaBridge.Communication/Grpc/SitePairChannelProvider.cs index 2e8a5e0b..d1330ecf 100644 --- a/src/ZB.MOM.WW.ScadaBridge.Communication/Grpc/SitePairChannelProvider.cs +++ b/src/ZB.MOM.WW.ScadaBridge.Communication/Grpc/SitePairChannelProvider.cs @@ -35,7 +35,8 @@ public sealed class SiteChannelUnavailableException(string siteId) /// /// /// Retry safety. Unavailable can mean a lost response after application. Only explicitly -/// retry-safe operations or a server-owned not-started trailer permit replay on the peer. +/// retry-safe operations, a server-owned not-started trailer, or a connect-phase failure that +/// provably never sent the request () permit replay on the peer. /// Ambiguous mutations fail with Aborted / OUTCOME_UNKNOWN; the next independent call uses the /// peer. Deadlines are never replayed. Readiness rejection remains safe before actor dispatch. /// @@ -324,16 +325,20 @@ public sealed class SitePairChannelProvider : IDisposable catch (RpcException ex) when (ex.StatusCode == StatusCode.Unavailable) { if (secondary is not null) FlipTo(pair, toIsA: !primaryIsA); - if (!retrySafe && !GrpcRetrySafety.WasNotDispatched(ex)) + // ScadaBridge#136: a connect-phase failure (refused / unresolvable — the stopped node of + // a pair) provably never sent the request, so a mutation fails over within this call + // just like an explicit pre-dispatch rejection. Anything that may have been delivered + // stays OUTCOME_UNKNOWN and is never replayed. + if (!retrySafe && !GrpcRetrySafety.ProvablyNotApplied(ex)) throw GrpcRetrySafety.OutcomeUnknown(ex); if (secondary is null) throw; - // Only reads/idempotent operations or an explicit pre-dispatch rejection may replay. + // Only reads/idempotent operations or a provably unapplied request may replay. _logger.LogWarning( "Site {SiteId} node {FromNode} unavailable; failing over to node {ToNode}", siteId, primaryIsA ? "A" : "B", primaryIsA ? "B" : "A"); try { return await call(secondary, ct).ConfigureAwait(false); } catch (RpcException retryError) when (retryError.StatusCode == StatusCode.Unavailable - && !retrySafe && !GrpcRetrySafety.WasNotDispatched(retryError)) + && !retrySafe && !GrpcRetrySafety.ProvablyNotApplied(retryError)) { throw GrpcRetrySafety.OutcomeUnknown(retryError); } diff --git a/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/Grpc/GrpcRetrySafetyNeverSentTests.cs b/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/Grpc/GrpcRetrySafetyNeverSentTests.cs new file mode 100644 index 00000000..61caf93a --- /dev/null +++ b/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/Grpc/GrpcRetrySafetyNeverSentTests.cs @@ -0,0 +1,73 @@ +using System.Net.Http; +using System.Net.Sockets; +using Grpc.Core; +using ZB.MOM.WW.ScadaBridge.Communication.Grpc; + +namespace ZB.MOM.WW.ScadaBridge.Communication.Tests.Grpc; + +/// +/// ScadaBridge#136: admits only connect-phase failures +/// (the request provably never left the client) and rejects every shape that may have delivered it. +/// +public sealed class GrpcRetrySafetyNeverSentTests +{ + private static RpcException Rpc(StatusCode code, string detail, Exception? debug) + => new(new Status(code, detail, debug)); + + private static SocketException Sock(SocketError error) => new((int)error); + + /// The balancer's subchannel connect failure is pre-send, whatever the socket error. + /// The socket error the subchannel connect saw. + [Theory] + [InlineData(SocketError.ConnectionRefused)] + [InlineData(SocketError.HostNotFound)] + [InlineData(SocketError.TryAgain)] + [InlineData(SocketError.HostUnreachable)] + [InlineData(SocketError.TimedOut)] + public void SubchannelConnectFailure_IsNeverSent(SocketError error) + { + var ex = Rpc(StatusCode.Unavailable, GrpcRetrySafety.SubchannelConnectFailureDetail, Sock(error)); + + Assert.True(GrpcRetrySafety.WasNeverSent(ex)); + Assert.True(GrpcRetrySafety.ProvablyNotApplied(ex)); + } + + /// An HttpRequestException connect error with a connect-only socket error is pre-send. + /// The HTTP request error classification. + /// The socket error. + [Theory] + [InlineData(HttpRequestError.ConnectionError, SocketError.ConnectionRefused)] + [InlineData(HttpRequestError.NameResolutionError, SocketError.HostNotFound)] + [InlineData(HttpRequestError.NameResolutionError, SocketError.TryAgain)] + public void HttpConnectError_WithConnectOnlySocketError_IsNeverSent(HttpRequestError kind, SocketError error) + { + var ex = Rpc(StatusCode.Unavailable, "Error starting gRPC call.", + new HttpRequestException(kind, "connect failed", Sock(error))); + + Assert.True(GrpcRetrySafety.WasNeverSent(ex)); + } + + /// Shapes that may follow delivery stay ambiguous. + [Fact] + public void FailuresThatMayFollowDelivery_AreNotNeverSent() + { + // Premature end / reset on an established connection. + Assert.False(GrpcRetrySafety.WasNeverSent(Rpc(StatusCode.Unavailable, "Error starting gRPC call.", + new HttpRequestException(HttpRequestError.ResponseEnded, "ended", + new IOException("reset", Sock(SocketError.ConnectionReset)))))); + // A ConnectionError classification with a post-connect socket error. + Assert.False(GrpcRetrySafety.WasNeverSent(Rpc(StatusCode.Unavailable, "Error starting gRPC call.", + new HttpRequestException(HttpRequestError.ConnectionError, "x", Sock(SocketError.ConnectionReset))))); + // A server-sent Unavailable (no transport exception at all). + Assert.False(GrpcRetrySafety.WasNeverSent(Rpc(StatusCode.Unavailable, "modelled fault", null))); + // A bare socket error outside the subchannel-pick path. + Assert.False(GrpcRetrySafety.WasNeverSent(Rpc(StatusCode.Unavailable, "Error starting gRPC call.", + Sock(SocketError.ConnectionRefused)))); + // Subchannel detail without a socket cause. + Assert.False(GrpcRetrySafety.WasNeverSent(Rpc(StatusCode.Unavailable, + GrpcRetrySafety.SubchannelConnectFailureDetail, new InvalidOperationException()))); + // Not Unavailable. + Assert.False(GrpcRetrySafety.WasNeverSent(Rpc(StatusCode.DeadlineExceeded, + GrpcRetrySafety.SubchannelConnectFailureDetail, Sock(SocketError.ConnectionRefused)))); + } +} diff --git a/tests/ZB.MOM.WW.ScadaBridge.Host.Tests/GrpcSiteTransportFailoverTests.cs b/tests/ZB.MOM.WW.ScadaBridge.Host.Tests/GrpcSiteTransportFailoverTests.cs index f0539e60..928aea3d 100644 --- a/tests/ZB.MOM.WW.ScadaBridge.Host.Tests/GrpcSiteTransportFailoverTests.cs +++ b/tests/ZB.MOM.WW.ScadaBridge.Host.Tests/GrpcSiteTransportFailoverTests.cs @@ -1,3 +1,5 @@ +using System.Net; +using System.Net.Sockets; using Grpc.Core; using Grpc.Net.Client; using Microsoft.AspNetCore.Builder; @@ -32,6 +34,7 @@ public sealed class GrpcSiteTransportFailoverTests : IAsyncLifetime private StubSiteCommandService _stubB = null!; private SitePairChannelProvider _provider = null!; private volatile bool _preferredReachable; + private readonly Dictionary> _extraHandlers = new(StringComparer.Ordinal); /// Starts the stub servers and builds the channel provider under test. /// A task that represents the asynchronous operation. @@ -47,7 +50,9 @@ public sealed class GrpcSiteTransportFailoverTests : IAsyncLifetime new FixedPskProvider(SiteKey), Options.Create(new CommunicationOptions()), NullLogger.Instance, - handlerFactory: endpoint => endpoint == EndpointA ? handlerA : handlerB, + handlerFactory: endpoint => endpoint == EndpointA ? handlerA + : endpoint == EndpointB ? handlerB + : _extraHandlers[endpoint](), reachabilityProbe: (_, _) => Task.FromResult(_preferredReachable)); _provider.UpdateSite(SiteId, EndpointA, EndpointB); @@ -193,6 +198,93 @@ public sealed class GrpcSiteTransportFailoverTests : IAsyncLifetime Assert.Equal(0, _stubB.LifecycleCalls); } + /// + /// ScadaBridge#136: NodeA is a stopped node — nothing listens on its port, so the TCP connect is + /// refused before any request byte is written. A mutation (here a lifecycle command, the same + /// path as an instance deploy) must fail over to NodeB within the SAME call instead of returning + /// OUTCOME_UNKNOWN, and stay sticky on B. + /// + /// A task that represents the asynchronous operation. + [Fact] + public async Task ConnectionRefused_OnNodeA_MutationFailsOverToNodeB_WithinTheCall() + { + var refused = $"http://127.0.0.1:{ClosedLoopbackPort()}"; + _extraHandlers[refused] = () => new SocketsHttpHandler(); + _provider.UpdateSite(SiteId, refused, EndpointB); + + var reply = await CallAsync(); + + Assert.Equal(LifecycleReply.ReplyOneofCase.InstanceLifecycle, reply.ReplyCase); + Assert.Equal(1, _stubB.LifecycleCalls); + Assert.False(_provider.IsOnPreferredNode(SiteId)); + } + + /// + /// ScadaBridge#136, the docker-rig shape: a stopped container's hostname no longer resolves, so + /// the connect fails at name resolution. The request was never sent; the mutation fails over. + /// + /// A task that represents the asynchronous operation. + [Fact] + public async Task UnresolvableNodeA_MutationFailsOverToNodeB_WithinTheCall() + { + const string unresolvable = "http://scadabridge-136-stopped-node.invalid:8083"; + _extraHandlers[unresolvable] = () => new SocketsHttpHandler(); + _provider.UpdateSite(SiteId, unresolvable, EndpointB); + + var reply = await CallAsync(); + + Assert.Equal(LifecycleReply.ReplyOneofCase.InstanceLifecycle, reply.ReplyCase); + Assert.Equal(1, _stubB.LifecycleCalls); + Assert.False(_provider.IsOnPreferredNode(SiteId)); + } + + /// + /// ScadaBridge#136 guard: the request REACHED NodeA (its stub ran the command) and the connection + /// then died before the response — modelled by a handler that forwards to A and then throws the + /// transport's premature-end error. That outcome is genuinely unknown: it must surface as + /// OUTCOME_UNKNOWN and must NOT be replayed on NodeB. + /// + /// A task that represents the asynchronous operation. + [Fact] + public async Task ConnectionLostAfterDelivery_StaysOutcomeUnknown_AndIsNotReplayed() + { + const string dropsAfterSend = "http://node-a-drops-after-send"; + var inner = _hostA.GetTestServer().CreateHandler(); + _extraHandlers[dropsAfterSend] = () => new DropAfterDeliveryHandler(inner); + _provider.UpdateSite(SiteId, dropsAfterSend, EndpointB); + + var error = await Assert.ThrowsAsync(() => CallAsync()); + + Assert.Equal(StatusCode.Aborted, error.StatusCode); + Assert.Contains("OUTCOME_UNKNOWN", error.Status.Detail); + Assert.Equal(1, _stubA.LifecycleCalls); + Assert.Equal(0, _stubB.LifecycleCalls); + } + + /// A loopback port with nothing listening (bound, read, released). + private static int ClosedLoopbackPort() + { + using var listener = new TcpListener(IPAddress.Loopback, 0); + listener.Start(); + var port = ((IPEndPoint)listener.LocalEndpoint).Port; + listener.Stop(); + return port; + } + + /// Delivers the request to the real server, then fails as if the connection reset mid-response. + private sealed class DropAfterDeliveryHandler(HttpMessageHandler inner) : DelegatingHandler(inner) + { + /// + protected override async Task SendAsync(HttpRequestMessage request, CancellationToken ct) + { + using var delivered = await base.SendAsync(request, ct); + await delivered.Content.ReadAsByteArrayAsync(ct); + throw new HttpRequestException(HttpRequestError.ResponseEnded, + "The response ended prematurely.", new IOException("connection reset", + new SocketException((int)SocketError.ConnectionReset))); + } + } + /// Verifies the case: unknown Site, throws, site Channel Unavailable. /// A task that represents the asynchronous operation. [Fact]