fix(communication): fail a site command over within the call when the stopped node never received it (#136)

The first mutation (instance deploy) to a site pair whose sticky node was down
returned OUTCOME_UNKNOWN: SitePairChannelProvider.ExecuteAsync treated every
Unavailable on a non-retry-safe command as possibly-applied unless the server
sent the not-dispatched trailer, flipped stickiness, and threw. The retry then
succeeded on the live peer.

A connect-phase failure provably never sent the request, so it is now treated
like the pre-dispatch rejection: GrpcRetrySafety.WasNeverSent recognises
Grpc.Net.Client's subchannel connect failure ("Error connecting to subchannel."
with a SocketException cause) and an HttpRequestException name-resolution /
connection error with a connect-only socket error. Anything that may have been
delivered (reset, premature end, GOAWAY, server Unavailable) stays
OUTCOME_UNKNOWN and is not replayed. The site->central reconcile path uses the
same ProvablyNotApplied rule.

Claude-Session: https://claude.ai/code/session_01GT67ACmpMR7b9LGRPLijMi

src/ZB.MOM.WW.ScadaBridge.Communication/Grpc/GrpcCentralTransport.cs

@@ -28,7 +28,8 @@ namespace ZB.MOM.WW.ScadaBridge.Communication.Grpc;
 /// <para>
 /// <b>Cross-node retry is operation-specific.</b> 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.
 /// </para>
 /// <para>
@@ -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;

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");
 
+    /// <summary>
+    /// 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.
+    /// </summary>
+    internal const string SubchannelConnectFailureDetail = "Error connecting to subchannel.";
+
+    /// <summary>
+    /// 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 <see cref="StatusCode.Unavailable"/> with a
+    /// <see cref="SocketException"/> in the debug-exception chain:
+    /// <list type="bullet">
+    /// <item>Grpc.Net.Client's subchannel connect failure (<see cref="SubchannelConnectFailureDetail"/>):
+    /// the balancer never handed the call a connection, whatever the socket error.</item>
+    /// <item>An <see cref="HttpRequestException"/> 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).</item>
+    /// </list>
+    /// Every other transport failure — a reset, GOAWAY, keepalive timeout or premature end on an
+    /// established connection — may have delivered the request and stays ambiguous.
+    /// </summary>
+    /// <param name="error">The RPC failure to inspect.</param>
+    /// <returns><c>true</c> when the request was never sent.</returns>
+    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;
+    }
+
+    /// <summary>
+    /// True when a mutation may be replayed on the peer node: the server rejected it before
+    /// dispatch (<see cref="WasNotDispatched"/>), or the client never sent it (<see cref="WasNeverSent"/>).
+    /// </summary>
+    /// <param name="error">The RPC failure to inspect.</param>
+    /// <returns><c>true</c> when no node can have applied the request.</returns>
+    public static bool ProvablyNotApplied(RpcException error) => WasNotDispatched(error) || WasNeverSent(error);
+
     /// <summary>Reports an ambiguous mutation without inviting a blind replay.</summary>
     /// <param name="error">The original RPC failure whose trailers are preserved.</param>
     /// <returns>An <see cref="RpcException"/> with <c>Aborted</c> status marking the outcome as unknown.</returns>

src/ZB.MOM.WW.ScadaBridge.Communication/Grpc/SitePairChannelProvider.cs

@@ -35,7 +35,8 @@ public sealed class SiteChannelUnavailableException(string siteId)
 /// <remarks>
 /// <para>
 /// <b>Retry safety.</b> 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 (<see cref="GrpcRetrySafety.WasNeverSent"/>) 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.
 /// </para>
@@ -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);
             }

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;
+
+/// <summary>
+/// ScadaBridge#136: <see cref="GrpcRetrySafety.WasNeverSent"/> admits only connect-phase failures
+/// (the request provably never left the client) and rejects every shape that may have delivered it.
+/// </summary>
+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);
+
+    /// <summary>The balancer's subchannel connect failure is pre-send, whatever the socket error.</summary>
+    /// <param name="error">The socket error the subchannel connect saw.</param>
+    [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));
+    }
+
+    /// <summary>An HttpRequestException connect error with a connect-only socket error is pre-send.</summary>
+    /// <param name="kind">The HTTP request error classification.</param>
+    /// <param name="error">The socket error.</param>
+    [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));
+    }
+
+    /// <summary>Shapes that may follow delivery stay ambiguous.</summary>
+    [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))));
+    }
+}

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<string, Func<HttpMessageHandler>> _extraHandlers = new(StringComparer.Ordinal);
 
     /// <summary>Starts the stub servers and builds the channel provider under test.</summary>
     /// <returns>A task that represents the asynchronous operation.</returns>
@@ -47,7 +50,9 @@ public sealed class GrpcSiteTransportFailoverTests : IAsyncLifetime
             new FixedPskProvider(SiteKey),
             Options.Create(new CommunicationOptions()),
             NullLogger<SitePairChannelProvider>.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);
     }
 
+    /// <summary>
+    /// 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.
+    /// </summary>
+    /// <returns>A task that represents the asynchronous operation.</returns>
+    [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));
+    }
+
+    /// <summary>
+    /// 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.
+    /// </summary>
+    /// <returns>A task that represents the asynchronous operation.</returns>
+    [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));
+    }
+
+    /// <summary>
+    /// 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.
+    /// </summary>
+    /// <returns>A task that represents the asynchronous operation.</returns>
+    [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<RpcException>(() => CallAsync());
+
+        Assert.Equal(StatusCode.Aborted, error.StatusCode);
+        Assert.Contains("OUTCOME_UNKNOWN", error.Status.Detail);
+        Assert.Equal(1, _stubA.LifecycleCalls);
+        Assert.Equal(0, _stubB.LifecycleCalls);
+    }
+
+    /// <summary>A loopback port with nothing listening (bound, read, released).</summary>
+    private static int ClosedLoopbackPort()
+    {
+        using var listener = new TcpListener(IPAddress.Loopback, 0);
+        listener.Start();
+        var port = ((IPEndPoint)listener.LocalEndpoint).Port;
+        listener.Stop();
+        return port;
+    }
+
+    /// <summary>Delivers the request to the real server, then fails as if the connection reset mid-response.</summary>
+    private sealed class DropAfterDeliveryHandler(HttpMessageHandler inner) : DelegatingHandler(inner)
+    {
+        /// <inheritdoc />
+        protected override async Task<HttpResponseMessage> 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)));
+        }
+    }
+
     /// <summary>Verifies the case: unknown Site, throws, site Channel Unavailable.</summary>
     /// <returns>A task that represents the asynchronous operation.</returns>
     [Fact]