diff --git a/.github/workflows/fossa.yaml b/.github/workflows/fossa.yaml deleted file mode 100644 index 0da5dc0..0000000 --- a/.github/workflows/fossa.yaml +++ /dev/null @@ -1,29 +0,0 @@ -name: FOSSA - -on: - push: - pull_request: - workflow_dispatch: - merge_group: - types: [checks_requested] - -jobs: - fossa: - runs-on: ubuntu-latest - - steps: - - name: Checkout repository - uses: actions/checkout@v7 - - - name: Install FOSSA CLI - run: | - curl -H 'Cache-Control: no-cache' https://raw.githubusercontent.com/fossas/fossa-cli/master/install-latest.sh | bash - - - name: Set FOSSA API Key - run: echo "FOSSA_API_KEY=${{ secrets.FOSSA_PUB_API_KEY }}" >> $GITHUB_ENV - - - name: Run FOSSA Analysis - run: fossa analyze - - - name: Run FOSSA Test - run: fossa test \ No newline at end of file diff --git a/Cargo.lock b/Cargo.lock index a13573d..66e90bf 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -341,7 +341,7 @@ dependencies = [ "cap-primitives", "cap-std", "io-lifetimes", - "windows-sys 0.59.0", + "windows-sys 0.52.0", ] [[package]] @@ -370,7 +370,7 @@ dependencies = [ "maybe-owned", "rustix 1.1.4", "rustix-linux-procfs", - "windows-sys 0.59.0", + "windows-sys 0.52.0", "winx", ] @@ -516,7 +516,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "117725a109d387c937a1533ce01b450cbde6b88abceea8473c4d7a85853cda3c" dependencies = [ "lazy_static", - "windows-sys 0.59.0", + "windows-sys 0.52.0", ] [[package]] @@ -910,7 +910,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "39cab71617ae0d63f51a36d69f866391735b51691dbda63cf6f96d042b63efeb" dependencies = [ "libc", - "windows-sys 0.61.2", + "windows-sys 0.52.0", ] [[package]] @@ -992,7 +992,7 @@ checksum = "0ce92ff622d6dadf7349484f42c93271a0d49b7cc4d466a936405bacbe10aa78" dependencies = [ "cfg-if", "rustix 1.1.4", - "windows-sys 0.59.0", + "windows-sys 0.52.0", ] [[package]] @@ -1051,7 +1051,7 @@ checksum = "94e7099f6313ecacbe1256e8ff9d617b75d1bcb16a6fddef94866d225a01a14a" dependencies = [ "io-lifetimes", "rustix 1.1.4", - "windows-sys 0.59.0", + "windows-sys 0.52.0", ] [[package]] @@ -1847,7 +1847,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "2285ddfe3054097ef4b2fe909ef8c3bcd1ea52a8f0d274416caebeef39f04a65" dependencies = [ "io-lifetimes", - "windows-sys 0.59.0", + "windows-sys 0.52.0", ] [[package]] @@ -1870,7 +1870,7 @@ checksum = "3640c1c38b8e4e43584d8df18be5fc6b0aa314ce6ebf51b53313d4306cca8e46" dependencies = [ "hermit-abi", "libc", - "windows-sys 0.61.2", + "windows-sys 0.52.0", ] [[package]] @@ -2951,7 +2951,7 @@ dependencies = [ "errno", "libc", "linux-raw-sys 0.4.15", - "windows-sys 0.59.0", + "windows-sys 0.52.0", ] [[package]] @@ -2964,7 +2964,7 @@ dependencies = [ "errno", "libc", "linux-raw-sys 0.12.1", - "windows-sys 0.61.2", + "windows-sys 0.52.0", ] [[package]] @@ -3361,7 +3361,7 @@ dependencies = [ "fd-lock", "io-lifetimes", "rustix 0.38.44", - "windows-sys 0.59.0", + "windows-sys 0.52.0", "winx", ] @@ -3384,10 +3384,10 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "32497e9a4c7b38532efcdebeef879707aa9f794296a4f0244f6f69e9bc8574bd" dependencies = [ "fastrand", - "getrandom 0.4.3", + "getrandom 0.3.4", "once_cell", "rustix 1.1.4", - "windows-sys 0.61.2", + "windows-sys 0.52.0", ] [[package]] @@ -4340,6 +4340,7 @@ dependencies = [ [[package]] name = "wasmtime-wasi-http" version = "36.0.12" +source = "git+https://github.com/G-Core/wasmtime.git?branch=release-36.0.0#62166f65ac52a1f164419097a2a533ff50e4d07b" dependencies = [ "anyhow", "async-trait", @@ -4355,7 +4356,7 @@ dependencies = [ "tracing", "wasmtime", "wasmtime-wasi", - "wasmtime-wasi-io 36.0.12", + "wasmtime-wasi-io 36.0.12 (git+https://github.com/G-Core/wasmtime.git?branch=release-36.0.0)", "webpki-roots 0.26.11", ] @@ -4543,7 +4544,7 @@ version = "0.1.11" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "c2a7b1c03c876122aa43f3020e6c3c3ee5c05081c9a00739faf7503aeba10d22" dependencies = [ - "windows-sys 0.61.2", + "windows-sys 0.52.0", ] [[package]] @@ -4681,15 +4682,6 @@ dependencies = [ "windows-targets 0.52.6", ] -[[package]] -name = "windows-sys" -version = "0.59.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "1e38bc4d79ed67fd075bcc251a1c39b32a1776bbe92e5bef1f0bf1f8c531853b" -dependencies = [ - "windows-targets 0.52.6", -] - [[package]] name = "windows-sys" version = "0.60.2" @@ -4862,7 +4854,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "3f3fd376f71958b862e7afb20cfe5a22830e1963462f3a17f49d82a6c1d1f42d" dependencies = [ "bitflags 2.13.1", - "windows-sys 0.59.0", + "windows-sys 0.52.0", ] [[package]] diff --git a/crates/http-service/src/executor/http.rs b/crates/http-service/src/executor/http.rs index 8530934..880b043 100644 --- a/crates/http-service/src/executor/http.rs +++ b/crates/http-service/src/executor/http.rs @@ -126,7 +126,18 @@ where let mut store = store_builder.build(state)?; - let instance = self.instance_pre.instantiate_async(&mut store).await?; + let instance = match self.instance_pre.instantiate_async(&mut store).await { + Ok(instance) => instance, + Err(error) => { + // A denied memory growth during instantiation (e.g. the module's + // declared minimum memory exceeds `mem_limit`) is recorded by the + // limiter; classify it as out-of-memory instead of a generic error. + if store.is_oom() { + return Err(runtime::store::OutOfMemory(error).into()); + } + return Err(error); + } + }; let http_handler = instance.get_export_index(&mut store, None, "gcore:fastedge/http-handler"); let process = instance @@ -396,34 +407,36 @@ mod tests { name: SmolStr, cfg: &App, engine: &WasmEngine>, - ) -> anyhow::Result { - let mut dictionary = Dictionary::new(); - for (k, v) in cfg.env.iter() { - dictionary.insert(k.to_string(), v.to_string()); + ) -> impl std::future::Future> + Send { + async move { + let mut dictionary = Dictionary::new(); + for (k, v) in cfg.env.iter() { + dictionary.insert(k.to_string(), v.to_string()); + } + let env = cfg.env.iter().collect::>(); + + let logger = self.make_logger(name.clone(), cfg); + + let version = WasiVersion::Preview2; + let store_builder = engine + .store_builder(version) + .set_env(&env) + .max_memory_size(cfg.mem_limit) + .max_epoch_ticks(cfg.max_duration) + .dictionary(dictionary) + .logger(logger); + + let component = self.loader().load_component(cfg.binary_id)?; + let instance_pre = engine.component_instantiate_pre(&component)?; + tracing::debug!("Added '{}' to cache", name); + Ok(HttpExecutorImpl::new( + instance_pre, + store_builder, + self.backend(), + false, + cfg.app_id, + )) } - let env = cfg.env.iter().collect::>(); - - let logger = self.make_logger(name.clone(), cfg); - - let version = WasiVersion::Preview2; - let store_builder = engine - .store_builder(version) - .set_env(&env) - .max_memory_size(cfg.mem_limit) - .max_epoch_ticks(cfg.max_duration) - .dictionary(dictionary) - .logger(logger); - - let component = self.loader().load_component(cfg.binary_id)?; - let instance_pre = engine.component_instantiate_pre(&component)?; - tracing::debug!("Added '{}' to cache", name); - Ok(HttpExecutorImpl::new( - instance_pre, - store_builder, - self.backend(), - false, - cfg.app_id, - )) } } diff --git a/crates/http-service/src/executor/mod.rs b/crates/http-service/src/executor/mod.rs index 1d446fb..1c91073 100644 --- a/crates/http-service/src/executor/mod.rs +++ b/crates/http-service/src/executor/mod.rs @@ -36,12 +36,16 @@ pub trait HttpExecutor { pub trait ExecutorFactory { type Executor; + /// Get (or build and cache) the executor for an app. Async so that a cache + /// miss can offload the blocking load + instantiate work to the blocking + /// pool instead of stalling the calling worker (or migrating its run + /// queue, as `block_in_place` does). fn get_executor( &self, name: SmolStr, app: &App, engine: &WasmEngine, - ) -> Result; + ) -> impl std::future::Future> + Send; } pub(crate) fn get_properties(headers: &HeaderMap) -> HashMap { diff --git a/crates/http-service/src/executor/wasi_http.rs b/crates/http-service/src/executor/wasi_http.rs index 0d534ff..90a083c 100644 --- a/crates/http-service/src/executor/wasi_http.rs +++ b/crates/http-service/src/executor/wasi_http.rs @@ -148,21 +148,45 @@ where .context("new response outparam")?; let proxy_pre = ProxyPre::new(instance_pre)?; - let proxy = proxy_pre.instantiate_async(&mut store).await?; + let proxy = match proxy_pre.instantiate_async(&mut store).await { + Ok(proxy) => proxy, + Err(error) => { + // A denied memory growth during instantiation (e.g. the module's + // declared minimum memory exceeds `mem_limit`) is recorded by the + // limiter; classify it as out-of-memory instead of a generic error. + if store.is_oom() { + return Err(runtime::store::OutOfMemory(error).into()); + } + return Err(error); + } + }; let task_stats = stats.clone(); let task = tokio::task::spawn( async move { let duration = Duration::from_millis(store.data().timeout); - if let Err(e) = tokio::time::timeout( + let exec_result = match tokio::time::timeout( duration, proxy .wasi_http_incoming_handler() .call_handle(&mut store, req, out), ) - .await? + .await { + Ok(inner) => inner, + // tokio timeout elapsed (outer deadline hit). + Err(elapsed) => Err(elapsed.into()), + }; + if let Err(e) = exec_result { tracing::warn!(cause=?e, "incoming handler"); + // Record the failure reason on the shared stats. The response + // headers may already have been flushed (the guest called + // `response-outparam::set` before trapping mid-body), in which + // case `receiver.await` already returned `Ok` and the request + // would otherwise be accounted in stats as a successful `200`. + // Setting `fail_reason` makes the stats row reflect the actual + // failure. See `crate::fail_reason_of`. + task_stats.fail_reason(crate::fail_reason_of(&e) as i32); // log to application logger error if let Some(ref logger) = store.data().logger { logger.write_msg(format!("Execution error: {}", e)).await; diff --git a/crates/http-service/src/lib.rs b/crates/http-service/src/lib.rs index 46eca34..db64351 100644 --- a/crates/http-service/src/lib.rs +++ b/crates/http-service/src/lib.rs @@ -285,199 +285,247 @@ where Ok(app_name) => app_name, }; - let span = tracing::info_span!("http", app = %app_name, traceparent = %traceparent); - let _enter = span.enter(); - - // lookup for application config and binary_id - tracing::debug!("Processing request URL: {}", request.uri()); - let lookup = match app_name { - AppName::Id(id) => self.context.lookup_by_id(id).instrument(span.clone()).await, - AppName::Name(name) => self + let span = tracing::error_span!("http", app = %app_name, traceparent = %traceparent); + + // Instrument the whole request body with the span. Using `span.enter()` + // here would be unsound: the returned guard is thread-local and does not + // re-enter the span after `.await` points, so synchronous log events in + // this body (e.g. the `execute` warning) could lose the span context when + // the task resumes on a different worker thread. Wrapping the body in an + // instrumented future re-enters the span on every poll. + async move { + // lookup for application config and binary_id + tracing::debug!("Processing request URL: {}", request.uri()); + let lookup = match app_name { + AppName::Id(id) => self.context.lookup_by_id(id).await, + AppName::Name(name) => self + .context + .lookup_by_name(&name) + .await + .map(|cfg| (name, cfg)), + }; + + let (app_name, cfg) = match lookup { + None => { + #[cfg(feature = "metrics")] + metrics::metrics(AppResult::UNKNOWN, HTTP_LABEL, None, None); + tracing::info!("Request for unknown application on URL: {}", request.uri()); + return not_found(); + } + Some((app_name, cfg)) + if cfg.status == Status::Draft || cfg.status == Status::Disabled => + { + tracing::info!( + "Request for disabled application '{}' on URL: {}", + app_name, + request.uri() + ); + return not_found(); + } + Some((app_name, cfg)) if cfg.status == Status::RateLimited => { + tracing::info!( + "Request for rate limited application '{}' on URL: {}", + app_name, + request.uri() + ); + return too_many_requests(); + } + Some((app_name, cfg)) if cfg.status == Status::Suspended => { + tracing::info!( + "Request for suspended application '{}' on URL: {}", + app_name, + request.uri() + ); + return not_acceptable(); + } + + Some((app_name, cfg)) => (app_name, cfg), + }; + + // get cached execute context for this application + let executor = match self .context - .lookup_by_name(&name) - .instrument(span.clone()) + .get_executor(app_name.clone(), &cfg, &self.engine) .await - .map(|cfg| (name, cfg)), - }; - - let (app_name, cfg) = match lookup { - None => { - #[cfg(feature = "metrics")] - metrics::metrics(AppResult::UNKNOWN, HTTP_LABEL, None, None); - tracing::info!("Request for unknown application on URL: {}", request.uri()); - return not_found(); - } - Some((app_name, cfg)) - if cfg.status == Status::Draft || cfg.status == Status::Disabled => { - tracing::info!( - "Request for disabled application '{}' on URL: {}", - app_name, - request.uri() - ); - return not_found(); - } - Some((app_name, cfg)) if cfg.status == Status::RateLimited => { - tracing::info!( - "Request for rate limited application '{}' on URL: {}", - app_name, - request.uri() - ); - return too_many_requests(); - } - Some((app_name, cfg)) if cfg.status == Status::Suspended => { - tracing::info!( - "Request for suspended application '{}' on URL: {}", - app_name, - request.uri() - ); - return not_acceptable(); - } - - Some((app_name, cfg)) => (app_name, cfg), - }; + Ok(executor) => executor, + Err(error) => { + #[cfg(feature = "metrics")] + metrics::metrics(AppResult::UNKNOWN, HTTP_LABEL, None, None); + tracing::warn!(cause=?error, app=%app_name, + "failure on getting context" + ); + return internal_fastedge_error("context error", INTERNAL_STATUS_CONTEXT_ERROR); + } + }; - // get cached execute context for this application - let executor = match self - .context - .get_executor(app_name.clone(), &cfg, &self.engine) - { - Ok(executor) => executor, - Err(error) => { - #[cfg(feature = "metrics")] - metrics::metrics(AppResult::UNKNOWN, HTTP_LABEL, None, None); - tracing::warn!(cause=?error, app=%app_name, - "failure on getting context" - ); - return internal_fastedge_error("context error", INTERNAL_STATUS_CONTEXT_ERROR); - } - }; + let stats = self.context.new_stats_row(&traceparent, &app_name, &cfg); - let stats = self.context.new_stats_row(&traceparent, &app_name, &cfg); + let response = match executor.execute(request, stats.clone()).await { + Ok(mut response) => { + #[cfg(feature = "metrics")] + metrics::metrics( + AppResult::SUCCESS, + &["http"], + Some(stats.get_time_elapsed()), + Some(stats.get_memory_used()), + ); - let response = match executor - .execute(request, stats.clone()) - .instrument(span.clone()) - .await - { - Ok(mut response) => { - #[cfg(feature = "metrics")] - metrics::metrics( - AppResult::SUCCESS, - &["http"], - Some(stats.get_time_elapsed()), - Some(stats.get_memory_used()), - ); - - response.headers_mut().extend(app_res_headers(cfg)); - response - } - Err(error) => { - tracing::warn!(cause=?error, "execute"); - let (status_code, fail_reason, msg, internal_code) = map_err(error); - stats.status_code(status_code); - stats.fail_reason(fail_reason as i32); - tracing::debug!(?fail_reason, ?traceparent, "stats"); + response.headers_mut().extend(app_res_headers(cfg)); + response + } + Err(error) => { + tracing::warn!(cause=?error, "execute"); + let (status_code, fail_reason, msg, internal_code) = map_err(error); + stats.status_code(status_code); + stats.fail_reason(fail_reason as i32); + tracing::debug!(?fail_reason, ?traceparent, "stats"); + + #[cfg(feature = "metrics")] + metrics::metrics( + fail_reason, + HTTP_LABEL, + Some(stats.get_time_elapsed()), + None, + ); + + let builder = hyper::Response::builder() + .status(status_code) + .header(X_CDN_INTERNAL_STATUS, internal_code); + let res_headers = app_res_headers(cfg); + let builder = res_headers + .iter() + .fold(builder, |builder, (k, v)| builder.header(k, v)); + + builder.body(msg)? + } + }; + Ok(response) + } + .instrument(span) + .await + } +} - #[cfg(feature = "metrics")] - metrics::metrics( - fail_reason, - HTTP_LABEL, - Some(stats.get_time_elapsed()), - None, - ); - - let builder = hyper::Response::builder() - .status(status_code) - .header(X_CDN_INTERNAL_STATUS, internal_code); - let res_headers = app_res_headers(cfg); - let builder = res_headers - .iter() - .fold(builder, |builder, (k, v)| builder.header(k, v)); - - builder.body(msg)? - } - }; - Ok(response) +/// Classify an execution error into an [`AppResult`] fail reason. +/// +/// Kept consistent with the fail-reason mapping in [`map_err`]. Used to record +/// failures that surface *after* the response headers were already sent (e.g. an +/// epoch-interrupt timeout during body streaming). Without this, such a request +/// is accounted in stats as a successful `200`, because `execute` returns `Ok` +/// as soon as the guest sets the response, before the trap occurs. +pub(crate) fn fail_reason_of(error: &Error) -> AppResult { + let root_cause = error.root_cause(); + if error.chain().any(|e| e.is::()) { + AppResult::OOM + } else if let Some(exit) = root_cause.downcast_ref::() { + if exit.0 == 0 { + AppResult::SUCCESS + } else { + AppResult::OTHER + } + } else if let Some(trap) = root_cause.downcast_ref::() { + match trap { + wasmtime::Trap::Interrupt => AppResult::TIMEOUT, + wasmtime::Trap::UnreachableCodeReached => AppResult::OOM, + _ => AppResult::OTHER, + } + } else if root_cause.downcast_ref::().is_some() + || root_cause.to_string().ends_with("deadline has elapsed") + { + AppResult::TIMEOUT + } else { + AppResult::OTHER } } fn map_err(error: Error) -> (u16, AppResult, HyperOutgoingBody, u16) { let root_cause = error.root_cause(); - let (status_code, fail_reason, msg, internal_code) = - if let Some(exit) = root_cause.downcast_ref::() { - if exit.0 == 0 { - ( - StatusCode::OK.as_u16(), - AppResult::SUCCESS, - Empty::new().map_err(|never| match never {}).boxed(), - 0, - ) - } else { - ( - FASTEDGE_EXECUTION_PANIC, - AppResult::OTHER, - Full::new(Bytes::from("fastedge: App failed")) - .map_err(|never| match never {}) - .boxed(), - INTERNAL_STATUS_APP_EXIT_ERROR, - ) - } - } else if let Some(trap) = root_cause.downcast_ref::() { - match trap { - wasmtime::Trap::Interrupt => ( - FASTEDGE_EXECUTION_TIMEOUT, - AppResult::TIMEOUT, - Full::new(Bytes::from("fastedge: Execution timeout")) - .map_err(|never| match never {}) - .boxed(), - INTERNAL_STATUS_TIMEOUT_INTERRUPT, - ), - wasmtime::Trap::UnreachableCodeReached => ( - FASTEDGE_OUT_OF_MEMORY, - AppResult::OOM, - Full::new(Bytes::from("fastedge: Out of memory")) - .map_err(|never| match never {}) - .boxed(), - INTERNAL_STATUS_OUT_OF_MEMORY, - ), - _ => ( - FASTEDGE_EXECUTION_PANIC, - AppResult::OTHER, - Full::new(Bytes::from("fastedge: App failed")) - .map_err(|never| match never {}) - .boxed(), - INTERNAL_STATUS_WASM_TRAP_OTHER, - ), - } - } else if let Some(_elapsed) = root_cause.downcast_ref::() { + // `OutOfMemory` wraps the underlying wasmtime error as its source, so it + // sits above `root_cause`; scan the whole chain for it. + let is_oom = error.chain().any(|e| e.is::()); + let (status_code, fail_reason, msg, internal_code) = if is_oom { + ( + FASTEDGE_OUT_OF_MEMORY, + AppResult::OOM, + Full::new(Bytes::from("fastedge: Out of memory")) + .map_err(|never| match never {}) + .boxed(), + INTERNAL_STATUS_OUT_OF_MEMORY, + ) + } else if let Some(exit) = root_cause.downcast_ref::() { + if exit.0 == 0 { ( - FASTEDGE_EXECUTION_TIMEOUT, - AppResult::TIMEOUT, - Full::new(Bytes::from("fastedge: Execution timeout")) + StatusCode::OK.as_u16(), + AppResult::SUCCESS, + Empty::new().map_err(|never| match never {}).boxed(), + 0, + ) + } else { + ( + FASTEDGE_EXECUTION_PANIC, + AppResult::OTHER, + Full::new(Bytes::from("fastedge: App failed")) .map_err(|never| match never {}) .boxed(), - INTERNAL_STATUS_TIMEOUT_ELAPSED, + INTERNAL_STATUS_APP_EXIT_ERROR, ) - } else if root_cause.to_string().ends_with("deadline has elapsed") { - ( + } + } else if let Some(trap) = root_cause.downcast_ref::() { + match trap { + wasmtime::Trap::Interrupt => ( FASTEDGE_EXECUTION_TIMEOUT, AppResult::TIMEOUT, Full::new(Bytes::from("fastedge: Execution timeout")) .map_err(|never| match never {}) .boxed(), - INTERNAL_STATUS_TIMEOUT_DEADLINE, - ) - } else { - ( - FASTEDGE_INTERNAL_ERROR, + INTERNAL_STATUS_TIMEOUT_INTERRUPT, + ), + wasmtime::Trap::UnreachableCodeReached => ( + FASTEDGE_OUT_OF_MEMORY, + AppResult::OOM, + Full::new(Bytes::from("fastedge: Out of memory")) + .map_err(|never| match never {}) + .boxed(), + INTERNAL_STATUS_OUT_OF_MEMORY, + ), + _ => ( + FASTEDGE_EXECUTION_PANIC, AppResult::OTHER, - Full::new(Bytes::from("fastedge: Execute error")) + Full::new(Bytes::from("fastedge: App failed")) .map_err(|never| match never {}) .boxed(), - INTERNAL_STATUS_EXECUTE_ERROR, - ) - }; + INTERNAL_STATUS_WASM_TRAP_OTHER, + ), + } + } else if let Some(_elapsed) = root_cause.downcast_ref::() { + ( + FASTEDGE_EXECUTION_TIMEOUT, + AppResult::TIMEOUT, + Full::new(Bytes::from("fastedge: Execution timeout")) + .map_err(|never| match never {}) + .boxed(), + INTERNAL_STATUS_TIMEOUT_ELAPSED, + ) + } else if root_cause.to_string().ends_with("deadline has elapsed") { + ( + FASTEDGE_EXECUTION_TIMEOUT, + AppResult::TIMEOUT, + Full::new(Bytes::from("fastedge: Execution timeout")) + .map_err(|never| match never {}) + .boxed(), + INTERNAL_STATUS_TIMEOUT_DEADLINE, + ) + } else { + ( + FASTEDGE_INTERNAL_ERROR, + AppResult::OTHER, + Full::new(Bytes::from("fastedge: Execute error")) + .map_err(|never| match never {}) + .boxed(), + INTERNAL_STATUS_EXECUTE_ERROR, + ) + }; (status_code, fail_reason, msg, internal_code) } diff --git a/crates/key-value-store/src/redis_impl.rs b/crates/key-value-store/src/redis_impl.rs index b88ca7a..e564100 100644 --- a/crates/key-value-store/src/redis_impl.rs +++ b/crates/key-value-store/src/redis_impl.rs @@ -2,6 +2,8 @@ use crate::Store; use reactor::gcore::fastedge::key_value::{Error, Value}; use redis::aio::{ConnectionManager, ConnectionManagerConfig}; use redis::{AsyncCommands, AsyncIter}; +use std::sync::Arc; +use std::sync::atomic::{AtomicUsize, Ordering}; use std::time::Duration; /// Fail-fast timeouts for the KV-store Redis connection. Redis sits on the @@ -25,34 +27,58 @@ fn connection_manager_config() -> ConnectionManagerConfig { #[derive(Clone)] pub struct RedisStore { - inner: ConnectionManager, + /// Pool of multiplexed connections. Each `ConnectionManager` owns its own + /// socket and background driver task, so spreading commands across the pool + /// round-robin keeps a burst from serializing behind a single connection + /// (which would push tail latency past the response timeout). Wrapped in + /// `Arc` so cloning a `RedisStore` shares the same pool and cursor. + conns: Arc>, + next: Arc, } impl RedisStore { - /// Open a store backed by `ConnectionManager`, which holds a multiplexed - /// connection and transparently reconnects with exponential backoff when - /// the underlying socket dies (e.g. broken pipe on Redis restart). The - /// command that hits the dead socket still surfaces as an error, but - /// follow-up calls land on the freshly re-established connection. - pub async fn open(params: &str) -> Result { + /// Open a store backed by a pool of `ConnectionManager`s. Each connection + /// holds a multiplexed connection and transparently reconnects with + /// exponential backoff when the underlying socket dies (e.g. broken pipe on + /// Redis restart). The command that hits the dead socket still surfaces as + /// an error, but follow-up calls land on a freshly re-established + /// connection. `pool_size` is clamped to at least 1. + pub async fn open(params: &str, pool_size: usize) -> Result { + let pool_size = pool_size.max(1); let client = ::redis::Client::open(params).map_err(|error| { tracing::warn!(error = ?error, "kv-store: redis open"); Error::InternalError })?; - let conn = ConnectionManager::new_with_config(client, connection_manager_config()) - .await - .map_err(|error| { - tracing::warn!(error = ?error, "kv-store: redis open"); - Error::InternalError - })?; - Ok(Self { inner: conn }) + let mut conns = Vec::with_capacity(pool_size); + for _ in 0..pool_size { + let conn = + ConnectionManager::new_with_config(client.clone(), connection_manager_config()) + .await + .map_err(|error| { + tracing::warn!(error = ?error, "kv-store: redis open"); + Error::InternalError + })?; + conns.push(conn); + } + Ok(Self { + conns: Arc::new(conns), + next: Arc::new(AtomicUsize::new(0)), + }) + } + + /// Pick the next connection from the pool (round-robin). Clones are cheap: + /// `ConnectionManager` is internally reference-counted and shares its + /// socket, so this just hands back another handle to a pooled connection. + fn conn(&self) -> ConnectionManager { + let idx = self.next.fetch_add(1, Ordering::Relaxed) % self.conns.len(); + self.conns[idx].clone() } } #[async_trait::async_trait] impl Store for RedisStore { async fn get(&self, key: &str) -> Result, Error> { - self.inner.clone().get(key).await.map_err(|error| { + self.conn().get(key).await.map_err(|error| { tracing::warn!(cause = ?error, key, "kv-store: redis get"); Error::InternalError }) @@ -64,8 +90,7 @@ impl Store for RedisStore { min: f64, max: f64, ) -> Result, Error> { - self.inner - .clone() + self.conn() .zrangebyscore_withscores(key, min, max) .await .map_err(|error| { @@ -75,7 +100,7 @@ impl Store for RedisStore { } async fn scan(&self, pattern: &str) -> Result, Error> { - let mut conn = self.inner.clone(); + let mut conn = self.conn(); let mut it = conn.scan_match(pattern).await.map_err(|error| { tracing::warn!(cause = ?error, pattern, "kv-store: redis scan_match"); Error::InternalError @@ -91,7 +116,7 @@ impl Store for RedisStore { } async fn zscan(&self, key: &str, pattern: &str) -> Result, Error> { - let mut conn = self.inner.clone(); + let mut conn = self.conn(); let mut it: AsyncIter<(Value, f64)> = conn.zscan_match(key, pattern).await.map_err(|error| { tracing::warn!(cause = ?error, key, pattern, "kv-store: redis zscan_match"); @@ -111,7 +136,7 @@ impl Store for RedisStore { redis::cmd("BF.EXISTS") .arg(key) .arg(item) - .query_async(&mut self.inner.clone()) + .query_async(&mut self.conn()) .await .map_err(|error| { tracing::warn!(cause = ?error, key, item, "kv-store: redis bf_exists"); diff --git a/crates/runtime/src/app.rs b/crates/runtime/src/app.rs index b982044..ee5acde 100644 --- a/crates/runtime/src/app.rs +++ b/crates/runtime/src/app.rs @@ -79,6 +79,7 @@ pub struct SecretOption { pub enum Log { #[default] None, + Kafka, Victoria, } diff --git a/crates/runtime/src/instances.rs b/crates/runtime/src/instances.rs new file mode 100644 index 0000000..9dd1e89 --- /dev/null +++ b/crates/runtime/src/instances.rs @@ -0,0 +1,142 @@ +//! Accounting for concurrently live WASM instances. +//! +//! Every request builds a [`crate::store::Store`], instantiates a module into it, and drops +//! both when the request finishes. For that whole span the instance holds one slot in each +//! of the wasmtime pooling allocator's pools — linear memory, async stack, table, core +//! instance — all sized by `max_execution_stacks` in `wasm-config`. +//! +//! Exhausting those pools is *not* graceful: the allocator returns "maximum concurrent limit +//! of N for ... reached", which surfaces as an instantiation error rather than the adaptive +//! run-queue shedding path. Sizing the pool therefore needs the observed concurrency, which +//! is what this module exports. +//! +//! The guard lives in [`crate::Data`], the store's data type, so every executor that goes +//! through `StoreBuilder::build` — ProxyWasm, `http-handler`, `wasi:http` — is counted +//! against the same pool it actually draws from. There is deliberately no `executor` label: +//! the pools are per-`Engine` and shared, so only the total is meaningful for sizing. + +#[cfg(feature = "metrics")] +mod imp { + use lazy_static::lazy_static; + use prometheus::{IntGauge, register_int_gauge}; + use std::sync::atomic::{AtomicI64, Ordering}; + + lazy_static! { + static ref WASM_INSTANCES_LIVE: IntGauge = register_int_gauge!( + "fastedge_wasm_instances_live", + "WASM instances currently alive, each holding one slot in every wasmtime pooling-allocator pool" + ) + .unwrap(); + + /// Peak since the previous scrape. A plain gauge is sampled at the scrape + /// interval and will miss the sub-second concurrency spikes that a stalled backend + /// produces — exactly the peaks the pool has to be sized for. This one cannot. + static ref WASM_INSTANCES_PEAK: IntGauge = register_int_gauge!( + "fastedge_wasm_instances_peak", + "Highest number of concurrently live WASM instances observed since the \ + previous scrape (resets on scrape)" + ) + .unwrap(); + } + + /// Running max since the previous scrape, kept separately so the peak can be updated + /// with an atomic `fetch_max` — `IntGauge` only offers `set`, which cannot express + /// "raise to" — and drained with `swap(0)` by [`flush_peak`] on scrape. + static PEAK: AtomicI64 = AtomicI64::new(0); + + pub(super) fn acquire() { + WASM_INSTANCES_LIVE.inc(); + // Reading back after `inc` may observe another thread's concurrent increment. That + // is still a level that genuinely occurred, so it is a valid sample for the peak. + let live = WASM_INSTANCES_LIVE.get(); + PEAK.fetch_max(live, Ordering::Relaxed); + } + + pub(super) fn release() { + WASM_INSTANCES_LIVE.dec(); + } + + /// Currently live instances. Test/diagnostic accessor. + pub fn live() -> i64 { + WASM_INSTANCES_LIVE.get() + } + + /// Peak since the previous [`flush_peak`]. Test/diagnostic accessor. + pub fn peak() -> i64 { + PEAK.load(Ordering::Relaxed) + } + + /// Export and reset the peak gauge; call on each Prometheus scrape. Never reports + /// less than the currently live count, so a long-running steady load can't read as 0. + pub fn flush_peak() { + let peak = PEAK + .swap(0, Ordering::Relaxed) + .max(WASM_INSTANCES_LIVE.get()); + WASM_INSTANCES_PEAK.set(peak); + } +} + +#[cfg(not(feature = "metrics"))] +mod imp { + pub(super) fn acquire() {} + pub(super) fn release() {} + pub fn flush_peak() {} +} + +pub use imp::flush_peak; +#[cfg(feature = "metrics")] +pub use imp::{live, peak}; + +/// RAII counter for one live WASM instance. +/// +/// Held by [`crate::Data`] so the count spans exactly the lifetime of the store that owns +/// the pooled slots, including the unwind path when a guest traps or the request is +/// cancelled mid-flight. +#[derive(Debug)] +pub struct LiveInstanceGuard; + +impl LiveInstanceGuard { + pub fn new() -> Self { + imp::acquire(); + LiveInstanceGuard + } +} + +impl Default for LiveInstanceGuard { + fn default() -> Self { + Self::new() + } +} + +impl Drop for LiveInstanceGuard { + fn drop(&mut self) { + imp::release(); + } +} + +#[cfg(all(test, feature = "metrics"))] +mod tests { + use super::*; + + /// Guards from concurrent tests share the process-global gauge, so assertions are on + /// deltas rather than absolute values. + #[test] + fn guard_tracks_live_count_and_raises_peak() { + let before = live(); + + let first = LiveInstanceGuard::new(); + assert_eq!(live(), before + 1); + let second = LiveInstanceGuard::new(); + assert_eq!(live(), before + 2); + assert!(peak() >= before + 2, "peak must cover the observed level"); + + let peak_at_top = peak(); + drop(second); + assert_eq!(live(), before + 1); + drop(first); + assert_eq!(live(), before); + + // The high-water mark is monotonic: dropping instances must not lower it. + assert_eq!(peak(), peak_at_top); + } +} diff --git a/crates/runtime/src/lib.rs b/crates/runtime/src/lib.rs index 6d2c0d2..d195e71 100644 --- a/crates/runtime/src/lib.rs +++ b/crates/runtime/src/lib.rs @@ -22,6 +22,7 @@ use wasmtime::{ use wit_component::ComponentEncoder; pub mod app; +pub mod instances; mod limiter; pub mod logger; mod registry; @@ -116,6 +117,9 @@ pub struct Data { pub epoch_pause_ms: Arc, /// Whether elapsed time of external HTTP should refund epoch ticks. pub pause_epoch_timeout_for_external_http: bool, + /// Counts this instance in `fastedge_wasm_instances_live` for as long as the store — + /// and therefore its pooling-allocator slots — is alive. Held only for its `Drop`. + _live_instance: crate::instances::LiveInstanceGuard, } pub trait BackendRequest { @@ -320,6 +324,20 @@ pub struct WasmEngine { module_linker: ModuleLinker, } +// Manual impl: `derive(Clone)` would incorrectly require `T: Clone`, but the +// engine and linkers are internally reference-counted and clone cheaply for +// any `T`. Needed so executor factories can move an engine handle into +// `spawn_blocking` closures. +impl Clone for WasmEngine { + fn clone(&self) -> Self { + Self { + inner: self.inner.clone(), + component_linker: self.component_linker.clone(), + module_linker: self.module_linker.clone(), + } + } +} + /// A builder interface for configuring a new [`WasmEngine`]. /// /// A new [`WasmEngineBuilder`] can be obtained with [`WasmEngine::builder`]. @@ -420,7 +438,11 @@ pub trait ContextT { } pub trait ExecutorCache { - fn remove(&self, name: &str); + /// Invalidate the cached executor for a single app. + fn remove(&self, name: &str) -> impl std::future::Future + Send; + /// Invalidate all cached executors. Prefer per-app [`ExecutorCache::remove`]: + /// a full flush forces a cold start (load + instantiate) for every active + /// app on the next request, causing a latency burst. fn remove_all(&self); } diff --git a/crates/runtime/src/limiter.rs b/crates/runtime/src/limiter.rs index ed2830f..355e7d1 100644 --- a/crates/runtime/src/limiter.rs +++ b/crates/runtime/src/limiter.rs @@ -6,6 +6,12 @@ use wasmtime::{ResourceLimiter, StoreLimits}; #[derive(Clone, Debug)] pub(crate) struct ProxyLimiter { pub(crate) allocated: usize, + /// Set when a memory growth request is denied because the desired size + /// exceeds the configured limit. This covers the instantiation-time case + /// where a module's declared minimum memory already exceeds the limit + /// (surfaced by wasmtime as "memory minimum size of N pages exceeds memory + /// limits"), letting callers classify the failure as out-of-memory. + pub(crate) oom: bool, inner: StoreLimits, } @@ -17,6 +23,7 @@ impl ProxyLimiter { .build(); Self { allocated: 0, + oom: false, inner, } } @@ -26,6 +33,7 @@ impl Default for ProxyLimiter { fn default() -> Self { ProxyLimiter { allocated: 0, + oom: false, inner: Default::default(), } } @@ -43,6 +51,10 @@ impl ResourceLimiter for ProxyLimiter { // increment used memory if ret { self.allocated += desired - current; + } else { + // Growth denied because `desired` exceeds the configured limit. + // Record it so the failure can be classified as out-of-memory. + self.oom = true; } Ok(ret) } @@ -77,3 +89,29 @@ impl ResourceLimiter for ProxyLimiter { self.inner.memories() } } + +#[cfg(test)] +mod tests { + use super::*; + + const PAGE: usize = 64 * 1024; + + #[test] + fn grow_within_limit_sets_no_oom() { + let mut limiter = ProxyLimiter::new(2 * PAGE); + let ret = limiter.memory_growing(0, PAGE, None).unwrap(); + assert!(ret); + assert!(!limiter.oom); + assert_eq!(limiter.allocated, PAGE); + } + + #[test] + fn grow_exceeding_limit_sets_oom() { + // Requesting more than the configured limit is denied and flagged as OOM. + let mut limiter = ProxyLimiter::new(PAGE); + let ret = limiter.memory_growing(0, 2 * PAGE, None).unwrap(); + assert!(!ret); + assert!(limiter.oom); + assert_eq!(limiter.allocated, 0); + } +} diff --git a/crates/runtime/src/store.rs b/crates/runtime/src/store.rs index cb4860f..0435c92 100644 --- a/crates/runtime/src/store.rs +++ b/crates/runtime/src/store.rs @@ -63,6 +63,37 @@ impl Store { pub fn memory_used(&self) -> usize { self.inner.data().store_limits.allocated } + + /// Returns `true` if a memory growth request was denied because the + /// desired size exceeded the configured limit (out-of-memory). Notably + /// covers instantiation failures where a module's declared minimum memory + /// already exceeds the limit. + pub fn is_oom(&self) -> bool { + self.inner.data().store_limits.oom + } +} + +/// Error indicating a wasm operation failed because it required more memory +/// than the app's configured limit. Surfaced as a typed error (rather than the +/// opaque wasmtime message) so callers can classify the failure as +/// out-of-memory. The originating wasmtime error is preserved as the source. +/// +/// Covers the instantiation-time case where a module's declared minimum memory +/// already exceeds the limit (wasmtime reports "memory minimum size of N pages +/// exceeds memory limits"). +#[derive(Debug)] +pub struct OutOfMemory(pub anyhow::Error); + +impl std::fmt::Display for OutOfMemory { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + write!(f, "out of memory: {}", self.0) + } +} + +impl std::error::Error for OutOfMemory { + fn source(&self) -> Option<&(dyn std::error::Error + 'static)> { + Some(self.0.as_ref()) + } } impl Deref for Store { @@ -345,6 +376,7 @@ impl StoreBuilder { cache: cache_impl, epoch_pause_ms: epoch_pause_ms.clone(), pause_epoch_timeout_for_external_http: self.epoch_exclude_http_wait, + _live_instance: crate::instances::LiveInstanceGuard::new(), }, ); inner.limiter(|state| &mut state.store_limits); @@ -565,4 +597,87 @@ mod tests { // end of the bounded loop without trapping. result.expect("guest must complete when epoch credit is deposited"); } + + // ── live-instance accounting ────────────────────────────────────────── + + /// No-op stats sink; `StoreBuilder::build` only needs `HasStats` to wire the + /// key-value store and utils host state. + #[cfg(feature = "metrics")] + struct NoStats; + + #[cfg(feature = "metrics")] + mod no_stats_impls { + use super::NoStats; + use crate::util::stats::{CdnPhase, ReadStats, StatsVisitor}; + use http_backend::stats::ExtRequestStats; + use std::time::Duration; + use utils::UserDiagStats; + + impl ReadStats for NoStats { + fn count_kv_read(&self, _: i32) {} + fn count_kv_byod_read(&self, _: i32) {} + } + impl UserDiagStats for NoStats { + fn set_user_diag(&self, _: &str) {} + } + impl ExtRequestStats for NoStats { + fn observe_ext(&self, _: Duration) {} + } + impl StatsVisitor for NoStats { + fn status_code(&self, _: u16) {} + fn memory_used(&self, _: u64) {} + fn fail_reason(&self, _: i32) {} + fn observe(&self, _: Duration) {} + fn get_time_elapsed(&self) -> u64 { + 0 + } + fn get_memory_used(&self) -> u64 { + 0 + } + fn cdn_phase(&self, _: CdnPhase) {} + } + } + + #[cfg(feature = "metrics")] + impl HasStats for NoStats { + fn get_stats(&self) -> Arc { + Arc::new(NoStats) + } + } + + /// End-to-end wiring check: a store built the way every executor builds it must be + /// counted in `fastedge_wasm_instances_live` for exactly as long as it is alive, since + /// that is the window in which it holds pooling-allocator slots. + #[cfg(feature = "metrics")] + #[test] + fn store_lifetime_is_counted_as_a_live_instance() { + use crate::instances; + + let engine = make_engine(); + let before = instances::live(); + + let store = StoreBuilder::new(engine.clone(), WasiVersion::Preview1) + .build(NoStats) + .expect("build store"); + assert_eq!( + instances::live(), + before + 1, + "building a store must count a live instance" + ); + + let second = StoreBuilder::new(engine, WasiVersion::Preview1) + .build(NoStats) + .expect("build store"); + assert_eq!(instances::live(), before + 2); + assert!(instances::peak() >= before + 2); + + drop(second); + assert_eq!(instances::live(), before + 1); + drop(store); + assert_eq!( + instances::live(), + before, + "dropping the store must release the count" + ); + } } diff --git a/crates/runtime/src/util/metrics.rs b/crates/runtime/src/util/metrics.rs index eefa490..7d5b31a 100644 --- a/crates/runtime/src/util/metrics.rs +++ b/crates/runtime/src/util/metrics.rs @@ -14,7 +14,8 @@ lazy_static! { .unwrap(); static ref ERROR_COUNT: IntCounterVec = register_int_counter_vec!( "fastedge_error_total_count", - "Number of failed app calls.", + "Total number of failed app calls, by executor and failure reason \ + (counter; the non-conventional *_count suffix is kept for backwards compatibility).", &["executor", "reason"] ) .unwrap(); @@ -30,7 +31,10 @@ lazy_static! { .unwrap(); static ref MEMORY_USAGE: IntCounterVec = register_int_counter_vec!( "fastedge_wasm_memory_used", - "WASM Memory usage", + "Cumulative WASM linear-memory bytes consumed by app calls (counter, NOT a \ + current-usage gauge; the name is kept for backwards compatibility). \ + rate() = memory-zeroing throughput; divide by rate(fastedge_call_count) \ + for average memory per request.", &["executor"] ) .unwrap(); diff --git a/src/context.rs b/src/context.rs index 8163a3d..1a9a990 100644 --- a/src/context.rs +++ b/src/context.rs @@ -103,7 +103,7 @@ impl ContextT for Context { impl ExecutorFactory>> for Context { type Executor = RunExecutor; - fn get_executor( + async fn get_executor( &self, name: SmolStr, app: &App, @@ -155,7 +155,7 @@ impl ExecutorFactory>> for Context { } impl ExecutorCache for Context { - fn remove(&self, _name: &str) { + async fn remove(&self, _name: &str) { unreachable!() } diff --git a/src/key_value.rs b/src/key_value.rs index c2d6526..47f5250 100644 --- a/src/key_value.rs +++ b/src/key_value.rs @@ -17,7 +17,7 @@ impl StoreManager for CliStoreManager { let Some(opts) = self.stores.iter().find(|store| store.param == param) else { return Err(Error::NoSuchStore); }; - let store = RedisStore::open(&opts.param).await?; + let store = RedisStore::open(&opts.param, 1).await?; Ok(Arc::new(store)) } }