repl

Author: dieriba

backend/windmill-common/src/worker.rs

@@ -307,15 +307,16 @@ pub async fn store_suspended_pull_query(wc: &WorkerConfig) {
 }
 
 pub fn make_pull_query(tags: &[String]) -> String {
-    format_pull_query(format!(
+    let query = format_pull_query(format!(
         "SELECT id
         FROM v2_job_queue
         WHERE running = false AND tag IN ({}) AND scheduled_for <= now()
         ORDER BY priority DESC NULLS LAST, scheduled_for
         FOR UPDATE SKIP LOCKED
         LIMIT 1",
         tags.iter().map(|x| format!("'{x}'")).join(", ")
-    ))
+    ));
+    query
 }
 
 pub async fn store_pull_query(wc: &WorkerConfig) {
@@ -382,10 +383,7 @@ fn normalize_path(path: &Path) -> PathBuf {
     ret
 }
 
-pub fn is_allowed_file_location(
-    job_dir: &str,
-    user_defined_path: &str,
-) -> error::Result<PathBuf> {
+pub fn is_allowed_file_location(job_dir: &str, user_defined_path: &str) -> error::Result<PathBuf> {
     let job_dir = Path::new(job_dir);
     let user_path = PathBuf::from(user_defined_path);
 

backend/windmill-queue/src/jobs.rs

@@ -2155,13 +2155,13 @@ pub struct PulledJob {
     pub permissioned_as_folders: Option<Vec<serde_json::Value>>,
 }
 
-#[derive(Serialize, Deserialize)]
+#[derive(Debug, Serialize, Deserialize)]
 pub enum PrecomputedAgentInfo {
     Bun { local: String, remote: String },
     Python { py_version: Option<u32>, requirements: Option<String> },
 }
 
-#[derive(Serialize, Deserialize)]
+#[derive(Debug, Serialize, Deserialize)]
 pub struct JobAndPerms {
     pub job: MiniPulledJob,
     pub raw_code: Option<String>,
@@ -2313,11 +2313,11 @@ pub async fn pull(
     db: &Pool<Postgres>,
     suspend_first: bool,
     worker_name: &str,
-    query_o: Option<(String, String)>,
+    query_o: Option<&(String, String)>,
     #[cfg(feature = "benchmark")] bench: &mut BenchmarkIter,
 ) -> windmill_common::error::Result<PulledJobResult> {
     loop {
-        if let Some((query_suspended, query_no_suspend)) = query_o.as_ref() {
+        if let Some((query_suspended, query_no_suspend)) = query_o {
             let njob = {
                 let job = sqlx::query_as::<_, PulledJob>(query_suspended)
                     .bind(worker_name)
@@ -2572,7 +2572,6 @@ async fn pull_single_job_and_mark_as_running_no_concurrency_limit<'c>(
             tokio::time::sleep(tokio::time::Duration::from_secs(1)).await;
             return Ok((None, false));
         }
-
         let r = if suspend_first {
             // tracing::info!("Pulling job with query: {}", query);
             sqlx::query_as::<_, PulledJob>(&query)