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] <windmill-internal-app[bot]@users.noreply.github.com>

backend/ee-repo-ref.txt

@@ -1 +1 @@
-0bfcfe263622f48c086331b628c1239d5cea0b37
+1549849fadc4e5634334a384bfe52343eb1e93f0

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<PgListener> {
-    let mut listener = match PgListener::connect(url).await {
+async fn listen_pg(db: &Pool<Postgres>) -> Option<PgListener> {
+    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<PgListener> {
     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<Postgres>) -> 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();

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<sqlx::postgres::PgConnectOptions, Error> {
+        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<DatabaseUrl, Error> {
 }
 
 pub async fn initial_connection() -> Result<sqlx::Pool<sqlx::Postgres>, 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