fix(jsonata): bound maxDepth by expression nesting, not input size

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.
+ *
+ * <p>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--;
+    }
+}

plugin-transform-json/src/main/java/io/kestra/plugin/transform/jsonata/JSONataInterface.java

@@ -13,8 +13,14 @@ public interface JSONataInterface {
     Property<String> 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")

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<JsonNode>();
             var errorRef = new AtomicReference<Throwable>();

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`.

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<Map<String, Object>> 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<Map> transformationResult = FileSerde.readAll(new InputStreamReader(is), new TypeReference<Map>() {
+        }).collectList().block();
+
+        Assertions.assertEquals(1, transformationResult.size());
+        Assertions.assertEquals(itemCount, ((List<?>) transformationResult.getFirst().get("items")).size());
+    }
+
     @Test
     void shouldTransformJsonInputWithDefaultIonMapper() throws Exception {
         // Given

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.