fix: avoid closing an idle RpcClient that is in use or still has in-flight requests

trpc-core/src/main/java/com/tencent/trpc/core/cluster/RpcClusterClientManager.java

@@ -128,19 +128,57 @@ public static void scanUnusedClient() {
         });
         unusedClientMap.forEach((bConfig, value) -> value.forEach(e -> {
             try {
+                RpcClientProxy proxy = (RpcClientProxy) e;
+                // Double-check before closing: ensure the client is still idle to avoid closing a client in use.
+                // This prevents race condition where getOrCreateClient() gets a client right before it's closed.
+                if (!isIdleTimeout(bConfig, proxy)) {
+                    // lastUsedNanos was updated, meaning a business thread is using it, try to put it back
+                    Map<String, RpcClientProxy> clientMap = CLUSTER_MAP.get(bConfig);
+                    if (clientMap != null) {
+                        RpcClientProxy existing = clientMap.putIfAbsent(
+                                proxy.getProtocolConfig().toUniqId(), proxy);
+                        if (existing == null) {
+                            // Successfully put back, do not close
+                            logger.info("RpcClient {} rescued from closing due to recent usage",
+                                    proxy.getProtocolConfig().toSimpleString());
+                            return;
+                        }
+                    }
+                    // Failed to put back (a new client already exists), still need to close the old one
+                }
                 e.close();
-            } finally {
                 logger.warn("RpcClient in clusterName={}, naming={}, remove rpc client{}, due to unused time > {} ms",
                         bConfig.getName(), bConfig.getNamingOptions().getServiceNaming(),
                         e.getProtocolConfig().toSimpleString(), bConfig.getIdleTimeout());
+            } catch (Exception ex) {
+                logger.error("Failed to close RpcClient in clusterName={}, naming={}, client={}",
+                        bConfig.getName(), bConfig.getNamingOptions().getServiceNaming(),
+                        e.getProtocolConfig().toSimpleString(), ex);
             }
         }));
     }
 
     private static boolean isIdleTimeout(BackendConfig bConfig, RpcClientProxy clientProxy) {
         long unusedNanosLimit = TimeUnit.MILLISECONDS.toNanos(bConfig.getIdleTimeout());
         long lastUsedNanos = clientProxy.getLastUsedNanos();
-        return lastUsedNanos > 0 && unusedNanosLimit > 0 && (System.nanoTime() - lastUsedNanos) > unusedNanosLimit;
+        boolean idleTooLong = lastUsedNanos > 0 && unusedNanosLimit > 0
+                && (System.nanoTime() - lastUsedNanos) > unusedNanosLimit;
+        if (!idleTooLong) {
+            return false;
+        }
+        // The client is idle for a long time, but there are still requests in flight on it.
+        // Skip cleaning in this round and re-check in the next one, otherwise closeClient() would
+        // forcibly fail all those in-flight requests with "Client(...) stop".
+        int pending = clientProxy.getPendingRequestCount();
+        if (pending > 0) {
+            logger.warn("RpcClient in clusterName={}, naming={}, client={} idle > {} ms, "
+                            + "but {} request(s) still in flight, skip cleaning this round",
+                    bConfig.getName(), bConfig.getNamingOptions().getServiceNaming(),
+                    clientProxy.getProtocolConfig().toSimpleString(),
+                    bConfig.getIdleTimeout(), pending);
+            return false;
+        }
+        return true;
     }
 
     /**
@@ -306,6 +344,11 @@ public ProtocolConfig getProtocolConfig() {
             return delegate.getProtocolConfig();
         }
 
+        @Override
+        public int getPendingRequestCount() {
+            return delegate.getPendingRequestCount();
+        }
+
         @Override
         public int hashCode() {
             return Objects.hash(delegate);
@@ -327,4 +370,4 @@ public boolean equals(Object obj) {
         }
     }
 
-}
+}

trpc-core/src/main/java/com/tencent/trpc/core/cluster/def/DefClusterInvoker.java

@@ -72,6 +72,9 @@ protected CompletionStage<Response> doInvoke(Request request, CompletionStage<Se
     protected ConsumerInvokerProxy<T> getInvoker(ServiceInstance instance) {
         String key = toUniqKey(instance);
         ConsumerInvokerProxy<T> result = invokerCache.get(key);
+        // Keep consistent with createInvoker: the invoker must be rebuilt once its client is closed
+        // (or is being closed), otherwise the request would be sent to a client which is being torn
+        // down by RpcClusterClientManager#scanUnusedClient.
         if (result != null && result.isAvailable()) {
             return result;
         }

trpc-core/src/main/java/com/tencent/trpc/core/rpc/RpcClient.java

@@ -61,4 +61,20 @@ public interface RpcClient {
      */
     ProtocolConfig getProtocolConfig();
 
+    /**
+     * Get the number of requests which are still in flight on this client.
+     *
+     * <p>It is used by the idle client cleaner to skip a client which still has in-flight requests,
+     * otherwise closing the client would forcibly fail those requests with {@code Client(...) stop}.</p>
+     *
+     * <p>A {@code default} implementation is provided to keep binary compatibility with the existing
+     * third-party implementations, which simply reports "no in-flight request" and thus keeps the old
+     * cleaning behavior.</p>
+     *
+     * @return the number of in-flight requests, 0 if unknown
+     */
+    default int getPendingRequestCount() {
+        return 0;
+    }
+
 }

trpc-core/src/test/java/com/tencent/trpc/core/cluster/RpcClusterClientManagerTest.java

@@ -21,6 +21,7 @@
 import com.tencent.trpc.core.rpc.ConsumerInvoker;
 import com.tencent.trpc.core.rpc.RpcClient;
 import java.lang.reflect.Field;
+import java.lang.reflect.Method;
 import java.util.Map;
 import org.junit.Assert;
 import org.junit.Test;
@@ -98,45 +99,203 @@ public void testScanWithEmptyCluster() {
         RpcClusterClientManager.scanUnusedClient();
     }
 
+    /**
+     * An idle client which still has in-flight requests must be kept, otherwise those requests would be
+     * forcibly failed with "Client(...) stop" by {@code DefResponseFutureManager#closeClient}.
+     * Once the requests are done, the client is expected to be cleaned in the next round.
+     */
+    @Test
+    public void testScanSkipsClientWithInFlightRequest() throws Exception {
+        BackendConfig backendConfig = new BackendConfig();
+        backendConfig.setIdleTimeout(1);
+        backendConfig.setNamingUrl("ip://127.0.0.1:8085");
+        InFlightProtocolConfigTest config = new InFlightProtocolConfigTest("127.0.0.1", 8085);
+        RpcClusterClientManager.getOrCreateClient(backendConfig, config);
+        Map<String, Object> innerMap = innerClientsOf(backendConfig);
+        assertEquals(1, innerMap.size());
+
+        Thread.sleep(10);
+        config.setPendingRequestCount(1);
+        RpcClusterClientManager.scanUnusedClient();
+        assertEquals("idle client with in-flight request should not be cleaned", 1, innerMap.size());
+
+        config.setPendingRequestCount(0);
+        RpcClusterClientManager.scanUnusedClient();
+        assertEquals("idle client without in-flight request should be cleaned", 0, innerMap.size());
+        RpcClusterClientManager.shutdownBackendConfig(backendConfig);
+    }
+
+    /**
+     * A client may be used by a business thread after it has been picked as unused but before it is really
+     * closed. In that case it must be put back instead of being closed.
+     */
+    @Test
+    public void testScanRescuesClientUsedAgainBeforeClosing() throws Exception {
+        BackendConfig backendConfig = new BackendConfig();
+        backendConfig.setIdleTimeout(1);
+        backendConfig.setNamingUrl("ip://127.0.0.1:8086");
+
+        ProtocolConfigTest configA = new ProtocolConfigTest("127.0.0.1", 8087);
+        ProtocolConfigTest configB = new ProtocolConfigTest("127.0.0.1", 8088);
+        RpcClusterClientManager.getOrCreateClient(backendConfig, configA);
+        RpcClusterClientManager.getOrCreateClient(backendConfig, configB);
+
+        Map<String, Object> innerMap = innerClientsOf(backendConfig);
+        assertEquals(2, innerMap.size());
+        Object clientA = innerMap.get(configA.toUniqId());
+        Object clientB = innerMap.get(configB.toUniqId());
+
+        // Whichever client is closed first refreshes the other one, which simulates a business thread
+        // using it during the cleaning process. The refreshed one must be rescued.
+        configA.setOnClose(() -> updateLastUsedNanos(clientB));
+        configB.setOnClose(() -> updateLastUsedNanos(clientA));
+
+        Thread.sleep(10);
+        RpcClusterClientManager.scanUnusedClient();
+        assertEquals("the client used again before closing should be rescued", 1, innerMap.size());
+        RpcClusterClientManager.shutdownBackendConfig(backendConfig);
+    }
+
+    /**
+     * A failure while closing one client must not break the cleaning of the others.
+     */
+    @Test
+    public void testScanContinuesWhenClosingFails() throws Exception {
+        BackendConfig backendConfig = new BackendConfig();
+        backendConfig.setIdleTimeout(1);
+        backendConfig.setNamingUrl("ip://127.0.0.1:8089");
+        ProtocolConfigTest config = new ProtocolConfigTest("127.0.0.1", 8090);
+        config.setOnClose(() -> {
+            throw new IllegalStateException("close failed");
+        });
+        RpcClusterClientManager.getOrCreateClient(backendConfig, config);
+        Map<String, Object> innerMap = innerClientsOf(backendConfig);
+        assertEquals(1, innerMap.size());
+
+        Thread.sleep(10);
+        RpcClusterClientManager.scanUnusedClient();
+        assertEquals("client should be removed even if closing it throws", 0, innerMap.size());
+        RpcClusterClientManager.shutdownBackendConfig(backendConfig);
+    }
+
+    @SuppressWarnings("unchecked")
+    private static Map<String, Object> innerClientsOf(BackendConfig backendConfig) throws Exception {
+        Field field = RpcClusterClientManager.class.getDeclaredField("CLUSTER_MAP");
+        field.setAccessible(true);
+        Map<BackendConfig, Map<String, Object>> clusterMap =
+                (Map<BackendConfig, Map<String, Object>>) field.get(null);
+        Map<String, Object> innerMap = clusterMap.get(backendConfig);
+        Assert.assertNotNull(innerMap);
+        return innerMap;
+    }
+
+    private static void updateLastUsedNanos(Object clientProxy) {
+        try {
+            Method method = clientProxy.getClass().getMethod("updateLastUsedNanos");
+            method.setAccessible(true);
+            method.invoke(clientProxy);
+        } catch (Exception ex) {
+            throw new IllegalStateException("failed to update lastUsedNanos", ex);
+        }
+    }
+
     private static class ProtocolConfigTest extends ProtocolConfig {
 
+        private Runnable onClose = () -> {
+        };
+
+        ProtocolConfigTest() {
+        }
+
+        ProtocolConfigTest(String ip, int port) {
+            setIp(ip);
+            setPort(port);
+            setNetwork("tcp");
+        }
+
+        void setOnClose(Runnable onClose) {
+            this.onClose = onClose;
+        }
+
         @Override
         public RpcClient createClient() {
-            return new RpcClient() {
+            return new TestRpcClient(this, () -> onClose.run());
+        }
+    }
 
-                @Override
-                public void open() throws TRpcException {
-                }
+    private static class InFlightProtocolConfigTest extends ProtocolConfig {
 
-                @Override
-                public boolean isClosed() {
-                    return false;
-                }
+        private volatile int pendingRequestCount;
 
-                @Override
-                public boolean isAvailable() {
-                    return true;
-                }
+        InFlightProtocolConfigTest(String ip, int port) {
+            setIp(ip);
+            setPort(port);
+            setNetwork("tcp");
+        }
 
-                @Override
-                public ProtocolConfig getProtocolConfig() {
-                    return ProtocolConfigTest.this;
-                }
+        void setPendingRequestCount(int pendingRequestCount) {
+            this.pendingRequestCount = pendingRequestCount;
+        }
 
+        @Override
+        public RpcClient createClient() {
+            return new TestRpcClient(this, () -> {
+            }) {
                 @Override
-                public void close() {
+                public int getPendingRequestCount() {
+                    return pendingRequestCount;
                 }
+            };
+        }
+    }
 
-                @Override
-                public <T> ConsumerInvoker<T> createInvoker(ConsumerConfig<T> consumerConfig) {
-                    return null;
-                }
+    /**
+     * Test client which does not override {@link RpcClient#getPendingRequestCount()}, so the default
+     * implementation (no in-flight request) is exercised as well.
+     */
+    private static class TestRpcClient implements RpcClient {
 
-                @Override
-                public CloseFuture<Void> closeFuture() {
-                    return new CloseFuture<Void>();
-                }
-            };
+        private final ProtocolConfig protocolConfig;
+
+        private final Runnable onClose;
+
+        TestRpcClient(ProtocolConfig protocolConfig, Runnable onClose) {
+            this.protocolConfig = protocolConfig;
+            this.onClose = onClose;
+        }
+
+        @Override
+        public void open() throws TRpcException {
+        }
+
+        @Override
+        public boolean isClosed() {
+            return false;
+        }
+
+        @Override
+        public boolean isAvailable() {
+            return true;
+        }
+
+        @Override
+        public ProtocolConfig getProtocolConfig() {
+            return protocolConfig;
+        }
+
+        @Override
+        public void close() {
+            onClose.run();
+        }
+
+        @Override
+        public <T> ConsumerInvoker<T> createInvoker(ConsumerConfig<T> consumerConfig) {
+            return null;
+        }
+
+        @Override
+        public CloseFuture<Void> closeFuture() {
+            return new CloseFuture<Void>();
         }
     }
 }

trpc-proto/trpc-rpc-support/src/main/java/com/tencent/trpc/proto/support/DefResponseFutureManager.java

@@ -79,6 +79,19 @@ public DefResponseFuture newFuture(RpcClientContext context, ConsumerInvoker<?>
         return future;
     }
 
+    /**
+     * Get the number of requests which are still in flight, i.e. the {@link DefResponseFuture}s that have
+     * not been completed (by a response or by the timeout watcher) yet.
+     *
+     * <p>It is used by the idle client cleaner to avoid closing a client which still has in-flight
+     * requests, otherwise those requests would be forcibly failed with {@code Client(...) stop}.</p>
+     *
+     * @return the number of in-flight requests
+     */
+    public int getPendingCount() {
+        return futureMap.size();
+    }
+
     /**
      * Removes and force stops all {@link DefResponseFuture}s related to a tRPC client.
      * Should be called when a tRPC client closes.

trpc-proto/trpc-rpc-support/src/main/java/com/tencent/trpc/proto/support/DefRpcClient.java

@@ -85,6 +85,14 @@ public boolean isAvailable() {
         return super.isAvailable() && (transport != null && transport.isConnected());
     }
 
+    /**
+     * {@inheritDoc}
+     */
+    @Override
+    public int getPendingRequestCount() {
+        return futureManager.getPendingCount();
+    }
+
     /**
      * {@inheritDoc}
      *