Fix warning

Author: owjs3901

crates/vespera_inprocess/benches/dispatch.rs

@@ -789,6 +789,103 @@ fn bench_async_spawn_pattern(c: &mut Criterion) {
     drop(runtime);
 }
 
+/// Same-run A/B for the `RequestSourceCloser` hardening: the request-source
+/// close hook is now invoked under `catch_unwind` so a panicking hook running
+/// from `Drop` during unwind cannot double-panic -> `abort()` the host JVM.
+/// This isolates the added `catch_unwind` landing-pad cost vs a direct call,
+/// with BOTH arms in the SAME run so the measurement is immune to the
+/// cross-run thermal/load drift that swamps the dispatch-level `streaming_path`
+/// comparison (the close hook fires once per bidirectional dispatch, after the
+/// response body is fully drained, so its cost is amortised over an entire
+/// dispatch — this micro-A/B is the only instrument fine enough to resolve it).
+fn bench_close_hook_ab(c: &mut Criterion) {
+    use std::panic::AssertUnwindSafe;
+    let mut group = c.benchmark_group("close_hook_ab");
+
+    // `pre`: the previous direct `close()` call.  `post`: the hardened
+    // `catch_unwind(AssertUnwindSafe(close))`.  The closure does a tiny
+    // black-boxed op so it is neither optimised away nor large enough to
+    // dwarf the landing-pad cost being measured.
+    group.bench_function("direct_call_pre", |b| {
+        b.iter(|| {
+            let f = || std::hint::black_box(1u64).wrapping_mul(3);
+            std::hint::black_box(f())
+        });
+    });
+
+    group.bench_function("catch_unwind_post", |b| {
+        b.iter(|| {
+            let f = || std::hint::black_box(1u64).wrapping_mul(3);
+            std::hint::black_box(std::panic::catch_unwind(AssertUnwindSafe(f)).unwrap_or(0))
+        });
+    });
+
+    group.finish();
+}
+
+/// Same-run A/B for the Oracle-flagged `dispatchAsync` completion-isolation
+/// question: does completing the Java `CompletableFuture` from a
+/// `spawn_blocking` thread (so a blocking / re-entrant Java continuation runs
+/// OFF the core Tokio workers) cost enough to matter on the async path?
+///
+/// - `complete_inline_pre`: the future is completed inline on the dispatch
+///   worker (the pre-change behaviour) — no isolation hop.
+/// - `complete_spawn_blocking_post`: the completion is moved to a
+///   `spawn_blocking` thread — isolates Java continuations from the core
+///   workers at the cost of one blocking-pool hand-off.
+///
+/// Both arms run in the SAME run (drift-immune).  The delta is the per-async-
+/// dispatch cost isolation would add, and decides whether to isolate
+/// unconditionally or document the `thenApplyAsync` contract instead (speed is
+/// the stated priority, so a large hop argues for the zero-cost doc contract).
+///
+/// VERDICT (measured, AMD Ryzen 9 9950X): `complete_inline_pre` ~1.5 µs vs
+/// `complete_spawn_blocking_post` ~24.5 µs — a ~16x per-dispatch regression.
+/// Forced isolation is therefore REJECTED (it violates the speed-first
+/// priority); the worker-thread completion is kept and the threading contract
+/// is documented on `dispatchAsync` instead (callers use `*Async` continuations
+/// and avoid blocking / re-entrant inline continuations).  This A/B stays as
+/// the permanent regression-decision guard so the 16x cost is not re-discovered.
+fn bench_async_completion_isolation_ab(c: &mut Criterion) {
+    let runtime = tokio::runtime::Builder::new_multi_thread()
+        .worker_threads(4)
+        .enable_all()
+        .build()
+        .expect("multi-thread runtime");
+    let mut group = c.benchmark_group("async_completion_ab");
+
+    group.bench_function("complete_inline_pre", |b| {
+        b.iter(|| {
+            runtime.block_on(async {
+                tokio::spawn(async move {
+                    let resp = std::hint::black_box(vec![0u8; 64]);
+                    std::hint::black_box(resp.len())
+                })
+                .await
+                .unwrap()
+            })
+        });
+    });
+
+    group.bench_function("complete_spawn_blocking_post", |b| {
+        b.iter(|| {
+            runtime.block_on(async {
+                tokio::spawn(async move {
+                    let resp = std::hint::black_box(vec![0u8; 64]);
+                    tokio::task::spawn_blocking(move || std::hint::black_box(resp.len()))
+                        .await
+                        .unwrap()
+                })
+                .await
+                .unwrap()
+            })
+        });
+    });
+
+    group.finish();
+    drop(runtime);
+}
+
 /// Hand-rolled wire-header serde vs `serde_json` (within-run A/B).
 ///
 /// Gates the Oracle-ranked #2 change: replacing `serde_json` on the
@@ -1059,7 +1156,9 @@ criterion_group!(
     bench_registry_ab,
     bench_headers_path,
     bench_streaming_path,
-    bench_async_spawn_pattern
+    bench_async_spawn_pattern,
+    bench_close_hook_ab,
+    bench_async_completion_isolation_ab
 );
 
 // The within-run A/B groups compare the production hand-rolled paths against

crates/vespera_inprocess/src/streaming.rs

@@ -567,11 +567,23 @@ impl<C: FnOnce()> RequestSourceCloser<C> {
     /// close hook is consumed on the first call, so later calls (including the
     /// one in `Drop`) are no-ops. If the producer never started the hook is
     /// dropped uncalled — there is nothing to close.
+    ///
+    /// The hook runs under `catch_unwind`: `close_if_started` is also invoked
+    /// from `Drop`, which can run while a panic is already unwinding out of the
+    /// handler or the response-body poll, where a hook panic would be a
+    /// double-panic → process `abort()` (taking the host JVM down with it). The
+    /// close is best-effort cleanup (unblock a producer parked in a blocking
+    /// read) that runs only AFTER the response is fully drained, so a panicking
+    /// hook is contained rather than allowed to abort the process or fail an
+    /// already-produced response.
     fn close_if_started(&mut self) {
         if let Some(close) = self.close.take()
             && producer_was_started(&self.producer_handle)
         {
-            close();
+            // `AssertUnwindSafe`: the hook is `FnOnce()` best-effort cleanup and
+            // the producer is being torn down regardless, so swallowing its
+            // panic leaves no observable state inconsistent.
+            let _ = std::panic::catch_unwind(std::panic::AssertUnwindSafe(close));
         }
     }
 }
@@ -748,3 +760,39 @@ async fn await_request_producer(producer_handle: &RequestProducerHandle) {
         let _ = handle.await;
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::{RequestProducerHandle, RequestSourceCloser};
+    use std::sync::{Arc, Mutex};
+
+    /// A panicking user close hook must be CONTAINED by `close_if_started`:
+    /// the method also runs from `Drop` during unwind, where an escaping panic
+    /// would be a double-panic → process `abort()`.  Build a "started" producer
+    /// handle (a real `JoinHandle`, so `producer_was_started` is true and the
+    /// hook actually runs), then assert the call returns normally despite the
+    /// hook panicking, and that a second call is a consumed-hook no-op.
+    ///
+    /// Without the `catch_unwind` in `close_if_started`, the first call would
+    /// unwind out of this `#[test]` (and, on a real `Drop`-during-unwind path,
+    /// abort the process).
+    #[test]
+    fn close_hook_panic_is_contained() {
+        let runtime = tokio::runtime::Builder::new_current_thread()
+            .build()
+            .expect("current-thread runtime");
+        // `Runtime::spawn` hands back a live `JoinHandle` without entering the
+        // runtime (the empty task is never driven or awaited) — we only need a
+        // handle present so the producer counts as "started".
+        let join_handle = runtime.spawn(async {});
+        let producer_handle: RequestProducerHandle = Arc::new(Mutex::new(Some(join_handle)));
+
+        let mut closer =
+            RequestSourceCloser::new(Arc::clone(&producer_handle), || panic!("hook boom"));
+        // Returns normally — the panic is caught inside `close_if_started`.
+        closer.close_if_started();
+        // Idempotent: the hook was consumed on the first call, so this is a
+        // no-op and does not panic a second time.
+        closer.close_if_started();
+    }
+}

crates/vespera_jni/src/jni_impl.rs

@@ -328,6 +328,26 @@ mod direct;
 ///   The future is always completed with a valid wire response —
 ///   it is never left dangling, even on internal errors.
 ///
+/// # Threading contract (IMPORTANT)
+///
+/// The future is completed **on a Tokio runtime worker thread**, so any
+/// *non-async* `CompletableFuture` continuation (`thenApply`, `thenAccept`,
+/// `whenComplete`, …) runs **inline on that worker**.  Callers MUST therefore:
+/// - attach heavy / blocking continuations with the `*Async` variants
+///   (`thenApplyAsync`, `whenCompleteAsync`, …) on their own executor, and
+/// - never re-enter a blocking vespera dispatch (`dispatchBytes` /
+///   `dispatchDirect`) from an inline continuation — that nests a `block_on`
+///   inside the runtime and degrades to a caught-panic `500`.
+///
+/// Completing the future off the worker (via `spawn_blocking`) was measured at
+/// ~16x the per-dispatch cost (`vespera_inprocess` `benches/dispatch.rs`,
+/// 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.
+///
 /// Cancellation: Java's `future.cancel(true)` does NOT abort the
 /// in-flight Rust task in this iteration (defer to follow-up).
 /// Java callers may still observe cancellation via `future.isCancelled()`.

crates/vespera_macro/src/schema_macro/defaults.rs

@@ -75,10 +75,22 @@ pub(super) fn generate_sea_orm_default_attrs(
             let fn_name = format!("default_{struct_name}_{field_name}");
             let fn_ident = syn::Ident::new(&fn_name, proc_macro2::Span::call_site());
 
+            // Validate the literal against the field's type at macro-expansion
+            // time: a malformed `default_value` (e.g. `"abc"` on an `i32`, or
+            // `"300"` on a `u8`) becomes a COMPILE error pointing at the field
+            // instead of the runtime panic the generated `#value.parse().unwrap()`
+            // would raise the first time serde fills a missing field.  A valid
+            // literal keeps the byte-identical prior `.parse().unwrap()` body, so
+            // no currently-valid default changes behaviour.
+            let fn_body = match validate_literal_default(value, original_ty) {
+                Ok(()) => quote! { #value.parse().unwrap() },
+                Err(msg) => syn::Error::new_spanned(original_ty, msg).to_compile_error(),
+            };
+
             default_functions.push(quote! {
                 #[allow(non_snake_case)]
                 fn #fn_ident() -> #field_ty {
-                    #value.parse().unwrap()
+                    #fn_body
                 }
             });
 
@@ -180,6 +192,64 @@ pub(super) fn is_parseable_type(ty: &syn::Type) -> bool {
     type_utils::PRIMITIVE_TYPE_NAMES.contains(&segment.ident.to_string().as_str())
 }
 
+/// Validate a literal `default_value` against the field's type **at
+/// macro-expansion time**, mirroring exactly the runtime `#value.parse()`
+/// the generated default function performs (no trimming — the generated
+/// code does not trim either, so this predicts the runtime result precisely).
+///
+/// Returns `Err(msg)` when the literal cannot parse to the concrete field
+/// type, so the caller emits a `compile_error!` (pointing at the field)
+/// instead of generating a `.parse().unwrap()` that panics the first time
+/// serde fills a missing field.  Types whose `FromStr` cannot be faithfully
+/// reproduced here return `Ok(())`:
+/// - `String` — its `FromStr` is infallible.
+/// - `Decimal` — needs the `rust_decimal` runtime crate; left to runtime.
+/// - any non-primitive / unknown type — already gated out by
+///   [`is_parseable_type`] before this is reached.
+fn validate_literal_default(value: &str, ty: &syn::Type) -> Result<(), String> {
+    let syn::Type::Path(type_path) = ty else {
+        return Ok(());
+    };
+    let Some(segment) = type_path.path.segments.last() else {
+        return Ok(());
+    };
+    let type_name = segment.ident.to_string();
+
+    // Parse against the EXACT field type so a range violation (e.g. `"300"`
+    // on a `u8`) is caught, not just a syntactic one.  The message carries
+    // the offending value and type plus the underlying `FromStr` error — the
+    // same error the runtime `.unwrap()` would have panicked with.
+    macro_rules! check {
+        ($t:ty) => {
+            value
+                .parse::<$t>()
+                .map(|_| ())
+                .map_err(|e| format!("invalid default_value {value:?} for `{type_name}`: {e}"))
+        };
+    }
+
+    match type_name.as_str() {
+        "i8" => check!(i8),
+        "i16" => check!(i16),
+        "i32" => check!(i32),
+        "i64" => check!(i64),
+        "i128" => check!(i128),
+        "isize" => check!(isize),
+        "u8" => check!(u8),
+        "u16" => check!(u16),
+        "u32" => check!(u32),
+        "u64" => check!(u64),
+        "u128" => check!(u128),
+        "usize" => check!(usize),
+        "f32" => check!(f32),
+        "f64" => check!(f64),
+        "bool" => check!(bool),
+        // `String::FromStr` is infallible; `Decimal` needs the runtime crate.
+        // Everything else is gated out by `is_parseable_type` before this call.
+        _ => Ok(()),
+    }
+}
+
 #[cfg(test)]
 mod tests {
     use std::collections::HashMap;
@@ -196,10 +266,102 @@ mod tests {
         items.into_iter().map(|s| (s.name.clone(), s)).collect()
     }
 
+    // ======================================
+    // validate_literal_default tests
+    // ======================================
+
+    #[test]
+    fn validate_literal_default_accepts_valid_primitives() {
+        let i32_ty: syn::Type = syn::parse_str("i32").unwrap();
+        assert!(validate_literal_default("42", &i32_ty).is_ok());
+        let u8_ty: syn::Type = syn::parse_str("u8").unwrap();
+        assert!(validate_literal_default("255", &u8_ty).is_ok());
+        let f64_ty: syn::Type = syn::parse_str("f64").unwrap();
+        assert!(validate_literal_default("0.7", &f64_ty).is_ok());
+        let bool_ty: syn::Type = syn::parse_str("bool").unwrap();
+        assert!(validate_literal_default("true", &bool_ty).is_ok());
+        // String FromStr is infallible; Decimal is intentionally left to runtime.
+        let string_ty: syn::Type = syn::parse_str("String").unwrap();
+        assert!(validate_literal_default("anything at all", &string_ty).is_ok());
+        let decimal_ty: syn::Type = syn::parse_str("Decimal").unwrap();
+        assert!(validate_literal_default("not-validated-here", &decimal_ty).is_ok());
+    }
+
+    #[test]
+    fn validate_literal_default_rejects_unparseable_and_out_of_range() {
+        let i32_ty: syn::Type = syn::parse_str("i32").unwrap();
+        assert!(validate_literal_default("abc", &i32_ty).is_err());
+        // Range violation caught against the EXACT type, not a generic integer.
+        let u8_ty: syn::Type = syn::parse_str("u8").unwrap();
+        assert!(validate_literal_default("300", &u8_ty).is_err());
+        let bool_ty: syn::Type = syn::parse_str("bool").unwrap();
+        assert!(validate_literal_default("maybe", &bool_ty).is_err());
+        let f64_ty: syn::Type = syn::parse_str("f64").unwrap();
+        assert!(validate_literal_default("3.14.15", &f64_ty).is_err());
+    }
+
     // ======================================
     // generate_sea_orm_default_attrs tests
     // ======================================
 
+    #[test]
+    fn test_sea_orm_default_attrs_valid_literal_keeps_parse_unwrap() {
+        let attrs: Vec<syn::Attribute> = vec![syn::parse_quote!(#[sea_orm(default_value = "42")])];
+        let struct_name = syn::Ident::new("Test", proc_macro2::Span::call_site());
+        let ty: syn::Type = syn::parse_str("i32").unwrap();
+        let mut fns = Vec::new();
+        let (serde, _schema) = generate_sea_orm_default_attrs(
+            &attrs,
+            &struct_name,
+            "count",
+            &ty,
+            &ty,
+            false,
+            &mut fns,
+        );
+        assert!(serde.to_string().contains("serde"));
+        assert_eq!(fns.len(), 1);
+        let body = fns[0].to_string();
+        assert!(body.contains("parse"), "valid literal keeps parse: {body}");
+        assert!(body.contains("unwrap"), "valid literal keeps unwrap: {body}");
+        assert!(
+            !body.contains("compile_error"),
+            "valid literal must not emit compile_error: {body}"
+        );
+    }
+
+    #[test]
+    fn test_sea_orm_default_attrs_invalid_literal_emits_compile_error() {
+        // `"abc"` cannot parse to i32: the generated default function body must
+        // be a compile_error (pointing at the field) instead of a runtime
+        // `.parse().unwrap()` that would panic when serde fills a missing field.
+        let attrs: Vec<syn::Attribute> =
+            vec![syn::parse_quote!(#[sea_orm(default_value = "abc")])];
+        let struct_name = syn::Ident::new("Test", proc_macro2::Span::call_site());
+        let ty: syn::Type = syn::parse_str("i32").unwrap();
+        let mut fns = Vec::new();
+        let (serde, _schema) = generate_sea_orm_default_attrs(
+            &attrs,
+            &struct_name,
+            "count",
+            &ty,
+            &ty,
+            false,
+            &mut fns,
+        );
+        assert!(serde.to_string().contains("serde"));
+        assert_eq!(fns.len(), 1);
+        let body = fns[0].to_string();
+        assert!(
+            body.contains("compile_error"),
+            "invalid literal must emit compile_error: {body}"
+        );
+        assert!(
+            !body.contains("unwrap"),
+            "invalid literal must not emit a runtime parse().unwrap(): {body}"
+        );
+    }
+
     #[test]
     fn test_sea_orm_default_attrs_optional_field_skips() {
         let attrs: Vec<syn::Attribute> = vec![syn::parse_quote!(#[sea_orm(default_value = "42")])];