From 7ed91cd3c2068da4f9aa90f4944c818c2924d2c7 Mon Sep 17 00:00:00 2001 From: Alexander Petric Date: Fri, 23 Jan 2026 07:09:40 -0500 Subject: [PATCH] fix: use pgoptions for iam rds connection (#7660) * use pgoptions * fix: use pgoptions for iam rds connection * ee ref * chore: update ee-repo-ref to 1549849fadc4e5634334a384bfe52343eb1e93f0 This commit updates the EE repository reference after PR #388 was merged in windmill-ee-private. Previous ee-repo-ref: ffc1de1498a8018a9cbc2daba846e6c57d500a1c New ee-repo-ref: 1549849fadc4e5634334a384bfe52343eb1e93f0 Automated by sync-ee-ref workflow. --------- Co-authored-by: windmill-internal-app[bot] --- backend/ee-repo-ref.txt | 2 +- backend/src/main.rs | 22 +++++++-------- backend/windmill-common/src/lib.rs | 43 +++++++++++++++++++++++------- 3 files changed, 44 insertions(+), 23 deletions(-) diff --git a/backend/ee-repo-ref.txt b/backend/ee-repo-ref.txt index d4623bb1da..14c88d243a 100644 --- a/backend/ee-repo-ref.txt +++ b/backend/ee-repo-ref.txt @@ -1 +1 @@ -0bfcfe263622f48c086331b628c1239d5cea0b37 +1549849fadc4e5634334a384bfe52343eb1e93f0 diff --git a/backend/src/main.rs b/backend/src/main.rs index f2a8f7b689..e789cfb767 100644 --- a/backend/src/main.rs +++ b/backend/src/main.rs @@ -16,7 +16,7 @@ use monitor::{ send_current_log_file_to_object_store, send_logs_to_object_store, WORKERS_NAMES, }; use rand::Rng; -use sqlx::postgres::PgListener; +use sqlx::{postgres::PgListener, Pool, Postgres}; use std::{ collections::HashMap, fs::{create_dir_all, DirBuilder}, @@ -35,7 +35,6 @@ use windmill_common::ee_oss::{ use windmill_common::{ agent_workers::build_agent_http_client, - get_database_url, global_settings::{ APP_WORKSPACED_ROUTE_SETTING, BASE_URL_SETTING, BUNFIG_INSTALL_SCOPES_SETTING, CRITICAL_ALERTS_ON_DB_OVERSIZE_SETTING, CRITICAL_ALERT_MUTE_UI_SETTING, @@ -901,10 +900,9 @@ Windmill Community Edition {GIT_VERSION} match conn { Connection::Sql(ref db) => { let base_internal_url = base_internal_url.to_string(); - let db_url = get_database_url().await?; let db = db.clone(); let h = tokio::spawn(async move { - let mut listener = retry_listen_pg(&db_url.as_str().await).await; + let mut listener = retry_listen_pg(&db).await; let mut last_listener_refresh = Instant::now(); let mut monitor_iteration: u64 = 0; let rd_shift: u8 = rand::rng().random_range(0..200); @@ -1245,14 +1243,14 @@ Windmill Community Edition {GIT_VERSION} }, Err(e) => { tracing::error!(error = %e, "Could not receive notification, attempting to reconnect listener"); - let db_url = db_url.clone(); + let db = db.clone(); tokio::select! { biased; _ = monitor_killpill_rx.recv() => { tracing::info!("received killpill for monitor job"); break; }, - new_listener = async move { retry_listen_pg(&db_url.as_str().await).await } => { + new_listener = async move { retry_listen_pg(&db).await } => { listener = new_listener; continue; } @@ -1266,7 +1264,7 @@ Windmill Community Edition {GIT_VERSION} if let Err(e) = listener.unlisten_all().await { tracing::error!(error = %e, "Could not unlisten to database"); } - listener = retry_listen_pg(&db_url.as_str().await).await; + listener = retry_listen_pg(&db).await; initial_load( &conn, tx.clone(), @@ -1451,8 +1449,8 @@ Windmill Community Edition {GIT_VERSION} std::process::exit(0); } -async fn listen_pg(url: &str) -> Option { - let mut listener = match PgListener::connect(url).await { +async fn listen_pg(db: &Pool) -> Option { + let mut listener = match PgListener::connect_with(db).await { Ok(l) => l, Err(e) => { tracing::error!(error = %e, "Could not connect to database"); @@ -1485,13 +1483,13 @@ async fn listen_pg(url: &str) -> Option { return Some(listener); } -async fn retry_listen_pg(url: &str) -> PgListener { - let mut listener = listen_pg(url).await; +async fn retry_listen_pg(db: &Pool) -> PgListener { + let mut listener = listen_pg(db).await; loop { if listener.is_none() { tracing::info!("Retrying listening to pg listen in 5 seconds"); tokio::time::sleep(Duration::from_secs(5)).await; - listener = listen_pg(url).await; + listener = listen_pg(db).await; } else { tracing::info!("Successfully connected to pg listen"); return listener.unwrap(); diff --git a/backend/windmill-common/src/lib.rs b/backend/windmill-common/src/lib.rs index 02cf7beff0..3eca5ce0b7 100644 --- a/backend/windmill-common/src/lib.rs +++ b/backend/windmill-common/src/lib.rs @@ -515,6 +515,9 @@ pub enum DatabaseUrl { } impl DatabaseUrl { + /// Get the database URL as a string. + /// Note: For IAM RDS, this returns the original URL (for metadata extraction). + /// For actual database connections, use connect_options() instead. pub async fn as_str(&self) -> String { match self { #[cfg(all(feature = "enterprise", feature = "private"))] @@ -526,6 +529,24 @@ impl DatabaseUrl { } } + /// Get PgConnectOptions for this database URL. + /// For IAM RDS, this returns options built directly from the token to avoid double-encoding + /// issues with temporary credentials (IRSA/Pod Identity). + /// For static URLs, this parses the URL string. + pub async fn connect_options(&self) -> Result { + match self { + #[cfg(all(feature = "enterprise", feature = "private"))] + DatabaseUrl::IamRds(rds_url) => { + let guard = rds_url.read().await; + Ok(guard.connect_options()) + } + DatabaseUrl::Static(url) => { + sqlx::postgres::PgConnectOptions::from_str(url) + .map_err(|e| Error::InternalErr(format!("Failed to parse database URL: {}", e))) + } + } + } + pub async fn refresh(&self) -> anyhow::Result<()> { match self { #[cfg(all(feature = "enterprise", feature = "private"))] @@ -622,10 +643,10 @@ pub async fn get_database_url() -> Result { } pub async fn initial_connection() -> Result, error::Error> { - let database_url = get_database_url().await?.as_str().await; + let connect_options = get_database_url().await?.connect_options().await?; sqlx::postgres::PgPoolOptions::new() .max_connections(2) - .connect_with(sqlx::postgres::PgConnectOptions::from_str(&database_url)?) + .connect_with(connect_options) .await .map_err(|err| Error::ConnectingToDatabase(err.to_string())) } @@ -679,14 +700,16 @@ pub async fn connect_db( let new_url = tokio::time::timeout(std::time::Duration::from_secs(10), get_database_url()).await; match new_url { Ok(Ok(new_url)) => { - let new_url = new_url.as_str().await; - let connect_options = sqlx::postgres::PgConnectOptions::from_str(&new_url); - if let Err(e) = connect_options { - tracing::error!("Error parsing IAM RDS URL as connect options, retrying in 10s: {}", e); - continue; + match new_url.connect_options().await { + Ok(connect_options) => { + pool2.set_connect_options(connect_options); + tracing::info!("Refreshed IAM RDS URL successfully"); + } + Err(e) => { + tracing::error!("Error getting IAM RDS connect options, retrying in 10s: {}", e); + continue; + } } - pool2.set_connect_options(connect_options.unwrap()); - tracing::info!("Refreshed IAM RDS URL successfully"); } Ok(Err(e)) => { tracing::error!("Error refreshing IAM RDS URL, trying again in 10s: {}", e); @@ -757,7 +780,7 @@ pub async fn connect( } }) .connect_with( - sqlx::postgres::PgConnectOptions::from_str(&database_url.as_str().await)? + database_url.connect_options().await? .statement_cache_capacity(400), ) .await