Adds recover_fn to many routes.

If a route matches, we want to run its handler, and upon failure,
the last thing we want is for another overlapping route to be executed
as a fallback.

To avoid this, we call recover on each individual routes that have no
overlap with routes of lesser priority.

quickwit/quickwit-serve/src/cluster_api/rest_handler.rs

@@ -23,6 +23,7 @@ use quickwit_cluster::{Cluster, ClusterSnapshot, NodeIdSchema};
 use warp::{Filter, Rejection};
 
 use crate::format::extract_format_from_qs;
+use crate::rest::recover_fn;
 use crate::rest_api_response::into_rest_api_response;
 
 #[derive(utoipa::OpenApi)]
@@ -43,6 +44,7 @@ pub fn cluster_handler(
         .then(get_cluster)
         .and(extract_format_from_qs())
         .map(into_rest_api_response)
+        .recover(recover_fn)
 }
 
 #[utoipa::path(

quickwit/quickwit-serve/src/delete_task_api/handler.rs

@@ -31,6 +31,7 @@ use serde::Deserialize;
 use warp::{Filter, Rejection};
 
 use crate::format::extract_format_from_qs;
+use crate::rest::recover_fn;
 use crate::rest_api_response::into_rest_api_response;
 use crate::with_arg;
 
@@ -61,7 +62,9 @@ pub struct DeleteQueryRequest {
 pub fn delete_task_api_handlers(
     metastore: MetastoreServiceClient,
 ) -> impl Filter<Extract = (impl warp::Reply,), Error = Rejection> + Clone {
-    get_delete_tasks_handler(metastore.clone()).or(post_delete_tasks_handler(metastore.clone()))
+    get_delete_tasks_handler(metastore.clone())
+        .or(post_delete_tasks_handler(metastore.clone()))
+        .recover(recover_fn)
 }
 
 pub fn get_delete_tasks_handler(

quickwit/quickwit-serve/src/developer_api/mod.rs

@@ -27,6 +27,7 @@ use quickwit_cluster::Cluster;
 pub(crate) use server::DeveloperApiServer;
 use warp::{Filter, Rejection};
 
+use crate::rest::recover_fn;
 use crate::EnvFilterReloadFn;
 
 #[derive(utoipa::OpenApi)]
@@ -39,4 +40,5 @@ pub(crate) fn developer_api_routes(
 ) -> impl Filter<Extract = (impl warp::Reply,), Error = Rejection> + Clone {
     warp::path!("api" / "developer" / ..)
         .and(debug_handler(cluster.clone()).or(log_level_handler(env_filter_reload_fn.clone())))
+        .recover(recover_fn)
 }

quickwit/quickwit-serve/src/elasticsearch_api/bulk.rs

@@ -35,6 +35,7 @@ use crate::elasticsearch_api::make_elastic_api_response;
 use crate::elasticsearch_api::model::{BulkAction, ElasticBulkOptions, ElasticsearchError};
 use crate::format::extract_format_from_qs;
 use crate::ingest_api::lines;
+use crate::rest::recover_fn;
 use crate::{with_arg, Body};
 
 /// POST `_elastic/_bulk`
@@ -50,6 +51,7 @@ pub fn es_compat_bulk_handler(
         })
         .and(extract_format_from_qs())
         .map(make_elastic_api_response)
+        .recover(recover_fn)
 }
 
 /// POST `_elastic/<index>/_bulk`
@@ -73,6 +75,7 @@ pub fn es_compat_index_bulk_handler(
         )
         .and(extract_format_from_qs())
         .map(make_elastic_api_response)
+        .recover(recover_fn)
 }
 
 async fn elastic_ingest_bulk(

quickwit/quickwit-serve/src/elasticsearch_api/filter.rs

@@ -177,8 +177,8 @@ pub(crate) fn elastic_index_count_filter(
 pub(crate) fn elastic_delete_index_filter(
 ) -> impl Filter<Extract = (Vec<String>, DeleteQueryParams), Error = Rejection> + Clone {
     warp::path!("_elastic" / String)
-        .and_then(extract_index_id_patterns)
         .and(warp::delete())
+        .and_then(extract_index_id_patterns)
         .and(serde_qs::warp::query(serde_qs::Config::default()))
 }
 

quickwit/quickwit-serve/src/elasticsearch_api/mod.rs

@@ -45,6 +45,7 @@ use serde::{Deserialize, Serialize};
 use warp::{Filter, Rejection};
 
 use crate::elasticsearch_api::model::ElasticsearchError;
+use crate::rest::recover_fn;
 use crate::rest_api_response::RestApiResponse;
 use crate::{BodyFormat, BuildInfo};
 
@@ -79,6 +80,7 @@ pub fn elastic_api_handlers(
         .or(es_compat_stats_handler(metastore.clone()))
         .or(es_compat_index_cat_indices_handler(metastore.clone()))
         .or(es_compat_cat_indices_handler(metastore.clone()))
+        .recover(recover_fn)
     // Register newly created handlers here.
 }
 

quickwit/quickwit-serve/src/elasticsearch_api/rest_handler.rs

@@ -62,6 +62,7 @@ use super::model::{
 };
 use super::{make_elastic_api_response, TrackTotalHits};
 use crate::format::BodyFormat;
+use crate::rest::recover_fn;
 use crate::rest_api_response::{RestApiError, RestApiResponse};
 use crate::{with_arg, BuildInfo};
 
@@ -139,6 +140,7 @@ pub fn es_compat_stats_handler(
         .and(with_arg(search_service))
         .then(es_compat_stats)
         .map(|result| make_elastic_api_response(result, BodyFormat::default()))
+        .recover(recover_fn)
 }
 
 /// GET _elastic/{index}/_stats
@@ -149,6 +151,7 @@ pub fn es_compat_index_stats_handler(
         .and(with_arg(search_service))
         .then(es_compat_index_stats)
         .map(|result| make_elastic_api_response(result, BodyFormat::default()))
+        .recover(recover_fn)
 }
 
 /// GET _elastic/_cat/indices
@@ -159,6 +162,7 @@ pub fn es_compat_cat_indices_handler(
         .and(with_arg(search_service))
         .then(es_compat_cat_indices)
         .map(|result| make_elastic_api_response(result, BodyFormat::default()))
+        .recover(recover_fn)
 }
 
 /// GET _elastic/_cat/indices/{index}
@@ -169,6 +173,7 @@ pub fn es_compat_index_cat_indices_handler(
         .and(with_arg(search_service))
         .then(es_compat_index_cat_indices)
         .map(|result| make_elastic_api_response(result, BodyFormat::default()))
+        .recover(recover_fn)
 }
 
 /// GET or POST _elastic/{index}/_search

quickwit/quickwit-serve/src/health_check_api/handler.rs

@@ -26,6 +26,7 @@ use warp::hyper::StatusCode;
 use warp::reply::with_status;
 use warp::{Filter, Rejection};
 
+use crate::rest::recover_fn;
 use crate::with_arg;
 
 #[derive(utoipa::OpenApi)]
@@ -50,6 +51,7 @@ fn liveness_handler(
         .and(with_arg(indexer_service_opt))
         .and(with_arg(janitor_service_opt))
         .then(get_liveness)
+        .recover(recover_fn)
 }
 
 fn readiness_handler(

quickwit/quickwit-serve/src/index_api/rest_handler.rs

@@ -47,6 +47,7 @@ use tracing::{info, warn};
 use warp::{Filter, Rejection};
 
 use crate::format::{extract_config_format, extract_format_from_qs};
+use crate::rest::recover_fn;
 use crate::rest_api_response::into_rest_api_response;
 use crate::simple_list::{from_simple_list, to_simple_list};
 use crate::with_arg;
@@ -107,6 +108,7 @@ pub fn index_management_handlers(
         .or(analyze_request_handler())
         // Parse query into query AST handler.
         .or(parse_query_request_handler())
+        .recover(recover_fn)
 }
 
 fn json_body<T: DeserializeOwned + Send>(

quickwit/quickwit-serve/src/indexing_api/rest_handler.rs

@@ -25,6 +25,7 @@ use warp::{Filter, Rejection};
 
 use crate::format::extract_format_from_qs;
 use crate::require;
+use crate::rest::recover_fn;
 use crate::rest_api_response::into_rest_api_response;
 
 #[derive(utoipa::OpenApi)]
@@ -60,4 +61,5 @@ pub fn indexing_get_handler(
         .then(indexing_endpoint)
         .and(extract_format_from_qs())
         .map(into_rest_api_response)
+        .recover(recover_fn)
 }

quickwit/quickwit-serve/src/jaeger_api/rest_handler.rs

@@ -39,6 +39,7 @@ use crate::jaeger_api::model::{
     JaegerError, JaegerResponseBody, JaegerSpan, JaegerTrace, TracesSearchQueryParams,
     DEFAULT_NUMBER_OF_TRACES,
 };
+use crate::rest::recover_fn;
 use crate::rest_api_response::RestApiResponse;
 use crate::search_api::extract_index_id_patterns;
 use crate::{require, BodyFormat};
@@ -66,6 +67,7 @@ pub(crate) fn jaeger_api_handlers(
         ))
         .or(jaeger_traces_search_handler(jaeger_service_opt.clone()))
         .or(jaeger_traces_handler(jaeger_service_opt.clone()))
+        .recover(recover_fn)
 }
 
 fn jaeger_api_path_filter() -> impl Filter<Extract = (Vec<String>,), Error = Rejection> + Clone {

quickwit/quickwit-serve/src/node_info_handler.rs

@@ -23,6 +23,7 @@ use quickwit_config::NodeConfig;
 use serde_json::json;
 use warp::{Filter, Rejection};
 
+use crate::rest::recover_fn;
 use crate::{with_arg, BuildInfo, RuntimeInfo};
 
 #[derive(utoipa::OpenApi)]
@@ -34,7 +35,9 @@ pub fn node_info_handler(
     runtime_info: &'static RuntimeInfo,
     config: Arc<NodeConfig>,
 ) -> impl Filter<Extract = (impl warp::Reply,), Error = Rejection> + Clone {
-    node_version_handler(build_info, runtime_info).or(node_config_handler(config))
+    node_version_handler(build_info, runtime_info)
+        .or(node_config_handler(config))
+        .recover(recover_fn)
 }
 
 #[utoipa::path(get, tag = "Node Info", path = "/version")]

quickwit/quickwit-serve/src/otlp_api/rest_handler.rs

@@ -35,6 +35,7 @@ use serde::{self, Serialize};
 use tracing::error;
 use warp::{Filter, Rejection};
 
+use crate::rest::recover_fn;
 use crate::rest_api_response::into_rest_api_response;
 use crate::{require, with_arg, BodyFormat};
 
@@ -47,6 +48,7 @@ pub(crate) fn otlp_ingest_api_handlers(
         .or(otlp_default_traces_handler(otlp_traces_service.clone()))
         .or(otlp_logs_handler(otlp_logs_service))
         .or(otlp_ingest_traces_handler(otlp_traces_service))
+        .recover(recover_fn)
 }
 
 pub(crate) fn otlp_default_logs_handler(