chore: Use shotgun for shutdown signal

backend/Cargo.toml

@@ -31,3 +31,4 @@ serde_json = "1.0"
 tokio = { version = "1", features = ["full"] }
 tokio-util = { version = "0.7.13", features = ["io"] }
 uuid = { version = "1.12.1", features = ["v4"] }
+shotgun = { git = "https://github.com/its-laika/shotgun.git", version = "0.1.2" }

backend/src/api/server.rs

@@ -6,11 +6,11 @@ use axum::{
 };
 use sea_orm::DatabaseConnection;
 use std::io;
-use tokio::{net::TcpListener, sync::broadcast};
+use tokio::net::TcpListener;
 
 pub async fn listen(
     connection: DatabaseConnection,
-    mut shutdown: broadcast::Receiver<()>,
+    shutdown: shotgun::Receiver<()>,
 ) -> io::Result<()> {
     let app = Router::new()
         .route("/files", post(routes::upload::handler))

backend/src/cleanup.rs

@@ -1,14 +1,7 @@
-use crate::{
-    database,
-    error::{Error, Result},
-    file,
-};
+use crate::{database, error::Result, file};
 use sea_orm::DatabaseConnection;
 use std::time::Duration;
-use tokio::{
-    sync::broadcast::{self, error::TryRecvError},
-    time,
-};
+use tokio::{select, time};
 use uuid::Uuid;
 
 /// The interval in seconds between each cleanup operation.
@@ -20,20 +13,23 @@ const CLEANUP_INTERVAL_SECONDS: u64 = 10 * 60; /* 10 minutes */
 /// # Arguments
 ///
 /// * `database_connection` - A connection to the database.
-/// * `shutdown` - A broadcast receiver to listen for shutdown signals.
+/// * `shutdown` - A shotgun receiver to listen for shutdown signal.
 ///
 /// # Returns
 ///
 /// * `Result<()>` - Returns an Ok result if the cleanup process runs successfully,
 ///   or an error if something goes wrong.
 pub async fn run(
     database_connection: DatabaseConnection,
-    mut shutdown: broadcast::Receiver<()>,
+    shutdown: shotgun::Receiver<()>,
 ) -> Result<()> {
     loop {
-        if !wait(&mut shutdown).await? {
-            return Ok(());
-        }
+        let shutdown = shutdown.clone();
+
+        select! {
+            _ = time::sleep(Duration::from_secs(CLEANUP_INTERVAL_SECONDS)) => (),
+            _ = shutdown => return Ok(()),
+        };
 
         log::info!("Cleaning up outdating files...");
 
@@ -42,33 +38,6 @@ pub async fn run(
     }
 }
 
-/// Waits for defined cleanup interval, returning early if `shutdown` is received.
-///
-/// # Arguments
-///
-/// * `shutdown` - A broadcast receiver to listen for shutdown signals.
-///
-/// # Returns
-///
-/// * `Result<bool>` - Returns an Ok result if the wait completes successfully,
-///   with `true` if the wait was not interrupted by a shutdown signal, or `false` otherwise.
-async fn wait(shutdown: &mut broadcast::Receiver<()>) -> Result<bool> {
-    let sleep_time = Duration::from_secs(1);
-
-    for _ in 0..CLEANUP_INTERVAL_SECONDS {
-        /* Prevent locking for 10 mins after a SIGINT like a spin lock */
-        match shutdown.try_recv() {
-            Err(TryRecvError::Empty) => (),
-            Err(_) => return Err(Error::BroadcastRecvFailed),
-            Ok(()) => return Ok(false),
-        };
-
-        time::sleep(sleep_time).await;
-    }
-
-    Ok(true)
-}
-
 /// Deletes outdated files from the file system.
 ///
 /// # Arguments

backend/src/error.rs

@@ -21,7 +21,6 @@ pub enum Error {
     InvalidEncryptionData(String),
     HashingFailure(String),
     HashVerificationFailure(String),
-    BroadcastRecvFailed,
 }
 
 impl fmt::Debug for Error {
@@ -44,7 +43,6 @@ impl fmt::Debug for Error {
             Self::InvalidEncryptionData(inner) => write!(f, "Invalid encryption data: {inner}"),
             Self::HashingFailure(inner) => write!(f, "Hashing failure: {inner}"),
             Self::HashVerificationFailure(inner) => write!(f, "Hash verification failure: {inner}"),
-            Self::BroadcastRecvFailed => write!(f, "Broadcast recv error"),
         }
     }
 }

backend/src/main.rs

@@ -2,7 +2,7 @@ use configuration::CONFIGURATION;
 use migration::{Migrator, MigratorTrait};
 use sea_orm::{ConnectOptions, Database, DatabaseConnection};
 use std::{process, time::Duration};
-use tokio::{signal::ctrl_c, sync::broadcast, task::JoinSet};
+use tokio::{signal::ctrl_c, task::JoinSet};
 
 mod api;
 mod cleanup;
@@ -31,10 +31,10 @@ async fn main() {
     };
 
     let mut join_set = JoinSet::new();
-    let (shutdown_tx, shutdown_rx) = broadcast::channel(1);
+    let (shotgun_tx, shotgun_rx) = shotgun::channel();
 
     let api_database_connection = database_connection.clone();
-    let api_shutdown_rx = shutdown_rx.resubscribe();
+    let api_shutdown_rx = shotgun_rx.clone();
 
     join_set.spawn(async move {
         if let Err(error) = api::listen(api_database_connection, api_shutdown_rx).await {
@@ -43,7 +43,7 @@ async fn main() {
     });
 
     let cleanup_database_connection = database_connection.clone();
-    let cleanup_shutdown_rx = shutdown_rx.resubscribe();
+    let cleanup_shutdown_rx = shotgun_rx.clone();
 
     join_set.spawn(async move {
         if let Err(error) = cleanup::run(cleanup_database_connection, cleanup_shutdown_rx).await {
@@ -62,11 +62,7 @@ async fn main() {
 
         info!("Received ctrl+c (SIGINT)");
 
-        if let Err(error) = shutdown_tx.send(()) {
-            error!("Could not inform about shutdown: {error}");
-            error!("Exiting process. Bye.");
-            process::exit(1);
-        };
+        shotgun_tx.send(());
     });
 
     join_set.join_all().await;