Completed futures should not park

vertx-async-await-incubator/src/main/java/io/vertx/await/Async.java

@@ -1,16 +1,16 @@
 package io.vertx.await;
 
 import io.netty.channel.EventLoop;
+import io.vertx.await.impl.DefaultScheduler;
 import io.vertx.await.impl.EventLoopScheduler;
 import io.vertx.await.impl.Scheduler;
 import io.vertx.await.impl.VirtualThreadContext;
-import io.vertx.await.impl.DefaultScheduler;
 import io.vertx.core.Future;
 import io.vertx.core.Handler;
 import io.vertx.core.Vertx;
 import io.vertx.core.impl.ContextInternal;
 
-import java.util.concurrent.CompletionStage;
+import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.locks.Lock;
 
 public class Async {
@@ -58,7 +58,7 @@ public static void lock(Lock lock) {
     ctx.lock(lock);
   }
 
-  public static <T> T await(CompletionStage<T> future) {
+  public static <T> T await(CompletableFuture<T> future) {
     VirtualThreadContext ctx = virtualThreadContext();
     return ctx.await(future);
   }

vertx-async-await-incubator/src/main/java/io/vertx/await/impl/VirtualThreadContext.java

@@ -158,7 +158,14 @@ public void lock(Lock lock) {
     }
   }
 
-  public <T> T await(CompletionStage<T> fut) {
+  public <T> T await(CompletableFuture<T> fut) {
+    if (fut.state() == java.util.concurrent.Future.State.SUCCESS) {
+      return fut.resultNow();
+    }
+    if (fut.state() == java.util.concurrent.Future.State.FAILED) {
+      throwAsUnchecked(fut.exceptionNow());
+      return null;
+    }
     inThread.remove();
     Consumer<Runnable> cont = scheduler.unschedule();
     CompletableFuture<T> latch = new CompletableFuture<>();

vertx-async-await-incubator/src/test/java/io/vertx/await/VirtualThreadContextTestBase.java

@@ -12,6 +12,7 @@
 
 import java.util.ArrayList;
 import java.util.List;
+import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.locks.ReentrantLock;
 
 public abstract class VirtualThreadContextTestBase extends VertxTestBase {
@@ -109,14 +110,49 @@ public void testAwaitCompoundFuture() {
     await();
   }
 
+  @Test
+  public void testImmediateCompletedFuture() {
+    var flag = new AtomicInteger();
+    async.run(v -> {
+      var completed = Future.succeededFuture("HELLO");
+      vertx.getOrCreateContext().runOnContext(v2 -> {
+        assertEquals(2, flag.incrementAndGet());
+        testComplete();
+      });
+      Async.await(completed);
+      assertEquals(1, flag.incrementAndGet());
+    });
+    await();
+  }
+
+  @Test
+  public void testImmediateFailedFuture() {
+    var flag = new AtomicInteger();
+    async.run(v -> {
+      var completed = Future.failedFuture("FAILED");
+      vertx.getOrCreateContext().runOnContext(v2 -> {
+        assertEquals(2, flag.incrementAndGet());
+        testComplete();
+      });
+      try {
+        Async.await(completed);
+      } catch (Throwable t) {
+        assertEquals(1, flag.incrementAndGet());
+        return;
+      }
+      fail("shouldn't reach");
+    });
+    await();
+  }
+
   @Test
   public void testDuplicateUseSameThread() {
     int num = 1000;
     waitFor(num);
     async.run(v -> {
       ContextInternal context = (ContextInternal) vertx.getOrCreateContext();
       Thread th = Thread.currentThread();
-      for (int i = 0;i < num;i++) {
+      for (int i = 0; i < num; i++) {
         ContextInternal duplicate = context.duplicate();
         duplicate.runOnContext(v2 -> {
           // assertSame(th, Thread.currentThread());
@@ -135,7 +171,7 @@ public void testDuplicateConcurrentAwait() {
       ContextInternal context = (ContextInternal) vertx.getOrCreateContext();
       Object lock = new Object();
       List<Promise<Void>> list = new ArrayList<>();
-      for (int i = 0;i < num;i++) {
+      for (int i = 0; i < num; i++) {
         ContextInternal duplicate = context.duplicate();
         duplicate.runOnContext(v2 -> {
           Promise<Void> promise = duplicate.promise();