From 66e1d0a686291526abff117e6b5d8c1c53cce8c6 Mon Sep 17 00:00:00 2001 From: Ruslan Pislari Date: Tue, 28 Jul 2026 18:26:42 +0300 Subject: [PATCH 01/13] feat(memory): classify denied memory growth as out-of-memory during instantiation --- crates/http-service/src/executor/http.rs | 13 ++++++- crates/http-service/src/executor/wasi_http.rs | 13 ++++++- crates/http-service/src/lib.rs | 15 +++++++- crates/runtime/src/limiter.rs | 38 +++++++++++++++++++ crates/runtime/src/store.rs | 31 +++++++++++++++ 5 files changed, 106 insertions(+), 4 deletions(-) diff --git a/crates/http-service/src/executor/http.rs b/crates/http-service/src/executor/http.rs index 8530934..2a541cf 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 diff --git a/crates/http-service/src/executor/wasi_http.rs b/crates/http-service/src/executor/wasi_http.rs index 0d534ff..9a16b5c 100644 --- a/crates/http-service/src/executor/wasi_http.rs +++ b/crates/http-service/src/executor/wasi_http.rs @@ -148,7 +148,18 @@ 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( diff --git a/crates/http-service/src/lib.rs b/crates/http-service/src/lib.rs index 46eca34..2b0ee0a 100644 --- a/crates/http-service/src/lib.rs +++ b/crates/http-service/src/lib.rs @@ -404,8 +404,19 @@ where 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::() { + // `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 { ( StatusCode::OK.as_u16(), 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..fba7b4d 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 { From 2c7b771b0e86ce2dfab63d750540ad7b9b9d9750 Mon Sep 17 00:00:00 2001 From: Ruslan Pislari Date: Wed, 5 Aug 2026 12:45:08 +0300 Subject: [PATCH 02/13] feat(logging): revert Kafka log variant to Log enum --- crates/runtime/src/app.rs | 1 + 1 file changed, 1 insertion(+) 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, } From 363ccf9814f0930160c5528f7c08cc5006b07379 Mon Sep 17 00:00:00 2001 From: Ruslan Pislari Date: Wed, 5 Aug 2026 14:20:11 +0300 Subject: [PATCH 03/13] feat(metrics): add live/peak gauges for concurrent WASM instances MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Exhausting a wasmtime pooling-allocator pool is not graceful: the allocator returns "maximum concurrent limit of N reached", which surfaces as an instantiation error rather than the adaptive run-queue shedding path. Sizing max_execution_stacks needed observed concurrency, which no existing metric provided. Track it with an RAII guard held in runtime::Data, so the count spans exactly the lifetime of the store that owns the pooled slots, including the unwind path. Placing it there rather than in the ProxyWasm executor covers every executor sharing the single Engine — a proxywasm-only gauge would systematically undercount pool occupancy. fastedge_wasm_instances_peak is a monotonic high-water mark: a plain gauge is scrape-limited and would miss the sub-second spikes a stalled backend produces, which are the peaks the pool must be sized for. --- crates/runtime/src/instances.rs | 133 ++++++++++++++++++++++++++++++++ crates/runtime/src/lib.rs | 4 + crates/runtime/src/store.rs | 84 ++++++++++++++++++++ 3 files changed, 221 insertions(+) create mode 100644 crates/runtime/src/instances.rs diff --git a/crates/runtime/src/instances.rs b/crates/runtime/src/instances.rs new file mode 100644 index 0000000..de33c1a --- /dev/null +++ b/crates/runtime/src/instances.rs @@ -0,0 +1,133 @@ +//! 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(); + + /// High-water mark since process start. 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 process start" + ) + .unwrap(); + } + + /// Mirror of the peak, kept separately so the high-water mark can be updated with an + /// atomic `fetch_max` — `IntGauge` only offers `set`, which cannot express "raise to". + 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(); + if PEAK.fetch_max(live, Ordering::Relaxed) < live { + // Publish the resolved maximum rather than `live`: if two threads race here, + // both write the same (largest) value instead of the smaller one winning. + WASM_INSTANCES_PEAK.set(PEAK.load(Ordering::Relaxed)); + } + } + + pub(super) fn release() { + WASM_INSTANCES_LIVE.dec(); + } + + /// Currently live instances. Test/diagnostic accessor. + pub fn live() -> i64 { + WASM_INSTANCES_LIVE.get() + } + + /// High-water mark since process start. Test/diagnostic accessor. + pub fn peak() -> i64 { + PEAK.load(Ordering::Relaxed) + } +} + +#[cfg(not(feature = "metrics"))] +mod imp { + pub(super) fn acquire() {} + pub(super) fn release() {} +} + +#[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..81054f2 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 { diff --git a/crates/runtime/src/store.rs b/crates/runtime/src/store.rs index fba7b4d..0435c92 100644 --- a/crates/runtime/src/store.rs +++ b/crates/runtime/src/store.rs @@ -376,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); @@ -596,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" + ); + } } From 695bde47f0faaef4eadb191865f98ea1f285a566 Mon Sep 17 00:00:00 2001 From: Ruslan Pislari Date: Tue, 11 Aug 2026 11:37:37 +0300 Subject: [PATCH 04/13] feat(redis): implement connection pooling for Redis backend --- crates/key-value-store/src/redis_impl.rs | 65 ++++++++++++++++-------- 1 file changed, 45 insertions(+), 20 deletions(-) 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"); From 5d00aafe4d5153c078f4efd24742e310f0a56cc9 Mon Sep 17 00:00:00 2001 From: Ruslan Pislari Date: Tue, 11 Aug 2026 12:15:14 +0300 Subject: [PATCH 05/13] feat(kvrocks): add pool size limits for RedisStore connections --- src/key_value.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) 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)) } } From 00d69c9f6e8046a29f45b87ca1f00a2ba27b2d90 Mon Sep 17 00:00:00 2001 From: Ruslan Pislari Date: Thu, 13 Aug 2026 14:59:58 +0300 Subject: [PATCH 06/13] fix(instances): update peak gauge description and logic for accurate scraping --- crates/runtime/src/instances.rs | 29 +++++++++++++++++++---------- 1 file changed, 19 insertions(+), 10 deletions(-) diff --git a/crates/runtime/src/instances.rs b/crates/runtime/src/instances.rs index de33c1a..8b265c5 100644 --- a/crates/runtime/src/instances.rs +++ b/crates/runtime/src/instances.rs @@ -28,18 +28,20 @@ mod imp { ) .unwrap(); - /// High-water mark since process start. A plain gauge is sampled at the scrape + /// 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 process start" + "Highest number of concurrently live WASM instances observed since the \ + previous scrape (resets on scrape)" ) .unwrap(); } - /// Mirror of the peak, kept separately so the high-water mark can be updated with an - /// atomic `fetch_max` — `IntGauge` only offers `set`, which cannot express "raise to". + /// 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() { @@ -47,11 +49,7 @@ mod imp { // 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(); - if PEAK.fetch_max(live, Ordering::Relaxed) < live { - // Publish the resolved maximum rather than `live`: if two threads race here, - // both write the same (largest) value instead of the smaller one winning. - WASM_INSTANCES_PEAK.set(PEAK.load(Ordering::Relaxed)); - } + PEAK.fetch_max(live, Ordering::Relaxed); } pub(super) fn release() { @@ -63,20 +61,31 @@ mod imp { WASM_INSTANCES_LIVE.get() } - /// High-water mark since process start. Test/diagnostic accessor. + /// 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() {} } #[cfg(feature = "metrics")] pub use imp::{live, peak}; +pub use imp::flush_peak; /// RAII counter for one live WASM instance. /// From 37d3c92be2378ec33c27671644da72476844b0d7 Mon Sep 17 00:00:00 2001 From: Ruslan Pislari Date: Wed, 19 Aug 2026 10:21:42 +0300 Subject: [PATCH 07/13] fix(tracing): change span type from info to error for better logging --- crates/http-service/src/lib.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/crates/http-service/src/lib.rs b/crates/http-service/src/lib.rs index 2b0ee0a..b8b8f0e 100644 --- a/crates/http-service/src/lib.rs +++ b/crates/http-service/src/lib.rs @@ -285,7 +285,7 @@ where Ok(app_name) => app_name, }; - let span = tracing::info_span!("http", app = %app_name, traceparent = %traceparent); + let span = tracing::error_span!("http", app = %app_name, traceparent = %traceparent); let _enter = span.enter(); // lookup for application config and binary_id From 22157afdaa0feb0c8e8f6fe457fb5d7b13f06b3f Mon Sep 17 00:00:00 2001 From: Ruslan Pislari Date: Wed, 19 Aug 2026 11:21:17 +0300 Subject: [PATCH 08/13] fix(http): classify execution errors and update timeout handling --- crates/http-service/src/executor/wasi_http.rs | 17 ++++++++-- crates/http-service/src/lib.rs | 32 +++++++++++++++++++ 2 files changed, 47 insertions(+), 2 deletions(-) diff --git a/crates/http-service/src/executor/wasi_http.rs b/crates/http-service/src/executor/wasi_http.rs index 9a16b5c..90a083c 100644 --- a/crates/http-service/src/executor/wasi_http.rs +++ b/crates/http-service/src/executor/wasi_http.rs @@ -165,15 +165,28 @@ where 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 2b0ee0a..cf705ea 100644 --- a/crates/http-service/src/lib.rs +++ b/crates/http-service/src/lib.rs @@ -402,6 +402,38 @@ where } } +/// 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() { + AppResult::TIMEOUT + } else if 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(); // `OutOfMemory` wraps the underlying wasmtime error as its source, so it From 3c27649cef78629282ecb21e6c3e8f71245a494d Mon Sep 17 00:00:00 2001 From: Ruslan Pislari Date: Wed, 19 Aug 2026 11:36:01 +0300 Subject: [PATCH 09/13] fix(tracing): ensure proper span context during async request processing --- crates/http-service/src/lib.rs | 216 +++++++++++++++++---------------- 1 file changed, 110 insertions(+), 106 deletions(-) diff --git a/crates/http-service/src/lib.rs b/crates/http-service/src/lib.rs index b8b8f0e..775471a 100644 --- a/crates/http-service/src/lib.rs +++ b/crates/http-service/src/lib.rs @@ -286,119 +286,123 @@ where }; let span = tracing::error_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 - .context - .lookup_by_name(&name) - .instrument(span.clone()) - .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(); - } + // 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)), + }; - Some((app_name, cfg)) => (app_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(); + } - // 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); - } - }; + Some((app_name, cfg)) => (app_name, cfg), + }; - let stats = self.context.new_stats_row(&traceparent, &app_name, &cfg); + // 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 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()), - ); + let stats = self.context.new_stats_row(&traceparent, &app_name, &cfg); - 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"); + 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()), + ); - #[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) + 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 } } From 7f8a5df31500c6352c921bf004ad1e653ef63872 Mon Sep 17 00:00:00 2001 From: Ruslan Pislari Date: Mon, 24 Aug 2026 11:52:56 +0300 Subject: [PATCH 10/13] chore(metrics): remove redundant tokio metrics, add client/kind labels, fix docs - Remove tokio_park_unpark_count: always exactly 2x tokio_park_count by construction (tokio increments it on both park and unpark), so the metric carried no information. - Remove tokio_global_queue_depth: strictly dominated by tokio_global_queue_depth_peak, which is fed by a 10ms sampler and floored at the current reading on scrape; the instantaneous value at scrape time was pure noise. - Remove tokio_busy_duration_us: tokio only publishes worker busy time just-before-park / on periodic maintenance, so the value freezes exactly when a runtime is saturated; superseded by the live-sampled fastedge_workload_15s. - fastedge_wasm_connections: IntGauge -> IntGaugeVec with a client label (nginx = V1/V2a/V2b/V3-yamux, core-proxy = V2/V2c). Counting moved from accept_loop into serve via a ConnectionGauge RAII guard, created after the handshake when the peer identity is known; Drop accounts for every exit path. - fastedge_wasm_commands_total: IntCounter -> IntCounterVec with a kind label (send = fire-and-forget, request_reply = round-trip). The request_reply path was previously not counted at all; it is now counted before the additional_info fast path, so the delta against fastedge_wasm_request_reply_duration_count measures locally-served zero-cost host calls. - Fix misleading HELP texts for fastedge_error_total_count and fastedge_wasm_memory_used (cumulative counter, not a usage gauge); names kept for backwards compatibility. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- crates/runtime/src/util/metrics.rs | 8 ++++++-- 1 file changed, 6 insertions(+), 2 deletions(-) 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(); From 755ec1975afcc129fc2cc548f423ad0c8be65277 Mon Sep 17 00:00:00 2001 From: Ruslan Pislari Date: Mon, 24 Aug 2026 12:43:15 +0300 Subject: [PATCH 11/13] perf: selective executor cache invalidation and non-blocking cold starts Two fixes for periodic p99 latency spikes: 1. Selective invalidation on config poll: on a revision bump the poller used to call remove_all(), wiping every cached executor and forcing a cold start for all active apps every ~120s. Now Config::stale_apps() diffs old vs new config and only removed/changed apps are evicted. 2. get_executor is now async (RPITIT): executor caches switch to moka::future::Cache and the blocking build (load_component / instantiate_pre) runs via spawn_blocking instead of block_in_place, which migrated the calling worker's run queue on every cold start. moka coalesces concurrent misses for the same app into one build. Supporting changes: manual Clone for WasmEngine (no T: Clone bound), ExecutorCache::remove is async (future cache invalidate is async), and config pub/sub handlers await the eviction. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- crates/http-service/src/executor/http.rs | 56 ++++++----- crates/http-service/src/executor/mod.rs | 6 +- crates/http-service/src/lib.rs | 121 ++++++++++++----------- crates/runtime/src/instances.rs | 2 +- crates/runtime/src/lib.rs | 20 +++- src/context.rs | 4 +- 6 files changed, 117 insertions(+), 92 deletions(-) diff --git a/crates/http-service/src/executor/http.rs b/crates/http-service/src/executor/http.rs index 2a541cf..880b043 100644 --- a/crates/http-service/src/executor/http.rs +++ b/crates/http-service/src/executor/http.rs @@ -407,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/lib.rs b/crates/http-service/src/lib.rs index 1d06ce4..ae21782 100644 --- a/crates/http-service/src/lib.rs +++ b/crates/http-service/src/lib.rs @@ -346,6 +346,7 @@ where let executor = match self .context .get_executor(app_name.clone(), &cfg, &self.engine) + .await { Ok(executor) => executor, Err(error) => { @@ -453,78 +454,78 @@ fn map_err(error: Error) -> (u16, AppResult, HyperOutgoingBody, u16) { INTERNAL_STATUS_OUT_OF_MEMORY, ) } else 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::() { + 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/runtime/src/instances.rs b/crates/runtime/src/instances.rs index 8b265c5..9dd1e89 100644 --- a/crates/runtime/src/instances.rs +++ b/crates/runtime/src/instances.rs @@ -83,9 +83,9 @@ mod imp { pub fn flush_peak() {} } +pub use imp::flush_peak; #[cfg(feature = "metrics")] pub use imp::{live, peak}; -pub use imp::flush_peak; /// RAII counter for one live WASM instance. /// diff --git a/crates/runtime/src/lib.rs b/crates/runtime/src/lib.rs index 81054f2..d195e71 100644 --- a/crates/runtime/src/lib.rs +++ b/crates/runtime/src/lib.rs @@ -324,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`]. @@ -424,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/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!() } From ec4ffa7e387ae7c5cfa3fa7fd13a06a3506d7ccf Mon Sep 17 00:00:00 2001 From: Ruslan Pislari Date: Tue, 1 Sep 2026 13:18:35 +0300 Subject: [PATCH 12/13] fix: unify timeout handling for elapsed deadlines --- Cargo.lock | 42 ++++++++++++++-------------------- crates/http-service/src/lib.rs | 6 ++--- 2 files changed, 20 insertions(+), 28 deletions(-) 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/lib.rs b/crates/http-service/src/lib.rs index ae21782..db64351 100644 --- a/crates/http-service/src/lib.rs +++ b/crates/http-service/src/lib.rs @@ -430,9 +430,9 @@ pub(crate) fn fail_reason_of(error: &Error) -> AppResult { wasmtime::Trap::UnreachableCodeReached => AppResult::OOM, _ => AppResult::OTHER, } - } else if root_cause.downcast_ref::().is_some() { - AppResult::TIMEOUT - } else if root_cause.to_string().ends_with("deadline has elapsed") { + } else if root_cause.downcast_ref::().is_some() + || root_cause.to_string().ends_with("deadline has elapsed") + { AppResult::TIMEOUT } else { AppResult::OTHER From 7a808f62a63a4e14c886d9539ad09ecb19a3f912 Mon Sep 17 00:00:00 2001 From: Ruslan Pislari Date: Tue, 1 Sep 2026 14:16:28 +0300 Subject: [PATCH 13/13] chore: remove FOSSA configuration file --- .github/workflows/fossa.yaml | 29 ----------------------------- 1 file changed, 29 deletions(-) delete mode 100644 .github/workflows/fossa.yaml 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