cm-async: Remove assertion when disposing a task (#13092)

* cm-async: Remove assertion when disposing a task

This commit removes a now-older assertion when disposing of a guest task
that all threads are empty. This assertion is hit in a situation where
`TypedFunc::call_async` is used where the guest can't be entered due to
backpressure. The assertion is a bit dated now and originally this
helper function handled reparenting, but that's all gone now too. This
commit removes the assertion entirely meaning that in this situation the
guest thread will "leak" into the host, but the store is locked down at
that point anyway, so this is in theory reasonable.

* Fix miri testing

Author: alexcrichton

crates/wasmtime/src/runtime/component/concurrent.rs

@@ -1578,7 +1578,7 @@ impl StoreOpaque {
         let state = self.concurrent_state_mut();
         let task = state.get_mut(thread.task)?;
         if task.ready_to_delete() {
-            state.delete(thread.task)?.dispose(state)?;
+            state.delete(thread.task)?;
         }
 
         Ok(())
@@ -4612,12 +4612,6 @@ impl GuestTask {
             async_function,
         })
     }
-
-    /// Dispose of this guest task.
-    fn dispose(self, _state: &mut ConcurrentState) -> Result<()> {
-        assert!(self.threads.is_empty());
-        Ok(())
-    }
 }
 
 impl TableDebug for GuestTask {
@@ -4778,7 +4772,7 @@ impl Waitable {
             }
             Self::Guest(task) => {
                 log::trace!("delete guest task {task:?}");
-                state.delete(*task)?.dispose(state)?;
+                state.delete(*task)?;
             }
             Self::Transmit(task) => {
                 state.delete(*task)?;
@@ -5335,13 +5329,14 @@ impl TaskId {
     /// and delete the task when all threads are done.
     pub(crate) fn host_future_dropped<T>(&self, store: StoreContextMut<T>) -> Result<()> {
         let task = store.0.concurrent_state_mut().get_mut(self.task)?;
-        if !task.already_lowered_parameters() {
-            Waitable::Guest(self.task).delete_from(store.0.concurrent_state_mut())?
+        let delete = if !task.already_lowered_parameters() {
+            true
         } else {
             task.host_future_state = HostFutureState::Dropped;
-            if task.ready_to_delete() {
-                Waitable::Guest(self.task).delete_from(store.0.concurrent_state_mut())?
-            }
+            task.ready_to_delete()
+        };
+        if delete {
+            Waitable::Guest(self.task).delete_from(store.0.concurrent_state_mut())?
         }
         Ok(())
     }

tests/all/component_model/async.rs

@@ -1324,3 +1324,60 @@ async fn bytes_stream_producer() -> Result<()> {
 
     Ok(())
 }
+
+#[tokio::test]
+#[cfg_attr(miri, ignore)]
+async fn drop_deadlocked_typed_future() -> Result<()> {
+    let mut config = Config::new();
+    config.wasm_component_model_async(true);
+
+    let engine = Engine::new(&config)?;
+    let component = Component::new(
+        &engine,
+        r#"
+            (component
+              (core func $backpressure_inc (canon backpressure.inc))
+
+              (core module $m
+                (import "" "backpressure.inc" (func $backpressure_inc))
+                (func (export "set-backpressure") (call $backpressure_inc))
+                (func (export "target") (result i32) unreachable)
+                (func (export "callback") (param i32 i32 i32) (result i32) unreachable)
+              )
+
+              (core instance $i (instantiate $m
+                (with "" (instance (export "backpressure.inc" (func $backpressure_inc))))
+              ))
+
+              (func (export "set-backpressure")
+                (canon lift (core func $i "set-backpressure")))
+
+              (func (export "target")
+                (canon lift (core func $i "target") async (callback (func $i "callback"))))
+            )
+        "#,
+    )?;
+    let mut store = Store::new(&engine, ());
+    let linker = Linker::<()>::new(&engine);
+    let instance = linker.instantiate_async(&mut store, &component).await?;
+
+    // Force the instance to deadlock the next call by turning on backpressure
+    // meaning it can't enter the instance.
+    instance
+        .get_typed_func::<(), ()>(&mut store, "set-backpressure")?
+        .call_async(&mut store, ())
+        .await?;
+
+    // This call should fail due to deadlock since no progress is possible. When
+    // the future is dropped that shouldn't cause anything to go awry...
+    let result = instance
+        .get_typed_func::<(), ()>(&mut store, "target")?
+        .call_async(&mut store, ())
+        .await;
+
+    assert!(result.is_err(), "expected an error, got Ok");
+    let trap = result.unwrap_err().downcast::<Trap>()?;
+    assert_eq!(trap, Trap::AsyncDeadlock);
+
+    Ok(())
+}