feat: allow interrupting repeating task (#3186)

Adds a `.run` method to a repeating task that lets us interrupt the
timeout to run immediately (debounced) before rescheduling the task
to run again in the future.

Author: achingbrain

packages/utils/src/repeating-task.ts

@@ -1,5 +1,6 @@
 import { anySignal } from 'any-signal'
 import { setMaxListeners } from 'main-event'
+import { debounce } from './debounce.ts'
 import type { AbortOptions } from '@libp2p/interface'
 
 export interface RepeatingTask {
@@ -25,6 +26,13 @@ export interface RepeatingTask {
    */
   setTimeout(ms: number): void
 
+  /**
+   * Schedule the task to be run immediately - if the task is not running it
+   * will run after a short delay in order to debounce multiple `.run()`
+   * invocations.
+   */
+  run(): void
+
   /**
    * Start the task running
    */
@@ -49,11 +57,20 @@ export interface RepeatingTaskOptions {
    * @default false
    */
   runImmediately?: boolean
+
+  /**
+   * When `.run()` is called to run the task outside of the current interval,
+   * debounce repeated calls to `.run()` by this amount.
+   *
+   * @default 100
+   */
+  debounce?: number
 }
 
 export function repeatingTask (fn: (options?: AbortOptions) => void | Promise<void>, interval: number, options?: RepeatingTaskOptions): RepeatingTask {
   let timeout: ReturnType<typeof setTimeout>
   let shutdownController: AbortController
+  let running = false
 
   function runTask (): void {
     const opts: AbortOptions = {
@@ -67,11 +84,15 @@ export function repeatingTask (fn: (options?: AbortOptions) => void | Promise<vo
       opts.signal = signal
     }
 
+    running = true
+
     Promise.resolve().then(async () => {
       await fn(opts)
     })
       .catch(() => {})
       .finally(() => {
+        running = false
+
         if (shutdownController.signal.aborted) {
           // task has been cancelled, bail
           return
@@ -82,6 +103,8 @@ export function repeatingTask (fn: (options?: AbortOptions) => void | Promise<vo
       })
   }
 
+  const runTaskDebounced = debounce(runTask, options?.debounce ?? 100)
+
   let started = false
 
   return {
@@ -103,6 +126,14 @@ export function repeatingTask (fn: (options?: AbortOptions) => void | Promise<vo
       options ??= {}
       options.timeout = ms
     },
+    run: (): void => {
+      if (running) {
+        return
+      }
+
+      clearTimeout(timeout)
+      runTaskDebounced()
+    },
     start: (): void => {
       if (started) {
         return

packages/utils/test/repeating-task.spec.ts

@@ -2,12 +2,19 @@ import { expect } from 'aegir/chai'
 import delay from 'delay'
 import pDefer from 'p-defer'
 import { repeatingTask } from '../src/repeating-task.js'
+import type { RepeatingTask } from '../src/repeating-task.js'
 
 describe('repeating-task', () => {
+  let task: RepeatingTask
+
+  afterEach(() => {
+    task?.stop()
+  })
+
   it('should repeat a task', async () => {
     let count = 0
 
-    const task = repeatingTask(() => {
+    task = repeatingTask(() => {
       count++
     }, 100)
     task.start()
@@ -22,7 +29,7 @@ describe('repeating-task', () => {
   it('should run a task immediately', async () => {
     let count = 0
 
-    const task = repeatingTask(() => {
+    task = repeatingTask(() => {
       count++
     }, 60000, {
       runImmediately: true
@@ -31,15 +38,13 @@ describe('repeating-task', () => {
 
     await delay(10)
 
-    task.stop()
-
     expect(count).to.equal(1)
   })
 
   it('should time out a task', async () => {
     const deferred = pDefer()
 
-    const task = repeatingTask((opts) => {
+    task = repeatingTask((opts) => {
       opts?.signal?.addEventListener('abort', () => {
         deferred.resolve()
       })
@@ -49,29 +54,26 @@ describe('repeating-task', () => {
     task.start()
 
     await deferred.promise
-    task.stop()
   })
 
   it('should repeat a task that throws', async () => {
     let count = 0
 
-    const task = repeatingTask(() => {
+    task = repeatingTask(() => {
       count++
       throw new Error('Urk!')
     }, 100)
     task.start()
 
     await delay(1000)
 
-    task.stop()
-
     expect(count).to.be.greaterThan(1)
   })
 
   it('should update the interval of a task', async () => {
     let count = 0
 
-    const task = repeatingTask(() => {
+    task = repeatingTask(() => {
       count++
 
       if (count === 1) {
@@ -82,15 +84,13 @@ describe('repeating-task', () => {
 
     await delay(1000)
 
-    task.stop()
-
     expect(count).to.equal(1)
   })
 
   it('should update the timeout of a task', async () => {
     let count = 0
 
-    const task = repeatingTask(async (options) => {
+    task = repeatingTask(async (options) => {
       // simulate a delay
       await delay(100)
 
@@ -109,15 +109,13 @@ describe('repeating-task', () => {
 
     await delay(1000)
 
-    task.stop()
-
     expect(count).to.equal(1)
   })
 
   it('should not reschedule the task if the interval is updated to the same value', async () => {
     let count = 0
 
-    const task = repeatingTask(() => {
+    task = repeatingTask(() => {
       count++
     }, 1_000, {
       runImmediately: true
@@ -134,7 +132,69 @@ describe('repeating-task', () => {
 
     await delay(100)
 
-    task.stop()
+    expect(count).to.equal(2)
+  })
+
+  it('should allow interrupting the timeout to run the task immediately', async () => {
+    let count = 0
+
+    task = repeatingTask(() => {
+      count++
+    }, 1_000)
+    task.start()
+
+    // run immediately
+    task.run()
+
+    // less than the repeat interval
+    await delay(200)
+
+    expect(count).to.equal(1)
+  })
+
+  it('should debounce interrupting the timeout to run the task immediately', async () => {
+    let count = 0
+
+    task = repeatingTask(() => {
+      count++
+    }, 1_000, {
+      debounce: 10
+    })
+    task.start()
+
+    // run immediately
+    task.run()
+    task.run()
+    task.run()
+    task.run()
+    task.run()
+
+    // less than the repeat interval
+    await delay(50)
+
+    expect(count).to.equal(1)
+  })
+
+  it('should schedule re-running the task after interrupting the timeout', async () => {
+    let count = 0
+
+    task = repeatingTask(() => {
+      count++
+    }, 100, {
+      debounce: 10
+    })
+    task.start()
+
+    // run immediately
+    task.run()
+
+    // less than the repeat interval
+    await delay(50)
+
+    expect(count).to.equal(1)
+
+    // wait longer than the repeat interval
+    await delay(150)
 
     expect(count).to.equal(2)
   })