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, Weak},
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::Error as WasiHttpError;
48use wasmtime_wasi_http::handler::{
49    HandlerState, Instance, Proxy, ShouldAccept, WorkerExpiration, WorkerState, 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    /// The application name, resolved once for use as the `app_id` telemetry attribute.
102    app_id: Arc<str>,
103    // Component ID -> component trigger config
104    component_trigger_configs: HashMap<spin_http::routes::TriggerLookupKey, HttpTriggerConfig>,
105    // Component ID -> handler type
106    component_handler_types: HashMap<String, HandlerType<HttpHandlerState<F>>>,
107}
108
109impl<F: RuntimeFactors> HttpServer<F> {
110    /// Create a new [`HttpServer`].
111    pub fn new(
112        listen_addr: SocketAddr,
113        tls_config: Option<TlsConfig>,
114        find_free_port: bool,
115        trigger_app: TriggerApp<F>,
116        http1_max_buf_size: Option<usize>,
117        reuse_config: InstanceReuseConfig,
118        output_format: OutputFormat,
119    ) -> anyhow::Result<Self> {
120        // This needs to be a vec before building the router to handle duplicate routes
121        let component_trigger_configs = trigger_app
122            .app()
123            .trigger_configs::<HttpTriggerConfig>("http")?
124            .into_iter()
125            .map(|(trigger_id, config)| config.lookup_key(trigger_id).map(|k| (k, config)))
126            .collect::<Result<Vec<_>, _>>()?;
127
128        // Build router
129        let component_routes = component_trigger_configs
130            .iter()
131            .map(|(key, config)| (key, &config.route));
132        let mut duplicate_routes = Vec::new();
133        let router = Router::build("/", component_routes, Some(&mut duplicate_routes))?;
134        if !duplicate_routes.is_empty() {
135            tracing::error!(
136                "The following component routes are duplicates and will never be used:"
137            );
138            for dup in &duplicate_routes {
139                tracing::error!(
140                    "  {}: {} (duplicate of {})",
141                    dup.replaced_id,
142                    dup.route(),
143                    dup.effective_id,
144                );
145            }
146        }
147        if router.contains_reserved_route() {
148            tracing::error!(
149                "Routes under {} are handled by the Spin runtime and will never be reached",
150                spin_http::WELL_KNOWN_PREFIX
151            );
152        }
153        tracing::trace!(
154            "Constructed router: {:?}",
155            router.routes().collect::<Vec<_>>()
156        );
157
158        // Now that router is built we can merge duplicate routes by component
159        let component_trigger_configs = HashMap::from_iter(component_trigger_configs);
160
161        let trigger_app = Arc::new(trigger_app);
162
163        let app_id: Arc<str> = trigger_app
164            .app()
165            .get_metadata(APP_NAME_KEY)?
166            .unwrap_or_else(|| "<unnamed>".into())
167            .into();
168
169        let component_handler_types = component_trigger_configs
170            .iter()
171            .filter_map(|(key, trigger_config)| match key {
172                spin_http::routes::TriggerLookupKey::Component(component) => Some(
173                    Self::handler_type_for_component(
174                        &trigger_app,
175                        component,
176                        &trigger_config.executor,
177                        reuse_config,
178                    )
179                    .map(|ht| (component.clone(), ht)),
180                ),
181                spin_http::routes::TriggerLookupKey::Trigger(_) => None,
182            })
183            .collect::<anyhow::Result<_>>()?;
184        Ok(Self {
185            listen_addr,
186            local_addr: OnceLock::new(),
187            tls_config,
188            find_free_port,
189            router,
190            trigger_app,
191            app_id,
192            http1_max_buf_size,
193            component_trigger_configs,
194            component_handler_types,
195            output_format,
196            request_deadline: reuse_config.request_deadline,
197        })
198    }
199
200    fn handler_type_for_component(
201        trigger_app: &Arc<TriggerApp<F>>,
202        component_id: &str,
203        executor: &Option<HttpExecutorType>,
204        reuse_config: InstanceReuseConfig,
205    ) -> anyhow::Result<HandlerType<HttpHandlerState<F>>> {
206        let pre = trigger_app.get_instance_pre(component_id)?;
207        let handler_type = match executor {
208            None | Some(HttpExecutorType::Http) => HandlerType::from_instance_pre(
209                pre,
210                HttpHandlerState {
211                    component_id: component_id.into(),
212                    reuse_config,
213                    server: Default::default(),
214                    self_scheme: Default::default(),
215                },
216            )?,
217            Some(HttpExecutorType::Wagi(wagi_config)) => {
218                anyhow::ensure!(
219                    wagi_config.entrypoint == "_start",
220                    "Wagi component '{component_id}' cannot use deprecated 'entrypoint' field"
221                );
222                HandlerType::Wagi(
223                    CommandIndices::new(pre)
224                        .map_err(anyhow::Error::from)
225                        .context("failed to find wasi command interface for wagi executor")?,
226                )
227            }
228        };
229        Ok(handler_type)
230    }
231
232    /// Serve incoming requests over the provided [`TcpListener`].
233    pub async fn serve(self: Arc<Self>) -> anyhow::Result<()> {
234        let listener: TcpListener = if self.find_free_port {
235            self.search_for_free_port().await?
236        } else {
237            TcpListener::bind(self.listen_addr).await.map_err(|err| {
238                if err.kind() == ErrorKind::AddrInUse {
239                    anyhow::anyhow!("{} is already in use. To have Spin search for a free port, use the --find-free-port option.", self.listen_addr)
240                } else {
241                    anyhow::anyhow!("Unable to listen on {}: {err:?}", self.listen_addr)
242                }
243            })?
244        };
245
246        let _ = self.local_addr.set(listener.local_addr()?);
247
248        if let Some(tls_config) = self.tls_config.clone() {
249            self.serve_https(listener, tls_config).await?;
250        } else {
251            self.serve_http(listener).await?;
252        }
253        Ok(())
254    }
255
256    async fn search_for_free_port(&self) -> anyhow::Result<TcpListener> {
257        let mut found_listener = None;
258        let mut addr = self.listen_addr;
259
260        for _ in 1..=MAX_RETRIES {
261            if addr.port() == u16::MAX {
262                anyhow::bail!(
263                    "Couldn't find a free port as we've reached the maximum port number. Consider retrying with a lower base port."
264                );
265            }
266
267            match TcpListener::bind(addr).await {
268                Ok(listener) => {
269                    found_listener = Some(listener);
270                    break;
271                }
272                Err(err) if err.kind() == ErrorKind::AddrInUse => {
273                    addr.set_port(addr.port() + 1);
274                    continue;
275                }
276                Err(err) => anyhow::bail!("Unable to listen on {addr}: {err:?}",),
277            }
278        }
279
280        found_listener.ok_or_else(|| anyhow::anyhow!(
281            "Couldn't find a free port in the range {}-{}. Consider retrying with a different base port.",
282            self.listen_addr.port(),
283            self.listen_addr.port() + MAX_RETRIES
284        ))
285    }
286
287    async fn serve_http(self: Arc<Self>, listener: TcpListener) -> anyhow::Result<()> {
288        self.print_startup_msgs("http", &listener)?;
289        loop {
290            let (stream, client_addr) = listener.accept().await?;
291            self.clone()
292                .serve_connection(stream, Scheme::HTTP, client_addr);
293        }
294    }
295
296    async fn serve_https(
297        self: Arc<Self>,
298        listener: TcpListener,
299        tls_config: TlsConfig,
300    ) -> anyhow::Result<()> {
301        self.print_startup_msgs("https", &listener)?;
302        let acceptor = tls_config.server_config()?;
303        loop {
304            let (stream, client_addr) = listener.accept().await?;
305            match acceptor.accept(stream).await {
306                Ok(stream) => self
307                    .clone()
308                    .serve_connection(stream, Scheme::HTTPS, client_addr),
309                Err(err) => tracing::error!(?err, "Failed to start TLS session"),
310            }
311        }
312    }
313
314    /// Handles incoming requests using an HTTP executor.
315    ///
316    /// This method handles well known paths and routes requests to the handler when the router
317    /// matches the requests path.
318    pub async fn handle(
319        self: &Arc<Self>,
320        mut req: Request<Body>,
321        server_scheme: Scheme,
322        client_addr: SocketAddr,
323    ) -> anyhow::Result<Response<Body>> {
324        strip_forbidden_headers(&mut req);
325
326        spin_telemetry::extract_trace_context(&req);
327
328        let path = req.uri().path().to_string();
329
330        tracing::info!("Processing request on path '{path}'");
331
332        // Handle well-known spin paths
333        if let Some(well_known) = path.strip_prefix(spin_http::WELL_KNOWN_PREFIX) {
334            return match well_known {
335                "health" => Ok(MatchedRoute::with_response_extension(
336                    Response::new(body::full(Bytes::from_static(b"OK"))),
337                    path,
338                )),
339                "info" => self.app_info(path),
340                _ => Self::not_found(NotFoundRouteKind::WellKnown),
341            };
342        }
343
344        match self.router.route(&path) {
345            Ok(route_match) => {
346                self.handle_trigger_route(req, route_match, server_scheme, client_addr)
347                    .await
348            }
349            Err(_) => Self::not_found(NotFoundRouteKind::Normal(path.to_string())),
350        }
351    }
352
353    /// Handles a successful route match.
354    pub async fn handle_trigger_route(
355        self: &Arc<Self>,
356        mut req: Request<Body>,
357        route_match: RouteMatch<'_, '_>,
358        server_scheme: Scheme,
359        client_addr: SocketAddr,
360    ) -> anyhow::Result<Response<Body>> {
361        set_req_uri(&mut req, server_scheme)?;
362        let lookup_key = route_match.lookup_key();
363
364        spin_telemetry::metrics::counter!(
365            spin.request_count = 1,
366            trigger_type = "http",
367            app_id = self.app_id.clone(),
368            component_id = lookup_key.to_string()
369        );
370
371        let trigger_config = self
372            .component_trigger_configs
373            .get(lookup_key)
374            .with_context(|| format!("unknown routing destination '{lookup_key}'"))?;
375
376        match (&trigger_config.component, &trigger_config.static_response) {
377            (Some(component), None) => {
378                self.respond_wasm_component(
379                    req,
380                    route_match,
381                    client_addr,
382                    component,
383                    &trigger_config.executor,
384                )
385                .await
386            }
387            (None, Some(static_response)) => Self::respond_static_response(static_response),
388            // These error cases should have been ruled out by this point but belt and braces
389            (None, None) => Err(anyhow::anyhow!(
390                "Triggers must specify either component or static_response - neither is specified for {}",
391                route_match.raw_route()
392            )),
393            (Some(_), Some(_)) => Err(anyhow::anyhow!(
394                "Triggers must specify either component or static_response - both are specified for {}",
395                route_match.raw_route()
396            )),
397        }
398    }
399
400    fn get_local_addr(&self) -> SocketAddr {
401        self.local_addr.get().copied().unwrap_or(self.listen_addr)
402    }
403
404    async fn respond_wasm_component(
405        self: &Arc<Self>,
406        req: Request<Body>,
407        route_match: RouteMatch<'_, '_>,
408        client_addr: SocketAddr,
409        component_id: &str,
410        executor: &Option<HttpExecutorType>,
411    ) -> anyhow::Result<Response<Body>> {
412        // Prepare HTTP executor
413        let handler_type = self
414            .component_handler_types
415            .get(component_id)
416            .with_context(|| format!("unknown component ID {component_id:?}"))?;
417        let executor = executor.as_ref().unwrap_or(&HttpExecutorType::Http);
418
419        let res = match executor {
420            HttpExecutorType::Http => match handler_type {
421                HandlerType::Spin => {
422                    SpinHttpExecutor
423                        .execute(self, &route_match, req, client_addr, component_id)
424                        .await
425                }
426                HandlerType::Wasi0_3(handler) => {
427                    Wasip3HttpExecutor(handler)
428                        .execute(self, &route_match, req, client_addr)
429                        .await
430                }
431                HandlerType::Wasi0_2(_)
432                | HandlerType::Wasi2023_11_10(_)
433                | HandlerType::Wasi2023_10_18(_)
434                | HandlerType::Wasi2026_03_15(_) => {
435                    WasiHttpExecutor { handler_type }
436                        .execute(self, &route_match, req, client_addr, component_id)
437                        .await
438                }
439                HandlerType::Wagi(_) => unreachable!(),
440            },
441            HttpExecutorType::Wagi(wagi_config) => {
442                let indices = match handler_type {
443                    HandlerType::Wagi(indices) => indices,
444                    _ => unreachable!(),
445                };
446                let executor = WagiHttpExecutor {
447                    wagi_config,
448                    indices,
449                };
450                executor
451                    .execute(self, &route_match, req, client_addr, component_id)
452                    .await
453            }
454        };
455        match res {
456            Ok(res) => Ok(MatchedRoute::with_response_extension(
457                res,
458                route_match.raw_route(),
459            )),
460            Err(err) => {
461                tracing::error!("Error processing request: {err:?}");
462                instrument_error(&err);
463                Self::internal_error(None, route_match.raw_route())
464            }
465        }
466    }
467
468    pub(crate) fn trigger_instance_builder(
469        self: &'_ Arc<Self>,
470        component_id: &str,
471        self_scheme: Option<&Scheme>,
472    ) -> anyhow::Result<TriggerInstanceBuilder<'_, F>> {
473        let mut instance_builder = self.trigger_app.prepare(component_id)?;
474
475        // Set up outbound HTTP request origin and service chaining
476        // The outbound HTTP factor is required since both inbound and outbound wasi HTTP
477        // implementations assume they use the same underlying wasmtime resource storage.
478        // Eventually, we may be able to factor this out to a separate factor.
479        let outbound_http = instance_builder
480            .factor_builder::<OutboundHttpFactor>()
481            .context(
482            "The wasi HTTP trigger was configured without the required wasi outbound http support",
483        )?;
484
485        let self_scheme = self_scheme.cloned().unwrap_or(Scheme::HTTPS);
486        let self_addr = self.get_local_addr();
487        let origin = SelfRequestOrigin::create(self_scheme, &self_addr.to_string())?;
488        outbound_http.set_self_request_origin(origin);
489        outbound_http.set_request_interceptor(OutboundHttpInterceptor::new(self.clone()))?;
490        Ok(instance_builder)
491    }
492
493    fn respond_static_response(
494        sr: &spin_http::config::StaticResponse,
495    ) -> anyhow::Result<Response<Body>> {
496        let mut response = Response::builder();
497
498        response = response.status(sr.status());
499        for (header_name, header_value) in sr.headers() {
500            response = response.header(header_name, header_value);
501        }
502
503        let body = match sr.body() {
504            Some(b) => body::full(b.clone().into()),
505            None => body::empty(),
506        };
507
508        Ok(response.body(body)?)
509    }
510
511    /// Returns spin status information.
512    fn app_info(&self, route: String) -> anyhow::Result<Response<Body>> {
513        let info = AppInfo::new(self.trigger_app.app());
514        let body = serde_json::to_vec_pretty(&info)?;
515        Ok(MatchedRoute::with_response_extension(
516            Response::builder()
517                .header("content-type", "application/json")
518                .body(body::full(body.into()))?,
519            route,
520        ))
521    }
522
523    /// Creates an HTTP 500 response.
524    fn internal_error(
525        body: Option<&str>,
526        route: impl Into<String>,
527    ) -> anyhow::Result<Response<Body>> {
528        let body = match body {
529            Some(body) => body::full(Bytes::copy_from_slice(body.as_bytes())),
530            None => body::empty(),
531        };
532
533        Ok(MatchedRoute::with_response_extension(
534            Response::builder()
535                .status(StatusCode::INTERNAL_SERVER_ERROR)
536                .body(body)?,
537            route,
538        ))
539    }
540
541    /// Creates an HTTP 404 response.
542    fn not_found(kind: NotFoundRouteKind) -> anyhow::Result<Response<Body>> {
543        use std::sync::atomic::{AtomicBool, Ordering};
544        static SHOWN_GENERIC_404_WARNING: AtomicBool = AtomicBool::new(false);
545        if let NotFoundRouteKind::Normal(route) = kind
546            && !SHOWN_GENERIC_404_WARNING.fetch_or(true, Ordering::Relaxed)
547            && std::io::stderr().is_terminal()
548        {
549            terminal::warn!(
550                "Request to {route} matched no pattern, and received a generic 404 response. To serve a more informative 404 page, add a catch-all (/...) route."
551            );
552        }
553        Ok(Response::builder()
554            .status(StatusCode::NOT_FOUND)
555            .body(body::empty())?)
556    }
557
558    fn serve_connection<S: AsyncRead + AsyncWrite + Unpin + Send + 'static>(
559        self: Arc<Self>,
560        stream: S,
561        server_scheme: Scheme,
562        client_addr: SocketAddr,
563    ) {
564        task::spawn(async move {
565            let mut server_builder = Builder::new(TokioExecutor::new());
566
567            if let Some(http1_max_buf_size) = self.http1_max_buf_size {
568                server_builder.http1().max_buf_size(http1_max_buf_size);
569            }
570
571            if let Err(err) = server_builder
572                .serve_connection(
573                    TokioIo::new(stream),
574                    service_fn(move |request| {
575                        self.clone().instrumented_service_fn(
576                            server_scheme.clone(),
577                            client_addr,
578                            request,
579                        )
580                    }),
581                )
582                .await
583            {
584                tracing::warn!("Error serving HTTP connection: {err:?}");
585            }
586        });
587    }
588
589    async fn instrumented_service_fn(
590        self: Arc<Self>,
591        server_scheme: Scheme,
592        client_addr: SocketAddr,
593        request: Request<Incoming>,
594    ) -> anyhow::Result<Response<HyperOutgoingBody>> {
595        let span = http_span!(request, client_addr);
596        let method = request.method().to_string();
597        async {
598            let result = self
599                .handle(
600                    request.map(|body: Incoming| body.map_err(WasiHttpError::from).boxed_unsync()),
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 RequestData = ();
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::RequestData,
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<Weak<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(|| Arc::downgrade(server));
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 server = self
825            .server
826            .get()
827            .expect("server should have been set")
828            .upgrade()
829            .ok_or_else(|| wasmtime::format_err!("HTTP server is no longer available"))?;
830        let (instance, mut store) = server
831            .trigger_instance_builder(&self.component_id, self.self_scheme.get())
832            .to_wasmtime_result()?
833            .instantiate(())
834            .await
835            .to_wasmtime_result()?;
836        set_request_deadline(&mut store, self.reuse_config.request_deadline);
837
838        let mut store = store.into_inner();
839
840        let proxy = Proxy::P3(Service::new(&mut store, &instance).unwrap());
841
842        let request_timeout = self
843            .reuse_config
844            .request_timeout
845            .map(|range| rand::rng().random_range(range))
846            .unwrap_or(Duration::MAX);
847
848        Ok(Instance {
849            store,
850            proxy,
851            view: |data| {
852                spin_factor_outbound_http::OutboundHttpFactor::get_wasi_http_impl(
853                    data.factors_instance_state_mut(),
854                )
855                .unwrap()
856            },
857            expiration: HttpWorkerExpiration {
858                idle_timeout: rand::rng().random_range(self.reuse_config.idle_instance_timeout),
859                request_timeout,
860                sleep: tokio::time::sleep(Duration::MAX),
861            },
862            state: HttpWorkerState {
863                request_timeout,
864                max_instance_reuse_count: rand::rng()
865                    .random_range(self.reuse_config.max_instance_reuse_count),
866                max_instance_concurrent_reuse_count: rand::rng()
867                    .random_range(self.reuse_config.max_instance_concurrent_reuse_count),
868                _phantom: PhantomData,
869            },
870        })
871    }
872}
spin_trigger_http/server.rs

spin_trigger_http/
server.rs
