Add futures-util and use their now_or_never

Author: gefjon

Cargo.lock

@@ -182,6 +182,7 @@ fs_extra = "1.3.0"
 fs2 = "0.4.3"
 futures = "0.3"
 futures-channel = "0.3"
+futures-util = "0.3"
 getrandom02 = { package = "getrandom", version = "0.2" }
 glob = "0.3.1"
 hashbrown = { version = "0.15", default-features = false, features = ["equivalent", "inline-more"] }

Cargo.toml

@@ -58,6 +58,7 @@ enum-map.workspace = true
 flate2.workspace = true
 fs2.workspace = true
 futures.workspace = true
+futures-util.workspace = true
 hashbrown = { workspace = true, features = ["rayon", "serde"] }
 hex.workspace = true
 hostname.workspace = true

crates/core/Cargo.toml

@@ -6,8 +6,8 @@ use crate::energy::ReducerBudget;
 use crate::host::instance_env::InstanceEnv;
 use crate::host::wasm_common::module_host_actor::{DescribeError, InitializationError};
 use crate::host::wasm_common::*;
-use crate::util::poll_once_executor::poll_once;
 use crate::util::string_from_utf8_lossy_owned;
+use futures_util::FutureExt;
 use spacetimedb_primitives::errno::HOST_CALL_FAILURE;
 use wasmtime::{
     AsContext, AsContextMut, ExternType, Instance, InstancePre, Linker, Store, TypedFunc, WasmBacktrace, WasmParams,
@@ -94,15 +94,16 @@ const CALL_FAILURE: i32 = HOST_CALL_FAILURE.get() as i32;
 /// Our Wasmtime is configured for `async` execution, and will panic if we use the non-async [`TypedFunc::call`].
 /// The `async` config is necessary to allow procedures to suspend, e.g. when making HTTP calls or acquiring transactions.
 /// However, most of the WASM we execute, incl. reducers and startup functions, should never block/yield.
-/// Rather than crossing our fingers and trusting, we run [`TypedFunc::call_async`] in [`crate::util::poll_once_executor`],
-/// a custom "async executor" which invokes [`std::task::Future::poll`] exactly once.
+/// Rather than crossing our fingers and trusting, we run [`TypedFunc::call_async`] in [`FutureExt::now_or_never`],
+/// an "async executor" which invokes [`std::task::Future::poll`] exactly once.
 fn call_sync_typed_func<Args: WasmParams, Ret: WasmResults>(
     typed_func: &TypedFunc<Args, Ret>,
     store: &mut Store<WasmInstanceEnv>,
     args: Args,
 ) -> anyhow::Result<Ret> {
     let fut = typed_func.call_async(store, args);
-    poll_once(fut).expect("`call_async` of supposedly synchronous WASM function returned `Poll::Pending`")
+    fut.now_or_never()
+        .expect("`call_async` of supposedly synchronous WASM function returned `Poll::Pending`")
 }
 
 impl module_host_actor::WasmInstancePre for WasmtimeModule {
@@ -113,7 +114,8 @@ impl module_host_actor::WasmInstancePre for WasmtimeModule {
         let mut store = Store::new(self.module.module().engine(), env);
         let instance_fut = self.module.instantiate_async(&mut store);
 
-        let instance = poll_once(instance_fut)
+        let instance = instance_fut
+            .now_or_never()
             .expect("`instantiate_async` did not immediately return `Ready`")
             .map_err(InitializationError::Instantiation)?;
 

crates/core/src/host/wasmtime/wasmtime_module.rs

@@ -9,7 +9,6 @@ pub mod prometheus_handle;
 pub mod jobs;
 pub mod notify_once;
 pub mod slow;
-pub mod poll_once_executor;
 
 // TODO: use String::from_utf8_lossy_owned once stabilized
 pub(crate) fn string_from_utf8_lossy_owned(v: Vec<u8>) -> String {