Fix jni

Author: owjs3901

crates/vespera/src/multipart.rs

@@ -310,8 +310,9 @@ impl IntoResponse for TypedMultipartError {
             | Self::TooManyFields { .. } => StatusCode::PAYLOAD_TOO_LARGE,
             Self::Other { .. } => StatusCode::INTERNAL_SERVER_ERROR,
         };
-        // Serialize the canonical 422 envelope (see `error_body` /
-        // module-scope `MultipartErrorEnvelope`).
+        // Serialize the canonical JSON error envelope (see `error_body` /
+        // module-scope `MultipartErrorEnvelope`); the status varies (400/413/
+        // 422/500) but the body shape is identical.
         let body = self.error_body();
         (
             status,
@@ -580,7 +581,16 @@ tokio::task_local! {
     static MULTIPART_AGGREGATE: RefCell<MultipartAggregateState>;
 }
 
-fn register_multipart_field() -> Result<(), TypedMultipartError> {
+/// Count one multipart PART against the request-wide `max_fields` limit.
+///
+/// Invoked by the derived `TryFromMultipart` loop **once per wire part** —
+/// before the field name is resolved — so EVERY part (known, unknown, or
+/// nameless) is counted exactly once.  Counting inside the per-known-field
+/// parsers instead let unknown parts in non-strict mode (the `_ => {}`
+/// dispatch arm) slip past the cap entirely, so a request with thousands of
+/// unknown parts could burn unbounded parser/boundary-scan work without ever
+/// tripping `TooManyFields`.
+pub fn register_multipart_part() -> Result<(), TypedMultipartError> {
     MULTIPART_AGGREGATE
         .try_with(|state| {
             let mut state = state.borrow_mut();
@@ -592,9 +602,8 @@ fn register_multipart_field() -> Result<(), TypedMultipartError> {
             }
             Ok(())
         })
-        // Field parsers can be unit-tested outside the extractor. In that shape
-        // there is no request aggregate to update, so per-field limits remain the
-        // only active guard instead of failing spuriously.
+        // The derived impl can be unit-tested outside the extractor scope; with
+        // no request aggregate present, counting no-ops rather than failing.
         .unwrap_or(Ok(()))
 }
 
@@ -705,7 +714,8 @@ async fn read_field_data(
     limit: Option<usize>,
     initial_capacity: usize,
 ) -> Result<(Field<'_>, Vec<u8>), TypedMultipartError> {
-    register_multipart_field()?;
+    // Part counting now happens once per part in the derived loop
+    // (`register_multipart_part`), so the field parsers no longer count.
     // Initial capacity is independent from the hard byte limit: tiny scalar
     // fields keep the 256B cap without preallocating 256B per bool/number.
     let capacity = limit.map_or(initial_capacity, |limit| initial_capacity.min(limit));
@@ -940,7 +950,8 @@ impl<S: Send + Sync> TryFromFieldWithState<S> for tempfile::NamedTempFile {
         limit_bytes: Option<usize>,
         _state: &S,
     ) -> Result<Self, TypedMultipartError> {
-        register_multipart_field()?;
+        // Part counting happens once per part in the derived loop
+        // (`register_multipart_part`); the temp-file parser no longer counts.
         // Temp-file creation AND reopen() are both blocking syscalls —
         // run them together on the blocking pool so neither stalls the
         // async worker (the reopen previously ran inline on the async

crates/vespera/tests/multipart_wire.rs

@@ -424,6 +424,35 @@ fn typed_multipart_aggregate_field_count_cap_rejected_413() {
     assert_eq!(header["status"].as_u64(), Some(413), "header={header:#}");
 }
 
+#[test]
+fn typed_multipart_unknown_fields_count_toward_max_fields() {
+    // Regression: in non-strict mode an UNKNOWN part (the generated `_ => {}`
+    // dispatch arm) must still count against `max_fields`.  Before the fix,
+    // counting happened only inside the per-known-field parsers, so a flood of
+    // unknown parts bypassed the cap entirely (a DoS-adjacent gap) and this
+    // request would instead fail later with a 400 missing-field error.  With
+    // `MAX_FIELDS = 0`, even one part — known OR unknown — must be rejected 413.
+    install_router_once();
+    let runtime = Builder::new_current_thread()
+        .enable_all()
+        .build()
+        .expect("tokio runtime");
+
+    let wire = encode_multipart_text(
+        "----UnknownFieldCountBoundary",
+        "/text-aggregate-field-count",
+        "definitely_not_a_known_field",
+        b"x",
+    );
+    let resp = dispatch_from_bytes(wire, &runtime);
+    let (header, _body) = decode_wire(&resp);
+    assert_eq!(
+        header["status"].as_u64(),
+        Some(413),
+        "an unknown multipart part must count against max_fields, got header={header:#}"
+    );
+}
+
 #[test]
 fn typed_multipart_aggregate_under_limit_passes() {
     install_router_once();

crates/vespera/tests/validated_extractor.rs

@@ -442,7 +442,7 @@ async fn multiple_per_rule_violations_all_appear_in_envelope() {
 // - Consumed by Java decoders and client libraries
 //
 // Multi-error coverage: triggers 2+ field errors to verify the full
-// envelope structure (path before message, array ordering, etc.).
+// envelope structure (message before path, array ordering, etc.).
 
 #[tokio::test]
 async fn byte_snapshot_422_envelope_multi_error() {

crates/vespera_inprocess/src/dispatch.rs

@@ -286,7 +286,11 @@ async fn finish_buffered_wire(
     // with zero reallocations.
     let header_cap = header_capacity_estimate(&headers, &metadata).max(WIRE_HEADER_RESERVE);
     let body_cap = usize::try_from(body.size_hint().exact().unwrap_or(0)).unwrap_or(0);
-    let mut out = Vec::with_capacity(4 + header_cap + body_cap);
+    // Saturating so a pathological/oversized exact body hint cannot wrap the
+    // capacity arithmetic (debug panic / release wrap → under-reserve); the
+    // common case computes the identical value, and `finish_direct_write`
+    // already uses the same saturating accounting for its overflow reporting.
+    let mut out = Vec::with_capacity(4usize.saturating_add(header_cap).saturating_add(body_cap));
     if !write_wire_header_into_vec(&mut out, status, &headers, &metadata) {
         // Unreachable for a real `HeaderMap` (4 GiB+ of header JSON); never
         // panic on the response path — emit a 500 wire response instead.

crates/vespera_inprocess/src/lib.rs

@@ -35,7 +35,7 @@
 //!                        (request)  { "v":1, "method", "path",
 //!                                     "query"?, "headers"? }
 //!                        (response) { "v":1, "status", "headers",
-//!                                     "metadata" }
+//!                                     "metadata", "validation_errors"? }
 //!    bytes 4+N..end  : raw body bytes (UTF-8 text or binary —
 //!                       no encoding applied)
 //!    ```

crates/vespera_inprocess/src/wire/hoist.rs

@@ -45,8 +45,9 @@ fn body_is_json(headers: &http::HeaderMap) -> bool {
 /// This is the **fast strict path**: the common, framework-generated envelope
 /// has all-string fields, so the plain derive parses it with no per-field
 /// visitor overhead.  A body with a wrong-typed field (`"code": 123`) fails
-/// this strict parse and is retried via [`LenientHoistEnvelope`], so the
-/// hoist stays genuinely best-effort without taxing the common case.
+/// this strict parse and is retried via the inline `serde_json::Value`
+/// fallback walk in [`try_hoist_validation_errors`], so the hoist stays
+/// genuinely best-effort without taxing the common case.
 #[derive(Deserialize)]
 struct HoistEnvelope {
     errors: Vec<HoistErrorIn>,

crates/vespera_jni/src/jni_impl.rs

@@ -402,10 +402,15 @@ pub extern "system" fn Java_com_devfive_vespera_bridge_VesperaBridge_dispatchAsy
 /// first to commit the HTTP status + response headers, then
 /// continue serving the streamed body bytes.
 ///
-/// Failure modes mirror [`Java_...dispatchBytes`]: malformed wire,
-/// version mismatch, no app registered, or Rust panic produce a
-/// regular `error_wire(...)` response (header + small body) and
-/// the `OutputStream` is **not** written to.
+/// Failure modes mirror [`Java_...dispatchBytes`]: a **pre-streaming**
+/// failure (malformed wire, version mismatch, no app registered, or a panic
+/// before the first body frame) produces a regular `error_wire(...)` response
+/// (header + small body) and the `OutputStream` is **not** written to.  A
+/// failure that occurs **after** the first body frame (the host
+/// `OutputStream` erroring mid-drain, or a body-stream error) may leave
+/// partial bytes already written to the `OutputStream`; it is still reported
+/// as a `500` `error_wire(...)` header return, so the caller must treat a
+/// `5xx` header returned after streaming has begun as a truncated response.
 #[unsafe(no_mangle)]
 pub extern "system" fn Java_com_devfive_vespera_bridge_VesperaBridge_dispatchStreaming<'local>(
     mut unowned_env: EnvUnowned<'local>,

crates/vespera_macro/src/cron_impl.rs

@@ -47,14 +47,36 @@ pub struct StoredCronInfo {
 pub static CRON_STORAGE: LazyLock<Mutex<HashMap<String, Vec<StoredCronInfo>>>> =
     LazyLock::new(|| Mutex::new(HashMap::new()));
 
-/// Append a `#[cron]` metadata entry to the current crate's bucket.
+fn same_cron_source(left: &StoredCronInfo, right: &StoredCronInfo) -> bool {
+    left.fn_name == right.fn_name
+        && left
+            .file_path
+            .as_deref()
+            .unwrap_or_default()
+            .replace('\\', "/")
+            == right
+                .file_path
+                .as_deref()
+                .unwrap_or_default()
+                .replace('\\', "/")
+}
+
+/// Replace-insert a `#[cron]` metadata entry in the current crate's bucket.
 pub fn register_cron(info: StoredCronInfo) {
-    CRON_STORAGE
+    let mut guard = CRON_STORAGE
         .lock()
-        .unwrap_or_else(std::sync::PoisonError::into_inner)
+        .unwrap_or_else(std::sync::PoisonError::into_inner);
+    let bucket = guard
         .entry(crate::schema_impl::current_crate_key())
-        .or_default()
-        .push(info);
+        .or_default();
+    if let Some(existing) = bucket
+        .iter_mut()
+        .find(|existing| same_cron_source(existing, &info))
+    {
+        *existing = info;
+    } else {
+        bucket.push(info);
+    }
 }
 
 /// Snapshot (clone) of the current crate's registered cron jobs, so the
@@ -305,6 +327,29 @@ mod tests {
         assert!(err.contains("must take no parameters"));
     }
 
+    #[test]
+    fn test_register_cron_replaces_same_file_and_function() {
+        let file_path = Some("/tmp/vespera/tasks/replaced.rs".to_string());
+        let fn_name = "__test_replace_cron".to_string();
+        register_cron(StoredCronInfo {
+            fn_name: fn_name.clone(),
+            expression: "0 */5 * * * *".to_string(),
+            file_path: file_path.clone(),
+        });
+        register_cron(StoredCronInfo {
+            fn_name: fn_name.clone(),
+            expression: "0 */10 * * * *".to_string(),
+            file_path,
+        });
+
+        let matches: Vec<_> = current_crate_crons()
+            .into_iter()
+            .filter(|entry| entry.fn_name == fn_name)
+            .collect();
+        assert_eq!(matches.len(), 1, "same source cron should replace");
+        assert_eq!(matches[0].expression, "0 */10 * * * *");
+    }
+
     // ===== Compile-time cron-syntax validation (gated by the `cron` feature) =====
 
     #[cfg(feature = "cron")]

crates/vespera_macro/src/lib.rs

@@ -121,6 +121,9 @@ pub fn derive_schema(input: TokenStream) -> TokenStream {
 
     let input = syn::parse_macro_input!(input as syn::DeriveInput);
     let (metadata, expanded) = schema_impl::process_derive_schema(&input);
+    let Some(metadata) = metadata else {
+        return TokenStream::from(expanded);
+    };
     let name = metadata.name.clone();
 
     // Register into the current crate's bucket (see `current_crate_key`).

crates/vespera_macro/src/metadata.rs

@@ -83,6 +83,13 @@ pub struct StructMetadata {
     /// Populated by `#[derive(Schema)]` to avoid AST re-parsing in `vespera!()`.
     #[serde(default, skip_serializing_if = "BTreeMap::is_empty")]
     pub field_defaults: BTreeMap<String, serde_json::Value>,
+    /// Stable source identity for proc-macro-server re-expansions.
+    ///
+    /// This is not part of the OpenAPI output. It lets `#[derive(Schema)]`
+    /// replace metadata for the same source item after an IDE edit while still
+    /// rejecting two distinct items that claim the same OpenAPI schema name.
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub source_identity: Option<String>,
 }
 
 const fn default_include_in_openapi() -> bool {
@@ -96,6 +103,7 @@ impl Default for StructMetadata {
             definition: String::new(),
             include_in_openapi: true,
             field_defaults: BTreeMap::new(),
+            source_identity: None,
         }
     }
 }
@@ -108,6 +116,7 @@ impl StructMetadata {
             definition,
             include_in_openapi: true,
             field_defaults: BTreeMap::new(),
+            source_identity: None,
         }
     }
 
@@ -118,8 +127,16 @@ impl StructMetadata {
             definition,
             include_in_openapi: false,
             field_defaults: BTreeMap::new(),
+            source_identity: None,
         }
     }
+
+    /// Attach the source identity used for same-item replacement in global storage.
+    #[must_use]
+    pub fn with_source_identity(mut self, source_identity: String) -> Self {
+        self.source_identity = Some(source_identity);
+        self
+    }
 }
 
 /// Cron job metadata