From 737fd0427e549c0cef9a58b3dd5bd1edeefa5685 Mon Sep 17 00:00:00 2001 From: Jeong Min Oh Date: Sun, 15 Nov 2026 09:44:15 +0900 Subject: [PATCH 1/2] fix(bridge): round-7 audit fixes - VesperaRequest: a read that fails because the native side closed the request source after the response drained no longer fails the send - proxy: DispatchEvent.mode reports SYNC / STREAMING when a DIRECT request is downgraded or re-routed; an unknown-app 404 reports its path - VesperaBridge: isInitialized() is published only after the pending configure* values were applied - dispatchDirectPooled: the heap fallback parks on a virtual thread - gradle plugin: extraNativeLibs without cargoRoot/targetDir bundles only the prebuilt libraries (fails when none) - jacoco merges demo-app isolatedJvmTest.exec - docs: DIRECT overflow stash (handler runs once), version check, configureStreaming ranges, request stream close, ASYNC callers Co-Authored-By: Claude Opus 5.5 --- .../changepack_log_bridge-round7.json | 1 + .github/workflows/CI.yml | 3 +- crates/vespera_jni/src/jni_impl_async.rs | 10 ++- .../java/demo-app/build.gradle.kts | 2 +- libs/vespera-bridge-gradle-plugin/README.md | 6 +- .../devfive/vespera/VesperaBridgeExtension.kt | 5 +- .../kr/devfive/vespera/VesperaBridgePlugin.kt | 20 +++++ .../VesperaBridgePluginFunctionalTest.kt | 22 ++++++ libs/vespera-bridge/README.md | 6 +- libs/vespera-bridge/build.gradle.kts | 5 ++ .../vespera/bridge/DirectModeDispatcher.java | 25 ++++-- .../vespera/bridge/DirectOverflowMemory.java | 13 ++-- .../devfive/vespera/bridge/ProxyExchange.java | 12 +++ .../devfive/vespera/bridge/VesperaBridge.java | 56 +++++++++----- .../VesperaBridgeAutoConfiguration.java | 10 ++- .../bridge/VesperaBridgeProperties.java | 22 ++++-- .../bridge/VesperaDirectBufferPool.java | 15 +++- .../bridge/VesperaProxyController.java | 41 +++++----- .../vespera/bridge/VesperaRequest.java | 19 ++++- .../vespera/bridge/VesperaVersions.java | 6 +- .../bridge/ConfigureStreamingTest.java | 2 +- .../vespera/bridge/ProxyOperabilityTest.java | 1 + .../bridge/ProxyResolverAndCorsTest.java | 2 +- .../vespera/bridge/VesperaBridgeInitTest.java | 40 +++++++++- .../vespera/bridge/VesperaClientTest.java | 76 +++++++++++++++++++ .../bridge/VesperaDirectWrapperTest.java | 23 +++++- .../VesperaProxyControllerCoverageTest.java | 60 ++++++++++++++- 27 files changed, 411 insertions(+), 92 deletions(-) create mode 100644 .changepacks/changepack_log_bridge-round7.json diff --git a/.changepacks/changepack_log_bridge-round7.json b/.changepacks/changepack_log_bridge-round7.json new file mode 100644 index 00000000..0fc8c198 --- /dev/null +++ b/.changepacks/changepack_log_bridge-round7.json @@ -0,0 +1 @@ +{"changes":{"crates/vespera/Cargo.toml":"Patch","crates/vespera_core/Cargo.toml":"Patch","crates/vespera_macro/Cargo.toml":"Patch","crates/vespera_inprocess/Cargo.toml":"Patch","crates/vespera_jni/Cargo.toml":"Patch","libs/vespera-bridge/build.gradle.kts":"Patch","libs/vespera-bridge-gradle-plugin/build.gradle.kts":"Patch"},"note":"vespera-bridge: a VesperaRequest InputStream read that fails because the native side closed the source after the response drained no longer fails the send; DispatchEvent.mode reports SYNC / STREAMING when a DIRECT request is downgraded or re-routed, and an unknown-app 404 reports its path; isInitialized() is published only after pending configure* values were applied; the public dispatchDirectPooled heap fallback parks on a virtual thread instead of pinning the carrier; Gradle plugin: extraNativeLibs without cargoRoot / targetDir bundles only the prebuilt libraries; docs on DIRECT overflow, version check, configureStreaming ranges and JaCoCo merge of isolatedJvmTest fixed","date":"2026-11-15T00:00:00.000Z"} diff --git a/.github/workflows/CI.yml b/.github/workflows/CI.yml index 23faf505..08fc7247 100644 --- a/.github/workflows/CI.yml +++ b/.github/workflows/CI.yml @@ -956,7 +956,7 @@ jobs: # Regenerates the bridge's JaCoCo report now that demo-app has run, so # it also credits the classes only reachable with a loaded cdylib — the # JNI wrappers and the Spring proxy, which the unit suite cannot drive. - # The report task reads demo-app/build/jacoco/test.exec + # The report task reads demo-app/build/jacoco/{test,isolatedJvmTest}.exec # (and demo-reactive-app's) when present. shell: bash working-directory: libs/vespera-bridge @@ -968,6 +968,7 @@ jobs: name: jni-e2e-${{ matrix.os }}-jdk${{ matrix.java }}-test-results path: | examples/rust-jni-demo/java/demo-app/build/test-results/test/*.xml + examples/rust-jni-demo/java/demo-app/build/test-results/isolatedJvmTest/*.xml examples/rust-jni-demo/java/demo-reactive-app/build/test-results/test/*.xml libs/vespera-bridge/build/test-results/test/*.xml libs/vespera-bridge-gradle-plugin/build/test-results/test/*.xml diff --git a/crates/vespera_jni/src/jni_impl_async.rs b/crates/vespera_jni/src/jni_impl_async.rs index f2d49b93..fb2f5ddc 100644 --- a/crates/vespera_jni/src/jni_impl_async.rs +++ b/crates/vespera_jni/src/jni_impl_async.rs @@ -69,9 +69,13 @@ use crate::streaming_closures::{CompleteAttempt, complete_future, complete_futur /// group `async_completion_ab`: ~1.5 µs inline vs ~24.5 µs hand-off), so the /// worker-thread completion is kept and this contract is documented instead — /// matching how Netty / async HTTP clients complete futures from their I/O -/// threads. The autoconfigured Spring proxy never selects `ASYNC` (its -/// `SmartDispatchModeResolver` uses DIRECT / SYNC / streaming), so this path is -/// opt-in for callers doing their own `CompletableFuture` composition. +/// threads. Java callers that reach this symbol: the WebFlux proxy +/// (`VesperaReactiveProxyHandler`, every buffered request), `VesperaClient`'s +/// `sendAsync()`, SYNC on a virtual thread with +/// `-Dvespera.virtualThread.pollFirst=false` (`ParkingDispatch`), a servlet +/// proxy whose resolver picks `ASYNC`, and callers doing their own +/// `CompletableFuture` composition (the servlet proxy's default +/// `SmartDispatchModeResolver` never selects `ASYNC`). /// /// Cancellation: the spawned dispatch is registered with /// [`vespera_inprocess::cancellable`] under the returned token. The Java diff --git a/examples/rust-jni-demo/java/demo-app/build.gradle.kts b/examples/rust-jni-demo/java/demo-app/build.gradle.kts index 0a374578..7bfdda8a 100644 --- a/examples/rust-jni-demo/java/demo-app/build.gradle.kts +++ b/examples/rust-jni-demo/java/demo-app/build.gradle.kts @@ -3,7 +3,7 @@ plugins { // Records execution data for the vespera-bridge classes this suite drives // through a REAL loaded cdylib — the JNI wrappers and the Spring proxy that // the bridge's own unit tests cannot reach. libs/vespera-bridge merges this - // `build/jacoco/test.exec` into its report. + // `build/jacoco/test.exec` and `isolatedJvmTest.exec` into its report. jacoco id("org.springframework.boot") version "4.1.2" id("io.spring.dependency-management") version "1.1.7" diff --git a/libs/vespera-bridge-gradle-plugin/README.md b/libs/vespera-bridge-gradle-plugin/README.md index 1de1b265..156ef9a0 100644 --- a/libs/vespera-bridge-gradle-plugin/README.md +++ b/libs/vespera-bridge-gradle-plugin/README.md @@ -8,11 +8,11 @@ Gradle plugin `kr.devfive.vespera-bridge` (Maven Central / Plugin Portal: `kr.de |---|---|---| | `crateName` | — (required) | Cargo package (`my-app`) or its underscore form; selects the library file name | | `packageName` / `libName` | derived | `cargo build -p` spec / custom `[lib] name` | -| `cargoRoot`, `targetDir`, `cargoProfile` | `/target`, `release` | where the cdylib is read from | +| `cargoRoot`, `targetDir`, `cargoProfile` | `/target`, `release` | where the cdylib is read from (leave both locations unset to bundle only `extraNativeLibs`) | | `autoBuildCargo` | `false` | run `cargoBuild` before `bundleNativeLib` | | `targets` | empty | extra cross builds: platform → Rust triple | -| `extraNativeLibs` | unset | prebuilt `native//` merged into the app jar | -| `bundleNative` | `true` | `false`: `processResources` bundles nothing (the app takes its library from native jars) | +| `extraNativeLibs` | unset | prebuilt `native//` merged into the app jar (alone, without `cargoRoot` / `targetDir`: the only libraries bundled) | +| `bundleNative` | `true` | `false`: `processResources` bundles nothing, `extraNativeLibs` included (the app takes its library from native jars) | | `nativeJarPlatforms` | empty | platforms to package as classifier jars (`nativeJar`, `nativeJars`) | | `nativeJarsDir` | `extraNativeLibs` | source of the native jars (`native//`) | | `bridgeVersion` | the bridge release this plugin was built with | `vespera-bridge` dependency version | diff --git a/libs/vespera-bridge-gradle-plugin/src/main/kotlin/kr/devfive/vespera/VesperaBridgeExtension.kt b/libs/vespera-bridge-gradle-plugin/src/main/kotlin/kr/devfive/vespera/VesperaBridgeExtension.kt index 86b53b96..015c6ae2 100644 --- a/libs/vespera-bridge-gradle-plugin/src/main/kotlin/kr/devfive/vespera/VesperaBridgeExtension.kt +++ b/libs/vespera-bridge-gradle-plugin/src/main/kotlin/kr/devfive/vespera/VesperaBridgeExtension.kt @@ -107,7 +107,10 @@ abstract class VesperaBridgeExtension @Inject constructor(objects: ObjectFactory * `native/-/` (for example CI artifacts from other * platforms). Everything under its `native/` folder is merged into the * jar next to the locally built libraries; a locally built library wins - * when both provide the same file. + * when both provide the same file. Without [cargoRoot] / [targetDir] + * only these prebuilt libraries are bundled (no local build is + * expected; the build fails when the folder holds none). With + * `bundleNative = false` they are not bundled either. */ abstract val extraNativeLibs: DirectoryProperty diff --git a/libs/vespera-bridge-gradle-plugin/src/main/kotlin/kr/devfive/vespera/VesperaBridgePlugin.kt b/libs/vespera-bridge-gradle-plugin/src/main/kotlin/kr/devfive/vespera/VesperaBridgePlugin.kt index d2da7d94..ae151b10 100644 --- a/libs/vespera-bridge-gradle-plugin/src/main/kotlin/kr/devfive/vespera/VesperaBridgePlugin.kt +++ b/libs/vespera-bridge-gradle-plugin/src/main/kotlin/kr/devfive/vespera/VesperaBridgePlugin.kt @@ -128,6 +128,9 @@ class VesperaBridgePlugin : Plugin { val base = targetBase.orNull val crate = ext.crateName.orNull val profile = ext.cargoProfile.get() + // Prebuilt-only jar: extraNativeLibs without cargoRoot / + // targetDir bundles just those libraries, no local build. + val prebuiltDir = if (base == null) ext.extraNativeLibs.orNull?.asFile else null val expected = if (base == null) { emptyList() } else { @@ -137,6 +140,10 @@ class VesperaBridgePlugin : Plugin { } t.doLast(object : org.gradle.api.Action { override fun execute(@Suppress("UNUSED_PARAMETER") task: Task) { + if (prebuiltDir != null) { + requirePrebuilt(prebuiltDir) + return + } requireLocation(crate, base) expected.forEach { lib -> lib.requireExists(crate.orEmpty(), profile) } } @@ -250,6 +257,19 @@ class VesperaBridgePlugin : Plugin { } } + /** + * Prebuilt-only bundling (`extraNativeLibs` without `cargoRoot` / + * `targetDir`): fail when the directory holds no library at all, + * where the `Sync` would be NO-SOURCE and the jar ship none. + */ + fun requirePrebuilt(dir: File) { + require(File(dir, "native").walk().any { it.isFile }) { + "vespera.extraNativeLibs ($dir) holds no native// file and neither " + + "vespera.cargoRoot nor vespera.targetDir is set: the jar would ship no native " + + "library." + } + } + /** The embedded default `bridgeVersion`, or `null` when the resource is missing. */ fun embeddedBridgeVersion(): String? = VesperaBridgePlugin::class.java.getResourceAsStream(VERSION_RESOURCE)?.use { input -> diff --git a/libs/vespera-bridge-gradle-plugin/src/test/kotlin/kr/devfive/vespera/VesperaBridgePluginFunctionalTest.kt b/libs/vespera-bridge-gradle-plugin/src/test/kotlin/kr/devfive/vespera/VesperaBridgePluginFunctionalTest.kt index f897c820..c331aaa9 100644 --- a/libs/vespera-bridge-gradle-plugin/src/test/kotlin/kr/devfive/vespera/VesperaBridgePluginFunctionalTest.kt +++ b/libs/vespera-bridge-gradle-plugin/src/test/kotlin/kr/devfive/vespera/VesperaBridgePluginFunctionalTest.kt @@ -227,6 +227,28 @@ class VesperaBridgePluginFunctionalTest { assertTrue(result.output.contains("vespera.targetDir"), result.output) } + @Test + fun extraNativeLibsWithoutCargoRootBundlesOnlyThePrebuiltLibraries() { + write("prebuilt/native/$extraPlatform/libmy_app.dylib", "prebuilt") + setUpWithoutCargoRoot(""" extraNativeLibs.set(layout.projectDirectory.dir("prebuilt"))""") + + val result = run("processResources").build() + + assertEquals(TaskOutcome.SUCCESS, result.task(":bundleNativeLib")?.outcome) + assertEquals("prebuilt", resource("$extraPlatform/libmy_app.dylib").readText()) + } + + @Test + fun emptyExtraNativeLibsWithoutCargoRootFailsTheBuild() { + File(dir, "prebuilt").mkdirs() + setUpWithoutCargoRoot(""" extraNativeLibs.set(layout.projectDirectory.dir("prebuilt"))""") + + val result = run("processResources").buildAndFail() + + assertTrue(result.output.contains("holds no native// file"), result.output) + assertFalse(File(dir, "build/resources/main/native").exists()) + } + @Test fun hostKeyMappedToMuslStillBundlesTheHostLibrary() { assumeTrue(host == "linux-x86_64", "needs a glibc linux-x86_64 build host") diff --git a/libs/vespera-bridge/README.md b/libs/vespera-bridge/README.md index af7c05e0..c3f1ef5f 100644 --- a/libs/vespera-bridge/README.md +++ b/libs/vespera-bridge/README.md @@ -70,7 +70,7 @@ vespera { } ``` -Each `targets` key is a bridge platform (`-`, os `windows` / `macos` / `linux`); its library is read from `///` and bundled under `native/-/` with that platform's file name (`.dll` / `lib*.dylib` / `lib*.so`). A plain `linux-` key built for a `*-linux-musl*` triple is bundled under `native/linux--musl/` (where the bridge looks first on a musl JVM); `linux--musl` keys are accepted too, so one jar can carry both builds, e.g. `targets.put("linux-x86_64-musl", "x86_64-unknown-linux-musl")` next to the glibc host build. A target bundled into the host platform's folder replaces the host build (on an Alpine build host the host platform is `linux--musl`); the resolved folder counts, not the key, so `targets.put("linux-x86_64", "x86_64-unknown-linux-musl")` on a glibc `linux-x86_64` host keeps the host build in `native/linux-x86_64/` next to `native/linux-x86_64-musl/`. The cross toolchain/linker must already be installed. `bundleNativeLib` depends on `verifyNativeLibs`, which fails the build with `Native library not found` when an expected library was not built (also when none was; set `vespera.autoBuildCargo = true` or build it first); with `bundleNative = false` neither task runs. Everything under `extraNativeLibs/native/` is merged in; a locally built library wins over a prebuilt one with the same path. `cargoBuild` is never `UP-TO-DATE` (Cargo's own incremental check decides) and declares no file inputs. +Each `targets` key is a bridge platform (`-`, os `windows` / `macos` / `linux`); its library is read from `///` and bundled under `native/-/` with that platform's file name (`.dll` / `lib*.dylib` / `lib*.so`). A plain `linux-` key built for a `*-linux-musl*` triple is bundled under `native/linux--musl/` (where the bridge looks first on a musl JVM); `linux--musl` keys are accepted too, so one jar can carry both builds, e.g. `targets.put("linux-x86_64-musl", "x86_64-unknown-linux-musl")` next to the glibc host build. A target bundled into the host platform's folder replaces the host build (on an Alpine build host the host platform is `linux--musl`); the resolved folder counts, not the key, so `targets.put("linux-x86_64", "x86_64-unknown-linux-musl")` on a glibc `linux-x86_64` host keeps the host build in `native/linux-x86_64/` next to `native/linux-x86_64-musl/`. The cross toolchain/linker must already be installed. `bundleNativeLib` depends on `verifyNativeLibs`, which fails the build with `Native library not found` when an expected library was not built (also when none was; set `vespera.autoBuildCargo = true` or build it first); with `bundleNative = false` neither task runs. Everything under `extraNativeLibs/native/` is merged in; a locally built library wins over a prebuilt one with the same path. With neither `cargoRoot` nor `targetDir` set, only the `extraNativeLibs` libraries are bundled (no local build is expected; `verifyNativeLibs` fails when that folder holds none); `bundleNative = false` bundles none of them either. `cargoBuild` is never `UP-TO-DATE` (Cargo's own incremental check decides) and declares no file inputs. #### Per-platform native jars (publish once per platform, consume one) @@ -1157,8 +1157,8 @@ runs. Configuration precedence (first hit wins, then cached): ```java // Configure streaming parameters before init VesperaBridge.configureStreaming( - 131072, // chunkBytes: 128 KiB (clamped to 4 KiB – 8 MiB) - 32 // channelCapacity: 32 slots (clamped to 1 – 1024) + 131072, // chunkBytes: 128 KiB (must be within 4 KiB – 8 MiB) + 32 // channelCapacity: 32 slots (must be within 1 – 1024) ); VesperaBridge.init("my_rust_lib"); ``` diff --git a/libs/vespera-bridge/build.gradle.kts b/libs/vespera-bridge/build.gradle.kts index d46110e9..5f0b97a9 100644 --- a/libs/vespera-bridge/build.gradle.kts +++ b/libs/vespera-bridge/build.gradle.kts @@ -225,6 +225,10 @@ tasks.named("test") { // match because demo-app resolves this exact build's JAR from mavenLocal. val demoAppExecutionData = rootProject.layout.projectDirectory .file("../../examples/rust-jni-demo/java/demo-app/build/jacoco/test.exec") +// demo-app's `isolated-jvm` tests (library-name auto-init, the buffered-response +// cap, ...) run in its `isolatedJvmTest` task, which writes its own file. +val demoAppIsolatedExecutionData = rootProject.layout.projectDirectory + .file("../../examples/rust-jni-demo/java/demo-app/build/jacoco/isolatedJvmTest.exec") // The WebFlux E2E app (demo-reactive-app) drives VesperaReactiveProxyHandler // through the same real cdylib. val demoReactiveAppExecutionData = rootProject.layout.projectDirectory @@ -236,6 +240,7 @@ tasks.named("jacocoTestReport") { files( layout.buildDirectory.file("jacoco/test.exec"), demoAppExecutionData, + demoAppIsolatedExecutionData, demoReactiveAppExecutionData, ).filter { it.exists() }, ) diff --git a/libs/vespera-bridge/src/main/java/com/devfive/vespera/bridge/DirectModeDispatcher.java b/libs/vespera-bridge/src/main/java/com/devfive/vespera/bridge/DirectModeDispatcher.java index 1daed707..dfb3f660 100644 --- a/libs/vespera-bridge/src/main/java/com/devfive/vespera/bridge/DirectModeDispatcher.java +++ b/libs/vespera-bridge/src/main/java/com/devfive/vespera/bridge/DirectModeDispatcher.java @@ -7,20 +7,27 @@ /** * The proxy's {@link DispatchMode#DIRECT} path: a pooled direct-buffer - * dispatch plus its two overflow fallbacks (a response stashed on the Rust - * side, and a safe-method re-run through response streaming), and the - * adaptive {@link DirectOverflowMemory} that streams known-large routes up - * front. + * dispatch plus its overflow handling, and the adaptive + * {@link DirectOverflowMemory} that streams known-large routes up front. + * + *

An overflow normally runs the handler once: the finished + * response is kept on the Rust side (the overflow stash) and either fetched + * into a grown pooled buffer or, beyond the pool cap / for an event stream, + * streamed from the stash. Only when the native side hands back no stash + * entry does the legacy fallback apply: a safe-method re-run through + * response streaming ({@code direct-retry-on-overflow}), which runs the + * handler a second time. * *

Only safe methods reach {@link #dispatch}: the controller - * downgrades every other method to {@code SYNC} first, because an overflow - * retry re-runs the Rust handler. + * downgrades every other method to {@code SYNC} first, because of that + * legacy re-run. A request served through another path is recorded on the + * {@link ProxyExchange} ({@link DispatchEvent#mode()}). */ final class DirectModeDispatcher { // Adaptive DIRECT-overflow avoidance: routes that overflowed the pooled // direct buffer once are streamed up front thereafter, removing the - // repeated DIRECT-overflow-then-stream double dispatch a known-large + // repeated overflow (whole response stashed, then streamed) a known-large // (download) route would otherwise pay on every request. Internal state, // so it is created directly rather than injected. private final DirectOverflowMemory overflowMemory = new DirectOverflowMemory(); @@ -63,7 +70,7 @@ void dispatch( HttpServletResponse response, String appName, String method, String path, String query, VesperaBridge.HeaderSource headers, byte[] body, - Boolean currentThreadIsVirtual) throws IOException { + Boolean currentThreadIsVirtual, ProxyExchange exchange) throws IOException { boolean retry = shouldRetry(retryOnOverflow, method); ByteBuffer wireResp; try { @@ -93,6 +100,7 @@ void dispatch( // no handler ran and no response overflowed, so stream this one // request (any method — it runs once) without remembering the // route as an overflowing one. + ProxyExchange.servedAs(exchange, DispatchMode.STREAMING); streamResponse(response, appName, method, path, query, headers, body); return; } catch (VesperaBridge.BufferTooSmallException overflow) { @@ -110,6 +118,7 @@ void dispatch( // Remember this route so the NEXT request to it streams up // front — recorded only here, where we actually fall back. overflowMemory.recordOverflow(appName, method, path, query); + ProxyExchange.servedAs(exchange, DispatchMode.STREAMING); streamResponse(response, appName, method, path, query, headers, body); return; } diff --git a/libs/vespera-bridge/src/main/java/com/devfive/vespera/bridge/DirectOverflowMemory.java b/libs/vespera-bridge/src/main/java/com/devfive/vespera/bridge/DirectOverflowMemory.java index 49aba47e..79966a21 100644 --- a/libs/vespera-bridge/src/main/java/com/devfive/vespera/bridge/DirectOverflowMemory.java +++ b/libs/vespera-bridge/src/main/java/com/devfive/vespera/bridge/DirectOverflowMemory.java @@ -9,11 +9,14 @@ * those targets directly on subsequent requests. * *

Without this, a known-large (e.g. download) route routed to - * {@link DispatchMode#DIRECT} pays the DIRECT-overflow-then-stream - * double dispatch on every request: the Rust handler - * runs once into the pooled direct buffer (overflows), then again through - * response streaming. After the first overflow this memory downgrades the route - * to {@link DispatchMode#STREAMING} up front, so later requests dispatch once. + * {@link DispatchMode#DIRECT} pays the overflow on every request: the + * handler runs once, but its whole response is first held on the Rust side + * (the overflow stash) and only then streamed to the client — a full + * buffered copy plus an extra stash round-trip, instead of chunk-by-chunk + * streaming. (With a native library that hands back no stash entry the + * fallback even re-runs the handler through response streaming.) After the + * first overflow this memory downgrades the route to + * {@link DispatchMode#STREAMING} up front. * *

Thread-safe and bounded. {@link #shouldAvoidDirect} is a single volatile * read until the first overflow is recorded, so apps that never overflow DIRECT diff --git a/libs/vespera-bridge/src/main/java/com/devfive/vespera/bridge/ProxyExchange.java b/libs/vespera-bridge/src/main/java/com/devfive/vespera/bridge/ProxyExchange.java index 85fef7c9..3bc30371 100644 --- a/libs/vespera-bridge/src/main/java/com/devfive/vespera/bridge/ProxyExchange.java +++ b/libs/vespera-bridge/src/main/java/com/devfive/vespera/bridge/ProxyExchange.java @@ -49,6 +49,18 @@ final class ProxyExchange { this.requestId = requestId; } + /** + * Record that the request is actually served by {@code served} (a DIRECT + * request downgraded to SYNC or re-routed through STREAMING), so the + * reported {@link DispatchEvent#mode()} names the path that ran. + * {@code exchange} may be {@code null} (direct calls from tests). + */ + static void servedAs(ProxyExchange exchange, DispatchMode served) { + if (exchange != null) { + exchange.mode = served; + } + } + /** Whether {@link #complete} already ran. */ boolean isCompleted() { return completed.get(); diff --git a/libs/vespera-bridge/src/main/java/com/devfive/vespera/bridge/VesperaBridge.java b/libs/vespera-bridge/src/main/java/com/devfive/vespera/bridge/VesperaBridge.java index 96180ca0..e676a34f 100644 --- a/libs/vespera-bridge/src/main/java/com/devfive/vespera/bridge/VesperaBridge.java +++ b/libs/vespera-bridge/src/main/java/com/devfive/vespera/bridge/VesperaBridge.java @@ -70,7 +70,14 @@ default VesperaPrincipal principal() { } } + /** + * Published ({@link #isInitialized()}) only after the pending + * configuration was applied: a dispatcher that sees it can no longer fix + * the engine defaults before a pending {@code configure*} value lands. + */ private static volatile boolean loaded = false; + /** The cdylib is loaded (set before the pending configuration is applied); guarded by the class lock. */ + private static boolean nativeLoaded = false; private static volatile boolean shutDown = false; /** Name passed to the first successful {@link #init(String)} — used to * reject a later re-init with a different library name. */ @@ -162,7 +169,7 @@ public byte[] bodyBytes() { */ public static synchronized void init(String libraryName) { Objects.requireNonNull(libraryName, "libraryName"); - if (loaded) { + if (nativeLoaded) { // Re-init with the SAME library is a no-op (friendly for test // harness resets / repeated Spring context starts). A DIFFERENT // name is a bug — a JVM process loads exactly one vespera cdylib @@ -181,14 +188,27 @@ public static synchronized void init(String libraryName) { // Bridge JAR <-> native library release check (VesperaVersions); a // major.minor mismatch throws before the library is marked loaded. VesperaVersions.verify(VesperaBridge::nativeVersion0); - // Mark the native library as loaded once System.load succeeded and the - // versions match: the post-load configuration hooks below may still - // throw, but a later init() must not load the same cdylib again. - loaded = true; + completeLoad(libraryName, VesperaNativeConfig::applyAfterLoad); + } + + /** + * Mark the native library as loaded once System.load succeeded and the + * versions match (a later init() must not load the same cdylib again, + * even when the post-load configuration hooks throw), apply the pending + * configuration (set via configure* before init; pending values beat + * system properties: Rust-side setter > env > default), and only + * then publish {@link #isInitialized()}. Publishing first would let a + * concurrent first dispatch fix the defaults before the pending values + * are applied. + */ + static synchronized void completeLoad(String libraryName, Runnable applyPendingConfig) { + nativeLoaded = true; loadedLibraryName = libraryName; - // Apply pending config (set via configure* before init); pending - // values beat system properties (Rust-side setter > env > default). - VesperaNativeConfig.applyAfterLoad(); + try { + applyPendingConfig.run(); + } finally { + loaded = true; + } } /** Whether {@link #init(String)} has loaded the native library in this class loader. */ @@ -210,7 +230,7 @@ public static boolean isShutdown() { */ public static synchronized boolean shutdown(Duration timeout) { Objects.requireNonNull(timeout, "timeout"); - if (!loaded) { + if (!nativeLoaded) { return false; } long millis = shutdownTimeoutMillis(timeout); @@ -271,7 +291,7 @@ static long shutdownTimeoutMillis(Duration timeout) { */ public static synchronized void configureStreaming(int chunkBytes, int channelCapacity) { BridgeInputChecks.requireStreamingConfig(chunkBytes, channelCapacity); - VesperaNativeConfig.setStreaming(chunkBytes, channelCapacity, loaded); + VesperaNativeConfig.setStreaming(chunkBytes, channelCapacity, nativeLoaded); } /** @@ -286,9 +306,9 @@ public static synchronized void configureStreaming(int chunkBytes, int channelCa * @return {@code false} when a value could not be applied (logged) */ public static synchronized boolean configureRuntime(int workerThreads, int maxBlockingThreads) { - boolean workers = VesperaNativeConfig.RUNTIME_WORKER_THREADS.set(workerThreads, loaded); + boolean workers = VesperaNativeConfig.RUNTIME_WORKER_THREADS.set(workerThreads, nativeLoaded); boolean blocking = - VesperaNativeConfig.RUNTIME_MAX_BLOCKING_THREADS.set(maxBlockingThreads, loaded); + VesperaNativeConfig.RUNTIME_MAX_BLOCKING_THREADS.set(maxBlockingThreads, nativeLoaded); return workers && blocking; } @@ -307,11 +327,11 @@ public static synchronized boolean configureRuntime(int workerThreads, int maxBl */ public static synchronized boolean configureLimits( long maxRequestBytes, long maxStreamingRequestBytes, long maxBufferedResponseBytes) { - boolean request = VesperaNativeConfig.MAX_REQUEST_BYTES.set(maxRequestBytes, loaded); + boolean request = VesperaNativeConfig.MAX_REQUEST_BYTES.set(maxRequestBytes, nativeLoaded); boolean streaming = VesperaNativeConfig.MAX_STREAMING_REQUEST_BYTES - .set(maxStreamingRequestBytes, loaded); + .set(maxStreamingRequestBytes, nativeLoaded); boolean response = VesperaNativeConfig.MAX_BUFFERED_RESPONSE_BYTES - .set(maxBufferedResponseBytes, loaded); + .set(maxBufferedResponseBytes, nativeLoaded); return request && streaming && response; } @@ -326,7 +346,7 @@ public static synchronized boolean configureLimits( * @return {@code false} when the value could not be applied (logged) */ public static synchronized boolean configureDispatchTimeout(long timeoutMillis) { - return VesperaNativeConfig.DISPATCH_TIMEOUT_MILLIS.set(timeoutMillis, loaded); + return VesperaNativeConfig.DISPATCH_TIMEOUT_MILLIS.set(timeoutMillis, nativeLoaded); } /** @@ -337,7 +357,7 @@ public static synchronized boolean configureDispatchTimeout(long timeoutMillis) * @return {@code false} when the value could not be applied (logged) */ public static synchronized boolean configureAsyncMaxInFlight(int maxInFlight) { - return VesperaNativeConfig.ASYNC_MAX_IN_FLIGHT.set(maxInFlight, loaded); + return VesperaNativeConfig.ASYNC_MAX_IN_FLIGHT.set(maxInFlight, nativeLoaded); } /** @@ -351,7 +371,7 @@ public static synchronized boolean configureAsyncMaxInFlight(int maxInFlight) { * @return {@code false} when the value could not be applied (logged) */ public static synchronized boolean configureMaxRequestProducers(int maxProducers) { - return VesperaNativeConfig.MAX_REQUEST_PRODUCERS.set(maxProducers, loaded); + return VesperaNativeConfig.MAX_REQUEST_PRODUCERS.set(maxProducers, nativeLoaded); } /** diff --git a/libs/vespera-bridge/src/main/java/com/devfive/vespera/bridge/VesperaBridgeAutoConfiguration.java b/libs/vespera-bridge/src/main/java/com/devfive/vespera/bridge/VesperaBridgeAutoConfiguration.java index ea905a6d..08294a1b 100644 --- a/libs/vespera-bridge/src/main/java/com/devfive/vespera/bridge/VesperaBridgeAutoConfiguration.java +++ b/libs/vespera-bridge/src/main/java/com/devfive/vespera/bridge/VesperaBridgeAutoConfiguration.java @@ -159,10 +159,12 @@ public DispatchModeResolver vesperaBridgeBidirectionalStreamingDispatchModeResol * *

The two trade-offs callers accept on the new default: *

    - *
  • DIRECT retries (re-runs the Rust handler) once when a - * response exceeds {@code vespera.direct.maxBufferBytes} - * (default 4 MiB). This is why DIRECT is restricted to safe - * methods (GET/HEAD/OPTIONS).
    • + *
  • A DIRECT response that overflows the pooled buffer is kept on + * the Rust side and fetched into a grown buffer, or streamed + * from there beyond {@code vespera.direct.maxBufferBytes} + * (default 4 MiB): the handler runs once. Only the rare + * stash-less fallback re-runs it, which is why DIRECT stays + * restricted to safe methods (GET/HEAD/OPTIONS).
    • *
  • SYNC buffers the full response on the JVM heap. The * 256 KiB request-size gate keeps the response size * reasonable for JSON-RPC-shaped traffic.
    • diff --git a/libs/vespera-bridge/src/main/java/com/devfive/vespera/bridge/VesperaBridgeProperties.java b/libs/vespera-bridge/src/main/java/com/devfive/vespera/bridge/VesperaBridgeProperties.java index 8c46422d..38f9f674 100644 --- a/libs/vespera-bridge/src/main/java/com/devfive/vespera/bridge/VesperaBridgeProperties.java +++ b/libs/vespera-bridge/src/main/java/com/devfive/vespera/bridge/VesperaBridgeProperties.java @@ -113,10 +113,13 @@ public class VesperaBridgeProperties { * falls back to bidirectional streaming. Measured 2.2 µs * (DIRECT) / 3.2 µs (SYNC) vs 24.1 µs (bidirectional) on * a small {@code GET /health} round-trip. Trade-offs: - * DIRECT re-runs the handler when a response overflows the - * pooled buffer ({@code vespera.direct.maxBufferBytes}, - * default 4 MiB) — acceptable for safe requests - * only; SYNC fully buffers the response on the JVM heap. + * a DIRECT response that overflows the pooled buffer + * ({@code vespera.direct.maxBufferBytes}, default 4 MiB) is + * kept on the Rust side and fetched into a grown buffer, or + * streamed from there beyond the cap; the handler runs once + * (only the rare stash-less fallback re-runs it, which is why + * DIRECT stays limited to safe requests); SYNC fully buffers + * the response on the JVM heap. *
  • {@code bidirectional-streaming} — opt-out, restores the * pre-0.2.0 default: every request that may carry a body * streams both ways; safe for any payload size; the @@ -127,10 +130,13 @@ public class VesperaBridgeProperties { private String dispatchMode = "smart"; /** - * Whether the Spring proxy may retry a DIRECT response-buffer overflow - * for safe methods. Default {@code true} preserves the 0.2.x - * behavior (grow the direct response buffer once and re-run the Rust - * handler). Set {@code false} to surface + * Whether the Spring proxy may re-dispatch a DIRECT response-buffer + * overflow for safe methods when the overflow stash cannot be used. + * Normally an overflowing response is kept on the Rust side and + * fetched into a grown buffer (or streamed from there beyond the + * pool cap) without re-running the handler, so this only matters for + * that rare stash-less fallback: {@code true} (default) re-runs the + * Rust handler through response streaming; {@code false} surfaces * {@link VesperaBridge.BufferTooSmallException} as a 500 instead, * avoiding any automatic double execution. */ diff --git a/libs/vespera-bridge/src/main/java/com/devfive/vespera/bridge/VesperaDirectBufferPool.java b/libs/vespera-bridge/src/main/java/com/devfive/vespera/bridge/VesperaDirectBufferPool.java index 93d21de7..79e5fc17 100644 --- a/libs/vespera-bridge/src/main/java/com/devfive/vespera/bridge/VesperaDirectBufferPool.java +++ b/libs/vespera-bridge/src/main/java/com/devfive/vespera/bridge/VesperaDirectBufferPool.java @@ -102,6 +102,15 @@ interface HeapDispatcher { byte[] dispatch(byte[] wireRequest); } + /** + * The buffered fallback for a request that cannot use the pool: on a + * virtual thread it parks ({@link ParkingDispatch#park}) instead of + * pinning the carrier inside {@code dispatchBytes}. + */ + static HeapDispatcher heapDispatcher(boolean currentThreadIsVirtual) { + return currentThreadIsVirtual ? ParkingDispatch::park : VesperaBridge::dispatchBytes; + } + private static OversizePolicy oversizePolicy() { String configured = System.getProperty("vespera.direct.oversize-policy", "heap-fallback"); if ("throw".equalsIgnoreCase(configured)) { @@ -359,7 +368,7 @@ static ByteBuffer dispatchDirectPooled(byte[] wireRequest, boolean retryOnOverfl static ByteBuffer dispatchDirectPooled( byte[] wireRequest, boolean retryOnOverflow, boolean currentThreadIsVirtual) { return dispatchDirectPooled( - wireRequest, retryOnOverflow, currentThreadIsVirtual, VesperaBridge::dispatchBytes); + wireRequest, retryOnOverflow, currentThreadIsVirtual, heapDispatcher(currentThreadIsVirtual)); } static ByteBuffer dispatchDirectPooled( @@ -402,7 +411,7 @@ static ByteBuffer dispatchDirectPooled( return dispatchDirectPooled( appName, method, path, query, headers, body, retryOnOverflow, VesperaDirectBufferPool::currentThreadIsVirtual, - VesperaBridge::dispatchBytes); + wire -> ParkingDispatch.dispatchBytes(wire, currentThreadIsVirtual())); } static ByteBuffer dispatchDirectPooled( @@ -470,7 +479,7 @@ static ByteBuffer dispatchDirectPooled( boolean currentThreadIsVirtual) { return dispatchDirectPooled( appName, method, path, query, headers, body, - retryOnOverflow, currentThreadIsVirtual, VesperaBridge::dispatchBytes); + retryOnOverflow, currentThreadIsVirtual, heapDispatcher(currentThreadIsVirtual)); } static ByteBuffer dispatchDirectPooled( diff --git a/libs/vespera-bridge/src/main/java/com/devfive/vespera/bridge/VesperaProxyController.java b/libs/vespera-bridge/src/main/java/com/devfive/vespera/bridge/VesperaProxyController.java index 205eeeae..83035ba1 100644 --- a/libs/vespera-bridge/src/main/java/com/devfive/vespera/bridge/VesperaProxyController.java +++ b/libs/vespera-bridge/src/main/java/com/devfive/vespera/bridge/VesperaProxyController.java @@ -303,6 +303,14 @@ static boolean isValidRequestId(String value) { private Object dispatch(HttpServletRequest request, ProxyServletResponse response, ProxyExchange exchange) throws IOException { final RequestShape shape = RequestShape.capture(request); + // Path RELATIVE to the servlet context: a Spring app deployed under + // a non-root context (e.g. server.servlet.context-path=/api) must + // still forward `/health` — not `/api/health` — so the Rust router + // sees exactly the URL published in the generated openapi.json. + // Recorded before the app is resolved: an unknown-app 404 still + // reports its path (DispatchEvent.path is never null). + final String path = pathWithinApplication(request); + exchange.path = path; final String appName = resolveAppName(request, exchange); final Boolean currentThreadIsVirtual; final DispatchMode mode; @@ -319,18 +327,12 @@ private Object dispatch(HttpServletRequest request, ProxyServletResponse respons mode = modeResolver.resolveMode(request); } final String method = shape.method; - // Path RELATIVE to the servlet context: a Spring app deployed under - // a non-root context (e.g. server.servlet.context-path=/api) must - // still forward `/health` — not `/api/health` — so the Rust router - // sees exactly the URL published in the generated openapi.json. - final String path = pathWithinApplication(request); exchange.app = appName; - exchange.path = path; final String query = Objects.toString(request.getQueryString(), ""); // Adaptive DIRECT-overflow avoidance: a route that overflowed the - // pooled direct buffer before is streamed up front, so it dispatches - // ONCE instead of paying the DIRECT-overflow-then-stream double - // dispatch again. `shouldAvoidDirect` is a single volatile read until + // pooled direct buffer before is streamed up front instead of paying + // the overflow (whole response stashed natively, then streamed) + // again. `shouldAvoidDirect` is a single volatile read until // the first overflow is recorded, so non-overflowing apps pay nothing. final DispatchMode effectiveMode = formSafeMode(effectiveMode( mode, @@ -389,7 +391,7 @@ private Object dispatch(HttpServletRequest request, ProxyServletResponse respons case DIRECT: default: dispatchDirectMode(response, appName, method, path, query, headers, body, - currentThreadIsVirtual); + currentThreadIsVirtual, exchange); return null; } } @@ -911,23 +913,23 @@ static boolean detachRequestBody( return true; } - /** - * {@link DispatchMode#DIRECT} boundary: a safe method goes to - * {@link DirectModeDispatcher}; every other method is served via - * {@code SYNC}, because a DIRECT overflow retry re-runs the handler - * (e.g. DELETE → 204 then 404), so the controller never double-executes - * a handler whose response could change. - */ /** Whether a recorded DIRECT overflow makes this route stream up front (tests). */ boolean avoidsDirect(String appName, String method, String path, String query) { return direct.shouldAvoidDirect(appName, method, path, query); } + /** + * {@link DispatchMode#DIRECT} boundary: a safe method goes to + * {@link DirectModeDispatcher}; every other method is served via + * {@code SYNC} (recorded on {@code exchange}, which may be {@code null}), + * because a DIRECT overflow that cannot use the stash re-runs the handler + * (e.g. DELETE → 204 then 404). + */ void dispatchDirectMode( HttpServletResponse response, String appName, String method, String path, String query, VesperaBridge.HeaderSource headers, byte[] body, - Boolean currentThreadIsVirtual) throws IOException { + Boolean currentThreadIsVirtual, ProxyExchange exchange) throws IOException { if (!HttpMethods.isSafe(method)) { // DIRECT runs the Rust handler on the FIRST dispatch before any // overflow is known; for an UNSAFE method an overflow would 500 @@ -949,12 +951,13 @@ void dispatchDirectMode( log.debug("DispatchModeResolver routed unsafe method {} to DIRECT; " + "downgrading to SYNC.", method); } + ProxyExchange.servedAs(exchange, DispatchMode.SYNC); dispatchSync(response, appName, method, path, query, headers, body, maxBufferedResponseBytes); return; } direct.dispatch(response, appName, method, path, query, headers, body, - currentThreadIsVirtual); + currentThreadIsVirtual, exchange); } /** diff --git a/libs/vespera-bridge/src/main/java/com/devfive/vespera/bridge/VesperaRequest.java b/libs/vespera-bridge/src/main/java/com/devfive/vespera/bridge/VesperaRequest.java index f9dac0c1..cf15a9c2 100644 --- a/libs/vespera-bridge/src/main/java/com/devfive/vespera/bridge/VesperaRequest.java +++ b/libs/vespera-bridge/src/main/java/com/devfive/vespera/bridge/VesperaRequest.java @@ -234,8 +234,12 @@ public VesperaRequest body(byte[] bytes, @Nullable String contentType) { * Streamed request body: pulled chunk-by-chunk by Rust through the * bidirectional streaming path (no full copy in memory; capped by * {@code vespera.limits.maxStreamingRequestBytes}). The response is - * still collected in memory. The stream is read but not closed, and can - * be sent once; {@link #sendAsync()} needs an {@link Executor} + * still collected in memory. The native side closes the stream once the + * dispatch is done with it (after the response drained, when the handler + * read from it); a read still blocked in it then fails with that close, + * which does not fail the send. Callers should not rely on the stream + * staying open, and it can be sent once; {@link #sendAsync()} needs an + * {@link Executor} * ({@link #sendAsync(Executor)}) because the call blocks. * * @param contentType sent as {@code content-type}; {@code null} leaves the header alone @@ -440,6 +444,14 @@ private void releaseUnreadStream(ReadTracking tracked) { static final class ReadTracking extends java.io.FilterInputStream { volatile boolean touched; volatile IOException readFailure; + /** + * Set by {@link #close()} BEFORE the delegate closes. The native side + * closes the source once the response drained (the handler will not + * read more); a read still parked on another thread then fails + * because of that close (e.g. {@code PipedInputStream}: "Pipe + * closed"), which is not a request-body failure. + */ + volatile boolean closed; ReadTracking(InputStream in) { super(in); @@ -468,7 +480,7 @@ public int read(byte[] b, int off, int len) throws IOException { } private void recordFailure(IOException e) { - if (readFailure == null) { + if (readFailure == null && !closed) { readFailure = e; } } @@ -482,6 +494,7 @@ public long skip(long n) throws IOException { @Override public void close() throws IOException { touched = true; + closed = true; super.close(); } } diff --git a/libs/vespera-bridge/src/main/java/com/devfive/vespera/bridge/VesperaVersions.java b/libs/vespera-bridge/src/main/java/com/devfive/vespera/bridge/VesperaVersions.java index f98f7e8b..61d7b911 100644 --- a/libs/vespera-bridge/src/main/java/com/devfive/vespera/bridge/VesperaVersions.java +++ b/libs/vespera-bridge/src/main/java/com/devfive/vespera/bridge/VesperaVersions.java @@ -23,8 +23,10 @@ * default {@code strict} mode), or only a WARNING with * {@code -Dvespera.native.versionCheck=warn}; {@code off} skips the * check;
    • - *
  • a native library that predates {@code nativeVersion0()} (≤ 0.5.x) or - * a bridge whose version is unknown: a WARNING, never a failure.
    • + *
  • a native library that predates {@code nativeVersion0()} (≤ 0.5.x): + * a WARNING, never a failure;
    • + *
  • a bridge whose own version is unknown (not embedded, e.g. an IDE + * build): nothing to compare, so the check is skipped silently.
    • *
* *

Package-private: an internal collaborator of {@link VesperaBridge}. diff --git a/libs/vespera-bridge/src/test/java/com/devfive/vespera/bridge/ConfigureStreamingTest.java b/libs/vespera-bridge/src/test/java/com/devfive/vespera/bridge/ConfigureStreamingTest.java index 5f22fe3a..527c839c 100644 --- a/libs/vespera-bridge/src/test/java/com/devfive/vespera/bridge/ConfigureStreamingTest.java +++ b/libs/vespera-bridge/src/test/java/com/devfive/vespera/bridge/ConfigureStreamingTest.java @@ -71,7 +71,7 @@ void nothingIsPendingUntilConfigured() { @Test void postInitMissingOptionalNativeHookDoesNotThrowRawLinkageError() throws Exception { - Field loadedField = VesperaBridge.class.getDeclaredField("loaded"); + Field loadedField = VesperaBridge.class.getDeclaredField("nativeLoaded"); Field nameField = VesperaBridge.class.getDeclaredField("loadedLibraryName"); loadedField.setAccessible(true); nameField.setAccessible(true); diff --git a/libs/vespera-bridge/src/test/java/com/devfive/vespera/bridge/ProxyOperabilityTest.java b/libs/vespera-bridge/src/test/java/com/devfive/vespera/bridge/ProxyOperabilityTest.java index d6aa6a92..cf025485 100644 --- a/libs/vespera-bridge/src/test/java/com/devfive/vespera/bridge/ProxyOperabilityTest.java +++ b/libs/vespera-bridge/src/test/java/com/devfive/vespera/bridge/ProxyOperabilityTest.java @@ -391,6 +391,7 @@ void unknownAppIsAPlainText404ObservedWithTheRequestId() throws Exception { DispatchEvent event = observer.events.get(0); assertEquals(AppTags.UNKNOWN, event.app(), "the rejected name is never reported"); assertNull(event.mode()); + assertEquals("/users", event.path(), "the path is known before the app is resolved"); assertEquals("GET", event.method()); assertEquals(404, event.status()); assertFalse(event.panicked()); diff --git a/libs/vespera-bridge/src/test/java/com/devfive/vespera/bridge/ProxyResolverAndCorsTest.java b/libs/vespera-bridge/src/test/java/com/devfive/vespera/bridge/ProxyResolverAndCorsTest.java index 0eb97d0c..0a1a2452 100644 --- a/libs/vespera-bridge/src/test/java/com/devfive/vespera/bridge/ProxyResolverAndCorsTest.java +++ b/libs/vespera-bridge/src/test/java/com/devfive/vespera/bridge/ProxyResolverAndCorsTest.java @@ -92,7 +92,7 @@ void oversizePolicyRejectionStreamsWithoutRecordingAnOverflow() { // 500 or remembering the route as an overflowing one. assertThrows(UnsatisfiedLinkError.class, () -> controller.dispatchDirectMode( response, null, "GET", "/health", "", sink -> { }, - VesperaWireCodec.EMPTY_BODY, Boolean.TRUE)); + VesperaWireCodec.EMPTY_BODY, Boolean.TRUE, null)); assertEquals(0, response.getContentAsByteArray().length, "no proxy-written 500"); assertFalse(controller.avoidsDirect(null, "GET", "/health", ""), "no overflow recorded"); } diff --git a/libs/vespera-bridge/src/test/java/com/devfive/vespera/bridge/VesperaBridgeInitTest.java b/libs/vespera-bridge/src/test/java/com/devfive/vespera/bridge/VesperaBridgeInitTest.java index 3d44ba7b..32e31ea5 100644 --- a/libs/vespera-bridge/src/test/java/com/devfive/vespera/bridge/VesperaBridgeInitTest.java +++ b/libs/vespera-bridge/src/test/java/com/devfive/vespera/bridge/VesperaBridgeInitTest.java @@ -26,7 +26,7 @@ class VesperaBridgeInitTest { @Test void reInitWithDifferentLibraryThrowsAndSameNameIsNoOp() throws Exception { - Field loadedField = VesperaBridge.class.getDeclaredField("loaded"); + Field loadedField = VesperaBridge.class.getDeclaredField("nativeLoaded"); Field nameField = VesperaBridge.class.getDeclaredField("loadedLibraryName"); loadedField.setAccessible(true); nameField.setAccessible(true); @@ -49,6 +49,44 @@ void reInitWithDifferentLibraryThrowsAndSameNameIsNoOp() throws Exception { } } + @Test + void isInitializedIsPublishedOnlyAfterThePendingConfigurationWasApplied() throws Exception { + Field loadedField = VesperaBridge.class.getDeclaredField("loaded"); + Field nativeField = VesperaBridge.class.getDeclaredField("nativeLoaded"); + Field nameField = VesperaBridge.class.getDeclaredField("loadedLibraryName"); + loadedField.setAccessible(true); + nativeField.setAccessible(true); + nameField.setAccessible(true); + boolean prevLoaded = loadedField.getBoolean(null); + boolean prevNative = nativeField.getBoolean(null); + Object prevName = nameField.get(null); + try { + AtomicInteger applied = new AtomicInteger(); + VesperaBridge.completeLoad("libA", () -> { + // A concurrent dispatcher gating on isInitialized() must not + // run before the pending configure* values reach Rust ... + assertFalse(VesperaBridge.isInitialized()); + // ... while configure* (and a re-init) already treat the + // library as loaded. + assertDoesNotThrow(() -> VesperaBridge.init("libA")); + applied.incrementAndGet(); + }); + assertEquals(1, applied.get()); + assertTrue(VesperaBridge.isInitialized()); + + // A failing configuration hook still marks the library loaded. + loadedField.setBoolean(null, false); + assertThrows(IllegalStateException.class, () -> VesperaBridge.completeLoad("libA", () -> { + throw new IllegalStateException("hook failed"); + })); + assertTrue(VesperaBridge.isInitialized()); + } finally { + loadedField.setBoolean(null, prevLoaded); + nativeField.setBoolean(null, prevNative); + nameField.set(null, prevName); + } + } + @Test void nativeLoadDecisionFallsBackOnlyForAbsentBundledResource() { AtomicInteger bundledCalls = new AtomicInteger(); diff --git a/libs/vespera-bridge/src/test/java/com/devfive/vespera/bridge/VesperaClientTest.java b/libs/vespera-bridge/src/test/java/com/devfive/vespera/bridge/VesperaClientTest.java index 43e330bf..8cef4507 100644 --- a/libs/vespera-bridge/src/test/java/com/devfive/vespera/bridge/VesperaClientTest.java +++ b/libs/vespera-bridge/src/test/java/com/devfive/vespera/bridge/VesperaClientTest.java @@ -23,7 +23,9 @@ import java.util.Map; import java.util.concurrent.CompletableFuture; import java.util.concurrent.CompletionException; +import java.util.concurrent.CountDownLatch; import java.util.concurrent.ExecutionException; +import java.util.concurrent.TimeUnit; import org.junit.jupiter.api.Test; /** {@link VesperaClient} / {@link VesperaRequest} / {@link VesperaResponse} over a fake transport. */ @@ -472,6 +474,80 @@ public int read() throws IOException { assertSame(broken, async.getCause().getCause()); } + @Test + void aParkedReadFailingBecauseTheNativeSideClosedTheSourceIsNotATransportFailure() + throws Exception { + CountDownLatch parked = new CountDownLatch(1); + CountDownLatch closed = new CountDownLatch(1); + // A socket-like source: once drained of its first byte, a read parks + // until the stream is closed, then fails with the close. + InputStream socketLike = new InputStream() { + private boolean first = true; + + @Override + public int read() throws IOException { + byte[] one = new byte[1]; + return read(one, 0, 1) < 0 ? -1 : one[0]; + } + + @Override + public int read(byte[] b, int off, int len) throws IOException { + if (first) { + first = false; + b[off] = 'x'; + return 1; + } + parked.countDown(); + try { + closed.await(); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + throw new IOException("Socket closed"); + } + + @Override + public void close() { + closed.countDown(); + } + }; + VesperaClient closing = new VesperaClient(new FakeTransport() { + @Override + public byte[] dispatchStreaming(byte[] header, InputStream in, OutputStream out) + throws IOException { + // The native producer: pulls on its own thread and parks in + // the second read. + Thread producer = new Thread(() -> { + byte[] chunk = new byte[16]; + try { + while (in.read(chunk, 0, chunk.length) >= 0) { + // keep pulling + } + } catch (IOException expected) { + // swallowed natively + } + }); + producer.start(); + try { + assertTrue(parked.await(10, TimeUnit.SECONDS)); + // The handler answered and the response drained: Rust + // closes the request source to unblock the producer. + in.close(); + producer.join(10_000); + } catch (InterruptedException e) { + throw new AssertionError(e); + } + assertFalse(producer.isAlive()); + return responseWire(200, "{}", null, "done"); + } + }, null); + + VesperaResponse response = closing.post("/s").body(socketLike, null).send(); + + assertEquals(200, response.status()); + assertEquals("done", response.bodyString()); + } + @Test void theDefaultAppNameReportsAsNull() { assertNull(client.forApp("_default").appName()); diff --git a/libs/vespera-bridge/src/test/java/com/devfive/vespera/bridge/VesperaDirectWrapperTest.java b/libs/vespera-bridge/src/test/java/com/devfive/vespera/bridge/VesperaDirectWrapperTest.java index 9a8f1886..d9bad894 100644 --- a/libs/vespera-bridge/src/test/java/com/devfive/vespera/bridge/VesperaDirectWrapperTest.java +++ b/libs/vespera-bridge/src/test/java/com/devfive/vespera/bridge/VesperaDirectWrapperTest.java @@ -294,13 +294,21 @@ void invalidOversizePolicyIsRejectedAndHeapFallbackReachesHeapNativeBoundary() { assertTrue(invalid.getMessage().contains("heap-fallback"), invalid.getMessage()); System.setProperty("vespera.direct.oversize-policy", "HeAp-FaLlBaCk"); - assertThrows(UnsatisfiedLinkError.class, - () -> VesperaDirectBufferPool.dispatchDirectPooled(new byte[1], false, true)); + // On a virtual thread the heap fallback parks (ParkingDispatch) + // instead of pinning the carrier in dispatchBytes. + assertParked(assertThrows(UnsatisfiedLinkError.class, + () -> VesperaDirectBufferPool.dispatchDirectPooled(new byte[1], false, true))); VesperaBridge.HeaderSource headers = sink -> sink.put("x-test", "yes"); - assertThrows(UnsatisfiedLinkError.class, + assertParked(assertThrows(UnsatisfiedLinkError.class, () -> VesperaDirectBufferPool.dispatchDirectPooled( - null, "GET", "/", null, headers, null, false, true)); + null, "GET", "/", null, headers, null, false, true))); + // A platform thread keeps the direct dispatchBytes call. + UnsatisfiedLinkError pinned = assertThrows(UnsatisfiedLinkError.class, + () -> VesperaDirectBufferPool.dispatchDirectPooled( + null, "GET", "/", null, headers, new byte[5 * 1024 * 1024], false, false)); + assertFalse(java.util.Arrays.stream(pinned.getStackTrace()) + .anyMatch(f -> f.getClassName().equals(ParkingDispatch.class.getName()))); } finally { restoreProperty("vespera.direct.oversize-policy", previous); } @@ -580,6 +588,13 @@ private static boolean throwChecked(Thread ignored) throws java.io.IOException { throw new java.io.IOException("checked"); } + /** The unlinked native call was reached through {@link ParkingDispatch} (no cdylib here). */ + private static void assertParked(UnsatisfiedLinkError unlinked) { + assertTrue(java.util.Arrays.stream(unlinked.getStackTrace()) + .anyMatch(f -> f.getClassName().equals(ParkingDispatch.class.getName())), + "heap fallback did not park: " + java.util.Arrays.toString(unlinked.getStackTrace())); + } + private static void restoreProperty(String name, String value) { if (value == null) { System.clearProperty(name); diff --git a/libs/vespera-bridge/src/test/java/com/devfive/vespera/bridge/VesperaProxyControllerCoverageTest.java b/libs/vespera-bridge/src/test/java/com/devfive/vespera/bridge/VesperaProxyControllerCoverageTest.java index 3c7cc9b8..d7e0a445 100644 --- a/libs/vespera-bridge/src/test/java/com/devfive/vespera/bridge/VesperaProxyControllerCoverageTest.java +++ b/libs/vespera-bridge/src/test/java/com/devfive/vespera/bridge/VesperaProxyControllerCoverageTest.java @@ -237,22 +237,76 @@ void repeatedUnsafeDirectDowngradeTakesDebugLoggingBranch() throws Exception { request -> null, request -> DispatchMode.DIRECT, Runnable::run, false); assertThrows(UnsatisfiedLinkError.class, () -> controller.dispatchDirectMode( new MockHttpServletResponse(), null, "POST", "/echo", "", - sink -> {}, new byte[] {1}, Boolean.FALSE)); + sink -> {}, new byte[] {1}, Boolean.FALSE, null)); assertThrows(UnsatisfiedLinkError.class, () -> controller.dispatchDirectMode( new MockHttpServletResponse(), null, "POST", "/echo", "", - sink -> {}, new byte[] {1}, Boolean.FALSE)); + sink -> {}, new byte[] {1}, Boolean.FALSE, null)); assertTrue(warned.get()); logger.setLevel(ch.qos.logback.classic.Level.INFO); assertThrows(UnsatisfiedLinkError.class, () -> controller.dispatchDirectMode( new MockHttpServletResponse(), null, "POST", "/echo", "", - sink -> {}, new byte[] {1}, Boolean.FALSE)); + sink -> {}, new byte[] {1}, Boolean.FALSE, null)); + + // With an exchange the downgrade is what gets reported. + ProxyExchange exchange = new ProxyExchange( + org.slf4j.LoggerFactory.getLogger(VesperaProxyControllerCoverageTest.class), + java.util.List.of(), "POST", null); + exchange.mode = DispatchMode.DIRECT; + assertThrows(UnsatisfiedLinkError.class, () -> controller.dispatchDirectMode( + new MockHttpServletResponse(), null, "POST", "/echo", "", + sink -> {}, new byte[] {1}, Boolean.FALSE, exchange)); + assertEquals(DispatchMode.SYNC, exchange.mode); } finally { logger.setLevel(previousLevel); warned.set(previousWarned); } } + @Test + void aDirectRequestServedThroughAnotherPathReportsThatMode() throws Exception { + Field loaded = VesperaBridge.class.getDeclaredField("loaded"); + loaded.setAccessible(true); + boolean wasLoaded = loaded.getBoolean(null); + previousOversizePolicy = System.getProperty("vespera.direct.oversize-policy"); + java.util.List events = new java.util.concurrent.CopyOnWriteArrayList<>(); + VesperaProxyController controller = new VesperaProxyController( + request -> null, request -> DispatchMode.DIRECT, Runnable::run, true, 0, 0, null, + java.util.List.of(events::add)); + try { + // Past the isInitialized() gate; every native call then fails + // with UnsatisfiedLinkError (answered 503) after the mode is final. + loaded.setBoolean(null, true); + + // An unsafe method is downgraded to SYNC. + MockHttpServletRequest post = new MockHttpServletRequest("POST", "/echo"); + post.setContent(new byte[] {1}); + MockHttpServletResponse posted = new MockHttpServletResponse(); + controller.proxy(post, posted); + assertEquals(503, posted.getStatus()); + assertEquals(DispatchMode.SYNC, events.get(0).mode()); + assertEquals("/echo", events.get(0).path()); + + // oversize-policy=throw: a request beyond the pool cap is re-routed + // through response streaming before anything is dispatched. + System.setProperty("vespera.direct.oversize-policy", "throw"); + MockHttpServletRequest big = new MockHttpServletRequest("GET", "/big"); + big.setContent(new byte[4 * 1024 * 1024 + 1]); + MockHttpServletResponse bigResponse = new MockHttpServletResponse(); + controller.proxy(big, bigResponse); + assertEquals(503, bigResponse.getStatus()); + assertEquals(DispatchMode.STREAMING, events.get(1).mode()); + + // A plain safe DIRECT request keeps reporting DIRECT. + MockHttpServletResponse got = new MockHttpServletResponse(); + controller.proxy(new MockHttpServletRequest("GET", "/small"), got); + assertEquals(503, got.getStatus()); + assertEquals(DispatchMode.DIRECT, events.get(2).mode()); + } finally { + loaded.setBoolean(null, wasLoaded); + } + } + @Test void cyclicAsyncFailureTerminatesCauseScanAndPrewrappedFailureIsPreserved() { RuntimeException cyclic = new RuntimeException("cycle") { From d1d5051079c5fb6c7b8b54a227dbd00c512eaab9 Mon Sep 17 00:00:00 2001 From: Jeong Min Oh Date: Sun, 15 Nov 2026 09:58:05 +0900 Subject: [PATCH 2/2] test(demo): pass the exchange to dispatchDirectMode Co-Authored-By: Claude Opus 5.5 --- .../bridge/VesperaProxyControllerOverflowIntegrationTest.java | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/examples/rust-jni-demo/java/demo-app/src/test/java/com/devfive/vespera/bridge/VesperaProxyControllerOverflowIntegrationTest.java b/examples/rust-jni-demo/java/demo-app/src/test/java/com/devfive/vespera/bridge/VesperaProxyControllerOverflowIntegrationTest.java index 888f4b82..3ceb8a17 100644 --- a/examples/rust-jni-demo/java/demo-app/src/test/java/com/devfive/vespera/bridge/VesperaProxyControllerOverflowIntegrationTest.java +++ b/examples/rust-jni-demo/java/demo-app/src/test/java/com/devfive/vespera/bridge/VesperaProxyControllerOverflowIntegrationTest.java @@ -79,7 +79,7 @@ void rejectedPooledFallbackStreamsOnceAndIsNotRememberedAsAnOverflow(boolean ret MockHttpServletResponse response = new MockHttpServletResponse(); controller.dispatchDirectMode( - response, null, "GET", path, "", sink -> {}, VesperaWireCodec.EMPTY_BODY, Boolean.TRUE); + response, null, "GET", path, "", sink -> {}, VesperaWireCodec.EMPTY_BODY, Boolean.TRUE, null); assertEquals(200, response.getStatus()); assertEquals("x".repeat(16), response.getContentAsString(StandardCharsets.UTF_8)); @@ -101,7 +101,7 @@ void realOverflowStreamsFromTheStashAndTheRouteAvoidsDirectNextTime() throws Exc MockHttpServletResponse overflow = new MockHttpServletResponse(); controller.dispatchDirectMode( - overflow, null, "GET", path, "", sink -> {}, VesperaWireCodec.EMPTY_BODY, Boolean.FALSE); + overflow, null, "GET", path, "", sink -> {}, VesperaWireCodec.EMPTY_BODY, Boolean.FALSE, null); assertEquals(200, overflow.getStatus()); assertEquals(BEYOND_POOL_CAP, overflow.getContentAsByteArray().length);