Wire task_spawner into all routes

beacon_node/http_api/src/task_spawner.rs

@@ -71,14 +71,66 @@ impl<E: EthSpec> TaskSpawner<E> {
         }
     }
 
-    pub async fn async_task<F, T>(
-        &self,
+    pub async fn blocking_response_task<F, T>(
+        self,
         priority: Priority,
-        func: impl Future<Output = Result<T, warp::Rejection>> + Send + Sync + 'static,
+        func: F,
     ) -> Result<Response, warp::Rejection>
     where
-        T: Serialize + Send + 'static,
+        F: FnOnce() -> Result<T, warp::Rejection> + Send + Sync + 'static,
+        T: Reply + Send + 'static,
     {
+        if let Some(beacon_processor_send) = &self.beacon_processor_send {
+            // Create a closure that will execute `func` and send the result to
+            // a channel held by this thread.
+            let (tx, rx) = oneshot::channel();
+            let process_fn = move || {
+                // Execute the function, collect the return value.
+                let func_result = func();
+                // Send the result down the channel. Ignore any failures; the
+                // send can only fail if the receiver is dropped.
+                let _ = tx.send(func_result);
+            };
+            let process_fn = BlockingOrAsync::Blocking(Box::new(process_fn));
+
+            let error_message =
+                match beacon_processor_send.try_send(priority.work_event(process_fn)) {
+                    Ok(()) => {
+                        match rx.await {
+                // The beacon processor executed the task and sent a result.
+                Ok(func_result) => {
+                    return func_result.map(Reply::into_response)
+                }
+                // The beacon processor dropped the channel without sending a
+                // result. The beacon processor dropped this task because its
+                // queues are full or it's shutting down.
+                Err(_) => "The task did not execute. The server is overloaded or shutting down.",
+            }
+                    }
+                    Err(TrySendError::Full(_)) => "The task was dropped. The server is overloaded.",
+                    Err(TrySendError::Closed(_)) => {
+                        "The task was dropped. The server is shutting down."
+                    }
+                };
+
+            let error_response = warp::reply::with_status(
+                warp::reply::json(&error_message),
+                eth2::StatusCode::INTERNAL_SERVER_ERROR,
+            )
+            .into_response();
+            Ok(error_response)
+        } else {
+            // There is no beacon processor so spawn a task directly on the
+            // tokio executor.
+            warp_utils::task::blocking_response_task(func).await
+        }
+    }
+
+    pub async fn spawn_async_with_rejection(
+        self,
+        priority: Priority,
+        func: impl Future<Output = Result<Response, warp::Rejection>> + Send + Sync + 'static,
+    ) -> Result<Response, warp::Rejection> {
         if let Some(beacon_processor_send) = &self.beacon_processor_send {
             // Create a wrapper future that will execute `func` and send the
             // result to a channel held by this thread.
@@ -90,29 +142,103 @@ impl<E: EthSpec> TaskSpawner<E> {
                 // send can only fail if the receiver is dropped.
                 let _ = tx.send(func_result);
             };
+            let process_fn = BlockingOrAsync::Async(Box::pin(process_fn));
 
             // Send the function to the beacon processor for execution at some arbitrary time.
-            send_to_beacon_processor(
-                beacon_processor_send,
-                priority,
-                BlockingOrAsync::Async(Box::pin(process_fn)),
-                rx,
+            let error_message =
+                match beacon_processor_send.try_send(priority.work_event(process_fn)) {
+                    Ok(()) => {
+                        match rx.await {
+                // The beacon processor executed the task and sent a result.
+                Ok(func_result) => {
+                    return func_result
+                }
+                // The beacon processor dropped the channel without sending a
+                // result. The beacon processor dropped this task because its
+                // queues are full or it's shutting down.
+                Err(_) => "The task did not execute. The server is overloaded or shutting down.",
+            }
+                    }
+                    Err(TrySendError::Full(_)) => "The task was dropped. The server is overloaded.",
+                    Err(TrySendError::Closed(_)) => {
+                        "The task was dropped. The server is shutting down."
+                    }
+                };
+
+            let error_response = warp::reply::with_status(
+                warp::reply::json(&error_message),
+                eth2::StatusCode::INTERNAL_SERVER_ERROR,
             )
-            .await
+            .into_response();
+            Ok(error_response)
         } else {
             // There is no beacon processor so spawn a task directly on the
             // tokio executor.
-            tokio::task::spawn(func)
-                .await
-                .map(|r| r.map(|r| warp::reply::json(&r).into_response()))
-                .unwrap_or_else(|e| {
-                    let response = warp::reply::with_status(
-                        warp::reply::json(&format!("Tokio did not execute task: {e:?}")),
-                        eth2::StatusCode::INTERNAL_SERVER_ERROR,
-                    )
-                    .into_response();
-                    Ok(response)
-                })
+            tokio::task::spawn(func).await.unwrap_or_else(|e| {
+                let response = warp::reply::with_status(
+                    warp::reply::json(&format!("Tokio did not execute task: {e:?}")),
+                    eth2::StatusCode::INTERNAL_SERVER_ERROR,
+                )
+                .into_response();
+                Ok(response)
+            })
+        }
+    }
+
+    pub async fn spawn_async(
+        self,
+        priority: Priority,
+        func: impl Future<Output = Response> + Send + Sync + 'static,
+    ) -> Response {
+        if let Some(beacon_processor_send) = &self.beacon_processor_send {
+            // Create a wrapper future that will execute `func` and send the
+            // result to a channel held by this thread.
+            let (tx, rx) = oneshot::channel();
+            let process_fn = async move {
+                // Await the future, collect the return value.
+                let func_result = func.await;
+                // Send the result down the channel. Ignore any failures; the
+                // send can only fail if the receiver is dropped.
+                let _ = tx.send(func_result);
+            };
+            let process_fn = BlockingOrAsync::Async(Box::pin(process_fn));
+
+            // Send the function to the beacon processor for execution at some arbitrary time.
+            let error_message =
+                match beacon_processor_send.try_send(priority.work_event(process_fn)) {
+                    Ok(()) => {
+                        match rx.await {
+                // The beacon processor executed the task and sent a result.
+                Ok(func_result) => {
+                    return func_result
+                }
+                // The beacon processor dropped the channel without sending a
+                // result. The beacon processor dropped this task because its
+                // queues are full or it's shutting down.
+                Err(_) => "The task did not execute. The server is overloaded or shutting down.",
+            }
+                    }
+                    Err(TrySendError::Full(_)) => "The task was dropped. The server is overloaded.",
+                    Err(TrySendError::Closed(_)) => {
+                        "The task was dropped. The server is shutting down."
+                    }
+                };
+
+            warp::reply::with_status(
+                warp::reply::json(&error_message),
+                eth2::StatusCode::INTERNAL_SERVER_ERROR,
+            )
+            .into_response()
+        } else {
+            // There is no beacon processor so spawn a task directly on the
+            // tokio executor.
+            tokio::task::spawn(func).await.unwrap_or_else(|e| {
+                warp::reply::with_status(
+                    warp::reply::json(&format!("Tokio did not execute task: {e:?}")),
+                    eth2::StatusCode::INTERNAL_SERVER_ERROR,
+                )
+                .into_response()
+            })
         }
     }
 }