refactor(server): coordinate graph rpc lifecycle

docs/server-runtime.md

@@ -481,6 +481,9 @@ Task database workers release their owned task and graph transactions on the wor
 thread. A failure while closing one owner must not skip cleanup of other owners or
 server-info resources. Cleanup retains the first exception and attaches later cleanup
 failures as suppressed exceptions.
+For the local Standard scheduler, a deleted task disappears from public lookup after its persisted record is
+removed, while its actual execution still counts toward drain until native cleanup finishes; late completion saves from
+that owner do not recreate the removed record.
 
 Synchronous task cancellation preserves a transaction already open on its caller,
 including uncommitted graph writes. Cancellation still runs the normal completion and
@@ -555,6 +558,84 @@ 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.
+PD deletion and startup coordinate RPC removal for the same graph instance. Startup does not
+register an owner whose RPC removal is pending, even after its placement-event token is consumed,
+and checks logical closure under the actual owner's RPC lifecycle monitor before registration.
+Any retained provider configuration must be unregistered before startup can export services;
+a cleanup failure fails startup instead of re-exporting that owner. Graph clearing, task drains
+and full graph close remain outside this RPC publication gate.
+
+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, with a repeated reference to the primary failure ignored.
+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 close failure retains the same local graph, factory identity and configuration for DELETE retry,
+even when access has already been fenced by logical closure. Only a successful clear by that
+instance's local drop permits retry to skip clearing and finish its retained resource cleanup.
+An ordinary close does not provide that certificate. A failure before clearing keeps the open
+owner's existing RPC services. A later cleanup failure does not preserve or restore 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.
+Retained failed candidates are excluded from RPC startup even while they are still open.
+An old graph's cleanup cannot remove or re-export a replacement graph's services.
+A PD graph-add notification blocked by an existing owner is coalesced until that owner is
+released, including when independent watches deliver the new ADD before the old REMOVE.
+Duplicate ADD notifications alone keep the healthy owner. Server consumes the PD acknowledgement
+of its own local graph creation without retaining a replacement ADD unless that exact owner already
+has failed peer cleanup; the existing cleanup retry takes priority. A healthy creation acknowledgement
+cannot republish the graph during a later reload. After removal, Server rereads the latest configuration;
+a missing configuration creates nothing. Reload consumes pending ADDs only after each old owner is
+successfully removed, then loads current metadata together instead of replaying ADDs during removal.
+After a peer removal fails, a later same-name PD add retries non-clearing cleanup of that exact
+owner before loading its replacement. Another cleanup failure keeps the owner for the next add
+or normal reload; no background retry is scheduled. A replacement load failure is logged and
+remains pending for a later add or normal reload. The completed old-owner DELETE still succeeds.
+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

@@ -20,7 +20,9 @@
 import java.util.ArrayList;
 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,9 +41,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 +58,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,28 +69,119 @@ 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) {
+                // Retain the cleanup owner even when refer never returned a proxy.
+                this.bootstraps.add(bootstrap);
+                throwFailure(this.cleanup(bootstrap, 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) {
+                    Throwable failure = this.cleanup(bootstrap, null);
+                    if (failure != null) {
+                        throwFailure(failure);
+                    }
+                    removed[0] = true;
+                    break;
+                }
+            }
+            for (ConsumerBootstrap<?> bootstrap : this.bootstraps) {
+                if (bootstrap.getConsumerConfig() == config) {
+                    return config;
+                }
+            }
+            return null;
+        });
+        return removed[0];
     }
 
     @Override
     public void removeAllServiceProxy() {
+        Throwable failure = null;
         for (ConsumerBootstrap<?> bootstrap : this.bootstraps) {
-            bootstrap.unRefer();
+            try {
+                bootstrap.unRefer();
+            } catch (RuntimeException | Error cleanup) {
+                failure = addFailure(failure, cleanup);
+            }
+        }
+        if (failure != null) {
+            throwFailure(failure);
         }
     }
 
     public void destroy() {
-        Set<Cluster> clusters = Sets.newHashSet();
+        Throwable failure = null;
         for (ConsumerBootstrap<?> bootstrap : this.bootstraps) {
+            failure = this.cleanup(bootstrap, failure);
+        }
+        if (failure != null) {
+            throwFailure(failure);
+        }
+    }
+
+    private Throwable cleanup(ConsumerBootstrap<?> bootstrap, Throwable failure) {
+        boolean complete = true;
+        try {
             bootstrap.unRefer();
-            clusters.add(bootstrap.getCluster());
+        } catch (RuntimeException | Error cleanup) {
+            complete = false;
+            failure = addFailure(failure, cleanup);
+        }
+        try {
+            Cluster cluster = bootstrap.getCluster();
+            if (cluster != null) {
+                cluster.destroy();
+            }
+        } catch (RuntimeException | Error cleanup) {
+            complete = false;
+            failure = addFailure(failure, cleanup);
+        }
+        if (complete) {
+            this.bootstraps.remove(bootstrap);
+        }
+        return failure;
+    }
+
+    private static Throwable addFailure(Throwable failure, Throwable cleanup) {
+        if (failure == null) {
+            return cleanup;
+        }
+        if (failure == cleanup) {
+            return failure;
         }
-        for (Cluster cluster : clusters) {
-            cluster.destroy();
+        if (cleanup instanceof Error && !(failure instanceof Error)) {
+            cleanup.addSuppressed(failure);
+            return cleanup;
         }
+        failure.addSuppressed(cleanup);
+        return failure;
+    }
+
+    private static void throwFailure(Throwable failure) {
+        if (failure instanceof Error) {
+            throw (Error) failure;
+        }
+        throw (RuntimeException) failure;
     }
 
     private <T> ConsumerConfig<T> consumerConfig(String graph,
@@ -103,15 +193,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 +220,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());
     }