diff --git a/backend/src/main.rs b/backend/src/main.rs index 1386a4b9bf..6468ef3072 100644 --- a/backend/src/main.rs +++ b/backend/src/main.rs @@ -533,51 +533,6 @@ fn print_help() { println!("- At startup, Windmill logs currently set configuration keys for visibility."); } -async fn resync_custom_instance_user_pwd_if_needed(db: &Pool) { - use windmill_common::utils::get_custom_pg_instance_password; - use windmill_common::{get_database_url, PgDatabase}; - - let user_pwd = match get_custom_pg_instance_password(db).await { - Ok(pwd) => pwd, - Err(_) => { - // Setting doesn't exist yet (fresh install or pre-migration), skip check - return; - } - }; - - let mut pg_creds = match get_database_url().await { - Ok(url) => match PgDatabase::parse_uri(&url.as_str().await) { - Ok(creds) => creds, - Err(e) => { - tracing::warn!("Failed to parse database URL for custom_instance_user check: {e}"); - return; - } - }, - Err(e) => { - tracing::warn!("Failed to get database URL for custom_instance_user check: {e}"); - return; - } - }; - - pg_creds.user = Some("custom_instance_user".to_string()); - pg_creds.password = Some(user_pwd); - - match pg_creds.connect().await { - Ok(_) => { - tracing::info!("custom_instance_user password is in sync"); - } - Err(e) => { - tracing::warn!("custom_instance_user password is out of sync ({e}), refreshing..."); - if let Err(e) = windmill_api_settings::refresh_custom_instance_user_pwd_inner(db).await - { - tracing::error!("Failed to refresh custom_instance_user password: {e}"); - } else { - tracing::info!("Successfully refreshed custom_instance_user password"); - } - } - } -} - async fn windmill_main() -> anyhow::Result<()> { let (killpill_tx, mut killpill_rx) = KillpillSender::new(2); let mut monitor_killpill_rx = killpill_tx.subscribe(); @@ -973,11 +928,6 @@ async fn windmill_main() -> anyhow::Result<()> { // NOTE: Variable/resource cache initialization moved to API server in windmill-api - // Check if custom_instance_user password is in sync - if server_mode { - resync_custom_instance_user_pwd_if_needed(&db).await; - } - Connection::Sql(db) }; diff --git a/backend/windmill-api-settings/src/lib.rs b/backend/windmill-api-settings/src/lib.rs index 24079f8ae6..dbd61531bf 100644 --- a/backend/windmill-api-settings/src/lib.rs +++ b/backend/windmill-api-settings/src/lib.rs @@ -997,49 +997,12 @@ async fn list_custom_instance_pg_databases( return Ok(Json(result)); } -pub async fn refresh_custom_instance_user_pwd_inner(db: &DB) -> Result<()> { - // 20251208123907_safety_custom_instance_db_user_pwd.up - let query = r#" - DO $$ - DECLARE - pwd text; - BEGIN - SELECT gen_random_uuid()::text INTO pwd; - - IF EXISTS (SELECT 1 FROM pg_roles WHERE rolname = 'custom_instance_user') THEN - EXECUTE format('ALTER USER custom_instance_user WITH PASSWORD %L', pwd); - RAISE NOTICE 'Updated password for existing user custom_instance_user'; - ELSE - EXECUTE format('CREATE USER custom_instance_user WITH PASSWORD %L', pwd); - RAISE NOTICE 'Created new user custom_instance_user'; - END IF; - - IF NOT EXISTS (SELECT 1 FROM global_settings WHERE name = 'custom_instance_pg_databases') THEN - INSERT INTO global_settings (name, value) - VALUES ('custom_instance_pg_databases', jsonb_build_object( - 'user_pwd', pwd::text, - 'databases', jsonb_build_object() - )); - RAISE NOTICE 'Inserted new global setting for custom_instance_pg_databases'; - ELSE - UPDATE global_settings - SET value = jsonb_set(COALESCE(value, '{}'::jsonb), '{user_pwd}', to_jsonb(pwd::text)::jsonb) - WHERE name = 'custom_instance_pg_databases'; - RAISE NOTICE 'Updated user_pwd in existing global setting for custom_instance_pg_databases'; - END IF; - END - $$; - "#; - sqlx::query(query).execute(db).await?; - Ok(()) -} - async fn refresh_custom_instance_user_pwd( authed: ApiAuthed, Extension(db): Extension, ) -> JsonResult<()> { require_super_admin(&db, &authed.email).await?; - refresh_custom_instance_user_pwd_inner(&db).await?; + windmill_common::utils::refresh_custom_instance_user_pwd(&db).await?; Ok(Json(())) } @@ -1140,7 +1103,7 @@ async fn setup_custom_instance_pg_database_inner( } // We have to connect to the newly created database as admin to grant permissions - let (client, connection) = pg_creds.connect().await?; + let (client, connection) = pg_creds.connect(Some(db)).await?; let join_handle = tokio::spawn(async move { connection.await }); logs.db_connect = "OK".to_string(); diff --git a/backend/windmill-api-workspaces/src/workspaces.rs b/backend/windmill-api-workspaces/src/workspaces.rs index dccd260449..5e8c2494b1 100644 --- a/backend/windmill-api-workspaces/src/workspaces.rs +++ b/backend/windmill-api-workspaces/src/workspaces.rs @@ -1397,7 +1397,7 @@ async fn get_datatable_schema(db: &DB, w_id: &str, datatable_name: &str) -> Resu .map_err(|e| Error::internal_err(format!("Failed to parse database credentials: {}", e)))?; // Connect to the datatable database - let (client, connection) = pg_db.connect().await?; + let (client, connection) = pg_db.connect(Some(db)).await?; // Spawn the connection handler tokio::spawn(async move { @@ -1733,7 +1733,7 @@ async fn create_pg_database( } else { let source_pg = resolve_pg_source_checked(&db, &user_db, &authed, &w_id, &req.source).await?; - let (client, connection) = source_pg.connect().await?; + let (client, connection) = source_pg.connect(Some(&db)).await?; let join_handle = tokio::spawn(async move { connection.await }); let row = client @@ -1865,7 +1865,7 @@ async fn get_datatable_full_schema( Json(req): Json, ) -> JsonResult { let pg = resolve_pg_source_checked(&db, &user_db, &authed, &w_id, &req.source).await?; - let (client, connection) = pg.connect().await?; + let (client, connection) = pg.connect(Some(&db)).await?; let join_handle = tokio::spawn(async move { connection.await }); let result = windmill_common::query_builders::pg_get_full_schema(&client) @@ -4190,7 +4190,7 @@ async fn snapshot_datatable_schema( let pg = get_datatable_resource_from_db_unchecked(db, parent_w_id, dt_name).await?; let pg: PgDatabase = serde_json::from_value(pg) .map_err(|e| Error::internal_err(format!("Failed to parse db credentials: {}", e)))?; - let (client, connection) = pg.connect().await?; + let (client, connection) = pg.connect(Some(db)).await?; let join_handle = tokio::spawn(async move { connection.await }); let schema = windmill_common::query_builders::pg_get_full_schema(&client) diff --git a/backend/windmill-api-workspaces/src/workspaces_extra.rs b/backend/windmill-api-workspaces/src/workspaces_extra.rs index 231f7cb429..24d4192299 100644 --- a/backend/windmill-api-workspaces/src/workspaces_extra.rs +++ b/backend/windmill-api-workspaces/src/workspaces_extra.rs @@ -983,7 +983,7 @@ pub async fn drop_forked_datatable_databases( continue; } - match parent_pg.connect().await { + match parent_pg.connect(Some(&db)).await { Ok((client, connection)) => { let join_handle = tokio::spawn(async move { connection.await }); if let Err(e) = client diff --git a/backend/windmill-common/src/ai_bedrock.rs b/backend/windmill-common/src/ai_bedrock.rs index 9a7c5baeeb..a740b202cb 100644 --- a/backend/windmill-common/src/ai_bedrock.rs +++ b/backend/windmill-common/src/ai_bedrock.rs @@ -410,9 +410,7 @@ fn mime_to_document_format(mime_type: &str) -> DocumentFormat { "application/vnd.openxmlformats-officedocument.wordprocessingml.document" => { DocumentFormat::Docx } - "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet" => { - DocumentFormat::Xlsx - } + "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet" => DocumentFormat::Xlsx, _ => DocumentFormat::Pdf, } } diff --git a/backend/windmill-common/src/ai_google.rs b/backend/windmill-common/src/ai_google.rs index b76d976a4c..f0591b9b81 100644 --- a/backend/windmill-common/src/ai_google.rs +++ b/backend/windmill-common/src/ai_google.rs @@ -356,9 +356,7 @@ pub fn convert_content_to_gemini_parts(content: &OpenAIContent) -> Vec { parse_data_url(&file.file_data).map(|(mime_type, data)| { - GeminiPart::InlineData { - inline_data: GeminiInlineData { mime_type, data }, - } + GeminiPart::InlineData { inline_data: GeminiInlineData { mime_type, data } } }) } // S3Objects are handled by the worker diff --git a/backend/windmill-common/src/flows.rs b/backend/windmill-common/src/flows.rs index a0923ccabf..01d714ea01 100644 --- a/backend/windmill-common/src/flows.rs +++ b/backend/windmill-common/src/flows.rs @@ -81,8 +81,7 @@ pub async fn get_full_hub_flow_by_path( .map_err(to_anyhow) { Ok(response) => response, - Err(_) if hub_base_url != DEFAULT_HUB_BASE_URL && flow_id < PRIVATE_HUB_MIN_VERSION => - { + Err(_) if hub_base_url != DEFAULT_HUB_BASE_URL && flow_id < PRIVATE_HUB_MIN_VERSION => { tracing::info!("Not found on private hub, fallback to default hub for hub flow {path}"); let fallback_url = format!("{DEFAULT_HUB_BASE_URL}/flows/{flow_id}/json"); http_get_from_hub(http_client, &fallback_url, false, None, db) diff --git a/backend/windmill-common/src/lib.rs b/backend/windmill-common/src/lib.rs index fa48415f80..a607e2dea5 100644 --- a/backend/windmill-common/src/lib.rs +++ b/backend/windmill-common/src/lib.rs @@ -464,6 +464,32 @@ impl PgDatabase { pub async fn connect( &self, + main_db: Option<&DB>, + ) -> Result<(tokio_postgres::Client, TokioPgConnection), error::Error> { + match self.connect_inner().await { + Ok(result) => Ok(result), + Err(e) => { + let err_str = e.to_string(); + if err_str.contains("password authentication failed for user") && err_str.contains("custom_instance_user") + { + if let Some(db) = main_db { + tracing::warn!( + "custom_instance_user password auth failed, refreshing and retrying..." + ); + crate::utils::refresh_custom_instance_user_pwd(db).await?; + let new_pwd = crate::utils::get_custom_pg_instance_password(db).await?; + let mut retried = self.clone(); + retried.password = Some(new_pwd); + return retried.connect_inner().await; + } + } + Err(e) + } + } + } + + async fn connect_inner( + &self, ) -> Result<(tokio_postgres::Client, TokioPgConnection), error::Error> { use native_tls::{Certificate, TlsConnector}; use postgres_native_tls::MakeTlsConnector; @@ -757,7 +783,7 @@ pub async fn create_custom_instance_database( // Grant permissions to custom_instance_user let wmill_pg_creds = PgDatabase::parse_uri(&get_database_url().await?.as_str().await)?; let new_pg_creds = PgDatabase { dbname: dbname.to_string(), ..wmill_pg_creds }; - let (client, connection) = new_pg_creds.connect().await?; + let (client, connection) = new_pg_creds.connect(Some(db)).await?; let join_handle = tokio::spawn(async move { connection.await }); if let Err(e) = client diff --git a/backend/windmill-common/src/secret_backend/mod.rs b/backend/windmill-common/src/secret_backend/mod.rs index 8caaed6012..4c9389cac9 100644 --- a/backend/windmill-common/src/secret_backend/mod.rs +++ b/backend/windmill-common/src/secret_backend/mod.rs @@ -17,10 +17,10 @@ pub mod database; #[cfg(feature = "private")] pub mod vault_ee; -pub mod vault_oss; #[cfg(feature = "private")] pub mod azure_kv_ee; pub mod azure_kv_oss; +pub mod vault_oss; #[cfg(feature = "private")] pub mod aws_sm_ee; diff --git a/backend/windmill-common/src/tracing_init.rs b/backend/windmill-common/src/tracing_init.rs index 1a2ca33518..3b009a36eb 100644 --- a/backend/windmill-common/src/tracing_init.rs +++ b/backend/windmill-common/src/tracing_init.rs @@ -273,4 +273,3 @@ where } } } - diff --git a/backend/windmill-common/src/utils.rs b/backend/windmill-common/src/utils.rs index 10b1a3b408..e35adc53d2 100644 --- a/backend/windmill-common/src/utils.rs +++ b/backend/windmill-common/src/utils.rs @@ -986,6 +986,38 @@ impl ExpiringCacheEntry { } } +pub async fn refresh_custom_instance_user_pwd(db: &DB) -> Result<()> { + let query = r#" + DO $$ + DECLARE + pwd text; + BEGIN + SELECT gen_random_uuid()::text INTO pwd; + + IF EXISTS (SELECT 1 FROM pg_roles WHERE rolname = 'custom_instance_user') THEN + EXECUTE format('ALTER USER custom_instance_user WITH PASSWORD %L', pwd); + ELSE + EXECUTE format('CREATE USER custom_instance_user WITH PASSWORD %L', pwd); + END IF; + + IF NOT EXISTS (SELECT 1 FROM global_settings WHERE name = 'custom_instance_pg_databases') THEN + INSERT INTO global_settings (name, value) + VALUES ('custom_instance_pg_databases', jsonb_build_object( + 'user_pwd', pwd::text, + 'databases', jsonb_build_object() + )); + ELSE + UPDATE global_settings + SET value = jsonb_set(COALESCE(value, '{}'::jsonb), '{user_pwd}', to_jsonb(pwd::text)::jsonb) + WHERE name = 'custom_instance_pg_databases'; + END IF; + END + $$; + "#; + sqlx::query(query).execute(db).await?; + Ok(()) +} + pub async fn get_custom_pg_instance_password(db: &DB) -> Result { sqlx::query_scalar!( "SELECT value->>'user_pwd' FROM global_settings WHERE name = 'custom_instance_pg_databases';" diff --git a/backend/windmill-worker/src/pg_executor.rs b/backend/windmill-worker/src/pg_executor.rs index 2e29e4ef89..3a70db5ecb 100644 --- a/backend/windmill-worker/src/pg_executor.rs +++ b/backend/windmill-worker/src/pg_executor.rs @@ -28,7 +28,7 @@ use windmill_common::worker::{ to_raw_value, Connection, SqlResultCollectionStrategy, CLOUD_HOSTED, }; use windmill_common::workspaces::get_datatable_resource_from_db_unchecked; -use windmill_common::{PgDatabase, PrepareQueryColumnInfo, PrepareQueryResult}; +use windmill_common::{PgDatabase, PrepareQueryColumnInfo, PrepareQueryResult, DB}; use windmill_object_store::convert_json_line_stream; use windmill_parser::{Arg, Typ}; use windmill_parser_sql::{ @@ -67,6 +67,7 @@ pub async fn clear_pg_cache() { async fn new_pg_connection( database: &PgDatabase, _use_iam_auth: bool, + main_db: Option<&DB>, ) -> error::Result<(tokio_postgres::Client, tokio::task::JoinHandle<()>)> { let (client, connection) = if _use_iam_auth { #[cfg(all(feature = "enterprise", feature = "private"))] @@ -80,7 +81,7 @@ async fn new_pg_connection( )); } } else { - database.connect().await? + database.connect(main_db).await? }; let handle = tokio::spawn(async move { if let Err(e) = connection.await { @@ -361,18 +362,18 @@ pub async fn do_postgresql( } drop(guard); cached_client = None; - new_client = Some(new_pg_connection(&database, use_iam_auth).await?); + new_client = Some(new_pg_connection(&database, use_iam_auth, conn.as_sql()).await?); } } else { // Release the lock before connecting so the post-query caching // code can re-acquire it. drop(guard); cached_client = None; - new_client = Some(new_pg_connection(&database, use_iam_auth).await?); + new_client = Some(new_pg_connection(&database, use_iam_auth, conn.as_sql()).await?); } } else { cached_client = None; - new_client = Some(new_pg_connection(&database, use_iam_auth).await?); + new_client = Some(new_pg_connection(&database, use_iam_auth, conn.as_sql()).await?); } let (sig, typed_schema) = parse_pgsql_sig_with_typed_schema(&query)