fix: offload php signature parsing from async workers (#11027)

* fix: reduce php parser stack use in debug workers

* test: document php stack regression expression depth

* fix: offload php signature parsing from async workers

* fix: address php parser review nits

backend/tests/worker.rs

@@ -3091,16 +3091,16 @@ async fn test_php_job(db: Pool<Postgres>) -> anyhow::Result<()> {
     let server = ApiServer::start(db.clone()).await?;
     let port = server.addr.port();
 
-    let content = r#"
+    let content = r#"// schema_validation
 <?php
 
-function main(string $name): string {
-    return "hello " . $name;
+function main(string $name, string $prefix = "hello "): string {
+    return $prefix . $name;
 }
 "#
     .to_owned();
 
-    let result = RunJob::from(JobPayload::Code(RawCode {
+    let code = RawCode {
         hash: None,
         content,
         path: None,
@@ -3114,14 +3114,28 @@ function main(string $name): string {
         debouncing_settings: windmill_common::runnable_settings::DebouncingSettings::default(),
         modules: None,
         tag: None,
-    }))
-    .arg("name", json!("world"))
-    .run_until_complete(&db, false, port)
-    .await
-    .json_result()
-    .unwrap();
+    };
+    let completed = RunJob::from(JobPayload::Code(code.clone()))
+        .arg("name", json!("world"))
+        .run_until_complete(&db, false, port)
+        .await;
 
-    assert_eq!(result, serde_json::json!("hello world"));
+    assert!(completed.success, "{:?}", completed.result);
+    assert_eq!(
+        completed.json_result().unwrap(),
+        serde_json::json!("hello world")
+    );
+
+    let invalid = RunJob::from(JobPayload::Code(code))
+        .arg("name", json!(42))
+        .run_until_complete(&db, false, port)
+        .await;
+    // PHP coerces numbers to strings; rejection proves inferred validation ran.
+    assert!(!invalid.success, "{:?}", invalid.result);
+    assert!(invalid.json_result().unwrap()["error"]["message"]
+        .as_str()
+        .unwrap()
+        .contains("Argument `name` should be a string"));
     Ok(())
 }
 

backend/windmill-worker/src/ai/utils.rs

@@ -26,11 +26,11 @@ use windmill_queue::{flow_status::get_step_of_flow_status, MiniPulledJob};
 
 use crate::parse_sig_of_lang;
 
-pub fn parse_raw_script_schema(
+pub async fn parse_raw_script_schema(
     content: &str,
     language: &ScriptLang,
 ) -> Result<Box<RawValue>, Error> {
-    let main_arg_signature = parse_sig_of_lang(content, Some(&language), None)?
+    let main_arg_signature = parse_sig_of_lang(content, Some(&language), None).await?
         .ok_or_else(|| Error::BadConfig(format!(
             "Cannot parse signature for language {:?}. The language parser may not be enabled in this build.",
             language

backend/windmill-worker/src/ai_executor.rs

@@ -621,7 +621,7 @@ pub async fn handle_ai_agent_job(
                     (schema, input_transforms, derived_description)
                 }
                 FlowModuleValue::RawScript { content, language, input_transforms, .. } => {
-                    let schema = Some(parse_raw_script_schema(&content, &language)?);
+                    let schema = Some(parse_raw_script_schema(&content, &language).await?);
                     (schema, input_transforms, None)
                 }
                 FlowModuleValue::AIAgent { input_transforms, .. } => {

backend/windmill-worker/src/php_executor.rs

@@ -14,7 +14,7 @@ use windmill_common::{
 };
 use windmill_queue::MiniPulledJob;
 
-use windmill_parser::Typ;
+use windmill_parser::{MainArgSignature, Typ};
 use windmill_queue::{append_logs, CanceledBy};
 
 use crate::{
@@ -40,6 +40,43 @@ lazy_static::lazy_static! {
 
 const COMPOSER_LOCK_SPLIT: &str = "\nLOCK\n";
 
+static PHP_PARSER_SLOT: tokio::sync::Semaphore = tokio::sync::Semaphore::const_new(1);
+
+pub(crate) async fn parse_php_signature(
+    code: &str,
+    main_override: Option<String>,
+) -> Result<MainArgSignature> {
+    parse_php_signature_with_slot(code, main_override, &PHP_PARSER_SLOT).await
+}
+
+async fn parse_php_signature_with_slot(
+    code: &str,
+    main_override: Option<String>,
+    slot: &'static tokio::sync::Semaphore,
+) -> Result<MainArgSignature> {
+    let acquire = slot.acquire();
+    tokio::pin!(acquire);
+    // Retain the acquisition across the warning to preserve its FIFO queue position.
+    let permit = tokio::select! {
+        permit = &mut acquire => permit,
+        _ = tokio::time::sleep(std::time::Duration::from_secs(1)) => {
+            tracing::warn!("Waiting over a second for PHP signature parser capacity");
+            acquire.await
+        }
+    }
+    .map_err(to_anyhow)?;
+    let code = code.to_owned();
+    tokio::task::spawn_blocking(move || {
+        // Parsing walks the entire AST. Keep its stack off async workers and retain
+        // the process-wide CPU limit even if the awaiting job is cancelled.
+        let _permit = permit;
+        windmill_parser_php::parse_php_signature(&code, main_override)
+    })
+    .await
+    .map_err(|e| error::Error::internal_err(format!("PHP signature parsing task failed: {e}")))?
+    .map_err(Into::into)
+}
+
 pub fn parse_php_imports(code: &str) -> anyhow::Result<Option<String>> {
     let find_requirements = code
         .lines()
@@ -333,11 +370,9 @@ pub async fn handle_php_job(
     let main_override = job.script_entrypoint_override.as_deref();
 
     let write_wrapper_f = async {
-        let args = windmill_parser_php::parse_php_signature(
-            inner_content,
-            main_override.map(ToString::to_string),
-        )?
-        .args;
+        let args = parse_php_signature(inner_content, main_override.map(ToString::to_string))
+            .await?
+            .args;
 
         let args_to_include = args
             .iter()
@@ -491,3 +526,51 @@ try {{
     .await?;
     read_result(job_dir, None).await
 }
+
+#[cfg(test)]
+mod tests {
+    use super::parse_php_signature_with_slot;
+
+    #[test]
+    fn cancelled_parse_retains_slot_until_blocking_work_finishes() {
+        static SLOT: tokio::sync::Semaphore = tokio::sync::Semaphore::const_new(1);
+        let runtime = tokio::runtime::Builder::new_current_thread()
+            .max_blocking_threads(1)
+            .enable_all()
+            .build()
+            .unwrap();
+        let (release_tx, release_rx) = std::sync::mpsc::channel();
+        let (started_tx, started_rx) = tokio::sync::oneshot::channel();
+        let blocker = runtime.spawn_blocking(move || {
+            started_tx.send(()).unwrap();
+            let _ = release_rx.recv();
+        });
+
+        runtime.block_on(async {
+            started_rx.await.unwrap();
+            let parse = tokio::spawn(parse_php_signature_with_slot(
+                "<?php function main() {}",
+                None,
+                &SLOT,
+            ));
+            tokio::time::timeout(std::time::Duration::from_secs(5), async {
+                while SLOT.available_permits() != 0 {
+                    tokio::task::yield_now().await;
+                }
+            })
+            .await
+            .unwrap();
+
+            parse.abort();
+            assert!(parse.await.unwrap_err().is_cancelled());
+            assert!(SLOT.try_acquire().is_err());
+
+            release_tx.send(()).unwrap();
+            blocker.await.unwrap();
+            let _permit = tokio::time::timeout(std::time::Duration::from_secs(5), SLOT.acquire())
+                .await
+                .unwrap()
+                .unwrap();
+        });
+    }
+}

backend/windmill-worker/src/worker.rs

@@ -5159,7 +5159,7 @@ async fn try_validate_schema(
                             code,
                             language,
                             job.script_entrypoint_override.clone(),
-                        )? {
+                        ).await? {
                             Ok(Some(schema_validator_from_main_arg_sig(&sig)))
                         } else {
                             Err(anyhow!("Job was expected to validate the arguments schema, but no schema was provided and couldn't be inferred from the script for language `{language:?}`. Try removing schema validation for this job").into())
@@ -7051,7 +7051,7 @@ mount {{
     result
 }
 
-pub fn parse_sig_of_lang(
+pub async fn parse_sig_of_lang(
     code: &str,
     language: Option<&ScriptLang>,
     main_override: Option<String>,
@@ -7086,10 +7086,9 @@ pub fn parse_sig_of_lang(
             ScriptLang::DuckDb => Some(windmill_parser_sql::parse_duckdb_sig(code)?),
             ScriptLang::OracleDB => Some(windmill_parser_sql::parse_oracledb_sig(code)?),
             #[cfg(feature = "php")]
-            ScriptLang::Php => Some(windmill_parser_php::parse_php_signature(
-                code,
-                main_override,
-            )?),
+            ScriptLang::Php => {
+                Some(crate::php_executor::parse_php_signature(code, main_override).await?)
+            }
             #[cfg(not(feature = "php"))]
             ScriptLang::Php => None,
             #[cfg(feature = "rust")]