fix(core): Ensure we don't lose the waker when polling an empty JoinSet (#19655)

This is a reproduction and fix for a very obscure bug where the Deno
runtime locks up we end up polling an empty JoinSet and attempt to
resolve ops after-the-fact. There's a small footgun in the JoinSet API
where polling it while empty returns Ready(None), which means that it
never holds on to the waker. This means that if we aren't testing for
this particular return value and don't stash the waker ourselves for a
future async op to eventually queue, we can end up losing the waker
entirely and the op wakes up, notifies tokio, which notifies the
JoinSet, which then has nobody to notify 😢.

Co-authored-by: Luca Casonato <hello@lcas.dev>
Co-authored-by: Bartek Iwańczuk <biwanczuk@gmail.com>

1 files changed, 7 insertions, 16 deletions

diff --git a/core/runtime/ops.rs b/core/runtime/ops.rs
index 5e51414d3..1c871cda0 100644
--- a/core/runtime/ops.rs
+++ b/core/runtime/ops.rs
@@ -25,10 +25,11 @@ pub fn queue_fast_async_op<R: serde::Serialize + 'static>(
     state.get_error_class_fn
   };
   let fut = op.map(|result| crate::_ops::to_op_result(get_class, result));
-  // SAFETY: this is guaranteed to be running on a current-thread executor
-  ctx.context_state.borrow_mut().pending_ops.spawn(unsafe {
-    crate::task::MaskFutureAsSend::new(OpCall::new(ctx, promise_id, fut))
-  });
+  ctx
+    .context_state
+    .borrow_mut()
+    .pending_ops
+    .spawn(OpCall::new(ctx, promise_id, fut));
 }
 
 #[inline]
@@ -123,12 +124,7 @@ pub fn queue_async_op<'s>(
     Poll::Pending => {}
     Poll::Ready(mut res) => {
       if deferred {
-        ctx
-          .context_state
-          .borrow_mut()
-          .pending_ops
-          // SAFETY: this is guaranteed to be running on a current-thread executor
-          .spawn(unsafe { crate::task::MaskFutureAsSend::new(ready(res)) });
+        ctx.context_state.borrow_mut().pending_ops.spawn(ready(res));
         return None;
       } else {
         ctx.state.borrow_mut().tracker.track_async_completed(ctx.id);
@@ -137,12 +133,7 @@ pub fn queue_async_op<'s>(
     }
   }
 
-  ctx
-    .context_state
-    .borrow_mut()
-    .pending_ops
-    // SAFETY: this is guaranteed to be running on a current-thread executor
-    .spawn(unsafe { crate::task::MaskFutureAsSend::new(pinned) });
+  ctx.context_state.borrow_mut().pending_ops.spawn(pinned);
   None
 }