Fix jni

Author: owjs3901

crates/vespera/src/multipart.rs

@@ -607,7 +607,28 @@ pub fn register_multipart_part() -> Result<(), TypedMultipartError> {
         .unwrap_or(Ok(()))
 }
 
-fn register_multipart_bytes(field_name: &str, chunk_len: usize) -> Result<(), TypedMultipartError> {
+/// Count `chunk_len` bytes of one multipart field against the request-wide
+/// `max_total_bytes` aggregate limit, returning [`TypedMultipartError::RequestTooLarge`]
+/// once the running total crosses the cap.
+///
+/// The public counterpart of [`register_multipart_part`] for the **byte**
+/// dimension of [`MultipartLimits`]. Vespera's built-in field parsers
+/// ([`read_field_data`] / the `NamedTempFile` path) already call this once per
+/// `field.chunk()`, so typed multipart structs are accounted automatically.
+///
+/// A **custom [`TryFromFieldWithState`] implementation that consumes a field's
+/// bytes itself** (via `field.chunk()` / `field.bytes()`) MUST call this once
+/// per chunk to participate in the aggregate cap — otherwise that field's bytes
+/// are invisible to `max_total_bytes` and a single custom-parsed field can read
+/// unboundedly past the configured policy. The per-field `limit_bytes` passed to
+/// the trait method still bounds that one field, but only this call enforces the
+/// request-wide total. Mirrors the cooperative contract of
+/// [`register_multipart_part`]: outside the extractor's task-local scope (e.g. a
+/// direct unit test of a derived parser) it no-ops rather than failing.
+pub fn register_multipart_bytes(
+    field_name: &str,
+    chunk_len: usize,
+) -> Result<(), TypedMultipartError> {
     MULTIPART_AGGREGATE
         .try_with(|state| {
             let mut state = state.borrow_mut();

crates/vespera/src/multipart/tests.rs

@@ -76,6 +76,45 @@ fn temp_file_default_limit_is_bounded_and_configurable() {
     );
 }
 
+#[test]
+fn register_multipart_bytes_lets_custom_parsers_enforce_aggregate_cap() {
+    // A custom `TryFromFieldWithState` impl that consumes a field's bytes itself
+    // can now call the public `register_multipart_bytes` to participate in the
+    // request-wide `max_total_bytes` cap — previously impossible (the counter was
+    // private), so a single custom-parsed field could read unboundedly past the
+    // configured `MultipartLimits`.
+    let rt = tokio::runtime::Builder::new_current_thread()
+        .build()
+        .expect("current-thread runtime");
+    let outcome = rt.block_on(async {
+        let limits = MultipartLimits::new(10, DEFAULT_MULTIPART_MAX_FIELDS);
+        MULTIPART_AGGREGATE
+            .scope(RefCell::new(MultipartAggregateState::new(limits)), async {
+                // Under the cap: two 4-byte chunks accepted (8 <= 10).
+                register_multipart_bytes("custom", 4)?;
+                register_multipart_bytes("custom", 4)?;
+                // Crossing the cap (8 + 4 = 12 > 10) trips RequestTooLarge.
+                register_multipart_bytes("custom", 4)
+            })
+            .await
+    });
+    assert!(
+        matches!(
+            outcome,
+            Err(TypedMultipartError::RequestTooLarge {
+                limit_bytes: 10,
+                ..
+            })
+        ),
+        "custom-parser byte accounting must trip the aggregate cap, got {outcome:?}"
+    );
+
+    // Cooperative contract (mirrors `register_multipart_part`): outside the
+    // extractor's task-local scope it no-ops rather than erroring, so a derived
+    // parser can be unit-tested without a live request aggregate.
+    assert!(register_multipart_bytes("custom", usize::MAX).is_ok());
+}
+
 // ─── Display tests for all error variants ───────────────────────────
 
 #[test]

crates/vespera_core/src/schema.rs

@@ -560,6 +560,17 @@ impl Serialize for Schema {
                 "invalid Schema: nullable `$ref` serializes through anyOf and cannot also carry explicit any_of",
             ));
         }
+        // A nullable `$ref` is emitted as `anyOf: [{$ref}, {type:null}]`; a
+        // sibling `type` would then describe the SAME node twice and produce
+        // ambiguous/invalid output (`anyOf` AND `type` at one level).  Vespera's
+        // own `Schema::nullable_reference` always leaves `schema_type` None, so
+        // this only fires for a hand-built `Schema` that mixed the two — reject
+        // it like the `any_of` case above instead of serializing broken OpenAPI.
+        if nullable_ref && self.schema_type.is_some() {
+            return Err(serde::ser::Error::custom(
+                "invalid Schema: nullable `$ref` serializes through anyOf and cannot also carry an explicit type; build it via Schema::nullable_reference",
+            ));
+        }
         let mut out = serializer.serialize_struct("Schema", 42)?;
         if let Some(ref_path) = &self.ref_path {
             if nullable_ref {

crates/vespera_core/src/schema/tests.rs

@@ -328,6 +328,27 @@ fn nullable_reference_with_explicit_any_of_returns_clean_serialization_error() {
     );
 }
 
+#[test]
+fn nullable_reference_with_explicit_type_returns_clean_serialization_error() {
+    // A hand-built nullable `$ref` that ALSO carries a `schema_type` would
+    // serialize both `anyOf` and a sibling `type` — ambiguous/invalid OpenAPI.
+    // It must fail with a clean serialization error like the `any_of` case,
+    // not silently emit the broken shape. (Vespera's own `nullable_reference`
+    // leaves `schema_type` None, so this only guards external manual construction.)
+    let schema = Schema {
+        schema_type: Some(SchemaType::Object),
+        ..Schema::nullable_reference("#/components/schemas/User".to_owned())
+    };
+
+    let err = serde_json::to_string(&schema).unwrap_err();
+
+    assert!(
+        err.to_string()
+            .contains("cannot also carry an explicit type"),
+        "unexpected error: {err}",
+    );
+}
+
 #[test]
 fn nullable_primitive_emits_type_array_with_null() {
     let schema = Schema {

crates/vespera_jni/src/jni_impl.rs

@@ -198,8 +198,8 @@ fn panic_wire() -> Vec<u8> {
 #[path = "jni_impl_support.rs"]
 mod support;
 use support::{
-    push_unless_header_failed, setup_full_stream, setup_full_stream_with_header, setup_stream,
-    setup_stream_with_header, should_fire_fallback_header, throw_streaming_abort,
+    PanicHeaderAction, panic_post_header_action, push_unless_header_failed, setup_full_stream,
+    setup_full_stream_with_header, setup_stream, setup_stream_with_header, throw_streaming_abort,
 };
 
 /// `com.devfive.vespera.bridge.VesperaBridge.dispatchBytes(byte[]) -> byte[]`
@@ -641,8 +641,11 @@ pub extern "system" fn Java_com_devfive_vespera_bridge_VesperaBridge_dispatchStr
             // the consumer once with a 500 header below so the documented
             // "header consumer invoked exactly once on every code path"
             // contract holds and the Java caller is not left hanging.  A
-            // panic AFTER the header fired leaves Spring's response partially
-            // committed — unrecoverable, but the contract is already met.
+            // panic AFTER the header fired truncates the body past a header the
+            // host already committed; `panic_post_header_action` then throws
+            // IOException to abort the response (symmetric with the body-error /
+            // sink-stop abort on the `Ok` branch) instead of finishing cleanly
+            // over a short body.
             let header_sent = Arc::new(AtomicBool::new(false));
             let header_failed = Arc::new(AtomicBool::new(false));
             let header_sent_cb = Arc::clone(&header_sent);
@@ -694,24 +697,29 @@ pub extern "system" fn Java_com_devfive_vespera_bridge_VesperaBridge_dispatchStr
                     }
                 }
                 Err(_) => {
-                    // See `should_fire_fallback_header`: a panic re-enters the
-                    // header consumer ONLY when it was never invoked (neither
-                    // succeeded nor threw), upholding the "invoked exactly once on
-                    // every code path" contract.
-                    if should_fire_fallback_header(
+                    // A panic unwound out of the dispatch future.  The action
+                    // depends on whether the response header was already committed
+                    // (see `panic_post_header_action`).
+                    match panic_post_header_action(
                         header_sent.load(Ordering::Relaxed),
                         header_failed.load(Ordering::Acquire),
                     ) {
-                        let err = panic_wire();
-                        // On the JNI entry thread `header_consumer` is still a
-                        // valid LOCAL ref, so deliver the mandatory fallback
-                        // header through it directly. Promoting it to a
-                        // `Global` here added an avoidable allocation AND a
-                        // failure point: a failed `new_global_ref` (e.g. OOM)
-                        // silently skipped the required single callback and
-                        // hung the Java caller. `call_header_consumer_local`
-                        // exists for exactly this cold on-thread fallback.
-                        let _ = call_header_consumer_local(env, &header_consumer, &err);
+                        // Header never reached the consumer: deliver the one-shot
+                        // 500 fallback through the still-valid LOCAL `header_consumer`
+                        // ref (no `Global` promotion to fail first and hang the
+                        // caller), upholding "invoked exactly once on every code path".
+                        PanicHeaderAction::FireFallbackHeader => {
+                            let err = panic_wire();
+                            let _ = call_header_consumer_local(env, &header_consumer, &err);
+                        }
+                        // Header already committed (or its delivery threw): the body
+                        // is now truncated past a header the host already wrote, so
+                        // throw IOException to abort the response instead of finishing
+                        // cleanly over a short body — symmetric with the body-error /
+                        // sink-stop abort on the `Ok` branch above.
+                        PanicHeaderAction::ThrowAbort => {
+                            throw_streaming_abort(env, header_failed.load(Ordering::Acquire));
+                        }
                     }
                 }
             }
@@ -851,24 +859,29 @@ pub extern "system" fn Java_com_devfive_vespera_bridge_VesperaBridge_dispatchFul
                     }
                 }
                 Err(_) => {
-                    // See `should_fire_fallback_header`: a panic re-enters the
-                    // header consumer ONLY when it was never invoked (neither
-                    // succeeded nor threw), upholding the "invoked exactly once on
-                    // every code path" contract.
-                    if should_fire_fallback_header(
+                    // A panic unwound out of the dispatch future.  The action
+                    // depends on whether the response header was already committed
+                    // (see `panic_post_header_action`).
+                    match panic_post_header_action(
                         header_sent.load(Ordering::Relaxed),
                         header_failed.load(Ordering::Acquire),
                     ) {
-                        let err = panic_wire();
-                        // On the JNI entry thread `header_consumer` is still a
-                        // valid LOCAL ref, so deliver the mandatory fallback
-                        // header through it directly. Promoting it to a
-                        // `Global` here added an avoidable allocation AND a
-                        // failure point: a failed `new_global_ref` (e.g. OOM)
-                        // silently skipped the required single callback and
-                        // hung the Java caller. `call_header_consumer_local`
-                        // exists for exactly this cold on-thread fallback.
-                        let _ = call_header_consumer_local(env, &header_consumer, &err);
+                        // Header never reached the consumer: deliver the one-shot
+                        // 500 fallback through the still-valid LOCAL `header_consumer`
+                        // ref (no `Global` promotion to fail first and hang the
+                        // caller), upholding "invoked exactly once on every code path".
+                        PanicHeaderAction::FireFallbackHeader => {
+                            let err = panic_wire();
+                            let _ = call_header_consumer_local(env, &header_consumer, &err);
+                        }
+                        // Header already committed (or its delivery threw): the body
+                        // is now truncated past a header the host already wrote, so
+                        // throw IOException to abort the response instead of finishing
+                        // cleanly over a short body — symmetric with the body-error /
+                        // sink-stop abort on the `Ok` branch above.
+                        PanicHeaderAction::ThrowAbort => {
+                            throw_streaming_abort(env, header_failed.load(Ordering::Acquire));
+                        }
                     }
                 }
             }

crates/vespera_jni/src/jni_impl_streaming_abort_tests.rs

@@ -1,7 +1,10 @@
 use std::ops::ControlFlow;
 use std::sync::atomic::{AtomicBool, Ordering};
 
-use super::{push_unless_header_failed, should_fire_fallback_header};
+use super::support::{
+    PanicHeaderAction, panic_post_header_action, push_unless_header_failed,
+    should_fire_fallback_header,
+};
 
 #[test]
 fn push_gate_aborts_without_writing_when_header_delivery_failed() {
@@ -70,3 +73,36 @@ fn fallback_header_fires_only_when_consumer_never_invoked() {
     // re-fire.
     assert!(!should_fire_fallback_header(true, true));
 }
+
+#[test]
+fn panic_post_header_action_aborts_once_header_is_committed() {
+    // Panic BEFORE the header was ever delivered: the Java caller has no header,
+    // so the one-shot 500 fallback must be delivered (never an abort, which
+    // would leave the caller with neither a header nor a result).
+    assert_eq!(
+        panic_post_header_action(false, false),
+        PanicHeaderAction::FireFallbackHeader
+    );
+
+    // Header already SUCCEEDED, then the dispatch future panicked mid-body: the
+    // body is truncated past a committed header, so the transport must be
+    // aborted — re-firing the consumer is forbidden (already invoked once).
+    assert_eq!(
+        panic_post_header_action(true, false),
+        PanicHeaderAction::ThrowAbort
+    );
+
+    // Header delivery THREW (consumer already invoked, response already broken):
+    // a later panic must abort rather than re-enter the consumer.
+    assert_eq!(
+        panic_post_header_action(false, true),
+        PanicHeaderAction::ThrowAbort
+    );
+
+    // Defensive: both flags set never co-occurs, but must still abort, never
+    // double-invoke the consumer.
+    assert_eq!(
+        panic_post_header_action(true, true),
+        PanicHeaderAction::ThrowAbort
+    );
+}

crates/vespera_jni/src/jni_impl_support.rs

@@ -58,6 +58,43 @@ pub(super) fn should_fire_fallback_header(header_sent: bool, header_failed: bool
     !header_sent && !header_failed
 }
 
+/// What the panic landing-pad of a streaming-with-header dispatch must do after
+/// a Rust panic unwound out of the dispatch future, given whether the response
+/// header was already delivered.
+///
+/// Mirror image of the SUCCESS branch's truncation handling: that branch throws
+/// [`throw_streaming_abort`] when the body errors or the sink stops *after* the
+/// header was committed (`failed_header || BodyError | SinkStopped`).  A panic
+/// after a committed header is the SAME failure shape — the body is truncated
+/// past a header the host already wrote — so it must abort the transport too,
+/// not return cleanly over a short body.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub(super) enum PanicHeaderAction {
+    /// The header consumer was never invoked (`!header_sent && !header_failed`):
+    /// deliver the one-shot `500` fallback header so the Java caller is never
+    /// left without a header.
+    FireFallbackHeader,
+    /// The header was already committed (or its delivery threw): a panic now
+    /// truncates the body past a committed header, so throw `IOException` to
+    /// abort the response — symmetric with the body-error / sink-stop abort on
+    /// the success branch.
+    ThrowAbort,
+}
+
+/// Decide the panic-branch action from the two header flags.  Splitting it out
+/// (like [`should_fire_fallback_header`], which it reuses) keeps the decision
+/// unit-testable without a live JVM — see `jni_impl_streaming_abort_tests.rs`.
+pub(super) fn panic_post_header_action(
+    header_sent: bool,
+    header_failed: bool,
+) -> PanicHeaderAction {
+    if should_fire_fallback_header(header_sent, header_failed) {
+        PanicHeaderAction::FireFallbackHeader
+    } else {
+        PanicHeaderAction::ThrowAbort
+    }
+}
+
 /// Promoted refs + a checked-out chunk buffer for a response
 /// streaming-with-header dispatch.  Aliased so the helper return type stays
 /// under clippy's `type_complexity` cap.