avoid redundant CallStarted or CallReturned events (#104)

If we're returning `Status::Started` or `Status::Returned`, don't _also_ deliver
redundant events.

Fixes #103

Signed-off-by: Joel Dice <joel.dice@fermyon.com>

Author: dicej

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

@@ -1675,8 +1675,12 @@ impl ComponentInstance {
         let mut status = if task.lower_params.is_some() {
             Status::Starting
         } else if task.lift_result.is_some() {
+            let event = Waitable::Guest(guest_task).take_event(self)?;
+            assert!(matches!(event, Some(Event::CallStarted)));
             Status::Started
         } else {
+            let event = Waitable::Guest(guest_task).take_event(self)?;
+            assert!(matches!(event, Some(Event::CallReturned)));
             Status::Returned
         };
 
@@ -1689,7 +1693,7 @@ impl ComponentInstance {
                 self.waitable_tables()[caller_instance]
                     .insert(guest_task.rep(), WaitableState::GuestTask)?
             } else {
-                let caller = if let Caller::Guest { task, .. } = &task.caller {
+                let caller = if let Caller::Guest { task, .. } = &self.get(guest_task)?.caller {
                     *task
                 } else {
                     unreachable!()
@@ -3108,6 +3112,15 @@ impl Waitable {
         })
     }
 
+    fn take_event(&self, instance: &mut ComponentInstance) -> Result<Option<Event>> {
+        let common = self.common(instance)?;
+        let event = common.event.take();
+        if let Some(set) = self.common(instance)?.set {
+            instance.get_mut(set)?.ready.remove(self);
+        }
+        Ok(event)
+    }
+
     fn mark_ready(&self, instance: &mut ComponentInstance) -> Result<()> {
         if let Some(set) = self.common(instance)?.set {
             instance.get_mut(set)?.ready.insert(*self);