fix(scheduler): improve error recovery

packages/runtime-core/__tests__/scheduler.spec.ts

@@ -1,6 +1,7 @@
 import {
   type SchedulerJob,
   SchedulerJobFlags,
+  flushOnAppMount,
   flushPostFlushCbs,
   flushPreFlushCbs,
   nextTick,
@@ -501,6 +502,67 @@ describe('scheduler', () => {
     await nextTick()
   })
 
+  test('flushOnAppMount error recovery', () => {
+    const err = new Error('test')
+    let shouldThrow = true
+
+    const job1: SchedulerJob = vi.fn(() => {
+      if (shouldThrow) {
+        shouldThrow = false
+        throw err
+      }
+    })
+
+    queuePostFlushCb(job1)
+
+    try {
+      flushOnAppMount()
+    } catch (e: any) {
+      expect(e).toBe(err)
+    }
+
+    expect(job1).toHaveBeenCalledTimes(1)
+
+    queuePostFlushCb(job1)
+
+    flushOnAppMount()
+
+    expect(job1).toHaveBeenCalledTimes(2)
+  })
+
+  test('pre jobs can be re-queued after an error', () => {
+    const err = new Error('test')
+    let shouldThrow = true
+
+    const job1: SchedulerJob = vi.fn(() => {
+      if (shouldThrow) {
+        shouldThrow = false
+        throw err
+      }
+    })
+    const job2: SchedulerJob = vi.fn()
+
+    queueJob(job1, undefined, true)
+    queueJob(job2, undefined, true)
+
+    try {
+      flushPreFlushCbs()
+    } catch (e: any) {
+      expect(e).toBe(err)
+    }
+
+    expect(job1).toHaveBeenCalledTimes(1)
+    expect(job2).toHaveBeenCalledTimes(0)
+
+    queueJob(job1, undefined, true)
+    queueJob(job2, undefined, true)
+
+    flushPreFlushCbs()
+
+    expect(job1).toHaveBeenCalledTimes(2)
+    expect(job2).toHaveBeenCalledTimes(1)
+  })
+
   test('jobs can be re-queued after an error', async () => {
     const err = new Error('test')
     let shouldThrow = true
@@ -537,6 +599,58 @@ describe('scheduler', () => {
     expect(job2).toHaveBeenCalledTimes(1)
   })
 
+  test('post jobs can be re-queued after an error', async () => {
+    const err = new Error('test')
+    let shouldThrow = true
+
+    const job1: SchedulerJob = vi.fn(() => {
+      if (shouldThrow) {
+        shouldThrow = false
+        throw err
+      }
+    })
+    const job2: SchedulerJob = vi.fn()
+
+    queuePostFlushCb(job1, 1)
+    queuePostFlushCb(job2, 2)
+
+    try {
+      await nextTick()
+    } catch (e: any) {
+      expect(e).toBe(err)
+    }
+
+    expect(job1).toHaveBeenCalledTimes(1)
+    expect(job2).toHaveBeenCalledTimes(0)
+
+    queuePostFlushCb(job1, 1)
+    queuePostFlushCb(job2, 2)
+
+    await nextTick()
+
+    expect(job1).toHaveBeenCalledTimes(2)
+    expect(job2).toHaveBeenCalledTimes(1)
+  })
+
+  test('post job error should not leave newly queued main jobs pending', async () => {
+    const calls: string[] = []
+
+    const job2: SchedulerJob = () => {
+      calls.push('job2')
+    }
+
+    const job1: SchedulerJob = () => {
+      queueJob(job2, 2)
+      throw new Error('test')
+    }
+
+    queuePostFlushCb(job1, 1)
+
+    await expect(nextTick()).rejects.toThrow('test')
+    await nextTick()
+    expect(calls).toEqual(['job2'])
+  })
+
   test('should prevent self-triggering jobs by default', async () => {
     let count = 0
     const job = () => {
@@ -634,6 +748,31 @@ describe('scheduler', () => {
     expect(job2).toHaveBeenCalledTimes(2)
   })
 
+  test(`recursive post jobs can't be re-queued by other jobs`, async () => {
+    let recurse = true
+
+    const job1: SchedulerJob = () => {
+      if (recurse) {
+        // job2 is already queued, so this shouldn't do anything
+        queuePostFlushCb(job2, 2)
+        recurse = false
+      }
+    }
+    const job2: SchedulerJob = vi.fn(() => {
+      if (recurse) {
+        queuePostFlushCb(job1, 1)
+        queuePostFlushCb(job2, 2)
+      }
+    })
+    job2.flags = SchedulerJobFlags.ALLOW_RECURSE
+
+    queuePostFlushCb(job2, 2)
+
+    await nextTick()
+
+    expect(job2).toHaveBeenCalledTimes(2)
+  })
+
   test('jobs are de-duplicated correctly when calling flushPreFlushCbs', async () => {
     let recurse = true
 

packages/runtime-core/src/scheduler.ts

@@ -140,6 +140,11 @@ const doFlushJobs = () => {
     flushJobs()
   } catch (e) {
     currentFlushPromise = null
+    // If a nested pre/post flush throws after queueing more work, defer the
+    // leftovers to a fresh microtask
+    if (jobsLength || postJobs.length) {
+      queueFlush()
+    }
     throw e
   }
 }
@@ -195,9 +200,12 @@ export function flushPreFlushCbs(
     if (cb.flags! & SchedulerJobFlags.ALLOW_RECURSE) {
       cb.flags! &= ~SchedulerJobFlags.QUEUED
     }
-    cb()
-    if (!(cb.flags! & SchedulerJobFlags.ALLOW_RECURSE)) {
-      cb.flags! &= ~SchedulerJobFlags.QUEUED
+    try {
+      cb()
+    } finally {
+      if (!(cb.flags! & SchedulerJobFlags.ALLOW_RECURSE)) {
+        cb.flags! &= ~SchedulerJobFlags.QUEUED
+      }
     }
   }
 }
@@ -218,25 +226,34 @@ export function flushPostFlushCbs(seen?: CountMap): void {
       seen = seen || new Map()
     }
 
-    while (postFlushIndex < activePostJobs.length) {
-      const cb = activePostJobs[postFlushIndex++]
-      if (__DEV__ && checkRecursiveUpdates(seen!, cb)) {
-        continue
-      }
-      if (cb.flags! & SchedulerJobFlags.ALLOW_RECURSE) {
-        cb.flags! &= ~SchedulerJobFlags.QUEUED
-      }
-      if (!(cb.flags! & SchedulerJobFlags.DISPOSED)) {
-        try {
-          cb()
-        } finally {
+    try {
+      while (postFlushIndex < activePostJobs.length) {
+        const cb = activePostJobs[postFlushIndex++]
+        if (__DEV__ && checkRecursiveUpdates(seen!, cb)) {
+          continue
+        }
+        if (cb.flags! & SchedulerJobFlags.ALLOW_RECURSE) {
           cb.flags! &= ~SchedulerJobFlags.QUEUED
         }
+        if (!(cb.flags! & SchedulerJobFlags.DISPOSED)) {
+          try {
+            cb()
+          } finally {
+            if (!(cb.flags! & SchedulerJobFlags.ALLOW_RECURSE)) {
+              cb.flags! &= ~SchedulerJobFlags.QUEUED
+            }
+          }
+        }
+      }
+    } finally {
+      // If there was an error we still need to clear the QUEUED flags
+      while (postFlushIndex < activePostJobs.length) {
+        activePostJobs[postFlushIndex++].flags! &= ~SchedulerJobFlags.QUEUED
       }
-    }
 
-    activePostJobs = null
-    postFlushIndex = 0
+      activePostJobs = null
+      postFlushIndex = 0
+    }
   }
 }
 
@@ -247,9 +264,12 @@ let isFlushing = false
 export function flushOnAppMount(instance?: GenericComponentInstance): void {
   if (!isFlushing) {
     isFlushing = true
-    flushPreFlushCbs(instance)
-    flushPostFlushCbs()
-    isFlushing = false
+    try {
+      flushPreFlushCbs(instance)
+      flushPostFlushCbs()
+    } finally {
+      isFlushing = false
+    }
   }
 }
 
@@ -297,10 +317,11 @@ function flushJobs(seen?: CountMap) {
 
     flushPostFlushCbs(seen)
 
-    currentFlushPromise = null
     // If new jobs have been added to either queue, keep flushing
     if (jobsLength || postJobs.length) {
       flushJobs(seen)
+    } else {
+      currentFlushPromise = null
     }
   }
 }