This is an automated email from the ASF dual-hosted git repository. rvesse pushed a commit to branch main in repository https://gitbox.apache.org/repos/asf/jena.git
commit e035938081d6fffecc0f37b59c11e734feae8ea6 Author: Rob Vesse <[email protected]> AuthorDate: Mon Sep 28 14:20:50 2026 +0100 Cap afn:wait() function wait The afn:wait() function is intended only for testing but can be used to request excessive waits that block query processing threads indefinitely. This commit improves the implementation as follows: - Refuses to wait longer than the query timeouts (if any) - Refuses to wait if elapsed time plus wait time would exceed overall query timeout - Refuses to wait longer than a static controlled maximum - Refuses to wait if query excecution has been cancelled --- .../java/org/apache/jena/sparql/ARQConstants.java | 2 + .../apache/jena/sparql/function/library/wait.java | 84 +++++++++++- .../java/org/apache/jena/sparql/util/Context.java | 1 + .../org/apache/jena/sparql/expr/TestFunctions.java | 144 ++++++++++++++++++++- 4 files changed, 219 insertions(+), 12 deletions(-) diff --git a/jena-arq/src/main/java/org/apache/jena/sparql/ARQConstants.java b/jena-arq/src/main/java/org/apache/jena/sparql/ARQConstants.java index dc6adb6844..5071f627d2 100644 --- a/jena-arq/src/main/java/org/apache/jena/sparql/ARQConstants.java +++ b/jena-arq/src/main/java/org/apache/jena/sparql/ARQConstants.java @@ -266,6 +266,8 @@ public class ARQConstants /** Context key for the current time of query execution */ public static final Symbol sysCurrentTime = Symbol.create(systemVarNS+"now") ; + /** Context key for the current time of query execution expressed as epoch milliseconds */ + public static final Symbol sysCurrentTimeEpoch = Symbol.create(systemVarNS+"nowEpoch"); /** Context key for ARQ version */ public static final Symbol sysVersionARQ = Symbol.create(systemVarNS+"version/ARQ") ; diff --git a/jena-arq/src/main/java/org/apache/jena/sparql/function/library/wait.java b/jena-arq/src/main/java/org/apache/jena/sparql/function/library/wait.java index d9988479e1..0d59241642 100644 --- a/jena-arq/src/main/java/org/apache/jena/sparql/function/library/wait.java +++ b/jena-arq/src/main/java/org/apache/jena/sparql/function/library/wait.java @@ -22,19 +22,91 @@ package org.apache.jena.sparql.function.library; import org.apache.jena.atlas.lib.Lib; +import org.apache.jena.graph.Node; +import org.apache.jena.sparql.ARQConstants; +import org.apache.jena.sparql.engine.Timeouts; import org.apache.jena.sparql.expr.ExprEvalException; import org.apache.jena.sparql.expr.NodeValue; import org.apache.jena.sparql.function.FunctionBase1; +import org.apache.jena.sparql.function.FunctionEnv; +import org.apache.jena.sparql.util.Context; -/** Function that pauses for N milliseconds whenever it is called (for testing) */ +import java.time.Instant; +import java.util.List; +import java.util.concurrent.atomic.AtomicBoolean; + +/** + * Function that pauses for N milliseconds whenever it is called (for testing) + */ public class wait extends FunctionBase1 { + /** + * The default maximum wait time (5 seconds) used for {@link #MAX_WAIT} + */ + public static final int DEFAULT_MAX_WAIT = 5_000; + /** + * Maximum permitted wait time in milliseconds, if a longer wait is requested then no wait happens + */ + public static int MAX_WAIT = DEFAULT_MAX_WAIT; + @Override - public NodeValue exec(NodeValue nv) { - if ( !nv.isInteger() ) - throw new ExprEvalException("Not an integer"); - int x = nv.getInteger().intValue(); - Lib.sleep(x); + protected NodeValue exec(List<NodeValue> args, FunctionEnv env) { + Timeouts.Timeout timeout = Timeouts.extractQueryTimeout(env.getContext()); + + int waitMillis = getWaitMillis(args, timeout); + if (waitMillis > MAX_WAIT) { + throw new ExprEvalException("Requested wait time too large"); + } + AtomicBoolean cancelSignal = Context.getCancelSignal(env.getContext()); + if (cancelSignal != null && cancelSignal.get()) { + throw new ExprEvalException("Cannot wait as query is cancelled"); + } + + Long startTime = getStartTimeEpoch(env.getContext()); + if (startTime != null && timeout.hasOverallTimeout()) { + long elapsed = System.currentTimeMillis() - startTime; + if (elapsed + waitMillis > timeout.overallTimeoutMillis()) { + throw new ExprEvalException("Cannot wait as requested wait would exceed query timeout"); + } + } + Lib.sleep(waitMillis); return NodeValue.TRUE; } + + private static Long getStartTimeEpoch(Context context) { + Long startTime = context.get(ARQConstants.sysCurrentTimeEpoch); + if (startTime != null) { + return startTime; + } + Node executionTime = context.get(ARQConstants.sysCurrentTime); + if (executionTime == null) { + return null; + } + startTime = Instant.parse(executionTime.getLiteralLexicalForm()).toEpochMilli(); + context.set(ARQConstants.sysCurrentTimeEpoch, startTime); + return startTime; + } + + private static int getWaitMillis(List<NodeValue> args, Timeouts.Timeout timeout) { + if (args.size() != 1) { + throw new ExprEvalException("afn:wait() takes a single argument but got " + args.size()); + } + NodeValue nv = args.getFirst(); + if (!nv.isInteger()) { + throw new ExprEvalException("Not an integer"); + } + int waitMillis = nv.getInteger().intValue(); + if (waitMillis <= 0) { + throw new ExprEvalException("Cannot wait for zero/negative milliseconds"); + } + if ((timeout.hasInitialTimeout() && waitMillis > timeout.initialTimeoutMillis()) || (timeout.hasOverallTimeout() && waitMillis > timeout.overallTimeoutMillis())) { + throw new ExprEvalException("Cannot wait longer than the query timeout"); + } + return waitMillis; + } + + @Override + public NodeValue exec(NodeValue v) { + throw new ExprEvalException("Must supply a FunctionEnv in order to enforce query timeout"); + } } diff --git a/jena-arq/src/main/java/org/apache/jena/sparql/util/Context.java b/jena-arq/src/main/java/org/apache/jena/sparql/util/Context.java index 7e141822fb..bc8bdce2f8 100644 --- a/jena-arq/src/main/java/org/apache/jena/sparql/util/Context.java +++ b/jena-arq/src/main/java/org/apache/jena/sparql/util/Context.java @@ -449,6 +449,7 @@ public class Context { public static void setCurrentDateTime(Context context) { //context.set(ARQConstants.sysCurrentTime, NodeFactoryExtra.nowUTC()); context.set(ARQConstants.sysCurrentTime, NodeFactoryExtra.nowAsDateTime()); + context.set(ARQConstants.sysCurrentTimeEpoch, System.currentTimeMillis()); } /** Set the current time in the context if there is no setting */ diff --git a/jena-arq/src/test/java/org/apache/jena/sparql/expr/TestFunctions.java b/jena-arq/src/test/java/org/apache/jena/sparql/expr/TestFunctions.java index eb47b514b8..9fe5e99838 100644 --- a/jena-arq/src/test/java/org/apache/jena/sparql/expr/TestFunctions.java +++ b/jena-arq/src/test/java/org/apache/jena/sparql/expr/TestFunctions.java @@ -21,20 +21,39 @@ package org.apache.jena.sparql.expr; -import static org.junit.jupiter.api.Assertions.assertEquals; -import static org.junit.jupiter.api.Assertions.assertTrue; -import static org.junit.jupiter.api.Assertions.fail; - import java.text.ParseException; import java.text.SimpleDateFormat; +import java.time.Duration; +import java.time.ZonedDateTime; +import java.time.temporal.ChronoUnit; +import java.time.temporal.TemporalUnit; +import java.util.Calendar; import java.util.Date; import java.util.TimeZone; +import java.util.concurrent.TimeUnit; import java.util.function.Predicate; +import java.util.stream.Stream; +import org.apache.commons.lang3.Strings; +import org.apache.jena.atlas.lib.DateTimeUtils; +import org.apache.jena.datatypes.xsd.XSDDatatype; +import org.apache.jena.graph.NodeFactory; +import org.apache.jena.sparql.engine.Timeouts; +import org.apache.jena.sparql.function.FunctionEnv; +import org.apache.jena.sparql.function.FunctionEnvBase; +import org.apache.jena.sparql.util.Context; +import org.apache.jena.sparql.util.NodeFactoryExtra; +import org.apache.jena.vocabulary.XSD; +import org.awaitility.Awaitility; import org.junit.jupiter.api.Test; import org.apache.jena.sparql.ARQConstants; import org.apache.jena.sparql.util.ExprUtils; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.Arguments; +import org.junit.jupiter.params.provider.MethodSource; + +import static org.junit.jupiter.api.Assertions.*; public class TestFunctions { @@ -129,6 +148,116 @@ public class TestFunctions @Test public void localDateTime_3() { test("afn:nowtz() = NOW()", NodeValue.TRUE); } + @Test + public void wait_1() { + test("afn:wait(100)", NodeValue.TRUE); + } + + @Test + public void givenQueryTimeout_whenWaitExceedsInitialTimeout_thenEvalException() { + // Given + Context context = new Context(); + Context.setCurrentDateTimeIfUndef(context); + Timeouts.setQueryTimeout(context, new Timeouts.Timeout(Timeouts.DurationWithUnit.of(100, TimeUnit.MILLISECONDS), Timeouts.DurationWithUnit.UNSET)); + FunctionEnv env = new FunctionEnvBase(context); + + // When + Expr expr = ExprUtils.parse("afn:wait(1000)"); + + // Then + ExprEvalException e = assertThrows(ExprEvalException.class, () -> expr.eval(null, env)); + assertTrue(Strings.CI.contains(e.getMessage(), "longer than the query timeout")); + } + + @Test + public void givenQueryTimeout_whenWaitExceedsOverallTimeout_thenEvalException() { + // Given + Context context = new Context(); + Context.setCurrentDateTimeIfUndef(context); + Timeouts.setQueryTimeout(context, new Timeouts.Timeout(Timeouts.DurationWithUnit.UNSET, Timeouts.DurationWithUnit.of(100, TimeUnit.MILLISECONDS))); + FunctionEnv env = new FunctionEnvBase(context); + + // When + Expr expr = ExprUtils.parse("afn:wait(1000)"); + + // Then + ExprEvalException e = assertThrows(ExprEvalException.class, () -> expr.eval(null, env)); + assertTrue(Strings.CI.contains(e.getMessage(), "longer than the query timeout")); + } + + @Test + public void givenQueryTimeout_whenWaitPlusElapsedExceedsOverallTimeout_thenEvalException() { + // Given + Context context = new Context(); + context.set(ARQConstants.sysCurrentTime, NodeFactory.createLiteralDT(DateTimeUtils.asXSDDateTimeString( + ZonedDateTime.now().minus(4500, ChronoUnit.MILLIS)), XSDDatatype.XSDdateTime)); + Timeouts.setQueryTimeout(context, new Timeouts.Timeout(Timeouts.DurationWithUnit.UNSET, Timeouts.DurationWithUnit.of(5, TimeUnit.SECONDS))); + FunctionEnv env = new FunctionEnvBase(context); + + // When + Expr expr = ExprUtils.parse("afn:wait(1000)"); + + // Then + ExprEvalException e = assertThrows(ExprEvalException.class, () -> expr.eval(null, env)); + assertTrue(Strings.CI.contains(e.getMessage(), "wait would exceed query timeout")); + } + + @Test + public void givenQueryTimeout_whenWaitsEventuallyExceedRemainingTimeout_thenEvalException() { + // Given + Context context = new Context(); + Context.setCurrentDateTimeIfUndef(context); + Timeouts.setQueryTimeout(context, new Timeouts.Timeout(Timeouts.DurationWithUnit.UNSET, Timeouts.DurationWithUnit.of(5, TimeUnit.SECONDS))); + FunctionEnv env = new FunctionEnvBase(context); + + // When + Expr expr = ExprUtils.parse("afn:wait(1000)"); + + // Then + Awaitility.await("Repeatedly evaluating afn:wait should eventually throw ExprEvalException") + .atMost(Duration.ofMillis(5500)) + .pollInterval(Duration.ZERO) + .untilAsserted(() -> { + ExprEvalException e = assertThrows(ExprEvalException.class, () -> expr.eval(null, env)); + assertTrue(Strings.CI.contains(e.getMessage(), "wait would exceed query timeout")); + }); + } + + @Test + public void givenQueryCancelled_whenWait_thenEvalException() { + // Given + Context context = new Context(); + Context.setCurrentDateTimeIfUndef(context); + Context.getOrSetCancelSignal(context).set(true); + FunctionEnv env = new FunctionEnvBase(context); + + // When + Expr expr = ExprUtils.parse("afn:wait(10)"); + + // Then + ExprEvalException e = assertThrows(ExprEvalException.class, () -> expr.eval(null, env)); + assertTrue(Strings.CI.contains(e.getMessage(), "query is cancelled")); + } + + public static Stream<Arguments> badWaits() { + return Stream.of(Arguments.of("2.0e16", "not an integer"), + Arguments.of("false", "not an integer"), + Arguments.of("60000", "too large"), + Arguments.of("-1", "zero/negative"), + Arguments.of("?x", "not bound") + ); + } + + @ParameterizedTest + @MethodSource("badWaits") + public void wait_bad(String waitArg, String expectedMessage) { + // Given and When + ExprEvalException e = testEvalException("afn:wait(" + waitArg +")"); + + // Then + assertTrue(Strings.CI.contains(e.getMessage(), expectedMessage)); + } + private void test(String exprStr, NodeValue result) { Expr expr = ExprUtils.parse(exprStr); NodeValue r = expr.eval(null, LibTestExpr.createTest()); @@ -147,12 +276,15 @@ public class TestFunctions test(exprStr, rExpected); } - private void testEvalException(String exprStr) { + private ExprEvalException testEvalException(String exprStr) { Expr expr = ExprUtils.parse(exprStr); try { NodeValue r = expr.eval(null, LibTestExpr.createTest()); fail("No exception raised"); + return null; + } + catch (ExprEvalException ex) { + return ex; } - catch (ExprEvalException ex) {} } }
