rest/
server.rs

1use crate::{
2    route::RoutePolicies, AdaptiveLoadShed, ConcurrencyLimit, ContentEncryption, HttpMetrics,
3    LoggingMiddleware, MetricsMiddleware, MultipartConfig, Recover, RequestBodyLimit, RequestId,
4    ResponsePolicy, RouteGroupConfig, RouteMiddleware, SecurityHeaders, StaticAssets, Timeout,
5    TraceContextMiddleware,
6};
7use actix_web::{
8    body::{self, BoxBody},
9    dev::{Service, ServiceResponse},
10    http::{header::HeaderMap, Method, Uri},
11    middleware::Condition,
12    web::{self, ServiceConfig},
13    App, Error, HttpServer,
14};
15use rust_zero_core::{AdaptiveShedder, LoadShedderConfig, Metrics};
16use serde::{Deserialize, Serialize};
17use std::{
18    collections::HashMap,
19    fmt,
20    future::Future,
21    io,
22    net::{SocketAddr, TcpListener},
23    path::PathBuf,
24    rc::Rc,
25    sync::Arc,
26    time::Duration,
27};
28
29macro_rules! standard_app {
30    ($config:expr, $http_metrics:expr, $adaptive_shedder:expr, $route_policies:expr, $response_policy:expr, $content_encryption:expr, $static_assets:expr, $configure:expr) => {{
31        let config = $config;
32        let http_metrics = $http_metrics;
33        let mut multipart_config = MultipartConfig::new(
34            config.max_multipart_field_bytes,
35            config.max_multipart_file_bytes,
36            config.max_multipart_total_bytes,
37        );
38        if let Some(temp_dir) = &config.multipart_temp_dir {
39            multipart_config = multipart_config.with_temp_dir(temp_dir);
40        }
41        let mut timeout = Timeout::new(Duration::from_millis(config.request_timeout_ms));
42        let mut concurrency = ConcurrencyLimit::new(config.max_concurrent_requests)
43            .with_priority_reserve(config.priority_concurrency_reserve);
44        if config.metrics {
45            timeout = timeout.with_metrics(http_metrics.clone());
46            concurrency = concurrency.with_metrics(http_metrics.clone());
47        }
48        let mut adaptive_load = AdaptiveLoadShed::new($adaptive_shedder);
49        if config.metrics {
50            adaptive_load = adaptive_load.with_metrics(http_metrics.clone());
51        }
52        let app = App::new()
53            .app_data(web::JsonConfig::default().limit(config.max_body_bytes))
54            .app_data(web::FormConfig::default().limit(config.max_body_bytes))
55            .app_data(web::Data::new(multipart_config))
56            .app_data(web::Data::new($response_policy))
57            .wrap(Condition::new(config.logging, LoggingMiddleware))
58            .wrap(Condition::new(config.recovery, Recover::new()))
59            .wrap(Condition::new(
60                config.security_headers,
61                SecurityHeaders::new(),
62            ))
63            .wrap(Condition::new(config.request_ids, RequestId::new()))
64            .wrap(Condition::new(
65                config.tracing,
66                TraceContextMiddleware::new(),
67            ))
68            .wrap(Condition::new(
69                config.metrics,
70                MetricsMiddleware::new(http_metrics),
71            ))
72            .wrap(timeout)
73            .wrap(concurrency)
74            .wrap(Condition::new(config.adaptive_load_shedding, adaptive_load))
75            .wrap(
76                RequestBodyLimit::new(config.max_body_bytes)
77                    .decompress_gzip(config.decompress_gzip),
78            )
79            .wrap(
80                $content_encryption
81                    .unwrap_or_else(|| ContentEncryption::disabled(config.max_body_bytes)),
82            )
83            .wrap($route_policies)
84            .configure($configure);
85        if let Some(static_assets) = $static_assets {
86            app.app_data(web::Data::new(static_assets))
87                .default_service(web::to(
88                    |request: actix_web::HttpRequest, assets: web::Data<StaticAssets>| async move {
89                        assets.serve(request).await
90                    },
91                ))
92        } else {
93            app.default_service(web::to(actix_web::HttpResponse::NotFound))
94        }
95    }};
96}
97
98/// Socket-free HTTP request accepted by [`ServerlessHandler`].
99#[derive(Debug, Clone)]
100pub struct ServerlessRequest {
101    pub method: Method,
102    pub uri: Uri,
103    pub headers: HeaderMap,
104    pub body: web::Bytes,
105}
106
107impl ServerlessRequest {
108    pub fn new(method: Method, uri: Uri, body: impl Into<web::Bytes>) -> Self {
109        Self {
110            method,
111            uri,
112            headers: HeaderMap::new(),
113            body: body.into(),
114        }
115    }
116}
117
118/// Fully buffered response returned to a serverless platform adapter.
119#[derive(Debug)]
120pub struct ServerlessResponse {
121    pub status: actix_web::http::StatusCode,
122    pub headers: HeaderMap,
123    pub body: web::Bytes,
124}
125
126/// Prebuilt, socket-free instance of the standard REST middleware and routing stack.
127#[derive(Clone)]
128pub struct ServerlessHandler {
129    service: actix_service::boxed::RcService<actix_http::Request, ServiceResponse<BoxBody>, Error>,
130}
131
132impl ServerlessHandler {
133    /// Dispatches one platform-neutral request through the prebuilt REST stack.
134    pub async fn call(&self, request: ServerlessRequest) -> Result<ServerlessResponse, Error> {
135        let uri = request.uri.to_string();
136        let mut builder = actix_web::test::TestRequest::default()
137            .method(request.method)
138            .uri(&uri)
139            .set_payload(request.body);
140        for (name, value) in request.headers.iter() {
141            builder = builder.append_header((name.clone(), value.clone()));
142        }
143
144        let response = self.service.call(builder.to_request()).await?;
145        let status = response.status();
146        let headers = response.headers().clone();
147        let body = body::to_bytes(response.into_body())
148            .await
149            .map_err(actix_web::error::ErrorInternalServerError)?;
150        Ok(ServerlessResponse {
151            status,
152            headers,
153            body,
154        })
155    }
156}
157
158/// Configuration for the standard rust-zero REST server stack.
159///
160/// Durations are expressed in milliseconds so the same representation works naturally in
161/// JSON, TOML, and YAML configuration files.
162#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
163#[serde(default)]
164pub struct RestServerConfig {
165    pub address: SocketAddr,
166    pub workers: usize,
167    pub shutdown_timeout_ms: u64,
168    pub request_timeout_ms: u64,
169    pub max_body_bytes: usize,
170    pub max_multipart_field_bytes: usize,
171    pub max_multipart_file_bytes: usize,
172    pub max_multipart_total_bytes: usize,
173    pub multipart_temp_dir: Option<PathBuf>,
174    pub max_concurrent_requests: usize,
175    pub priority_concurrency_reserve: usize,
176    pub adaptive_load_shedding: bool,
177    pub load_shed_cpu_threshold_percent: u8,
178    pub load_shed_bucket_ms: u64,
179    pub load_shed_buckets: usize,
180    pub load_shed_cooldown_ms: u64,
181    pub logging: bool,
182    pub recovery: bool,
183    pub tracing: bool,
184    pub metrics: bool,
185    pub security_headers: bool,
186    pub request_ids: bool,
187    pub decompress_gzip: bool,
188    pub metrics_namespace: String,
189    pub route_groups: Vec<RouteGroupConfig>,
190}
191
192impl Default for RestServerConfig {
193    fn default() -> Self {
194        Self {
195            address: "0.0.0.0:8080"
196                .parse()
197                .expect("default REST address is valid"),
198            workers: 1,
199            shutdown_timeout_ms: 30_000,
200            request_timeout_ms: 10_000,
201            max_body_bytes: 4 * 1024 * 1024,
202            max_multipart_field_bytes: 64 * 1024,
203            max_multipart_file_bytes: 4 * 1024 * 1024,
204            max_multipart_total_bytes: 4 * 1024 * 1024,
205            multipart_temp_dir: None,
206            max_concurrent_requests: 1_024,
207            priority_concurrency_reserve: 256,
208            adaptive_load_shedding: true,
209            load_shed_cpu_threshold_percent: 90,
210            load_shed_bucket_ms: 1_000,
211            load_shed_buckets: 10,
212            load_shed_cooldown_ms: 1_000,
213            logging: true,
214            recovery: true,
215            tracing: true,
216            metrics: true,
217            security_headers: true,
218            request_ids: true,
219            decompress_gzip: true,
220            metrics_namespace: "rust_zero".to_owned(),
221            route_groups: Vec::new(),
222        }
223    }
224}
225
226impl RestServerConfig {
227    pub fn validate(&self) -> Result<(), RestServerConfigError> {
228        if self.workers == 0 {
229            return Err(RestServerConfigError::Invalid(
230                "workers must be greater than zero",
231            ));
232        }
233        if self.shutdown_timeout_ms == 0 {
234            return Err(RestServerConfigError::Invalid(
235                "shutdown_timeout_ms must be greater than zero",
236            ));
237        }
238        if self.request_timeout_ms == 0 {
239            return Err(RestServerConfigError::Invalid(
240                "request_timeout_ms must be greater than zero",
241            ));
242        }
243        if self.max_body_bytes == 0 {
244            return Err(RestServerConfigError::Invalid(
245                "max_body_bytes must be greater than zero",
246            ));
247        }
248        if self.max_multipart_field_bytes == 0 {
249            return Err(RestServerConfigError::Invalid(
250                "max_multipart_field_bytes must be greater than zero",
251            ));
252        }
253        if self.max_multipart_file_bytes == 0 {
254            return Err(RestServerConfigError::Invalid(
255                "max_multipart_file_bytes must be greater than zero",
256            ));
257        }
258        if self.max_multipart_total_bytes == 0 {
259            return Err(RestServerConfigError::Invalid(
260                "max_multipart_total_bytes must be greater than zero",
261            ));
262        }
263        if self.max_multipart_field_bytes > self.max_multipart_total_bytes {
264            return Err(RestServerConfigError::Invalid(
265                "max_multipart_field_bytes must not exceed max_multipart_total_bytes",
266            ));
267        }
268        if self.max_multipart_file_bytes > self.max_multipart_total_bytes {
269            return Err(RestServerConfigError::Invalid(
270                "max_multipart_file_bytes must not exceed max_multipart_total_bytes",
271            ));
272        }
273        if self.max_multipart_total_bytes > self.max_body_bytes {
274            return Err(RestServerConfigError::Invalid(
275                "max_multipart_total_bytes must not exceed max_body_bytes",
276            ));
277        }
278        if self.max_concurrent_requests == 0 {
279            return Err(RestServerConfigError::Invalid(
280                "max_concurrent_requests must be greater than zero",
281            ));
282        }
283        if self.priority_concurrency_reserve == 0 {
284            return Err(RestServerConfigError::Invalid(
285                "priority_concurrency_reserve must be greater than zero",
286            ));
287        }
288        if !(1..=100).contains(&self.load_shed_cpu_threshold_percent) {
289            return Err(RestServerConfigError::Invalid(
290                "load_shed_cpu_threshold_percent must be between 1 and 100",
291            ));
292        }
293        if self.load_shed_bucket_ms == 0 {
294            return Err(RestServerConfigError::Invalid(
295                "load_shed_bucket_ms must be greater than zero",
296            ));
297        }
298        if self.load_shed_buckets == 0 {
299            return Err(RestServerConfigError::Invalid(
300                "load_shed_buckets must be greater than zero",
301            ));
302        }
303        if self.load_shed_cooldown_ms == 0 {
304            return Err(RestServerConfigError::Invalid(
305                "load_shed_cooldown_ms must be greater than zero",
306            ));
307        }
308        if self.metrics && self.metrics_namespace.trim().is_empty() {
309            return Err(RestServerConfigError::Invalid(
310                "metrics_namespace must not be empty when metrics are enabled",
311            ));
312        }
313        RoutePolicies::compile(&self.route_groups).map_err(RestServerConfigError::RoutePolicy)?;
314        Ok(())
315    }
316}
317
318#[derive(Debug)]
319pub enum RestServerConfigError {
320    Invalid(&'static str),
321    Metrics(rust_zero_core::MetricsError),
322    RoutePolicy(String),
323    RouteMiddleware(String),
324}
325
326impl fmt::Display for RestServerConfigError {
327    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
328        match self {
329            Self::Invalid(message) => formatter.write_str(message),
330            Self::Metrics(error) => write!(formatter, "failed to configure REST metrics: {error}"),
331            Self::RoutePolicy(error) => write!(formatter, "invalid REST route policy: {error}"),
332            Self::RouteMiddleware(error) => {
333                write!(formatter, "invalid REST route middleware: {error}")
334            }
335        }
336    }
337}
338
339impl std::error::Error for RestServerConfigError {}
340
341/// Assembles and runs an Actix server with rust-zero's standard production middleware.
342#[derive(Clone)]
343pub struct RestServer {
344    config: RestServerConfig,
345    metrics: Arc<Metrics>,
346    http_metrics: HttpMetrics,
347    route_policies: RoutePolicies,
348    response_policy: ResponsePolicy,
349    content_encryption: Option<ContentEncryption>,
350    route_middleware: HashMap<String, Arc<dyn RouteMiddleware>>,
351    static_assets: Option<StaticAssets>,
352    adaptive_shedder: AdaptiveShedder,
353}
354
355impl RestServer {
356    pub fn new(config: RestServerConfig) -> Result<Self, RestServerConfigError> {
357        Self::with_metrics(config, Arc::new(Metrics::new()))
358    }
359
360    pub fn with_metrics(
361        config: RestServerConfig,
362        metrics: Arc<Metrics>,
363    ) -> Result<Self, RestServerConfigError> {
364        config.validate()?;
365        let http_metrics = HttpMetrics::new(&metrics, config.metrics_namespace.clone())
366            .map_err(RestServerConfigError::Metrics)?;
367        let route_policies = RoutePolicies::compile(&config.route_groups)
368            .map_err(RestServerConfigError::RoutePolicy)?;
369        let adaptive_shedder = AdaptiveShedder::new(
370            LoadShedderConfig::production(config.max_concurrent_requests)
371                .with_cpu_threshold(f64::from(config.load_shed_cpu_threshold_percent) / 100.0)
372                .with_rolling_window(
373                    Duration::from_millis(config.load_shed_bucket_ms),
374                    config.load_shed_buckets,
375                )
376                .with_cooldown(Duration::from_millis(config.load_shed_cooldown_ms)),
377        );
378        Ok(Self {
379            config,
380            metrics,
381            http_metrics,
382            route_policies,
383            response_policy: ResponsePolicy::new(),
384            content_encryption: None,
385            route_middleware: HashMap::new(),
386            static_assets: None,
387            adaptive_shedder,
388        })
389    }
390
391    /// Installs an opt-in static-directory and/or embedded-asset fallback.
392    pub fn with_static_assets(mut self, static_assets: StaticAssets) -> Self {
393        self.static_assets = Some(static_assets);
394        self
395    }
396
397    /// Registers application middleware referenced by declarative route groups.
398    pub fn with_route_middleware<M>(
399        mut self,
400        name: impl Into<String>,
401        middleware: M,
402    ) -> Result<Self, RestServerConfigError>
403    where
404        M: RouteMiddleware,
405    {
406        let name = name.into();
407        if name.trim().is_empty() {
408            return Err(RestServerConfigError::RouteMiddleware(
409                "middleware name must not be empty".to_owned(),
410            ));
411        }
412        if self
413            .route_middleware
414            .insert(name.clone(), Arc::new(middleware))
415            .is_some()
416        {
417            return Err(RestServerConfigError::RouteMiddleware(format!(
418                "middleware '{name}' is already registered"
419            )));
420        }
421        Ok(self)
422    }
423
424    /// Installs the response policy as Actix application data for all registered handlers.
425    pub fn with_response_policy(mut self, response_policy: ResponsePolicy) -> Self {
426        self.response_policy = response_policy;
427        self
428    }
429
430    /// Enables authenticated request decryption and buffered response encryption.
431    ///
432    /// Install this only for APIs whose clients implement the versioned content-encryption wire
433    /// format. Streaming routes such as SSE must remain on a server without this middleware.
434    pub fn with_content_encryption(mut self, content_encryption: ContentEncryption) -> Self {
435        self.content_encryption = Some(content_encryption);
436        self
437    }
438
439    pub fn config(&self) -> &RestServerConfig {
440        &self.config
441    }
442
443    pub fn metrics(&self) -> Arc<Metrics> {
444        Arc::clone(&self.metrics)
445    }
446
447    pub fn response_policy(&self) -> &ResponsePolicy {
448        &self.response_policy
449    }
450
451    /// Builds a reusable socket-free handler with the same routes, policies, middleware, and
452    /// static fallback as [`RestServer::run`].
453    pub async fn serverless_handler<F>(
454        &self,
455        configure: F,
456    ) -> Result<ServerlessHandler, RestServerConfigError>
457    where
458        F: FnOnce(&mut ServiceConfig) + 'static,
459    {
460        let route_policies = self
461            .route_policies
462            .clone()
463            .with_middleware(self.route_middleware.clone())
464            .map_err(RestServerConfigError::RouteMiddleware)?;
465        let app = standard_app!(
466            &self.config,
467            self.http_metrics.clone(),
468            self.adaptive_shedder.clone(),
469            route_policies,
470            self.response_policy.clone(),
471            self.content_encryption.clone(),
472            self.static_assets.clone(),
473            configure
474        );
475        let service = actix_web::test::init_service(app).await;
476        let service = Rc::new(service);
477        let service = actix_web::dev::fn_service(move |request| {
478            let future = service.call(request);
479            async move { future.await.map(ServiceResponse::map_into_boxed_body) }
480        });
481        Ok(ServerlessHandler {
482            service: actix_service::boxed::rc_service(service),
483        })
484    }
485
486    /// Binds the configured listener and installs the standard stack around application routes.
487    ///
488    /// The configure callback is cloned per Actix worker and can register ordinary routes,
489    /// resources, and scoped route groups.
490    pub fn run<F>(&self, configure: F) -> io::Result<actix_web::dev::Server>
491    where
492        F: Fn(&mut ServiceConfig) + Clone + Send + 'static,
493    {
494        let listener = TcpListener::bind(self.config.address)?;
495        self.run_on(listener, configure)
496    }
497
498    /// Runs the configured stack on an existing listener.
499    ///
500    /// Supplying the listener is useful for socket activation and lets tests reserve an ephemeral
501    /// port without a bind race.
502    pub fn run_on<F>(
503        &self,
504        listener: TcpListener,
505        configure: F,
506    ) -> io::Result<actix_web::dev::Server>
507    where
508        F: Fn(&mut ServiceConfig) + Clone + Send + 'static,
509    {
510        let config = self.config.clone();
511        let http_metrics = self.http_metrics.clone();
512        let route_policies = self
513            .route_policies
514            .clone()
515            .with_middleware(self.route_middleware.clone())
516            .map_err(|error| io::Error::new(io::ErrorKind::InvalidInput, error))?;
517        let response_policy = self.response_policy.clone();
518        let content_encryption = self.content_encryption.clone();
519        let static_assets = self.static_assets.clone();
520        let adaptive_shedder = self.adaptive_shedder.clone();
521        let shutdown_seconds = config.shutdown_timeout_ms.div_ceil(1_000);
522        let workers = config.workers;
523
524        HttpServer::new(move || {
525            standard_app!(
526                &config,
527                http_metrics.clone(),
528                adaptive_shedder.clone(),
529                route_policies.clone(),
530                response_policy.clone(),
531                content_encryption.clone(),
532                static_assets.clone(),
533                configure.clone()
534            )
535        })
536        .workers(workers)
537        .shutdown_timeout(shutdown_seconds)
538        .listen(listener)
539        .map(HttpServer::run)
540    }
541
542    /// Serves until the supplied shutdown signal resolves, then gracefully drains requests.
543    pub async fn serve_until<C, F>(&self, configure: C, shutdown: F) -> io::Result<()>
544    where
545        C: Fn(&mut ServiceConfig) + Clone + Send + 'static,
546        F: Future<Output = ()>,
547    {
548        let server = self.run(configure)?;
549        drain_on_signal(server, shutdown).await
550    }
551
552    /// Listener-based variant of [`RestServer::serve_until`].
553    pub async fn serve_on_until<C, F>(
554        &self,
555        listener: TcpListener,
556        configure: C,
557        shutdown: F,
558    ) -> io::Result<()>
559    where
560        C: Fn(&mut ServiceConfig) + Clone + Send + 'static,
561        F: Future<Output = ()>,
562    {
563        let server = self.run_on(listener, configure)?;
564        drain_on_signal(server, shutdown).await
565    }
566}
567
568async fn drain_on_signal<F>(server: actix_web::dev::Server, shutdown: F) -> io::Result<()>
569where
570    F: Future<Output = ()>,
571{
572    use futures::future::{select, Either};
573
574    let handle = server.handle();
575    match select(Box::pin(server), Box::pin(shutdown)).await {
576        Either::Left((result, _)) => result,
577        Either::Right(((), server)) => {
578            let (_, result) = futures::future::join(handle.stop(true), server).await;
579            result
580        }
581    }
582}
583
584#[cfg(test)]
585mod tests {
586    use super::*;
587    use actix_web::{
588        dev::ServiceRequest,
589        http::{header::HeaderValue, StatusCode},
590        test as actix_test, web, HttpResponse,
591    };
592    use rust_zero_core::{parse_config, ConfigFormat};
593    use std::path::Path;
594    use tokio::sync::Notify;
595
596    #[test]
597    fn parses_and_validates_transport_configuration() {
598        let config: RestServerConfig = parse_config(
599            r#"
600address = "127.0.0.1:9000"
601workers = 2
602request_timeout_ms = 250
603adaptive_load_shedding = true
604load_shed_cpu_threshold_percent = 85
605load_shed_bucket_ms = 250
606load_shed_buckets = 8
607load_shed_cooldown_ms = 750
608max_body_bytes = 16777216
609max_multipart_field_bytes = 32768
610max_multipart_file_bytes = 8388608
611max_multipart_total_bytes = 12582912
612multipart_temp_dir = "/tmp/rust-zero-uploads"
613
614[[route_groups]]
615prefix = "/api"
616timeout_ms = 100
617middleware = ["audit"]
618
619[[route_groups.routes]]
620method = "GET"
621path = "/users/{id}"
622public = true
623priority = true
624"#,
625            ConfigFormat::Toml,
626        )
627        .unwrap();
628
629        assert_eq!(config.address, "127.0.0.1:9000".parse().unwrap());
630        assert_eq!(config.workers, 2);
631        assert_eq!(config.request_timeout_ms, 250);
632        assert!(config.adaptive_load_shedding);
633        assert_eq!(config.load_shed_cpu_threshold_percent, 85);
634        assert_eq!(config.load_shed_bucket_ms, 250);
635        assert_eq!(config.load_shed_buckets, 8);
636        assert_eq!(config.load_shed_cooldown_ms, 750);
637        assert_eq!(config.max_multipart_field_bytes, 32 * 1024);
638        assert_eq!(config.max_multipart_file_bytes, 8 * 1024 * 1024);
639        assert_eq!(config.max_multipart_total_bytes, 12 * 1024 * 1024);
640        assert_eq!(
641            config.multipart_temp_dir.as_deref(),
642            Some(Path::new("/tmp/rust-zero-uploads"))
643        );
644        assert_eq!(config.route_groups[0].prefix, "/api");
645        assert_eq!(config.route_groups[0].middleware, ["audit"]);
646        assert_eq!(config.route_groups[0].routes[0].path, "/users/{id}");
647        config.validate().unwrap();
648    }
649
650    #[test]
651    fn rejects_zero_limits_before_binding() {
652        let error = RestServer::new(RestServerConfig {
653            max_body_bytes: 0,
654            ..RestServerConfig::default()
655        })
656        .err()
657        .unwrap();
658        assert!(error.to_string().contains("max_body_bytes"));
659
660        let error = RestServer::new(RestServerConfig {
661            max_multipart_field_bytes: 3,
662            max_multipart_file_bytes: 5,
663            max_multipart_total_bytes: 4,
664            ..RestServerConfig::default()
665        })
666        .err()
667        .unwrap();
668        assert!(error.to_string().contains("max_multipart_file_bytes"));
669
670        let error = RestServer::new(RestServerConfig {
671            load_shed_cpu_threshold_percent: 0,
672            ..RestServerConfig::default()
673        })
674        .err()
675        .unwrap();
676        assert!(error
677            .to_string()
678            .contains("load_shed_cpu_threshold_percent"));
679    }
680
681    #[actix_rt::test]
682    async fn configured_stack_wraps_registered_routes() {
683        let config = RestServerConfig::default();
684        let metrics = Metrics::new();
685        let http_metrics = HttpMetrics::new(&metrics, config.metrics_namespace.clone()).unwrap();
686        let app = actix_test::init_service(
687            App::new()
688                .wrap(LoggingMiddleware)
689                .wrap(Recover::new())
690                .wrap(RequestId::new())
691                .wrap(TraceContextMiddleware::new())
692                .wrap(MetricsMiddleware::new(http_metrics))
693                .wrap(Timeout::new(Duration::from_millis(
694                    config.request_timeout_ms,
695                )))
696                .wrap(ConcurrencyLimit::new(config.max_concurrent_requests))
697                .wrap(RequestBodyLimit::new(config.max_body_bytes))
698                .route("/healthz", web::get().to(HttpResponse::Ok)),
699        )
700        .await;
701
702        let response = actix_test::call_service(
703            &app,
704            actix_test::TestRequest::get().uri("/healthz").to_request(),
705        )
706        .await;
707        assert_eq!(response.status(), StatusCode::OK);
708        assert!(response.headers().contains_key("x-request-id"));
709        assert!(response.headers().contains_key("traceparent"));
710        assert!(metrics.render().contains("http_requests_total"));
711    }
712
713    #[actix_rt::test]
714    async fn serverless_handler_reuses_routes_middleware_metrics_and_static_fallback() {
715        let config = RestServerConfig {
716            route_groups: vec![RouteGroupConfig {
717                prefix: "/api".to_owned(),
718                middleware: vec!["tag".to_owned()],
719                routes: vec![crate::RoutePolicyConfig {
720                    method: "GET".to_owned(),
721                    path: "/value".to_owned(),
722                    public: true,
723                    jwt: None,
724                    timeout_ms: None,
725                    max_body_bytes: None,
726                    priority: None,
727                    sse: None,
728                }],
729                ..RouteGroupConfig::default()
730            }],
731            ..RestServerConfig::default()
732        };
733        let server = RestServer::new(config)
734            .unwrap()
735            .with_route_middleware(
736                "tag",
737                |request: ServiceRequest, next: crate::RouteMiddlewareNext| async move {
738                    let mut response = next.call(request).await?;
739                    response.headers_mut().insert(
740                        "x-route-middleware".parse().unwrap(),
741                        HeaderValue::from_static("yes"),
742                    );
743                    Ok(response)
744                },
745            )
746            .unwrap()
747            .with_static_assets(
748                StaticAssets::embedded([(
749                    "index.html",
750                    crate::EmbeddedAsset::inferred("serverless home"),
751                )])
752                .unwrap(),
753            );
754        let metrics = server.metrics();
755        let handler = server
756            .serverless_handler(|routes| {
757                routes.route("/api/value", web::get().to(|| async { "value" }));
758            })
759            .await
760            .unwrap();
761
762        let api = handler
763            .call(ServerlessRequest::new(
764                Method::GET,
765                Uri::from_static("/api/value"),
766                web::Bytes::new(),
767            ))
768            .await
769            .unwrap();
770        assert_eq!(api.status, StatusCode::OK);
771        assert_eq!(api.body, "value");
772        assert_eq!(api.headers.get("x-route-middleware").unwrap(), "yes");
773        assert!(api.headers.contains_key("x-request-id"));
774        assert!(api.headers.contains_key("traceparent"));
775
776        let static_response = handler
777            .call(ServerlessRequest::new(
778                Method::GET,
779                Uri::from_static("/"),
780                web::Bytes::new(),
781            ))
782            .await
783            .unwrap();
784        assert_eq!(static_response.status, StatusCode::OK);
785        assert_eq!(static_response.body, "serverless home");
786        assert!(metrics.render().contains("http_requests_total"));
787    }
788
789    #[actix_rt::test]
790    async fn shutdown_signal_gracefully_drains_an_inflight_request() {
791        let listener = TcpListener::bind("127.0.0.1:0").unwrap();
792        let address = listener.local_addr().unwrap();
793        let started = Arc::new(Notify::new());
794        let release = Arc::new(Notify::new());
795        let (shutdown_sender, shutdown_receiver) = tokio::sync::oneshot::channel();
796        let server = RestServer::new(RestServerConfig {
797            address,
798            shutdown_timeout_ms: 2_000,
799            request_timeout_ms: 2_000,
800            ..RestServerConfig::default()
801        })
802        .unwrap();
803
804        let server_task = actix_rt::spawn({
805            let started = Arc::clone(&started);
806            let release = Arc::clone(&release);
807            async move {
808                server
809                    .serve_on_until(
810                        listener,
811                        move |routes| {
812                            routes.route(
813                                "/slow",
814                                web::get().to({
815                                    let started = Arc::clone(&started);
816                                    let release = Arc::clone(&release);
817                                    move || {
818                                        let started = Arc::clone(&started);
819                                        let release = Arc::clone(&release);
820                                        async move {
821                                            started.notify_one();
822                                            release.notified().await;
823                                            HttpResponse::Ok().body("finished")
824                                        }
825                                    }
826                                }),
827                            );
828                        },
829                        async move {
830                            let _ = shutdown_receiver.await;
831                        },
832                    )
833                    .await
834            }
835        });
836
837        let request_task =
838            actix_rt::spawn(async move { reqwest::get(format!("http://{address}/slow")).await });
839        actix_rt::time::timeout(Duration::from_secs(1), started.notified())
840            .await
841            .unwrap();
842        shutdown_sender.send(()).unwrap();
843        actix_rt::time::sleep(Duration::from_millis(20)).await;
844        assert!(!request_task.is_finished());
845
846        release.notify_one();
847        let response = request_task.await.unwrap().unwrap();
848        assert_eq!(response.status(), reqwest::StatusCode::OK);
849        assert_eq!(response.text().await.unwrap(), "finished");
850        server_task.await.unwrap().unwrap();
851    }
852}
rest/server.rs

rest/
server.rs
