This is an automated email from the ASF dual-hosted git repository.
davsclaus pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/camel.git
The following commit(s) were added to refs/heads/main by this push:
new a08483cc21a6 CAMEL-25456: camel-wasm - discard the Wasm instance after
a failed call (#27593)
a08483cc21a6 is described below
commit a08483cc21a67ea063b2102489427c59ad73de6d
Author: allthingssecurity <[email protected]>
AuthorDate: Sat Oct 10 01:20:41 2026 +0530
CAMEL-25456: camel-wasm - discard the Wasm instance after a failed call
(#27593)
When the guest function trapped or the thread was interrupted, WasmFunction
kept the instance, so guest allocations leaked on every failed message and
the
cleanup could replace the original exception. A call that does not complete
now
makes no further call into the instance, closes and discards it and rethrows
the original exception; the next call creates a new instance. The docs now
say
the module is instantiated per producer (or per wasm expression).
Co-Authored-By: Claude Opus 5.5 <[email protected]>
---
.../apache/camel/catalog/docs/wasm-component.adoc | 15 ++
.../apache/camel/catalog/docs/wasm-language.adoc | 14 ++
components/camel-wasm/pom.xml | 6 +
.../camel-wasm/src/main/docs/wasm-component.adoc | 15 ++
.../camel-wasm/src/main/docs/wasm-language.adoc | 14 ++
.../apache/camel/component/wasm/WasmProducer.java | 3 +
.../java/org/apache/camel/wasm/WasmFunction.java | 116 ++++++++++----
.../org/apache/camel/wasm/WasmFunctionTest.java | 177 +++++++++++++++++++++
.../ROOT/pages/camel-4x-upgrade-guide-4_23.adoc | 6 +
9 files changed, 338 insertions(+), 28 deletions(-)
diff --git
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/wasm-component.adoc
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/wasm-component.adoc
index 5b09ac8f5c33..ad9aa3daf956 100644
---
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/wasm-component.adoc
+++
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/wasm-component.adoc
@@ -152,6 +152,21 @@ pub extern fn process(ptr: u32, len: u32) -> u64 {
}
----
+==== Errors and module state
+
+The module is instantiated once per producer (each `to` of a `wasm` endpoint
in a route has its own instance) and that
+instance handles every message, one at a time, so data the module keeps in its
memory or globals carries over from one
+message to the next.
+
+* When the function returns with the error bit set, the error data is released
with `dealloc` using its size without
+the error bit, the error data becomes the message of the exception, and the
instance is kept.
+* When the call fails without returning, for example on a Rust `panic` or any
other Wasm trap, the instance is
+discarded and a new one is created from the module for the next message.
Whatever the failed call allocated can no
+longer be released, so keeping the instance would grow its memory with every
failure. Any state the module kept in
+the discarded instance is lost.
+On a route where every call fails this way, the module is instantiated again
for every message (its data segments
+are copied and any start function runs), which costs more than a call on a
kept instance.
+
== Examples
Supposing we have compiled a Wasm module containing the function above, then
it can be called in a Camel Route by its name and module resource location:
diff --git
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/wasm-language.adoc
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/wasm-language.adoc
index 72208ba9c90e..340d82b8157b 100644
---
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/wasm-language.adoc
+++
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/wasm-language.adoc
@@ -126,6 +126,20 @@ pub extern fn transform(ptr: u32, len: u32) -> u64 {
}
----
+=== Errors and module state
+
+The module is instantiated once per `wasm` expression and that instance
handles every message, one at a time, so data
+the module keeps in its memory or globals carries over from one message to the
next.
+
+* When the function returns with the error bit set, the error data is released
with `dealloc` using its size without
+the error bit, the error data becomes the message of the exception, and the
instance is kept.
+* When the call fails without returning, for example on a Rust `panic` or any
other Wasm trap, the instance is
+discarded and a new one is created from the module for the next message.
Whatever the failed call allocated can no
+longer be released, so keeping the instance would grow its memory with every
failure. Any state the module kept in
+the discarded instance is lost.
+On a route where every call fails this way, the module is instantiated again
for every message (its data segments
+are copied and any start function runs), which costs more than a call on a
kept instance.
+
== Examples
Supposing we have compiled a Wasm module containing the function above, then
it can be called in a Camel Route by its name and module resource location:
diff --git a/components/camel-wasm/pom.xml b/components/camel-wasm/pom.xml
index 7cf013987204..d340565ec437 100644
--- a/components/camel-wasm/pom.xml
+++ b/components/camel-wasm/pom.xml
@@ -74,6 +74,12 @@
<artifactId>assertj-core</artifactId>
<scope>test</scope>
</dependency>
+ <dependency>
+ <groupId>run.endive</groupId>
+ <artifactId>wabt</artifactId>
+ <version>${endive-version}</version>
+ <scope>test</scope>
+ </dependency>
</dependencies>
diff --git a/components/camel-wasm/src/main/docs/wasm-component.adoc
b/components/camel-wasm/src/main/docs/wasm-component.adoc
index 5b09ac8f5c33..ad9aa3daf956 100644
--- a/components/camel-wasm/src/main/docs/wasm-component.adoc
+++ b/components/camel-wasm/src/main/docs/wasm-component.adoc
@@ -152,6 +152,21 @@ pub extern fn process(ptr: u32, len: u32) -> u64 {
}
----
+==== Errors and module state
+
+The module is instantiated once per producer (each `to` of a `wasm` endpoint
in a route has its own instance) and that
+instance handles every message, one at a time, so data the module keeps in its
memory or globals carries over from one
+message to the next.
+
+* When the function returns with the error bit set, the error data is released
with `dealloc` using its size without
+the error bit, the error data becomes the message of the exception, and the
instance is kept.
+* When the call fails without returning, for example on a Rust `panic` or any
other Wasm trap, the instance is
+discarded and a new one is created from the module for the next message.
Whatever the failed call allocated can no
+longer be released, so keeping the instance would grow its memory with every
failure. Any state the module kept in
+the discarded instance is lost.
+On a route where every call fails this way, the module is instantiated again
for every message (its data segments
+are copied and any start function runs), which costs more than a call on a
kept instance.
+
== Examples
Supposing we have compiled a Wasm module containing the function above, then
it can be called in a Camel Route by its name and module resource location:
diff --git a/components/camel-wasm/src/main/docs/wasm-language.adoc
b/components/camel-wasm/src/main/docs/wasm-language.adoc
index 72208ba9c90e..340d82b8157b 100644
--- a/components/camel-wasm/src/main/docs/wasm-language.adoc
+++ b/components/camel-wasm/src/main/docs/wasm-language.adoc
@@ -126,6 +126,20 @@ pub extern fn transform(ptr: u32, len: u32) -> u64 {
}
----
+=== Errors and module state
+
+The module is instantiated once per `wasm` expression and that instance
handles every message, one at a time, so data
+the module keeps in its memory or globals carries over from one message to the
next.
+
+* When the function returns with the error bit set, the error data is released
with `dealloc` using its size without
+the error bit, the error data becomes the message of the exception, and the
instance is kept.
+* When the call fails without returning, for example on a Rust `panic` or any
other Wasm trap, the instance is
+discarded and a new one is created from the module for the next message.
Whatever the failed call allocated can no
+longer be released, so keeping the instance would grow its memory with every
failure. Any state the module kept in
+the discarded instance is lost.
+On a route where every call fails this way, the module is instantiated again
for every message (its data segments
+are copied and any start function runs), which costs more than a call on a
kept instance.
+
== Examples
Supposing we have compiled a Wasm module containing the function above, then
it can be called in a Camel Route by its name and module resource location:
diff --git
a/components/camel-wasm/src/main/java/org/apache/camel/component/wasm/WasmProducer.java
b/components/camel-wasm/src/main/java/org/apache/camel/component/wasm/WasmProducer.java
index c81f0f0669fb..4f37664da67b 100644
---
a/components/camel-wasm/src/main/java/org/apache/camel/component/wasm/WasmProducer.java
+++
b/components/camel-wasm/src/main/java/org/apache/camel/component/wasm/WasmProducer.java
@@ -67,6 +67,9 @@ public class WasmProducer extends DefaultProducer {
public void doStop() throws Exception {
super.doStop();
+ if (this.function != null) {
+ this.function.close();
+ }
this.function = null;
}
diff --git
a/components/camel-wasm/src/main/java/org/apache/camel/wasm/WasmFunction.java
b/components/camel-wasm/src/main/java/org/apache/camel/wasm/WasmFunction.java
index d7890c743e0b..2c3ef70f1ed4 100644
---
a/components/camel-wasm/src/main/java/org/apache/camel/wasm/WasmFunction.java
+++
b/components/camel-wasm/src/main/java/org/apache/camel/wasm/WasmFunction.java
@@ -16,24 +16,30 @@
*/
package org.apache.camel.wasm;
+import java.nio.charset.StandardCharsets;
import java.util.Objects;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
import run.endive.runtime.ExportFunction;
import run.endive.runtime.Instance;
import run.endive.wasm.WasmModule;
public class WasmFunction implements AutoCloseable {
+ private static final Logger LOG =
LoggerFactory.getLogger(WasmFunction.class);
+
private final Lock lock;
private final WasmModule module;
private final String functionName;
- private final Instance instance;
- private final ExportFunction function;
- private final ExportFunction alloc;
- private final ExportFunction dealloc;
+ // guarded by lock; null after a failed call until the next call creates a
new instance
+ private Instance instance;
+ private ExportFunction function;
+ private ExportFunction alloc;
+ private ExportFunction dealloc;
public WasmFunction(WasmModule module, String functionName) {
this.lock = new ReentrantLock();
@@ -41,19 +47,22 @@ public class WasmFunction implements AutoCloseable {
this.module = Objects.requireNonNull(module);
this.functionName = Objects.requireNonNull(functionName);
- this.instance = Instance.builder(this.module).build();
- this.function = this.instance.export(this.functionName);
- this.alloc = this.instance.export(Wasm.FN_ALLOC);
- this.dealloc = this.instance.export(Wasm.FN_DEALLOC);
+ // the instance fields are not final: create the first instance under
the lock, so that a first run() on
+ // another thread sees them
+ lock.lock();
+ try {
+ createInstance();
+ } finally {
+ lock.unlock();
+ }
}
public byte[] run(byte[] in) throws Exception {
Objects.requireNonNull(in);
- int inPtr = -1;
- int inSize = in.length;
- int outPtr = -1;
- int outSize = 0;
+ final int inSize = in.length;
+ final byte[] out;
+ final boolean error;
//
// Wasm execution is not thread safe so we must put a
@@ -61,41 +70,92 @@ public class WasmFunction implements AutoCloseable {
//
lock.lock();
try {
+ if (instance == null) {
+ createInstance();
+ }
+
try {
- inPtr = (int) alloc.apply(inSize)[0];
+ int inPtr = (int) alloc.apply(inSize)[0];
instance.memory().write(inPtr, in);
long[] results = function.apply(inPtr, inSize);
long ptrAndSize = results[0];
- outPtr = (int) (ptrAndSize >> 32);
- outSize = (int) ptrAndSize;
+ int outPtr = (int) (ptrAndSize >> 32);
+ int outSize = (int) ptrAndSize;
// assume the max output is 31 bit, leverage the first bit for
// error detection
- if (isError(outSize)) {
- int errSize = errSize(outSize);
- String errData = instance.memory().readString(outPtr,
errSize);
-
- throw new RuntimeException(errData);
+ error = isError(outSize);
+ if (error) {
+ outSize = errSize(outSize);
}
- return instance.memory().readBytes(outPtr, outSize);
- } finally {
- if (inPtr != -1) {
- dealloc.apply(inPtr, inSize);
- }
- if (outPtr != -1) {
- dealloc.apply(outPtr, outSize);
- }
+ out = instance.memory().readBytes(outPtr, outSize);
+
+ dealloc.apply(inPtr, inSize);
+ dealloc.apply(outPtr, outSize);
+ } catch (RuntimeException | Error e) {
+ // The guest trapped, was interrupted or failed otherwise: its
memory and globals are left as the
+ // failed call left them, and whatever it allocated before the
failure can never be released.
+ // Discard the instance instead of calling into it again, so
the original exception is propagated
+ // as is, and create a new one on the next call.
+ discardInstance(e);
+ throw e;
}
} finally {
lock.unlock();
}
+
+ if (error) {
+ throw new RuntimeException(new String(out,
StandardCharsets.UTF_8));
+ }
+
+ return out;
}
@Override
public void close() throws Exception {
+ lock.lock();
+ try {
+ if (instance != null) {
+ instance.close();
+ }
+ clearInstance();
+ } finally {
+ lock.unlock();
+ }
+ }
+
+ private void createInstance() {
+ Instance newInstance = Instance.builder(this.module).build();
+ ExportFunction newFunction = newInstance.export(this.functionName);
+ ExportFunction newAlloc = newInstance.export(Wasm.FN_ALLOC);
+ ExportFunction newDealloc = newInstance.export(Wasm.FN_DEALLOC);
+
+ // all four fields refer to the same instance, or are left as they
were if an export is missing
+ this.instance = newInstance;
+ this.function = newFunction;
+ this.alloc = newAlloc;
+ this.dealloc = newDealloc;
+ }
+
+ private void discardInstance(Throwable cause) {
+ LOG.debug("Discarding the Wasm instance of function {} after a failed
call: {}", functionName, cause.getMessage());
+
+ try {
+ instance.close();
+ } catch (RuntimeException e) {
+ cause.addSuppressed(e);
+ }
+ clearInstance();
+ }
+
+ private void clearInstance() {
+ this.instance = null;
+ this.function = null;
+ this.alloc = null;
+ this.dealloc = null;
}
private static boolean isError(int number) {
diff --git
a/components/camel-wasm/src/test/java/org/apache/camel/wasm/WasmFunctionTest.java
b/components/camel-wasm/src/test/java/org/apache/camel/wasm/WasmFunctionTest.java
new file mode 100644
index 000000000000..3df5ad17aa00
--- /dev/null
+++
b/components/camel-wasm/src/test/java/org/apache/camel/wasm/WasmFunctionTest.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.camel.wasm;
+
+import java.util.Arrays;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicReference;
+
+import org.junit.jupiter.api.Test;
+import run.endive.runtime.TrapException;
+import run.endive.runtime.WasmInterruptedException;
+import run.endive.wabt.Wat2Wasm;
+import run.endive.wasm.Parser;
+import run.endive.wasm.WasmModule;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+import static org.awaitility.Awaitility.await;
+
+public class WasmFunctionTest {
+
+ /**
+ * A guest with one page of memory (64 KiB) and an arena allocator that
only gets its memory back when every
+ * allocation has been released. The process function looks at the first
byte of its input:
+ * <ul>
+ * <li>{@code t} allocates 32 KiB and traps before returning, like a Rust
panic on bad input</li>
+ * <li>{@code e} returns the error message {@code boom} with the error bit
set on its size</li>
+ * <li>{@code s} spins until the calling thread is interrupted</li>
+ * <li>anything else is echoed back</li>
+ * </ul>
+ * The dealloc function traps when it is given a size that does not fit in
31 bits.
+ */
+ private static final String GUEST = """
+ (module
+ (memory (export "memory") 1 1)
+ (data (i32.const 16) "boom")
+ (global $base i32 (i32.const 1024))
+ (global $next (mut i32) (i32.const 1024))
+ (global $live (mut i32) (i32.const 0))
+
+ (func $alloc (export "alloc") (param $size i32) (result i32)
+ (local $ptr i32)
+ (local.set $ptr (global.get $next))
+ (if (i32.gt_u (i32.add (local.get $ptr) (local.get $size))
(i32.const 65536))
+ (then unreachable))
+ (global.set $next (i32.add (local.get $ptr) (local.get $size)))
+ (global.set $live (i32.add (global.get $live) (i32.const 1)))
+ (local.get $ptr))
+
+ (func (export "dealloc") (param $ptr i32) (param $len i32)
+ (if (i32.lt_s (local.get $len) (i32.const 0))
+ (then unreachable))
+ (global.set $live (i32.sub (global.get $live) (i32.const 1)))
+ (if (i32.eqz (global.get $live))
+ (then (global.set $next (global.get $base)))))
+
+ (func (export "process") (param $ptr i32) (param $len i32)
(result i64)
+ (local $op i32)
+ (local $out i32)
+ (if (i32.gt_u (local.get $len) (i32.const 0))
+ (then (local.set $op (i32.load8_u (local.get $ptr)))))
+ (if (i32.eq (local.get $op) (i32.const 116))
+ (then
+ (drop (call $alloc (i32.const 32768)))
+ unreachable))
+ (if (i32.eq (local.get $op) (i32.const 101))
+ (then
+ (local.set $out (call $alloc (i32.const 4)))
+ (memory.copy (local.get $out) (i32.const 16) (i32.const 4))
+ (return (i64.or
+ (i64.shl (i64.extend_i32_u (local.get $out)) (i64.const
32))
+ (i64.const 0x80000004)))))
+ (if (i32.eq (local.get $op) (i32.const 115))
+ (then (loop $spin (br $spin))))
+ (local.set $out (call $alloc (local.get $len)))
+ (memory.copy (local.get $out) (local.get $ptr) (local.get
$len))
+ (i64.or
+ (i64.shl (i64.extend_i32_u (local.get $out)) (i64.const 32))
+ (i64.extend_i32_u (local.get $len))))
+ )
+ """;
+
+ private static WasmFunction newFunction() {
+ WasmModule module = Parser.parse(Wat2Wasm.parse(GUEST));
+ return new WasmFunction(module, "process");
+ }
+
+ private static byte[] data(char first, int size) {
+ byte[] data = new byte[size];
+ Arrays.fill(data, (byte) 'a');
+ data[0] = (byte) first;
+ return data;
+ }
+
+ @Test
+ public void testTrapDoesNotLeakGuestMemory() throws Exception {
+ WasmFunction function = newFunction();
+
+ // each trap leaves 32 KiB allocated in the guest, more than half of
its memory
+ for (int i = 0; i < 3; i++) {
+ assertThatThrownBy(() -> function.run(data('t', 1)))
+ .isInstanceOf(TrapException.class);
+ }
+
+ // needs 2 x 16 KiB: only fits if the memory of the failed calls was
given back
+ byte[] in = data('a', 16 * 1024);
+ assertThat(function.run(in)).isEqualTo(in);
+ }
+
+ @Test
+ public void testErrorIsDeallocatedWithItsSize() throws Exception {
+ WasmFunction function = newFunction();
+
+ // dealloc traps on a size with the error bit set, which would replace
the error of the function
+ assertThatThrownBy(() -> function.run(data('e', 1)))
+ .isExactlyInstanceOf(RuntimeException.class)
+ .hasMessage("boom");
+
+ byte[] in = data('a', 16);
+ assertThat(function.run(in)).isEqualTo(in);
+ }
+
+ @Test
+ public void testInterruptDoesNotLeakGuestMemory() throws Exception {
+ WasmFunction function = newFunction();
+ AtomicReference<Throwable> failure = new AtomicReference<>();
+
+ // the input of the interrupted call takes 40 KiB of the guest memory
until it is released
+ Thread caller = new Thread(() -> {
+ try {
+ function.run(data('s', 40 * 1024));
+ } catch (Throwable e) {
+ failure.set(e);
+ }
+ }, "wasm-caller");
+ // a daemon, so that a guest that is never interrupted cannot keep the
test JVM from exiting
+ caller.setDaemon(true);
+ caller.start();
+
+ try {
+ await().atMost(10, TimeUnit.SECONDS).until(() ->
isRunningGuestCode(caller));
+ caller.interrupt();
+ caller.join(TimeUnit.SECONDS.toMillis(10));
+
+ assertThat(caller.isAlive()).isFalse();
+
assertThat(failure.get()).isInstanceOf(WasmInterruptedException.class);
+
+ // needs 2 x 16 KiB: only fits if the input of the interrupted
call was given back
+ byte[] in = data('a', 16 * 1024);
+ assertThat(function.run(in)).isEqualTo(in);
+ } finally {
+ // stop the spinning guest even when an assertion above failed
+ caller.interrupt();
+ caller.join(TimeUnit.SECONDS.toMillis(10));
+ }
+ }
+
+ private static boolean isRunningGuestCode(Thread thread) {
+ // the package rather than a class, so that an engine switch or an
internal rename does not break the probe
+ return Arrays.stream(thread.getStackTrace())
+ .anyMatch(e -> e.getClassName().startsWith("run.endive."));
+ }
+}
diff --git
a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_23.adoc
b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_23.adoc
index 77e4031403e2..0652619a5217 100644
--- a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_23.adoc
+++ b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_23.adoc
@@ -3015,6 +3015,12 @@ Routes that deliberately read a `Camel`-prefixed header
produced by the Tika par
themselves after the `tika:parse` step, for example with a `setHeader` reading
the corresponding
non-prefixed metadata name.
+=== camel-wasm - a failed call discards the Wasm instance
+
+When a Wasm function fails without returning (a trap, such as a Rust `panic`),
the `wasm` producer or `wasm` expression
+now discards its module instance and creates a new one for the next message,
instead of reusing an instance whose memory
+kept everything the failed call had allocated. State that the module keeps
between messages is reset at that point.
+
=== camel-xpath
The XPath language now parses the message with the same hardened XML parser
for every `documentType`.
