fix(server): register cache rpc for late graphs

docs/server-runtime.md

@@ -516,6 +516,61 @@ compatibility evidence. Run the read-only application examples against represent
 existing data and validate the serializers, client versions and authentication
 combinations actually used by the application.
 
+## Cache RPC graph lifecycle
+
+When cache RPC is enabled and a backend declares shared storage, Server registers both
+graph and schema cache notifiers for graphs loaded during startup and graphs created or
+loaded later. A new graph's RPC registration must succeed before its local publication.
+Graph services use the graph space and graph name together as their service identity.
+
+Internal graph loading registers RPC services in Server's administrative scope and
+restores the caller's identity afterward. Direct registration and cleanup remain
+administrative operations.
+
+Successful graph close releases that graph's notifier listeners, providers and consumer
+proxies. A failed task drain during ordinary close or non-clearing local removal retains
+the owner and its RPC resources. In PD-enabled deployments, RPC cleanup failures during
+graph DELETE retain the local owner and bindings for retry; partial cleanup does not imply
+a healthy service. This DELETE path first closes task admission and drains local executions,
+then cleans up RPC before metadata deletion and backend clearing. A failed drain or RPC
+cleanup can be retried with the same DELETE. Logical graph closure fences access before
+all resource cleanup completes. Any close failure retains the local owner and bindings;
+retrying the same DELETE retries cleanup without repeating backend clearing. A completed clear
+is remembered for that exact graph instance even if an external idle transaction delays close;
+retry after that transaction exits. A replacement graph with the same name is cleared independently.
+A graph closed by a non-clearing removal has no successful-clear certificate. A later clearing DELETE
+refuses that closed owner and keeps its binding, rather than reporting success while old backend data remains.
+Raft provider, node and an owned listener are closed independently. The first fatal Java Error takes
+priority over ordinary close failures; otherwise the first failure is retained. Other failures are
+reported as suppressed, except when that would create an exception cycle, in which case they are logged.
+A graph using Server's shared Raft listener closes
+only its own node. Server releases that listener and its address registration on shutdown;
+deleting one graph keeps the listener available to other local graphs. An interrupted node
+join retains the graph owner and interrupt status so a later DELETE can finish cleanup.
+The context's backend, snapshot and optional read-index executors stop after node join and
+drain queued work before close completes. Their wait uses `task.wait_timeout`; timeout or
+interruption retains the graph for retry without cancelling queued operations.
+RPC cleanup can start before graph close finishes. Clearing PD DELETE unregisters RPC before
+metadata removal and backend clear. Factory, binding and quota removal require a successful graph close;
+a failed close retains their owner even if its RPC services have already been unregistered.
+One failing graph does not prevent other graphs in a PD removal batch from being processed.
+Fatal Java Errors retain their original identity and
+propagate out of the removal batch. Failures after backend clearing do not undo deletion;
+retries do not restore deleted data or placement metadata.
+
+For local deletion without PD, backend clearing precedes RPC cleanup during graph close.
+A later cleanup failure does not preserve the deleted data.
+
+Failed creation or local loading closes the candidate in Server's administrative scope.
+If rollback close fails, Server retains the local owner and factory identity for cleanup retry.
+An old graph's cleanup cannot remove or re-export a replacement graph's services.
+Events queued for a closed notifier are ignored; an RPC request already sent cannot be recalled.
+
+Cache RPC uses the configured peer addresses; local provider export does not establish
+that every remote consumer connection is ready. Notifications are asynchronous, and
+connection or delivery failures are logged. This lifecycle handling does not provide
+acknowledged delivery or a distributed cache consistency protocol. Validate actual DML
+and cache behavior against the deployment's shared backend and topology.
 
 ## Deleting a graph with local tasks
 

hugegraph-commons/hugegraph-common/src/main/java/org/apache/hugegraph/rpc/RpcServiceConfig4Client.java

@@ -32,4 +32,8 @@ default <T> T serviceProxy(String graph, Class<T> clazz) {
     }
 
     void removeAllServiceProxy();
+
+    default boolean removeServiceProxy(String graph, String interfaceId, Object proxy) {
+        throw new UnsupportedOperationException("Removing an individual RPC proxy is not supported");
+    }
 }

hugegraph-commons/hugegraph-common/src/main/java/org/apache/hugegraph/rpc/RpcServiceConfig4Server.java

@@ -25,5 +25,9 @@ public interface RpcServiceConfig4Server {
 
     void removeService(String serviceId);
 
+    default boolean removeService(String serviceId, Object serviceImpl) {
+        throw new UnsupportedOperationException("Removing an RPC service by implementation is not supported");
+    }
+
     void removeAllService();
 }

hugegraph-commons/hugegraph-rpc/src/main/java/org/apache/hugegraph/rpc/RpcConsumerConfig.java

@@ -21,6 +21,9 @@
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.atomic.AtomicReference;
 
 import org.apache.hugegraph.config.HugeConfig;
 import org.apache.hugegraph.config.RpcOptions;
@@ -39,8 +42,6 @@
 import com.alipay.sofa.rpc.core.response.SofaResponse;
 import com.alipay.sofa.rpc.ext.Extension;
 import com.alipay.sofa.rpc.ext.ExtensionLoaderFactory;
-import com.google.common.collect.Lists;
-import com.google.common.collect.Maps;
 import com.google.common.collect.Sets;
 
 public class RpcConsumerConfig implements RpcServiceConfig4Client {
@@ -59,8 +60,8 @@ public RpcConsumerConfig(HugeConfig config, String remoteUrls) {
         RpcCommonConfig.initRpcConfigs(config);
         this.conf = config;
         this.remoteUrls = remoteUrls;
-        this.configs = Maps.newHashMap();
-        this.bootstraps = Lists.newArrayList();
+        this.configs = new ConcurrentHashMap<>();
+        this.bootstraps = new CopyOnWriteArrayList<>();
     }
 
     @Override
@@ -70,10 +71,57 @@ public <T> T serviceProxy(String interfaceId) {
 
     @Override
     public <T> T serviceProxy(String graph, String interfaceId) {
-        ConsumerConfig<T> config = this.consumerConfig(graph, interfaceId);
-        ConsumerBootstrap<T> bootstrap = Bootstraps.from(config);
-        this.bootstraps.add(bootstrap);
-        return bootstrap.refer();
+        String serviceId = graph == null ? interfaceId : interfaceId + ":" + graph;
+        AtomicReference<T> proxy = new AtomicReference<>();
+        this.configs.compute(serviceId, (key, existing) -> {
+            @SuppressWarnings("unchecked")
+            ConsumerConfig<T> config = existing == null ? this.consumerConfig(graph, interfaceId) :
+                                       (ConsumerConfig<T>) existing;
+            ConsumerBootstrap<T> bootstrap = Bootstraps.from(config);
+            try {
+                proxy.set(bootstrap.refer());
+                this.bootstraps.add(bootstrap);
+            } catch (RuntimeException | Error failure) {
+                try {
+                    bootstrap.unRefer();
+                    if (bootstrap.getCluster() != null) {
+                        bootstrap.getCluster().destroy();
+                    }
+                } catch (RuntimeException | Error cleanup) {
+                    failure.addSuppressed(cleanup);
+                }
+                throw failure;
+            }
+            return config;
+        });
+        return proxy.get();
+    }
+
+    @Override
+    public boolean removeServiceProxy(String graph, String interfaceId, Object proxy) {
+        String serviceId = graph == null ? interfaceId : interfaceId + ":" + graph;
+        boolean[] removed = {false};
+        this.configs.computeIfPresent(serviceId, (key, config) -> {
+            for (ConsumerBootstrap<?> bootstrap : this.bootstraps) {
+                if (bootstrap.getConsumerConfig() == config && bootstrap.getProxyIns() == proxy) {
+                    try {
+                        bootstrap.unRefer();
+                    } finally {
+                        bootstrap.getCluster().destroy();
+                    }
+                    this.bootstraps.remove(bootstrap);
+                    removed[0] = true;
+                    break;
+                }
+            }
+            for (ConsumerBootstrap<?> bootstrap : this.bootstraps) {
+                if (bootstrap.getConsumerConfig() == config) {
+                    return config;
+                }
+            }
+            return null;
+        });
+        return removed[0];
     }
 
     @Override
@@ -103,15 +151,7 @@ private <T> ConsumerConfig<T> consumerConfig(String graph,
             serviceId = interfaceId;
         }
 
-        @SuppressWarnings("unchecked")
-        ConsumerConfig<T> consumerConfig = (ConsumerConfig<T>)
-                                           this.configs.get(serviceId);
-        if (consumerConfig != null) {
-            return consumerConfig;
-        }
-
-        assert consumerConfig == null;
-        consumerConfig = new ConsumerConfig<>();
+        ConsumerConfig<T> consumerConfig = new ConsumerConfig<>();
 
         HugeConfig conf = this.conf;
         String protocol = conf.get(RpcOptions.RPC_PROTOCOL);
@@ -138,7 +178,6 @@ private <T> ConsumerConfig<T> consumerConfig(String graph,
                       .setSerialization(conf.get(RpcOptions.RPC_SERIALIZATION))
                       .setLoadBalancer(loadBalancer);
 
-        this.configs.put(serviceId, consumerConfig);
         return consumerConfig;
     }
 

hugegraph-commons/hugegraph-rpc/src/main/java/org/apache/hugegraph/rpc/RpcProviderConfig.java

@@ -18,14 +18,15 @@
 package org.apache.hugegraph.rpc;
 
 import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
 
 import com.alipay.sofa.rpc.config.ProviderConfig;
+import com.alipay.sofa.rpc.config.ServerConfig;
 import org.apache.hugegraph.util.E;
-import com.google.common.collect.Maps;
 
 public class RpcProviderConfig implements RpcServiceConfig4Server {
 
-    private final Map<String, ProviderConfig<?>> configs = Maps.newHashMap();
+    private final Map<String, ProviderConfig<?>> configs = new ConcurrentHashMap<>();
 
     @Override
     public <T, S extends T> String addService(Class<T> clazz, S serviceImpl) {
@@ -54,27 +55,58 @@ private <T, S extends T> String addService(String graph,
         providerConfig.setInterfaceId(interfaceId)
                       .setRef(serviceImpl);
 
-        E.checkArgument(!this.configs.containsKey(serviceId),
+        E.checkArgument(this.configs.putIfAbsent(serviceId, providerConfig) == null,
                         "Not allowed to add service already exist: '%s'",
                         serviceId);
-        this.configs.put(serviceId, providerConfig);
         return serviceId;
     }
 
     @Override
     public void removeService(String serviceId) {
-        ProviderConfig<?> config = this.configs.remove(serviceId);
-        E.checkArgument(config != null,
-                        "The service '%s' doesn't exist", serviceId);
-        config.unExport();
+        boolean[] removed = {false};
+        this.configs.computeIfPresent(serviceId, (key, config) -> {
+            config.unExport();
+            removed[0] = true;
+            return null;
+        });
+        E.checkArgument(removed[0], "The service '%s' doesn't exist", serviceId);
     }
 
     @Override
-    public void removeAllService() {
-        for (ProviderConfig<?> config : this.configs.values()) {
+    public boolean removeService(String serviceId, Object serviceImpl) {
+        boolean[] removed = {false};
+        this.configs.computeIfPresent(serviceId, (key, config) -> {
+            if (config.getRef() != serviceImpl) {
+                return config;
+            }
             config.unExport();
+            removed[0] = true;
+            return null;
+        });
+        return removed[0];
+    }
+
+    @Override
+    public void removeAllService() {
+        for (String serviceId : this.configs.keySet()) {
+            this.configs.computeIfPresent(serviceId, (key, config) -> {
+                config.unExport();
+                return null;
+            });
+        }
+    }
+
+    void exportAll(ServerConfig server, int timeout) {
+        for (String serviceId : this.configs.keySet()) {
+            this.configs.computeIfPresent(serviceId, (key, config) -> {
+                // Sofa appends a single server; retain other bindings without adding duplicates.
+                if (config.getServer() == null || config.getServer().stream().noneMatch(bound -> bound == server)) {
+                    config.setServer(server);
+                }
+                config.setTimeout(timeout).export();
+                return config;
+            });
         }
-        this.configs.clear();
     }
 
     public Map<String, ProviderConfig<?>> configs() {

hugegraph-commons/hugegraph-rpc/src/main/java/org/apache/hugegraph/rpc/RpcServer.java

@@ -104,11 +104,7 @@ public void exportAll() {
             return;
         }
         int timeout = this.conf.get(RpcOptions.RPC_SERVER_TIMEOUT) * 1000;
-        for (ProviderConfig<?> providerConfig : configs.values()) {
-            providerConfig.setServer(this.serverConfig)
-                          .setTimeout(timeout)
-                          .export();
-        }
+        this.configs.exportAll(this.serverConfig, timeout);
         LOG.info("RpcServer started success on port {}", this.port());
     }
 

hugegraph-commons/hugegraph-rpc/src/test/java/org/apache/hugegraph/unit/BaseUnitTest.java

@@ -20,12 +20,19 @@
 import java.net.URI;
 import java.net.URL;
 import java.nio.file.Paths;
+import java.util.concurrent.TimeUnit;
 
 import org.junit.BeforeClass;
 
 import org.apache.hugegraph.config.HugeConfig;
 import org.apache.hugegraph.config.OptionSpace;
 import org.apache.hugegraph.rpc.RpcServer;
+import org.apache.hugegraph.testutil.Assert;
+import org.apache.hugegraph.testutil.Whitebox;
+
+import com.alipay.sofa.rpc.config.ServerConfig;
+import com.alipay.sofa.rpc.server.Server;
+import io.netty.channel.ChannelFuture;
 
 public class BaseUnitTest {
 
@@ -53,6 +60,34 @@ protected static void startServer(RpcServer rpcServer) {
     }
 
     protected static void stopServer(RpcServer rpcServer) {
-        rpcServer.destroy();
+        closeServer(rpcServer, rpcServer::destroy);
+    }
+
+    protected static void unexportServer(RpcServer rpcServer) {
+        closeServer(rpcServer, rpcServer::unexportAll);
+    }
+
+    protected static ChannelFuture listener(RpcServer rpcServer) {
+        if (!rpcServer.enabled()) {
+            return null;
+        }
+        ServerConfig config = Whitebox.getInternalState(rpcServer, "serverConfig");
+        Server server = config.getServer();
+        if (server == null || !server.isStarted()) {
+            return null;
+        }
+        Object remoting = Whitebox.getInternalState(server, "remotingServer");
+        return Whitebox.getInternalState(remoting, "channelFuture");
+    }
+
+    private static void closeServer(RpcServer rpcServer, Runnable close) {
+        ChannelFuture binding = listener(rpcServer);
+        close.run();
+        if (binding != null) {
+            // Bolt stops asynchronously; the next fixture may reuse this port.
+            ChannelFuture closed = binding.channel().closeFuture();
+            Assert.assertTrue(closed.awaitUninterruptibly(5, TimeUnit.SECONDS));
+            Assert.assertTrue(closed.isSuccess());
+        }
     }
 }

hugegraph-commons/hugegraph-rpc/src/test/java/org/apache/hugegraph/unit/RpcConfigTest.java

@@ -171,13 +171,41 @@ public void testConsumerConfigFromOptions() {
         Assert.assertEquals("g1", graph.getUniqueId());
         Assert.assertEquals("fanout", graph.getCluster());
 
-        // Configs are cached per service id
-        Assert.assertSame(plain, consumerConfig(consumer, null, HELLO));
-        Assert.assertSame(graph, consumerConfig(consumer, "g1", HELLO));
-        Assert.assertNotSame(graph, consumerConfig(consumer, "g2", HELLO));
         consumer.destroy();
     }
 
+    @Test
+    public void testConsumerConfigCachedThroughServiceProxy() {
+        RpcServer server = new RpcServer(config("server-random"));
+        RpcConsumerConfig consumer = null;
+        try {
+            server.config().addService(HelloService.class, new HelloServiceImpl());
+            server.config().addService("g1", HelloService.class, new HelloServiceImpl());
+            server.config().addService("g2", HelloService.class, new HelloServiceImpl());
+            server.exportAll();
+            consumer = new RpcConsumerConfig(config("client"), server.host() + ":" + server.port());
+            ConsumerConfig<?> plain = referredConfig(consumer, null);
+            ConsumerConfig<?> graph = referredConfig(consumer, "g1");
+            Assert.assertSame(plain, referredConfig(consumer, null));
+            Assert.assertSame(graph, referredConfig(consumer, "g1"));
+            Assert.assertNotSame(graph, referredConfig(consumer, "g2"));
+        } finally {
+            if (consumer != null) {
+                consumer.destroy();
+            }
+            stopServer(server);
+        }
+    }
+
+    private static ConsumerConfig<?> referredConfig(RpcConsumerConfig consumer, String graph) {
+        HelloService proxy = consumer.serviceProxy(graph, HELLO);
+        Assert.assertEquals("hello tom!", proxy.hello("tom"));
+        Map<String, ConsumerConfig<?>> configs = Whitebox.getInternalState(consumer, "configs");
+        String id = graph == null ? HELLO : HELLO + ":" + graph;
+        Assert.assertNotNull(configs.get(id));
+        return configs.get(id);
+    }
+
     @Test
     public void testServerDestroyClosesServices() {
         RpcServer server = new RpcServer(config("server-random"));