diff --git a/plugin-transform-json/src/main/java/io/kestra/plugin/transform/jsonata/EvaluationBounds.java b/plugin-transform-json/src/main/java/io/kestra/plugin/transform/jsonata/EvaluationBounds.java new file mode 100644 index 0000000..f7a9edf --- /dev/null +++ b/plugin-transform-json/src/main/java/io/kestra/plugin/transform/jsonata/EvaluationBounds.java @@ -0,0 +1,71 @@ +package io.kestra.plugin.transform.jsonata; + +import com.dashjoin.jsonata.JException; +import com.dashjoin.jsonata.Jsonata; + +/** + * Depth and timeout guard for a single JSONata evaluation. + * + *

Replaces {@code Jsonata.Frame#setRuntimeBounds(long, int)}, whose {@code Timebox} skips the + * increment and the decrement whenever the frame carries {@code isParallelCall}. Entry and exit do + * not observe the same flag for a given {@code evaluate()} call, so the counter gains one per input + * item and never unwinds — making {@code maxDepth} bound input size instead of recursion depth + * (kestra-io/plugin-transform#102). Counting unconditionally keeps the two sides symmetric, which is + * safe because the Java port evaluates path steps sequentially. + */ +final class EvaluationBounds { + + private final int maxDepth; + private final long timeoutMillis; + private final long startedAt; + + private int depth; + + static void register(Jsonata.Frame frame, int maxDepth, long timeoutMillis) { + var bounds = new EvaluationBounds(maxDepth, timeoutMillis); + frame.setEvaluateEntryCallback((expression, input, environment) -> bounds.enter()); + frame.setEvaluateExitCallback((expression, input, environment, result) -> bounds.exit()); + } + + private EvaluationBounds(int maxDepth, long timeoutMillis) { + this.maxDepth = maxDepth; + this.timeoutMillis = timeoutMillis; + this.startedAt = System.currentTimeMillis(); + } + + private void enter() { + checkDepth(); + checkTimeout(); + } + + private void checkDepth() { + if (++depth > maxDepth) { + throw new JException( + "JSONata expression exceeded maxDepth=" + maxDepth + " nested evaluation levels. " + + "Raise maxDepth if the expression legitimately recurses that deep, otherwise check " + + "for a recursive function that never reaches its terminating case.", + -1 + ); + } + } + + /** + * Checked on entry only, unlike {@code Timebox.checkRunnaway()} which also checked on exit. Every + * node that can extend the evaluation enters before it runs, so the entry check bounds the overrun + * to a single node; neither side can interrupt an in-flight call anyway. + */ + private void checkTimeout() { + if (System.currentTimeMillis() - startedAt > timeoutMillis) { + throw new JException( + "JSONata evaluation exceeded timeout=" + timeoutMillis + "ms. " + + "Raise timeout for genuinely long transformations, otherwise check for a " + + "non-terminating expression.", + -1 + ); + } + } + + private void exit() { + depth--; + } +} diff --git a/plugin-transform-json/src/main/java/io/kestra/plugin/transform/jsonata/JSONataInterface.java b/plugin-transform-json/src/main/java/io/kestra/plugin/transform/jsonata/JSONataInterface.java index dd1fba0..00dff34 100644 --- a/plugin-transform-json/src/main/java/io/kestra/plugin/transform/jsonata/JSONataInterface.java +++ b/plugin-transform-json/src/main/java/io/kestra/plugin/transform/jsonata/JSONataInterface.java @@ -13,8 +13,14 @@ public interface JSONataInterface { Property getExpression(); @Schema( - title = "The maximum number of recursive calls allowed for the JSONata transformation", - description = "Limits recursive JSONata function call depth. Each recursive call adds a frame to the chain traversed by variable lookup, so high values can cause a JVM StackOverflowError on platforms with small default thread stacks (e.g. Windows ~256 KB vs Linux ~512 KB). Raise only for expressions with proven deep recursion needs." + title = "The maximum number of nested evaluation levels allowed for the JSONata expression", + description = """ + Bounds how deeply the expression may nest while it is evaluated, which is what caps runaway \ + recursive functions. The limit depends only on the expression, never on how many records are \ + processed: the default of 1000 is far above what ordinary expressions reach, and a batch of \ + 1,000,000 records needs no higher a value than a batch of 10 does. Raise it only for \ + expressions with proven deep recursion needs — each level adds frames to the chain traversed \ + by variable lookup, so very high values trade the clean error for a JVM StackOverflowError.""" ) @NotNull @PluginProperty(group = "main") diff --git a/plugin-transform-json/src/main/java/io/kestra/plugin/transform/jsonata/Transform.java b/plugin-transform-json/src/main/java/io/kestra/plugin/transform/jsonata/Transform.java index d117376..39b0504 100644 --- a/plugin-transform-json/src/main/java/io/kestra/plugin/transform/jsonata/Transform.java +++ b/plugin-transform-json/src/main/java/io/kestra/plugin/transform/jsonata/Transform.java @@ -108,7 +108,7 @@ protected JsonNode evaluateExpression(RunContext runContext, JsonNode jsonNode) var data = MAPPER.convertValue(jsonNode, Object.class); var frame = this.parsedExpression.createFrame(); - frame.setRuntimeBounds(timeoutInMilli, rMaxDepth); + EvaluationBounds.register(frame, rMaxDepth, timeoutInMilli); var resultRef = new AtomicReference(); var errorRef = new AtomicReference(); diff --git a/plugin-transform-json/src/main/resources/doc/io.kestra.plugin.transform.jsonata.md b/plugin-transform-json/src/main/resources/doc/io.kestra.plugin.transform.jsonata.md index f7c362c..584c014 100644 --- a/plugin-transform-json/src/main/resources/doc/io.kestra.plugin.transform.jsonata.md +++ b/plugin-transform-json/src/main/resources/doc/io.kestra.plugin.transform.jsonata.md @@ -13,3 +13,19 @@ Query and transform JSON data using [JSONata](https://jsonata.org/) expressions: Expressions follow the standard JSONata syntax. See the [JSONata documentation](https://docs.jsonata.org/) for operators, built-in functions, and examples. `TransformValue` works on a value already in the flow, while `TransformItems` streams over a file in internal storage record by record. + +`maxDepth` bounds how deeply an expression nests during evaluation, which is what stops a runaway recursive function. It does not depend on how much data you process, so the default of 1000 needs no adjustment as batches grow. + +### Prefer path projection over `$map` with a lambda + +Building an object around a batch mapped through a user-defined function is quadratic in the underlying engine, so it degrades sharply with batch size — around 18 s for 1000 items where the equivalent path projection takes 18 ms. Both forms produce the same output, so prefer the projection: + +``` +# slow — avoid +{ "items": $map($, function($r) { { "id": $r.eventId, "amount": $r.value } }) } + +# fast — same result +{ "items": [ $.{ "id": eventId, "amount": value } ] } +``` + +It takes both ingredients to trigger: the same object wrapper around a builtin function (`$map($, $string)`) stays fast, and so does the lambda on its own without the wrapper. The cost is in [`dashjoin/jsonata-java`](https://github.com/dashjoin/jsonata-java), not in this plugin. Reach for `$map` with a lambda only when the transformation genuinely needs a function value, such as passing it to `$reduce` or `$sort`. diff --git a/plugin-transform-json/src/test/java/io/kestra/plugin/transform/jsonata/TransformItemsTest.java b/plugin-transform-json/src/test/java/io/kestra/plugin/transform/jsonata/TransformItemsTest.java index f36cf47..18505e9 100644 --- a/plugin-transform-json/src/test/java/io/kestra/plugin/transform/jsonata/TransformItemsTest.java +++ b/plugin-transform-json/src/test/java/io/kestra/plugin/transform/jsonata/TransformItemsTest.java @@ -10,6 +10,8 @@ import jakarta.inject.Inject; import org.junit.jupiter.api.Assertions; import org.junit.jupiter.api.Test; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.ValueSource; import reactor.core.publisher.Flux; import java.io.InputStream; @@ -223,6 +225,52 @@ void shouldHandleLargeDatasetWithFlatFieldLookupOnConstrainedStack() throws Exce Assertions.assertEquals(recordCount, output.getProcessedItemsTotal()); } + @ParameterizedTest + @ValueSource(ints = {100, 5_000}) + void shouldCollapseBatchIntoOneRecordOnDefaultMaxDepth(int itemCount) throws Exception { + // Regression: kestra-io/plugin-transform#102. One record holding a batch of N items, folded into a + // single object — the reported shape. Frame.setRuntimeBounds leaked depth per item, so the default + // maxDepth=1000 failed above roughly 300 items here with "Depth=1001 max=1000" while passing at 100. + // The nested object constructor is what leaked; the projection form keeps the test linear in time. + RunContext runContext = runContextFactory.of(); + final Path outputFilePath = runContext.workingDir().createTempFile(".ion"); + + List> batch = new ArrayList<>(itemCount); + for (int i = 0; i < itemCount; i++) { + batch.add(Map.of("eventId", "e" + i, "value", i, "currency", "USD")); + } + + try (Writer writer = new OutputStreamWriter(Files.newOutputStream(outputFilePath))) { + FileSerde.writeAll(writer, Flux.just(batch)).block(); + writer.flush(); + } + URI uri = runContext.storage().putFile(outputFilePath.toFile()); + + TransformItems task = TransformItems.builder() + .from(Property.ofValue(uri.toString())) + .explodeArray(Property.ofValue(false)) + .expression(Property.ofValue(""" + { + "items": [ $.{ + "eventId": eventId, + "deposit": [ { "value": value, "currency": currency } ] + } ] + } + """)) + .build(); + + TransformItems.Output output = task.run(runContext); + + Assertions.assertEquals(1, output.getProcessedItemsTotal()); + + InputStream is = runContext.storage().getFile(output.getUri()); + List transformationResult = FileSerde.readAll(new InputStreamReader(is), new TypeReference() { + }).collectList().block(); + + Assertions.assertEquals(1, transformationResult.size()); + Assertions.assertEquals(itemCount, ((List) transformationResult.getFirst().get("items")).size()); + } + @Test void shouldTransformJsonInputWithDefaultIonMapper() throws Exception { // Given diff --git a/plugin-transform-json/src/test/java/io/kestra/plugin/transform/jsonata/TransformValueTest.java b/plugin-transform-json/src/test/java/io/kestra/plugin/transform/jsonata/TransformValueTest.java index 4bc0b62..5b14531 100644 --- a/plugin-transform-json/src/test/java/io/kestra/plugin/transform/jsonata/TransformValueTest.java +++ b/plugin-transform-json/src/test/java/io/kestra/plugin/transform/jsonata/TransformValueTest.java @@ -13,6 +13,10 @@ import org.junit.jupiter.params.ParameterizedTest; import org.junit.jupiter.params.provider.ValueSource; +import java.time.Duration; +import java.util.stream.Collectors; +import java.util.stream.IntStream; + import static org.assertj.core.api.Assertions.assertThat; import static org.assertj.core.api.Assertions.assertThatThrownBy; @@ -206,6 +210,87 @@ void shouldContinueWorkingAfterStackOverflowError() throws Exception { assertThat(output.getValue().toString()).isEqualTo("42"); } + // Regression: kestra-io/plugin-transform#102. + // Frame.setRuntimeBounds counted one un-unwound level per input item, so maxDepth silently bounded + // input size — a 1000-item batch failed on the default maxDepth=1000 with a message blaming + // recursion the expression never contained. EvaluationBounds counts entry/exit symmetrically, so the + // budget a given expression needs is a property of the expression alone. + + @ParameterizedTest + @ValueSource(ints = {10, 400}) + void shouldRequireTheSameMaxDepthRegardlessOfInputSize(int itemCount) throws Exception { + // maxDepth=12 leaves ample headroom over the 4 levels this expression actually nests, while staying + // far below the itemCount + 3 the leaking counter reached — so it fails pre-fix at both sizes and + // passes post-fix at both. 400 is the upper size because $map with a lambda is still quadratic + // upstream (defect 2 of #102); the size-independence claim is carried to 5000 items by + // TransformItemsTest#shouldCollapseBatchIntoOneRecordOnDefaultMaxDepth, which uses the linear form. + RunContext runContext = runContextFactory.of(); + TransformValue task = TransformValue.builder() + .from(Property.ofValue(jsonArrayOfItems(itemCount))) + .expression(Property.ofValue(""" + { + "items": $append([], $map($, function($r) { + { + "eventId": $r.eventId, + "deposit": [ { "value": $r.value, "currency": $r.currency } ] + } + })) + } + """)) + .maxDepth(Property.ofValue(12)) + .build(); + + TransformValue.Output output = task.run(runContext); + + assertThat(output.getValue()).isNotNull(); + JsonNode result = new ObjectMapper().valueToTree(output.getValue()); + assertThat(result.get("items").size()).isEqualTo(itemCount); + } + + @Test + void shouldReportMaxDepthBreachWithActionableMessage() throws Exception { + RunContext runContext = runContextFactory.of(); + TransformValue task = TransformValue.builder() + .from(Property.ofValue("{}")) + .expression(Property.ofValue("($f := function($n) { $n > 0 ? $f($n - 1) + 0 : 0 }; $f(500))")) + .maxDepth(Property.ofValue(100)) + .build(); + + assertThatThrownBy(() -> task.run(runContext)) + .isInstanceOf(RuntimeException.class) + .hasCauseInstanceOf(JException.class) + .cause() + .hasMessageContaining("exceeded maxDepth=100"); + } + + @Test + void shouldStillEnforceTimeout() throws Exception { + // Guards that replacing setRuntimeBounds did not drop timeout enforcement along with the depth + // counter. The expression is tail-recursive, so TCO holds it at ~5 nested levels: it runs long + // purely through iteration count and cannot be stopped by the depth bound instead. + // Wall-clock assertion by nature: 5M interpreted iterations against a 200ms budget leaves a wide + // margin. Widen it further (more iterations, lower timeout) if CI contention ever makes it flake — + // this is the only guard that timeout enforcement survived the move off Timebox. + RunContext runContext = runContextFactory.of(); + TransformValue task = TransformValue.builder() + .from(Property.ofValue("{}")) + .expression(Property.ofValue("($f := function($n) { $n > 0 ? $f($n - 1) : 0 }; $f(5000000))")) + .timeout(Property.ofValue(Duration.ofMillis(200))) + .build(); + + assertThatThrownBy(() -> task.run(runContext)) + .isInstanceOf(RuntimeException.class) + .hasCauseInstanceOf(JException.class) + .cause() + .hasMessageContaining("exceeded timeout=200ms"); + } + + private static String jsonArrayOfItems(int itemCount) { + return IntStream.range(0, itemCount) + .mapToObj(i -> "{\"eventId\":\"e%d\",\"value\":%d,\"currency\":\"USD\"}".formatted(i, i)) + .collect(Collectors.joining(",", "[", "]")); + } + @Test void shouldIsolateStackOverflowInEvalThreadWhenMaxDepthExceedsStackCapacity() throws Exception { // User sets maxDepth high enough that bounds check never fires before stack exhaustion.