Compare commits

...

5 commits

Author SHA1 Message Date
fd762e2bd2 feat: perform controlled shutdown on ctrl-c
All checks were successful
Rust / build (push) Successful in 1m50s
ci/woodpecker/push/cron-docker-builder Pipeline was successful
ci/woodpecker/push/push-next Pipeline was successful
ci/woodpecker/push/tag-created Pipeline was successful
ci/woodpecker/cron/cron-docker-builder Pipeline was successful
ci/woodpecker/cron/push-next Pipeline was successful
ci/woodpecker/cron/tag-created Pipeline was successful
Closes kemitix/git-next#94

Controlled shutdown includes attempting to unregister webhooks.
2024-07-11 19:19:04 +01:00
681b2c4c10 refactor: split messages and handlers for server-actor
All checks were successful
Rust / build (push) Successful in 1m46s
ci/woodpecker/push/cron-docker-builder Pipeline was successful
ci/woodpecker/push/push-next Pipeline was successful
ci/woodpecker/push/tag-created Pipeline was successful
2024-07-11 19:19:01 +01:00
7578ab3144 feat: log as an error when webhook url ends with a slash
All checks were successful
Rust / build (push) Successful in 1m26s
ci/woodpecker/push/cron-docker-builder Pipeline was successful
ci/woodpecker/push/push-next Pipeline was successful
ci/woodpecker/push/tag-created Pipeline was successful
Closes kemitix/git-next#87
2024-07-11 19:18:58 +01:00
7212154037 refactor: split ReceiveServerConfig handler
All checks were successful
Rust / build (push) Successful in 1m48s
ci/woodpecker/push/cron-docker-builder Pipeline was successful
ci/woodpecker/push/push-next Pipeline was successful
ci/woodpecker/push/tag-created Pipeline was successful
First handler, with original name, validates the server config.

The new second handler, ReceiveValidServerConfig, can then (re)start the
server without needing to validate the settings.
2024-07-11 19:18:55 +01:00
4276964f4d refactor: split server storage creation out from startup
All checks were successful
Rust / build (push) Successful in 1m34s
ci/woodpecker/push/cron-docker-builder Pipeline was successful
ci/woodpecker/push/push-next Pipeline was successful
ci/woodpecker/push/tag-created Pipeline was successful
Closes kemitix/git-next#75
2024-07-11 19:18:50 +01:00
13 changed files with 263 additions and 82 deletions

View file

@ -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();

View file

@ -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 }

View file

@ -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);
}
}

View file

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

View file

@ -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,
);
}
}

View file

@ -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);
}
}

View file

@ -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);
}
}
}

View file

@ -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");
}
}

View file

@ -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");

View file

@ -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()
}

View file

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

View file

@ -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")));
}

View file

@ -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();