diff --git a/packages/runtime-core/__tests__/scheduler.spec.ts b/packages/runtime-core/__tests__/scheduler.spec.ts
index 5c5b04673ab..cf961667e68 100644
--- a/packages/runtime-core/__tests__/scheduler.spec.ts
+++ b/packages/runtime-core/__tests__/scheduler.spec.ts
@@ -517,6 +517,45 @@ describe('scheduler', () => {
     await nextTick()
   })
 
+  test('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
+      }
+    })
+    job1.id = 1
+
+    const job2: SchedulerJob = vi.fn()
+    job2.id = 2
+
+    queueJob(job1)
+    queueJob(job2)
+
+    try {
+      await nextTick()
+    } catch (e: any) {
+      expect(e).toBe(err)
+    }
+    expect(
+      `Unhandled error during execution of scheduler flush`,
+    ).toHaveBeenWarned()
+
+    expect(job1).toHaveBeenCalledTimes(1)
+    expect(job2).toHaveBeenCalledTimes(0)
+
+    queueJob(job1)
+    queueJob(job2)
+
+    await nextTick()
+
+    expect(job1).toHaveBeenCalledTimes(2)
+    expect(job2).toHaveBeenCalledTimes(1)
+  })
+
   test('should prevent self-triggering jobs by default', async () => {
     let count = 0
     const job = () => {
@@ -558,6 +597,113 @@ describe('scheduler', () => {
     expect(count).toBe(5)
   })
 
+  test('recursive jobs can only be queued once non-recursively', async () => {
+    const job: SchedulerJob = vi.fn()
+    job.id = 1
+    job.flags = SchedulerJobFlags.ALLOW_RECURSE
+
+    queueJob(job)
+    queueJob(job)
+
+    await nextTick()
+
+    expect(job).toHaveBeenCalledTimes(1)
+  })
+
+  test('recursive jobs can only be queued once recursively', async () => {
+    let recurse = true
+
+    const job: SchedulerJob = vi.fn(() => {
+      if (recurse) {
+        queueJob(job)
+        queueJob(job)
+        recurse = false
+      }
+    })
+    job.id = 1
+    job.flags = SchedulerJobFlags.ALLOW_RECURSE
+
+    queueJob(job)
+
+    await nextTick()
+
+    expect(job).toHaveBeenCalledTimes(2)
+  })
+
+  test(`recursive 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
+        queueJob(job2)
+        recurse = false
+      }
+    }
+    job1.id = 1
+
+    const job2: SchedulerJob = vi.fn(() => {
+      if (recurse) {
+        queueJob(job1)
+        queueJob(job2)
+      }
+    })
+    job2.id = 2
+    job2.flags = SchedulerJobFlags.ALLOW_RECURSE
+
+    queueJob(job2)
+
+    await nextTick()
+
+    expect(job2).toHaveBeenCalledTimes(2)
+  })
+
+  test('jobs are de-duplicated correctly when calling flushPreFlushCbs', async () => {
+    let recurse = true
+
+    const job1: SchedulerJob = vi.fn(() => {
+      queueJob(job3)
+      queueJob(job3)
+      flushPreFlushCbs()
+    })
+    job1.id = 1
+    job1.flags = SchedulerJobFlags.PRE
+
+    const job2: SchedulerJob = vi.fn(() => {
+      if (recurse) {
+        // job2 does not allow recurse, so this shouldn't do anything
+        queueJob(job2)
+
+        // job3 is already queued, so this shouldn't do anything
+        queueJob(job3)
+        recurse = false
+      }
+    })
+    job2.id = 2
+    job2.flags = SchedulerJobFlags.PRE
+
+    const job3: SchedulerJob = vi.fn(() => {
+      if (recurse) {
+        queueJob(job2)
+        queueJob(job3)
+
+        // The jobs are already queued, so these should have no effect
+        queueJob(job2)
+        queueJob(job3)
+      }
+    })
+    job3.id = 3
+    job3.flags = SchedulerJobFlags.ALLOW_RECURSE | SchedulerJobFlags.PRE
+
+    queueJob(job1)
+
+    await nextTick()
+
+    expect(job1).toHaveBeenCalledTimes(1)
+    expect(job2).toHaveBeenCalledTimes(1)
+    expect(job3).toHaveBeenCalledTimes(2)
+  })
+
   // #1947 flushPostFlushCbs should handle nested calls
   // e.g. app.mount inside app.mount
   test('flushPostFlushCbs', async () => {
diff --git a/packages/runtime-core/src/scheduler.ts b/packages/runtime-core/src/scheduler.ts
index 1eac06e5bf0..798f850b7de 100644
--- a/packages/runtime-core/src/scheduler.ts
+++ b/packages/runtime-core/src/scheduler.ts
@@ -162,7 +162,9 @@ export function flushPreFlushCbs(
         cb.flags! &= ~SchedulerJobFlags.QUEUED
       }
       cb()
-      cb.flags! &= ~SchedulerJobFlags.QUEUED
+      if (!(cb.flags! & SchedulerJobFlags.ALLOW_RECURSE)) {
+        cb.flags! &= ~SchedulerJobFlags.QUEUED
+      }
     }
   }
 }
@@ -239,7 +241,9 @@ function flushJobs(seen?: CountMap) {
           job.i,
           job.i ? ErrorCodes.COMPONENT_UPDATE : ErrorCodes.SCHEDULER,
         )
-        job.flags! &= ~SchedulerJobFlags.QUEUED
+        if (!(job.flags! & SchedulerJobFlags.ALLOW_RECURSE)) {
+          job.flags! &= ~SchedulerJobFlags.QUEUED
+        }
       }
     }
   } finally {