feat(server): coordinate loading owners

docs/server-runtime.md

@@ -535,3 +535,63 @@ configuration changes rather than hot reload, whose recovery is handled separate
 Deletion is not transactional: a failure after destructive cleanup needs operator
 inspection, and retry does not restore removed data. An owner-cleanup failure must not
 be treated as a successful drain; inspect it instead of force-removing the scheduler.
+
+## Local loading ownership
+
+Server keeps ordinary import ownership in memory on each graph instance. The first
+owner changes NONE to LOADING; releasing one owner leaves LOADING while others are
+active, and the last release restores NONE. Manual LOADING, RESTORING and MERGING
+cannot be adopted. Original enum setters reject every request while owners are active,
+including bare LOADING; their original behavior remains available without active owners.
+
+`GET /graphspaces/{space}/graphs/{graph}/mode/owners` returns
+`{"protocol":1,"instance":"<UUID>","epoch":0,"revision":0}`. `PUT` on that path accepts
+`{"action":"acquire","instance":"<UUID>","epoch":0,"token":"<import UUID>"}`;
+`release` uses the same tuple. Receipts echo protocol/instance/epoch/token and return
+`state` (`active` or `released`) and the effective `mode`. Both operations use existing
+graph-owner/STATUS WRITE permissions; the authenticated principal owns the token.
+The request principal is also bound for graphs opened without the authentication proxy.
+Each new acquire, actual release and successful recovery writes an INFO audit entry
+with the graph instance, previous/resulting mode, epoch and owner counts. Idempotent
+retries do not add entries. These audit entries omit tokens and authenticated principal values.
+
+Keep one tuple and identity for the whole import attempt, including uncertain acquire
+outcomes. Release after that import's workers stop. Acquire/release retries are
+idempotent. An unknown release cancels a late acquire; a released handle cannot be
+reused. Terminal records are bounded to 1024 per epoch. Epoch advancement retains
+active old handles and permanently retires all other old handles.
+
+This protection coordinates Loader processes connected directly to the same stable
+Server and graph instance. It does not fence ordinary data requests across load
+balancers, node failover, graph replacement or Server restart. There is no automatic
+expiry: after kill -9 or permanent disconnection, stop the import and its in-flight
+writes before retrying release with the same handle and identity. If that handle is
+lost, an administrator can recover without that token:
+
+1. Stop every related import and drain its in-flight writes. Keep ordinary writers
+   stopped too; the owner protocol does not fence graph data requests.
+2. Read the current instance, epoch and revision from the owners endpoint on the same Server.
+3. Send `PUT` to that endpoint with
+   `{"action":"recover","instance":"<UUID>","epoch":0,"revision":0,"confirm":"I'm sure all imports have stopped"}`.
+
+Recovery requires administrator permission at both the REST and graph-proxy layers,
+the exact current instance/epoch/revision and
+the explicit confirmation. It clears owners, advances the epoch and returns NONE;
+late acquire/release requests from the retired epoch cannot register an owner again.
+The receipt contains the new epoch and `state: recovered`, without tokens. A stale
+recovery request returns a conflict and cannot clear an import started in a newer
+epoch. Owner registration/release and manual mode changes advance the revision, so
+an import started after inspection also rejects the stale recovery request even within
+the same epoch. After an uncertain result, inspect current state rather than blindly
+retrying with a newly fetched epoch/revision. This operation does not stop or drain imports itself.
+
+The same recovery procedure can reset manual RESTORING or MERGING after that work
+has stopped and drained; restore completion keeps its historical RESTORING mode.
+Before ordinary import on an owner-capable Server, explicitly recover or set NONE
+only after inspecting the completed work and confirming there are no active imports.
+Expected loading-owner and manual-mode conflicts return HTTP 409; invalid requests
+return 400 and unauthorized recovery is rejected by the authentication layer.
+
+Loaders retain historical ordinary-import and restore behavior with Servers that do
+not provide this API. Concurrent owner protection applies only when this API is
+supported; RESTORING keeps its existing enum mode path.

hugegraph-server/README.md

@@ -16,6 +16,8 @@ HugeGraph Server consists of two layers of functionality: the graph engine layer
 
 See [launcher GC options](../docs/server-runtime.md#launcher-gc-options) for explicit collector selection
 and caller JVM tuning.
+See [local loading ownership](../docs/server-runtime.md#local-loading-ownership) for
+concurrent import mode protection and its compatibility limits.
 
 ## Backend Evolution and Compatibility
 

hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/api/profile/GraphsAPI.java

@@ -31,6 +31,7 @@
 import org.apache.commons.lang3.StringUtils;
 import org.apache.hugegraph.exception.HugeException;
 import org.apache.hugegraph.HugeGraph;
+import org.apache.hugegraph.LoadingMode;
 import org.apache.hugegraph.api.API;
 import org.apache.hugegraph.api.filter.StatusFilter;
 import org.apache.hugegraph.auth.AuthManager;
@@ -56,6 +57,7 @@
 import jakarta.annotation.security.RolesAllowed;
 import jakarta.inject.Singleton;
 import jakarta.ws.rs.Consumes;
+import jakarta.ws.rs.ClientErrorException;
 import jakarta.ws.rs.DELETE;
 import jakarta.ws.rs.ForbiddenException;
 import jakarta.ws.rs.GET;
@@ -68,6 +70,7 @@
 import jakarta.ws.rs.QueryParam;
 import jakarta.ws.rs.core.Context;
 import jakarta.ws.rs.core.SecurityContext;
+import jakarta.ws.rs.core.Response;
 
 @Path("graphspaces/{graphspace}/graphs")
 @Singleton
@@ -78,6 +81,7 @@ public class GraphsAPI extends API {
 
     private static final String CONFIRM_CLEAR = "I'm sure to delete all data";
     private static final String CONFIRM_DROP = "I'm sure to drop the graph";
+    private static final String CONFIRM_RECOVER = "I'm sure all imports have stopped";
     private static final String GRAPH_DESCRIPTION = "description";
     private static final String GRAPH_ACTION = "action";
     private static final String UPDATE = "update";
@@ -674,7 +678,11 @@ public Map<String, GraphMode> mode(@Context GraphManager manager,
 
         E.checkArgument(mode != null, "Graph mode can't be null");
         HugeGraph g = graph(manager, graphSpace, name);
-        g.mode(mode);
+        try {
+            g.mode(mode);
+        } catch (LoadingMode.ConflictException e) {
+            throw new ClientErrorException(e.getMessage(), Response.Status.CONFLICT, e);
+        }
         return ImmutableMap.of("mode", mode);
     }
 
@@ -693,6 +701,60 @@ public Map<String, GraphMode> mode(@Context GraphManager manager,
         return ImmutableMap.of("mode", g.mode());
     }
 
+    @GET
+    @Timed
+    @Path("{name}/mode/owners")
+    @Produces(APPLICATION_JSON_WITH_CHARSET)
+    @RolesAllowed({"space", "$owner=$name"})
+    public Map<String, Object> loadingMode(@Context GraphManager manager,
+                                          @PathParam("graphspace") String graphSpace,
+                                          @PathParam("name") String name) {
+        return graph(manager, graphSpace, name).loadingMode();
+    }
+
+    @PUT
+    @Timed
+    @Path("{name}/mode/owners")
+    @Consumes(APPLICATION_JSON)
+    @Produces(APPLICATION_JSON_WITH_CHARSET)
+    @RolesAllowed({"space", "$owner=$name"})
+    public Map<String, Object> loadingMode(@Context GraphManager manager,
+                                          @PathParam("graphspace") String graphSpace,
+                                          @PathParam("name") String name,
+                                          JsonLoadingOwner owner,
+                                          @Context SecurityContext securityContext) {
+        E.checkArgument(owner != null && owner.epoch != null, "Loading owner body and epoch are required");
+        if ("recover".equals(owner.action)) {
+            if (securityContext == null || !securityContext.isUserInRole("admin")) {
+                throw new ForbiddenException("Loading recovery requires administrator permission");
+            }
+            E.checkArgument(CONFIRM_RECOVER.equals(owner.confirm),
+                            "Stop and drain all related imports, then confirm with '%s'", CONFIRM_RECOVER);
+            E.checkArgument(owner.revision != null && owner.revision >= 0,
+                            "Current loading state revision is required for recovery");
+        }
+        try {
+            HugeGraph graph = graph(manager, graphSpace, name);
+            if ("recover".equals(owner.action)) {
+                return graph.recoverLoadingMode(owner.instance, owner.epoch, owner.revision);
+            }
+            return graph.loadingMode(owner.action, owner.instance, owner.epoch, owner.token,
+                                     HugeGraphAuthProxy.username());
+        } catch (LoadingMode.ConflictException e) {
+            throw new ClientErrorException(e.getMessage(), Response.Status.CONFLICT, e);
+        }
+    }
+
+    public static class JsonLoadingOwner {
+
+        public String action;
+        public String instance;
+        public Long epoch;
+        public String token;
+        public String confirm;
+        public Long revision;
+    }
+
     @PUT
     @Timed
     @Path("{name}/graph_read_mode")

hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/auth/HugeGraphAuthProxy.java

@@ -812,6 +812,27 @@ public void mode(GraphMode mode) {
         this.hugegraph.mode(mode);
     }
 
+    @Override
+    public Map<String, Object> loadingMode() {
+        this.verifyPermission(HugePermission.WRITE, ResourceType.STATUS);
+        return this.hugegraph.loadingMode();
+    }
+
+    @Override
+    public Map<String, Object> loadingMode(String action, String instance, long epoch,
+                                            String token, String principal) {
+        this.verifyPermission(HugePermission.WRITE, ResourceType.STATUS);
+        Context context = getContext();
+        E.checkState(context != null, "Missing authenticated loading owner context");
+        return this.hugegraph.loadingMode(action, instance, epoch, token, context.user().username());
+    }
+
+    @Override
+    public Map<String, Object> recoverLoadingMode(String instance, long epoch, long revision) {
+        this.verifyAdminPermission();
+        return this.hugegraph.recoverLoadingMode(instance, epoch, revision);
+    }
+
     @Override
     public GraphReadMode readMode() {
         this.verifyStatusPermission();

hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/HugeGraph.java

@@ -23,6 +23,7 @@
 import java.util.HashSet;
 import java.util.Iterator;
 import java.util.List;
+import java.util.Map;
 import java.util.Set;
 import java.util.function.Predicate;
 
@@ -216,6 +217,19 @@ public interface HugeGraph extends Graph, HugeGraphSupplier {
 
     void mode(GraphMode mode);
 
+    default Map<String, Object> loadingMode() {
+        throw new UnsupportedOperationException("Loading ownership is not supported by this graph implementation");
+    }
+
+    default Map<String, Object> loadingMode(String action, String instance, long epoch,
+                                            String token, String principal) {
+        throw new UnsupportedOperationException("Loading ownership is not supported by this graph implementation");
+    }
+
+    default Map<String, Object> recoverLoadingMode(String instance, long epoch, long revision) {
+        throw new UnsupportedOperationException("Loading recovery is not supported by this graph implementation");
+    }
+
     GraphReadMode readMode();
 
     void readMode(GraphReadMode readMode);