server.rs

use std::{
    collections::HashMap,
    io::{ErrorKind, IsTerminal},
    marker::PhantomData,
    net::SocketAddr,
    pin::Pin,
    sync::{Arc, OnceLock},
    task::{Context, Poll},
    time::{Duration, Instant},
};

use anyhow::{Context as _, bail};
use http::{
    Request, Response, StatusCode, Uri,
    uri::{Authority, Scheme},
};
use http_body_util::BodyExt;
use hyper::{
    body::{Bytes, Incoming},
    service::service_fn,
};
use hyper_util::{
    rt::{TokioExecutor, TokioIo},
    server::conn::auto::Builder,
};
use pin_project_lite::pin_project;
use rand::RngExt;
use spin_app::{APP_DESCRIPTION_KEY, APP_NAME_KEY};
use spin_factor_outbound_http::{OutboundHttpFactor, SelfRequestOrigin};
use spin_factors::RuntimeFactors;
use spin_factors_executor::InstanceState;
use spin_http::{
    app_info::AppInfo,
    body,
    config::{HttpExecutorType, HttpTriggerConfig},
    routes::{RouteInfo, RouteMatch, Router},
    trigger::HandlerType,
};
use tokio::{
    io::{AsyncRead, AsyncWrite},
    net::TcpListener,
    task,
};
use tracing::Instrument;
use wasmtime::{Store, StoreContextMut, ToWasmtimeResult, component::GuestTaskId};
use wasmtime_wasi::p2::bindings::CommandIndices;
use wasmtime_wasi_http::handler::{
    HandlerState, Instance, Proxy, ShouldAccept, ViewFn, WorkerExpiration, WorkerState,
    WorkerStatus,
};
use wasmtime_wasi_http::p2::body::HyperOutgoingBody;
use wasmtime_wasi_http::p3::bindings::Service;

use crate::{
    Body, InstanceReuseConfig, NotFoundRouteKind, OutputFormat, TlsConfig, TriggerApp,
    TriggerInstanceBuilder,
    headers::strip_forbidden_headers,
    instrument::{MatchedRoute, finalize_http_span, http_span, instrument_error},
    outbound_http::OutboundHttpInterceptor,
    spin::SpinHttpExecutor,
    wagi::WagiHttpExecutor,
    wasi::WasiHttpExecutor,
    wasip3::Wasip3HttpExecutor,
};

pub const MAX_RETRIES: u16 = 10;

pub(crate) fn set_request_deadline<T>(
    store: &mut spin_core::Store<T>,
    request_deadline: Option<Duration>,
) {
    if let Some(timeout) = request_deadline {
        store.set_deadline(Instant::now() + timeout);
    }
}

/// An HTTP server which runs Spin apps.
pub struct HttpServer<F: RuntimeFactors> {
    /// The address the server was configured to listen on (the `--listen` value).
    listen_addr: SocketAddr,
    /// The address the server is actually bound to, captured once after binding.
    ///
    /// This can differ from `listen_addr` when the OS assigns the port — e.g.
    /// `--listen 127.0.0.1:0` or `--find-free-port`. Self-request origins must use
    /// this real address rather than the configured one.
    local_addr: OnceLock<SocketAddr>,
    /// The TLS configuration for the server.
    tls_config: Option<TlsConfig>,
    /// The maximum buffer size for an HTTP1 connection.
    http1_max_buf_size: Option<usize>,
    /// Whether to find a free port if the specified port is already in use.
    find_free_port: bool,
    /// The output format for the server's startup information.
    output_format: OutputFormat,
    /// Hard Wasmtime request deadline for direct HTTP executor paths.
    request_deadline: Option<Duration>,
    /// Request router.
    router: Router,
    /// The app being triggered.
    trigger_app: Arc<TriggerApp<F>>,
    /// The application name, resolved once for use as the `app_id` telemetry attribute.
    app_id: String,
    // Component ID -> component trigger config
    component_trigger_configs: HashMap<spin_http::routes::TriggerLookupKey, HttpTriggerConfig>,
    // Component ID -> handler type
    component_handler_types: HashMap<String, HandlerType<HttpHandlerState<F>>>,
}

impl<F: RuntimeFactors> HttpServer<F> {
    /// Create a new [`HttpServer`].
    pub fn new(
        listen_addr: SocketAddr,
        tls_config: Option<TlsConfig>,
        find_free_port: bool,
        trigger_app: TriggerApp<F>,
        http1_max_buf_size: Option<usize>,
        reuse_config: InstanceReuseConfig,
        output_format: OutputFormat,
    ) -> anyhow::Result<Self> {
        // This needs to be a vec before building the router to handle duplicate routes
        let component_trigger_configs = trigger_app
            .app()
            .trigger_configs::<HttpTriggerConfig>("http")?
            .into_iter()
            .map(|(trigger_id, config)| config.lookup_key(trigger_id).map(|k| (k, config)))
            .collect::<Result<Vec<_>, _>>()?;

        // Build router
        let component_routes = component_trigger_configs
            .iter()
            .map(|(key, config)| (key, &config.route));
        let mut duplicate_routes = Vec::new();
        let router = Router::build("/", component_routes, Some(&mut duplicate_routes))?;
        if !duplicate_routes.is_empty() {
            tracing::error!(
                "The following component routes are duplicates and will never be used:"
            );
            for dup in &duplicate_routes {
                tracing::error!(
                    "  {}: {} (duplicate of {})",
                    dup.replaced_id,
                    dup.route(),
                    dup.effective_id,
                );
            }
        }
        if router.contains_reserved_route() {
            tracing::error!(
                "Routes under {} are handled by the Spin runtime and will never be reached",
                spin_http::WELL_KNOWN_PREFIX
            );
        }
        tracing::trace!(
            "Constructed router: {:?}",
            router.routes().collect::<Vec<_>>()
        );

        // Now that router is built we can merge duplicate routes by component
        let component_trigger_configs = HashMap::from_iter(component_trigger_configs);

        let trigger_app = Arc::new(trigger_app);

        let app_id = trigger_app
            .app()
            .get_metadata(APP_NAME_KEY)?
            .unwrap_or_else(|| "<unnamed>".into());

        let component_handler_types = component_trigger_configs
            .iter()
            .filter_map(|(key, trigger_config)| match key {
                spin_http::routes::TriggerLookupKey::Component(component) => Some(
                    Self::handler_type_for_component(
                        &trigger_app,
                        component,
                        &trigger_config.executor,
                        reuse_config,
                    )
                    .map(|ht| (component.clone(), ht)),
                ),
                spin_http::routes::TriggerLookupKey::Trigger(_) => None,
            })
            .collect::<anyhow::Result<_>>()?;
        Ok(Self {
            listen_addr,
            local_addr: OnceLock::new(),
            tls_config,
            find_free_port,
            router,
            trigger_app,
            app_id,
            http1_max_buf_size,
            component_trigger_configs,
            component_handler_types,
            output_format,
            request_deadline: reuse_config.request_deadline,
        })
    }

    fn handler_type_for_component(
        trigger_app: &Arc<TriggerApp<F>>,
        component_id: &str,
        executor: &Option<HttpExecutorType>,
        reuse_config: InstanceReuseConfig,
    ) -> anyhow::Result<HandlerType<HttpHandlerState<F>>> {
        let pre = trigger_app.get_instance_pre(component_id)?;
        let handler_type = match executor {
            None | Some(HttpExecutorType::Http) => HandlerType::from_instance_pre(
                pre,
                HttpHandlerState {
                    component_id: component_id.into(),
                    reuse_config,
                    server: Default::default(),
                    self_scheme: Default::default(),
                },
            )?,
            Some(HttpExecutorType::Wagi(wagi_config)) => {
                anyhow::ensure!(
                    wagi_config.entrypoint == "_start",
                    "Wagi component '{component_id}' cannot use deprecated 'entrypoint' field"
                );
                HandlerType::Wagi(
                    CommandIndices::new(pre)
                        .map_err(anyhow::Error::from)
                        .context("failed to find wasi command interface for wagi executor")?,
                )
            }
        };
        Ok(handler_type)
    }

    /// Serve incoming requests over the provided [`TcpListener`].
    pub async fn serve(self: Arc<Self>) -> anyhow::Result<()> {
        let listener: TcpListener = if self.find_free_port {
            self.search_for_free_port().await?
        } else {
            TcpListener::bind(self.listen_addr).await.map_err(|err| {
                if err.kind() == ErrorKind::AddrInUse {
                    anyhow::anyhow!("{} is already in use. To have Spin search for a free port, use the --find-free-port option.", self.listen_addr)
                } else {
                    anyhow::anyhow!("Unable to listen on {}: {err:?}", self.listen_addr)
                }
            })?
        };

        let _ = self.local_addr.set(listener.local_addr()?);

        if let Some(tls_config) = self.tls_config.clone() {
            self.serve_https(listener, tls_config).await?;
        } else {
            self.serve_http(listener).await?;
        }
        Ok(())
    }

    async fn search_for_free_port(&self) -> anyhow::Result<TcpListener> {
        let mut found_listener = None;
        let mut addr = self.listen_addr;

        for _ in 1..=MAX_RETRIES {
            if addr.port() == u16::MAX {
                anyhow::bail!(
                    "Couldn't find a free port as we've reached the maximum port number. Consider retrying with a lower base port."
                );
            }

            match TcpListener::bind(addr).await {
                Ok(listener) => {
                    found_listener = Some(listener);
                    break;
                }
                Err(err) if err.kind() == ErrorKind::AddrInUse => {
                    addr.set_port(addr.port() + 1);
                    continue;
                }
                Err(err) => anyhow::bail!("Unable to listen on {addr}: {err:?}",),
            }
        }

        found_listener.ok_or_else(|| anyhow::anyhow!(
            "Couldn't find a free port in the range {}-{}. Consider retrying with a different base port.",
            self.listen_addr.port(),
            self.listen_addr.port() + MAX_RETRIES
        ))
    }

    async fn serve_http(self: Arc<Self>, listener: TcpListener) -> anyhow::Result<()> {
        self.print_startup_msgs("http", &listener)?;
        loop {
            let (stream, client_addr) = listener.accept().await?;
            self.clone()
                .serve_connection(stream, Scheme::HTTP, client_addr);
        }
    }

    async fn serve_https(
        self: Arc<Self>,
        listener: TcpListener,
        tls_config: TlsConfig,
    ) -> anyhow::Result<()> {
        self.print_startup_msgs("https", &listener)?;
        let acceptor = tls_config.server_config()?;
        loop {
            let (stream, client_addr) = listener.accept().await?;
            match acceptor.accept(stream).await {
                Ok(stream) => self
                    .clone()
                    .serve_connection(stream, Scheme::HTTPS, client_addr),
                Err(err) => tracing::error!(?err, "Failed to start TLS session"),
            }
        }
    }

    /// Handles incoming requests using an HTTP executor.
    ///
    /// This method handles well known paths and routes requests to the handler when the router
    /// matches the requests path.
    pub async fn handle(
        self: &Arc<Self>,
        mut req: Request<Body>,
        server_scheme: Scheme,
        client_addr: SocketAddr,
    ) -> anyhow::Result<Response<Body>> {
        strip_forbidden_headers(&mut req);

        spin_telemetry::extract_trace_context(&req);

        let path = req.uri().path().to_string();

        tracing::info!("Processing request on path '{path}'");

        // Handle well-known spin paths
        if let Some(well_known) = path.strip_prefix(spin_http::WELL_KNOWN_PREFIX) {
            return match well_known {
                "health" => Ok(MatchedRoute::with_response_extension(
                    Response::new(body::full(Bytes::from_static(b"OK"))),
                    path,
                )),
                "info" => self.app_info(path),
                _ => Self::not_found(NotFoundRouteKind::WellKnown),
            };
        }

        match self.router.route(&path) {
            Ok(route_match) => {
                self.handle_trigger_route(req, route_match, server_scheme, client_addr)
                    .await
            }
            Err(_) => Self::not_found(NotFoundRouteKind::Normal(path.to_string())),
        }
    }

    /// Handles a successful route match.
    pub async fn handle_trigger_route(
        self: &Arc<Self>,
        mut req: Request<Body>,
        route_match: RouteMatch<'_, '_>,
        server_scheme: Scheme,
        client_addr: SocketAddr,
    ) -> anyhow::Result<Response<Body>> {
        set_req_uri(&mut req, server_scheme)?;
        let lookup_key = route_match.lookup_key();

        spin_telemetry::metrics::monotonic_counter!(
            spin.request_count = 1,
            trigger_type = "http",
            app_id = self.app_id.as_str(),
            component_id = lookup_key.to_string()
        );

        let trigger_config = self
            .component_trigger_configs
            .get(lookup_key)
            .with_context(|| format!("unknown routing destination '{lookup_key}'"))?;

        match (&trigger_config.component, &trigger_config.static_response) {
            (Some(component), None) => {
                self.respond_wasm_component(
                    req,
                    route_match,
                    client_addr,
                    component,
                    &trigger_config.executor,
                )
                .await
            }
            (None, Some(static_response)) => Self::respond_static_response(static_response),
            // These error cases should have been ruled out by this point but belt and braces
            (None, None) => Err(anyhow::anyhow!(
                "Triggers must specify either component or static_response - neither is specified for {}",
                route_match.raw_route()
            )),
            (Some(_), Some(_)) => Err(anyhow::anyhow!(
                "Triggers must specify either component or static_response - both are specified for {}",
                route_match.raw_route()
            )),
        }
    }

    fn get_local_addr(&self) -> SocketAddr {
        self.local_addr.get().copied().unwrap_or(self.listen_addr)
    }

    async fn respond_wasm_component(
        self: &Arc<Self>,
        req: Request<Body>,
        route_match: RouteMatch<'_, '_>,
        client_addr: SocketAddr,
        component_id: &str,
        executor: &Option<HttpExecutorType>,
    ) -> anyhow::Result<Response<Body>> {
        // Prepare HTTP executor
        let handler_type = self
            .component_handler_types
            .get(component_id)
            .with_context(|| format!("unknown component ID {component_id:?}"))?;
        let executor = executor.as_ref().unwrap_or(&HttpExecutorType::Http);

        let res = match executor {
            HttpExecutorType::Http => match handler_type {
                HandlerType::Spin => {
                    SpinHttpExecutor
                        .execute(self, &route_match, req, client_addr, component_id)
                        .await
                }
                HandlerType::Wasi0_3(handler) => {
                    Wasip3HttpExecutor(handler)
                        .execute(self, &route_match, req, client_addr)
                        .await
                }
                HandlerType::Wasi0_2(_)
                | HandlerType::Wasi2023_11_10(_)
                | HandlerType::Wasi2023_10_18(_)
                | HandlerType::Wasi2026_03_15(_) => {
                    WasiHttpExecutor { handler_type }
                        .execute(self, &route_match, req, client_addr, component_id)
                        .await
                }
                HandlerType::Wagi(_) => unreachable!(),
            },
            HttpExecutorType::Wagi(wagi_config) => {
                let indices = match handler_type {
                    HandlerType::Wagi(indices) => indices,
                    _ => unreachable!(),
                };
                let executor = WagiHttpExecutor {
                    wagi_config,
                    indices,
                };
                executor
                    .execute(self, &route_match, req, client_addr, component_id)
                    .await
            }
        };
        match res {
            Ok(res) => Ok(MatchedRoute::with_response_extension(
                res,
                route_match.raw_route(),
            )),
            Err(err) => {
                tracing::error!("Error processing request: {err:?}");
                instrument_error(&err);
                Self::internal_error(None, route_match.raw_route())
            }
        }
    }

    pub(crate) fn trigger_instance_builder(
        self: &'_ Arc<Self>,
        component_id: &str,
        self_scheme: Option<&Scheme>,
    ) -> anyhow::Result<TriggerInstanceBuilder<'_, F>> {
        let mut instance_builder = self.trigger_app.prepare(component_id)?;

        // Set up outbound HTTP request origin and service chaining
        // The outbound HTTP factor is required since both inbound and outbound wasi HTTP
        // implementations assume they use the same underlying wasmtime resource storage.
        // Eventually, we may be able to factor this out to a separate factor.
        let outbound_http = instance_builder
            .factor_builder::<OutboundHttpFactor>()
            .context(
            "The wasi HTTP trigger was configured without the required wasi outbound http support",
        )?;

        let self_scheme = self_scheme.cloned().unwrap_or(Scheme::HTTPS);
        let self_addr = self.get_local_addr();
        let origin = SelfRequestOrigin::create(self_scheme, &self_addr.to_string())?;
        outbound_http.set_self_request_origin(origin);
        outbound_http.set_request_interceptor(OutboundHttpInterceptor::new(self.clone()))?;
        Ok(instance_builder)
    }

    fn respond_static_response(
        sr: &spin_http::config::StaticResponse,
    ) -> anyhow::Result<Response<Body>> {
        let mut response = Response::builder();

        response = response.status(sr.status());
        for (header_name, header_value) in sr.headers() {
            response = response.header(header_name, header_value);
        }

        let body = match sr.body() {
            Some(b) => body::full(b.clone().into()),
            None => body::empty(),
        };

        Ok(response.body(body)?)
    }

    /// Returns spin status information.
    fn app_info(&self, route: String) -> anyhow::Result<Response<Body>> {
        let info = AppInfo::new(self.trigger_app.app());
        let body = serde_json::to_vec_pretty(&info)?;
        Ok(MatchedRoute::with_response_extension(
            Response::builder()
                .header("content-type", "application/json")
                .body(body::full(body.into()))?,
            route,
        ))
    }

    /// Creates an HTTP 500 response.
    fn internal_error(
        body: Option<&str>,
        route: impl Into<String>,
    ) -> anyhow::Result<Response<Body>> {
        let body = match body {
            Some(body) => body::full(Bytes::copy_from_slice(body.as_bytes())),
            None => body::empty(),
        };

        Ok(MatchedRoute::with_response_extension(
            Response::builder()
                .status(StatusCode::INTERNAL_SERVER_ERROR)
                .body(body)?,
            route,
        ))
    }

    /// Creates an HTTP 404 response.
    fn not_found(kind: NotFoundRouteKind) -> anyhow::Result<Response<Body>> {
        use std::sync::atomic::{AtomicBool, Ordering};
        static SHOWN_GENERIC_404_WARNING: AtomicBool = AtomicBool::new(false);
        if let NotFoundRouteKind::Normal(route) = kind
            && !SHOWN_GENERIC_404_WARNING.fetch_or(true, Ordering::Relaxed)
            && std::io::stderr().is_terminal()
        {
            terminal::warn!(
                "Request to {route} matched no pattern, and received a generic 404 response. To serve a more informative 404 page, add a catch-all (/...) route."
            );
        }
        Ok(Response::builder()
            .status(StatusCode::NOT_FOUND)
            .body(body::empty())?)
    }

    fn serve_connection<S: AsyncRead + AsyncWrite + Unpin + Send + 'static>(
        self: Arc<Self>,
        stream: S,
        server_scheme: Scheme,
        client_addr: SocketAddr,
    ) {
        task::spawn(async move {
            let mut server_builder = Builder::new(TokioExecutor::new());

            if let Some(http1_max_buf_size) = self.http1_max_buf_size {
                server_builder.http1().max_buf_size(http1_max_buf_size);
            }

            if let Err(err) = server_builder
                .serve_connection(
                    TokioIo::new(stream),
                    service_fn(move |request| {
                        self.clone().instrumented_service_fn(
                            server_scheme.clone(),
                            client_addr,
                            request,
                        )
                    }),
                )
                .await
            {
                tracing::warn!("Error serving HTTP connection: {err:?}");
            }
        });
    }

    async fn instrumented_service_fn(
        self: Arc<Self>,
        server_scheme: Scheme,
        client_addr: SocketAddr,
        request: Request<Incoming>,
    ) -> anyhow::Result<Response<HyperOutgoingBody>> {
        let span = http_span!(request, client_addr);
        let method = request.method().to_string();
        async {
            let result = self
                .handle(
                    request.map(|body: Incoming| {
                        body.map_err(wasmtime_wasi_http::p2::hyper_response_error)
                            .boxed_unsync()
                    }),
                    server_scheme,
                    client_addr,
                )
                .await;
            finalize_http_span(result, method)
        }
        .instrument(span)
        .await
    }

    fn get_description_for_route(
        &self,
        key: &spin_http::routes::TriggerLookupKey,
    ) -> anyhow::Result<Option<String>> {
        if let spin_http::routes::TriggerLookupKey::Component(component_id) = key {
            self.trigger_app
                .app()
                .get_component(component_id)
                .and_then(|c| c.get_metadata(APP_DESCRIPTION_KEY).transpose())
                .transpose()
                .map_err(Into::into)
        } else {
            Ok(None)
        }
    }

    fn print_startup_msgs(&self, scheme: &str, listener: &TcpListener) -> anyhow::Result<()> {
        let local_addr = listener.local_addr()?;
        let base_url = format!("{scheme}://{local_addr:?}");
        tracing::info!("Serving {base_url}");

        match self.output_format {
            OutputFormat::Plain => {
                terminal::step!("\nServing", "{base_url}");
                println!("Available Routes:");
                for (route, key) in self.router.routes() {
                    println!("  {key}: {base_url}{route}");
                    if let Some(description) = self.get_description_for_route(key)? {
                        println!("    {description}");
                    }
                }
            }
            OutputFormat::Json => {
                #[derive(serde::Serialize)]
                struct RoutesOutput {
                    base_url: String,
                    routes: Vec<RouteEntry>,
                }

                #[derive(serde::Serialize)]
                struct RouteEntry {
                    id: String,
                    route: String,
                    wildcard: bool,
                    #[serde(skip_serializing_if = "Option::is_none")]
                    description: Option<String>,
                }
                let mut routes = Vec::new();
                for (route, key) in self.router.routes() {
                    routes.push(RouteEntry {
                        id: key.to_string(),
                        route: route.path().to_string(),
                        wildcard: route.is_wildcard(),
                        description: self.get_description_for_route(key)?,
                    });
                }

                let output = RoutesOutput { base_url, routes };
                println!("{}", serde_json::to_string_pretty(&output)?);
            }
        }
        Ok(())
    }

    pub(crate) fn request_deadline(&self) -> Option<Duration> {
        self.request_deadline
    }
}

/// The incoming request's scheme and authority
///
/// The incoming request's URI is relative to the server, so we need to set the scheme and authority.
/// Either the `Host` header or the request's URI's authority is used as the source of truth for the authority.
/// This function will error if the authority cannot be unambiguously determined.
fn set_req_uri(req: &mut Request<Body>, scheme: Scheme) -> anyhow::Result<()> {
    let uri = req.uri().clone();
    let mut parts = uri.into_parts();
    let headers = req.headers();
    let header_authority = headers
        .get(http::header::HOST)
        .map(|h| -> anyhow::Result<Authority> {
            let host_header = h.to_str().context("'Host' header is not valid UTF-8")?;
            host_header
                .parse()
                .context("'Host' header contains an invalid authority")
        })
        .transpose()?;
    let uri_authority = parts.authority;

    // Get authority either from request URI or from 'Host' header
    let authority = match (header_authority, uri_authority) {
        (None, None) => bail!("no 'Host' header present in request"),
        (None, Some(a)) => a,
        (Some(a), None) => a,
        (Some(a1), Some(a2)) => {
            // Ensure that if `req.authority` is set, it matches what was in the `Host` header
            // https://github.com/hyperium/hyper/issues/1612
            if a1 != a2 {
                return Err(anyhow::anyhow!(
                    "authority in 'Host' header does not match authority in URI"
                ));
            }
            a1
        }
    };
    parts.scheme = Some(scheme);
    parts.authority = Some(authority);
    *req.uri_mut() = Uri::from_parts(parts).unwrap();
    Ok(())
}

pin_project! {
    pub(crate) struct HttpWorkerExpiration {
        idle_timeout: Duration,
        request_timeout: Duration,
        #[pin]
        sleep: tokio::time::Sleep,
    }
}

impl WorkerExpiration for HttpWorkerExpiration {
    fn poll(
        self: Pin<&mut Self>,
        cx: &mut Context<'_>,
        status: WorkerStatus,
        start: Instant,
    ) -> Poll<()> {
        let mut me = self.project();

        let timeout = match status {
            WorkerStatus::Idle => *me.idle_timeout,
            // TODO: add a dedicated `post_return_timeout` config setting
            // instead of reusing `request_timeout` for
            // `WorkerStatus::PostReturn` here
            WorkerStatus::Requests | WorkerStatus::PostReturn => *me.request_timeout,
        };

        if let Some(deadline) = start.checked_add(timeout) {
            let deadline = deadline.into();
            if deadline != me.sleep.deadline() {
                me.sleep.as_mut().reset(deadline);
            }
            me.sleep.poll(cx)
        } else {
            Poll::Pending
        }
    }
}

pub(crate) struct HttpWorkerState<F: RuntimeFactors> {
    request_timeout: Duration,
    max_instance_reuse_count: usize,
    max_instance_concurrent_reuse_count: usize,
    _phantom: PhantomData<F>,
}

impl<F: RuntimeFactors> WorkerState for HttpWorkerState<F> {
    type StoreData = InstanceState<F::InstanceState, ()>;
    type RequestId = ();

    fn should_accept_request(&self, concurrent_count: usize, total_count: usize) -> ShouldAccept {
        if total_count >= self.max_instance_reuse_count {
            ShouldAccept::Never
        } else if concurrent_count >= self.max_instance_concurrent_reuse_count {
            ShouldAccept::No
        } else {
            ShouldAccept::Yes
        }
    }

    fn on_request_start(
        &self,
        _: StoreContextMut<'_, Self::StoreData>,
        _: Self::RequestId,
        _: GuestTaskId,
    ) -> Pin<Box<dyn Future<Output = ()> + Send + Sync + 'static>> {
        Box::pin(tokio::time::sleep(self.request_timeout))
    }

    fn drop(&self, store: Store<Self::StoreData>, result: Result<(), wasmtime::Error>) {
        if let Err(error) = result {
            eprintln!("worker failed: {error:?}");
        }

        drop(store);
    }
}

pub(crate) struct HttpHandlerState<F: RuntimeFactors> {
    component_id: String,
    reuse_config: InstanceReuseConfig,
    server: OnceLock<Arc<HttpServer<F>>>,
    self_scheme: OnceLock<Scheme>,
}

impl<F: RuntimeFactors> HttpHandlerState<F> {
    pub(crate) fn init_once(&self, server: &Arc<HttpServer<F>>, first_uri: &Uri) {
        self.server.get_or_init(|| server.clone());
        if let Some(scheme) = first_uri.scheme() {
            self.self_scheme.get_or_init(|| scheme.clone());
        }
    }
}

impl<F: RuntimeFactors> HandlerState for HttpHandlerState<F> {
    type StoreData = InstanceState<F::InstanceState, ()>;
    type WorkerExpiration = HttpWorkerExpiration;
    type WorkerState = HttpWorkerState<F>;

    async fn instantiate(
        &self,
    ) -> wasmtime::Result<Instance<Self::StoreData, Self::WorkerExpiration, Self::WorkerState>>
    {
        let (instance, mut store) = self
            .server
            .get()
            .expect("server should have been set")
            .trigger_instance_builder(&self.component_id, self.self_scheme.get())
            .to_wasmtime_result()?
            .instantiate(())
            .await
            .to_wasmtime_result()?;
        set_request_deadline(&mut store, self.reuse_config.request_deadline);

        let mut store = store.into_inner();

        let proxy = Proxy::P3(Service::new(&mut store, &instance).unwrap());

        let request_timeout = self
            .reuse_config
            .request_timeout
            .map(|range| rand::rng().random_range(range))
            .unwrap_or(Duration::MAX);

        Ok(Instance {
            store,
            proxy,
            view: ViewFn::P3(|data| {
                spin_factor_outbound_http::OutboundHttpFactor::get_wasi_p3_http_impl(
                    data.factors_instance_state_mut(),
                )
                .unwrap()
            }),
            expiration: HttpWorkerExpiration {
                idle_timeout: rand::rng().random_range(self.reuse_config.idle_instance_timeout),
                request_timeout,
                sleep: tokio::time::sleep(Duration::MAX),
            },
            state: HttpWorkerState {
                request_timeout,
                max_instance_reuse_count: rand::rng()
                    .random_range(self.reuse_config.max_instance_reuse_count),
                max_instance_concurrent_reuse_count: rand::rng()
                    .random_range(self.reuse_config.max_instance_concurrent_reuse_count),
                _phantom: PhantomData,
            },
        })
    }
}