spin_trigger_http/
server.rs

1use std::{
2    collections::HashMap,
3    io::{ErrorKind, IsTerminal},
4    marker::PhantomData,
5    net::SocketAddr,
6    pin::Pin,
7    sync::{Arc, OnceLock},
8    task::{Context, Poll},
9    time::{Duration, Instant},
10};
11
12use anyhow::{Context as _, bail};
13use http::{
14    Request, Response, StatusCode, Uri,
15    uri::{Authority, Scheme},
16};
17use http_body_util::BodyExt;
18use hyper::{
19    body::{Bytes, Incoming},
20    service::service_fn,
21};
22use hyper_util::{
23    rt::{TokioExecutor, TokioIo},
24    server::conn::auto::Builder,
25};
26use pin_project_lite::pin_project;
27use rand::RngExt;
28use spin_app::{APP_DESCRIPTION_KEY, APP_NAME_KEY};
29use spin_factor_outbound_http::{OutboundHttpFactor, SelfRequestOrigin};
30use spin_factors::RuntimeFactors;
31use spin_factors_executor::InstanceState;
32use spin_http::{
33    app_info::AppInfo,
34    body,
35    config::{HttpExecutorType, HttpTriggerConfig},
36    routes::{RouteInfo, RouteMatch, Router},
37    trigger::HandlerType,
38};
39use tokio::{
40    io::{AsyncRead, AsyncWrite},
41    net::TcpListener,
42    task,
43};
44use tracing::Instrument;
45use wasmtime::{Store, StoreContextMut, ToWasmtimeResult, component::GuestTaskId};
46use wasmtime_wasi::p2::bindings::CommandIndices;
47use wasmtime_wasi_http::handler::{
48    HandlerState, Instance, Proxy, ShouldAccept, ViewFn, WorkerExpiration, WorkerState,
49    WorkerStatus,
50};
51use wasmtime_wasi_http::p2::body::HyperOutgoingBody;
52use wasmtime_wasi_http::p3::bindings::Service;
53
54use crate::{
55    Body, InstanceReuseConfig, NotFoundRouteKind, OutputFormat, TlsConfig, TriggerApp,
56    TriggerInstanceBuilder,
57    headers::strip_forbidden_headers,
58    instrument::{MatchedRoute, finalize_http_span, http_span, instrument_error},
59    outbound_http::OutboundHttpInterceptor,
60    spin::SpinHttpExecutor,
61    wagi::WagiHttpExecutor,
62    wasi::WasiHttpExecutor,
63    wasip3::Wasip3HttpExecutor,
64};
65
66pub const MAX_RETRIES: u16 = 10;
67
68pub(crate) fn set_request_deadline<T>(
69    store: &mut spin_core::Store<T>,
70    request_deadline: Option<Duration>,
71) {
72    if let Some(timeout) = request_deadline {
73        store.set_deadline(Instant::now() + timeout);
74    }
75}
76
77/// An HTTP server which runs Spin apps.
78pub struct HttpServer<F: RuntimeFactors> {
79    /// The address the server was configured to listen on (the `--listen` value).
80    listen_addr: SocketAddr,
81    /// The address the server is actually bound to, captured once after binding.
82    ///
83    /// This can differ from `listen_addr` when the OS assigns the port — e.g.
84    /// `--listen 127.0.0.1:0` or `--find-free-port`. Self-request origins must use
85    /// this real address rather than the configured one.
86    local_addr: OnceLock<SocketAddr>,
87    /// The TLS configuration for the server.
88    tls_config: Option<TlsConfig>,
89    /// The maximum buffer size for an HTTP1 connection.
90    http1_max_buf_size: Option<usize>,
91    /// Whether to find a free port if the specified port is already in use.
92    find_free_port: bool,
93    /// The output format for the server's startup information.
94    output_format: OutputFormat,
95    /// Hard Wasmtime request deadline for direct HTTP executor paths.
96    request_deadline: Option<Duration>,
97    /// Request router.
98    router: Router,
99    /// The app being triggered.
100    trigger_app: Arc<TriggerApp<F>>,
101    // Component ID -> component trigger config
102    component_trigger_configs: HashMap<spin_http::routes::TriggerLookupKey, HttpTriggerConfig>,
103    // Component ID -> handler type
104    component_handler_types: HashMap<String, HandlerType<HttpHandlerState<F>>>,
105}
106
107impl<F: RuntimeFactors> HttpServer<F> {
108    /// Create a new [`HttpServer`].
109    pub fn new(
110        listen_addr: SocketAddr,
111        tls_config: Option<TlsConfig>,
112        find_free_port: bool,
113        trigger_app: TriggerApp<F>,
114        http1_max_buf_size: Option<usize>,
115        reuse_config: InstanceReuseConfig,
116        output_format: OutputFormat,
117    ) -> anyhow::Result<Self> {
118        // This needs to be a vec before building the router to handle duplicate routes
119        let component_trigger_configs = trigger_app
120            .app()
121            .trigger_configs::<HttpTriggerConfig>("http")?
122            .into_iter()
123            .map(|(trigger_id, config)| config.lookup_key(trigger_id).map(|k| (k, config)))
124            .collect::<Result<Vec<_>, _>>()?;
125
126        // Build router
127        let component_routes = component_trigger_configs
128            .iter()
129            .map(|(key, config)| (key, &config.route));
130        let mut duplicate_routes = Vec::new();
131        let router = Router::build("/", component_routes, Some(&mut duplicate_routes))?;
132        if !duplicate_routes.is_empty() {
133            tracing::error!(
134                "The following component routes are duplicates and will never be used:"
135            );
136            for dup in &duplicate_routes {
137                tracing::error!(
138                    "  {}: {} (duplicate of {})",
139                    dup.replaced_id,
140                    dup.route(),
141                    dup.effective_id,
142                );
143            }
144        }
145        if router.contains_reserved_route() {
146            tracing::error!(
147                "Routes under {} are handled by the Spin runtime and will never be reached",
148                spin_http::WELL_KNOWN_PREFIX
149            );
150        }
151        tracing::trace!(
152            "Constructed router: {:?}",
153            router.routes().collect::<Vec<_>>()
154        );
155
156        // Now that router is built we can merge duplicate routes by component
157        let component_trigger_configs = HashMap::from_iter(component_trigger_configs);
158
159        let trigger_app = Arc::new(trigger_app);
160
161        let component_handler_types = component_trigger_configs
162            .iter()
163            .filter_map(|(key, trigger_config)| match key {
164                spin_http::routes::TriggerLookupKey::Component(component) => Some(
165                    Self::handler_type_for_component(
166                        &trigger_app,
167                        component,
168                        &trigger_config.executor,
169                        reuse_config,
170                    )
171                    .map(|ht| (component.clone(), ht)),
172                ),
173                spin_http::routes::TriggerLookupKey::Trigger(_) => None,
174            })
175            .collect::<anyhow::Result<_>>()?;
176        Ok(Self {
177            listen_addr,
178            local_addr: OnceLock::new(),
179            tls_config,
180            find_free_port,
181            router,
182            trigger_app,
183            http1_max_buf_size,
184            component_trigger_configs,
185            component_handler_types,
186            output_format,
187            request_deadline: reuse_config.request_deadline,
188        })
189    }
190
191    fn handler_type_for_component(
192        trigger_app: &Arc<TriggerApp<F>>,
193        component_id: &str,
194        executor: &Option<HttpExecutorType>,
195        reuse_config: InstanceReuseConfig,
196    ) -> anyhow::Result<HandlerType<HttpHandlerState<F>>> {
197        let pre = trigger_app.get_instance_pre(component_id)?;
198        let handler_type = match executor {
199            None | Some(HttpExecutorType::Http) => HandlerType::from_instance_pre(
200                pre,
201                HttpHandlerState {
202                    component_id: component_id.into(),
203                    reuse_config,
204                    server: Default::default(),
205                    self_scheme: Default::default(),
206                },
207            )?,
208            Some(HttpExecutorType::Wagi(wagi_config)) => {
209                anyhow::ensure!(
210                    wagi_config.entrypoint == "_start",
211                    "Wagi component '{component_id}' cannot use deprecated 'entrypoint' field"
212                );
213                HandlerType::Wagi(
214                    CommandIndices::new(pre)
215                        .map_err(anyhow::Error::from)
216                        .context("failed to find wasi command interface for wagi executor")?,
217                )
218            }
219        };
220        Ok(handler_type)
221    }
222
223    /// Serve incoming requests over the provided [`TcpListener`].
224    pub async fn serve(self: Arc<Self>) -> anyhow::Result<()> {
225        let listener: TcpListener = if self.find_free_port {
226            self.search_for_free_port().await?
227        } else {
228            TcpListener::bind(self.listen_addr).await.map_err(|err| {
229                if err.kind() == ErrorKind::AddrInUse {
230                    anyhow::anyhow!("{} is already in use. To have Spin search for a free port, use the --find-free-port option.", self.listen_addr)
231                } else {
232                    anyhow::anyhow!("Unable to listen on {}: {err:?}", self.listen_addr)
233                }
234            })?
235        };
236
237        let _ = self.local_addr.set(listener.local_addr()?);
238
239        if let Some(tls_config) = self.tls_config.clone() {
240            self.serve_https(listener, tls_config).await?;
241        } else {
242            self.serve_http(listener).await?;
243        }
244        Ok(())
245    }
246
247    async fn search_for_free_port(&self) -> anyhow::Result<TcpListener> {
248        let mut found_listener = None;
249        let mut addr = self.listen_addr;
250
251        for _ in 1..=MAX_RETRIES {
252            if addr.port() == u16::MAX {
253                anyhow::bail!(
254                    "Couldn't find a free port as we've reached the maximum port number. Consider retrying with a lower base port."
255                );
256            }
257
258            match TcpListener::bind(addr).await {
259                Ok(listener) => {
260                    found_listener = Some(listener);
261                    break;
262                }
263                Err(err) if err.kind() == ErrorKind::AddrInUse => {
264                    addr.set_port(addr.port() + 1);
265                    continue;
266                }
267                Err(err) => anyhow::bail!("Unable to listen on {addr}: {err:?}",),
268            }
269        }
270
271        found_listener.ok_or_else(|| anyhow::anyhow!(
272            "Couldn't find a free port in the range {}-{}. Consider retrying with a different base port.",
273            self.listen_addr.port(),
274            self.listen_addr.port() + MAX_RETRIES
275        ))
276    }
277
278    async fn serve_http(self: Arc<Self>, listener: TcpListener) -> anyhow::Result<()> {
279        self.print_startup_msgs("http", &listener)?;
280        loop {
281            let (stream, client_addr) = listener.accept().await?;
282            self.clone()
283                .serve_connection(stream, Scheme::HTTP, client_addr);
284        }
285    }
286
287    async fn serve_https(
288        self: Arc<Self>,
289        listener: TcpListener,
290        tls_config: TlsConfig,
291    ) -> anyhow::Result<()> {
292        self.print_startup_msgs("https", &listener)?;
293        let acceptor = tls_config.server_config()?;
294        loop {
295            let (stream, client_addr) = listener.accept().await?;
296            match acceptor.accept(stream).await {
297                Ok(stream) => self
298                    .clone()
299                    .serve_connection(stream, Scheme::HTTPS, client_addr),
300                Err(err) => tracing::error!(?err, "Failed to start TLS session"),
301            }
302        }
303    }
304
305    /// Handles incoming requests using an HTTP executor.
306    ///
307    /// This method handles well known paths and routes requests to the handler when the router
308    /// matches the requests path.
309    pub async fn handle(
310        self: &Arc<Self>,
311        mut req: Request<Body>,
312        server_scheme: Scheme,
313        client_addr: SocketAddr,
314    ) -> anyhow::Result<Response<Body>> {
315        strip_forbidden_headers(&mut req);
316
317        spin_telemetry::extract_trace_context(&req);
318
319        let path = req.uri().path().to_string();
320
321        tracing::info!("Processing request on path '{path}'");
322
323        // Handle well-known spin paths
324        if let Some(well_known) = path.strip_prefix(spin_http::WELL_KNOWN_PREFIX) {
325            return match well_known {
326                "health" => Ok(MatchedRoute::with_response_extension(
327                    Response::new(body::full(Bytes::from_static(b"OK"))),
328                    path,
329                )),
330                "info" => self.app_info(path),
331                _ => Self::not_found(NotFoundRouteKind::WellKnown),
332            };
333        }
334
335        match self.router.route(&path) {
336            Ok(route_match) => {
337                self.handle_trigger_route(req, route_match, server_scheme, client_addr)
338                    .await
339            }
340            Err(_) => Self::not_found(NotFoundRouteKind::Normal(path.to_string())),
341        }
342    }
343
344    /// Handles a successful route match.
345    pub async fn handle_trigger_route(
346        self: &Arc<Self>,
347        mut req: Request<Body>,
348        route_match: RouteMatch<'_, '_>,
349        server_scheme: Scheme,
350        client_addr: SocketAddr,
351    ) -> anyhow::Result<Response<Body>> {
352        set_req_uri(&mut req, server_scheme)?;
353        let app_id = self
354            .trigger_app
355            .app()
356            .get_metadata(APP_NAME_KEY)?
357            .unwrap_or_else(|| "<unnamed>".into());
358
359        let lookup_key = route_match.lookup_key();
360
361        spin_telemetry::metrics::monotonic_counter!(
362            spin.request_count = 1,
363            trigger_type = "http",
364            app_id = app_id,
365            component_id = lookup_key.to_string()
366        );
367
368        let trigger_config = self
369            .component_trigger_configs
370            .get(lookup_key)
371            .with_context(|| format!("unknown routing destination '{lookup_key}'"))?;
372
373        match (&trigger_config.component, &trigger_config.static_response) {
374            (Some(component), None) => {
375                self.respond_wasm_component(
376                    req,
377                    route_match,
378                    client_addr,
379                    component,
380                    &trigger_config.executor,
381                )
382                .await
383            }
384            (None, Some(static_response)) => Self::respond_static_response(static_response),
385            // These error cases should have been ruled out by this point but belt and braces
386            (None, None) => Err(anyhow::anyhow!(
387                "Triggers must specify either component or static_response - neither is specified for {}",
388                route_match.raw_route()
389            )),
390            (Some(_), Some(_)) => Err(anyhow::anyhow!(
391                "Triggers must specify either component or static_response - both are specified for {}",
392                route_match.raw_route()
393            )),
394        }
395    }
396
397    fn get_local_addr(&self) -> SocketAddr {
398        self.local_addr.get().copied().unwrap_or(self.listen_addr)
399    }
400
401    async fn respond_wasm_component(
402        self: &Arc<Self>,
403        req: Request<Body>,
404        route_match: RouteMatch<'_, '_>,
405        client_addr: SocketAddr,
406        component_id: &str,
407        executor: &Option<HttpExecutorType>,
408    ) -> anyhow::Result<Response<Body>> {
409        // Prepare HTTP executor
410        let handler_type = self
411            .component_handler_types
412            .get(component_id)
413            .with_context(|| format!("unknown component ID {component_id:?}"))?;
414        let executor = executor.as_ref().unwrap_or(&HttpExecutorType::Http);
415
416        let res = match executor {
417            HttpExecutorType::Http => match handler_type {
418                HandlerType::Spin => {
419                    SpinHttpExecutor
420                        .execute(self, &route_match, req, client_addr, component_id)
421                        .await
422                }
423                HandlerType::Wasi0_3(handler) => {
424                    Wasip3HttpExecutor(handler)
425                        .execute(self, &route_match, req, client_addr)
426                        .await
427                }
428                HandlerType::Wasi0_2(_)
429                | HandlerType::Wasi2023_11_10(_)
430                | HandlerType::Wasi2023_10_18(_)
431                | HandlerType::Wasi2026_03_15(_) => {
432                    WasiHttpExecutor { handler_type }
433                        .execute(self, &route_match, req, client_addr, component_id)
434                        .await
435                }
436                HandlerType::Wagi(_) => unreachable!(),
437            },
438            HttpExecutorType::Wagi(wagi_config) => {
439                let indices = match handler_type {
440                    HandlerType::Wagi(indices) => indices,
441                    _ => unreachable!(),
442                };
443                let executor = WagiHttpExecutor {
444                    wagi_config,
445                    indices,
446                };
447                executor
448                    .execute(self, &route_match, req, client_addr, component_id)
449                    .await
450            }
451        };
452        match res {
453            Ok(res) => Ok(MatchedRoute::with_response_extension(
454                res,
455                route_match.raw_route(),
456            )),
457            Err(err) => {
458                tracing::error!("Error processing request: {err:?}");
459                instrument_error(&err);
460                Self::internal_error(None, route_match.raw_route())
461            }
462        }
463    }
464
465    pub(crate) fn trigger_instance_builder(
466        self: &'_ Arc<Self>,
467        component_id: &str,
468        self_scheme: Option<&Scheme>,
469    ) -> anyhow::Result<TriggerInstanceBuilder<'_, F>> {
470        let mut instance_builder = self.trigger_app.prepare(component_id)?;
471
472        // Set up outbound HTTP request origin and service chaining
473        // The outbound HTTP factor is required since both inbound and outbound wasi HTTP
474        // implementations assume they use the same underlying wasmtime resource storage.
475        // Eventually, we may be able to factor this out to a separate factor.
476        let outbound_http = instance_builder
477            .factor_builder::<OutboundHttpFactor>()
478            .context(
479            "The wasi HTTP trigger was configured without the required wasi outbound http support",
480        )?;
481
482        let self_scheme = self_scheme.cloned().unwrap_or(Scheme::HTTPS);
483        let self_addr = self.get_local_addr();
484        let origin = SelfRequestOrigin::create(self_scheme, &self_addr.to_string())?;
485        outbound_http.set_self_request_origin(origin);
486        outbound_http.set_request_interceptor(OutboundHttpInterceptor::new(self.clone()))?;
487        Ok(instance_builder)
488    }
489
490    fn respond_static_response(
491        sr: &spin_http::config::StaticResponse,
492    ) -> anyhow::Result<Response<Body>> {
493        let mut response = Response::builder();
494
495        response = response.status(sr.status());
496        for (header_name, header_value) in sr.headers() {
497            response = response.header(header_name, header_value);
498        }
499
500        let body = match sr.body() {
501            Some(b) => body::full(b.clone().into()),
502            None => body::empty(),
503        };
504
505        Ok(response.body(body)?)
506    }
507
508    /// Returns spin status information.
509    fn app_info(&self, route: String) -> anyhow::Result<Response<Body>> {
510        let info = AppInfo::new(self.trigger_app.app());
511        let body = serde_json::to_vec_pretty(&info)?;
512        Ok(MatchedRoute::with_response_extension(
513            Response::builder()
514                .header("content-type", "application/json")
515                .body(body::full(body.into()))?,
516            route,
517        ))
518    }
519
520    /// Creates an HTTP 500 response.
521    fn internal_error(
522        body: Option<&str>,
523        route: impl Into<String>,
524    ) -> anyhow::Result<Response<Body>> {
525        let body = match body {
526            Some(body) => body::full(Bytes::copy_from_slice(body.as_bytes())),
527            None => body::empty(),
528        };
529
530        Ok(MatchedRoute::with_response_extension(
531            Response::builder()
532                .status(StatusCode::INTERNAL_SERVER_ERROR)
533                .body(body)?,
534            route,
535        ))
536    }
537
538    /// Creates an HTTP 404 response.
539    fn not_found(kind: NotFoundRouteKind) -> anyhow::Result<Response<Body>> {
540        use std::sync::atomic::{AtomicBool, Ordering};
541        static SHOWN_GENERIC_404_WARNING: AtomicBool = AtomicBool::new(false);
542        if let NotFoundRouteKind::Normal(route) = kind
543            && !SHOWN_GENERIC_404_WARNING.fetch_or(true, Ordering::Relaxed)
544            && std::io::stderr().is_terminal()
545        {
546            terminal::warn!(
547                "Request to {route} matched no pattern, and received a generic 404 response. To serve a more informative 404 page, add a catch-all (/...) route."
548            );
549        }
550        Ok(Response::builder()
551            .status(StatusCode::NOT_FOUND)
552            .body(body::empty())?)
553    }
554
555    fn serve_connection<S: AsyncRead + AsyncWrite + Unpin + Send + 'static>(
556        self: Arc<Self>,
557        stream: S,
558        server_scheme: Scheme,
559        client_addr: SocketAddr,
560    ) {
561        task::spawn(async move {
562            let mut server_builder = Builder::new(TokioExecutor::new());
563
564            if let Some(http1_max_buf_size) = self.http1_max_buf_size {
565                server_builder.http1().max_buf_size(http1_max_buf_size);
566            }
567
568            if let Err(err) = server_builder
569                .serve_connection(
570                    TokioIo::new(stream),
571                    service_fn(move |request| {
572                        self.clone().instrumented_service_fn(
573                            server_scheme.clone(),
574                            client_addr,
575                            request,
576                        )
577                    }),
578                )
579                .await
580            {
581                tracing::warn!("Error serving HTTP connection: {err:?}");
582            }
583        });
584    }
585
586    async fn instrumented_service_fn(
587        self: Arc<Self>,
588        server_scheme: Scheme,
589        client_addr: SocketAddr,
590        request: Request<Incoming>,
591    ) -> anyhow::Result<Response<HyperOutgoingBody>> {
592        let span = http_span!(request, client_addr);
593        let method = request.method().to_string();
594        async {
595            let result = self
596                .handle(
597                    request.map(|body: Incoming| {
598                        body.map_err(wasmtime_wasi_http::p2::hyper_response_error)
599                            .boxed_unsync()
600                    }),
601                    server_scheme,
602                    client_addr,
603                )
604                .await;
605            finalize_http_span(result, method)
606        }
607        .instrument(span)
608        .await
609    }
610
611    fn get_description_for_route(
612        &self,
613        key: &spin_http::routes::TriggerLookupKey,
614    ) -> anyhow::Result<Option<String>> {
615        if let spin_http::routes::TriggerLookupKey::Component(component_id) = key {
616            self.trigger_app
617                .app()
618                .get_component(component_id)
619                .and_then(|c| c.get_metadata(APP_DESCRIPTION_KEY).transpose())
620                .transpose()
621                .map_err(Into::into)
622        } else {
623            Ok(None)
624        }
625    }
626
627    fn print_startup_msgs(&self, scheme: &str, listener: &TcpListener) -> anyhow::Result<()> {
628        let local_addr = listener.local_addr()?;
629        let base_url = format!("{scheme}://{local_addr:?}");
630        tracing::info!("Serving {base_url}");
631
632        match self.output_format {
633            OutputFormat::Plain => {
634                terminal::step!("\nServing", "{base_url}");
635                println!("Available Routes:");
636                for (route, key) in self.router.routes() {
637                    println!("  {key}: {base_url}{route}");
638                    if let Some(description) = self.get_description_for_route(key)? {
639                        println!("    {description}");
640                    }
641                }
642            }
643            OutputFormat::Json => {
644                #[derive(serde::Serialize)]
645                struct RoutesOutput {
646                    base_url: String,
647                    routes: Vec<RouteEntry>,
648                }
649
650                #[derive(serde::Serialize)]
651                struct RouteEntry {
652                    id: String,
653                    route: String,
654                    wildcard: bool,
655                    #[serde(skip_serializing_if = "Option::is_none")]
656                    description: Option<String>,
657                }
658                let mut routes = Vec::new();
659                for (route, key) in self.router.routes() {
660                    routes.push(RouteEntry {
661                        id: key.to_string(),
662                        route: route.path().to_string(),
663                        wildcard: route.is_wildcard(),
664                        description: self.get_description_for_route(key)?,
665                    });
666                }
667
668                let output = RoutesOutput { base_url, routes };
669                println!("{}", serde_json::to_string_pretty(&output)?);
670            }
671        }
672        Ok(())
673    }
674
675    pub(crate) fn request_deadline(&self) -> Option<Duration> {
676        self.request_deadline
677    }
678}
679
680/// The incoming request's scheme and authority
681///
682/// The incoming request's URI is relative to the server, so we need to set the scheme and authority.
683/// Either the `Host` header or the request's URI's authority is used as the source of truth for the authority.
684/// This function will error if the authority cannot be unambiguously determined.
685fn set_req_uri(req: &mut Request<Body>, scheme: Scheme) -> anyhow::Result<()> {
686    let uri = req.uri().clone();
687    let mut parts = uri.into_parts();
688    let headers = req.headers();
689    let header_authority = headers
690        .get(http::header::HOST)
691        .map(|h| -> anyhow::Result<Authority> {
692            let host_header = h.to_str().context("'Host' header is not valid UTF-8")?;
693            host_header
694                .parse()
695                .context("'Host' header contains an invalid authority")
696        })
697        .transpose()?;
698    let uri_authority = parts.authority;
699
700    // Get authority either from request URI or from 'Host' header
701    let authority = match (header_authority, uri_authority) {
702        (None, None) => bail!("no 'Host' header present in request"),
703        (None, Some(a)) => a,
704        (Some(a), None) => a,
705        (Some(a1), Some(a2)) => {
706            // Ensure that if `req.authority` is set, it matches what was in the `Host` header
707            // https://github.com/hyperium/hyper/issues/1612
708            if a1 != a2 {
709                return Err(anyhow::anyhow!(
710                    "authority in 'Host' header does not match authority in URI"
711                ));
712            }
713            a1
714        }
715    };
716    parts.scheme = Some(scheme);
717    parts.authority = Some(authority);
718    *req.uri_mut() = Uri::from_parts(parts).unwrap();
719    Ok(())
720}
721
722pin_project! {
723    pub(crate) struct HttpWorkerExpiration {
724        idle_timeout: Duration,
725        request_timeout: Duration,
726        #[pin]
727        sleep: tokio::time::Sleep,
728    }
729}
730
731impl WorkerExpiration for HttpWorkerExpiration {
732    fn poll(
733        self: Pin<&mut Self>,
734        cx: &mut Context<'_>,
735        status: WorkerStatus,
736        start: Instant,
737    ) -> Poll<()> {
738        let mut me = self.project();
739
740        let timeout = match status {
741            WorkerStatus::Idle => *me.idle_timeout,
742            // TODO: add a dedicated `post_return_timeout` config setting
743            // instead of reusing `request_timeout` for
744            // `WorkerStatus::PostReturn` here
745            WorkerStatus::Requests | WorkerStatus::PostReturn => *me.request_timeout,
746        };
747
748        if let Some(deadline) = start.checked_add(timeout) {
749            let deadline = deadline.into();
750            if deadline != me.sleep.deadline() {
751                me.sleep.as_mut().reset(deadline);
752            }
753            me.sleep.poll(cx)
754        } else {
755            Poll::Pending
756        }
757    }
758}
759
760pub(crate) struct HttpWorkerState<F: RuntimeFactors> {
761    request_timeout: Duration,
762    max_instance_reuse_count: usize,
763    max_instance_concurrent_reuse_count: usize,
764    _phantom: PhantomData<F>,
765}
766
767impl<F: RuntimeFactors> WorkerState for HttpWorkerState<F> {
768    type StoreData = InstanceState<F::InstanceState, ()>;
769    type RequestId = ();
770
771    fn should_accept_request(&self, concurrent_count: usize, total_count: usize) -> ShouldAccept {
772        if total_count >= self.max_instance_reuse_count {
773            ShouldAccept::Never
774        } else if concurrent_count >= self.max_instance_concurrent_reuse_count {
775            ShouldAccept::No
776        } else {
777            ShouldAccept::Yes
778        }
779    }
780
781    fn on_request_start(
782        &self,
783        _: StoreContextMut<'_, Self::StoreData>,
784        _: Self::RequestId,
785        _: GuestTaskId,
786    ) -> Pin<Box<dyn Future<Output = ()> + Send + Sync + 'static>> {
787        Box::pin(tokio::time::sleep(self.request_timeout))
788    }
789
790    fn drop(&self, store: Store<Self::StoreData>, result: Result<(), wasmtime::Error>) {
791        if let Err(error) = result {
792            eprintln!("worker failed: {error:?}");
793        }
794
795        drop(store);
796    }
797}
798
799pub(crate) struct HttpHandlerState<F: RuntimeFactors> {
800    component_id: String,
801    reuse_config: InstanceReuseConfig,
802    server: OnceLock<Arc<HttpServer<F>>>,
803    self_scheme: OnceLock<Scheme>,
804}
805
806impl<F: RuntimeFactors> HttpHandlerState<F> {
807    pub(crate) fn init_once(&self, server: &Arc<HttpServer<F>>, first_uri: &Uri) {
808        self.server.get_or_init(|| server.clone());
809        if let Some(scheme) = first_uri.scheme() {
810            self.self_scheme.get_or_init(|| scheme.clone());
811        }
812    }
813}
814
815impl<F: RuntimeFactors> HandlerState for HttpHandlerState<F> {
816    type StoreData = InstanceState<F::InstanceState, ()>;
817    type WorkerExpiration = HttpWorkerExpiration;
818    type WorkerState = HttpWorkerState<F>;
819
820    async fn instantiate(
821        &self,
822    ) -> wasmtime::Result<Instance<Self::StoreData, Self::WorkerExpiration, Self::WorkerState>>
823    {
824        let (instance, mut store) = self
825            .server
826            .get()
827            .expect("server should have been set")
828            .trigger_instance_builder(&self.component_id, self.self_scheme.get())
829            .to_wasmtime_result()?
830            .instantiate(())
831            .await
832            .to_wasmtime_result()?;
833        set_request_deadline(&mut store, self.reuse_config.request_deadline);
834
835        let mut store = store.into_inner();
836
837        let proxy = Proxy::P3(Service::new(&mut store, &instance).unwrap());
838
839        let request_timeout = self
840            .reuse_config
841            .request_timeout
842            .map(|range| rand::rng().random_range(range))
843            .unwrap_or(Duration::MAX);
844
845        Ok(Instance {
846            store,
847            proxy,
848            view: ViewFn::P3(|data| {
849                spin_factor_outbound_http::OutboundHttpFactor::get_wasi_p3_http_impl(
850                    data.factors_instance_state_mut(),
851                )
852                .unwrap()
853            }),
854            expiration: HttpWorkerExpiration {
855                idle_timeout: rand::rng().random_range(self.reuse_config.idle_instance_timeout),
856                request_timeout,
857                sleep: tokio::time::sleep(Duration::MAX),
858            },
859            state: HttpWorkerState {
860                request_timeout,
861                max_instance_reuse_count: rand::rng()
862                    .random_range(self.reuse_config.max_instance_reuse_count),
863                max_instance_concurrent_reuse_count: rand::rng()
864                    .random_range(self.reuse_config.max_instance_concurrent_reuse_count),
865                _phantom: PhantomData,
866            },
867        })
868    }
869}
spin_trigger_http/server.rs

spin_trigger_http/
server.rs
