Improve

owjs3901 · owjs3901 · commit 090530d7ee88 · 2026-06-21T03:58:03.000+09:00
diff --git a/crates/vespera_inprocess/tests/wire_contract.rs b/crates/vespera_inprocess/tests/wire_contract.rs
@@ -14,7 +14,7 @@ use std::sync::Once;
 use axum::Router;
 use axum::http::{HeaderMap, HeaderName};
 use axum::response::{IntoResponse, Response};
-use axum::routing::get;
+use axum::routing::{get, post};
 use serde_json::Value;
 use tokio::runtime::Builder;
 use vespera_inprocess::{dispatch_from_bytes, error_wire, register_app};
@@ -31,10 +31,21 @@ async fn contract_headers() -> Response {
     (headers, "ok").into_response()
 }
 
+/// Echo the raw request body back — used by the cross-language golden
+/// test so a matched `POST /users` proves the header/body split + routing
+/// on the exact bytes the Java encoder produces.
+async fn echo_body(body: axum::body::Bytes) -> axum::body::Bytes {
+    body
+}
+
 fn install_router() {
     static INIT: Once = Once::new();
     INIT.call_once(|| {
-        register_app(|| Router::new().route("/contract", get(contract_headers)));
+        register_app(|| {
+            Router::new()
+                .route("/contract", get(contract_headers))
+                .route("/users", post(echo_body))
+        });
     });
 }
 
@@ -133,6 +144,51 @@ fn error_wire_bytes_are_locked() {
     );
 }
 
+/// **Cross-language golden (request direction)** — dispatches the
+/// byte-identical wire frame the Java encoder produces and asserts the
+/// Rust parser accepts it and routes correctly.
+///
+/// The header JSON + body below are byte-identical to the Java side's
+/// shared golden (`VesperaWireTest.CANONICAL_REQUEST_HEADER_JSON` /
+/// `CANONICAL_REQUEST_BODY`).  Java asserts its encoder emits exactly
+/// these bytes; this test asserts Rust parses exactly these bytes and
+/// routes `POST /users` with the body intact.  Together they lock the two
+/// independent hand-rolled wire implementations against silent drift: a
+/// change to either side's field order / structure / framing breaks its
+/// own golden assertion.
+#[test]
+fn cross_language_request_golden_routes() {
+    install_router();
+    let runtime = Builder::new_current_thread()
+        .enable_all()
+        .build()
+        .expect("tokio runtime");
+
+    // Byte-identical to the Java cross-language golden — do NOT edit one
+    // side without the other (see VesperaWireTest).
+    let header_json =
+        br#"{"v":1,"method":"POST","path":"/users","query":"page=1","headers":{"content-type":"application/json"}}"#;
+    let body = br#"{"x":1}"#;
+    let mut wire = Vec::with_capacity(4 + header_json.len() + body.len());
+    wire.extend_from_slice(&u32::try_from(header_json.len()).unwrap().to_be_bytes());
+    wire.extend_from_slice(header_json);
+    wire.extend_from_slice(body);
+
+    let resp = dispatch_from_bytes(wire, &runtime);
+    let (header, resp_body) = split_wire(&resp);
+
+    // Body round-trip proves the parser split header/body at the exact
+    // offset and routed the echo handler; 200 proves `POST /users` matched.
+    assert_eq!(
+        resp_body, body,
+        "cross-language request golden: echo body must round-trip (header/body split + routing)"
+    );
+    assert!(
+        header.contains(r#""status":200"#),
+        "cross-language request golden: POST /users must route 200 — got: {header}"
+    );
+}
+
 /// Golden: 422 hoisting shape — `validation_errors` appears as the
 /// LAST field, after `metadata`, with `path`/`message` entry order.
 #[test]
diff --git a/libs/vespera-bridge/README.md b/libs/vespera-bridge/README.md
@@ -394,6 +394,45 @@ end-to-end:
 `DispatchMode.BIDIRECTIONAL_STREAMING` is safe for virtual threads
 and handles all payload sizes without pooling.
 
+### Synchronous dispatch runtime semantics (background tasks)
+
+`SYNC` and `DIRECT` drive each request on a **per-calling-thread
+current-thread Tokio runtime** (one runtime per Java request thread,
+created lazily, blocking pool capped at 4 threads). The request future and
+any task the handler `.await`s run to completion normally. The one caveat:
+a handler that **detaches** background work with `tokio::spawn(...)` and
+returns *without awaiting it* makes progress on that detached task only
+while a later request reuses the same Java thread's runtime in a
+`block_on` — there are no dedicated worker threads on this path.
+
+If your handlers fire-and-forget background tasks, route them through a
+mode backed by the shared multi-threaded runtime instead:
+
+- `dispatchAsync` / the `CompletableFuture` API, or
+- `vespera.bridge.dispatch-mode=bidirectional-streaming`
+
+(both use the shared `RUNTIME`, whose worker count is tunable via
+`vespera.runtime.workerThreads` / `VESPERA_RUNTIME_WORKERS`). Handlers that
+only `.await` their own work are unaffected.
+
+### Operational sizing (threads & off-heap)
+
+The `DIRECT` fast path keeps per-platform-thread pooled direct
+`ByteBuffer`s (64 KiB → `vespera.direct.maxBufferBytes`, default 4 MiB),
+and `SYNC`/`DIRECT` keep one current-thread runtime per Java request
+thread. On a large servlet pool (e.g. 200 Tomcat threads) the steady-state
+cost is therefore bounded but not negligible:
+
+- **Off-heap**: up to `poolThreads × maxBufferBytes` retained direct
+  memory once each thread has served a large response (adaptive shrink
+  reclaims it after idle dispatches). Lower `vespera.direct.maxBufferBytes`
+  if off-heap pressure matters more than the DIRECT copy savings.
+- **Threads**: worst case `poolThreads × 4` Tokio blocking threads if many
+  handlers use `spawn_blocking` concurrently (the multipart temp-file
+  extractor does). Cap Rust's shared async runtime with
+  `vespera.runtime.workerThreads` when the JVM's own pools compete for the
+  same cores, or when a container CPU limit is below the host CPU count.
+
 ## Direct API (without the proxy controller)
 
 For custom integrations bypassing Spring:
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
@@ -329,18 +329,7 @@ private static void dispatchSync(
      */
     private static void writeWireResponse(byte[] wire, HttpServletResponse response)
             throws IOException {
-        if (wire == null || wire.length < 4) {
-            throw new IllegalArgumentException(
-                    "wire response too short: "
-                            + (wire == null ? "null" : wire.length + " bytes"));
-        }
-        int headerLen = ((wire[0] & 0xFF) << 24) | ((wire[1] & 0xFF) << 16)
-                | ((wire[2] & 0xFF) << 8) | (wire[3] & 0xFF);
-        if (headerLen < 0 || (long) 4 + headerLen > wire.length) {
-            throw new IllegalArgumentException(
-                    "wire header_len " + headerLen
-                            + " overflows response (" + wire.length + " bytes)");
-        }
+        int headerLen = VesperaWireCodec.readHeaderLength(wire);
         int[] statusHolder = {500};
         WireHeaderReader.apply(
                 ByteBuffer.wrap(wire), 4, headerLen,
@@ -517,25 +506,13 @@ private void dispatchDirectMode(
      * already apply.
      */
     static int readValidatedHeaderLen(ByteBuffer wire) {
-        int limit = wire.limit();
-        if (limit < 4) {
-            throw new IllegalArgumentException("wire response too short: " + limit + " bytes");
-        }
-        // Decode the u32 BE length prefix from absolute bytes (order-independent)
-        // instead of wire.getInt(0), which honours the buffer's CURRENT byte
-        // order — a LITTLE_ENDIAN view (e.g. a caller that called order(...) on
-        // the buffer, or a future change) would misparse the big-endian wire
-        // prefix. Matches the manual big-endian decode the heap byte[] paths
-        // (writeWireResponse / buildResponseEntityFromWire) already use.
-        int headerLen = ((wire.get(0) & 0xFF) << 24)
-                | ((wire.get(1) & 0xFF) << 16)
-                | ((wire.get(2) & 0xFF) << 8)
-                | (wire.get(3) & 0xFF);
-        if (headerLen < 0 || (long) 4 + headerLen > limit) {
-            throw new IllegalArgumentException(
-                    "wire header_len " + headerLen + " overflows response (" + limit + " bytes)");
-        }
-        return headerLen;
+        // Delegates to the single source of truth in VesperaWireCodec so the
+        // u32 BE prefix decode + bounds contract stays byte-identical across
+        // every wire-frame split site (heap byte[] and direct ByteBuffer). The
+        // helper decodes from absolute bytes (order-independent) — never
+        // wire.getInt(0), which honours the buffer's CURRENT byte order — so a
+        // LITTLE_ENDIAN view can never misparse the big-endian wire prefix.
+        return VesperaWireCodec.readHeaderLength(wire);
     }
 
     // Package-private so tests can verify DIRECT header/body-length behavior
@@ -805,16 +782,7 @@ private static void applyDecodedHeader(byte[] headerBytes,
      * response executor instead of the native completion thread.
      */
     private static ResponseEntity<?> buildResponseEntityFromWire(byte[] wire) {
-        if (wire == null || wire.length < 4) {
-            throw new IllegalArgumentException(
-                    "wire response too short: " + (wire == null ? "null" : wire.length + " bytes"));
-        }
-        int headerLen = ((wire[0] & 0xFF) << 24) | ((wire[1] & 0xFF) << 16)
-                | ((wire[2] & 0xFF) << 8) | (wire[3] & 0xFF);
-        if (headerLen < 0 || (long) 4 + headerLen > wire.length) {
-            throw new IllegalArgumentException(
-                    "wire header_len " + headerLen + " overflows response (" + wire.length + " bytes)");
-        }
+        int headerLen = VesperaWireCodec.readHeaderLength(wire);
         HttpHeaders httpHeaders = new HttpHeaders();
         int[] statusHolder = {500};
         WireHeaderReader.apply(
diff --git a/libs/vespera-bridge/src/main/java/com/devfive/vespera/bridge/VesperaWireCodec.java b/libs/vespera-bridge/src/main/java/com/devfive/vespera/bridge/VesperaWireCodec.java
@@ -251,6 +251,63 @@ static int wireTotalLength(int headerLen, int bodyLen) {
         return (int) total;
     }
 
+    /**
+     * Decode and validate the u32 BE header-length prefix at bytes {@code 0..4}
+     * of a heap wire frame — the <strong>single source of truth</strong> for
+     * the frame split shared by {@link #decodeResponse} and the
+     * {@link VesperaProxyController} write/build paths, so the bounds contract
+     * can never drift between the (previously duplicated) call sites.
+     *
+     * <p>The prefix is read from absolute bytes (big-endian, order-independent),
+     * never {@code ByteBuffer.getInt} which honours the buffer's current byte
+     * order.
+     *
+     * @return the header JSON length {@code N} (so the body is {@code wire[4+N..]})
+     * @throws IllegalArgumentException if {@code wire} is shorter than the
+     *     4-byte prefix, or the decoded length is negative or overflows the frame
+     */
+    static int readHeaderLength(byte[] wire) {
+        if (wire == null || wire.length < 4) {
+            throw new IllegalArgumentException(
+                    "wire response too short: "
+                            + (wire == null ? "null" : wire.length + " bytes"));
+        }
+        int headerLen = ((wire[0] & 0xFF) << 24) | ((wire[1] & 0xFF) << 16)
+                | ((wire[2] & 0xFF) << 8) | (wire[3] & 0xFF);
+        if (headerLen < 0 || 4L + headerLen > wire.length) {
+            throw new IllegalArgumentException(
+                    "wire header_len " + headerLen
+                            + " overflows response (" + wire.length + " bytes)");
+        }
+        return headerLen;
+    }
+
+    /**
+     * {@link ByteBuffer} sibling of {@link #readHeaderLength(byte[])} — decodes
+     * the u32 BE header-length prefix from absolute bytes {@code 0..4} of
+     * {@code wire} (honouring neither the buffer's position nor its byte order),
+     * validating against {@code wire.limit()}.
+     *
+     * @return the header JSON length {@code N}
+     * @throws IllegalArgumentException if the buffer is shorter than the 4-byte
+     *     prefix, or the decoded length is negative or overflows the limit
+     */
+    static int readHeaderLength(ByteBuffer wire) {
+        int limit = wire.limit();
+        if (limit < 4) {
+            throw new IllegalArgumentException("wire response too short: " + limit + " bytes");
+        }
+        int headerLen = ((wire.get(0) & 0xFF) << 24)
+                | ((wire.get(1) & 0xFF) << 16)
+                | ((wire.get(2) & 0xFF) << 8)
+                | (wire.get(3) & 0xFF);
+        if (headerLen < 0 || 4L + headerLen > limit) {
+            throw new IllegalArgumentException(
+                    "wire header_len " + headerLen + " overflows response (" + limit + " bytes)");
+        }
+        return headerLen;
+    }
+
     /** Internal: write {@code [u32 BE len | headerJson[0..headerLen] | body]} at position 0. */
     static int assembleInto(byte[] headerJson, int headerLen, byte[] body, ByteBuffer target) {
         int total = wireTotalLength(headerLen, body.length);
@@ -499,18 +556,7 @@ private static void writeJsonString(ExposedByteArrayOutputStream out, String s)
      * @throws IllegalArgumentException if the wire bytes are malformed
      */
     static DecodedResponse decodeResponse(byte[] wire) {
-        if (wire == null || wire.length < 4) {
-            throw new IllegalArgumentException(
-                    "wire response too short: "
-                            + (wire == null ? "null" : wire.length + " bytes"));
-        }
-        int headerLen = ((wire[0] & 0xFF) << 24) | ((wire[1] & 0xFF) << 16)
-                | ((wire[2] & 0xFF) << 8) | (wire[3] & 0xFF);
-        if (headerLen < 0 || (long) 4 + headerLen > wire.length) {
-            throw new IllegalArgumentException(
-                    "wire header_len " + headerLen
-                            + " overflows response (" + wire.length + " bytes)");
-        }
+        int headerLen = readHeaderLength(wire);
         // Manual decode via the allocation-lean WireHeaderReader tokenizer
         // (the same parser the DIRECT / streaming header callbacks use)
         // instead of a Jackson JsonParser — drops the per-response parser +
diff --git a/libs/vespera-bridge/src/test/java/com/devfive/vespera/bridge/VesperaWireTest.java b/libs/vespera-bridge/src/test/java/com/devfive/vespera/bridge/VesperaWireTest.java
@@ -92,6 +92,56 @@ void encodeRequest_includes_query_and_headers_when_present() throws Exception {
         assertEquals("{\"x\":1}", new String(body, StandardCharsets.UTF_8));
     }
 
+    /**
+     * Canonical request header JSON — the <strong>shared cross-language
+     * golden</strong> that locks the Java encoder against the Rust
+     * {@code serde_json}/hand-rolled parser. The Rust counterpart
+     * ({@code crates/vespera_inprocess/tests/wire_contract.rs ::
+     * cross_language_request_golden_routes}) dispatches the byte-identical
+     * frame and asserts it routes, so the two independent hand-rolled wire
+     * implementations cannot silently drift: a change to either side's field
+     * order / escaping / structure breaks its own golden assertion.
+     *
+     * <p>Field order is fixed by {@code VesperaWireCodec.fillHeaderJson}:
+     * {@code v, method, path, query?, headers?, app?}.
+     */
+    static final String CANONICAL_REQUEST_HEADER_JSON =
+            "{\"v\":1,\"method\":\"POST\",\"path\":\"/users\",\"query\":\"page=1\","
+                    + "\"headers\":{\"content-type\":\"application/json\"}}";
+
+    /** Canonical request body paired with {@link #CANONICAL_REQUEST_HEADER_JSON}. */
+    static final byte[] CANONICAL_REQUEST_BODY = "{\"x\":1}".getBytes(StandardCharsets.UTF_8);
+
+    @Test
+    void crossLanguage_request_golden_bytes_are_locked() {
+        Map<String, String> headers = new LinkedHashMap<>();
+        headers.put("content-type", "application/json");
+
+        byte[] wire = VesperaBridge.encodeRequest(
+                "POST", "/users", "page=1", headers, CANONICAL_REQUEST_BODY);
+
+        byte[] expectedHeader =
+                CANONICAL_REQUEST_HEADER_JSON.getBytes(StandardCharsets.UTF_8);
+
+        // Length prefix == exact canonical header byte length (big-endian).
+        int headerLen = ByteBuffer.wrap(wire).order(ByteOrder.BIG_ENDIAN).getInt();
+        assertEquals(expectedHeader.length, headerLen,
+                "encoded header length drifted from the cross-language golden");
+
+        // Header JSON bytes are byte-identical to the shared golden (locks the
+        // Java encoder's field order + structure the Rust parser is asserted
+        // to accept verbatim in wire_contract.rs).
+        byte[] headerJson = new byte[headerLen];
+        System.arraycopy(wire, 4, headerJson, 0, headerLen);
+        assertArrayEquals(expectedHeader, headerJson,
+                "request header JSON drifted from the cross-language golden — WIRE FORMAT BREAK");
+
+        // Body follows the header verbatim.
+        byte[] body = new byte[wire.length - 4 - headerLen];
+        System.arraycopy(wire, 4 + headerLen, body, 0, body.length);
+        assertArrayEquals(CANONICAL_REQUEST_BODY, body, "request body must follow header verbatim");
+    }
+
     @Test
     void encodeRequestRejectsNullMethodAndPathWithFieldName() {
         NullPointerException method = assertThrows(
