Fix header callback

Author: owjs3901

AGENTS.md

@@ -81,8 +81,9 @@ vespera/
 | `vespera_macro/src/openapi_generator.rs` | ~808 | OpenAPI doc assembly |
 | `vespera_macro/src/collector.rs` | ~707 | Filesystem route scanning |
 | `vespera_inprocess/src/lib.rs` | ~1184 | In-process dispatch + app factory + streaming + binary wire |
-| `vespera_jni/src/jni_impl.rs` | ~833 | JNI RUNTIME + jni_app! macro + 7 JNI symbols (incl. direct-buffer path) |
-| `vespera_jni/src/streaming_closures.rs` | ~406 | Streaming closure factories (`make_pull_closure`, `make_push_closure`, `call_header_consumer`, `complete_future`) + `OnceLock<MethodCache>` caching `JMethodID`+`GlobalRef<JClass>` for `InputStream.read`, `OutputStream.write`, `Consumer.accept`, `CompletableFuture.complete` — `call_method_unchecked` on the hot path |
+| `vespera_jni/src/jni_impl.rs` | ~880 | JNI RUNTIME + jni_app! macro + 7 JNI symbols (incl. direct-buffer path) |
+| `vespera_jni/src/streaming_closures.rs` | ~410 | Streaming closure factories (`make_pull_closure`, `make_push_closure`, `call_header_consumer`, `complete_future`) + `OnceLock<MethodCache>` caching `JMethodID`+`GlobalRef<JClass>` for `InputStream.read`, `OutputStream.write`, `Consumer.accept`, `CompletableFuture.complete` — `call_method_unchecked` on the hot path. Pull/push/header closures attach via [`daemon_env::with_cached_daemon_env`] (TLS-cached daemon attach), not `attach_current_thread` per chunk |
+| `vespera_jni/src/daemon_env.rs` | ~130 | `with_cached_daemon_env(jvm, cb)` — attaches the current OS thread once as a daemon (`AttachCurrentThreadAsDaemon`), caches the `JNIEnv` in a `thread_local!` `Cell`, and reuses it for every JNI callback on that thread (streaming chunk pull/push, header callbacks, async `CompletableFuture.complete`). Replaces the prior per-chunk attach/detach churn; per-call local frame + exception scrub preserved |
 
 ## CRATE DEPENDENCY GRAPH
 

crates/vespera/src/multipart.rs

@@ -509,7 +509,10 @@ impl<S: Send + Sync> TryFromFieldWithState<S> for tempfile::NamedTempFile {
 
         let mut total = 0usize;
         while let Some(chunk) = field.chunk().await? {
-            total += chunk.len();
+            // `saturating_add` (matching `read_field_data`) prevents a
+            // pathological chunk size from wrapping `total` and slipping
+            // past the limit check below.
+            total = total.saturating_add(chunk.len());
             if let Some(limit) = limit_bytes
                 && total > limit
             {

crates/vespera_core/src/schema.rs

@@ -59,10 +59,21 @@ where
 {
     match value {
         Some(v) if v.fract() == 0.0 => {
-            // Practical OpenAPI constraints are well within i64 range
+            // Float→int casts saturate in Rust, so an out-of-range
+            // constraint (e.g. `1e20`) would silently become `i64::MAX`
+            // and corrupt the generated spec.  Emit the integer form
+            // only when it round-trips exactly back to the original
+            // value; otherwise keep the `f64` rendering.
             #[allow(clippy::cast_possible_truncation)]
             let int_val = *v as i64;
-            serializer.serialize_some(&int_val)
+            // Exact round-trip check is intentional: we emit the integer
+            // form only when `i64 → f64` reproduces the original bits.
+            #[allow(clippy::cast_precision_loss, clippy::float_cmp)]
+            if int_val as f64 == *v {
+                serializer.serialize_some(&int_val)
+            } else {
+                serializer.serialize_some(v)
+            }
         }
         Some(v) => serializer.serialize_some(v),
         None => serializer.serialize_none(),
@@ -503,6 +514,29 @@ mod tests {
         );
     }
 
+    #[test]
+    fn serialize_out_of_i64_range_constraint_stays_float() {
+        // A whole-number constraint beyond i64 range must NOT saturate to
+        // i64::MAX — it stays a float so the spec keeps the real value.
+        let schema = Schema {
+            maximum: Some(1e20),
+            ..Schema::number()
+        };
+        let json = serde_json::to_string(&schema).unwrap();
+        assert!(
+            !json.contains(&i64::MAX.to_string()),
+            "must not saturate to i64::MAX: {json}"
+        );
+        // Parse back: the constraint value must be preserved exactly,
+        // regardless of serde's float formatting.
+        let parsed: serde_json::Value = serde_json::from_str(&json).unwrap();
+        assert_eq!(
+            parsed["maximum"].as_f64(),
+            Some(1e20),
+            "constraint value must be preserved: {json}"
+        );
+    }
+
     #[test]
     fn serialize_multiple_of_whole_number_as_integer() {
         let schema = Schema {

crates/vespera_inprocess/benches/dispatch.rs

@@ -34,8 +34,8 @@ use criterion::{BenchmarkId, Criterion, Throughput, criterion_group, criterion_m
 use serde::{Deserialize, Serialize};
 use tokio::runtime::Runtime;
 use vespera_inprocess::{
-    RequestEnvelope, dispatch_bidirectional_streaming, dispatch_from_bytes, dispatch_owned,
-    dispatch_streaming_async, dispatch_typed, register_app,
+    RequestChunk, RequestEnvelope, dispatch_bidirectional_streaming, dispatch_from_bytes,
+    dispatch_owned, dispatch_streaming_async, dispatch_typed, register_app,
 };
 
 // ── Test fixtures ────────────────────────────────────────────────────
@@ -426,7 +426,13 @@ fn bench_streaming_path(c: &mut Criterion) {
             |b, _| {
                 b.iter(|| {
                     let chunks_iter = Mutex::new(request_chunks.clone().into_iter());
-                    let pull = move || -> Option<Vec<u8>> { chunks_iter.lock().unwrap().next() };
+                    let pull = move || -> RequestChunk {
+                        chunks_iter
+                            .lock()
+                            .unwrap()
+                            .next()
+                            .map_or(RequestChunk::End, RequestChunk::Data)
+                    };
                     let mut sink = 0usize;
                     runtime.block_on(dispatch_bidirectional_streaming(
                         header_only.clone(),
@@ -448,14 +454,14 @@ fn bench_streaming_path(c: &mut Criterion) {
             |b, _| {
                 b.iter(|| {
                     let remaining = Mutex::new(body_kb * 1024);
-                    let pull = move || -> Option<Vec<u8>> {
+                    let pull = move || -> RequestChunk {
                         let mut remaining = remaining.lock().unwrap();
                         if *remaining == 0 {
-                            return None;
+                            return RequestChunk::End;
                         }
                         let len = (*remaining).min(pull_chunk_size);
                         *remaining -= len;
-                        Some(vec![0xA5u8; len])
+                        RequestChunk::Data(vec![0xA5u8; len])
                     };
                     let mut sink = 0usize;
                     runtime.block_on(dispatch_bidirectional_streaming(

crates/vespera_inprocess/src/internal.rs

@@ -56,9 +56,14 @@ pub async fn dispatch_parts<'h>(
         builder = builder.header("content-type", "application/json");
     }
 
-    let request = builder
-        .body(Body::from(body_bytes))
-        .expect("request construction should not fail with valid URI");
+    // A malformed wire `path` (e.g. a raw space → not a valid
+    // `http::Uri`) or an invalid header name/value surfaces here as a
+    // builder error; convert it to a 400 so the contract "every failure
+    // returns a wire response" holds instead of panicking.
+    let request = match builder.body(Body::from(body_bytes)) {
+        Ok(req) => req,
+        Err(e) => return Err((400, format!("invalid request: {e}"))),
+    };
 
     let response = router
         .oneshot(request)
@@ -122,9 +127,14 @@ where
         builder = builder.header("content-type", "application/json");
     }
 
-    let request = builder
-        .body(Body::from(body_bytes))
-        .expect("request construction should not fail with valid URI");
+    // A malformed wire `path` (e.g. a raw space → not a valid
+    // `http::Uri`) or an invalid header name/value surfaces here as a
+    // builder error; convert it to a 400 so the contract "every failure
+    // returns a wire response" holds instead of panicking.
+    let request = match builder.body(Body::from(body_bytes)) {
+        Ok(req) => req,
+        Err(e) => return Err((400, format!("invalid request: {e}"))),
+    };
 
     let response = router
         .oneshot(request)
@@ -205,7 +215,12 @@ async fn collect_response_parts(response: axum::response::Response) -> ResponseP
 /// [`http::HeaderMap`].
 pub fn to_response_envelope_text(parts: ResponseParts) -> ResponseEnvelope {
     let (status, headers, body_bytes, metadata) = parts;
-    let body = String::from_utf8(body_bytes.to_vec()).unwrap_or_default();
+    // `Vec::from(Bytes)` reuses the underlying buffer when the `Bytes`
+    // is uniquely owned (the common case for a collected response body),
+    // copying only for a shared/static slice — unlike `to_vec()`, which
+    // always allocates and copies.  Semantics preserved: a non-UTF-8
+    // body still yields the empty string.
+    let body = String::from_utf8(Vec::from(body_bytes)).unwrap_or_default();
     ResponseEnvelope {
         status,
         headers: collect_header_map(&headers),
@@ -250,9 +265,12 @@ pub async fn dispatch_and_split<'h>(
         builder = builder.header("content-type", "application/json");
     }
 
-    let request = builder
-        .body(body)
-        .expect("request construction should not fail with valid URI");
+    // Same contract as dispatch_parts: a malformed path/header must
+    // surface as a 400 wire response, not a panic.
+    let request = match builder.body(body) {
+        Ok(req) => req,
+        Err(e) => return Err((400, format!("invalid request: {e}"))),
+    };
 
     let response = router
         .oneshot(request)
@@ -267,3 +285,84 @@ pub async fn dispatch_and_split<'h>(
         body,
     ))
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    fn block_on<F: std::future::Future>(fut: F) -> F::Output {
+        tokio::runtime::Builder::new_current_thread()
+            .enable_all()
+            .build()
+            .expect("build current-thread runtime")
+            .block_on(fut)
+    }
+
+    /// A wire `path` that cannot be parsed into an [`http::Uri`] (a raw
+    /// space is illegal) must surface as an `Err((4xx, _))` the caller
+    /// turns into a wire response — never a panic.  Guards the
+    /// "all failure modes return a valid wire response" contract for
+    /// every `request_builder` call site.
+    #[test]
+    fn malformed_path_returns_error_not_panic() {
+        let result = block_on(async {
+            dispatch_parts(
+                crate::Router::new(),
+                "GET",
+                "bad path with spaces",
+                "",
+                std::iter::empty(),
+                Bytes::new(),
+            )
+            .await
+        });
+        match result {
+            Err((status, _)) => assert!(
+                (400..500).contains(&status),
+                "expected 4xx for malformed path, got {status}"
+            ),
+            Ok(_) => panic!("malformed path should not produce a successful dispatch"),
+        }
+    }
+
+    #[test]
+    fn malformed_path_streaming_returns_error_not_panic() {
+        let result = block_on(async {
+            let mut sink = |_: &[u8]| {};
+            dispatch_response_streaming(
+                crate::Router::new(),
+                "GET",
+                "bad path with spaces",
+                "",
+                std::iter::empty(),
+                Bytes::new(),
+                &mut sink,
+            )
+            .await
+        });
+        assert!(
+            result.is_err(),
+            "streaming dispatch must reject malformed path"
+        );
+    }
+
+    #[test]
+    fn malformed_path_split_returns_error_not_panic() {
+        let result = block_on(async {
+            dispatch_and_split(
+                crate::Router::new(),
+                "GET",
+                "bad path with spaces",
+                "",
+                std::iter::empty(),
+                Body::empty(),
+                false,
+            )
+            .await
+        });
+        assert!(
+            result.is_err(),
+            "dispatch_and_split must reject malformed path"
+        );
+    }
+}

crates/vespera_inprocess/src/lib.rs

@@ -80,7 +80,8 @@ pub use envelope::{
 };
 pub use registry::{DEFAULT_APP_NAME, register_app, register_app_named};
 pub use streaming::{
-    dispatch_bidirectional_streaming, dispatch_bidirectional_streaming_with_header,
-    dispatch_streaming_async, dispatch_streaming_with_header_async,
+    RequestChunk, StreamAbort, dispatch_bidirectional_streaming,
+    dispatch_bidirectional_streaming_with_header, dispatch_streaming_async,
+    dispatch_streaming_with_header_async,
 };
 pub use wire::error_wire;