Moving a ready subtask to another waitable set leaves a stale DeliverEvent: "BUG: delivering non-present event" #14878

When a component's callback-lifted task is waiting on a waitable set and one of the set's subtasks becomes ready, Wasmtime queues a DeliverEvent for that set. If another task then moves the ready subtask to a different set with waitable.join before the queued delivery runs, the delivery still runs. The old set is now empty, get_event returns None, and handle_guest_call hits bail_bug!("delivering non-present event"). Debug builds panic. Release builds return the error and the store is unusable.

Moving a waitable between async waitable sets is allowed: waitable.join handles it and traps only for synchronous-call sets. So I believe this is reachable from valid guests. It shows up in practice with wit-bindgen's inter-task-wakeup runtime, which unregisters and re-registers a subtask's waitable when a future created in one export task is awaited in another.

Versions

• wasmtime 49.0.3, 49.0.4 and 50.0.0-rc.1 (all reproduce it; concurrent.rs is identical between 49.0.3 and 49.0.4)
• features: std, runtime, cranelift, component-model-async, wat
• tokio 1.53.2, rustc 1.98.0, Linux x86_64

Reproduction

The component has one async import hold and two async exports with callbacks. start calls hold, joins the subtask to a new set and waits on it. move re-joins the same subtask handle to 0, then to a new set of its own, and waits.

The host releases hold and queues move before the ready event reaches start's callback. Run with --control for the order where move runs before hold is released; that order completes normally.

repro.wat (at the crate root, next to Cargo.toml):

(component
  (type $unit (func async))
  (import "hold" (func $hold (type $unit)))
  (core func $hold (canon lower (func $hold) async))
  (core func $new (canon waitable-set.new))
  (core func $join (canon waitable.join))
  (core func $done (canon task.return))
  (core func $drop (canon subtask.drop))
  (core module $m
    (import "" "hold" (func $hold (result i32)))
    (import "" "new" (func $new (result i32)))
    (import "" "join" (func $join (param i32 i32)))
    (import "" "done" (func $done))
    (import "" "drop" (func $drop (param i32)))
    (global $task (mut i32) (i32.const 0))
    (func (export "start") (result i32) (local $set i32)
      call $hold i32.const 4 i32.shr_u global.set $task
      call $new local.tee $set
      global.get $task local.get $set call $join
      i32.const 4 i32.shl i32.const 2 i32.or)
    (func (export "move") (result i32) (local $set i32)
      global.get $task i32.const 0 call $join
      call $new local.set $set
      global.get $task local.get $set call $join
      local.get $set i32.const 4 i32.shl i32.const 2 i32.or)
    (func (export "callback") (param i32 i32 i32) (result i32)
      local.get 1 call $drop
      call $done
      i32.const 0))
  (core instance $imports
    (export "hold" (func $hold)) (export "new" (func $new))
    (export "join" (func $join)) (export "done" (func $done))
    (export "drop" (func $drop)))
  (core instance $i (instantiate $m (with "" (instance $imports))))
  (func (export "start") (type $unit)
    (canon lift (core func $i "start") async (callback (core func $i "callback"))))
  (func (export "move") (type $unit)
    (canon lift (core func $i "move") async (callback (core func $i "callback")))))

Cargo.toml:

[package]
name = "event-delivery-repro"
version = "0.0.0"
edition = "2024"

[dependencies]
wasmtime = { version = "=49.0.4", default-features = false, features = ["std", "runtime", "cranelift", "component-model-async", "wat"] }
tokio = { version = "=1.53.2", default-features = false, features = ["rt", "sync"] }

src/main.rs (49.x; for 50.0.0-rc.1 replace the two AccessorTask structs with the equivalent async closures, a.spawn(async move |a: &Accessor<()>| start.call_concurrent(a, ()).await) and a.spawn(async |_: &Accessor<()>| Ok(()))):

use std::{
    future::Future,
    pin::pin,
    task::{Context, Waker},
};
use tokio::sync::oneshot;
use wasmtime::{
    Config, Engine, Store,
    component::{Accessor, AccessorTask, Component, Linker, TypedFunc},
};
struct Start(TypedFunc<(), ()>);
impl AccessorTask<()> for Start {
    async fn run(self, a: &Accessor<()>) -> wasmtime::Result<()> {
        self.0.call_concurrent(a, ()).await
    }
}
struct Noop;
impl AccessorTask<()> for Noop {
    async fn run(self, _: &Accessor<()>) -> wasmtime::Result<()> {
        Ok(())
    }
}
fn main() -> wasmtime::Result<()> {
    let stale = !std::env::args().any(|arg| arg == "--control");
    let runtime = tokio::runtime::Builder::new_current_thread()
        .build()
        .unwrap();
    runtime.block_on(run(stale))
}
async fn run(stale: bool) -> wasmtime::Result<()> {
    let mut cfg = Config::new();
    cfg.wasm_component_model_async(true);
    let engine = Engine::new(&cfg)?;
    let component = Component::new(&engine, include_str!("../repro.wat"))?;
    let mut linker = Linker::<()>::new(&engine);
    let (entered, ready) = oneshot::channel();
    let (release, released) = oneshot::channel();
    let (completed, completion) = oneshot::channel();
    let holder = std::sync::Mutex::new(Some((entered, released, completed)));
    linker
        .root()
        .func_wrap_concurrent("hold", move |_, (): ()| {
            let (entered, released, completed) = holder.lock().unwrap().take().unwrap();
            Box::pin(async move {
                entered.send(()).unwrap();
                released.await.unwrap();
                completed.send(()).unwrap();
                Ok(())
            })
        })?;
    let mut store = Store::new(&engine, ());
    let instance = linker.instantiate_async(&mut store, &component).await?;
    let start = instance.get_typed_func::<(), ()>(&mut store, "start")?;
    let moving = instance.get_typed_func::<(), ()>(&mut store, "move")?;
    store
        .run_concurrent(async |a| -> wasmtime::Result<()> {
            a.spawn(Start(start))?;
            ready.await.unwrap();
            if stale {
                // Keep one queued item ahead of host completion so the second
                // export can be queued before the ready event is delivered.
                a.spawn(Noop)?;
                release.send(()).unwrap();
                completion.await.unwrap();
                moving.call_concurrent(a, ()).await
            } else {
                // Move the still-pending waitable before its host result is ready.
                let mut call = pin!(moving.call_concurrent(a, ()));
                assert!(
                    call.as_mut()
                        .poll(&mut Context::from_waker(Waker::noop()))
                        .is_pending()
                );
                release.send(()).unwrap();
                call.await
            }
        })
        .await??;
    Ok(())
}

Expected

Both cargo run and cargo run -- --control exit 0. In the fault order, start's callback either gets no stale delivery once its only ready waitable has left the set, or it keeps waiting. move's callback receives the subtask's Returned event.

Actual

cargo run -- --control exits 0 (5/5, debug and release). cargo run fails every time (5/5, debug and release):

Debug (exit 101):

thread 'main' panicked at .../wasmtime-49.0.4/src/runtime/bug.rs:63:13:
BUG: delivering non-present event

Release (exit 1):

Error: BUG: delivering non-present event
location: .../wasmtime-49.0.4/src/runtime/component/concurrent.rs:948
version: 49.0.4

50.0.0-rc.1 gives the same result, at concurrent.rs:983.

Where it comes from

Line numbers are for 49.0.4 (src/runtime/component/concurrent.rs):

1. The subtask's host future finishes. set_event → Waitable::mark_ready (5461-5530) inserts the waitable into set.ready, pops the set's waiter (5466) and pushes a high-priority GuestCall { kind: DeliverEvent { set: Some(set) } } (5513-5526).
2. Before that item runs, the move task calls waitable.join. Waitable::join (5387-5416) removes the waitable from the old set's ready (5399) and marks it ready in the new set. The queued DeliverEvent for the old set stays in the queue, and the old set's waiter is not restored.
3. The stale item runs. handle_guest_call calls Instance::get_event (2595-2634). The old set's ready.pop_first() (2618) is empty, so it returns None, and line 948 raises bail_bug!("delivering non-present event").

When the old set happens to hold another ready waitable at step 3, the stale delivery hands that one over instead and nothing fails. That makes the bug intermittent in larger programs. The reproduction above leaves the old set empty, so it fails every time.

A possible fix: when join takes the last ready waitable out of a set whose callback delivery is already queued, either drop that queued delivery and restore the waiter, or have DeliverEvent treat an empty set as "keep waiting" instead of a bug.