From ab3bc97cd92b6480327029bcf018280442462af7 Mon Sep 17 00:00:00 2001 From: Pyra <92104930+pyranota@users.noreply.github.com> Date: Fri, 19 Jun 2026 13:11:35 +0000 Subject: [PATCH] feat(python, windows): enable S3 to cache wheels (#5199) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * add .minio-data to .gitignore * feat(python): make S3 cache arch specific Current schema: S3-Bucket ├── python_311 │   ├── wheel==1.0 │   └── wheel2==1.0 └── python_312 ├── wheel==1.0 └── wheel2==1.0 New schema: S3-Bucket ├── linux_aarch64 │ └── ... └── linux_x86_64 ├── python_311 │   ├── wheel==1.0 │   └── wheel2==1.0 └── python_312 ├── wheel==1.0 └── wheel2==1.0 * remove .minio-data from .gitignore * remove unneeded tracing::error * feat(python, windows): enable S3 to cache wheels * fix(python, windows): drop residual unix cfg gates on S3 cache items Post-merge, PIPTAR_UPLOAD_CHANNEL and its call sites were already cross-platform, but the types/functions they reference (PiptarUploadTask, handle_piptar_uploads, pull_from_tar, OBJECT_STORE_SETTINGS) remained unix-gated, breaking the Windows build. Co-Authored-By: Claude Opus 4.7 (1M context) --------- Co-authored-by: Claude Opus 4.7 (1M context) --- .../windmill-worker/src/python_executor.rs | 30 ++++++++++--------- 1 file changed, 16 insertions(+), 14 deletions(-) diff --git a/backend/windmill-worker/src/python_executor.rs b/backend/windmill-worker/src/python_executor.rs index c121266e34..409dc362af 100644 --- a/backend/windmill-worker/src/python_executor.rs +++ b/backend/windmill-worker/src/python_executor.rs @@ -24,8 +24,10 @@ use tokio::{ use windmill_queue::MiniPulledJob; use uuid::Uuid; -#[cfg(all(feature = "enterprise", feature = "parquet", unix))] + +#[cfg(all(feature = "enterprise", feature = "parquet"))] use windmill_common::ee_oss::{get_license_plan, LicensePlan}; + use windmill_common::{ error::{ self, @@ -72,7 +74,7 @@ lazy_static::lazy_static! { static ref EPHEMERAL_TOKEN_CMD: Option = var("EPHEMERAL_TOKEN_CMD").ok(); } -#[cfg(all(feature = "enterprise", feature = "parquet", unix))] +#[cfg(all(feature = "enterprise", feature = "parquet"))] lazy_static::lazy_static! { static ref PIPTAR_UPLOAD_CHANNEL: tokio::sync::mpsc::UnboundedSender = { let (tx, rx) = tokio::sync::mpsc::unbounded_channel(); @@ -84,14 +86,14 @@ lazy_static::lazy_static! { }; } -#[cfg(all(feature = "enterprise", feature = "parquet", unix))] +#[cfg(all(feature = "enterprise", feature = "parquet"))] #[derive(Debug)] struct PiptarUploadTask { venv_path: String, cache_dir: String, } -#[cfg(all(feature = "enterprise", feature = "parquet", unix))] +#[cfg(all(feature = "enterprise", feature = "parquet"))] async fn handle_piptar_uploads(mut rx: tokio::sync::mpsc::UnboundedReceiver) { use crate::global_cache::build_tar_and_push; use windmill_object_store::get_object_store; @@ -137,10 +139,10 @@ pub fn has_relative_imports(content: &str) -> bool { RELATIVE_IMPORT_REGEX.is_match(content) } -#[cfg(all(feature = "enterprise", feature = "parquet", unix))] +#[cfg(all(feature = "enterprise", feature = "parquet"))] use crate::global_cache::pull_from_tar; -#[cfg(all(feature = "enterprise", feature = "parquet", unix))] +#[cfg(all(feature = "enterprise", feature = "parquet"))] use windmill_object_store::OBJECT_STORE_SETTINGS; use crate::{ @@ -2388,12 +2390,12 @@ pub async fn handle_python_reqs( instant: std::time::Instant, conn: &Connection, ) { - #[cfg(not(all(feature = "enterprise", feature = "parquet", unix)))] + #[cfg(not(all(feature = "enterprise", feature = "parquet")))] { (s3_pull, s3_push) = (false, false); } - #[cfg(all(feature = "enterprise", feature = "parquet", unix))] + #[cfg(all(feature = "enterprise", feature = "parquet"))] if OBJECT_STORE_SETTINGS.read().await.is_none() { (s3_pull, s3_push) = (false, false); } @@ -2648,7 +2650,7 @@ pub async fn handle_python_reqs( let mut handles = Vec::with_capacity(total_to_install); // let mem_peak_thread_safe = Arc::new(tokio::sync::Mutex::new(0)); - #[cfg(all(feature = "enterprise", feature = "parquet", unix))] + #[cfg(all(feature = "enterprise", feature = "parquet"))] let is_not_pro = !matches!(get_license_plan().await, LicensePlan::Pro); let total_time = std::time::Instant::now(); @@ -2696,7 +2698,7 @@ pub async fn handle_python_reqs( let pids = pids.clone(); let worker_dir = worker_dir.clone(); - #[cfg(all(feature = "enterprise", feature = "parquet", unix))] + #[cfg(all(feature = "enterprise", feature = "parquet"))] let py_version = py_version.clone(); handles.push(task::spawn(async move { @@ -2714,7 +2716,7 @@ pub async fn handle_python_reqs( ); let start = std::time::Instant::now(); - #[cfg(all(feature = "enterprise", feature = "parquet", unix))] + #[cfg(all(feature = "enterprise", feature = "parquet"))] if is_not_pro { if let Some(os) = windmill_object_store::get_object_store().await { tokio::select! { @@ -2896,10 +2898,10 @@ pub async fn handle_python_reqs( } }; - #[cfg(all(feature = "enterprise", feature = "parquet", unix))] + #[cfg(all(feature = "enterprise", feature = "parquet"))] let s3_push = is_not_pro; - #[cfg(not(all(feature = "enterprise", feature = "parquet", unix)))] + #[cfg(not(all(feature = "enterprise", feature = "parquet")))] let s3_push = false; if is_sandboxing_enabled() { @@ -2953,7 +2955,7 @@ pub async fn handle_python_reqs( ) .await; - #[cfg(all(feature = "enterprise", feature = "parquet", unix))] + #[cfg(all(feature = "enterprise", feature = "parquet"))] if s3_push { // Send to upload channel for sequential processing let upload_task = PiptarUploadTask {