Check java buffer

Author: owjs3901

crates/vespera/src/multipart.rs

@@ -171,9 +171,13 @@ impl TypedMultipartError {
 impl IntoResponse for TypedMultipartError {
     fn into_response(self) -> Response {
         let status = match &self {
-            Self::InvalidRequest { .. }
-            | Self::InvalidRequestBody { .. }
-            | Self::MissingField { .. }
+            // Preserve the SOURCE rejection / stream status so an over-limit
+            // multipart body surfaces as `413 Payload Too Large` (axum's body
+            // limit), an unsupported media type as `415`, etc. — instead of
+            // collapsing every transport-level failure to a generic `400`.
+            Self::InvalidRequest { source } => source.status(),
+            Self::InvalidRequestBody { source } => source.status(),
+            Self::MissingField { .. }
             | Self::DuplicateField { .. }
             | Self::UnknownField { .. }
             | Self::InvalidEnumValue { .. }
@@ -185,16 +189,40 @@ impl IntoResponse for TypedMultipartError {
             Self::FieldTooLarge { .. } => StatusCode::PAYLOAD_TOO_LARGE,
             Self::Other { .. } => StatusCode::INTERNAL_SERVER_ERROR,
         };
-        // Canonical JSON error envelope — the SAME shape `Validated<T>`
-        // emits ({"errors":[{"path","message"}]}), so multipart failures are
-        // consumed uniformly across the API instead of as ad-hoc plain text;
-        // under JNI a 422 body is additionally hoisted into the wire header
-        // exactly like a `Validated` rejection.  `path` is the offending
-        // field name when known, else empty.
-        let path = self.field_name().unwrap_or("").to_owned();
+        // Canonical JSON error envelope, byte-identical to `Validated<T>`'s
+        // 422 envelope — `{"errors":[{"message":...,"path":...}]}` (message
+        // before path) — so multipart failures are consumed uniformly and,
+        // under JNI, the 422 body hoists into the wire header exactly like a
+        // `Validated` rejection.  Serialized through a borrowing `Serialize`
+        // (no `serde_json::Value` map/array/object intermediate). `path` is
+        // the offending field name when known, else empty.
+        #[derive(serde::Serialize)]
+        struct OneError<'a> {
+            message: &'a str,
+            path: &'a str,
+        }
+        #[derive(serde::Serialize)]
+        struct Envelope<'a> {
+            errors: [OneError<'a>; 1],
+        }
+        let path = self.field_name().unwrap_or("");
         let message = self.response_message();
-        let body = serde_json::json!({ "errors": [{ "path": path, "message": message }] });
-        (status, axum::Json(body)).into_response()
+        let body = serde_json::to_vec(&Envelope {
+            errors: [OneError {
+                message: &message,
+                path,
+            }],
+        })
+        .expect("multipart error envelope is infallible to serialize");
+        (
+            status,
+            [(
+                axum::http::header::CONTENT_TYPE,
+                axum::http::HeaderValue::from_static("application/json"),
+            )],
+            body,
+        )
+            .into_response()
     }
 }
 

crates/vespera_core/src/openapi.rs

@@ -286,20 +286,16 @@ impl OpenApi {
             self.external_docs = other.external_docs;
         }
 
-        // Merge tags, de-duplicating by name in a single pass.  `seen` starts
-        // with the existing tag names and grows as incoming tags are appended,
-        // so an incoming tag is kept only when its name collides with neither
-        // an existing tag nor an already-appended incoming one (first-wins,
-        // incoming insertion order preserved).  A name is cloned only when the
-        // tag is actually kept — a duplicate is detected by borrow and skipped
-        // without cloning.
+        // Merge tags, de-duplicating by name in a single pass with first-wins
+        // semantics (existing tags and already-appended incoming tags both
+        // win; incoming insertion order preserved).  Tag lists are tiny, so a
+        // linear membership scan over `self_tags` beats a `HashSet` here: it
+        // allocates nothing and clones nothing — the kept tag is *moved* in,
+        // and a duplicate is detected by borrow and skipped.
         if let Some(other_tags) = other.tags {
             let self_tags = self.tags.get_or_insert_with(Vec::new);
-            let mut seen: std::collections::HashSet<String> =
-                self_tags.iter().map(|tag| tag.name.clone()).collect();
             for tag in other_tags {
-                if !seen.contains(tag.name.as_str()) {
-                    seen.insert(tag.name.clone());
+                if !self_tags.iter().any(|existing| existing.name == tag.name) {
                     self_tags.push(tag);
                 }
             }

crates/vespera_core/src/schema.rs

@@ -191,10 +191,20 @@ pub struct Schema {
         serialize_with = "serialize_number_constraint"
     )]
     pub maximum: Option<f64>,
-    /// Exclusive minimum
+    /// Exclusive minimum.
+    ///
+    /// NOTE: currently modeled as the OpenAPI 3.0 / draft-04 **boolean
+    /// flag** (paired with `minimum`).  Migrating this to the JSON Schema
+    /// 2020-12 / OpenAPI 3.1 **numeric** form is tracked as a deliberate,
+    /// breaking spec-conformance change (it alters generated output and the
+    /// `#[schema(exclusive_minimum)]` attribute semantics) — see the 3.1
+    /// conformance decision, not done here to avoid a half-migrated model.
     #[serde(skip_serializing_if = "Option::is_none")]
     pub exclusive_minimum: Option<bool>,
-    /// Exclusive maximum
+    /// Exclusive maximum.
+    ///
+    /// See [`Schema::exclusive_minimum`]: still the OpenAPI 3.0 boolean
+    /// flag, pending the bundled strict-3.1 conformance migration.
     #[serde(skip_serializing_if = "Option::is_none")]
     pub exclusive_maximum: Option<bool>,
     /// Multiple of
@@ -411,17 +421,23 @@ impl Schema {
             Ok(schema) => schema,
             Err(e) => {
                 // Surface the (in-practice-unreachable) macro/serde drift in
-                // debug / CI builds while degrading gracefully in release.
-                // `debug_assert!` keeps `e` referenced in both profiles (its
-                // release expansion is a dead `if false` branch), so there is
-                // no unused-binding warning.
+                // debug / CI builds via `debug_assert!`.  In release, degrade
+                // to a VISIBLE sentinel schema (a description-only object)
+                // rather than a silent `Schema::default()`, so a drift never
+                // disappears unnoticed from the generated spec yet never
+                // panics in downstream user code.
                 debug_assert!(
                     false,
                     "vespera: Schema::from_compiled_json failed to parse macro-emitted \
-                     JSON ({e}); falling back to Schema::default(). This indicates a \
+                     JSON ({e}); emitting a sentinel schema. This indicates a \
                      vespera bug — the macro serialized a Schema that cannot round-trip."
                 );
-                Self::default()
+                Self {
+                    description: Some(format!(
+                        "vespera: schema unavailable — macro/serde drift ({e})"
+                    )),
+                    ..Self::default()
+                }
             }
         }
     }
@@ -489,6 +505,10 @@ pub enum SecuritySchemeType {
     /// `mutualTls` the container rule would produce).
     #[serde(rename = "mutualTLS")]
     MutualTls,
+    /// OpenAPI's canonical wire name is `oauth2`; the `camelCase` container
+    /// rule would otherwise lowercase only the leading char and emit the
+    /// invalid `oAuth2`.
+    #[serde(rename = "oauth2")]
     OAuth2,
     OpenIdConnect,
 }
@@ -662,6 +682,38 @@ mod tests {
         );
     }
 
+    // ── CORE: OpenAPI 3.1 conformance of the schema model ────────────
+
+    #[test]
+    fn oauth2_security_scheme_serializes_to_canonical_lowercase() {
+        // OpenAPI's canonical wire name is `oauth2`.  serde's `camelCase`
+        // container rule lowercases only the leading char, which would emit
+        // the invalid `oAuth2` without the explicit `#[serde(rename)]`.
+        let json = serde_json::to_string(&SecuritySchemeType::OAuth2).unwrap();
+        assert_eq!(json, "\"oauth2\"", "must be exactly \"oauth2\"");
+    }
+
+    #[rstest]
+    #[case(SecuritySchemeType::ApiKey, "\"apiKey\"")]
+    #[case(SecuritySchemeType::Http, "\"http\"")]
+    #[case(SecuritySchemeType::MutualTls, "\"mutualTLS\"")]
+    #[case(SecuritySchemeType::OAuth2, "\"oauth2\"")]
+    #[case(SecuritySchemeType::OpenIdConnect, "\"openIdConnect\"")]
+    fn security_scheme_type_uses_openapi_canonical_wire_names(
+        #[case] ty: SecuritySchemeType,
+        #[case] expected: &str,
+    ) {
+        assert_eq!(serde_json::to_string(&ty).unwrap(), expected);
+    }
+
+    #[test]
+    #[should_panic(expected = "from_compiled_json failed to parse")]
+    fn from_compiled_json_invalid_input_trips_debug_assert() {
+        // In debug / test builds the (in-practice-unreachable) macro/serde
+        // drift guard fires loudly so a bug never goes unnoticed in CI.
+        let _ = Schema::from_compiled_json("{not valid json");
+    }
+
     // ── CORE-04: typed `additionalProperties` (untagged) ─────────────
     //
     // The untagged enum MUST serialize to the bare JSON Schema wire form

crates/vespera_inprocess/src/internal.rs

@@ -292,7 +292,13 @@ where
                     && !data.is_empty()
                     && on_chunk(data.as_ref()).is_break()
                 {
-                    break;
+                    // The chunk sink asked to stop EARLY (e.g. the host's
+                    // OutputStream failed mid-stream).  The bytes already
+                    // delivered are truncated, so surface a 500 — exactly
+                    // like the body-error arm below — instead of falling
+                    // through to the original success header, which would
+                    // report a short, truncated response as a clean success.
+                    return Err((500, "response body sink stopped before completion".to_owned()));
                 }
             }
             Some(Err(_)) => {

crates/vespera_inprocess/src/streaming.rs

@@ -379,12 +379,26 @@ where
     C: FnOnce(),
 {
     let mut header_bytes: Vec<u8> = Vec::with_capacity(4 + WIRE_HEADER_RESERVE);
-    {
+    let outcome = {
         let on_header = |h: &[u8]| header_bytes.extend_from_slice(h);
         bidirectional_streaming_inner(input_header, pull_chunk, on_chunk, on_header, request_close)
-            .await;
+            .await
+    };
+    match outcome {
+        // `Complete` covers a clean drain AND the pre-dispatch error paths
+        // (which deliver a full `error_wire(...)` via `on_header`), so the
+        // captured bytes are authoritative.
+        StreamOutcome::Complete => header_bytes,
+        // The response body errored, or the chunk sink stopped, AFTER the
+        // success header was captured into `header_bytes` — the delivered
+        // body is truncated.  Replace the captured success header with a 500
+        // so a truncated bidirectional response is never returned as a clean
+        // success (mirrors `dispatch_streaming_async`).
+        StreamOutcome::BodyError => error_wire(500, "response body stream error"),
+        StreamOutcome::SinkStopped => {
+            error_wire(500, "response body sink stopped before completion")
+        }
     }
-    header_bytes
 }
 
 /// **Bidirectional streaming with explicit header callback** — the
@@ -725,7 +739,20 @@ fn spawn_request_producer(
         // handler aborted mid-stream, so we stop pulling.
         let mut consecutive_empty: u32 = 0;
         loop {
-            match pull() {
+            // A panic inside the user / JNI-supplied `pull()` must NOT be
+            // turned into a clean end-of-stream — that would accept a
+            // TRUNCATED upload as a complete request body (silent data
+            // loss).  Catch it and forward a `StreamAbort`, exactly like the
+            // explicit `RequestChunk::Error` path, so axum/the handler
+            // rejects the body instead of seeing a short, "successful" read.
+            let next = match std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| pull())) {
+                Ok(next) => next,
+                Err(_) => {
+                    let _ = tx.blocking_send(Err(StreamAbort));
+                    break;
+                }
+            };
+            match next {
                 RequestChunk::Data(chunk) => {
                     if chunk.is_empty() {
                         // A conformant blocking `InputStream.read(byte[])`

crates/vespera_inprocess/tests/streaming_with_header.rs

@@ -1044,7 +1044,13 @@ async fn streaming_with_header_chunk_break_returns_sink_stopped_outcome() {
 }
 
 #[tokio::test(flavor = "multi_thread", worker_threads = 4)]
-async fn response_streaming_stops_draining_when_chunk_callback_breaks() {
+async fn response_streaming_chunk_break_returns_500_not_silent_success() {
+    // When the chunk sink returns `Break` (the host output sink failed
+    // mid-stream), the non-header `dispatch_streaming_async` must surface a
+    // 500 — NOT the original success header — so a TRUNCATED response is never
+    // reported as a clean success.  (Mirrors the header-first
+    // `...sink_stopped_outcome` and direct-write
+    // `...body_error_yields_500_not_truncated_success` contracts.)
     install_router();
     let wire = encode_wire("GET", "/multi-chunk", HashMap::new(), &[]);
     let body_buf: Arc<Mutex<Vec<u8>>> = Arc::new(Mutex::new(Vec::new()));
@@ -1056,12 +1062,39 @@ async fn response_streaming_stops_draining_when_chunk_callback_breaks() {
     })
     .await;
 
-    let (header_json, header_body) = decode_wire(&header);
-    assert_eq!(header_json["status"].as_u64(), Some(200));
-    assert!(header_body.is_empty());
+    let (header_json, _header_body) = decode_wire(&header);
+    assert_eq!(
+        header_json["status"].as_u64(),
+        Some(500),
+        "a chunk-sink break must yield 500, not a truncated 200 success"
+    );
+    // The first chunk was already delivered to the sink before the break fired.
     assert_eq!(body_buf.lock().unwrap().as_slice(), b"first");
 }
 
+#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+async fn bidirectional_chunk_break_returns_500_not_silent_success() {
+    // The non-header BIDIRECTIONAL path must also surface a 500 when the chunk
+    // sink breaks mid-response, instead of returning the captured success
+    // header (which would report a truncated bidirectional response as a clean
+    // success).  Mirrors `response_streaming_chunk_break_returns_500...`.
+    install_router();
+    let wire = encode_wire("GET", "/multi-chunk", HashMap::new(), &[]);
+    let header = dispatch_bidirectional_streaming_closing(
+        wire,
+        || RequestChunk::End,            // no request body
+        |_chunk| ControlFlow::Break(()), // sink fails on the first chunk
+        || {},                           // no-op request-source close
+    )
+    .await;
+    let (header_json, _) = decode_wire(&header);
+    assert_eq!(
+        header_json["status"].as_u64(),
+        Some(500),
+        "a bidirectional chunk-sink break must yield 500, not truncated success"
+    );
+}
+
 #[tokio::test(flavor = "multi_thread", worker_threads = 4)]
 async fn direct_write_body_error_yields_500_not_truncated_success() {
     // Direct-write path: the response is buffered into the caller's slice and

crates/vespera_jni/src/jni_impl_direct.rs

@@ -50,12 +50,24 @@ fn ranges_overlap(a0: usize, a_len: usize, b0: usize, b_len: usize) -> bool {
     a0 < b1 && b0 < a1
 }
 
-fn write_response_to_out(out_addr: *mut u8, out_cap: usize, response: &[u8]) -> jint {
+/// Copy `response` into the caller's direct out buffer, returning the
+/// `dispatchDirect0` code (`>= 0` bytes written, `-(required)` on overflow,
+/// [`DIRECT_UNREPRESENTABLE`] when the size exceeds `i32::MAX`).
+///
+/// # Safety
+///
+/// `out_addr` must point to a writable region of at least `out_cap` bytes
+/// that stays valid for the whole call (a JNI direct buffer pinned by a
+/// live `JByteBuffer` local ref) and must NOT alias `response` (callers
+/// pass a Rust-owned wire `Vec`).  Encoded as `unsafe fn` so every call
+/// site acknowledges the raw-pointer contract instead of it being an
+/// unchecked promise on a safe function.
+unsafe fn write_response_to_out(out_addr: *mut u8, out_cap: usize, response: &[u8]) -> jint {
     if response.len() <= out_cap {
-        // SAFETY: `response.len() <= out_cap` and the caller
-        // guarantees `out_addr..out_addr+out_cap` is writable.
-        // Source and destination cannot overlap: `response` is a
-        // Rust-owned Vec, the destination is a Java direct buffer.
+        // SAFETY: `response.len() <= out_cap` and the caller's `# Safety`
+        // contract guarantees `out_addr..out_addr+out_cap` is writable and
+        // non-aliasing with `response` (a Rust-owned Vec → a Java direct
+        // buffer).
         unsafe {
             std::ptr::copy_nonoverlapping(response.as_ptr(), out_addr, response.len());
         }
@@ -151,7 +163,7 @@ pub extern "system" fn Java_com_devfive_vespera_bridge_VesperaBridge_dispatchDir
                         400,
                         "invalid in_len (negative or exceeds buffer capacity)",
                     );
-                    return Ok(write_response_to_out(out_addr, out_cap, &err));
+                    return Ok(unsafe { write_response_to_out(out_addr, out_cap, &err) });
                 }
             };
 
@@ -169,7 +181,7 @@ pub extern "system" fn Java_com_devfive_vespera_bridge_VesperaBridge_dispatchDir
                     400,
                     "in_buf and out_buf must not overlap (aliasing would be undefined behavior)",
                 );
-                return Ok(write_response_to_out(out_addr, out_cap, &err));
+                return Ok(unsafe { write_response_to_out(out_addr, out_cap, &err) });
             }
 
             let dispatched = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
@@ -197,7 +209,7 @@ pub extern "system" fn Java_com_devfive_vespera_bridge_VesperaBridge_dispatchDir
                 }
                 Err(_) => {
                     let err = vespera_inprocess::error_wire(500, "panic in Rust engine");
-                    write_response_to_out(out_addr, out_cap, &err)
+                    unsafe { write_response_to_out(out_addr, out_cap, &err) }
                 }
             };
             Ok(code)