Impl response body error

Author: owjs3901

crates/vespera_core/src/schema.rs

@@ -617,4 +617,78 @@ mod tests {
             "must not contain 2.0: {json}"
         );
     }
+
+    // ── CORE-04: typed `additionalProperties` (untagged) ─────────────
+    //
+    // The untagged enum MUST serialize to the bare JSON Schema wire form
+    // (a `true`/`false` or the schema object/`$ref`) — byte-identical to
+    // the previous `serde_json::Value` representation — and round-trip
+    // back to the right variant.  Untagged deserialization is
+    // order-sensitive, so these lock the contract.
+
+    #[test]
+    fn additional_properties_bool_serializes_bare() {
+        let schema = Schema {
+            additional_properties: Some(AdditionalProperties::Bool(false)),
+            ..Schema::object()
+        };
+        let json = serde_json::to_string(&schema).unwrap();
+        assert!(
+            json.contains("\"additionalProperties\":false"),
+            "bool must serialize as a bare boolean, got: {json}"
+        );
+    }
+
+    #[test]
+    fn additional_properties_schema_ref_serializes_as_ref() {
+        let schema = Schema {
+            additional_properties: Some(AdditionalProperties::Schema(SchemaRef::Ref(
+                Reference::schema("User"),
+            ))),
+            ..Schema::object()
+        };
+        let json = serde_json::to_string(&schema).unwrap();
+        assert!(
+            json.contains("\"additionalProperties\":{\"$ref\":\"#/components/schemas/User\"}"),
+            "schema-ref must serialize as a bare $ref object, got: {json}"
+        );
+    }
+
+    #[test]
+    fn additional_properties_roundtrips_each_variant() {
+        // bool → Bool
+        let v: AdditionalProperties = serde_json::from_str("true").unwrap();
+        assert!(matches!(v, AdditionalProperties::Bool(true)));
+        // {"$ref":...} → Schema(Ref)
+        let v: AdditionalProperties =
+            serde_json::from_str(r##"{"$ref":"#/components/schemas/X"}"##).unwrap();
+        assert!(matches!(v, AdditionalProperties::Schema(SchemaRef::Ref(_))));
+        // inline schema object → Schema(Inline)
+        let v: AdditionalProperties = serde_json::from_str(r#"{"type":"string"}"#).unwrap();
+        assert!(matches!(
+            v,
+            AdditionalProperties::Schema(SchemaRef::Inline(_))
+        ));
+    }
+
+    // ── CORE-03: nullable-reference constructor ──────────────────────
+
+    #[test]
+    fn nullable_reference_emits_ref_plus_nullable_only() {
+        let schema = Schema::nullable_reference("#/components/schemas/User".to_owned());
+        let json = serde_json::to_string(&schema).unwrap();
+        assert!(
+            json.contains("\"$ref\":\"#/components/schemas/User\""),
+            "must carry the $ref: {json}"
+        );
+        assert!(
+            json.contains("\"nullable\":true"),
+            "must be nullable: {json}"
+        );
+        // schema_type stays None so no stray `"type"` is emitted alongside.
+        assert!(
+            !json.contains("\"type\":"),
+            "a nullable reference must not also emit a type: {json}"
+        );
+    }
 }

crates/vespera_inprocess/benches/dispatch.rs

@@ -402,7 +402,7 @@ fn bench_direct_write_path(c: &mut Criterion) {
 
 /// P2 isolation (within-run A/B): default-app resolution via the
 /// lock-free `OnceLock` fast path vs named-app resolution through the
-/// `RwLock<HashMap>` slow path.  Identical router, identical wire
+/// lock-free `ArcSwap` load (INP-07).  Identical router, identical wire
 /// request shape — the only difference is the `"app"` header field.
 fn bench_resolve_path(c: &mut Criterion) {
     static INIT_NAMED: std::sync::Once = std::sync::Once::new();
@@ -434,11 +434,14 @@ fn bench_resolve_path(c: &mut Criterion) {
 /// many OS threads against one shared multi-thread runtime.
 ///
 /// `default` resolves through the lock-free `OnceLock` fast path;
-/// `named` goes through the `RwLock<HashMap>`.  Under reader pressure
-/// the RwLock path can park threads — the delta between the two
-/// captures exactly what the single-threaded `resolve_path` group
-/// cannot.  Excluded from the CI regression gate (heavily
-/// scheduler-dependent); run locally for the numbers.
+/// `named` resolves through the lock-free `ArcSwap` load (INP-07).
+/// Both stay lock-free under reader pressure — the residual delta is
+/// the `OnceLock` single-atomic-load advantage over the `ArcSwap`
+/// load-plus-hash-lookup, which the single-threaded `resolve_path`
+/// group cannot isolate.  See `registry_ab` for the RwLock-vs-ArcSwap
+/// before/after.
+/// Excluded from the CI regression gate (heavily scheduler-dependent);
+/// run locally for the numbers.
 fn bench_contended_path(c: &mut Criterion) {
     static INIT_NAMED: std::sync::Once = std::sync::Once::new();
 

crates/vespera_inprocess/src/registry.rs

@@ -42,15 +42,15 @@ static APP_ROUTERS: LazyLock<ArcSwap<HashMap<String, Router>>> =
 ///
 /// The overwhelmingly common dispatch case is a wire header without
 /// an `"app"` field — routing to [`DEFAULT_APP_NAME`].  Resolving it
-/// through `APP_ROUTERS` costs an `RwLock` read acquisition per
-/// request, which parks threads under high concurrency.  This
-/// `OnceLock` mirror is set (exactly once, inside the registration
-/// write lock so it can never diverge from the map) by the first
-/// successful `_default` registration and read with a single atomic
-/// load + `Router::clone` (`Arc` refcount bump) on every dispatch.
-///
-/// Named apps keep using the `RwLock<HashMap>` — they are the rare
-/// multi-app case and can be registered at any time.
+/// through `APP_ROUTERS` still costs an `ArcSwap` load + hash lookup
+/// per request.  This `OnceLock` mirror is set (exactly once, by the
+/// first successful `_default` registration so it can never diverge
+/// from the map) and read with a single atomic load + `Router::clone`
+/// (`Arc` refcount bump) on every dispatch — skipping even the hash
+/// lookup.
+///
+/// Named apps resolve through the lock-free [`ArcSwap`] load — they are
+/// the rare multi-app case and can be registered at any time.
 static DEFAULT_ROUTER: OnceLock<Router> = OnceLock::new();
 
 /// Validate an app name for registration / lookup.
@@ -119,9 +119,9 @@ where
 ///
 /// # Panic safety
 ///
-/// The `factory` closure is invoked **outside** the internal
-/// `RwLock`'s write guard.  A panic in `factory` cannot poison the
-/// map; the registration is simply discarded and the slot remains
+/// The `factory` closure is invoked **outside** the [`ArcSwap`]
+/// copy-on-write update.  A panic in `factory` cannot corrupt the
+/// registry; the registration is simply discarded and the slot remains
 /// available for retry.
 ///
 /// # Invalid names

crates/vespera_inprocess/tests/request_size_cap.rs

@@ -5,9 +5,15 @@
 //! unlimited behaviour).  Both tests pin the same cap so they are
 //! order-independent under the parallel test runner.
 
+use std::cell::{Cell, RefCell};
+use std::ops::ControlFlow;
+
 use serde_json::Value;
 use tokio::runtime::Builder;
-use vespera_inprocess::{dispatch_from_bytes, set_max_request_bytes};
+use vespera_inprocess::{
+    dispatch_from_bytes, dispatch_streaming_async, dispatch_streaming_with_header_async,
+    set_max_request_bytes,
+};
 
 /// Small enough that a tiny valid header passes but a padded request
 /// trips the cap.
@@ -19,15 +25,28 @@ fn ensure_cap() {
     let _ = set_max_request_bytes(CAP);
 }
 
+/// Parse the JSON header out of a `[u32 BE len | header JSON | body]`
+/// wire response.
+fn parse_header_json(resp: &[u8]) -> Value {
+    assert!(resp.len() >= 4, "wire response too short");
+    let header_len = u32::from_be_bytes(resp[..4].try_into().unwrap()) as usize;
+    serde_json::from_slice(&resp[4..4 + header_len]).expect("response header JSON")
+}
+
 fn dispatch(wire: Vec<u8>) -> Value {
     let runtime = Builder::new_current_thread()
         .enable_all()
         .build()
         .expect("build runtime");
-    let resp = dispatch_from_bytes(wire, &runtime);
-    assert!(resp.len() >= 4, "wire response too short");
-    let header_len = u32::from_be_bytes(resp[..4].try_into().unwrap()) as usize;
-    serde_json::from_slice(&resp[4..4 + header_len]).expect("response header JSON")
+    parse_header_json(&dispatch_from_bytes(wire, &runtime))
+}
+
+fn block_on<F: std::future::Future>(fut: F) -> F::Output {
+    Builder::new_current_thread()
+        .enable_all()
+        .build()
+        .expect("build runtime")
+        .block_on(fut)
 }
 
 fn wire_with_body(body_len: usize) -> Vec<u8> {
@@ -66,3 +85,68 @@ fn within_limit_request_is_not_capped() {
         "a request within the cap must not be rejected as oversized"
     );
 }
+
+// ── Streaming-path ingress cap (INP-01) ──────────────────────────────
+//
+// Response streaming still buffers the full *request* in memory, so it
+// must enforce the same cap as the buffered entry points — unlike
+// bidirectional streaming, which pulls the request chunk-by-chunk and
+// is intentionally exempt.
+
+#[test]
+fn oversized_streaming_request_returns_413() {
+    ensure_cap();
+    let wire = wire_with_body(200);
+    assert!(wire.len() > CAP);
+
+    let chunks = Cell::new(0usize);
+    let header_bytes = block_on(dispatch_streaming_async(wire, |_chunk: &[u8]| {
+        chunks.set(chunks.get() + 1);
+        ControlFlow::Continue(())
+    }));
+
+    let header = parse_header_json(&header_bytes);
+    assert_eq!(
+        header["status"].as_u64(),
+        Some(413),
+        "response streaming buffers the full request, so an over-cap request must be 413"
+    );
+    assert_eq!(
+        chunks.get(),
+        0,
+        "a capped request must never stream body chunks"
+    );
+}
+
+#[test]
+fn oversized_streaming_with_header_request_returns_413() {
+    ensure_cap();
+    let wire = wire_with_body(200);
+    assert!(wire.len() > CAP);
+
+    let header_seen: RefCell<Option<Vec<u8>>> = RefCell::new(None);
+    let chunks = Cell::new(0usize);
+    block_on(dispatch_streaming_with_header_async(
+        wire,
+        |header: &[u8]| *header_seen.borrow_mut() = Some(header.to_vec()),
+        |_chunk: &[u8]| {
+            chunks.set(chunks.get() + 1);
+            ControlFlow::Continue(())
+        },
+    ));
+
+    let header_bytes = header_seen
+        .into_inner()
+        .expect("the header callback must fire exactly once, even on the 413 cap path");
+    let header = parse_header_json(&header_bytes);
+    assert_eq!(
+        header["status"].as_u64(),
+        Some(413),
+        "the 413 must be delivered through the header callback"
+    );
+    assert_eq!(
+        chunks.get(),
+        0,
+        "a capped request must never stream body chunks"
+    );
+}

crates/vespera_inprocess/tests/response_body_error.rs

@@ -0,0 +1,52 @@
+//! Regression test for INP-04: a response body that errors mid-stream
+//! must surface as a `500` wire response — never the original status
+//! with a silently-truncated (empty) body.
+//!
+//! Runs in its own test binary because [`register_app`] is a
+//! process-global first-wins registration; isolating it keeps this
+//! erroring app from leaking into other integration tests.
+
+use axum::body::Body;
+use axum::response::Response;
+use axum::routing::get;
+use futures_util::stream;
+use tokio::runtime::Builder;
+use vespera_inprocess::{Router, dispatch_from_bytes, register_app};
+
+/// A `200 OK` whose body's only frame is an error — collecting it fails
+/// partway, which the buffered dispatch path must report as a `500`.
+async fn erroring_body() -> Response {
+    let s =
+        stream::once(async { Err::<bytes::Bytes, std::io::Error>(std::io::Error::other("boom")) });
+    Response::new(Body::from_stream(s))
+}
+
+fn assemble_wire(method: &str, path: &str) -> Vec<u8> {
+    let header = format!(r#"{{"v":1,"method":"{method}","path":"{path}"}}"#);
+    let mut wire = Vec::new();
+    wire.extend_from_slice(&u32::try_from(header.len()).unwrap().to_be_bytes());
+    wire.extend_from_slice(header.as_bytes());
+    wire
+}
+
+#[test]
+fn response_body_stream_error_becomes_500() {
+    register_app(|| Router::new().route("/boom", get(erroring_body)));
+
+    let runtime = Builder::new_current_thread()
+        .enable_all()
+        .build()
+        .expect("build runtime");
+    let resp = dispatch_from_bytes(assemble_wire("GET", "/boom"), &runtime);
+
+    assert!(resp.len() >= 4, "wire response too short");
+    let header_len = u32::from_be_bytes(resp[..4].try_into().unwrap()) as usize;
+    let header: serde_json::Value =
+        serde_json::from_slice(&resp[4..4 + header_len]).expect("response header JSON");
+    assert_eq!(
+        header["status"].as_u64(),
+        Some(500),
+        "a mid-stream response body error must become a 500, not a silent empty success \
+         (the handler's 200 status must NOT be reported with a truncated body)"
+    );
+}

crates/vespera_jni/src/jni_impl.rs

@@ -315,6 +315,17 @@ const _: () = assert!(DIRECT_UNREPRESENTABLE < -i32::MAX);
 /// `out_addr` must point to a writable region of at least `out_cap`
 /// bytes that stays valid for the duration of this call (a JNI
 /// direct buffer pinned by the live `JByteBuffer` local ref).
+/// Whether `[a0, a0+a_len)` and `[b0, b0+b_len)` overlap (addresses as
+/// `usize`).  Used to reject aliasing `in_buf` / `out_buf` direct-buffer
+/// ranges in [`Java_..._dispatchDirect0`] before creating a shared `&[u8]`
+/// and an exclusive `&mut [u8]` over them (SEC-1).  `saturating_add`
+/// keeps the bound arithmetic panic-free for any address.
+fn ranges_overlap(a0: usize, a_len: usize, b0: usize, b_len: usize) -> bool {
+    let a1 = a0.saturating_add(a_len);
+    let b1 = b0.saturating_add(b_len);
+    a0 < b1 && b0 < a1
+}
+
 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
@@ -383,6 +394,11 @@ fn write_response_to_out(out_addr: *mut u8, out_cap: usize, response: &[u8]) ->
 ///    keeps the backing memory valid throughout and the borrow never
 ///    escapes the `block_on`, so nothing borrowed from the buffer
 ///    outlives the call.
+/// 4. `in_buf` and `out_buf` are proven **non-overlapping** (SEC-1)
+///    before the shared `&[u8]` / exclusive `&mut [u8]` are created, so
+///    they never alias the same memory; and `out_buf` is **writable**
+///    (the Java wrapper rejects read-only buffers — SEC-2), so the
+///    `&mut [u8]` write target is valid.
 #[unsafe(no_mangle)]
 pub extern "system" fn Java_com_devfive_vespera_bridge_VesperaBridge_dispatchDirect0<'local>(
     mut unowned_env: EnvUnowned<'local>,
@@ -415,6 +431,23 @@ pub extern "system" fn Java_com_devfive_vespera_bridge_VesperaBridge_dispatchDir
                 }
             };
 
+            // SEC-1: reject overlapping `in_buf` / `out_buf` ranges.
+            // Below we create a shared `&[u8]` over the input and an
+            // exclusive `&mut [u8]` over the output; if they alias the
+            // same direct-buffer memory (the caller passed the same
+            // buffer, or overlapping `slice()`/`duplicate()` views) that
+            // is instant UB.  The Java wrapper cannot detect this (it has
+            // no native address), so the check lives here.  `out_buf` is
+            // writable by the wrapper's `isReadOnly()` guard (SEC-2), so
+            // writing the error response into it is sound.
+            if ranges_overlap(in_addr as usize, in_len, out_addr as usize, out_cap) {
+                let err = vespera_inprocess::error_wire(
+                    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));
+            }
+
             let dispatched = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
                 // SAFETY: invariants 1–3 above.  `in_addr..in_addr+in_len`
                 // (`in_len <= in_cap`) is a readable region and