diff --git a/docs/server-runtime.md b/docs/server-runtime.md index 5a27fb6b0c..586d2d11cf 100644 --- a/docs/server-runtime.md +++ b/docs/server-runtime.md @@ -574,3 +574,73 @@ 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":"","epoch":0,"revision":0}`. This response is illustrative; +recovery must use the actual values returned by the latest GET. `PUT` on that path accepts +`{"action":"acquire","instance":"","epoch":0,"token":""}`; +`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. +Each live graph admits at most 1024 concurrent owners. At that limit, a new owner +receives HTTP 409 without changing state; existing-owner retries, release and +administrator recovery remain available. Releasing an owner frees a slot. +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 name, 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. An initial +acquire prepared before epoch advancement can therefore be rejected even if it +never became active. The client must not automatically refresh that retired handle: +a new token could undo a prior cancellation or administrator recovery. After a +terminal acquire rejection, stop and drain that import attempt before creating a +fresh Loader context to retry. + +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":"","epoch":0,"revision":0,"confirm":"I'm sure all imports have stopped"}`. + +Recovery requires the authenticated global administrator role even when graph-space +permission checks are disabled; a proxied graph also enforces administrator permission. +With Server authentication disabled, the existing trusted anonymous administrator semantics apply. +Recovery also requires the exact current instance/epoch/revision and 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 every successful enum mode write advance the +revision, including a write of the current mode. Either operation after inspection +therefore 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. diff --git a/hugegraph-server/README.md b/hugegraph-server/README.md index fbd529f4ac..ee5f8cff64 100644 --- a/hugegraph-server/README.md +++ b/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. See [pairwise same neighbors](../docs/same-neighbors-batch.md) for the batch traverser request and response. ## Backend Evolution and Compatibility diff --git a/hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/api/profile/GraphsAPI.java b/hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/api/profile/GraphsAPI.java index 95505be80b..66343de6eb 100644 --- a/hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/api/profile/GraphsAPI.java +++ b/hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/api/profile/GraphsAPI.java @@ -31,12 +31,15 @@ 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.api.filter.AuthenticationFilter.Authorizer; import org.apache.hugegraph.auth.AuthManager; import org.apache.hugegraph.auth.HugeAuthenticator.RequiredPerm; import org.apache.hugegraph.auth.HugeGraphAuthProxy; import org.apache.hugegraph.auth.HugePermission; +import org.apache.hugegraph.auth.RolePermission; import org.apache.hugegraph.config.HugeConfig; import org.apache.hugegraph.core.GraphManager; import org.apache.hugegraph.space.GraphSpace; @@ -56,6 +59,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; @@ -67,7 +71,9 @@ import jakarta.ws.rs.Produces; import jakarta.ws.rs.QueryParam; import jakarta.ws.rs.core.Context; +import jakarta.ws.rs.container.ContainerRequestContext; import jakarta.ws.rs.core.SecurityContext; +import jakarta.ws.rs.core.Response; @Path("graphspaces/{graphspace}/graphs") @Singleton @@ -78,6 +84,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 +681,11 @@ public Map 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 +704,63 @@ public Map mode(@Context GraphManager manager, return ImmutableMap.of("mode", g.mode()); } + @GET + @Timed + @Path("{name}/mode/owners") + @Produces(APPLICATION_JSON_WITH_CHARSET) + @RolesAllowed({"space", "$graphspace=$graphspace $owner=$name"}) + public Map 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", "$graphspace=$graphspace $owner=$name"}) + public Map loadingMode(@Context GraphManager manager, + @PathParam("graphspace") String graphSpace, + @PathParam("name") String name, + JsonLoadingOwner owner, + @Context ContainerRequestContext requestContext) { + E.checkArgument(owner != null && owner.epoch != null, "Loading owner body and epoch are required"); + if ("recover".equals(owner.action)) { + // Read the filter's Authorizer rather than Jersey's injected SecurityContext wrapper. + SecurityContext securityContext = requestContext == null ? null : requestContext.getSecurityContext(); + if (!(securityContext instanceof Authorizer) || + !RolePermission.isAdmin(((Authorizer) securityContext).role())) { + 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") diff --git a/hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/auth/HugeGraphAuthProxy.java b/hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/auth/HugeGraphAuthProxy.java index f4066cf154..1967ed0724 100644 --- a/hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/auth/HugeGraphAuthProxy.java +++ b/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 loadingMode() { + this.verifyPermission(HugePermission.WRITE, ResourceType.STATUS); + return this.hugegraph.loadingMode(); + } + + @Override + public Map 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 recoverLoadingMode(String instance, long epoch, long revision) { + this.verifyAdminPermission(); + return this.hugegraph.recoverLoadingMode(instance, epoch, revision); + } + @Override public GraphReadMode readMode() { this.verifyStatusPermission(); diff --git a/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/HugeGraph.java b/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/HugeGraph.java index 56bc00c0bb..661453e75f 100644 --- a/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/HugeGraph.java +++ b/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 loadingMode() { + throw new UnsupportedOperationException("Loading ownership is not supported by this graph implementation"); + } + + default Map 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 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); diff --git a/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/LoadingMode.java b/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/LoadingMode.java new file mode 100644 index 0000000000..11b68f3354 --- /dev/null +++ b/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/LoadingMode.java @@ -0,0 +1,183 @@ +/* + * 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.hugegraph; + +import java.util.HashMap; +import java.util.Map; +import java.util.Objects; +import java.util.UUID; +import java.util.function.Supplier; + +import org.apache.hugegraph.type.define.GraphMode; +import org.apache.hugegraph.util.E; +import org.apache.hugegraph.util.Log; +import org.slf4j.Logger; + +import com.google.common.collect.ImmutableMap; + +/** Same graph-instance ownership; this state is neither persisted nor distributed. */ +public final class LoadingMode { + + private static final Logger LOG = Log.logger(LoadingMode.class); + + private static final int MAX_ACTIVE = 1024; + private static final int MAX_RELEASED = 1024; + + private final Supplier graphName; + private final String instance = UUID.randomUUID().toString(); + private final Map active = new HashMap<>(); + private final Map released = new HashMap<>(); + private long epoch; + private long revision; + private volatile GraphMode mode = GraphMode.NONE; + + public LoadingMode(Supplier graphName) { + this.graphName = Objects.requireNonNull(graphName, "graphName"); + } + + public GraphMode mode() { + return this.mode; + } + + public synchronized void mode(GraphMode mode) { + Objects.requireNonNull(mode, "mode"); + checkState(this.active.isEmpty(), "Loading owners are active; release them before setting graph mode"); + this.advanceRevision(); + this.mode = mode; + } + + public synchronized Map capability() { + return ImmutableMap.of("protocol", 1, "instance", this.instance, "epoch", this.epoch, + "revision", this.revision); + } + + public synchronized Map recover(String instance, long epoch, long revision) { + E.checkArgument(this.instance.equals(instance), "Loading owner belongs to another graph instance"); + E.checkArgument(epoch >= 0 && revision >= 0, "Invalid loading recovery epoch or revision"); + checkState(epoch == this.epoch && revision == this.revision, + "Loading state changed; inspect it before recovery"); + checkState(this.epoch < Long.MAX_VALUE, "Loading owner epoch exhausted"); + GraphMode previousMode = this.mode; + int clearedOwners = this.active.size(); + this.advanceRevision(); + this.epoch++; + this.active.clear(); + this.released.clear(); + this.mode = GraphMode.NONE; + this.logChange("recover", previousMode, clearedOwners); + return ImmutableMap.builder().putAll(this.capability()) + .put("state", "recovered").put("mode", this.mode).build(); + } + + public synchronized Map update(String action, String instance, long epoch, + String token, String principal) { + E.checkArgument("acquire".equals(action) || "release".equals(action), + "Loading owner action must be acquire or release"); + E.checkArgument(this.instance.equals(instance), "Loading owner belongs to another graph instance"); + E.checkArgument(epoch >= 0 && epoch <= this.epoch, "Invalid loading owner epoch"); + E.checkArgument(token != null && token.length() == 36 && UUID.fromString(token).toString().equals(token), + "Loading owner token must be a canonical UUID"); + Objects.requireNonNull(principal, "principal"); + GraphMode previousMode = this.mode; + String key = epoch + "/" + token; + String owner = this.active.get(key); + if (owner != null) { + checkState(owner.equals(principal), "Loading owner belongs to another authenticated principal"); + if ("acquire".equals(action)) { + return this.receipt(epoch, token, "active"); + } + // Exhaustion must fail before changing an active registration. + checkState(epoch != this.epoch || this.released.size() < MAX_RELEASED || + this.epoch < Long.MAX_VALUE, "Loading owner epoch exhausted"); + this.advanceRevision(); + this.active.remove(key); + if (this.active.isEmpty()) { + this.mode = GraphMode.NONE; + } + } else if ("acquire".equals(action)) { + if (epoch < this.epoch || this.released.containsKey(key)) { + return this.receipt(epoch, token, "released"); + } + checkState(this.active.isEmpty() ? this.mode == GraphMode.NONE : this.mode == GraphMode.LOADING, + "Cannot acquire loading ownership of manually set graph mode %s", this.mode); + checkState(this.active.size() < MAX_ACTIVE, "Loading owner limit %s reached", MAX_ACTIVE); + this.advanceRevision(); + this.active.put(key, principal); + this.mode = GraphMode.LOADING; + this.logChange("acquire", previousMode, 0); + return this.receipt(epoch, token, "active"); + } + // Unknown release cancels an acquire which may still be in transit. + if (epoch == this.epoch) { + String releasedOwner = this.released.get(key); + checkState(releasedOwner == null || releasedOwner.equals(principal), + "Loading owner belongs to another authenticated principal"); + if (releasedOwner != null) { + return this.receipt(epoch, token, "released"); + } + if (this.released.size() == MAX_RELEASED) { + checkState(this.epoch < Long.MAX_VALUE, "Loading owner epoch exhausted"); + if (owner == null) { + this.advanceRevision(); + } + this.epoch++; + this.released.clear(); + // Old nonactive tuples are terminal; active old tuples remain registered. + } else { + if (owner == null) { + this.advanceRevision(); + } + this.released.put(key, principal); + } + } + if (owner != null) { + this.logChange("release", previousMode, 1); + } + return this.receipt(epoch, token, "released"); + } + + private void logChange(String action, GraphMode previousMode, int clearedOwners) { + LOG.info("Loading ownership {} for graph '{}' instance {}: mode {} -> {}, epoch {}, " + + "activeOwners {}, clearedOwners {}", action, this.graphName.get(), this.instance, previousMode, this.mode, + this.epoch, this.active.size(), clearedOwners); + } + + private void advanceRevision() { + checkState(this.revision < Long.MAX_VALUE, "Loading state revision exhausted"); + this.revision++; + } + + private static void checkState(boolean condition, String message, Object... args) { + if (!condition) { + throw new ConflictException(String.format(message, args)); + } + } + + public static class ConflictException extends IllegalStateException { + + public ConflictException(String message) { + super(message); + } + } + + private Map receipt(long epoch, String token, String state) { + return ImmutableMap.builder().put("protocol", 1).put("instance", this.instance) + .put("epoch", epoch).put("token", token).put("state", state) + .put("mode", this.mode).build(); + } +} diff --git a/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/StandardHugeGraph.java b/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/StandardHugeGraph.java index 157ec6b74b..551d12556a 100644 --- a/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/StandardHugeGraph.java +++ b/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/StandardHugeGraph.java @@ -24,6 +24,7 @@ import java.util.Date; import java.util.Iterator; import java.util.List; +import java.util.Map; import java.util.Set; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.Future; @@ -176,7 +177,7 @@ public class StandardHugeGraph implements HugeGraph { private final RamTable ramtable; private volatile boolean started; private volatile boolean closed; - private volatile GraphMode mode; + private final LoadingMode loadingMode = new LoadingMode(this::spaceGraphName); private volatile GraphReadMode readMode; private volatile HugeVariables variables; private String graphSpace; @@ -224,7 +225,6 @@ public StandardHugeGraph(HugeConfig config) { this.started = false; this.closed = false; - this.mode = GraphMode.NONE; this.readMode = GraphReadMode.OLTP_ONLY; // Init process-wide static configs before lock, so that validation @@ -438,13 +438,29 @@ private void closeTx() { @Override public GraphMode mode() { - return this.mode; + return this.loadingMode.mode(); } @Override public void mode(GraphMode mode) { + this.loadingMode.mode(mode); LOG.info("Graph {} will work in {} mode", this, mode); - this.mode = mode; + } + + @Override + public Map loadingMode() { + return this.loadingMode.capability(); + } + + @Override + public Map loadingMode(String action, String instance, long epoch, + String token, String principal) { + return this.loadingMode.update(action, instance, epoch, token, principal); + } + + @Override + public Map recoverLoadingMode(String instance, long epoch, long revision) { + return this.loadingMode.recover(instance, epoch, revision); } @Override diff --git a/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/unit/UnitTestSuite.java b/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/unit/UnitTestSuite.java index 2ab8d4e8fa..810ddad60f 100644 --- a/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/unit/UnitTestSuite.java +++ b/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/unit/UnitTestSuite.java @@ -18,6 +18,7 @@ package org.apache.hugegraph.unit; import org.apache.hugegraph.HugeFactoryTest; +import org.apache.hugegraph.unit.core.LoadingModeTest; import org.apache.hugegraph.GraphDropPendingTest; import org.apache.hugegraph.StandardTaskAdmissionTest; import org.apache.hugegraph.DistributedTaskAdmissionTest; @@ -161,6 +162,7 @@ StandardAuthManagerV2Test.class, ContextTaskTest.class, HugeFactoryTest.class, + LoadingModeTest.class, GraphDropPendingTest.class, StandardTaskAdmissionTest.class, DistributedTaskAdmissionTest.class, diff --git a/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/unit/api/space/GraphSpaceAPITest.java b/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/unit/api/space/GraphSpaceAPITest.java index 1a0b585b0c..6783ff2b7e 100644 --- a/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/unit/api/space/GraphSpaceAPITest.java +++ b/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/unit/api/space/GraphSpaceAPITest.java @@ -19,20 +19,28 @@ import static org.apache.hugegraph.HugeFactory.SYS_GRAPH; +import java.io.IOException; import java.lang.reflect.Method; +import java.net.URI; import java.util.Collections; import java.util.Map; import java.util.Set; import java.util.concurrent.ConcurrentHashMap; import org.apache.hugegraph.HugeGraph; +import org.apache.hugegraph.HugeFactory; +import org.apache.hugegraph.LoadingMode; import org.apache.hugegraph.api.auth.ManagerAPI; +import org.apache.hugegraph.api.filter.AuthenticationFilter.Authorizer; import org.apache.hugegraph.api.profile.GraphsAPI; import org.apache.hugegraph.api.space.GraphSpaceAPI; import org.apache.hugegraph.auth.AuthManager; import org.apache.hugegraph.auth.HugeAuthenticator; import org.apache.hugegraph.auth.HugeDefaultRole; import org.apache.hugegraph.auth.HugeGraphAuthProxy; +import org.apache.hugegraph.auth.HugePermission; +import org.apache.hugegraph.auth.HugeResource; +import org.apache.hugegraph.auth.ResourceType; import org.apache.hugegraph.auth.HugeUser; import org.apache.hugegraph.auth.RolePermission; import org.apache.hugegraph.core.GraphManager; @@ -43,15 +51,27 @@ import org.apache.hugegraph.space.GraphSpace; import org.apache.hugegraph.testutil.Assert; import org.apache.hugegraph.testutil.Whitebox; +import org.apache.hugegraph.type.define.GraphMode; import org.apache.hugegraph.unit.BaseUnitTest; import org.apache.hugegraph.unit.FakeObjects; import org.apache.tinkerpop.gremlin.structure.Graph; import org.junit.After; import org.junit.Test; +import org.glassfish.jersey.internal.MapPropertiesDelegate; +import org.glassfish.jersey.server.ContainerRequest; +import org.glassfish.jersey.server.filter.RolesAllowedDynamicFeature; +import org.mockito.ArgumentCaptor; import org.mockito.Mockito; import jakarta.ws.rs.ForbiddenException; +import jakarta.ws.rs.container.ContainerRequestFilter; +import jakarta.ws.rs.container.ResourceInfo; +import jakarta.ws.rs.core.FeatureContext; +import jakarta.ws.rs.ClientErrorException; import jakarta.ws.rs.core.SecurityContext; +import jakarta.ws.rs.container.ContainerRequestContext; +import jakarta.ws.rs.core.MultivaluedHashMap; +import jakarta.ws.rs.core.UriInfo; public class GraphSpaceAPITest extends BaseUnitTest { @@ -243,6 +263,194 @@ public void testPdGraphsIncludesLoadedLocalConfigGraphs() { Assert.assertFalse(result.contains(SYS_GRAPH)); } + @Test + public void testUnproxiedLoadingOwnerUsesAuthenticatedPrincipal() throws Exception { + HugeConfig config = FakeObjects.newConfig(); + config.setProperty("backend", "memory"); + config.setProperty("serializer", "text"); + config.setProperty("store", "unproxied_loading_owner"); + HugeGraph graph = HugeFactory.open(config); + try { + GraphManager manager = Mockito.mock(GraphManager.class); + Mockito.when(manager.graph("DEFAULT", "hugegraph")).thenReturn(graph); + GraphsAPI api = new GraphsAPI(); + GraphsAPI.JsonLoadingOwner request = new GraphsAPI.JsonLoadingOwner(); + request.instance = (String) graph.loadingMode().get("instance"); + request.epoch = 0L; + request.token = java.util.UUID.randomUUID().toString(); + request.action = "acquire"; + setContext("alice"); + Assert.assertEquals("active", api.loadingMode(manager, "DEFAULT", "hugegraph", request, null).get("state")); + request.action = "release"; + setContext("bob"); + ClientErrorException conflict = org.junit.Assert.assertThrows(ClientErrorException.class, + () -> api.loadingMode(manager, "DEFAULT", "hugegraph", request, null)); + Assert.assertEquals(409, conflict.getResponse().getStatus()); + Assert.assertEquals(GraphMode.LOADING, graph.mode()); + setContext("alice"); + Assert.assertEquals("released", api.loadingMode(manager, "DEFAULT", "hugegraph", request, null).get("state")); + Assert.assertEquals(GraphMode.NONE, graph.mode()); + } finally { + graph.close(); + HugeFactory.remove(graph); + } + } + + @Test + public void testLoadingOwnerRoutesAllowScopedStatusWriter() throws Exception { + RolePermission role = RolePermission.role(GRAPHSPACE, GRAPH, HugePermission.WRITE); + role.roles().get(GRAPHSPACE).get(GRAPH).put(HugePermission.WRITE, + Collections.singletonMap(ResourceType.STATUS.name(), + Collections.singletonList(new HugeResource(ResourceType.STATUS, "*", null)))); + int endpoints = 0; + for (Method method : GraphsAPI.class.getDeclaredMethods()) { + if (!method.getName().equals("loadingMode")) { + continue; + } + endpoints++; + authorizeLoadingOwnerRoute(method, role, GRAPHSPACE, GRAPH); + authorizeLoadingOwnerRoute(method, RolePermission.admin(), GRAPHSPACE, GRAPH); + authorizeLoadingOwnerRoute(method, RolePermission.role(GRAPHSPACE, GRAPH, HugePermission.SPACE), + GRAPHSPACE, GRAPH); + Assert.assertThrows(ForbiddenException.class, + () -> authorizeLoadingOwnerRoute(method, role, "other_space", GRAPH)); + Assert.assertThrows(ForbiddenException.class, + () -> authorizeLoadingOwnerRoute(method, role, GRAPHSPACE, "other_graph")); + Assert.assertThrows(ForbiddenException.class, + () -> authorizeLoadingOwnerRoute(method, RolePermission.none(), GRAPHSPACE, GRAPH)); + } + Assert.assertEquals(2, endpoints); + } + + private static void authorizeLoadingOwnerRoute(Method method, RolePermission role, + String space, String graph) throws IOException { + GraphSpace graphSpace = new GraphSpace(space); + graphSpace.auth(true); + GraphManager manager = Mockito.mock(GraphManager.class); + Mockito.when(manager.graphSpace(space)).thenReturn(graphSpace); + MultivaluedHashMap parameters = new MultivaluedHashMap<>(); + parameters.putSingle("graphspace", space); + parameters.putSingle("name", graph); + UriInfo uri = Mockito.mock(UriInfo.class); + Mockito.when(uri.getPathParameters()).thenReturn(parameters); + Authorizer authorizer = new Authorizer(manager, new HugeAuthenticator.User("writer", role), uri); + ContainerRequestContext request = Mockito.mock(ContainerRequestContext.class); + Mockito.when(request.getSecurityContext()).thenReturn(authorizer); + ResourceInfo resource = Mockito.mock(ResourceInfo.class); + Mockito.when(resource.getResourceMethod()).thenReturn(method); + Mockito.doReturn(GraphsAPI.class).when(resource).getResourceClass(); + FeatureContext feature = Mockito.mock(FeatureContext.class); + new RolesAllowedDynamicFeature().configure(resource, feature); + ArgumentCaptor registered = ArgumentCaptor.forClass(Object.class); + Mockito.verify(feature).register(registered.capture()); + ((ContainerRequestFilter) registered.getValue()).filter(request); + } + + @Test + public void testAuthDisabledSpaceRecoveryStillRequiresGlobalAdmin() throws Exception { + HugeConfig config = FakeObjects.newConfig(); + config.setProperty("backend", "memory"); + config.setProperty("serializer", "text"); + config.setProperty("store", "unproxied_loading_recovery"); + HugeGraph graph = HugeFactory.open(config); + try { + Assert.assertFalse(graph instanceof HugeGraphAuthProxy); + String space = "open_space"; + GraphSpace graphSpace = new GraphSpace(space); + graphSpace.auth(false); + GraphManager manager = Mockito.mock(GraphManager.class); + Mockito.when(manager.requireAuthentication()).thenReturn(true); + Mockito.when(manager.graphSpace(space)).thenReturn(graphSpace); + Mockito.when(manager.graph(space, "hugegraph")).thenReturn(graph); + UriInfo uri = Mockito.mock(UriInfo.class); + MultivaluedHashMap parameters = new MultivaluedHashMap<>(); + parameters.putSingle("graphspace", space); + Mockito.when(uri.getPathParameters()).thenReturn(parameters); + Authorizer restricted = new Authorizer(manager, + new HugeAuthenticator.User("bob", RolePermission.none()), uri); + Assert.assertTrue(restricted.isUserInRole("admin")); + Assert.assertFalse(RolePermission.isAdmin(restricted.role())); + URI endpoint = URI.create("http://localhost/graphspaces/open_space/graphs/hugegraph/mode/owners"); + ContainerRequest context = new ContainerRequest(endpoint, endpoint, "PUT", restricted, + new MapPropertiesDelegate()); + GraphsAPI api = new GraphsAPI(); + Map state = graph.loadingMode(); + String instance = (String) state.get("instance"); + String token = java.util.UUID.randomUUID().toString(); + graph.loadingMode("acquire", instance, 0, token, "alice"); + Map before = api.loadingMode(manager, space, "hugegraph"); + GraphsAPI.JsonLoadingOwner request = new GraphsAPI.JsonLoadingOwner(); + request.action = "recover"; + request.instance = instance; + request.epoch = (Long) before.get("epoch"); + request.revision = (Long) before.get("revision"); + request.confirm = "I'm sure all imports have stopped"; + ForbiddenException denied = org.junit.Assert.assertThrows(ForbiddenException.class, + () -> api.loadingMode(manager, space, "hugegraph", request, context)); + Assert.assertEquals(403, denied.getResponse().getStatus()); + Assert.assertEquals(before, graph.loadingMode()); + Assert.assertEquals(GraphMode.LOADING, graph.mode()); + + for (HugeAuthenticator.User user : new HugeAuthenticator.User[]{HugeAuthenticator.User.ADMIN, + HugeAuthenticator.User.ANONYMOUS}) { + Mockito.when(manager.requireAuthentication()).thenReturn(user != HugeAuthenticator.User.ANONYMOUS); + context.setSecurityContext(new Authorizer(manager, user, uri)); + state = graph.loadingMode(); + request.epoch = (Long) state.get("epoch"); + graph.loadingMode("acquire", instance, request.epoch, token, "alice"); + request.revision = (Long) graph.loadingMode().get("revision"); + Assert.assertEquals("recovered", api.loadingMode(manager, space, "hugegraph", request, + context).get("state")); + Assert.assertEquals(GraphMode.NONE, graph.mode()); + } + } finally { + graph.close(); + HugeFactory.remove(graph); + } + } + + @Test + public void testLoadingRecoveryConfirmationAndConflictResponse() { + GraphManager manager = Mockito.mock(GraphManager.class); + HugeGraph graph = Mockito.mock(HugeGraph.class); + Mockito.when(manager.graph("DEFAULT", "hugegraph")).thenReturn(graph); + GraphsAPI api = new GraphsAPI(); + GraphsAPI.JsonLoadingOwner request = new GraphsAPI.JsonLoadingOwner(); + request.action = "recover"; + request.instance = "instance"; + request.epoch = 0L; + ContainerRequestContext requestContext = Mockito.mock(ContainerRequestContext.class); + Authorizer authorizer = Mockito.mock(Authorizer.class); + Mockito.when(requestContext.getSecurityContext()).thenReturn(authorizer); + Mockito.when(authorizer.role()).thenReturn(RolePermission.none()); + Assert.assertThrows(ForbiddenException.class, + () -> api.loadingMode(manager, "DEFAULT", "hugegraph", request, requestContext)); + Mockito.verify(graph, Mockito.never()).recoverLoadingMode(Mockito.anyString(), Mockito.anyLong(), + Mockito.anyLong()); + Mockito.when(authorizer.role()).thenReturn(RolePermission.admin()); + Assert.assertThrows(IllegalArgumentException.class, + () -> api.loadingMode(manager, "DEFAULT", "hugegraph", request, requestContext)); + Mockito.verify(graph, Mockito.never()).loadingMode(Mockito.anyString(), Mockito.anyString(), + Mockito.anyLong(), Mockito.any(), Mockito.anyString()); + request.confirm = "I'm sure all imports have stopped"; + Assert.assertThrows(IllegalArgumentException.class, + () -> api.loadingMode(manager, "DEFAULT", "hugegraph", request, requestContext)); + request.revision = 0L; + Map receipt = Collections.singletonMap("state", "recovered"); + Mockito.when(graph.recoverLoadingMode("instance", 0, 0)).thenReturn(receipt); + Assert.assertSame(receipt, api.loadingMode(manager, "DEFAULT", "hugegraph", request, requestContext)); + + Mockito.when(graph.recoverLoadingMode("instance", 0, 0)) + .thenThrow(new LoadingMode.ConflictException("stale epoch")); + ClientErrorException ownerConflict = org.junit.Assert.assertThrows(ClientErrorException.class, + () -> api.loadingMode(manager, "DEFAULT", "hugegraph", request, requestContext)); + Assert.assertEquals(409, ownerConflict.getResponse().getStatus()); + Mockito.doThrow(new LoadingMode.ConflictException("active owners")).when(graph).mode(GraphMode.NONE); + ClientErrorException modeConflict = org.junit.Assert.assertThrows(ClientErrorException.class, + () -> api.mode(manager, "DEFAULT", "hugegraph", GraphMode.NONE)); + Assert.assertEquals(409, modeConflict.getResponse().getStatus()); + } + @Test public void testAdminCanListGraphsWithoutGraphReadRole() { GraphManager manager = managerWithLoadedDefaultGraph(ADMIN, true); diff --git a/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/unit/auth/HugeGraphAuthProxyTest.java b/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/unit/auth/HugeGraphAuthProxyTest.java index 13216d2164..0756724b1b 100644 --- a/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/unit/auth/HugeGraphAuthProxyTest.java +++ b/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/unit/auth/HugeGraphAuthProxyTest.java @@ -62,6 +62,42 @@ public class HugeGraphAuthProxyTest extends BaseUnitTest { + @Test + public void testLoadingOwnershipRequiresStatusWriteAndUsesAuthenticatedPrincipal() { + HugeGraph graph = Mockito.mock(HugeGraph.class); + HugeConfig config = Mockito.mock(HugeConfig.class); + Mockito.when(graph.name()).thenReturn("tc13_mode_auth"); + Mockito.when(graph.graphSpace()).thenReturn("DEFAULT"); + Mockito.when(graph.spaceGraphName()).thenReturn("tc13_mode_auth"); + Mockito.when(graph.configuration()).thenReturn(config); + Mockito.when(graph.authManager()).thenReturn(Mockito.mock(AuthManager.class)); + Mockito.when(graph.taskScheduler()).thenReturn(Mockito.mock(TaskScheduler.class)); + Mockito.when(config.get(AuthOptions.AUTH_CACHE_EXPIRE)).thenReturn(3600L); + Mockito.when(config.get(AuthOptions.AUTH_CACHE_CAPACITY)).thenReturn(100L); + Mockito.when(config.get(AuthOptions.AUTH_AUDIT_LOG_RATE)).thenReturn(1000D); + HugeGraphAuthProxy proxy = new HugeGraphAuthProxy(graph); + setContext(new HugeGraphAuthProxy.Context(new HugeAuthenticator.User( + "reader", RolePermission.role("DEFAULT", "tc13_mode_auth", HugePermission.READ)))); + Assert.assertThrows(jakarta.ws.rs.ForbiddenException.class, (Assert.ThrowableRunnable) proxy::loadingMode); + Assert.assertThrows(jakarta.ws.rs.ForbiddenException.class, + () -> proxy.loadingMode("acquire", "instance", 0, "token", "forged")); + Mockito.verify(graph, Mockito.never()).loadingMode(); + Mockito.verify(graph, Mockito.never()).loadingMode(Mockito.anyString(), Mockito.anyString(), + Mockito.anyLong(), Mockito.anyString(), Mockito.anyString()); + setContext(new HugeGraphAuthProxy.Context(new HugeAuthenticator.User( + "writer", RolePermission.role("DEFAULT", "tc13_mode_auth", HugePermission.WRITE)))); + proxy.loadingMode(); + proxy.loadingMode("acquire", "instance", 0, "token", "forged"); + Mockito.verify(graph).loadingMode(); + Mockito.verify(graph).loadingMode("acquire", "instance", 0, "token", "writer"); + Assert.assertThrows(jakarta.ws.rs.ForbiddenException.class, + () -> proxy.recoverLoadingMode("instance", 0, 0)); + Mockito.verify(graph, Mockito.never()).recoverLoadingMode("instance", 0, 0); + setContext(new HugeGraphAuthProxy.Context(new HugeAuthenticator.User("admin", RolePermission.admin()))); + proxy.recoverLoadingMode("instance", 0, 0); + Mockito.verify(graph).recoverLoadingMode("instance", 0, 0); + } + @Test public void testJdk17ReflectionFilters() { Reflection.registerFieldsToFilter(ReflectionFilterTarget.class, "field"); diff --git a/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/unit/core/LoadingModeTest.java b/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/unit/core/LoadingModeTest.java new file mode 100644 index 0000000000..2431379c6e --- /dev/null +++ b/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/unit/core/LoadingModeTest.java @@ -0,0 +1,360 @@ +/* + * 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.hugegraph.unit.core; + +import java.util.ArrayList; +import java.util.List; +import java.util.Map; +import java.util.UUID; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicReference; + +import org.apache.hugegraph.LoadingMode; +import org.apache.hugegraph.type.define.GraphMode; +import org.apache.logging.log4j.Level; +import org.apache.logging.log4j.LogManager; +import org.apache.logging.log4j.core.LogEvent; +import org.apache.logging.log4j.core.LoggerContext; +import org.apache.logging.log4j.core.appender.AbstractAppender; +import org.apache.logging.log4j.core.config.LoggerConfig; +import org.apache.logging.log4j.core.config.Property; +import org.junit.Assert; +import org.junit.Test; + +public class LoadingModeTest { + + @Test + public void testOwnerAuditRecordsOnlySuccessfulChangesWithoutTokens() { + List events = new ArrayList<>(); + AbstractAppender appender = new AbstractAppender("LoadingModeTestAppender", null, null, + true, Property.EMPTY_ARRAY) { + @Override + public void append(LogEvent event) { + events.add(event.toImmutable()); + } + }; + appender.start(); + LoggerContext context = (LoggerContext) LogManager.getContext(false); + org.apache.logging.log4j.core.config.Configuration configuration = context.getConfiguration(); + String loggerName = LoadingMode.class.getName(); + LoggerConfig previous = configuration.getLoggers().get(loggerName); + LoggerConfig logger = new LoggerConfig(loggerName, Level.INFO, false); + logger.addAppender(appender, Level.INFO, null); + configuration.addLogger(loggerName, logger); + context.updateLoggers(); + try { + AtomicReference graphName = new AtomicReference<>("test-space-test-graph"); + LoadingMode mode = new LoadingMode(graphName::get); + Map capability = mode.capability(); + String first = token(); + String second = token(); + update(mode, capability, first, "acquire"); + update(mode, capability, first, "acquire"); + update(mode, capability, second, "acquire"); + Assert.assertThrows(LoadingMode.ConflictException.class, + () -> mode.update("release", (String) capability.get("instance"), + 0, first, "other-user")); + update(mode, capability, first, "release"); + update(mode, capability, first, "release"); + update(mode, capability, second, "release"); + update(mode, capability, first, "acquire"); + Assert.assertEquals(4, events.size()); + Assert.assertTrue(message(events, 0).contains("mode NONE -> LOADING, epoch 0, activeOwners 1")); + Assert.assertTrue(message(events, 1).contains("mode LOADING -> LOADING, epoch 0, activeOwners 2")); + Assert.assertTrue(message(events, 2).contains("activeOwners 1, clearedOwners 1")); + Assert.assertTrue(message(events, 3).contains("mode LOADING -> NONE, epoch 0, activeOwners 0")); + update(mode, capability, token(), "acquire"); + update(mode, capability, token(), "acquire"); + Map current = mode.capability(); + mode.recover((String) current.get("instance"), 0, ((Number) current.get("revision")).longValue()); + Assert.assertTrue(message(events, 6).contains("mode LOADING -> NONE, epoch 1, " + + "activeOwners 0, clearedOwners 2")); + Assert.assertThrows(LoadingMode.ConflictException.class, + () -> mode.recover((String) capability.get("instance"), 0, 0)); + current = mode.capability(); + mode.recover((String) current.get("instance"), 1, ((Number) current.get("revision")).longValue()); + Assert.assertEquals(8, events.size()); + Assert.assertTrue(message(events, 7).contains("mode NONE -> NONE, epoch 2, " + + "activeOwners 0, clearedOwners 0")); + for (LogEvent event : events) { + Assert.assertEquals(Level.INFO, event.getLevel()); + Assert.assertTrue(event.getMessage().getFormattedMessage().contains( + "graph 'test-space-test-graph' instance " + capability.get("instance"))); + Assert.assertFalse(event.getMessage().getFormattedMessage().contains(first)); + Assert.assertFalse(event.getMessage().getFormattedMessage().contains(second)); + Assert.assertFalse(event.getMessage().getFormattedMessage().contains("other-user")); + } + graphName.set("renamed-space-test-graph"); + recover(mode, mode.capability()); + Assert.assertEquals(9, events.size()); + Assert.assertTrue(message(events, 8).contains("graph 'renamed-space-test-graph' instance " + + capability.get("instance"))); + } finally { + configuration.removeLogger(loggerName); + if (previous != null) { + configuration.addLogger(loggerName, previous); + } + context.updateLoggers(); + appender.stop(); + } + } + + private static String message(List events, int index) { + return events.get(index).getMessage().getFormattedMessage(); + } + + @Test + public void testTwoOwnersAndIdempotentRetry() { + LoadingMode mode = new LoadingMode(() -> "test-space-test-graph"); + Map capability = mode.capability(); + String a = token(); + String b = token(); + Assert.assertEquals("active", update(mode, capability, a, "acquire").get("state")); + update(mode, capability, a, "acquire"); + update(mode, capability, b, "acquire"); + update(mode, capability, a, "release"); + update(mode, capability, a, "release"); + Assert.assertEquals(GraphMode.LOADING, mode.mode()); + Assert.assertEquals("released", update(mode, capability, a, "acquire").get("state")); + update(mode, capability, b, "release"); + Assert.assertEquals(GraphMode.NONE, mode.mode()); + } + + @Test + public void testUnknownReleaseCancelsLateAcquire() { + LoadingMode mode = new LoadingMode(() -> "test-space-test-graph"); + Map capability = mode.capability(); + String token = token(); + update(mode, capability, token, "release"); + Assert.assertEquals("released", update(mode, capability, token, "acquire").get("state")); + Assert.assertEquals(GraphMode.NONE, mode.mode()); + } + + @Test + public void testTerminalCompactionKeepsOldActiveAndRejectsLateAcquire() { + LoadingMode mode = new LoadingMode(() -> "test-space-test-graph"); + Map original = mode.capability(); + String b = token(); + String cancelled = token(); + update(mode, original, b, "acquire"); + update(mode, original, cancelled, "release"); + for (int i = 0; i < 2200; i++) { + update(mode, mode.capability(), token(), "release"); + } + Assert.assertTrue(((Number) mode.capability().get("epoch")).longValue() >= 2); + Assert.assertEquals("released", update(mode, original, cancelled, "acquire").get("state")); + Assert.assertEquals("released", update(mode, original, token(), "acquire").get("state")); + Assert.assertEquals("active", update(mode, original, b, "acquire").get("state")); + Assert.assertEquals(GraphMode.LOADING, mode.mode()); + String c = token(); + Map current = mode.capability(); + update(mode, current, c, "acquire"); + update(mode, original, b, "release"); + Assert.assertEquals(GraphMode.LOADING, mode.mode()); + update(mode, current, c, "release"); + Assert.assertEquals(GraphMode.NONE, mode.mode()); + } + + @Test + public void testLegacySetterCannotOverwriteOwnersOrAdoptManualMode() { + LoadingMode mode = new LoadingMode(() -> "test-space-test-graph"); + Map capability = mode.capability(); + String a = token(); + for (GraphMode manual : GraphMode.values()) { + if (manual == GraphMode.NONE) { + continue; + } + mode.mode(manual); + Assert.assertThrows(IllegalStateException.class, () -> update(mode, capability, a, "acquire")); + Assert.assertEquals(manual, mode.mode()); + } + mode.mode(GraphMode.NONE); + update(mode, capability, a, "acquire"); + for (GraphMode legacy : GraphMode.values()) { + Assert.assertThrows(IllegalStateException.class, () -> mode.mode(legacy)); + Assert.assertEquals(GraphMode.LOADING, mode.mode()); + } + update(mode, capability, a, "release"); + mode.mode(GraphMode.NONE); + Assert.assertThrows(NullPointerException.class, () -> mode.mode(null)); + } + + @Test + public void testInstanceAndPrincipalBoundaries() { + LoadingMode mode = new LoadingMode(() -> "test-space-test-graph"); + Map capability = mode.capability(); + String a = token(); + update(mode, capability, a, "acquire"); + String instance = (String) capability.get("instance"); + Assert.assertThrows(IllegalStateException.class, + () -> mode.update("release", instance, 0, a, "another-user")); + Assert.assertEquals(GraphMode.LOADING, mode.mode()); + Assert.assertThrows(IllegalArgumentException.class, + () -> mode.update("acquire", token(), 0, a, "user")); + Assert.assertThrows(IllegalArgumentException.class, + () -> mode.update("acquire", instance, 1, a, "user")); + update(mode, capability, a, "release"); + } + + @Test + public void testRecoveryRetiresOwnersAndLateRequests() { + LoadingMode mode = new LoadingMode(() -> "test-space-test-graph"); + Map before = mode.capability(); + String lost = token(); + update(mode, before, lost, "acquire"); + + Map recovered = recover(mode, mode.capability()); + Assert.assertEquals("recovered", recovered.get("state")); + Assert.assertEquals(1L, recovered.get("epoch")); + Assert.assertFalse(recovered.containsKey("token")); + Assert.assertEquals(GraphMode.NONE, mode.mode()); + Assert.assertEquals("released", update(mode, before, lost, "acquire").get("state")); + Assert.assertEquals("released", update(mode, before, token(), "acquire").get("state")); + update(mode, before, lost, "release"); + + Map current = mode.capability(); + update(mode, current, token(), "acquire"); + Assert.assertThrows(IllegalStateException.class, + () -> recover(mode, before)); + Assert.assertEquals(GraphMode.LOADING, mode.mode()); + Assert.assertEquals(1L, mode.capability().get("epoch")); + } + + @Test + public void testRecoveryCannotClearOwnerAddedAfterInspection() { + LoadingMode mode = new LoadingMode(() -> "test-space-test-graph"); + Map capability = mode.capability(); + String first = token(); + String second = token(); + update(mode, capability, first, "acquire"); + Map inspected = mode.capability(); + update(mode, inspected, second, "acquire"); + Assert.assertThrows(IllegalStateException.class, + () -> recover(mode, inspected)); + Assert.assertEquals(GraphMode.LOADING, mode.mode()); + update(mode, capability, first, "release"); + Assert.assertEquals(GraphMode.LOADING, mode.mode()); + update(mode, capability, second, "release"); + Assert.assertEquals(GraphMode.NONE, mode.mode()); + } + + @Test + public void testRecoveryAllowsConfirmedManualModeReset() { + LoadingMode mode = new LoadingMode(() -> "test-space-test-graph"); + mode.mode(GraphMode.RESTORING); + Map before = mode.capability(); + recover(mode, before); + Assert.assertEquals(GraphMode.NONE, mode.mode()); + Assert.assertEquals("active", update(mode, mode.capability(), token(), "acquire").get("state")); + } + + @Test + public void testSuccessfulSameModeWriteInvalidatesRecoverySnapshot() { + LoadingMode mode = new LoadingMode(() -> "test-space-test-graph"); + for (GraphMode manual : GraphMode.values()) { + mode.mode(manual); + Map inspected = mode.capability(); + mode.mode(manual); + Assert.assertEquals(((Number) inspected.get("revision")).longValue() + 1, + mode.capability().get("revision")); + Assert.assertThrows(LoadingMode.ConflictException.class, () -> recover(mode, inspected)); + Assert.assertEquals(manual, mode.mode()); + } + } + + @Test + public void testOwnerAdmissionLimitPreservesExistingOwnersAndRecovery() { + LoadingMode mode = new LoadingMode(() -> "test-space-test-graph"); + Map capability = mode.capability(); + String first = token(); + update(mode, capability, first, "acquire"); + for (int i = 1; i < 1024; i++) { + update(mode, capability, token(), "acquire"); + } + Map full = mode.capability(); + String rejected = token(); + LoadingMode.ConflictException failure = Assert.assertThrows(LoadingMode.ConflictException.class, + () -> update(mode, capability, rejected, "acquire")); + Assert.assertTrue(failure.getMessage().contains("1024")); + Assert.assertEquals(full, mode.capability()); + Assert.assertEquals(GraphMode.LOADING, mode.mode()); + Assert.assertEquals("active", update(mode, capability, first, "acquire").get("state")); + Assert.assertEquals(full, mode.capability()); + Assert.assertEquals("released", update(mode, capability, first, "release").get("state")); + Assert.assertEquals("active", update(mode, capability, rejected, "acquire").get("state")); + Map recovered = recover(mode, mode.capability()); + Assert.assertEquals(GraphMode.NONE, recovered.get("mode")); + Assert.assertEquals("active", update(mode, mode.capability(), token(), "acquire").get("state")); + } + + @Test + public void testConcurrentAcquireAndLegacySetterAreAtomic() throws Exception { + LoadingMode mode = new LoadingMode(() -> "test-space-test-graph"); + Map capability = mode.capability(); + String a = token(); + CountDownLatch start = new CountDownLatch(1); + ExecutorService executor = Executors.newFixedThreadPool(2); + try { + Future acquire = executor.submit(() -> { + Assert.assertTrue(start.await(5, TimeUnit.SECONDS)); + try { + update(mode, capability, a, "acquire"); + return true; + } catch (IllegalStateException expected) { + return false; + } + }); + Future legacy = executor.submit(() -> { + Assert.assertTrue(start.await(5, TimeUnit.SECONDS)); + try { + mode.mode(GraphMode.RESTORING); + return true; + } catch (IllegalStateException expected) { + return false; + } + }); + start.countDown(); + boolean acquired = acquire.get(10, TimeUnit.SECONDS); + Assert.assertNotEquals(acquired, legacy.get(10, TimeUnit.SECONDS)); + Assert.assertEquals(acquired ? GraphMode.LOADING : GraphMode.RESTORING, mode.mode()); + } finally { + executor.shutdownNow(); + Assert.assertTrue(executor.awaitTermination(10, TimeUnit.SECONDS)); + } + } + + private static String token() { + return UUID.randomUUID().toString(); + } + + private static Map recover(LoadingMode mode, Map capability) { + return mode.recover((String) capability.get("instance"), + ((Number) capability.get("epoch")).longValue(), + ((Number) capability.get("revision")).longValue()); + } + + private static Map update(LoadingMode mode, Map capability, + String token, String action) { + return mode.update(action, (String) capability.get("instance"), + ((Number) capability.get("epoch")).longValue(), token, "user"); + } +}