Make VM run microtasks after an event handler throws.

Tested: Test added for fix.
Change-Id: Ifbefb01ef0caf9de80e44c44e5de4cc51bb129cc
Reviewed-on: https://dart-review.googlesource.com/c/sdk/+/489080
Reviewed-by: Martin Kustermann <kustermann@google.com>
Commit-Queue: Lasse Nielsen <lrn@google.com>

runtime/engine/engine.h

@@ -54,6 +54,10 @@ class Engine {
   void HandleMessage(Dart_Isolate isolate);
 
   // Drains the microtasks queue, requires an active isolate.
+  //
+  // If a microtask throws, the error is returned to the caller,
+  // and the queue may still contain more entries.
+  // The caller should continue to drain the queue after handling the error.
   Dart_Handle DrainMicrotasksQueue();
 
   // Sets a callback to be called when Dart_HandleMessage returns an error.

runtime/engine/include/dart_engine.h

@@ -116,6 +116,10 @@ DART_EXPORT void DartEngine_SetHandleMessageErrorCallback(
  * isolate message, but when the engine calls into Dart, it might be
  * required to manually drain the microtasks queue.
  *
+ * If a microtask throws, the error is returned to the caller,
+ * and the queue may still contain more entries.
+ * The caller should continue to drain the queue after handling the error.
+ *
  * \return Dart_Handle invocation result.
  */
 DART_EXPORT Dart_Handle DartEngine_DrainMicrotasksQueue();

runtime/vm/dart_entry.h

@@ -317,6 +317,9 @@ class DartLibraryCalls : public AllStatic {
   static ObjectPtr LookupOpenPorts();
 
   // Returns null on success, an ErrorPtr on failure.
+  //
+  // On an error, the caller should continue to drain the microtask
+  // queue after processing the error.
   static ObjectPtr DrainMicrotaskQueue();
 
   // Runs the `_rehashObjects()` function in `dart:compact_hash`.

runtime/vm/isolate.cc

@@ -1551,11 +1551,17 @@ MessageHandler::MessageStatus IsolateMessageHandler::HandleMessage(
       }
     }
   } else {
-    const Object& msg_handler = Object::Handle(
+    Object& msg_handler = Object::Handle(
         zone, DartLibraryCalls::HandleMessage(message->dest_port(), msg));
-    if (msg_handler.IsError()) {
+    while (msg_handler.IsError()) {
       status = ProcessUnhandledException(Error::Cast(msg_handler));
-    } else if (msg_handler.IsNull()) {
+      if (status == kOK) {
+        msg_handler = DartLibraryCalls::DrainMicrotaskQueue();
+      } else {
+        break;
+      }
+    }
+    if (msg_handler.IsNull()) {
       // If the port has been closed then the message will be dropped at this
       // point. Make sure to post to the delivery failure port in that case.
     } else {

sdk/lib/_internal/vm/lib/isolate_patch.dart

@@ -187,9 +187,8 @@ final class _RawReceivePort implements RawReceivePort {
     if (handler == null) {
       return null;
     }
-    // TODO(floitsch): this relies on the fact that any exception aborts the
-    // VM. Once we have non-fatal global exceptions we need to catch errors
-    // so that we can run the immediate callbacks.
+    // If handler or microtasks throw, the VM will drain microtasks again
+    // after handling the error.
     handler(message);
     _runPendingImmediateCallback();
     return handler;

tests/lib/isolate/throws_in_microtask_test.dart

@@ -0,0 +1,126 @@
+// Copyright (c) 2026, the Dart project authors.  Please see the AUTHORS file
+// for details. All rights reserved. Use of this source code is governed by a
+// BSD-style license that can be found in the LICENSE file.
+
+// Tests that the microtask queue is not broken if a microtask throws.
+
+import 'dart:isolate';
+import "dart:async";
+
+import 'package:expect/async_helper.dart';
+import 'package:expect/expect.dart';
+
+void main() async {
+  asyncStart();
+  // Runs code in a new isolate with `errorsAreFatal` set to `fatal`.
+  // The isolate code:
+  // - schedules two microtasks, each sending an event when they run,
+  //   and then they throw.
+  // - if `timer` is true, also scheduled a timer which reports
+  //   running, but doesn't throw. (To check see that a microtask
+  //   doesn't get postponed to after the timer.)
+  // - runs the code that schedules the microtasks either
+  //   synchronously in the isolate entry point (if `start` is `"sync"`),
+  //   as a microtask (if it's `"microtask"`) or as a zero-duration
+  //   timer (if it's `"timer"`).
+
+  for (var fatal in [true, false]) {
+    for (var timer in [false, true]) {
+      for (var start in ["sync", "microtask", "timer"]) {
+        // ID to keep cases apart.
+        var id = 'ID-${fatal ? 'F' : ''}-${timer ? 'T' : ''}-$start';
+        // Expectation.
+        var expect = [
+          // Always runs once microtask.
+          "M:$id#1", "E:$id#1",
+          if (!fatal) // If not fatal ...
+          ...[
+            // Also runs second microtask,
+            "M:$id#2", "E:$id#2",
+            // and timer if requested, in that order.
+            if (timer) "T:$id",
+          ],
+          "done",
+        ];
+        Expect.listEquals(
+          expect,
+          await test(id, fatal: fatal, timer: timer, start: start),
+          "(fatal: $fatal, timer: $timer, start: $start)",
+        );
+      }
+    }
+  }
+  asyncEnd();
+}
+
+/// Spawns isolate with given [fatal] running test with the remaining parameters.
+///
+/// Collects sent messages and uncaught errors, plus a final `"done"` when
+/// the isolate closes, and returns the list.
+Future<List<String>> test(
+  String id, {
+  required bool fatal,
+  required bool timer,
+  required String start,
+}) async {
+  var log = <String>[];
+  var done = Completer<void>();
+  var port = RawReceivePort();
+  port.handler = (m) {
+    switch (m) {
+      case null:
+        log.add("done");
+        done.complete();
+        port.close();
+      case [var e, _]:
+        log.add("E:$e");
+      case var o:
+        log.add("$o");
+    }
+  };
+  await Isolate.spawn(
+    run,
+    (id, fatal, timer, start, port.sendPort),
+    errorsAreFatal: fatal,
+    onError: port.sendPort,
+    onExit: port.sendPort,
+  );
+  await done.future;
+  return log;
+}
+
+/// Remote isolate entry point.
+///
+/// Unpacks parameters and runs [runTasks] either synchronously
+/// or as a timer event.
+void run((String id, bool fatal, bool timer, String start, SendPort) message) {
+  var (id, fatal, timer, start, output) = message;
+  switch (start) {
+    case "sync":
+      runTasks(id, timer, output);
+    case "microtask":
+      Zone.current.scheduleMicrotask(() {
+        runTasks(id, timer, output);
+      });
+    case "timer":
+      Zone.current.createTimer(Duration.zero, () {
+        runTasks(id, timer, output);
+      });
+  }
+}
+
+void runTasks(String id, bool timer, SendPort output) {
+  Zone.current.scheduleMicrotask(() {
+    output.send("M:$id#1");
+    throw "$id#1";
+  });
+  Zone.current.scheduleMicrotask(() {
+    output.send("M:$id#2");
+    throw "$id#2";
+  });
+  if (timer) {
+    Zone.current.createTimer(Duration.zero, () {
+      output.send("T:$id");
+    });
+  }
+}