Fix #16344: graceful HTTP/2 GOAWAY migration to eliminate request failure window

dubbo-remoting/dubbo-remoting-netty4/src/main/java/org/apache/dubbo/remoting/transport/netty4/AbstractNettyConnectionClient.java

@@ -209,7 +209,8 @@ public void onConnected(Object channel) {
             return;
         }
 
-        // Close the existing channel before setting a new channel
+        // Close the existing channel before setting a new channel.
+        // With the closeFuture CAS fix, closing the old channel won't wipe the new one.
         io.netty.channel.Channel current = getNettyChannel();
         if (current != null) {
             current.close();
@@ -229,10 +230,16 @@ public void onConnected(Object channel) {
         connectedPromise.trySuccess(null);
 
         if (logger.isDebugEnabled()) {
-            logger.debug("Connection:{} connected", this);
+            logger.debug("Connection:{} connected, graceful migration complete", this);
         }
     }
 
+    /**
+     * Fallback onGoaway: CAS-null the channel and close it.
+     * In the graceful migration flow (see {@link NettyConnectionHandler#onGoAway}),
+     * this is only called when doConnect() fails after GOAWAY, as a last resort
+     * to let the connectivity-scheduler handle recovery.
+     */
     @Override
     public void onGoaway(Object channel) {
         if (!(channel instanceof io.netty.channel.Channel)) {
@@ -247,7 +254,7 @@ public void onGoaway(Object channel) {
             }
             NettyChannel.removeChannelIfDisconnected(nettyChannel);
             if (logger.isDebugEnabled()) {
-                logger.debug("Connection:{} goaway", this);
+                logger.debug("Connection:{} goaway fallback: channel nulled, awaiting scheduler recovery", this);
             }
         }
     }
@@ -269,6 +276,14 @@ protected void clearNettyChannel() {
         channelRef.set(null);
     }
 
+    /**
+     * CAS-clear the channel reference: only clears if the current reference equals the expected channel.
+     * Used by closeFuture listeners to prevent wiping a newly-swapped channel during graceful migration.
+     */
+    protected boolean compareAndClearNettyChannel(io.netty.channel.Channel expected) {
+        return channelRef.compareAndSet(expected, null);
+    }
+
     @Override
     @SuppressWarnings("unchecked")
     public <T> T getChannel(Boolean generalizable) {
@@ -316,6 +331,24 @@ public static AbstractConnectionClient getConnectionClientFromChannel(io.netty.c
         return channel.attr(CONNECTION).get();
     }
 
+    /**
+     * Package-private accessor for {@link NettyConnectionHandler} to schedule
+     * graceful-migration tasks on the connectivity executor (instead of the
+     * Netty I/O EventLoop, which would block the I/O thread during doConnect).
+     */
+    java.util.concurrent.ScheduledExecutorService getConnectivityExecutor() {
+        return connectivityExecutor;
+    }
+
+    /**
+     * Package-private accessor for the configured reconnect interval, used by
+     * {@link NettyConnectionHandler} when scheduling the retry that follows a
+     * failed graceful migration attempt.
+     */
+    long getReconnectDuration() {
+        return reconnectDuration;
+    }
+
     public ChannelFuture write(Object request) throws RemotingException {
         if (!isAvailable()) {
             throw new RemotingException(

dubbo-remoting/dubbo-remoting-netty4/src/main/java/org/apache/dubbo/remoting/transport/netty4/NettyChannel.java

@@ -150,6 +150,23 @@ static void removeChannel(Channel ch) {
         }
     }
 
+    /**
+     * Look up the existing NettyChannel from cache without creating a new one.
+     * Used by handlers (e.g. {@link NettyConnectionHandler#channelInactive}) that
+     * only need to read state from an already-registered channel and must not
+     * allocate transient unregistered NettyChannel instances when the underlying
+     * Netty channel is no longer active.
+     *
+     * @param ch netty channel
+     * @return cached NettyChannel or null if absent
+     */
+    static NettyChannel getChannelIfPresent(Channel ch) {
+        if (ch == null) {
+            return null;
+        }
+        return CHANNEL_MAP.get(ch);
+    }
+
     @Override
     public InetSocketAddress getLocalAddress() {
         return AddressUtils.getLocalAddress(this);

dubbo-remoting/dubbo-remoting-netty4/src/main/java/org/apache/dubbo/remoting/transport/netty4/NettyConnectionClient.java

@@ -122,8 +122,10 @@ protected void initChannel(SocketChannel ch) {
                             new Http2ClientSettingsHandler(connectionPrefaceReceivedPromiseRef));
                 }
 
-                // set null but do not close this client, it will be reconnecting in the future
-                ch.closeFuture().addListener(channelFuture -> clearNettyChannel());
+                // Use compareAndSet to prevent wiping a newly-swapped channel during graceful migration.
+                // When onConnected() sets the new channel and then the old channel closes,
+                // the old channel's closeFuture must only clear its own reference, not the new one.
+                ch.closeFuture().addListener(channelFuture -> compareAndClearNettyChannel(ch));
                 // TODO support Socks5
 
                 // set channel initialized promise to success if necessary.

dubbo-remoting/dubbo-remoting-netty4/src/main/java/org/apache/dubbo/remoting/transport/netty4/NettyConnectionHandler.java

@@ -18,9 +18,9 @@
 
 import org.apache.dubbo.common.logger.ErrorTypeAwareLogger;
 import org.apache.dubbo.common.logger.LoggerFactory;
-import org.apache.dubbo.remoting.Constants;
 import org.apache.dubbo.remoting.api.connection.ConnectionHandler;
 
+import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.TimeUnit;
 
 import io.netty.channel.Channel;
@@ -30,15 +30,37 @@
 import io.netty.util.Attribute;
 import io.netty.util.AttributeKey;
 
+import static org.apache.dubbo.common.constants.LoggerCodeConstants.TRANSPORT_FAILED_RECONNECT;
 import static org.apache.dubbo.common.constants.LoggerCodeConstants.TRANSPORT_UNEXPECTED_EXCEPTION;
 
+/**
+ * Connection lifecycle handler for Netty-based Dubbo connections.
+ *
+ * <p>Implements graceful HTTP/2 GOAWAY migration: when a GOAWAY frame is received
+ * (e.g., from Envoy/Higress max_requests_per_connection), the old channel is kept
+ * alive for in-flight requests while a new connection is established in the background.
+ * Once the new connection is ready, {@link AbstractNettyConnectionClient#onConnected} atomically
+ * swaps the channel reference and closes the old one - eliminating any unavailability window.
+ *
+ * <p>Fallback: if the new connection fails (e.g., mTLS handshake timeout), the handler
+ * calls {@link AbstractNettyConnectionClient#onGoaway} to null the channel and let the
+ * connectivity-scheduler handle recovery (original behavior).
+ */
 @Sharable
 public class NettyConnectionHandler extends ChannelInboundHandlerAdapter implements ConnectionHandler {
 
     private static final ErrorTypeAwareLogger LOGGER =
             LoggerFactory.getErrorTypeAwareLogger(NettyConnectionHandler.class);
 
     private static final AttributeKey<Boolean> GO_AWAY_KEY = AttributeKey.valueOf("dubbo_channel_goaway");
+
+    /**
+     * Delay before initiating new connection after GOAWAY (milliseconds).
+     * Gives the gateway (Envoy/Higress) time to complete drain setup
+     * while keeping the migration window short.
+     */
+    private static final long GRACEFUL_RECONNECT_DELAY_MS = 200;
+
     private final AbstractNettyConnectionClient connectionClient;
 
     public NettyConnectionHandler(AbstractNettyConnectionClient connectionClient) {
@@ -55,18 +77,85 @@ public void onGoAway(Object channel) {
         if (Boolean.TRUE.equals(attr.get())) {
             return;
         }
-
         attr.set(true);
-        if (connectionClient != null) {
-            connectionClient.setAttribute(Constants.CHANNEL_ATTRIBUTE_READONLY_KEY, true);
-            connectionClient.onGoaway(nettyChannel);
-        }
+
         if (LOGGER.isDebugEnabled()) {
-            LOGGER.debug("Channel {} go away ,schedule reconnect", nettyChannel);
+            LOGGER.debug(String.format(
+                    "Received GOAWAY on %s -> %s, initiating graceful migration (delay=%dms)",
+                    nettyChannel.localAddress(), nettyChannel.remoteAddress(), GRACEFUL_RECONNECT_DELAY_MS));
+        }
+
+        if (connectionClient.isClosed()) {
+            LOGGER.info("The client has been closed and will not reconnect.");
+            return;
+        }
+
+        // Use connectivityExecutor (a dedicated scheduler) instead of the Netty I/O
+        // EventLoop. doConnect() internally awaits the connect promise via
+        // awaitUninterruptibly(getConnectTimeout()), which would block the I/O thread
+        // and stall in-flight requests on this very channel.
+        final ScheduledExecutorService scheduler = connectionClient.getConnectivityExecutor();
+        scheduler.schedule(
+                () -> attemptGracefulMigration(nettyChannel, scheduler, true),
+                GRACEFUL_RECONNECT_DELAY_MS,
+                TimeUnit.MILLISECONDS);
+    }
+
+    /**
+     * Try to establish a new connection on behalf of {@link #onGoAway}.
+     *
+     * <p>On success, {@link AbstractNettyConnectionClient#onConnected} swaps the channel
+     * reference atomically (closeFuture CAS in initBootstrap protects against
+     * the old channel's close handler wiping the new reference).
+     *
+     * <p>On failure, the first attempt schedules one extra retry through the
+     * same connectivity executor. The second failure falls back to
+     * {@link AbstractNettyConnectionClient#onGoaway}, which nulls the channel and lets
+     * the connectivity scheduler take over recovery.
+     */
+    private void attemptGracefulMigration(
+            final Channel nettyChannel, final ScheduledExecutorService scheduler, final boolean canRetry) {
+        if (connectionClient.isClosed()) {
+            return;
+        }
+        try {
+            connectionClient.doConnect();
+            // Success: onConnected() will close old channel and set new one.
+            // The closeFuture CAS in initBootstrap() ensures no race condition.
+        } catch (Throwable e) {
+            if (canRetry) {
+                long retryDelay = Math.max(connectionClient.getReconnectDuration(), GRACEFUL_RECONNECT_DELAY_MS);
+                LOGGER.warn(
+                        TRANSPORT_FAILED_RECONNECT,
+                        "",
+                        "",
+                        String.format(
+                                "Graceful migration attempt failed for %s, scheduling one retry in %dms",
+                                nettyChannel.remoteAddress(), retryDelay),
+                        e);
+                scheduler.schedule(
+                        () -> attemptGracefulMigration(nettyChannel, scheduler, false),
+                        retryDelay,
+                        TimeUnit.MILLISECONDS);
+                return;
+            }
+            // Graceful migration failed twice (e.g., persistent mTLS timeout).
+            // Fallback: null the channel to let connectivity-scheduler take over.
+            LOGGER.error(
+                    TRANSPORT_FAILED_RECONNECT,
+                    "",
+                    "",
+                    String.format(
+                            "Graceful migration failed for %s, falling back to scheduler recovery",
+                            nettyChannel.remoteAddress()),
+                    e);
+            connectionClient.onGoaway(nettyChannel);
         }
-        reconnect(nettyChannel);
     }
 
+    /**
+     * Reconnect with longer delay - used for unexpected disconnects (non-GOAWAY).
+     */
     @Override
     public void reconnect(Object channel) {
         if (!(channel instanceof Channel)) {
@@ -76,7 +165,7 @@ public void reconnect(Object channel) {
             LOGGER.debug("Connection:{} is reconnecting, attempt={}", connectionClient, 1);
         }
         if (connectionClient.isClosed()) {
-            LOGGER.info("The connection {} has been closed and will not reconnect", connectionClient);
+            LOGGER.info("The client has been closed and will not reconnect.");
             return;
         }
         connectionClient.scheduleReconnect(1, TimeUnit.SECONDS);
@@ -106,13 +195,17 @@ public void channelActive(ChannelHandlerContext ctx) {
     public void channelInactive(ChannelHandlerContext ctx) throws Exception {
         super.channelInactive(ctx);
         Channel ch = ctx.channel();
-        NettyChannel channel = NettyChannel.getOrAddChannel(ch, connectionClient.getUrl(), connectionClient);
+        // Look up the cached NettyChannel without allocating a new one.
+        // The channel is already inactive at this point, so getOrAddChannel would
+        // construct a transient unregistered NettyChannel only to log it.
+        NettyChannel channel = NettyChannel.getChannelIfPresent(ch);
         try {
             Attribute<Boolean> goawayAttr = ch.attr(GO_AWAY_KEY);
             if (!Boolean.TRUE.equals(goawayAttr.get())) {
+                // Only reconnect for unexpected disconnects (not GOAWAY).
+                // GOAWAY disconnects are already handled by the graceful migration above.
                 reconnect(ch);
             }
-
             if (LOGGER.isInfoEnabled() && channel != null) {
                 LOGGER.info(
                         "The connection {} of {} -> {} is disconnected.",

dubbo-remoting/dubbo-remoting-netty4/src/test/java/org/apache/dubbo/remoting/transport/netty4/TripleGoAwayTest.java

@@ -0,0 +1,98 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.dubbo.remoting.transport.netty4;
+
+import org.apache.dubbo.common.URL;
+import org.apache.dubbo.common.constants.CommonConstants;
+import org.apache.dubbo.common.utils.NetUtils;
+import org.apache.dubbo.config.ApplicationConfig;
+import org.apache.dubbo.config.context.ConfigManager;
+import org.apache.dubbo.remoting.api.connection.AbstractConnectionClient;
+import org.apache.dubbo.remoting.api.connection.ConnectionManager;
+import org.apache.dubbo.remoting.api.connection.MultiplexProtocolConnectionManager;
+import org.apache.dubbo.remoting.api.pu.DefaultPuHandler;
+import org.apache.dubbo.rpc.model.ApplicationModel;
+import org.apache.dubbo.rpc.model.ModuleModel;
+
+import java.time.Duration;
+
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Test;
+
+import static org.apache.dubbo.common.constants.CommonConstants.EXECUTOR_MANAGEMENT_MODE_DEFAULT;
+import static org.awaitility.Awaitility.await;
+
+public class TripleGoAwayTest {
+
+    private static URL url;
+
+    private static NettyPortUnificationServer server;
+
+    private static ConnectionManager connectionManager;
+
+    @BeforeAll
+    public static void init() throws Throwable {
+        int port = NetUtils.getAvailablePort();
+        url = URL.valueOf("tri://127.0.0.1:" + port + "?foo=bar");
+        ApplicationModel applicationModel = ApplicationModel.defaultModel();
+        ApplicationConfig applicationConfig = new ApplicationConfig("provider-app");
+        applicationConfig.setExecutorManagementMode(EXECUTOR_MANAGEMENT_MODE_DEFAULT);
+        applicationModel.getApplicationConfigManager().setApplication(applicationConfig);
+        ConfigManager configManager = new ConfigManager(applicationModel);
+        configManager.setApplication(applicationConfig);
+        configManager.getApplication();
+        applicationModel.setConfigManager(configManager);
+        url = url.setScopeModel(applicationModel);
+        ModuleModel moduleModel = applicationModel.getDefaultModule();
+        url = url.putAttribute(CommonConstants.SCOPE_MODEL, moduleModel);
+        server = new NettyPortUnificationServer(url, new DefaultPuHandler());
+        server.bind();
+        connectionManager = url.getOrDefaultFrameworkModel()
+                .getExtensionLoader(ConnectionManager.class)
+                .getExtension(MultiplexProtocolConnectionManager.NAME);
+    }
+
+    @AfterAll
+    public static void close() {
+        try {
+            server.close();
+        } catch (Throwable e) {
+            // ignored
+        }
+    }
+
+    @Test
+    void testGoawayCausesNoChannelFailure() {
+        final AbstractConnectionClient connectionClient = connectionManager.connect(url, new DefaultPuHandler());
+        Assertions.assertTrue(
+                connectionClient.isAvailable(), "Precondition: connection should be available before GOAWAY");
+        io.netty.channel.Channel channelBeforeGoaway = connectionClient.getChannel(true);
+        NettyConnectionHandler handler = new NettyConnectionHandler((AbstractNettyConnectionClient) connectionClient);
+        handler.onGoAway(channelBeforeGoaway);
+        io.netty.channel.Channel channelAfterGoaway = connectionClient.getChannel(true);
+        Assertions.assertNotNull(
+                channelAfterGoaway,
+                "Channel should NOT be null immediately after GOAWAY "
+                        + "— old channel should stay alive until new one is ready");
+        await().atMost(Duration.ofSeconds(10)).until(() -> connectionClient.isAvailable());
+        Assertions.assertTrue(
+                connectionClient.isAvailable(), "Connection should be available after GOAWAY migration completes");
+        connectionClient.close();
+    }
+}