diff --git a/backend/src/main.rs b/backend/src/main.rs index 48bc518cdc..37c401f58b 100644 --- a/backend/src/main.rs +++ b/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::().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::().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::(); + 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( +pub async fn monitor_db( db: &Pool, rx: tokio::sync::broadcast::Receiver<()>, base_internal_url: &str, rsmq: Option, -) { +) -> tokio::task::JoinHandle<()> { let db1 = db.clone(); let db2 = db.clone(); @@ -285,7 +292,7 @@ pub fn monitor_db handle_zombie_jobs_periodically(&db1, rx, &base_internal_url, rsmq), windmill_api::delete_expired_items_perdiodically(&db2, rx2) ); - }); + }) } pub async fn run_workers( diff --git a/backend/tests/worker.rs b/backend/tests/worker.rs index 12610e3c00..9ad80f8230 100644 --- a/backend/tests/worker.rs +++ b/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::(); - 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 }; } diff --git a/backend/windmill-api/src/lib.rs b/backend/windmill-api/src/lib.rs index a5335f11ba..10cec6dbbd 100644 --- a/backend/windmill-api/src/lib.rs +++ b/backend/windmill-api/src/lib.rs @@ -140,6 +140,7 @@ pub async fn run_server( rsmq: Option, addr: SocketAddr, mut rx: tokio::sync::broadcast::Receiver<()>, + port_tx: tokio::sync::oneshot::Sender, ) -> 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 }); diff --git a/backend/windmill-api/src/main3.rs b/backend/windmill-api/src/main3.rs deleted file mode 100644 index bcfb107829..0000000000 --- a/backend/windmill-api/src/main3.rs +++ /dev/null @@ -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::().ok()) -// .unwrap_or(DEFAULT_NUM_WORKERS as i32); - -// let metrics_addr: Option = std::env::var("METRICS_ADDR") -// .ok() -// .map(|s| { -// s.parse::() -// .map(|b| b.then(|| SocketAddr::from(([0, 0, 0, 0], 8001)))) -// .or_else(|_| s.parse::().map(Some)) -// }) -// .transpose()? -// .flatten(); - -// let server_mode = !std::env::var("DISABLE_SERVER") -// .ok() -// .and_then(|x| x.parse::().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(()) -// }