refactor: properly cleanup invalidated jobs

packages/scheduler/__tests__/scheduler.spec.ts

@@ -1,4 +1,4 @@
-import { queueJob, nextTick } from '../src/index'
+import { queueJob, queuePostCommitCb, nextTick } from '../src/index'
 
 describe('scheduler', () => {
   it('queueJob', async () => {
@@ -32,13 +32,15 @@ describe('scheduler', () => {
     expect(calls).toEqual(['job1', 'job2'])
   })
 
-  it('queueJob w/ postFlushCb', async () => {
+  it('queueJob w/ postCommitCb', async () => {
     const calls: any = []
     const job1 = () => {
       calls.push('job1')
+      queuePostCommitCb(cb1)
     }
     const job2 = () => {
       calls.push('job2')
+      queuePostCommitCb(cb2)
     }
     const cb1 = () => {
       calls.push('cb1')
@@ -46,21 +48,24 @@ describe('scheduler', () => {
     const cb2 = () => {
       calls.push('cb2')
     }
-    queueJob(job1, cb1)
-    queueJob(job2, cb2)
+    queueJob(job1)
+    queueJob(job2)
     await nextTick()
-    expect(calls).toEqual(['job1', 'job2', 'cb1', 'cb2'])
+    // post commit cbs are called in reverse!
+    expect(calls).toEqual(['job1', 'job2', 'cb2', 'cb1'])
   })
 
   it('queueJob w/ postFlushCb while flushing', async () => {
     const calls: any = []
     const job1 = () => {
       calls.push('job1')
+      queuePostCommitCb(cb1)
       // job1 queues job2
-      queueJob(job2, cb2)
+      queueJob(job2)
     }
     const job2 = () => {
       calls.push('job2')
+      queuePostCommitCb(cb2)
     }
     const cb1 = () => {
       calls.push('cb1')
@@ -68,10 +73,10 @@ describe('scheduler', () => {
     const cb2 = () => {
       calls.push('cb2')
     }
-    queueJob(job1, cb1)
+    queueJob(job1)
     expect(calls).toEqual([])
     await nextTick()
-    expect(calls).toEqual(['job1', 'job2', 'cb1', 'cb2'])
+    expect(calls).toEqual(['job1', 'job2', 'cb2', 'cb1'])
   })
 
   it('should dedupe queued tasks', async () => {
@@ -91,26 +96,28 @@ describe('scheduler', () => {
     expect(calls).toEqual(['job1', 'job2'])
   })
 
-  it('queueJob inside postFlushCb', async () => {
+  it('queueJob inside postCommitCb', async () => {
     const calls: any = []
     const job1 = () => {
       calls.push('job1')
+      queuePostCommitCb(cb1)
     }
     const cb1 = () => {
       // queue another job in postFlushCb
       calls.push('cb1')
-      queueJob(job2, cb2)
+      queueJob(job2)
     }
     const job2 = () => {
       calls.push('job2')
+      queuePostCommitCb(cb2)
     }
     const cb2 = () => {
       calls.push('cb2')
     }
 
-    queueJob(job1, cb1)
-    queueJob(job2, cb2)
+    queueJob(job1)
+    queueJob(job2)
     await nextTick()
-    expect(calls).toEqual(['job1', 'job2', 'cb1', 'cb2', 'job2', 'cb2'])
+    expect(calls).toEqual(['job1', 'job2', 'cb2', 'cb1', 'job2', 'cb2'])
   })
 })

packages/scheduler/src/experimental.ts

@@ -1,8 +1,11 @@
+// TODO infinite updates detection
+
 import { Op, setCurrentOps } from './patchNodeOps'
 
 interface Job extends Function {
   ops: Op[]
-  post: Function | null
+  post: Function[]
+  cleanup: Function | null
   expiration: number
 }
 
@@ -12,6 +15,8 @@ const enum Priorities {
 
 type ErrorHandler = (err: Error) => any
 
+let currentJob: Job | null = null
+
 let start: number = 0
 const getNow = () => window.performance.now()
 const frameBudget = __JSDOM__ ? Infinity : 1000 / 60
@@ -85,10 +90,10 @@ export function handleSchedulerError(handler: ErrorHandler) {
 
 let hasPendingFlush = false
 
-export function queueJob(rawJob: Function, postJob?: Function | null) {
+export function queueJob(rawJob: Function) {
   const job = rawJob as Job
-  job.post = postJob || null
   job.ops = job.ops || []
+  job.post = job.post || []
   // 1. let's see if this invalidates any work that
   // has already been done.
   const commitIndex = commitQueue.indexOf(job)
@@ -118,6 +123,24 @@ export function queueJob(rawJob: Function, postJob?: Function | null) {
   }
 }
 
+export function queuePostCommitCb(fn: Function) {
+  if (currentJob) {
+    currentJob.post.push(fn)
+  } else {
+    postCommitQueue.push(fn)
+  }
+}
+
+export function flushPostCommitCbs() {
+  // post commit hooks (updated, mounted)
+  // this queue is flushed in reverse becuase these hooks should be invoked
+  // child first
+  let job
+  while ((job = postCommitQueue.pop())) {
+    job()
+  }
+}
+
 function flush(): void {
   let job
   while (true) {
@@ -139,14 +162,12 @@ function flush(): void {
     // all done, time to commit!
     while ((job = commitQueue.shift())) {
       commitJob(job)
-      if (job.post && postCommitQueue.indexOf(job.post) < 0) {
-        postCommitQueue.push(job.post)
+      if (job.post) {
+        postCommitQueue.push(...job.post)
+        job.post.length = 0
       }
     }
-    // post commit hooks (updated, mounted)
-    while ((job = postCommitQueue.shift())) {
-      job()
-    }
+    flushPostCommitCbs()
     // some post commit hook triggered more updates...
     if (patchQueue.length > 0) {
       if (!__COMPAT__ && getNow() - start > frameBudget) {
@@ -174,7 +195,9 @@ function patchJob(job: Job) {
   // job with existing ops means it's already been patched in a low priority queue
   if (job.ops.length === 0) {
     setCurrentOps(job.ops)
-    job()
+    currentJob = job
+    job.cleanup = job()
+    currentJob = null
     setCurrentOps(null)
     commitQueue.push(job)
   }
@@ -190,4 +213,9 @@ function commitJob({ ops }: Job) {
 
 function invalidateJob(job: Job) {
   job.ops.length = 0
+  job.post.length = 0
+  if (job.cleanup) {
+    job.cleanup()
+    job.cleanup = null
+  }
 }