avoid sending redundant Status::Started for subtask (#11980)

Previously, we sent this event both immediately after lowering parameters and
when the async-with-callback-lifted export returned.  The latter was redundant,
but it didn't matter in most cases because the second event usually overwrote
the first one such that it was only delivered once.  However, in cases of three
or more components composed together, both events got delivered, and that made
`wit-bindgen` justifiably upset.

This removes the redundant event and adds a three component composition test to
cover the scenario.

Signed-off-by: Joel Dice <[email protected]>

Author: dicej

crates/misc/component-async-tests/tests/scenario/round_trip.rs

@@ -42,6 +42,16 @@ pub async fn async_round_trip_stackless_plus_stackless() -> Result<()> {
     .await
 }
 
+#[tokio::test]
+pub async fn async_round_trip_stackless_plus_stackless_plus_stackless() -> Result<()> {
+    test_round_trip_composed_more(
+        test_programs_artifacts::ASYNC_ROUND_TRIP_STACKLESS_COMPONENT,
+        test_programs_artifacts::ASYNC_ROUND_TRIP_STACKLESS_COMPONENT,
+        test_programs_artifacts::ASYNC_ROUND_TRIP_STACKLESS_COMPONENT,
+    )
+    .await
+}
+
 #[tokio::test]
 async fn async_round_trip_synchronous_plus_stackless() -> Result<()> {
     test_round_trip_composed(
@@ -478,3 +488,27 @@ pub async fn test_round_trip_composed(a: &str, b: &str) -> Result<()> {
     )
     .await
 }
+
+pub async fn test_round_trip_composed_more(a: &str, b: &str, c: &str) -> Result<()> {
+    test_round_trip(
+        &[a, b, c],
+        &[
+            (
+                "hello, world!",
+                "hello, world! - entered guest - entered guest - entered guest - entered host \
+                 - exited host - exited guest - exited guest - exited guest",
+            ),
+            (
+                "¡hola, mundo!",
+                "¡hola, mundo! - entered guest - entered guest - entered guest - entered host \
+                 - exited host - exited guest - exited guest - exited guest",
+            ),
+            (
+                "hi y'all!",
+                "hi y'all! - entered guest - entered guest - entered guest - entered host \
+                 - exited host - exited guest - exited guest - exited guest",
+            ),
+        ],
+    )
+    .await
+}

crates/misc/component-async-tests/tests/scenario/util.rs

@@ -98,12 +98,18 @@ pub async fn make_component(engine: &Engine, components: &[&str]) -> Result<Comp
     }
 
     async fn compile(engine: &Engine, components: &[&str]) -> Result<Vec<u8>> {
-        match components {
-            [component] => engine.precompile_component(&fs::read(component).await?),
-            [a, b] => engine
-                .precompile_component(&compose(&fs::read(a).await?, &fs::read(b).await?).await?),
-            _ => Err(anyhow!("expected one or two paths")),
+        let mut composed = None::<Vec<u8>>;
+        for component in components {
+            let component = fs::read(component).await?;
+            if let Some(other) = composed.take() {
+                composed = Some(compose(&other, &component).await?);
+            } else {
+                composed = Some(component);
+            }
         }
+        engine.precompile_component(
+            &composed.ok_or_else(|| anyhow!("expected at least one component"))?,
+        )
     }
 
     async fn load(engine: &Engine, components: &[&str]) -> Result<Vec<u8>> {

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

@@ -845,7 +845,7 @@ fn handle_guest_call(store: &mut dyn VMStore, call: GuestCall) -> Result<()> {
 
             store.maybe_pop_call_context(call.thread.task)?;
 
-            instance.handle_callback_code(store, call.thread, runtime_instance, code, false)?;
+            instance.handle_callback_code(store, call.thread, runtime_instance, code)?;
 
             store.concurrent_state_mut().guest_thread = old_thread;
             log::trace!("GuestCallKind::DeliverEvent: restored {old_thread:?} as current thread");
@@ -1579,34 +1579,18 @@ impl Instance {
 
     /// Handle the `CallbackCode` returned from an async-lifted export or its
     /// callback.
-    ///
-    /// If `initial_call` is `true`, then the code was received from the
-    /// async-lifted export; otherwise, it was received from its callback.
     fn handle_callback_code(
         self,
         store: &mut StoreOpaque,
         guest_thread: QualifiedThreadId,
         runtime_instance: RuntimeComponentInstanceIndex,
         code: u32,
-        initial_call: bool,
     ) -> Result<()> {
         let (code, set) = unpack_callback_code(code);
 
         log::trace!("received callback code from {guest_thread:?}: {code} (set: {set})");
 
         let state = store.concurrent_state_mut();
-        if !state.get_mut(guest_thread.task)?.returned_or_cancelled() {
-            if initial_call {
-                // Notify any current or future waiters that this subtask has
-                // started.
-                Waitable::Guest(guest_thread.task).set_event(
-                    state,
-                    Some(Event::Subtask {
-                        status: Status::Started,
-                    }),
-                )?;
-            }
-        }
 
         let get_set = |store, handle| {
             if handle == 0 {
@@ -1876,7 +1860,7 @@ impl Instance {
                 // function returns a `i32` result.
                 let code = unsafe { storage[0].assume_init() }.get_i32() as u32;
 
-                self.handle_callback_code(store, guest_thread, callee_instance, code, true)?;
+                self.handle_callback_code(store, guest_thread, callee_instance, code)?;
 
                 Ok(())
             }) as Box<dyn FnOnce(&mut dyn VMStore) -> Result<()> + Send + Sync>