Sync/executor cache metrics cleanup

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<HttpState<FastEdgeConnector>>,
-        ) -> anyhow::Result<Self::Executor> {
-            let mut dictionary = Dictionary::new();
-            for (k, v) in cfg.env.iter() {
-                dictionary.insert(k.to_string(), v.to_string());
+        ) -> impl std::future::Future<Output = anyhow::Result<Self::Executor>> + 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::<Vec<(&SmolStr, &SmolStr)>>();
+
+                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::<Vec<(&SmolStr, &SmolStr)>>();
-
-            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,
-            ))
         }
     }
 

crates/http-service/src/executor/mod.rs

@@ -36,12 +36,16 @@ pub trait HttpExecutor {
 
 pub trait ExecutorFactory<C> {
     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<C>,
-    ) -> Result<Self::Executor>;
+    ) -> impl std::future::Future<Output = Result<Self::Executor>> + Send;
 }
 
 pub(crate) fn get_properties(headers: &HeaderMap<HeaderValue>) -> HashMap<String, String> {

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;