diff --git a/backend/tests/worker.rs b/backend/tests/worker.rs index 419be6fdf3..22f95aefb8 100644 --- a/backend/tests/worker.rs +++ b/backend/tests/worker.rs @@ -3091,16 +3091,16 @@ async fn test_php_job(db: Pool) -> anyhow::Result<()> { let server = ApiServer::start(db.clone()).await?; let port = server.addr.port(); - let content = r#" + let content = r#"// schema_validation Result, 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 diff --git a/backend/windmill-worker/src/ai_executor.rs b/backend/windmill-worker/src/ai_executor.rs index 2c29721ef6..cd17c57f59 100644 --- a/backend/windmill-worker/src/ai_executor.rs +++ b/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, .. } => { diff --git a/backend/windmill-worker/src/php_executor.rs b/backend/windmill-worker/src/php_executor.rs index 1ede0933cb..03328b82cb 100644 --- a/backend/windmill-worker/src/php_executor.rs +++ b/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, +) -> Result { + parse_php_signature_with_slot(code, main_override, &PHP_PARSER_SLOT).await +} + +async fn parse_php_signature_with_slot( + code: &str, + main_override: Option, + slot: &'static tokio::sync::Semaphore, +) -> Result { + 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> { 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( + ", main_override: Option, @@ -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")]