feat: add rust-protobuf support to extism-convert (#653)

(code from @zshipko, thank you!)

---------

Co-authored-by: Zach Shipko <zach@dylibso.com>

convert/Cargo.toml

@@ -14,6 +14,7 @@ anyhow = "1.0.75"
 base64 = "~0.21"
 bytemuck = {version = "1.14.0", optional = true }
 prost = { version = "0.12.0", optional = true }
+protobuf = { version = "3.2.0", optional = true }
 rmp-serde = { version = "1.1.2", optional = true }
 serde = "1.0.186"
 serde_json = "1.0.105"
@@ -22,7 +23,6 @@ serde_json = "1.0.105"
 serde = { version = "1.0.186", features = ["derive"] }
 
 [features]
-default = ["msgpack", "protobuf", "raw"]
+default = ["msgpack", "prost", "raw"]
 msgpack = ["rmp-serde"]
-protobuf = ["prost"]
 raw = ["bytemuck"]

convert/src/encoding.rs

@@ -112,19 +112,19 @@ impl FromBytesOwned for Base64<String> {
 /// Protobuf encoding
 ///
 /// Allows for `prost` Protobuf messages to be used as arguments to Extism plugin calls
-#[cfg(feature = "protobuf")]
+#[cfg(feature = "prost")]
 #[derive(Debug)]
-pub struct Protobuf<T: prost::Message>(pub T);
+pub struct Prost<T: prost::Message>(pub T);
 
-#[cfg(feature = "protobuf")]
-impl<T: prost::Message> From<T> for Protobuf<T> {
+#[cfg(feature = "prost")]
+impl<T: prost::Message> From<T> for Prost<T> {
     fn from(data: T) -> Self {
         Self(data)
     }
 }
 
-#[cfg(feature = "protobuf")]
-impl<'a, T: prost::Message> ToBytes<'a> for Protobuf<T> {
+#[cfg(feature = "prost")]
+impl<'a, T: prost::Message> ToBytes<'a> for Prost<T> {
     type Bytes = Vec<u8>;
 
     fn to_bytes(&self) -> Result<Self::Bytes, Error> {
@@ -132,10 +132,32 @@ impl<'a, T: prost::Message> ToBytes<'a> for Protobuf<T> {
     }
 }
 
-#[cfg(feature = "protobuf")]
-impl<T: Default + prost::Message> FromBytesOwned for Protobuf<T> {
+#[cfg(feature = "prost")]
+impl<T: Default + prost::Message> FromBytesOwned for Prost<T> {
     fn from_bytes_owned(data: &[u8]) -> Result<Self, Error> {
-        Ok(Protobuf(T::decode(data)?))
+        Ok(Prost(T::decode(data)?))
+    }
+}
+
+/// Protobuf encoding
+///
+/// Allows for `rust-protobuf` Protobuf messages to be used as arguments to Extism plugin calls
+#[cfg(feature = "protobuf")]
+pub struct Protobuf<T: protobuf::Message>(pub T);
+
+#[cfg(feature = "protobuf")]
+impl<'a, T: protobuf::Message> ToBytes<'a> for Protobuf<T> {
+    type Bytes = Vec<u8>;
+
+    fn to_bytes(&self) -> Result<Self::Bytes, Error> {
+        Ok(self.0.write_to_bytes()?)
+    }
+}
+
+#[cfg(feature = "protobuf")]
+impl<T: Default + protobuf::Message> FromBytesOwned for Protobuf<T> {
+    fn from_bytes_owned(data: &[u8]) -> Result<Self, Error> {
+        Ok(Protobuf(T::parse_from_bytes(data)?))
     }
 }
 

convert/src/lib.rs

@@ -18,6 +18,9 @@ pub use encoding::{Base64, Json};
 #[cfg(feature = "msgpack")]
 pub use encoding::Msgpack;
 
+#[cfg(feature = "prost")]
+pub use encoding::Prost;
+
 #[cfg(feature = "protobuf")]
 pub use encoding::Protobuf;
 

kernel/src/lib.rs

@@ -382,11 +382,40 @@ pub unsafe fn free(p: Pointer) {
 }
 
 /// Get the length of an allocated memory block
+///
+/// Note: this should only be called on memory handles returned
+/// by a call to `alloc` - it will return garbage on invalid offsets
+#[no_mangle]
+pub unsafe fn length_unsafe(p: Pointer) -> Length {
+    if p == 0 {
+        return 0;
+    }
+
+    if !MemoryRoot::pointer_in_bounds_fast(p) {
+        return 0;
+    }
+
+    let ptr = p - core::mem::size_of::<MemoryBlock>() as u64;
+    let block = &mut *(ptr as *mut MemoryBlock);
+
+    // Simplest sanity check to verify the pointer is a block
+    if block.status.load(Ordering::Acquire) != MemoryStatus::Active as u8 {
+        return 0;
+    }
+
+    block.used as Length
+}
+
+/// Get the length but returns 0 if the offset is not a valid handle.
+///
+/// Note: this function walks each node in the allocations list, which ensures correctness, but is also
+/// slow
 #[no_mangle]
 pub unsafe fn length(p: Pointer) -> Length {
     if p == 0 {
         return 0;
     }
+
     if let Some(block) = MemoryRoot::new().find_block(p) {
         block.used as Length
     } else {

runtime/Cargo.toml

@@ -9,8 +9,8 @@ repository.workspace = true
 version.workspace = true
 
 [dependencies]
-wasmtime = ">= 14.0.0, < 16.0.0"
-wasmtime-wasi = ">= 14.0.0, < 16.0.0"
+wasmtime = ">= 14.0.0, < 17.0.0"
+wasmtime-wasi = ">= 14.0.0, < 17.0.0"
 anyhow = "1"
 serde = {version = "1", features = ["derive"]}
 serde_json = "1"
@@ -33,7 +33,7 @@ register-filesystem = [] # enables wasm to be loaded from disk
 http = ["ureq"]          # enables extism_http_request
 
 [build-dependencies]
-cbindgen = "0.26"
+cbindgen = { version = "0.26", default-features = false }
 
 [dev-dependencies]
 criterion = "0.5.1"

runtime/extism.h

@@ -173,13 +173,6 @@ void extism_function_free(ExtismFunction *f);
  */
 void extism_function_set_namespace(ExtismFunction *ptr, const char *namespace_);
 
-/**
- * Set the cost of an `ExtismFunction`, when set to 0 this has no effect, when set to `1` this will add
- * the runtime of the function back to the plugin timer.
- */
-void extism_function_set_cost(ExtismFunction *ptr,
-                              double cost);
-
 /**
  * Create a new plugin with host functions, the functions passed to this function no longer need to be manually freed using
  *

runtime/src/current_plugin.rs

@@ -246,6 +246,26 @@ impl CurrentPlugin {
         Ok(len)
     }
 
+    pub fn memory_length_unsafe(&mut self, offs: u64) -> Result<u64, Error> {
+        let (linker, mut store) = self.linker_and_store();
+        let output = &mut [Val::I64(0)];
+        if let Some(f) = linker.get(&mut store, EXTISM_ENV_MODULE, "length_unsafe") {
+            f.into_func()
+                .unwrap()
+                .call(&mut store, &[Val::I64(offs as i64)], output)?;
+        } else {
+            anyhow::bail!("unable to locate an extism kernel function: length_unsafe",)
+        }
+        let len = output[0].unwrap_i64() as u64;
+        trace!(
+            plugin = self.id.to_string(),
+            "memory_length_unsafe({}) = {}",
+            offs,
+            len
+        );
+        Ok(len)
+    }
+
     /// Access a plugin's variables
     pub fn vars(&self) -> &std::collections::BTreeMap<String, Vec<u8>> {
         &self.vars

runtime/src/function.rs

@@ -155,7 +155,7 @@ unsafe impl<T> Sync for UserData<T> {}
 unsafe impl Send for CPtr {}
 unsafe impl Sync for CPtr {}
 
-pub(crate) type FunctionInner = dyn Fn(wasmtime::Caller<CurrentPlugin>, &[wasmtime::Val], &mut [wasmtime::Val]) -> Result<(), Error>
+type FunctionInner = dyn Fn(wasmtime::Caller<CurrentPlugin>, &[wasmtime::Val], &mut [wasmtime::Val]) -> Result<(), Error>
     + Sync
     + Send;
 
@@ -174,9 +174,6 @@ pub struct Function {
     /// Function handle
     pub(crate) f: Arc<FunctionInner>,
 
-    /// Function cost (in terms of time)
-    pub(crate) cost: f64,
-
     /// UserData
     pub(crate) _user_data: UserDataHandle,
 }
@@ -207,12 +204,10 @@ impl Function {
             ty,
             f: Arc::new(
                 move |mut caller: Caller<_>, inp: &[Val], outp: &mut [Val]| {
-                    let plugin = caller.data_mut();
                     let x = data.clone();
-                    f(plugin, inp, outp, x)
+                    f(caller.data_mut(), inp, outp, x)
                 },
             ),
-            cost: 0.0,
             namespace: None,
             _user_data: match &user_data {
                 UserData::C(ptr) => UserDataHandle::C(ptr.clone()),
@@ -244,23 +239,6 @@ impl Function {
         self
     }
 
-    /// Function cost
-    pub fn cost(&self) -> f64 {
-        self.cost
-    }
-
-    /// Set host function cost
-    pub fn set_cost(&mut self, cost: f64) {
-        trace!("Setting cost for {} to {cost}", self.name);
-        self.cost = cost;
-    }
-
-    /// Update host function cost
-    pub fn with_cost(mut self, cost: f64) -> Self {
-        self.set_cost(cost);
-        self
-    }
-
     /// Get function type
     pub fn ty(&self) -> &wasmtime::FuncType {
         &self.ty

runtime/src/lib.rs

@@ -14,7 +14,6 @@ pub(crate) mod manifest;
 pub(crate) mod pdk;
 mod plugin;
 mod plugin_builder;
-mod timeout_manager;
 mod timer;
 
 /// Extism C API
@@ -28,7 +27,6 @@ pub use plugin::{CancelHandle, Plugin, WasmInput, EXTISM_ENV_MODULE, EXTISM_USER
 pub use plugin_builder::{DebugOptions, PluginBuilder};
 
 pub(crate) use internal::{Internal, Wasi};
-pub(crate) use timeout_manager::TimeoutManager;
 pub(crate) use timer::{Timer, TimerAction};
 pub(crate) use tracing::{debug, error, trace, warn};
 

runtime/src/manifest.rs

@@ -117,10 +117,17 @@ pub(crate) fn load(
     match input {
         WasmInput::Data(data) => {
             let has_magic = data.len() >= 4 && data[0..4] == WASM_MAGIC;
-            let is_wat = data.starts_with(b"(module") || data.starts_with(b";;");
+            let s = std::str::from_utf8(&data);
+            let is_wat = s.is_ok_and(|s| {
+                let s = s.trim_start();
+                let starts_with_module = s.len() > 2
+                    && data[0] == b'('   // First character is `(`
+                    && s[1..].trim_start().starts_with("module"); // Then `module` (after any whitespace)
+                starts_with_module || s.starts_with(";;") || s.starts_with("(;")
+            });
             if !has_magic && !is_wat {
                 trace!("Loading manifest");
-                if let Ok(s) = std::str::from_utf8(&data) {
+                if let Ok(s) = s {
                     let t = if let Ok(t) = toml::from_str::<extism_manifest::Manifest>(s) {
                         trace!("Manifest is TOML");
                         modules(engine, &t, &mut mods)?;

runtime/src/plugin.rs

@@ -238,20 +238,6 @@ impl Plugin {
             CurrentPlugin::new(manifest, with_wasi, available_pages, id)?,
         );
         store.set_epoch_deadline(1);
-        store.call_hook(|data, hook| {
-            if hook.entering_host() {
-                let tx = Timer::tx();
-                tx.send(TimerAction::EnterHost {
-                    id: data.id.clone(),
-                })?;
-            } else if hook.exiting_host() {
-                let tx = Timer::tx();
-                tx.send(TimerAction::ExitHost {
-                    id: data.id.clone(),
-                })?;
-            }
-            Ok(())
-        });
 
         let mut linker = Linker::new(&engine);
         linker.allow_shadowing(true);
@@ -298,13 +284,9 @@ impl Plugin {
         for f in &mut imports {
             let name = f.name().to_string();
             let ns = f.namespace().unwrap_or(EXTISM_USER_MODULE);
-            let cost = f.cost();
-            let inner: &function::FunctionInner = unsafe { &*(f.f.as_ref() as *const _) };
-            linker.func_new(ns, &name, f.ty().clone(), move |caller, params, results| {
-                let _timeout =
-                    TimeoutManager::new(caller.data()).map(|x| x.with_cost(cost.clone()));
-                inner(caller, params, results)
-            })?;
+            unsafe {
+                linker.func_new(ns, &name, f.ty().clone(), &*(f.f.as_ref() as *const _))?;
+            }
         }
 
         let instance_pre = linker.instantiate_pre(main)?;
@@ -842,6 +824,24 @@ impl Plugin {
             .and_then(move |_| self.output())
     }
 
+    /// Similar to `Plugin::call`, but returns the Extism error code along with the
+    /// `Error`. It is assumed if `Ok(_)` is returned that the error code was `0`.
+    ///
+    /// All Extism plugin calls return an error code, `Plugin::call` consumes the error code,
+    /// while `Plugin::call_get_error_code` preserves it - this function should only be used
+    /// when you need to inspect the actual return value of a plugin function when it fails.
+    pub fn call_get_error_code<'a, 'b, T: ToBytes<'a>, U: FromBytes<'b>>(
+        &'b mut self,
+        name: impl AsRef<str>,
+        input: T,
+    ) -> Result<U, (Error, i32)> {
+        let lock = self.instance.clone();
+        let mut lock = lock.lock().unwrap();
+        let data = input.to_bytes().map_err(|e| (e, -1))?;
+        self.raw_call(&mut lock, name, data)
+            .and_then(move |_| self.output().map_err(|e| (e, -1)))
+    }
+
     /// Get a `CancelHandle`, which can be used from another thread to cancel a running plugin
     pub fn cancel_handle(&self) -> CancelHandle {
         self.cancel_handle.clone()

runtime/src/sdk.rs

@@ -258,18 +258,6 @@ pub unsafe extern "C" fn extism_function_set_namespace(
     }
 }
 
-/// Set the cost of an `ExtismFunction`, when set to 0 this has no effect, when set to `1` this will add
-/// the runtime of the function back to the plugin timer.
-#[no_mangle]
-pub unsafe extern "C" fn extism_function_set_cost(ptr: *mut ExtismFunction, cost: f64) {
-    let f = &mut *ptr;
-    if let Some(x) = f.0.get_mut() {
-        x.set_cost(cost);
-    } else {
-        debug!("Trying to set the cost of already registered function")
-    }
-}
-
 /// Create a new plugin with host functions, the functions passed to this function no longer need to be manually freed using
 ///
 /// `wasm`: is a WASM module (wat or wasm) or a JSON encoded manifest

runtime/src/tests/kernel.rs

@@ -22,6 +22,20 @@ fn extism_length<T>(mut store: &mut wasmtime::Store<T>, instance: &mut Instance,
     out[0].unwrap_i64() as u64
 }
 
+fn extism_length_unsafe<T>(
+    mut store: &mut wasmtime::Store<T>,
+    instance: &mut Instance,
+    p: u64,
+) -> u64 {
+    let out = &mut [Val::I64(0)];
+    instance
+        .get_func(&mut store, "length_unsafe")
+        .unwrap()
+        .call(&mut store, &[Val::I64(p as i64)], out)
+        .unwrap();
+    out[0].unwrap_i64() as u64
+}
+
 fn extism_load_u8<T>(mut store: &mut wasmtime::Store<T>, instance: &mut Instance, p: u64) -> u8 {
     let out = &mut [Val::I32(0)];
     instance
@@ -173,6 +187,7 @@ fn test_kernel_allocations() {
     let first_alloc = p;
     assert!(p > 0);
     assert_eq!(extism_length(&mut store, instance, p), 1);
+    assert_eq!(extism_length_unsafe(&mut store, instance, p), 1);
     extism_free(&mut store, instance, p);
 
     // 2 bytes
@@ -180,18 +195,21 @@ fn test_kernel_allocations() {
     assert!(x > 0);
     assert!(x != p);
     assert_eq!(extism_length(&mut store, instance, x), 2);
+    assert_eq!(extism_length_unsafe(&mut store, instance, x), 2);
     extism_free(&mut store, instance, x);
 
     for i in 0..64 {
         let p = extism_alloc(&mut store, instance, 64 - i);
         assert!(p > 0);
         assert_eq!(extism_length(&mut store, instance, p), 64 - i);
+        assert_eq!(extism_length_unsafe(&mut store, instance, p), 64 - i);
         extism_free(&mut store, instance, p);
 
         // should re-use the last allocation
         let q = extism_alloc(&mut store, instance, 64 - i);
         assert_eq!(p, q);
         assert_eq!(extism_length(&mut store, instance, q), 64 - i);
+        assert_eq!(extism_length_unsafe(&mut store, instance, q), 64 - i);
         extism_free(&mut store, instance, q);
     }
 

runtime/src/tests/runtime.rs

@@ -235,43 +235,6 @@ fn test_timeout() {
     assert!(err == "timeout");
 }
 
-fn hello_world_timeout_manager(
-    plugin: &mut CurrentPlugin,
-    inputs: &[Val],
-    outputs: &mut [Val],
-    _user_data: UserData<()>,
-) -> Result<(), Error> {
-    let mgr = plugin.timeout_manager().map(|x| x.add_on_drop());
-    std::thread::sleep(std::time::Duration::from_secs(3));
-    outputs[0] = inputs[0].clone();
-    drop(mgr);
-    Ok(())
-}
-
-#[test]
-fn test_timeout_manager() {
-    let f = Function::new(
-        "hello_world",
-        [PTR],
-        [PTR],
-        UserData::default(),
-        hello_world_timeout_manager,
-    );
-
-    let manifest = Manifest::new([extism_manifest::Wasm::data(WASM)])
-        .with_timeout(std::time::Duration::from_secs(1));
-    let mut plugin = Plugin::new(manifest, [f], true).unwrap();
-
-    let start = std::time::Instant::now();
-    let output: Result<&[u8], Error> = plugin.call("count_vowels", "testing");
-    println!("Result {:?}", output);
-    assert!(output.is_ok());
-    let end = std::time::Instant::now();
-    let time = end - start;
-    println!("Plugin ran for {:?}", time);
-    assert!(time.as_secs() >= 3);
-}
-
 typed_plugin!(pub TestTypedPluginGenerics {
     count_vowels<T: FromBytes<'a>>(&str) -> T
 });

runtime/src/timeout_manager.rs

@@ -1,59 +0,0 @@
-use crate::*;
-
-/// `TimeoutManager` is used to control `Plugin` timeouts from within host functions
-///
-/// It can be used to add or subtract time from a plug-in's timeout. If a plugin is not
-/// configured to have a timeout then this will have no effect.
-pub(crate) struct TimeoutManager {
-    start_time: std::time::Instant,
-    id: uuid::Uuid,
-    tx: std::sync::mpsc::Sender<TimerAction>,
-    cost: f64,
-}
-
-impl TimeoutManager {
-    /// Create a new `TimeoutManager` from the `CurrentPlugin`, this will return `None` if no timeout
-    /// is configured
-    pub fn new(plugin: &CurrentPlugin) -> Option<TimeoutManager> {
-        plugin.manifest.timeout_ms.map(|_| TimeoutManager {
-            start_time: std::time::Instant::now(),
-            id: plugin.id.clone(),
-            tx: Timer::tx(),
-            cost: 0.0,
-        })
-    }
-
-    /// Add the amount of time this value has existed to the configured timeout
-    pub fn add_elapsed(&mut self) -> Result<(), Error> {
-        let cost = self.cost.abs();
-        let d = self.start_time.elapsed().mul_f64(cost);
-        let mut d: timer::ExtendTimeout = d.into();
-        if self.cost.is_sign_negative() {
-            d = -d;
-        }
-        self.tx.send(TimerAction::Extend {
-            id: self.id.clone(),
-            duration: d,
-        })?;
-        self.start_time = std::time::Instant::now();
-        Ok(())
-    }
-
-    pub fn with_cost(mut self, cost: f64) -> Self {
-        self.cost = cost;
-        self
-    }
-}
-
-impl Drop for TimeoutManager {
-    fn drop(&mut self) {
-        let x = self.add_elapsed();
-        if let Err(e) = x {
-            error!(
-                plugin = self.id.to_string(),
-                "unable to extend timeout: {}",
-                e.to_string()
-            );
-        }
-    }
-}

runtime/src/timer.rs

@@ -1,51 +1,17 @@
 use crate::*;
 
-#[derive(Debug)]
-pub(crate) struct ExtendTimeout {
-    duration: std::time::Duration,
-    negative: bool,
-}
-
-impl From<std::time::Duration> for ExtendTimeout {
-    fn from(value: std::time::Duration) -> Self {
-        ExtendTimeout {
-            duration: value,
-            negative: false,
-        }
-    }
-}
-
-impl std::ops::Neg for ExtendTimeout {
-    type Output = ExtendTimeout;
-
-    fn neg(mut self) -> Self::Output {
-        self.negative = !self.negative;
-        self
-    }
-}
-
 pub(crate) enum TimerAction {
     Start {
         id: uuid::Uuid,
         engine: Engine,
         duration: Option<std::time::Duration>,
     },
-    Extend {
-        id: uuid::Uuid,
-        duration: ExtendTimeout,
-    },
     Stop {
         id: uuid::Uuid,
     },
     Cancel {
         id: uuid::Uuid,
     },
-    EnterHost {
-        id: uuid::Uuid,
-    },
-    ExitHost {
-        id: uuid::Uuid,
-    },
     Shutdown,
 }
 
@@ -65,8 +31,6 @@ extern "C" fn cleanup_timer() {
 
 static mut TIMER: std::sync::Mutex<Option<Timer>> = std::sync::Mutex::new(None);
 
-type TimerMap = std::collections::BTreeMap<uuid::Uuid, (Engine, Option<std::time::Instant>)>;
-
 impl Timer {
     pub(crate) fn tx() -> std::sync::mpsc::Sender<TimerAction> {
         let mut timer = match unsafe { TIMER.lock() } {
@@ -85,8 +49,7 @@ impl Timer {
     pub fn init(timer: &mut Option<Timer>) -> std::sync::mpsc::Sender<TimerAction> {
         let (tx, rx) = std::sync::mpsc::channel();
         let thread = std::thread::spawn(move || {
-            let mut plugins = TimerMap::new();
-            let mut in_host = TimerMap::new();
+            let mut plugins = std::collections::BTreeMap::new();
 
             macro_rules! handle {
                 ($x:expr) => {
@@ -107,16 +70,12 @@ impl Timer {
                         TimerAction::Stop { id } => {
                             trace!(plugin = id.to_string(), "handling stop event");
                             plugins.remove(&id);
-                            in_host.remove(&id);
                         }
                         TimerAction::Cancel { id } => {
                             trace!(plugin = id.to_string(), "handling cancel event");
                             if let Some((engine, _)) = plugins.remove(&id) {
                                 engine.increment_epoch();
                             }
-                            if let Some((engine, _)) = in_host.remove(&id) {
-                                engine.increment_epoch();
-                            }
                         }
                         TimerAction::Shutdown => {
                             trace!("Shutting down timer");
@@ -124,42 +83,8 @@ impl Timer {
                                 trace!(plugin = id.to_string(), "handling shutdown event");
                                 engine.increment_epoch();
                             }
-
-                            for (id, (engine, _)) in in_host.iter() {
-                                trace!(plugin = id.to_string(), "handling shutdown event");
-                                engine.increment_epoch();
-                            }
                             return;
                         }
-                        TimerAction::Extend { id, duration } => {
-                            if let Some((_engine, Some(timeout))) = plugins.get_mut(&id) {
-                                let x = if duration.negative {
-                                    timeout.checked_sub(duration.duration)
-                                } else {
-                                    timeout.checked_add(duration.duration)
-                                };
-                                if let Some(t) = x {
-                                    *timeout = t;
-                                } else {
-                                    error!(
-                                        plugin = id.to_string(),
-                                        "unable to extend timeout by {:?}", duration.duration
-                                    );
-                                }
-                            }
-                        }
-                        TimerAction::EnterHost { id } => {
-                            trace!(plugin = id.to_string(), "enter host function");
-                            if let Some(x) = plugins.remove(&id) {
-                                in_host.insert(id, x);
-                            }
-                        }
-                        TimerAction::ExitHost { id } => {
-                            trace!(plugin = id.to_string(), "exit host function");
-                            if let Some(x) = in_host.remove(&id) {
-                                plugins.insert(id, x);
-                            }
-                        }
                     }
                 };
             }
@@ -171,10 +96,6 @@ impl Timer {
                     }
                 }
 
-                for x in rx.try_iter() {
-                    handle!(x)
-                }
-
                 plugins = plugins
                     .into_iter()
                     .filter(|(_k, (engine, end))| {
@@ -188,6 +109,10 @@ impl Timer {
                         true
                     })
                     .collect();
+
+                for x in rx.try_iter() {
+                    handle!(x)
+                }
             }
         });
         *timer = Some(Timer {