Rm clippy warning

Author: owjs3901

crates/vespera_core/src/openapi.rs

@@ -316,12 +316,17 @@ impl OpenApi {
         // Merge tags, de-duplicating by name with first-wins semantics while
         // preserving deterministic output order (existing tags first, then
         // incoming tags in their original order).
+        //
+        // A linear `any` scan beats a `HashSet<String>` here: tag sets are
+        // tiny (OpenAPI tags are top-level operation groupings — a handful,
+        // rarely past a few dozen even for large APIs), so the O(n²) short-
+        // string compare over an already-resident `Vec` is cheaper than
+        // allocating a set and cloning every existing + incoming tag name.
+        // Net: zero allocations and zero `String` clones on the merge path.
         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.insert(tag.name.clone()) {
+                if !self_tags.iter().any(|existing| existing.name == tag.name) {
                     self_tags.push(tag);
                 }
             }

crates/vespera_inprocess/benches/dispatch/serde_ab.rs

@@ -8,6 +8,11 @@
 //! `#[cfg(feature = "bench-support")]`). Wired into the parent `ab_benches`
 //! criterion group.
 
+// This is a `#[path]` bench submodule of `dispatch.rs`; it intentionally
+// re-uses the parent bench file's imports (criterion types, header types,
+// wire-assembly helpers) rather than re-listing them.  The glob is the
+// idiomatic shape for a bench helper split out only to honour the file-size cap.
+#[allow(clippy::wildcard_imports)]
 use super::*;
 
 /// `request_parse_*` / `response_serialize_*` within-run A/B: the hand-rolled

crates/vespera_inprocess/src/streaming.rs

@@ -543,13 +543,24 @@ where
     outcome
 }
 
+/// Lock the producer handle, transparently recovering the guard if the mutex
+/// was poisoned.  A poison here only means a prior holder panicked while the
+/// streaming path was tearing down; the guarded `Option<JoinHandle>` is still
+/// structurally valid, so recovering and proceeding is correct — and keeps this
+/// FFI-adjacent path free of the `unwrap` panic site each of the three call
+/// sites would otherwise carry.  (Same idiom as the registry/bench read paths.)
+fn lock_producer_handle(
+    producer_handle: &RequestProducerHandle,
+) -> std::sync::MutexGuard<'_, Option<tokio::task::JoinHandle<()>>> {
+    producer_handle
+        .lock()
+        .unwrap_or_else(std::sync::PoisonError::into_inner)
+}
+
 /// Whether the request producer task was started — i.e. the handler read
 /// at least one body chunk, which lazily spawns the producer.
 fn producer_was_started(producer_handle: &RequestProducerHandle) -> bool {
-    match producer_handle.lock() {
-        Ok(guard) => guard.is_some(),
-        Err(poisoned) => poisoned.into_inner().is_some(),
-    }
+    lock_producer_handle(producer_handle).is_some()
 }
 
 /// RAII guard that closes the request body source **exactly once** if the
@@ -750,24 +761,13 @@ fn store_request_producer_handle(
     producer_handle: &RequestProducerHandle,
     handle: tokio::task::JoinHandle<()>,
 ) {
-    match producer_handle.lock() {
-        Ok(mut guard) => *guard = Some(handle),
-        Err(poisoned) => {
-            let mut guard = poisoned.into_inner();
-            *guard = Some(handle);
-        }
-    }
+    *lock_producer_handle(producer_handle) = Some(handle);
 }
 
 async fn await_request_producer(producer_handle: &RequestProducerHandle) {
-    let handle = match producer_handle.lock() {
-        Ok(mut guard) => guard.take(),
-        Err(poisoned) => {
-            let mut guard = poisoned.into_inner();
-            guard.take()
-        }
-    };
-
+    // Take the handle and release the guard on the same statement: a
+    // `MutexGuard` is not `Send` and must never be held across the `.await`.
+    let handle = lock_producer_handle(producer_handle).take();
     if let Some(handle) = handle {
         let _ = handle.await;
     }

crates/vespera_inprocess/src/wire/header_write.rs

@@ -182,15 +182,20 @@ fn write_headers<S: JsonSink>(sink: &mut S, headers: &http::HeaderMap) {
         return;
     }
     if key_count == 1 {
-        let name = headers
-            .keys()
-            .next()
-            .expect("keys_len()==1 yields exactly one name");
-        sink.put(b"{");
-        write_header_name_json_string(sink, name.as_str());
-        sink.put(b":");
-        write_header_value(sink, headers, name.as_str());
-        sink.put(b"}");
+        // `keys_len() == 1` guarantees exactly one key.  On the impossible
+        // `None` we emit an empty object instead of panicking, keeping this
+        // FFI-adjacent response serializer free of unwind sites (mirrors the
+        // no-panic/unwind discipline the dispatch internals document).
+        if let Some(name) = headers.keys().next() {
+            sink.put(b"{");
+            write_header_name_json_string(sink, name.as_str());
+            sink.put(b":");
+            write_header_value(sink, headers, name.as_str());
+            sink.put(b"}");
+        } else {
+            debug_assert!(false, "keys_len()==1 yields exactly one name");
+            sink.put(b"{}");
+        }
         return;
     }
 
@@ -241,9 +246,14 @@ fn write_header_name_json_string<S: JsonSink>(sink: &mut S, name: &str) {
 /// (first, second, then the rest) — byte-identical, no second hash lookup.
 fn write_header_value<S: JsonSink>(sink: &mut S, headers: &http::HeaderMap, name: &str) {
     let mut values = headers.get_all(name).iter();
-    let first = values
-        .next()
-        .expect("write_header_value is only called for present names");
+    // `write_header_value` is only invoked for names taken from `headers.keys()`,
+    // so `get_all(name)` is always non-empty.  On the impossible `None` we emit an
+    // empty-string value rather than panicking on this response hot path.
+    let Some(first) = values.next() else {
+        debug_assert!(false, "write_header_value is only called for present names");
+        write_json_string(sink, "");
+        return;
+    };
     match values.next() {
         // Single value: emit the scalar string.
         None => write_json_string(sink, first.to_str().unwrap_or("")),

crates/vespera_jni/src/jni_impl.rs

@@ -199,7 +199,7 @@ fn panic_wire() -> Vec<u8> {
 mod support;
 use support::{
     push_unless_header_failed, setup_full_stream, setup_full_stream_with_header, setup_stream,
-    setup_stream_with_header, throw_streaming_abort,
+    setup_stream_with_header, should_fire_fallback_header, throw_streaming_abort,
 };
 
 /// `com.devfive.vespera.bridge.VesperaBridge.dispatchBytes(byte[]) -> byte[]`
@@ -684,8 +684,14 @@ pub extern "system" fn Java_com_devfive_vespera_bridge_VesperaBridge_dispatchStr
                     }
                 }
                 Err(_) => {
-                    if !header_sent.load(Ordering::Relaxed)
-                        && let Ok(fallback) = env.new_global_ref(&header_consumer)
+                    // 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(
+                        header_sent.load(Ordering::Relaxed),
+                        header_failed.load(Ordering::Acquire),
+                    ) && let Ok(fallback) = env.new_global_ref(&header_consumer)
                     {
                         let err = panic_wire();
                         let _ = call_header_consumer(env, &fallback, &err);
@@ -828,8 +834,14 @@ pub extern "system" fn Java_com_devfive_vespera_bridge_VesperaBridge_dispatchFul
                     }
                 }
                 Err(_) => {
-                    if !header_sent.load(Ordering::Relaxed)
-                        && let Ok(fallback) = env.new_global_ref(&header_consumer)
+                    // 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(
+                        header_sent.load(Ordering::Relaxed),
+                        header_failed.load(Ordering::Acquire),
+                    ) && let Ok(fallback) = env.new_global_ref(&header_consumer)
                     {
                         let err = panic_wire();
                         let _ = call_header_consumer(env, &fallback, &err);

crates/vespera_jni/src/jni_impl_streaming_abort_tests.rs

@@ -1,7 +1,7 @@
 use std::ops::ControlFlow;
 use std::sync::atomic::{AtomicBool, Ordering};
 
-use super::push_unless_header_failed;
+use super::{push_unless_header_failed, should_fire_fallback_header};
 
 #[test]
 fn push_gate_aborts_without_writing_when_header_delivery_failed() {
@@ -49,3 +49,24 @@ fn push_gate_delegates_when_header_delivery_succeeded() {
         push_unless_header_failed(&header_failed, &mut |_| ControlFlow::Continue(()), b"x");
     assert!(stopped.is_break());
 }
+
+#[test]
+fn fallback_header_fires_only_when_consumer_never_invoked() {
+    // Panic unwound BEFORE the header callback was ever reached: the Java caller
+    // has no header yet, so the one-shot 500 fallback MUST fire.
+    assert!(should_fire_fallback_header(false, false));
+
+    // Header callback already SUCCEEDED: re-firing would deliver the header
+    // twice — forbidden by the "invoked exactly once on every code path" contract.
+    assert!(!should_fire_fallback_header(true, false));
+
+    // Header callback already THREW (it WAS invoked): a later panic must not
+    // re-enter the (possibly broken / already-committed) consumer a second time.
+    // This is the edge the prior `!header_sent`-only guard mishandled by
+    // double-invoking the consumer.
+    assert!(!should_fire_fallback_header(false, true));
+
+    // Defensive: both flags set never co-occurs in practice, but must still not
+    // re-fire.
+    assert!(!should_fire_fallback_header(true, true));
+}

crates/vespera_jni/src/jni_impl_support.rs

@@ -38,6 +38,26 @@ pub(super) fn push_unless_header_failed(
     }
 }
 
+/// Whether the panic-path fallback header (a one-shot `500`) should be delivered
+/// after a Rust panic unwound out of a streaming-with-header dispatch.
+///
+/// It fires ONLY when the header consumer was never invoked: `header_sent`
+/// records a SUCCESSFUL invocation and `header_failed` records one that THREW —
+/// either flag means "already invoked once", so a later panic must NOT re-enter
+/// the consumer.  Re-entry would break the documented "header consumer invoked
+/// exactly once on every code path" contract and re-deliver to a consumer that
+/// may already be in a failed / partially-committed state.  Only a panic that
+/// unwound BEFORE the callback was ever reached (both flags false) earns the
+/// fallback, so the Java caller is never left without a header.
+///
+/// (The prior inline guard tested `!header_sent` alone, which double-invoked the
+/// consumer in the rare "callback threw, then the dispatch future panicked"
+/// edge; this predicate closes that gap and is unit-tested in
+/// `jni_impl_streaming_abort_tests.rs`.)
+pub(super) fn should_fire_fallback_header(header_sent: bool, header_failed: bool) -> bool {
+    !header_sent && !header_failed
+}
+
 /// 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.

libs/vespera-bridge/src/main/java/com/devfive/vespera/bridge/VesperaBridgeAutoConfiguration.java

@@ -138,8 +138,14 @@ public DispatchModeResolver vesperaBridgeDispatchModeResolver(VesperaBridgePrope
 
     @Bean("vesperaBridgeAsyncResponseExecutor")
     @ConditionalOnMissingBean(name = "vesperaBridgeAsyncResponseExecutor")
-    public ExecutorService vesperaBridgeAsyncResponseExecutor() {
-        int threads = Math.max(2, Math.min(4, Runtime.getRuntime().availableProcessors()));
+    public ExecutorService vesperaBridgeAsyncResponseExecutor(VesperaBridgeProperties props) {
+        // Default (asyncPoolSize <= 0) preserves the historical sizing:
+        // Math.max(2, Math.min(4, cpus)). A positive vespera.bridge.async-pool-size
+        // overrides the cap for high-concurrency async dispatch (clamped to >= 1).
+        int configured = props.getAsyncPoolSize();
+        int threads = configured > 0
+                ? configured
+                : Math.max(2, Math.min(4, Runtime.getRuntime().availableProcessors()));
         AtomicInteger seq = new AtomicInteger(1);
         ThreadFactory factory = task -> {
             Thread thread = new Thread(task, "vespera-bridge-async-response-" + seq.getAndIncrement());

libs/vespera-bridge/src/main/java/com/devfive/vespera/bridge/VesperaBridgeProperties.java

@@ -90,6 +90,16 @@ public class VesperaBridgeProperties {
      */
     private long maxBufferedRequestBytes = 0;
 
+    /**
+     * Thread count for the autoconfigured {@code vesperaBridgeAsyncResponseExecutor}
+     * — the JVM-side pool that parses the ASYNC wire response off the native
+     * completion thread.  Default {@code 0} preserves the historical sizing
+     * ({@code Math.max(2, Math.min(4, availableProcessors()))}).  Set a positive
+     * value to override the cap for high-concurrency async dispatch; the value
+     * is clamped to at least {@code 1}.
+     */
+    private int asyncPoolSize = 0;
+
     public String getAppHeader() {
         return appHeader;
     }
@@ -129,4 +139,12 @@ public long getMaxBufferedRequestBytes() {
     public void setMaxBufferedRequestBytes(long maxBufferedRequestBytes) {
         this.maxBufferedRequestBytes = maxBufferedRequestBytes;
     }
+
+    public int getAsyncPoolSize() {
+        return asyncPoolSize;
+    }
+
+    public void setAsyncPoolSize(int asyncPoolSize) {
+        this.asyncPoolSize = asyncPoolSize;
+    }
 }

libs/vespera-bridge/src/main/java/com/devfive/vespera/bridge/VesperaDirectBufferPool.java

@@ -121,7 +121,16 @@ static boolean currentThreadIsVirtual() {
         }
         try {
             return (boolean) IS_VIRTUAL.invokeExact(Thread.currentThread());
-        } catch (Throwable ignoredFallBackToPooled) {
+        } catch (RuntimeException | Error fatalMustPropagate) {
+            // JVM Errors (OutOfMemoryError, StackOverflowError, …) and runtime
+            // exceptions are never the reflective-fallback case — let them
+            // propagate instead of silently reporting "not virtual".
+            throw fatalMustPropagate;
+        } catch (Throwable reflectiveFailureFallBackToPooled) {
+            // MethodHandle.invokeExact is declared `throws Throwable`; the only
+            // residual checked failure here is a reflective/linkage problem
+            // resolving Thread.isVirtual() — fall back to the non-virtual
+            // (pooled) path, preserving the prior behavior.
             return false;
         }
     }
@@ -243,17 +252,17 @@ static ByteBuffer dispatchDirectPooled(
             // request (> cap): byte[] fallback is safe for any method
             // because no dispatch has run yet.  The reusable header buffer
             // is consumed here, before any other fillHeaderJson call.
-            return ByteBuffer.wrap(
-                    VesperaBridge.dispatchBytes(
-                            VesperaWireCodec.assembleWire(hdr.backingArray(), headerLen, bodyBytes)))
-                    .asReadOnlyBuffer();
+            byte[] wire = VesperaWireCodec.assembleWire(hdr.backingArray(), headerLen, bodyBytes);
+            VesperaWireCodec.shrinkHeaderBufferIfOversized(hdr);
+            return ByteBuffer.wrap(VesperaBridge.dispatchBytes(wire)).asReadOnlyBuffer();
         }
         ByteBuffer[] pool = directPool();
         if (pool[0].capacity() < total) {
             pool[0] = ByteBuffer.allocateDirect(grownCapacity(total));
         }
         // Consume the reusable header buffer into the pooled direct buffer.
         int written = VesperaWireCodec.assembleInto(hdr.backingArray(), headerLen, bodyBytes, pool[0]);
+        VesperaWireCodec.shrinkHeaderBufferIfOversized(hdr);
         if (written != total) {
             throw new IllegalStateException(
                     "assembleInto wrote " + written + ", expected " + total);
@@ -289,16 +298,16 @@ static ByteBuffer dispatchDirectPooled(
         int headerLen = hdr.size();
         int total = VesperaWireCodec.wireTotalLength(headerLen, bodyBytes.length);
         if (currentThreadIsVirtual || total > DIRECT_MAX_CAPACITY) {
-            return ByteBuffer.wrap(
-                    VesperaBridge.dispatchBytes(
-                            VesperaWireCodec.assembleWire(hdr.backingArray(), headerLen, bodyBytes)))
-                    .asReadOnlyBuffer();
+            byte[] wire = VesperaWireCodec.assembleWire(hdr.backingArray(), headerLen, bodyBytes);
+            VesperaWireCodec.shrinkHeaderBufferIfOversized(hdr);
+            return ByteBuffer.wrap(VesperaBridge.dispatchBytes(wire)).asReadOnlyBuffer();
         }
         ByteBuffer[] pool = directPool();
         if (pool[0].capacity() < total) {
             pool[0] = ByteBuffer.allocateDirect(grownCapacity(total));
         }
         int written = VesperaWireCodec.assembleInto(hdr.backingArray(), headerLen, bodyBytes, pool[0]);
+        VesperaWireCodec.shrinkHeaderBufferIfOversized(hdr);
         if (written != total) {
             throw new IllegalStateException(
                     "assembleInto wrote " + written + ", expected " + total);