diff --git a/backend/Cargo.lock b/backend/Cargo.lock index 3220347e6f..012e435173 100644 --- a/backend/Cargo.lock +++ b/backend/Cargo.lock @@ -1499,9 +1499,9 @@ dependencies = [ [[package]] name = "chrono-tz" -version = "0.8.4" +version = "0.8.5" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "e23185c0e21df6ed832a12e2bda87c7d1def6842881fb634a8511ced741b0d76" +checksum = "91d7b79e99bfaa0d47da0687c43aa3b7381938a62ad3a6498599039321f660b7" dependencies = [ "chrono", "chrono-tz-build", @@ -1551,9 +1551,9 @@ dependencies = [ [[package]] name = "clap" -version = "4.4.11" +version = "4.4.12" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "bfaff671f6b22ca62406885ece523383b9b64022e341e53e009a62ebc47a45f2" +checksum = "dcfab8ba68f3668e89f6ff60f5b205cea56aa7b769451a59f34b8682f51c056d" dependencies = [ "clap_builder", "clap_derive", @@ -1561,9 +1561,9 @@ dependencies = [ [[package]] name = "clap_builder" -version = "4.4.11" +version = "4.4.12" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "a216b506622bb1d316cd51328dce24e07bdff4a6128a47c7e7fad11878d5adbb" +checksum = "fb7fb5e4e979aec3be7791562fcba452f94ad85e954da024396433e0e25a79e9" dependencies = [ "anstream", "anstyle", @@ -3742,6 +3742,15 @@ dependencies = [ "syn 2.0.43", ] +[[package]] +name = "itertools" +version = "0.9.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "284f18f85651fe11e8a991b2adb42cb078325c996ed026d994719efcfca1d54b" +dependencies = [ + "either", +] + [[package]] name = "itertools" version = "0.10.5" @@ -3813,6 +3822,15 @@ dependencies = [ "simple_asn1", ] +[[package]] +name = "keccak" +version = "0.1.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8f6d5ed8676d904364de097082f4e7d240b571b67989ced0240f08b7f966f940" +dependencies = [ + "cpufeatures", +] + [[package]] name = "keyed_priority_queue" version = "0.4.2" @@ -4104,9 +4122,9 @@ dependencies = [ [[package]] name = "mail-send" -version = "0.4.3" +version = "0.4.4" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "e373c9ec9178397af6ea2964568e49e1d486b2d80d17b06aa19da75566416422" +checksum = "4e38e7453c5cb25cbeaa251e480c74592ec121360a6bcb822928926bf69e3d1e" dependencies = [ "base64 0.21.5", "gethostname", @@ -4121,9 +4139,9 @@ dependencies = [ [[package]] name = "malachite" -version = "0.4.4" +version = "0.3.2" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "220cb36c52aa6eff45559df497abe0e2a4c1209f92279a746a399f622d7b95c7" +checksum = "f6cf7f4730c30071ba374fac86ad35b1cb7a0716f774737768667ea3fa1828e3" dependencies = [ "malachite-base", "malachite-nz", @@ -4132,19 +4150,22 @@ dependencies = [ [[package]] name = "malachite-base" -version = "0.4.4" +version = "0.3.2" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "6538136c5daf04126d6be4899f7fe4879b7f8de896dd1b4210fe6de5b94f2555" +checksum = "2b06bfa98a4b4802af5a4263b4ad4660e28e51e8490f6354eb9336c70767e1c5" dependencies = [ - "itertools 0.11.0", + "itertools 0.9.0", + "rand 0.7.3", + "rand_chacha 0.2.2", "ryu", + "sha3", ] [[package]] name = "malachite-bigint" -version = "0.1.1" +version = "0.1.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "76c3eca3b5df299486144c8423c45c24bdf9e82e2452c8a1eeda547c4d8b5d41" +checksum = "8a5110aee54537b0cef214efbebdd7df79b7408db8eef4f6a4b6db9d0d8fc01b" dependencies = [ "derive_more", "malachite", @@ -4155,22 +4176,22 @@ dependencies = [ [[package]] name = "malachite-nz" -version = "0.4.4" +version = "0.3.2" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "5f0b05577b7a3f09433106460b10304f97fc572f0baabf6640e6cb1e23f5fc52" +checksum = "c89e21c64b7af5be3dc8cef16f786243faf59459fe4ba93b44efdeb264e5ade4" dependencies = [ "embed-doc-image", - "itertools 0.11.0", + "itertools 0.9.0", "malachite-base", ] [[package]] name = "malachite-q" -version = "0.4.4" +version = "0.3.2" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "a1cfdb4016292e6acd832eaee261175f3af8bbee62afeefe4420ebce4c440cb5" +checksum = "3755e541d5134b5016594c9043094172c4dda9259b3ce824a7b8101941850360" dependencies = [ - "itertools 0.11.0", + "itertools 0.9.0", "malachite-base", "malachite-nz", ] @@ -4219,9 +4240,9 @@ dependencies = [ [[package]] name = "memchr" -version = "2.6.4" +version = "2.7.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "f665ee40bc4a3c5590afb1e9677db74a508659dfd71e126420da8274909a0167" +checksum = "523dc4f511e55ab87b694dc30d0f820d60906ef06413f93d4d7a1385599cc149" [[package]] name = "memmap2" @@ -7000,6 +7021,18 @@ dependencies = [ "digest 0.10.7", ] +[[package]] +name = "sha3" +version = "0.9.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f81199417d4e5de3f04b1e871023acea7389672c4135918f05aa9cbf2f2fa809" +dependencies = [ + "block-buffer 0.9.0", + "digest 0.9.0", + "keccak", + "opaque-debug", +] + [[package]] name = "sharded-slab" version = "0.1.7" @@ -7116,9 +7149,9 @@ dependencies = [ [[package]] name = "smtp-proto" -version = "0.1.1" +version = "0.1.2" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "d4b756ac662e92a0e5b360349bea5f0b0784d4be4541eff2972049dfdfd7f862" +checksum = "0f7278d054ed6844babe3d760d03ccae24e26207c3c27b68d8385a03d3324619" [[package]] name = "snafu" @@ -8156,15 +8189,15 @@ checksum = "cfb5fa503293557c5158bd215fdc225695e567a77e453f5d4452a50a193969bd" [[package]] name = "tempfile" -version = "3.8.1" +version = "3.9.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "7ef1adac450ad7f4b3c28589471ade84f25f731a7a0fe30d71dfa9f60fd808e5" +checksum = "01ce4141aa927a6d1bd34a041795abd0db1cccba5d5f24b009f694bdf3a1f3fa" dependencies = [ "cfg-if", "fastrand 2.0.1", "redox_syscall 0.4.1", "rustix 0.38.28", - "windows-sys 0.48.0", + "windows-sys 0.52.0", ] [[package]] @@ -9452,7 +9485,6 @@ dependencies = [ "async-recursion", "async-stripe", "async_zip", - "aws-config", "aws-sdk-s3", "axum", "base64 0.21.5", @@ -9550,6 +9582,8 @@ name = "windmill-common" version = "1.232.0" dependencies = [ "anyhow", + "aws-config", + "aws-sdk-s3", "axum", "chrono", "cron", @@ -9996,9 +10030,9 @@ checksum = "dff9641d1cd4be8d1a070daf9e3773c5f67e78b4d9d42263020c057706765c04" [[package]] name = "winnow" -version = "0.5.30" +version = "0.5.31" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "9b5c3db89721d50d0e2a673f5043fc4722f76dcc352d7b1ab8b8288bed4ed2c5" +checksum = "97a4882e6b134d6c28953a387571f1acdd3496830d5e36c5e3a1075580ea641c" dependencies = [ "memchr", ] @@ -10024,9 +10058,9 @@ dependencies = [ [[package]] name = "xattr" -version = "1.1.3" +version = "1.2.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "a7dae5072fe1f8db8f8d29059189ac175196e410e40ba42d5d4684ae2f750995" +checksum = "914566e6413e7fa959cc394fb30e563ba80f3541fbd40816d4c05a0fc3f2a0f1" dependencies = [ "libc", "linux-raw-sys 0.4.12", diff --git a/backend/Cargo.toml b/backend/Cargo.toml index c30b7c51a3..4fea5d9c43 100644 --- a/backend/Cargo.toml +++ b/backend/Cargo.toml @@ -208,4 +208,7 @@ candle-transformers = "0.3.0" candle-nn = "0.3.0" tiberius = { version = "0.12.2", default-features = false, features = ["rustls", "tds73", "chrono"] } aws-sdk-s3 = "1.5.0" -aws-config = "1.0.3" \ No newline at end of file +aws-config = "1.0.3" +polars = { version = "0.35.2", features = ["lazy", "parquet", "aws", "csv", "dtype-full"] } +polars-io = { version = "0.35.2", features = ["csv"] } +object_store = { version = "0.8.0", features = ["aws"] } diff --git a/backend/windmill-api/Cargo.toml b/backend/windmill-api/Cargo.toml index 5e28418e1e..37d205c337 100644 --- a/backend/windmill-api/Cargo.toml +++ b/backend/windmill-api/Cargo.toml @@ -84,7 +84,6 @@ candle-core.workspace = true candle-transformers.workspace = true candle-nn.workspace = true aws-sdk-s3.workspace = true -aws-config.workspace = true -polars = { version = "0.35.2", features = ["lazy", "parquet", "aws", "csv", "dtype-full"] } -polars-io = { version = "0.35.2", features = ["csv"] } -object_store = { version = "0.8.0", features = ["aws"] } +polars.workspace = true +polars-io.workspace = true +object_store.workspace = true diff --git a/backend/windmill-api/src/job_helpers.rs b/backend/windmill-api/src/job_helpers.rs index f0cc5023b3..e797e5f098 100644 --- a/backend/windmill-api/src/job_helpers.rs +++ b/backend/windmill-api/src/job_helpers.rs @@ -1,12 +1,8 @@ use std::{cmp, time::Duration}; -use crate::{ - db::DB, resources::get_resource_value_interpolated_internal, users::Tokened, - workspaces::LargeFileStorage, -}; +use crate::{db::DB, resources::get_resource_value_interpolated_internal, users::Tokened}; use anyhow::Context; use aws_sdk_s3::{ - config::{BehaviorVersion, Credentials, Region}, presigning::PresigningConfig, primitives::ByteStream, types::{CompletedMultipartUpload, CompletedPart}, @@ -32,7 +28,11 @@ use polars::{ }; use serde::{Deserialize, Serialize}; use tower_http::cors::{Any, CorsLayer}; -use windmill_common::{db::UserDB, error}; +use windmill_common::{ + db::UserDB, + error, + s3_helpers::{build_s3_client, render_endpoint, LargeFileStorage, S3Resource}, +}; use crate::db::ApiAuthed; @@ -87,23 +87,6 @@ pub fn workspaced_service() -> Router { ) } -#[derive(Debug, Deserialize, Serialize, Clone)] -struct S3Resource { - #[serde(rename = "bucket")] - bucket: String, - region: String, - #[serde(rename = "endPoint")] - endpoint: String, - #[serde(rename = "useSSL")] - use_ssl: bool, - #[serde(rename = "accessKey")] - access_key: Option, - #[serde(rename = "secretKey")] - secret_key: Option, - #[serde(rename = "pathStyle")] - path_style: bool, -} - #[derive(Deserialize)] struct DuckdbConnectionSettingsQuery { s3_resource: S3Resource, @@ -346,7 +329,10 @@ async fn test_connection( .max_keys(1) .send() .await - .map_err(|err| error::Error::InternalErr(err.to_string()))?; + .map_err(|err| { + tracing::error!("Error testing connection to S3 bucket: {:?}", err); + error::Error::InternalErr(err.to_string()) + })?; return Ok(Json(())); } @@ -911,6 +897,11 @@ async fn multipart_upload_s3_file( })); } +#[derive(Deserialize)] +pub struct S3Object { + pub s3: String, +} + async fn get_workspace_s3_resource<'c>( authed: &ApiAuthed, user_db: &UserDB, @@ -983,30 +974,6 @@ async fn get_s3_resource<'c>( return Ok(Some(s3_resource)); } -fn build_s3_client(s3_resource_ref: &S3Resource) -> aws_sdk_s3::Client { - let s3_resource = s3_resource_ref.clone(); - - let endpoint = render_endpoint(&s3_resource); - let mut s3_config_builder = aws_sdk_s3::Config::builder() - .endpoint_url(endpoint) - .behavior_version(BehaviorVersion::latest()) - .region(Region::new(s3_resource.region)); - if s3_resource.access_key.is_some() { - s3_config_builder = s3_config_builder.credentials_provider(Credentials::new( - s3_resource.access_key.unwrap_or_default(), - s3_resource.secret_key.unwrap_or_default(), - None, - None, - "s3_storage", - )); - } - if s3_resource.path_style { - s3_config_builder = s3_config_builder.force_path_style(true); - } - let s3_config = s3_config_builder.build(); - return aws_sdk_s3::Client::from_conf(s3_config); -} - fn build_polars_s3_config(s3_resource_ref: &S3Resource) -> CloudOptions { let s3_resource = s3_resource_ref.to_owned(); let mut s3_configs: Vec<(AmazonS3ConfigKey, String)> = vec![ @@ -1034,16 +1001,6 @@ fn build_polars_s3_config(s3_resource_ref: &S3Resource) -> CloudOptions { return CloudOptions::default().with_aws(s3_configs); } -fn render_endpoint(s3_resource: &S3Resource) -> String { - if s3_resource.endpoint.starts_with("http://") || s3_resource.endpoint.starts_with("https://") { - s3_resource.endpoint.clone() - } else if s3_resource.use_ssl { - format!("https://{}", s3_resource.endpoint) - } else { - format!("http://{}", s3_resource.endpoint) - } -} - async fn read_s3_text_object_head( s3_client: &aws_sdk_s3::Client, s3_bucket: &str, diff --git a/backend/windmill-api/src/workspaces.rs b/backend/windmill-api/src/workspaces.rs index fd7650815d..fe36a655be 100644 --- a/backend/windmill-api/src/workspaces.rs +++ b/backend/windmill-api/src/workspaces.rs @@ -40,6 +40,7 @@ use stripe::CustomerId; use uuid::Uuid; use windmill_audit::{audit_log, ActionKind}; use windmill_common::db::UserDB; +use windmill_common::s3_helpers::LargeFileStorage; use windmill_common::schedule::Schedule; use windmill_common::users::username_to_permissioned_as; use windmill_common::worker::CLOUD_HOSTED; @@ -281,18 +282,6 @@ pub struct EditErrorHandler { pub error_handler_muted_on_cancel: Option, } -#[derive(Serialize, Deserialize, Debug)] -#[serde(tag = "type")] -pub enum LargeFileStorage { - S3Storage(S3Storage), - // TODO: Add a filesystem type here in the future if needed -} - -#[derive(Serialize, Deserialize, Debug)] -pub struct S3Storage { - pub s3_resource_path: String, -} - async fn list_pending_invites( authed: ApiAuthed, Extension(user_db): Extension, diff --git a/backend/windmill-common/Cargo.toml b/backend/windmill-common/Cargo.toml index 5bff350550..981b4989e1 100644 --- a/backend/windmill-common/Cargo.toml +++ b/backend/windmill-common/Cargo.toml @@ -48,4 +48,6 @@ itertools.workspace = true regex.workspace = true git-version.workspace = true cron.workspace = true -tracing-loki = { version = "^0", optional = true } \ No newline at end of file +tracing-loki = { version = "^0", optional = true } +aws-sdk-s3.workspace = true +aws-config.workspace = true \ No newline at end of file diff --git a/backend/windmill-common/src/lib.rs b/backend/windmill-common/src/lib.rs index 834b8e2cd5..f103bf46f2 100644 --- a/backend/windmill-common/src/lib.rs +++ b/backend/windmill-common/src/lib.rs @@ -26,6 +26,7 @@ pub mod global_settings; pub mod jobs; pub mod more_serde; pub mod oauth2; +pub mod s3_helpers; pub mod schedule; pub mod scripts; pub mod server; diff --git a/backend/windmill-common/src/s3_helpers.rs b/backend/windmill-common/src/s3_helpers.rs new file mode 100644 index 0000000000..148e5288d9 --- /dev/null +++ b/backend/windmill-common/src/s3_helpers.rs @@ -0,0 +1,96 @@ +use aws_config::{BehaviorVersion, Region}; +use aws_sdk_s3::config::Credentials; +use serde::{Deserialize, Serialize}; + +#[derive(Serialize, Deserialize, Debug)] +#[serde(tag = "type")] +pub enum LargeFileStorage { + S3Storage(S3Storage), + // TODO: Add a filesystem type here in the future if needed +} + +#[derive(Serialize, Deserialize, Debug)] +pub struct S3Storage { + pub s3_resource_path: String, +} + +#[derive(Debug, Deserialize, Serialize, Clone)] +pub struct S3Resource { + #[serde(rename = "bucket")] + pub bucket: String, + pub region: String, + #[serde(rename = "endPoint")] + pub endpoint: String, + #[serde(rename = "useSSL")] + pub use_ssl: bool, + #[serde(rename = "accessKey")] + pub access_key: Option, + #[serde(rename = "secretKey")] + pub secret_key: Option, + #[serde(rename = "pathStyle")] + pub path_style: bool, + pub port: Option, +} + +#[derive(Deserialize, Clone)] +pub struct S3Object { + pub s3: String, +} + +pub async fn get_etag_or_empty(s3_resource: &S3Resource, s3_object: S3Object) -> Option { + let s3_client = build_s3_client(s3_resource); + + let s3_bucket = s3_resource.bucket.clone(); + let s3_key = s3_object.s3; + let s3_object_metadata = s3_client + .head_object() + .bucket(s3_bucket) + .key(s3_key) + .send() + .await; + s3_object_metadata + .ok() + .map(|res| res.e_tag().map(|et| et.to_string())) + .flatten() +} + +pub fn render_endpoint(s3_resource: &S3Resource) -> String { + let url_with_prefix = if s3_resource.endpoint.starts_with("http://") + || s3_resource.endpoint.starts_with("https://") + { + s3_resource.endpoint.clone() + } else if s3_resource.use_ssl { + format!("https://{}", s3_resource.endpoint) + } else { + format!("http://{}", s3_resource.endpoint) + }; + if s3_resource.port.is_some() { + format!("{}:{}", url_with_prefix, s3_resource.port.unwrap()) + } else { + url_with_prefix + } +} + +pub fn build_s3_client(s3_resource_ref: &S3Resource) -> aws_sdk_s3::Client { + let s3_resource = s3_resource_ref.clone(); + + let endpoint = render_endpoint(&s3_resource); + let mut s3_config_builder = aws_sdk_s3::Config::builder() + .endpoint_url(endpoint) + .behavior_version(BehaviorVersion::latest()) + .region(Region::new(s3_resource.region)); + if s3_resource.access_key.is_some() { + s3_config_builder = s3_config_builder.credentials_provider(Credentials::new( + s3_resource.access_key.unwrap_or_default(), + s3_resource.secret_key.unwrap_or_default(), + None, + None, + "s3_storage", + )); + } + if s3_resource.path_style { + s3_config_builder = s3_config_builder.force_path_style(true); + } + let s3_config = s3_config_builder.build(); + return aws_sdk_s3::Client::from_conf(s3_config); +} diff --git a/backend/windmill-worker/src/common.rs b/backend/windmill-worker/src/common.rs index 76c5dd3538..4d9da070d6 100644 --- a/backend/windmill-worker/src/common.rs +++ b/backend/windmill-worker/src/common.rs @@ -3,12 +3,13 @@ use itertools::Itertools; use nix::sys::signal::{self, Signal}; use nix::unistd::Pid; use regex::Regex; -use serde::Serialize; +use serde::{Deserialize, Serialize}; use serde_json::value::RawValue; use serde_json::{json, Value}; use sqlx::{Pool, Postgres}; use tokio::process::Command; use tokio::{fs::File, io::AsyncReadExt}; +use windmill_common::s3_helpers::{get_etag_or_empty, LargeFileStorage, S3Object, S3Resource}; use windmill_common::worker::{CLOUD_HOSTED, WORKER_CONFIG}; use windmill_common::{ error::{self, Error}, @@ -31,7 +32,7 @@ use std::{ use tracing::{trace_span, Instrument}; use uuid::Uuid; -use windmill_common::variables; +use windmill_common::{variables, DB}; use tokio::{ io::{AsyncBufReadExt, AsyncWriteExt, BufReader}, @@ -841,13 +842,26 @@ fn append_with_limit(dst: &mut String, src: &str, limit: &mut usize, remove_x00: } } -pub fn hash_args(v: &Option>>>) -> String { +pub async fn hash_args( + db: &DB, + client: &AuthedClient, + workspace_id: &str, + job_id: &Uuid, + v: &Option>>>, +) -> String { if let Some(vs) = v { let mut dh = DefaultHasher::new(); let hm = &vs.0; for k in hm.keys().sorted() { k.hash(&mut dh); - hm.get(k).unwrap().get().hash(&mut dh); + let arg_value = hm.get(k).unwrap(); + let arg_additions = + arg_value_hash_additions(db, client, workspace_id, job_id, hm.get(k).unwrap()) + .await; + arg_value.get().hash(&mut dh); + for (_, arg_addition) in arg_additions { + arg_addition.hash(&mut dh); + } } hex::encode(dh.finish().to_be_bytes()) } else { @@ -855,21 +869,152 @@ pub fn hash_args(v: &Option>>>) } } -#[derive(Serialize)] -struct StoreCachedResource<'a> { - expire: i64, - value: &'a RawValue, +async fn get_workspace_s3_resource_path( + db: &DB, + client: &AuthedClient, + workspace_id: &str, + job_id: &Uuid, +) -> Option { + let raw_lfs_opt = sqlx::query_scalar!( + "SELECT large_file_storage FROM workspace_settings WHERE workspace_id = $1", + workspace_id + ) + .fetch_one(db) + .await + .ok() + .flatten() + .map(|val| serde_json::from_value::(val).ok()) + .flatten(); + + if let Some(LargeFileStorage::S3Storage(s3_storage)) = raw_lfs_opt { + let resource_path = s3_storage.s3_resource_path.trim_start_matches("$res:"); + client + .get_resource_value_interpolated::(&resource_path, Some(job_id.to_string())) + .await + .ok() + } else { + return None; + } } -pub async fn save_in_cache<'a>( +async fn arg_value_hash_additions( + db: &DB, + client: &AuthedClient, + workspace_id: &str, + job_id: &Uuid, + raw_value: &Box, +) -> HashMap { + let mut result: HashMap = HashMap::new(); + let parsed_s3_values: &mut Vec = &mut Vec::new(); + extract_all_s3_object_from_raw_value(raw_value, parsed_s3_values); + if parsed_s3_values.is_empty() { + // no s3 object, nothing to return + return result; + } + + let s3_resource_opt = get_workspace_s3_resource_path(db, client, workspace_id, job_id).await; + if let Some(s3_resource) = s3_resource_opt { + for s3_object in parsed_s3_values { + let etag = get_etag_or_empty(&s3_resource, s3_object.clone()).await; + tracing::warn!("Enriching s3 arg value with etag: {:?}", etag); + result.insert(s3_object.s3.clone(), etag.unwrap_or_default()); // TODO: maybe inject a random value to invalidate the cache? + } + } + + return result; +} + +fn extract_all_s3_object_from_raw_value(raw_value: &Box, result: &mut Vec) { + let parsed_value = serde_json::from_str::(raw_value.get()); + if let Ok(parsed_value) = parsed_value { + result.push(parsed_value); + } else { + let parsed_value = serde_json::from_str::>>(raw_value.get()); + if let Ok(parsed_value) = parsed_value { + for (_, v) in parsed_value { + extract_all_s3_object_from_raw_value(&v, result); + } + } else { + let parsed_value = serde_json::from_str::>>(raw_value.get()); + if let Ok(parsed_value) = parsed_value { + for v in parsed_value { + extract_all_s3_object_from_raw_value(&v, result); + } + } + } + } +} + +#[derive(Deserialize, Serialize)] +struct CachedResource { + expire: i64, + #[serde(skip_serializing_if = "Option::is_none")] + s3_etags: Option>, + value: Box, +} + +pub async fn get_cached_resource_value_if_valid( + db: &DB, + client: &AuthedClient, + job_id: &Uuid, + workspace_id: &str, + cached_res_path: &str, +) -> Option> { + let resource_opt = client + .get_resource_value::(cached_res_path) + .await + .ok(); + if let Some(cached_resource) = resource_opt { + if cached_resource.expire <= chrono::Utc::now().timestamp() { + // cache expired + return None; + } + let s3_etags = cached_resource.s3_etags.unwrap_or_default(); + let s3_resource_opt: Option = if s3_etags.is_empty() { + None + } else { + get_workspace_s3_resource_path(db, &client, workspace_id, job_id).await + }; + if !s3_etags.is_empty() && s3_resource_opt.is_none() { + tracing::warn!("Cached result references s3 files that are not retrievable anymore because the workspace S3 resource can't be fetched. Cache will be invalidated"); + return None; + } + for (s3_file_key, s3_file_etag) in s3_etags { + if let Some(s3_resource) = s3_resource_opt.clone() { + let etag = + get_etag_or_empty(&s3_resource, S3Object { s3: s3_file_key.clone() }).await; + if etag.is_none() || etag.clone().unwrap() != s3_file_etag { + tracing::warn!("S3 file etag for '{}' has changed. Value from cache is {:?} while current value from S3 is {:?}. Cache will be invalidated", s3_file_key.clone(), s3_file_etag, etag); + return None; + } + } + } + return Some(cached_resource.value); + } + return None; +} + +pub async fn save_in_cache( db: &Pool, + client: &AuthedClient, job: &QueuedJob, cached_path: String, - r: &'a RawValue, + r: &Box, ) { let expire = chrono::Utc::now().timestamp() + job.cache_ttl.unwrap() as i64; - let store_cache_resource = StoreCachedResource { expire, value: r }; + let s3_etags = + arg_value_hash_additions(db, client, job.workspace_id.as_str(), &job.id, r).await; + + let store_cache_resource = CachedResource { + expire, + s3_etags: if s3_etags.is_empty() { + None + } else { + Some(s3_etags) + }, + value: r.clone(), + }; let raw_json = sqlx::types::Json(store_cache_resource); if let Err(e) = sqlx::query!( @@ -879,7 +1024,7 @@ pub async fn save_in_cache<'a>( DO UPDATE SET value = $3", job.workspace_id, cached_path, - raw_json as sqlx::types::Json, + raw_json as sqlx::types::Json, "cache" ) .execute(db) diff --git a/backend/windmill-worker/src/worker.rs b/backend/windmill-worker/src/worker.rs index 9428b61bf2..b421b2ae28 100644 --- a/backend/windmill-worker/src/worker.rs +++ b/backend/windmill-worker/src/worker.rs @@ -76,7 +76,10 @@ use windmill_queue::{add_completed_job, add_completed_job_error}; use crate::{ bash_executor::{handle_bash_job, handle_powershell_job, ANSI_ESCAPE_RE}, bun_executor::{gen_lockfile, get_trusted_deps, handle_bun_job}, - common::{build_args_map, hash_args, read_result, save_in_cache, write_file}, + common::{ + build_args_map, get_cached_resource_value_if_valid, hash_args, read_result, save_in_cache, + write_file, + }, deno_executor::{generate_deno_lock, handle_deno_job}, go_executor::{handle_go_job, install_go_dependencies}, graphql_executor::do_graphql, @@ -2084,7 +2087,7 @@ pub async fn process_completed_job, -} - #[derive(Deserialize, Serialize, Default)] pub struct PreviousResult<'a> { #[serde(borrow)] @@ -2399,7 +2396,14 @@ async fn handle_queued_job( } else { "none".to_string() }; - let args_hash = hash_args(&job.args); + let args_hash = hash_args( + db, + &client.get_authed().await, + &job.workspace_id, + &job.id, + &job.args, + ) + .await; if job.is_flow_step { let flow_path = sqlx::query_scalar!( "SELECT script_path FROM queue WHERE id = $1", @@ -2424,33 +2428,32 @@ async fn handle_queued_job( if let Some(cached_res_path) = cached_res_path.clone() { let authed_client = client.get_authed().await; - let resource = authed_client - .get_resource_value::(&cached_res_path) - .await; - if let Ok(resource) = resource { - let expire = resource.expire; - if expire > chrono::Utc::now().timestamp() { - let result = resource.value; - let logs = - "Job skipped because args & path found in cache and not expired".to_string(); + let cached_resource_value_maybe = get_cached_resource_value_if_valid( + db, + &authed_client, + &job.id, + &job.workspace_id, + &cached_res_path, + ) + .await; + if let Some(cached_resource_value) = cached_resource_value_maybe { + let logs = "Job skipped because args & path found in cache and not expired".to_string(); + job_completed_tx + .send(JobCompleted { + job: job, + result: cached_resource_value, + logs, + mem_peak: 0, + canceled_by: None, + success: true, + cached_res_path: None, + token: authed_client.token, + }) + .await + .expect("send job completed"); - job_completed_tx - .send(JobCompleted { - job: job, - result, - logs, - mem_peak: 0, - canceled_by: None, - success: true, - cached_res_path: None, - token: authed_client.token, - }) - .await - .expect("send job completed"); - - return Ok(()); - } + return Ok(()); } }; match job.job_kind { diff --git a/backend/windmill-worker/src/worker_flow.rs b/backend/windmill-worker/src/worker_flow.rs index 9050958315..aa71fa640b 100644 --- a/backend/windmill-worker/src/worker_flow.rs +++ b/backend/windmill-worker/src/worker_flow.rs @@ -633,7 +633,8 @@ pub async fn update_flow_status_after_job_completion_internal< } else { if flow_job.cache_ttl.is_some() { let cached_res_path = { - let args_hash = hash_args(&flow_job.args); + let args_hash = + hash_args(db, client, w_id, job_id_for_status, &flow_job.args).await; let flow_path = flow_job.script_path(); let version_hash = if let Some(rc) = flow_job.raw_flow.as_ref() { use std::hash::Hasher; @@ -648,7 +649,7 @@ pub async fn update_flow_status_after_job_completion_internal< format!("{flow_path}/cache/{version_hash}/{args_hash}") }; - save_in_cache(db, &flow_job, cached_res_path, &nresult).await; + save_in_cache(db, client, &flow_job, cached_res_path, &nresult).await; } let success = success && !is_failure_step && !skip_error_handler; if success {