feat: perform controlled shutdown on ctrl-c

Closes kemitix/git-next#94

Controlled shutdown includes attempting to unregister webhooks.

crates/repo-actor/src/tests/handlers/validate_repo.rs

@@ -354,7 +354,7 @@ async fn should_reject_message_with_expired_token() -> TestResult {
 }
 
 #[test_log::test(actix::test)]
-// NOTE: failed then passed on retry: count = 2
+// NOTE: failed then passed on retry: count = 3
 async fn should_send_validate_repo_when_retryable_error() -> TestResult {
     //given
     let fs = given::a_filesystem();

crates/server-actor/Cargo.toml

@@ -20,6 +20,7 @@ kxio = { workspace = true }
 
 # boilerplate
 derive_more = { workspace = true }
+derive-with = { workspace = true }
 
 # Actors
 actix = { workspace = true }
@@ -27,6 +28,8 @@ actix = { workspace = true }
 [dev-dependencies]
 # Testing
 # assert2 = { workspace = true }
+test-log = { workspace = true }
+tokio = { workspace = true }
 
 [lints.clippy]
 nursery = { level = "warn", priority = -1 }

crates/server-actor/src/handlers/file_updated.rs

@@ -0,0 +1,21 @@
+//-
+use actix::prelude::*;
+use git_next_config::server::ServerConfig;
+use git_next_file_watcher_actor::FileUpdated;
+
+use crate::{messages::ReceiveServerConfig, Server};
+
+impl Handler<FileUpdated> for Server {
+    type Result = ();
+
+    fn handle(&mut self, _msg: FileUpdated, ctx: &mut Self::Context) -> Self::Result {
+        let server_config = match ServerConfig::load(&self.fs) {
+            Ok(server_config) => server_config,
+            Err(err) => {
+                tracing::error!("Failed to load config file. Error: {}", err);
+                return;
+            }
+        };
+        self.do_send(ReceiveServerConfig::new(server_config), ctx);
+    }
+}

crates/server-actor/src/handlers/mod.rs

@@ -0,0 +1,4 @@
+mod file_updated;
+mod receive_server_config;
+mod receive_valid_server_config;
+mod shutdown;

crates/server-actor/src/handlers/receive_server_config.rs

@@ -0,0 +1,38 @@
+use actix::prelude::*;
+
+use crate::{
+    messages::{ReceiveServerConfig, ReceiveValidServerConfig, ValidServerConfig},
+    Server,
+};
+
+impl Handler<ReceiveServerConfig> for Server {
+    type Result = ();
+
+    #[allow(clippy::cognitive_complexity)]
+    fn handle(&mut self, msg: ReceiveServerConfig, ctx: &mut Self::Context) -> Self::Result {
+        tracing::info!("recieved server config");
+        let Ok(socket_addr) = msg.http() else {
+            tracing::error!("Unable to parse http.addr");
+            return;
+        };
+
+        let Some(server_storage) = self.server_storage(&msg) else {
+            tracing::error!("Server storage not available");
+            return;
+        };
+
+        if msg.webhook().base_url().ends_with('/') {
+            tracing::error!("webhook.url must not end with a '/'");
+            return;
+        }
+
+        self.do_send(
+            ReceiveValidServerConfig::new(ValidServerConfig::new(
+                msg.unwrap(),
+                socket_addr,
+                server_storage,
+            )),
+            ctx,
+        );
+    }
+}

crates/server-actor/src/handlers/receive_valid_server_config.rs

@@ -0,0 +1,39 @@
+use actix::prelude::*;
+use git_next_webhook_actor::{AddWebhookRecipient, ShutdownWebhook, WebhookActor, WebhookRouter};
+
+use crate::{
+    messages::{ReceiveValidServerConfig, ValidServerConfig},
+    Server,
+};
+
+impl Handler<ReceiveValidServerConfig> for Server {
+    type Result = ();
+
+    fn handle(&mut self, msg: ReceiveValidServerConfig, _ctx: &mut Self::Context) -> Self::Result {
+        let ValidServerConfig {
+            server_config,
+            socket_address,
+            server_storage,
+        } = msg.unwrap();
+        if let Some(webhook) = self.webhook.take() {
+            webhook.do_send(ShutdownWebhook);
+        }
+        self.generation.inc();
+        // Webhook Server
+        tracing::info!("Starting Webhook Server...");
+        let webhook_router = WebhookRouter::default().start();
+        let webhook = server_config.webhook();
+        // Forge Actors
+        for (forge_alias, forge_config) in server_config.forges() {
+            self.create_forge_repos(forge_config, forge_alias.clone(), &server_storage, webhook)
+                .into_iter()
+                .map(|a| self.start_actor(a))
+                .map(|(repo_alias, addr)| {
+                    AddWebhookRecipient::new(forge_alias.clone(), repo_alias, addr.recipient())
+                })
+                .for_each(|msg| webhook_router.do_send(msg));
+        }
+        let webhook = WebhookActor::new(socket_address, webhook_router.recipient()).start();
+        self.webhook.replace(webhook);
+    }
+}

crates/server-actor/src/handlers/shutdown.rs

@@ -0,0 +1,15 @@
+//-
+use actix::prelude::*;
+use git_next_webhook_actor::ShutdownWebhook;
+
+use crate::{messages::Shutdown, Server};
+
+impl Handler<Shutdown> for Server {
+    type Result = ();
+
+    fn handle(&mut self, _msg: Shutdown, _ctx: &mut Self::Context) -> Self::Result {
+        if let Some(webhook) = self.webhook.take() {
+            webhook.do_send(ShutdownWebhook);
+        }
+    }
+}

crates/server-actor/src/lib.rs

@@ -1,25 +1,28 @@
 //
+#[cfg(test)]
+mod tests;
+
+mod handlers;
+pub mod messages;
+
 use actix::prelude::*;
-use git_next_actor_macros::message;
 use git_next_config as config;
 use git_next_config::server::{ServerConfig, ServerStorage, Webhook};
 use git_next_config::{ForgeAlias, ForgeConfig, GitDir, RepoAlias, ServerRepoConfig};
-use git_next_file_watcher_actor::FileUpdated;
 use git_next_git::{Generation, RepoDetails};
 use git_next_repo_actor::{messages::CloneRepo, RepoActor};
 use git_next_webhook_actor as webhook;
 use kxio::{fs::FileSystem, network::Network};
-use std::path::PathBuf;
-use tracing::{error, info, warn};
-use webhook::{AddWebhookRecipient, ShutdownWebhook, WebhookActor, WebhookRouter};
+use std::{
+    path::PathBuf,
+    sync::{Arc, RwLock},
+};
+use tracing::{error, info};
+use webhook::WebhookActor;
 
 pub use git_next_git::repository::{factory::real as repository_factory, RepositoryFactory};
 
-message!(ReceiveServerConfig: ServerConfig: "Notification of newly loaded server configuration.
-
-This message will prompt the `git-next` server to stop and restart all repo-actors.
-
-Contains the new server configuration.");
+use crate::messages::ReceiveServerConfig;
 
 #[derive(Debug, derive_more::Display, derive_more::From)]
 pub enum Error {
@@ -37,6 +40,8 @@ pub enum Error {
 }
 type Result<T> = core::result::Result<T, Error>;
 
+#[derive(derive_with::With)]
+#[with(message_log)]
 pub struct Server {
     generation: Generation,
     webhook: Option<Addr<WebhookActor>>,
@@ -44,80 +49,14 @@ pub struct Server {
     net: Network,
     repository_factory: Box<dyn RepositoryFactory>,
     sleep_duration: std::time::Duration,
+
+    // testing
+    message_log: Option<Arc<RwLock<Vec<String>>>>,
 }
 impl Actor for Server {
     type Context = Context<Self>;
 }
-impl Handler<FileUpdated> for Server {
-    type Result = ();
 
-    fn handle(&mut self, _msg: FileUpdated, ctx: &mut Self::Context) -> Self::Result {
-        let server_config = match ServerConfig::load(&self.fs) {
-            Ok(server_config) => server_config,
-            Err(err) => {
-                error!("Failed to load config file. Error: {}", err);
-                return;
-            }
-        };
-        ctx.notify(ReceiveServerConfig::new(server_config));
-    }
-}
-impl Handler<ReceiveServerConfig> for Server {
-    type Result = ();
-
-    #[allow(clippy::cognitive_complexity)] // TODO: (#75) reduce complexity
-    fn handle(&mut self, msg: ReceiveServerConfig, _ctx: &mut Self::Context) -> Self::Result {
-        let Ok(socket_addr) = msg.http() else {
-            warn!("Unable to parse http.addr");
-            return;
-        };
-        if let Some(webhook) = self.webhook.take() {
-            webhook.do_send(ShutdownWebhook);
-        }
-        self.generation.inc();
-        let server_config = msg;
-        // Server Storage
-        let dir = server_config.storage().path();
-        if !dir.exists() {
-            info!(?dir, "server storage doesn't exist - creating it");
-            if let Err(err) = self.fs.dir_create(dir) {
-                error!(?err, ?dir, "Failed to create server storage");
-                return;
-            }
-        }
-        let Ok(canon) = dir.canonicalize() else {
-            error!(?dir, "Failed to confirm server storage");
-            return;
-        };
-        info!(dir = ?canon, "server storage");
-
-        // Forge directories in Server Storage
-        let server_storage = server_config.storage();
-        if let Err(err) = self.create_forge_data_directories(&server_config, dir) {
-            error!(?err, "Failure creating forge storage");
-            return;
-        }
-
-        // Webhook Server
-        info!("Starting Webhook Server...");
-        let webhook_router = WebhookRouter::default().start();
-        let webhook = server_config.webhook();
-
-        // Forge Actors
-        for (forge_alias, forge_config) in server_config.forges() {
-            self.create_forge_repos(forge_config, forge_alias.clone(), server_storage, webhook)
-                .into_iter()
-                .map(|a| self.start_actor(a))
-                .map(|(repo_alias, addr)| {
-                    AddWebhookRecipient::new(forge_alias.clone(), repo_alias, addr.recipient())
-                })
-                .for_each(|msg| webhook_router.do_send(msg));
-        }
-
-        let webhook = WebhookActor::new(socket_addr, webhook_router.recipient()).start();
-        self.webhook.replace(webhook);
-    }
-}
 impl Server {
     pub fn new(
         fs: FileSystem,
@@ -133,6 +72,7 @@ impl Server {
             net,
             repository_factory: repo,
             sleep_duration,
+            message_log: None,
         }
     }
     fn create_forge_data_directories(
@@ -247,4 +187,43 @@ impl Server {
         info!("Started");
         (repo_alias, addr)
     }
+
+    fn server_storage(&self, server_config: &ReceiveServerConfig) -> Option<ServerStorage> {
+        let server_storage = server_config.storage().clone();
+        let dir = server_storage.path();
+        if !dir.exists() {
+            if let Err(err) = self.fs.dir_create(dir) {
+                error!(?err, ?dir, "Failed to create server storage");
+                return None;
+            }
+        }
+        let Ok(canon) = dir.canonicalize() else {
+            error!(?dir, "Failed to confirm server storage");
+            return None;
+        };
+        if let Err(err) = self.create_forge_data_directories(server_config, &canon) {
+            error!(?err, "Failure creating forge storage");
+            return None;
+        }
+        Some(server_storage)
+    }
+
+    fn do_send<M>(&mut self, msg: M, _ctx: &mut <Self as actix::Actor>::Context)
+    where
+        M: actix::Message + Send + 'static + std::fmt::Debug,
+        Self: actix::Handler<M>,
+        <M as actix::Message>::Result: Send,
+    {
+        tracing::info!(?msg, "send");
+        if let Some(message_log) = &self.message_log {
+            let log_message = format!("send: {:?}", msg);
+            tracing::debug!(log_message);
+            if let Ok(mut log) = message_log.write() {
+                log.push(log_message);
+            }
+        }
+        #[cfg(not(test))]
+        _ctx.address().do_send(msg);
+        tracing::info!("sent");
+    }
 }

crates/server-actor/src/messages.rs

@@ -0,0 +1,23 @@
+//-
+use derive_more::Constructor;
+use git_next_actor_macros::message;
+use git_next_config::server::{ServerConfig, ServerStorage};
+use std::net::SocketAddr;
+
+// receive server config
+message!(ReceiveServerConfig: ServerConfig: "Notification of newly loaded server configuration.
+
+This message will prompt the `git-next` server to stop and restart all repo-actors.
+
+Contains the new server configuration.");
+
+// receive valid server config
+#[derive(Clone, Debug, PartialEq, Eq, Constructor)]
+pub struct ValidServerConfig {
+    pub server_config: ServerConfig,
+    pub socket_address: SocketAddr,
+    pub server_storage: ServerStorage,
+}
+message!(ReceiveValidServerConfig: ValidServerConfig: "Notification of validated server configuration.");
+
+message!(Shutdown: "Notification to shutdown the server actor");

crates/server-actor/src/tests/given.rs

@@ -0,0 +1,7 @@
+pub fn a_filesystem() -> kxio::fs::FileSystem {
+    kxio::fs::temp().unwrap_or_else(|e| panic!("{}", e))
+}
+
+pub fn a_network() -> kxio::network::MockNetwork {
+    kxio::network::MockNetwork::new()
+}

crates/server-actor/src/tests/mod.rs

@@ -0,0 +1,3 @@
+mod receive_server_config;
+
+mod given;

crates/server-actor/src/tests/receive_server_config.rs

@@ -0,0 +1,49 @@
+//
+use crate::{tests::given, ReceiveServerConfig, Server};
+use actix::prelude::*;
+use git_next_config::server::{Http, ServerConfig, ServerStorage, Webhook};
+use std::{
+    collections::BTreeMap,
+    sync::{Arc, RwLock},
+};
+
+#[test_log::test(actix::test)]
+async fn when_webhook_url_has_trailing_slash_should_not_send() {
+    //given
+    // parameters
+    let fs = given::a_filesystem();
+    let net = given::a_network();
+    let repo = git_next_git::repository::factory::mock();
+    let duration = std::time::Duration::from_millis(1);
+
+    // sut
+    let server = Server::new(fs.clone(), net.into(), repo, duration);
+
+    // collaborators
+    let http = Http::new("0.0.0.0".to_string(), 80);
+    let webhook = Webhook::new("http://localhost/".to_string()); // With trailing slash
+    let server_storage = ServerStorage::new((fs.base()).to_path_buf());
+    let repos = BTreeMap::default();
+
+    // debugging
+    let message_log: Arc<RwLock<Vec<String>>> = Arc::new(RwLock::new(vec![]));
+    let server = server.with_message_log(Some(message_log.clone()));
+
+    //when
+    server
+        .start()
+        .do_send(ReceiveServerConfig::new(ServerConfig::new(
+            http,
+            webhook,
+            server_storage,
+            repos,
+        )));
+    tokio::time::sleep(std::time::Duration::from_millis(1)).await;
+
+    //then
+    // INFO: assert that ReceiveValidServerConfig is NOT sent
+    tracing::debug!(?message_log, "");
+    assert!(message_log.read().iter().any(|log| !log
+        .iter()
+        .any(|line| line != "send: ReceiveValidServerConfig")));
+}

crates/server/src/lib.rs

@@ -40,12 +40,11 @@ pub fn start(
 
     info!("Starting Server...");
     let execution = async move {
-        //-
         let server = Server::new(fs.clone(), net.clone(), repo, sleep_duration).start();
         server.do_send(FileUpdated);
 
         info!("Starting File Watcher...");
-        let fw = match FileWatcher::new("git-next-server.toml".into(), server.recipient()) {
+        let fw = match FileWatcher::new("git-next-server.toml".into(), server.clone().recipient()) {
             Ok(fw) => fw,
             Err(err) => {
                 error!(?err, "Failed to start file watcher");
@@ -57,6 +56,7 @@ pub fn start(
         info!("Server running - Press Ctrl-C to stop...");
         let _ = actix_rt::signal::ctrl_c().await;
         info!("Ctrl-C received, shutting down...");
+        server.do_send(git_next_server_actor::messages::Shutdown);
         System::current().stop();
     };
     let system = System::new();