Skip to content

Commit d1d4768

Browse files
owjs3901owjs3901
committed
Add e2e test
1 parent 015a444 commit d1d4768

5 files changed

Lines changed: 423 additions & 8 deletions

File tree

‎.github/workflows/CI.yml‎

Lines changed: 57 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -109,3 +109,60 @@ jobs:
109109
outputs:
110110
changepacks: ${{ steps.changepacks.outputs.changepacks }}
111111
release_assets_urls: ${{ steps.changepacks.outputs.release_assets_urls }}
112+
113+
# JNI end-to-end tests — builds the rust-jni-demo cdylib, publishes the
114+
# vespera-bridge JAR to mavenLocal (so the demo-app Gradle plugin can
115+
# resolve kr.devfive:vespera-bridge:1.0.0), then runs the full
116+
# :demo-app:test suite (StreamingClosureStressTest + JNI dispatch tests)
117+
# across all three target host OSes. This is the project's only Java/JNI
118+
# coverage gate — until now the workflow ran zero JNI tests.
119+
#
120+
# Runs unconditionally on every push/PR (matching the existing CI job's
121+
# style — no per-job paths-filter). The whole workflow already inherits
122+
# the workflow-level `paths-ignore` for docs-only changes.
123+
jni-e2e:
124+
name: JNI E2E (${{ matrix.os }})
125+
runs-on: ${{ matrix.os }}
126+
timeout-minutes: 25
127+
strategy:
128+
fail-fast: false
129+
matrix:
130+
os: [ubuntu-latest, windows-latest, macos-latest]
131+
steps:
132+
- uses: actions/checkout@v6
133+
- uses: actions/setup-java@v5
134+
with:
135+
distribution: 'temurin'
136+
java-version: '17'
137+
cache: 'gradle'
138+
- uses: actions-rust-lang/setup-rust-toolchain@v1
139+
- name: Build rust-jni-demo cdylib (release)
140+
# The vespera-bridge Gradle plugin's bundleNativeLib task copies
141+
# this cdylib from target/release into demo-app's resources, so it
142+
# must exist before `:demo-app:test` (processResources) runs.
143+
run: cargo build -p rust-jni-demo --release
144+
- name: Make gradlew executable (unix)
145+
if: runner.os != 'Windows'
146+
run: |
147+
chmod +x libs/vespera-bridge/gradlew
148+
chmod +x examples/rust-jni-demo/java/gradlew
149+
- name: Publish vespera-bridge to mavenLocal
150+
# demo-app resolves kr.devfive:vespera-bridge:1.0.0 from mavenLocal
151+
# (see examples/rust-jni-demo/java/demo-app/build.gradle.kts —
152+
# bridgeVersion.set("1.0.0")).
153+
shell: bash
154+
working-directory: libs/vespera-bridge
155+
run: ./gradlew publishToMavenLocal --console=plain --no-daemon
156+
- name: Run demo-app JNI E2E tests
157+
# Includes StreamingClosureStressTest (1000 × 1 MiB SHA256
158+
# bidirectional round-trip). Bench knobs are NOT propagated —
159+
# gated bench tests stay skipped in CI.
160+
shell: bash
161+
working-directory: examples/rust-jni-demo/java
162+
run: ./gradlew :demo-app:test --console=plain --no-daemon
163+
- name: Upload demo-app test results
164+
if: always()
165+
uses: actions/upload-artifact@v4
166+
with:
167+
name: jni-e2e-${{ matrix.os }}-test-results
168+
path: examples/rust-jni-demo/java/demo-app/build/test-results/test/*.xml

‎crates/vespera_jni/src/jni_impl.rs‎

Lines changed: 81 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -1,7 +1,13 @@
1-
use std::sync::LazyLock;
1+
use std::{
2+
cell::Cell,
3+
ffi::c_void,
4+
panic::{AssertUnwindSafe, catch_unwind, resume_unwind},
5+
ptr,
6+
sync::LazyLock,
7+
};
28

39
use jni::EnvUnowned;
4-
use jni::errors::ThrowRuntimeExAndDefault;
10+
use jni::errors::{ThrowRuntimeExAndDefault, jni_error_code_to_result};
511
use jni::objects::{Global, JByteArray, JByteBuffer, JClass, JObject};
612
use jni::sys::{jbyteArray, jint};
713

@@ -31,6 +37,75 @@ const MAX_RUNTIME_WORKERS: usize = 1024;
3137

3238
static RUNTIME_WORKER_THREADS: std::sync::OnceLock<Option<usize>> = std::sync::OnceLock::new();
3339

40+
thread_local! {
41+
static ASYNC_DAEMON_ENV: Cell<*mut jni::sys::JNIEnv> = const { Cell::new(ptr::null_mut()) };
42+
}
43+
44+
fn attach_async_daemon_thread(jvm: &jni::JavaVM) -> jni::errors::Result<*mut jni::sys::JNIEnv> {
45+
let raw_vm = jvm.get_raw();
46+
let mut env_ptr = ptr::null_mut::<c_void>();
47+
let mut args = jni::sys::JavaVMAttachArgs {
48+
version: jni::JNIVersion::V1_4.into(),
49+
name: ptr::null_mut(),
50+
group: ptr::null_mut(),
51+
};
52+
53+
// SAFETY: `raw_vm` comes from `Env::get_java_vm()` and is therefore a valid
54+
// JavaVM pointer for this process. JNI 1.4 provides
55+
// `AttachCurrentThreadAsDaemon`; the returned `JNIEnv` is valid only on the
56+
// current OS thread and is cached in thread-local storage below.
57+
let res = unsafe {
58+
((*(*raw_vm)).v1_4.AttachCurrentThreadAsDaemon)(
59+
raw_vm,
60+
&raw mut env_ptr,
61+
(&raw mut args).cast::<c_void>(),
62+
)
63+
};
64+
jni_error_code_to_result(res)?;
65+
if env_ptr.is_null() {
66+
return Err(jni::errors::Error::NullPtr("AttachCurrentThreadAsDaemon"));
67+
}
68+
69+
Ok(env_ptr.cast())
70+
}
71+
72+
fn with_async_daemon_env<F, T, E>(jvm: &jni::JavaVM, callback: F) -> std::result::Result<T, E>
73+
where
74+
F: FnOnce(&mut jni::Env<'_>) -> std::result::Result<T, E>,
75+
E: From<jni::errors::Error>,
76+
{
77+
ASYNC_DAEMON_ENV.with(|env_cell| {
78+
let mut env_ptr = env_cell.get();
79+
if env_ptr.is_null() {
80+
env_ptr = attach_async_daemon_thread(jvm)?;
81+
env_cell.set(env_ptr);
82+
}
83+
84+
// SAFETY: the pointer was produced for this exact Tokio worker thread
85+
// by `AttachCurrentThreadAsDaemon` and is never shared across threads
86+
// (TLS confines it). Tokio workers for the static runtime live until
87+
// process teardown, and daemon attachment means they do not keep the JVM
88+
// alive during shutdown. The per-call local frame prevents local-ref
89+
// accumulation on the permanently attached daemon thread. The explicit
90+
// post-call exception cleanup below replaces jni-rs scoped-detach
91+
// cleanup, which daemon attachments intentionally do not run.
92+
let mut guard = unsafe { jni::AttachGuard::from_unowned(env_ptr) };
93+
let env = guard.borrow_env_mut();
94+
let result = catch_unwind(AssertUnwindSafe(|| {
95+
env.with_local_frame(jni::DEFAULT_LOCAL_FRAME_CAPACITY, callback)
96+
}));
97+
98+
if env.exception_check() {
99+
env.exception_clear();
100+
}
101+
102+
match result {
103+
Ok(callback_result) => callback_result,
104+
Err(payload) => resume_unwind(payload),
105+
}
106+
})
107+
}
108+
34109
/// Worker thread count for the shared [`RUNTIME`], resolved once
35110
/// (first hit wins, then fixed for the process lifetime):
36111
///
@@ -391,10 +466,10 @@ pub extern "system" fn Java_com_devfive_vespera_bridge_VesperaBridge_dispatchAsy
391466
.await
392467
.unwrap_or_else(|_| vespera_inprocess::error_wire(500, "panic in Rust engine"));
393468

394-
// Re-attach to JVM on this worker thread; subsequent
395-
// dispatches on the same thread will hit the TLS fast
396-
// path (cheap).
397-
let _ = jvm.attach_current_thread(|env| -> jni::errors::Result<()> {
469+
// Complete on a cached daemon attachment for this Tokio
470+
// worker. This avoids attach/detach churn without making
471+
// runtime workers block JVM shutdown.
472+
let _ = with_async_daemon_env(&jvm, |env| -> jni::errors::Result<()> {
398473
complete_future(env, &future_global, &response)
399474
});
400475
});

‎examples/rust-jni-demo/java/demo-app/src/test/java/kr/go/demo/AsyncDispatchExceptionHygieneTest.java‎

Lines changed: 59 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,59 @@
1+
package kr.go.demo;
2+
3+
import static org.junit.jupiter.api.Assertions.assertEquals;
4+
import static org.junit.jupiter.api.Assertions.assertTrue;
5+
6+
import com.devfive.vespera.bridge.VesperaBridge;
7+
import java.util.Map;
8+
import java.util.concurrent.CompletableFuture;
9+
import java.util.concurrent.CountDownLatch;
10+
import java.util.concurrent.TimeUnit;
11+
import java.util.concurrent.atomic.AtomicInteger;
12+
import org.junit.jupiter.api.BeforeAll;
13+
import org.junit.jupiter.api.Test;
14+
15+
class AsyncDispatchExceptionHygieneTest {
16+
private static final Map<String, String> HEADERS = Map.of("accept", "application/json");
17+
private static final int TIMEOUT_SECONDS = 10;
18+
19+
@BeforeAll
20+
static void setUp() {
21+
System.setProperty("vespera.runtime.workerThreads", "1");
22+
VesperaBridge.init("rust_jni_demo");
23+
}
24+
25+
@Test
26+
void throwingFutureCompleteDoesNotPoisonNextAsyncCompletion() throws Exception {
27+
poisonAsyncCompletion();
28+
29+
CompletableFuture<byte[]> healthy = new CompletableFuture<>();
30+
VesperaBridge.dispatchAsync(healthy, healthRequest());
31+
32+
byte[] wireResponse = healthy.get(TIMEOUT_SECONDS, TimeUnit.SECONDS);
33+
assertEquals(200, VesperaBridge.decodeResponse(wireResponse).status());
34+
}
35+
36+
private static void poisonAsyncCompletion() throws InterruptedException {
37+
CountDownLatch completeCalled = new CountDownLatch(1);
38+
AtomicInteger completeCalls = new AtomicInteger();
39+
CompletableFuture<byte[]> throwingFuture = new CompletableFuture<>() {
40+
@Override
41+
public boolean complete(byte[] value) {
42+
completeCalls.incrementAndGet();
43+
completeCalled.countDown();
44+
throw new RuntimeException("intentional complete() failure");
45+
}
46+
};
47+
48+
VesperaBridge.dispatchAsync(throwingFuture, healthRequest());
49+
50+
assertTrue(
51+
completeCalled.await(TIMEOUT_SECONDS, TimeUnit.SECONDS),
52+
"poison future complete() must be invoked");
53+
assertEquals(1, completeCalls.get(), "poison future complete() call count");
54+
}
55+
56+
private static byte[] healthRequest() {
57+
return VesperaBridge.encodeRequest(null, "GET", "/health", null, HEADERS, null);
58+
}
59+
}

‎examples/rust-jni-demo/java/demo-app/src/test/java/kr/go/demo/SmallRequestLatencyBenchTest.java‎

Lines changed: 22 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -8,6 +8,8 @@
88
import java.io.OutputStream;
99
import java.nio.ByteBuffer;
1010
import java.util.Map;
11+
import java.util.concurrent.CompletableFuture;
12+
import java.util.concurrent.TimeUnit;
1113
import org.junit.jupiter.api.BeforeAll;
1214
import org.junit.jupiter.api.Test;
1315
import org.junit.jupiter.api.condition.EnabledIfSystemProperty;
@@ -88,6 +90,18 @@ private static int streamingOnce() throws IOException {
8890
return status[0];
8991
}
9092

93+
private static int asyncOnce() {
94+
byte[] wire = VesperaBridge.encodeRequest(null, "GET", "/health", null, HEADERS, null);
95+
CompletableFuture<byte[]> future = new CompletableFuture<>();
96+
VesperaBridge.dispatchAsync(future, wire);
97+
try {
98+
byte[] resp = future.get(30, TimeUnit.SECONDS);
99+
return VesperaBridge.decodeResponse(resp).status();
100+
} catch (Exception e) {
101+
throw new RuntimeException(e);
102+
}
103+
}
104+
91105
/** Response-streaming only — no request pull thread (empty body inline). */
92106
private static int responseStreamingOnce() {
93107
byte[] wire = VesperaBridge.encodeRequest(null, "GET", "/health", null, HEADERS, null);
@@ -130,11 +144,17 @@ void smallRequestLatencyByMode() throws IOException {
130144
SmallRequestLatencyBenchTest::responseStreamingOnce);
131145
long streaming =
132146
measure("bidirectional_streaming", SmallRequestLatencyBenchTest::streamingOnce);
147+
long async =
148+
measure(
149+
"async_completable_future",
150+
SmallRequestLatencyBenchTest::asyncOnce);
133151
System.out.printf(
134152
"VESPERA_BENCH summary direct_vs_streaming=%.2fx direct_vs_sync=%.2fx"
135-
+ " resp_only_vs_bidi=%.2fx%n",
153+
+ " resp_only_vs_bidi=%.2fx async_vs_sync=%.2fx async_vs_direct=%.2fx%n",
136154
(double) streaming / direct,
137155
(double) sync / direct,
138-
(double) streaming / respStreaming);
156+
(double) streaming / respStreaming,
157+
(double) async / sync,
158+
(double) async / direct);
139159
}
140160
}

0 commit comments

Comments
 (0)