feat: Various minor improvements

- Parameter order consistency.
- Further `TaskTracker` adoption.
- Exit early if no plugin initializes successfully.
- Move the log CLI arguments to the bottom and simplify argument help messages.

Author: celarye

src/cli.rs

@@ -24,13 +24,10 @@ impl From<String> for CliLogParametersFileRotation {
 #[derive(Parser)]
 #[command(about, long_about = None, version, author)]
 pub struct Cli {
-    #[command(flatten)]
-    pub log_parameters: CliLogParameters,
-
     #[arg(default_value = "./config.yaml", short, long, value_name = "FILE PATH", help = "The path to the program its configuration file", long_help = None)]
     pub config_file: PathBuf,
 
-    #[arg(default_value = "./.env", short, long, value_name = "FILE PATH", help = "The path to an env file, used by the program its configuration file for env var interpolation", long_help = None)]
+    #[arg(default_value = "./.env", short, long, value_name = "FILE PATH", help = "The path to the program its env file", long_help = None)]
     pub env_file: PathBuf,
 
     #[arg(default_value = "./plugins", short, long, value_name = "DIRECTORY PATH", help = "The path to the program its plugin directory", long_help = None)]
@@ -41,6 +38,9 @@ pub struct Cli {
 
     #[arg(default_value_t = false, short, long, help = "Run in restricted mode, in this case plugin permissions are opt in", long_help = None)]
     pub restricted: bool,
+
+    #[command(flatten)]
+    pub log_parameters: CliLogParameters,
 }
 
 #[derive(Args)]

src/config.rs

@@ -6,6 +6,7 @@ use std::{collections::HashMap, fs, path::Path};
 use anyhow::Result;
 use serde::Deserialize;
 use tracing::info;
+use uuid::Uuid;
 
 use crate::config::{plugins::ConfigPlugin, services::ConfigServices};
 
@@ -14,13 +15,18 @@ pub mod services;
 
 #[derive(Deserialize)]
 pub struct Config {
+    #[serde(default = "Config::default_name")]
     pub name: String,
     #[serde(default)]
     pub services: ConfigServices,
     pub plugins: HashMap<String, ConfigPlugin>,
 }
 
 impl Config {
+    fn default_name() -> String {
+        Uuid::new_v4().to_string()
+    }
+
     pub fn new(file_path: &Path, restricted: bool) -> Result<Self> {
         info!("Loading and parsing the config file");
 

src/main.rs

@@ -101,21 +101,21 @@ async fn main() -> Result<ExitCode> {
     let database = database::new(&cli.database_directory)?;
 
     let message_handler = message_handler(
-        database.clone(),
         Arc::new(RwLock::new(Some(channels.core.runtime_tx))),
         Arc::new(RwLock::new(channels.core.job_scheduler_tx)),
         Arc::new(RwLock::new(channels.core.discord_tx)),
-        Arc::new(shutdown_signal_listener),
         channels.core.rx,
+        database.clone(),
+        Arc::new(shutdown_signal_listener),
     );
 
     let setup_result = setup(
         cli.plugin_directory,
-        database,
-        channels.services,
-        channels.runtime,
         config,
         secrets,
+        channels.services,
+        channels.runtime,
+        database,
     )
     .await;
 
@@ -129,12 +129,12 @@ async fn main() -> Result<ExitCode> {
 }
 
 fn message_handler(
-    database: Database,
     runtime_tx: Arc<RwLock<Option<UnboundedSender<RuntimeMessages>>>>,
     job_scheduler_tx: Arc<RwLock<Option<UnboundedSender<JobSchedulerMessages>>>>,
     discord_tx: Arc<RwLock<Option<UnboundedSender<DiscordMessages>>>>,
-    shutdown_signal_listener: Arc<JoinHandle<()>>,
     mut rx: UnboundedReceiver<CoreMessages>,
+    database: Database,
+    shutdown_signal_listener: Arc<JoinHandle<()>>,
 ) -> JoinHandle<Result<()>> {
     debug!("Starting the message handler");
 
@@ -158,11 +158,11 @@ fn message_handler(
                 }
                 CoreMessages::Shutdown(shutdown_kind) => {
                     tokio::spawn(shutdown(
-                        shutdown_kind,
                         runtime_tx.clone(),
                         job_scheduler_tx.clone(),
                         discord_tx.clone(),
                         shutdown_signal_listener.clone(),
+                        shutdown_kind,
                     ));
                 }
             }
@@ -174,27 +174,27 @@ fn message_handler(
 
 async fn setup(
     plugin_directory_path: PathBuf,
-    database: Database,
-    service_channels: ChannelsServices,
-    runtime_channels: ChannelsRuntime,
     config: Config,
     secrets: Secrets,
+    service_channels: ChannelsServices,
+    runtime_channels: ChannelsRuntime,
+    database: Database,
 ) -> Result<()> {
     let config_name = Arc::new(config.name);
 
     let available_plugins = registry::get_plugins(
         &plugin_directory_path,
-        database.clone(),
         config_name.clone(),
         config.plugins,
+        database.clone(),
     )
     .await?;
 
     services::setup(
         config.services,
         secrets.services,
-        database.clone(),
         service_channels,
+        database.clone(),
     )
     .await?;
 
@@ -204,9 +204,9 @@ async fn setup(
         .initialize_plugins(
             plugin_directory_path,
             config_name,
-            available_plugins,
-            database,
             runtime_channels.core_tx,
+            database,
+            available_plugins,
         )
         .await?;
 
@@ -293,11 +293,11 @@ fn shutdown_signal_listener(core_tx: UnboundedSender<CoreMessages>) -> JoinHandl
 }
 
 async fn shutdown(
-    shutdown_kind: Shutdown,
     runtime_tx: Arc<RwLock<Option<UnboundedSender<RuntimeMessages>>>>,
     job_scheduler_tx: Arc<RwLock<Option<UnboundedSender<JobSchedulerMessages>>>>,
     discord_tx: Arc<RwLock<Option<UnboundedSender<DiscordMessages>>>>,
     shutdown_signal_listener: Arc<JoinHandle<()>>,
+    shutdown_kind: Shutdown,
 ) {
     let mut shutdown_guard = SHUTDOWN.write().await;
 

src/registry.rs

@@ -27,9 +27,9 @@ static DEFAULT_NAMESPACE_ID: &str = "wpbs-rs";
 #[hotpath::measure]
 pub async fn get_plugins(
     plugin_directory_path: &Path,
-    database: Database,
     config_name: Arc<String>,
     config_plugins: HashMap<String, ConfigPlugin>,
+    database: Database,
 ) -> Result<Vec<(Uuid, AvailablePlugin)>> {
     info!("Getting all plugins from their respective registries");
 

src/runtime.rs

@@ -6,7 +6,7 @@ pub mod plugins;
 
 use std::{collections::HashMap, path::PathBuf, sync::Arc};
 
-use anyhow::Result;
+use anyhow::{Result, bail};
 use fjall::Database;
 use tokio::{
     fs,
@@ -172,26 +172,25 @@ impl Runtime {
         &self,
         plugins_directory_path: PathBuf,
         config_name: Arc<String>,
-        available_plugins: Vec<(Uuid, AvailablePlugin)>,
-        database: Database,
         core_tx: UnboundedSender<CoreMessages>,
+        database: Database,
+        available_plugins: Vec<(Uuid, AvailablePlugin)>,
     ) -> Result<()> {
         info!("Initializing the plugins");
 
         let plugins_directory_path = Arc::new(plugins_directory_path);
 
-        let mut tasks = Vec::new();
+        let task_tracker = TaskTracker::new();
 
-        // TODO: Bail on no successful plugin initializations
         for (plugin_uuid, plugin_metadata) in available_plugins {
             let plugins_directory_path = plugins_directory_path.clone();
             let config_name = config_name.clone();
-            let plugins = self.plugins.clone();
-            let plugin_builder = self.plugin_builder.clone();
             let database = database.clone();
             let core_tx = core_tx.clone();
+            let plugins = self.plugins.clone();
+            let plugin_builder = self.plugin_builder.clone();
 
-            tasks.push(tokio::spawn(async move {
+            task_tracker.spawn(async move {
                 let plugin_binary_path = if let Some(content_digest) = &plugin_metadata.content_digest {
                     match content_digest {
                         ContentDigest::Sha256 { hex } => plugins_directory_path.join("binaries").join("remote").join(format!("sha256:{hex}"))
@@ -205,7 +204,7 @@ impl Runtime {
                         .join(plugin_metadata.version.to_string()).join("plugin.wasm")
                 };
 
-                // TODO: Make this configurable
+                // TODO: Make workspaces configurable
                 let plugin_workspace_path = plugins_directory_path
                     .join("workspaces")
                     .join(&*config_name)
@@ -326,11 +325,14 @@ impl Runtime {
                 });
 
                 plugins.write().await.insert(plugin_uuid, plugin_context);
-            }));
+            });
         }
 
-        for task in tasks {
-            task.await.unwrap();
+        task_tracker.close();
+        task_tracker.wait().await;
+
+        if self.plugins.read().await.is_empty() {
+            bail!("No plugin initialized successfully")
         }
 
         Ok(())

src/services.rs

@@ -21,8 +21,8 @@ pub mod job_scheduler;
 pub async fn setup(
     config: ConfigServices,
     secrets: SecretsServices,
-    database: Database,
     channels: ChannelsServices,
+    database: Database,
 ) -> Result<()> {
     // TODO:
     // - Make service starts concurrent
@@ -39,9 +39,9 @@ pub async fn setup(
         let discord = Discord::new(
             config.discord.settings,
             secrets.discord.unwrap(),
-            database,
             discord_channels.core_tx,
             discord_channels.rx,
+            database,
         )
         .await?;
 

src/services/discord.rs

@@ -49,9 +49,9 @@ impl Discord {
     pub async fn new(
         config: ConfigDiscordSettings,
         secrets: SecretsDiscord,
-        database: Database,
         core_tx: UnboundedSender<CoreMessages>,
         rx: UnboundedReceiver<DiscordMessages>,
+        database: Database,
     ) -> Result<Self> {
         info!("Creating the Discord service");
 
@@ -109,17 +109,17 @@ impl Discord {
     pub fn run(mut self) -> JoinHandle<()> {
         info!("Starting the Discord service");
 
-        let mut shard_tasks = Vec::with_capacity(self.shards.len());
+        let shard_task_tracker = TaskTracker::new();
         let http_task_tracker = TaskTracker::new();
 
         for shard in self.shards.drain(..) {
-            shard_tasks.push(tokio::spawn(Self::shard_runner(
+            shard_task_tracker.spawn(Self::shard_runner(
                 self.database.clone(),
                 self.cache.clone(),
                 self.core_tx.clone(),
                 shard.0,
                 shard.1,
-            )));
+            ));
         }
 
         tokio::spawn(async move {
@@ -152,7 +152,7 @@ impl Discord {
             http_task_tracker.close();
             http_task_tracker.wait().await;
 
-            self.shutdown(shard_tasks).await;
+            self.shutdown(shard_task_tracker).await;
         })
     }
 
@@ -187,15 +187,14 @@ impl Discord {
         }
     }
 
-    async fn shutdown(&self, tasks: Vec<JoinHandle<()>>) {
+    async fn shutdown(&self, shard_task_tracker: TaskTracker) {
         info!("Shutting the Discord service down");
 
         for shard_message_sender in self.shard_message_senders.iter() {
-            _ = shard_message_sender.close(CloseFrame::NORMAL);
+            let _ = shard_message_sender.close(CloseFrame::NORMAL);
         }
 
-        for task in tasks {
-            task.await.unwrap();
-        }
+        shard_task_tracker.close();
+        shard_task_tracker.wait().await;
     }
 }

src/services/discord/requests.rs

@@ -23,6 +23,7 @@ use crate::{
 };
 
 impl Discord {
+    // TODO: Split up in sub functions
     #[allow(clippy::too_many_lines)]
     pub async fn request(
         http_client: Arc<Client>,