diff --git a/docs/server-runtime.md b/docs/server-runtime.md index f370951a7f..fcd8c76d12 100644 --- a/docs/server-runtime.md +++ b/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":"","epoch":0,"revision":0}`. `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. +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":"","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. diff --git a/hugegraph-server/README.md b/hugegraph-server/README.md index dc2ab3618e..d533d82cf7 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. ## 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..845876a0d0 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,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 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 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 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 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") 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..02bd6d2fe7 --- /dev/null +++ b/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/LoadingMode.java @@ -0,0 +1,177 @@ +/* + * 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 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_RELEASED = 1024; + + 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 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"); + if (this.mode != 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); + 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.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..6bc061819c 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(); 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 3cd65b0d8f..300edb4b10 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; @@ -157,6 +158,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..b044579843 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 @@ -26,6 +26,8 @@ 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.profile.GraphsAPI; import org.apache.hugegraph.api.space.GraphSpaceAPI; @@ -43,6 +45,7 @@ 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; @@ -51,6 +54,7 @@ import org.mockito.Mockito; import jakarta.ws.rs.ForbiddenException; +import jakarta.ws.rs.ClientErrorException; import jakarta.ws.rs.core.SecurityContext; public class GraphSpaceAPITest extends BaseUnitTest { @@ -243,6 +247,78 @@ 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 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; + SecurityContext securityContext = Mockito.mock(SecurityContext.class); + Assert.assertThrows(ForbiddenException.class, + () -> api.loadingMode(manager, "DEFAULT", "hugegraph", request, securityContext)); + Mockito.verify(graph, Mockito.never()).recoverLoadingMode(Mockito.anyString(), Mockito.anyLong(), + Mockito.anyLong()); + Mockito.when(securityContext.isUserInRole("admin")).thenReturn(true); + Assert.assertThrows(IllegalArgumentException.class, + () -> api.loadingMode(manager, "DEFAULT", "hugegraph", request, securityContext)); + 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, securityContext)); + 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, securityContext)); + + 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, securityContext)); + 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..4ff3a8ae2e --- /dev/null +++ b/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/unit/core/LoadingModeTest.java @@ -0,0 +1,312 @@ +/* + * 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 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 { + LoadingMode mode = new LoadingMode(); + 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.assertFalse(event.getMessage().getFormattedMessage().contains(first)); + Assert.assertFalse(event.getMessage().getFormattedMessage().contains(second)); + Assert.assertFalse(event.getMessage().getFormattedMessage().contains("other-user")); + } + } 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(); + 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(); + 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(); + 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(); + 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(); + 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(); + 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(); + 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(); + 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 testConcurrentAcquireAndLegacySetterAreAtomic() throws Exception { + LoadingMode mode = new LoadingMode(); + 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"); + } +}