fix(core): poll async ops eagerly (#12385)

Currently all async ops are polled lazily, which means that op
initialization code is postponed until control is yielded to the event
loop. This has some weird consequences, e.g.

```js
let listener = Deno.listen(...);
let conn_promise = listener.accept();
listener.close();
// `BadResource` is thrown. A reasonable error would be `Interrupted`.
let conn = await conn_promise;
```

JavaScript promises are expected to be eagerly evaluated. This patch
makes ops actually do that.

1 files changed, 64 insertions, 2 deletions

diff --git a/core/ops.rs b/core/ops.rs
index 80bb30eda..05b91f32f 100644
--- a/core/ops.rs
+++ b/core/ops.rs
@@ -6,6 +6,11 @@ use crate::gotham_state::GothamState;
 use crate::ops_metrics::OpsTracker;
 use crate::resources::ResourceTable;
 use crate::runtime::GetErrorClassFn;
+use futures::future::maybe_done;
+use futures::future::FusedFuture;
+use futures::future::MaybeDone;
+use futures::ready;
+use futures::task::noop_waker;
 use futures::Future;
 use indexmap::IndexMap;
 use rusty_v8 as v8;
@@ -17,10 +22,67 @@ use std::ops::Deref;
 use std::ops::DerefMut;
 use std::pin::Pin;
 use std::rc::Rc;
+use std::task::Context;
+use std::task::Poll;
+
+/// Wrapper around a Future, which causes that Future to be polled immediately.
+/// (Background: ops are stored in a `FuturesUnordered` structure which polls
+/// them, but without the `OpCall` wrapper this doesn't happen until the next
+/// turn of the event loop, which is too late for certain ops.)
+pub struct OpCall<T>(MaybeDone<Pin<Box<dyn Future<Output = T>>>>);
+
+impl<T> OpCall<T> {
+  /// Wraps a future, and polls the inner future immediately.
+  /// This should be the default choice for ops.
+  pub fn eager(fut: impl Future<Output = T> + 'static) -> Self {
+    let boxed = Box::pin(fut) as Pin<Box<dyn Future<Output = T>>>;
+    let mut inner = maybe_done(boxed);
+    let waker = noop_waker();
+    let mut cx = Context::from_waker(&waker);
+    let mut pinned = Pin::new(&mut inner);
+    let _ = pinned.as_mut().poll(&mut cx);
+    Self(inner)
+  }
+
+  /// Wraps a future; the inner future is polled the usual way (lazily).
+  pub fn lazy(fut: impl Future<Output = T> + 'static) -> Self {
+    let boxed = Box::pin(fut) as Pin<Box<dyn Future<Output = T>>>;
+    let inner = maybe_done(boxed);
+    Self(inner)
+  }
+
+  /// Create a future by specifying its output. This is basically the same as
+  /// `async { value }` or `futures::future::ready(value)`.
+  pub fn ready(value: T) -> Self {
+    Self(MaybeDone::Done(value))
+  }
+}
+
+impl<T> Future for OpCall<T> {
+  type Output = T;
+
+  fn poll(
+    self: std::pin::Pin<&mut Self>,
+    cx: &mut std::task::Context<'_>,
+  ) -> std::task::Poll<Self::Output> {
+    let inner = unsafe { &mut self.get_unchecked_mut().0 };
+    let mut pinned = Pin::new(inner);
+    ready!(pinned.as_mut().poll(cx));
+    Poll::Ready(pinned.as_mut().take_output().unwrap())
+  }
+}
+
+impl<F> FusedFuture for OpCall<F>
+where
+  F: Future,
+{
+  fn is_terminated(&self) -> bool {
+    self.0.is_terminated()
+  }
+}
 
 pub type PromiseId = u64;
-pub type OpAsyncFuture =
-  Pin<Box<dyn Future<Output = (PromiseId, OpId, OpResult)>>>;
+pub type OpAsyncFuture = OpCall<(PromiseId, OpId, OpResult)>;
 pub type OpFn = dyn Fn(Rc<RefCell<OpState>>, OpPayload) -> Op + 'static;
 pub type OpId = usize;