feat(meta): try to report the root cause of recovery

src/meta/src/barrier/rpc.rs

@@ -15,12 +15,13 @@
 use std::collections::HashMap;
 use std::future::Future;
 use std::sync::Arc;
+use std::time::Duration;
 
 use anyhow::anyhow;
 use fail::fail_point;
-use futures::future::try_join_all;
+use futures::future::{select, Either};
 use futures::stream::FuturesUnordered;
-use futures::{FutureExt, StreamExt};
+use futures::{pin_mut, FutureExt, StreamExt};
 use itertools::Itertools;
 use risingwave_common::bail;
 use risingwave_common::hash::ActorId;
@@ -40,7 +41,7 @@ use uuid::Uuid;
 use super::command::CommandContext;
 use super::{BarrierCompletion, GlobalBarrierManagerContext};
 use crate::manager::{MetaSrvEnv, WorkerId};
-use crate::MetaResult;
+use crate::{MetaError, MetaResult};
 
 pub(super) struct BarrierRpcManager {
     context: GlobalBarrierManagerContext,
@@ -268,11 +269,18 @@ impl StreamRpcManager {
     ) -> MetaResult<Vec<RSP>> {
         let pool = self.env.stream_client_pool();
         let f = &f;
-        Ok(try_join_all(request.map(|(node, input)| async move {
-            let client = pool.get(node).await?;
-            f(client, input).await
-        }))
-        .await?)
+        let iters = request.map(|(node, input)| async move {
+            let client = pool.get(node).await.map_err(|e| (node.id, e))?;
+            f(client, input).await.map_err(|e| (node.id, e))
+        });
+
+        let (results_ok, results_err) =
+            try_join_all_with_error_timeout(iters, Duration::from_secs(3)).await;
+        if results_err.is_empty() {
+            return Ok(results_ok);
+        }
+        let merged_error = merge_node_rpc_errors("merged RPC Error", results_err);
+        Err(merged_error)
     }
 
     async fn broadcast<RSP, Fut: Future<Output = Result<RSP, RpcError>> + 'static>(
@@ -391,3 +399,67 @@ impl StreamRpcManager {
         Ok(())
     }
 }
+
+/// This function is similar to `try_join_all`, but it attempts to collect many error as possible within `error_timeout`.
+async fn try_join_all_with_error_timeout<I, RSP, E, F>(
+    iters: I,
+    error_timeout: Duration,
+) -> (Vec<RSP>, Vec<E>)
+where
+    I: IntoIterator<Item = F>,
+    F: Future<Output = Result<RSP, E>>,
+{
+    let stream = FuturesUnordered::from_iter(iters);
+    pin_mut!(stream);
+    let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel();
+    let mut results_ok = vec![];
+    let mut results_err = vec![];
+    let mut is_err_timeout = false;
+    loop {
+        let rx = rx.recv();
+        pin_mut!(rx);
+        match select(rx, stream.next()).await {
+            Either::Left((_, _)) => {
+                // error_timeout
+                break;
+            }
+            Either::Right((None, _)) => {
+                break;
+            }
+            Either::Right((Some(Ok(rsp)), _)) => {
+                results_ok.push(rsp);
+            }
+            Either::Right((Some(Err(err)), _)) => {
+                results_err.push(err);
+                if is_err_timeout {
+                    continue;
+                }
+                // Start `error_timeout` when the first error occurs.
+                is_err_timeout = true;
+                let tx = tx.clone();
+                tokio::spawn(async move {
+                    tokio::time::sleep(error_timeout).await;
+                    let _ = tx.send(());
+                });
+            }
+        }
+    }
+    (results_ok, results_err)
+}
+
+fn merge_node_rpc_errors(
+    message: &str,
+    errors: impl IntoIterator<Item = (WorkerId, RpcError)>,
+) -> MetaError {
+    use std::fmt::Write;
+
+    use thiserror_ext::AsReport;
+
+    let concat: String = errors
+        .into_iter()
+        .fold(format!("{message}:"), |mut s, (w, e)| {
+            write!(&mut s, " worker node {}, {};", w, e.as_report()).unwrap();
+            s
+        });
+    anyhow::anyhow!(concat).into()
+}