Feedback

Author: slinkydeveloper

cli/src/clients/admin_interface.rs

@@ -65,7 +65,7 @@ pub trait AdminClientInterface {
         id: &str,
     ) -> impl Future<Output = reqwest::Result<Envelope<()>>> + Send + 'static;
 
-    async fn resume_invocation(
+    fn resume_invocation(
         &self,
         id: &str,
         deployment: Option<&str>,
@@ -180,6 +180,18 @@ impl AdminClientInterface for AdminClient {
         )
     }
 
+    fn resume_invocation(
+        &self,
+        id: &str,
+        deployment: Option<&str>,
+    ) -> impl Future<Output = reqwest::Result<Envelope<()>>> + Send + 'static {
+        let mut url = self.versioned_url(["invocations", id, "resume"]);
+        if let Some(deployment) = deployment {
+            url.set_query(Some(&format!("deployment={deployment}")));
+        }
+        self.run(reqwest::Method::PATCH, url)
+    }
+
     fn kill_invocation(
         &self,
         id: &str,
@@ -209,18 +221,6 @@ impl AdminClientInterface for AdminClient {
         self.run(reqwest::Method::PATCH, url)
     }
 
- fn resume_invocation(
-        &self,
-        id: &str,
-        deployment: Option<&str>,
-    ) -> impl Future<Output = reqwest::Result<Envelope<()>>> + Send + 'static {
-        let mut url = self.versioned_url(["invocations", id, "resume"]);
-        if let Some(deployment) = deployment {
-            url.set_query(Some(&format!("deployment={deployment}")));
-        }
-        self.run(reqwest::Method::PATCH, url)
-    }
-
     fn pause_invocation(
         &self,
         id: &str,

cli/src/commands/invocations/cancel.rs

@@ -20,7 +20,8 @@ use crate::clients::batch_execute;
 use crate::clients::datafusion_helpers::find_active_invocations_simple;
 use crate::clients::{self, AdminClientInterface};
 use crate::commands::invocations::{
-    DEFAULT_BATCH_INVOCATIONS_OPERATION_LIMIT, create_query_filter,
+    DEFAULT_BATCH_INVOCATIONS_OPERATION_LIMIT, DEFAULT_BATCH_INVOCATIONS_OPERATION_PRINT_LIMIT,
+    create_query_filter,
 };
 use crate::ui::invocations::render_simple_invocation_list;
 use crate::ui::with_progress;
@@ -43,7 +44,7 @@ pub struct Cancel {
     #[clap(long)]
     pub(super) kill: bool,
     /// Limit the number of fetched invocations
-    #[clap(long, default_value = "DEFAULT_BATCH_INVOCATIONS_OPERATION_LIMIT")]
+    #[clap(long, default_value_t = DEFAULT_BATCH_INVOCATIONS_OPERATION_LIMIT)]
     pub(super) limit: usize,
 }
 
@@ -70,7 +71,10 @@ pub async fn run_cancel(State(env): State<CliEnv>, opts: &Cancel) -> Result<()>
         );
     };
 
-    render_simple_invocation_list(&invocations, DEFAULT_BATCH_INVOCATIONS_OPERATION_LIMIT);
+    render_simple_invocation_list(
+        &invocations,
+        DEFAULT_BATCH_INVOCATIONS_OPERATION_PRINT_LIMIT,
+    );
 
     // Get the invocation and confirm
     let prompt = format!(

cli/src/commands/invocations/kill.rs

@@ -7,6 +7,9 @@
 // As of the Change Date specified in that file, in accordance with
 // the Business Source License, use of this software will be governed
 // by the Apache License, Version 2.0.
+
+use super::DEFAULT_BATCH_INVOCATIONS_OPERATION_LIMIT;
+
 use anyhow::Result;
 use cling::prelude::*;
 
@@ -28,7 +31,7 @@ pub struct Kill {
     /// * `workflowName/key/handler`
     query: String,
     /// Limit the number of fetched invocations
-    #[clap(long, default_value = "500")]
+    #[clap(long, default_value_t = DEFAULT_BATCH_INVOCATIONS_OPERATION_LIMIT)]
     limit: usize,
 }
 

cli/src/commands/invocations/mod.rs

@@ -20,7 +20,9 @@ mod resume;
 use cling::prelude::*;
 use restate_types::identifiers::InvocationId;
 
-const DEFAULT_BATCH_INVOCATIONS_OPERATION_LIMIT: usize = 100;
+const DEFAULT_BATCH_INVOCATIONS_OPERATION_LIMIT: usize = 500;
+const DEFAULT_BATCH_INVOCATIONS_OPERATION_PRINT_LIMIT: usize =
+    DEFAULT_BATCH_INVOCATIONS_OPERATION_LIMIT;
 
 #[derive(Run, Subcommand, Clone)]
 pub enum Invocations {

cli/src/commands/invocations/pause.rs

@@ -13,7 +13,9 @@ use crate::clients::datafusion_helpers::find_active_invocations_simple;
 use crate::clients::{self, AdminClientInterface, batch_execute};
 use crate::ui::invocations::render_simple_invocation_list;
 
-use crate::commands::invocations::DEFAULT_BATCH_INVOCATIONS_OPERATION_LIMIT;
+use crate::commands::invocations::{
+    DEFAULT_BATCH_INVOCATIONS_OPERATION_LIMIT, DEFAULT_BATCH_INVOCATIONS_OPERATION_PRINT_LIMIT,
+};
 use anyhow::{Result, anyhow, bail};
 use cling::prelude::*;
 use comfy_table::{Cell, Color, Table};
@@ -33,7 +35,7 @@ pub struct Pause {
     /// * `virtualObjectName/key/handler`
     query: String,
     /// Limit the number of fetched invocations
-    #[clap(long, default_value = "DEFAULT_BATCH_INVOCATIONS_OPERATION_LIMIT")]
+    #[clap(long, default_value_t = DEFAULT_BATCH_INVOCATIONS_OPERATION_LIMIT)]
     limit: usize,
 }
 
@@ -67,7 +69,10 @@ pub async fn run_pause(State(env): State<CliEnv>, opts: &Pause) -> Result<()> {
         );
     };
 
-    render_simple_invocation_list(&invocations, DEFAULT_BATCH_INVOCATIONS_OPERATION_LIMIT);
+    render_simple_invocation_list(
+        &invocations,
+        DEFAULT_BATCH_INVOCATIONS_OPERATION_PRINT_LIMIT,
+    );
 
     // Get the invocation and confirm
     confirm_or_exit("Are you sure you want to pause these invocations?")?;

cli/src/commands/invocations/purge.rs

@@ -12,7 +12,8 @@ use crate::cli_env::CliEnv;
 use crate::clients::datafusion_helpers::find_active_invocations_simple;
 use crate::clients::{self, AdminClientInterface, batch_execute};
 use crate::commands::invocations::{
-    DEFAULT_BATCH_INVOCATIONS_OPERATION_LIMIT, create_query_filter,
+    DEFAULT_BATCH_INVOCATIONS_OPERATION_LIMIT, DEFAULT_BATCH_INVOCATIONS_OPERATION_PRINT_LIMIT,
+    create_query_filter,
 };
 use crate::ui::invocations::render_simple_invocation_list;
 
@@ -38,7 +39,7 @@ pub struct Purge {
     /// * `workflowName/key/handler`
     query: String,
     /// Limit the number of fetched invocations
-    #[clap(long, default_value = "DEFAULT_BATCH_INVOCATIONS_OPERATION_LIMIT")]
+    #[clap(long, default_value_t = DEFAULT_BATCH_INVOCATIONS_OPERATION_LIMIT)]
     limit: usize,
 }
 
@@ -61,7 +62,10 @@ pub async fn run_purge(State(env): State<CliEnv>, opts: &Purge) -> Result<()> {
         );
     };
 
-    render_simple_invocation_list(&invocations, DEFAULT_BATCH_INVOCATIONS_OPERATION_LIMIT);
+    render_simple_invocation_list(
+        &invocations,
+        DEFAULT_BATCH_INVOCATIONS_OPERATION_PRINT_LIMIT,
+    );
 
     // Get the invocation and confirm
     confirm_or_exit("Are you sure you want to purge these invocations?")?;

cli/src/commands/invocations/restart_as_new.rs

@@ -14,7 +14,8 @@ use crate::clients::{self, AdminClientInterface, batch_execute};
 use crate::ui::invocations::render_simple_invocation_list;
 
 use crate::commands::invocations::{
-    DEFAULT_BATCH_INVOCATIONS_OPERATION_LIMIT, create_query_filter,
+    DEFAULT_BATCH_INVOCATIONS_OPERATION_LIMIT, DEFAULT_BATCH_INVOCATIONS_OPERATION_PRINT_LIMIT,
+    create_query_filter,
 };
 use anyhow::{Result, anyhow, bail};
 use cling::prelude::*;
@@ -35,7 +36,7 @@ pub struct RestartAsNew {
     /// * `virtualObjectName/key/handler`
     query: String,
     /// Limit the number of fetched invocations
-    #[clap(long, default_value = "DEFAULT_BATCH_INVOCATIONS_OPERATION_LIMIT")]
+    #[clap(long, default_value_t = DEFAULT_BATCH_INVOCATIONS_OPERATION_LIMIT)]
     limit: usize,
 }
 
@@ -57,7 +58,10 @@ pub async fn run_restart_as_new(State(env): State<CliEnv>, opts: &RestartAsNew)
         );
     };
 
-    render_simple_invocation_list(&invocations, DEFAULT_BATCH_INVOCATIONS_OPERATION_LIMIT);
+    render_simple_invocation_list(
+        &invocations,
+        DEFAULT_BATCH_INVOCATIONS_OPERATION_PRINT_LIMIT,
+    );
 
     // Get the invocation and confirm
     confirm_or_exit("Are you sure you want to restart these invocations?")?;

cli/src/commands/invocations/resume.rs

@@ -14,7 +14,8 @@ use crate::clients::{self, AdminClientInterface, batch_execute};
 use crate::ui::invocations::render_simple_invocation_list;
 
 use crate::commands::invocations::{
-    DEFAULT_BATCH_INVOCATIONS_OPERATION_LIMIT, create_query_filter,
+    DEFAULT_BATCH_INVOCATIONS_OPERATION_LIMIT, DEFAULT_BATCH_INVOCATIONS_OPERATION_PRINT_LIMIT,
+    create_query_filter,
 };
 use anyhow::{Result, anyhow, bail};
 use cling::prelude::*;
@@ -41,7 +42,7 @@ pub struct Resume {
     #[clap(long)]
     deployment: Option<String>,
     /// Limit the number of fetched invocations
-    #[clap(long, default_value = "DEFAULT_BATCH_INVOCATIONS_OPERATION_LIMIT")]
+    #[clap(long, default_value_t = DEFAULT_BATCH_INVOCATIONS_OPERATION_LIMIT)]
     limit: usize,
 }
 
@@ -64,21 +65,30 @@ pub async fn run_resume(State(env): State<CliEnv>, opts: &Resume) -> Result<()>
         );
     };
 
-    render_simple_invocation_list(&invocations, DEFAULT_BATCH_INVOCATIONS_OPERATION_LIMIT);
+    render_simple_invocation_list(
+        &invocations,
+        DEFAULT_BATCH_INVOCATIONS_OPERATION_PRINT_LIMIT,
+    );
 
     // Get the invocation and confirm
     confirm_or_exit("Are you sure you want to resume these invocations?")?;
 
     // Resume invocations
-    let deployment = opts.deployment.as_deref();
-    let (resumed, failed_to_resume) =
-        batch_execute(client, invocations, |client, invocation| async move {
+    let deployment = opts.deployment.clone();
+    let (resumed, failed_to_resume) = batch_execute(
+        client,
+        invocations
+            .into_iter()
+            .map(|i| (i, deployment.clone()))
+            .collect(),
+        |client, (invocation, deployment)| async move {
             client
-                .resume_invocation(&invocation.id, deployment)
+                .resume_invocation(&invocation.id, deployment.as_deref())
                 .await
                 .map_err(anyhow::Error::from)
-        })
-        .await;
+        },
+    )
+    .await;
     let succeeded_count = resumed.len();
     let failed_count = failed_to_resume.len();
 
@@ -90,7 +100,7 @@ pub async fn run_resume(State(env): State<CliEnv>, opts: &Resume) -> Result<()>
         c_warn!("Failed to resume:");
         let mut failed_to_restart_table = Table::new_styled();
         failed_to_restart_table.set_styled_header(vec!["ID", "REASON"]);
-        for (inv, reason) in failed_to_resume {
+        for ((inv, _), reason) in failed_to_resume {
             failed_to_restart_table.add_row(vec![
                 Cell::new(&inv.id),
                 Cell::new(reason).fg(Color::DarkRed),