fix: make workers bind their http servers on any available port for OCI compliance

backend/src/main.rs

@@ -12,7 +12,7 @@ use monitor::handle_zombie_jobs_periodically;
 use sqlx::{Pool, Postgres};
 use std::{
     net::{IpAddr, Ipv4Addr, SocketAddr},
-    sync::{Arc},
+    sync::Arc,
 };
 use tokio::{
     fs::{metadata, DirBuilder},
@@ -65,17 +65,6 @@ async fn main() -> anyhow::Result<()> {
         IpAddr::V4(Ipv4Addr::new(127, 0, 0, 1))
     };
 
-    let port: u16 = std::env::var("PORT")
-        .ok()
-        .and_then(|x| x.parse::<u16>().ok())
-        .unwrap_or(DEFAULT_PORT as u16);
-
-    if std::env::var("BASE_INTERNAL_URL").is_ok() {
-        tracing::warn!("BASE_INTERNAL_URL is now unecessary and ignored, you can remove it.");
-    }
-
-    let base_internal_url: String = format!("http://localhost:{}", port.to_string());
-
     let rsmq_config = std::env::var("REDIS_URL").ok().map(|x| {
         let url = x.parse::<url::Url>().unwrap();
         let mut config = rsmq_async::RsmqOptions { ..Default::default() };
@@ -202,15 +191,31 @@ Windmill Community Edition {GIT_VERSION}
         }
     }
     if server_mode || num_workers > 0 {
+        let port_var = std::env::var("PORT").ok().and_then(|x| x.parse().ok());
+
+        let port = if server_mode {
+            port_var.unwrap_or(DEFAULT_PORT as u16)
+        } else {
+            port_var.unwrap_or(0)
+        };
+
+        if std::env::var("BASE_INTERNAL_URL").is_ok() {
+            tracing::warn!("BASE_INTERNAL_URL is now unecessary and ignored, you can remove it.");
+        }
+
         let addr = SocketAddr::from((server_bind_address, port));
 
         let rsmq2 = rsmq.clone();
+        let (port_tx, port_rx) = tokio::sync::oneshot::channel::<u16>();
+
         let server_f = async {
-            windmill_api::run_server(db.clone(), rsmq2, addr, rx.resubscribe()).await?;
+            windmill_api::run_server(db.clone(), rsmq2, addr, rx.resubscribe(), port_tx).await?;
             Ok(()) as anyhow::Result<()>
         };
 
         let workers_f = async {
+            let port = port_rx.await?;
+            let base_internal_url: String = format!("http://localhost:{}", port.to_string());
             if num_workers > 0 {
                 run_workers(
                     db.clone(),
@@ -229,7 +234,9 @@ Windmill Community Edition {GIT_VERSION}
         let rsmq2 = rsmq.clone();
         let monitor_f = async {
             if server_mode {
-                monitor_db(&db, rx.resubscribe(), &base_internal_url, rsmq2);
+                // since it's only on server mode, the port is statically defined
+                let base_internal_url: String = format!("http://localhost:{}", port.to_string());
+                monitor_db(&db, rx.resubscribe(), &base_internal_url, rsmq2).await;
             }
             Ok(()) as anyhow::Result<()>
         };
@@ -269,12 +276,12 @@ fn display_config(envs: Vec<&str>) {
     )
 }
 
-pub fn monitor_db<R: rsmq_async::RsmqConnection + Send + Sync + Clone + 'static>(
+pub async fn monitor_db<R: rsmq_async::RsmqConnection + Send + Sync + Clone + 'static>(
     db: &Pool<Postgres>,
     rx: tokio::sync::broadcast::Receiver<()>,
     base_internal_url: &str,
     rsmq: Option<R>,
-) {
+) -> tokio::task::JoinHandle<()> {
     let db1 = db.clone();
     let db2 = db.clone();
 
@@ -285,7 +292,7 @@ pub fn monitor_db<R: rsmq_async::RsmqConnection + Send + Sync + Clone + 'static>
             handle_zombie_jobs_periodically(&db1, rx, &base_internal_url, rsmq),
             windmill_api::delete_expired_items_perdiodically(&db2, rx2)
         );
-    });
+    })
 }
 
 pub async fn run_workers<R: rsmq_async::RsmqConnection + Send + Sync + Clone + 'static>(

backend/tests/worker.rs

@@ -1,4 +1,4 @@
-use std::sync::{Arc};
+use std::sync::Arc;
 
 use futures::{stream, Stream};
 use serde::Deserialize;
@@ -92,8 +92,15 @@ impl ApiServer {
 
         let addr = sock.local_addr().unwrap();
         drop(sock);
+        let (port_tx, _port_rx) = tokio::sync::oneshot::channel::<u16>();
 
-        let task = tokio::task::spawn(windmill_api::run_server(db.clone(), None, addr, rx));
+        let task = tokio::task::spawn(windmill_api::run_server(
+            db.clone(),
+            None,
+            addr,
+            rx,
+            port_tx,
+        ));
 
         return Self { addr, tx, task };
     }

backend/windmill-api/src/lib.rs

@@ -140,6 +140,7 @@ pub async fn run_server(
     rsmq: Option<rsmq_async::MultiplexedRsmq>,
     addr: SocketAddr,
     mut rx: tokio::sync::broadcast::Receiver<()>,
+    port_tx: tokio::sync::oneshot::Sender<u16>,
 ) -> anyhow::Result<()> {
     let user_db = UserDB::new(db.clone());
 
@@ -243,12 +244,22 @@ pub async fn run_server(
     let instance_name = rd_string(5);
 
     tracing::info!(addr = %addr.to_string(), instance = %instance_name, "server started listening");
-    let server = axum::Server::bind(&addr)
-        .serve(app.into_make_service())
-        .with_graceful_shutdown(async {
-            rx.recv().await.ok();
-            println!("Graceful shutdown of server");
-        });
+    let server = axum::Server::bind(&addr).serve(app.into_make_service());
+
+    let port = server.local_addr().port();
+    tracing::info!(
+        "server started on port={} and addr={}",
+        port,
+        server.local_addr().ip()
+    );
+    port_tx
+        .send(server.local_addr().port())
+        .expect("Failed to send port");
+
+    let server = server.with_graceful_shutdown(async {
+        rx.recv().await.ok();
+        println!("Graceful shutdown of server");
+    });
 
     tokio::spawn(async move { auth_cache.monitor().await });
 

backend/windmill-api/src/main3.rs

@@ -1,69 +0,0 @@
-// /*
-//  * Author: Ruben Fiszel
-//  * Copyright: Windmill Labs, Inc 2022
-//  * This file and its contents are licensed under the AGPLv3 License.
-//  * Please see the included NOTICE for copyright information and
-//  * LICENSE-AGPL for a copy of the license.
-//  */
-// use std::net::SocketAddr;
-
-// use anyhow::Ok;
-
-// pub const DEFAULT_NUM_WORKERS: usize = 3;
-
-// #[tokio::main]
-// async fn main() -> anyhow::Result<()> {
-//     windmill_common::tracing_init::initialize_tracing();
-
-//     let db = windmill_common::connect_db(true).await?;
-
-//     let num_workers = std::env::var("NUM_WORKERS")
-//         .ok()
-//         .and_then(|x| x.parse::<i32>().ok())
-//         .unwrap_or(DEFAULT_NUM_WORKERS as i32);
-
-//     let metrics_addr: Option<SocketAddr> = std::env::var("METRICS_ADDR")
-//         .ok()
-//         .map(|s| {
-//             s.parse::<bool>()
-//                 .map(|b| b.then(|| SocketAddr::from(([0, 0, 0, 0], 8001))))
-//                 .or_else(|_| s.parse::<SocketAddr>().map(Some))
-//         })
-//         .transpose()?
-//         .flatten();
-
-//     let server_mode = !std::env::var("DISABLE_SERVER")
-//         .ok()
-//         .and_then(|x| x.parse::<bool>().ok())
-//         .unwrap_or(false);
-
-//     if server_mode {
-//         windmill_api::migrate_db(&db).await?;
-//     }
-
-//     let (tx, rx) = tokio::sync::broadcast::channel::<()>(3);
-//     let shutdown_signal = windmill_common::shutdown_signal(tx);
-
-//     if server_mode || num_workers > 0 {
-//         let addr = SocketAddr::from(([0, 0, 0, 0], 8000));
-
-//         let server_f = async {
-//             if server_mode {
-//                 windmill_api::run_server(db.clone(), addr, rx.resubscribe()).await?;
-//             }
-//             Ok(()) as anyhow::Result<()>
-//         };
-
-//         let metrics_f = async {
-//             match metrics_addr {
-//                 Some(addr) => windmill_common::serve_metrics(addr, rx.resubscribe())
-//                     .await
-//                     .map_err(anyhow::Error::from),
-//                 None => Ok(()),
-//             }
-//         };
-
-//         futures::try_join!(shutdown_signal, server_f, metrics_f)?;
-//     }
-//     Ok(())
-// }