fix: stop reading an array job result as wm_failure or http response (#11154)

* fix: only read wm_failure and wm_labels from an object job result

* fix: serve an array sync result as json, not a composite response

backend/windmill-api-jobs/src/execution.rs

@@ -38,8 +38,8 @@ use windmill_common::{
     FlowVersionInfo, DB,
 };
 use windmill_queue::{
-    cancel_job, get_result_and_success_by_id_from_flow, push, PushArgs, PushArgsOwned,
-    PushIsolationLevel,
+    cancel_job, get_result_and_success_by_id_from_flow, parse_result_object, push, PushArgs,
+    PushArgsOwned, PushIsolationLevel,
 };
 
 use crate::types::RunJobQuery;
@@ -374,9 +374,9 @@ pub async fn run_wait_result_internal(
 }
 
 pub fn result_to_response(result: Box<RawValue>, success: bool) -> error::Result<Response> {
-    let composite_result = serde_json::from_str::<WindmillCompositeResult>(result.get());
+    let composite_result = parse_result_object::<WindmillCompositeResult>(result.get());
     match composite_result {
-        Ok(WindmillCompositeResult {
+        Some(WindmillCompositeResult {
             windmill_status_code,
             windmill_content_type,
             windmill_headers,
@@ -1192,4 +1192,13 @@ mod result_to_response_tests {
             assert!(res.is_err(), "hop-by-hop header must be rejected: {name}");
         }
     }
+
+    #[tokio::test]
+    async fn array_result_is_not_a_composite_response() {
+        let json = r#"[201,"text/html",null,null,"<h1>hi</h1>"]"#;
+        let resp = result_to_response(raw(json), true).expect("response");
+
+        assert_eq!(resp.status(), StatusCode::OK);
+        assert_eq!(body_bytes(resp).await, json.as_bytes());
+    }
 }

backend/windmill-queue/src/jobs.rs

@@ -654,6 +654,16 @@ pub struct ResultMetadata {
     pub wm_failure: Option<String>,
 }
 
+/// Parses a marker struct out of a job result, which only an object can carry.
+/// A derived `Deserialize` also accepts an array, filling fields by position, so
+/// without the check a result like `[[], "boom"]` reads as `wm_failure: "boom"`.
+pub fn parse_result_object<T: serde::de::DeserializeOwned>(result: &str) -> Option<T> {
+    if !result.trim_start().starts_with('{') {
+        return None;
+    }
+    serde_json::from_str(result).ok()
+}
+
 /// Sentinel `error.name` we inject into a result when retagging a successful
 /// run as a failure due to `wm_failure`. Used downstream to detect that
 /// the result is already in the standard `{ error: { name, message }, ... }`
@@ -674,8 +684,7 @@ pub fn is_pre_shaped_wm_failure_result(result: &str) -> bool {
     struct NameOnly {
         name: String,
     }
-    serde_json::from_str::<Marker>(result)
-        .ok()
+    parse_result_object::<Marker>(result)
         .and_then(|m| m.error)
         .map(|e| e.name == MANUAL_FAILURE_ERROR_NAME)
         .unwrap_or(false)
@@ -721,7 +730,7 @@ impl ValidableJson for Box<RawValue> {
     }
 
     fn result_metadata(&self) -> ResultMetadata {
-        serde_json::from_str::<ResultMetadata>(self.get()).unwrap_or_default()
+        parse_result_object::<ResultMetadata>(self.get()).unwrap_or_default()
     }
 
     fn size(&self) -> usize {
@@ -774,6 +783,10 @@ impl ValidableJson for serde_json::Value {
     }
 
     fn result_metadata(&self) -> ResultMetadata {
+        // An array would decode positionally, see `parse_result_object`.
+        if !self.is_object() {
+            return ResultMetadata::default();
+        }
         serde_json::from_value::<ResultMetadata>(self.clone()).unwrap_or_default()
     }
 
@@ -7876,3 +7889,36 @@ mod git_sync_concurrency_key_tests {
         assert!(a.len() <= 255 && b.len() <= 255);
     }
 }
+
+#[cfg(test)]
+mod result_metadata_tests {
+    use super::{ResultMetadata, ValidableJson};
+    use serde_json::value::RawValue;
+
+    fn from_raw(json: &str) -> ResultMetadata {
+        RawValue::from_string(json.to_string())
+            .unwrap()
+            .result_metadata()
+    }
+
+    fn from_value(json: &str) -> ResultMetadata {
+        serde_json::from_str::<serde_json::Value>(json)
+            .unwrap()
+            .result_metadata()
+    }
+
+    #[test]
+    fn array_result_carries_no_markers() {
+        for json in [r#"[["label"], "boom"]"#, r#"[null, "boom"]"#] {
+            for meta in [from_raw(json), from_value(json)] {
+                assert!(
+                    meta.wm_labels.is_none() && meta.wm_failure.is_none(),
+                    "{json}"
+                );
+            }
+        }
+        let meta = from_raw(r#"{"wm_labels": ["label"], "wm_failure": "boom"}"#);
+        assert_eq!(meta.wm_labels, Some(vec!["label".to_string()]));
+        assert_eq!(meta.wm_failure.as_deref(), Some("boom"));
+    }
+}

backend/windmill-worker/src/result_processor.rs

@@ -32,8 +32,8 @@ use windmill_common::bench::{BenchmarkInfo, BenchmarkIter};
 
 use windmill_queue::{
     append_logs, asset_dispatch, get_mini_completed_job, is_pre_shaped_wm_failure_result,
-    CanceledBy, FlowRunners, JobCompleted, MiniCompletedJob, MiniPulledJob, ValidableJson,
-    WrappedError, INIT_SCRIPT_TAG, MANUAL_FAILURE_ERROR_NAME,
+    parse_result_object, CanceledBy, FlowRunners, JobCompleted, MiniCompletedJob, MiniPulledJob,
+    ValidableJson, WrappedError, INIT_SCRIPT_TAG, MANUAL_FAILURE_ERROR_NAME,
 };
 
 use serde_json::{json, value::RawValue, Value};
@@ -72,13 +72,11 @@ struct NestedErrorMessage {
 /// named `name`/`message`), and we want OTel to record the ManualFailure
 /// rather than the user's sibling fields.
 fn extract_error_message(raw: &str) -> Option<ErrorMessage> {
-    let nested = serde_json::from_str::<NestedErrorMessage>(raw)
-        .ok()
-        .map(|n| n.error);
+    let nested = parse_result_object::<NestedErrorMessage>(raw).map(|n| n.error);
     if matches!(&nested, Some(em) if em.name == MANUAL_FAILURE_ERROR_NAME) {
         return nested;
     }
-    if let Ok(em) = serde_json::from_str::<ErrorMessage>(raw) {
+    if let Some(em) = parse_result_object::<ErrorMessage>(raw) {
         return Some(em);
     }
     nested