runtime: Depend on wasmtime PR #1761

core/src/subgraph/instance_manager.rs

@@ -909,12 +909,12 @@ async fn process_triggers<B: BlockStreamBuilder, T: RuntimeHostBuilder, S: Send
             .await
             .map_err(move |e| match transaction_id {
                 Some(tx_hash) => format_err!(
-                    "Failed to process trigger in block {}, transaction {:x}: {}",
+                    "Failed to process trigger in block {}, transaction {:x}: {:#}",
                     block_ptr,
                     tx_hash,
                     e
                 ),
-                None => format_err!("Failed to process trigger: {}", e),
+                None => format_err!("Failed to process trigger: {:#}", e),
             })?;
         let elapsed = start.elapsed().as_secs_f64();
         subgraph_metrics.observe_trigger_processing_duration(elapsed, trigger_type);

graphql/src/subscription/mod.rs

@@ -167,7 +167,7 @@ fn map_source_to_response_stream(
                     resolver.clone(),
                     query.clone(),
                     event,
-                    timeout.clone(),
+                    timeout,
                     max_first,
                 )
                 .boxed(),

runtime/wasm/Cargo.toml

@@ -24,7 +24,7 @@ bytes = "0.5"
 # PR is merged. https://github.com/rustonaut/maybe-owned/pull/9
 # See also 92cd8019-0136-4011-96a0-40b3eec37f73
 maybe-owned = { git = "https://github.com/rustonaut/maybe-owned", branch = "master" }
-wasmtime = { git = "https://github.com/leoyvens/wasmtime.git", branch = "from-error-for-trap" }
+wasmtime = { git = "https://github.com/yurydelendik/wasmtime.git", branch = "mv-sig-registry" }
 
 [dev-dependencies]
 graphql-parser = "0.2.3"

runtime/wasm/src/host_exports.rs

@@ -389,7 +389,7 @@ impl HostExports {
                     valid_module.clone(),
                     ctx.derive_with_empty_block_state(),
                     host_metrics.clone(),
-                    module.timeout.clone(),
+                    module.timeout,
                 )?;
                 let result = module.handle_json_callback(&callback, &sv.value, &user_data)?;
                 // Log progress every 15s

runtime/wasm/src/mapping.rs

@@ -21,6 +21,8 @@ pub fn spawn_module(
     runtime: tokio::runtime::Handle,
     timeout: Option<Duration>,
 ) -> Result<mpsc::Sender<MappingRequest>, anyhow::Error> {
+    let valid_module = Arc::new(ValidModule::new(&raw_module).unwrap());
+
     // Create channel for event handling requests
     let (mapping_request_sender, mapping_request_receiver) = mpsc::channel(100);
 
@@ -33,7 +35,6 @@ pub fn spawn_module(
     let conf =
         thread::Builder::new().name(format!("mapping-{}-{}", &subgraph_id, uuid::Uuid::new_v4()));
     conf.spawn(move || {
-        let valid_module = Arc::new(ValidModule::new(&raw_module).unwrap());
         runtime.enter(|| {
             // Pass incoming triggers to the WASM module and return entity changes;
             // Stop when canceled because all RuntimeHosts and their senders were dropped.
@@ -181,9 +182,8 @@ impl ValidModule {
         config.strategy(wasmtime::Strategy::Cranelift).unwrap();
         config.interruptable(true); // For timeouts.
         config.cranelift_nan_canonicalization(true); // For NaN determinism.
-
-        let store = wasmtime::Store::new(&wasmtime::Engine::new(&config));
-        let module = wasmtime::Module::from_binary(&store, raw_module)?;
+        let engine = &wasmtime::Engine::new(&config);
+        let module = wasmtime::Module::from_binary(&engine, raw_module)?;
 
         let mut import_name_to_modules: BTreeMap<String, Vec<String>> = BTreeMap::new();
         for (name, module) in module

runtime/wasm/src/module/mod.rs

@@ -31,6 +31,8 @@ use into_wasm_ret::IntoWasmRet;
 #[cfg(test)]
 mod test;
 
+const TRAP_TIMEOUT: &str = "trap: interrupt";
+
 /// Handle to a WASM instance, which is terminated if and only if this is dropped.
 pub(crate) struct WasmInstanceHandle {
     // This is the only reference to `WasmInstance` that's not within the instance itself,
@@ -177,8 +179,20 @@ impl WasmInstanceHandle {
             .get_func(handler)
             .with_context(|| format!("function {} not found", handler))?;
 
-        func.get1()?(arg.wasm_ptr())
-            .with_context(|| format!("Failed to invoke handler '{}'", handler))
+        func.get1()?(arg.wasm_ptr()).map_err(|e| {
+            if e.to_string().contains(TRAP_TIMEOUT) {
+                anyhow::Error::context(
+                    e.into(),
+                    format!(
+                        "Handler '{}' hit the timeout of '{}' seconds",
+                        handler,
+                        self.instance().timeout.unwrap().as_secs()
+                    ),
+                )
+            } else {
+                anyhow::Error::context(e.into(), format!("Failed to invoke handler '{}'", handler))
+            }
+        })
     }
 }
 
@@ -210,7 +224,7 @@ impl WasmInstance {
         host_metrics: Arc<HostMetrics>,
         timeout: Option<Duration>,
     ) -> Result<WasmInstanceHandle, anyhow::Error> {
-        let mut linker = wasmtime::Linker::new(valid_module.module.store());
+        let mut linker = wasmtime::Linker::new(&wasmtime::Store::new(valid_module.module.engine()));
 
         // Used by exports to access the instance context. It is `None` while the module is not yet
         // instantiated. A desirable consequence is that start function cannot access host exports.
@@ -391,7 +405,7 @@ impl WasmInstance {
             .get1()?;
 
         let should_interrupt = Arc::new(AtomicBool::new(true));
-        if let Some(timeout) = timeout.clone() {
+        if let Some(timeout) = timeout {
             // This task is likely to outlive the instance, which is fine.
             let interrupt_handle = instance.store().interrupt_handle().unwrap();
             let should_interrupt = should_interrupt.clone();

runtime/wasm/src/module/test/abi.rs

@@ -10,7 +10,7 @@ async fn unbounded_loop() {
     );
     let func = module.get_func("loop").get0().unwrap();
     let res: Result<(), _> = func();
-    assert!(res.unwrap_err().to_string().contains("interrupt"));
+    assert!(res.unwrap_err().to_string().contains(TRAP_TIMEOUT));
 }
 
 #[tokio::test]