From 5fc74e81dd09f5453025f78ef657718d734fdc85 Mon Sep 17 00:00:00 2001 From: contrueCT Date: Thu, 17 Sep 2026 14:49:38 +0800 Subject: [PATCH 1/2] fix(hstore): fall back when ordered scan workers are denied --- .../store/client/OrderedScanSecurityTest.java | 108 ++++++++++++ .../apache/hugegraph/unit/UnitTestSuite.java | 2 + .../store/client/OrderedKvIterator.java | 33 +++- .../store/client/OrderedKvIteratorTest.java | 157 +++++++++++++++++- 4 files changed, 294 insertions(+), 6 deletions(-) create mode 100644 hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/store/client/OrderedScanSecurityTest.java diff --git a/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/store/client/OrderedScanSecurityTest.java b/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/store/client/OrderedScanSecurityTest.java new file mode 100644 index 0000000000..6d81e66487 --- /dev/null +++ b/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/store/client/OrderedScanSecurityTest.java @@ -0,0 +1,108 @@ +/* + * 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.store.client; + +import java.util.Arrays; +import java.util.Collections; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.ThreadPoolExecutor; +import java.util.concurrent.TimeUnit; + +import javax.script.ScriptException; +import javax.script.SimpleBindings; + +import org.apache.hugegraph.security.HugeSecurityManager; +import org.apache.hugegraph.store.HgKvEntry; +import org.apache.hugegraph.store.HgKvIterator; +import org.apache.hugegraph.store.client.util.ExecutorPool; +import org.apache.tinkerpop.gremlin.groovy.jsr223.GremlinGroovyScriptEngine; +import org.junit.Assert; +import org.junit.Test; +import org.mockito.Mockito; + +public class OrderedScanSecurityTest { + + @Test + public void testOrderedScanFromGremlin() throws Exception { + assertScanUnderSandbox(null); + } + + @Test + public void testOrderedScanAfterWorkerExpiry() throws Exception { + ThreadPoolExecutor executor = ExecutorPool.createExecutor( + "ordered-scan-expiry-test", 1L, 0, 2, + new ThreadPoolExecutor.AbortPolicy()); + try { + executor.submit(() -> { }).get(3L, TimeUnit.SECONDS); + long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(5L); + while (executor.getPoolSize() != 0 && System.nanoTime() < deadline) { + Thread.sleep(10L); + } + Assert.assertEquals(0, executor.getPoolSize()); + assertScanUnderSandbox(executor); + } finally { + executor.shutdownNow(); + } + } + + @SuppressWarnings("unchecked") + private static void assertScanUnderSandbox(ExecutorService executor) throws Exception { + HgKvIterator first = Mockito.mock(HgKvIterator.class); + HgKvIterator second = Mockito.mock(HgKvIterator.class); + GremlinGroovyScriptEngine engine = new GremlinGroovyScriptEngine(); + try (OrderedKvIterator scan = executor == null ? + new OrderedKvIterator(Arrays.asList(first, second), 0L) : + new OrderedKvIterator(Arrays.asList(first, second), 0L, executor)) { + engine.eval("1 + 1"); + SimpleBindings bindings = new SimpleBindings(); + // Prime class loading without creating an initializer worker. + HgKvIterator warmup = Mockito.mock(HgKvIterator.class); + try (OrderedKvIterator single = new OrderedKvIterator( + Collections.singletonList(warmup), 0L)) { + bindings.put("scan", single); + Assert.assertEquals(false, engine.eval("scan.hasNext()", bindings)); + } + bindings.put("scan", scan); + String factoryScript = "org.apache.hugegraph.store.client.util.ExecutorPool." + + "newThreadFactory('untrusted').newThread({} as Runnable)"; + engine.eval(factoryScript); + SecurityManager previous = System.getSecurityManager(); + String name = Thread.currentThread().getName(); + Thread.currentThread().setName("gremlin-server-exec-ordered-scan-test"); + System.setSecurityManager(new HugeSecurityManager()); + try { + ScriptException denied = Assert.assertThrows(ScriptException.class, + () -> engine.eval("new Thread()")); + Assert.assertTrue(denied.getMessage().contains( + "Not allowed to access thread group via Gremlin")); + Assert.assertEquals(false, engine.eval("scan.hasNext()", bindings)); + denied = Assert.assertThrows(ScriptException.class, + () -> engine.eval(factoryScript)); + Assert.assertTrue(denied.getMessage().contains( + "Not allowed to access thread group via Gremlin")); + } finally { + System.setSecurityManager(previous); + Thread.currentThread().setName(name); + } + } + Mockito.verify(first).hasNext(); + Mockito.verify(second).hasNext(); + Mockito.verify(first).close(); + Mockito.verify(second).close(); + } +} 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 d48738b840..d5e43a37ad 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 @@ -27,6 +27,7 @@ import org.apache.hugegraph.meta.EtcdMetaDriverTest; import org.apache.hugegraph.meta.MetaManagerSchemaCacheClearEventTest; import org.apache.hugegraph.meta.managers.AuthMetaManagerTest; +import org.apache.hugegraph.store.client.OrderedScanSecurityTest; import org.apache.hugegraph.traversal.optimize.TraversalUtilOptimizeTest; import org.apache.hugegraph.unit.api.auth.LoginAPITest; import org.apache.hugegraph.unit.api.filter.AccessLogFilterTest; @@ -158,6 +159,7 @@ QueryResultsTest.class, RangeTest.class, SecurityManagerTest.class, + OrderedScanSecurityTest.class, RolePermissionTest.class, ExceptionTest.class, GraphManagerAdminInitTest.class, diff --git a/hugegraph-store/hg-store-client/src/main/java/org/apache/hugegraph/store/client/OrderedKvIterator.java b/hugegraph-store/hg-store-client/src/main/java/org/apache/hugegraph/store/client/OrderedKvIterator.java index 3410b216e8..ed65706300 100644 --- a/hugegraph-store/hg-store-client/src/main/java/org/apache/hugegraph/store/client/OrderedKvIterator.java +++ b/hugegraph-store/hg-store-client/src/main/java/org/apache/hugegraph/store/client/OrderedKvIterator.java @@ -192,9 +192,14 @@ private void initialize() { new ExecutorCompletionService<>(this.initializer); int nextSource = 0; int inFlight = 0; + boolean useExecutor = true; while (nextSource < this.iterators.size() || inFlight > 0) { while (nextSource < this.iterators.size() && inFlight < INITIALIZE_THREADS) { + if (!useExecutor) { + this.addFirst(this.firstEntry(nextSource++)); + continue; + } int source = nextSource; try { futures.add(completions.submit( @@ -209,6 +214,13 @@ private void initialize() { break; } this.addFirst(this.firstEntry(nextSource++)); + } catch (SecurityException e) { + // Gremlin's sandbox can deny lazy worker creation, + // including after idle workers expire. Keep its policy + // intact and initialize remaining sources on the caller. + // Submitted tasks are still drained below; source errors + // retain the normal cancellation/close behavior. + useExecutor = false; } } if (inFlight > 0) { @@ -221,16 +233,19 @@ private void initialize() { } } } catch (InterruptedException e) { - this.cancel(futures); Thread.currentThread().interrupt(); + this.cancel(futures, e); throw this.initializationFailure( new IllegalStateException( "Interrupted while initializing ordered scan", e)); } catch (ExecutionException e) { - this.cancel(futures); + this.cancel(futures, e.getCause()); throw this.initializationFailure(e.getCause()); } catch (RuntimeException | Error e) { - this.cancel(futures); + if (interruption(e) != null) { + Thread.currentThread().interrupt(); + } + this.cancel(futures, e); this.closeAfterFailure(e); throw e; } @@ -264,10 +279,18 @@ private SourceEntry firstEntry(int source) { return new SourceEntry(source, iterator.next()); } - private void cancel(List> futures) { + private void cancel(List> futures, Throwable failure) { for (Future future : futures) { if (!future.isDone()) { - future.cancel(true); + try { + future.cancel(true); + } catch (RuntimeException | Error cancelFailure) { + // The sandbox may also deny interrupting an existing worker. + // Keep closing sources and preserve the original scan error. + if (cancelFailure != failure) { + failure.addSuppressed(cancelFailure); + } + } } } } diff --git a/hugegraph-store/hg-store-test/src/main/java/org/apache/hugegraph/store/client/OrderedKvIteratorTest.java b/hugegraph-store/hg-store-test/src/main/java/org/apache/hugegraph/store/client/OrderedKvIteratorTest.java index 3cad083b60..cc4648bd7f 100644 --- a/hugegraph-store/hg-store-test/src/main/java/org/apache/hugegraph/store/client/OrderedKvIteratorTest.java +++ b/hugegraph-store/hg-store-test/src/main/java/org/apache/hugegraph/store/client/OrderedKvIteratorTest.java @@ -23,17 +23,21 @@ import java.util.List; import java.util.NoSuchElementException; import java.util.concurrent.AbstractExecutorService; +import java.util.concurrent.Callable; import java.util.concurrent.CountDownLatch; import java.util.concurrent.ExecutionException; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import java.util.concurrent.Future; +import java.util.concurrent.FutureTask; import java.util.concurrent.RejectedExecutionException; +import java.util.concurrent.RunnableFuture; import java.util.concurrent.SynchronousQueue; import java.util.concurrent.ThreadPoolExecutor; import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeoutException; import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicInteger; import java.util.concurrent.atomic.AtomicReference; import org.apache.hugegraph.store.HgKvEntry; @@ -43,6 +47,131 @@ public class OrderedKvIteratorTest { + @Test + public void testSecurityDeniedSubmissionFallsBackWithoutRetrying() { + for (int allowed : new int[]{0, 1}) { + DirectExecutorService executor = new DirectExecutorService(); + executor.allowedExecutions = allowed; + TestIterator first = new TestIterator(1, 4); + TestIterator second = new TestIterator(2, 3); + TestIterator third = new TestIterator(5); + try (OrderedKvIterator iterator = new OrderedKvIterator( + Arrays.asList(first, second, third), 0L, executor)) { + Assert.assertEquals(Arrays.asList(1, 2, 3, 4, 5), keys(iterator)); + Assert.assertEquals(allowed + 1, executor.attempts); + Assert.assertTrue(first.closed); + Assert.assertTrue(second.closed); + Assert.assertTrue(third.closed); + } + } + } + + @Test + public void testSecurityFallbackStillPropagatesSourceFailure() { + DirectExecutorService executor = new DirectExecutorService(); + executor.allowedExecutions = 0; + TestIterator first = new TestIterator(1); + TestIterator second = new TestIterator(2); + first.failOnHasNextAfter(0); + OrderedKvIterator iterator = new OrderedKvIterator(Arrays.asList(first, second), 0L, executor); + Assert.assertThrows(IllegalStateException.class, iterator::hasNext); + Assert.assertTrue(first.closed); + Assert.assertTrue(second.closed); + } + + @Test + public void testSecurityFallbackDrainsAlreadyRunningSource() throws Exception { + AtomicInteger attempts = new AtomicInteger(); + CountDownLatch denied = new CountDownLatch(1); + ExecutorService initializer = new ThreadPoolExecutor( + 0, 2, 60L, TimeUnit.SECONDS, new SynchronousQueue<>(), task -> { + if (attempts.getAndIncrement() > 0) { + denied.countDown(); + throw new SecurityException("Worker creation denied"); + } + return new Thread(task); + }); + ExecutorService caller = Executors.newSingleThreadExecutor(); + CountDownLatch started = new CountDownLatch(1); + CountDownLatch release = new CountDownLatch(1); + TestIterator first = new TestIterator(1, 4); + TestIterator second = new TestIterator(2, 3); + first.blockFirstHasNext(started, release); + Future> result = caller.submit(() -> { + try (OrderedKvIterator iterator = new OrderedKvIterator( + Arrays.asList(first, second), 0L, initializer)) { + return keys(iterator); + } + }); + try { + Assert.assertTrue(started.await(3L, TimeUnit.SECONDS)); + Assert.assertTrue(denied.await(3L, TimeUnit.SECONDS)); + Assert.assertFalse(result.isDone()); + release.countDown(); + Assert.assertEquals(Arrays.asList(1, 2, 3, 4), result.get(3L, TimeUnit.SECONDS)); + Assert.assertEquals(2, attempts.get()); + Assert.assertTrue(first.closed); + Assert.assertTrue(second.closed); + } finally { + release.countDown(); + result.cancel(true); + caller.shutdownNow(); + initializer.shutdownNow(); + } + } + + @Test + public void testSourceSecurityExceptionIsNotSubmissionFallback() { + for (int allowed : new int[]{0, Integer.MAX_VALUE}) { + DirectExecutorService executor = new DirectExecutorService(); + executor.allowedExecutions = allowed; + TestIterator first = new TestIterator(1); + TestIterator second = new TestIterator(2); + SecurityException failure = new SecurityException("Source access denied"); + first.initializationFailure = failure; + OrderedKvIterator iterator = new OrderedKvIterator( + Arrays.asList(first, second), 0L, executor); + Assert.assertSame(failure, Assert.assertThrows(SecurityException.class, iterator::hasNext)); + Assert.assertTrue(first.closed); + Assert.assertTrue(second.closed); + } + } + + @Test + public void testSecurityFallbackPreservesCallerInterrupt() { + DirectExecutorService executor = new DirectExecutorService(); + executor.allowedExecutions = 0; + TestIterator first = new TestIterator(1); + TestIterator second = new TestIterator(2); + first.initializationFailure = new IllegalStateException(new InterruptedException()); + OrderedKvIterator iterator = new OrderedKvIterator(Arrays.asList(first, second), 0L, executor); + try { + Assert.assertThrows(IllegalStateException.class, iterator::hasNext); + Assert.assertTrue(Thread.currentThread().isInterrupted()); + Assert.assertTrue(first.closed); + Assert.assertTrue(second.closed); + } finally { + Thread.interrupted(); + } + } + + @Test + public void testSecurityFallbackClosesSourcesWhenCancellationIsDenied() { + DirectExecutorService executor = new DirectExecutorService(); + executor.allowedExecutions = 1; + executor.deferExecution = true; + executor.cancelFailure = new SecurityException("Worker interruption denied"); + TestIterator first = new TestIterator(1); + TestIterator second = new TestIterator(2); + RuntimeException failure = new IllegalStateException("Source failed"); + second.initializationFailure = failure; + OrderedKvIterator iterator = new OrderedKvIterator(Arrays.asList(first, second), 0L, executor); + Assert.assertSame(failure, Assert.assertThrows(IllegalStateException.class, iterator::hasNext)); + Assert.assertArrayEquals(new Throwable[]{executor.cancelFailure}, failure.getSuppressed()); + Assert.assertTrue(first.closed); + Assert.assertTrue(second.closed); + } + @Test public void testMergeInterleavedSourcesByUnsignedKey() { TestIterator first = new TestIterator(1, 4); @@ -453,6 +582,23 @@ private static final class DirectExecutorService private boolean shutdown; private int executions; + private int attempts; + private int allowedExecutions = Integer.MAX_VALUE; + private boolean deferExecution; + private RuntimeException cancelFailure; + + @Override + protected RunnableFuture newTaskFor(Callable task) { + return new FutureTask(task) { + @Override + public boolean cancel(boolean mayInterruptIfRunning) { + if (DirectExecutorService.this.cancelFailure != null) { + throw DirectExecutorService.this.cancelFailure; + } + return super.cancel(mayInterruptIfRunning); + } + }; + } private int executions() { return this.executions; @@ -489,8 +635,13 @@ public void execute(Runnable command) { if (this.shutdown) { throw new RejectedExecutionException(); } + if (this.attempts++ >= this.allowedExecutions) { + throw new SecurityException("Worker creation denied"); + } this.executions++; - command.run(); + if (!this.deferExecution) { + command.run(); + } } } @@ -520,6 +671,7 @@ private static final class TestIterator implements HgKvIterator { private boolean firstHasNextBlocked; private boolean failAfterFirstHasNextRelease; private boolean restoreInterrupt; + private RuntimeException initializationFailure; private TestIterator(Integer... keys) { this.entries = new ArrayList<>(keys.length); @@ -567,6 +719,9 @@ private void blockFirstHasNextWithoutRestoringInterrupt( @Override public boolean hasNext() { + if (this.initializationFailure != null) { + throw this.initializationFailure; + } if (this.nextCalls == this.failOnHasNextAfter) { throw new IllegalStateException("injected failure"); } From cd5ccfcc738f85d1198af12f9eeb4770df142a35 Mon Sep 17 00:00:00 2001 From: imbajin Date: Thu, 17 Sep 2026 22:47:24 +0800 Subject: [PATCH 2/2] fix(hstore): drain completed scans before fallback - propagate completed worker failures before inline scans - keep draining nonblocking while other workers are active - cover original failure preservation and source cleanup --- .../hugegraph/store/client/OrderedKvIterator.java | 6 ++++++ .../store/client/OrderedKvIteratorTest.java | 15 +++++++++++++++ 2 files changed, 21 insertions(+) diff --git a/hugegraph-store/hg-store-client/src/main/java/org/apache/hugegraph/store/client/OrderedKvIterator.java b/hugegraph-store/hg-store-client/src/main/java/org/apache/hugegraph/store/client/OrderedKvIterator.java index ed65706300..f395431a06 100644 --- a/hugegraph-store/hg-store-client/src/main/java/org/apache/hugegraph/store/client/OrderedKvIterator.java +++ b/hugegraph-store/hg-store-client/src/main/java/org/apache/hugegraph/store/client/OrderedKvIterator.java @@ -197,6 +197,12 @@ private void initialize() { while (nextSource < this.iterators.size() && inFlight < INITIALIZE_THREADS) { if (!useExecutor) { + Future completed = completions.poll(); + while (completed != null) { + this.addFirst(completed.get()); + inFlight--; + completed = completions.poll(); + } this.addFirst(this.firstEntry(nextSource++)); continue; } diff --git a/hugegraph-store/hg-store-test/src/main/java/org/apache/hugegraph/store/client/OrderedKvIteratorTest.java b/hugegraph-store/hg-store-test/src/main/java/org/apache/hugegraph/store/client/OrderedKvIteratorTest.java index cc4648bd7f..3073c1bfec 100644 --- a/hugegraph-store/hg-store-test/src/main/java/org/apache/hugegraph/store/client/OrderedKvIteratorTest.java +++ b/hugegraph-store/hg-store-test/src/main/java/org/apache/hugegraph/store/client/OrderedKvIteratorTest.java @@ -66,6 +66,21 @@ public void testSecurityDeniedSubmissionFallsBackWithoutRetrying() { } } + @Test + public void testSecurityFallbackDrainsSubmittedFailureBeforeInlineSource() { + DirectExecutorService executor = new DirectExecutorService(); + executor.allowedExecutions = 1; + TestIterator first = new TestIterator(1); + TestIterator second = new TestIterator(2); + RuntimeException failure = new IllegalStateException("Submitted source failed"); + first.initializationFailure = failure; + second.initializationFailure = new IllegalStateException("Inline source should not start"); + OrderedKvIterator iterator = new OrderedKvIterator(Arrays.asList(first, second), 0L, executor); + Assert.assertSame(failure, Assert.assertThrows(IllegalStateException.class, iterator::hasNext)); + Assert.assertTrue(first.closed); + Assert.assertTrue(second.closed); + } + @Test public void testSecurityFallbackStillPropagatesSourceFailure() { DirectExecutorService executor = new DirectExecutorService();