fix(utils): guard onProgress cycles in Job dispatch (#3485)

* fix(utils): guard against re-entrant progress dispatch in Job

When two jobs share progress callbacks that form a cycle, a single
progress event would recurse synchronously through the recipients
forEach loop until V8 hit `Maximum call stack size exceeded` and the
process crashed. The cycle most often appears in DialQueue, where two
parallel dials of the same peer share a job via `join()` and propagate
progress events back to each other.

Add a per-Job `dispatchingProgress` flag so a synchronous re-entry into
the same job's synthesised onProgress short-circuits instead of
recursing. Non-cyclic dispatches behave identically; only the cycle is
broken.

Fixes #3484

Co-authored-by: tabcat <tabcat00@proton.me>
Co-authored-by: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Author: 3 people

packages/utils/src/queue/job.ts

@@ -27,6 +27,7 @@ export class Job <JobOptions extends AbortOptions & ProgressOptions = AbortOptio
   public status: JobStatus
   public readonly timeline: JobTimeline
   private readonly controller: AbortController
+  private dispatchingProgress: boolean
 
   constructor (fn: (options: JobOptions) => Promise<JobReturnType>, options: any) {
     this.id = randomId()
@@ -41,6 +42,8 @@ export class Job <JobOptions extends AbortOptions & ProgressOptions = AbortOptio
     this.controller = new AbortController()
     setMaxListeners(Infinity, this.controller.signal)
 
+    this.dispatchingProgress = false
+
     this.onAbort = this.onAbort.bind(this)
   }
 
@@ -80,9 +83,21 @@ export class Job <JobOptions extends AbortOptions & ProgressOptions = AbortOptio
         ...(this.options ?? {}),
         signal: this.controller.signal,
         onProgress: (evt: any): void => {
-          this.recipients.forEach(recipient => {
-            recipient.onProgress?.(evt)
-          })
+          // Recipients can transitively re-enter this dispatcher; without
+          // this guard a single event recurses until the stack overflows.
+          if (this.dispatchingProgress) {
+            return
+          }
+
+          this.dispatchingProgress = true
+
+          try {
+            this.recipients.forEach(recipient => {
+              recipient.onProgress?.(evt)
+            })
+          } finally {
+            this.dispatchingProgress = false
+          }
         }
       }), this.controller.signal)
 

packages/utils/test/queue.spec.ts

@@ -934,4 +934,229 @@ describe('queue', () => {
 
     expect(events).to.have.lengthOf(2)
   })
+
+  it('should not recurse infinitely when two jobs progress-feed each other', async () => {
+    interface ProgressJobOptions extends AbortOptions, ProgressOptions {
+
+    }
+
+    const queueA = new Queue<string, ProgressJobOptions>({ concurrency: 1 })
+    const queueB = new Queue<string, ProgressJobOptions>({ concurrency: 1 })
+
+    let aSynthOP: ((evt: any) => void) | undefined
+    let bSynthOP: ((evt: any) => void) | undefined
+
+    const aReady = pDefer<void>()
+    const bReady = pDefer<void>()
+    const aHold = pDefer<void>()
+    const bHold = pDefer<void>()
+
+    const eventsA: any[] = []
+    const eventsB: any[] = []
+
+    const pA = queueA.add(async (options) => {
+      aSynthOP = options.onProgress
+      aReady.resolve()
+      await aHold.promise
+      return 'a'
+    }, {
+      onProgress: (evt) => {
+        eventsA.push(evt)
+        bSynthOP?.(evt)
+      }
+    })
+
+    const pB = queueB.add(async (options) => {
+      bSynthOP = options.onProgress
+      bReady.resolve()
+      await bHold.promise
+      return 'b'
+    }, {
+      onProgress: (evt) => {
+        eventsB.push(evt)
+        aSynthOP?.(evt)
+      }
+    })
+
+    await Promise.all([aReady.promise, bReady.promise])
+
+    expect(() => {
+      aSynthOP?.(new CustomProgressEvent('kick'))
+    }).to.not.throw()
+
+    expect(eventsA).to.have.lengthOf(1)
+    expect(eventsB).to.have.lengthOf(1)
+
+    aHold.resolve()
+    bHold.resolve()
+
+    await Promise.all([pA, pB])
+  })
+
+  it('should not recurse infinitely on a 3-job triangle (A -> B -> C -> A)', async () => {
+    interface ProgressJobOptions extends AbortOptions, ProgressOptions {
+
+    }
+
+    const queueA = new Queue<string, ProgressJobOptions>({ concurrency: 1 })
+    const queueB = new Queue<string, ProgressJobOptions>({ concurrency: 1 })
+    const queueC = new Queue<string, ProgressJobOptions>({ concurrency: 1 })
+
+    let aSynthOP: ((evt: any) => void) | undefined
+    let bSynthOP: ((evt: any) => void) | undefined
+    let cSynthOP: ((evt: any) => void) | undefined
+
+    const aReady = pDefer<void>()
+    const bReady = pDefer<void>()
+    const cReady = pDefer<void>()
+    const hold = pDefer<void>()
+
+    const eventsA: any[] = []
+    const eventsB: any[] = []
+    const eventsC: any[] = []
+
+    const pA = queueA.add(async (options) => {
+      aSynthOP = options.onProgress
+      aReady.resolve()
+      await hold.promise
+      return 'a'
+    }, {
+      onProgress: (evt) => {
+        eventsA.push(evt)
+        bSynthOP?.(evt)
+      }
+    })
+
+    const pB = queueB.add(async (options) => {
+      bSynthOP = options.onProgress
+      bReady.resolve()
+      await hold.promise
+      return 'b'
+    }, {
+      onProgress: (evt) => {
+        eventsB.push(evt)
+        cSynthOP?.(evt)
+      }
+    })
+
+    const pC = queueC.add(async (options) => {
+      cSynthOP = options.onProgress
+      cReady.resolve()
+      await hold.promise
+      return 'c'
+    }, {
+      onProgress: (evt) => {
+        eventsC.push(evt)
+        aSynthOP?.(evt)
+      }
+    })
+
+    await Promise.all([aReady.promise, bReady.promise, cReady.promise])
+
+    expect(() => {
+      aSynthOP?.(new CustomProgressEvent('kick'))
+    }).to.not.throw()
+
+    expect(eventsA).to.have.lengthOf(1)
+    expect(eventsB).to.have.lengthOf(1)
+    expect(eventsC).to.have.lengthOf(1)
+
+    hold.resolve()
+    await Promise.all([pA, pB, pC])
+  })
+
+  it('should keep dispatching after a previous cycle completes', async () => {
+    interface ProgressJobOptions extends AbortOptions, ProgressOptions {
+
+    }
+
+    const queueA = new Queue<string, ProgressJobOptions>({ concurrency: 1 })
+    const queueB = new Queue<string, ProgressJobOptions>({ concurrency: 1 })
+
+    let aSynthOP: ((evt: any) => void) | undefined
+    let bSynthOP: ((evt: any) => void) | undefined
+
+    const aReady = pDefer<void>()
+    const bReady = pDefer<void>()
+    const hold = pDefer<void>()
+
+    const eventsA: any[] = []
+    const eventsB: any[] = []
+
+    const pA = queueA.add(async (options) => {
+      aSynthOP = options.onProgress
+      aReady.resolve()
+      await hold.promise
+      return 'a'
+    }, {
+      onProgress: (evt) => {
+        eventsA.push(evt)
+        bSynthOP?.(evt)
+      }
+    })
+
+    const pB = queueB.add(async (options) => {
+      bSynthOP = options.onProgress
+      bReady.resolve()
+      await hold.promise
+      return 'b'
+    }, {
+      onProgress: (evt) => {
+        eventsB.push(evt)
+        aSynthOP?.(evt)
+      }
+    })
+
+    await Promise.all([aReady.promise, bReady.promise])
+
+    aSynthOP?.(new CustomProgressEvent('first'))
+    aSynthOP?.(new CustomProgressEvent('second'))
+
+    expect(eventsA.map(e => e.type)).to.deep.equal(['first', 'second'])
+    expect(eventsB.map(e => e.type)).to.deep.equal(['first', 'second'])
+
+    hold.resolve()
+    await Promise.all([pA, pB])
+  })
+
+  it('resets the dispatch flag even if a recipient throws', async () => {
+    interface ProgressJobOptions extends AbortOptions, ProgressOptions {
+
+    }
+
+    const queue = new Queue<string, ProgressJobOptions>({ concurrency: 1 })
+
+    let synthOP: ((evt: any) => void) | undefined
+
+    const ready = pDefer<void>()
+    const hold = pDefer<void>()
+
+    const seen: any[] = []
+    let throwOnce = true
+
+    const p = queue.add(async (options) => {
+      synthOP = options.onProgress
+      ready.resolve()
+      await hold.promise
+      return 'done'
+    }, {
+      onProgress: (evt) => {
+        seen.push(evt)
+        if (throwOnce) {
+          throwOnce = false
+          throw new Error('boom')
+        }
+      }
+    })
+
+    await ready.promise
+
+    expect(() => synthOP?.(new CustomProgressEvent('first'))).to.throw('boom')
+    expect(() => synthOP?.(new CustomProgressEvent('second'))).to.not.throw()
+
+    expect(seen.map(e => e.type)).to.deep.equal(['first', 'second'])
+
+    hold.resolve()
+    await p
+  })
 })