add PG_LISTENER_REFRESH_PERIOD_SECS

backend/src/main.rs

@@ -19,7 +19,7 @@ use std::{
     collections::HashMap,
     fs::{create_dir_all, DirBuilder},
     net::{IpAddr, Ipv4Addr, SocketAddr},
-    time::Duration,
+    time::{Duration, Instant},
 };
 use tokio::{fs::File, io::AsyncReadExt, task::JoinHandle};
 use uuid::Uuid;
@@ -115,6 +115,12 @@ where
     rt.block_on(future)
 }
 
+lazy_static::lazy_static! {
+    static ref PG_LISTENER_REFRESH_PERIOD_SECS: Option<u64> = std::env::var("PG_LISTENER_REFRESH_PERIOD_SECS")
+        .ok()
+        .and_then(|x| x.parse::<u64>().ok());
+}
+
 pub fn main() -> anyhow::Result<()> {
     // https://github.com/denoland/deno/blob/main/cli/main.rs#L477
     #[cfg(feature = "deno_core")]
@@ -650,6 +656,7 @@ Windmill Community Edition {GIT_VERSION}
 
             let h = tokio::spawn(async move {
                 let mut listener = retry_listen_pg(&db_url).await;
+                let mut last_listener_refresh = Instant::now();
                 loop {
                     tokio::select! {
                         biased;
@@ -859,6 +866,23 @@ Windmill Community Edition {GIT_VERSION}
                             };
                         },
                         _ = tokio::time::sleep(Duration::from_secs(30))    => {
+                            if PG_LISTENER_REFRESH_PERIOD_SECS.is_some_and(|x| last_listener_refresh.elapsed() > Duration::from_secs(x)) {
+                                tracing::info!("Refreshing pg listener");
+                                if let Err(e) = listener.unlisten_all().await {
+                                    tracing::error!(error = %e, "Could not unlisten to database");
+                                }
+                                monitor_db(
+                                    &db,
+                                    &base_internal_url,
+                                    server_mode,
+                                    worker_mode,
+                                    true,
+                                    tx.clone(),
+                                ).await;
+                                listener = retry_listen_pg(&db_url).await;
+                                last_listener_refresh = Instant::now();
+                            }
+
                             tracing::info!("monitor task started");
                             monitor_db(
                                 &db,

backend/windmill-common/src/global_settings.rs

@@ -40,7 +40,7 @@ pub const JWT_SECRET_SETTING: &str = "jwt_secret";
 pub const EMAIL_DOMAIN_SETTING: &str = "email_domain";
 pub const OTEL_SETTING: &str = "otel";
 
-pub const ENV_SETTINGS: [&str; 57] = [
+pub const ENV_SETTINGS: [&str; 58] = [
     "DISABLE_NSJAIL",
     "MODE",
     "NUM_WORKERS",
@@ -98,6 +98,7 @@ pub const ENV_SETTINGS: [&str; 57] = [
     "OTEL_LOGS",
     "DISABLE_S3_STORE",
     "PG_SCHEMA",
+    "PG_LISTENER_REFRESH_PERIOD_SECS",
 ];
 
 use crate::error;