Use net_util::run_hyper_server to run admin http server

crates/admin/Cargo.toml

@@ -39,6 +39,7 @@ futures = { workspace = true }
 http = { workspace = true }
 http-body = { workspace = true }
 http-body-util = { workspace = true }
+hyper-util = { workspace = true }
 okapi-operation = { version = "0.3.0-rc2", features = ["axum-integration"] }
 prost = { workspace = true }
 prost-dto = { workspace = true }

crates/admin/src/rest_api/deployments.rs

@@ -11,14 +11,14 @@
 use super::error::*;
 use crate::state::AdminServiceState;
 
-use crate::rest_api::log_error;
 use crate::schema_registry::{ApplyMode, Force};
 use axum::extract::{Path, Query, State};
 use axum::http::{header, StatusCode};
 use axum::response::IntoResponse;
 use axum::Json;
 use okapi_operation::*;
 use restate_admin_rest_model::deployments::*;
+use restate_errors::fmt::CodedErrorResultExt;
 use restate_service_client::Endpoint;
 use restate_service_protocol::discovery::DiscoverEndpoint;
 use restate_types::identifiers::{DeploymentId, InvalidLambdaARN};
@@ -96,16 +96,10 @@ pub async fn create_deployment<V>(
     };
 
     let (id, services) = state
-        .task_center
-        .run_in_scope("create-deployment", None, async {
-            log_error(
-                state
-                    .schema_registry
-                    .register_deployment(discover_endpoint, force, apply_mode)
-                    .await,
-            )
-        })
-        .await?;
+        .schema_registry
+        .register_deployment(discover_endpoint, force, apply_mode)
+        .await
+        .warn_it()?;
 
     let response_body = RegisterDeploymentResponse { id, services };
 
@@ -136,10 +130,8 @@ pub async fn get_deployment<V>(
     Path(deployment_id): Path<DeploymentId>,
 ) -> Result<Json<DetailedDeploymentResponse>, MetaApiError> {
     let (deployment, services) = state
-        .task_center
-        .run_in_scope_sync("get-deployment", None, || {
-            state.schema_registry.get_deployment(deployment_id)
-        })
+        .schema_registry
+        .get_deployment(deployment_id)
         .ok_or_else(|| MetaApiError::DeploymentNotFound(deployment_id))?;
 
     Ok(DetailedDeploymentResponse {
@@ -161,10 +153,8 @@ pub async fn list_deployments<V>(
     State(state): State<AdminServiceState<V>>,
 ) -> Json<ListDeploymentsResponse> {
     let deployments = state
-        .task_center
-        .run_in_scope_sync("list-deployments", None, || {
-            state.schema_registry.list_deployments()
-        })
+        .schema_registry
+        .list_deployments()
         .into_iter()
         .map(|(deployment, services)| DeploymentResponse {
             id: deployment.id,
@@ -226,7 +216,11 @@ pub async fn delete_deployment<V>(
     Query(DeleteDeploymentParams { force }): Query<DeleteDeploymentParams>,
 ) -> Result<StatusCode, MetaApiError> {
     if let Some(true) = force {
-        log_error(state.schema_registry.delete_deployment(deployment_id).await)?;
+        state
+            .schema_registry
+            .delete_deployment(deployment_id)
+            .await
+            .warn_it()?;
         Ok(StatusCode::ACCEPTED)
     } else {
         Ok(StatusCode::NOT_IMPLEMENTED)

crates/admin/src/rest_api/handlers.rs

@@ -33,11 +33,7 @@ pub async fn list_service_handlers<V>(
     State(state): State<AdminServiceState<V>>,
     Path(service_name): Path<String>,
 ) -> Result<Json<ListServiceHandlersResponse>, MetaApiError> {
-    match state
-        .task_center
-        .run_in_scope_sync("list-service-handlers", None, || {
-            state.schema_registry.list_service_handlers(&service_name)
-        }) {
+    match state.schema_registry.list_service_handlers(&service_name) {
         Some(handlers) => Ok(ListServiceHandlersResponse { handlers }.into()),
         None => Err(MetaApiError::ServiceNotFound(service_name)),
     }
@@ -67,12 +63,9 @@ pub async fn get_service_handler<V>(
     Path((service_name, handler_name)): Path<(String, String)>,
 ) -> Result<Json<HandlerMetadata>, MetaApiError> {
     match state
-        .task_center
-        .run_in_scope_sync("get-service-handler", None, || {
-            state
-                .schema_registry
-                .get_service_handler(&service_name, &handler_name)
-        }) {
+        .schema_registry
+        .get_service_handler(&service_name, &handler_name)
+    {
         Some(metadata) => Ok(metadata.into()),
         _ => Err(MetaApiError::HandlerNotFound {
             service_name,

crates/admin/src/rest_api/invocations.rs

@@ -93,17 +93,11 @@ pub async fn delete_invocation<V>(
 
     let partition_key = invocation_id.partition_key();
 
-    let result = state
-        .task_center
-        .run_in_scope(
-            "delete_invocation",
-            None,
-            append_envelope_to_bifrost(
-                &state.bifrost,
-                Envelope::new(create_envelope_header(partition_key), cmd),
-            ),
-        )
-        .await;
+    let result = append_envelope_to_bifrost(
+        &state.bifrost,
+        Envelope::new(create_envelope_header(partition_key), cmd),
+    )
+    .await;
 
     if let Err(err) = result {
         warn!("Could not append invocation termination command to Bifrost: {err}");

crates/admin/src/rest_api/mod.rs

@@ -19,10 +19,8 @@ mod services;
 mod subscriptions;
 mod version;
 
-use codederror::CodedError;
 use okapi_operation::axum_integration::{delete, get, patch, post};
 use okapi_operation::*;
-use restate_errors::warn_it;
 use restate_types::identifiers::PartitionKey;
 use restate_types::schema::subscriptions::SubscriptionValidator;
 use restate_wal_protocol::{Destination, Header, Source};
@@ -108,11 +106,3 @@ fn create_envelope_header(partition_key: PartitionKey) -> Header {
         },
     }
 }
-
-#[inline]
-fn log_error<T, E: CodedError>(result: Result<T, E>) -> Result<T, E> {
-    result.map_err(|err| {
-        warn_it!(err);
-        err
-    })
-}

crates/admin/src/rest_api/services.rs

@@ -8,8 +8,8 @@
 // the Business Source License, use of this software will be governed
 // by the Apache License, Version 2.0.
 
+use super::create_envelope_header;
 use super::error::*;
-use super::{create_envelope_header, log_error};
 use crate::schema_registry::ModifyServiceChange;
 use crate::state::AdminServiceState;
 
@@ -20,6 +20,7 @@ use http::StatusCode;
 use okapi_operation::*;
 use restate_admin_rest_model::services::ListServicesResponse;
 use restate_admin_rest_model::services::*;
+use restate_errors::fmt::CodedErrorResultExt;
 use restate_types::identifiers::{ServiceId, WithPartitionKey};
 use restate_types::schema::service::ServiceMetadata;
 use restate_types::state_mut::ExternalStateMutation;
@@ -36,11 +37,7 @@ use tracing::{debug, warn};
 pub async fn list_services<V>(
     State(state): State<AdminServiceState<V>>,
 ) -> Result<Json<ListServicesResponse>, MetaApiError> {
-    let services = state
-        .task_center
-        .run_in_scope_sync("list-services", None, || {
-            state.schema_registry.list_services()
-        });
+    let services = state.schema_registry.list_services();
 
     Ok(ListServicesResponse { services }.into())
 }
@@ -62,10 +59,8 @@ pub async fn get_service<V>(
     Path(service_name): Path<String>,
 ) -> Result<Json<ServiceMetadata>, MetaApiError> {
     state
-        .task_center
-        .run_in_scope_sync("get-service", None, || {
-            state.schema_registry.get_service(&service_name)
-        })
+        .schema_registry
+        .get_service(&service_name)
         .map(Into::into)
         .ok_or_else(|| MetaApiError::ServiceNotFound(service_name))
 }
@@ -112,16 +107,10 @@ pub async fn modify_service<V>(
     }
 
     let response = state
-        .task_center
-        .run_in_scope("modify-service", None, async {
-            log_error(
-                state
-                    .schema_registry
-                    .modify_service(service_name, modify_request)
-                    .await,
-            )
-        })
-        .await?;
+        .schema_registry
+        .modify_service(service_name, modify_request)
+        .await
+        .warn_it()?;
 
     Ok(response.into())
 }
@@ -156,12 +145,7 @@ pub async fn modify_service_state<V>(
         new_state,
     }): Json<ModifyServiceStateRequest>,
 ) -> Result<StatusCode, MetaApiError> {
-    if let Some(svc) = state
-        .task_center
-        .run_in_scope_sync("get-service", None, || {
-            state.schema_registry.get_service(&service_name)
-        })
-    {
+    if let Some(svc) = state.schema_registry.get_service(&service_name) {
         if !svc.ty.has_state() {
             return Err(MetaApiError::UnsupportedOperation("modify state", svc.ty));
         }
@@ -189,20 +173,14 @@ pub async fn modify_service_state<V>(
         state: new_state,
     };
 
-    let result = state
-        .task_center
-        .run_in_scope(
-            "modify_service_state",
-            None,
-            append_envelope_to_bifrost(
-                &state.bifrost,
-                Envelope::new(
-                    create_envelope_header(partition_key),
-                    Command::PatchState(patch_state),
-                ),
-            ),
-        )
-        .await;
+    let result = append_envelope_to_bifrost(
+        &state.bifrost,
+        Envelope::new(
+            create_envelope_header(partition_key),
+            Command::PatchState(patch_state),
+        ),
+    )
+    .await;
 
     if let Err(err) = result {
         warn!("Could not append state patching command to Bifrost: {err}");

crates/admin/src/rest_api/subscriptions.rs

@@ -14,12 +14,12 @@ use crate::state::AdminServiceState;
 use restate_admin_rest_model::subscriptions::*;
 use restate_types::schema::subscriptions::{ListSubscriptionFilter, SubscriptionValidator};
 
-use crate::rest_api::log_error;
 use axum::extract::Query;
 use axum::extract::{Path, State};
 use axum::http::StatusCode;
 use axum::{http, Json};
 use okapi_operation::*;
+use restate_errors::fmt::CodedErrorResultExt;
 use restate_types::identifiers::SubscriptionId;
 
 /// Create subscription.
@@ -42,12 +42,11 @@ pub async fn create_subscription<V: SubscriptionValidator>(
     State(state): State<AdminServiceState<V>>,
     #[request_body(required = true)] Json(payload): Json<CreateSubscriptionRequest>,
 ) -> Result<impl axum::response::IntoResponse, MetaApiError> {
-    let subscription = log_error(
-        state
-            .schema_registry
-            .create_subscription(payload.source, payload.sink, payload.options)
-            .await,
-    )?;
+    let subscription = state
+        .schema_registry
+        .create_subscription(payload.source, payload.sink, payload.options)
+        .await
+        .warn_it()?;
 
     Ok((
         StatusCode::CREATED,
@@ -76,10 +75,8 @@ pub async fn get_subscription<V>(
     Path(subscription_id): Path<SubscriptionId>,
 ) -> Result<Json<SubscriptionResponse>, MetaApiError> {
     let subscription = state
-        .task_center
-        .run_in_scope_sync("get-subscription", None, || {
-            state.schema_registry.get_subscription(subscription_id)
-        })
+        .schema_registry
+        .get_subscription(subscription_id)
         .ok_or_else(|| MetaApiError::SubscriptionNotFound(subscription_id))?;
 
     Ok(SubscriptionResponse::from(subscription).into())
@@ -126,11 +123,7 @@ pub async fn list_subscriptions<V>(
         _ => vec![],
     };
 
-    let subscriptions = state
-        .task_center
-        .run_in_scope_sync("list-subscriptions", None, || {
-            state.schema_registry.list_subscriptions(&filters)
-        });
+    let subscriptions = state.schema_registry.list_subscriptions(&filters);
 
     ListSubscriptionsResponse {
         subscriptions: subscriptions
@@ -166,11 +159,10 @@ pub async fn delete_subscription<V>(
     State(state): State<AdminServiceState<V>>,
     Path(subscription_id): Path<SubscriptionId>,
 ) -> Result<StatusCode, MetaApiError> {
-    log_error(
-        state
-            .schema_registry
-            .delete_subscription(subscription_id)
-            .await,
-    )?;
+    state
+        .schema_registry
+        .delete_subscription(subscription_id)
+        .await
+        .warn_it()?;
     Ok(StatusCode::ACCEPTED)
 }