add support clean by group name (-g/--group flag in the clean command)

This command allows to clean tasks only in the selected group

Co-authored-by: Arne Beer <privat@arne.beer>

CHANGELOG.md

@@ -11,6 +11,7 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 - Introduce the `rm` (remove), `re` (restart) and `fo` (follow) subcommand aliases [#245](https://github.com/Nukesor/pueue/issues/245).
 - Allow to set the amount of parallel tasks at group creation by [Spyros Roum](https://github.com/SpyrosRoum) [#245](https://github.com/Nukesor/pueue/issues/249).
 - When calling `pueue` without a subcommand, the `status` command will be called by default [#247](https://github.com/Nukesor/pueue/issues/247).
+- Add the `--group` parameter to the `pueue clean` command [#248](https://github.com/Nukesor/pueue/issues/248)
 
 ## [1.0.3] - 2021-09-15
 

Cargo.lock

@@ -1282,7 +1282,7 @@ dependencies = [
 [[package]]
 name = "pueue-lib"
 version = "0.18.2-alpha.0"
-source = "git+https://github.com/Nukesor/pueue-lib?branch=master#edefd46bdaf40f6bf35ff688d31539fda13bf140"
+source = "git+https://github.com/Nukesor/pueue-lib?branch=master#b14796f452b06967924b04b68453f0dfc166901e"
 dependencies = [
  "async-std",
  "async-tls",

client/cli.rs

@@ -357,6 +357,10 @@ pub enum SubCommand {
         /// Only clean tasks that finished successfully.
         #[clap(short, long)]
         successful_only: bool,
+
+        /// Only clean tasks of a specific group
+        #[clap(short, long)]
+        group: Option<String>,
     },
 
     /// Kill all tasks, clean up afterwards and reset EVERYTHING!

client/client.rs

@@ -505,9 +505,13 @@ impl Client {
                 };
                 Ok(Message::StreamRequest(message))
             }
-            SubCommand::Clean { successful_only } => {
+            SubCommand::Clean {
+                successful_only,
+                group,
+            } => {
                 let message = CleanMessage {
                     successful_only: *successful_only,
+                    group: group.clone(),
                 };
 
                 Ok(Message::Clean(message))

daemon/network/message_handler/clean.rs

@@ -14,9 +14,15 @@ fn construct_success_clean_message(message: CleanMessage) -> String {
         ""
     };
 
+    let group_fix = if let Some(group) = message.group {
+        format!(" in group '{}'", group)
+    } else {
+        String::new()
+    };
+
     format!(
-        "All{} finished tasks have been removed",
-        successfull_only_fix
+        "All{} finished tasks have been removed{}",
+        successfull_only_fix, group_fix
     )
 }
 
@@ -33,10 +39,19 @@ pub fn clean(message: CleanMessage, state: &SharedState) -> Message {
         if !is_task_removable(&state, task_id, &[]) {
             continue;
         }
-        // Check if we should ignore this task, if only successful tasks should be removed.
-        if message.successful_only {
+
+        if message.successful_only || message.group.is_some() {
             if let Some(task) = state.tasks.get(task_id) {
-                if !matches!(task.status, TaskStatus::Done(TaskResult::Success)) {
+                // Check if we should ignore this task, if only successful tasks should be removed.
+                if message.successful_only
+                    && !matches!(task.status, TaskStatus::Done(TaskResult::Success))
+                {
+                    continue;
+                }
+
+                // User's can specify a specific group to be cleaned.
+                // Skip the task if that's the case and the task's group doesn't match.
+                if message.group.is_some() && message.group.as_deref() != Some(&task.group) {
                     continue;
                 }
             }
@@ -58,8 +73,11 @@ mod tests {
     use pretty_assertions::assert_eq;
     use tempfile::TempDir;
 
-    fn get_message(successful_only: bool) -> CleanMessage {
-        CleanMessage { successful_only }
+    fn get_message(successful_only: bool, group: Option<String>) -> CleanMessage {
+        CleanMessage {
+            successful_only,
+            group,
+        }
     }
 
     trait TaskAddable {
@@ -73,18 +91,25 @@ mod tests {
         }
     }
 
-    fn get_clean_test_state() -> (SharedState, TempDir) {
+    /// gets the clean test state with the required groups
+    fn get_clean_test_state(groups: &[&str]) -> (SharedState, TempDir) {
         let (state, tempdir) = get_state();
 
         {
             let mut state = state.lock().unwrap();
 
-            state.add_stub_task("0", group, TaskResult::Success);
-            state.add_stub_task("1", group, TaskResult::Failed(1));
-            state.add_stub_task("2", group, TaskResult::FailedToSpawn("error".to_string()));
-            state.add_stub_task("3", group, TaskResult::Killed);
-            state.add_stub_task("4", group, TaskResult::Errored);
-            state.add_stub_task("5", group, TaskResult::DependencyFailed);
+            for &group in groups {
+                if !state.groups.contains_key(group) {
+                    state.create_group(group);
+                }
+
+                state.add_stub_task("0", group, TaskResult::Success);
+                state.add_stub_task("1", group, TaskResult::Failed(1));
+                state.add_stub_task("2", group, TaskResult::FailedToSpawn("error".to_string()));
+                state.add_stub_task("3", group, TaskResult::Killed);
+                state.add_stub_task("4", group, TaskResult::Errored);
+                state.add_stub_task("5", group, TaskResult::DependencyFailed);
+            }
         }
 
         (state, tempdir)
@@ -95,7 +120,7 @@ mod tests {
         let (state, _tempdir) = get_stub_state();
 
         // Only task 1 will be removed, since it's the only TaskStatus with `Done`.
-        let message = clean(get_message(false), &state);
+        let message = clean(get_message(false, None), &state);
 
         // Return message is correct
         assert!(matches!(message, Message::Success(_)));
@@ -109,10 +134,10 @@ mod tests {
 
     #[test]
     fn clean_normal_for_all_results() {
-        let (state, _tempdir) = get_clean_test_state();
+        let (state, _tempdir) = get_clean_test_state(&[PUEUE_DEFAULT_GROUP]);
 
         // All finished tasks should removed when calling default `clean`.
-        let message = clean(get_message(false), &state);
+        let message = clean(get_message(false, None), &state);
 
         // Return message is correct
         assert!(matches!(message, Message::Success(_)));
@@ -126,11 +151,11 @@ mod tests {
 
     #[test]
     fn clean_successful_only() {
-        let (state, _tempdir) = get_clean_test_state();
+        let (state, _tempdir) = get_clean_test_state(&[PUEUE_DEFAULT_GROUP]);
 
         // Only successfully finished tasks should get removed when
         // calling `clean` with the `successful_only` flag.
-        let message = clean(get_message(true), &state);
+        let message = clean(get_message(true, None), &state);
 
         // Return message is correct
         assert!(matches!(message, Message::Success(_)));
@@ -143,4 +168,50 @@ mod tests {
         assert_eq!(state.tasks.len(), 5);
         assert!(state.tasks.get(&0).is_none());
     }
+
+    #[test]
+    fn clean_only_in_selected_group() {
+        let (state, _tempdir) = get_clean_test_state(&[PUEUE_DEFAULT_GROUP, "other"]);
+
+        // All finished tasks should removed in selected group (other)
+        let message = clean(get_message(false, Some("other".into())), &state);
+
+        // Return message is correct
+        assert!(matches!(message, Message::Success(_)));
+
+        if let Message::Success(text) = message {
+            assert_eq!(
+                text,
+                "All finished tasks have been removed in group 'other'"
+            );
+        };
+
+        // Assert that only the 'other' group has been cleared
+        let state = state.lock().unwrap();
+        assert_eq!(state.tasks.len(), 6);
+        assert!(state.tasks.iter().all(|(_, task)| &task.group != "other"));
+    }
+
+    #[test]
+    fn clean_only_successful_only_in_selected_group() {
+        let (state, _tempdir) = get_clean_test_state(&[PUEUE_DEFAULT_GROUP, "other"]);
+
+        // Only successfully finished tasks should removed in the 'other' group
+        let message = clean(get_message(true, Some("other".into())), &state);
+
+        // Return message is correct
+        assert!(matches!(message, Message::Success(_)));
+
+        if let Message::Success(text) = message {
+            assert_eq!(
+                text,
+                "All successfully finished tasks have been removed in group 'other'"
+            );
+        };
+
+        // Assert that only the first entry has been deleted from the 'other' group (TaskResult::Success)
+        let state = state.lock().unwrap();
+        assert_eq!(state.tasks.len(), 11);
+        assert!(state.tasks.get(&6).is_none());
+    }
 }

tests/unix/clean.rs

@@ -19,6 +19,7 @@ async fn test_normal_clean() -> Result<()> {
     // Send the clean message
     let clean_message = CleanMessage {
         successful_only: false,
+        group: None,
     };
     send_message(shared, Message::Clean(clean_message)).await?;
 
@@ -46,6 +47,7 @@ async fn test_successful_only_clean() -> Result<()> {
     // Send the clean message
     let clean_message = CleanMessage {
         successful_only: true,
+        group: None,
     };
     send_message(shared, Message::Clean(clean_message)).await?;
 
@@ -57,3 +59,83 @@ async fn test_successful_only_clean() -> Result<()> {
 
     Ok(())
 }
+
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+/// Ensure only tasks of the selected group are cleaned up
+async fn test_clean_in_selected_group() -> Result<()> {
+    let (settings, _tempdir, _pid) = threaded_setup()?;
+    let shared = &settings.shared;
+
+    fixtures::add_group_with_slots(shared, "other", 1).await?;
+
+    for group in &[PUEUE_DEFAULT_GROUP, "other"] {
+        for command in &["failing", "ls", "sleep 60", "ls"] {
+            assert_success(fixtures::add_task_to_group(shared, command, group).await?);
+        }
+    }
+
+    // Wait for task6 to start. This implies task[4,5] in the 'other' group being finished.
+    wait_for_task_condition(shared, 6, |task| task.is_running()).await?;
+
+    // Send the clean message
+    let clean_message = CleanMessage {
+        successful_only: false,
+        group: Some("other".to_string()),
+    };
+    send_message(shared, Message::Clean(clean_message)).await?;
+
+    // Assert that task 0 and 1 are still there
+    let state = get_state(shared).await?;
+    assert!(state.tasks.contains_key(&0));
+    assert!(state.tasks.contains_key(&1));
+    assert!(state.tasks.contains_key(&2));
+    assert!(state.tasks.contains_key(&3));
+    // Assert that task 4 and 5 have both been removed
+    assert!(!state.tasks.contains_key(&4));
+    assert!(!state.tasks.contains_key(&5));
+    assert!(state.tasks.contains_key(&6));
+    assert!(state.tasks.contains_key(&7));
+
+    Ok(())
+}
+
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+/// Ensure only successful tasks are removed, if the `-s` flag is set.
+async fn test_clean_successful_only_in_selected_group() -> Result<()> {
+    let (settings, _tempdir, _pid) = threaded_setup()?;
+    let shared = &settings.shared;
+
+    fixtures::add_group_with_slots(shared, "other", 1).await?;
+
+    for group in &[PUEUE_DEFAULT_GROUP, "other"] {
+        for command in &["failing", "ls", "sleep 60", "ls"] {
+            assert_success(fixtures::add_task_to_group(shared, command, group).await?);
+        }
+    }
+
+    // Wait for task6 to start. This implies task[4,5] in the 'other' group being finished.
+    wait_for_task_condition(shared, 6, |task| task.is_running()).await?;
+
+    // Send the clean message
+    let clean_message = CleanMessage {
+        successful_only: true,
+        group: Some("other".to_string()),
+    };
+    send_message(shared, Message::Clean(clean_message)).await?;
+
+    let state = get_state(shared).await?;
+    // group default
+    assert!(state.tasks.contains_key(&0));
+    assert!(state.tasks.contains_key(&1));
+    assert!(state.tasks.contains_key(&2));
+    assert!(state.tasks.contains_key(&3));
+
+    // group other
+    assert!(state.tasks.contains_key(&4));
+    // Task 5 should have been removed.
+    assert!(!state.tasks.contains_key(&5));
+    assert!(state.tasks.contains_key(&6));
+    assert!(state.tasks.contains_key(&7));
+
+    Ok(())
+}